noot

Author: elee1766

cmd/volmetd/main.go

@@ -66,7 +66,7 @@ func main() {
 	capacity := collector.NewCapacityCollector()
 
 	// Create and register volume collector
-	vc := collector.NewVolumeCollector(multi, diskstats, capacity)
+	vc := collector.NewVolumeCollector(multi, cfg.HostProcPath, diskstats, capacity)
 	prometheus.MustRegister(vc)
 
 	// HTTP server

pkg/collector/collector.go

@@ -9,6 +9,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/gfx-labs/volmetd/pkg/discovery"
+	"github.com/gfx-labs/volmetd/pkg/diskstats"
 )
 
 // Collector collects metrics for discovered volumes
@@ -41,13 +42,18 @@ var (
 type VolumeCollector struct {
 	discoverer *discovery.MultiDiscoverer
 	collectors []Collector
+	procPath   string
 }
 
 // NewVolumeCollector creates a new volume collector
-func NewVolumeCollector(discoverer *discovery.MultiDiscoverer, collectors ...Collector) *VolumeCollector {
+func NewVolumeCollector(discoverer *discovery.MultiDiscoverer, procPath string, collectors ...Collector) *VolumeCollector {
+	if procPath == "" {
+		procPath = "/proc"
+	}
 	return &VolumeCollector{
 		discoverer: discoverer,
 		collectors: collectors,
+		procPath:   procPath,
 	}
 }
 
@@ -76,6 +82,9 @@ func (v *VolumeCollector) Collect(ch chan<- prometheus.Metric) {
 	ch <- prometheus.MustNewConstMetric(scrapeSuccessDesc, prometheus.GaugeValue, 1, "discovery")
 	ch <- prometheus.MustNewConstMetric(volumesDiscoveredDesc, prometheus.GaugeValue, float64(len(volumes)))
 
+	// Resolve device names from diskstats before running collectors
+	v.resolveDeviceNames(volumes)
+
 	// Run collectors in parallel
 	wg := sync.WaitGroup{}
 	wg.Add(len(v.collectors))
@@ -104,3 +113,21 @@ func (v *VolumeCollector) execute(c Collector, volumes []*discovery.VolumeInfo,
 	}
 	ch <- prometheus.MustNewConstMetric(scrapeSuccessDesc, prometheus.GaugeValue, 1, c.Name())
 }
+
+// resolveDeviceNames resolves device names from diskstats using device IDs
+func (v *VolumeCollector) resolveDeviceNames(volumes []*discovery.VolumeInfo) {
+	stats, err := diskstats.Parse(v.procPath + "/diskstats")
+	if err != nil {
+		log.Printf("failed to parse diskstats for device name resolution: %v", err)
+		return
+	}
+
+	for _, vol := range volumes {
+		// Try to resolve device name from device ID
+		if vol.DeviceID != "" {
+			if s, ok := stats.ByDeviceID[vol.DeviceID]; ok {
+				vol.DeviceName = s.DeviceName
+			}
+		}
+	}
+}

pkg/collector/diskstats.go

@@ -76,10 +76,16 @@ func (d *DiskstatsCollector) Update(volumes []*discovery.VolumeInfo, ch chan<- p
 
 	wg := sync.WaitGroup{}
 	for _, vol := range volumes {
-		s, ok := stats[vol.DeviceName]
+		// Device name should already be resolved by VolumeCollector
+		if vol.DeviceName == "" {
+			continue
+		}
+
+		s, ok := stats.ByName[vol.DeviceName]
 		if !ok {
 			continue
 		}
+
 		wg.Add(1)
 		go func(vol *discovery.VolumeInfo, s *diskstats.Stats) {
 			defer wg.Done()

pkg/discovery/csi.go

@@ -115,6 +115,9 @@ func (d *CSIDiscoverer) discoverCSIVolumes(ctx context.Context, podUID, csiDir s
 		// Resolve symlinks to get actual device for diskstats
 		resolvedPath, deviceName := mounts.ResolveDevice(mount.Device)
 
+		// Get device ID from mount point for reliable diskstats lookup
+		deviceID, _ := mounts.GetDeviceID(mountPath)
+
 		vol := &VolumeInfo{
 			PVName:        volData.VolumeName,
 			PVCName:       extractPVCName(volData.VolumeName),
@@ -127,6 +130,7 @@ func (d *CSIDiscoverer) discoverCSIVolumes(ctx context.Context, podUID, csiDir s
 			CSIDevicePath: mount.Device,
 			DevicePath:    resolvedPath,
 			DeviceName:    deviceName,
+			DeviceID:      deviceID,
 			MountPath:     mountPath,
 		}
 

pkg/discovery/k8sapi.go

@@ -3,6 +3,7 @@ package discovery
 import (
 	"context"
 	"fmt"
+	"log"
 	"os"
 	"path/filepath"
 	"strings"
@@ -43,6 +44,7 @@ func NewK8sAPIDiscoverer(kubeletPath, mountsPath string, namespaces []string) (*
 	}
 
 	nodeName := detectNodeName()
+	log.Printf("k8sapi: detected node name: %q", nodeName)
 
 	if kubeletPath == "" {
 		kubeletPath = "/var/lib/kubelet"
@@ -93,12 +95,21 @@ func (d *K8sAPIDiscoverer) Name() string {
 }
 
 func (d *K8sAPIDiscoverer) Available(ctx context.Context) bool {
-	if d.client == nil || d.nodeName == "" {
+	if d.client == nil {
+		log.Printf("k8sapi: client is nil")
+		return false
+	}
+	if d.nodeName == "" {
+		log.Printf("k8sapi: node name not detected")
 		return false
 	}
 	// Quick check that we can talk to the API
 	_, err := d.client.CoreV1().Nodes().Get(ctx, d.nodeName, metav1.GetOptions{})
-	return err == nil
+	if err != nil {
+		log.Printf("k8sapi: cannot get node %s: %v", d.nodeName, err)
+		return false
+	}
+	return true
 }
 
 func (d *K8sAPIDiscoverer) Discover(ctx context.Context) ([]*VolumeInfo, error) {
@@ -167,6 +178,9 @@ func (d *K8sAPIDiscoverer) Discover(ctx context.Context) ([]*VolumeInfo, error)
 			// Resolve symlinks to get actual device for diskstats
 			resolvedPath, deviceName := mounts.ResolveDevice(mount.Device)
 
+			// Get device ID from mount point for reliable diskstats lookup
+			deviceID, _ := mounts.GetDeviceID(mountPath)
+
 			// Find container mount path
 			containerMountPath := findContainerMountPath(&pod, vol.Name)
 
@@ -182,6 +196,7 @@ func (d *K8sAPIDiscoverer) Discover(ctx context.Context) ([]*VolumeInfo, error)
 				CSIDevicePath:      mount.Device,
 				DevicePath:         resolvedPath,
 				DeviceName:         deviceName,
+				DeviceID:           deviceID,
 				MountPath:          mountPath,
 				ContainerMountPath: containerMountPath,
 			}

pkg/discovery/types.go

@@ -1,6 +1,9 @@
 package discovery
 
-import "context"
+import (
+	"context"
+	"log"
+)
 
 // VolumeInfo represents a discovered PVC volume
 type VolumeInfo struct {
@@ -22,6 +25,7 @@ type VolumeInfo struct {
 	// Node-local info
 	DevicePath         string // resolved device path, e.g., /dev/sda
 	DeviceName         string // device name for diskstats, e.g., sda
+	DeviceID           string // major:minor device ID for diskstats lookup, e.g., "8:0"
 	CSIDevicePath      string // original CSI device path, e.g., /dev/disk/by-id/scsi-0DO_Volume_...
 	MountPath          string // host path, e.g., /var/lib/kubelet/pods/.../volumes/...
 	ContainerMountPath string // path inside container, e.g., /data
@@ -51,28 +55,37 @@ func NewMultiDiscoverer(discoverers ...Discoverer) *MultiDiscoverer {
 
 // Discover tries all discoverers and returns merged results
 func (m *MultiDiscoverer) Discover(ctx context.Context) ([]*VolumeInfo, error) {
-	seen := make(map[string]*VolumeInfo) // key by device name
+	seen := make(map[string]*VolumeInfo) // key by device ID (preferred) or device name
 
 	for _, d := range m.discoverers {
 		if !d.Available(ctx) {
+			log.Printf("discoverer %s not available", d.Name())
 			continue
 		}
 
 		volumes, err := d.Discover(ctx)
 		if err != nil {
-			// Log but continue with other discoverers
+			log.Printf("discoverer %s error: %v", d.Name(), err)
 			continue
 		}
 
+		log.Printf("discoverer %s found %d volumes", d.Name(), len(volumes))
+
 		for _, v := range volumes {
-			if v.DeviceName == "" {
+			// Use device ID as key if available, otherwise device name
+			key := v.DeviceID
+			if key == "" {
+				key = v.DeviceName
+			}
+			if key == "" {
 				continue
 			}
-			if existing, exists := seen[v.DeviceName]; exists {
+
+			if existing, exists := seen[key]; exists {
 				// Merge: fill in empty fields from new discoverer
 				mergeVolumeInfo(existing, v)
 			} else {
-				seen[v.DeviceName] = v
+				seen[key] = v
 			}
 		}
 	}
@@ -119,6 +132,9 @@ func mergeVolumeInfo(dst, src *VolumeInfo) {
 	if dst.DevicePath == "" {
 		dst.DevicePath = src.DevicePath
 	}
+	if dst.DeviceID == "" {
+		dst.DeviceID = src.DeviceID
+	}
 	if dst.CSIDevicePath == "" {
 		dst.CSIDevicePath = src.CSIDevicePath
 	}

pkg/diskstats/diskstats.go

@@ -53,8 +53,14 @@ func (s *Stats) WriteBytesTotal() uint64 {
 	return s.SectorsWritten * 512
 }
 
+// StatsMap holds diskstats indexed by both device name and major:minor
+type StatsMap struct {
+	ByName     map[string]*Stats // keyed by device name (e.g., "sda")
+	ByDeviceID map[string]*Stats // keyed by "major:minor" (e.g., "8:0")
+}
+
 // Parse reads /proc/diskstats and returns stats for all devices
-func Parse(path string) (map[string]*Stats, error) {
+func Parse(path string) (*StatsMap, error) {
 	if path == "" {
 		path = "/proc/diskstats"
 	}
@@ -65,15 +71,20 @@ func Parse(path string) (map[string]*Stats, error) {
 	}
 	defer f.Close()
 
-	result := make(map[string]*Stats)
+	result := &StatsMap{
+		ByName:     make(map[string]*Stats),
+		ByDeviceID: make(map[string]*Stats),
+	}
 	scanner := bufio.NewScanner(f)
 
 	for scanner.Scan() {
 		stats, err := parseLine(scanner.Text())
 		if err != nil {
 			continue // skip malformed lines
 		}
-		result[stats.DeviceName] = stats
+		result.ByName[stats.DeviceName] = stats
+		deviceID := fmt.Sprintf("%d:%d", stats.Major, stats.Minor)
+		result.ByDeviceID[deviceID] = stats
 	}
 
 	if err := scanner.Err(); err != nil {

pkg/mounts/mounts.go

@@ -111,6 +111,22 @@ func ResolveDevice(devicePath string) (resolvedPath, deviceName string) {
 	return resolved, name
 }
 
+// GetDeviceID returns the major:minor device ID for a mount point
+// This works by stat'ing the mount point and extracting the device ID
+func GetDeviceID(mountPoint string) (string, error) {
+	var stat syscall.Stat_t
+	if err := syscall.Stat(mountPoint, &stat); err != nil {
+		return "", fmt.Errorf("stat %s: %w", mountPoint, err)
+	}
+
+	// Dev contains major:minor encoded
+	// On Linux: major = (dev >> 8) & 0xfff, minor = (dev & 0xff) | ((dev >> 12) & 0xfff00)
+	major := (stat.Dev >> 8) & 0xfff
+	minor := (stat.Dev & 0xff) | ((stat.Dev >> 12) & 0xfff00)
+
+	return fmt.Sprintf("%d:%d", major, minor), nil
+}
+
 // evalSymlinks resolves all symlinks in a path
 func evalSymlinks(path string) (string, error) {
 	// Use filepath.EvalSymlinks equivalent