metrics: collect mount items for detail metrics

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 (pvc, inline, dynamic types), 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

@@ -76,15 +76,21 @@ var (
 		},
 	)
 
-	NodeMountedStaticImages = prometheus.NewGauge(
+	NodeMountedPVCModels = prometheus.NewGauge(
 		prometheus.GaugeOpts{
-			Name: Prefix + "node_mounted_static_images",
+			Name: Prefix + "node_mounted_pvc_models",
 		},
 	)
 
-	NodeMountedDynamicImages = prometheus.NewGauge(
+	NodeMountedInlineModels = prometheus.NewGauge(
 		prometheus.GaugeOpts{
-			Name: Prefix + "node_mounted_dynamic_images",
+			Name: Prefix + "node_mounted_inline_models",
+		},
+	)
+
+	NodeMountedDynamicModels = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Name: Prefix + "node_mounted_dynamic_models",
 		},
 	)
 
@@ -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,
@@ -155,8 +163,9 @@ func init() {
 		ControllerOpLatency,
 
 		NodeCacheSizeInBytes,
-		NodeMountedStaticImages,
-		NodeMountedDynamicImages,
+		NodeMountedPVCModels,
+		NodeMountedInlineModels,
+		NodeMountedDynamicModels,
 		NodePullLayerTooLong,
 	)
 }

pkg/server/server_test.go

@@ -558,7 +558,7 @@ func TestServer(t *testing.T) {
 	require.NoError(t, err)
 	cfg.Get().RootDir = rootDir
 	cfg.Get().PullConfig.ProxyURL = ""
-	service.CacheSacnInterval = 1 * time.Second
+	service.CacheScanInterval = 1 * time.Second
 
 	service.NewPuller = func(ctx context.Context, pullCfg *config.PullConfig, hook *status.Hook, diskQuotaChecker *service.DiskQuotaChecker) service.Puller {
 		return &mockPuller{

pkg/service/cache.go

@@ -8,34 +8,35 @@ 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
+var CacheScanInterval = 60 * time.Second
+
+const (
+	mountTypePVC = "pvc"
+	mountTypeInline = "inline"
+	mountTypeDynamic = "dynamic"
+)
 
 type CacheManager struct {
 	cfg *config.Config
+	sm *status.StatusManager
 }
 
 func (cm *CacheManager) getCacheSize() (int64, error) {
-	var total int64
-	if err := filepath.Walk(cm.cfg.Get().RootDir, func(path string, info os.FileInfo, err error) error {
-		if err != nil {
-			return err
-		}
-		if info.IsDir() {
-			return nil
-		}
-		total += info.Size()
-		return nil
-	}); err != nil {
-		return 0, err
+	size, err := getUsedSize(cm.cfg.Get().RootDir)
+	if err != nil {
+		return 0, errors.Wrapf(err, "get used size: %s", cm.cfg.Get().RootDir)
 	}
-	return total, nil
+
+	return size, nil
 }
 
 func (cm *CacheManager) scanModels() error {
-	staticModels := 0
+	pvcModels := 0
+	inlineModels := 0
 	dynamicModels := 0
 	volumesDir := cm.cfg.Get().GetVolumesDir()
 	volumeDirs, err := os.ReadDir(volumesDir)
@@ -45,29 +46,73 @@ 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:        mountTypePVC,
+					VolumeName: volumeName,
+					MountID:    modelStatus.MountID,
+				})
+				pvcModels += 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 err != nil {
+				if os.IsNotExist(err) {
+					// This is potentially an inline model, the status file is expected
+					// to be directly under the volume directory.
+					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:        mountTypeInline,
+							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() {
 					continue
 				}
-				dynamicModels += 1
+				statusPath := filepath.Join(modelsDir, modelDir.Name(), "status.json")
+				modelStatus, err := cm.sm.Get(statusPath)
+				if err == nil {
+					mountItems = append(mountItems, metrics.MountItem{
+						Reference:   modelStatus.Reference,
+						Type:        mountTypeDynamic,
+						VolumeName: volumeName,
+						MountID:    modelStatus.MountID,
+					})
+					dynamicModels += 1
+				}
 			}
 		}
 	}
-	metrics.NodeMountedStaticImages.Set(float64(staticModels))
-	metrics.NodeMountedDynamicImages.Set(float64(dynamicModels))
+
+	metrics.MountItems.Set(mountItems)
+	metrics.NodeMountedPVCModels.Set(float64(pvcModels))
+	metrics.NodeMountedInlineModels.Set(float64(inlineModels))
+	metrics.NodeMountedDynamicModels.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)
 		}
 	}()