[processor/tailsampling] Refactor internal telemetry

processor/tailsamplingprocessor/factory.go

@@ -7,38 +7,17 @@ package tailsamplingprocessor // import "github.com/open-telemetry/opentelemetry
 
 import (
 	"context"
-	"sync"
 	"time"
 
-	"go.opencensus.io/stats/view"
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/featuregate"
 	"go.opentelemetry.io/collector/processor"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/metadata"
 )
 
-var onceMetrics sync.Once
-
-var metricStatCountSpansSampledFeatureGate = featuregate.GlobalRegistry().MustRegister(
-	"processor.tailsamplingprocessor.metricstatcountspanssampled",
-	featuregate.StageAlpha,
-	featuregate.WithRegisterDescription("When enabled, a new metric stat_count_spans_sampled will be available in the tail sampling processor. Differently from stat_count_traces_sampled, this metric will count the number of spans sampled or not per sampling policy, where the original counts traces."),
-)
-
-func isMetricStatCountSpansSampledEnabled() bool {
-	return metricStatCountSpansSampledFeatureGate.IsEnabled()
-}
-
 // NewFactory returns a new factory for the Tail Sampling processor.
 func NewFactory() processor.Factory {
-	onceMetrics.Do(func() {
-		// TODO: this is hardcoding the metrics level and skips error handling
-		_ = view.Register(samplingProcessorMetricViews(configtelemetry.LevelNormal)...)
-	})
-
 	return processor.NewFactory(
 		metadata.Type,
 		createDefaultConfig,

processor/tailsamplingprocessor/internal/telemetry/featureflag.go

@@ -0,0 +1,16 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package telemetry // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/telemetry"
+
+import "go.opentelemetry.io/collector/featuregate"
+
+var metricStatCountSpansSampledFeatureGate = featuregate.GlobalRegistry().MustRegister(
+	"processor.tailsamplingprocessor.metricstatcountspanssampled",
+	featuregate.StageAlpha,
+	featuregate.WithRegisterDescription("When enabled, a new metric stat_count_spans_sampled will be available in the tail sampling processor. Differently from stat_count_traces_sampled, this metric will count the number of spans sampled or not per sampling policy, where the original counts traces."),
+)
+
+func isMetricStatCountSpansSampledEnabled() bool {
+	return metricStatCountSpansSampledFeatureGate.IsEnabled()
+}

processor/tailsamplingprocessor/metrics.go → processor/tailsamplingprocessor/internal/telemetry/metrics_oc.go

@@ -1,16 +1,19 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package tailsamplingprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor"
+package telemetry // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/telemetry"
 
 import (
+	"context"
+
 	"go.opencensus.io/stats"
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/tag"
 	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/processor/processorhelper"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/metadata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/sampling"
 )
 
 // Variables related to metrics specific to tail sampling.
@@ -19,6 +22,9 @@ var (
 	tagSampledKey, _   = tag.NewKey("sampled")
 	tagSourceFormat, _ = tag.NewKey("source_format")
 
+	tagMutatorSampled    = []tag.Mutator{tag.Upsert(tagSampledKey, "true")}
+	tagMutatorNotSampled = []tag.Mutator{tag.Upsert(tagSampledKey, "false")}
+
 	statDecisionLatencyMicroSec  = stats.Int64("sampling_decision_latency", "Latency (in microseconds) of a given sampling policy", "µs")
 	statOverallDecisionLatencyUs = stats.Int64("sampling_decision_timer_latency", "Latency (in microseconds) of each run of the sampling decision timer", "µs")
 
@@ -36,6 +42,74 @@ var (
 	statTracesOnMemoryGauge     = stats.Int64("sampling_traces_on_memory", "Tracks the number of traces current on memory", stats.UnitDimensionless)
 )
 
+func contextForPolicyOC(ctx context.Context, configName, format string) (context.Context, error) {
+	return tag.New(ctx, tag.Upsert(tagPolicyKey, configName), tag.Upsert(tagSourceFormat, format))
+}
+
+func recordFinalDecisionOC(ctx context.Context, latencyMicroSec, droppedTooEarly, evaluationErrors, tracesOnMemory int64, decision sampling.Decision) {
+	stats.Record(ctx,
+		statOverallDecisionLatencyUs.M(latencyMicroSec),
+		statDroppedTooEarlyCount.M(droppedTooEarly),
+		statPolicyEvaluationErrorCount.M(evaluationErrors),
+		statTracesOnMemoryGauge.M(tracesOnMemory),
+	)
+
+	var mutators []tag.Mutator
+	switch decision {
+	case sampling.Sampled:
+		mutators = tagMutatorSampled
+	case sampling.NotSampled:
+		mutators = tagMutatorNotSampled
+	}
+
+	_ = stats.RecordWithTags(
+		ctx,
+		mutators,
+		statCountGlobalTracesSampled.M(int64(1)),
+	)
+}
+
+func recordPolicyLatencyOC(ctx context.Context, latencyMicroSec int64) {
+	stats.Record(ctx,
+		statDecisionLatencyMicroSec.M(latencyMicroSec),
+	)
+}
+
+func recordPolicyDecisionOC(ctx context.Context, sampled bool, numSpans int64) {
+	var mutators []tag.Mutator
+	if sampled {
+		mutators = tagMutatorSampled
+	} else {
+		mutators = tagMutatorNotSampled
+	}
+
+	_ = stats.RecordWithTags(
+		ctx,
+		mutators,
+		statCountTracesSampled.M(int64(1)),
+	)
+	if isMetricStatCountSpansSampledEnabled() {
+		_ = stats.RecordWithTags(
+			ctx,
+			mutators,
+			statCountSpansSampled.M(numSpans),
+		)
+	}
+
+}
+
+func recordNewTraceIDsOC(ctx context.Context, count int64) {
+	stats.Record(ctx, statNewTraceIDReceivedCount.M(count))
+}
+
+func recordLateSpanOC(ctx context.Context, ageSec int64) {
+	stats.Record(ctx, statLateSpanArrivalAfterDecision.M(ageSec))
+}
+
+func recordTraceRemovalAgeOC(ctx context.Context, ageSec int64) {
+	stats.Record(ctx, statTraceRemovalAgeSec.M(ageSec))
+}
+
 // samplingProcessorMetricViews return the metrics views according to given telemetry level.
 func samplingProcessorMetricViews(level configtelemetry.Level) []*view.View {
 	if level == configtelemetry.LevelNone {

processor/tailsamplingprocessor/internal/telemetry/telemetry.go

@@ -0,0 +1,42 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package telemetry // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/telemetry"
+
+import (
+	"context"
+	"sync"
+
+	"go.opencensus.io/stats/view"
+	"go.opentelemetry.io/collector/config/configtelemetry"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/sampling"
+)
+
+var onceTelemetry sync.Once
+
+type T struct {
+	ContextForPolicy      func(ctx context.Context, configName, format string) (context.Context, error)
+	RecordPolicyLatency   func(ctx context.Context, latencyMicroSec int64)
+	RecordPolicyDecision  func(ctx context.Context, sampled bool, numSpans int64)
+	RecordNewTraceIDs     func(ctx context.Context, count int64)
+	RecordLateSpan        func(ctx context.Context, ageSec int64)
+	RecordTraceRemovalAge func(ctx context.Context, ageSec int64)
+	RecordFinalDecision   func(ctx context.Context, latencyMicroSec, droppedTooEarly, evaluationErrors, tracesOnMemory int64, decision sampling.Decision)
+}
+
+func New() *T {
+	onceTelemetry.Do(func() {
+		_ = view.Register(samplingProcessorMetricViews(configtelemetry.LevelNormal)...)
+	})
+
+	return &T{
+		ContextForPolicy:      contextForPolicyOC,
+		RecordPolicyLatency:   recordPolicyLatencyOC,
+		RecordPolicyDecision:  recordPolicyDecisionOC,
+		RecordNewTraceIDs:     recordNewTraceIDsOC,
+		RecordLateSpan:        recordLateSpanOC,
+		RecordTraceRemovalAge: recordTraceRemovalAgeOC,
+		RecordFinalDecision:   recordFinalDecisionOC,
+	}
+}