From 84af69433b761cadc8ed7b2252b7e3384816a03a Mon Sep 17 00:00:00 2001 From: Spandan Naik Date: Thu, 27 Aug 2026 15:34:19 +0000 Subject: [PATCH 1/3] fix(neuron): stop losing per-core Neuron metrics on multi-runtime nodes On a node with more than one Neuron runtime, roughly half of every per-core Neuron metric is silently replaced by zero before it leaves the agent. A NeuronCore pinned at 75% utilization reports 0. Measured on an inf2.xlarge (2 NeuronCores, one process pinned per core). neuron-monitor's own /metrics is correct -- it emits the full (neuroncore x runtime_tag) cross-product with only the matching pairs non-zero: neuroncore_utilization_ratio{neuroncore="0",runtime_tag="...core0"} 0.7523 <- real neuroncore_utilization_ratio{neuroncore="1",runtime_tag="...core0"} 0.0 neuroncore_utilization_ratio{neuroncore="0",runtime_tag="...core1"} 0.0 neuroncore_utilization_ratio{neuroncore="1",runtime_tag="...core1"} 0.7517 <- real A debug exporter on the live pipeline shows what the agent then produces: ONE ResourceMetrics containing FOUR datapoints with only TWO distinct attribute sets -- aws.neuron.core=0 Value: 75.154815 <- real aws.neuron.core=1 Value: 0.000000 aws.neuron.core=0 Value: 0.000000 <- same identity, shadows the real one aws.neuron.core=1 Value: 75.363968 Cause, in transform/cw_k8s_ci_v0_neuron_promote: 1. It runs in `context: datapoint` but writes `resource.attributes`. Resource attributes are per-ResourceMetrics, so with N runtimes in one RM the statement executes N times and the last write wins -- which is why a single arbitrary runtime tag ends up labelling the whole node. 2. It then deletes `runtime_tag` from the datapoint. That was the only attribute distinguishing the real reading from the other runtime's zero, so two datapoints collapse to one identity with different values. Nothing downstream reconciles them (awsattributelimit early-returns below 150 attributes; batch only concatenates), so both go over the wire and one is arbitrarily dropped at ingest. Blast radius is wider than utilization: neuron_runtime_memory_used_bytes has 4 series upstream (2 memory_location x 2 runtime_tag) and arrives as 2, so one runtime's memory is dropped as well. Fix: make `runtime_tag` a groupbyattrs grouping key so each runtime gets its own ResourceMetrics. That removes the key from the datapoint as a side effect, so each datapoint identity stays unique and the last-write-wins clobber becomes a no-op. The promote then moves to `context: resource`, which is where the attribute now lives -- and is how the DCGM pipeline already does it (containerinsights/dcgm.yaml:66-83). The six pod-identity statements are dropped because they are dead code: groupbyattrs has already deleted those keys from the datapoint by then, so `attributes["k8s.pod.name"]` is always nil and they never fire. Verified on the live cluster: core 0 goes from 0 to 75.18, and CloudWatch now carries 4 correctly-labelled series instead of 2. Trade-off: this publishes the full cross-product, so series count grows multiplicatively with runtime count (2 -> 4 here; a 32-core trn1 with 4 runtimes goes from 32 to 128 series per metric). The alternative is to aggregate the runtime dimension away in-agent, which keeps the count flat but loses per-runtime attribution; that trade-off is rejected here because the PromQL surface can carry the runtime dimension and a runtime legitimately spans multiple cores. Note: the legacy EMF path already handles this correctly, via gpuattributes/internal/awsneuron_metric_modifier.go -- it flattens runtime_tag to "DEFAULT" then aggregates per core with max (commit dc672ff52). The OTel rewrite reimplemented synthesis and scaling but not the aggregation. There is no multi-runtime test case in plugins/processors/awsneuron/processor_test.go, which is why this shipped; adding pipeline-level coverage is recommended follow-up. --- .../opentelemetry/combined_v1_v2_eks_config.yaml | 11 +++-------- .../opentelemetry/container_insights_node_config.yaml | 11 +++-------- .../opentelemetry/default_otel_config_aks.yaml | 11 +++-------- .../opentelemetry/default_otel_config_gke.yaml | 11 +++-------- .../opentelemetry/containerinsights/neuron.yaml | 11 +++-------- 5 files changed, 15 insertions(+), 40 deletions(-) diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.yaml b/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.yaml index 64f2cfd821..d4f71d6466 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.yaml +++ b/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.yaml @@ -642,6 +642,7 @@ processors: - k8s.pod.name - k8s.namespace.name - k8s.container.name + - runtime_tag groupbyattrs/files: keys: - log.file.name @@ -1129,15 +1130,9 @@ processors: transform/cw_k8s_ci_v0_neuron_promote: error_mode: ignore metric_statements: - - context: datapoint + - context: resource statements: - - set(resource.attributes["k8s.pod.name"], attributes["k8s.pod.name"]) where attributes["k8s.pod.name"] != nil - - set(resource.attributes["k8s.namespace.name"], attributes["k8s.namespace.name"]) where attributes["k8s.namespace.name"] != nil - - set(resource.attributes["k8s.container.name"], attributes["k8s.container.name"]) where attributes["k8s.container.name"] != nil - - set(resource.attributes["aws.neuron.runtime.tag"], attributes["runtime_tag"]) where attributes["runtime_tag"] != nil - - delete_key(attributes, "k8s.pod.name") where attributes["k8s.pod.name"] != nil - - delete_key(attributes, "k8s.namespace.name") where attributes["k8s.namespace.name"] != nil - - delete_key(attributes, "k8s.container.name") where attributes["k8s.container.name"] != nil + - set(attributes["aws.neuron.runtime.tag"], attributes["runtime_tag"]) where attributes["runtime_tag"] != nil - delete_key(attributes, "runtime_tag") where attributes["runtime_tag"] != nil transform/cw_k8s_ci_v0_promote_node_name: error_mode: ignore diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_node_config.yaml b/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_node_config.yaml index 9d1a005ce8..109a7e9546 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_node_config.yaml +++ b/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_node_config.yaml @@ -209,6 +209,7 @@ processors: - k8s.pod.name - k8s.namespace.name - k8s.container.name + - runtime_tag k8sattributes/cw_k8s_ci_v0_node: auth_type: serviceAccount exclude: @@ -765,15 +766,9 @@ processors: transform/cw_k8s_ci_v0_neuron_promote: error_mode: ignore metric_statements: - - context: datapoint + - context: resource statements: - - set(resource.attributes["k8s.pod.name"], attributes["k8s.pod.name"]) where attributes["k8s.pod.name"] != nil - - set(resource.attributes["k8s.namespace.name"], attributes["k8s.namespace.name"]) where attributes["k8s.namespace.name"] != nil - - set(resource.attributes["k8s.container.name"], attributes["k8s.container.name"]) where attributes["k8s.container.name"] != nil - - set(resource.attributes["aws.neuron.runtime.tag"], attributes["runtime_tag"]) where attributes["runtime_tag"] != nil - - delete_key(attributes, "k8s.pod.name") where attributes["k8s.pod.name"] != nil - - delete_key(attributes, "k8s.namespace.name") where attributes["k8s.namespace.name"] != nil - - delete_key(attributes, "k8s.container.name") where attributes["k8s.container.name"] != nil + - set(attributes["aws.neuron.runtime.tag"], attributes["runtime_tag"]) where attributes["runtime_tag"] != nil - delete_key(attributes, "runtime_tag") where attributes["runtime_tag"] != nil transform/cw_k8s_ci_v0_node_logs_set_log_destination: error_mode: ignore diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/default_otel_config_aks.yaml b/translator/tocwconfig/sampleConfig/opentelemetry/default_otel_config_aks.yaml index 08c01102f6..36bce8c16c 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/default_otel_config_aks.yaml +++ b/translator/tocwconfig/sampleConfig/opentelemetry/default_otel_config_aks.yaml @@ -281,6 +281,7 @@ processors: - k8s.pod.name - k8s.namespace.name - k8s.container.name + - runtime_tag k8sattributes/cw_k8s_ci_v0_node: auth_type: serviceAccount exclude: @@ -763,15 +764,9 @@ processors: transform/cw_k8s_ci_v0_neuron_promote: error_mode: ignore metric_statements: - - context: datapoint + - context: resource statements: - - set(resource.attributes["k8s.pod.name"], attributes["k8s.pod.name"]) where attributes["k8s.pod.name"] != nil - - set(resource.attributes["k8s.namespace.name"], attributes["k8s.namespace.name"]) where attributes["k8s.namespace.name"] != nil - - set(resource.attributes["k8s.container.name"], attributes["k8s.container.name"]) where attributes["k8s.container.name"] != nil - - set(resource.attributes["aws.neuron.runtime.tag"], attributes["runtime_tag"]) where attributes["runtime_tag"] != nil - - delete_key(attributes, "k8s.pod.name") where attributes["k8s.pod.name"] != nil - - delete_key(attributes, "k8s.namespace.name") where attributes["k8s.namespace.name"] != nil - - delete_key(attributes, "k8s.container.name") where attributes["k8s.container.name"] != nil + - set(attributes["aws.neuron.runtime.tag"], attributes["runtime_tag"]) where attributes["runtime_tag"] != nil - delete_key(attributes, "runtime_tag") where attributes["runtime_tag"] != nil transform/cw_k8s_ci_v0_promote_node_name: error_mode: ignore diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/default_otel_config_gke.yaml b/translator/tocwconfig/sampleConfig/opentelemetry/default_otel_config_gke.yaml index f9d782e291..3d28332843 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/default_otel_config_gke.yaml +++ b/translator/tocwconfig/sampleConfig/opentelemetry/default_otel_config_gke.yaml @@ -281,6 +281,7 @@ processors: - k8s.pod.name - k8s.namespace.name - k8s.container.name + - runtime_tag k8sattributes/cw_k8s_ci_v0_node: auth_type: serviceAccount exclude: @@ -762,15 +763,9 @@ processors: transform/cw_k8s_ci_v0_neuron_promote: error_mode: ignore metric_statements: - - context: datapoint + - context: resource statements: - - set(resource.attributes["k8s.pod.name"], attributes["k8s.pod.name"]) where attributes["k8s.pod.name"] != nil - - set(resource.attributes["k8s.namespace.name"], attributes["k8s.namespace.name"]) where attributes["k8s.namespace.name"] != nil - - set(resource.attributes["k8s.container.name"], attributes["k8s.container.name"]) where attributes["k8s.container.name"] != nil - - set(resource.attributes["aws.neuron.runtime.tag"], attributes["runtime_tag"]) where attributes["runtime_tag"] != nil - - delete_key(attributes, "k8s.pod.name") where attributes["k8s.pod.name"] != nil - - delete_key(attributes, "k8s.namespace.name") where attributes["k8s.namespace.name"] != nil - - delete_key(attributes, "k8s.container.name") where attributes["k8s.container.name"] != nil + - set(attributes["aws.neuron.runtime.tag"], attributes["runtime_tag"]) where attributes["runtime_tag"] != nil - delete_key(attributes, "runtime_tag") where attributes["runtime_tag"] != nil transform/cw_k8s_ci_v0_promote_node_name: error_mode: ignore diff --git a/translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron.yaml b/translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron.yaml index beeed7e846..cd7970b39a 100644 --- a/translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron.yaml +++ b/translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron.yaml @@ -89,18 +89,13 @@ processors: - k8s.pod.name - k8s.namespace.name - k8s.container.name + - runtime_tag transform/cw_k8s_ci_v0_neuron_promote: error_mode: ignore metric_statements: - - context: datapoint + - context: resource statements: - - set(resource.attributes["k8s.pod.name"], attributes["k8s.pod.name"]) where attributes["k8s.pod.name"] != nil - - set(resource.attributes["k8s.namespace.name"], attributes["k8s.namespace.name"]) where attributes["k8s.namespace.name"] != nil - - set(resource.attributes["k8s.container.name"], attributes["k8s.container.name"]) where attributes["k8s.container.name"] != nil - - set(resource.attributes["aws.neuron.runtime.tag"], attributes["runtime_tag"]) where attributes["runtime_tag"] != nil - - delete_key(attributes, "k8s.pod.name") where attributes["k8s.pod.name"] != nil - - delete_key(attributes, "k8s.namespace.name") where attributes["k8s.namespace.name"] != nil - - delete_key(attributes, "k8s.container.name") where attributes["k8s.container.name"] != nil + - set(attributes["aws.neuron.runtime.tag"], attributes["runtime_tag"]) where attributes["runtime_tag"] != nil - delete_key(attributes, "runtime_tag") where attributes["runtime_tag"] != nil transform/cw_k8s_ci_v0_neuron_hw_attrs: error_mode: ignore From 767fb4b16e2ce87ad2b84f9620c6362e21acada2 Mon Sep 17 00:00:00 2001 From: Spandan Naik Date: Mon, 31 Aug 2026 11:58:17 +0000 Subject: [PATCH 2/3] test(neuron): pipeline-level coverage for multi-runtime per-core data loss Builds the real groupbyattrs + promote processors from the embedded neuron.yaml and feeds them the shape neuron-monitor emits with two runtimes on one node (2 cores x 2 runtimes, non-owning pairs at zero, all four sharing one pod). Four tests: KeepsEveryCorePerRuntime all four (core, tag, value) triples survive SeparatesRuntimesIntoResources one ResourceMetrics per runtime PromotesPodIdentity pod/ns/container on resource, off datapoints -- pins the six promote statements the fix removed, which groupbyattrs already performs PreFixConfigLosesData negative control: the datapoint-context promote collapses to one tag and two datapoints per core Config is read from the embedded template, not restated, so a revert fails the test. Verified by mutation: removing `- runtime_tag` from the groupbyattrs keys in neuron.yaml fails three of the four. Runs in ~25ms with no cluster or Neuron hardware, which is the gap that let the original defect ship -- it needed >=2 runtimes on one node to appear and no test environment had that. --- .../neuron_multiruntime_test.go | 336 ++++++++++++++++++ 1 file changed, 336 insertions(+) create mode 100644 translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron_multiruntime_test.go diff --git a/translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron_multiruntime_test.go b/translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron_multiruntime_test.go new file mode 100644 index 0000000000..ca59ddfff6 --- /dev/null +++ b/translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron_multiruntime_test.go @@ -0,0 +1,336 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +// Per-core Neuron data loss on multi-runtime nodes. +// +// The shipped neuron.yaml is run for real here -- groupbyattrs then the promote +// transform, built from the embedded template -- against synthetic metrics in the +// shape neuron-monitor emits when two Neuron runtimes share a node: every core +// reported by every runtime, with the non-owning pairs at zero. +// +// The defect: the promote ran in `context: datapoint` while writing +// resource.attributes. Resource attributes are per-ResourceMetrics, so the +// statement ran once per datapoint and the last write won; it then deleted +// runtime_tag from the datapoint, the only attribute separating one runtime's real +// reading for a core from another runtime's zero for that same core. Two datapoints +// collapsed onto one identity and a core pinned at 75% reported 0. +// +// TestNeuronMultiRuntimePreFixConfigLosesData pins that failure mode, so the +// assertions below are known to discriminate rather than merely pass. + +package containerinsights + +import ( + "bytes" + "context" + "fmt" + "sort" + "testing" + "text/template" + + "github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbyattrsprocessor" + "github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/collector/component" + "go.opentelemetry.io/collector/confmap" + "go.opentelemetry.io/collector/consumer" + "go.opentelemetry.io/collector/consumer/consumertest" + "go.opentelemetry.io/collector/pdata/pmetric" + "go.opentelemetry.io/collector/processor" + "go.opentelemetry.io/collector/processor/processortest" + "gopkg.in/yaml.v3" +) + +const ( + neuronGroupByAttrs = "groupbyattrs/cw_k8s_ci_v0_neuron" + neuronPromote = "transform/cw_k8s_ci_v0_neuron_promote" + + testPod = "neuron-burn-abcde-12345" + testNamespace = "default" + testContainer = "burn" + tagA = "burn-core0" + tagB = "burn-core1" +) + +// renderedNeuronProcessors returns the processors block of the shipped neuron.yaml. +// Read from the embedded template rather than restated, so reverting the config +// fails this test. +func renderedNeuronProcessors(t *testing.T) map[string]any { + t.Helper() + tmpl, err := template.New("neuron").Parse(neuronYAML) + require.NoError(t, err) + var buf bytes.Buffer + require.NoError(t, tmpl.Execute(&buf, templateData{ + ClusterName: "test-cluster", + Region: "us-west-2", + CollectionInterval: "30s", + ScrapeTimeout: "10s", + NodeName: "test-node", + HostIP: "127.0.0.1", + })) + + var parsed map[string]any + require.NoError(t, yaml.Unmarshal(buf.Bytes(), &parsed)) + processors, ok := parsed["processors"].(map[string]any) + require.True(t, ok, "neuron.yaml has no processors block") + return processors +} + +func processorConfig(t *testing.T, processors map[string]any, name string) map[string]any { + t.Helper() + raw, ok := processors[name] + require.True(t, ok, "neuron.yaml is missing %s", name) + cfg, ok := raw.(map[string]any) + require.True(t, ok, "%s is not a mapping", name) + return cfg +} + +// buildChain wires groupbyattrs -> promote -> sink from the given configs. +func buildChain(t *testing.T, gbaCfg, promoteCfg map[string]any) (processor.Metrics, *consumertest.MetricsSink) { + t.Helper() + sink := new(consumertest.MetricsSink) + + promote := newProcessor(t, transformprocessor.NewFactory(), "transform", promoteCfg, sink) + gba := newProcessor(t, groupbyattrsprocessor.NewFactory(), "groupbyattrs", gbaCfg, promote) + return gba, sink +} + +func newProcessor( + t *testing.T, factory processor.Factory, typ string, cfg map[string]any, next consumer.Metrics, +) processor.Metrics { + t.Helper() + c := factory.CreateDefaultConfig() + require.NoError(t, confmap.NewFromStringMap(cfg).Unmarshal(c)) + p, err := factory.CreateMetrics( + context.Background(), processortest.NewNopSettings(component.MustNewType(typ)), c, next) + require.NoError(t, err) + require.NoError(t, p.Start(context.Background(), componenttestNopHost{})) + return p +} + +type componenttestNopHost struct{} + +func (componenttestNopHost) GetExtensions() map[component.ID]component.Component { return nil } + +// multiRuntimeMetrics is what neuron-monitor emits for 2 cores x 2 runtimes on one +// node: the owning pair real, the other zero. Pod identity is identical across all +// four, which is the collapse-prone case -- one pod holding both cores, as measured +// on solstice-gpu-test. +func multiRuntimeMetrics() pmetric.Metrics { + md := pmetric.NewMetrics() + sm := md.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty() + m := sm.Metrics().AppendEmpty() + m.SetName("neuroncore_utilization_ratio") + dps := m.SetEmptyGauge().DataPoints() + + add := func(core, tag string, value float64) { + dp := dps.AppendEmpty() + dp.SetDoubleValue(value) + a := dp.Attributes() + a.PutStr("neuroncore", core) + a.PutStr("runtime_tag", tag) + a.PutStr("k8s.pod.name", testPod) + a.PutStr("k8s.namespace.name", testNamespace) + a.PutStr("k8s.container.name", testContainer) + } + add("0", tagA, 75.1) + add("1", tagA, 0) + add("0", tagB, 0) + add("1", tagB, 75.3) + return md +} + +// observed flattens the sink into "core= tag= value=" triples, reading the +// tag from the RESOURCE (where the promote puts it) and the core from the datapoint. +func observed(t *testing.T, sink *consumertest.MetricsSink) []string { + t.Helper() + var out []string + for _, md := range sink.AllMetrics() { + for i := 0; i < md.ResourceMetrics().Len(); i++ { + rm := md.ResourceMetrics().At(i) + tag := "" + if v, ok := rm.Resource().Attributes().Get("aws.neuron.runtime.tag"); ok { + tag = v.Str() + } + for j := 0; j < rm.ScopeMetrics().Len(); j++ { + ms := rm.ScopeMetrics().At(j).Metrics() + for k := 0; k < ms.Len(); k++ { + dps := ms.At(k).Gauge().DataPoints() + for d := 0; d < dps.Len(); d++ { + dp := dps.At(d) + core, _ := dp.Attributes().Get("neuroncore") + out = append(out, fmt.Sprintf("core=%s tag=%s value=%.1f", + core.Str(), tag, dp.DoubleValue())) + } + } + } + } + } + sort.Strings(out) + return out +} + +func consume(t *testing.T, chain processor.Metrics, md pmetric.Metrics) { + t.Helper() + require.NoError(t, chain.ConsumeMetrics(context.Background(), md)) +} + +// TestNeuronMultiRuntimeKeepsEveryCorePerRuntime runs the SHIPPED config and +// requires all four (core, runtime) readings to survive with their real values. +func TestNeuronMultiRuntimeKeepsEveryCorePerRuntime(t *testing.T) { + processors := renderedNeuronProcessors(t) + chain, sink := buildChain(t, + processorConfig(t, processors, neuronGroupByAttrs), + processorConfig(t, processors, neuronPromote)) + consume(t, chain, multiRuntimeMetrics()) + + assert.Equal(t, []string{ + "core=0 tag=burn-core0 value=75.1", + "core=0 tag=burn-core1 value=0.0", + "core=1 tag=burn-core0 value=0.0", + "core=1 tag=burn-core1 value=75.3", + }, observed(t, sink), + "both busy cores must survive with their own runtime tag; a missing 75.x "+ + "reading is the per-core data-loss defect") +} + +// TestNeuronMultiRuntimeSeparatesRuntimesIntoResources pins the mechanism: one +// ResourceMetrics per runtime is what stops the datapoints colliding. +func TestNeuronMultiRuntimeSeparatesRuntimesIntoResources(t *testing.T) { + processors := renderedNeuronProcessors(t) + chain, sink := buildChain(t, + processorConfig(t, processors, neuronGroupByAttrs), + processorConfig(t, processors, neuronPromote)) + consume(t, chain, multiRuntimeMetrics()) + + tags := map[string]int{} + total := 0 + for _, md := range sink.AllMetrics() { + for i := 0; i < md.ResourceMetrics().Len(); i++ { + rm := md.ResourceMetrics().At(i) + v, ok := rm.Resource().Attributes().Get("aws.neuron.runtime.tag") + require.True(t, ok, "resource is missing aws.neuron.runtime.tag") + tags[v.Str()]++ + total++ + } + } + assert.Equal(t, 2, total, "expected one ResourceMetrics per runtime") + assert.Equal(t, map[string]int{tagA: 1, tagB: 1}, tags) +} + +// TestNeuronMultiRuntimePromotesPodIdentity guards the six promote statements +// removed by the fix: groupbyattrs already moves pod/namespace/container to the +// resource and deletes the datapoint copies, so dropping them changed nothing. +func TestNeuronMultiRuntimePromotesPodIdentity(t *testing.T) { + processors := renderedNeuronProcessors(t) + chain, sink := buildChain(t, + processorConfig(t, processors, neuronGroupByAttrs), + processorConfig(t, processors, neuronPromote)) + consume(t, chain, multiRuntimeMetrics()) + + for _, md := range sink.AllMetrics() { + for i := 0; i < md.ResourceMetrics().Len(); i++ { + rm := md.ResourceMetrics().At(i) + for _, kv := range []struct{ key, want string }{ + {"k8s.pod.name", testPod}, + {"k8s.namespace.name", testNamespace}, + {"k8s.container.name", testContainer}, + } { + v, ok := rm.Resource().Attributes().Get(kv.key) + require.True(t, ok, "resource is missing %s", kv.key) + assert.Equal(t, kv.want, v.Str()) + } + + for j := 0; j < rm.ScopeMetrics().Len(); j++ { + ms := rm.ScopeMetrics().At(j).Metrics() + for k := 0; k < ms.Len(); k++ { + dps := ms.At(k).Gauge().DataPoints() + for d := 0; d < dps.Len(); d++ { + attrs := dps.At(d).Attributes() + for _, key := range []string{ + "k8s.pod.name", "k8s.namespace.name", "k8s.container.name", "runtime_tag", + } { + _, present := attrs.Get(key) + assert.False(t, present, + "%s must be moved off the datapoint, not left on it", key) + } + } + } + } + } + } +} + +// TestNeuronMultiRuntimePreFixConfigLosesData is the negative control. It runs the +// PRE-FIX config -- groupbyattrs without the runtime_tag key, promote in +// `context: datapoint` -- and asserts the loss, so the tests above are known to +// discriminate. If this ever starts passing the collapse has stopped reproducing +// and these assertions no longer prove anything. +func TestNeuronMultiRuntimePreFixConfigLosesData(t *testing.T) { + preFixGBA := map[string]any{ + "keys": []any{"k8s.pod.name", "k8s.namespace.name", "k8s.container.name"}, + } + preFixPromote := map[string]any{ + "error_mode": "ignore", + "metric_statements": []any{ + map[string]any{ + "context": "datapoint", + "statements": []any{ + `set(resource.attributes["k8s.pod.name"], attributes["k8s.pod.name"]) where attributes["k8s.pod.name"] != nil`, + `set(resource.attributes["k8s.namespace.name"], attributes["k8s.namespace.name"]) where attributes["k8s.namespace.name"] != nil`, + `set(resource.attributes["k8s.container.name"], attributes["k8s.container.name"]) where attributes["k8s.container.name"] != nil`, + `set(resource.attributes["aws.neuron.runtime.tag"], attributes["runtime_tag"]) where attributes["runtime_tag"] != nil`, + `delete_key(attributes, "k8s.pod.name") where attributes["k8s.pod.name"] != nil`, + `delete_key(attributes, "k8s.namespace.name") where attributes["k8s.namespace.name"] != nil`, + `delete_key(attributes, "k8s.container.name") where attributes["k8s.container.name"] != nil`, + `delete_key(attributes, "runtime_tag") where attributes["runtime_tag"] != nil`, + }, + }, + }, + } + + chain, sink := buildChain(t, preFixGBA, preFixPromote) + consume(t, chain, multiRuntimeMetrics()) + got := observed(t, sink) + + // One resource, one surviving tag: the four datapoints keep only two identities. + assert.Len(t, got, 4, "the pre-fix chain still emits four datapoints") + tags := map[string]struct{}{} + for _, md := range sink.AllMetrics() { + for i := 0; i < md.ResourceMetrics().Len(); i++ { + if v, ok := md.ResourceMetrics().At(i).Resource().Attributes().Get("aws.neuron.runtime.tag"); ok { + tags[v.Str()] = struct{}{} + } + } + } + assert.Len(t, tags, 1, + "pre-fix, all runtimes collapse onto ONE resource tag; got %v", tags) + + // With one tag on the resource, (core, tag) is no longer unique: each core + // appears twice, once with its real value and once with another runtime's zero. + // Downstream that is a duplicate identity and one value is lost. + perCore := map[string][]float64{} + for _, md := range sink.AllMetrics() { + for i := 0; i < md.ResourceMetrics().Len(); i++ { + rm := md.ResourceMetrics().At(i) + for j := 0; j < rm.ScopeMetrics().Len(); j++ { + ms := rm.ScopeMetrics().At(j).Metrics() + for k := 0; k < ms.Len(); k++ { + dps := ms.At(k).Gauge().DataPoints() + for d := 0; d < dps.Len(); d++ { + dp := dps.At(d) + core, _ := dp.Attributes().Get("neuroncore") + perCore[core.Str()] = append(perCore[core.Str()], dp.DoubleValue()) + } + } + } + } + } + for core, values := range perCore { + assert.Len(t, values, 2, + "pre-fix, core %s carries two datapoints under one identity %v — whichever "+ + "arrives last wins, and for one core that is a legitimate-looking 0", + core, values) + } +} From 76f9b94ff845af0a6360441bff20c0528a774e83 Mon Sep 17 00:00:00 2001 From: Spandan Naik Date: Mon, 31 Aug 2026 14:53:41 +0000 Subject: [PATCH 3/3] test(neuron): frame the multi-runtime tests around the invariant, not the bug Review feedback on #2263: the file's preface read as a post-mortem of the fix rather than something a future reader could use. Lead with the invariant every (core, runtime) reading must satisfy; keep the collapsed-config mechanism on the negative control, which is the one test that is genuinely about that shape. Rename PreFixConfigLosesData -> CollapsedConfigLosesData so the name describes the configuration rather than when it existed. No assertions changed. Also drop an internal cluster name from a comment. --- .../neuron_multiruntime_test.go | 67 ++++++++++--------- 1 file changed, 34 insertions(+), 33 deletions(-) diff --git a/translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron_multiruntime_test.go b/translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron_multiruntime_test.go index ca59ddfff6..e9ea43662a 100644 --- a/translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron_multiruntime_test.go +++ b/translator/translate/otel/pipeline/opentelemetry/containerinsights/neuron_multiruntime_test.go @@ -1,22 +1,19 @@ // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: MIT -// Per-core Neuron data loss on multi-runtime nodes. +// Per-core Neuron attribution on multi-runtime nodes. // -// The shipped neuron.yaml is run for real here -- groupbyattrs then the promote -// transform, built from the embedded template -- against synthetic metrics in the -// shape neuron-monitor emits when two Neuron runtimes share a node: every core -// reported by every runtime, with the non-owning pairs at zero. +// Invariant: when several Neuron runtimes share a node, every (core, runtime) +// reading must reach the exporter under its own identity. neuron-monitor reports +// every core from every runtime -- the owning pair real, the rest zero -- so +// runtime_tag is the only attribute separating a core's real reading from another +// runtime's zero for that same core. Stop separating them and two datapoints share +// an identity, the last write wins, and for one core that is a plausible-looking 0. // -// The defect: the promote ran in `context: datapoint` while writing -// resource.attributes. Resource attributes are per-ResourceMetrics, so the -// statement ran once per datapoint and the last write won; it then deleted -// runtime_tag from the datapoint, the only attribute separating one runtime's real -// reading for a core from another runtime's zero for that same core. Two datapoints -// collapsed onto one identity and a core pinned at 75% reported 0. -// -// TestNeuronMultiRuntimePreFixConfigLosesData pins that failure mode, so the -// assertions below are known to discriminate rather than merely pass. +// These run the shipped neuron.yaml -- groupbyattrs then the promote transform, +// rendered from the embedded template rather than restated, so a config change that +// breaks the invariant fails here. TestNeuronMultiRuntimeCollapsedConfigLosesData +// runs a config that violates it, proving the assertions discriminate. package containerinsights @@ -54,8 +51,6 @@ const ( ) // renderedNeuronProcessors returns the processors block of the shipped neuron.yaml. -// Read from the embedded template rather than restated, so reverting the config -// fails this test. func renderedNeuronProcessors(t *testing.T) map[string]any { t.Helper() tmpl, err := template.New("neuron").Parse(neuronYAML) @@ -115,8 +110,8 @@ func (componenttestNopHost) GetExtensions() map[component.ID]component.Component // multiRuntimeMetrics is what neuron-monitor emits for 2 cores x 2 runtimes on one // node: the owning pair real, the other zero. Pod identity is identical across all -// four, which is the collapse-prone case -- one pod holding both cores, as measured -// on solstice-gpu-test. +// four -- one pod holding both cores, which occurs in practice and is the +// collapse-prone case, since pod identity alone cannot separate the datapoints. func multiRuntimeMetrics() pmetric.Metrics { md := pmetric.NewMetrics() sm := md.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty() @@ -176,8 +171,8 @@ func consume(t *testing.T, chain processor.Metrics, md pmetric.Metrics) { require.NoError(t, chain.ConsumeMetrics(context.Background(), md)) } -// TestNeuronMultiRuntimeKeepsEveryCorePerRuntime runs the SHIPPED config and -// requires all four (core, runtime) readings to survive with their real values. +// TestNeuronMultiRuntimeKeepsEveryCorePerRuntime requires all four (core, runtime) +// readings to survive the shipped config with their real values. func TestNeuronMultiRuntimeKeepsEveryCorePerRuntime(t *testing.T) { processors := renderedNeuronProcessors(t) chain, sink := buildChain(t, @@ -192,7 +187,7 @@ func TestNeuronMultiRuntimeKeepsEveryCorePerRuntime(t *testing.T) { "core=1 tag=burn-core1 value=75.3", }, observed(t, sink), "both busy cores must survive with their own runtime tag; a missing 75.x "+ - "reading is the per-core data-loss defect") + "reading means one runtime's datapoint was shadowed by another's zero") } // TestNeuronMultiRuntimeSeparatesRuntimesIntoResources pins the mechanism: one @@ -219,9 +214,9 @@ func TestNeuronMultiRuntimeSeparatesRuntimesIntoResources(t *testing.T) { assert.Equal(t, map[string]int{tagA: 1, tagB: 1}, tags) } -// TestNeuronMultiRuntimePromotesPodIdentity guards the six promote statements -// removed by the fix: groupbyattrs already moves pod/namespace/container to the -// resource and deletes the datapoint copies, so dropping them changed nothing. +// TestNeuronMultiRuntimePromotesPodIdentity requires pod identity to land on the +// resource and NOT remain on the datapoint. groupbyattrs moves its grouping keys +// rather than copying them, so nothing downstream needs to re-promote them. func TestNeuronMultiRuntimePromotesPodIdentity(t *testing.T) { processors := renderedNeuronProcessors(t) chain, sink := buildChain(t, @@ -262,16 +257,22 @@ func TestNeuronMultiRuntimePromotesPodIdentity(t *testing.T) { } } -// TestNeuronMultiRuntimePreFixConfigLosesData is the negative control. It runs the -// PRE-FIX config -- groupbyattrs without the runtime_tag key, promote in -// `context: datapoint` -- and asserts the loss, so the tests above are known to -// discriminate. If this ever starts passing the collapse has stopped reproducing -// and these assertions no longer prove anything. -func TestNeuronMultiRuntimePreFixConfigLosesData(t *testing.T) { - preFixGBA := map[string]any{ +// TestNeuronMultiRuntimeCollapsedConfigLosesData is the negative control: it runs a +// config that drops the runtime dimension and asserts the loss, so the tests above +// are known to discriminate rather than merely pass. If it ever starts passing, the +// collapse has stopped reproducing and those assertions prove nothing. +// +// The collapse needs both halves. groupbyattrs omits the runtime_tag key, so the +// runtimes are not split into separate resources; the promote then runs in +// `context: datapoint` while writing resource.attributes, which are +// per-ResourceMetrics -- so the statement executes once per datapoint and the last +// write wins -- and deletes runtime_tag from the datapoint. This is the shape the +// agent shipped with. +func TestNeuronMultiRuntimeCollapsedConfigLosesData(t *testing.T) { + collapsedGBA := map[string]any{ "keys": []any{"k8s.pod.name", "k8s.namespace.name", "k8s.container.name"}, } - preFixPromote := map[string]any{ + collapsedPromote := map[string]any{ "error_mode": "ignore", "metric_statements": []any{ map[string]any{ @@ -290,7 +291,7 @@ func TestNeuronMultiRuntimePreFixConfigLosesData(t *testing.T) { }, } - chain, sink := buildChain(t, preFixGBA, preFixPromote) + chain, sink := buildChain(t, collapsedGBA, collapsedPromote) consume(t, chain, multiRuntimeMetrics()) got := observed(t, sink)