Make all receivers take "next" on creation (#477)

* Make all receivers take "next" on creation

This is required in order to move the code to use factories for the receivers. The "next" parameter will be passed to the factory and then to the receiver constructor, this way Start[Trace|Metrics]Reception doesn't need the next parameter anymore. Since typically the Start* methods launch goroutines took the opportunity to add a channel to report asynchronous errors.

Opportunistic change: renamed WrapWithSpanSink and related variables.

Remaing items from previous commit

* Improve test coverage for zipkin-scribe

Improving code coverage for zipkin-scribe that went a tad down with the initial PR.

* Improve a little code coverage  zipkin receiver

* Fix test broken during rebase

Author: Paulo Janotti

cmd/ocagent/main.go

@@ -122,7 +122,7 @@ func runOCAgent() {
 	commonMetricsSink := multiconsumer.NewMetricsProcessor(metricsExporters)
 
 	// Add other receivers here as they are implemented
-	ocReceiverDoneFn, err := runOCReceiver(logger, &agentConfig, commonSpanSink, commonMetricsSink)
+	ocReceiverDoneFn, err := runOCReceiver(logger, &agentConfig, commonSpanSink, commonMetricsSink, asyncErrorChan)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -139,15 +139,15 @@ func runOCAgent() {
 	// If the Zipkin receiver is enabled, then run it
 	if agentConfig.ZipkinReceiverEnabled() {
 		zipkinReceiverAddr := agentConfig.ZipkinReceiverAddress()
-		zipkinReceiverDoneFn, err := runZipkinReceiver(zipkinReceiverAddr, commonSpanSink)
+		zipkinReceiverDoneFn, err := runZipkinReceiver(zipkinReceiverAddr, commonSpanSink, asyncErrorChan)
 		if err != nil {
 			log.Fatal(err)
 		}
 		closeFns = append(closeFns, zipkinReceiverDoneFn)
 	}
 
 	if agentConfig.ZipkinScribeReceiverEnabled() {
-		zipkinScribeDoneFn, err := runZipkinScribeReceiver(agentConfig.ZipkinScribeConfig(), commonSpanSink)
+		zipkinScribeDoneFn, err := runZipkinScribeReceiver(agentConfig.ZipkinScribeConfig(), commonSpanSink, asyncErrorChan)
 		if err != nil {
 			log.Fatal(err)
 		}
@@ -156,7 +156,7 @@ func runOCAgent() {
 
 	if agentConfig.JaegerReceiverEnabled() {
 		collectorHTTPPort, collectorThriftPort := agentConfig.JaegerReceiverPorts()
-		jaegerDoneFn, err := runJaegerReceiver(collectorThriftPort, collectorHTTPPort, commonSpanSink)
+		jaegerDoneFn, err := runJaegerReceiver(collectorThriftPort, collectorHTTPPort, commonSpanSink, asyncErrorChan)
 		if err != nil {
 			log.Fatal(err)
 		}
@@ -165,7 +165,7 @@ func runOCAgent() {
 
 	// If the Prometheus receiver is enabled, then run it.
 	if agentConfig.PrometheusReceiverEnabled() {
-		promDoneFn, err := runPrometheusReceiver(viperCfg, commonMetricsSink)
+		promDoneFn, err := runPrometheusReceiver(viperCfg, commonMetricsSink, asyncErrorChan)
 		if err != nil {
 			log.Fatal(err)
 		}
@@ -214,14 +214,16 @@ func runZPages(port int) func() error {
 	return srv.Close
 }
 
-func runOCReceiver(logger *zap.Logger, acfg *config.Config, tdp consumer.TraceConsumer, mdp consumer.MetricsConsumer) (doneFn func() error, err error) {
+func runOCReceiver(logger *zap.Logger, acfg *config.Config, tc consumer.TraceConsumer, mc consumer.MetricsConsumer, asyncErrorChan chan<- error) (doneFn func() error, err error) {
 	tlsCredsOption, hasTLSCreds, err := acfg.OpenCensusReceiverTLSCredentialsServerOption()
 	if err != nil {
 		return nil, fmt.Errorf("OpenCensus receiver TLS Credentials: %v", err)
 	}
 	addr := acfg.OpenCensusReceiverAddress()
 	corsOrigins := acfg.OpenCensusReceiverCorsAllowedOrigins()
 	ocr, err := opencensusreceiver.New(addr,
+		tc,
+		mc,
 		tlsCredsOption,
 		opencensusreceiver.WithCorsOrigins(corsOrigins))
 
@@ -242,19 +244,19 @@ func runOCReceiver(logger *zap.Logger, acfg *config.Config, tdp consumer.TraceCo
 
 	switch {
 	case acfg.CanRunOpenCensusTraceReceiver() && acfg.CanRunOpenCensusMetricsReceiver():
-		if err := ocr.Start(ctx, tdp, mdp); err != nil {
+		if err := ocr.Start(ctx); err != nil {
 			return nil, fmt.Errorf("failed to start Trace and Metrics Receivers: %v", err)
 		}
 		log.Printf("Running OpenCensus Trace and Metrics receivers as a gRPC service at %q", addr)
 
 	case acfg.CanRunOpenCensusTraceReceiver():
-		if err := ocr.StartTraceReception(ctx, tdp); err != nil {
+		if err := ocr.StartTraceReception(ctx, asyncErrorChan); err != nil {
 			return nil, fmt.Errorf("failed to start TraceReceiver: %v", err)
 		}
 		log.Printf("Running OpenCensus Trace receiver as a gRPC service at %q", addr)
 
 	case acfg.CanRunOpenCensusMetricsReceiver():
-		if err := ocr.StartMetricsReception(ctx, mdp); err != nil {
+		if err := ocr.StartMetricsReception(ctx, asyncErrorChan); err != nil {
 			return nil, fmt.Errorf("failed to start MetricsReceiver: %v", err)
 		}
 		log.Printf("Running OpenCensus Metrics receiver as a gRPC service at %q", addr)
@@ -271,19 +273,20 @@ func runOCReceiver(logger *zap.Logger, acfg *config.Config, tdp consumer.TraceCo
 	return doneFn, nil
 }
 
-func runJaegerReceiver(collectorThriftPort, collectorHTTPPort int, next consumer.TraceConsumer) (doneFn func() error, err error) {
-	jtr, err := jaegerreceiver.New(context.Background(), &jaegerreceiver.Configuration{
+func runJaegerReceiver(collectorThriftPort, collectorHTTPPort int, next consumer.TraceConsumer, asyncErrorChan chan<- error) (doneFn func() error, err error) {
+	config := &jaegerreceiver.Configuration{
 		CollectorThriftPort: collectorThriftPort,
 		CollectorHTTPPort:   collectorHTTPPort,
 
 		// TODO: (@odeke-em, @pjanotti) send a change
 		// to dynamically retrieve the Jaeger Agent's ports
 		// and not use their defaults of 5778, 6831, 6832
-	})
+	}
+	jtr, err := jaegerreceiver.New(context.Background(), config, next)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create new Jaeger receiver: %v", err)
 	}
-	if err := jtr.StartTraceReception(context.Background(), next); err != nil {
+	if err := jtr.StartTraceReception(context.Background(), asyncErrorChan); err != nil {
 		return nil, fmt.Errorf("failed to start Jaeger receiver: %v", err)
 	}
 	doneFn = func() error {
@@ -293,13 +296,13 @@ func runJaegerReceiver(collectorThriftPort, collectorHTTPPort int, next consumer
 	return doneFn, nil
 }
 
-func runZipkinReceiver(addr string, next consumer.TraceConsumer) (doneFn func() error, err error) {
-	zi, err := zipkinreceiver.New(addr)
+func runZipkinReceiver(addr string, next consumer.TraceConsumer, asyncErrorChan chan<- error) (doneFn func() error, err error) {
+	zi, err := zipkinreceiver.New(addr, next)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create the Zipkin receiver: %v", err)
 	}
 
-	if err := zi.StartTraceReception(context.Background(), next); err != nil {
+	if err := zi.StartTraceReception(context.Background(), asyncErrorChan); err != nil {
 		return nil, fmt.Errorf("cannot start Zipkin receiver with address %q: %v", addr, err)
 	}
 	doneFn = func() error {
@@ -309,13 +312,13 @@ func runZipkinReceiver(addr string, next consumer.TraceConsumer) (doneFn func()
 	return doneFn, nil
 }
 
-func runZipkinScribeReceiver(config *config.ScribeReceiverConfig, next consumer.TraceConsumer) (doneFn func() error, err error) {
-	zs, err := zipkinscribereceiver.NewReceiver(config.Address, config.Port, config.Category)
+func runZipkinScribeReceiver(config *config.ScribeReceiverConfig, next consumer.TraceConsumer, asyncErrorChan chan<- error) (doneFn func() error, err error) {
+	zs, err := zipkinscribereceiver.NewReceiver(config.Address, config.Port, config.Category, next)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create the Zipkin Scribe receiver: %v", err)
 	}
 
-	if err := zs.StartTraceReception(context.Background(), next); err != nil {
+	if err := zs.StartTraceReception(context.Background(), asyncErrorChan); err != nil {
 		return nil, fmt.Errorf("cannot start Zipkin Scribe receiver with %v: %v", config, err)
 	}
 	doneFn = func() error {
@@ -325,12 +328,12 @@ func runZipkinScribeReceiver(config *config.ScribeReceiverConfig, next consumer.
 	return doneFn, nil
 }
 
-func runPrometheusReceiver(v *viper.Viper, next consumer.MetricsConsumer) (doneFn func() error, err error) {
-	pmr, err := prometheusreceiver.New(v.Sub("receivers.prometheus"))
+func runPrometheusReceiver(v *viper.Viper, next consumer.MetricsConsumer, asyncErrorChan chan<- error) (doneFn func() error, err error) {
+	pmr, err := prometheusreceiver.New(v.Sub("receivers.prometheus"), next)
 	if err != nil {
 		return nil, err
 	}
-	if err := pmr.StartMetricsReception(context.Background(), next); err != nil {
+	if err := pmr.StartMetricsReception(context.Background(), asyncErrorChan); err != nil {
 		return nil, err
 	}
 	doneFn = func() error {

cmd/occollector/app/collector/collector.go

@@ -89,7 +89,7 @@ func (app *Application) execute() {
 	var closeFns []func()
 	app.processor, closeFns = startProcessor(app.v, app.logger)
 
-	app.receivers = createReceivers(app.v, app.logger, app.processor)
+	app.receivers = createReceivers(app.v, app.logger, app.processor, asyncErrorChannel)
 
 	err = initTelemetry(asyncErrorChannel, app.v, app.logger)
 	if err != nil {

cmd/occollector/app/collector/receivers.go

@@ -30,10 +30,10 @@ import (
 	"github.com/census-instrumentation/opencensus-service/receiver"
 )
 
-func createReceivers(v *viper.Viper, logger *zap.Logger, traceConsumers consumer.TraceConsumer) []receiver.TraceReceiver {
+func createReceivers(v *viper.Viper, logger *zap.Logger, traceConsumers consumer.TraceConsumer, asyncErrorChan chan<- error) []receiver.TraceReceiver {
 	var someReceiverEnabled bool
 	receivers := []struct {
-		runFn   func(*zap.Logger, *viper.Viper, consumer.TraceConsumer) (receiver.TraceReceiver, error)
+		runFn   func(*zap.Logger, *viper.Viper, consumer.TraceConsumer, chan<- error) (receiver.TraceReceiver, error)
 		enabled bool
 	}{
 		{jaegerreceiver.Start, builder.JaegerReceiverEnabled(v)},
@@ -45,7 +45,7 @@ func createReceivers(v *viper.Viper, logger *zap.Logger, traceConsumers consumer
 	var startedTraceReceivers []receiver.TraceReceiver
 	for _, receiver := range receivers {
 		if receiver.enabled {
-			rec, err := receiver.runFn(logger, v, traceConsumers)
+			rec, err := receiver.runFn(logger, v, traceConsumers, asyncErrorChan)
 			if err != nil {
 				// TODO: (@pjanotti) better shutdown, for now just try to stop any started receiver before terminating.
 				for _, startedTraceReceiver := range startedTraceReceivers {

exporter/zipkinexporter/zipkin_test.go

@@ -168,13 +168,13 @@ zipkin:
 	tes[0].(*zipkinExporter).reporter = mzr
 
 	// Run the Zipkin receiver to "receive spans upload from a client application"
-	zi, err := zipkinreceiver.New(":0")
+	zexp := multiconsumer.NewTraceProcessor(tes)
+	zi, err := zipkinreceiver.New(":0", zexp)
 	if err != nil {
 		t.Fatalf("Failed to create a new Zipkin receiver: %v", err)
 	}
 
-	zexp := multiconsumer.NewTraceProcessor(tes)
-	if err := zi.StartTraceReception(context.Background(), zexp); err != nil {
+	if err := zi.StartTraceReception(context.Background(), nil); err != nil {
 		t.Fatalf("Failed to start trace reception: %v", err)
 	}
 	defer zi.StopTraceReception(context.Background())

internal/collector/jaeger/receiver.go

@@ -35,22 +35,23 @@ import (
 )
 
 // Start starts the Jaeger receiver endpoint.
-func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsumer) (receiver.TraceReceiver, error) {
+func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsumer, asyncErrorChan chan<- error) (receiver.TraceReceiver, error) {
 	rOpts, err := builder.NewDefaultJaegerReceiverCfg().InitFromViper(v)
 	if err != nil {
 		return nil, err
 	}
 
 	ctx := context.Background()
-	jtr, err := jaegerreceiver.New(ctx, &jaegerreceiver.Configuration{
+	config := &jaegerreceiver.Configuration{
 		CollectorThriftPort: rOpts.ThriftTChannelPort,
 		CollectorHTTPPort:   rOpts.ThriftHTTPPort,
-	})
+	}
+	jtr, err := jaegerreceiver.New(ctx, config, traceConsumer)
 	if err != nil {
 		return nil, err
 	}
 
-	if err := jtr.StartTraceReception(ctx, traceConsumer); err != nil {
+	if err := jtr.StartTraceReception(ctx, asyncErrorChan); err != nil {
 		return nil, err
 	}
 

internal/collector/opencensus/receiver.go

@@ -31,7 +31,7 @@ import (
 )
 
 // Start starts the OpenCensus receiver endpoint.
-func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsumer) (receiver.TraceReceiver, error) {
+func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsumer, asyncErrorChan chan<- error) (receiver.TraceReceiver, error) {
 	rOpts, err := builder.NewDefaultOpenCensusReceiverCfg().InitFromViper(v)
 	if err != nil {
 		return nil, err
@@ -43,12 +43,12 @@ func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsu
 	}
 
 	addr := ":" + strconv.FormatInt(int64(rOpts.Port), 10)
-	ocr, err := opencensusreceiver.New(addr, tlsCredsOption)
+	ocr, err := opencensusreceiver.New(addr, traceConsumer, nil, tlsCredsOption)
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create the OpenCensus trace receiver: %v", err)
 	}
 
-	if err := ocr.StartTraceReception(context.Background(), traceConsumer); err != nil {
+	if err := ocr.StartTraceReception(context.Background(), asyncErrorChan); err != nil {
 		return nil, fmt.Errorf("Cannot bind Opencensus receiver to address %q: %v", addr, err)
 	}
 

internal/collector/zipkin/receiver.go

@@ -31,19 +31,19 @@ import (
 )
 
 // Start starts the Zipkin receiver endpoint.
-func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsumer) (receiver.TraceReceiver, error) {
+func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsumer, asyncErrorChan chan<- error) (receiver.TraceReceiver, error) {
 	rOpts, err := builder.NewDefaultZipkinReceiverCfg().InitFromViper(v)
 	if err != nil {
 		return nil, err
 	}
 
 	addr := ":" + strconv.FormatInt(int64(rOpts.Port), 10)
-	zi, err := zipkinreceiver.New(addr)
+	zi, err := zipkinreceiver.New(addr, traceConsumer)
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create the Zipkin receiver: %v", err)
 	}
 
-	if err := zi.StartTraceReception(context.Background(), traceConsumer); err != nil {
+	if err := zi.StartTraceReception(context.Background(), asyncErrorChan); err != nil {
 		return nil, fmt.Errorf("Cannot start Zipkin receiver to address %q: %v", addr, err)
 	}
 

internal/collector/zipkin/scribe/receiver.go

@@ -30,18 +30,18 @@ import (
 )
 
 // Start starts the Zipkin Scribe receiver endpoint.
-func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsumer) (receiver.TraceReceiver, error) {
+func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsumer, asyncErrorChan chan<- error) (receiver.TraceReceiver, error) {
 	rOpts, err := builder.NewDefaultZipkinScribeReceiverCfg().InitFromViper(v)
 	if err != nil {
 		return nil, err
 	}
 
-	sr, err := zipkinscribereceiver.NewReceiver(rOpts.Address, rOpts.Port, rOpts.Category)
+	sr, err := zipkinscribereceiver.NewReceiver(rOpts.Address, rOpts.Port, rOpts.Category, traceConsumer)
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create the Zipkin Scribe receiver: %v", err)
 	}
 
-	if err := sr.StartTraceReception(context.Background(), traceConsumer); err != nil {
+	if err := sr.StartTraceReception(context.Background(), asyncErrorChan); err != nil {
 		return nil, fmt.Errorf("Cannot start Zipkin Scribe receiver %+v: %v", rOpts, err)
 	}
 

receiver/end_to_end_test.go

@@ -33,7 +33,12 @@ func Example_endToEnd() {
 	// This is what the cmd/ocagent code would look like this.
 	// A trace receiver as per the trace receiver
 	// configs that have been parsed.
-	tr, err := opencensusreceiver.New("localhost:55678")
+	lte, err := loggingexporter.NewTraceExporter(zap.NewNop())
+	if err != nil {
+		log.Fatalf("Failed to create logging exporter: %v", err)
+	}
+
+	tr, err := opencensusreceiver.New("localhost:55678", lte, nil)
 	if err != nil {
 		log.Fatalf("Failed to create trace receiver: %v", err)
 	}
@@ -43,9 +48,8 @@ func Example_endToEnd() {
 
 	// Once we have the span receiver which will connect to the
 	// various exporter pipeline i.e. *tracepb.Span->OpenCensus.SpanData
-	lsr, _ := loggingexporter.NewTraceExporter(zap.NewNop())
 	for _, tr := range trl {
-		if err := tr.StartTraceReception(context.Background(), lsr); err != nil {
+		if err := tr.StartTraceReception(context.Background(), nil); err != nil {
 			log.Fatalf("Failed to start trace receiver: %v", err)
 		}
 	}

receiver/factorytemplate/factorytemplate_test.go

@@ -531,7 +531,7 @@ func (mr *mockReceiver) TraceSource() string {
 	return "mockReceiver"
 }
 
-func (mr *mockReceiver) StartTraceReception(ctx context.Context, nextConsumer consumer.TraceConsumer) error {
+func (mr *mockReceiver) StartTraceReception(ctx context.Context, asyncErrChan chan<- error) error {
 	if mr == nil {
 		panic("mockReceiver is nil")
 	}
@@ -552,7 +552,7 @@ func (mr *mockReceiver) MetricsSource() string {
 	return "mockReceiver"
 }
 
-func (mr *mockReceiver) StartMetricsReception(ctx context.Context, nextConsumer consumer.MetricsConsumer) error {
+func (mr *mockReceiver) StartMetricsReception(ctx context.Context, asyncErrChan chan<- error) error {
 	if mr == nil {
 		panic("mockReceiver is nil")
 	}