Merge pull request #1914 from ktock/fuse-manager-tests

CI: test fuse manager

Author: ktock

.github/workflows/tests.yml

@@ -56,6 +56,7 @@ jobs:
         builtin: ["true", "false"]
         metadata-store: ["memory", "db"]
         fuse-passthrough: ["true", "false"]
+        fuse-manager: ["true", "false"]
         transfer-service: ["true", "false"]
         exclude:
         - buildargs: ""
@@ -70,6 +71,10 @@ jobs:
           buildargs: "--build-arg=CONTAINERD_VERSION=main"
         - fuse-passthrough: "true"
           metadata-store: "db"
+        - fuse-manager: "true"
+          builtin: "true"
+        - fuse-manager: "true"
+          buildargs: "--build-arg=CONTAINERD_VERSION=main"
         - transfer-service: "true"
           buildargs: "--build-arg=CONTAINERD_VERSION=main"
         - transfer-service: "true"
@@ -88,6 +93,7 @@ jobs:
         BUILTIN_SNAPSHOTTER: ${{ matrix.builtin }}
         METADATA_STORE: ${{ matrix.metadata-store }}
         FUSE_PASSTHROUGH: ${{ matrix.fuse-passthrough }}
+        FUSE_MANAGER: ${{ matrix.fuse-manager }}
         TRANSFER_SERVICE: ${{ matrix.transfer-service }}
       run: make integration
 
@@ -159,6 +165,7 @@ jobs:
         builtin: ["true", "false"]
         metadata-store: ["memory", "db"]
         fuse-passthrough: ["true", "false"]
+        fuse-manager: ["true", "false"]
         transfer-service: ["true", "false"]
         exclude:
         - buildargs: ""
@@ -173,6 +180,10 @@ jobs:
           buildargs: "--build-arg=CONTAINERD_VERSION=main"
         - fuse-passthrough: "true"
           metadata-store: "db"
+        - fuse-manager: "true"
+          builtin: "true"
+        - fuse-manager: "true"
+          buildargs: "--build-arg=CONTAINERD_VERSION=main"
         - transfer-service: "true"
           buildargs: "--build-arg=CONTAINERD_VERSION=main"
         - transfer-service: "true"
@@ -189,6 +200,7 @@ jobs:
         BUILTIN_SNAPSHOTTER: ${{ matrix.builtin }}
         METADATA_STORE: ${{ matrix.metadata-store }}
         FUSE_PASSTHROUGH: ${{ matrix.fuse-passthrough }}
+        FUSE_MANAGER: ${{ matrix.fuse-manager }}
         TRANSFER_SERVICE: ${{ matrix.transfer-service }}
       run: make test-cri-containerd
 

Dockerfile

@@ -88,7 +88,8 @@ COPY . $GOPATH/src/github.com/containerd/stargz-snapshotter
 ARG CGO_ENABLED
 RUN cd $GOPATH/src/github.com/containerd/stargz-snapshotter && \
     PREFIX=/out/ GOARCH=${TARGETARCH:-amd64} GO_BUILD_FLAGS=${SNAPSHOTTER_BUILD_FLAGS} make containerd-stargz-grpc && \
-    PREFIX=/out/ GOARCH=${TARGETARCH:-amd64} GO_BUILD_FLAGS=${CTR_REMOTE_BUILD_FLAGS} make ctr-remote
+    PREFIX=/out/ GOARCH=${TARGETARCH:-amd64} GO_BUILD_FLAGS=${CTR_REMOTE_BUILD_FLAGS} make ctr-remote && \
+    PREFIX=/out/ GOARCH=${TARGETARCH:-amd64} GO_BUILD_FLAGS=${CTR_REMOTE_BUILD_FLAGS} make stargz-fuse-manager
 
 # Build stargz store
 FROM golang-base AS stargz-store-dev

cmd/containerd-stargz-grpc/fsopts/fsopts.go

@@ -34,6 +34,7 @@ import (
 type Config struct {
 	EnableIpfs    bool
 	MetadataStore string
+	OpenBoltDB    func(string) (*bolt.DB, error)
 }
 
 const (
@@ -62,12 +63,10 @@ func getMetadataStore(rootDir string, config *Config) (metadata.Store, error) {
 	case "", memoryMetadataType:
 		return memorymetadata.NewReader, nil
 	case dbMetadataType:
-		bOpts := bolt.Options{
-			NoFreelistSync:  true,
-			InitialMmapSize: 64 * 1024 * 1024,
-			FreelistType:    bolt.FreelistMapType,
+		if config.OpenBoltDB == nil {
+			return nil, fmt.Errorf("bolt DB is not configured")
 		}
-		db, err := bolt.Open(filepath.Join(rootDir, "metadata.db"), 0600, &bOpts)
+		db, err := config.OpenBoltDB(filepath.Join(rootDir, "metadata.db"))
 		if err != nil {
 			return nil, err
 		}

cmd/containerd-stargz-grpc/main.go

@@ -44,6 +44,7 @@ import (
 	sddaemon "github.com/coreos/go-systemd/v22/daemon"
 	metrics "github.com/docker/go-metrics"
 	"github.com/pelletier/go-toml"
+	bolt "go.etcd.io/bbolt"
 	"golang.org/x/sys/unix"
 	"google.golang.org/grpc"
 )
@@ -55,42 +56,48 @@ const (
 	defaultRootDir             = "/var/lib/containerd-stargz-grpc"
 	defaultImageServiceAddress = "/run/containerd/containerd.sock"
 	defaultFuseManagerAddress  = "/run/containerd-stargz-grpc/fuse-manager.sock"
-
-	fuseManagerBin     = "stargz-fuse-manager"
-	fuseManagerAddress = "fuse-manager.sock"
+	fuseManagerBin             = "stargz-fuse-manager"
 )
 
 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")
-	detachFuseManager = flag.Bool("detach-fuse-manager", false, "whether detach fusemanager or not")
-	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")
+	printVersion = flag.Bool("version", false, "print the version")
 )
 
 type snapshotterConfig struct {
 	service.Config
 
 	// MetricsAddress is address for the metrics API
-	MetricsAddress string `toml:"metrics_address"`
+	MetricsAddress string `toml:"metrics_address" json:"metrics_address"`
 
 	// NoPrometheus is a flag to disable the emission of the metrics
-	NoPrometheus bool `toml:"no_prometheus"`
+	NoPrometheus bool `toml:"no_prometheus" json:"no_prometheus"`
 
 	// DebugAddress is a Unix domain socket address where the snapshotter exposes /debug/ endpoints.
-	DebugAddress string `toml:"debug_address"`
+	DebugAddress string `toml:"debug_address" json:"debug_address"`
 
 	// IPFS is a flag to enbale lazy pulling from IPFS.
-	IPFS bool `toml:"ipfs"`
+	IPFS bool `toml:"ipfs" json:"ipfs"`
 
 	// MetadataStore is the type of the metadata store to use.
-	MetadataStore string `toml:"metadata_store" default:"memory"`
-	// FuseManagerAddress is address for the fusemanager's GRPC server
-	FuseManagerAddress string `toml:"fusemanager_address"`
+	MetadataStore string `toml:"metadata_store" default:"memory" json:"metadata_store"`
 
-	// FuseManagerPath is path to the fusemanager's executable
-	FuseManagerPath string `toml:"fusemanager_path"`
+	// FuseManagerConfig is configuration for fusemanager
+	FuseManagerConfig `toml:"fuse_manager" json:"fuse_manager"`
+}
+
+type FuseManagerConfig struct {
+	// Enable is whether detach fusemanager or not
+	Enable bool `toml:"enable" default:"false" json:"enable"`
+
+	// Address is address for the fusemanager's GRPC server (default: "/run/containerd-stargz-grpc/fuse-manager.sock")
+	Address string `toml:"address" json:"address"`
+
+	// Path is path to the fusemanager's executable (default: looking for a binary "stargz-fuse-manager")
+	Path string `toml:"path" json:"path"`
 }
 
 func main() {
@@ -148,16 +155,17 @@ func main() {
 	}
 
 	var rs snapshots.Snapshotter
-	if *detachFuseManager {
-		fmPath := config.FuseManagerPath
+	fuseManagerConfig := config.FuseManagerConfig
+	if fuseManagerConfig.Enable {
+		fmPath := fuseManagerConfig.Path
 		if fmPath == "" {
 			var err error
 			fmPath, err = exec.LookPath(fuseManagerBin)
 			if err != nil {
 				log.G(ctx).WithError(err).Fatalf("failed to find fusemanager bin")
 			}
 		}
-		fmAddr := config.FuseManagerAddress
+		fmAddr := fuseManagerConfig.Address
 		if fmAddr == "" {
 			fmAddr = defaultFuseManagerAddress
 		}
@@ -171,7 +179,7 @@ func main() {
 		}
 
 		fuseManagerConfig := fusemanager.Config{
-			Config:                     &config.Config,
+			Config:                     config.Config,
 			IPFS:                       config.IPFS,
 			MetadataStore:              config.MetadataStore,
 			DefaultImageServiceAddress: defaultImageServiceAddress,
@@ -195,14 +203,50 @@ func main() {
 		}
 		log.G(ctx).Infof("Start snapshotter with fusemanager mode")
 	} else {
-		credsFuncs, err := keychainconfig.ConfigKeychain(ctx, rpc, &keyChainConfig)
+		crirpc := rpc
+		// For CRI keychain, if listening path is different from stargz-snapshotter's socket, prepare for the dedicated grpc server and the socket.
+		serveCRISocket := config.CRIKeychainConfig.EnableKeychain && config.ListenPath != "" && config.ListenPath != *address
+		if serveCRISocket {
+			crirpc = grpc.NewServer()
+		}
+		credsFuncs, err := keychainconfig.ConfigKeychain(ctx, crirpc, &keyChainConfig)
 		if err != nil {
 			log.G(ctx).WithError(err).Fatalf("failed to configure keychain")
 		}
+		if serveCRISocket {
+			addr := config.ListenPath
+			// Prepare the directory for the socket
+			if err := os.MkdirAll(filepath.Dir(addr), 0700); err != nil {
+				log.G(ctx).WithError(err).Fatalf("failed to create directory %q", filepath.Dir(addr))
+			}
+
+			// Try to remove the socket file to avoid EADDRINUSE
+			if err := os.RemoveAll(addr); err != nil {
+				log.G(ctx).WithError(err).Fatalf("failed to remove %q", addr)
+			}
+
+			// Listen and serve
+			l, err := net.Listen("unix", addr)
+			if err != nil {
+				log.G(ctx).WithError(err).Fatalf("error on listen socket %q", addr)
+			}
+			go func() {
+				if err := crirpc.Serve(l); err != nil {
+					log.G(ctx).WithError(err).Errorf("error on serving CRI via socket %q", addr)
+				}
+			}()
+		}
 
 		fsConfig := fsopts.Config{
 			EnableIpfs:    config.IPFS,
 			MetadataStore: config.MetadataStore,
+			OpenBoltDB: func(p string) (*bolt.DB, error) {
+				return bolt.Open(p, 0600, &bolt.Options{
+					NoFreelistSync:  true,
+					InitialMmapSize: 64 * 1024 * 1024,
+					FreelistType:    bolt.FreelistMapType,
+				})
+			},
 		}
 		fsOpts, err := fsopts.ConfigFsOpts(ctx, *rootDir, &fsConfig)
 		if err != nil {
@@ -232,7 +276,7 @@ func main() {
 	// commanded via SIGINT. The user can use SIGINT to gracefully killing the FUSE
 	// manager before rebooting the node for ensuring that the all snapshots are
 	// unmounted with cleaning up associated temporary resources.
-	if cleanup || !*detachFuseManager {
+	if cleanup || !fuseManagerConfig.Enable {
 		log.G(ctx).Debug("Closing the snapshotter")
 		rs.Close()
 	}

cmd/stargz-fuse-manager/main.go

@@ -17,17 +17,26 @@
 package main
 
 import (
+	"fmt"
+	"net"
+	"os"
+	"path/filepath"
+
+	"github.com/containerd/log"
+
 	"github.com/containerd/stargz-snapshotter/cmd/containerd-stargz-grpc/fsopts"
 	fusemanager "github.com/containerd/stargz-snapshotter/fusemanager"
 	"github.com/containerd/stargz-snapshotter/service"
 	"github.com/containerd/stargz-snapshotter/service/keychain/keychainconfig"
+	"google.golang.org/grpc"
 )
 
 func init() {
 	fusemanager.RegisterConfigFunc(func(cc *fusemanager.ConfigContext) ([]service.Option, error) {
 		fsConfig := fsopts.Config{
 			EnableIpfs:    cc.Config.IPFS,
 			MetadataStore: cc.Config.MetadataStore,
+			OpenBoltDB:    cc.OpenBoltDB,
 		}
 		fsOpts, err := fsopts.ConfigFsOpts(cc.Ctx, cc.RootDir, &fsConfig)
 		if err != nil {
@@ -44,10 +53,41 @@ func init() {
 			DefaultImageServiceAddress: cc.Config.DefaultImageServiceAddress,
 			ImageServicePath:           cc.Config.Config.ImageServicePath,
 		}
-		credsFuncs, err := keychainconfig.ConfigKeychain(cc.Ctx, cc.Server, &keyChainConfig)
+		if cc.Config.Config.CRIKeychainConfig.EnableKeychain && cc.Config.Config.ListenPath == "" || cc.Config.Config.ListenPath == cc.Address {
+			return nil, fmt.Errorf("listen path of CRI server must be specified as a separated socket from FUSE manager server")
+		}
+		// For CRI keychain, if listening path is different from stargz-snapshotter's socket, prepare for the dedicated grpc server and the socket.
+		serveCRISocket := cc.Config.Config.CRIKeychainConfig.EnableKeychain && cc.Config.Config.ListenPath != "" && cc.Config.Config.ListenPath != cc.Address
+		if serveCRISocket {
+			cc.CRIServer = grpc.NewServer()
+		}
+		credsFuncs, err := keychainconfig.ConfigKeychain(cc.Ctx, cc.CRIServer, &keyChainConfig)
 		if err != nil {
 			return nil, err
 		}
+		if serveCRISocket {
+			addr := cc.Config.Config.ListenPath
+			// Prepare the directory for the socket
+			if err := os.MkdirAll(filepath.Dir(addr), 0700); err != nil {
+				return nil, fmt.Errorf("failed to create directory %q: %w", filepath.Dir(addr), err)
+			}
+
+			// Try to remove the socket file to avoid EADDRINUSE
+			if err := os.RemoveAll(addr); err != nil {
+				return nil, fmt.Errorf("failed to remove %q: %w", addr, err)
+			}
+
+			// Listen and serve
+			l, err := net.Listen("unix", addr)
+			if err != nil {
+				return nil, fmt.Errorf("error on listen socket %q: %w", addr, err)
+			}
+			go func() {
+				if err := cc.CRIServer.Serve(l); err != nil {
+					log.G(cc.Ctx).WithError(err).Errorf("error on serving CRI via socket %q", addr)
+				}
+			}()
+		}
 		return []service.Option{service.WithCredsFuncs(credsFuncs...)}, nil
 	})
 }

docs/overview.md

@@ -108,7 +108,12 @@ Remote snapshots are mounted using FUSE, and its filesystem processes are attach
 
 To avoid this, we use a fuse daemon called the fuse manager to handle filesystem processes. The fuse manager is responsible for mounting and unmounting remote snapshotters. Its process is detached from the stargz snapshotter main process to an independent one in a shim-like way during the snapshotter's startup. This design ensures that the restart of the snapshotter won't affect the filesystem processes it manages, keeping mountpoints and running containers available during the restart. However, it is important to note that the restart of the fuse manager itself triggers a remount, so it is recommended to keep the fuse manager running in a good state.
 
-You can enable the fuse manager by adding the flag `--detach-fuse-manager=true` to the stargz snapshotter.
+You can enable the fuse manager by adding the following configuration.
+
+```toml
+[fusem_anager]
+enable = true
+```
 
 ## Killing and restarting Stargz Snapshotter
 
@@ -173,13 +178,23 @@ The snapshotter acquires registry creds by scanning requests.
 
 You must specify `--image-service-endpoint=unix:///run/containerd-stargz-grpc/containerd-stargz-grpc.sock` option to kubelet.
 
+You can specify the backing image service's socket using `image_service_path`.
+The default is the containerd's socket (`/run/containerd/containerd.sock`).
+
 ```toml
 # Stargz Snapshotter proxies CRI Image Service into containerd socket.
 [cri_keychain]
 enable_keychain = true
 image_service_path = "/run/containerd/containerd.sock"
 ```
 
+The default path where containerd-stargz-grpc serves the CRI Image Service API is `unix:///run/containerd-stargz-grpc/containerd-stargz-grpc.sock`.
+You can also change this path using `listen_path` field.
+
+> Note that if you enabled the FUSE manager and CRI-based authentication together, `listen_path` is a mandatory field with some caveats:
+> - This path must be different from the FUSE manager's socket path (`/run/containerd-stargz-grpc/fuse-manager.sock`) because they have different lifecycle. Specifically, the CRI socket is recreted on each reload of the configuration to the FUSE manager.
+> - containerd-stargz-grpc's socket path (`/run/containerd-stargz-grpc/containerd-stargz-grpc.sock`) can't be used as `listen_path` because the CRI socket is served by the FUSE manager process (not containerd-stargz-grpc process).
+
 #### kubeconfig-based authentication
 
 This is another way to enable lazy pulling of private images on Kubernetes.