Add fuse-manager

Signed-off-by: Zuti He <ilyeeelihe@gmail.com>

Makefile

@@ -23,7 +23,7 @@ VERSION=$(shell git describe --match 'v[0-9]*' --dirty='.m' --always --tags)
 REVISION=$(shell git rev-parse HEAD)$(shell if ! git diff --no-ext-diff --quiet --exit-code; then echo .m; fi)
 GO_LD_FLAGS=-ldflags '-s -w -X $(PKG)/version.Version=$(VERSION) -X $(PKG)/version.Revision=$(REVISION) $(GO_EXTRA_LDFLAGS)'
 
-CMD=containerd-stargz-grpc ctr-remote stargz-store
+CMD=containerd-stargz-grpc ctr-remote stargz-store stargz-fuse-manager
 
 CMD_BINARIES=$(addprefix $(PREFIX),$(CMD))
 
@@ -44,6 +44,9 @@ ctr-remote: FORCE
 stargz-store: FORCE
 	GO111MODULE=$(GO111MODULE_VALUE) go build -o $(PREFIX)$@ $(GO_BUILD_FLAGS) $(GO_LD_FLAGS) -v ./cmd/stargz-store
 
+stargz-fuse-manager: FORCE
+	GO111MODULE=$(GO111MODULE_VALUE) go build -o $(PREFIX)$@ $(GO_BUILD_FLAGS) $(GO_LD_FLAGS) -v ./cmd/stargz-fuse-manager
+
 check:
 	@echo "$@"
 	@GO111MODULE=$(GO111MODULE_VALUE) golangci-lint run

cmd/containerd-stargz-grpc/main.go

@@ -39,6 +39,7 @@ import (
 	"github.com/containerd/stargz-snapshotter/service/keychain/dockerconfig"
 	"github.com/containerd/stargz-snapshotter/service/keychain/kubeconfig"
 	"github.com/containerd/stargz-snapshotter/service/resolver"
+	fusemanager "github.com/containerd/stargz-snapshotter/stargzfusemanager"
 	"github.com/containerd/stargz-snapshotter/version"
 	sddaemon "github.com/coreos/go-systemd/v22/daemon"
 	metrics "github.com/docker/go-metrics"
@@ -60,11 +61,13 @@ const (
 )
 
 var (
-	address      = flag.String("address", defaultAddress, "address for the snapshotter's GRPC server")
-	configPath   = flag.String("config", defaultConfigPath, "path to the configuration file")
-	logLevel     = flag.String("log-level", defaultLogLevel.String(), "set the logging level [trace, debug, info, warn, error, fatal, panic]")
-	rootDir      = flag.String("root", defaultRootDir, "path to the root directory for this snapshotter")
-	printVersion = flag.Bool("version", false, "print the version")
+	address         = flag.String("address", defaultAddress, "address for the snapshotter's GRPC server")
+	configPath      = flag.String("config", defaultConfigPath, "path to the configuration file")
+	logLevel        = flag.String("log-level", defaultLogLevel.String(), "set the logging level [trace, debug, info, warn, error, fatal, panic]")
+	rootDir         = flag.String("root", defaultRootDir, "path to the root directory for this snapshotter")
+	fuseManagerAddr = flag.String("fusemanager-address", "", "address for the fusemanager's GRPC server")
+	fuseManagerPath = flag.String("fusemanager-path", "", "path to the fusemanager's executable")
+	printVersion    = flag.Bool("version", false, "print the version")
 )
 
 type snapshotterConfig struct {
@@ -153,7 +156,21 @@ func main() {
 		runtime.RegisterImageServiceServer(rpc, criServer)
 		credsFuncs = append(credsFuncs, f)
 	}
-	rs, err := service.NewStargzSnapshotterService(ctx, *rootDir, &config.Config, service.WithCredsFuncs(credsFuncs...))
+
+	opts := []service.Option{service.WithCredsFuncs(credsFuncs...)}
+	if *fuseManagerAddr != "" {
+		err := service.StartFuseManager(ctx, *fuseManagerPath, *fuseManagerAddr, filepath.Join(*rootDir, "fusestore.db"), *logLevel, filepath.Join(*rootDir, "stargz-fuse-manager.log"))
+		if err != nil {
+			log.G(ctx).WithError(err).Fatalf("failed to start fuse manager")
+		}
+		fs, err := fusemanager.NewManagerClient(ctx, *rootDir, *fuseManagerAddr, &config.Config)
+		if err != nil {
+			log.G(ctx).WithError(err).Fatalf("failed to configure fuse manager")
+		}
+		opts = append(opts, service.WithFuseManagerCli(fs))
+	}
+
+	rs, err := service.NewStargzSnapshotterService(ctx, *rootDir, &config.Config, opts...)
 	if err != nil {
 		log.G(ctx).WithError(err).Fatalf("failed to configure snapshotter")
 	}

cmd/stargz-fuse-manager/main.go

@@ -0,0 +1,25 @@
+/*
+   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 main
+
+import (
+	fusemanager "github.com/containerd/stargz-snapshotter/stargzfusemanager"
+)
+
+func main() {
+	fusemanager.Run()
+}

fs/fs.go

@@ -64,6 +64,13 @@ import (
 
 const defaultMaxConcurrency = 2
 
+var (
+	nsLock = sync.Mutex{}
+
+	ns         *metrics.Namespace
+	metricsCtr *fsmetrics.Controller
+)
+
 type Option func(*options)
 
 type options struct {
@@ -97,13 +104,19 @@ func NewFilesystem(root string, cfg config.Config, opts ...Option) (_ snapshot.F
 		return nil, errors.Wrapf(err, "failed to setup resolver")
 	}
 
-	var ns *metrics.Namespace
-	if !cfg.NoPrometheus {
-		ns = metrics.NewNamespace("stargz", "fs", nil)
-	}
-	c := fsmetrics.NewLayerMetrics(ns)
-	if ns != nil {
-		metrics.Register(ns)
+	nsLock.Lock()
+	defer nsLock.Unlock()
+
+	var c *fsmetrics.Controller
+	if cfg.NoPrometheus {
+		c = fsmetrics.NewLayerMetrics(nil)
+	} else {
+		if ns == nil {
+			ns = metrics.NewNamespace("stargz", "fs", nil)
+			metrics.Register(ns)
+			metricsCtr = fsmetrics.NewLayerMetrics(ns)
+		}
+		c = metricsCtr
 	}
 
 	return &filesystem{

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/docker/docker-credential-helpers v0.6.3 // indirect
 	github.com/docker/go-metrics v0.0.1
 	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e
+	github.com/golang/protobuf v1.4.3
 	github.com/hanwen/go-fuse/v2 v2.1.0
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/golang-lru v0.5.3 // indirect
@@ -28,6 +29,7 @@ require (
 	github.com/sirupsen/logrus v1.8.1
 	github.com/stretchr/testify v1.7.0 // indirect
 	github.com/urfave/cli v1.22.2
+	go.etcd.io/bbolt v1.3.5
 	golang.org/x/sync v0.0.0-20201207232520-09787c993a3a
 	golang.org/x/sys v0.0.0-20210324051608-47abb6519492
 	google.golang.org/grpc v1.35.0

service/service.go

@@ -18,6 +18,7 @@ package service
 
 import (
 	"context"
+	"os"
 	"os/exec"
 	"path/filepath"
 
@@ -27,17 +28,22 @@ import (
 	stargzfs "github.com/containerd/stargz-snapshotter/fs"
 	"github.com/containerd/stargz-snapshotter/fs/source"
 	"github.com/containerd/stargz-snapshotter/service/resolver"
+	"github.com/containerd/stargz-snapshotter/snapshot"
 	snbase "github.com/containerd/stargz-snapshotter/snapshot"
 	"github.com/hashicorp/go-multierror"
 	"github.com/pkg/errors"
 )
 
-const fusermountBin = "fusermount"
+const (
+	fusermountBin          = "fusermount"
+	defaultFuseManagerPath = "/usr/bin/stargz-fuse-manager"
+)
 
 type Option func(*options)
 
 type options struct {
-	credsFuncs []resolver.Credential
+	credsFuncs     []resolver.Credential
+	fuseManagerCli snbase.FileSystem
 }
 
 func WithCredsFuncs(creds ...resolver.Credential) Option {
@@ -46,8 +52,30 @@ func WithCredsFuncs(creds ...resolver.Credential) Option {
 	}
 }
 
+func WithFuseManagerCli(cli snbase.FileSystem) Option {
+	return func(o *options) {
+		o.fuseManagerCli = cli
+	}
+}
+
 // NewStargzSnapshotterService returns stargz snapshotter.
 func NewStargzSnapshotterService(ctx context.Context, root string, config *Config, opts ...Option) (snapshots.Snapshotter, error) {
+	fs, enableFuseManager, err := NewFileSystem(ctx, root, config, opts...)
+	if err != nil {
+		log.G(ctx).WithError(err).Fatalf("failed to configure filesystem")
+	}
+
+	snOpts := make([]snbase.Opt, 0)
+	snOpts = append(snOpts, snbase.AsynchronousRemove)
+	if !enableFuseManager {
+		snOpts = append(snOpts, snbase.CleanupCommitted)
+		snOpts = append(snOpts, snbase.EnableRestoreSnapshots)
+	}
+
+	return snbase.NewSnapshotter(ctx, snapshotterRoot(root), fs, snOpts...)
+}
+
+func NewFileSystem(ctx context.Context, root string, config *Config, opts ...Option) (snapshot.FileSystem, bool, error) {
 	var sOpts options
 	for _, o := range opts {
 		o(&sOpts)
@@ -56,6 +84,10 @@ func NewStargzSnapshotterService(ctx context.Context, root string, config *Confi
 	// Use RegistryHosts based on ResolverConfig and keychain
 	hosts := resolver.RegistryHostsFromConfig(resolver.Config(config.ResolverConfig), sOpts.credsFuncs...)
 
+	if sOpts.fuseManagerCli != nil {
+		return sOpts.fuseManagerCli, true, nil
+	}
+
 	// Configure filesystem and snapshotter
 	fs, err := stargzfs.NewFilesystem(fsRoot(root),
 		config.Config,
@@ -65,10 +97,47 @@ func NewStargzSnapshotterService(ctx context.Context, root string, config *Confi
 		)),
 	)
 	if err != nil {
-		log.G(ctx).WithError(err).Fatalf("failed to configure filesystem")
+		return nil, false, err
+	}
+
+	return fs, false, nil
+}
+
+func StartFuseManager(ctx context.Context, executable, address, fusestore, logLevel, logPath string) error {
+	// if socket exists, do not start it
+	if _, err := os.Stat(address); err == nil {
+		return nil
+	} else if !os.IsNotExist(err) {
+		return err
+	}
+
+	if executable == "" {
+		executable = defaultFuseManagerPath
+	}
+
+	if _, err := os.Stat(executable); err != nil {
+		log.G(ctx).WithError(err).Errorf("failed to stat fusemanager binary: %s", executable)
+		return err
+	}
+
+	args := []string{
+		"-action", "start",
+		"-address", address,
+		"-fusestore-path", fusestore,
+		"-log-level", logLevel,
+		"-log-path", logPath,
+	}
+
+	cmd := exec.Command(executable, args...)
+	if err := cmd.Start(); err != nil {
+		return err
+	}
+
+	if err := cmd.Wait(); err != nil {
+		return err
 	}
 
-	return snbase.NewSnapshotter(ctx, snapshotterRoot(root), fs, snbase.AsynchronousRemove)
+	return nil
 }
 
 func snapshotterRoot(root string) string {

snapshot/snapshot.go

@@ -73,7 +73,9 @@ type FileSystem interface {
 
 // SnapshotterConfig is used to configure the remote snapshotter instance
 type SnapshotterConfig struct {
-	asyncRemove bool
+	asyncRemove            bool
+	cleanupCommitted       bool
+	enableRestoreSnapshots bool
 }
 
 // Opt is an option to configure the remote snapshotter
@@ -88,10 +90,24 @@ func AsynchronousRemove(config *SnapshotterConfig) error {
 	return nil
 }
 
+// CleanupCommitted cleans committed snapshots when closing snapshotter
+func CleanupCommitted(config *SnapshotterConfig) error {
+	config.cleanupCommitted = true
+	return nil
+}
+
+// EnableRestoreSnapshots indicates whether restore snapshots when
+// starting snapshotter
+func EnableRestoreSnapshots(config *SnapshotterConfig) error {
+	config.enableRestoreSnapshots = true
+	return nil
+}
+
 type snapshotter struct {
-	root        string
-	ms          *storage.MetaStore
-	asyncRemove bool
+	root             string
+	ms               *storage.MetaStore
+	asyncRemove      bool
+	cleanupCommitted bool
 
 	// fs is a filesystem that this snapshotter recognizes.
 	fs        FileSystem
@@ -146,8 +162,10 @@ func NewSnapshotter(ctx context.Context, root string, targetFs FileSystem, opts
 		userxattr:   userxattr,
 	}
 
-	if err := o.restoreRemoteSnapshot(ctx); err != nil {
-		return nil, errors.Wrap(err, "failed to restore remote snapshot")
+	if config.enableRestoreSnapshots {
+		if err := o.restoreRemoteSnapshot(ctx); err != nil {
+			return nil, errors.Wrap(err, "failed to restore remote snapshot")
+		}
 	}
 
 	return o, nil
@@ -632,9 +650,8 @@ func (o *snapshotter) workPath(id string) string {
 // Close closes the snapshotter
 func (o *snapshotter) Close() error {
 	// unmount all mounts including Committed
-	const cleanupCommitted = true
 	ctx := context.Background()
-	if err := o.cleanup(ctx, cleanupCommitted); err != nil {
+	if err := o.cleanup(ctx, o.cleanupCommitted); err != nil {
 		log.G(ctx).WithError(err).Warn("failed to cleanup")
 	}
 	return o.ms.Close()

snapshot/snapshot_test.go

@@ -60,7 +60,7 @@ func TestRemotePrepare(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer os.RemoveAll(root)
-	sn, err := NewSnapshotter(context.TODO(), root, bindFileSystem(t))
+	sn, err := NewSnapshotter(context.TODO(), root, bindFileSystem(t), CleanupCommitted, EnableRestoreSnapshots)
 	if err != nil {
 		t.Fatalf("failed to make new remote snapshotter: %q", err)
 	}
@@ -111,7 +111,7 @@ func TestRemoteOverlay(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer os.RemoveAll(root)
-	sn, err := NewSnapshotter(context.TODO(), root, bindFileSystem(t))
+	sn, err := NewSnapshotter(context.TODO(), root, bindFileSystem(t), CleanupCommitted, EnableRestoreSnapshots)
 	if err != nil {
 		t.Fatalf("failed to make new remote snapshotter: %q", err)
 	}
@@ -170,7 +170,7 @@ func TestRemoteCommit(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer os.RemoveAll(root)
-	sn, err := NewSnapshotter(context.TODO(), root, bindFileSystem(t))
+	sn, err := NewSnapshotter(context.TODO(), root, bindFileSystem(t), CleanupCommitted, EnableRestoreSnapshots)
 	if err != nil {
 		t.Fatalf("failed to make new remote snapshotter: %q", err)
 	}
@@ -313,7 +313,7 @@ func TestFailureDetection(t *testing.T) {
 			}
 			defer os.RemoveAll(root)
 			fi := bindFileSystem(t)
-			sn, err := NewSnapshotter(context.TODO(), root, fi)
+			sn, err := NewSnapshotter(context.TODO(), root, fi, CleanupCommitted, EnableRestoreSnapshots)
 			if err != nil {
 				t.Fatalf("failed to make new Snapshotter: %q", err)
 			}
@@ -439,7 +439,8 @@ func (fs *dummyFs) Unmount(ctx context.Context, mountpoint string) error {
 // Tests backword-comaptibility of overlayfs snapshotter.
 
 func newSnapshotter(ctx context.Context, root string) (snapshots.Snapshotter, func() error, error) {
-	snapshotter, err := NewSnapshotter(context.TODO(), root, dummyFileSystem())
+	snapshotter, err := NewSnapshotter(context.TODO(), root, dummyFileSystem(),
+		CleanupCommitted, EnableRestoreSnapshots)
 	if err != nil {
 		return nil, nil, err
 	}