[receiver/kubeletstats] Collect network metrics from all interfaces (#38737)

#### Description
Revives #34287 by picking it's changes.
Also changed changed approach from feature gate to configuration
parameters (disabled by default).
In context of network metrics - default behavior of component hasn't
changed, so I consider it as a non-breaking change.

#### Link to tracking issue
Fixes #30196 

#### Testing
Unit test were updated to cover new functionality

#### Documentation
Documentation updated to cover new opt-in functionality

Original authors are included in the commit:
Co-authored-by: ChrsMark <[email protected]>
Co-authored-by: jinja2 <[email protected]>

---------

Co-authored-by: Christos Markou <[email protected]>
Co-authored-by: Jina Jain <[email protected]>

Author: 3 people

.chloggen/feat_kubelet-network-all-inerfaces.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: kubeletstatsreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adds support for collecting Node and Pod network IO/error metrics for all network interfaces
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [30196]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/kubeletstatsreceiver/README.md

@@ -32,7 +32,7 @@ to connect and authenticate to the API server and how often to collect data
 and send it to the next consumer.
 
 Kubelet Stats Receiver supports both secure Kubelet endpoint exposed at port 10250 by default and read-only
-Kubelet endpoint exposed at port 10255. If `auth_type` set to `none`, the read-only endpoint will be used. The secure 
+Kubelet endpoint exposed at port 10255. If `auth_type` set to `none`, the read-only endpoint will be used. The secure
 endpoint will be used if `auth_type` set to any of the following values:
 
 - `tls` tells the receiver to use TLS for auth and requires that the fields
@@ -109,14 +109,14 @@ node's network namespace.
 
 #### Custom CA
 
-The service account client, by default, uses the CA certificate located at 
-`/var/run/secrets/kubernetes.io/serviceaccount/ca.crt` to validate the kubelet certificate. 
-If the kubelet server uses a certificate issued by a different CA, 
+The service account client, by default, uses the CA certificate located at
+`/var/run/secrets/kubernetes.io/serviceaccount/ca.crt` to validate the kubelet certificate.
+If the kubelet server uses a certificate issued by a different CA,
 specify the custom CA certificate path using the `ca_file` option.
 
 ##### AKS Custom CA example
 
-This use case applies to AKS cluster, where the kubelet certificate is issued by 
+This use case applies to AKS cluster, where the kubelet certificate is issued by
 `/etc/kubernetes/certs/kubeletserver.crt`
 
 ```yaml
@@ -169,6 +169,7 @@ service:
       receivers: [kubeletstats]
       exporters: [file]
 ```
+
 Note that using `auth_type` `kubeConfig`, the endpoint should only be the node name as the communication to the kubelet is proxied by the API server configured in the `kubeConfig`.
 `insecure_skip_verify` still applies by overriding the `kubeConfig` settings.
 If no `context` is specified, the current context or the default context is used.
@@ -244,6 +245,23 @@ receivers:
       - pod
 ```
 
+### Network metrics from all interfaces for Node and Pod
+
+By default, `k8s.[node|pod].network.*` metrics are collected only for the default network interface (e.g. `eth0`). To enable network IO/error metrics collection from all available interfaces on Node/Pod level - you can use `collect_all_network_interfaces` configuration parameters. Please be aware that enabling this options will increase the amount of produced network metrics and increase network metrics cardinality, because of `interface` attribute.
+For example, if you would like to have network IO/error metrics from all network interfaces for both Pod and Node level you can use the following configuration.
+
+```yaml
+receivers:
+  kubeletstats:
+    collection_interval: 10s
+    auth_type: "serviceAccount"
+    endpoint: "${env:K8S_NODE_NAME}:10250"
+    insecure_skip_verify: true
+    collect_all_network_interfaces:
+      pod: true
+      node: true
+```
+
 ### Collect `k8s.{container,pod}.{cpu,memory}.node.utilization` as ratio of total node's capacity
 
 In order to calculate the `k8s.container.cpu.node.utilization`, `k8s.pod.cpu.node.utilization`,
@@ -259,6 +277,7 @@ env:
       fieldRef:
         fieldPath: spec.nodeName
 ```
+
 Then set `node` value to `${env:K8S_NODE_NAME}` in the receiver's configuration:
 
 ```yaml
@@ -304,7 +323,7 @@ rules:
   - apiGroups: [""]
     resources: ["nodes/stats"]
     verbs: ["get"]
-    
+
   # Only needed if you are using extra_metadata_labels or
   # are collecting the request/limit utilization metrics
   - apiGroups: [""]
@@ -333,4 +352,4 @@ You can enable the usage of the deprecated metrics by disabling the `receiver.ku
 - removed three releases after stable.
 
 More information about the deprecation plan and
-the background reasoning can be found at https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/27885.
+the background reasoning can be found at <https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/27885>.

receiver/kubeletstatsreceiver/config.go

@@ -55,6 +55,15 @@ type Config struct {
 
 	// MetricsBuilderConfig allows customizing scraped metrics/attributes representation.
 	metadata.MetricsBuilderConfig `mapstructure:",squash"`
+
+	// NetworkCollectAllInterfaces allows to enable collecting metrics from all network interfaces instead of default one
+	// Can be set separately for Pod and Node network metrics
+	NetworkCollectAllInterfaces NetworkInterfacesEnablerConfig `mapstructure:"collect_all_network_interfaces"`
+}
+
+type NetworkInterfacesEnablerConfig struct {
+	PodMetrics  bool `mapstructure:"pod"`
+	NodeMetrics bool `mapstructure:"node"`
 }
 
 // getReceiverOptions returns scraperOptions is the config is valid,
@@ -70,6 +79,11 @@ func (cfg *Config) getReceiverOptions() (*scraperOptions, error) {
 		return nil, err
 	}
 
+	ifaces := map[kubelet.MetricGroup]bool{
+		kubelet.NodeMetricGroup: cfg.NetworkCollectAllInterfaces.NodeMetrics,
+		kubelet.PodMetricGroup:  cfg.NetworkCollectAllInterfaces.PodMetrics,
+	}
+
 	var k8sAPIClient kubernetes.Interface
 	if cfg.K8sAPIConfig != nil {
 		k8sAPIClient, err = k8sconfig.MakeClient(*cfg.K8sAPIConfig)
@@ -82,6 +96,7 @@ func (cfg *Config) getReceiverOptions() (*scraperOptions, error) {
 		collectionInterval:    cfg.CollectionInterval,
 		extraMetadataLabels:   cfg.ExtraMetadataLabels,
 		metricGroupsToCollect: mgs,
+		allNetworkInterfaces:  ifaces,
 		k8sAPIClient:          k8sAPIClient,
 	}, nil
 }

receiver/kubeletstatsreceiver/config_test.go

@@ -339,6 +339,10 @@ func TestGetReceiverOptions(t *testing.T) {
 					kubelet.NodeMetricGroup: true,
 					kubelet.PodMetricGroup:  true,
 				},
+				allNetworkInterfaces: map[kubelet.MetricGroup]bool{
+					kubelet.NodeMetricGroup: false,
+					kubelet.PodMetricGroup:  false,
+				},
 				collectionInterval: 10 * time.Second,
 			},
 		},

receiver/kubeletstatsreceiver/internal/kubelet/accumulator.go

@@ -38,6 +38,7 @@ type metricDataAccumulator struct {
 	metadata              Metadata
 	logger                *zap.Logger
 	metricGroupsToCollect map[MetricGroup]bool
+	allNetworkInterfaces  map[MetricGroup]bool
 	time                  time.Time
 	mbs                   *metadata.MetricsBuilders
 }
@@ -59,7 +60,7 @@ func (a *metricDataAccumulator) nodeStats(s stats.NodeStats) {
 	addCPUMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeCPUMetrics, s.CPU, currentTime, resources{}, 0)
 	addMemoryMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeMemoryMetrics, s.Memory, currentTime, resources{}, 0)
 	addFilesystemMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeFilesystemMetrics, s.Fs, currentTime)
-	addNetworkMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeNetworkMetrics, s.Network, currentTime)
+	addNetworkMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeNetworkMetrics, s.Network, currentTime, a.allNetworkInterfaces[NodeMetricGroup])
 	// todo s.Runtime.ImageFs
 	rb := a.mbs.NodeMetricsBuilder.NewResourceBuilder()
 	rb.SetK8sNodeName(s.NodeName)
@@ -79,7 +80,7 @@ func (a *metricDataAccumulator) podStats(s stats.PodStats) {
 	addCPUMetrics(a.mbs.PodMetricsBuilder, metadata.PodCPUMetrics, s.CPU, currentTime, a.metadata.podResources[s.PodRef.UID], a.metadata.nodeInfo.CPUCapacity)
 	addMemoryMetrics(a.mbs.PodMetricsBuilder, metadata.PodMemoryMetrics, s.Memory, currentTime, a.metadata.podResources[s.PodRef.UID], a.metadata.nodeInfo.MemoryCapacity)
 	addFilesystemMetrics(a.mbs.PodMetricsBuilder, metadata.PodFilesystemMetrics, s.EphemeralStorage, currentTime)
-	addNetworkMetrics(a.mbs.PodMetricsBuilder, metadata.PodNetworkMetrics, s.Network, currentTime)
+	addNetworkMetrics(a.mbs.PodMetricsBuilder, metadata.PodNetworkMetrics, s.Network, currentTime, a.allNetworkInterfaces[PodMetricGroup])
 
 	rb := a.mbs.PodMetricsBuilder.NewResourceBuilder()
 	rb.SetK8sPodUID(s.PodRef.UID)

receiver/kubeletstatsreceiver/internal/kubelet/metrics.go

@@ -17,12 +17,14 @@ func MetricsData(
 	logger *zap.Logger, summary *stats.Summary,
 	metadata Metadata,
 	metricGroupsToCollect map[MetricGroup]bool,
+	allNetworkInterfaces map[MetricGroup]bool,
 	mbs *metadata.MetricsBuilders,
 ) []pmetric.Metrics {
 	acc := &metricDataAccumulator{
 		metadata:              metadata,
 		logger:                logger,
 		metricGroupsToCollect: metricGroupsToCollect,
+		allNetworkInterfaces:  allNetworkInterfaces,
 		time:                  time.Now(),
 		mbs:                   mbs,
 	}

receiver/kubeletstatsreceiver/internal/kubelet/metrics_test.go

@@ -39,12 +39,16 @@ func TestMetricAccumulator(t *testing.T) {
 		ContainerMetricsBuilder: metadata.NewMetricsBuilder(metadata.DefaultMetricsBuilderConfig(), receivertest.NewNopSettings(metadata.Type)),
 		OtherMetricsBuilder:     metadata.NewMetricsBuilder(metadata.DefaultMetricsBuilderConfig(), receivertest.NewNopSettings(metadata.Type)),
 	}
-	requireMetricsOk(t, MetricsData(zap.NewNop(), summary, k8sMetadata, ValidMetricGroups, mbs))
+	ifaces := map[MetricGroup]bool{
+		NodeMetricGroup: true,
+		PodMetricGroup:  true,
+	}
+	requireMetricsOk(t, MetricsData(zap.NewNop(), summary, k8sMetadata, ValidMetricGroups, ifaces, mbs))
 	// Disable all groups
 	mbs.NodeMetricsBuilder.Reset()
 	mbs.PodMetricsBuilder.Reset()
 	mbs.OtherMetricsBuilder.Reset()
-	require.Empty(t, MetricsData(zap.NewNop(), summary, k8sMetadata, map[MetricGroup]bool{}, mbs))
+	require.Empty(t, MetricsData(zap.NewNop(), summary, k8sMetadata, map[MetricGroup]bool{}, map[MetricGroup]bool{}, mbs))
 }
 
 func requireMetricsOk(t *testing.T, mds []pmetric.Metrics) {
@@ -142,6 +146,10 @@ func TestUptime(t *testing.T) {
 		PodMetricGroup:       true,
 		NodeMetricGroup:      true,
 	}
+	ifaces := map[MetricGroup]bool{
+		NodeMetricGroup: true,
+		PodMetricGroup:  true,
+	}
 
 	cfg := metadata.DefaultMetricsBuilderConfig()
 	cfg.Metrics.K8sNodeUptime.Enabled = true
@@ -154,7 +162,7 @@ func TestUptime(t *testing.T) {
 		ContainerMetricsBuilder: metadata.NewMetricsBuilder(cfg, receivertest.NewNopSettings(metadata.Type)),
 	}
 
-	metrics := indexedFakeMetrics(MetricsData(zap.NewNop(), summary, Metadata{}, mgs, mbs))
+	metrics := indexedFakeMetrics(MetricsData(zap.NewNop(), summary, Metadata{}, mgs, ifaces, mbs))
 
 	requireContains(t, metrics, "k8s.node.uptime")
 	requireContains(t, metrics, "k8s.pod.uptime")
@@ -214,11 +222,15 @@ func fakeMetrics() []pmetric.Metrics {
 		PodMetricGroup:       true,
 		NodeMetricGroup:      true,
 	}
+	ifaces := map[MetricGroup]bool{
+		NodeMetricGroup: true,
+		PodMetricGroup:  true,
+	}
 	mbs := &metadata.MetricsBuilders{
 		NodeMetricsBuilder:      metadata.NewMetricsBuilder(metadata.DefaultMetricsBuilderConfig(), receivertest.NewNopSettings(metadata.Type)),
 		PodMetricsBuilder:       metadata.NewMetricsBuilder(metadata.DefaultMetricsBuilderConfig(), receivertest.NewNopSettings(metadata.Type)),
 		ContainerMetricsBuilder: metadata.NewMetricsBuilder(metadata.DefaultMetricsBuilderConfig(), receivertest.NewNopSettings(metadata.Type)),
 		OtherMetricsBuilder:     metadata.NewMetricsBuilder(metadata.DefaultMetricsBuilderConfig(), receivertest.NewNopSettings(metadata.Type)),
 	}
-	return MetricsData(zap.NewNop(), summary, Metadata{}, mgs, mbs)
+	return MetricsData(zap.NewNop(), summary, Metadata{}, mgs, ifaces, mbs)
 }

receiver/kubeletstatsreceiver/internal/kubelet/network.go

@@ -12,11 +12,23 @@ import (
 
 type getNetworkDataFunc func(s *stats.NetworkStats) (rx *uint64, tx *uint64)
 
-func addNetworkMetrics(mb *metadata.MetricsBuilder, networkMetrics metadata.NetworkMetrics, s *stats.NetworkStats, currentTime pcommon.Timestamp) {
+type getInterfaceDataFunc func(s *stats.InterfaceStats) (rx *uint64, tx *uint64)
+
+func addNetworkMetrics(mb *metadata.MetricsBuilder, networkMetrics metadata.NetworkMetrics, s *stats.NetworkStats, currentTime pcommon.Timestamp, allInterfaces bool) {
 	if s == nil {
 		return
 	}
 
+	if allInterfaces {
+		for i := range s.Interfaces {
+			recordInterfaceDataPoint(mb, networkMetrics.IO, &s.Interfaces[i], getInterfaceIO, currentTime)
+			recordInterfaceDataPoint(mb, networkMetrics.Errors, &s.Interfaces[i], getInterfaceErrors, currentTime)
+		}
+		// Because stats.NetworkStats.Interfaces contains metrics for all interfaces, including default,
+		// we don't need to iterate over stats.NetworkStats.InterfaceStats for it, hence we return here
+		return
+	}
+
 	recordNetworkDataPoint(mb, networkMetrics.IO, s, getNetworkIO, currentTime)
 	recordNetworkDataPoint(mb, networkMetrics.Errors, s, getNetworkErrors, currentTime)
 }
@@ -40,3 +52,23 @@ func getNetworkIO(s *stats.NetworkStats) (*uint64, *uint64) {
 func getNetworkErrors(s *stats.NetworkStats) (*uint64, *uint64) {
 	return s.RxErrors, s.TxErrors
 }
+
+func recordInterfaceDataPoint(mb *metadata.MetricsBuilder, recordDataPoint metadata.RecordIntDataPointWithDirectionFunc, s *stats.InterfaceStats, getData getInterfaceDataFunc, currentTime pcommon.Timestamp) {
+	rx, tx := getData(s)
+
+	if rx != nil {
+		recordDataPoint(mb, currentTime, int64(*rx), s.Name, metadata.AttributeDirectionReceive)
+	}
+
+	if tx != nil {
+		recordDataPoint(mb, currentTime, int64(*tx), s.Name, metadata.AttributeDirectionTransmit)
+	}
+}
+
+func getInterfaceIO(s *stats.InterfaceStats) (*uint64, *uint64) {
+	return s.RxBytes, s.TxBytes
+}
+
+func getInterfaceErrors(s *stats.InterfaceStats) (*uint64, *uint64) {
+	return s.RxErrors, s.TxErrors
+}

receiver/kubeletstatsreceiver/scraper.go

@@ -30,6 +30,7 @@ type scraperOptions struct {
 	collectionInterval    time.Duration
 	extraMetadataLabels   []kubelet.MetadataLabel
 	metricGroupsToCollect map[kubelet.MetricGroup]bool
+	allNetworkInterfaces  map[kubelet.MetricGroup]bool
 	k8sAPIClient          kubernetes.Interface
 }
 
@@ -39,6 +40,7 @@ type kubeletScraper struct {
 	logger                *zap.Logger
 	extraMetadataLabels   []kubelet.MetadataLabel
 	metricGroupsToCollect map[kubelet.MetricGroup]bool
+	allNetworkInterfaces  map[kubelet.MetricGroup]bool
 	k8sAPIClient          kubernetes.Interface
 	cachedVolumeSource    map[string]v1.PersistentVolumeSource
 	mbs                   *metadata.MetricsBuilders
@@ -80,6 +82,7 @@ func newKubeletScraper(
 		logger:                set.Logger,
 		extraMetadataLabels:   rOptions.extraMetadataLabels,
 		metricGroupsToCollect: rOptions.metricGroupsToCollect,
+		allNetworkInterfaces:  rOptions.allNetworkInterfaces,
 		k8sAPIClient:          rOptions.k8sAPIClient,
 		cachedVolumeSource:    make(map[string]v1.PersistentVolumeSource),
 		mbs: &metadata.MetricsBuilders{
@@ -138,7 +141,7 @@ func (r *kubeletScraper) scrape(context.Context) (pmetric.Metrics, error) {
 
 	metaD := kubelet.NewMetadata(r.extraMetadataLabels, podsMetadata, nodeInfo, r.detailedPVCLabelsSetter())
 
-	mds := kubelet.MetricsData(r.logger, summary, metaD, r.metricGroupsToCollect, r.mbs)
+	mds := kubelet.MetricsData(r.logger, summary, metaD, r.metricGroupsToCollect, r.allNetworkInterfaces, r.mbs)
 	md := pmetric.NewMetrics()
 	for i := range mds {
 		mds[i].ResourceMetrics().MoveAndAppendTo(md.ResourceMetrics())

receiver/kubeletstatsreceiver/scraper_test.go

@@ -82,6 +82,42 @@ func TestScraper(t *testing.T) {
 		pmetrictest.IgnoreMetricsOrder()))
 }
 
+func TestScraperWithInterfacesMetrics(t *testing.T) {
+	options := &scraperOptions{
+		metricGroupsToCollect: allMetricGroups,
+		allNetworkInterfaces: map[kubelet.MetricGroup]bool{
+			kubelet.NodeMetricGroup: true,
+			kubelet.PodMetricGroup:  true,
+		},
+	}
+	r, err := newKubeletScraper(
+		&fakeRestClient{},
+		receivertest.NewNopSettings(metadata.Type),
+		options,
+		metadata.DefaultMetricsBuilderConfig(),
+		"worker-42",
+	)
+	require.NoError(t, err)
+
+	md, err := r.ScrapeMetrics(context.Background())
+	require.NoError(t, err)
+
+	require.Equal(t, dataLen+numPods*4+numNodes*4, md.DataPointCount())
+	expectedFile := filepath.Join("testdata", "scraper", "test_scraper_with_interfaces_metrics.yaml")
+
+	// Uncomment to regenerate '*_expected.yaml' files
+	// golden.WriteMetrics(t, expectedFile, md)
+
+	expectedMetrics, err := golden.ReadMetrics(expectedFile)
+	require.NoError(t, err)
+	require.NoError(t, pmetrictest.CompareMetrics(expectedMetrics, md,
+		pmetrictest.IgnoreStartTimestamp(),
+		pmetrictest.IgnoreResourceMetricsOrder(),
+		pmetrictest.IgnoreMetricDataPointsOrder(),
+		pmetrictest.IgnoreTimestamp(),
+		pmetrictest.IgnoreMetricsOrder()))
+}
+
 func TestScraperWithCPUNodeUtilization(t *testing.T) {
 	watcherStarted := make(chan struct{})
 	// Create the fake client.

receiver/kubeletstatsreceiver/testdata/config.yaml

@@ -52,3 +52,9 @@ kubeletstats/pod_memory_node_utilization:
   metrics:
     k8s.pod.memory.node.utilization:
       enabled: true
+kubeletstats/all_network_interfaces_metrics:
+  collection_interval: 10s
+  metric_groups: [ container, pod, node ]
+  collect_all_network_interfaces:
+    pod: true
+    node: true