Update all exporters to use the new helper TraceExporter. (#488)

* Update all exporters to use the new helper TraceExporter.

* Add todo for the service name span attribute.

Author: Bogdan Drutu

exporter/awsexporter/aws_xray.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/census-instrumentation/opencensus-service/consumer"
 	"github.com/census-instrumentation/opencensus-service/data"
+	"github.com/census-instrumentation/opencensus-service/exporter/exporterhelper"
 	"github.com/census-instrumentation/opencensus-service/exporter/exporterwrapper"
 )
 
@@ -57,8 +58,6 @@ type awsXRayExporter struct {
 	defaultOptions     []xray.Option
 }
 
-var _ consumer.TraceConsumer = (*awsXRayExporter)(nil)
-
 // AWSXRayTraceExportersFromViper unmarshals the viper and returns an consumer.TraceConsumer targeting
 // AWS X-Ray according to the configuration settings.
 func AWSXRayTraceExportersFromViper(v *viper.Viper) (tps []consumer.TraceConsumer, mps []consumer.MetricsConsumer, doneFns []func() error, err error) {
@@ -84,7 +83,17 @@ func AWSXRayTraceExportersFromViper(v *viper.Viper) (tps []consumer.TraceConsume
 		defaultServiceName:     xc.DefaultServiceName,
 	}
 
-	tps = append(tps, axe)
+	axte, err := exporterhelper.NewTraceExporter(
+		"aws_xray",
+		axe.PushTraceData,
+		exporterhelper.WithSpanName("ocservice.exporter.AwsXray.ConsumeTraceData"),
+		exporterhelper.WithRecordMetrics(true),
+	)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	tps = append(tps, axte)
 	doneFns = append(doneFns, func() error {
 		axe.Flush()
 		return nil
@@ -146,25 +155,13 @@ func transformConfigToXRayOptions(axrCfg *awsXRayConfig) (xopts []xray.Option, e
 
 // ExportSpans is the method that translates OpenCensus-Proto Traces into AWS X-Ray spans.
 // It uniquely maintains
-func (axe *awsXRayExporter) ConsumeTraceData(ctx context.Context, td data.TraceData) (xerr error) {
-	ctx, span := trace.StartSpan(ctx,
-		"opencensus.service.exporter.aws_xray.ExportSpans",
-		trace.WithSampler(trace.NeverSample()))
-
-	defer func() {
-		if xerr != nil && span.IsRecordingEvents() {
-			span.SetStatus(trace.Status{
-				Code:    int32(trace.StatusCodeUnknown),
-				Message: xerr.Error(),
-			})
-		}
-		span.End()
-	}()
-
+func (axe *awsXRayExporter) PushTraceData(ctx context.Context, td data.TraceData) (int, error) {
 	serviceName := td.Node.GetServiceInfo().GetName()
 	if serviceName == "" {
 		serviceName = axe.defaultServiceName
 	}
+	// TODO: Consider to do this in the exporterhelper.
+	span := trace.FromContext(ctx)
 	if span.IsRecordingEvents() {
 		span.Annotate([]trace.Attribute{
 			trace.StringAttribute("service_name", serviceName),
@@ -173,7 +170,7 @@ func (axe *awsXRayExporter) ConsumeTraceData(ctx context.Context, td data.TraceD
 
 	exp, err := axe.getOrMakeExporterByServiceName(serviceName)
 	if err != nil {
-		return err
+		return len(td.Spans), err
 	}
 	return exporterwrapper.PushOcProtoSpansToOCTraceExporter(exp, td)
 }

exporter/datadogexporter/datadog.go

@@ -71,10 +71,15 @@ func DatadogTraceExportersFromViper(v *viper.Viper) (tps []consumer.TraceConsume
 		return nil
 	})
 
+	dgte, err := exporterwrapper.NewExporterWrapper("datadog", "ocservice.exporter.DataDog.ConsumeTraceData", de)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
 	// TODO: Examine the Datadog exporter to see
 	// if trace.ExportSpan was constraining and if perhaps the
 	// upload can use the context and information from the Node.
-	tps = append(tps, exporterwrapper.NewExporterWrapper("datadog", de))
+	tps = append(tps, dgte)
 
 	// TODO: (@odeke-em, @songya23) implement ExportMetrics for Datadog.
 	// mes = append(mes, oexp)

exporter/exporterwrapper/exporterwrapper.go

@@ -27,8 +27,9 @@ import (
 	"go.opencensus.io/trace"
 
 	tracepb "github.com/census-instrumentation/opencensus-proto/gen-go/trace/v1"
-	"github.com/census-instrumentation/opencensus-service/consumer"
 	"github.com/census-instrumentation/opencensus-service/data"
+	"github.com/census-instrumentation/opencensus-service/exporter"
+	"github.com/census-instrumentation/opencensus-service/exporter/exporterhelper"
 	"github.com/census-instrumentation/opencensus-service/internal"
 	spandatatranslator "github.com/census-instrumentation/opencensus-service/translator/trace/spandata"
 )
@@ -41,42 +42,22 @@ import (
 // by various vendors and contributors. Eventually the goal is to
 // get those exporters converted to directly receive
 // OpenCensus Proto TraceData.
-func NewExporterWrapper(exporterName string, ocExporter trace.Exporter) consumer.TraceConsumer {
-	return &ocExporterWrapper{spanName: "opencensus.service.exporter." + exporterName + ".ExportTrace", ocExporter: ocExporter}
-}
-
-type ocExporterWrapper struct {
-	spanName   string
-	ocExporter trace.Exporter
-}
-
-var _ consumer.TraceConsumer = (*ocExporterWrapper)(nil)
-
-func (octew *ocExporterWrapper) ConsumeTraceData(ctx context.Context, td data.TraceData) (aerr error) {
-	ctx, span := trace.StartSpan(ctx,
-		octew.spanName, trace.WithSampler(trace.NeverSample()))
-
-	if span.IsRecordingEvents() {
-		span.Annotate([]trace.Attribute{
-			trace.Int64Attribute("n_spans", int64(len(td.Spans))),
-		}, "")
-	}
-
-	defer func() {
-		if aerr != nil && span.IsRecordingEvents() {
-			span.SetStatus(trace.Status{Code: trace.StatusCodeInternal, Message: aerr.Error()})
-		}
-		span.End()
-	}()
-
-	return PushOcProtoSpansToOCTraceExporter(octew.ocExporter, td)
+func NewExporterWrapper(exporterName string, spanName string, ocExporter trace.Exporter) (exporter.TraceExporter, error) {
+	return exporterhelper.NewTraceExporter(
+		exporterName,
+		func(ctx context.Context, td data.TraceData) (int, error) {
+			return PushOcProtoSpansToOCTraceExporter(ocExporter, td)
+		},
+		exporterhelper.WithSpanName(spanName),
+		exporterhelper.WithRecordMetrics(true),
+	)
 }
 
 // TODO: Remove PushOcProtoSpansToOCTraceExporter after aws-xray is changed to ExporterWrapper.
 
 // PushOcProtoSpansToOCTraceExporter pushes TraceData to the given trace.Exporter by converting the
 // protos to trace.SpanData.
-func PushOcProtoSpansToOCTraceExporter(ocExporter trace.Exporter, td data.TraceData) error {
+func PushOcProtoSpansToOCTraceExporter(ocExporter trace.Exporter, td data.TraceData) (int, error) {
 	var errs []error
 	var goodSpans []*tracepb.Span
 	for _, span := range td.Spans {
@@ -89,5 +70,5 @@ func PushOcProtoSpansToOCTraceExporter(ocExporter trace.Exporter, td data.TraceD
 		}
 	}
 
-	return internal.CombineErrors(errs)
+	return len(td.Spans) - len(goodSpans), internal.CombineErrors(errs)
 }

exporter/honeycombexporter/honeycomb.go

@@ -47,7 +47,12 @@ func HoneycombTraceExportersFromViper(v *viper.Viper) (tps []consumer.TraceConsu
 
 	rawExp := honeycomb.NewExporter(hc.WriteKey, hc.DatasetName)
 
-	tps = append(tps, exporterwrapper.NewExporterWrapper("honeycomb", rawExp))
+	hcte, err := exporterwrapper.NewExporterWrapper("honeycomb", "ocservice.exporter.HoneyComb.ConsumeTraceData", rawExp)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	tps = append(tps, hcte)
 	doneFns = append(doneFns, func() error {
 		rawExp.Close()
 		return nil

exporter/jaegerexporter/jaeger.go

@@ -61,9 +61,14 @@ func JaegerExportersFromViper(v *viper.Viper) (tps []consumer.TraceConsumer, mps
 		je.Flush()
 		return nil
 	})
+
+	jte, err := exporterwrapper.NewExporterWrapper("jaeger", "ocservice.exporter.Jaeger.ConsumeTraceData", je)
+	if err != nil {
+		return nil, nil, nil, err
+	}
 	// TODO: Examine "contrib.go.opencensus.io/exporter/jaeger" to see
 	// if trace.ExportSpan was constraining and if perhaps the Jaeger
 	// upload can use the context and information from the Node.
-	tps = append(tps, exporterwrapper.NewExporterWrapper("jaeger", je))
+	tps = append(tps, jte)
 	return
 }

exporter/kafkaexporter/kafka.go

@@ -53,7 +53,12 @@ func KafkaExportersFromViper(v *viper.Viper) (tps []consumer.TraceConsumer, mps
 		return nil, nil, nil, fmt.Errorf("Cannot configure Kafka Trace exporter: %v", kerr)
 	}
 
-	tps = append(tps, exporterwrapper.NewExporterWrapper("kafka", kde))
+	kte, err := exporterwrapper.NewExporterWrapper("kafka", "ocservice.exporter.Kafka.ConsumeTraceData", kde)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	tps = append(tps, kte)
 	doneFns = append(doneFns, func() error {
 		kde.Flush()
 		return nil

exporter/opencensusexporter/opencensus.go

@@ -118,7 +118,15 @@ func OpenCensusTraceExportersFromViper(v *viper.Viper) (tps []consumer.TraceCons
 	}
 
 	oce := &ocagentExporter{exporters: exporters}
-	oexp, _ := exporterhelper.NewTraceExporter("oc_trace", oce.PushTraceData, exporterhelper.WithRecordMetrics(true))
+	oexp, err := exporterhelper.NewTraceExporter(
+		"oc_trace",
+		oce.PushTraceData,
+		exporterhelper.WithSpanName("ocservice.exporter.OpenCensus.ConsumeTraceData"),
+		exporterhelper.WithRecordMetrics(true))
+
+	if err != nil {
+		return nil, nil, nil, err
+	}
 
 	tps = append(tps, oexp)
 

exporter/stackdriverexporter/stackdriver.go

@@ -88,11 +88,16 @@ func StackdriverTraceExportersFromViper(v *viper.Viper) (tps []consumer.TraceCon
 		exporter: sde,
 	}
 
+	sdte, err := exporterwrapper.NewExporterWrapper("stackdriver_trace", "ocservice.exporter.Stackdriver.ConsumeTraceData", sde)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
 	// TODO: Examine "contrib.go.opencensus.io/exporter/stackdriver" to see
 	// if trace.ExportSpan was constraining and if perhaps the Stackdriver
 	// upload can use the context and information from the Node.
 	if sc.EnableTracing {
-		tps = append(tps, exporterwrapper.NewExporterWrapper("stackdriver", sde))
+		tps = append(tps, sdte)
 	}
 
 	if sc.EnableMetrics {