Sth/kcm 5797 (#3810)

Signed-off-by: Sean Holcomb <seanholcomb@gmail.com>

core/pkg/clustercache/helper.go

@@ -0,0 +1,15 @@
+package clustercache
+
+func GetLoadBalancerIngressAddress(service *Service) []string {
+	var addresses []string
+	for _, loadBalancerIngress := range service.Status.LoadBalancer.Ingress {
+		address := loadBalancerIngress.IP
+		// Some cloud providers use hostname rather than IP
+		if address == "" {
+			address = loadBalancerIngress.Hostname
+		}
+		addresses = append(addresses, address)
+
+	}
+	return addresses
+}

core/pkg/clustercache/helper_test.go

@@ -0,0 +1,89 @@
+package clustercache
+
+import (
+	"testing"
+
+	v1 "k8s.io/api/core/v1"
+)
+
+func TestGetLoadBalancerIngressAddress(t *testing.T) {
+	tests := []struct {
+		name     string
+		service  *Service
+		expected []string
+	}{
+		{
+			name:     "no ingresses",
+			service:  &Service{},
+			expected: nil,
+		},
+		{
+			name: "single IP ingress",
+			service: &Service{
+				Status: v1.ServiceStatus{
+					LoadBalancer: v1.LoadBalancerStatus{
+						Ingress: []v1.LoadBalancerIngress{
+							{IP: "1.2.3.4"},
+						},
+					},
+				},
+			},
+			expected: []string{"1.2.3.4"},
+		},
+		{
+			name: "single hostname ingress",
+			service: &Service{
+				Status: v1.ServiceStatus{
+					LoadBalancer: v1.LoadBalancerStatus{
+						Ingress: []v1.LoadBalancerIngress{
+							{Hostname: "lb.example.com"},
+						},
+					},
+				},
+			},
+			expected: []string{"lb.example.com"},
+		},
+		{
+			name: "IP takes priority over hostname",
+			service: &Service{
+				Status: v1.ServiceStatus{
+					LoadBalancer: v1.LoadBalancerStatus{
+						Ingress: []v1.LoadBalancerIngress{
+							{IP: "1.2.3.4", Hostname: "lb.example.com"},
+						},
+					},
+				},
+			},
+			expected: []string{"1.2.3.4"},
+		},
+		{
+			name: "multiple ingresses",
+			service: &Service{
+				Status: v1.ServiceStatus{
+					LoadBalancer: v1.LoadBalancerStatus{
+						Ingress: []v1.LoadBalancerIngress{
+							{IP: "1.2.3.4"},
+							{Hostname: "lb2.example.com"},
+							{IP: "5.6.7.8"},
+						},
+					},
+				},
+			},
+			expected: []string{"1.2.3.4", "lb2.example.com", "5.6.7.8"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := GetLoadBalancerIngressAddress(tt.service)
+			if len(got) != len(tt.expected) {
+				t.Fatalf("got %v, want %v", got, tt.expected)
+			}
+			for i := range tt.expected {
+				if got[i] != tt.expected[i] {
+					t.Errorf("index %d: got %q, want %q", i, got[i], tt.expected[i])
+				}
+			}
+		})
+	}
+}

core/pkg/model/kubemodel/kubemodel.go

@@ -7,24 +7,24 @@ import (
 
 // @bingen:generate[stringtable,streamable]:KubeModelSet
 type KubeModelSet struct {
-	Metadata               *Metadata                         `json:"meta"`                   // @bingen:field[version=1]
-	Window                 Window                            `json:"window"`                 // @bingen:field[version=1]
-	Cluster                *Cluster                          `json:"cluster"`                // @bingen:field[version=1]
-	Namespaces             map[string]*Namespace             `json:"namespaces"`             // @bingen:field[version=1]
-	ResourceQuotas         map[string]*ResourceQuota         `json:"resourceQuotas"`         // @bingen:field[version=1]
-	Containers             map[string]*Container             `json:"containers,omitempty"`   // @bingen:field[version=2]
-	Deployments            map[string]*Deployment            `json:"deployments,omitempty"`  // @bingen:field[version=2]
-	StatefulSets           map[string]*StatefulSet           `json:"statefulSets,omitempty"` // @bingen:field[version=2]
-	DaemonSets             map[string]*DaemonSet             `json:"daemonSets,omitempty"`   // @bingen:field[version=2]
-	Jobs                   map[string]*Job                   `json:"jobs,omitempty"`         // @bingen:field[version=2]
-	CronJobs               map[string]*CronJob               `json:"cronJobs,omitempty"`     // @bingen:field[version=2]
-	ReplicaSets            map[string]*ReplicaSet            `json:"replicaSets,omitempty"`  // @bingen:field[version=2]
-	Nodes                  map[string]*Node                  `json:"nodes,omitempty"`        // @bingen:field[version=2]
-	Pods                   map[string]*Pod                   `json:"pods,omitempty"`         // @bingen:field[version=2]
-	PersistentVolumeClaims map[string]*PersistentVolumeClaim `json:"pvcs,omitempty"`         // @bingen:field[version=2]
-	Services               map[string]*Service               `json:"services,omitempty"`     // @bingen:field[version=2]
-	PersistentVolumes      map[string]*PersistentVolume      `json:"volumes,omitempty"`      // @bingen:field[version=2]
-	DCGMDevices            map[string]*DCGMDevice            `json:"dcgmDevices,omitempty"`  // @bingen:field[version=2]
+	Metadata               *Metadata                         `json:"meta"`              // @bingen:field[version=1]
+	Window                 Window                            `json:"window"`            // @bingen:field[version=1]
+	Cluster                *Cluster                          `json:"cluster"`           // @bingen:field[version=1]
+	Namespaces             map[string]*Namespace             `json:"namespaces"`        // @bingen:field[version=1]
+	ResourceQuotas         map[string]*ResourceQuota         `json:"resourceQuotas"`    // @bingen:field[version=1]
+	Containers             map[string]*Container             `json:"containers"`        // @bingen:field[version=2]
+	Deployments            map[string]*Deployment            `json:"deployments"`       // @bingen:field[version=2]
+	StatefulSets           map[string]*StatefulSet           `json:"statefulSets"`      // @bingen:field[version=2]
+	DaemonSets             map[string]*DaemonSet             `json:"daemonSets"`        // @bingen:field[version=2]
+	Jobs                   map[string]*Job                   `json:"jobs"`              // @bingen:field[version=2]
+	CronJobs               map[string]*CronJob               `json:"cronJobs"`          // @bingen:field[version=2]
+	ReplicaSets            map[string]*ReplicaSet            `json:"replicaSets"`       // @bingen:field[version=2]
+	Nodes                  map[string]*Node                  `json:"nodes"`             // @bingen:field[version=2]
+	Pods                   map[string]*Pod                   `json:"pods"`              // @bingen:field[version=2]
+	PersistentVolumeClaims map[string]*PersistentVolumeClaim `json:"pvcs"`              // @bingen:field[version=2]
+	Services               map[string]*Service               `json:"services"`          // @bingen:field[version=2]
+	PersistentVolumes      map[string]*PersistentVolume      `json:"persistentVolumes"` // @bingen:field[version=2]
+	DCGMDevices            map[string]*DCGMDevice            `json:"dcgmDevices"`       // @bingen:field[version=2]
 	idx                    *kubeModelSetIndexes              // @bingen:field[ignore]
 }
 

core/pkg/model/kubemodel/kubemodel_codecs.go

@@ -4103,7 +4103,6 @@ func (target *KubeModelSet) UnmarshalBinaryWithContext(ctx *DecodingContext) (er
 		if buff.ReadUInt8() == uint8(0) {
 			target.Cluster = nil
 		} else {
-
 			// --- [begin][read][struct](Cluster) ---
 			c := new(Cluster)
 			buff.ReadInt() // [compatibility, unused]
@@ -4143,7 +4142,6 @@ func (target *KubeModelSet) UnmarshalBinaryWithContext(ctx *DecodingContext) (er
 				if buff.ReadUInt8() == uint8(0) {
 					z = nil
 				} else {
-
 					// --- [begin][read][struct](Namespace) ---
 					l := new(Namespace)
 					buff.ReadInt() // [compatibility, unused]
@@ -4327,7 +4325,6 @@ func (target *KubeModelSet) UnmarshalBinaryWithContext(ctx *DecodingContext) (er
 				if buff.ReadUInt8() == uint8(0) {
 					zzzzz = nil
 				} else {
-
 					// --- [begin][read][struct](StatefulSet) ---
 					oo := new(StatefulSet)
 					buff.ReadInt() // [compatibility, unused]
@@ -4465,7 +4462,6 @@ func (target *KubeModelSet) UnmarshalBinaryWithContext(ctx *DecodingContext) (er
 				if buff.ReadUInt8() == uint8(0) {
 					zzzzzzzz = nil
 				} else {
-
 					// --- [begin][read][struct](CronJob) ---
 					lll := new(CronJob)
 					buff.ReadInt() // [compatibility, unused]
@@ -4511,7 +4507,6 @@ func (target *KubeModelSet) UnmarshalBinaryWithContext(ctx *DecodingContext) (er
 				if buff.ReadUInt8() == uint8(0) {
 					zzzzzzzzz = nil
 				} else {
-
 					// --- [begin][read][struct](ReplicaSet) ---
 					rrr := new(ReplicaSet)
 					buff.ReadInt() // [compatibility, unused]
@@ -4695,7 +4690,6 @@ func (target *KubeModelSet) UnmarshalBinaryWithContext(ctx *DecodingContext) (er
 				if buff.ReadUInt8() == uint8(0) {
 					zzzzzzzzzzzzz = nil
 				} else {
-
 					// --- [begin][read][struct](Service) ---
 					uuuu := new(Service)
 					buff.ReadInt() // [compatibility, unused]
@@ -4787,7 +4781,6 @@ func (target *KubeModelSet) UnmarshalBinaryWithContext(ctx *DecodingContext) (er
 				if buff.ReadUInt8() == uint8(0) {
 					zzzzzzzzzzzzzzz = nil
 				} else {
-
 					// --- [begin][read][struct](DCGMDevice) ---
 					lllll := new(DCGMDevice)
 					buff.ReadInt() // [compatibility, unused]
@@ -4995,7 +4988,6 @@ func (stream *KubeModelSetStream) Stream() iter.Seq2[BingenFieldInfo, *BingenVal
 					if buff.ReadUInt8() == uint8(0) {
 						z = nil
 					} else {
-
 						// --- [begin][read][struct](Namespace) ---
 						n := new(Namespace)
 						buff.ReadInt() // [compatibility, unused]
@@ -5235,7 +5227,6 @@ func (stream *KubeModelSetStream) Stream() iter.Seq2[BingenFieldInfo, *BingenVal
 					if buff.ReadUInt8() == uint8(0) {
 						zzzzz = nil
 					} else {
-
 						// --- [begin][read][struct](StatefulSet) ---
 						mm := new(StatefulSet)
 						buff.ReadInt() // [compatibility, unused]
@@ -5415,7 +5406,6 @@ func (stream *KubeModelSetStream) Stream() iter.Seq2[BingenFieldInfo, *BingenVal
 					if buff.ReadUInt8() == uint8(0) {
 						zzzzzzzz = nil
 					} else {
-
 						// --- [begin][read][struct](CronJob) ---
 						ddd := new(CronJob)
 						buff.ReadInt() // [compatibility, unused]
@@ -5475,7 +5465,6 @@ func (stream *KubeModelSetStream) Stream() iter.Seq2[BingenFieldInfo, *BingenVal
 					if buff.ReadUInt8() == uint8(0) {
 						zzzzzzzzz = nil
 					} else {
-
 						// --- [begin][read][struct](ReplicaSet) ---
 						lll := new(ReplicaSet)
 						buff.ReadInt() // [compatibility, unused]
@@ -5715,7 +5704,6 @@ func (stream *KubeModelSetStream) Stream() iter.Seq2[BingenFieldInfo, *BingenVal
 					if buff.ReadUInt8() == uint8(0) {
 						zzzzzzzzzzzzz = nil
 					} else {
-
 						// --- [begin][read][struct](Service) ---
 						hhhh := new(Service)
 						buff.ReadInt() // [compatibility, unused]
@@ -5835,7 +5823,6 @@ func (stream *KubeModelSetStream) Stream() iter.Seq2[BingenFieldInfo, *BingenVal
 					if buff.ReadUInt8() == uint8(0) {
 						zzzzzzzzzzzzzzz = nil
 					} else {
-
 						// --- [begin][read][struct](DCGMDevice) ---
 						uuuu := new(DCGMDevice)
 						buff.ReadInt() // [compatibility, unused]
@@ -7845,7 +7832,6 @@ func (target *Pod) MarshalBinaryWithContext(ctx *EncodingContext) (err error) {
 		// --- [begin][write][slice]([]NetworkTrafficDetail) ---
 		buff.WriteInt(len(target.NetworkTrafficDetails)) // slice length
 		for ii := range target.NetworkTrafficDetails {
-
 			// --- [begin][write][struct](NetworkTrafficDetail) ---
 			buff.WriteInt(0) // [compatibility, unused]
 			errC := target.NetworkTrafficDetails[ii].MarshalBinaryWithContext(ctx)
@@ -8099,7 +8085,6 @@ func (target *Pod) UnmarshalBinaryWithContext(ctx *DecodingContext) (err error)
 		tt := buff.ReadInt() // slice len
 		ss := make([]NetworkTrafficDetail, tt)
 		for ii := range tt {
-
 			// --- [begin][read][struct](NetworkTrafficDetail) ---
 			ww := new(NetworkTrafficDetail)
 			buff.ReadInt() // [compatibility, unused]
@@ -8342,6 +8327,7 @@ func (target *ReplicaSet) MarshalBinaryWithContext(ctx *EncodingContext) (err er
 		// --- [begin][write][slice]([]Owner) ---
 		buff.WriteInt(len(target.Owners)) // slice length
 		for i := range target.Owners {
+
 			// --- [begin][write][struct](Owner) ---
 			buff.WriteInt(0) // [compatibility, unused]
 			errA := target.Owners[i].MarshalBinaryWithContext(ctx)
@@ -8516,6 +8502,7 @@ func (target *ReplicaSet) UnmarshalBinaryWithContext(ctx *DecodingContext) (err
 		n := buff.ReadInt() // slice len
 		m := make([]Owner, n)
 		for i := range n {
+
 			// --- [begin][read][struct](Owner) ---
 			p := new(Owner)
 			buff.ReadInt() // [compatibility, unused]
@@ -8794,7 +8781,6 @@ func (target *ResourceQuantity) UnmarshalBinaryWithContext(ctx *DecodingContext)
 	}
 	// field version check
 	if uint8(1) <= version {
-
 		// --- [begin][read][alias](Unit) ---
 		var e string
 		var g string
@@ -10055,6 +10041,13 @@ func (target *Service) MarshalBinaryWithContext(ctx *EncodingContext) (err error
 
 	}
 
+	if ctx.IsStringTable() {
+		l := ctx.Table.AddOrGet(target.LBIngressAddress)
+		buff.WriteInt(l) // write table index
+	} else {
+		buff.WriteString(target.LBIngressAddress) // write string
+	}
+
 	return nil
 }
 
@@ -10212,6 +10205,16 @@ func (target *Service) UnmarshalBinaryWithContext(ctx *DecodingContext) (err err
 
 	}
 
+	var hh string
+	if ctx.IsStringTable() {
+		ll := buff.ReadInt() // read string index
+		hh = ctx.Table.At(ll)
+	} else {
+		hh = buff.ReadString() // read string
+	}
+	gg := hh
+	target.LBIngressAddress = gg
+
 	return nil
 }
 

core/pkg/model/kubemodel/pod.go

@@ -18,7 +18,7 @@ type Pod struct {
 	NodeUID               string                 `json:"nodeUid"`
 	Name                  string                 `json:"name"`
 	Owners                []Owner                `json:"owners"`
-	PVCVolumes            []PodPVCVolumes        `json:"pvcVolumes"`
+	PVCVolumes            []PodPVCVolumes        `json:"pvcVolumes,omitempty"`
 	Labels                map[string]string      `json:"labels,omitempty"`
 	Annotations           map[string]string      `json:"annotations,omitempty"`
 	NetworkTrafficDetails []NetworkTrafficDetail `json:"networkTrafficDetails,omitempty"`

core/pkg/model/kubemodel/service.go

@@ -44,7 +44,8 @@ type Service struct {
 	// Label selector to identify pods/containers targeted by this service
 	// Maps label keys to values (e.g., {"app": "nginx", "tier": "frontend"})
 	// Pods with matching labels will receive traffic from this service
-	Selector map[string]string `json:"selector,omitempty"`
+	Selector         map[string]string `json:"selector,omitempty"`
+	LBIngressAddress string            `json:"lbIngressAddress,omitempty"`
 }
 
 func (s *Service) ValidateService(window Window) error {

core/pkg/source/datasource.go

@@ -17,6 +17,10 @@ type MetricsQuerier interface {
 	QueryLocalStorageUsedMax(start, end time.Time) *Future[LocalStorageUsedMaxResult]
 	QueryLocalStorageBytes(start, end time.Time) *Future[LocalStorageBytesResult]
 
+	QueryKMLocalStorageUsedAvg(start, end time.Time) *Future[NodeUIDValueResult]
+	QueryKMLocalStorageUsedMax(start, end time.Time) *Future[NodeUIDValueResult]
+	QueryKMLocalStorageBytes(start, end time.Time) *Future[UIDValueResult]
+
 	// Nodes
 	QueryNodeInfo(start, end time.Time) *Future[NodeInfoResult]
 	QueryNodeUptime(start, end time.Time) *Future[UptimeResult]
@@ -93,6 +97,7 @@ type MetricsQuerier interface {
 	QueryPodPVCAllocation(start, end time.Time) *Future[PodPVCAllocationResult]
 	QueryPVCBytesRequested(start, end time.Time) *Future[PVCBytesRequestedResult]
 	QueryPVCInfo(start, end time.Time) *Future[PVCInfoResult]
+	QueryKMPVCInfo(start, end time.Time) *Future[PVCInfoResult]
 	QueryPVCUptime(start, end time.Time) *Future[UptimeResult]
 
 	// PV

core/pkg/source/decoders.go

@@ -37,6 +37,7 @@ const (
 	ServiceNameLabel     = "service_name"
 	ServiceTypeLabel     = "service_type"
 	IngressIPLabel       = "ingress_ip"
+	LBIngressAddress     = "lb_ingress_address"
 	ProvisionerNameLabel = "provisioner_name"
 	UIDLabel             = "uid"
 	KubernetesNodeLabel  = "kubernetes_node"
@@ -67,6 +68,35 @@ const (
 	NoneLabelValue = "<none>"
 )
 
+type UIDValueResult struct {
+	UID   string
+	Value float64
+}
+
+func DecodeUIDValueResult(result *QueryResult) *UIDValueResult {
+	return decodeValueResult(result, UIDLabel)
+}
+
+type NodeUIDValueResult UIDValueResult
+
+func DecodeNodeUIDValueResult(result *QueryResult) *NodeUIDValueResult {
+	return (*NodeUIDValueResult)(decodeValueResult(result, NodeUIDLabel))
+}
+
+func decodeValueResult(result *QueryResult, uidLabel string) *UIDValueResult {
+	uid, _ := result.GetString(uidLabel)
+	var value float64
+	if len(result.Values) > 0 {
+		value = result.Values[0].Value
+	} else {
+		log.Warnf("Error decoding value for uid '%s': empty value returned", uid)
+	}
+	return &UIDValueResult{
+		UID:   uid,
+		Value: value,
+	}
+}
+
 // UptimeResult represents the first and last recorded sample timestamp within the query window
 type UptimeResult struct {
 	UID   string
@@ -1637,11 +1667,12 @@ func DecodeServiceLabelsResult(result *QueryResult) *ServiceLabelsResult {
 }
 
 type ServiceInfoResult struct {
-	UID          string
-	Cluster      string
-	NamespaceUID string
-	Service      string
-	ServiceType  string
+	UID              string
+	Cluster          string
+	NamespaceUID     string
+	Service          string
+	ServiceType      string
+	LBIngressAddress string
 }
 
 func DecodeServiceInfoResult(result *QueryResult) *ServiceInfoResult {
@@ -1650,13 +1681,15 @@ func DecodeServiceInfoResult(result *QueryResult) *ServiceInfoResult {
 	namespaceUID, _ := result.GetString(NamespaceUIDLabel)
 	service, _ := result.GetString(ServiceLabel)
 	serviceType, _ := result.GetString(ServiceTypeLabel)
+	lbIngressAddress, _ := result.GetString(LBIngressAddress)
 
 	return &ServiceInfoResult{
-		UID:          uid,
-		Cluster:      cluster,
-		NamespaceUID: namespaceUID,
-		Service:      service,
-		ServiceType:  serviceType,
+		UID:              uid,
+		Cluster:          cluster,
+		NamespaceUID:     namespaceUID,
+		Service:          service,
+		ServiceType:      serviceType,
+		LBIngressAddress: lbIngressAddress,
 	}
 }
 

modules/collector-source/pkg/collector/collector.go

@@ -26,6 +26,10 @@ func NewOpenCostMetricStore() metric.MetricStore {
 	memStore.Register(NewLocalStorageUsedMaxMetricCollector())
 	memStore.Register(NewLocalStorageBytesMetricCollector())
 	memStore.Register(NewLocalStorageActiveMinutesMetricCollector())
+	memStore.Register(NewKMLocalStorageUsedAverageMetricCollector())
+	memStore.Register(NewKMLocalStorageUsedMaxMetricCollector())
+	memStore.Register(NewKMLocalStorageBytesMetricCollector())
+	memStore.Register(NewKMPVCInfoMetricCollector())
 	memStore.Register(NewNodeInfoMetricCollector())
 	memStore.Register(NewNodeUptimeMetricCollector())
 	memStore.Register(NewNodeResourceCapacitiesMetricCollector())
@@ -397,6 +401,54 @@ func NewLocalStorageBytesMetricCollector() *metric.MetricCollector {
 	)
 }
 
+func NewKMLocalStorageUsedAverageMetricCollector() *metric.MetricCollector {
+	return metric.NewMetricCollector(
+		metric.KMLocalStorageUsedAverageID,
+		metric.ContainerFSUsageBytes,
+		[]string{
+			source.NodeUIDLabel,
+		},
+		aggregator.AverageOverTime,
+		nil,
+	)
+}
+
+func NewKMLocalStorageUsedMaxMetricCollector() *metric.MetricCollector {
+	return metric.NewMetricCollector(
+		metric.KMLocalStorageUsedMaxID,
+		metric.ContainerFSUsageBytes,
+		[]string{
+			source.NodeUIDLabel,
+		},
+		aggregator.MaxOverTime,
+		nil,
+	)
+}
+
+func NewKMLocalStorageBytesMetricCollector() *metric.MetricCollector {
+	return metric.NewMetricCollector(
+		metric.KMLocalStorageBytesID,
+		metric.NodeFSCapacityBytes,
+		[]string{
+			source.UIDLabel,
+		},
+		aggregator.AverageOverTime,
+		nil,
+	)
+}
+
+func NewKMPVCInfoMetricCollector() *metric.MetricCollector {
+	return metric.NewMetricCollector(
+		metric.KMPVCInfoID,
+		metric.KubePersistentVolumeClaimInfo,
+		[]string{
+			source.UIDLabel,
+		},
+		aggregator.Info,
+		nil,
+	)
+}
+
 // count(
 //
 //	kube_node_labels{

modules/collector-source/pkg/collector/metricsquerier.go

@@ -384,6 +384,22 @@ func (c *collectorMetricsQuerier) QueryPVCInfo(start, end time.Time) *source.Fut
 	return queryCollector(c, start, end, metric.PVCInfoID, source.DecodePVCInfoResult)
 }
 
+func (c *collectorMetricsQuerier) QueryKMPVCInfo(start, end time.Time) *source.Future[source.PVCInfoResult] {
+	return queryCollector(c, start, end, metric.KMPVCInfoID, source.DecodePVCInfoResult)
+}
+
+func (c *collectorMetricsQuerier) QueryKMLocalStorageUsedAvg(start, end time.Time) *source.Future[source.NodeUIDValueResult] {
+	return queryCollector(c, start, end, metric.KMLocalStorageUsedAverageID, source.DecodeNodeUIDValueResult)
+}
+
+func (c *collectorMetricsQuerier) QueryKMLocalStorageUsedMax(start, end time.Time) *source.Future[source.NodeUIDValueResult] {
+	return queryCollector(c, start, end, metric.KMLocalStorageUsedMaxID, source.DecodeNodeUIDValueResult)
+}
+
+func (c *collectorMetricsQuerier) QueryKMLocalStorageBytes(start, end time.Time) *source.Future[source.UIDValueResult] {
+	return queryCollector(c, start, end, metric.KMLocalStorageBytesID, source.DecodeUIDValueResult)
+}
+
 func (c *collectorMetricsQuerier) QueryPVCUptime(start, end time.Time) *source.Future[source.UptimeResult] {
 	return queryCollector(c, start, end, metric.PVCUptimeID, source.DecodeUptimeResult)
 }

modules/collector-source/pkg/metric/collector.go

@@ -19,6 +19,7 @@ const (
 	PVUsedAverageID                            MetricCollectorID = "PVUsedAverage"
 	PVUsedMaxID                                MetricCollectorID = "PVUsedMax"
 	PVCInfoID                                  MetricCollectorID = "PVCInfo"
+	KMPVCInfoID                                MetricCollectorID = "KMPVCInfo"
 	PVCUptimeID                                MetricCollectorID = "PVCUptime"
 	PVActiveMinutesID                          MetricCollectorID = "PVActiveMinutes"
 	PVUptimeID                                 MetricCollectorID = "PVUptime"
@@ -27,6 +28,9 @@ const (
 	LocalStorageUsedMaxID                      MetricCollectorID = "LocalStorageUsedMax"
 	LocalStorageBytesID                        MetricCollectorID = "LocalStorageBytesID"
 	LocalStorageActiveMinutesID                MetricCollectorID = "LocalStorageActiveMinutes"
+	KMLocalStorageUsedAverageID                MetricCollectorID = "KMLocalStorageUsedAverage"
+	KMLocalStorageUsedMaxID                    MetricCollectorID = "KMLocalStorageUsedMax"
+	KMLocalStorageBytesID                      MetricCollectorID = "KMLocalStorageBytes"
 	NodeCPUCoresCapacityID                     MetricCollectorID = "NodeCPUCoresCapacity"
 	NodeCPUCoresAllocatableID                  MetricCollectorID = "NodeCPUCoresAllocatable"
 	NodeRAMBytesCapacityID                     MetricCollectorID = "NodeRAMBytesCapacity"

modules/collector-source/pkg/metric/repository.go

@@ -126,12 +126,6 @@ func (r *resolutionStores) update(
 	defer r.lock.Unlock()
 	limit := r.resolution.Limit()
 	if updateSet.Timestamp.Before(limit) {
-		log.Debugf(
-			"skipping update on resolution '%s' because Timestamp '%s' is before the limit '%s",
-			r.resolution.Interval(),
-			updateSet.Timestamp.Format(time.RFC3339),
-			limit.Format(time.RFC3339),
-		)
 		return
 	}
 

modules/collector-source/pkg/scrape/clustercache.go

@@ -64,7 +64,7 @@ func (ccs *ClusterCacheScraper) Scrape() []metric.Update {
 		ccs.GetScrapePods(pods, pvcs, nodeNameToUID, namespaceNameToUID, pvcNameToUID),
 		ccs.GetScrapePVCs(pvcs, namespaceNameToUID, pvNameToUID),
 		ccs.GetScrapePVs(pvs),
-		ccs.GetScrapeServices(services),
+		ccs.GetScrapeServices(services, namespaceNameToUID),
 		ccs.GetScrapeStatefulSets(statefulSets, namespaceNameToUID),
 		ccs.GetScrapeDaemonSets(daemonSets, namespaceNameToUID),
 		ccs.GetScrapeJobs(jobs, namespaceNameToUID),
@@ -333,6 +333,10 @@ func (ccs *ClusterCacheScraper) scrapePods(
 
 	var scrapeResults []metric.Update
 	for _, pod := range pods {
+		// pods without a set node name are not running
+		if pod.Spec.NodeName == "" {
+			continue
+		}
 		nodeUID, ok := nodeIndex[pod.Spec.NodeName]
 		if !ok {
 			log.Debugf("pod nodeUID missing from index for node name '%s'", pod.Spec.NodeName)
@@ -708,20 +712,36 @@ func (ccs *ClusterCacheScraper) scrapePVs(pvs []*clustercache.PersistentVolume)
 	return scrapeResults
 }
 
-func (ccs *ClusterCacheScraper) GetScrapeServices(services []*clustercache.Service) ScrapeFunc {
+func (ccs *ClusterCacheScraper) GetScrapeServices(
+	services []*clustercache.Service,
+	namespaceIndex map[string]types.UID,
+) ScrapeFunc {
 	return func() []metric.Update {
-		return ccs.scrapeServices(services)
+		return ccs.scrapeServices(services, namespaceIndex)
 	}
 }
 
-func (ccs *ClusterCacheScraper) scrapeServices(services []*clustercache.Service) []metric.Update {
+func (ccs *ClusterCacheScraper) scrapeServices(
+	services []*clustercache.Service,
+	namespaceIndex map[string]types.UID,
+) []metric.Update {
 	var scrapeResults []metric.Update
 	for _, service := range services {
+		namespaceUID := namespaceIndex[service.Namespace]
+
+		// Assuming one address for now
+		var lbIngressAddress string
+		lbIngressAddresses := clustercache.GetLoadBalancerIngressAddress(service)
+		if len(lbIngressAddresses) > 0 {
+			lbIngressAddress = lbIngressAddresses[0]
+		}
 		serviceInfo := map[string]string{
-			source.UIDLabel:         string(service.UID),
-			source.ServiceLabel:     service.Name,
-			source.NamespaceLabel:   service.Namespace,
-			source.ServiceTypeLabel: string(service.Type),
+			source.UIDLabel:          string(service.UID),
+			source.ServiceLabel:      service.Name,
+			source.NamespaceLabel:    service.Namespace,
+			source.NamespaceUIDLabel: string(namespaceUID),
+			source.ServiceTypeLabel:  string(service.Type),
+			source.LBIngressAddress:  lbIngressAddress,
 		}
 
 		scrapeResults = append(scrapeResults, metric.Update{

modules/collector-source/pkg/scrape/clustercache_test.go

@@ -1176,6 +1176,7 @@ func Test_kubernetesScraper_scrapeServices(t *testing.T) {
 	}
 	tests := []struct {
 		name     string
+		nsSetup  func(map[string]types.UID)
 		scrapes  []scrape
 		expected []metric.Update
 	}{
@@ -1201,26 +1202,32 @@ func Test_kubernetesScraper_scrapeServices(t *testing.T) {
 				{
 					Name: metric.ServiceInfo,
 					Labels: map[string]string{
-						source.UIDLabel:         "uuid1",
-						source.ServiceLabel:     "service1",
-						source.NamespaceLabel:   "namespace1",
-						source.ServiceTypeLabel: "",
+						source.UIDLabel:          "uuid1",
+						source.ServiceLabel:      "service1",
+						source.NamespaceLabel:    "namespace1",
+						source.NamespaceUIDLabel: "",
+						source.ServiceTypeLabel:  "",
+						source.LBIngressAddress:  "",
 					},
 					Value: 0,
 					AdditionalInfo: map[string]string{
-						source.UIDLabel:         "uuid1",
-						source.ServiceLabel:     "service1",
-						source.NamespaceLabel:   "namespace1",
-						source.ServiceTypeLabel: "",
+						source.UIDLabel:          "uuid1",
+						source.ServiceLabel:      "service1",
+						source.NamespaceLabel:    "namespace1",
+						source.NamespaceUIDLabel: "",
+						source.ServiceTypeLabel:  "",
+						source.LBIngressAddress:  "",
 					},
 				},
 				{
 					Name: metric.ServiceSelectorLabels,
 					Labels: map[string]string{
-						source.UIDLabel:         "uuid1",
-						source.ServiceLabel:     "service1",
-						source.NamespaceLabel:   "namespace1",
-						source.ServiceTypeLabel: "",
+						source.UIDLabel:          "uuid1",
+						source.ServiceLabel:      "service1",
+						source.NamespaceLabel:    "namespace1",
+						source.NamespaceUIDLabel: "",
+						source.ServiceTypeLabel:  "",
+						source.LBIngressAddress:  "",
 					},
 					Value: 0,
 					AdditionalInfo: map[string]string{
@@ -1230,13 +1237,131 @@ func Test_kubernetesScraper_scrapeServices(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "with namespace index",
+			nsSetup: func(nsIndex map[string]types.UID) {
+				nsIndex["namespace1"] = "ns-uuid1"
+			},
+			scrapes: []scrape{
+				{
+					Services: []*clustercache.Service{
+						{
+							Name:      "service1",
+							Namespace: "namespace1",
+							UID:       "uuid1",
+						},
+					},
+					Timestamp: start1,
+				},
+			},
+			expected: []metric.Update{
+				{
+					Name: metric.ServiceInfo,
+					Labels: map[string]string{
+						source.UIDLabel:          "uuid1",
+						source.ServiceLabel:      "service1",
+						source.NamespaceLabel:    "namespace1",
+						source.NamespaceUIDLabel: "ns-uuid1",
+						source.ServiceTypeLabel:  "",
+						source.LBIngressAddress:  "",
+					},
+					Value: 0,
+					AdditionalInfo: map[string]string{
+						source.UIDLabel:          "uuid1",
+						source.ServiceLabel:      "service1",
+						source.NamespaceLabel:    "namespace1",
+						source.NamespaceUIDLabel: "ns-uuid1",
+						source.ServiceTypeLabel:  "",
+						source.LBIngressAddress:  "",
+					},
+				},
+				{
+					Name: metric.ServiceSelectorLabels,
+					Labels: map[string]string{
+						source.UIDLabel:          "uuid1",
+						source.ServiceLabel:      "service1",
+						source.NamespaceLabel:    "namespace1",
+						source.NamespaceUIDLabel: "ns-uuid1",
+						source.ServiceTypeLabel:  "",
+						source.LBIngressAddress:  "",
+					},
+					Value:          0,
+					AdditionalInfo: map[string]string{},
+				},
+			},
+		},
+		{
+			name: "with LB ingress IP",
+			nsSetup: func(nsIndex map[string]types.UID) {
+				nsIndex["namespace1"] = "ns-uuid1"
+			},
+			scrapes: []scrape{
+				{
+					Services: []*clustercache.Service{
+						{
+							Name:      "service1",
+							Namespace: "namespace1",
+							UID:       "uuid1",
+							Type:      v1.ServiceTypeLoadBalancer,
+							Status: v1.ServiceStatus{
+								LoadBalancer: v1.LoadBalancerStatus{
+									Ingress: []v1.LoadBalancerIngress{
+										{IP: "1.2.3.4"},
+									},
+								},
+							},
+						},
+					},
+					Timestamp: start1,
+				},
+			},
+			expected: []metric.Update{
+				{
+					Name: metric.ServiceInfo,
+					Labels: map[string]string{
+						source.UIDLabel:          "uuid1",
+						source.ServiceLabel:      "service1",
+						source.NamespaceLabel:    "namespace1",
+						source.NamespaceUIDLabel: "ns-uuid1",
+						source.ServiceTypeLabel:  "LoadBalancer",
+						source.LBIngressAddress:  "1.2.3.4",
+					},
+					Value: 0,
+					AdditionalInfo: map[string]string{
+						source.UIDLabel:          "uuid1",
+						source.ServiceLabel:      "service1",
+						source.NamespaceLabel:    "namespace1",
+						source.NamespaceUIDLabel: "ns-uuid1",
+						source.ServiceTypeLabel:  "LoadBalancer",
+						source.LBIngressAddress:  "1.2.3.4",
+					},
+				},
+				{
+					Name: metric.ServiceSelectorLabels,
+					Labels: map[string]string{
+						source.UIDLabel:          "uuid1",
+						source.ServiceLabel:      "service1",
+						source.NamespaceLabel:    "namespace1",
+						source.NamespaceUIDLabel: "ns-uuid1",
+						source.ServiceTypeLabel:  "LoadBalancer",
+						source.LBIngressAddress:  "1.2.3.4",
+					},
+					Value:          0,
+					AdditionalInfo: map[string]string{},
+				},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			ks := &ClusterCacheScraper{}
+			nsIndex := make(map[string]types.UID)
+			if tt.nsSetup != nil {
+				tt.nsSetup(nsIndex)
+			}
 			var scrapeResults []metric.Update
 			for _, s := range tt.scrapes {
-				res := ks.scrapeServices(s.Services)
+				res := ks.scrapeServices(s.Services, nsIndex)
 				scrapeResults = append(scrapeResults, res...)
 			}
 

modules/prometheus-source/pkg/prom/metricsquerier.go

@@ -129,6 +129,24 @@ func (pds *PrometheusMetricsQuerier) QueryPVCInfo(start, end time.Time) *source.
 	return source.NewFuture(source.DecodePVCInfoResult, ctx.QueryAtTime(queryPVCInfo, end))
 }
 
+func (pds *PrometheusMetricsQuerier) QueryKMPVCInfo(start, end time.Time) *source.Future[source.PVCInfoResult] {
+	const queryName = "QueryKMPVCInfo"
+	const queryFmt = `avg(avg_over_time(kube_persistentvolumeclaim_info{volumename != "", %s}[%s])) by (uid, namespace_uid, persistentvolumeclaim, namespace, storageclass, volumename, persistentvolume_uid, %s)`
+
+	cfg := pds.promConfig
+
+	durStr := timeutil.DurationString(end.Sub(start))
+	if durStr == "" {
+		panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
+	}
+
+	q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
+	log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q)
+
+	ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
+	return source.NewFuture(source.DecodePVCInfoResult, ctx.QueryAtTime(q, end))
+}
+
 func (pds *PrometheusMetricsQuerier) QueryPVActiveMinutes(start, end time.Time) *source.Future[source.PVActiveMinutesResult] {
 	const queryName = "QueryPVActiveMinutes"
 	const pvActiveMinsQuery = `avg(kube_persistentvolume_capacity_bytes{%s}) by (%s, persistentvolume, uid)[%s:%dm]`
@@ -203,6 +221,61 @@ func (pds *PrometheusMetricsQuerier) QueryLocalStorageBytes(start, end time.Time
 	return source.NewFuture(source.DecodeLocalStorageBytesResult, ctx.QueryAtTime(queryLocalStorageBytes, end))
 }
 
+func (pds *PrometheusMetricsQuerier) QueryKMLocalStorageUsedAvg(start, end time.Time) *source.Future[source.NodeUIDValueResult] {
+	const queryName = "QueryKMLocalStorageUsedAvg"
+	const queryFmt = `avg(avg_over_time(container_fs_usage_bytes{%s}[%s])) by (node_uid, %s)`
+
+	cfg := pds.promConfig
+
+	durStr := timeutil.DurationString(end.Sub(start))
+	if durStr == "" {
+		panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
+	}
+
+	q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
+	log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q)
+
+	ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
+	return source.NewFuture(source.DecodeNodeUIDValueResult, ctx.QueryAtTime(q, end))
+}
+
+func (pds *PrometheusMetricsQuerier) QueryKMLocalStorageUsedMax(start, end time.Time) *source.Future[source.NodeUIDValueResult] {
+	const queryName = "QueryKMLocalStorageUsedMax"
+	const queryFmt = `max(max_over_time(container_fs_usage_bytes{%s}[%s])) by (node_uid, %s)`
+
+	cfg := pds.promConfig
+
+	durStr := timeutil.DurationString(end.Sub(start))
+	if durStr == "" {
+		panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
+	}
+
+	q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
+	log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q)
+
+	ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
+	return source.NewFuture(source.DecodeNodeUIDValueResult, ctx.QueryAtTime(q, end))
+}
+
+func (pds *PrometheusMetricsQuerier) QueryKMLocalStorageBytes(start, end time.Time) *source.Future[source.UIDValueResult] {
+	const queryName = "QueryKMLocalStorageBytes"
+	const queryFmt = `avg_over_time(node_fs_capacity_bytes{%s}[%s:%dm]) by (uid, %s)`
+
+	cfg := pds.promConfig
+	minsPerResolution := cfg.DataResolutionMinutes
+
+	durStr := pds.durationStringFor(start, end, minsPerResolution, false)
+	if durStr == "" {
+		panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
+	}
+
+	q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
+	log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q)
+
+	ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
+	return source.NewFuture(source.DecodeUIDValueResult, ctx.QueryAtTime(q, end))
+}
+
 func (pds *PrometheusMetricsQuerier) QueryLocalStorageActiveMinutes(start, end time.Time) *source.Future[source.LocalStorageActiveMinutesResult] {
 	const queryName = "QueryLocalStorageActiveMinutes"
 	const localStorageActiveMinutesQuery = `count(node_total_hourly_cost{%s}) by (%s, node, uid, instance, provider_id)[%s:%dm]`

modules/prometheus-source/pkg/prom/metricsquerier_test.go

@@ -103,6 +103,9 @@ func TestQueryLogs(t *testing.T) {
 		"QueryLocalStorageUsedAvg":                      func(s, e time.Time) { querier.QueryLocalStorageUsedAvg(s, e) },
 		"QueryLocalStorageUsedMax":                      func(s, e time.Time) { querier.QueryLocalStorageUsedMax(s, e) },
 		"QueryLocalStorageBytes":                        func(s, e time.Time) { querier.QueryLocalStorageBytes(s, e) },
+		"QueryKMLocalStorageUsedAvg":                    func(s, e time.Time) { querier.QueryKMLocalStorageUsedAvg(s, e) },
+		"QueryKMLocalStorageUsedMax":                    func(s, e time.Time) { querier.QueryKMLocalStorageUsedMax(s, e) },
+		"QueryKMLocalStorageBytes":                      func(s, e time.Time) { querier.QueryKMLocalStorageBytes(s, e) },
 		"QueryNodeActiveMinutes":                        func(s, e time.Time) { querier.QueryNodeActiveMinutes(s, e) },
 		"QueryNodeCPUCoresCapacity":                     func(s, e time.Time) { querier.QueryNodeCPUCoresCapacity(s, e) },
 		"QueryNodeCPUCoresAllocatable":                  func(s, e time.Time) { querier.QueryNodeCPUCoresAllocatable(s, e) },
@@ -141,6 +144,7 @@ func TestQueryLogs(t *testing.T) {
 		"QueryPodPVCAllocation":                         func(s, e time.Time) { querier.QueryPodPVCAllocation(s, e) },
 		"QueryPVCBytesRequested":                        func(s, e time.Time) { querier.QueryPVCBytesRequested(s, e) },
 		"QueryPVCInfo":                                  func(s, e time.Time) { querier.QueryPVCInfo(s, e) },
+		"QueryKMPVCInfo":                                func(s, e time.Time) { querier.QueryKMPVCInfo(s, e) },
 		"QueryPVBytes":                                  func(s, e time.Time) { querier.QueryPVBytes(s, e) },
 		"QueryPVPricePerGiBHour":                        func(s, e time.Time) { querier.QueryPVPricePerGiBHour(s, e) },
 		"QueryPVInfo":                                   func(s, e time.Time) { querier.QueryPVInfo(s, e) },

pkg/costmodel/costmodel.go

@@ -1377,15 +1377,7 @@ func (cm *CostModel) GetLBCost() (map[serviceKey]*costAnalyzerCloud.LoadBalancer
 				return nil, err
 			}
 			newLoadBalancer := *loadBalancer
-			for _, loadBalancerIngress := range service.Status.LoadBalancer.Ingress {
-				address := loadBalancerIngress.IP
-				// Some cloud providers use hostname rather than IP
-				if address == "" {
-					address = loadBalancerIngress.Hostname
-				}
-				newLoadBalancer.IngressIPAddresses = append(newLoadBalancer.IngressIPAddresses, address)
-
-			}
+			newLoadBalancer.IngressIPAddresses = clustercache.GetLoadBalancerIngressAddress(service)
 			loadBalancerMap[key] = &newLoadBalancer
 		}
 	}

pkg/kubemodel/kubemodel.go

@@ -197,14 +197,13 @@ func (km *KubeModel) computeNodes(kms *kubemodel.KubeModelSet, start, end time.T
 	nodeInfoResultFuture := source.WithGroup(grp, metrics.QueryNodeInfo(start, end))
 	nodeUptimeResultFuture := source.WithGroup(grp, metrics.QueryNodeUptime(start, end))
 	nodeLabelsResultFuture := source.WithGroup(grp, metrics.QueryNodeLabels(start, end))
-	// TODO make sure that UID is being populated correctly here
 	nodeIsSpotResultFuture := source.WithGroup(grp, metrics.QueryNodeIsSpot(start, end))
 	nodeResourceCapacitiesFuture := source.WithGroup(grp, metrics.QueryNodeResourceCapacities(start, end))
 	nodeResourcesAllocatableFuture := source.WithGroup(grp, metrics.QueryNodeResourcesAllocatable(start, end))
 
-	localStorageBytesFuture := source.WithGroup(grp, metrics.QueryLocalStorageBytes(start, end))
-	localStorageUsedAvgFuture := source.WithGroup(grp, metrics.QueryLocalStorageUsedAvg(start, end))
-	localStorageUsedMaxFuture := source.WithGroup(grp, metrics.QueryLocalStorageUsedMax(start, end))
+	localStorageBytesFuture := source.WithGroup(grp, metrics.QueryKMLocalStorageBytes(start, end))
+	localStorageUsedAvgFuture := source.WithGroup(grp, metrics.QueryKMLocalStorageUsedAvg(start, end))
+	localStorageUsedMaxFuture := source.WithGroup(grp, metrics.QueryKMLocalStorageUsedMax(start, end))
 
 	nodeMap := make(map[string]*kubemodel.Node)
 
@@ -279,24 +278,24 @@ func (km *KubeModel) computeNodes(kms *kubemodel.KubeModelSet, start, end time.T
 	localStorageBytesResult, _ := localStorageBytesFuture.Await()
 	for _, res := range localStorageBytesResult {
 		node, ok := nodeMap[res.UID]
-		if ok && len(res.Data) > 0 {
-			node.FileSystem.CapacityBytes = res.Data[0].Value
+		if ok {
+			node.FileSystem.CapacityBytes = res.Value
 		}
 	}
 
 	localStorageUsedAvgResult, _ := localStorageUsedAvgFuture.Await()
 	for _, res := range localStorageUsedAvgResult {
 		node, ok := nodeMap[res.UID]
-		if ok && len(res.Data) > 0 {
-			node.FileSystem.UsageByteAvg = res.Data[0].Value
+		if ok {
+			node.FileSystem.UsageByteAvg = res.Value
 		}
 	}
 
 	localStorageUsedMaxResult, _ := localStorageUsedMaxFuture.Await()
 	for _, res := range localStorageUsedMaxResult {
 		node, ok := nodeMap[res.UID]
-		if ok && len(res.Data) > 0 {
-			node.FileSystem.UsageByteMax = res.Data[0].Value
+		if ok {
+			node.FileSystem.UsageByteMax = res.Value
 		}
 	}
 
@@ -1353,10 +1352,11 @@ func (km *KubeModel) computeServices(kms *kubemodel.KubeModelSet, start, end tim
 	serviceInfoResult, _ := serviceInfoResultFuture.Await()
 	for _, res := range serviceInfoResult {
 		serviceMap[res.UID] = &kubemodel.Service{
-			UID:          res.UID,
-			NamespaceUID: res.NamespaceUID,
-			Name:         res.Service,
-			Type:         kubemodel.ParseServiceType(res.ServiceType),
+			UID:              res.UID,
+			NamespaceUID:     res.NamespaceUID,
+			Name:             res.Service,
+			Type:             kubemodel.ParseServiceType(res.ServiceType),
+			LBIngressAddress: res.LBIngressAddress,
 		}
 	}
 
@@ -1450,7 +1450,7 @@ func (km *KubeModel) computePersistentVolumeClaims(kms *kubemodel.KubeModelSet,
 	grp := source.NewQueryGroup()
 	metrics := km.ds.Metrics()
 
-	pvcInfoResultFuture := source.WithGroup(grp, metrics.QueryPVCInfo(start, end))
+	pvcInfoResultFuture := source.WithGroup(grp, metrics.QueryKMPVCInfo(start, end))
 	pvcUptimeResultFuture := source.WithGroup(grp, metrics.QueryPVCUptime(start, end))
 	pvcBytesRequestedResultFuture := source.WithGroup(grp, metrics.QueryPVCBytesRequested(start, end))