metrics: collect mount items for detail metrics

Introduce a new Prometheus metric for tracking detailed information
about mounted items and refines the cache manager to export richer metrics.
It also adds corresponding tests to ensure correct metric collection.

The changes enhance observability of mounted volumes, distinguishing between
static, dynamic, and inline models.

Signed-off-by: imeoer <[email protected]>

Author: imeoer

go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/opencontainers/image-spec v1.1.1
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.22.0
+	github.com/prometheus/client_model v0.6.2
 	github.com/rexray/gocsi v1.2.2
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.11.1
@@ -86,6 +87,7 @@ require (
 	github.com/kevinburke/ssh_config v1.2.0 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/labstack/gommon v0.4.2 // indirect
 	github.com/libgit2/git2go/v34 v34.0.0 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
@@ -99,7 +101,6 @@ require (
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pjbgf/sha1cd v0.3.2 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.63.0 // indirect
 	github.com/prometheus/procfs v0.16.1 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect

pkg/metrics/mount_collector.go

@@ -0,0 +1,61 @@
+package metrics
+
+import (
+	"sync/atomic"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type MountItem struct {
+    Reference  string
+    Type       string
+    VolumeName string
+    MountID    string
+}
+
+type MountItemCollector struct {
+    desc  *prometheus.Desc
+    items atomic.Value // stores []MountItem
+}
+
+func NewMountItemCollector() *MountItemCollector {
+    c := &MountItemCollector{
+        desc: prometheus.NewDesc(
+            Prefix+"mount_item",
+            "Mounted item list (static snapshot); value is always 1 for existing items.",
+            []string{"reference", "type", "volume_name", "mount_id"},
+            nil,
+        ),
+    }
+    c.items.Store([]MountItem(nil))
+    return c
+}
+
+func (c *MountItemCollector) Set(items []MountItem) {
+    c.items.Store(append([]MountItem(nil), items...))
+}
+
+func (c *MountItemCollector) Describe(ch chan<- *prometheus.Desc) {
+    ch <- c.desc
+}
+
+func (c *MountItemCollector) Collect(ch chan<- prometheus.Metric) {
+    v := c.items.Load()
+    if v == nil {
+        return
+    }
+    items := v.([]MountItem)
+    for _, it := range items {
+        ch <- prometheus.MustNewConstMetric(
+            c.desc,
+            prometheus.GaugeValue,
+            1,
+            it.Reference,
+            it.Type,
+            it.VolumeName,
+            it.MountID,
+        )
+    }
+}
+
+var MountItems = NewMountItemCollector()

pkg/metrics/registry.go

@@ -82,6 +82,12 @@ var (
 		},
 	)
 
+	NodeMountedInlineImages = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Name: Prefix + "node_mounted_inline_images",
+		},
+	)
+
 	NodeMountedDynamicImages = prometheus.NewGauge(
 		prometheus.GaugeOpts{
 			Name: Prefix + "node_mounted_dynamic_images",
@@ -140,7 +146,9 @@ func NodePullOpObserve(op string, size int64, start time.Time, err error) {
 func init() {
 	DummyRegistry.MustRegister()
 
-	DetailRegistry.MustRegister()
+	DetailRegistry.MustRegister(
+		MountItems,
+	)
 
 	Registry.MustRegister(
 		NodeNotReady,
@@ -156,6 +164,7 @@ func init() {
 
 		NodeCacheSizeInBytes,
 		NodeMountedStaticImages,
+		NodeMountedInlineImages,
 		NodeMountedDynamicImages,
 		NodePullLayerTooLong,
 	)

pkg/service/cache.go

@@ -8,13 +8,15 @@ import (
 	"github.com/modelpack/model-csi-driver/pkg/config"
 	"github.com/modelpack/model-csi-driver/pkg/logger"
 	"github.com/modelpack/model-csi-driver/pkg/metrics"
+	"github.com/modelpack/model-csi-driver/pkg/status"
 	"github.com/pkg/errors"
 )
 
-var CacheSacnInterval = 60 * time.Second
+const CacheScanInterval = 60 * time.Second
 
 type CacheManager struct {
 	cfg *config.Config
+	sm *status.StatusManager
 }
 
 func (cm *CacheManager) getCacheSize() (int64, error) {
@@ -36,6 +38,7 @@ func (cm *CacheManager) getCacheSize() (int64, error) {
 
 func (cm *CacheManager) scanModels() error {
 	staticModels := 0
+	inlineModels := 0
 	dynamicModels := 0
 	volumesDir := cm.cfg.Get().GetVolumesDir()
 	volumeDirs, err := os.ReadDir(volumesDir)
@@ -45,29 +48,71 @@ func (cm *CacheManager) scanModels() error {
 		}
 		return errors.Wrapf(err, "read volume dirs from %s", volumesDir)
 	}
+
+	mountItems := []metrics.MountItem{}
 	for _, volumeDir := range volumeDirs {
 		if !volumeDir.IsDir() {
 			continue
 		}
-		if isStaticVolume(volumeDir.Name()) {
-			staticModels += 1
+		volumeName := volumeDir.Name()
+		if isStaticVolume(volumeName) {
+			statusPath := filepath.Join(volumesDir, volumeName, "status.json")
+			modelStatus, err := cm.sm.Get(statusPath)
+			if err == nil {
+				mountItems = append(mountItems, metrics.MountItem{
+					Reference:   modelStatus.Reference,
+					Type:        "static",
+					VolumeName: volumeName,
+					MountID:    modelStatus.MountID,
+				})
+				staticModels += 1
+			}
 		}
-		if isDynamicVolume(volumeDir.Name()) {
-			modelsDir := cm.cfg.Get().GetModelsDirForDynamic(volumeDir.Name())
+		if isDynamicVolume(volumeName) {
+			modelsDir := cm.cfg.Get().GetModelsDirForDynamic(volumeName)
 			modelDirs, err := os.ReadDir(modelsDir)
 			if err != nil {
-				return errors.Wrapf(err, "read model dirs from %s", modelsDir)
+				if os.IsNotExist(err) {
+					statusPath := filepath.Join(modelsDir, "status.json")
+					modelStatus, err := cm.sm.Get(statusPath)
+					if err == nil {
+						mountItems = append(mountItems, metrics.MountItem{
+							Reference:   modelStatus.Reference,
+							Type:        "inline",
+							VolumeName: volumeName,
+							MountID:    modelStatus.MountID,
+						})
+						inlineModels += 1
+					}
+					continue
+				}
+				logger.Logger().WithError(err).Warnf("read model dirs from %s", modelsDir)
+				continue
 			}
-			for _, modelDir := range modelDirs {
-				if !modelDir.IsDir() {
+			for _, modelID := range modelDirs {
+				if !modelID.IsDir() {
 					continue
 				}
-				dynamicModels += 1
+				statusPath := filepath.Join(modelsDir, modelID.Name(), "status.json")
+				modelStatus, err := cm.sm.Get(statusPath)
+				if err == nil {
+					mountItems = append(mountItems, metrics.MountItem{
+						Reference:   modelStatus.Reference,
+						Type:        "dynamic",
+						VolumeName: volumeName,
+						MountID:    modelStatus.MountID,
+					})
+					dynamicModels += 1
+				}
 			}
 		}
 	}
+
+	metrics.MountItems.Set(mountItems)
 	metrics.NodeMountedStaticImages.Set(float64(staticModels))
+	metrics.NodeMountedInlineImages.Set(float64(inlineModels))
 	metrics.NodeMountedDynamicImages.Set(float64(dynamicModels))
+
 	return nil
 }
 
@@ -87,17 +132,18 @@ func (cm *CacheManager) Scan() error {
 	return nil
 }
 
-func NewCacheManager(cfg *config.Config) (*CacheManager, error) {
+func NewCacheManager(cfg *config.Config, sm *status.StatusManager) (*CacheManager, error) {
 	cm := CacheManager{
 		cfg: cfg,
+		sm:  sm,
 	}
 
 	go func() {
 		for {
 			if err := cm.Scan(); err != nil && !errors.Is(err, os.ErrNotExist) {
 				logger.Logger().WithError(err).Warnf("scan cache failed")
 			}
-			time.Sleep(CacheSacnInterval)
+			time.Sleep(CacheScanInterval)
 		}
 	}()
 

pkg/service/cache_test.go

@@ -0,0 +1,113 @@
+package service
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/modelpack/model-csi-driver/pkg/config"
+	"github.com/modelpack/model-csi-driver/pkg/metrics"
+	"github.com/modelpack/model-csi-driver/pkg/status"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCacheManagerScanUpdatesMetrics(t *testing.T) {
+	tempDir := t.TempDir()
+
+	rawCfg := &config.RawConfig{ServiceName: "test", RootDir: tempDir}
+	cfg := config.NewWithRaw(rawCfg)
+
+	sm, err := status.NewStatusManager()
+	require.NoError(t, err)
+
+	// Create a static volume status
+	staticStatusPath := filepath.Join(tempDir, "volumes", "pvc-static", "status.json")
+	_, err = sm.Set(staticStatusPath, status.Status{Reference: "ref-static", MountID: "m-static"})
+	require.NoError(t, err)
+
+	// Create a dynamic volume status under models/<mountID>/status.json
+	dynamicStatusPath := filepath.Join(tempDir, "volumes", "csi-dyn", "models", "mount-1", "status.json")
+	_, err = sm.Set(dynamicStatusPath, status.Status{Reference: "ref-dyn", MountID: "mount-1"})
+	require.NoError(t, err)
+
+	// An extra file to ensure cache size covers arbitrary files under RootDir.
+	extraPath := filepath.Join(tempDir, "extra.bin")
+	require.NoError(t, os.WriteFile(extraPath, []byte("abc"), 0o644))
+
+	paths := []string{staticStatusPath, dynamicStatusPath, extraPath}
+	var expectedSize int64
+	for _, p := range paths {
+		st, statErr := os.Stat(p)
+		require.NoError(t, statErr)
+		expectedSize += st.Size()
+	}
+
+	cm := &CacheManager{cfg: cfg, sm: sm}
+	require.NoError(t, cm.Scan())
+
+	require.Equal(t, float64(expectedSize), testutil.ToFloat64(metrics.NodeCacheSizeInBytes))
+	require.Equal(t, float64(1), testutil.ToFloat64(metrics.NodeMountedStaticImages))
+	require.Equal(t, float64(0), testutil.ToFloat64(metrics.NodeMountedInlineImages))
+	require.Equal(t, float64(1), testutil.ToFloat64(metrics.NodeMountedDynamicImages))
+
+	// Verify mount item metrics are exported as a snapshot without Reset/Delete races.
+	reg := prometheus.NewRegistry()
+	reg.MustRegister(metrics.MountItems)
+
+	mfs, err := reg.Gather()
+	require.NoError(t, err)
+
+	mf := findMetricFamily(t, mfs, metrics.Prefix+"mount_item")
+	require.Len(t, mf.Metric, 2)
+
+	require.True(t, hasLabels(mf.Metric[0], map[string]string{
+		"reference": "ref-static",
+		"type": "static",
+		"volume_name": "pvc-static",
+		"mount_id": "m-static",
+	}) || hasLabels(mf.Metric[1], map[string]string{
+		"reference": "ref-static",
+		"type": "static",
+		"volume_name": "pvc-static",
+		"mount_id": "m-static",
+	}))
+
+	require.True(t, hasLabels(mf.Metric[0], map[string]string{
+		"reference": "ref-dyn",
+		"type": "dynamic",
+		"volume_name": "csi-dyn",
+		"mount_id": "mount-1",
+	}) || hasLabels(mf.Metric[1], map[string]string{
+		"reference": "ref-dyn",
+		"type": "dynamic",
+		"volume_name": "csi-dyn",
+		"mount_id": "mount-1",
+	}))
+}
+
+func findMetricFamily(t *testing.T, mfs []*dto.MetricFamily, name string) *dto.MetricFamily {
+	t.Helper()
+	for _, mf := range mfs {
+		if mf.GetName() == name {
+			return mf
+		}
+	}
+	require.FailNow(t, "metric family not found", name)
+	return nil
+}
+
+func hasLabels(m *dto.Metric, want map[string]string) bool {
+	labels := map[string]string{}
+	for _, lp := range m.GetLabel() {
+		labels[lp.GetName()] = lp.GetValue()
+	}
+	for k, v := range want {
+		if labels[k] != v {
+			return false
+		}
+	}
+	return true
+}

pkg/service/service.go

@@ -75,7 +75,7 @@ func New(cfg *config.Config) (*Service, error) {
 		if err != nil {
 			return nil, errors.Wrap(err, "create worker")
 		}
-		cm, err := NewCacheManager(cfg)
+		cm, err := NewCacheManager(cfg, sm)
 		if err != nil {
 			return nil, errors.Wrap(err, "create cache manager")
 		}