Merge branch 'main' into fricounet/token-renewals

Signed-off-by: Baptiste Girard-Carrabin <baptiste.girardcarrabin@datadoghq.com>

Author: Fricounet

config/daemonconfig/daemonconfig.go

@@ -112,6 +112,7 @@ type BackendConfig struct {
 }
 
 type DeviceConfig struct {
+	ID      string `json:"id,omitempty"`
 	Backend struct {
 		BackendType string        `json:"type"`
 		Config      BackendConfig `json:"config"`

config/daemonconfig/fuse.go

@@ -60,13 +60,17 @@ func LoadFuseConfig(p string) (*FuseDaemonConfig, error) {
 	return &cfg, nil
 }
 
-func (c *FuseDaemonConfig) Supplement(host, repo, _ string, params map[string]string) {
+func (c *FuseDaemonConfig) Supplement(host, repo, snapshotID string, params map[string]string) {
 	if host != "" {
 		c.Device.Backend.Config.Host = host
 	}
 	if repo != "" {
 		c.Device.Backend.Config.Repo = repo
 	}
+	// Temporary fix while https://github.com/containerd/nydus-snapshotter/issues/712 is being addressed
+	if snapshotID != "" {
+		c.Device.ID = "/" + snapshotID
+	}
 	c.Device.Cache.Config.WorkDir = params[CacheDir]
 }
 

pkg/cache/manager.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"os"
 	"path"
+	"strings"
 	"time"
 
 	"github.com/pkg/errors"
@@ -120,3 +121,32 @@ func (m *Manager) RemoveBlobCache(blobID string) error {
 	}
 	return nil
 }
+
+// extractBlobIDFromFilename extracts the blob ID from a cache filename
+// Cache files can have formats like:
+// - <blobID>
+// - <blobID>.blob.data
+// - <blobID>.chunk_map
+// - <blobID>.blob.meta
+// - <blobID>.image.disk
+// - <blobID>.layer.disk
+func ExtractBlobIDFromFilename(filename string) string {
+	// Remove known suffixes
+	suffixes := []string{
+		dataFileSuffix,
+		dataFileSuffix + chunkMapFileSuffix,
+		chunkMapFileSuffix,
+		metaFileSuffix,
+		imageDiskFileSuffix,
+		layerDiskFileSuffix,
+	}
+
+	for _, suffix := range suffixes {
+		if strings.HasSuffix(filename, suffix) {
+			return strings.TrimSuffix(filename, suffix)
+		}
+	}
+
+	// If no suffix matches, assume it's already a blob ID
+	return filename
+}

pkg/cache/manager_test.go

@@ -0,0 +1,88 @@
+/*
+ * Copyright (c) 2025. Nydus Developers. All rights reserved.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package cache
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestExtractBlobIDFromFilename(t *testing.T) {
+	tests := map[string]struct {
+		filename string
+		expected string
+	}{
+		"plain blob ID": {
+			filename: "abc123def456",
+			expected: "abc123def456",
+		},
+		"blob with .blob.data suffix": {
+			filename: "abc123def456.blob.data",
+			expected: "abc123def456",
+		},
+		"blob with .chunk_map suffix": {
+			filename: "abc123def456.chunk_map",
+			expected: "abc123def456",
+		},
+		"blob with .blob.meta suffix": {
+			filename: "abc123def456.blob.meta",
+			expected: "abc123def456",
+		},
+		"blob with .image.disk suffix": {
+			filename: "abc123def456.image.disk",
+			expected: "abc123def456",
+		},
+		"blob with .layer.disk suffix": {
+			filename: "abc123def456.layer.disk",
+			expected: "abc123def456",
+		},
+		"blob with combined .blob.data.chunk_map suffix": {
+			filename: "abc123def456.blob.data.chunk_map",
+			expected: "abc123def456",
+		},
+		"real sha256 hash": {
+			filename: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
+			expected: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
+		},
+		"real sha256 hash with .blob.data": {
+			filename: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855.blob.data",
+			expected: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
+		},
+		"real sha256 hash with .chunk_map": {
+			filename: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855.chunk_map",
+			expected: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
+		},
+		"real sha256 hash with .blob.data.chunk_map": {
+			filename: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855.blob.data.chunk_map",
+			expected: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
+		},
+		"empty filename": {
+			filename: "",
+			expected: "",
+		},
+		"filename with unknown suffix": {
+			filename: "abc123def456.unknown",
+			expected: "abc123def456.unknown",
+		},
+		"filename with multiple dots but no known suffix": {
+			filename: "abc.def.ghi",
+			expected: "abc.def.ghi",
+		},
+		".blob.data.chunk_map is matched before .blob.data or .chunk_map": {
+			filename: "test.blob.data.chunk_map",
+			expected: "test",
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			result := ExtractBlobIDFromFilename(tc.filename)
+			assert.Equal(t, tc.expected, result)
+		})
+	}
+}

pkg/daemon/daemon.go

@@ -151,6 +151,10 @@ func (d *Daemon) AddRafsInstance(r *rafs.Rafs) {
 	r.DaemonID = d.ID()
 }
 
+func (d *Daemon) UpdateRafsInstance(r *rafs.Rafs) {
+	d.RafsCache.Add(r)
+}
+
 func (d *Daemon) RemoveRafsInstance(snapshotID string) {
 	d.RafsCache.Remove(snapshotID)
 	d.DecRef()

pkg/filesystem/fs.go

@@ -42,6 +42,7 @@ import (
 	"github.com/containerd/nydus-snapshotter/pkg/signature"
 	"github.com/containerd/nydus-snapshotter/pkg/stargz"
 	"github.com/containerd/nydus-snapshotter/pkg/tarfs"
+	"github.com/containerd/nydus-snapshotter/pkg/utils/erofs"
 )
 
 type Filesystem struct {
@@ -268,6 +269,45 @@ func (fs *Filesystem) WaitUntilReady(snapshotID string) error {
 			return err
 		}
 
+		// For shared daemons, we need to use the correct cache ID to query metrics.
+		// For fscache, the cache is registered with fscacheID (a digest), not the raw snapshot ID.
+		// For fusedev, the cache is registered with the snapshot ID.
+		sid := ""
+		if d.IsSharedDaemon() {
+			if rafs.GetFsDriver() == config.FsDriverFscache {
+				// For fscache, use the fscache ID from annotations if available
+				if fscacheID, ok := rafs.Annotations[racache.AnnoFsCacheID]; ok && fscacheID != "" {
+					sid = fscacheID
+				} else {
+					// Fallback: compute fscacheID if not in annotations yet
+					sid = erofs.FscacheID(rafs.SnapshotID)
+				}
+			} else {
+				// For fusedev, use the snapshot ID directly
+				sid = rafs.SnapshotID
+			}
+		}
+
+		cacheMetrics, err := d.GetCacheMetrics(sid)
+		if err != nil {
+			return errors.Wrapf(err, "failed to get cache metric")
+		}
+		log.L.Debugf("Found %d underlying files for rafs instance %s", len(cacheMetrics.UnderlyingFiles), rafs.SnapshotID)
+
+		// Lock the daemon's RafsCache when modifying rafs fields to prevent
+		// race conditions with concurrent reads (e.g., during cache cleanup)
+		d.RafsCache.Lock()
+		rafs.UnderlyingFiles = cacheMetrics.UnderlyingFiles
+		d.RafsCache.Unlock()
+
+		fsManager, err := fs.getManager(rafs.GetFsDriver())
+		if err != nil {
+			return errors.Wrap(err, "failed to get manager")
+		}
+		err = fsManager.UpdateRafsInstance(rafs)
+		if err != nil {
+			return errors.Wrap(err, "failed to update rafs instance")
+		}
 		log.L.Debugf("Nydus remote snapshot %s is ready", snapshotID)
 	}
 
@@ -603,6 +643,24 @@ func (fs *Filesystem) RemoveCache(blobDigest string) error {
 	return fs.cacheMgr.RemoveBlobCache(blobID)
 }
 
+// WalkManagers iterates over all enabled managers and calls the provided function
+func (fs *Filesystem) WalkManagers(fn func(*manager.Manager) error) error {
+	for _, mgr := range fs.enabledManagers {
+		if err := fn(mgr); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// GetCacheConfig returns the cache directory from the cache manager
+func (fs *Filesystem) GetCacheDir() (string, error) {
+	if fs.cacheMgr == nil {
+		return "", errors.New("cache manager is not initialized")
+	}
+	return fs.cacheMgr.CacheDir(), nil
+}
+
 // Try to stop all the running daemons if they are not referenced by any snapshots
 // Clean up resources along with the daemons.
 func (fs *Filesystem) Teardown(ctx context.Context) error {

pkg/manager/manager.go

@@ -146,6 +146,10 @@ func (m *Manager) AddRafsInstance(r *rafs.Rafs) error {
 	return m.store.AddRafsInstance(r)
 }
 
+func (m *Manager) UpdateRafsInstance(r *rafs.Rafs) error {
+	return m.store.UpdateRafsInstance(r)
+}
+
 func (m *Manager) RemoveRafsInstance(snapshotID string) error {
 	return m.store.DeleteRafsInstance(snapshotID)
 }

pkg/manager/store.go

@@ -25,6 +25,7 @@ type Store interface {
 	CleanupDaemons(ctx context.Context) error
 
 	AddRafsInstance(r *rafs.Rafs) error
+	UpdateRafsInstance(r *rafs.Rafs) error
 	DeleteRafsInstance(snapshotID string) error
 	WalkRafsInstances(ctx context.Context, cb func(*rafs.Rafs) error) error
 

pkg/metrics/data/snapshotter.go

@@ -79,4 +79,25 @@ var (
 			Help: "Thread counts of snapshotter.",
 		},
 	)
+
+	CacheBlobsDeleted = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "snapshotter_cache_blobs_deleted_total",
+			Help: "Total number of cache blobs deleted during cleanup.",
+		},
+	)
+
+	CacheBlobsInUse = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Name: "snapshotter_cache_blobs_in_use",
+			Help: "Number of cache blobs currently in use by running daemons.",
+		},
+	)
+
+	CacheBlobDeletionErrors = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "snapshotter_cache_blob_deletion_errors_total",
+			Help: "Total number of errors encountered while deleting cache blobs.",
+		},
+	)
 )

pkg/metrics/registry/registry.go

@@ -44,7 +44,10 @@ func init() {
 		data.CachePrefetchCumulativeTimeMillis,
 		data.CachePrefetchTotalDurationMillis,
 		data.CacheBufferedBackendSize,
-		data.CredentialRenewals,
+		data.CacheBlobsDeleted,
+		data.CacheBlobsInUse,
+		data.CacheBlobDeletionErrors,
+    data.CredentialRenewals,
 		data.CredentialStoreEntries,
 	)