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/efa/dedup_test.go

@@ -48,7 +48,7 @@ func TestEFANoDuplicateSeries(t *testing.T) {
 		`efa_rx_bytes{"@resource.k8s.cluster.name"="%s","@resource.k8s.node.name"="%s","aws.efa.device"="%s"}`,
 		escaped, pinNode, pinDevice,
 	)
-	pinned, err := client.Query(ctx, promql)
+	pinned, err := client.QueryComplete(ctx, promql)
 	require.NoError(t, err, "querying pinned efa_rx_bytes")
 	require.NotEmpty(t, pinned, "pinned efa_rx_bytes returned 0 results")
 	require.Equal(t, 1, len(pinned),

test/otel/efa/efa_test.go

@@ -406,7 +406,7 @@ func TestEFANodeGroupCoverage(t *testing.T) {
 		promql := fmt.Sprintf(
 			`efa_rx_bytes{"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`,
 			escaped, "c5n.9xlarge")
-		results, err := client.Query(context.Background(), promql)
+		results, err := client.QueryComplete(context.Background(), promql)
 		require.NoError(t, err, "querying efa_rx_bytes on c5n.9xlarge")
 		require.True(t, len(results) > 0,
 			"EFA metrics missing from c5n.9xlarge nodes — awsefareceiver not collecting?")

test/otel/efa/node_color_test.go

@@ -31,7 +31,7 @@ func TestEFAHasYellowNodeColor(t *testing.T) {
 			promql := fmt.Sprintf(
 				`%s{"@resource.k8s.cluster.name"="%s","@resource.%s"="yellow"}`,
 				metricName, escaped, nodeColorLabel)
-			results, err := client.Query(context.Background(), promql)
+			results, err := client.QueryComplete(context.Background(), promql)
 			require.NoError(t, err, "querying %s with node-color=yellow filter", metricName)
 			require.True(t, len(results) > 0,
 				"%s: no results from c5n.9xlarge nodes (expected node-color=yellow)", metricName)

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/gpu/dcgm_test.go

@@ -422,7 +422,7 @@ func TestDCGMNodeGroupCoverage(t *testing.T) {
 			promql := fmt.Sprintf(
 				`DCGM_FI_DEV_GPU_UTIL{"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`,
 				escaped, ng.InstanceType)
-			results, err := client.Query(context.Background(), promql)
+			results, err := client.QueryComplete(context.Background(), promql)
 			require.NoError(t, err, "querying DCGM_FI_DEV_GPU_UTIL on %s", ng.Description)
 			require.True(t, len(results) > 0,
 				"DCGM missing from %s nodes (%s) — dcgm-exporter not running?",

test/otel/gpu/dedup_test.go

@@ -50,7 +50,7 @@ func TestDCGMNoDuplicateSeries(t *testing.T) {
 		`DCGM_FI_DEV_GPU_UTIL{"@resource.k8s.cluster.name"="%s","@resource.k8s.node.name"="%s",UUID="%s"}`,
 		escaped, pinNode, pinUUID,
 	)
-	pinned, err := client.Query(ctx, promql)
+	pinned, err := client.QueryComplete(ctx, promql)
 	require.NoError(t, err, "querying pinned DCGM_FI_DEV_GPU_UTIL")
 	require.NotEmpty(t, pinned, "pinned DCGM_FI_DEV_GPU_UTIL returned 0 results")
 	require.Equal(t, 1, len(pinned),

test/otel/gpu/node_color_test.go

@@ -35,7 +35,7 @@ func TestDCGMHasGreenNodeColor(t *testing.T) {
 			promql := fmt.Sprintf(
 				`%s{"@resource.k8s.cluster.name"="%s","@resource.%s"="green"}`,
 				metricName, escaped, nodeColorLabel)
-			results, err := client.Query(context.Background(), promql)
+			results, err := client.QueryComplete(context.Background(), promql)
 			require.NoError(t, err, "querying %s with node-color=green filter", metricName)
 			require.True(t, len(results) > 0,
 				"%s: no results from GPU nodes (expected node-color=green)", metricName)

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/neuron/multi_device_test.go

@@ -273,7 +273,7 @@ func TestMultiNeuronCorrelatedCoreCount(t *testing.T) {
 	promql := fmt.Sprintf(
 		`neuroncore_utilization_ratio{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"neuron-burn-multi.*"}`,
 		escaped)
-	burnResults, err := client.Query(ctx, promql)
+	burnResults, err := client.QueryComplete(ctx, promql)
 	require.NoError(t, err, "querying neuroncore_utilization_ratio for neuron-burn-multi")
 	require.Equal(t, 2, len(burnResults), fmt.Sprintf("Expected 2 cores correlated to neuron-burn-multi (1 whole device × 2 cores), got %d",
 		len(burnResults)))
@@ -289,7 +289,7 @@ func TestMultiNeuronBurnDevicePodLabels(t *testing.T) {
 	promql := fmt.Sprintf(
 		`neuroncore_utilization_ratio{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"neuron-burn-multi.*"}`,
 		escaped)
-	burnResults, err := client.Query(ctx, promql)
+	burnResults, err := client.QueryComplete(ctx, promql)
 	require.NoError(t, err, "querying neuroncore_utilization_ratio for neuron-burn-multi")
 	require.Equal(t, 2, len(burnResults), fmt.Sprintf("Expected 2 cores correlated to neuron-burn-multi, got %d", len(burnResults)))
 

test/otel/neuron/neuron_test.go

@@ -528,7 +528,7 @@ func TestNeuronNodeGroupCoverage(t *testing.T) {
 			promql := fmt.Sprintf(
 				`neuroncore_utilization_ratio{"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`,
 				escaped, ng.InstanceType)
-			results, err := client.Query(context.Background(), promql)
+			results, err := client.QueryComplete(context.Background(), promql)
 			require.NoError(t, err, "querying neuroncore_utilization_ratio on %s", ng.Description)
 			require.True(t, len(results) > 0,
 				"Neuron metrics missing from %s nodes (%s) — neuron-monitor not running?",

test/otel/neuron/node_color_test.go

@@ -33,7 +33,7 @@ func TestNeuronHasRedNodeColor(t *testing.T) {
 			promql := fmt.Sprintf(
 				`%s{"@resource.k8s.cluster.name"="%s","@resource.%s"="red"}`,
 				metricName, escaped, nodeColorLabel)
-			results, err := client.Query(context.Background(), promql)
+			results, err := client.QueryComplete(context.Background(), promql)
 			require.NoError(t, err, "querying %s with node-color=red filter", metricName)
 			require.True(t, len(results) > 0,
 				"%s: no results from inf2.xlarge nodes (expected node-color=red)", metricName)
@@ -55,7 +55,7 @@ func TestNeuronHasPurpleNodeColor(t *testing.T) {
 			promql := fmt.Sprintf(
 				`%s{"@resource.k8s.cluster.name"="%s","@resource.%s"="purple"}`,
 				md.Name, escaped, nodeColorLabel)
-			results, err := client.Query(context.Background(), promql)
+			results, err := client.QueryComplete(context.Background(), promql)
 			require.NoError(t, err, "querying %s with node-color=purple filter", md.Name)
 			require.True(t, len(results) > 0,
 				"%s: no results from inf2.24xlarge node (expected node-color=purple)", md.Name)

test/otel/performance/performance_test.go

@@ -62,7 +62,7 @@ func getNodeAllocatable(t *testing.T, metricName string, resource string) (float
 	// and skew the percent-of-node denominator the thresholds are calibrated to.
 	query := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s", resource="%s"}`,
 		metricName, otelmetrics.EscapePromQLValue(cfg.ClusterName), resource)
-	results, err := client.Query(context.Background(), query)
+	results, err := client.QueryComplete(context.Background(), query)
 	require.NoError(t, err, "failed to query node allocatable for resource=%s", resource)
 	require.NotEmpty(t, results, "no kube_node_status_allocatable data for resource=%s — are KSM metrics being collected?", resource)
 	var sum float64

test/otel/standard/cadvisor_test.go

@@ -138,7 +138,7 @@ func TestCadvisorNginxWorkloadLabels(t *testing.T) {
 		t.Run(name, func(t *testing.T) {
 			promql := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"nginx-test.*"}`,
 				name, cfg.ClusterName)
-			nginx, err := client.Query(context.Background(), promql)
+			nginx, err := client.QueryComplete(context.Background(), promql)
 			require.NoError(t, err, "querying %s for nginx-test", name)
 			require.True(t, len(nginx) > 0, "No %s from nginx-test pods", name)
 			for _, r := range nginx {
@@ -237,7 +237,7 @@ func TestCadvisorNodeGroupCoverage(t *testing.T) {
 			promql := fmt.Sprintf(
 				`container_memory_working_set_bytes{"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`,
 				otelmetrics.EscapePromQLValue(cfg.ClusterName), ng.InstanceType)
-			results, err := client.Query(context.Background(), promql)
+			results, err := client.QueryComplete(context.Background(), promql)
 			require.NoError(t, err, "querying cadvisor on %s", ng.Description)
 			require.True(t, len(results) > 0, "cadvisor missing from %s (%s)", ng.Description, ng.InstanceType)
 		})

test/otel/standard/dedup_test.go

@@ -29,7 +29,7 @@ func TestNodeExporterNoDuplicateSeries(t *testing.T) {
 		`node_cpu_seconds_total{"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`,
 		escaped, clusterHostTypes[0],
 	)
-	results, err := client.Query(ctx, promql)
+	results, err := client.QueryComplete(ctx, promql)
 	require.NoError(t, err, "querying node_cpu_seconds_total")
 	require.NotEmpty(t, results, "node_cpu_seconds_total not available")
 
@@ -44,7 +44,7 @@ func TestNodeExporterNoDuplicateSeries(t *testing.T) {
 		`node_cpu_seconds_total{"@resource.k8s.cluster.name"="%s","@resource.k8s.node.name"="%s",cpu="%s",mode="%s"}`,
 		escaped, pinNode, pinCPU, pinMode,
 	)
-	pinned, err := client.Query(ctx, promql)
+	pinned, err := client.QueryComplete(ctx, promql)
 	require.NoError(t, err, "querying pinned node_cpu_seconds_total")
 	require.NotEmpty(t, pinned, "pinned node_cpu_seconds_total returned 0 results")
 	require.Equal(t, 1, len(pinned),
@@ -62,7 +62,7 @@ func TestCadvisorNoDuplicateSeries(t *testing.T) {
 		`container_memory_working_set_bytes{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"nginx-test.*"}`,
 		escaped,
 	)
-	results, err := client.Query(ctx, promql)
+	results, err := client.QueryComplete(ctx, promql)
 	require.NoError(t, err, "querying container_memory_working_set_bytes for nginx-test")
 	require.NotEmpty(t, results, "container_memory_working_set_bytes not available for nginx-test pods")
 
@@ -84,7 +84,7 @@ func TestCadvisorNoDuplicateSeries(t *testing.T) {
 		`container_memory_working_set_bytes{"@resource.k8s.cluster.name"="%s","@resource.k8s.node.name"="%s","@resource.k8s.pod.name"="%s","@resource.k8s.container.name"="%s"}`,
 		escaped, pinNode, pinPod, pinContainer,
 	)
-	pinned, err := client.Query(ctx, promql)
+	pinned, err := client.QueryComplete(ctx, promql)
 	require.NoError(t, err, "querying pinned container_memory_working_set_bytes")
 	require.NotEmpty(t, pinned, "pinned container_memory_working_set_bytes returned 0 results")
 	require.Equal(t, 1, len(pinned),
@@ -128,7 +128,7 @@ func TestKubeletstatsNoDuplicateSeries(t *testing.T) {
 		`{"__name__"="k8s.node.cpu.usage","@resource.k8s.cluster.name"="%s","@resource.k8s.node.name"="%s"}`,
 		escaped, pinNode,
 	)
-	pinned, err := client.Query(ctx, promql)
+	pinned, err := client.QueryComplete(ctx, promql)
 	require.NoError(t, err, "querying pinned k8s.node.cpu.usage")
 	require.NotEmpty(t, pinned, "pinned k8s.node.cpu.usage returned 0 results")
 	require.Equal(t, 1, len(pinned),