scraper_test.go

// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

package kubeletstatsreceiver

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"testing"

	"github.com/stretchr/testify/require"
	"go.opentelemetry.io/collector/receiver/receivertest"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/fake"

	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden"
	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/pmetrictest"
	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kubeletstatsreceiver/internal/kubelet"
	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kubeletstatsreceiver/internal/metadata"
)

const (
	dataLen = numContainers*containerMetrics + numPods*podMetrics + numNodes*nodeMetrics + numVolumes*volumeMetrics

	// Number of resources by type in testdata/stats-summary.json
	numContainers = 9
	numPods       = 9
	numNodes      = 1
	numVolumes    = 8

	// Number of metrics by resource
	nodeMetrics      = 15
	podMetrics       = 15
	containerMetrics = 11
	volumeMetrics    = 5
)

var allMetricGroups = map[kubelet.MetricGroup]bool{
	kubelet.ContainerMetricGroup: true,
	kubelet.PodMetricGroup:       true,
	kubelet.NodeMetricGroup:      true,
	kubelet.VolumeMetricGroup:    true,
}

func TestScraper(t *testing.T) {
	options := &scraperOptions{
		metricGroupsToCollect: allMetricGroups,
	}
	r, err := newKubletScraper(
		&fakeRestClient{},
		receivertest.NewNopCreateSettings(),
		options,
		metadata.DefaultMetricsBuilderConfig(),
	)
	require.NoError(t, err)

	md, err := r.Scrape(context.Background())
	require.NoError(t, err)
	require.Equal(t, dataLen, md.DataPointCount())
	expectedFile := filepath.Join("testdata", "scraper", "test_scraper_expected.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 TestScraperWithMetadata(t *testing.T) {
	tests := []struct {
		name           string
		metadataLabels []kubelet.MetadataLabel
		metricGroups   map[kubelet.MetricGroup]bool
		dataLen        int
		metricPrefix   string
		requiredLabel  string
	}{
		{
			name:           "Container_Metadata",
			metadataLabels: []kubelet.MetadataLabel{kubelet.MetadataLabelContainerID},
			metricGroups: map[kubelet.MetricGroup]bool{
				kubelet.ContainerMetricGroup: true,
			},
			dataLen:       numContainers * containerMetrics,
			metricPrefix:  "container.",
			requiredLabel: "container.id",
		},
		{
			name:           "Volume_Metadata",
			metadataLabels: []kubelet.MetadataLabel{kubelet.MetadataLabelVolumeType},
			metricGroups: map[kubelet.MetricGroup]bool{
				kubelet.VolumeMetricGroup: true,
			},
			dataLen:       numVolumes * volumeMetrics,
			metricPrefix:  "k8s.volume.",
			requiredLabel: "k8s.volume.type",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			options := &scraperOptions{
				extraMetadataLabels:   tt.metadataLabels,
				metricGroupsToCollect: tt.metricGroups,
			}
			r, err := newKubletScraper(
				&fakeRestClient{},
				receivertest.NewNopCreateSettings(),
				options,
				metadata.DefaultMetricsBuilderConfig(),
			)
			require.NoError(t, err)

			md, err := r.Scrape(context.Background())
			require.NoError(t, err)

			filename := "test_scraper_with_metadata_" + tt.name + "_expected.yaml"
			expectedFile := filepath.Join("testdata", "scraper", filename)

			// 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 TestScraperWithPercentMetrics(t *testing.T) {
	options := &scraperOptions{
		metricGroupsToCollect: map[kubelet.MetricGroup]bool{
			kubelet.ContainerMetricGroup: true,
			kubelet.PodMetricGroup:       true,
		},
	}
	metricsConfig := metadata.MetricsBuilderConfig{
		Metrics: metadata.MetricsConfig{
			ContainerCPUTime: metadata.MetricConfig{
				Enabled: false,
			},
			ContainerCPUUtilization: metadata.MetricConfig{
				Enabled: false,
			},
			ContainerFilesystemAvailable: metadata.MetricConfig{
				Enabled: false,
			},
			ContainerFilesystemCapacity: metadata.MetricConfig{
				Enabled: false,
			},
			ContainerFilesystemUsage: metadata.MetricConfig{
				Enabled: false,
			},
			ContainerMemoryAvailable: metadata.MetricConfig{
				Enabled: false,
			},
			ContainerMemoryMajorPageFaults: metadata.MetricConfig{
				Enabled: false,
			},
			ContainerMemoryPageFaults: metadata.MetricConfig{
				Enabled: false,
			},
			ContainerMemoryRss: metadata.MetricConfig{
				Enabled: false,
			},
			ContainerMemoryUsage: metadata.MetricConfig{
				Enabled: false,
			},
			K8sContainerCPULimitUtilization: metadata.MetricConfig{
				Enabled: true,
			},
			K8sContainerCPURequestUtilization: metadata.MetricConfig{
				Enabled: true,
			},
			K8sContainerMemoryLimitUtilization: metadata.MetricConfig{
				Enabled: true,
			},
			K8sContainerMemoryRequestUtilization: metadata.MetricConfig{
				Enabled: true,
			},
			ContainerMemoryWorkingSet: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeCPUTime: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeCPUUtilization: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeFilesystemAvailable: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeFilesystemCapacity: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeFilesystemUsage: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeMemoryAvailable: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeMemoryMajorPageFaults: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeMemoryPageFaults: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeMemoryRss: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeMemoryUsage: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeMemoryWorkingSet: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeNetworkErrors: metadata.MetricConfig{
				Enabled: false,
			},
			K8sNodeNetworkIo: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodCPUTime: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodCPUUtilization: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodFilesystemAvailable: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodFilesystemCapacity: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodFilesystemUsage: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodMemoryAvailable: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodMemoryMajorPageFaults: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodMemoryPageFaults: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodMemoryRss: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodMemoryUsage: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodCPULimitUtilization: metadata.MetricConfig{
				Enabled: true,
			},
			K8sPodCPURequestUtilization: metadata.MetricConfig{
				Enabled: true,
			},
			K8sPodMemoryLimitUtilization: metadata.MetricConfig{
				Enabled: true,
			},
			K8sPodMemoryRequestUtilization: metadata.MetricConfig{
				Enabled: true,
			},
			K8sPodMemoryWorkingSet: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodNetworkErrors: metadata.MetricConfig{
				Enabled: false,
			},
			K8sPodNetworkIo: metadata.MetricConfig{
				Enabled: false,
			},
			K8sVolumeAvailable: metadata.MetricConfig{
				Enabled: false,
			},
			K8sVolumeCapacity: metadata.MetricConfig{
				Enabled: false,
			},
			K8sVolumeInodes: metadata.MetricConfig{
				Enabled: false,
			},
			K8sVolumeInodesFree: metadata.MetricConfig{
				Enabled: false,
			},
			K8sVolumeInodesUsed: metadata.MetricConfig{
				Enabled: false,
			},
		},
		ResourceAttributes: metadata.DefaultResourceAttributesConfig(),
	}
	r, err := newKubletScraper(
		&fakeRestClient{},
		receivertest.NewNopCreateSettings(),
		options,
		metricsConfig,
	)
	require.NoError(t, err)

	md, err := r.Scrape(context.Background())
	require.NoError(t, err)

	expectedFile := filepath.Join("testdata", "scraper", "test_scraper_with_percent_expected.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.IgnoreMetricDataPointsOrder(),
		pmetrictest.IgnoreTimestamp(),
		pmetrictest.IgnoreMetricsOrder()))
}

func TestScraperWithMetricGroups(t *testing.T) {
	tests := []struct {
		name         string
		metricGroups map[kubelet.MetricGroup]bool
		dataLen      int
	}{
		{
			name:         "all_groups",
			metricGroups: allMetricGroups,
			dataLen:      dataLen,
		},
		{
			name: "only_container_group",
			metricGroups: map[kubelet.MetricGroup]bool{
				kubelet.ContainerMetricGroup: true,
			},
			dataLen: numContainers * containerMetrics,
		},
		{
			name: "only_pod_group",
			metricGroups: map[kubelet.MetricGroup]bool{
				kubelet.PodMetricGroup: true,
			},
			dataLen: numPods * podMetrics,
		},
		{
			name: "only_node_group",
			metricGroups: map[kubelet.MetricGroup]bool{
				kubelet.NodeMetricGroup: true,
			},
			dataLen: numNodes * nodeMetrics,
		},
		{
			name: "only_volume_group",
			metricGroups: map[kubelet.MetricGroup]bool{
				kubelet.VolumeMetricGroup: true,
			},
			dataLen: numVolumes * volumeMetrics,
		},
		{
			name: "pod_and_node_groups",
			metricGroups: map[kubelet.MetricGroup]bool{
				kubelet.PodMetricGroup:  true,
				kubelet.NodeMetricGroup: true,
			},
			dataLen: numNodes*nodeMetrics + numPods*podMetrics,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			r, err := newKubletScraper(
				&fakeRestClient{},
				receivertest.NewNopCreateSettings(),
				&scraperOptions{
					extraMetadataLabels:   []kubelet.MetadataLabel{kubelet.MetadataLabelContainerID},
					metricGroupsToCollect: test.metricGroups,
				},
				metadata.DefaultMetricsBuilderConfig(),
			)
			require.NoError(t, err)

			md, err := r.Scrape(context.Background())
			require.NoError(t, err)

			filename := "test_scraper_with_metric_groups_" + test.name + "_expected.yaml"
			expectedFile := filepath.Join("testdata", "scraper", filename)

			// 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()))
		})
	}
}

type expectedVolume struct {
	name   string
	typ    string
	labels map[string]string
}

func TestScraperWithPVCDetailedLabels(t *testing.T) {
	tests := []struct {
		name               string
		k8sAPIClient       kubernetes.Interface
		expectedVolumes    map[string]expectedVolume
		volumeClaimsToMiss map[string]bool
		dataLen            int
		numLogs            int
	}{
		{
			name:         "successful",
			k8sAPIClient: fake.NewSimpleClientset(getValidMockedObjects()...),
			expectedVolumes: map[string]expectedVolume{
				"volume_claim_1": {
					name: "storage-provisioner-token-qzlx6",
					typ:  "awsElasticBlockStore",
					labels: map[string]string{
						"aws.volume.id": "volume_id",
						"fs.type":       "fs_type",
						"partition":     "10",
					},
				},
				"volume_claim_2": {
					name: "kube-proxy",
					typ:  "gcePersistentDisk",
					labels: map[string]string{
						"gce.pd.name": "pd_name",
						"fs.type":     "fs_type",
						"partition":   "10",
					},
				},
				"volume_claim_3": {
					name: "coredns-token-dzc5t",
					typ:  "glusterfs",
					labels: map[string]string{
						"glusterfs.endpoints.name": "endpoints_name",
						"glusterfs.path":           "path",
					},
				},
			},
			dataLen: numVolumes,
		},
		{
			name:         "pvc_doesnot_exist",
			k8sAPIClient: fake.NewSimpleClientset(),
			dataLen:      numVolumes - 3,
			volumeClaimsToMiss: map[string]bool{
				"volume_claim_1": true,
				"volume_claim_2": true,
				"volume_claim_3": true,
			},
			numLogs: 3,
		},
		{
			name:         "empty_volume_name_in_pvc",
			k8sAPIClient: fake.NewSimpleClientset(getMockedObjectsWithEmptyVolumeName()...),
			expectedVolumes: map[string]expectedVolume{
				"volume_claim_1": {
					name: "storage-provisioner-token-qzlx6",
					typ:  "awsElasticBlockStore",
					labels: map[string]string{
						"aws.volume.id": "volume_id",
						"fs.type":       "fs_type",
						"partition":     "10",
					},
				},
				"volume_claim_2": {
					name: "kube-proxy",
					typ:  "gcePersistentDisk",
					labels: map[string]string{
						"gce.pd.name": "pd_name",
						"fs.type":     "fs_type",
						"partition":   "10",
					},
				},
			},
			// Two of mocked volumes are invalid and do not match any of volumes in
			// testdata/pods.json. Hence, don't expected to see metrics from them.
			dataLen: numVolumes - 1,
			volumeClaimsToMiss: map[string]bool{
				"volume_claim_3": true,
			},
			numLogs: 1,
		},
		{
			name:         "non_existent_volume_in_pvc",
			k8sAPIClient: fake.NewSimpleClientset(getMockedObjectsWithNonExistentVolumeName()...),
			expectedVolumes: map[string]expectedVolume{
				"volume_claim_1": {
					name: "storage-provisioner-token-qzlx6",
					typ:  "awsElasticBlockStore",
					labels: map[string]string{
						"aws.volume.id": "volume_id",
						"fs.type":       "fs_type",
						"partition":     "10",
					},
				},
				"volume_claim_2": {
					name: "kube-proxy",
					typ:  "gcePersistentDisk",
					labels: map[string]string{
						"gce.pd.name": "pd_name",
						"fs.type":     "fs_type",
						"partition":   "10",
					},
				},
			},
			// Two of mocked volumes are invalid and do not match any of volumes in
			// testdata/pods.json. Hence, don't expected to see metrics from them.
			dataLen: numVolumes - 1,
			volumeClaimsToMiss: map[string]bool{
				"volume_claim_3": true,
			},
			numLogs: 1,
		},
		{
			name:    "do_not_collect_detailed_labels",
			dataLen: numVolumes,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			r, err := newKubletScraper(
				&fakeRestClient{},
				receivertest.NewNopCreateSettings(),
				&scraperOptions{
					extraMetadataLabels: []kubelet.MetadataLabel{kubelet.MetadataLabelVolumeType},
					metricGroupsToCollect: map[kubelet.MetricGroup]bool{
						kubelet.VolumeMetricGroup: true,
					},
					k8sAPIClient: test.k8sAPIClient,
				},
				metadata.DefaultMetricsBuilderConfig(),
			)
			require.NoError(t, err)

			md, err := r.Scrape(context.Background())
			require.NoError(t, err)

			filename := "test_scraper_with_pvc_labels_" + test.name + "_expected.yaml"
			expectedFile := filepath.Join("testdata", "scraper", filename)

			// 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 TestClientErrors(t *testing.T) {
	tests := []struct {
		name                  string
		statsSummaryFail      bool
		podsFail              bool
		extraMetadataLabels   []kubelet.MetadataLabel
		metricGroupsToCollect map[kubelet.MetricGroup]bool
		numLogs               int
	}{
		{
			name:                  "no_errors_without_metadata",
			statsSummaryFail:      false,
			podsFail:              false,
			extraMetadataLabels:   nil,
			metricGroupsToCollect: allMetricGroups,
			numLogs:               0,
		},
		{
			name:                  "no_errors_with_metadata",
			statsSummaryFail:      false,
			podsFail:              false,
			extraMetadataLabels:   []kubelet.MetadataLabel{kubelet.MetadataLabelContainerID},
			metricGroupsToCollect: allMetricGroups,
			numLogs:               0,
		},
		{
			name:                  "stats_summary_endpoint_error",
			statsSummaryFail:      true,
			podsFail:              false,
			extraMetadataLabels:   nil,
			metricGroupsToCollect: allMetricGroups,
			numLogs:               1,
		},
		{
			name:                  "pods_endpoint_error",
			statsSummaryFail:      false,
			podsFail:              true,
			extraMetadataLabels:   []kubelet.MetadataLabel{kubelet.MetadataLabelContainerID},
			metricGroupsToCollect: allMetricGroups,
			numLogs:               1,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			core, observedLogs := observer.New(zap.ErrorLevel)
			logger := zap.New(core)
			settings := receivertest.NewNopCreateSettings()
			settings.Logger = logger
			options := &scraperOptions{
				extraMetadataLabels:   test.extraMetadataLabels,
				metricGroupsToCollect: test.metricGroupsToCollect,
			}
			r, err := newKubletScraper(
				&fakeRestClient{
					statsSummaryFail: test.statsSummaryFail,
					podsFail:         test.podsFail,
				},
				settings,
				options,
				metadata.DefaultMetricsBuilderConfig(),
			)
			require.NoError(t, err)

			_, err = r.Scrape(context.Background())
			if test.numLogs == 0 {
				require.NoError(t, err)
			} else {
				require.Error(t, err)
			}
			require.Equal(t, test.numLogs, observedLogs.Len())
		})
	}
}

var _ kubelet.RestClient = (*fakeRestClient)(nil)

type fakeRestClient struct {
	statsSummaryFail bool
	podsFail         bool
}

func (f *fakeRestClient) StatsSummary() ([]byte, error) {
	if f.statsSummaryFail {
		return nil, errors.New("")
	}
	return os.ReadFile("testdata/stats-summary.json")
}

func (f *fakeRestClient) Pods() ([]byte, error) {
	if f.podsFail {
		return nil, errors.New("")
	}
	return os.ReadFile("testdata/pods.json")
}