add context-based e2e pipeline processing duration metrics

Author: dashpole

cmd/otelcol/config.yaml

@@ -44,12 +44,10 @@ service:
 
     traces:
       receivers: [otlp, opencensus, jaeger, zipkin]
-      processors: [batch]
       exporters: [logging]
 
     metrics:
       receivers: [otlp, opencensus, prometheus]
-      processors: [batch]
       exporters: [logging]
 
   extensions: [health_check, pprof, zpages]

obsreport/obsreport.go

@@ -134,6 +134,15 @@ func AllViews() (views []*view.View) {
 	tagKeys = []tag.Key{tagKeyProcessor}
 	views = append(views, genViews(measures, tagKeys, view.Sum())...)
 
+	// Pipeline views.
+	views = append(views, &view.View{
+		Name:        mPipelineMetricLatencyMetrics.Name(),
+		Description: mPipelineMetricLatencyMetrics.Description(),
+		TagKeys:     []tag.Key{tagKeyPipeline},
+		Measure:     mPipelineMetricLatencyMetrics,
+		Aggregation: view.Distribution(0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1),
+	})
+
 	return views
 }
 

obsreport/obsreport_consumer.go

@@ -0,0 +1,79 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package obsreport
+
+import (
+	"context"
+
+	"go.opencensus.io/tag"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/pdata"
+)
+
+// NewMetrics wraps a metrics consumer and adds pipeline tags.
+func NewMetrics(pipeline string, mc consumer.Metrics) consumer.Metrics {
+	return metricsConsumer{next: mc, pipeline: pipeline}
+}
+
+type metricsConsumer struct {
+	next     consumer.Metrics
+	pipeline string
+}
+
+var _ consumer.Metrics = (*metricsConsumer)(nil)
+
+// ConsumeMetrics exports the pdata.Metrics to the next consumer after tagging.
+func (c metricsConsumer) ConsumeMetrics(ctx context.Context, md pdata.Metrics) error {
+	ctx, _ = tag.New(ctx,
+		tag.Upsert(tagKeyPipeline, c.pipeline, tag.WithTTL(tag.TTLNoPropagation)))
+	return c.next.ConsumeMetrics(ctx, md)
+}
+
+// NewTraces wraps a traces consumer and adds pipeline tags.
+func NewTraces(pipeline string, tc consumer.Traces) consumer.Traces {
+	return tracesConsumer{next: tc, pipeline: pipeline}
+}
+
+type tracesConsumer struct {
+	next     consumer.Traces
+	pipeline string
+}
+
+var _ consumer.Traces = (*tracesConsumer)(nil)
+
+// ConsumeTraces exports the pdata.Traces to the next consumer after tagging.
+func (c tracesConsumer) ConsumeTraces(ctx context.Context, md pdata.Traces) error {
+	ctx, _ = tag.New(ctx,
+		tag.Upsert(tagKeyPipeline, c.pipeline, tag.WithTTL(tag.TTLNoPropagation)))
+	return c.next.ConsumeTraces(ctx, md)
+}
+
+// NewLogs wraps a logs consumer and adds pipeline tags.
+func NewLogs(pipeline string, lc consumer.Logs) consumer.Logs {
+	return logsConsumer{next: lc, pipeline: pipeline}
+}
+
+type logsConsumer struct {
+	next     consumer.Logs
+	pipeline string
+}
+
+var _ consumer.Logs = (*logsConsumer)(nil)
+
+// ConsumeLogs exports the pdata.Logs to the next consumer after tagging.
+func (c logsConsumer) ConsumeLogs(ctx context.Context, md pdata.Logs) error {
+	ctx, _ = tag.New(ctx,
+		tag.Upsert(tagKeyPipeline, c.pipeline, tag.WithTTL(tag.TTLNoPropagation)))
+	return c.next.ConsumeLogs(ctx, md)
+}

obsreport/obsreport_exporter.go

@@ -167,6 +167,8 @@ func (eor *Exporter) recordMetrics(ctx context.Context, numSent, numFailedToSend
 		eor.mutators,
 		sentMeasure.M(numSent),
 		failedToSendMeasure.M(numFailedToSend))
+
+	recordPipelineDuration(ctx)
 }
 
 func endSpan(ctx context.Context, err error, numSent, numFailedToSend int64, sentItemsKey, failedToSendItemsKey string) {

obsreport/obsreport_pipeline.go

@@ -0,0 +1,55 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package obsreport
+
+import (
+	"context"
+	"time"
+
+	"go.opencensus.io/stats"
+	"go.opencensus.io/tag"
+)
+
+const (
+	// PipelineKey the name of the pipeline
+	PipelineKey = "pipeline"
+
+	MetricLatencyKey = "metric_processing_duration_seconds"
+)
+
+var (
+	tagKeyPipeline, _             = tag.NewKey(PipelineKey)
+	e2ePrefix                     = PipelineKey + nameSep
+	mPipelineMetricLatencyMetrics = stats.Float64(
+		e2ePrefix+MetricLatencyKey,
+		"Duration of handling a metric in the pipeline.",
+		stats.UnitSeconds)
+)
+
+type pipelineStartContextKey struct{}
+
+func recordPipelineStart(ctx context.Context) context.Context {
+	return context.WithValue(ctx, pipelineStartContextKey{}, time.Now())
+}
+
+func recordPipelineDuration(ctx context.Context) {
+	startTime, _ := ctx.Value(pipelineStartContextKey{}).(time.Time)
+	if startTime.IsZero() {
+		return
+	}
+	stats.Record(
+		ctx,
+		mPipelineMetricLatencyMetrics.M(time.Since(startTime).Seconds()))
+}

obsreport/obsreport_receiver.go

@@ -152,6 +152,7 @@ func StartTraceDataReceiveOp(
 	transport string,
 	opt ...StartReceiveOption,
 ) context.Context {
+	operationCtx = recordPipelineStart(operationCtx)
 	return traceReceiveOp(
 		operationCtx,
 		receiver,
@@ -186,6 +187,7 @@ func StartLogsReceiveOp(
 	transport string,
 	opt ...StartReceiveOption,
 ) context.Context {
+	operationCtx = recordPipelineStart(operationCtx)
 	return traceReceiveOp(
 		operationCtx,
 		receiver,
@@ -220,6 +222,7 @@ func StartMetricsReceiveOp(
 	transport string,
 	opt ...StartReceiveOption,
 ) context.Context {
+	operationCtx = recordPipelineStart(operationCtx)
 	return traceReceiveOp(
 		operationCtx,
 		receiver,

service/internal/builder/pipelines_builder.go

@@ -25,6 +25,7 @@ import (
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/consumer/fanoutconsumer"
+	"go.opentelemetry.io/collector/obsreport"
 )
 
 // builtPipeline is a pipeline that is built based on a config.
@@ -125,11 +126,11 @@ func (pb *pipelinesBuilder) buildPipeline(ctx context.Context, pipelineCfg *conf
 
 	switch pipelineCfg.InputType {
 	case config.TracesDataType:
-		tc = pb.buildFanoutExportersTraceConsumer(pipelineCfg.Exporters)
+		tc = obsreport.NewTraces(pipelineCfg.Name, pb.buildFanoutExportersTraceConsumer(pipelineCfg.Exporters))
 	case config.MetricsDataType:
-		mc = pb.buildFanoutExportersMetricsConsumer(pipelineCfg.Exporters)
+		mc = obsreport.NewMetrics(pipelineCfg.Name, pb.buildFanoutExportersMetricsConsumer(pipelineCfg.Exporters))
 	case config.LogsDataType:
-		lc = pb.buildFanoutExportersLogConsumer(pipelineCfg.Exporters)
+		lc = obsreport.NewLogs(pipelineCfg.Name, pb.buildFanoutExportersLogConsumer(pipelineCfg.Exporters))
 	}
 
 	mutatesConsumedData := false