Gather Kubernetes data about pod volumes and add K8s context to filesystem metrics (#613)

* Gather Kubernetes data about pod volumes and add K8s context to filesystem metrics

* Take into account block volumes as well

* Filter pod volume metrics to only include PVC-backed volumes and exclude kube-system

* Add RBAC permissions for PVCs and PVs in controller ClusterRole

* Deduplicate filesystem metrics

* Simplify FilesystemMetric and add in-tree volume source support

* Add k8s_pod_volume_metrics to E2E test

Author: ciprianfocsaneanu

api/v1/kube/kube_api.pb.go

@@ -12,6 +12,7 @@ service KubeAPI {
   rpc GetPod(GetPodRequest) returns (GetPodResponse);
   rpc GetNode(GetNodeRequest) returns (GetNodeResponse);
   rpc GetNodeStatsSummary(GetNodeStatsSummaryRequest) returns (GetNodeStatsSummaryResponse);
+  rpc GetPodVolumes(GetPodVolumesRequest) returns (GetPodVolumesResponse);
 }
 
 message GetClusterInfoRequest {}
@@ -152,3 +153,31 @@ message RuntimeStats {
   FsStats image_fs = 2;
   FsStats container_fs = 3;
 }
+
+message GetPodVolumesRequest {
+  string node_name = 1;
+}
+
+message GetPodVolumesResponse {
+  repeated PodVolumeInfo volumes = 1;
+}
+
+message PodVolumeInfo {
+  string namespace = 1;
+  string pod_name = 2;
+  string pod_uid = 3;
+  string controller_kind = 4;
+  string controller_name = 5;
+  string container_name = 6;
+  string volume_name = 7;
+  string mount_path = 8;
+  string pvc_name = 9;
+  string pvc_uid = 10;
+  int64 requested_size_bytes = 11;
+  string pv_name = 12;
+  string storage_class = 13;
+  string csi_driver = 14;
+  string csi_volume_handle = 15;
+  string volume_mode = 16;    // "Filesystem" or "Block"
+  string device_path = 17;    // For block volumes: container's volumeDevices[].devicePath
+}

api/v1/kube/kube_api.proto

@@ -257,6 +257,8 @@ rules:
       - namespaces
       - services
       - endpoints
+      - persistentvolumeclaims
+      - persistentvolumes
     verbs:
       - get
       - list

api/v1/kube/kube_api_grpc.pb.go

@@ -255,6 +255,7 @@ func (a *App) Run(ctx context.Context) error {
 	var blockDeviceMetricsWriter pipeline.BlockDeviceMetricsWriter
 	var filesystemMetricsWriter pipeline.FilesystemMetricsWriter
 	var nodeStatsSummaryWriter pipeline.NodeStatsSummaryWriter
+	var podVolumeMetricsWriter pipeline.K8sPodVolumeMetricsWriter
 	var storageInfoProvider pipeline.StorageInfoProvider
 	if cfg.Stats.StorageEnabled {
 		metricsClient, err := createMetricsClient(cfg)
@@ -263,12 +264,12 @@ func (a *App) Run(ctx context.Context) error {
 		}
 
 		go func() {
-			if err = metricsClient.Start(ctx); err != nil {
-				log.Warnf("metric client failed with:%v", err)
+			if err := metricsClient.Start(ctx); err != nil {
+				log.Warnf("metrics client failed: %v", err)
 			}
 		}()
 
-		blockDeviceMetricsWriter, filesystemMetricsWriter, nodeStatsSummaryWriter, err = setupStorageMetrics(metricsClient)
+		blockDeviceMetricsWriter, filesystemMetricsWriter, nodeStatsSummaryWriter, podVolumeMetricsWriter, err = setupStorageMetrics(metricsClient)
 		if err != nil {
 			return fmt.Errorf("failed to setup storage metrics: %w", err)
 		}
@@ -301,6 +302,7 @@ func (a *App) Run(ctx context.Context) error {
 		filesystemMetricsWriter,
 		storageInfoProvider,
 		nodeStatsSummaryWriter,
+		podVolumeMetricsWriter,
 	)
 
 	for _, namespace := range cfg.MutedNamespaces {
@@ -569,23 +571,28 @@ func waitWithTimeout(errg *errgroup.Group, timeout time.Duration) error {
 	}
 }
 
-func setupStorageMetrics(metricsClient custommetrics.MetricClient) (pipeline.BlockDeviceMetricsWriter, pipeline.FilesystemMetricsWriter, pipeline.NodeStatsSummaryWriter, error) {
+func setupStorageMetrics(metricsClient custommetrics.MetricClient) (pipeline.BlockDeviceMetricsWriter, pipeline.FilesystemMetricsWriter, pipeline.NodeStatsSummaryWriter, pipeline.K8sPodVolumeMetricsWriter, error) {
 	blockDeviceMetrics, err := pipeline.NewBlockDeviceMetricsWriter(metricsClient)
 	if err != nil {
-		return nil, nil, nil, fmt.Errorf("failed to create block device metrics writer: %w", err)
+		return nil, nil, nil, nil, fmt.Errorf("failed to create block device metrics writer: %w", err)
 	}
 
 	filesystemMetrics, err := pipeline.NewFilesystemMetricsWriter(metricsClient)
 	if err != nil {
-		return nil, nil, nil, fmt.Errorf("failed to create filesystem metrics writer: %w", err)
+		return nil, nil, nil, nil, fmt.Errorf("failed to create filesystem metrics writer: %w", err)
 	}
 
 	nodeStatsSummaryWriter, err := pipeline.NewNodeStatsSummaryWriter(metricsClient)
 	if err != nil {
-		return nil, nil, nil, fmt.Errorf("failed to create node storage stats summary writer: %w", err)
+		return nil, nil, nil, nil, fmt.Errorf("failed to create node storage stats summary writer: %w", err)
 	}
 
-	return blockDeviceMetrics, filesystemMetrics, nodeStatsSummaryWriter, nil
+	podVolumeMetricsWriter, err := pipeline.NewK8sPodVolumeMetricsWriter(metricsClient)
+	if err != nil {
+		return nil, nil, nil, nil, fmt.Errorf("failed to create pod volume metrics writer: %w", err)
+	}
+
+	return blockDeviceMetrics, filesystemMetrics, nodeStatsSummaryWriter, podVolumeMetricsWriter, nil
 }
 
 // resolveMetricsAddr transforms kvisor.* addresses to telemetry.* addresses