Remove SpanProcessor and replace it with TraceConsumer. (#491)

Author: Bogdan Drutu

cmd/occollector/app/collector/collector.go

@@ -29,7 +29,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/builder"
-	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
+	"github.com/census-instrumentation/opencensus-service/consumer"
 	"github.com/census-instrumentation/opencensus-service/internal/config/viperutils"
 	"github.com/census-instrumentation/opencensus-service/internal/pprofserver"
 	"github.com/census-instrumentation/opencensus-service/receiver"
@@ -45,7 +45,7 @@ type Application struct {
 	v           *viper.Viper
 	logger      *zap.Logger
 	healthCheck *healthcheck.HealthCheck
-	processor   processor.SpanProcessor
+	processor   consumer.TraceConsumer
 	receivers   []receiver.TraceReceiver
 }
 

cmd/occollector/app/collector/processors.go

@@ -34,6 +34,7 @@ import (
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor/tailsampling"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/sampling"
 	"github.com/census-instrumentation/opencensus-service/internal/config"
+	"github.com/census-instrumentation/opencensus-service/processor/multiconsumer"
 )
 
 func createExporters(v *viper.Viper, logger *zap.Logger) ([]func(), []consumer.TraceConsumer, []consumer.MetricsConsumer) {
@@ -59,11 +60,11 @@ func createExporters(v *viper.Viper, logger *zap.Logger) ([]func(), []consumer.T
 
 func buildQueuedSpanProcessor(
 	logger *zap.Logger, opts *builder.QueuedSpanProcessorCfg,
-) (closeFns []func(), queuedSpanProcessor processor.SpanProcessor, err error) {
+) (closeFns []func(), queuedSpanProcessor consumer.TraceConsumer, err error) {
 	logger.Info("Constructing queue processor with name", zap.String("name", opts.Name))
 
 	// build span batch sender from configured options
-	var spanSender processor.SpanProcessor
+	var spanSender consumer.TraceConsumer
 	switch opts.SenderType {
 	case builder.ThriftTChannelSenderType:
 		logger.Info("Initializing thrift-tChannel sender")
@@ -99,14 +100,12 @@ func buildQueuedSpanProcessor(
 		logger.Fatal("No senders or exporters configured.")
 	}
 
-	allSendersAndExporters := make([]processor.SpanProcessor, 0, 1+len(traceExporters))
+	allSendersAndExporters := make([]consumer.TraceConsumer, 0, 1+len(traceExporters))
 	if spanSender != nil {
 		allSendersAndExporters = append(allSendersAndExporters, spanSender)
 	}
 	for _, traceExporter := range traceExporters {
-		allSendersAndExporters = append(
-			allSendersAndExporters, processor.NewTraceExporterProcessor(traceExporter),
-		)
+		allSendersAndExporters = append(allSendersAndExporters, traceExporter)
 	}
 
 	var batchingOptions []nodebatcher.Option
@@ -137,11 +136,11 @@ func buildQueuedSpanProcessor(
 		}
 	}
 
-	queuedProcessors := make([]processor.SpanProcessor, 0, len(allSendersAndExporters))
+	queuedConsumers := make([]consumer.TraceConsumer, 0, len(allSendersAndExporters))
 	for _, senderOrExporter := range allSendersAndExporters {
 		// build queued span processor with underlying sender
-		queuedProcessors = append(
-			queuedProcessors,
+		queuedConsumers = append(
+			queuedConsumers,
 			queued.NewQueuedSpanProcessor(
 				senderOrExporter,
 				queued.Options.WithLogger(logger),
@@ -155,10 +154,10 @@ func buildQueuedSpanProcessor(
 			),
 		)
 	}
-	return doneFns, processor.NewMultiSpanProcessor(queuedProcessors), nil
+	return doneFns, processor.NewMultiSpanProcessor(queuedConsumers), nil
 }
 
-func buildSamplingProcessor(cfg *builder.SamplingCfg, nameToSpanProcessor map[string]processor.SpanProcessor, v *viper.Viper, logger *zap.Logger) (processor.SpanProcessor, error) {
+func buildSamplingProcessor(cfg *builder.SamplingCfg, nameToTraceConsumer map[string]consumer.TraceConsumer, v *viper.Viper, logger *zap.Logger) (consumer.TraceConsumer, error) {
 	var policies []*tailsampling.Policy
 	seenExporter := make(map[string]bool)
 	for _, polCfg := range cfg.Policies {
@@ -183,14 +182,14 @@ func buildSamplingProcessor(cfg *builder.SamplingCfg, nameToSpanProcessor map[st
 			return nil, fmt.Errorf("unknown sampling policy %s", polCfg.Name)
 		}
 
-		var policyProcessors []processor.SpanProcessor
+		var policyProcessors []consumer.TraceConsumer
 		for _, exporter := range polCfg.Exporters {
 			if _, ok := seenExporter[exporter]; ok {
 				return nil, fmt.Errorf("multiple sampling polices pointing to exporter %q", exporter)
 			}
 			seenExporter[exporter] = true
 
-			policyProcessor, ok := nameToSpanProcessor[exporter]
+			policyProcessor, ok := nameToTraceConsumer[exporter]
 			if !ok {
 				return nil, fmt.Errorf("invalid exporter %q for sampling policy %q", exporter, polCfg.Name)
 			}
@@ -225,12 +224,12 @@ func buildSamplingProcessor(cfg *builder.SamplingCfg, nameToSpanProcessor map[st
 	return tailSamplingProcessor, err
 }
 
-func startProcessor(v *viper.Viper, logger *zap.Logger) (processor.SpanProcessor, []func()) {
+func startProcessor(v *viper.Viper, logger *zap.Logger) (consumer.TraceConsumer, []func()) {
 	// Build pipeline from its end: 1st exporters, the OC-proto queue processor, and
 	// finally the receivers.
 	var closeFns []func()
-	var spanProcessors []processor.SpanProcessor
-	nameToSpanProcessor := make(map[string]processor.SpanProcessor)
+	var traceConsumers []consumer.TraceConsumer
+	nameToTraceConsumer := make(map[string]consumer.TraceConsumer)
 	exportersCloseFns, traceExporters, metricsExporters := createExporters(v, logger)
 	closeFns = append(closeFns, exportersCloseFns...)
 	if len(traceExporters) > 0 {
@@ -239,20 +238,19 @@ func startProcessor(v *viper.Viper, logger *zap.Logger) (processor.SpanProcessor
 		// TODO: (@pjanotti) we should avoid this step in the long run, its an extra hop just to re-use
 		// the exporters: this can lose node information and it is not ideal for performance and delegates
 		// the retry/buffering to the exporters (that are designed to run within the tracing process).
-		traceExpProc := processor.NewTraceExporterProcessor(traceExporters...)
-		nameToSpanProcessor["exporters"] = traceExpProc
-		spanProcessors = append(spanProcessors, traceExpProc)
+		traceExpProc := multiconsumer.NewTraceProcessor(traceExporters)
+		nameToTraceConsumer["exporters"] = traceExpProc
+		traceConsumers = append(traceConsumers, traceExpProc)
 	}
 
 	// TODO: (@pjanotti) make use of metrics exporters
 	_ = metricsExporters
 
 	if builder.LoggingExporterEnabled(v) {
-		tle, _ := loggingexporter.NewTraceExporter(logger)
-		dbgProc := processor.NewTraceExporterProcessor(tle)
+		dbgProc, _ := loggingexporter.NewTraceExporter(logger)
 		// TODO: Add this to the exporters list and avoid treating it specially. Don't know all the implications.
-		nameToSpanProcessor["debug"] = dbgProc
-		spanProcessors = append(spanProcessors, dbgProc)
+		nameToTraceConsumer["debug"] = dbgProc
+		traceConsumers = append(traceConsumers, dbgProc)
 	}
 
 	multiProcessorCfg := builder.NewDefaultMultiSpanProcessorCfg().InitFromViper(v)
@@ -263,21 +261,21 @@ func startProcessor(v *viper.Viper, logger *zap.Logger) (processor.SpanProcessor
 			logger.Error("Failed to build the queued span processor", zap.Error(err))
 			os.Exit(1)
 		}
-		nameToSpanProcessor[queuedJaegerProcessorCfg.Name] = queuedJaegerProcessor
-		spanProcessors = append(spanProcessors, queuedJaegerProcessor)
+		nameToTraceConsumer[queuedJaegerProcessorCfg.Name] = queuedJaegerProcessor
+		traceConsumers = append(traceConsumers, queuedJaegerProcessor)
 		closeFns = append(closeFns, doneFns...)
 	}
 
-	if len(spanProcessors) == 0 {
+	if len(traceConsumers) == 0 {
 		logger.Warn("Nothing to do: no processor was enabled. Shutting down.")
 		os.Exit(1)
 	}
 
-	var tailSamplingProcessor processor.SpanProcessor
+	var tailSamplingProcessor consumer.TraceConsumer
 	samplingProcessorCfg := builder.NewDefaultSamplingCfg().InitFromViper(v)
 	if samplingProcessorCfg.Mode == builder.TailSampling {
 		var err error
-		tailSamplingProcessor, err = buildSamplingProcessor(samplingProcessorCfg, nameToSpanProcessor, v, logger)
+		tailSamplingProcessor, err = buildSamplingProcessor(samplingProcessorCfg, nameToTraceConsumer, v, logger)
 		if err != nil {
 			logger.Error("Falied to build the sampling processor", zap.Error(err))
 			os.Exit(1)
@@ -287,7 +285,7 @@ func startProcessor(v *viper.Viper, logger *zap.Logger) (processor.SpanProcessor
 			{
 				Name:        "tail-always-sampling",
 				Evaluator:   sampling.NewAlwaysSample(),
-				Destination: processor.NewMultiSpanProcessor(spanProcessors),
+				Destination: processor.NewMultiSpanProcessor(traceConsumers),
 			},
 		}
 		var err error
@@ -301,7 +299,7 @@ func startProcessor(v *viper.Viper, logger *zap.Logger) (processor.SpanProcessor
 
 	if tailSamplingProcessor != nil {
 		// SpanProcessors are going to go all via the tail sampling processor.
-		spanProcessors = []processor.SpanProcessor{tailSamplingProcessor}
+		traceConsumers = []consumer.TraceConsumer{tailSamplingProcessor}
 	}
 
 	// Wraps processors in a single one to be connected to all enabled receivers.
@@ -320,5 +318,5 @@ func startProcessor(v *viper.Viper, logger *zap.Logger) (processor.SpanProcessor
 			),
 		)
 	}
-	return processor.NewMultiSpanProcessor(spanProcessors, processorOptions...), closeFns
+	return processor.NewMultiSpanProcessor(traceConsumers, processorOptions...), closeFns
 }

cmd/occollector/app/collector/receivers.go

@@ -22,18 +22,18 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/builder"
+	"github.com/census-instrumentation/opencensus-service/consumer"
 	jaegerreceiver "github.com/census-instrumentation/opencensus-service/internal/collector/jaeger"
 	ocreceiver "github.com/census-instrumentation/opencensus-service/internal/collector/opencensus"
-	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
 	zipkinreceiver "github.com/census-instrumentation/opencensus-service/internal/collector/zipkin"
 	zipkinscribereceiver "github.com/census-instrumentation/opencensus-service/internal/collector/zipkin/scribe"
 	"github.com/census-instrumentation/opencensus-service/receiver"
 )
 
-func createReceivers(v *viper.Viper, logger *zap.Logger, spanProcessor processor.SpanProcessor) []receiver.TraceReceiver {
+func createReceivers(v *viper.Viper, logger *zap.Logger, traceConsumers consumer.TraceConsumer) []receiver.TraceReceiver {
 	var someReceiverEnabled bool
 	receivers := []struct {
-		runFn   func(*zap.Logger, *viper.Viper, processor.SpanProcessor) (receiver.TraceReceiver, error)
+		runFn   func(*zap.Logger, *viper.Viper, consumer.TraceConsumer) (receiver.TraceReceiver, error)
 		enabled bool
 	}{
 		{jaegerreceiver.Start, builder.JaegerReceiverEnabled(v)},
@@ -45,7 +45,7 @@ func createReceivers(v *viper.Viper, logger *zap.Logger, spanProcessor processor
 	var startedTraceReceivers []receiver.TraceReceiver
 	for _, receiver := range receivers {
 		if receiver.enabled {
-			rec, err := receiver.runFn(logger, v, spanProcessor)
+			rec, err := receiver.runFn(logger, v, traceConsumers)
 			if err != nil {
 				// TODO: (@pjanotti) better shutdown, for now just try to stop any started receiver before terminating.
 				for _, startedTraceReceiver := range startedTraceReceivers {

cmd/occollector/app/sender/jaeger_thrift_http_sender.go

@@ -26,8 +26,8 @@ import (
 	"github.com/apache/thrift/lib/go/thrift"
 	"go.uber.org/zap"
 
+	"github.com/census-instrumentation/opencensus-service/consumer"
 	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
 	jaegertranslator "github.com/census-instrumentation/opencensus-service/translator/trace/jaeger"
 )
 
@@ -43,7 +43,7 @@ type JaegerThriftHTTPSender struct {
 	logger  *zap.Logger
 }
 
-var _ processor.SpanProcessor = (*JaegerThriftHTTPSender)(nil)
+var _ consumer.TraceConsumer = (*JaegerThriftHTTPSender)(nil)
 
 // HTTPOption sets a parameter for the HttpCollector
 type HTTPOption func(s *JaegerThriftHTTPSender)
@@ -83,8 +83,8 @@ func NewJaegerThriftHTTPSender(
 	return s
 }
 
-// ProcessSpans sends the received data to the configured Jaeger Thrift end-point.
-func (s *JaegerThriftHTTPSender) ProcessSpans(ctx context.Context, td data.TraceData) error {
+// ConsumeTraceData sends the received data to the configured Jaeger Thrift end-point.
+func (s *JaegerThriftHTTPSender) ConsumeTraceData(ctx context.Context, td data.TraceData) error {
 	// TODO: (@pjanotti) In case of failure the translation to Jaeger Thrift is going to be remade, cache it somehow.
 	tBatch, err := jaegertranslator.OCProtoToJaegerThrift(td)
 	if err != nil {

cmd/occollector/app/sender/jaeger_thrift_tchannel_sender.go

@@ -21,8 +21,8 @@ import (
 
 	reporter "github.com/jaegertracing/jaeger/cmd/agent/app/reporter"
 
+	"github.com/census-instrumentation/opencensus-service/consumer"
 	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
 	jaegertranslator "github.com/census-instrumentation/opencensus-service/translator/trace/jaeger"
 )
 
@@ -33,7 +33,7 @@ type JaegerThriftTChannelSender struct {
 	reporter reporter.Reporter
 }
 
-var _ processor.SpanProcessor = (*JaegerThriftTChannelSender)(nil)
+var _ consumer.TraceConsumer = (*JaegerThriftTChannelSender)(nil)
 
 // NewJaegerThriftTChannelSender creates new TChannel-based sender.
 func NewJaegerThriftTChannelSender(
@@ -46,8 +46,8 @@ func NewJaegerThriftTChannelSender(
 	}
 }
 
-// ProcessSpans sends the received data to the configured Jaeger Thrift end-point.
-func (s *JaegerThriftTChannelSender) ProcessSpans(ctx context.Context, td data.TraceData) error {
+// ConsumeTraceData sends the received data to the configured Jaeger Thrift end-point.
+func (s *JaegerThriftTChannelSender) ConsumeTraceData(ctx context.Context, td data.TraceData) error {
 	// TODO: (@pjanotti) In case of failure the translation to Jaeger Thrift is going to be remade, cache it somehow.
 	tBatch, err := jaegertranslator.OCProtoToJaegerThrift(td)
 	if err != nil {

internal/collector/jaeger/receiver.go

@@ -29,13 +29,14 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/builder"
+	"github.com/census-instrumentation/opencensus-service/consumer"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
 	"github.com/census-instrumentation/opencensus-service/receiver"
 	"github.com/census-instrumentation/opencensus-service/receiver/jaegerreceiver"
 )
 
 // Start starts the Jaeger receiver endpoint.
-func Start(logger *zap.Logger, v *viper.Viper, spanProc processor.SpanProcessor) (receiver.TraceReceiver, error) {
+func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsumer) (receiver.TraceReceiver, error) {
 	rOpts, err := builder.NewDefaultJaegerReceiverCfg().InitFromViper(v)
 	if err != nil {
 		return nil, err
@@ -50,7 +51,7 @@ func Start(logger *zap.Logger, v *viper.Viper, spanProc processor.SpanProcessor)
 		return nil, err
 	}
 
-	ss := processor.WrapWithSpanSink("jaeger", spanProc)
+	ss := processor.WithSourceName("jaeger", traceConsumer)
 	if err := jtr.StartTraceReception(ctx, ss); err != nil {
 		return nil, err
 	}

internal/collector/opencensus/receiver.go

@@ -25,13 +25,14 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/builder"
+	"github.com/census-instrumentation/opencensus-service/consumer"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
 	"github.com/census-instrumentation/opencensus-service/receiver"
 	"github.com/census-instrumentation/opencensus-service/receiver/opencensusreceiver"
 )
 
 // Start starts the OpenCensus receiver endpoint.
-func Start(logger *zap.Logger, v *viper.Viper, spanProc processor.SpanProcessor) (receiver.TraceReceiver, error) {
+func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsumer) (receiver.TraceReceiver, error) {
 	rOpts, err := builder.NewDefaultOpenCensusReceiverCfg().InitFromViper(v)
 	if err != nil {
 		return nil, err
@@ -47,7 +48,7 @@ func Start(logger *zap.Logger, v *viper.Viper, spanProc processor.SpanProcessor)
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create the OpenCensus trace receiver: %v", err)
 	}
-	ss := processor.WrapWithSpanSink("oc", spanProc)
+	ss := processor.WithSourceName("oc_trace", traceConsumer)
 	if err := ocr.StartTraceReception(context.Background(), ss); err != nil {
 		return nil, fmt.Errorf("Cannot bind Opencensus receiver to address %q: %v", addr, err)
 	}