Add OTLP trace data types, factories and helpers (#534)

This is step one in converting in-memory representation from OC to OTLP. This
change has no effect except in tests. The introduced code is not reachable.

Issue: #478

Author: tigrannajaryan

consumer/consumer.go

@@ -36,3 +36,10 @@ type MetricsConsumer interface {
 type TraceConsumer interface {
 	ConsumeTraceData(ctx context.Context, td consumerdata.TraceData) error
 }
+
+// OTLPTraceConsumer is an interface that receives consumerdata.OTLPTraceData, processes it
+// as needed, and sends it to the next processing node if any or to the destination.
+type OTLPTraceConsumer interface {
+	// ConsumeOTLPTrace receives consumerdata.OTLPTraceData for processing.
+	ConsumeOTLPTrace(ctx context.Context, td consumerdata.OTLPTraceData) error
+}

consumer/consumerdata/consumerdata.go

@@ -20,6 +20,7 @@ import (
 	metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
 	resourcepb "github.com/census-instrumentation/opencensus-proto/gen-go/resource/v1"
 	tracepb "github.com/census-instrumentation/opencensus-proto/gen-go/trace/v1"
+	otlptrace "github.com/open-telemetry/opentelemetry-proto/gen/go/trace/v1"
 )
 
 // MetricsData is a struct that groups proto metrics with a unique node and a resource.
@@ -36,3 +37,18 @@ type TraceData struct {
 	Spans        []*tracepb.Span
 	SourceFormat string
 }
+
+// OTLPTraceData is a struct that groups proto spans with a resource. This is the
+// newer version of TraceData, using OTLP-based representation.
+type OTLPTraceData struct {
+	ResourceSpanList []*otlptrace.ResourceSpans
+}
+
+// SpanCount calculates the total number of spans.
+func (td OTLPTraceData) SpanCount() int {
+	spanCount := 0
+	for _, rsl := range td.ResourceSpanList {
+		spanCount += len(rsl.Spans)
+	}
+	return spanCount
+}

exporter/exporter.go

@@ -25,13 +25,19 @@ type Exporter interface {
 	component.Component
 }
 
-// TraceExporter composes TraceConsumer with some additional exporter-specific functions.
+// TraceExporter is a TraceConsumer that is also an Exporter.
 type TraceExporter interface {
 	consumer.TraceConsumer
 	Exporter
 }
 
-// MetricsExporter composes MetricsConsumer with some additional exporter-specific functions.
+// OTLPTraceExporter is an OTLPTraceConsumer that is also an Exporter.
+type OTLPTraceExporter interface {
+	consumer.OTLPTraceConsumer
+	Exporter
+}
+
+// MetricsExporter is a MetricsConsumer that is also an Exporter.
 type MetricsExporter interface {
 	consumer.MetricsConsumer
 	Exporter

exporter/exporterhelper/constants.go

@@ -21,8 +21,8 @@ import (
 var (
 	// errNilConfig is returned when an empty name is given.
 	errNilConfig = errors.New("nil config")
-	// errNilPushTraceData is returned when a nil pushTraceData is given.
-	errNilPushTraceData = errors.New("nil pushTraceData")
+	// errNilPushTraceData is returned when a nil traceDataPusher is given.
+	errNilPushTraceData = errors.New("nil traceDataPusher")
 	// errNilPushMetricsData is returned when a nil pushMetricsData is given.
 	errNilPushMetricsData = errors.New("nil pushMetricsData")
 )

exporter/exporterhelper/tracehelper.go

@@ -26,25 +26,33 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector/observability"
 )
 
-// PushTraceData is a helper function that is similar to ConsumeTraceData but also returns
-// the number of dropped spans.
-type PushTraceData func(ctx context.Context, td consumerdata.TraceData) (droppedSpans int, err error)
+// Suffix to use for span names emitted by exporter for observability purposes.
+const spanNameSuffix = ".ExportTraceData"
 
+// traceDataPusher is a helper function that is similar to ConsumeTraceData but also
+// returns the number of dropped spans.
+type traceDataPusher func(ctx context.Context, td consumerdata.TraceData) (droppedSpans int, err error)
+
+// otlpTraceDataPusher is a helper function that is similar to ConsumeTraceData but also
+// returns the number of dropped spans.
+type otlpTraceDataPusher func(ctx context.Context, td consumerdata.OTLPTraceData) (droppedSpans int, err error)
+
+// traceExporter implements the exporter with additional helper options.
 type traceExporter struct {
 	exporterFullName string
-	pushTraceData    PushTraceData
+	dataPusher       traceDataPusher
 	shutdown         Shutdown
 }
 
-var _ (exporter.TraceExporter) = (*traceExporter)(nil)
+var _ exporter.TraceExporter = (*traceExporter)(nil)
 
 func (te *traceExporter) Start(host component.Host) error {
 	return nil
 }
 
 func (te *traceExporter) ConsumeTraceData(ctx context.Context, td consumerdata.TraceData) error {
 	exporterCtx := observability.ContextWithExporterName(ctx, te.exporterFullName)
-	_, err := te.pushTraceData(exporterCtx, td)
+	_, err := te.dataPusher(exporterCtx, td)
 	return err
 }
 
@@ -53,28 +61,34 @@ func (te *traceExporter) Shutdown() error {
 	return te.shutdown()
 }
 
-// NewTraceExporter creates an TraceExporter that can record metrics and can wrap every request with a Span.
-// If no options are passed it just adds the exporter format as a tag in the Context.
-// TODO: Add support for retries.
-func NewTraceExporter(config configmodels.Exporter, pushTraceData PushTraceData, options ...ExporterOption) (exporter.TraceExporter, error) {
+// NewTraceExporter creates an TraceExporter that can record metrics and can wrap every
+// request with a Span. If no options are passed it just adds the exporter format as a
+// tag in the Context.
+func NewTraceExporter(
+	config configmodels.Exporter,
+	dataPusher traceDataPusher,
+	options ...ExporterOption,
+) (exporter.TraceExporter, error) {
+
 	if config == nil {
 		return nil, errNilConfig
 	}
 
-	if pushTraceData == nil {
+	if dataPusher == nil {
 		return nil, errNilPushTraceData
 	}
 
 	opts := newExporterOptions(options...)
 	if opts.recordMetrics {
-		pushTraceData = pushTraceDataWithMetrics(pushTraceData)
+		dataPusher = dataPusher.withMetrics()
 	}
 
 	if opts.recordTrace {
-		pushTraceData = pushTraceDataWithSpan(pushTraceData, config.Name()+".ExportTraceData")
+		spanName := config.Name() + spanNameSuffix
+		dataPusher = dataPusher.withSpan(spanName)
 	}
 
-	// The default shutdown function returns nil.
+	// The default shutdown function does nothing.
 	if opts.shutdown == nil {
 		opts.shutdown = func() error {
 			return nil
@@ -83,27 +97,31 @@ func NewTraceExporter(config configmodels.Exporter, pushTraceData PushTraceData,
 
 	return &traceExporter{
 		exporterFullName: config.Name(),
-		pushTraceData:    pushTraceData,
+		dataPusher:       dataPusher,
 		shutdown:         opts.shutdown,
 	}, nil
 }
 
-func pushTraceDataWithMetrics(next PushTraceData) PushTraceData {
+// withMetrics wraps the current pusher into a function that records the metrics of the
+// pusher execution.
+func (p traceDataPusher) withMetrics() traceDataPusher {
 	return func(ctx context.Context, td consumerdata.TraceData) (int, error) {
-		// TODO: Add retry logic here if we want to support because we need to record special metrics.
-		droppedSpans, err := next(ctx, td)
+		// Forward the data to the next consumer (this pusher is the next).
+		droppedSpans, err := p(ctx, td)
 		// TODO: How to record the reason of dropping?
 		observability.RecordMetricsForTraceExporter(ctx, len(td.Spans), droppedSpans)
 		return droppedSpans, err
 	}
 }
 
-func pushTraceDataWithSpan(next PushTraceData, spanName string) PushTraceData {
+// withSpan wraps the current pusher into a function that records a span during
+// pusher execution.
+func (p traceDataPusher) withSpan(spanName string) traceDataPusher {
 	return func(ctx context.Context, td consumerdata.TraceData) (int, error) {
 		ctx, span := trace.StartSpan(ctx, spanName)
 		defer span.End()
-		// Call next stage.
-		droppedSpans, err := next(ctx, td)
+		// Forward the data to the next consumer (this pusher is the next).
+		droppedSpans, err := p(ctx, td)
 		if span.IsRecordingEvents() {
 			span.AddAttributes(
 				trace.Int64Attribute(numReceivedSpansAttribute, int64(len(td.Spans))),
@@ -116,3 +134,108 @@ func pushTraceDataWithSpan(next PushTraceData, spanName string) PushTraceData {
 		return droppedSpans, err
 	}
 }
+
+type otlpTraceExporter struct {
+	exporterFullName string
+	dataPusher       otlpTraceDataPusher
+	shutdown         Shutdown
+}
+
+var _ exporter.OTLPTraceExporter = (*otlpTraceExporter)(nil)
+
+func (te *otlpTraceExporter) Start(host component.Host) error {
+	return nil
+}
+
+func (te *otlpTraceExporter) ConsumeOTLPTrace(
+	ctx context.Context,
+	td consumerdata.OTLPTraceData,
+) error {
+	exporterCtx := observability.ContextWithExporterName(ctx, te.exporterFullName)
+	_, err := te.dataPusher(exporterCtx, td)
+	return err
+}
+
+// Shutdown stops the exporter and is invoked during shutdown.
+func (te *otlpTraceExporter) Shutdown() error {
+	return te.shutdown()
+}
+
+// NewOTLPTraceExporter creates an OTLPTraceExporter that can record metrics and can wrap
+// every request with a Span.
+func NewOTLPTraceExporter(
+	config configmodels.Exporter,
+	dataPusher otlpTraceDataPusher,
+	options ...ExporterOption,
+) (exporter.OTLPTraceExporter, error) {
+
+	if config == nil {
+		return nil, errNilConfig
+	}
+
+	if dataPusher == nil {
+		return nil, errNilPushTraceData
+	}
+
+	opts := newExporterOptions(options...)
+	if opts.recordMetrics {
+		dataPusher = dataPusher.withMetrics()
+	}
+
+	if opts.recordTrace {
+		spanName := config.Name() + spanNameSuffix
+		dataPusher = dataPusher.withSpan(spanName)
+	}
+
+	// The default shutdown function does nothing.
+	if opts.shutdown == nil {
+		opts.shutdown = func() error {
+			return nil
+		}
+	}
+
+	return &otlpTraceExporter{
+		exporterFullName: config.Name(),
+		dataPusher:       dataPusher,
+		shutdown:         opts.shutdown,
+	}, nil
+}
+
+// withMetrics wraps the current pusher into a function that records the metrics of the
+// pusher execution.
+func (p otlpTraceDataPusher) withMetrics() otlpTraceDataPusher {
+	return func(ctx context.Context, td consumerdata.OTLPTraceData) (int, error) {
+		// Forward the data to the next consumer (this pusher is the next).
+		droppedSpans, err := p(ctx, td)
+
+		// Record the results as metrics.
+		observability.RecordMetricsForTraceExporter(ctx, td.SpanCount(), droppedSpans)
+
+		return droppedSpans, err
+	}
+}
+
+// withSpan wraps the current pusher into a function that records a span during
+// pusher execution.
+func (p otlpTraceDataPusher) withSpan(spanName string) otlpTraceDataPusher {
+	return func(ctx context.Context, td consumerdata.OTLPTraceData) (int, error) {
+		// Start a span.
+		ctx, span := trace.StartSpan(ctx, spanName)
+
+		// End the span after this function is done.
+		defer span.End()
+
+		// Forward the data to the next consumer (this pusher is the next).
+		droppedSpans, err := p(ctx, td)
+		if span.IsRecordingEvents() {
+			span.AddAttributes(
+				trace.Int64Attribute(numReceivedSpansAttribute, int64(td.SpanCount())),
+				trace.Int64Attribute(numDroppedSpansAttribute, int64(droppedSpans)),
+			)
+			if err != nil {
+				span.SetStatus(errToStatus(err))
+			}
+		}
+		return droppedSpans, err
+	}
+}