Enable retry and recover for tarfs

.github/workflows/ci.yml

@@ -28,6 +28,14 @@ jobs:
           key: ${{ runner.os }}-go-${{ hashFiles('go.sum') }}
           restore-keys: |
             ${{ runner.os }}-go
+      - name: Setup Nydus
+        run: |
+          # Download nydus components
+          NYDUS_VER=v$(curl -s "https://api.github.com/repos/dragonflyoss/nydus/releases/latest" | jq -r .tag_name | sed 's/^v//')
+          wget https://github.com/dragonflyoss/nydus/releases/download/$NYDUS_VER/nydus-static-$NYDUS_VER-linux-amd64.tgz
+          tar xzvf nydus-static-$NYDUS_VER-linux-amd64.tgz
+          mkdir -p /usr/bin
+          sudo mv nydus-static/nydus-image /usr/bin/
       - name: Build
         run: |
           go install github.com/golangci/golangci-lint/cmd/[email protected]
@@ -135,6 +143,14 @@ jobs:
           key: ${{ runner.os }}-go-${{ hashFiles('go.sum') }}
           restore-keys: |
             ${{ runner.os }}-go
+      - name: Setup Nydus
+        run: |
+          # Download nydus components
+          NYDUS_VER=v$(curl --header 'authorization: Bearer ${{ secrets.GITHUB_TOKEN }}' -s "https://api.github.com/repos/dragonflyoss/nydus/releases/latest" | jq -r .tag_name | sed 's/^v//')
+          wget https://github.com/dragonflyoss/nydus/releases/download/$NYDUS_VER/nydus-static-$NYDUS_VER-linux-amd64.tgz
+          tar xzvf nydus-static-$NYDUS_VER-linux-amd64.tgz
+          mkdir -p /usr/bin
+          sudo mv nydus-static/nydus-image /usr/bin/
       - name: Run unit tests.
         run: make cover
       - name: Upload coverage to Codecov

pkg/daemon/config_test.go

@@ -0,0 +1,50 @@
+/*
+ * Copyright (c) 2023. Nydus Developers. All rights reserved.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package daemon
+
+import (
+	"testing"
+
+	"github.com/containerd/nydus-snapshotter/config"
+	"gotest.tools/assert"
+)
+
+func TestConfigOptions(t *testing.T) {
+	tmpDir := t.TempDir()
+	opts := []NewDaemonOpt{
+		WithSocketDir("/tmp/socket"),
+		WithRef(5),
+		WithLogDir(tmpDir),
+		WithLogToStdout(true),
+		WithLogLevel("Warning"),
+		WithLogRotationSize(1024),
+		WithConfigDir(tmpDir),
+		WithMountpoint("/tmp/mnt"),
+		WithNydusdThreadNum(4),
+		WithFsDriver("fscache"),
+		WithDaemonMode("dedicated"),
+	}
+
+	daemon, err := NewDaemon(opts...)
+	assert.Assert(t, err)
+	assert.Equal(t, daemon.States.APISocket, "/tmp/socket/"+daemon.ID()+"/api.sock")
+	assert.Equal(t, daemon.ref, int32(5))
+	assert.Equal(t, daemon.States.LogDir, tmpDir+"/"+daemon.ID())
+	assert.Equal(t, daemon.States.LogToStdout, true)
+	assert.Equal(t, daemon.States.LogLevel, "Warning")
+	assert.Equal(t, daemon.States.LogRotationSize, 1024)
+	assert.Equal(t, daemon.States.ConfigDir, tmpDir+"/"+daemon.ID())
+	assert.Equal(t, daemon.States.Mountpoint, "/tmp/mnt")
+	assert.Equal(t, daemon.States.ThreadNum, 4)
+	assert.Equal(t, daemon.States.FsDriver, "fscache")
+	assert.Equal(t, string(daemon.States.DaemonMode), "dedicated")
+
+}
+
+func String(daemonMode config.DaemonMode) {
+	panic("unimplemented")
+}

pkg/daemon/idgen_test.go

@@ -0,0 +1,22 @@
+/*
+ * Copyright (c) 2023. Nydus Developers. All rights reserved.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package daemon
+
+import (
+	"testing"
+
+	"gotest.tools/assert"
+)
+
+func TestIdGenerate(t *testing.T) {
+	id1 := newID()
+	id2 := newID()
+
+	assert.Assert(t, len(id1) > 0)
+	assert.Assert(t, len(id2) > 0)
+	assert.Assert(t, id1 != id2)
+}

pkg/filesystem/fs.go

@@ -67,7 +67,7 @@ func NewFileSystem(ctx context.Context, opt ...NewFSOpt) (*Filesystem, error) {
 	recoveringDaemons := make(map[string]*daemon.Daemon, 0)
 	liveDaemons := make(map[string]*daemon.Daemon, 0)
 	for _, fsManager := range fs.enabledManagers {
-		err := fsManager.Recover(ctx, &recoveringDaemons, &liveDaemons)
+		err := fsManager.Recover(ctx, fs.tarfsMgr, &recoveringDaemons, &liveDaemons)
 		if err != nil {
 			return nil, errors.Wrap(err, "reconnect daemons and recover filesystem instance")
 		}
@@ -351,7 +351,7 @@ func (fs *Filesystem) Mount(ctx context.Context, snapshotID string, labels map[s
 			err = errors.Wrapf(err, "mount file system by daemon %s, snapshot %s", d.ID(), snapshotID)
 		}
 	case config.FsDriverBlockdev:
-		err = fs.tarfsMgr.MountTarErofs(snapshotID, s, labels, rafs)
+		err = fs.tarfsMgr.MountErofs(snapshotID, s, labels, rafs)
 		if err != nil {
 			err = errors.Wrapf(err, "mount tarfs for snapshot %s", snapshotID)
 		}

pkg/filesystem/tarfs_adaptor.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/containerd/containerd/log"
 	snpkg "github.com/containerd/containerd/pkg/snapshotters"
+	"github.com/containerd/containerd/snapshots"
 	"github.com/containerd/containerd/snapshots/storage"
 	"github.com/containerd/nydus-snapshotter/pkg/label"
 	"github.com/opencontainers/go-digest"
@@ -63,8 +64,9 @@ func (fs *Filesystem) PrepareTarfsLayer(ctx context.Context, labels map[string]s
 	return nil
 }
 
-func (fs *Filesystem) MergeTarfsLayers(s storage.Snapshot, storageLocater func(string) string) error {
-	return fs.tarfsMgr.MergeLayers(s, storageLocater)
+func (fs *Filesystem) MergeTarfsLayers(ctx context.Context, s storage.Snapshot, storageLocater func(string) string,
+	infoGetter func(ctx context.Context, id string) (string, snapshots.Info, error)) error {
+	return fs.tarfsMgr.MergeLayers(ctx, s, storageLocater, infoGetter)
 }
 
 func (fs *Filesystem) DetachTarfsLayer(snapshotID string) error {

pkg/label/label.go

@@ -36,6 +36,8 @@ const (
 	NydusRefLayer = "containerd.io/snapshot/nydus-ref"
 	// The blobID of associated layer, also marking the layer as a nydus tarfs, set by the snapshotter
 	NydusTarfsLayer = "containerd.io/snapshot/nydus-tarfs"
+	// List of parent snapshot IDs, saved in `Rafs.Annotation`.
+	NydusTarfsParents = "containerd.io/snapshot/nydus-tarfs-parent-snapshot-list"
 	// Dm-verity information for image block device
 	NydusImageBlockInfo = "containerd.io/snapshot/nydus-image-block"
 	// Dm-verity information for layer block device

pkg/manager/manager.go

@@ -27,6 +27,7 @@ import (
 	"github.com/containerd/nydus-snapshotter/pkg/rafs"
 	"github.com/containerd/nydus-snapshotter/pkg/store"
 	"github.com/containerd/nydus-snapshotter/pkg/supervisor"
+	"github.com/containerd/nydus-snapshotter/pkg/tarfs"
 )
 
 // Manage RAFS filesystem instances and nydusd daemons.
@@ -121,12 +122,12 @@ func (m *Manager) CacheDir() string {
 // - Never ever delete any records from DB
 // - Only cache daemon information from DB, do not actually start/create daemons
 // - Only cache RAFS instance information from DB, do not actually recover RAFS runtime state.
-func (m *Manager) Recover(ctx context.Context,
+func (m *Manager) Recover(ctx context.Context, tarfsMgr *tarfs.Manager,
 	recoveringDaemons *map[string]*daemon.Daemon, liveDaemons *map[string]*daemon.Daemon) error {
 	if err := m.recoverDaemons(ctx, recoveringDaemons, liveDaemons); err != nil {
 		return errors.Wrapf(err, "recover nydusd daemons")
 	}
-	if err := m.recoverRafsInstances(ctx, recoveringDaemons, liveDaemons); err != nil {
+	if err := m.recoverRafsInstances(ctx, tarfsMgr, recoveringDaemons, liveDaemons); err != nil {
 		return errors.Wrapf(err, "recover RAFS instances")
 	}
 	return nil
@@ -150,7 +151,7 @@ func (m *Manager) RemoveRafsInstance(snapshotID string) error {
 	return m.store.DeleteRafsInstance(snapshotID)
 }
 
-func (m *Manager) recoverRafsInstances(ctx context.Context,
+func (m *Manager) recoverRafsInstances(ctx context.Context, tarfsMgr *tarfs.Manager,
 	recoveringDaemons *map[string]*daemon.Daemon, liveDaemons *map[string]*daemon.Daemon) error {
 	if err := m.store.WalkRafsInstances(ctx, func(r *rafs.Rafs) error {
 		if r.GetFsDriver() != m.FsDriver {
@@ -169,7 +170,12 @@ func (m *Manager) recoverRafsInstances(ctx context.Context,
 			}
 			rafs.RafsGlobalCache.Add(r)
 		} else if r.GetFsDriver() == config.FsDriverBlockdev {
-			rafs.RafsGlobalCache.Add(r)
+			err1 := tarfsMgr.RecoverRafsInstance(r)
+			if err1 != nil {
+				log.L.Errorf("failed to recover tarfs instance %s, %s", r.SnapshotID, err1)
+			} else {
+				rafs.RafsGlobalCache.Add(r)
+			}
 		}
 
 		return nil

pkg/rafs/rafs_test.go

@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2022. Nydus Developers. All rights reserved.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package rafs
+
+import (
+	"testing"
+
+	"github.com/containerd/nydus-snapshotter/config"
+	"github.com/containerd/nydus-snapshotter/internal/constant"
+	"gotest.tools/assert"
+)
+
+func TestRafsSetEmpty(t *testing.T) {
+	cache := NewRafsCache()
+
+	assert.Assert(t, cache.Get("rafs1") == nil)
+	assert.Equal(t, cache.Len(), 0)
+	assert.Assert(t, cache.Head() == nil)
+	instances := cache.List()
+	assert.Equal(t, len(instances), 0)
+}
+
+func TestRafs(t *testing.T) {
+	tmpDir := t.TempDir()
+	snapshotterConfig := config.SnapshotterConfig{}
+	snapshotterConfig.Root = tmpDir
+	snapshotterConfig.DaemonMode = constant.DaemonModeDedicated
+	assert.Assert(t, config.ProcessConfigurations(&snapshotterConfig))
+
+	rafs, err := NewRafs("snapshot1", "image1", "fscache")
+	assert.Assert(t, err)
+	assert.Equal(t, rafs, RafsGlobalCache.Get("snapshot1"))
+	assert.Equal(t, RafsGlobalCache.Len(), 1)
+	assert.Equal(t, rafs, RafsGlobalCache.Head())
+	instances := RafsGlobalCache.List()
+	assert.Equal(t, len(instances), 1)
+	assert.Equal(t, instances["snapshot1"].SnapshotID, "snapshot1")
+
+	RafsGlobalCache.Lock()
+	instances2 := RafsGlobalCache.ListLocked()
+	RafsGlobalCache.Unlock()
+	assert.Equal(t, len(instances2), 1)
+
+	RafsGlobalCache.SetIntances(instances)
+	assert.Equal(t, RafsGlobalCache.Len(), 1)
+	assert.Equal(t, RafsGlobalCache.Head().SnapshotID, "snapshot1")
+
+	assert.Equal(t, len(rafs.Annotations), 0)
+	rafs.AddAnnotation("key", "value")
+	assert.Equal(t, len(rafs.Annotations), 1)
+	assert.Equal(t, rafs.GetSnapshotDir(), tmpDir+"/snapshots/snapshot1")
+	assert.Equal(t, rafs.RelaMountpoint(), "/snapshot1")
+	assert.Equal(t, rafs.FscacheWorkDir(), tmpDir+"/snapshots/snapshot1/fs")
+	assert.Equal(t, rafs.GetFsDriver(), "fscache")
+	rafs.SetMountpoint("/tmp/mnt")
+	assert.Equal(t, rafs.GetMountpoint(), "/tmp/mnt")
+	_, err = rafs.BootstrapFile()
+	assert.Assert(t, err != nil)
+
+	RafsGlobalCache.Remove("snapshot1")
+	assert.Equal(t, RafsGlobalCache.Len(), 0)
+}