[processor/tailsampling] Refactor internal telemetry (#33108)

This PR is the first step in refactoring the internal telemetry for the
tail-sampling processor.

There are no expected changes in user behavior between this and the
current main, but all OpenCensus code is now isolated and can be
replaced without affecting other parts of the code.
Once this PR is merged, the next step is to create equivalent metrics
using the OpenTelemetry Go API.

Signed-off-by: Juraci Paixão Kröhling <[email protected]>

---------

Signed-off-by: Juraci Paixão Kröhling <[email protected]>

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,
+	}
+}