open-telemetry · Sep 30, 2022 · Nov 21, 2023
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
diff --git a/.github/workflows/auto-assign-owners.yml b/.github/workflows/auto-assign-owners.yml
diff --git a/.github/workflows/build-and-test-windows.yml b/.github/workflows/build-and-test-windows.yml
diff --git a/.github/workflows/build-and-test.yml b/.github/workflows/build-and-test.yml
diff --git a/.github/workflows/check-links.yaml b/.github/workflows/check-links.yaml
diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml
diff --git a/.github/workflows/load-tests.yml b/.github/workflows/load-tests.yml
diff --git a/.github/workflows/prometheus-compliance-tests.yml b/.github/workflows/prometheus-compliance-tests.yml
diff --git a/.github/workflows/scripts/dependabot-pr.sh b/.github/workflows/scripts/dependabot-pr.sh
diff --git a/.github/workflows/tracegen.yml b/.github/workflows/tracegen.yml
diff --git a/Makefile b/Makefile
@@ -162,23 +162,23 @@ gendependabot:
 	@echo "  - package-ecosystem: \"github-actions\"" >> ${DEPENDABOT_PATH}
 	@echo "    directory: \"/\"" >> ${DEPENDABOT_PATH}
 	@echo "    schedule:" >> ${DEPENDABOT_PATH}
-	@echo "      interval: \"weekly\"" >> ${DEPENDABOT_PATH}
+	@echo "      interval: \"monthly\"" >> ${DEPENDABOT_PATH}
 	@echo "Add entry for \"/\" docker"
 	@echo "  - package-ecosystem: \"docker\"" >> ${DEPENDABOT_PATH}
 	@echo "    directory: \"/\"" >> ${DEPENDABOT_PATH}
 	@echo "    schedule:" >> ${DEPENDABOT_PATH}
-	@echo "      interval: \"weekly\"" >> ${DEPENDABOT_PATH}
+	@echo "      interval: \"monthly\"" >> ${DEPENDABOT_PATH}
 	@echo "Add entry for \"/\" gomod"
 	@echo "  - package-ecosystem: \"gomod\"" >> ${DEPENDABOT_PATH}
 	@echo "    directory: \"/\"" >> ${DEPENDABOT_PATH}
 	@echo "    schedule:" >> ${DEPENDABOT_PATH}
-	@echo "      interval: \"weekly\"" >> ${DEPENDABOT_PATH}
+	@echo "      interval: \"monthly\"" >> ${DEPENDABOT_PATH}
 	@set -e; for dir in $(NONROOT_MODS); do \
 		echo "Add entry for \"$${dir:1}\""; \
 		echo "  - package-ecosystem: \"gomod\"" >> ${DEPENDABOT_PATH}; \
 		echo "    directory: \"$${dir:1}\"" >> ${DEPENDABOT_PATH}; \
 		echo "    schedule:" >> ${DEPENDABOT_PATH}; \
-		echo "      interval: \"weekly\"" >> ${DEPENDABOT_PATH}; \
+		echo "      interval: \"monthly\"" >> ${DEPENDABOT_PATH}; \
 	done
 
 # Define a delegation target for each module

diff --git a/processor/probabilisticsamplerprocessor/config.go b/processor/probabilisticsamplerprocessor/config.go
@@ -16,12 +16,16 @@ package probabilisticsamplerprocessor // import "github.com/open-telemetry/opent
 
 import (
 	"go.opentelemetry.io/collector/config"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterconfig"
 )
 
 // Config has the configuration guiding the trace sampler processor.
 type Config struct {
 	config.ProcessorSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct
 
+	filterconfig.MatchConfig `mapstructure:",squash"`
+
 	// SamplingPercentage is the percentage rate at which traces are going to be sampled. Defaults to zero, i.e.: no sample.
 	// Values greater or equal 100 are treated as "sample all traces".
 	SamplingPercentage float32 `mapstructure:"sampling_percentage"`

diff --git a/processor/probabilisticsamplerprocessor/go.mod b/processor/probabilisticsamplerprocessor/go.mod
@@ -13,16 +13,17 @@ require (
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/knadh/koanf v1.4.3 // indirect
-	github.com/kr/pretty v0.3.0 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/spf13/cast v1.5.0 // indirect
 	go.opencensus.io v0.23.0 // indirect
 	go.opentelemetry.io/otel v1.10.0 // indirect
 	go.opentelemetry.io/otel/metric v0.32.1 // indirect

diff --git a/processor/probabilisticsamplerprocessor/go.sum b/processor/probabilisticsamplerprocessor/go.sum
diff --git a/processor/probabilisticsamplerprocessor/probabilisticsampler.go b/processor/probabilisticsamplerprocessor/probabilisticsampler.go
@@ -23,6 +23,8 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/processor/processorhelper"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterspan"
 )
 
 // samplingPriority has the semantic result of parsing the "sampling.priority"
@@ -52,15 +54,30 @@ const (
 type tracesamplerprocessor struct {
 	scaledSamplingRate uint32
 	hashSeed           uint32
+	include            filterspan.Matcher
+	exclude            filterspan.Matcher
 }
 
 // newTracesProcessor returns a processor.TracesProcessor that will perform head sampling according to the given
 // configuration.
+
 func newTracesProcessor(ctx context.Context, set component.ProcessorCreateSettings, cfg *Config, nextConsumer consumer.Traces) (component.TracesProcessor, error) {
+
+	include, err := filterspan.NewMatcher(cfg.Include)
+	if err != nil {
+		return nil, err
+	}
+	exclude, err := filterspan.NewMatcher(cfg.Exclude)
+	if err != nil {
+		return nil, err
+	}
+
 	tsp := &tracesamplerprocessor{
 		// Adjust sampling percentage on private so recalculations are avoided.
 		scaledSamplingRate: uint32(cfg.SamplingPercentage * percentageScaleFactor),
 		hashSeed:           cfg.HashSeed,
+		include:            include,
+		exclude:            exclude,
 	}
 
 	return processorhelper.NewTracesProcessor(
@@ -76,6 +93,9 @@ func (tsp *tracesamplerprocessor) processTraces(_ context.Context, td ptrace.Tra
 	td.ResourceSpans().RemoveIf(func(rs ptrace.ResourceSpans) bool {
 		rs.ScopeSpans().RemoveIf(func(ils ptrace.ScopeSpans) bool {
 			ils.Spans().RemoveIf(func(s ptrace.Span) bool {
+				if filterspan.SkipSpan(tsp.include, tsp.exclude, s, rs.Resource(), ils.Scope()) {
+					return false
+				}
 				sp := parseSpanSamplingPriority(s)
 				if sp == doNotSampleSpan {
 					// The OpenTelemetry mentions this as a "hint" we take a stronger

diff --git a/processor/spanmetricsprocessor/config.go b/processor/spanmetricsprocessor/config.go
@@ -20,6 +20,8 @@ import (
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/featuregate"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterconfig"
 )
 
 const (
@@ -37,6 +39,8 @@ type Dimension struct {
 type Config struct {
 	config.ProcessorSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct
 
+	filterconfig.MatchConfig `mapstructure:",squash"`
+
 	// MetricsExporter is the name of the metrics exporter to use to ship metrics.
 	MetricsExporter string `mapstructure:"metrics_exporter"`
 

diff --git a/processor/spanmetricsprocessor/go.mod b/processor/spanmetricsprocessor/go.mod
@@ -6,8 +6,10 @@ require (
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/jaegerexporter v0.61.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusexporter v0.61.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.61.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/jaegerreceiver v0.61.0
 	github.com/stretchr/testify v1.8.0
+	go.opencensus.io v0.23.0
 	go.opentelemetry.io/collector v0.61.0
 	go.opentelemetry.io/collector/pdata v0.61.0
 	go.opentelemetry.io/collector/semconv v0.61.0
@@ -31,6 +33,7 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/gogo/googleapis v1.4.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/gorilla/handlers v1.5.1 // indirect
@@ -50,7 +53,6 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mostynb/go-grpc-compression v1.1.17 // indirect
-	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.61.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry v0.61.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.61.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheus v0.61.0 // indirect
@@ -73,7 +75,6 @@ require (
 	github.com/subosito/gotenv v1.3.0 // indirect
 	github.com/uber/jaeger-client-go v2.30.0+incompatible // indirect
 	github.com/uber/jaeger-lib v2.4.1+incompatible // indirect
-	go.opencensus.io v0.23.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.36.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.36.1 // indirect
 	go.opentelemetry.io/otel v1.10.0 // indirect

diff --git a/processor/spanmetricsprocessor/go.sum b/processor/spanmetricsprocessor/go.sum
diff --git a/processor/spanmetricsprocessor/processor.go b/processor/spanmetricsprocessor/processor.go
@@ -17,12 +17,15 @@ package spanmetricsprocessor // import "github.com/open-telemetry/opentelemetry-
 import (
 	"context"
 	"fmt"
+	"log"
 	"sort"
 	"strings"
 	"sync"
 	"time"
 	"unicode"
 
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
@@ -33,6 +36,7 @@ import (
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterspan"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanmetricsprocessor/internal/cache"
 )
 
@@ -51,8 +55,42 @@ var (
 	defaultLatencyHistogramBucketsMs = []float64{
 		2, 4, 6, 8, 10, 50, 100, 200, 400, 800, 1000, 1400, 2000, 5000, 10_000, 15_000,
 	}
+
+	dimensionsCacheSize   = stats.Int64("spanmetrics_dimensions_cache_entries", "size of metricKeyToDimensions LRU cache", stats.UnitDimensionless)
+	uniqueTimeSeriesCount = stats.Int64("spanmetrics_unique_time_series", "number of unique time series.", stats.UnitDimensionless)
+	spanIngestedCount     = stats.Int64("spanmetrics_spans_ingested_total", "number of spans ingested", stats.UnitDimensionless)
+	spanProcessedCount    = stats.Int64("spanmetrics_spans_processed_total", "number of spans processed", stats.UnitDimensionless)
 )
 
+type MetricKeyError struct {
+	key metricKey
+}
+
+func (r *MetricKeyError) Error() string {
+	return fmt.Sprintf("value not found in metricKeyToDimensions cache by key %q", r.key)
+}
+
+func metricViews() []*view.View {
+	return []*view.View{
+		{
+			Measure:     dimensionsCacheSize,
+			Aggregation: view.LastValue(),
+		},
+		{
+			Measure:     uniqueTimeSeriesCount,
+			Aggregation: view.LastValue(),
+		},
+		{
+			Measure:     spanIngestedCount,
+			Aggregation: view.Sum(),
+		},
+		{
+			Measure:     spanProcessedCount,
+			Aggregation: view.Sum(),
+		},
+	}
+}
+
 type exemplarData struct {
 	traceID pcommon.TraceID
 	value   float64
@@ -87,18 +125,30 @@ type processorImp struct {
 	// An LRU cache of dimension key-value maps keyed by a unique identifier formed by a concatenation of its values:
 	// e.g. { "foo/barOK": { "serviceName": "foo", "operation": "/bar", "status_code": "OK" }}
 	metricKeyToDimensions *cache.Cache
+
+	include filterspan.Matcher
+	exclude filterspan.Matcher
 }
 
 func newProcessor(logger *zap.Logger, config config.Processor, nextConsumer consumer.Traces) (*processorImp, error) {
 	logger.Info("Building spanmetricsprocessor")
 	pConfig := config.(*Config)
 
+	include, err := filterspan.NewMatcher(pConfig.Include)
+	if err != nil {
+		return nil, err
+	}
+	exclude, err := filterspan.NewMatcher(pConfig.Exclude)
+	if err != nil {
+		return nil, err
+	}
+
 	bounds := defaultLatencyHistogramBucketsMs
 	if pConfig.LatencyHistogramBuckets != nil {
 		bounds = mapDurationsToMillis(pConfig.LatencyHistogramBuckets)
 	}
 
-	if err := validateDimensions(pConfig.Dimensions, pConfig.skipSanitizeLabel); err != nil {
+	if err = validateDimensions(pConfig.Dimensions, pConfig.skipSanitizeLabel); err != nil {
 		return nil, err
 	}
 
@@ -126,6 +176,8 @@ func newProcessor(logger *zap.Logger, config config.Processor, nextConsumer cons
 		nextConsumer:          nextConsumer,
 		dimensions:            pConfig.Dimensions,
 		metricKeyToDimensions: metricKeyToDimensionsCache,
+		include:               include,
+		exclude:               exclude,
 	}, nil
 }
 
@@ -172,6 +224,12 @@ func validateDimensions(dimensions []Dimension, skipSanitizeLabel bool) error {
 	return nil
 }
 
+func init() {
+	if err := view.Register(metricViews()...); err != nil {
+		log.Fatalf(err.Error())
+	}
+}
+
 // Start implements the component.Component interface.
 func (p *processorImp) Start(ctx context.Context, host component.Host) error {
 	p.logger.Info("Starting spanmetricsprocessor")
@@ -202,6 +260,7 @@ func (p *processorImp) Start(ctx context.Context, host component.Host) error {
 		return fmt.Errorf("failed to find metrics exporter: '%s'; please configure metrics_exporter from one of: %+v",
 			p.config.MetricsExporter, availableMetricsExporters)
 	}
+
 	p.logger.Info("Started spanmetricsprocessor")
 	return nil
 }
@@ -228,7 +287,7 @@ func (p *processorImp) ConsumeTraces(ctx context.Context, traces ptrace.Traces)
 func (p *processorImp) tracesToMetrics(ctx context.Context, traces ptrace.Traces) error {
 	p.lock.Lock()
 
-	p.aggregateMetrics(traces)
+	p.aggregateMetrics(ctx, traces)
 	m, err := p.buildMetrics()
 
 	// Exemplars are only relevant to this batch of traces, so must be cleared within the lock,
@@ -340,14 +399,14 @@ func (p *processorImp) getDimensionsByMetricKey(k metricKey) (*pcommon.Map, erro
 		return nil, fmt.Errorf("type assertion of metricKeyToDimensions attributes failed, the key is %q", k)
 	}
 
-	return nil, fmt.Errorf("value not found in metricKeyToDimensions cache by key %q", k)
+	return nil, &MetricKeyError{k}
 }
 
 // aggregateMetrics aggregates the raw metrics from the input trace data.
 // Each metric is identified by a key that is built from the service name
 // and span metadata such as operation, kind, status_code and any additional
 // dimensions the user has configured.
-func (p *processorImp) aggregateMetrics(traces ptrace.Traces) {
+func (p *processorImp) aggregateMetrics(ctx context.Context, traces ptrace.Traces) {
 	for i := 0; i < traces.ResourceSpans().Len(); i++ {
 		rspans := traces.ResourceSpans().At(i)
 		r := rspans.Resource()
@@ -356,21 +415,33 @@ func (p *processorImp) aggregateMetrics(traces ptrace.Traces) {
 		if !ok {
 			continue
 		}
+
 		serviceName := attr.Str()
-		p.aggregateMetricsForServiceSpans(rspans, serviceName)
+		p.aggregateMetricsForServiceSpans(ctx, rspans, serviceName)
+
 	}
+
+	uniqueMetricsCount := len(p.callSum) + len(p.latencySum) + (len(p.latencyBucketCounts) * len(p.latencyBounds))
+
+	stats.Record(ctx, uniqueTimeSeriesCount.M(int64(uniqueMetricsCount)), dimensionsCacheSize.M(int64(p.metricKeyToDimensions.Len())))
 }
 
-func (p *processorImp) aggregateMetricsForServiceSpans(rspans ptrace.ResourceSpans, serviceName string) {
+func (p *processorImp) aggregateMetricsForServiceSpans(ctx context.Context, rspans ptrace.ResourceSpans, serviceName string) {
 	ilsSlice := rspans.ScopeSpans()
+	spanCounter := 0
 	for j := 0; j < ilsSlice.Len(); j++ {
 		ils := ilsSlice.At(j)
 		spans := ils.Spans()
 		for k := 0; k < spans.Len(); k++ {
 			span := spans.At(k)
+			if filterspan.SkipSpan(p.include, p.exclude, span, rspans.Resource(), ils.Scope()) {
+				continue
+			}
 			p.aggregateMetricsForSpan(serviceName, span, rspans.Resource().Attributes())
+			spanCounter++
 		}
 	}
+	stats.Record(ctx, spanProcessedCount.M(int64(spanCounter)))
 }
 
 func (p *processorImp) aggregateMetricsForSpan(serviceName string, span ptrace.Span, resourceAttr pcommon.Map) {