[processor/spanpruning] Fix spans with the same name not being aggregated (#47770)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
If spans at two different depths each have the same name, kind, and
status (such as two HTTP GET spans from different services) then highest
in the tree will overwrite the aggregation. This causes the group to not
be aggregated creating orphan spans.

Adding the depth to the key when building the parent group key fixes
this problem as each level will be considered on its own.

<!--Describe what testing was performed and which tests were added.-->
#### Testing
Unit test added and I have also verified the behavior on some internal
traces that exposed this bug.

Author: csmarchbanks

.chloggen/fix-parent-key-collision.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: processor/spanpruning
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix bug in span pruning where spans with the same name at different depths could become orphaned instead of deleted.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [47770]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

processor/spanpruningprocessor/grouping.go

@@ -133,12 +133,15 @@ func writeAttributeSliceKey(builder *strings.Builder, value pcommon.Slice) {
 }
 
 // buildParentGroupKey constructs a parent grouping key from name and status
-// only; attributes are intentionally excluded for parent aggregation.
-func (*spanPruningProcessor) buildParentGroupKey(span ptrace.Span) string {
+// only; attributes are intentionally excluded for parent aggregation. Depth is
+// required to avoid duplicate names and status entries overwriting each other.
+func (*spanPruningProcessor) buildParentGroupKey(span ptrace.Span, depth int) string {
 	builder := builderPool.Get().(*strings.Builder)
 	builder.Reset()
 	defer builderPool.Put(builder)
 
+	builder.WriteString(strconv.Itoa(depth))
+	builder.WriteByte('|')
 	builder.WriteString(span.Name())
 	builder.WriteString("|kind=")
 	builder.WriteString(span.Kind().String())

processor/spanpruningprocessor/processor.go

@@ -224,7 +224,7 @@ func (p *spanPruningProcessor) analyzeAggregationsWithTree(tree *traceTree) map[
 		// Group parent candidates by name + status
 		parentGroups := make(map[string][]*spanNode)
 		for _, node := range eligibleParents {
-			parentKey := p.buildParentGroupKey(node.span)
+			parentKey := p.buildParentGroupKey(node.span, depth)
 			parentGroups[parentKey] = append(parentGroups[parentKey], node)
 		}
 

processor/spanpruningprocessor/processor_test.go

@@ -1738,6 +1738,153 @@ func TestLeafSpanPruning_TraceStateGrouping_EmptyTraceState(t *testing.T) {
 	assert.Equal(t, int64(3), spanCount.Int())
 }
 
+// TestParentKeyCollisionAcrossDepths verifies that when the same span name
+// appears at multiple tree depths (causing buildParentGroupKey collisions),
+// nodes from overwritten groups are not incorrectly removed. This is a
+// regression test for a bug where the depth-2 parent group overwrites the
+// depth-1 entry in aggregationGroups, leaving depth-1 nodes marked for
+// removal but absent from the final plan.
+//
+// Trace structure (3 copies to meet MinSpansToAggregate=2 for parents):
+//
+//	root
+//	├── svc [depth-2 parent candidate]
+//	│   ├── svc [depth-1 parent candidate, SAME name as depth-2]
+//	│   │   ├── SELECT (leaf)
+//	│   │   └── SELECT (leaf)
+//	│   └── svc
+//	│       ├── SELECT (leaf)
+//	│       └── SELECT (leaf)
+//	├── svc
+//	│   ├── svc
+//	│   │   ├── SELECT (leaf)
+//	│   │   └── SELECT (leaf)
+//	│   └── svc
+//	│       ├── SELECT (leaf)
+//	│       └── SELECT (leaf)
+//	└── svc
+//	    ├── svc
+//	    │   ├── SELECT (leaf)
+//	    │   └── SELECT (leaf)
+//	    └── svc
+//	        ├── SELECT (leaf)
+//	        └── SELECT (leaf)
+//
+// Expected result (22 spans → 4 spans):
+//
+//	root
+//	├── summary(svc, aggregates 3 outer svc spans)
+//	│   └── summary(svc, aggregates 6 inner svc spans)
+//	│       └── summary(SELECT, aggregates 12 leaf spans)
+func TestParentKeyCollisionAcrossDepths(t *testing.T) {
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig().(*Config)
+	cfg.MinSpansToAggregate = 2
+	cfg.MaxParentDepth = -1
+
+	tp, err := factory.CreateTraces(t.Context(), processortest.NewNopSettings(metadata.Type), cfg, consumertest.NewNop())
+	require.NoError(t, err)
+
+	td := createTestTraceWithParentKeyCollision(t)
+
+	// Before: 1 root + 3 outer-svc + 6 inner-svc + 12 SELECT = 22 spans
+	originalSpanCount := countSpans(td)
+	assert.Equal(t, 22, originalSpanCount)
+
+	err = tp.ConsumeTraces(t.Context(), td)
+	require.NoError(t, err)
+
+	finalSpanCount := countSpans(td)
+
+	// Verify every non-summary span that remains is NOT an orphan: its parent
+	// must either be another span in the output or it must be the root.
+	remainingByID := make(map[pcommon.SpanID]ptrace.Span)
+	rss := td.ResourceSpans()
+	for i := 0; i < rss.Len(); i++ {
+		ilss := rss.At(i).ScopeSpans()
+		for j := 0; j < ilss.Len(); j++ {
+			spans := ilss.At(j).Spans()
+			for k := 0; k < spans.Len(); k++ {
+				span := spans.At(k)
+				remainingByID[span.SpanID()] = span
+			}
+		}
+	}
+	for id, span := range remainingByID {
+		parentID := span.ParentSpanID()
+		if parentID.IsEmpty() {
+			continue // root span
+		}
+		_, parentExists := remainingByID[parentID]
+		assert.True(t, parentExists, "span %s (name=%s) has dangling parent %s — parent was removed but span was kept",
+			id, span.Name(), parentID)
+	}
+
+	// With the bug, depth-1 "svc" nodes (6 spans) get incorrectly removed,
+	// dropping the count too low. The expected result: root + 3 summaries.
+	t.Logf("original=%d final=%d", originalSpanCount, finalSpanCount)
+	assert.Equal(t, 4, finalSpanCount,
+		"expected root + 3 summary spans (SELECT, inner svc, outer svc)")
+}
+
+func createTestTraceWithParentKeyCollision(t *testing.T) ptrace.Traces {
+	t.Helper()
+	td := ptrace.NewTraces()
+	rs := td.ResourceSpans().AppendEmpty()
+	ss := rs.ScopeSpans().AppendEmpty()
+
+	traceID := pcommon.TraceID([16]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16})
+	rootSpanID := pcommon.SpanID([8]byte{1, 0, 0, 0, 0, 0, 0, 0})
+
+	root := ss.Spans().AppendEmpty()
+	root.SetTraceID(traceID)
+	root.SetSpanID(rootSpanID)
+	root.SetName("root")
+
+	spanIDCounter := byte(2)
+	nextID := func() pcommon.SpanID {
+		id := pcommon.SpanID([8]byte{spanIDCounter, 0, 0, 0, 0, 0, 0, 0})
+		spanIDCounter++
+		return id
+	}
+
+	// 3 outer "svc" spans (same name/kind/status → same parent group key)
+	for range 3 {
+		outerID := nextID()
+		outer := ss.Spans().AppendEmpty()
+		outer.SetTraceID(traceID)
+		outer.SetSpanID(outerID)
+		outer.SetParentSpanID(rootSpanID)
+		outer.SetName("svc")
+		outer.Status().SetCode(ptrace.StatusCodeOk)
+
+		// Each outer has 2 inner "svc" spans (same name → key collision with outer)
+		for range 2 {
+			innerID := nextID()
+			inner := ss.Spans().AppendEmpty()
+			inner.SetTraceID(traceID)
+			inner.SetSpanID(innerID)
+			inner.SetParentSpanID(outerID)
+			inner.SetName("svc")
+			inner.Status().SetCode(ptrace.StatusCodeOk)
+
+			// Each inner has 2 leaf SELECT spans
+			for range 2 {
+				leaf := ss.Spans().AppendEmpty()
+				leaf.SetTraceID(traceID)
+				leaf.SetSpanID(nextID())
+				leaf.SetParentSpanID(innerID)
+				leaf.SetName("SELECT")
+				leaf.Status().SetCode(ptrace.StatusCodeOk)
+				leaf.SetStartTimestamp(pcommon.Timestamp(1000000000))
+				leaf.SetEndTimestamp(pcommon.Timestamp(1000000100))
+			}
+		}
+	}
+
+	return td
+}
+
 // Helper functions for TraceState tests
 
 func createTestTraceWithSameTraceState(t *testing.T, traceState string) ptrace.Traces {