Add the source name to the receiver interface. (#435)

Author: Bogdan Drutu

cmd/occollector/app/collector/receivers.go

@@ -33,14 +33,13 @@ import (
 func createReceivers(v *viper.Viper, logger *zap.Logger, spanProcessor processor.SpanProcessor) []receiver.TraceReceiver {
 	var someReceiverEnabled bool
 	receivers := []struct {
-		name    string
 		runFn   func(*zap.Logger, *viper.Viper, processor.SpanProcessor) (receiver.TraceReceiver, error)
 		enabled bool
 	}{
-		{"Jaeger", jaegerreceiver.Start, builder.JaegerReceiverEnabled(v)},
-		{"OpenCensus", ocreceiver.Start, builder.OpenCensusReceiverEnabled(v)},
-		{"Zipkin", zipkinreceiver.Start, builder.ZipkinReceiverEnabled(v)},
-		{"Zipkin-Scribe", zipkinscribereceiver.Start, builder.ZipkinScribeReceiverEnabled(v)},
+		{jaegerreceiver.Start, builder.JaegerReceiverEnabled(v)},
+		{ocreceiver.Start, builder.OpenCensusReceiverEnabled(v)},
+		{zipkinreceiver.Start, builder.ZipkinReceiverEnabled(v)},
+		{zipkinscribereceiver.Start, builder.ZipkinScribeReceiverEnabled(v)},
 	}
 
 	var startedTraceReceivers []receiver.TraceReceiver
@@ -52,7 +51,7 @@ func createReceivers(v *viper.Viper, logger *zap.Logger, spanProcessor processor
 				for _, startedTraceReceiver := range startedTraceReceivers {
 					startedTraceReceiver.StopTraceReception(context.Background())
 				}
-				logger.Fatal("Cannot run receiver for "+receiver.name, zap.Error(err))
+				logger.Fatal("Cannot run receiver for "+rec.TraceSource(), zap.Error(err))
 			}
 			startedTraceReceivers = append(startedTraceReceivers, rec)
 			someReceiverEnabled = true

receiver/jaegerreceiver/trace_receiver.go

@@ -89,6 +89,8 @@ const (
 	defaultZipkinThriftUDPPort  = 5775
 	defaultCompactThriftUDPPort = 6831
 	defaultBinaryThriftUDPPort  = 6832
+
+	traceSource string = "Jaeger"
 )
 
 // New creates a TraceReceiver that receives traffic as a collector with both Thrift and HTTP transports.
@@ -160,6 +162,10 @@ func (jr *jReceiver) agentBinaryThriftAddr() string {
 	return fmt.Sprintf(":%d", port)
 }
 
+func (jr *jReceiver) TraceSource() string {
+	return traceSource
+}
+
 func (jr *jReceiver) StartTraceReception(ctx context.Context, nextProcessor processor.TraceDataProcessor) error {
 	jr.mu.Lock()
 	defer jr.mu.Unlock()

receiver/opencensusreceiver/opencensus.go

@@ -64,6 +64,8 @@ var (
 	errAlreadyStopped = errors.New("already stopped")
 )
 
+const source string = "OpenCensus"
+
 // New just creates the OpenCensus receiver services. It is the caller's
 // responsibility to invoke the respective Start*Reception methods as well
 // as the various Stop*Reception methods or simply Stop to end it.
@@ -87,6 +89,11 @@ func New(addr string, opts ...Option) (*Receiver, error) {
 	return ocr, nil
 }
 
+// TraceSource returns the name of the trace data source.
+func (ocr *Receiver) TraceSource() string {
+	return source
+}
+
 // StartTraceReception exclusively runs the Trace receiver on the gRPC server.
 // To start both Trace and Metrics receivers/services, please use Start.
 func (ocr *Receiver) StartTraceReception(ctx context.Context, ts processor.TraceDataProcessor) error {
@@ -111,6 +118,11 @@ func (ocr *Receiver) registerTraceDataProcessor(ts processor.TraceDataProcessor)
 	return err
 }
 
+// MetricsSource returns the name of the metrics data source.
+func (ocr *Receiver) MetricsSource() string {
+	return source
+}
+
 // StartMetricsReception exclusively runs the Metrics receiver on the gRPC server.
 // To start both Trace and Metrics receivers/services, please use Start.
 func (ocr *Receiver) StartMetricsReception(ctx context.Context, ms processor.MetricsDataProcessor) error {

receiver/prometheusreceiver/metrics_receiver.go

@@ -59,6 +59,13 @@ var (
 	errNilScrapeConfig        = errors.New("expecting a non-nil ScrapeConfig")
 )
 
+const metricsSource string = "Prometheus"
+
+// MetricsSource returns the name of the metrics data source.
+func (pr *Preceiver) MetricsSource() string {
+	return metricsSource
+}
+
 // StartMetricsReception is the method that starts Prometheus scraping and it
 // is controlled by having previously defined a Configuration using perhaps New.
 func (pr *Preceiver) StartMetricsReception(ctx context.Context, nextProcessor processor.MetricsDataProcessor) error {

receiver/receiver.go

@@ -24,28 +24,37 @@ import (
 // A TraceReceiver is an "arbitrary data"-to-"trace proto span" converter.
 // Its purpose is to translate data from the wild into trace proto accompanied
 // by a *commonpb.Node to uniquely identify where that data comes from.
-// TraceReceiver feeds a TraceReceiverSink with data.
+// TraceReceiver feeds a processor.TraceDataProcessor with data.
 //
 // For example it could be Zipkin data source which translates
 // Zipkin spans into *tracepb.Span-s.
-//
-// StartTraceReception tells the receiver to start its processing.
-//
-// StopTraceReception tells the receiver that should stop reception,
-// giving it a chance to perform any necessary clean-up.
 type TraceReceiver interface {
+	// TraceSource returns the name of the trace data source.
+	TraceSource() string
+
+	// StartTraceReception tells the receiver to start its processing.
 	StartTraceReception(ctx context.Context, nextProcessor processor.TraceDataProcessor) error
+
+	// StopTraceReception tells the receiver that should stop reception,
+	// giving it a chance to perform any necessary clean-up.
 	StopTraceReception(ctx context.Context) error
 }
 
 // A MetricsReceiver is an "arbitrary data"-to-"metric proto" converter.
 // Its purpose is to translate data from the wild into metric proto accompanied
 // by a *commonpb.Node to uniquely identify where that data comes from.
-// MetricsReceiver feeds a MetricsReceiverSink with data.
+// MetricsReceiver feeds a processor.MetricsDataProcessor with data.
 //
 // For example it could be Prometheus data source which translates
 // Prometheus metrics into *metricpb.Metric-s.
 type MetricsReceiver interface {
+	// MetricsSource returns the name of the metrics data source.
+	MetricsSource() string
+
+	// StartMetricsReception tells the receiver to start its processing.
 	StartMetricsReception(ctx context.Context, nextProcessor processor.MetricsDataProcessor) error
+
+	// StopMetricsReception tells the receiver that should stop reception,
+	// giving it a chance to perform any necessary clean-up.
 	StopMetricsReception(ctx context.Context) error
 }

receiver/zipkinreceiver/scribe/scribe_receiver.go

@@ -65,6 +65,13 @@ func NewReceiver(addr string, port uint16, category string) (receiver.TraceRecei
 	return r, nil
 }
 
+const traceSource string = "Zipkin-Scribe"
+
+// TraceSource returns the name of the trace data source.
+func (r *scribeReceiver) TraceSource() string {
+	return traceSource
+}
+
 func (r *scribeReceiver) StartTraceReception(ctx context.Context, nextProcessor processor.TraceDataProcessor) error {
 	r.Lock()
 	defer r.Unlock()

receiver/zipkinreceiver/trace_receiver.go

@@ -84,6 +84,13 @@ var (
 	errAlreadyStopped = errors.New("already stopped")
 )
 
+const traceSource string = "Zipkin"
+
+// TraceSource returns the name of the trace data source.
+func (zr *ZipkinReceiver) TraceSource() string {
+	return traceSource
+}
+
 // StartTraceReception spins up the receiver's HTTP server and makes the receiver start its processing.
 func (zr *ZipkinReceiver) StartTraceReception(ctx context.Context, nextProcessor processor.TraceDataProcessor) error {
 	zr.mu.Lock()