Remove trace/metrics exporter sinks. (#426)

* Remove trace/metrics exporter sinks.

* Fix naming in exporter_processor

Author: Bogdan Drutu

cmd/ocagent/main.go

@@ -31,7 +31,6 @@ import (
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 
-	"github.com/census-instrumentation/opencensus-service/exporter"
 	"github.com/census-instrumentation/opencensus-service/internal"
 	"github.com/census-instrumentation/opencensus-service/internal/config"
 	"github.com/census-instrumentation/opencensus-service/internal/config/viperutils"
@@ -89,8 +88,8 @@ func runOCAgent() {
 		log.Fatalf("Config: failed to create exporters from YAML: %v", err)
 	}
 
-	commonSpanSink := exporter.MultiTraceExporters(traceExporters...)
-	commonMetricsSink := exporter.MultiMetricsExporters(metricsExporters...)
+	commonSpanSink := processor.NewMultiTraceDataProcessor(traceExporters)
+	commonMetricsSink := processor.NewMultiMetricsDataProcessor(metricsExporters)
 
 	// Add other receivers here as they are implemented
 	ocReceiverDoneFn, err := runOCReceiver(logger, agentConfig, commonSpanSink, commonMetricsSink)

cmd/occollector/app/collector/processors.go

@@ -26,16 +26,16 @@ import (
 
 	"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/builder"
 	"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/sender"
-	"github.com/census-instrumentation/opencensus-service/exporter"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor/nodebatcher"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor/queued"
-	"github.com/census-instrumentation/opencensus-service/internal/collector/processor/tail_sampling"
+	tailsampling "github.com/census-instrumentation/opencensus-service/internal/collector/processor/tail_sampling"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/sampling"
 	"github.com/census-instrumentation/opencensus-service/internal/config"
+	mainprocessor "github.com/census-instrumentation/opencensus-service/processor"
 )
 
-func createExporters(v *viper.Viper, logger *zap.Logger) ([]func(), []exporter.TraceExporter, []exporter.MetricsExporter) {
+func createExporters(v *viper.Viper, logger *zap.Logger) ([]func(), []mainprocessor.TraceDataProcessor, []mainprocessor.MetricsDataProcessor) {
 	// TODO: (@pjanotti) this is slightly modified from agent but in the end duplication, need to consolidate style and visibility.
 	traceExporters, metricsExporters, doneFns, err := config.ExportersFromViperConfig(logger, v)
 	if err != nil {

exporter/exporterparser/README.md exporter/README.mdexporter/exporterparser/README.md renamed to exporter/README.md

@@ -27,8 +27,8 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/exporter"
-	"github.com/census-instrumentation/opencensus-service/exporter/exporterparser"
+	"github.com/census-instrumentation/opencensus-service/exporter/exporterwrapper"
+	"github.com/census-instrumentation/opencensus-service/processor"
 )
 
 const defaultVersionForAWSXRayApplications = "latest"
@@ -57,11 +57,11 @@ type awsXRayExporter struct {
 	defaultOptions     []xray.Option
 }
 
-var _ exporter.TraceExporter = (*awsXRayExporter)(nil)
+var _ processor.TraceDataProcessor = (*awsXRayExporter)(nil)
 
-// AWSXRayTraceExportersFromViper unmarshals the viper and returns an exporter.TraceExporter targeting
+// AWSXRayTraceExportersFromViper unmarshals the viper and returns an processor.TraceDataProcessor targeting
 // AWS X-Ray according to the configuration settings.
-func AWSXRayTraceExportersFromViper(v *viper.Viper) (tes []exporter.TraceExporter, mes []exporter.MetricsExporter, doneFns []func() error, err error) {
+func AWSXRayTraceExportersFromViper(v *viper.Viper) (tdps []processor.TraceDataProcessor, mdps []processor.MetricsDataProcessor, doneFns []func() error, err error) {
 	var cfg struct {
 		AWSXRay *awsXRayConfig `mapstructure:"aws-xray"`
 	}
@@ -84,12 +84,12 @@ func AWSXRayTraceExportersFromViper(v *viper.Viper) (tes []exporter.TraceExporte
 		defaultServiceName:     xc.DefaultServiceName,
 	}
 
-	tes = append(tes, axe)
+	tdps = append(tdps, axe)
 	doneFns = append(doneFns, func() error {
 		axe.Flush()
 		return nil
 	})
-	return tes, mes, doneFns, nil
+	return
 }
 
 // Flush invokes .Flush() for every one of its underlying exporters.
@@ -146,13 +146,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) ExportSpans(ctx context.Context, td data.TraceData) (xerr error) {
+func (axe *awsXRayExporter) ProcessTraceData(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 {
+		if xerr != nil && span.IsRecordingEvents() {
 			span.SetStatus(trace.Status{
 				Code:    int32(trace.StatusCodeUnknown),
 				Message: xerr.Error(),
@@ -165,15 +165,17 @@ func (axe *awsXRayExporter) ExportSpans(ctx context.Context, td data.TraceData)
 	if serviceName == "" {
 		serviceName = axe.defaultServiceName
 	}
-	span.Annotate([]trace.Attribute{
-		trace.StringAttribute("service_name", serviceName),
-	}, "")
+	if span.IsRecordingEvents() {
+		span.Annotate([]trace.Attribute{
+			trace.StringAttribute("service_name", serviceName),
+		}, "")
+	}
 
 	exp, err := axe.getOrMakeExporterByServiceName(serviceName)
 	if err != nil {
 		return err
 	}
-	return exporterparser.OcProtoSpansToOCSpanDataInstrumented(ctx, "aws-xray", exp, td)
+	return exporterwrapper.PushOcProtoSpansToOCTraceExporter(exp, td)
 }
 
 func (axe *awsXRayExporter) getOrMakeExporterByServiceName(serviceName string) (*xray.Exporter, error) {

exporter/awsexporter/aws_xray.go

@@ -15,14 +15,11 @@
 package datadogexporter
 
 import (
-	"context"
-
 	datadog "github.com/DataDog/opencensus-go-exporter-datadog"
 	"github.com/spf13/viper"
 
-	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/exporter"
-	"github.com/census-instrumentation/opencensus-service/exporter/exporterparser"
+	"github.com/census-instrumentation/opencensus-service/exporter/exporterwrapper"
+	"github.com/census-instrumentation/opencensus-service/processor"
 )
 
 type datadogConfig struct {
@@ -44,13 +41,9 @@ type datadogConfig struct {
 	EnableMetrics bool `mapstructure:"enable_metrics,omitempty"`
 }
 
-type datadogExporter struct {
-	exporter *datadog.Exporter
-}
-
 // DatadogTraceExportersFromViper unmarshals the viper and returns an exporter.TraceExporter targeting
 // Datadog according to the configuration settings.
-func DatadogTraceExportersFromViper(v *viper.Viper) (tes []exporter.TraceExporter, mes []exporter.MetricsExporter, doneFns []func() error, err error) {
+func DatadogTraceExportersFromViper(v *viper.Viper) (tdps []processor.TraceDataProcessor, mdps []processor.MetricsDataProcessor, doneFns []func() error, err error) {
 	var cfg struct {
 		Datadog *datadogConfig `mapstructure:"datadog,omitempty"`
 	}
@@ -78,18 +71,13 @@ func DatadogTraceExportersFromViper(v *viper.Viper) (tes []exporter.TraceExporte
 		return nil
 	})
 
-	dexp := &datadogExporter{exporter: de}
-	tes = append(tes, dexp)
+	// 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.
+	tdps = append(tdps, exporterwrapper.NewExporterWrapper("datadog", de))
 
 	// TODO: (@odeke-em, @songya23) implement ExportMetrics for Datadog.
 	// mes = append(mes, oexp)
 
 	return
 }
-
-func (dde *datadogExporter) ExportSpans(ctx context.Context, td data.TraceData) error {
-	// 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.
-	return exporterparser.OcProtoSpansToOCSpanDataInstrumented(ctx, "datadog", dde.exporter, td)
-}

exporter/datadogexporter/datadog.go

@@ -12,14 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Package exporterparser provides support for parsing and creating the
-// respective exporters given a viper configuration.
+// Package exporterwrapper provides support for wrapping OC go library trace.Exporter into a
+// processor.TraceDataProcessor.
 // For now it currently only provides statically imported OpenCensus
 // exporters like:
 //  * Stackdriver Tracing and Monitoring
 //  * DataDog
 //  * Zipkin
-package exporterparser
+package exporterwrapper
 
 import (
 	"context"
@@ -29,41 +29,60 @@ import (
 	tracepb "github.com/census-instrumentation/opencensus-proto/gen-go/trace/v1"
 	"github.com/census-instrumentation/opencensus-service/data"
 	"github.com/census-instrumentation/opencensus-service/internal"
+	"github.com/census-instrumentation/opencensus-service/processor"
 	spandatatranslator "github.com/census-instrumentation/opencensus-service/translator/trace/spandata"
 )
 
-// OcProtoSpansToOCSpanDataInstrumented converts
-// OpenCensus Proto TraceData to OpenCensus-Go SpanData.
-// The "Instrumented" suffix serves to document that this
-// function is traced but also has stats for self-observability.
+// NewExporterWrapper returns a processor.TraceDataProcessor that converts OpenCensus Proto TraceData
+// to OpenCensus-Go SpanData and calls into the given trace.Exporter.
 //
 // This is a bootstrapping mechanism for us to re-use as many of
 // the OpenCensus-Go trace.SpanData exporters which were written
 // by various vendors and contributors. Eventually the goal is to
 // get those exporters converted to directly receive
 // OpenCensus Proto TraceData.
-func OcProtoSpansToOCSpanDataInstrumented(ctx context.Context, exporterName string, te trace.Exporter, td data.TraceData) (aerr error) {
+func NewExporterWrapper(exporterName string, ocExporter trace.Exporter) processor.TraceDataProcessor {
+	return &ocExporterWrapper{spanName: "opencensus.service.exporter." + exporterName + ".ExportTrace", ocExporter: ocExporter}
+}
+
+type ocExporterWrapper struct {
+	spanName   string
+	ocExporter trace.Exporter
+}
+
+var _ processor.TraceDataProcessor = (*ocExporterWrapper)(nil)
+
+func (octew *ocExporterWrapper) ProcessTraceData(ctx context.Context, td data.TraceData) (aerr error) {
 	ctx, span := trace.StartSpan(ctx,
-		"opencensus.service.exporter."+exporterName+".ExportTrace",
-		trace.WithSampler(trace.NeverSample()))
+		octew.spanName, trace.WithSampler(trace.NeverSample()))
 
-	span.Annotate([]trace.Attribute{
-		trace.Int64Attribute("n_spans", int64(len(td.Spans))),
-	}, "")
+	if span.IsRecordingEvents() {
+		span.Annotate([]trace.Attribute{
+			trace.Int64Attribute("n_spans", int64(len(td.Spans))),
+		}, "")
+	}
 
 	defer func() {
-		if aerr != nil {
+		if aerr != nil && span.IsRecordingEvents() {
 			span.SetStatus(trace.Status{Code: trace.StatusCodeInternal, Message: aerr.Error()})
 		}
 		span.End()
 	}()
 
+	return PushOcProtoSpansToOCTraceExporter(octew.ocExporter, td)
+}
+
+// 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 {
 	var errs []error
 	var goodSpans []*tracepb.Span
 	for _, span := range td.Spans {
 		sd, err := spandatatranslator.ProtoSpanToOCSpanData(span)
 		if err == nil {
-			te.ExportSpan(sd)
+			ocExporter.ExportSpan(sd)
 			goodSpans = append(goodSpans, span)
 		} else {
 			errs = append(errs, err)

exporter/exporterparser/exparser.go …orter/exporterwrapper/exporterwrapper.goexporter/exporterparser/exparser.go renamed to exporter/exporterwrapper/exporterwrapper.go exporter/exporterparser/exparser.go renamed to exporter/exporterwrapper/exporterwrapper.go

@@ -18,14 +18,11 @@ package honeycombexporter
 // ask them to make an exporter that uses OpenCensus-Proto instead of OpenCensus-Go.
 
 import (
-	"context"
-
 	"github.com/honeycombio/opencensus-exporter/honeycomb"
 	"github.com/spf13/viper"
 
-	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/exporter"
-	"github.com/census-instrumentation/opencensus-service/exporter/exporterparser"
+	"github.com/census-instrumentation/opencensus-service/exporter/exporterwrapper"
+	"github.com/census-instrumentation/opencensus-service/processor"
 )
 
 type honeycombConfig struct {
@@ -35,7 +32,7 @@ type honeycombConfig struct {
 
 // HoneycombTraceExportersFromViper unmarshals the viper and returns an exporter.TraceExporter
 // targeting Honeycomb according to the configuration settings.
-func HoneycombTraceExportersFromViper(v *viper.Viper) (tes []exporter.TraceExporter, mes []exporter.MetricsExporter, doneFns []func() error, err error) {
+func HoneycombTraceExportersFromViper(v *viper.Viper) (tdps []processor.TraceDataProcessor, mdps []processor.MetricsDataProcessor, doneFns []func() error, err error) {
 	var cfg struct {
 		Honeycomb *honeycombConfig `mapstructure:"honeycomb"`
 	}
@@ -49,20 +46,11 @@ func HoneycombTraceExportersFromViper(v *viper.Viper) (tes []exporter.TraceExpor
 	}
 
 	rawExp := honeycomb.NewExporter(hc.WriteKey, hc.DatasetName)
-	hce := &honeycombExporter{exporter: rawExp}
 
-	tes = append(tes, hce)
+	tdps = append(tdps, exporterwrapper.NewExporterWrapper("honeycomb", rawExp))
 	doneFns = append(doneFns, func() error {
 		rawExp.Close()
 		return nil
 	})
 	return
 }
-
-type honeycombExporter struct {
-	exporter *honeycomb.Exporter
-}
-
-func (hce *honeycombExporter) ExportSpans(ctx context.Context, td data.TraceData) error {
-	return exporterparser.OcProtoSpansToOCSpanDataInstrumented(ctx, "honeycomb", hce.exporter, td)
-}

exporter/honeycombexporter/honeycomb.go

@@ -15,14 +15,11 @@
 package jaegerexporter
 
 import (
-	"context"
-
 	"github.com/spf13/viper"
 	"go.opencensus.io/exporter/jaeger"
 
-	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/exporter"
-	"github.com/census-instrumentation/opencensus-service/exporter/exporterparser"
+	"github.com/census-instrumentation/opencensus-service/exporter/exporterwrapper"
+	"github.com/census-instrumentation/opencensus-service/processor"
 )
 
 // Slight modified version of go/src/go.opencensus.io/exporter/jaeger/jaeger.go
@@ -33,13 +30,9 @@ type jaegerConfig struct {
 	ServiceName       string `mapstructure:"service_name,omitempty"`
 }
 
-type jaegerExporter struct {
-	exporter *jaeger.Exporter
-}
-
 // JaegerExportersFromViper unmarshals the viper and returns exporter.TraceExporters targeting
 // Jaeger according to the configuration settings.
-func JaegerExportersFromViper(v *viper.Viper) (tes []exporter.TraceExporter, mes []exporter.MetricsExporter, doneFns []func() error, err error) {
+func JaegerExportersFromViper(v *viper.Viper) (tdps []processor.TraceDataProcessor, mdps []processor.MetricsDataProcessor, doneFns []func() error, err error) {
 	var cfg struct {
 		Jaeger *jaegerConfig `mapstructure:"jaeger"`
 	}
@@ -68,13 +61,9 @@ func JaegerExportersFromViper(v *viper.Viper) (tes []exporter.TraceExporter, mes
 		je.Flush()
 		return nil
 	})
-	tes = append(tes, &jaegerExporter{exporter: je})
-	return
-}
-
-func (je *jaegerExporter) ExportSpans(ctx context.Context, td data.TraceData) error {
 	// 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.
-	return exporterparser.OcProtoSpansToOCSpanDataInstrumented(ctx, "jaeger", je.exporter, td)
+	tdps = append(tdps, exporterwrapper.NewExporterWrapper("jaeger", je))
+	return
 }