Unverified Commit 8b857eef authored by Kohei Tokunaga's avatar Kohei Tokunaga Committed by GitHub

Ship Stargz Snapshotter (#2936)

* Ship Stargz Snapshotter Signed-off-by: 's avatarktock <ktokunaga.mail@gmail.com> * Bump github.com/containerd/stargz-snapshotter to v0.8.0 Signed-off-by: 's avatarKohei Tokunaga <ktokunaga.mail@gmail.com>
parent cf12a131
......@@ -77,9 +77,10 @@ require (
github.com/containerd/cgroups v1.0.1
github.com/containerd/containerd v1.5.5
github.com/containerd/fuse-overlayfs-snapshotter v1.0.3
github.com/containerd/stargz-snapshotter v0.8.0
github.com/coreos/go-iptables v0.5.0
github.com/coreos/go-systemd v0.0.0-20190719114852-fd7a80b32e1f
github.com/docker/docker v20.10.5+incompatible
github.com/docker/docker v20.10.7+incompatible
github.com/erikdubbelboer/gspt v0.0.0-20190125194910-e68493906b83
github.com/flannel-io/flannel v0.14.0
github.com/go-bindata/go-bindata v3.1.2+incompatible
......@@ -91,7 +92,7 @@ require (
github.com/gorilla/websocket v1.4.2
github.com/k3s-io/helm-controller v0.11.3
github.com/k3s-io/kine v0.8.0
github.com/klauspost/compress v1.12.2
github.com/klauspost/compress v1.13.5
github.com/kubernetes-sigs/cri-tools v0.0.0-00010101000000-000000000000
github.com/lib/pq v1.10.2
github.com/mattn/go-sqlite3 v1.14.8
......@@ -123,7 +124,7 @@ require (
golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2
golang.org/x/net v0.0.0-20210520170846-37e1c6afe023
golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c
google.golang.org/grpc v1.38.0
google.golang.org/grpc v1.40.0
gopkg.in/yaml.v2 v2.4.0
inet.af/tcpproxy v0.0.0-20200125044825-b6bb9b5b8252
k8s.io/api v0.22.1
......
......@@ -442,6 +442,12 @@ func get(ctx context.Context, envInfo *cmds.Agent, proxy proxy.Proxy) (*config.N
return nil, errors.Wrapf(err, "\"fuse-overlayfs\" snapshotter cannot be enabled for %q, try using \"native\"",
nodeConfig.Containerd.Root)
}
case "stargz":
if err := containerd.StargzSupported(nodeConfig.Containerd.Root); err != nil {
return nil, errors.Wrapf(err, "\"stargz\" snapshotter cannot be enabled for %q, try using \"overlayfs\" or \"native\"",
nodeConfig.Containerd.Root)
}
nodeConfig.AgentConfig.ImageServiceSocket = "/run/containerd-stargz-grpc/containerd-stargz-grpc.sock"
}
}
nodeConfig.Containerd.Opt = filepath.Join(envInfo.DataDir, "agent", "containerd")
......
......@@ -30,8 +30,46 @@ const ContainerdConfigTemplate = `
{{- if .NodeConfig.AgentConfig.Snapshotter }}
[plugins.cri.containerd]
disable_snapshot_annotations = true
snapshotter = "{{ .NodeConfig.AgentConfig.Snapshotter }}"
disable_snapshot_annotations = {{ if eq .NodeConfig.AgentConfig.Snapshotter "stargz" }}false{{else}}true{{end}}
{{ if eq .NodeConfig.AgentConfig.Snapshotter "stargz" }}
{{ if .NodeConfig.AgentConfig.ImageServiceSocket }}
[plugins.stargz]
cri_keychain_image_service_path = "{{ .NodeConfig.AgentConfig.ImageServiceSocket }}"
[plugins.stargz.cri_keychain]
enable_keychain = true
{{end}}
{{ if .PrivateRegistryConfig }}
{{ if .PrivateRegistryConfig.Mirrors }}
[plugins.stargz.registry.mirrors]{{end}}
{{range $k, $v := .PrivateRegistryConfig.Mirrors }}
[plugins.stargz.registry.mirrors."{{$k}}"]
endpoint = [{{range $i, $j := $v.Endpoints}}{{if $i}}, {{end}}{{printf "%q" .}}{{end}}]
{{if $v.Rewrites}}
[plugins.stargz.registry.mirrors."{{$k}}".rewrite]
{{range $pattern, $replace := $v.Rewrites}}
"{{$pattern}}" = "{{$replace}}"
{{end}}
{{end}}
{{end}}
{{range $k, $v := .PrivateRegistryConfig.Configs }}
{{ if $v.Auth }}
[plugins.stargz.registry.configs."{{$k}}".auth]
{{ if $v.Auth.Username }}username = {{ printf "%q" $v.Auth.Username }}{{end}}
{{ if $v.Auth.Password }}password = {{ printf "%q" $v.Auth.Password }}{{end}}
{{ if $v.Auth.Auth }}auth = {{ printf "%q" $v.Auth.Auth }}{{end}}
{{ if $v.Auth.IdentityToken }}identitytoken = {{ printf "%q" $v.Auth.IdentityToken }}{{end}}
{{end}}
{{ if $v.TLS }}
[plugins.stargz.registry.configs."{{$k}}".tls]
{{ if $v.TLS.CAFile }}ca_file = "{{ $v.TLS.CAFile }}"{{end}}
{{ if $v.TLS.CertFile }}cert_file = "{{ $v.TLS.CertFile }}"{{end}}
{{ if $v.TLS.KeyFile }}key_file = "{{ $v.TLS.KeyFile }}"{{end}}
{{ if $v.TLS.InsecureSkipVerify }}insecure_skip_verify = true{{end}}
{{end}}
{{end}}
{{end}}
{{end}}
{{end}}
{{- if not .NodeConfig.NoFlannel }}
......
......@@ -25,4 +25,5 @@ import (
_ "github.com/containerd/containerd/snapshots/native/plugin"
_ "github.com/containerd/containerd/snapshots/overlay/plugin"
_ "github.com/containerd/fuse-overlayfs-snapshotter/plugin"
_ "github.com/containerd/stargz-snapshotter/service/plugin"
)
......@@ -5,6 +5,7 @@ package containerd
import (
"github.com/containerd/containerd/snapshots/overlay/overlayutils"
fuseoverlayfs "github.com/containerd/fuse-overlayfs-snapshotter"
stargz "github.com/containerd/stargz-snapshotter/service"
)
func OverlaySupported(root string) error {
......@@ -14,3 +15,7 @@ func OverlaySupported(root string) error {
func FuseoverlayfsSupported(root string) error {
return fuseoverlayfs.Supported(root)
}
func StargzSupported(root string) error {
return stargz.Supported(root)
}
......@@ -108,6 +108,13 @@ func kubeletArgs(cfg *config.Agent) map[string]string {
} else if cfg.PauseImage != "" {
argsMap["pod-infra-container-image"] = cfg.PauseImage
}
if cfg.ImageServiceSocket != "" {
if strings.HasPrefix(cfg.ImageServiceSocket, unixPrefix) {
argsMap["image-service-endpoint"] = cfg.ImageServiceSocket
} else {
argsMap["image-service-endpoint"] = unixPrefix + cfg.ImageServiceSocket
}
}
if cfg.ListenAddress != "" {
argsMap["address"] = cfg.ListenAddress
}
......
......@@ -76,6 +76,7 @@ type Agent struct {
NodeExternalIP string
NodeExternalIPs []net.IP
RuntimeSocket string
ImageServiceSocket string
ListenAddress string
ClientCA string
CNIBinDir string
......
......@@ -21,6 +21,8 @@ echo "Did test-run-basics $?"
. ./scripts/test-run-compat
echo "Did test-run-compat $?"
. ./scripts/test-run-lazypull
echo "Did test-run-lazypull $?"
# ---
......
#!/bin/bash
all_services=(
coredns
local-path-provisioner
metrics-server
traefik
)
export NUM_SERVERS=1
export NUM_AGENTS=1
export WAIT_SERVICES="${all_services[@]}"
export SERVER_ARGS="--node-taint=CriticalAddonsOnly=true:NoExecute"
# ---
cluster-pre-hook() {
export SERVER_ARGS="${SERVER_ARGS}
--snapshotter=stargz
"
export AGENT_ARGS="${AGENT_ARGS}
--snapshotter=stargz
"
}
export -f cluster-pre-hook
# ---
start-test() {
local REMOTE_SNAPSHOT_LABEL="containerd.io/snapshot/remote"
local TEST_IMAGE="ghcr.io/stargz-containers/k3s-test-ubuntu:20.04-esgz"
local TEST_POD_NAME=testpod-$(head /dev/urandom | tr -dc a-z0-9 | head -c 10)
local TEST_CONTAINER_NAME=testcontainer-$(head /dev/urandom | tr -dc a-z0-9 | head -c 10)
# Create the target Pod
cat <<EOF | kubectl apply -f -
apiVersion: v1
kind: Pod
metadata:
name: ${TEST_POD_NAME}
spec:
containers:
- name: ${TEST_CONTAINER_NAME}
image: ${TEST_IMAGE}
command: ["sleep"]
args: ["infinity"]
EOF
wait-for-pod "${TEST_POD_NAME}"
# Check if all layers are remote snapshots
NODE=$(kubectl get pods "${TEST_POD_NAME}" -ojsonpath='{.spec.nodeName}')
LAYER=$(get-topmost-layer "${NODE}" "${TEST_CONTAINER_NAME}")
LAYERSNUM=0
for (( ; ; )) ; do
LAYER=$(docker exec -i "${NODE}" ctr --namespace="k8s.io" snapshot --snapshotter=stargz info "${LAYER}" | jq -r '.Parent')
if [ "${LAYER}" == "null" ] ; then
break
elif [ ${LAYERSNUM} -gt 100 ] ; then
echo "testing image contains too many layes > 100"
return 1
fi
((LAYERSNUM+=1))
LABEL=$(docker exec -i "${NODE}" ctr --namespace="k8s.io" snapshots --snapshotter=stargz info "${LAYER}" \
| jq -r ".Labels.\"${REMOTE_SNAPSHOT_LABEL}\"")
echo "Checking layer ${LAYER} : ${LABEL}"
if [ "${LABEL}" == "null" ] ; then
echo "layer ${LAYER} isn't remote snapshot"
return 1
fi
done
if [ ${LAYERSNUM} -eq 0 ] ; then
echo "cannot get layers"
return 1
fi
return 0
}
export -f start-test
wait-for-pod() {
local POD_NAME="${1}"
if [ "${POD_NAME}" == "" ] ; then
return 1
fi
IDX=0
DEADLINE=120
for (( ; ; )) ; do
STATUS=$(kubectl get pods "${POD_NAME}" -o 'jsonpath={..status.containerStatuses[0].state.running.startedAt}${..status.containerStatuses[0].state.waiting.reason}')
echo "Status: ${STATUS}"
STARTEDAT=$(echo "${STATUS}" | cut -f 1 -d '$')
if [ "${STARTEDAT}" != "" ] ; then
echo "Pod created"
break
elif [ ${IDX} -gt ${DEADLINE} ] ; then
echo "Deadline exeeded to wait for pod creation"
return 1
fi
((IDX+=1))
sleep 1
done
return 0
}
export -f wait-for-pod
get-topmost-layer() {
local NODE="${1}"
local CONTAINER="${2}"
local TARGET_CONTAINER=
if [ "${NODE}" == "" ] || [ "${CONTAINER}" == "" ] ; then
return 1
fi
for (( RETRY=1; RETRY<=50; RETRY++ )) ; do
TARGET_CONTAINER=$(docker exec -i "${NODE}" ctr --namespace="k8s.io" c ls -q labels."io.kubernetes.container.name"=="${CONTAINER}" | sed -n 1p)
if [ "${TARGET_CONTAINER}" != "" ] ; then
break
fi
sleep 3
done
if [ "${TARGET_CONTAINER}" == "" ] ; then
return 1
fi
LAYER=$(docker exec -i "${NODE}" ctr --namespace="k8s.io" c info "${TARGET_CONTAINER}" | jq -r '.SnapshotKey')
echo "${LAYER}"
}
export -f get-topmost-layer
# --- create a basic cluster and check for lazy pulling
LABEL=LAZYPULL run-test
The source code developed under the Stargz Snapshotter Project is licensed under Apache License 2.0.
However, the Stargz Snapshotter project contains modified subcomponents from Container Registry Filesystem Project with separate copyright notices and license terms. Your use of the source code for the subcomponent is subject to the terms and conditions as defined by the source project. Files in these subcomponents contain following file header.
```
Copyright 2019 The Go Authors. All rights reserved.
Use of this source code is governed by a BSD-style
license that can be found in the NOTICE.md file.
```
These source code is governed by a 3-Clause BSD license. The copyright notice, list of conditions and disclaimer are the following.
```
Copyright (c) 2019 Google LLC. All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:
* Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above
copyright notice, this list of conditions and the following disclaimer
in the documentation and/or other materials provided with the
distribution.
* Neither the name of Google Inc. nor the names of its
contributors may be used to endorse or promote products derived from
this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
```
The Stargz Snapshotter project also contains modified benchmarking code from HelloBench Project with separate copyright notices and license terms. Your use of the source code for the benchmarking code is subject to the terms and conditions as defined by the source project. These source code is governed by a MIT license. The copyright notice, condition and disclaimer are the following. The file in the benchmarking code contains it as the file header.
```
The MIT License (MIT)
Copyright (c) 2015 Tintri
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
```
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package errorutil
import (
"errors"
"fmt"
"strings"
)
// Aggregate combines a list of errors into a single new error.
func Aggregate(errs []error) error {
switch len(errs) {
case 0:
return nil
case 1:
return errs[0]
default:
points := make([]string, len(errs)+1)
points[0] = fmt.Sprintf("%d error(s) occurred:", len(errs))
for i, err := range errs {
points[i+1] = fmt.Sprintf("* %s", err)
}
return errors.New(strings.Join(points, "\n\t"))
}
}
module github.com/containerd/stargz-snapshotter/estargz
go 1.16
require (
github.com/klauspost/compress v1.13.5
github.com/opencontainers/go-digest v1.0.0
github.com/pkg/errors v0.9.1
golang.org/x/sync v0.0.0-20201207232520-09787c993a3a
)
github.com/klauspost/compress v1.13.5 h1:9O69jUPDcsT9fEm74W92rZL9FQY7rCdaXVneq+yyzl4=
github.com/klauspost/compress v1.13.5/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
golang.org/x/sync v0.0.0-20201207232520-09787c993a3a h1:DcqTD9SDLc+1P/r1EmRBwnVsrOwW+kk2vWf9n+1sGhs=
golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
/*
Copyright 2019 The Go Authors. All rights reserved.
Use of this source code is governed by a BSD-style
license that can be found in the LICENSE file.
*/
package estargz
import (
"archive/tar"
"bytes"
"compress/gzip"
"encoding/binary"
"encoding/json"
"fmt"
"hash"
"io"
"strconv"
digest "github.com/opencontainers/go-digest"
"github.com/pkg/errors"
)
type gzipCompression struct {
*gzipCompressor
*GzipDecompressor
}
func newGzipCompressionWithLevel(level int) Compression {
return &gzipCompression{
&gzipCompressor{level},
&GzipDecompressor{},
}
}
func NewGzipCompressorWithLevel(level int) Compressor {
return &gzipCompressor{level}
}
type gzipCompressor struct {
compressionLevel int
}
func (gc *gzipCompressor) Writer(w io.Writer) (io.WriteCloser, error) {
return gzip.NewWriterLevel(w, gc.compressionLevel)
}
func (gc *gzipCompressor) WriteTOCAndFooter(w io.Writer, off int64, toc *JTOC, diffHash hash.Hash) (digest.Digest, error) {
tocJSON, err := json.MarshalIndent(toc, "", "\t")
if err != nil {
return "", err
}
gz, _ := gzip.NewWriterLevel(w, gc.compressionLevel)
gw := io.Writer(gz)
if diffHash != nil {
gw = io.MultiWriter(gz, diffHash)
}
tw := tar.NewWriter(gw)
if err := tw.WriteHeader(&tar.Header{
Typeflag: tar.TypeReg,
Name: TOCTarName,
Size: int64(len(tocJSON)),
}); err != nil {
return "", err
}
if _, err := tw.Write(tocJSON); err != nil {
return "", err
}
if err := tw.Close(); err != nil {
return "", err
}
if err := gz.Close(); err != nil {
return "", err
}
if _, err := w.Write(gzipFooterBytes(off)); err != nil {
return "", err
}
return digest.FromBytes(tocJSON), nil
}
// gzipFooterBytes returns the 51 bytes footer.
func gzipFooterBytes(tocOff int64) []byte {
buf := bytes.NewBuffer(make([]byte, 0, FooterSize))
gz, _ := gzip.NewWriterLevel(buf, gzip.NoCompression) // MUST be NoCompression to keep 51 bytes
// Extra header indicating the offset of TOCJSON
// https://tools.ietf.org/html/rfc1952#section-2.3.1.1
header := make([]byte, 4)
header[0], header[1] = 'S', 'G'
subfield := fmt.Sprintf("%016xSTARGZ", tocOff)
binary.LittleEndian.PutUint16(header[2:4], uint16(len(subfield))) // little-endian per RFC1952
gz.Header.Extra = append(header, []byte(subfield)...)
gz.Close()
if buf.Len() != FooterSize {
panic(fmt.Sprintf("footer buffer = %d, not %d", buf.Len(), FooterSize))
}
return buf.Bytes()
}
type GzipDecompressor struct{}
func (gz *GzipDecompressor) Reader(r io.Reader) (io.ReadCloser, error) {
return gzip.NewReader(r)
}
func (gz *GzipDecompressor) ParseTOC(r io.Reader) (toc *JTOC, tocDgst digest.Digest, err error) {
return parseTOCEStargz(r)
}
func (gz *GzipDecompressor) ParseFooter(p []byte) (tocOffset, tocSize int64, err error) {
if len(p) != FooterSize {
return 0, 0, fmt.Errorf("invalid length %d cannot be parsed", len(p))
}
zr, err := gzip.NewReader(bytes.NewReader(p))
if err != nil {
return 0, 0, err
}
defer zr.Close()
extra := zr.Header.Extra
si1, si2, subfieldlen, subfield := extra[0], extra[1], extra[2:4], extra[4:]
if si1 != 'S' || si2 != 'G' {
return 0, 0, fmt.Errorf("invalid subfield IDs: %q, %q; want E, S", si1, si2)
}
if slen := binary.LittleEndian.Uint16(subfieldlen); slen != uint16(16+len("STARGZ")) {
return 0, 0, fmt.Errorf("invalid length of subfield %d; want %d", slen, 16+len("STARGZ"))
}
if string(subfield[16:]) != "STARGZ" {
return 0, 0, fmt.Errorf("STARGZ magic string must be included in the footer subfield")
}
tocOffset, err = strconv.ParseInt(string(subfield[:16]), 16, 64)
if err != nil {
return 0, 0, errors.Wrapf(err, "legacy: failed to parse toc offset")
}
return tocOffset, 0, nil
}
func (gz *GzipDecompressor) FooterSize() int64 {
return FooterSize
}
type legacyGzipDecompressor struct{}
func (gz *legacyGzipDecompressor) Reader(r io.Reader) (io.ReadCloser, error) {
return gzip.NewReader(r)
}
func (gz *legacyGzipDecompressor) ParseTOC(r io.Reader) (toc *JTOC, tocDgst digest.Digest, err error) {
return parseTOCEStargz(r)
}
func (gz *legacyGzipDecompressor) ParseFooter(p []byte) (tocOffset, tocSize int64, err error) {
if len(p) != legacyFooterSize {
return 0, 0, fmt.Errorf("legacy: invalid length %d cannot be parsed", len(p))
}
zr, err := gzip.NewReader(bytes.NewReader(p))
if err != nil {
return 0, 0, errors.Wrapf(err, "legacy: failed to get footer gzip reader")
}
defer zr.Close()
extra := zr.Header.Extra
if len(extra) != 16+len("STARGZ") {
return 0, 0, fmt.Errorf("legacy: invalid stargz's extra field size")
}
if string(extra[16:]) != "STARGZ" {
return 0, 0, fmt.Errorf("legacy: magic string STARGZ not found")
}
tocOffset, err = strconv.ParseInt(string(extra[:16]), 16, 64)
if err != nil {
return 0, 0, errors.Wrapf(err, "legacy: failed to parse toc offset")
}
return tocOffset, 0, nil
}
func (gz *legacyGzipDecompressor) FooterSize() int64 {
return legacyFooterSize
}
func parseTOCEStargz(r io.Reader) (toc *JTOC, tocDgst digest.Digest, err error) {
zr, err := gzip.NewReader(r)
if err != nil {
return nil, "", fmt.Errorf("malformed TOC gzip header: %v", err)
}
defer zr.Close()
zr.Multistream(false)
tr := tar.NewReader(zr)
h, err := tr.Next()
if err != nil {
return nil, "", fmt.Errorf("failed to find tar header in TOC gzip stream: %v", err)
}
if h.Name != TOCTarName {
return nil, "", fmt.Errorf("TOC tar entry had name %q; expected %q", h.Name, TOCTarName)
}
dgstr := digest.Canonical.Digester()
toc = new(JTOC)
if err := json.NewDecoder(io.TeeReader(tr, dgstr.Hash())).Decode(&toc); err != nil {
return nil, "", fmt.Errorf("error decoding TOC JSON: %v", err)
}
return toc, dgstr.Digest(), nil
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package zstdchunked
import (
"bytes"
"encoding/binary"
"encoding/json"
"fmt"
"hash"
"io"
"sync"
"github.com/containerd/stargz-snapshotter/estargz"
"github.com/klauspost/compress/zstd"
digest "github.com/opencontainers/go-digest"
"github.com/pkg/errors"
)
const (
ZstdChunkedManifestChecksumAnnotation = "io.containers.zstd-chunked.manifest-checksum"
ZstdChunkedManifestPositionAnnotation = "io.containers.zstd-chunked.manifest-position"
FooterSize = 40
manifestTypeCRFS = 1
)
var (
skippableFrameMagic = []byte{0x50, 0x2a, 0x4d, 0x18}
zstdFrameMagic = []byte{0x28, 0xb5, 0x2f, 0xfd}
zstdChunkedFrameMagic = []byte{0x47, 0x6e, 0x55, 0x6c, 0x49, 0x6e, 0x55, 0x78}
)
type Decompressor struct{}
func (zz *Decompressor) Reader(r io.Reader) (io.ReadCloser, error) {
decoder, err := zstd.NewReader(r)
if err != nil {
return nil, err
}
return &zstdReadCloser{decoder}, nil
}
func (zz *Decompressor) ParseTOC(r io.Reader) (toc *estargz.JTOC, tocDgst digest.Digest, err error) {
zr, err := zstd.NewReader(r)
if err != nil {
return nil, "", err
}
defer zr.Close()
dgstr := digest.Canonical.Digester()
toc = new(estargz.JTOC)
if err := json.NewDecoder(io.TeeReader(zr, dgstr.Hash())).Decode(&toc); err != nil {
return nil, "", errors.Wrap(err, "error decoding TOC JSON")
}
return toc, dgstr.Digest(), nil
}
func (zz *Decompressor) ParseFooter(p []byte) (tocOffset, tocSize int64, err error) {
offset := binary.LittleEndian.Uint64(p[0:8])
compressedLength := binary.LittleEndian.Uint64(p[8:16])
if !bytes.Equal(zstdChunkedFrameMagic, p[32:40]) {
return 0, 0, fmt.Errorf("invalid magic number")
}
return int64(offset), int64(compressedLength), nil
}
func (zz *Decompressor) FooterSize() int64 {
return FooterSize
}
type zstdReadCloser struct{ *zstd.Decoder }
func (z *zstdReadCloser) Close() error {
z.Decoder.Close()
return nil
}
type Compressor struct {
CompressionLevel zstd.EncoderLevel
Metadata map[string]string
pool sync.Pool
}
func (zc *Compressor) Writer(w io.Writer) (io.WriteCloser, error) {
if wc := zc.pool.Get(); wc != nil {
ec := wc.(*zstd.Encoder)
ec.Reset(w)
return &poolEncoder{ec, zc}, nil
}
ec, err := zstd.NewWriter(w, zstd.WithEncoderLevel(zc.CompressionLevel), zstd.WithLowerEncoderMem(true))
if err != nil {
return nil, err
}
return &poolEncoder{ec, zc}, nil
}
type poolEncoder struct {
*zstd.Encoder
zc *Compressor
}
func (w *poolEncoder) Close() error {
if err := w.Encoder.Close(); err != nil {
return err
}
w.zc.pool.Put(w.Encoder)
return nil
}
func (zc *Compressor) WriteTOCAndFooter(w io.Writer, off int64, toc *estargz.JTOC, diffHash hash.Hash) (digest.Digest, error) {
tocJSON, err := json.MarshalIndent(toc, "", "\t")
if err != nil {
return "", err
}
buf := new(bytes.Buffer)
encoder, err := zstd.NewWriter(buf, zstd.WithEncoderLevel(zc.CompressionLevel))
if err != nil {
return "", err
}
if _, err := encoder.Write(tocJSON); err != nil {
return "", err
}
if err := encoder.Close(); err != nil {
return "", err
}
compressedTOC := buf.Bytes()
_, err = io.Copy(w, bytes.NewReader(appendSkippableFrameMagic(compressedTOC)))
// 8 is the size of the zstd skippable frame header + the frame size
tocOff := uint64(off) + 8
if _, err := w.Write(appendSkippableFrameMagic(
zstdFooterBytes(tocOff, uint64(len(tocJSON)), uint64(len(compressedTOC)))),
); err != nil {
return "", err
}
if zc.Metadata != nil {
zc.Metadata[ZstdChunkedManifestChecksumAnnotation] = digest.FromBytes(compressedTOC).String()
zc.Metadata[ZstdChunkedManifestPositionAnnotation] = fmt.Sprintf("%d:%d:%d:%d",
tocOff, len(compressedTOC), len(tocJSON), manifestTypeCRFS)
}
return digest.FromBytes(tocJSON), err
}
// zstdFooterBytes returns the 40 bytes footer.
func zstdFooterBytes(tocOff, tocRawSize, tocCompressedSize uint64) []byte {
footer := make([]byte, FooterSize)
binary.LittleEndian.PutUint64(footer, tocOff)
binary.LittleEndian.PutUint64(footer[8:], tocCompressedSize)
binary.LittleEndian.PutUint64(footer[16:], tocRawSize)
binary.LittleEndian.PutUint64(footer[24:], manifestTypeCRFS)
copy(footer[32:40], zstdChunkedFrameMagic)
return footer
}
func appendSkippableFrameMagic(b []byte) []byte {
size := make([]byte, 4)
binary.LittleEndian.PutUint32(size, uint32(len(b)))
return append(append(skippableFrameMagic, size...), b...)
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
/*
Copyright 2019 The Go Authors. All rights reserved.
Use of this source code is governed by a BSD-style
license that can be found in the NOTICE.md file.
*/
package config
const (
// TargetSkipVerifyLabel is a snapshot label key that indicates to skip content
// verification for the layer.
TargetSkipVerifyLabel = "containerd.io/snapshot/remote/stargz.skipverify"
// TargetPrefetchSizeLabel is a snapshot label key that indicates size to prefetch
// the layer. If the layer is eStargz and contains prefetch landmarks, these config
// will be respeced.
TargetPrefetchSizeLabel = "containerd.io/snapshot/remote/stargz.prefetch"
)
type Config struct {
HTTPCacheType string `toml:"http_cache_type"`
FSCacheType string `toml:"filesystem_cache_type"`
ResolveResultEntry int `toml:"resolve_result_entry"`
PrefetchSize int64 `toml:"prefetch_size"`
PrefetchTimeoutSec int64 `toml:"prefetch_timeout_sec"`
NoPrefetch bool `toml:"noprefetch"`
NoBackgroundFetch bool `toml:"no_background_fetch"`
Debug bool `toml:"debug"`
AllowNoVerification bool `toml:"allow_no_verification"`
DisableVerification bool `toml:"disable_verification"`
MaxConcurrency int64 `toml:"max_concurrency"`
NoPrometheus bool `toml:"no_prometheus"`
// BlobConfig is config for layer blob management.
BlobConfig `toml:"blob"`
// DirectoryCacheConfig is config for directory-based cache.
DirectoryCacheConfig `toml:"directory_cache"`
FuseConfig `toml:"fuse"`
}
type BlobConfig struct {
ValidInterval int64 `toml:"valid_interval"`
CheckAlways bool `toml:"check_always"`
// ChunkSize is the granularity at which background fetch and on-demand reads
// are fetched from the remote registry.
ChunkSize int64 `toml:"chunk_size"`
FetchTimeoutSec int64 `toml:"fetching_timeout_sec"`
ForceSingleRangeMode bool `toml:"force_single_range_mode"`
// PrefetchChunkSize is the maximum bytes transferred per http GET from remote registry
// during prefetch. It is recommended to have PrefetchChunkSize > ChunkSize.
// If PrefetchChunkSize < ChunkSize prefetch bytes will be fetched as a single http GET,
// else total GET requests for prefetch = ceil(PrefetchSize / PrefetchChunkSize).
PrefetchChunkSize int64 `toml:"prefetch_chunk_size"`
}
type DirectoryCacheConfig struct {
MaxLRUCacheEntry int `toml:"max_lru_cache_entry"`
MaxCacheFds int `toml:"max_cache_fds"`
SyncAdd bool `toml:"sync_add"`
Direct bool `toml:"direct"`
}
type FuseConfig struct {
// AttrTimeout defines overall timeout attribute for a file system in seconds.
AttrTimeout int64 `toml:"attr_timeout"`
// EntryTimeout defines TTL for directory, name lookup in seconds.
EntryTimeout int64 `toml:"entry_timeout"`
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package commonmetrics
import (
"context"
"sync"
"time"
"github.com/containerd/containerd/log"
digest "github.com/opencontainers/go-digest"
"github.com/prometheus/client_golang/prometheus"
)
const (
// OperationLatencyKey is the key for stargz operation latency metrics.
OperationLatencyKey = "operation_duration"
// OperationCountKey is the key for stargz operation count metrics.
OperationCountKey = "operation_count"
// BytesServedKey is the key for any metric related to counting bytes served as the part of specific operation.
BytesServedKey = "bytes_served"
// Keep namespace as stargz and subsystem as fs.
namespace = "stargz"
subsystem = "fs"
)
// Lists all metric labels.
const (
// prometheus metrics
Mount = "mount"
RemoteRegistryGet = "remote_registry_get"
NodeReaddir = "node_readdir"
StargzHeaderGet = "stargz_header_get"
StargzFooterGet = "stargz_footer_get"
StargzTocGet = "stargz_toc_get"
DeserializeTocJSON = "stargz_toc_json_deserialize"
PrefetchesCompleted = "all_prefetches_completed"
ReadOnDemand = "read_on_demand"
MountLayerToLastOnDemandFetch = "mount_layer_to_last_on_demand_fetch"
OnDemandReadAccessCount = "on_demand_read_access_count"
OnDemandRemoteRegistryFetchCount = "on_demand_remote_registry_fetch_count"
OnDemandBytesServed = "on_demand_bytes_served"
OnDemandBytesFetched = "on_demand_bytes_fetched"
// logs metrics
PrefetchTotal = "prefetch_total"
PrefetchDownload = "prefetch_download"
PrefetchDecompress = "prefetch_decompress"
BackgroundFetchTotal = "background_fetch_total"
BackgroundFetchDownload = "background_fetch_download"
BackgroundFetchDecompress = "background_fetch_decompress"
PrefetchSize = "prefetch_size"
)
var (
// Buckets for OperationLatency metric in milliseconds.
latencyBuckets = []float64{1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384} // in milliseconds
// operationLatency collects operation latency numbers by operation
// type and layer digest.
operationLatency = prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: OperationLatencyKey,
Help: "Latency in milliseconds of stargz snapshotter operations. Broken down by operation type and layer sha.",
Buckets: latencyBuckets,
},
[]string{"operation_type", "layer"},
)
// operationCount collects operation count numbers by operation
// type and layer sha.
operationCount = prometheus.NewCounterVec(
prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: OperationCountKey,
Help: "The count of stargz snapshotter operations. Broken down by operation type and layer sha.",
},
[]string{"operation_type", "layer"},
)
// bytesCount reflects the number of bytes served as the part of specitic operation type per layer sha.
bytesCount = prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: BytesServedKey,
Help: "The number of bytes served per stargz snapshotter operations. Broken down by operation type and layer sha.",
},
[]string{"operation_type", "layer"},
)
)
var register sync.Once
// sinceInMilliseconds gets the time since the specified start in milliseconds.
// The division by 1e6 is made to have the milliseconds value as floating point number, since the native method
// .Milliseconds() returns an integer value and you can lost a precision for sub-millisecond values.
func sinceInMilliseconds(start time.Time) float64 {
return float64(time.Since(start).Nanoseconds()) / 1e6
}
// Register registers metrics. This is always called only once.
func Register() {
register.Do(func() {
prometheus.MustRegister(operationLatency)
prometheus.MustRegister(operationCount)
prometheus.MustRegister(bytesCount)
})
}
// MeasureLatency wraps the labels attachment as well as calling Observe into a single method.
// Right now we attach the operation and layer digest, so it's possible to see the breakdown for latency
// by operation and individual layers.
// If you want this to be layer agnostic, just pass the digest from empty string, e.g.
// layerDigest := digest.FromString("")
func MeasureLatency(operation string, layer digest.Digest, start time.Time) {
operationLatency.WithLabelValues(operation, layer.String()).Observe(sinceInMilliseconds(start))
}
// IncOperationCount wraps the labels attachment as well as calling Inc into a single method.
func IncOperationCount(operation string, layer digest.Digest) {
operationCount.WithLabelValues(operation, layer.String()).Inc()
}
// AddBytesCount wraps the labels attachment as well as calling Add into a single method.
func AddBytesCount(operation string, layer digest.Digest, bytes int64) {
bytesCount.WithLabelValues(operation, layer.String()).Add(float64(bytes))
}
// WriteLatencyLogValue wraps writing the log info record for latency in milliseconds. The log record breaks down by operation and layer digest.
func WriteLatencyLogValue(ctx context.Context, layer digest.Digest, operation string, start time.Time) {
ctx = log.WithLogger(ctx, log.G(ctx).WithField("metrics", "latency").WithField("operation", operation).WithField("layer_sha", layer.String()))
log.G(ctx).Infof("value=%v milliseconds", sinceInMilliseconds(start))
}
// WriteLatencyWithBytesLogValue wraps writing the log info record for latency in milliseconds with adding the size in bytes.
// The log record breaks down by operation, layer digest and byte value.
func WriteLatencyWithBytesLogValue(ctx context.Context, layer digest.Digest, latencyOperation string, start time.Time, bytesMetricName string, bytesMetricValue int64) {
ctx = log.WithLogger(ctx, log.G(ctx).WithField("metrics", "latency").WithField("operation", latencyOperation).WithField("layer_sha", layer.String()))
log.G(ctx).Infof("value=%v milliseconds; %v=%v bytes", sinceInMilliseconds(start), bytesMetricName, bytesMetricValue)
}
// LogLatencyForLastOnDemandFetch implements a special case for measuring the latency of last on demand fetch, which must be invoked at the end of
// background fetch operation only. Since this is expected to happen only once per container launch, it writes a log line,
// instead of directly emitting a metric.
// We do that in the following way:
// 1. We record the mount start time
// 2. We constantly record the timestamps when we do on demand fetch for each layer sha
// 3. On background fetch completed we measure the difference between the last on demand fetch and mount start time
// and record it as a metric
func LogLatencyForLastOnDemandFetch(ctx context.Context, layer digest.Digest, start time.Time, end time.Time) {
diffInMilliseconds := float64(end.Sub(start).Milliseconds())
// value can be negative if we pass the default value for time.Time as `end`
// this can happen if there were no on-demand fetch for the particular layer
if diffInMilliseconds > 0 {
ctx = log.WithLogger(ctx, log.G(ctx).WithField("metrics", "latency").WithField("operation", MountLayerToLastOnDemandFetch).WithField("layer_sha", layer.String()))
log.G(ctx).Infof("value=%v milliseconds", diffInMilliseconds)
}
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package layermetrics
import (
"github.com/containerd/stargz-snapshotter/fs/layer"
metrics "github.com/docker/go-metrics"
"github.com/prometheus/client_golang/prometheus"
)
var layerMetrics = []*metric{
{
name: "layer_fetched_size",
help: "Total fetched size of the layer",
unit: metrics.Bytes,
vt: prometheus.CounterValue,
getValues: func(l layer.Layer) []value {
return []value{
{
v: float64(l.Info().FetchedSize),
},
}
},
},
{
name: "layer_prefetch_size",
help: "Total prefetched size of the layer",
unit: metrics.Bytes,
vt: prometheus.CounterValue,
getValues: func(l layer.Layer) []value {
return []value{
{
v: float64(l.Info().PrefetchSize),
},
}
},
},
{
name: "layer_size",
help: "Total size of the layer",
unit: metrics.Bytes,
vt: prometheus.CounterValue,
getValues: func(l layer.Layer) []value {
return []value{
{
v: float64(l.Info().Size),
},
}
},
},
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package layermetrics
import (
"sync"
"github.com/containerd/stargz-snapshotter/fs/layer"
metrics "github.com/docker/go-metrics"
"github.com/prometheus/client_golang/prometheus"
)
func NewLayerMetrics(ns *metrics.Namespace) *Controller {
if ns == nil {
return &Controller{}
}
c := &Controller{
ns: ns,
layer: make(map[string]layer.Layer),
}
c.metrics = append(c.metrics, layerMetrics...)
ns.Add(c)
return c
}
type Controller struct {
ns *metrics.Namespace
metrics []*metric
layer map[string]layer.Layer
layerMu sync.RWMutex
}
func (c *Controller) Describe(ch chan<- *prometheus.Desc) {
for _, e := range c.metrics {
ch <- e.desc(c.ns)
}
}
func (c *Controller) Collect(ch chan<- prometheus.Metric) {
c.layerMu.RLock()
wg := &sync.WaitGroup{}
for mp, l := range c.layer {
mp, l := mp, l
wg.Add(1)
go func() {
defer wg.Done()
for _, e := range c.metrics {
e.collect(mp, l, c.ns, ch)
}
}()
}
c.layerMu.RUnlock()
wg.Wait()
}
func (c *Controller) Add(key string, l layer.Layer) {
if c.ns == nil {
return
}
c.layerMu.Lock()
c.layer[key] = l
c.layerMu.Unlock()
}
func (c *Controller) Remove(key string) {
if c.ns == nil {
return
}
c.layerMu.Lock()
delete(c.layer, key)
c.layerMu.Unlock()
}
type value struct {
v float64
l []string
}
type metric struct {
name string
help string
unit metrics.Unit
vt prometheus.ValueType
labels []string
// getValues returns the value and labels for the data
getValues func(l layer.Layer) []value
}
func (m *metric) desc(ns *metrics.Namespace) *prometheus.Desc {
return ns.NewDesc(m.name, m.help, m.unit, append([]string{"digest", "mountpoint"}, m.labels...)...)
}
func (m *metric) collect(mountpoint string, l layer.Layer, ns *metrics.Namespace, ch chan<- prometheus.Metric) {
values := m.getValues(l)
for _, v := range values {
ch <- prometheus.MustNewConstMetric(m.desc(ns), m.vt, v.v, append([]string{l.Info().Digest.String(), mountpoint}, v.l...)...)
}
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
/*
Copyright 2019 The Go Authors. All rights reserved.
Use of this source code is governed by a BSD-style
license that can be found in the NOTICE.md file.
*/
package remote
// region is HTTP-range-request-compliant range.
// "b" is beginning byte of the range and "e" is the end.
// "e" is must be inclusive along with HTTP's range expression.
type region struct{ b, e int64 }
func (c region) size() int64 {
return c.e - c.b + 1
}
func superRegion(regs []region) region {
s := regs[0]
for _, reg := range regs {
if reg.b < s.b {
s.b = reg.b
}
if reg.e > s.e {
s.e = reg.e
}
}
return s
}
// regionSet is a set of regions
type regionSet struct {
rs []region // must be kept sorted
}
// add attempts to merge r to rs.rs with squashing the regions as
// small as possible. This operation takes O(n).
// TODO: more efficient way to do it.
func (rs *regionSet) add(r region) {
// Iterate over the sorted region slice from the tail.
// a) When an overwrap occurs, adjust `r` to fully contain the looking region
// `l` and remove `l` from region slice.
// b) Once l.e become less than r.b, no overwrap will occur again. So immediately
// insert `r` which fully contains all overwrapped regions, to the region slice.
// Here, `r` is inserted to the region slice with keeping it sorted, without
// overwrapping to any regions.
// *) If any `l` contains `r`, we don't need to do anything so return immediately.
for i := len(rs.rs) - 1; i >= 0; i-- {
l := &rs.rs[i]
// *) l contains r
if l.b <= r.b && r.e <= l.e {
return
}
// a) r overwraps to l so adjust r to fully contain l and reomve l
// from region slice.
if l.b <= r.b && r.b <= l.e+1 && l.e <= r.e {
r.b = l.b
rs.rs = append(rs.rs[:i], rs.rs[i+1:]...)
continue
}
if r.b <= l.b && l.b <= r.e+1 && r.e <= l.e {
r.e = l.e
rs.rs = append(rs.rs[:i], rs.rs[i+1:]...)
continue
}
if r.b <= l.b && l.e <= r.e {
rs.rs = append(rs.rs[:i], rs.rs[i+1:]...)
continue
}
// b) No overwrap will occur after this iteration. Instert r to the
// region slice immediately.
if l.e < r.b {
rs.rs = append(rs.rs[:i+1], append([]region{r}, rs.rs[i+1:]...)...)
return
}
// No overwrap occurs yet. See the next region.
}
// r is the topmost region among regions in the slice.
rs.rs = append([]region{r}, rs.rs...)
}
func (rs *regionSet) totalSize() int64 {
var sz int64
for _, f := range rs.rs {
sz += f.size()
}
return sz
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package source
import (
"context"
"fmt"
"strings"
"github.com/containerd/containerd/images"
"github.com/containerd/containerd/labels"
"github.com/containerd/containerd/reference"
"github.com/containerd/containerd/remotes/docker"
"github.com/containerd/stargz-snapshotter/fs/config"
digest "github.com/opencontainers/go-digest"
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
)
// GetSources is a function for converting snapshot labels into typed blob sources
// information. This package defines a default converter which provides source
// information based on some labels but implementations aren't required to use labels.
// Implementations are allowed to return several sources (registry config + image refs)
// about the blob.
type GetSources func(labels map[string]string) (source []Source, err error)
// RegistryHosts returns a list of registries that provides the specified image.
type RegistryHosts func(reference.Spec) ([]docker.RegistryHost, error)
// Source is a typed blob source information. This contains information about
// a blob stored in registries and some contexts of the blob.
type Source struct {
// Hosts is a registry configuration where this blob is stored.
Hosts RegistryHosts
// Name is an image reference which contains this blob.
Name reference.Spec
// Target is a descriptor of this blob.
Target ocispec.Descriptor
// Manifest is an image manifest which contains the blob. This will
// be used by the filesystem to pre-resolve some layers contained in
// the manifest.
// Currently, only layer digests (Manifest.Layers.Digest) will be used.
Manifest ocispec.Manifest
}
const (
// targetRefLabel is a label which contains image reference.
targetRefLabel = "containerd.io/snapshot/remote/stargz.reference"
// targetDigestLabel is a label which contains layer digest.
targetDigestLabel = "containerd.io/snapshot/remote/stargz.digest"
// targetImageLayersLabel is a label which contains layer digests contained in
// the target image.
targetImageLayersLabel = "containerd.io/snapshot/remote/stargz.layers"
)
// FromDefaultLabels returns a function for converting snapshot labels to
// source information based on labels.
func FromDefaultLabels(hosts RegistryHosts) GetSources {
return func(labels map[string]string) ([]Source, error) {
refStr, ok := labels[targetRefLabel]
if !ok {
return nil, fmt.Errorf("reference hasn't been passed")
}
refspec, err := reference.Parse(refStr)
if err != nil {
return nil, err
}
digestStr, ok := labels[targetDigestLabel]
if !ok {
return nil, fmt.Errorf("digest hasn't been passed")
}
target, err := digest.Parse(digestStr)
if err != nil {
return nil, err
}
var layersDgst []digest.Digest
if l, ok := labels[targetImageLayersLabel]; ok {
layersStr := strings.Split(l, ",")
for _, l := range layersStr {
d, err := digest.Parse(l)
if err != nil {
return nil, err
}
if d.String() != target.String() {
layersDgst = append(layersDgst, d)
}
}
}
var layers []ocispec.Descriptor
for _, dgst := range append([]digest.Digest{target}, layersDgst...) {
layers = append(layers, ocispec.Descriptor{Digest: dgst})
}
return []Source{
{
Hosts: hosts,
Name: refspec,
Target: ocispec.Descriptor{Digest: target},
Manifest: ocispec.Manifest{Layers: layers},
},
}, nil
}
}
// AppendDefaultLabelsHandlerWrapper makes a handler which appends image's basic
// information to each layer descriptor as annotations during unpack. These
// annotations will be passed to this remote snapshotter as labels and used to
// construct source information.
func AppendDefaultLabelsHandlerWrapper(ref string, prefetchSize int64) func(f images.Handler) images.Handler {
return func(f images.Handler) images.Handler {
return images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
children, err := f.Handle(ctx, desc)
if err != nil {
return nil, err
}
switch desc.MediaType {
case ocispec.MediaTypeImageManifest, images.MediaTypeDockerSchema2Manifest:
for i := range children {
c := &children[i]
if images.IsLayerType(c.MediaType) {
if c.Annotations == nil {
c.Annotations = make(map[string]string)
}
c.Annotations[targetRefLabel] = ref
c.Annotations[targetDigestLabel] = c.Digest.String()
var layers string
for _, l := range children[i:] {
if images.IsLayerType(l.MediaType) {
ls := fmt.Sprintf("%s,", l.Digest.String())
// This avoids the label hits the size limitation.
// Skipping layers is allowed here and only affects performance.
if err := labels.Validate(targetImageLayersLabel, layers+ls); err != nil {
break
}
layers += ls
}
}
c.Annotations[targetImageLayersLabel] = strings.TrimSuffix(layers, ",")
c.Annotations[config.TargetPrefetchSizeLabel] = fmt.Sprintf("%d", prefetchSize)
}
}
}
return children, nil
})
}
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package service
import (
"github.com/containerd/stargz-snapshotter/fs/config"
"github.com/containerd/stargz-snapshotter/service/resolver"
)
type Config struct {
config.Config
// KubeconfigKeychainConfig is config for kubeconfig-based keychain.
KubeconfigKeychainConfig `toml:"kubeconfig_keychain"`
// CRIKeychainConfig is config for CRI-based keychain.
CRIKeychainConfig `toml:"cri_keychain"`
// ResolverConfig is config for resolving registries.
ResolverConfig `toml:"resolver"`
}
// KubeconfigKeychainConfig is config for kubeconfig-based keychain.
type KubeconfigKeychainConfig struct {
// EnableKeychain enables kubeconfig-based keychain
EnableKeychain bool `toml:"enable_keychain"`
// KubeconfigPath is the path to kubeconfig which can be used to sync
// secrets on the cluster into this snapshotter.
KubeconfigPath string `toml:"kubeconfig_path"`
}
// CRIKeychainConfig is config for CRI-based keychain.
type CRIKeychainConfig struct {
// EnableKeychain enables CRI-based keychain
EnableKeychain bool `toml:"enable_keychain"`
// ImageServicePath is the path to the unix socket of backing CRI Image Service (e.g. containerd CRI plugin)
ImageServicePath string `toml:"image_service_path"`
}
// ResolverConfig is config for resolving registries.
type ResolverConfig resolver.Config
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package service
import (
"fmt"
"strings"
"github.com/containerd/containerd/reference"
"github.com/containerd/stargz-snapshotter/fs/source"
digest "github.com/opencontainers/go-digest"
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
)
const (
// targetRefLabel is a label which contains image reference passed from CRI plugin.
targetRefLabel = "containerd.io/snapshot/cri.image-ref"
// targetDigestLabel is a label which contains layer digest passed from CRI plugin.
targetDigestLabel = "containerd.io/snapshot/cri.layer-digest"
// targetImageLayersLabel is a label which contains layer digests contained in
// the target image and is passed from CRI plugin.
targetImageLayersLabel = "containerd.io/snapshot/cri.image-layers"
)
func sourceFromCRILabels(hosts source.RegistryHosts) source.GetSources {
return func(labels map[string]string) ([]source.Source, error) {
refStr, ok := labels[targetRefLabel]
if !ok {
return nil, fmt.Errorf("reference hasn't been passed")
}
refspec, err := reference.Parse(refStr)
if err != nil {
return nil, err
}
digestStr, ok := labels[targetDigestLabel]
if !ok {
return nil, fmt.Errorf("digest hasn't been passed")
}
target, err := digest.Parse(digestStr)
if err != nil {
return nil, err
}
var layersDgst []digest.Digest
if l, ok := labels[targetImageLayersLabel]; ok {
layersStr := strings.Split(l, ",")
for _, l := range layersStr {
d, err := digest.Parse(l)
if err != nil {
return nil, err
}
if d.String() != target.String() {
layersDgst = append(layersDgst, d)
}
}
}
var layers []ocispec.Descriptor
for _, dgst := range append([]digest.Digest{target}, layersDgst...) {
layers = append(layers, ocispec.Descriptor{Digest: dgst})
}
return []source.Source{
{
Hosts: hosts,
Name: refspec,
Target: ocispec.Descriptor{Digest: target},
Manifest: ocispec.Manifest{Layers: layers},
},
}, nil
}
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package cri
import (
"context"
"sync"
"time"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/reference"
distribution "github.com/containerd/containerd/reference/docker"
"github.com/containerd/stargz-snapshotter/service/resolver"
"github.com/pkg/errors"
runtime "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
)
// NewCRIKeychain provides creds passed through CRI PullImage API.
// This also returns a CRI image service server that works as a proxy backed by the specified CRI service.
// This server reads all PullImageRequest and uses PullImageRequest.AuthConfig for authenticating snapshots.
func NewCRIKeychain(ctx context.Context, connectCRI func() (runtime.ImageServiceClient, error)) (resolver.Credential, runtime.ImageServiceServer) {
server := &instrumentedService{config: make(map[string]*runtime.AuthConfig)}
go func() {
log.G(ctx).Debugf("Waiting for CRI service is started...")
for i := 0; i < 100; i++ {
client, err := connectCRI()
if err == nil {
server.criMu.Lock()
server.cri = client
server.criMu.Unlock()
log.G(ctx).Info("connected to backend CRI service")
return
}
log.G(ctx).WithError(err).Warnf("failed to connect to CRI")
time.Sleep(10 * time.Second)
}
log.G(ctx).Warnf("no connection is available to CRI")
}()
return server.credentials, server
}
type instrumentedService struct {
cri runtime.ImageServiceClient
criMu sync.Mutex
config map[string]*runtime.AuthConfig
configMu sync.Mutex
}
func (in *instrumentedService) credentials(host string, refspec reference.Spec) (string, string, error) {
if host == "docker.io" || host == "registry-1.docker.io" {
// Creds of "docker.io" is stored keyed by "https://index.docker.io/v1/".
host = "index.docker.io"
}
in.configMu.Lock()
defer in.configMu.Unlock()
if cfg, ok := in.config[refspec.String()]; ok {
return resolver.ParseAuth(cfg, host)
}
return "", "", nil
}
func (in *instrumentedService) getCRI() (c runtime.ImageServiceClient) {
in.criMu.Lock()
c = in.cri
in.criMu.Unlock()
return
}
func (in *instrumentedService) ListImages(ctx context.Context, r *runtime.ListImagesRequest) (res *runtime.ListImagesResponse, err error) {
cri := in.getCRI()
if cri == nil {
return nil, errors.New("server is not initialized yet")
}
return cri.ListImages(ctx, r)
}
func (in *instrumentedService) ImageStatus(ctx context.Context, r *runtime.ImageStatusRequest) (res *runtime.ImageStatusResponse, err error) {
cri := in.getCRI()
if cri == nil {
return nil, errors.New("server is not initialized yet")
}
return cri.ImageStatus(ctx, r)
}
func (in *instrumentedService) PullImage(ctx context.Context, r *runtime.PullImageRequest) (res *runtime.PullImageResponse, err error) {
cri := in.getCRI()
if cri == nil {
return nil, errors.New("server is not initialized yet")
}
refspec, err := parseReference(r.GetImage().GetImage())
if err != nil {
return nil, err
}
in.configMu.Lock()
in.config[refspec.String()] = r.GetAuth()
in.configMu.Unlock()
return cri.PullImage(ctx, r)
}
func (in *instrumentedService) RemoveImage(ctx context.Context, r *runtime.RemoveImageRequest) (_ *runtime.RemoveImageResponse, err error) {
cri := in.getCRI()
if cri == nil {
return nil, errors.New("server is not initialized yet")
}
refspec, err := parseReference(r.GetImage().GetImage())
if err != nil {
return nil, err
}
in.configMu.Lock()
delete(in.config, refspec.String())
in.configMu.Unlock()
return cri.RemoveImage(ctx, r)
}
func (in *instrumentedService) ImageFsInfo(ctx context.Context, r *runtime.ImageFsInfoRequest) (res *runtime.ImageFsInfoResponse, err error) {
cri := in.getCRI()
if cri == nil {
return nil, errors.New("server is not initialized yet")
}
return cri.ImageFsInfo(ctx, r)
}
func parseReference(ref string) (reference.Spec, error) {
namedRef, err := distribution.ParseDockerRef(ref)
if err != nil {
return reference.Spec{}, errors.Wrapf(err, "failed to parse image reference %q", ref)
}
return reference.Parse(namedRef.String())
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package dockerconfig
import (
"context"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/reference"
"github.com/containerd/stargz-snapshotter/service/resolver"
"github.com/docker/cli/cli/config"
)
func NewDockerconfigKeychain(ctx context.Context) resolver.Credential {
return func(host string, refspec reference.Spec) (string, string, error) {
cf, err := config.Load("")
if err != nil {
log.G(ctx).WithError(err).Warnf("failed to load docker config file")
return "", "", nil
}
if host == "docker.io" || host == "registry-1.docker.io" {
// Creds of docker.io is stored keyed by "https://index.docker.io/v1/".
host = "https://index.docker.io/v1/"
}
ac, err := cf.GetAuthConfig(host)
if err != nil {
return "", "", err
}
if ac.IdentityToken != "" {
return "", ac.IdentityToken, nil
}
return ac.Username, ac.Password, nil
}
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package kubeconfig
import (
"bytes"
"context"
"fmt"
"os"
"sync"
"time"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/reference"
"github.com/containerd/stargz-snapshotter/service/resolver"
dcfile "github.com/docker/cli/cli/config/configfile"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/clientcmd"
"k8s.io/client-go/util/workqueue"
)
const dockerconfigSelector = "type=" + string(corev1.SecretTypeDockerConfigJson)
type options struct {
kubeconfigPath string
}
type Option func(*options)
func WithKubeconfigPath(path string) Option {
return func(opts *options) {
opts.kubeconfigPath = path
}
}
// NewKubeconfigKeychain provides a keychain which can sync its contents with
// kubernetes API server by fetching all `kubernetes.io/dockerconfigjson`
// secrets in the cluster with provided kubeconfig. It's OK that config provides
// kubeconfig path but the file doesn't exist at that moment. In this case, this
// keychain keeps on trying to read the specified path periodically and when the
// file is actually provided, this keychain tries to access API server using the
// file. This is useful for some environments (e.g. single node cluster with
// containerized apiserver) where stargz snapshotter needs to start before
// everything, including booting containerd/kubelet/apiserver and configuring
// users/roles.
// TODO: support update of kubeconfig file
func NewKubeconfigKeychain(ctx context.Context, opts ...Option) resolver.Credential {
var kcOpts options
for _, o := range opts {
o(&kcOpts)
}
kc := newKeychain(ctx, kcOpts.kubeconfigPath)
return kc.credentials
}
func newKeychain(ctx context.Context, kubeconfigPath string) *keychain {
kc := &keychain{
config: make(map[string]*dcfile.ConfigFile),
}
ctx = log.WithLogger(ctx, log.G(ctx).WithField("kubeconfig", kubeconfigPath))
go func() {
if kubeconfigPath != "" {
log.G(ctx).Debugf("Waiting for kubeconfig being installed...")
for {
if _, err := os.Stat(kubeconfigPath); err == nil {
break
} else if !os.IsNotExist(err) {
log.G(ctx).WithError(err).
Warnf("failed to read; Disabling syncing")
return
}
time.Sleep(10 * time.Second)
}
}
// default loader for KUBECONFIG or `~/.kube/config`
// if no explicit path provided, KUBECONFIG will be used.
// if KUBECONFIG doesn't contain paths, `~/.kube/config` will be used.
loadingRule := clientcmd.NewDefaultClientConfigLoadingRules()
// explicitly provide path for kubeconfig.
// if path isn't "", this path will be respected.
loadingRule.ExplicitPath = kubeconfigPath
// load and merge config files
clientcfg, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
loadingRule, // loader for config files
&clientcmd.ConfigOverrides{}, // no overrides for config
).ClientConfig()
if err != nil {
log.G(ctx).WithError(err).Warnf("failed to load config; Disabling syncing")
return
}
client, err := kubernetes.NewForConfig(clientcfg)
if err != nil {
log.G(ctx).WithError(err).Warnf("failed to prepare client; Disabling syncing")
return
}
if err := kc.startSyncSecrets(ctx, client); err != nil {
log.G(ctx).WithError(err).Warnf("failed to sync secrets")
}
}()
return kc
}
type keychain struct {
config map[string]*dcfile.ConfigFile
configMu sync.Mutex
// the following entries are used for syncing secrets with API server.
// these fields are lazily filled after kubeconfig file is provided.
queue *workqueue.Type
informer cache.SharedIndexInformer
}
func (kc *keychain) credentials(host string, refspec reference.Spec) (string, string, error) {
if host == "docker.io" || host == "registry-1.docker.io" {
// Creds of "docker.io" is stored keyed by "https://index.docker.io/v1/".
host = "https://index.docker.io/v1/"
}
kc.configMu.Lock()
defer kc.configMu.Unlock()
for _, cfg := range kc.config {
if acfg, err := cfg.GetAuthConfig(host); err == nil {
if acfg.IdentityToken != "" {
return "", acfg.IdentityToken, nil
} else if !(acfg.Username == "" && acfg.Password == "") {
return acfg.Username, acfg.Password, nil
}
}
}
return "", "", nil
}
func (kc *keychain) startSyncSecrets(ctx context.Context, client kubernetes.Interface) error {
// don't let panics crash the process
defer utilruntime.HandleCrash()
// get informed on `kubernetes.io/dockerconfigjson` secrets in all namespaces
informer := cache.NewSharedIndexInformer(
&cache.ListWatch{
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
// TODO: support legacy image secret `kubernetes.io/dockercfg`
options.FieldSelector = dockerconfigSelector
return client.CoreV1().Secrets(metav1.NamespaceAll).List(ctx, options)
},
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
// TODO: support legacy image secret `kubernetes.io/dockercfg`
options.FieldSelector = dockerconfigSelector
return client.CoreV1().Secrets(metav1.NamespaceAll).Watch(ctx, options)
},
},
&corev1.Secret{},
0,
cache.Indexers{},
)
// use workqueue because each task possibly takes long for parsing config,
// wating for lock, etc...
queue := workqueue.New()
defer queue.ShutDown()
informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
key, err := cache.MetaNamespaceKeyFunc(obj)
if err == nil {
queue.Add(key)
}
},
UpdateFunc: func(old, new interface{}) {
key, err := cache.MetaNamespaceKeyFunc(new)
if err == nil {
queue.Add(key)
}
},
DeleteFunc: func(obj interface{}) {
key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
if err == nil {
queue.Add(key)
}
},
})
go informer.Run(ctx.Done())
if !cache.WaitForCacheSync(ctx.Done(), informer.HasSynced) {
return fmt.Errorf("Timed out for syncing cache")
}
// get informer and queue
kc.informer = informer
kc.queue = queue
// keep on syncing secrets
wait.Until(kc.runWorker, time.Second, ctx.Done())
return nil
}
func (kc *keychain) runWorker() {
for kc.processNextItem() {
// continue looping
}
}
// TODO: consider retrying?
func (kc *keychain) processNextItem() bool {
key, quit := kc.queue.Get()
if quit {
return false
}
defer kc.queue.Done(key)
obj, exists, err := kc.informer.GetIndexer().GetByKey(key.(string))
if err != nil {
utilruntime.HandleError(fmt.Errorf("failed to get object; don't sync %q: %v", key, err))
return true
}
if !exists {
kc.configMu.Lock()
delete(kc.config, key.(string))
kc.configMu.Unlock()
return true
}
// TODO: support legacy image secret `kubernetes.io/dockercfg`
data, ok := obj.(*corev1.Secret).Data[corev1.DockerConfigJsonKey]
if !ok {
utilruntime.HandleError(fmt.Errorf("no secret is provided; don't sync %q", key))
return true
}
configFile := dcfile.New("")
if err := configFile.LoadFromReader(bytes.NewReader(data)); err != nil {
utilruntime.HandleError(fmt.Errorf("broken data; don't sync %q: %v", key, err))
return true
}
kc.configMu.Lock()
kc.config[key.(string)] = configFile
kc.configMu.Unlock()
return true
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package plugin
import (
"net"
"os"
"path/filepath"
"time"
"github.com/containerd/containerd/defaults"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/pkg/dialer"
"github.com/containerd/containerd/platforms"
ctdplugin "github.com/containerd/containerd/plugin"
"github.com/containerd/stargz-snapshotter/service"
"github.com/containerd/stargz-snapshotter/service/keychain/cri"
"github.com/containerd/stargz-snapshotter/service/keychain/dockerconfig"
"github.com/containerd/stargz-snapshotter/service/keychain/kubeconfig"
"github.com/containerd/stargz-snapshotter/service/resolver"
"github.com/pkg/errors"
grpc "google.golang.org/grpc"
"google.golang.org/grpc/backoff"
runtime "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
)
// Config represents configuration for the stargz snapshotter plugin.
type Config struct {
service.Config
// RootPath is the directory for the plugin
RootPath string `toml:"root_path"`
// CRIKeychainImageServicePath is the path to expose CRI service wrapped by CRI keychain
CRIKeychainImageServicePath string `toml:"cri_keychain_image_service_path"`
// Registry is CRI-plugin-compatible registry configuration
Registry resolver.Registry `toml:"registry"`
}
func init() {
ctdplugin.Register(&ctdplugin.Registration{
Type: ctdplugin.SnapshotPlugin,
ID: "stargz",
Config: &Config{},
InitFn: func(ic *ctdplugin.InitContext) (interface{}, error) {
ic.Meta.Platforms = append(ic.Meta.Platforms, platforms.DefaultSpec())
ctx := ic.Context
config, ok := ic.Config.(*Config)
if !ok {
return nil, errors.New("invalid stargz snapshotter configuration")
}
root := ic.Root
if config.RootPath != "" {
root = config.RootPath
}
ic.Meta.Exports["root"] = root
// Configure keychain
credsFuncs := []resolver.Credential{dockerconfig.NewDockerconfigKeychain(ctx)}
if config.Config.KubeconfigKeychainConfig.EnableKeychain {
var opts []kubeconfig.Option
if kcp := config.Config.KubeconfigKeychainConfig.KubeconfigPath; kcp != "" {
opts = append(opts, kubeconfig.WithKubeconfigPath(kcp))
}
credsFuncs = append(credsFuncs, kubeconfig.NewKubeconfigKeychain(ctx, opts...))
}
if addr := config.CRIKeychainImageServicePath; config.Config.CRIKeychainConfig.EnableKeychain && addr != "" {
// connects to the backend CRI service (defaults to containerd socket)
criAddr := ic.Address
if cp := config.Config.CRIKeychainConfig.ImageServicePath; cp != "" {
criAddr = cp
}
if criAddr == "" {
return nil, errors.New("backend CRI service address is not specified")
}
connectCRI := func() (runtime.ImageServiceClient, error) {
// TODO: make gRPC options configurable from config.toml
backoffConfig := backoff.DefaultConfig
backoffConfig.MaxDelay = 3 * time.Second
connParams := grpc.ConnectParams{
Backoff: backoffConfig,
}
gopts := []grpc.DialOption{
grpc.WithInsecure(),
grpc.WithConnectParams(connParams),
grpc.WithContextDialer(dialer.ContextDialer),
grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(defaults.DefaultMaxRecvMsgSize)),
grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(defaults.DefaultMaxSendMsgSize)),
}
conn, err := grpc.Dial(dialer.DialAddress(criAddr), gopts...)
if err != nil {
return nil, err
}
return runtime.NewImageServiceClient(conn), nil
}
criCreds, criServer := cri.NewCRIKeychain(ctx, connectCRI)
// Create a gRPC server
rpc := grpc.NewServer()
runtime.RegisterImageServiceServer(rpc, criServer)
// Prepare the directory for the socket
if err := os.MkdirAll(filepath.Dir(addr), 0700); err != nil {
return nil, errors.Wrapf(err, "failed to create directory %q", filepath.Dir(addr))
}
// Try to remove the socket file to avoid EADDRINUSE
if err := os.RemoveAll(addr); err != nil {
return nil, errors.Wrapf(err, "failed to remove %q", addr)
}
// Listen and serve
l, err := net.Listen("unix", addr)
if err != nil {
return nil, errors.Wrapf(err, "error on listen socket %q", addr)
}
go func() {
if err := rpc.Serve(l); err != nil {
log.G(ctx).WithError(err).Warnf("error on serving via socket %q", addr)
}
}()
credsFuncs = append(credsFuncs, criCreds)
}
// TODO(ktock): print warn if old configuration is specified.
// TODO(ktock): should we respect old configuration?
return service.NewStargzSnapshotterService(ctx, root, &config.Config,
service.WithCustomRegistryHosts(resolver.RegistryHostsFromCRIConfig(ctx, config.Registry, credsFuncs...)))
},
})
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package resolver
import (
"net/http"
"time"
"github.com/containerd/containerd/reference"
"github.com/containerd/containerd/remotes/docker"
"github.com/containerd/stargz-snapshotter/fs/source"
)
const defaultRequestTimeoutSec = 30
// Config is config for resolving registries.
type Config struct {
Host map[string]HostConfig `toml:"host"`
}
type HostConfig struct {
Mirrors []MirrorConfig `toml:"mirrors"`
}
type MirrorConfig struct {
// Host is the hostname of the host.
Host string `toml:"host"`
// Insecure is true means use http scheme instead of https.
Insecure bool `toml:"insecure"`
// RequestTimeoutSec is timeout seconds of each request to the registry.
// RequestTimeoutSec == 0 indicates the default timeout (defaultRequestTimeoutSec).
// RequestTimeoutSec < 0 indicates no timeout.
RequestTimeoutSec int `toml:"request_timeout_sec"`
}
type Credential func(string, reference.Spec) (string, string, error)
// RegistryHostsFromConfig creates RegistryHosts (a set of registry configuration) from Config.
func RegistryHostsFromConfig(cfg Config, credsFuncs ...Credential) source.RegistryHosts {
return func(ref reference.Spec) (hosts []docker.RegistryHost, _ error) {
host := ref.Hostname()
for _, h := range append(cfg.Host[host].Mirrors, MirrorConfig{
Host: host,
}) {
tr := &http.Client{Transport: http.DefaultTransport.(*http.Transport).Clone()}
if h.RequestTimeoutSec >= 0 {
if h.RequestTimeoutSec == 0 {
tr.Timeout = defaultRequestTimeoutSec * time.Second
} else {
tr.Timeout = time.Duration(h.RequestTimeoutSec) * time.Second
}
} // h.RequestTimeoutSec < 0 means "no timeout"
config := docker.RegistryHost{
Client: tr,
Host: h.Host,
Scheme: "https",
Path: "/v2",
Capabilities: docker.HostCapabilityPull | docker.HostCapabilityResolve,
Authorizer: docker.NewDockerAuthorizer(
docker.WithAuthClient(tr),
docker.WithAuthCreds(multiCredsFuncs(ref, credsFuncs...))),
}
if localhost, _ := docker.MatchLocalhost(config.Host); localhost || h.Insecure {
config.Scheme = "http"
}
if config.Host == "docker.io" {
config.Host = "registry-1.docker.io"
}
hosts = append(hosts, config)
}
return
}
}
func multiCredsFuncs(ref reference.Spec, credsFuncs ...Credential) func(string) (string, string, error) {
return func(host string) (string, string, error) {
for _, f := range credsFuncs {
if username, secret, err := f(host, ref); err != nil {
return "", "", err
} else if !(username == "" && secret == "") {
return username, secret, nil
}
}
return "", "", nil
}
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package service
import (
"context"
"path/filepath"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/snapshots"
stargzfs "github.com/containerd/stargz-snapshotter/fs"
"github.com/containerd/stargz-snapshotter/fs/source"
"github.com/containerd/stargz-snapshotter/service/resolver"
snbase "github.com/containerd/stargz-snapshotter/snapshot"
"github.com/containerd/stargz-snapshotter/snapshot/overlayutils"
"github.com/hashicorp/go-multierror"
)
type Option func(*options)
type options struct {
credsFuncs []resolver.Credential
registryHosts source.RegistryHosts
}
// WithCredsFuncs specifies credsFuncs to be used for connecting to the registries.
func WithCredsFuncs(creds ...resolver.Credential) Option {
return func(o *options) {
o.credsFuncs = append(o.credsFuncs, creds...)
}
}
// WithCustomRegistryHosts is registry hosts to use instead.
func WithCustomRegistryHosts(hosts source.RegistryHosts) Option {
return func(o *options) {
o.registryHosts = hosts
}
}
// NewStargzSnapshotterService returns stargz snapshotter.
func NewStargzSnapshotterService(ctx context.Context, root string, config *Config, opts ...Option) (snapshots.Snapshotter, error) {
var sOpts options
for _, o := range opts {
o(&sOpts)
}
hosts := sOpts.registryHosts
if hosts == nil {
// Use RegistryHosts based on ResolverConfig and keychain
hosts = resolver.RegistryHostsFromConfig(resolver.Config(config.ResolverConfig), sOpts.credsFuncs...)
}
// Configure filesystem and snapshotter
fs, err := stargzfs.NewFilesystem(fsRoot(root),
config.Config,
stargzfs.WithGetSources(sources(
sourceFromCRILabels(hosts), // provides source info based on CRI labels
source.FromDefaultLabels(hosts), // provides source info based on default labels
)),
)
if err != nil {
log.G(ctx).WithError(err).Fatalf("failed to configure filesystem")
}
var snapshotter snapshots.Snapshotter
snapshotter, err = snbase.NewSnapshotter(ctx, snapshotterRoot(root), fs, snbase.AsynchronousRemove)
if err != nil {
log.G(ctx).WithError(err).Fatalf("failed to create new snapshotter")
}
return snapshotter, err
}
func snapshotterRoot(root string) string {
return filepath.Join(root, "snapshotter")
}
func fsRoot(root string) string {
return filepath.Join(root, "stargz")
}
func sources(ps ...source.GetSources) source.GetSources {
return func(labels map[string]string) (source []source.Source, allErr error) {
for _, p := range ps {
src, err := p(labels)
if err == nil {
return src, nil
}
allErr = multierror.Append(allErr, err)
}
return
}
}
// Supported returns nil when the remote snapshotter is functional on the system with the root directory.
// Supported is not called during plugin initialization, but exposed for downstream projects which uses
// this snapshotter as a library.
func Supported(root string) error {
// Remote snapshotter is implemented based on overlayfs snapshotter.
return overlayutils.Supported(snapshotterRoot(root))
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// =====
// NOTE: This file is ported from https://github.com/containerd/containerd/blob/v1.5.2/snapshots/overlay/overlayutils/check.go
// TODO: import this from containerd package once we drop support to continerd v1.4.x
// =====
package overlayutils
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/mount"
userns "github.com/containerd/containerd/sys"
"github.com/containerd/continuity/fs"
"github.com/pkg/errors"
)
// SupportsMultipleLowerDir checks if the system supports multiple lowerdirs,
// which is required for the overlay snapshotter. On 4.x kernels, multiple lowerdirs
// are always available (so this check isn't needed), and backported to RHEL and
// CentOS 3.x kernels (3.10.0-693.el7.x86_64 and up). This function is to detect
// support on those kernels, without doing a kernel version compare.
//
// Ported from moby overlay2.
func SupportsMultipleLowerDir(d string) error {
td, err := ioutil.TempDir(d, "multiple-lowerdir-check")
if err != nil {
return err
}
defer func() {
if err := os.RemoveAll(td); err != nil {
log.L.WithError(err).Warnf("Failed to remove check directory %v", td)
}
}()
for _, dir := range []string{"lower1", "lower2", "upper", "work", "merged"} {
if err := os.Mkdir(filepath.Join(td, dir), 0755); err != nil {
return err
}
}
opts := fmt.Sprintf("lowerdir=%s:%s,upperdir=%s,workdir=%s", filepath.Join(td, "lower2"), filepath.Join(td, "lower1"), filepath.Join(td, "upper"), filepath.Join(td, "work"))
m := mount.Mount{
Type: "overlay",
Source: "overlay",
Options: []string{opts},
}
dest := filepath.Join(td, "merged")
if err := m.Mount(dest); err != nil {
return errors.Wrap(err, "failed to mount overlay")
}
if err := mount.UnmountAll(dest, 0); err != nil {
log.L.WithError(err).Warnf("Failed to unmount check directory %v", dest)
}
return nil
}
// Supported returns nil when the overlayfs is functional on the system with the root directory.
// Supported is not called during plugin initialization, but exposed for downstream projects which uses
// this snapshotter as a library.
func Supported(root string) error {
if err := os.MkdirAll(root, 0700); err != nil {
return err
}
supportsDType, err := fs.SupportsDType(root)
if err != nil {
return err
}
if !supportsDType {
return fmt.Errorf("%s does not support d_type. If the backing filesystem is xfs, please reformat with ftype=1 to enable d_type support", root)
}
return SupportsMultipleLowerDir(root)
}
// NeedsUserXAttr returns whether overlayfs should be mounted with the "userxattr" mount option.
//
// The "userxattr" option is needed for mounting overlayfs inside a user namespace with kernel >= 5.11.
//
// The "userxattr" option is NOT needed for the initial user namespace (aka "the host").
//
// Also, Ubuntu (since circa 2015) and Debian (since 10) with kernel < 5.11 can mount
// the overlayfs in a user namespace without the "userxattr" option.
//
// The corresponding kernel commit: https://github.com/torvalds/linux/commit/2d2f2d7322ff43e0fe92bf8cccdc0b09449bf2e1
// > ovl: user xattr
// >
// > Optionally allow using "user.overlay." namespace instead of "trusted.overlay."
// > ...
// > Disable redirect_dir and metacopy options, because these would allow privilege escalation through direct manipulation of the
// > "user.overlay.redirect" or "user.overlay.metacopy" xattrs.
// > ...
//
// The "userxattr" support is not exposed in "/sys/module/overlay/parameters".
func NeedsUserXAttr(d string) (bool, error) {
if !userns.RunningInUserNS() {
// we are the real root (i.e., the root in the initial user NS),
// so we do never need "userxattr" opt.
return false, nil
}
// TODO: add fast path for kernel >= 5.11 .
//
// Keep in mind that distro vendors might be going to backport the patch to older kernels.
// So we can't completely remove the check.
tdRoot := filepath.Join(d, "userxattr-check")
if err := os.RemoveAll(tdRoot); err != nil {
log.L.WithError(err).Warnf("Failed to remove check directory %v", tdRoot)
}
if err := os.MkdirAll(tdRoot, 0700); err != nil {
return false, err
}
defer func() {
if err := os.RemoveAll(tdRoot); err != nil {
log.L.WithError(err).Warnf("Failed to remove check directory %v", tdRoot)
}
}()
td, err := ioutil.TempDir(tdRoot, "")
if err != nil {
return false, err
}
for _, dir := range []string{"lower1", "lower2", "upper", "work", "merged"} {
if err := os.Mkdir(filepath.Join(td, dir), 0755); err != nil {
return false, err
}
}
opts := []string{
fmt.Sprintf("lowerdir=%s:%s,upperdir=%s,workdir=%s", filepath.Join(td, "lower2"), filepath.Join(td, "lower1"), filepath.Join(td, "upper"), filepath.Join(td, "work")),
"userxattr",
}
m := mount.Mount{
Type: "overlay",
Source: "overlay",
Options: opts,
}
dest := filepath.Join(td, "merged")
if err := m.Mount(dest); err != nil {
// Probably the host is running Ubuntu/Debian kernel (< 5.11) with the userns patch but without the userxattr patch.
// Return false without error.
log.L.WithError(err).Debugf("cannot mount overlay with \"userxattr\", probably the kernel does not support userxattr")
return false, nil
}
if err := mount.UnmountAll(dest, 0); err != nil {
log.L.WithError(err).Warnf("Failed to unmount check directory %v", dest)
}
return true, nil
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package task
import (
"context"
"sync"
"sync/atomic"
"time"
"golang.org/x/sync/semaphore"
)
// NewBackgroundTaskManager provides a task manager. You can specify the
// concurrency of background tasks. When running a background task, this will be
// forced to wait until no prioritized task is running for some period. You can
// specify the period through the argument of this function, too.
func NewBackgroundTaskManager(concurrency int64, period time.Duration) *BackgroundTaskManager {
return &BackgroundTaskManager{
backgroundSem: semaphore.NewWeighted(concurrency),
prioritizedTaskSilencePeriod: period,
prioritizedTaskStartNotify: make(chan struct{}),
prioritizedTaskDoneCond: sync.NewCond(&sync.Mutex{}),
}
}
// BackgroundTaskManager is a task manager which manages prioritized tasks and
// background tasks execution. Background tasks are less important than
// prioritized tasks. You can let these background tasks not to use compute
// resources (CPU, NW, etc...) during more important tasks(=prioritized tasks)
// running.
//
// When you run a prioritised task and don't want background tasks to use
// resources you can tell it this manager by calling DoPrioritizedTask method.
// DonePrioritizedTask method must be called at the end of the prioritised task
// execution.
//
// For running a background task, you can use InvokeBackgroundTask method. The
// background task must be able to be cancelled via context.Context argument.
// The task is forced to wait until no prioritized task is running for some
// period. You can specify the period when making this manager instance. The
// limited number of background tasks run simultaneously and you can specify the
// concurrency when making this manager instance too. If a prioritized task
// starts during the execution of background tasks, all background tasks running
// will be cancelled via context. These cancelled tasks will be executed again
// later, same as other background tasks (when no prioritized task is running
// for some period).
type BackgroundTaskManager struct {
prioritizedTasks int64
backgroundSem *semaphore.Weighted
prioritizedTaskSilencePeriod time.Duration
prioritizedTaskStartNotify chan struct{}
prioritizedTaskStartNotifyMu sync.Mutex
prioritizedTaskDoneCond *sync.Cond
}
// DoPrioritizedTask tells the manager that we are running a prioritized task
// and don't want background tasks to disturb resources(CPU, NW, etc...)
func (ts *BackgroundTaskManager) DoPrioritizedTask() {
// Notify the prioritized task execution to background tasks.
ts.prioritizedTaskStartNotifyMu.Lock()
atomic.AddInt64(&ts.prioritizedTasks, 1)
close(ts.prioritizedTaskStartNotify)
ts.prioritizedTaskStartNotify = make(chan struct{})
ts.prioritizedTaskStartNotifyMu.Unlock()
}
// DonePrioritizedTask tells the manager that we've done a prioritized task
// and don't want background tasks to disturb resources(CPU, NW, etc...)
func (ts *BackgroundTaskManager) DonePrioritizedTask() {
go func() {
// Notify the task completion after `ts.prioritizedTaskSilencePeriod`
// so that background tasks aren't invoked immediately.
time.Sleep(ts.prioritizedTaskSilencePeriod)
atomic.AddInt64(&ts.prioritizedTasks, -1)
ts.prioritizedTaskDoneCond.L.Lock()
ts.prioritizedTaskDoneCond.Broadcast()
ts.prioritizedTaskDoneCond.L.Unlock()
}()
}
// InvokeBackgroundTask invokes a background task. The task is started only when
// no prioritized tasks are running. Prioritized task's execution stops the
// execution of all background tasks. Background task must be able to be
// cancelled via context.Context argument and be able to be restarted again.
func (ts *BackgroundTaskManager) InvokeBackgroundTask(do func(context.Context), timeout time.Duration) {
for {
// Wait until all prioritized tasks are done
for {
if atomic.LoadInt64(&ts.prioritizedTasks) <= 0 {
break
}
// waits until a prioritized task is done
ts.prioritizedTaskDoneCond.L.Lock()
if atomic.LoadInt64(&ts.prioritizedTasks) > 0 {
ts.prioritizedTaskDoneCond.Wait()
}
ts.prioritizedTaskDoneCond.L.Unlock()
}
// limited number of background tasks can run at once.
// if prioritized tasks are running, cancel this task.
if func() bool {
ts.backgroundSem.Acquire(context.Background(), 1)
defer ts.backgroundSem.Release(1)
// Get notify the prioritized tasks execution.
ts.prioritizedTaskStartNotifyMu.Lock()
ch := ts.prioritizedTaskStartNotify
tasks := atomic.LoadInt64(&ts.prioritizedTasks)
ts.prioritizedTaskStartNotifyMu.Unlock()
if tasks > 0 {
return false
}
// Invoke the background task. if some prioritized tasks added during
// execution, cancel it and try it later.
var (
done = make(chan struct{})
ctx, cancel = context.WithTimeout(context.Background(), timeout)
)
defer cancel()
go func() {
do(ctx)
close(done)
}()
// Wait until the background task is done or canceled.
select {
case <-ch: // some prioritized tasks started; retry it later
cancel()
return false
case <-done: // All tasks completed
}
return true
}() {
break
}
}
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package lrucache provides reference-count-aware lru cache.
package lrucache
import (
"sync"
"github.com/golang/groupcache/lru"
)
// Cache is "groupcache/lru"-like cache. The difference is that "groupcache/lru" immediately
// finalizes theevicted contents using OnEvicted callback but our version strictly tracks the
// reference counts of contents and calls OnEvicted when nobody refers to the evicted contents.
type Cache struct {
cache *lru.Cache
mu sync.Mutex
// OnEvicted optionally specifies a callback function to be
// executed when an entry is purged from the cache.
OnEvicted func(key string, value interface{})
}
// New creates new cache.
func New(maxEntries int) *Cache {
inner := lru.New(maxEntries)
inner.OnEvicted = func(key lru.Key, value interface{}) {
// Decrease the ref count incremented in Add().
// When nobody refers to this value, this value will be finalized via refCounter.
value.(*refCounter).finalize()
}
return &Cache{
cache: inner,
}
}
// Get retrieves the specified object from the cache and increments the reference counter of the
// target content. Client must call `done` callback to decrease the reference count when the value
// will no longer be used.
func (c *Cache) Get(key string) (value interface{}, done func(), ok bool) {
c.mu.Lock()
defer c.mu.Unlock()
o, ok := c.cache.Get(key)
if !ok {
return nil, nil, false
}
rc := o.(*refCounter)
rc.inc()
return rc.v, c.decreaseOnceFunc(rc), true
}
// Add adds object to the cache and returns the cached contents with incrementing the reference count.
// If the specified content already exists in the cache, this sets `added` to false and returns
// "already cached" content (i.e. doesn't replace the content with the new one). Client must call
// `done` callback to decrease the counter when the value will no longer be used.
func (c *Cache) Add(key string, value interface{}) (cachedValue interface{}, done func(), added bool) {
c.mu.Lock()
defer c.mu.Unlock()
if o, ok := c.cache.Get(key); ok {
rc := o.(*refCounter)
rc.inc()
return rc.v, c.decreaseOnceFunc(rc), false
}
rc := &refCounter{
key: key,
v: value,
onEvicted: c.OnEvicted,
}
rc.initialize() // Keep this object having at least 1 ref count (will be decreased in OnEviction)
rc.inc() // The client references this object (will be decreased on "done")
c.cache.Add(key, rc)
return rc.v, c.decreaseOnceFunc(rc), true
}
// Remove removes the specified contents from the cache. OnEvicted callback will be called when
// nobody refers to the removed content.
func (c *Cache) Remove(key string) {
c.mu.Lock()
defer c.mu.Unlock()
c.cache.Remove(key)
}
func (c *Cache) decreaseOnceFunc(rc *refCounter) func() {
var once sync.Once
return func() {
c.mu.Lock()
defer c.mu.Unlock()
once.Do(func() { rc.dec() })
}
}
type refCounter struct {
onEvicted func(key string, value interface{})
key string
v interface{}
refCounts int64
mu sync.Mutex
initializeOnce sync.Once
finalizeOnce sync.Once
}
func (r *refCounter) inc() {
r.mu.Lock()
defer r.mu.Unlock()
r.refCounts++
}
func (r *refCounter) dec() {
r.mu.Lock()
defer r.mu.Unlock()
r.refCounts--
if r.refCounts <= 0 && r.onEvicted != nil {
// nobody will refer this object
r.onEvicted(r.key, r.v)
}
}
func (r *refCounter) initialize() {
r.initializeOnce.Do(func() { r.inc() })
}
func (r *refCounter) finalize() {
r.finalizeOnce.Do(func() { r.dec() })
}
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package namedmutex provides NamedMutex that wraps sync.Mutex
// and provides namespaced mutex.
package namedmutex
import (
"sync"
)
// NamedMutex wraps sync.Mutex and provides namespaced mutex.
type NamedMutex struct {
muMap map[string]*sync.Mutex
refMap map[string]int
mu sync.Mutex
}
// Lock locks the mutex of the given name
func (nl *NamedMutex) Lock(name string) {
nl.mu.Lock()
if nl.muMap == nil {
nl.muMap = make(map[string]*sync.Mutex)
}
if nl.refMap == nil {
nl.refMap = make(map[string]int)
}
if _, ok := nl.muMap[name]; !ok {
nl.muMap[name] = &sync.Mutex{}
}
mu := nl.muMap[name]
nl.refMap[name]++
nl.mu.Unlock()
mu.Lock()
}
// Unlock unlocks the mutex of the given name
func (nl *NamedMutex) Unlock(name string) {
nl.mu.Lock()
mu := nl.muMap[name]
nl.refMap[name]--
if nl.refMap[name] <= 0 {
delete(nl.muMap, name)
delete(nl.refMap, name)
}
nl.mu.Unlock()
mu.Unlock()
}
......@@ -6,6 +6,7 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"github.com/docker/cli/cli/config/configfile"
"github.com/docker/cli/cli/config/credentials"
......@@ -23,17 +24,44 @@ const (
)
var (
configDir = os.Getenv("DOCKER_CONFIG")
initConfigDir = new(sync.Once)
configDir string
homeDir string
)
func init() {
// resetHomeDir is used in testing to reset the "homeDir" package variable to
// force re-lookup of the home directory between tests.
func resetHomeDir() {
homeDir = ""
}
func getHomeDir() string {
if homeDir == "" {
homeDir = homedir.Get()
}
return homeDir
}
// resetConfigDir is used in testing to reset the "configDir" package variable
// and its sync.Once to force re-lookup between tests.
func resetConfigDir() {
configDir = ""
initConfigDir = new(sync.Once)
}
func setConfigDir() {
if configDir != "" {
return
}
configDir = os.Getenv("DOCKER_CONFIG")
if configDir == "" {
configDir = filepath.Join(homedir.Get(), configFileDir)
configDir = filepath.Join(getHomeDir(), configFileDir)
}
}
// Dir returns the directory the configuration file is stored in
func Dir() string {
initConfigDir.Do(setConfigDir)
return configDir
}
......@@ -76,10 +104,15 @@ func LoadFromReader(configData io.Reader) (*configfile.ConfigFile, error) {
return &configFile, err
}
// TODO remove this temporary hack, which is used to warn about the deprecated ~/.dockercfg file
var printLegacyFileWarning bool
// Load reads the configuration files in the given directory, and sets up
// the auth config information and returns values.
// FIXME: use the internal golang config parser
func Load(configDir string) (*configfile.ConfigFile, error) {
printLegacyFileWarning = false
if configDir == "" {
configDir = Dir()
}
......@@ -88,11 +121,7 @@ func Load(configDir string) (*configfile.ConfigFile, error) {
configFile := configfile.New(filename)
// Try happy path first - latest config file
if _, err := os.Stat(filename); err == nil {
file, err := os.Open(filename)
if err != nil {
return configFile, errors.Wrap(err, filename)
}
if file, err := os.Open(filename); err == nil {
defer file.Close()
err = configFile.LoadFromReader(file)
if err != nil {
......@@ -106,22 +135,13 @@ func Load(configDir string) (*configfile.ConfigFile, error) {
}
// Can't find latest config file so check for the old one
homedir, err := os.UserHomeDir()
if err != nil {
return configFile, errors.Wrap(err, oldConfigfile)
}
confFile := filepath.Join(homedir, oldConfigfile)
if _, err := os.Stat(confFile); err != nil {
return configFile, nil //missing file is not an error
}
file, err := os.Open(confFile)
if err != nil {
return configFile, errors.Wrap(err, filename)
}
defer file.Close()
err = configFile.LegacyLoadFromReader(file)
if err != nil {
return configFile, errors.Wrap(err, filename)
filename = filepath.Join(getHomeDir(), oldConfigfile)
if file, err := os.Open(filename); err == nil {
printLegacyFileWarning = true
defer file.Close()
if err := configFile.LegacyLoadFromReader(file); err != nil {
return configFile, errors.Wrap(err, filename)
}
}
return configFile, nil
}
......@@ -133,6 +153,9 @@ func LoadDefaultConfigFile(stderr io.Writer) *configfile.ConfigFile {
if err != nil {
fmt.Fprintf(stderr, "WARNING: Error loading config file: %v\n", err)
}
if printLegacyFileWarning {
_, _ = fmt.Fprintln(stderr, "WARNING: Support for the legacy ~/.dockercfg configuration file and file-format is deprecated and will be removed in an upcoming release")
}
if !configFile.ContainsAuth() {
configFile.CredentialsStore = credentials.DetectDefaultStore(configFile.CredentialsStore)
}
......
......@@ -13,6 +13,7 @@ import (
"github.com/docker/cli/cli/config/credentials"
"github.com/docker/cli/cli/config/types"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
const (
......@@ -118,7 +119,7 @@ func (configFile *ConfigFile) LegacyLoadFromReader(configData io.Reader) error {
// LoadFromReader reads the configuration data given and sets up the auth config
// information with given directory and populates the receiver object
func (configFile *ConfigFile) LoadFromReader(configData io.Reader) error {
if err := json.NewDecoder(configData).Decode(&configFile); err != nil {
if err := json.NewDecoder(configData).Decode(&configFile); err != nil && !errors.Is(err, io.EOF) {
return err
}
var err error
......@@ -168,6 +169,13 @@ func (configFile *ConfigFile) SaveToWriter(writer io.Writer) error {
configFile.AuthConfigs = tmpAuthConfigs
defer func() { configFile.AuthConfigs = saveAuthConfigs }()
// User-Agent header is automatically set, and should not be stored in the configuration
for v := range configFile.HTTPHeaders {
if strings.EqualFold(v, "User-Agent") {
delete(configFile.HTTPHeaders, v)
}
}
data, err := json.MarshalIndent(configFile, "", "\t")
if err != nil {
return err
......@@ -177,7 +185,7 @@ func (configFile *ConfigFile) SaveToWriter(writer io.Writer) error {
}
// Save encodes and writes out all the authorization information
func (configFile *ConfigFile) Save() error {
func (configFile *ConfigFile) Save() (retErr error) {
if configFile.Filename == "" {
return errors.Errorf("Can't save config with empty filename")
}
......@@ -190,13 +198,33 @@ func (configFile *ConfigFile) Save() error {
if err != nil {
return err
}
defer func() {
temp.Close()
if retErr != nil {
if err := os.Remove(temp.Name()); err != nil {
logrus.WithError(err).WithField("file", temp.Name()).Debug("Error cleaning up temp file")
}
}
}()
err = configFile.SaveToWriter(temp)
temp.Close()
if err != nil {
os.Remove(temp.Name())
return err
}
return os.Rename(temp.Name(), configFile.Filename)
if err := temp.Close(); err != nil {
return errors.Wrap(err, "error closing temp file")
}
// Handle situation where the configfile is a symlink
cfgFile := configFile.Filename
if f, err := os.Readlink(cfgFile); err == nil {
cfgFile = f
}
// Try copying the current config file (if any) ownership and permissions
copyFilePermissions(cfgFile, temp.Name())
return os.Rename(temp.Name(), cfgFile)
}
// ParseProxyConfig computes proxy configuration by retrieving the config for the provided host and
......
// +build !windows
package configfile
import (
"os"
"syscall"
)
// copyFilePermissions copies file ownership and permissions from "src" to "dst",
// ignoring any error during the process.
func copyFilePermissions(src, dst string) {
var (
mode os.FileMode = 0600
uid, gid int
)
fi, err := os.Stat(src)
if err != nil {
return
}
if fi.Mode().IsRegular() {
mode = fi.Mode()
}
if err := os.Chmod(dst, mode); err != nil {
return
}
uid = int(fi.Sys().(*syscall.Stat_t).Uid)
gid = int(fi.Sys().(*syscall.Stat_t).Gid)
if uid > 0 && gid > 0 {
_ = os.Chown(dst, uid, gid)
}
}
package configfile
func copyFilePermissions(src, dst string) {
// TODO implement for Windows
}
package credentials
import (
"os/exec"
exec "golang.org/x/sys/execabs"
)
// DetectDefaultStore return the default credentials store for the platform if
......
......@@ -4,7 +4,8 @@ import (
"fmt"
"io"
"os"
"os/exec"
exec "golang.org/x/sys/execabs"
)
// Program is an interface to execute external programs.
......
package credentials
// Version holds a string describing the current version
const Version = "0.6.3"
const Version = "0.6.4"
cmd/snappytool/snappytool
testdata/bench
# These explicitly listed benchmark data files are for an obsolete version of
# snappy_test.go.
testdata/alice29.txt
testdata/asyoulik.txt
testdata/fireworks.jpeg
testdata/geo.protodata
testdata/html
testdata/html_x_4
testdata/kppkn.gtb
testdata/lcet10.txt
testdata/paper-100k.pdf
testdata/plrabn12.txt
testdata/urls.10K
# This is the official list of Snappy-Go authors for copyright purposes.
# This file is distinct from the CONTRIBUTORS files.
# See the latter for an explanation.
# Names should be added to this file as
# Name or Organization <email address>
# The email address is not required for organizations.
# Please keep the list sorted.
Amazon.com, Inc
Damian Gryski <dgryski@gmail.com>
Google Inc.
Jan Mercl <0xjnml@gmail.com>
Klaus Post <klauspost@gmail.com>
Rodolfo Carvalho <rhcarvalho@gmail.com>
Sebastien Binet <seb.binet@gmail.com>
# This is the official list of people who can contribute
# (and typically have contributed) code to the Snappy-Go repository.
# The AUTHORS file lists the copyright holders; this file
# lists people. For example, Google employees are listed here
# but not in AUTHORS, because Google holds the copyright.
#
# The submission process automatically checks to make sure
# that people submitting code are listed in this file (by email address).
#
# Names should be added to this file only after verifying that
# the individual or the individual's organization has agreed to
# the appropriate Contributor License Agreement, found here:
#
# http://code.google.com/legal/individual-cla-v1.0.html
# http://code.google.com/legal/corporate-cla-v1.0.html
#
# The agreement for individuals can be filled out on the web.
#
# When adding J Random Contributor's name to this file,
# either J's name or J's organization's name should be
# added to the AUTHORS file, depending on whether the
# individual or corporate CLA was used.
# Names should be added to this file like so:
# Name <email address>
# Please keep the list sorted.
Damian Gryski <dgryski@gmail.com>
Jan Mercl <0xjnml@gmail.com>
Jonathan Swinney <jswinney@amazon.com>
Kai Backman <kaib@golang.org>
Klaus Post <klauspost@gmail.com>
Marc-Antoine Ruel <maruel@chromium.org>
Nigel Tao <nigeltao@golang.org>
Rob Pike <r@golang.org>
Rodolfo Carvalho <rhcarvalho@gmail.com>
Russ Cox <rsc@golang.org>
Sebastien Binet <seb.binet@gmail.com>
The Snappy compression format in the Go programming language.
To download and install from source:
$ go get github.com/golang/snappy
Unless otherwise noted, the Snappy-Go source files are distributed
under the BSD-style license found in the LICENSE file.
Benchmarks.
The golang/snappy benchmarks include compressing (Z) and decompressing (U) ten
or so files, the same set used by the C++ Snappy code (github.com/google/snappy
and note the "google", not "golang"). On an "Intel(R) Core(TM) i7-3770 CPU @
3.40GHz", Go's GOARCH=amd64 numbers as of 2016-05-29:
"go test -test.bench=."
_UFlat0-8 2.19GB/s ± 0% html
_UFlat1-8 1.41GB/s ± 0% urls
_UFlat2-8 23.5GB/s ± 2% jpg
_UFlat3-8 1.91GB/s ± 0% jpg_200
_UFlat4-8 14.0GB/s ± 1% pdf
_UFlat5-8 1.97GB/s ± 0% html4
_UFlat6-8 814MB/s ± 0% txt1
_UFlat7-8 785MB/s ± 0% txt2
_UFlat8-8 857MB/s ± 0% txt3
_UFlat9-8 719MB/s ± 1% txt4
_UFlat10-8 2.84GB/s ± 0% pb
_UFlat11-8 1.05GB/s ± 0% gaviota
_ZFlat0-8 1.04GB/s ± 0% html
_ZFlat1-8 534MB/s ± 0% urls
_ZFlat2-8 15.7GB/s ± 1% jpg
_ZFlat3-8 740MB/s ± 3% jpg_200
_ZFlat4-8 9.20GB/s ± 1% pdf
_ZFlat5-8 991MB/s ± 0% html4
_ZFlat6-8 379MB/s ± 0% txt1
_ZFlat7-8 352MB/s ± 0% txt2
_ZFlat8-8 396MB/s ± 1% txt3
_ZFlat9-8 327MB/s ± 1% txt4
_ZFlat10-8 1.33GB/s ± 1% pb
_ZFlat11-8 605MB/s ± 1% gaviota
"go test -test.bench=. -tags=noasm"
_UFlat0-8 621MB/s ± 2% html
_UFlat1-8 494MB/s ± 1% urls
_UFlat2-8 23.2GB/s ± 1% jpg
_UFlat3-8 1.12GB/s ± 1% jpg_200
_UFlat4-8 4.35GB/s ± 1% pdf
_UFlat5-8 609MB/s ± 0% html4
_UFlat6-8 296MB/s ± 0% txt1
_UFlat7-8 288MB/s ± 0% txt2
_UFlat8-8 309MB/s ± 1% txt3
_UFlat9-8 280MB/s ± 1% txt4
_UFlat10-8 753MB/s ± 0% pb
_UFlat11-8 400MB/s ± 0% gaviota
_ZFlat0-8 409MB/s ± 1% html
_ZFlat1-8 250MB/s ± 1% urls
_ZFlat2-8 12.3GB/s ± 1% jpg
_ZFlat3-8 132MB/s ± 0% jpg_200
_ZFlat4-8 2.92GB/s ± 0% pdf
_ZFlat5-8 405MB/s ± 1% html4
_ZFlat6-8 179MB/s ± 1% txt1
_ZFlat7-8 170MB/s ± 1% txt2
_ZFlat8-8 189MB/s ± 1% txt3
_ZFlat9-8 164MB/s ± 1% txt4
_ZFlat10-8 479MB/s ± 1% pb
_ZFlat11-8 270MB/s ± 1% gaviota
For comparison (Go's encoded output is byte-for-byte identical to C++'s), here
are the numbers from C++ Snappy's
make CXXFLAGS="-O2 -DNDEBUG -g" clean snappy_unittest.log && cat snappy_unittest.log
BM_UFlat/0 2.4GB/s html
BM_UFlat/1 1.4GB/s urls
BM_UFlat/2 21.8GB/s jpg
BM_UFlat/3 1.5GB/s jpg_200
BM_UFlat/4 13.3GB/s pdf
BM_UFlat/5 2.1GB/s html4
BM_UFlat/6 1.0GB/s txt1
BM_UFlat/7 959.4MB/s txt2
BM_UFlat/8 1.0GB/s txt3
BM_UFlat/9 864.5MB/s txt4
BM_UFlat/10 2.9GB/s pb
BM_UFlat/11 1.2GB/s gaviota
BM_ZFlat/0 944.3MB/s html (22.31 %)
BM_ZFlat/1 501.6MB/s urls (47.78 %)
BM_ZFlat/2 14.3GB/s jpg (99.95 %)
BM_ZFlat/3 538.3MB/s jpg_200 (73.00 %)
BM_ZFlat/4 8.3GB/s pdf (83.30 %)
BM_ZFlat/5 903.5MB/s html4 (22.52 %)
BM_ZFlat/6 336.0MB/s txt1 (57.88 %)
BM_ZFlat/7 312.3MB/s txt2 (61.91 %)
BM_ZFlat/8 353.1MB/s txt3 (54.99 %)
BM_ZFlat/9 289.9MB/s txt4 (66.26 %)
BM_ZFlat/10 1.2GB/s pb (19.68 %)
BM_ZFlat/11 527.4MB/s gaviota (37.72 %)
// Copyright 2016 The Snappy-Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// +build !appengine
// +build gc
// +build !noasm
// +build amd64 arm64
package snappy
// decode has the same semantics as in decode_other.go.
//
//go:noescape
func decode(dst, src []byte) int
// Copyright 2016 The Snappy-Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// +build !appengine
// +build gc
// +build !noasm
// +build amd64 arm64
package snappy
// emitLiteral has the same semantics as in encode_other.go.
//
//go:noescape
func emitLiteral(dst, lit []byte) int
// emitCopy has the same semantics as in encode_other.go.
//
//go:noescape
func emitCopy(dst []byte, offset, length int) int
// extendMatch has the same semantics as in encode_other.go.
//
//go:noescape
func extendMatch(src []byte, i, j int) int
// encodeBlock has the same semantics as in encode_other.go.
//
//go:noescape
func encodeBlock(dst, src []byte) (d int)
Adam H. Leventhal <adam.leventhal@gmail.com>
Daniel Martí <mvdan@mvdan.cc>
Fazlul Shahriar <fshahriar@gmail.com>
Frederick Akalin <akalin@gmail.com>
Google Inc.
Haitao Li <lihaitao@gmail.com>
Jakob Unterwurzacher <jakobunt@gmail.com>
James D. Nurmi <james@abneptis.com>
Jeff <leterip@me.com>
Kaoet Ibe <kaoet.ibe@outlook.com>
Kirill Smelkov <kirr@nexedi.com>
Logan Hanks <logan@bitcasa.com>
Maria Shaldibina <mshaldibina@pivotal.io>
Nick Cooper <gh@smoogle.org>
Patrick Crosby <pcrosby@gmail.com>
Paul Jolly <paul@myitcv.org.uk>
Paul Warren <paul.warren@emc.com>
Shayan Pooya <shayan@arista.com>
Tommy Lindgren <tommy.lindgren@gmail.com>
Valient Gough <vgough@pobox.com>
Yongwoo Park <nnnlife@gmail.com>
// New BSD License
//
// Copyright (c) 2010 the Go-FUSE Authors. All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
// * Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
// * Neither the name of Ivan Krasin nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
Objective
=========
A high-performance FUSE API that minimizes pitfalls with writing
correct filesystems.
Decisions
=========
* Nodes contain references to their children. This is useful
because most filesystems will need to construct tree-like
structures.
* Nodes contain references to their parents. As a result, we can
derive the path for each Inode, and there is no need for a
separate PathFS.
* Nodes can be "persistent", meaning their lifetime is not under
control of the kernel. This is useful for constructing FS trees
in advance, rather than driven by LOOKUP.
* The NodeID for FS tree node must be defined on creation and are
immutable. By contrast, reusing NodeIds (eg. rsc/bazil FUSE, as
well as old go-fuse/fuse/nodefs) needs extra synchronization to
avoid races with notify and FORGET, and makes handling the inode
Generation more complicated.
* The mode of an Inode is defined on creation. Files cannot change
type during their lifetime. This also prevents the common error
of forgetting to return the filetype in Lookup/GetAttr.
* The NodeID (used for communicating with kernel) is equal to
Attr.Ino (value shown in Stat and Lstat return values.).
* No global treelock, to ensure scalability.
* Support for hard links. libfuse doesn't support this in the
high-level API. Extra care for race conditions is needed when
looking up the same file through different paths.
* do not issue Notify{Entry,Delete} as part of
AddChild/RmChild/MvChild: because NodeIDs are unique and
immutable, there is no confusion about which nodes are
invalidated, and the notification doesn't have to happen under
lock.
* Directory reading uses the DirStream. Semantics for rewinding
directory reads, and adding files after opening (but before
reading) are handled automatically. No support for directory
seeks.
* Method names are based on syscall names. Where there is no
syscall (eg. "open directory"), we bias towards writing
everything together (Opendir)
To do/To decide
=========
* Symlink []byte vs string.
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import (
"syscall"
"github.com/hanwen/go-fuse/v2/fuse"
)
// OK is the Errno return value to indicate absense of errors.
var OK = syscall.Errno(0)
// ToErrno exhumes the syscall.Errno error from wrapped error values.
func ToErrno(err error) syscall.Errno {
s := fuse.ToStatus(err)
return syscall.Errno(s)
}
// RENAME_EXCHANGE is a flag argument for renameat2()
const RENAME_EXCHANGE = 0x2
// seek to the next data
const _SEEK_DATA = 3
// seek to the next hole
const _SEEK_HOLE = 4
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import "syscall"
// ENOATTR indicates that an extended attribute was not present.
var ENOATTR = syscall.ENOATTR
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import "syscall"
// ENOATTR indicates that an extended attribute was not present.
var ENOATTR = syscall.ENODATA
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import (
"syscall"
"github.com/hanwen/go-fuse/v2/fuse"
)
type dirArray struct {
entries []fuse.DirEntry
}
func (a *dirArray) HasNext() bool {
return len(a.entries) > 0
}
func (a *dirArray) Next() (fuse.DirEntry, syscall.Errno) {
e := a.entries[0]
a.entries = a.entries[1:]
return e, 0
}
func (a *dirArray) Close() {
}
// NewListDirStream wraps a slice of DirEntry as a DirStream.
func NewListDirStream(list []fuse.DirEntry) DirStream {
return &dirArray{list}
}
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import (
"io"
"os"
"syscall"
"github.com/hanwen/go-fuse/v2/fuse"
)
func NewLoopbackDirStream(nm string) (DirStream, syscall.Errno) {
f, err := os.Open(nm)
if err != nil {
return nil, ToErrno(err)
}
defer f.Close()
var entries []fuse.DirEntry
for {
want := 100
infos, err := f.Readdir(want)
for _, info := range infos {
s := fuse.ToStatT(info)
if s == nil {
continue
}
entries = append(entries, fuse.DirEntry{
Name: info.Name(),
Mode: uint32(s.Mode),
Ino: s.Ino,
})
}
if len(infos) < want || err == io.EOF {
break
}
if err != nil {
return nil, ToErrno(err)
}
}
return &dirArray{entries}, OK
}
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import (
"sync"
"syscall"
"unsafe"
"github.com/hanwen/go-fuse/v2/fuse"
)
type loopbackDirStream struct {
buf []byte
todo []byte
// Protects fd so we can guard against double close
mu sync.Mutex
fd int
}
// NewLoopbackDirStream open a directory for reading as a DirStream
func NewLoopbackDirStream(name string) (DirStream, syscall.Errno) {
fd, err := syscall.Open(name, syscall.O_DIRECTORY, 0755)
if err != nil {
return nil, ToErrno(err)
}
ds := &loopbackDirStream{
buf: make([]byte, 4096),
fd: fd,
}
if err := ds.load(); err != 0 {
ds.Close()
return nil, err
}
return ds, OK
}
func (ds *loopbackDirStream) Close() {
ds.mu.Lock()
defer ds.mu.Unlock()
if ds.fd != -1 {
syscall.Close(ds.fd)
ds.fd = -1
}
}
func (ds *loopbackDirStream) HasNext() bool {
ds.mu.Lock()
defer ds.mu.Unlock()
return len(ds.todo) > 0
}
// Like syscall.Dirent, but without the [256]byte name.
type dirent struct {
Ino uint64
Off int64
Reclen uint16
Type uint8
Name [1]uint8 // align to 4 bytes for 32 bits.
}
func (ds *loopbackDirStream) Next() (fuse.DirEntry, syscall.Errno) {
ds.mu.Lock()
defer ds.mu.Unlock()
// We can't use syscall.Dirent here, because it declares a
// [256]byte name, which may run beyond the end of ds.todo.
// when that happens in the race detector, it causes a panic
// "converted pointer straddles multiple allocations"
de := (*dirent)(unsafe.Pointer(&ds.todo[0]))
nameBytes := ds.todo[unsafe.Offsetof(dirent{}.Name):de.Reclen]
ds.todo = ds.todo[de.Reclen:]
// After the loop, l contains the index of the first '\0'.
l := 0
for l = range nameBytes {
if nameBytes[l] == 0 {
break
}
}
nameBytes = nameBytes[:l]
result := fuse.DirEntry{
Ino: de.Ino,
Mode: (uint32(de.Type) << 12),
Name: string(nameBytes),
}
return result, ds.load()
}
func (ds *loopbackDirStream) load() syscall.Errno {
if len(ds.todo) > 0 {
return OK
}
n, err := syscall.Getdents(ds.fd, ds.buf)
if err != nil {
return ToErrno(err)
}
ds.todo = ds.buf[:n]
return OK
}
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import (
"context"
"sync"
// "time"
"syscall"
"github.com/hanwen/go-fuse/v2/fuse"
"golang.org/x/sys/unix"
)
// NewLoopbackFile creates a FileHandle out of a file descriptor. All
// operations are implemented. When using the Fd from a *os.File, call
// syscall.Dup() on the fd, to avoid os.File's finalizer from closing
// the file descriptor.
func NewLoopbackFile(fd int) FileHandle {
return &loopbackFile{fd: fd}
}
type loopbackFile struct {
mu sync.Mutex
fd int
}
var _ = (FileHandle)((*loopbackFile)(nil))
var _ = (FileReleaser)((*loopbackFile)(nil))
var _ = (FileGetattrer)((*loopbackFile)(nil))
var _ = (FileReader)((*loopbackFile)(nil))
var _ = (FileWriter)((*loopbackFile)(nil))
var _ = (FileGetlker)((*loopbackFile)(nil))
var _ = (FileSetlker)((*loopbackFile)(nil))
var _ = (FileSetlkwer)((*loopbackFile)(nil))
var _ = (FileLseeker)((*loopbackFile)(nil))
var _ = (FileFlusher)((*loopbackFile)(nil))
var _ = (FileFsyncer)((*loopbackFile)(nil))
var _ = (FileSetattrer)((*loopbackFile)(nil))
var _ = (FileAllocater)((*loopbackFile)(nil))
func (f *loopbackFile) Read(ctx context.Context, buf []byte, off int64) (res fuse.ReadResult, errno syscall.Errno) {
f.mu.Lock()
defer f.mu.Unlock()
r := fuse.ReadResultFd(uintptr(f.fd), off, len(buf))
return r, OK
}
func (f *loopbackFile) Write(ctx context.Context, data []byte, off int64) (uint32, syscall.Errno) {
f.mu.Lock()
defer f.mu.Unlock()
n, err := syscall.Pwrite(f.fd, data, off)
return uint32(n), ToErrno(err)
}
func (f *loopbackFile) Release(ctx context.Context) syscall.Errno {
f.mu.Lock()
defer f.mu.Unlock()
if f.fd != -1 {
err := syscall.Close(f.fd)
f.fd = -1
return ToErrno(err)
}
return syscall.EBADF
}
func (f *loopbackFile) Flush(ctx context.Context) syscall.Errno {
f.mu.Lock()
defer f.mu.Unlock()
// Since Flush() may be called for each dup'd fd, we don't
// want to really close the file, we just want to flush. This
// is achieved by closing a dup'd fd.
newFd, err := syscall.Dup(f.fd)
if err != nil {
return ToErrno(err)
}
err = syscall.Close(newFd)
return ToErrno(err)
}
func (f *loopbackFile) Fsync(ctx context.Context, flags uint32) (errno syscall.Errno) {
f.mu.Lock()
defer f.mu.Unlock()
r := ToErrno(syscall.Fsync(f.fd))
return r
}
const (
_OFD_GETLK = 36
_OFD_SETLK = 37
_OFD_SETLKW = 38
)
func (f *loopbackFile) Getlk(ctx context.Context, owner uint64, lk *fuse.FileLock, flags uint32, out *fuse.FileLock) (errno syscall.Errno) {
f.mu.Lock()
defer f.mu.Unlock()
flk := syscall.Flock_t{}
lk.ToFlockT(&flk)
errno = ToErrno(syscall.FcntlFlock(uintptr(f.fd), _OFD_GETLK, &flk))
out.FromFlockT(&flk)
return
}
func (f *loopbackFile) Setlk(ctx context.Context, owner uint64, lk *fuse.FileLock, flags uint32) (errno syscall.Errno) {
return f.setLock(ctx, owner, lk, flags, false)
}
func (f *loopbackFile) Setlkw(ctx context.Context, owner uint64, lk *fuse.FileLock, flags uint32) (errno syscall.Errno) {
return f.setLock(ctx, owner, lk, flags, true)
}
func (f *loopbackFile) setLock(ctx context.Context, owner uint64, lk *fuse.FileLock, flags uint32, blocking bool) (errno syscall.Errno) {
f.mu.Lock()
defer f.mu.Unlock()
if (flags & fuse.FUSE_LK_FLOCK) != 0 {
var op int
switch lk.Typ {
case syscall.F_RDLCK:
op = syscall.LOCK_SH
case syscall.F_WRLCK:
op = syscall.LOCK_EX
case syscall.F_UNLCK:
op = syscall.LOCK_UN
default:
return syscall.EINVAL
}
if !blocking {
op |= syscall.LOCK_NB
}
return ToErrno(syscall.Flock(f.fd, op))
} else {
flk := syscall.Flock_t{}
lk.ToFlockT(&flk)
var op int
if blocking {
op = _OFD_SETLKW
} else {
op = _OFD_SETLK
}
return ToErrno(syscall.FcntlFlock(uintptr(f.fd), op, &flk))
}
}
func (f *loopbackFile) Setattr(ctx context.Context, in *fuse.SetAttrIn, out *fuse.AttrOut) syscall.Errno {
if errno := f.setAttr(ctx, in); errno != 0 {
return errno
}
return f.Getattr(ctx, out)
}
func (f *loopbackFile) setAttr(ctx context.Context, in *fuse.SetAttrIn) syscall.Errno {
f.mu.Lock()
defer f.mu.Unlock()
var errno syscall.Errno
if mode, ok := in.GetMode(); ok {
errno = ToErrno(syscall.Fchmod(f.fd, mode))
if errno != 0 {
return errno
}
}
uid32, uOk := in.GetUID()
gid32, gOk := in.GetGID()
if uOk || gOk {
uid := -1
gid := -1
if uOk {
uid = int(uid32)
}
if gOk {
gid = int(gid32)
}
errno = ToErrno(syscall.Fchown(f.fd, uid, gid))
if errno != 0 {
return errno
}
}
mtime, mok := in.GetMTime()
atime, aok := in.GetATime()
if mok || aok {
ap := &atime
mp := &mtime
if !aok {
ap = nil
}
if !mok {
mp = nil
}
errno = f.utimens(ap, mp)
if errno != 0 {
return errno
}
}
if sz, ok := in.GetSize(); ok {
errno = ToErrno(syscall.Ftruncate(f.fd, int64(sz)))
if errno != 0 {
return errno
}
}
return OK
}
func (f *loopbackFile) Getattr(ctx context.Context, a *fuse.AttrOut) syscall.Errno {
f.mu.Lock()
defer f.mu.Unlock()
st := syscall.Stat_t{}
err := syscall.Fstat(f.fd, &st)
if err != nil {
return ToErrno(err)
}
a.FromStat(&st)
return OK
}
func (f *loopbackFile) Lseek(ctx context.Context, off uint64, whence uint32) (uint64, syscall.Errno) {
f.mu.Lock()
defer f.mu.Unlock()
n, err := unix.Seek(f.fd, int64(off), int(whence))
return uint64(n), ToErrno(err)
}
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import "github.com/hanwen/go-fuse/v2/fuse"
func setBlocks(out *fuse.Attr) {
}
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import (
"context"
"syscall"
"time"
"github.com/hanwen/go-fuse/v2/fuse"
)
func (f *loopbackFile) Allocate(ctx context.Context, off uint64, sz uint64, mode uint32) syscall.Errno {
f.mu.Lock()
defer f.mu.Unlock()
err := syscall.Fallocate(f.fd, mode, int64(off), int64(sz))
if err != nil {
return ToErrno(err)
}
return OK
}
// Utimens - file handle based version of loopbackFileSystem.Utimens()
func (f *loopbackFile) utimens(a *time.Time, m *time.Time) syscall.Errno {
var ts [2]syscall.Timespec
ts[0] = fuse.UtimeToTimespec(a)
ts[1] = fuse.UtimeToTimespec(m)
err := futimens(int(f.fd), &ts)
return ToErrno(err)
}
func setBlocks(out *fuse.Attr) {
if out.Blksize > 0 {
return
}
out.Blksize = 4096
pages := (out.Size + 4095) / 4096
out.Blocks = pages * 8
}
// +build darwin
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import (
"context"
"syscall"
"time"
"unsafe"
"github.com/hanwen/go-fuse/v2/fuse"
"github.com/hanwen/go-fuse/v2/internal/utimens"
)
func (n *LoopbackNode) Getxattr(ctx context.Context, attr string, dest []byte) (uint32, syscall.Errno) {
return 0, syscall.ENOSYS
}
func (n *LoopbackNode) Setxattr(ctx context.Context, attr string, data []byte, flags uint32) syscall.Errno {
return syscall.ENOSYS
}
func (n *LoopbackNode) Removexattr(ctx context.Context, attr string) syscall.Errno {
return syscall.ENOSYS
}
func (n *LoopbackNode) Listxattr(ctx context.Context, dest []byte) (uint32, syscall.Errno) {
return 0, syscall.ENOSYS
}
func (n *LoopbackNode) renameExchange(name string, newparent InodeEmbedder, newName string) syscall.Errno {
return syscall.ENOSYS
}
func (f *loopbackFile) Allocate(ctx context.Context, off uint64, sz uint64, mode uint32) syscall.Errno {
// TODO: Handle `mode` parameter.
// From `man fcntl` on OSX:
// The F_PREALLOCATE command operates on the following structure:
//
// typedef struct fstore {
// u_int32_t fst_flags; /* IN: flags word */
// int fst_posmode; /* IN: indicates offset field */
// off_t fst_offset; /* IN: start of the region */
// off_t fst_length; /* IN: size of the region */
// off_t fst_bytesalloc; /* OUT: number of bytes allocated */
// } fstore_t;
//
// The flags (fst_flags) for the F_PREALLOCATE command are as follows:
//
// F_ALLOCATECONTIG Allocate contiguous space.
//
// F_ALLOCATEALL Allocate all requested space or no space at all.
//
// The position modes (fst_posmode) for the F_PREALLOCATE command indicate how to use the offset field. The modes are as fol-
// lows:
//
// F_PEOFPOSMODE Allocate from the physical end of file.
//
// F_VOLPOSMODE Allocate from the volume offset.
k := struct {
Flags uint32 // u_int32_t
Posmode int64 // int
Offset int64 // off_t
Length int64 // off_t
Bytesalloc int64 // off_t
}{
0,
0,
int64(off),
int64(sz),
0,
}
// Linux version for reference:
// err := syscall.Fallocate(int(f.File.Fd()), mode, int64(off), int64(sz))
_, _, errno := syscall.Syscall(syscall.SYS_FCNTL, uintptr(f.fd), uintptr(syscall.F_PREALLOCATE), uintptr(unsafe.Pointer(&k)))
return errno
}
// timeToTimeval - Convert time.Time to syscall.Timeval
//
// Note: This does not use syscall.NsecToTimespec because
// that does not work properly for times before 1970,
// see https://github.com/golang/go/issues/12777
func timeToTimeval(t *time.Time) syscall.Timeval {
var tv syscall.Timeval
tv.Usec = int32(t.Nanosecond() / 1000)
tv.Sec = t.Unix()
return tv
}
// MacOS before High Sierra lacks utimensat() and UTIME_OMIT.
// We emulate using utimes() and extra Getattr() calls.
func (f *loopbackFile) utimens(a *time.Time, m *time.Time) syscall.Errno {
var attr fuse.AttrOut
if a == nil || m == nil {
errno := f.Getattr(context.Background(), &attr)
if errno != 0 {
return errno
}
}
tv := utimens.Fill(a, m, &attr.Attr)
err := syscall.Futimes(int(f.fd), tv)
return ToErrno(err)
}
func (n *LoopbackNode) CopyFileRange(ctx context.Context, fhIn FileHandle,
offIn uint64, out *Inode, fhOut FileHandle, offOut uint64,
len uint64, flags uint64) (uint32, syscall.Errno) {
return 0, syscall.ENOSYS
}
// +build linux
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import (
"context"
"path/filepath"
"syscall"
"golang.org/x/sys/unix"
)
func (n *LoopbackNode) Getxattr(ctx context.Context, attr string, dest []byte) (uint32, syscall.Errno) {
sz, err := unix.Lgetxattr(n.path(), attr, dest)
return uint32(sz), ToErrno(err)
}
func (n *LoopbackNode) Setxattr(ctx context.Context, attr string, data []byte, flags uint32) syscall.Errno {
err := unix.Lsetxattr(n.path(), attr, data, int(flags))
return ToErrno(err)
}
func (n *LoopbackNode) Removexattr(ctx context.Context, attr string) syscall.Errno {
err := unix.Lremovexattr(n.path(), attr)
return ToErrno(err)
}
func (n *LoopbackNode) Listxattr(ctx context.Context, dest []byte) (uint32, syscall.Errno) {
sz, err := unix.Llistxattr(n.path(), dest)
return uint32(sz), ToErrno(err)
}
func (n *LoopbackNode) renameExchange(name string, newparent InodeEmbedder, newName string) syscall.Errno {
fd1, err := syscall.Open(n.path(), syscall.O_DIRECTORY, 0)
if err != nil {
return ToErrno(err)
}
defer syscall.Close(fd1)
p2 := filepath.Join(n.RootData.Path, newparent.EmbeddedInode().Path(nil))
fd2, err := syscall.Open(p2, syscall.O_DIRECTORY, 0)
defer syscall.Close(fd2)
if err != nil {
return ToErrno(err)
}
var st syscall.Stat_t
if err := syscall.Fstat(fd1, &st); err != nil {
return ToErrno(err)
}
// Double check that nodes didn't change from under us.
inode := &n.Inode
if inode.Root() != inode && inode.StableAttr().Ino != n.RootData.idFromStat(&st).Ino {
return syscall.EBUSY
}
if err := syscall.Fstat(fd2, &st); err != nil {
return ToErrno(err)
}
newinode := newparent.EmbeddedInode()
if newinode.Root() != newinode && newinode.StableAttr().Ino != n.RootData.idFromStat(&st).Ino {
return syscall.EBUSY
}
return ToErrno(unix.Renameat2(fd1, name, fd2, newName, unix.RENAME_EXCHANGE))
}
func (n *LoopbackNode) CopyFileRange(ctx context.Context, fhIn FileHandle,
offIn uint64, out *Inode, fhOut FileHandle, offOut uint64,
len uint64, flags uint64) (uint32, syscall.Errno) {
lfIn, ok := fhIn.(*loopbackFile)
if !ok {
return 0, syscall.ENOTSUP
}
lfOut, ok := fhOut.(*loopbackFile)
if !ok {
return 0, syscall.ENOTSUP
}
signedOffIn := int64(offIn)
signedOffOut := int64(offOut)
count, err := unix.CopyFileRange(lfIn.fd, &signedOffIn, lfOut.fd, &signedOffOut, int(len), int(flags))
return uint32(count), ToErrno(err)
}
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import (
"context"
"sync"
"syscall"
"github.com/hanwen/go-fuse/v2/fuse"
)
// MemRegularFile is a filesystem node that holds a read-only data
// slice in memory.
type MemRegularFile struct {
Inode
mu sync.Mutex
Data []byte
Attr fuse.Attr
}
var _ = (NodeOpener)((*MemRegularFile)(nil))
var _ = (NodeReader)((*MemRegularFile)(nil))
var _ = (NodeWriter)((*MemRegularFile)(nil))
var _ = (NodeSetattrer)((*MemRegularFile)(nil))
var _ = (NodeFlusher)((*MemRegularFile)(nil))
func (f *MemRegularFile) Open(ctx context.Context, flags uint32) (fh FileHandle, fuseFlags uint32, errno syscall.Errno) {
return nil, fuse.FOPEN_KEEP_CACHE, OK
}
func (f *MemRegularFile) Write(ctx context.Context, fh FileHandle, data []byte, off int64) (uint32, syscall.Errno) {
f.mu.Lock()
defer f.mu.Unlock()
end := int64(len(data)) + off
if int64(len(f.Data)) < end {
n := make([]byte, end)
copy(n, f.Data)
f.Data = n
}
copy(f.Data[off:off+int64(len(data))], data)
return uint32(len(data)), 0
}
var _ = (NodeGetattrer)((*MemRegularFile)(nil))
func (f *MemRegularFile) Getattr(ctx context.Context, fh FileHandle, out *fuse.AttrOut) syscall.Errno {
f.mu.Lock()
defer f.mu.Unlock()
out.Attr = f.Attr
out.Attr.Size = uint64(len(f.Data))
return OK
}
func (f *MemRegularFile) Setattr(ctx context.Context, fh FileHandle, in *fuse.SetAttrIn, out *fuse.AttrOut) syscall.Errno {
f.mu.Lock()
defer f.mu.Unlock()
if sz, ok := in.GetSize(); ok {
f.Data = f.Data[:sz]
}
out.Attr = f.Attr
out.Size = uint64(len(f.Data))
return OK
}
func (f *MemRegularFile) Flush(ctx context.Context, fh FileHandle) syscall.Errno {
return 0
}
func (f *MemRegularFile) Read(ctx context.Context, fh FileHandle, dest []byte, off int64) (fuse.ReadResult, syscall.Errno) {
f.mu.Lock()
defer f.mu.Unlock()
end := int(off) + len(dest)
if end > len(f.Data) {
end = len(f.Data)
}
return fuse.ReadResultData(f.Data[off:end]), OK
}
// MemSymlink is an inode holding a symlink in memory.
type MemSymlink struct {
Inode
Attr fuse.Attr
Data []byte
}
var _ = (NodeReadlinker)((*MemSymlink)(nil))
func (l *MemSymlink) Readlink(ctx context.Context) ([]byte, syscall.Errno) {
return l.Data, OK
}
var _ = (NodeGetattrer)((*MemSymlink)(nil))
func (l *MemSymlink) Getattr(ctx context.Context, fh FileHandle, out *fuse.AttrOut) syscall.Errno {
out.Attr = l.Attr
return OK
}
// Copyright 2019 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import (
"time"
"github.com/hanwen/go-fuse/v2/fuse"
)
// Mount mounts the given NodeFS on the directory, and starts serving
// requests. This is a convenience wrapper around NewNodeFS and
// fuse.NewServer. If nil is given as options, default settings are
// applied, which are 1 second entry and attribute timeout.
func Mount(dir string, root InodeEmbedder, options *Options) (*fuse.Server, error) {
if options == nil {
oneSec := time.Second
options = &Options{
EntryTimeout: &oneSec,
AttrTimeout: &oneSec,
}
}
rawFS := NewNodeFS(root, options)
server, err := fuse.NewServer(rawFS, dir, &options.MountOptions)
if err != nil {
return nil, err
}
go server.Serve()
if err := server.WaitMount(); err != nil {
// we don't shutdown the serve loop. If the mount does
// not succeed, the loop won't work and exit.
return nil, err
}
return server, nil
}
// Copyright 2016 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fs
import (
"syscall"
"unsafe"
)
// futimens - futimens(3) calls utimensat(2) with "pathname" set to null and
// "flags" set to zero
func futimens(fd int, times *[2]syscall.Timespec) (err error) {
_, _, e1 := syscall.Syscall6(syscall.SYS_UTIMENSAT, uintptr(fd), 0, uintptr(unsafe.Pointer(times)), uintptr(0), 0, 0)
if e1 != 0 {
err = syscall.Errno(e1)
}
return
}
// Copyright 2016 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fuse
import (
"os"
"syscall"
"time"
)
func (a *Attr) IsFifo() bool { return (uint32(a.Mode) & syscall.S_IFMT) == syscall.S_IFIFO }
// IsChar reports whether the FileInfo describes a character special file.
func (a *Attr) IsChar() bool { return (uint32(a.Mode) & syscall.S_IFMT) == syscall.S_IFCHR }
// IsDir reports whether the FileInfo describes a directory.
func (a *Attr) IsDir() bool { return (uint32(a.Mode) & syscall.S_IFMT) == syscall.S_IFDIR }
// IsBlock reports whether the FileInfo describes a block special file.
func (a *Attr) IsBlock() bool { return (uint32(a.Mode) & syscall.S_IFMT) == syscall.S_IFBLK }
// IsRegular reports whether the FileInfo describes a regular file.
func (a *Attr) IsRegular() bool { return (uint32(a.Mode) & syscall.S_IFMT) == syscall.S_IFREG }
// IsSymlink reports whether the FileInfo describes a symbolic link.
func (a *Attr) IsSymlink() bool { return (uint32(a.Mode) & syscall.S_IFMT) == syscall.S_IFLNK }
// IsSocket reports whether the FileInfo describes a socket.
func (a *Attr) IsSocket() bool { return (uint32(a.Mode) & syscall.S_IFMT) == syscall.S_IFSOCK }
func (a *Attr) SetTimes(access *time.Time, mod *time.Time, chstatus *time.Time) {
if access != nil {
a.Atime = uint64(access.Unix())
a.Atimensec = uint32(access.Nanosecond())
}
if mod != nil {
a.Mtime = uint64(mod.Unix())
a.Mtimensec = uint32(mod.Nanosecond())
}
if chstatus != nil {
a.Ctime = uint64(chstatus.Unix())
a.Ctimensec = uint32(chstatus.Nanosecond())
}
}
func (a *Attr) ChangeTime() time.Time {
return time.Unix(int64(a.Ctime), int64(a.Ctimensec))
}
func (a *Attr) AccessTime() time.Time {
return time.Unix(int64(a.Atime), int64(a.Atimensec))
}
func (a *Attr) ModTime() time.Time {
return time.Unix(int64(a.Mtime), int64(a.Mtimensec))
}
func ToStatT(f os.FileInfo) *syscall.Stat_t {
s, _ := f.Sys().(*syscall.Stat_t)
if s != nil {
return s
}
return nil
}
func ToAttr(f os.FileInfo) *Attr {
if f == nil {
return nil
}
s := ToStatT(f)
if s != nil {
a := &Attr{}
a.FromStat(s)
return a
}
return nil
}
// Copyright 2016 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fuse
import (
"syscall"
)
func (a *Attr) FromStat(s *syscall.Stat_t) {
a.Ino = uint64(s.Ino)
a.Size = uint64(s.Size)
a.Blocks = uint64(s.Blocks)
a.Atime = uint64(s.Atimespec.Sec)
a.Atimensec = uint32(s.Atimespec.Nsec)
a.Mtime = uint64(s.Mtimespec.Sec)
a.Mtimensec = uint32(s.Mtimespec.Nsec)
a.Ctime = uint64(s.Ctimespec.Sec)
a.Ctimensec = uint32(s.Ctimespec.Nsec)
a.Mode = uint32(s.Mode)
a.Nlink = uint32(s.Nlink)
a.Uid = uint32(s.Uid)
a.Gid = uint32(s.Gid)
a.Rdev = uint32(s.Rdev)
}
// Copyright 2016 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fuse
import (
"syscall"
)
func (a *Attr) FromStat(s *syscall.Stat_t) {
a.Ino = uint64(s.Ino)
a.Size = uint64(s.Size)
a.Blocks = uint64(s.Blocks)
a.Atime = uint64(s.Atim.Sec)
a.Atimensec = uint32(s.Atim.Nsec)
a.Mtime = uint64(s.Mtim.Sec)
a.Mtimensec = uint32(s.Mtim.Nsec)
a.Ctime = uint64(s.Ctim.Sec)
a.Ctimensec = uint32(s.Ctim.Nsec)
a.Mode = s.Mode
a.Nlink = uint32(s.Nlink)
a.Uid = uint32(s.Uid)
a.Gid = uint32(s.Gid)
a.Rdev = uint32(s.Rdev)
a.Blksize = uint32(s.Blksize)
}
// Copyright 2016 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fuse
import (
"os"
"sync"
)
// bufferPool implements explicit memory management. It is used for
// minimizing the GC overhead of communicating with the kernel.
type bufferPool struct {
lock sync.Mutex
// For each page size multiple a list of slice pointers.
buffersBySize []*sync.Pool
}
var pageSize = os.Getpagesize()
func (p *bufferPool) getPool(pageCount int) *sync.Pool {
p.lock.Lock()
for len(p.buffersBySize) < pageCount+1 {
p.buffersBySize = append(p.buffersBySize, nil)
}
if p.buffersBySize[pageCount] == nil {
p.buffersBySize[pageCount] = &sync.Pool{
New: func() interface{} { return make([]byte, pageSize*pageCount) },
}
}
pool := p.buffersBySize[pageCount]
p.lock.Unlock()
return pool
}
// AllocBuffer creates a buffer of at least the given size. After use,
// it should be deallocated with FreeBuffer().
func (p *bufferPool) AllocBuffer(size uint32) []byte {
sz := int(size)
if sz < pageSize {
sz = pageSize
}
if sz%pageSize != 0 {
sz += pageSize
}
pages := sz / pageSize
b := p.getPool(pages).Get().([]byte)
return b[:size]
}
// FreeBuffer takes back a buffer if it was allocated through
// AllocBuffer. It is not an error to call FreeBuffer() on a slice
// obtained elsewhere.
func (p *bufferPool) FreeBuffer(slice []byte) {
if slice == nil {
return
}
if cap(slice)%pageSize != 0 || cap(slice) == 0 {
return
}
pages := cap(slice) / pageSize
slice = slice[:cap(slice)]
p.getPool(pages).Put(slice)
}
// Copyright 2016 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fuse
import (
"os"
"syscall"
)
const (
FUSE_ROOT_ID = 1
FUSE_UNKNOWN_INO = 0xffffffff
CUSE_UNRESTRICTED_IOCTL = (1 << 0)
FUSE_LK_FLOCK = (1 << 0)
FUSE_IOCTL_MAX_IOV = 256
FUSE_POLL_SCHEDULE_NOTIFY = (1 << 0)
CUSE_INIT_INFO_MAX = 4096
S_IFDIR = syscall.S_IFDIR
S_IFREG = syscall.S_IFREG
S_IFLNK = syscall.S_IFLNK
S_IFIFO = syscall.S_IFIFO
CUSE_INIT = 4096
O_ANYWRITE = uint32(os.O_WRONLY | os.O_RDWR | os.O_APPEND | os.O_CREATE | os.O_TRUNC)
logicalBlockSize = 512
)
// Copyright 2016 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fuse
// arbitrary values
const syscall_O_LARGEFILE = 1 << 29
const syscall_O_NOATIME = 1 << 30
// Copyright 2016 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fuse
import (
"syscall"
)
const syscall_O_LARGEFILE = syscall.O_LARGEFILE
const syscall_O_NOATIME = syscall.O_NOATIME
// Copyright 2016 the Go-FUSE Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package fuse
import (
"context"
"time"
)
// Context passes along cancelation signal and request data (PID, GID,
// UID). The name of this class predates the standard "context"
// package from Go, but it does implement the context.Context
// interface.
//
// When a FUSE request is canceled, the API routine should respond by
// returning the EINTR status code.
type Context struct {
Caller
Cancel <-chan struct{}
}
func (c *Context) Deadline() (time.Time, bool) {
return time.Time{}, false
}
func (c *Context) Done() <-chan struct{} {
return c.Cancel
}
func (c *Context) Err() error {
select {
case <-c.Cancel:
return context.Canceled
default:
return nil
}
}
type callerKeyType struct{}
var callerKey callerKeyType
func FromContext(ctx context.Context) (*Caller, bool) {
v, ok := ctx.Value(callerKey).(*Caller)
return v, ok
}
func NewContext(ctx context.Context, caller *Caller) context.Context {
return context.WithValue(ctx, callerKey, caller)
}
func (c *Context) Value(key interface{}) interface{} {
if key == callerKey {
return &c.Caller
}
return nil
}
var _ = context.Context((*Context)(nil))
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment