test(otel): run the every-series checks on every node and model the label filters

test/otel/attr_limit/setup_test.go

@@ -26,7 +26,8 @@ var (
 	queryCache *otelmetrics.QueryCache
 )
 
-var clusterHostTypes = []string{"t3.medium"}
+// Instance types in the attribute-limit cluster; OTEL_HOST_TYPES overrides them.
+var clusterHostTypes = otelmetrics.ResolveHostTypes([]string{"t3.medium"})
 
 func TestMain(m *testing.M) {
 	environment.RegisterEnvironmentMetaDataFlags()
@@ -80,6 +81,8 @@ func TestMain(m *testing.M) {
 		os.Exit(1)
 	}
 
+	fmt.Fprintf(os.Stderr, "Host types: %v (set %s to override)\n", clusterHostTypes, otelmetrics.HostTypesEnvVar)
+
 	hostMappings := []otelmetrics.SourceHostMapping{
 		{Source: otelmetrics.SourceNodeExporter, HostTypes: clusterHostTypes},
 		{Source: otelmetrics.SourceCadvisor, HostTypes: clusterHostTypes},

test/otel/ebs_csi/setup_test.go

@@ -29,8 +29,8 @@ var (
 // Instrumentation scope name constant.
 const scopePrometheus = "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver"
 
-// Instance types in the EBS CSI cluster.
-var clusterHostTypes = []string{"t3.medium"}
+// Instance types in the EBS CSI cluster; OTEL_HOST_TYPES overrides them.
+var clusterHostTypes = otelmetrics.ResolveHostTypes([]string{"t3.medium"})
 
 func TestMain(m *testing.M) {
 	environment.RegisterEnvironmentMetaDataFlags()
@@ -84,6 +84,8 @@ func TestMain(m *testing.M) {
 		os.Exit(1)
 	}
 
+	fmt.Fprintf(os.Stderr, "Host types: %v (set %s to override)\n", clusterHostTypes, otelmetrics.HostTypesEnvVar)
+
 	hostMappings := []otelmetrics.SourceHostMapping{
 		{Source: otelmetrics.SourceEBSCSI, HostTypes: nil},
 	}

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(), allNodesCache, 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/efa/setup_test.go

@@ -24,10 +24,12 @@ var (
 	cfg        otelmetrics.TestConfig
 	client     *otelmetrics.OtelMetricsClient
 	queryCache *otelmetrics.QueryCache
+	// allNodesCache reads every series on every node, for the attribute schema.
+	allNodesCache *otelmetrics.QueryCache
 )
 
-// Instance types in the EFA cluster.
-var clusterHostTypes = []string{"c5n.9xlarge"}
+// Instance types in the EFA cluster; OTEL_HOST_TYPES overrides them.
+var clusterHostTypes = otelmetrics.ResolveHostTypes([]string{"c5n.9xlarge"})
 
 func TestMain(m *testing.M) {
 	environment.RegisterEnvironmentMetaDataFlags()
@@ -81,24 +83,32 @@ func TestMain(m *testing.M) {
 		os.Exit(1)
 	}
 
+	fmt.Fprintf(os.Stderr, "Host types: %v (set %s to override)\n", clusterHostTypes, otelmetrics.HostTypesEnvVar)
+
 	hostMappings := []otelmetrics.SourceHostMapping{
 		{Source: otelmetrics.SourceNodeExporter, HostTypes: clusterHostTypes},
 		{Source: otelmetrics.SourceCadvisor, HostTypes: clusterHostTypes},
 		{Source: otelmetrics.SourceKubeletstats, HostTypes: clusterHostTypes},
 		{Source: otelmetrics.SourceEFA, HostTypes: clusterHostTypes},
 	}
 
-	registry := otelmetrics.NewSourceRegistry(clusterHostTypes, hostMappings,
-		otelmetrics.SourceMapping{Source: otelmetrics.SourceNodeExporter, Metrics: nodeExporterMetrics},
-		otelmetrics.SourceMapping{Source: otelmetrics.SourceCadvisor, Metrics: cadvisorMetrics},
-		otelmetrics.SourceMapping{Source: otelmetrics.SourceKubeletstats, Metrics: kubeletstatsMetrics},
-		otelmetrics.SourceMapping{Source: otelmetrics.SourceEFA, Metrics: efaMetrics},
-	)
+	sourceMappings := []otelmetrics.SourceMapping{
+		{Source: otelmetrics.SourceNodeExporter, Metrics: nodeExporterMetrics},
+		{Source: otelmetrics.SourceCadvisor, Metrics: cadvisorMetrics},
+		{Source: otelmetrics.SourceKubeletstats, Metrics: kubeletstatsMetrics},
+		{Source: otelmetrics.SourceEFA, Metrics: efaMetrics},
+	}
+	registry := otelmetrics.NewSourceRegistry(clusterHostTypes, hostMappings, sourceMappings...)
 
 	queryCache = otelmetrics.NewQueryCache(client, cfg.ClusterName,
 		otelmetrics.WithHostTypes(clusterHostTypes),
 		otelmetrics.WithSourceRegistry(registry),
 	)
+	// Same routing with no host types, sharing queryCache's concurrency limit.
+	allNodesCache = otelmetrics.NewQueryCache(client, cfg.ClusterName,
+		otelmetrics.WithSourceRegistry(otelmetrics.NewSourceRegistry(nil, nil, sourceMappings...)),
+		otelmetrics.WithSharedConcurrency(queryCache),
+	)
 
 	os.Exit(m.Run())
 }

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(), allNodesCache, 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

@@ -87,7 +87,7 @@ func TestDCGMNamespace(t *testing.T) {
 }
 
 // TestDCGMMultiNodeCoverage validates that DCGM_FI_DEV_GPU_UTIL is reported
-// from at least 2 GPU nodes (g4dn.xlarge + g4dn.12xlarge).
+// from at least 2 GPU nodes (the idle node and the busy node).
 func TestDCGMMultiNodeCoverage(t *testing.T) {
 	t.Parallel()
 	results, err := queryCache.Get(context.Background(), "DCGM_FI_DEV_GPU_UTIL")
@@ -104,8 +104,22 @@ func TestDCGMMultiNodeCoverage(t *testing.T) {
 		"Expected DCGM metrics from at least 2 GPU nodes, got %d", len(nodes))
 }
 
+// TestDCGMGPUCountMatchesAllocatable validates that every GPU node reports
+// one distinct gpu in DCGM_FI_DEV_GPU_UTIL per device in its nvidia.com/gpu
+// allocatable.
+func TestDCGMGPUCountMatchesAllocatable(t *testing.T) {
+	t.Parallel()
+	for _, node := range gpuRoles(t).Nodes {
+		want := allocatableGPUs(t, node.Name)
+		gpus := node.GPUs()
+		t.Logf("%s: nvidia.com/gpu allocatable %d, DCGM gpu %v", node.Name, want, gpus)
+		require.Equal(t, want, len(gpus), "Expected %d distinct gpu on %s (nvidia.com/gpu allocatable), got %d: %v",
+			want, node.Name, len(gpus), gpus)
+	}
+}
+
 // TestDCGMActiveNodeNonzero validates that at least one GPU node has
-// non-zero utilization (multi-gpu-burn should be running on g4dn.12xlarge).
+// non-zero utilization (multi-gpu-burn should be running on the busy node).
 func TestDCGMActiveNodeNonzero(t *testing.T) {
 	t.Parallel()
 	results, err := queryCache.Get(context.Background(), "DCGM_FI_DEV_GPU_UTIL")
@@ -211,7 +225,7 @@ func TestDCGMExpectedLabels(t *testing.T) {
 
 // TestDCGMGpuburnContainerName validates that multi-gpu-burn pod results have
 // k8s.container.name=gpu-burn for all DCGM metrics.
-// In this ephemeral cluster, multi-gpu-burn on g4dn.12xlarge is the active workload.
+// In this ephemeral cluster, multi-gpu-burn on the busy node is the active workload.
 func TestDCGMGpuburnContainerName(t *testing.T) {
 	t.Parallel()
 	for _, metricName := range dcgmMetricNamesList {
@@ -238,18 +252,19 @@ func TestDCGMGpuburnContainerName(t *testing.T) {
 }
 
 // TestDCGMIdleNodeNoContainerName validates that uncorrelated DCGM results
-// (from idle g4dn.xlarge) do NOT have k8s.container.name.
+// (from the idle node) do NOT have k8s.container.name.
 func TestDCGMIdleNodeNoContainerName(t *testing.T) {
 	t.Parallel()
+	nodes := idleNodes(t)
 	for _, metricName := range dcgmMetricNamesList {
 		metricName := metricName
 		t.Run(metricName, func(t *testing.T) {
 			t.Parallel()
 			results, err := queryCache.Get(context.Background(), metricName)
 			require.NoError(t, err, "querying %s", metricName)
 			require.NotEmpty(t, results, "%s not available (no GPU nodes?)", metricName)
-			idle := filterByHostType(results, "g4dn.xlarge")
-			require.NotEmpty(t, idle, "No %s results from idle g4dn.xlarge node", metricName)
+			idle := otelmetrics.SeriesOnNodes(results, nodes)
+			require.NotEmpty(t, idle, "No %s results from idle nodes %v", metricName, nodeNames(nodes))
 			for _, r := range idle {
 				_, has := r.Labels.Resource["k8s.container.name"]
 				require.False(t, has,
@@ -289,18 +304,19 @@ func TestDCGMGpuburnWorkloadLabels(t *testing.T) {
 }
 
 // TestDCGMIdleNodeNoWorkloadLabels validates that DCGM results from the idle
-// g4dn.xlarge node do NOT have k8s.workload.name.
+// node do NOT have k8s.workload.name.
 func TestDCGMIdleNodeNoWorkloadLabels(t *testing.T) {
 	t.Parallel()
+	nodes := idleNodes(t)
 	for _, metricName := range dcgmMetricNamesList {
 		metricName := metricName
 		t.Run(metricName, func(t *testing.T) {
 			t.Parallel()
 			results, err := queryCache.Get(context.Background(), metricName)
 			require.NoError(t, err, "querying %s", metricName)
 			require.NotEmpty(t, results, "%s not available (no GPU nodes?)", metricName)
-			idle := filterByHostType(results, "g4dn.xlarge")
-			require.NotEmpty(t, idle, "No results from idle g4dn.xlarge node for %s", metricName)
+			idle := otelmetrics.SeriesOnNodes(results, nodes)
+			require.NotEmpty(t, idle, "No results from idle nodes %v for %s", nodeNames(nodes), metricName)
 			for _, r := range idle {
 				_, hasName := r.Labels.Resource["k8s.workload.name"]
 				require.False(t, hasName,
@@ -389,18 +405,19 @@ func TestDCGMHasRawPromotedKeys(t *testing.T) {
 }
 
 // TestDCGMIdleNodeNoPodColor validates that DCGM results from the idle
-// g4dn.xlarge node do NOT have the pod color label.
+// node do NOT have the pod color label.
 func TestDCGMIdleNodeNoPodColor(t *testing.T) {
 	t.Parallel()
+	nodes := idleNodes(t)
 	for _, metricName := range dcgmMetricNamesList {
 		metricName := metricName
 		t.Run(metricName, func(t *testing.T) {
 			t.Parallel()
 			results, err := queryCache.Get(context.Background(), metricName)
 			require.NoError(t, err, "querying %s", metricName)
 			require.NotEmpty(t, results, "%s not available (no GPU nodes?)", metricName)
-			idle := filterByHostType(results, "g4dn.xlarge")
-			require.NotEmpty(t, idle, "No results from idle g4dn.xlarge node for %s", metricName)
+			idle := otelmetrics.SeriesOnNodes(results, nodes)
+			require.NotEmpty(t, idle, "No results from idle nodes %v for %s", nodeNames(nodes), metricName)
 			for _, r := range idle {
 				_, has := r.Labels.Resource[podColorLabel]
 				require.False(t, has,
@@ -422,7 +439,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),