test(otel): check node and pod labels, the attribute schema and KSM zones against the cluster

test/otel/efa/attribute_schema_test.go

@@ -0,0 +1,54 @@
+//go:build integration
+
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package efa
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics"
+)
+
+// ---------------------------------------------------------------------------
+// TestAttributeSchema — the closed set of resource and instrumentation-scope
+// attributes on EFA series. Every series must have every required key and
+// nothing undeclared, expected keys must be on at least one series, and keys set
+// together must arrive together. Node and pod labels are open by construction
+// (k8sattributes extracts "(.*)"). Datapoint scope is not covered.
+// ---------------------------------------------------------------------------
+
+// awsdevicepodcorrelation then efa_promote; only on devices a pod holds.
+var efaPodKeys = []string{"k8s.pod.name", "k8s.namespace.name", "k8s.container.name"}
+
+// awsefareceiver sets no resource attributes, so unlike the scraped pipelines
+// there are no server.* or service.* keys. This suite runs no workload that holds
+// a device, so the pod keys are optional rather than expected.
+var efaResourceSchema = otelmetrics.KeySchema{
+	Required: otelmetrics.ConcatKeys(otelmetrics.CloudKeys, otelmetrics.HostKeys, []string{
+		"k8s.cluster.name", // set_cluster_name; protected
+		"k8s.node.name",    // set_node_name, promote_node_name; protected
+		"k8s.node.uid",     // k8sattributes/cw_k8s_ci_v0_node
+	}),
+	Optional:     otelmetrics.ConcatKeys(efaPodKeys, []string{"k8s.pod.uid"}, otelmetrics.OwnerKindKeys, otelmetrics.WorkloadKeys),
+	OpenPrefixes: []string{"k8s.node.label.", "k8s.pod.label."},
+	Implies:      otelmetrics.MergeImplies(otelmetrics.WorkloadImplies(), otelmetrics.Together(efaPodKeys...)),
+}
+
+func TestAttributeSchema(t *testing.T) {
+	t.Parallel()
+	tally, err := otelmetrics.TallySchema(context.Background(), queryCache, efaMetricNamesList, otelmetrics.GroupSchema{
+		Resource:        efaResourceSchema,
+		Instrumentation: otelmetrics.KeySchema{Required: otelmetrics.ScopeKeys}, // set_scope_efa
+	}, nil)
+	require.NoError(t, err, "no EFA nodes?")
+	t.Logf("%d series across %d metrics", tally.Series, len(efaMetricNamesList))
+	require.NotZero(t, tally.Series, "no series checked")
+	for _, p := range tally.Problems() {
+		t.Errorf("%s: change the schema if intended", p)
+	}
+}

test/otel/gpu/attribute_schema_test.go

@@ -0,0 +1,54 @@
+//go:build integration
+
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package gpu
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics"
+)
+
+// ---------------------------------------------------------------------------
+// TestAttributeSchema — the closed set of resource and instrumentation-scope
+// attributes on DCGM series. Every series must have every required key and
+// nothing undeclared, expected keys must be on at least one series, and keys set
+// together must arrive together. Node and pod labels are open by construction
+// (k8sattributes extracts "(.*)"). Datapoint scope is not covered.
+// ---------------------------------------------------------------------------
+
+// dcgm_promote; dcgm-exporter maps a GPU to a pod only while the pod holds it.
+var dcgmPodKeys = []string{"k8s.pod.name", "k8s.namespace.name", "k8s.container.name"}
+
+var dcgmResourceSchema = otelmetrics.KeySchema{
+	Required: otelmetrics.ConcatKeys(otelmetrics.PrometheusReceiverKeys, otelmetrics.CloudKeys, otelmetrics.HostKeys, []string{
+		"k8s.cluster.name", // set_cluster_name; protected
+		"k8s.node.name",    // set_node_name, promote_node_name; protected
+		"k8s.node.uid",     // k8sattributes/cw_k8s_ci_v0_node
+	}),
+	// The multi-gpu-burn Deployment holds GPUs; TestDCGMPodName requires it.
+	Expected: otelmetrics.ConcatKeys(dcgmPodKeys, []string{"k8s.deployment.name", "k8s.replicaset.name"}, otelmetrics.WorkloadKeys),
+	Optional: otelmetrics.ConcatKeys([]string{"k8s.pod.uid"}, // k8sattributes/cw_k8s_ci_v0_pod
+		otelmetrics.Without(otelmetrics.OwnerKindKeys, "k8s.deployment.name", "k8s.replicaset.name")),
+	OpenPrefixes: []string{"k8s.node.label.", "k8s.pod.label."},
+	Implies:      otelmetrics.MergeImplies(otelmetrics.WorkloadImplies(), otelmetrics.Together(dcgmPodKeys...)),
+}
+
+func TestAttributeSchema(t *testing.T) {
+	t.Parallel()
+	tally, err := otelmetrics.TallySchema(context.Background(), queryCache, dcgmMetricNamesList, otelmetrics.GroupSchema{
+		Resource:        dcgmResourceSchema,
+		Instrumentation: otelmetrics.KeySchema{Required: otelmetrics.ScopeKeys}, // set_scope_dcgm
+	}, nil)
+	require.NoError(t, err, "no GPU nodes?")
+	t.Logf("%d series across %d metrics", tally.Series, len(dcgmMetricNamesList))
+	require.NotZero(t, tally.Series, "no series checked")
+	for _, p := range tally.Problems() {
+		t.Errorf("%s: change the schema if intended", p)
+	}
+}

test/otel/neuron/attribute_schema_test.go

@@ -0,0 +1,57 @@
+//go:build integration
+
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package neuron
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics"
+)
+
+// ---------------------------------------------------------------------------
+// TestAttributeSchema — the closed set of resource and instrumentation-scope
+// attributes on Neuron series. Every series must have every required key and
+// nothing undeclared, expected keys must be on at least one series, and keys set
+// together must arrive together. Node and pod labels are open by construction
+// (k8sattributes extracts "(.*)"). Datapoint scope is not covered.
+// ---------------------------------------------------------------------------
+
+// awsdevicepodcorrelation then groupbyattrs; only on cores or devices a pod holds.
+var neuronPodKeys = []string{"k8s.pod.name", "k8s.namespace.name", "k8s.container.name"}
+
+var neuronResourceSchema = otelmetrics.KeySchema{
+	Required: otelmetrics.ConcatKeys(otelmetrics.PrometheusReceiverKeys, otelmetrics.CloudKeys, otelmetrics.HostKeys, []string{
+		"k8s.cluster.name", // set_cluster_name; protected
+		"k8s.node.name",    // set_node_name, promote_node_name; protected
+		"k8s.node.uid",     // k8sattributes/cw_k8s_ci_v0_node
+	}),
+	// The neuron-burn Deployments hold cores and devices and run a Neuron runtime.
+	Expected: otelmetrics.ConcatKeys(neuronPodKeys, []string{
+		"aws.neuron.runtime.tag", // neuron_promote; zero-filled series of an idle device have no runtime_tag
+		"k8s.deployment.name", "k8s.replicaset.name",
+	}, otelmetrics.WorkloadKeys),
+	Optional: otelmetrics.ConcatKeys([]string{"k8s.pod.uid"}, // k8sattributes/cw_k8s_ci_v0_pod
+		otelmetrics.Without(otelmetrics.OwnerKindKeys, "k8s.deployment.name", "k8s.replicaset.name")),
+	OpenPrefixes: []string{"k8s.node.label.", "k8s.pod.label."},
+	Implies:      otelmetrics.MergeImplies(otelmetrics.WorkloadImplies(), otelmetrics.Together(neuronPodKeys...)),
+}
+
+func TestAttributeSchema(t *testing.T) {
+	t.Parallel()
+	tally, err := otelmetrics.TallySchema(context.Background(), queryCache, neuronMetricNamesList, otelmetrics.GroupSchema{
+		Resource:        neuronResourceSchema,
+		Instrumentation: otelmetrics.KeySchema{Required: otelmetrics.ScopeKeys}, // set_scope_neuron_monitor
+	}, nil)
+	require.NoError(t, err, "no Neuron nodes?")
+	t.Logf("%d series across %d metrics", tally.Series, len(neuronMetricNamesList))
+	require.NotZero(t, tally.Series, "no series checked")
+	for _, p := range tally.Problems() {
+		t.Errorf("%s: change the schema if intended", p)
+	}
+}

test/otel/standard/k8s_helpers_test.go

@@ -8,16 +8,16 @@ package standard
 import (
 	"context"
 	"fmt"
-	"os"
-	"path/filepath"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 )
 
@@ -45,22 +45,46 @@ func getGroundTruth(t *testing.T) *k8sGroundTruth {
 	return groundTruth
 }
 
-func buildGroundTruth() (*k8sGroundTruth, error) {
-	kubeconfig := os.Getenv("KUBECONFIG")
-	if kubeconfig == "" {
-		kubeconfig = filepath.Join(os.Getenv("HOME"), ".kube", "config")
-	}
+var (
+	k8sClient     *kubernetes.Clientset
+	k8sClientOnce sync.Once
+	k8sClientErr  error
+)
 
-	restConfig, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
-	if err != nil {
-		return nil, fmt.Errorf("building kubeconfig: %w", err)
-	}
+// sharedK8sClientset builds one clientset from the default kubeconfig loading rules
+// ($KUBECONFIG, then ~/.kube/config), falling back to the in-cluster configuration.
+func sharedK8sClientset() (*kubernetes.Clientset, error) {
+	k8sClientOnce.Do(func() {
+		restConfig, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
+			clientcmd.NewDefaultClientConfigLoadingRules(), nil).ClientConfig()
+		if err != nil {
+			if restConfig, err = rest.InClusterConfig(); err != nil {
+				k8sClientErr = fmt.Errorf("no kubeconfig or in-cluster config: %w", err)
+				return
+			}
+		}
+		k8sClient, k8sClientErr = kubernetes.NewForConfig(restConfig)
+	})
+	return k8sClient, k8sClientErr
+}
 
-	clientset, err := kubernetes.NewForConfig(restConfig)
+func k8sClientset(t *testing.T) *kubernetes.Clientset {
+	t.Helper()
+	clientset, err := sharedK8sClientset()
+	require.NoError(t, err, "creating K8s clientset")
+	return clientset
+}
+
+func buildGroundTruth() (*k8sGroundTruth, error) {
+	clientset, err := sharedK8sClientset()
 	if err != nil {
-		return nil, fmt.Errorf("creating K8s clientset: %w", err)
+		return nil, err
 	}
+	return snapshotGroundTruth(clientset)
+}
 
+// snapshotGroundTruth lists every node and pod now.
+func snapshotGroundTruth(clientset kubernetes.Interface) (*k8sGroundTruth, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
@@ -116,19 +140,7 @@ func imageTagFromPod(t *testing.T, gt *k8sGroundTruth, namespace, labelKey, labe
 // k8sServerVersion returns the Kubernetes API server version string.
 func k8sServerVersion(t *testing.T) string {
 	t.Helper()
-	kubeconfig := os.Getenv("KUBECONFIG")
-	if kubeconfig == "" {
-		kubeconfig = filepath.Join(os.Getenv("HOME"), ".kube", "config")
-	}
-	restConfig, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
-	if err != nil {
-		t.Fatalf("building kubeconfig: %v", err)
-	}
-	clientset, err := kubernetes.NewForConfig(restConfig)
-	if err != nil {
-		t.Fatalf("creating K8s clientset: %v", err)
-	}
-	sv, err := clientset.Discovery().ServerVersion()
+	sv, err := k8sClientset(t).Discovery().ServerVersion()
 	if err != nil {
 		t.Fatalf("getting K8s server version: %v", err)
 	}

test/otel/standard/ksm_availability_zone_test.go

@@ -0,0 +1,94 @@
+//go:build integration
+
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package standard
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics"
+)
+
+// ---------------------------------------------------------------------------
+// TestKSMAvailabilityZoneIsTheObjectsZone — the cluster scraper runs on one
+// node, so its own AZ is wrong for most kube-state-metrics series. On every KSM
+// series, cloud.availability_zone must be the zone of the node the series names:
+// nodemetadataenricher sets it, or else the node's topology.kubernetes.io/zone label
+// does, and a pod's own zone label replaces it. A series whose node k8sattributes
+// knows must carry it. A series that names no node, such as a Deployment's, must not.
+// The pod zone label itself is denied, so the API is the reference for it.
+// Series are read cluster-wide, not through queryCache.Get, whose host.type=""
+// filter on workload metrics would hide a series that gained the scraper's host keys.
+// ---------------------------------------------------------------------------
+
+const zoneLabel = "topology.kubernetes.io/zone"
+
+func TestKSMAvailabilityZoneIsTheObjectsZone(t *testing.T) {
+	gt := freshGroundTruth(t)
+	buckets := map[string][]string{
+		"node": ksmNodeBucket, "pod": ksmPodBucket, "container": ksmContainerBucket,
+		"object": otelmetrics.ConcatKeys(ksmWorkloadBucket.Deployment, ksmWorkloadBucket.DaemonSet,
+			ksmWorkloadBucket.StatefulSet, ksmWorkloadBucket.Job, ksmWorkloadBucket.CronJob, ksmClusterBucket),
+	}
+	for _, bucket := range []string{"node", "pod", "container", "object"} {
+		bucket := bucket
+		t.Run(bucket, func(t *testing.T) {
+			var series, withZone, fromPodLabel int
+			wrong, missing, absent, unexpected := map[string]int{}, map[string]int{}, map[string]int{}, map[string]int{}
+			for _, metricName := range buckets[bucket] {
+				for _, r := range queryAllSeries(t, metricName) {
+					series++
+					res := r.Labels.Resource
+					az, hasZone := res[ksmZoneKey]
+					if hasZone {
+						withZone++
+					}
+					nodeName := res["k8s.node.name"]
+					if nodeName == "" {
+						if hasZone {
+							unexpected[metricName]++
+						}
+						continue
+					}
+					node, known := gt.nodes[nodeName]
+					if !known {
+						continue // the node left between the API read and the scrape
+					}
+					if hasZone && az != node.Labels[zoneLabel] {
+						wrong[metricName+": "+az+" on a node in "+node.Labels[zoneLabel]]++
+					}
+					pod, ok := gt.pods[res["k8s.namespace.name"]+"/"+res["k8s.pod.name"]]
+					if want := pod.Labels[zoneLabel]; ok && want != "" && pod.Spec.NodeName == nodeName {
+						fromPodLabel++
+						if az != want {
+							missing[metricName]++
+						}
+					} else if !hasZone && node.Labels[zoneLabel] != "" && res["k8s.node.uid"] != "" {
+						absent[metricName]++ // transform/cw_k8s_ci_v0_ksm_zone takes the node's zone label
+					}
+				}
+			}
+			t.Logf("%d series, %d with an AZ, %d of a pod with a zone label", series, withZone, fromPodLabel)
+			require.NotZero(t, series, "no %s series", bucket)
+			for _, k := range otelmetrics.SortedKeys(wrong) {
+				t.Errorf("%s: wrong AZ on %d series", k, wrong[k])
+			}
+			for _, k := range otelmetrics.SortedKeys(missing) {
+				t.Errorf("%s: %d series of a pod with a zone label lack that AZ", k, missing[k])
+			}
+			for _, k := range otelmetrics.SortedKeys(absent) {
+				t.Errorf("%s: %d series of a known node with a zone label carry no AZ", k, absent[k])
+			}
+			for _, k := range otelmetrics.SortedKeys(unexpected) {
+				t.Errorf("%s: %d series name no node but carry an AZ, the cluster scraper's", k, unexpected[k])
+			}
+			if bucket == "node" || bucket == "pod" {
+				require.NotZero(t, withZone, "no %s series carries an AZ", bucket)
+			}
+		})
+	}
+}

test/otel/standard/ksm_test.go

@@ -13,9 +13,6 @@ import (
 
 	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/clientcmd"
 
 	"github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics"
 )
@@ -1571,18 +1568,8 @@ func TestKSM_LeaseExistence(t *testing.T) {
 	t.Parallel()
 	gt := getGroundTruth(t)
 
-	loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
-	kubeConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, nil)
-	restConfig, err := kubeConfig.ClientConfig()
-	if err != nil {
-		restConfig, err = rest.InClusterConfig()
-		require.NoError(t, err, "no kubeconfig or in-cluster config")
-	}
-	clientset, err := kubernetes.NewForConfig(restConfig)
-	require.NoError(t, err, "creating K8s clientset")
-
 	ctx := context.Background()
-	leases, err := clientset.CoordinationV1().Leases("amazon-cloudwatch").List(ctx, metav1.ListOptions{})
+	leases, err := k8sClientset(t).CoordinationV1().Leases("amazon-cloudwatch").List(ctx, metav1.ListOptions{})
 	require.NoError(t, err, "listing Leases")
 
 	leaseByNode := make(map[string]bool)