Remove processor to sink. (#497)

Bogdan Drutu · web-flow · commit f4c55919d5d9 · 2019-03-14T13:33:20.000-07:00
diff --git a/internal/collector/jaeger/receiver.go b/internal/collector/jaeger/receiver.go
@@ -30,7 +30,6 @@ import (
 
 	"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"
 )
@@ -51,8 +50,7 @@ func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsu
 		return nil, err
 	}
 
-	ss := processor.WithSourceName("jaeger", traceConsumer)
-	if err := jtr.StartTraceReception(ctx, ss); err != nil {
+	if err := jtr.StartTraceReception(ctx, traceConsumer); err != nil {
 		return nil, err
 	}
 
diff --git a/internal/collector/opencensus/receiver.go b/internal/collector/opencensus/receiver.go
@@ -26,7 +26,6 @@ import (
 
 	"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"
 )
@@ -48,8 +47,8 @@ func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsu
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create the OpenCensus trace receiver: %v", err)
 	}
-	ss := processor.WithSourceName("oc_trace", traceConsumer)
-	if err := ocr.StartTraceReception(context.Background(), ss); err != nil {
+
+	if err := ocr.StartTraceReception(context.Background(), traceConsumer); err != nil {
 		return nil, fmt.Errorf("Cannot bind Opencensus receiver to address %q: %v", addr, err)
 	}
 
diff --git a/internal/collector/processor/processor_to_sink.go b/internal/collector/processor/processor_to_sink.go
diff --git a/internal/collector/zipkin/receiver.go b/internal/collector/zipkin/receiver.go
@@ -26,7 +26,6 @@ import (
 
 	"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/zipkinreceiver"
 )
@@ -43,9 +42,8 @@ func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsu
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create the Zipkin receiver: %v", err)
 	}
-	ss := processor.WithSourceName("zipkin", traceConsumer)
 
-	if err := zi.StartTraceReception(context.Background(), ss); err != nil {
+	if err := zi.StartTraceReception(context.Background(), traceConsumer); err != nil {
 		return nil, fmt.Errorf("Cannot start Zipkin receiver to address %q: %v", addr, err)
 	}
 
diff --git a/internal/collector/zipkin/scribe/receiver.go b/internal/collector/zipkin/scribe/receiver.go
@@ -25,7 +25,6 @@ import (
 
 	"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/zipkinreceiver/scribe"
 )
@@ -41,9 +40,8 @@ func Start(logger *zap.Logger, v *viper.Viper, traceConsumer consumer.TraceConsu
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create the Zipkin Scribe receiver: %v", err)
 	}
-	ss := processor.WithSourceName("zipkin-scribe", traceConsumer)
 
-	if err := sr.StartTraceReception(context.Background(), ss); err != nil {
+	if err := sr.StartTraceReception(context.Background(), traceConsumer); err != nil {
 		return nil, fmt.Errorf("Cannot start Zipkin Scribe receiver %+v: %v", rOpts, err)
 	}
 
diff --git a/receiver/jaegerreceiver/trace_receiver.go b/receiver/jaegerreceiver/trace_receiver.go
@@ -249,6 +249,7 @@ func (jr *jReceiver) SubmitBatches(ctx thrift.Context, batches []*jaeger.Batch)
 
 		if err == nil {
 			ok = true
+			td.SourceFormat = "jaeger"
 			jr.nextConsumer.ConsumeTraceData(ctx, td)
 			// We MUST unconditionally record metrics from this reception.
 			observability.RecordTraceReceiverMetrics(ctxWithReceiverName, len(batch.Spans), len(batch.Spans)-len(td.Spans))
diff --git a/receiver/jaegerreceiver/trace_receiver_test.go b/receiver/jaegerreceiver/trace_receiver_test.go
@@ -140,7 +140,6 @@ func TestReception(t *testing.T) {
 					"int64":  "10000000",
 				},
 			},
-
 			Spans: []*tracepb.Span{
 				{
 					TraceId:      []byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F, 0x80},
@@ -209,6 +208,7 @@ func TestReception(t *testing.T) {
 					},
 				},
 			},
+			SourceFormat: "jaeger",
 		},
 	}
 
diff --git a/receiver/opencensusreceiver/octrace/opencensus.go b/receiver/opencensusreceiver/octrace/opencensus.go
@@ -123,9 +123,10 @@ func (ocr *Receiver) Export(tes agenttracepb.TraceService_ExportServer) error {
 		}
 
 		td := &data.TraceData{
-			Node:     lastNonNilNode,
-			Resource: resource,
-			Spans:    recv.Spans,
+			Node:         lastNonNilNode,
+			Resource:     resource,
+			Spans:        recv.Spans,
+			SourceFormat: "oc_trace",
 		}
 
 		ocr.messageChan <- &traceDataWithCtx{data: td, ctx: ctxWithReceiverName}
diff --git a/receiver/zipkinreceiver/scribe/scribe_receiver.go b/receiver/zipkinreceiver/scribe/scribe_receiver.go
@@ -174,6 +174,7 @@ func (sc *scribeCollector) Log(messages []*scribe.LogEntry) (r scribe.ResultCode
 
 	tdsSize := 0
 	for _, td := range tds {
+		td.SourceFormat = "zipkin-scribe"
 		sc.nextConsumer.ConsumeTraceData(sc.defaultCtx, td)
 		tdsSize += len(td.Spans)
 	}
diff --git a/receiver/zipkinreceiver/scribe/scribe_receiver_test.go b/receiver/zipkinreceiver/scribe/scribe_receiver_test.go
@@ -264,4 +264,5 @@ var wantTraceData = data.TraceData{
 			},
 		},
 	},
+	SourceFormat: "zipkin-scribe",
 }
diff --git a/receiver/zipkinreceiver/trace_receiver.go b/receiver/zipkinreceiver/trace_receiver.go
@@ -322,6 +322,7 @@ func (zr *ZipkinReceiver) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	ctxWithReceiverName := observability.ContextWithReceiverName(ctx, receiverTagValue)
 	tdsSize := 0
 	for _, td := range tds {
+		td.SourceFormat = "zipkin"
 		zr.nextConsumer.ConsumeTraceData(ctxWithReceiverName, td)
 		tdsSize += len(td.Spans)
 	}

Original file line number	Diff line number	Diff line change
`@@ -30,7 +30,6 @@ import (`
`30`	`30`
`31`	`31`	`"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/builder"`
`32`	`32`	`"github.com/census-instrumentation/opencensus-service/consumer"`
`33`		`- "github.com/census-instrumentation/opencensus-service/internal/collector/processor"`
`34`	`33`	`"github.com/census-instrumentation/opencensus-service/receiver"`
`35`	`34`	`"github.com/census-instrumentation/opencensus-service/receiver/jaegerreceiver"`
`36`	`35`	`)`
`@@ -51,8 +50,7 @@ func Start(logger zap.Logger, v viper.Viper, traceConsumer consumer.TraceConsu`
`51`	`50`	`return nil, err`
`52`	`51`	`}`
`53`	`52`
`54`		`- ss := processor.WithSourceName("jaeger", traceConsumer)`
`55`		`- if err := jtr.StartTraceReception(ctx, ss); err != nil {`
	`53`	`+ if err := jtr.StartTraceReception(ctx, traceConsumer); err != nil {`
`56`	`54`	`return nil, err`
`57`	`55`	`}`
`58`	`56`
Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,6 @@ import (`
`26`	`26`
`27`	`27`	`"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/builder"`
`28`	`28`	`"github.com/census-instrumentation/opencensus-service/consumer"`
`29`		`- "github.com/census-instrumentation/opencensus-service/internal/collector/processor"`
`30`	`29`	`"github.com/census-instrumentation/opencensus-service/receiver"`
`31`	`30`	`"github.com/census-instrumentation/opencensus-service/receiver/opencensusreceiver"`
`32`	`31`	`)`
`@@ -48,8 +47,8 @@ func Start(logger zap.Logger, v viper.Viper, traceConsumer consumer.TraceConsu`
`48`	`47`	`if err != nil {`
`49`	`48`	`return nil, fmt.Errorf("Failed to create the OpenCensus trace receiver: %v", err)`
`50`	`49`	`}`
`51`		`- ss := processor.WithSourceName("oc_trace", traceConsumer)`
`52`		`- if err := ocr.StartTraceReception(context.Background(), ss); err != nil {`
	`50`	`+`
	`51`	`+ if err := ocr.StartTraceReception(context.Background(), traceConsumer); err != nil {`
`53`	`52`	`return nil, fmt.Errorf("Cannot bind Opencensus receiver to address %q: %v", addr, err)`
`54`	`53`	`}`
`55`	`54`
Original file line number	Diff line number	Diff line change
`@@ -25,7 +25,6 @@ import (`
`25`	`25`
`26`	`26`	`"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/builder"`
`27`	`27`	`"github.com/census-instrumentation/opencensus-service/consumer"`
`28`		`- "github.com/census-instrumentation/opencensus-service/internal/collector/processor"`
`29`	`28`	`"github.com/census-instrumentation/opencensus-service/receiver"`
`30`	`29`	`"github.com/census-instrumentation/opencensus-service/receiver/zipkinreceiver/scribe"`
`31`	`30`	`)`
`@@ -41,9 +40,8 @@ func Start(logger zap.Logger, v viper.Viper, traceConsumer consumer.TraceConsu`
`41`	`40`	`if err != nil {`
`42`	`41`	`return nil, fmt.Errorf("Failed to create the Zipkin Scribe receiver: %v", err)`
`43`	`42`	`}`
`44`		`- ss := processor.WithSourceName("zipkin-scribe", traceConsumer)`
`45`	`43`
`46`		`- if err := sr.StartTraceReception(context.Background(), ss); err != nil {`
	`44`	`+ if err := sr.StartTraceReception(context.Background(), traceConsumer); err != nil {`
`47`	`45`	`return nil, fmt.Errorf("Cannot start Zipkin Scribe receiver %+v: %v", rOpts, err)`
`48`	`46`	`}`
`49`	`47`
Original file line number	Diff line number	Diff line change
`@@ -140,7 +140,6 @@ func TestReception(t *testing.T) {`
`140`	`140`	`"int64": "10000000",`
`141`	`141`	`},`
`142`	`142`	`},`
`143`		`-`
`144`	`143`	`Spans: []*tracepb.Span{`
`145`	`144`	`{`
`146`	`145`	`TraceId: []byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F, 0x80},`
`@@ -209,6 +208,7 @@ func TestReception(t *testing.T) {`
`209`	`208`	`},`
`210`	`209`	`},`
`211`	`210`	`},`
	`211`	`+ SourceFormat: "jaeger",`
`212`	`212`	`},`
`213`	`213`	`}`
`214`	`214`
Original file line number	Diff line number	Diff line change
`@@ -123,9 +123,10 @@ func (ocr *Receiver) Export(tes agenttracepb.TraceService_ExportServer) error {`
`123`	`123`	`}`
`124`	`124`
`125`	`125`	`td := &data.TraceData{`
`126`		`- Node: lastNonNilNode,`
`127`		`- Resource: resource,`
`128`		`- Spans: recv.Spans,`
	`126`	`+ Node: lastNonNilNode,`
	`127`	`+ Resource: resource,`
	`128`	`+ Spans: recv.Spans,`
	`129`	`+ SourceFormat: "oc_trace",`
`129`	`130`	`}`
`130`	`131`
`131`	`132`	`ocr.messageChan <- &traceDataWithCtx{data: td, ctx: ctxWithReceiverName}`
Original file line number	Diff line number	Diff line change
`@@ -174,6 +174,7 @@ func (sc scribeCollector) Log(messages []scribe.LogEntry) (r scribe.ResultCode`
`174`	`174`
`175`	`175`	`tdsSize := 0`
`176`	`176`	`for _, td := range tds {`
	`177`	`+ td.SourceFormat = "zipkin-scribe"`
`177`	`178`	`sc.nextConsumer.ConsumeTraceData(sc.defaultCtx, td)`
`178`	`179`	`tdsSize += len(td.Spans)`
`179`	`180`	`}`
Original file line number	Diff line number	Diff line change
`@@ -264,4 +264,5 @@ var wantTraceData = data.TraceData{`
`264`	`264`	`},`
`265`	`265`	`},`
`266`	`266`	`},`
	`267`	`+ SourceFormat: "zipkin-scribe",`
`267`	`268`	`}`