[processor/spanpruning] Add histogram support for latency distribution on aggregated summary spans (#47604)

#### Description

The span pruning processor currently emits min/max/avg/total duration
statistics on summary spans, but these aggregates hide the shape of the
latency distribution. A flat average can mask bimodal distributions and
tail latency problems that are critical for debugging.

This PR adds configurable cumulative histogram buckets to summary spans,
giving users latency distribution visibility without requiring a
separate metrics pipeline. Bucket boundaries default to the
OpenTelemetry SDK explicit bucket histogram boundaries (`[5ms, 10ms,
25ms, 50ms, 100ms, 250ms, 500ms, 1s, 2.5s, 5s, 10s]`) and can be
customized or disabled entirely by setting
`aggregation_histogram_buckets` to an empty list.

Summary spans gain two new slice attributes:
- `<prefix>histogram_bucket_bounds_s` -- upper bounds in seconds
(excludes +Inf)
- `<prefix>histogram_bucket_counts` -- cumulative counts per bucket
(includes +Inf)

Cumulative semantics were chosen to match Prometheus/OpenTelemetry
histogram conventions, where each bucket count includes all observations
at or below that boundary.

#### Testing

- Unit tests for cumulative bucket counting logic and the
disabled-histogram path (`stats_test.go`)
- Config validation tests covering valid buckets, negative values,
unsorted values, and empty list (`config_test.go`)
- Integration tests exercising the full processor pipeline with
histogram attributes enabled and disabled, asserting exact bucket bounds
and counts on output summary spans (`processor_test.go`)
- YAML unmarshal test verifying custom bucket configuration round-trips
through `TestLoadConfig`

#### Documentation

- Added `aggregation_histogram_buckets` to the configuration table in
README
- Added `histogram_bucket_bounds_s` and `histogram_bucket_counts` to the
summary span attributes table
- Added a "Histogram Buckets" section with a worked example showing
cumulative bucket semantics

Author: portertech

.chloggen/spanpruning-histogram-buckets.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: enhancement
+
+# 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: Add cumulative histogram bucket support for latency distribution on aggregated summary spans
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [47277]
+
+# (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/README.md

@@ -64,6 +64,11 @@ processors:
     # Prefix for aggregation statistics attributes
     # Default: "aggregation."
     aggregation_attribute_prefix: "batch."
+
+    # Upper bounds for histogram buckets (latency distribution)
+    # Default: [5ms, 10ms, 25ms, 50ms, 100ms, 250ms, 500ms, 1s, 2.5s, 5s, 10s]
+    # Set to empty list to disable histogram attributes
+    aggregation_histogram_buckets: [10ms, 50ms, 100ms, 500ms, 1s]
 ```
 
 ## Configuration Options
@@ -74,6 +79,7 @@ processors:
 | `min_spans_to_aggregate` | int | 5 | Minimum group size before aggregation occurs |
 | `max_parent_depth` | int | 1 | Max depth of parent aggregation (0=none, -1=unlimited) |
 | `aggregation_attribute_prefix` | string | "aggregation." | Prefix for aggregation statistics attributes |
+| `aggregation_histogram_buckets` | []time.Duration | `[5ms, 10ms, 25ms, 50ms, 100ms, 250ms, 500ms, 1s, 2.5s, 5s, 10s]` | Upper bounds for latency histogram buckets |
 
 ### Glob Pattern Support
 
@@ -129,6 +135,20 @@ The following attributes are added to the summary span (shown with default `aggr
 | `<prefix>duration_max_ns` | int64 | Maximum duration in nanoseconds |
 | `<prefix>duration_avg_ns` | int64 | Average duration in nanoseconds |
 | `<prefix>duration_total_ns` | int64 | Total duration in nanoseconds |
+| `<prefix>histogram_bucket_bounds_s` | []float64 | Bucket upper bounds in seconds (excludes +Inf) |
+| `<prefix>histogram_bucket_counts` | []int64 | Cumulative count per bucket (includes +Inf bucket) |
+
+### Histogram Buckets
+
+When `aggregation_histogram_buckets` is configured, summary spans include latency distribution data as cumulative histogram buckets. Cumulative means each bucket count includes all spans with duration less than or equal to that bucket boundary.
+
+**Worked example** with buckets `[10ms, 50ms, 100ms]` and span durations `[5ms, 15ms, 25ms, 75ms, 150ms]`:
+- `histogram_bucket_bounds_s`: `[0.01, 0.05, 0.1]`
+- `histogram_bucket_counts`: `[1, 3, 4, 5]`
+  - Bucket 0 (<=10ms): 1 span (5ms)
+  - Bucket 1 (<=50ms): 3 spans (5ms, 15ms, 25ms)
+  - Bucket 2 (<=100ms): 4 spans (5ms, 15ms, 25ms, 75ms)
+  - Bucket 3 (+Inf): 5 spans (all spans)
 
 ## Pipeline Placement
 

processor/spanpruningprocessor/aggregation.go

@@ -7,6 +7,7 @@ import (
 	"encoding/binary"
 	"math/rand/v2"
 	"sort"
+	"time"
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/ptrace"
@@ -174,5 +175,20 @@ func (p *spanPruningProcessor) createSummarySpanWithParent(group aggregationGrou
 		newSpan.Attributes().PutInt(prefix+"duration_avg_ns", int64(data.sumDuration)/data.count)
 	}
 
+	// Add histogram attributes if enabled.
+	if len(p.config.AggregationHistogramBuckets) > 0 {
+		// Add bucket bounds in seconds.
+		bucketBoundsSlice := newSpan.Attributes().PutEmptySlice(prefix + "histogram_bucket_bounds_s")
+		for _, bucket := range p.config.AggregationHistogramBuckets {
+			bucketBoundsSlice.AppendEmpty().SetDouble(float64(bucket) / float64(time.Second))
+		}
+
+		// Add cumulative bucket counts.
+		bucketCountsSlice := newSpan.Attributes().PutEmptySlice(prefix + "histogram_bucket_counts")
+		for _, count := range data.bucketCounts {
+			bucketCountsSlice.AppendEmpty().SetInt(count)
+		}
+	}
+
 	return newSpan
 }

processor/spanpruningprocessor/config.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/gobwas/glob"
 	"go.opentelemetry.io/collector/component"
@@ -40,6 +41,10 @@ type Config struct {
 	// are added to summary spans.
 	// Default: "aggregation."
 	AggregationAttributePrefix string `mapstructure:"aggregation_attribute_prefix"`
+
+	// AggregationHistogramBuckets lists cumulative histogram bucket upper bounds
+	// for latency tracking on aggregated spans. Empty slice disables histograms.
+	AggregationHistogramBuckets []time.Duration `mapstructure:"aggregation_histogram_buckets"`
 }
 
 var _ component.Config = (*Config)(nil)
@@ -75,5 +80,15 @@ func (cfg *Config) Validate() error {
 		}
 	}
 
+	// Validate histogram buckets
+	for i, bucket := range cfg.AggregationHistogramBuckets {
+		if bucket <= 0 {
+			return errors.New("histogram bucket values must be positive")
+		}
+		if i > 0 && bucket <= cfg.AggregationHistogramBuckets[i-1] {
+			return errors.New("histogram buckets must be sorted in ascending order")
+		}
+	}
+
 	return nil
 }

processor/spanpruningprocessor/config_test.go

@@ -6,6 +6,7 @@ package spanpruningprocessor
 import (
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -15,6 +16,28 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanpruningprocessor/internal/metadata"
 )
 
+var defaultHistogramBuckets = []time.Duration{
+	5 * time.Millisecond,
+	10 * time.Millisecond,
+	25 * time.Millisecond,
+	50 * time.Millisecond,
+	100 * time.Millisecond,
+	250 * time.Millisecond,
+	500 * time.Millisecond,
+	time.Second,
+	2500 * time.Millisecond,
+	5 * time.Second,
+	10 * time.Second,
+}
+
+var customHistogramBuckets = []time.Duration{
+	10 * time.Millisecond,
+	50 * time.Millisecond,
+	100 * time.Millisecond,
+	500 * time.Millisecond,
+	time.Second,
+}
+
 func TestLoadConfig(t *testing.T) {
 	t.Parallel()
 
@@ -26,19 +49,21 @@ func TestLoadConfig(t *testing.T) {
 		{
 			id: component.NewIDWithName(metadata.Type, ""),
 			expected: &Config{
-				GroupByAttributes:          []string{"db.operation"},
-				MinSpansToAggregate:        5,
-				MaxParentDepth:             1,
-				AggregationAttributePrefix: "aggregation.",
+				GroupByAttributes:           []string{"db.operation"},
+				MinSpansToAggregate:         5,
+				MaxParentDepth:              1,
+				AggregationAttributePrefix:  "aggregation.",
+				AggregationHistogramBuckets: defaultHistogramBuckets,
 			},
 		},
 		{
 			id: component.NewIDWithName(metadata.Type, "custom"),
 			expected: &Config{
-				GroupByAttributes:          []string{"db.operation", "db.name"},
-				MinSpansToAggregate:        3,
-				MaxParentDepth:             1,
-				AggregationAttributePrefix: "batch.",
+				GroupByAttributes:           []string{"db.operation", "db.name"},
+				MinSpansToAggregate:         3,
+				MaxParentDepth:              1,
+				AggregationAttributePrefix:  "batch.",
+				AggregationHistogramBuckets: customHistogramBuckets,
 			},
 		},
 	}
@@ -155,6 +180,56 @@ func TestConfig_Validate(t *testing.T) {
 			},
 			expectError: false,
 		},
+		{
+			name: "valid histogram buckets",
+			config: &Config{
+				MinSpansToAggregate:        2,
+				AggregationAttributePrefix: "aggregation.",
+				GroupByAttributes:          []string{"db.operation"},
+				AggregationHistogramBuckets: []time.Duration{
+					10 * time.Millisecond,
+					50 * time.Millisecond,
+					100 * time.Millisecond,
+				},
+			},
+			expectError: false,
+		},
+		{
+			name: "negative histogram bucket value",
+			config: &Config{
+				MinSpansToAggregate:        2,
+				AggregationAttributePrefix: "aggregation.",
+				GroupByAttributes:          []string{"db.operation"},
+				AggregationHistogramBuckets: []time.Duration{
+					10 * time.Millisecond,
+					-1 * time.Millisecond,
+				},
+			},
+			expectError: true,
+		},
+		{
+			name: "unsorted histogram buckets",
+			config: &Config{
+				MinSpansToAggregate:        2,
+				AggregationAttributePrefix: "aggregation.",
+				GroupByAttributes:          []string{"db.operation"},
+				AggregationHistogramBuckets: []time.Duration{
+					100 * time.Millisecond,
+					50 * time.Millisecond,
+				},
+			},
+			expectError: true,
+		},
+		{
+			name: "empty histogram buckets",
+			config: &Config{
+				MinSpansToAggregate:         2,
+				AggregationAttributePrefix:  "aggregation.",
+				GroupByAttributes:           []string{"db.operation"},
+				AggregationHistogramBuckets: []time.Duration{},
+			},
+			expectError: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

processor/spanpruningprocessor/factory.go

@@ -5,6 +5,7 @@ package spanpruningprocessor // import "github.com/open-telemetry/opentelemetry-
 
 import (
 	"context"
+	"time"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
@@ -29,6 +30,19 @@ func createDefaultConfig() component.Config {
 		MinSpansToAggregate:        5,
 		MaxParentDepth:             1,
 		AggregationAttributePrefix: "aggregation.",
+		AggregationHistogramBuckets: []time.Duration{
+			5 * time.Millisecond,
+			10 * time.Millisecond,
+			25 * time.Millisecond,
+			50 * time.Millisecond,
+			100 * time.Millisecond,
+			250 * time.Millisecond,
+			500 * time.Millisecond,
+			time.Second,
+			2500 * time.Millisecond,
+			5 * time.Second,
+			10 * time.Second,
+		},
 	}
 }
 

processor/spanpruningprocessor/processor_test.go

@@ -6,6 +6,7 @@ package spanpruningprocessor
 import (
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -246,6 +247,88 @@ func TestLeafSpanPruning_DurationStats(t *testing.T) {
 	assert.Equal(t, int64(600), totalDuration.Int())
 }
 
+func TestLeafSpanPruningProcessorWithHistogram(t *testing.T) {
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig().(*Config)
+	cfg.MinSpansToAggregate = 2
+	cfg.AggregationHistogramBuckets = []time.Duration{
+		10 * time.Millisecond,
+		50 * time.Millisecond,
+		100 * time.Millisecond,
+	}
+
+	tp, err := factory.CreateTraces(t.Context(), processortest.NewNopSettings(metadata.Type), cfg, consumertest.NewNop())
+	require.NoError(t, err)
+
+	td := createTestTraceWithKnownDurations(t, []int64{
+		int64(5 * time.Millisecond),
+		int64(15 * time.Millisecond),
+		int64(25 * time.Millisecond),
+		int64(75 * time.Millisecond),
+		int64(150 * time.Millisecond),
+	})
+
+	err = tp.ConsumeTraces(t.Context(), td)
+	require.NoError(t, err)
+
+	assert.Equal(t, 2, countSpans(td))
+
+	summarySpan, found := findSummarySpan(td)
+	require.True(t, found)
+
+	attrs := summarySpan.Attributes()
+
+	bounds, exists := attrs.Get("aggregation.histogram_bucket_bounds_s")
+	require.True(t, exists)
+	require.Equal(t, 3, bounds.Slice().Len())
+	expectedBounds := []float64{0.01, 0.05, 0.1}
+	for i, expected := range expectedBounds {
+		assert.InDelta(t, expected, bounds.Slice().At(i).Double(), 1e-9)
+	}
+
+	counts, exists := attrs.Get("aggregation.histogram_bucket_counts")
+	require.True(t, exists)
+	expectedCounts := []int64{1, 3, 4, 5}
+	require.Equal(t, len(expectedCounts), counts.Slice().Len())
+	for i, expected := range expectedCounts {
+		assert.Equal(t, expected, counts.Slice().At(i).Int())
+	}
+}
+
+func TestLeafSpanPruningProcessorWithHistogramDisabled(t *testing.T) {
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig().(*Config)
+	cfg.MinSpansToAggregate = 2
+	cfg.AggregationHistogramBuckets = []time.Duration{}
+
+	tp, err := factory.CreateTraces(t.Context(), processortest.NewNopSettings(metadata.Type), cfg, consumertest.NewNop())
+	require.NoError(t, err)
+
+	td := createTestTraceWithKnownDurations(t, []int64{
+		int64(5 * time.Millisecond),
+		int64(15 * time.Millisecond),
+		int64(25 * time.Millisecond),
+		int64(75 * time.Millisecond),
+		int64(150 * time.Millisecond),
+	})
+
+	err = tp.ConsumeTraces(t.Context(), td)
+	require.NoError(t, err)
+
+	assert.Equal(t, 2, countSpans(td))
+
+	summarySpan, found := findSummarySpan(td)
+	require.True(t, found)
+
+	attrs := summarySpan.Attributes()
+
+	_, exists := attrs.Get("aggregation.histogram_bucket_bounds_s")
+	assert.False(t, exists)
+
+	_, exists = attrs.Get("aggregation.histogram_bucket_counts")
+	assert.False(t, exists)
+}
+
 func TestLeafSpanPruning_GroupByNonStringAttributes(t *testing.T) {
 	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig().(*Config)