Remove Receiver Trace and Metrics sinks. (#424)

Author: Bogdan Drutu

cmd/ocagent/main.go

@@ -31,12 +31,11 @@ import (
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 
-	"github.com/census-instrumentation/opencensus-service/data"
 	"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"
-	"github.com/census-instrumentation/opencensus-service/receiver"
+	"github.com/census-instrumentation/opencensus-service/processor"
 	"github.com/census-instrumentation/opencensus-service/receiver/jaegerreceiver"
 	"github.com/census-instrumentation/opencensus-service/receiver/opencensusreceiver"
 	"github.com/census-instrumentation/opencensus-service/receiver/prometheusreceiver"
@@ -55,14 +54,6 @@ func main() {
 	}
 }
 
-type noopMetricsSink int
-
-var _ receiver.MetricsReceiverSink = (*noopMetricsSink)(nil)
-
-func (nms *noopMetricsSink) ReceiveMetricsData(ctx context.Context, metricsdata data.MetricsData) (*receiver.MetricsReceiverAcknowledgement, error) {
-	return &receiver.MetricsReceiverAcknowledgement{}, nil
-}
-
 func runOCAgent() {
 	yamlBlob, err := ioutil.ReadFile(configYAMLFile)
 	if err != nil {
@@ -189,7 +180,7 @@ func runZPages(port int) func() error {
 	return srv.Close
 }
 
-func runOCReceiver(logger *zap.Logger, acfg *config.Config, sr receiver.TraceReceiverSink, mr receiver.MetricsReceiverSink) (doneFn func() error, err error) {
+func runOCReceiver(logger *zap.Logger, acfg *config.Config, tdp processor.TraceDataProcessor, mdp processor.MetricsDataProcessor) (doneFn func() error, err error) {
 	tlsCredsOption, hasTLSCreds, err := acfg.OpenCensusReceiverTLSCredentialsServerOption()
 	if err != nil {
 		return nil, fmt.Errorf("OpenCensus receiver TLS Credentials: %v", err)
@@ -217,19 +208,19 @@ func runOCReceiver(logger *zap.Logger, acfg *config.Config, sr receiver.TraceRec
 
 	switch {
 	case acfg.CanRunOpenCensusTraceReceiver() && acfg.CanRunOpenCensusMetricsReceiver():
-		if err := ocr.Start(ctx, sr, mr); err != nil {
+		if err := ocr.Start(ctx, tdp, mdp); 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, sr); err != nil {
+		if err := ocr.StartTraceReception(ctx, tdp); 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, mr); err != nil {
+		if err := ocr.StartMetricsReception(ctx, mdp); 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)
@@ -246,7 +237,7 @@ func runOCReceiver(logger *zap.Logger, acfg *config.Config, sr receiver.TraceRec
 	return doneFn, nil
 }
 
-func runJaegerReceiver(collectorThriftPort, collectorHTTPPort int, sr receiver.TraceReceiverSink) (doneFn func() error, err error) {
+func runJaegerReceiver(collectorThriftPort, collectorHTTPPort int, next processor.TraceDataProcessor) (doneFn func() error, err error) {
 	jtr, err := jaegerreceiver.New(context.Background(), &jaegerreceiver.Configuration{
 		CollectorThriftPort: collectorThriftPort,
 		CollectorHTTPPort:   collectorHTTPPort,
@@ -258,7 +249,7 @@ func runJaegerReceiver(collectorThriftPort, collectorHTTPPort int, sr receiver.T
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create new Jaeger receiver: %v", err)
 	}
-	if err := jtr.StartTraceReception(context.Background(), sr); err != nil {
+	if err := jtr.StartTraceReception(context.Background(), next); err != nil {
 		return nil, fmt.Errorf("Failed to start Jaeger receiver: %v", err)
 	}
 	doneFn = func() error {
@@ -268,13 +259,13 @@ func runJaegerReceiver(collectorThriftPort, collectorHTTPPort int, sr receiver.T
 	return doneFn, nil
 }
 
-func runZipkinReceiver(addr string, sr receiver.TraceReceiverSink) (doneFn func() error, err error) {
+func runZipkinReceiver(addr string, next processor.TraceDataProcessor) (doneFn func() error, err error) {
 	zi, err := zipkinreceiver.New(addr)
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create the Zipkin receiver: %v", err)
 	}
 
-	if err := zi.StartTraceReception(context.Background(), sr); err != nil {
+	if err := zi.StartTraceReception(context.Background(), next); err != nil {
 		return nil, fmt.Errorf("Cannot start Zipkin receiver with address %q: %v", addr, err)
 	}
 	doneFn = func() error {
@@ -284,13 +275,13 @@ func runZipkinReceiver(addr string, sr receiver.TraceReceiverSink) (doneFn func(
 	return doneFn, nil
 }
 
-func runZipkinScribeReceiver(config *config.ScribeReceiverConfig, sr receiver.TraceReceiverSink) (doneFn func() error, err error) {
+func runZipkinScribeReceiver(config *config.ScribeReceiverConfig, next processor.TraceDataProcessor) (doneFn func() error, err error) {
 	zs, err := scribe.NewReceiver(config.Address, config.Port, config.Category)
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create the Zipkin Scribe receiver: %v", err)
 	}
 
-	if err := zs.StartTraceReception(context.Background(), sr); err != nil {
+	if err := zs.StartTraceReception(context.Background(), next); err != nil {
 		return nil, fmt.Errorf("Cannot start Zipkin Scribe receiver with %v: %v", config, err)
 	}
 	doneFn = func() error {
@@ -300,12 +291,12 @@ func runZipkinScribeReceiver(config *config.ScribeReceiverConfig, sr receiver.Tr
 	return doneFn, nil
 }
 
-func runPrometheusReceiver(promConfig *prometheusreceiver.Configuration, mr receiver.MetricsReceiverSink) (doneFn func() error, err error) {
+func runPrometheusReceiver(promConfig *prometheusreceiver.Configuration, next processor.MetricsDataProcessor) (doneFn func() error, err error) {
 	pmr, err := prometheusreceiver.New(promConfig)
 	if err != nil {
 		return nil, err
 	}
-	if err := pmr.StartMetricsReception(context.Background(), mr); err != nil {
+	if err := pmr.StartMetricsReception(context.Background(), next); err != nil {
 		return nil, err
 	}
 	doneFn = func() error {

exporter/metrics_exporter.go

@@ -18,7 +18,7 @@ import (
 	"context"
 
 	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/receiver"
+	"github.com/census-instrumentation/opencensus-service/processor"
 )
 
 // MetricsExporter is an interface that receives data.MetricsData, converts it as needed, and
@@ -30,13 +30,13 @@ type MetricsExporter interface {
 }
 
 // MultiMetricsExporters wraps multiple metrics exporters in a single one.
-func MultiMetricsExporters(mes ...MetricsExporter) receiver.MetricsReceiverSink {
+func MultiMetricsExporters(mes ...MetricsExporter) processor.MetricsDataProcessor {
 	return metricsExporters(mes)
 }
 
 type metricsExporters []MetricsExporter
 
-var _ receiver.MetricsReceiverSink = (*metricsExporters)(nil)
+var _ processor.MetricsDataProcessor = (*metricsExporters)(nil)
 
 // ExportMetricsData exports the MetricsData to all exporters wrapped by the current one.
 func (mes metricsExporters) ExportMetricsData(ctx context.Context, md data.MetricsData) error {
@@ -46,15 +46,12 @@ func (mes metricsExporters) ExportMetricsData(ctx context.Context, md data.Metri
 	return nil
 }
 
-// ReceiveTraceData receives the span data in the protobuf format, translates it, and forwards the transformed
+// ProcessMetricsData receives the span data in the protobuf format, translates it, and forwards the transformed
 // span data to all trace exporters wrapped by the current one.
-func (mes metricsExporters) ReceiveMetricsData(ctx context.Context, md data.MetricsData) (*receiver.MetricsReceiverAcknowledgement, error) {
+func (mes metricsExporters) ProcessMetricsData(ctx context.Context, md data.MetricsData) error {
 	for _, me := range mes {
 		_ = me.ExportMetricsData(ctx, md)
 	}
 
-	ack := &receiver.MetricsReceiverAcknowledgement{
-		SavedMetrics: uint64(len(md.Metrics)),
-	}
-	return ack, nil
+	return nil
 }

exporter/trace_exporter.go

@@ -18,7 +18,7 @@ import (
 	"context"
 
 	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/receiver"
+	"github.com/census-instrumentation/opencensus-service/processor"
 )
 
 // TraceExporter is a interface that receives OpenCensus data, converts it as needed, and
@@ -34,7 +34,7 @@ type TraceExporter interface {
 // transforms it to OpenCensus in memory data and sends it to the exporter.
 type TraceExporterSink interface {
 	TraceExporter
-	receiver.TraceReceiverSink
+	processor.TraceDataProcessor
 }
 
 // MultiTraceExporters wraps multiple trace exporters in a single one.
@@ -44,6 +44,9 @@ func MultiTraceExporters(tes ...TraceExporter) TraceExporterSink {
 
 type traceExporters []TraceExporter
 
+var _ TraceExporter = (*traceExporters)(nil)
+var _ processor.TraceDataProcessor = (*traceExporters)(nil)
+
 // ExportSpans exports the span data to all trace exporters wrapped by the current one.
 func (tes traceExporters) ExportSpans(ctx context.Context, td data.TraceData) error {
 	for _, te := range tes {
@@ -52,15 +55,12 @@ func (tes traceExporters) ExportSpans(ctx context.Context, td data.TraceData) er
 	return nil
 }
 
-// ReceiveTraceData receives the span data in the protobuf format, translates it, and forwards the transformed
+// ProcessTraceData receives the span data in the protobuf format, translates it, and forwards the transformed
 // span data to all trace exporters wrapped by the current one.
-func (tes traceExporters) ReceiveTraceData(ctx context.Context, td data.TraceData) (*receiver.TraceReceiverAcknowledgement, error) {
+func (tes traceExporters) ProcessTraceData(ctx context.Context, td data.TraceData) error {
 	for _, te := range tes {
 		_ = te.ExportSpans(ctx, td)
 	}
 
-	ack := &receiver.TraceReceiverAcknowledgement{
-		SavedSpans: uint64(len(td.Spans)),
-	}
-	return ack, nil
+	return nil
 }

internal/collector/processor/exporter_processor.go

@@ -34,9 +34,10 @@ func NewTraceExporterProcessor(traceExporters ...exporter.TraceExporter) SpanPro
 }
 
 func (sp *exporterSpanProcessor) ProcessSpans(batch *agenttracepb.ExportTraceServiceRequest, spanFormat string) (uint64, error) {
-	ack, err := sp.tes.ReceiveTraceData(context.Background(), data.TraceData{Node: batch.Node, Resource: batch.Resource, Spans: batch.Spans})
+	err := sp.tes.ProcessTraceData(context.Background(), data.TraceData{Node: batch.Node, Resource: batch.Resource, Spans: batch.Spans})
 	if err != nil {
-		return ack.DroppedSpans, err
+		// TODO: determine if the number of dropped spans is needed because it was wrong anyway.
+		return 0, err
 	}
 	return 0, nil
 }

internal/collector/processor/processor_to_sink.go

@@ -19,36 +19,31 @@ import (
 
 	agenttracepb "github.com/census-instrumentation/opencensus-proto/gen-go/agent/trace/v1"
 	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/receiver"
+	"github.com/census-instrumentation/opencensus-service/processor"
 )
 
 type protoProcessorSink struct {
 	sourceFormat   string
 	protoProcessor SpanProcessor
 }
 
-var _ (receiver.TraceReceiverSink) = (*protoProcessorSink)(nil)
+var _ (processor.TraceDataProcessor) = (*protoProcessorSink)(nil)
 
 // WrapWithSpanSink wraps a processor to be used as a span sink by receivers.
-func WrapWithSpanSink(format string, p SpanProcessor) receiver.TraceReceiverSink {
+func WrapWithSpanSink(format string, p SpanProcessor) processor.TraceDataProcessor {
 	return &protoProcessorSink{
 		sourceFormat:   format,
 		protoProcessor: p,
 	}
 }
 
-func (ps *protoProcessorSink) ReceiveTraceData(ctx context.Context, td data.TraceData) (*receiver.TraceReceiverAcknowledgement, error) {
+func (ps *protoProcessorSink) ProcessTraceData(ctx context.Context, td data.TraceData) error {
 	batch := &agenttracepb.ExportTraceServiceRequest{
 		Node:  td.Node,
 		Spans: td.Spans,
 	}
 
-	failures, err := ps.protoProcessor.ProcessSpans(batch, ps.sourceFormat)
+	_, err := ps.protoProcessor.ProcessSpans(batch, ps.sourceFormat)
 
-	ack := &receiver.TraceReceiverAcknowledgement{
-		SavedSpans:   uint64(len(batch.Spans)) - failures,
-		DroppedSpans: failures,
-	}
-
-	return ack, err
+	return err
 }

receiver/end_to_end_test.go

@@ -24,6 +24,7 @@ import (
 	"go.opencensus.io/trace"
 
 	"github.com/census-instrumentation/opencensus-service/data"
+	"github.com/census-instrumentation/opencensus-service/processor"
 	"github.com/census-instrumentation/opencensus-service/receiver"
 	"github.com/census-instrumentation/opencensus-service/receiver/opencensusreceiver"
 )
@@ -92,11 +93,11 @@ func Example_endToEnd() {
 
 type logSpanSink int
 
-var _ receiver.TraceReceiverSink = (*logSpanSink)(nil)
+var _ processor.TraceDataProcessor = (*logSpanSink)(nil)
 
-func (lsr *logSpanSink) ReceiveTraceData(ctx context.Context, td data.TraceData) (*receiver.TraceReceiverAcknowledgement, error) {
+func (lsr *logSpanSink) ProcessTraceData(ctx context.Context, td data.TraceData) error {
 	spansBlob, _ := json.MarshalIndent(td.Spans, " ", "  ")
 	log.Printf("\n****\nNode: %#v\nSpans: %s\n****\n", td.Node, spansBlob)
 
-	return &receiver.TraceReceiverAcknowledgement{SavedSpans: uint64(len(td.Spans))}, nil
+	return nil
 }

receiver/jaegerreceiver/trace_receiver.go

@@ -39,6 +39,7 @@ import (
 
 	"github.com/census-instrumentation/opencensus-service/data"
 	"github.com/census-instrumentation/opencensus-service/internal"
+	"github.com/census-instrumentation/opencensus-service/processor"
 	"github.com/census-instrumentation/opencensus-service/receiver"
 	jaegertranslator "github.com/census-instrumentation/opencensus-service/translator/trace/jaeger"
 )
@@ -60,7 +61,7 @@ type jReceiver struct {
 	// mu protects the fields of this type
 	mu sync.Mutex
 
-	spanSink receiver.TraceReceiverSink
+	nextProcessor processor.TraceDataProcessor
 
 	startOnce sync.Once
 	stopOnce  sync.Once
@@ -159,7 +160,7 @@ func (jr *jReceiver) agentBinaryThriftAddr() string {
 	return fmt.Sprintf(":%d", port)
 }
 
-func (jr *jReceiver) StartTraceReception(ctx context.Context, spanSink receiver.TraceReceiverSink) error {
+func (jr *jReceiver) StartTraceReception(ctx context.Context, nextProcessor processor.TraceDataProcessor) error {
 	jr.mu.Lock()
 	defer jr.mu.Unlock()
 
@@ -175,8 +176,8 @@ func (jr *jReceiver) StartTraceReception(ctx context.Context, spanSink receiver.
 			return
 		}
 
-		// Finally set the spanSink, since we never encountered an error.
-		jr.spanSink = spanSink
+		// Finally set the nextProcessor, since we never encountered an error.
+		jr.nextProcessor = nextProcessor
 
 		err = nil
 	})
@@ -236,7 +237,7 @@ func (jr *jReceiver) SubmitBatches(ctx thrift.Context, batches []*jaeger.Batch)
 
 		if err == nil && octrace != nil {
 			ok = true
-			jr.spanSink.ReceiveTraceData(ctx, data.TraceData{Node: octrace.Node, Spans: octrace.Spans})
+			jr.nextProcessor.ProcessTraceData(ctx, data.TraceData{Node: octrace.Node, Spans: octrace.Spans})
 			// We MUST unconditionally record metrics from this reception.
 			spansMetricsFn(octrace.Node, octrace.Spans)
 		}
@@ -268,7 +269,7 @@ func (jr *jReceiver) EmitBatch(batch *jaeger.Batch) error {
 
 	ctx := context.Background()
 	spansMetricsFn := internal.NewReceivedSpansRecorderStreaming(ctx, "jaeger-agent")
-	_, err = jr.spanSink.ReceiveTraceData(ctx, data.TraceData{Node: octrace.Node, Spans: octrace.Spans})
+	err = jr.nextProcessor.ProcessTraceData(ctx, data.TraceData{Node: octrace.Node, Spans: octrace.Spans})
 	// We MUST unconditionally record metrics from this reception.
 	spansMetricsFn(octrace.Node, octrace.Spans)
 

receiver/opencensusreceiver/ocmetrics/opencensus.go

@@ -29,22 +29,22 @@ import (
 	resourcepb "github.com/census-instrumentation/opencensus-proto/gen-go/resource/v1"
 	"github.com/census-instrumentation/opencensus-service/data"
 	"github.com/census-instrumentation/opencensus-service/internal"
-	"github.com/census-instrumentation/opencensus-service/receiver"
+	"github.com/census-instrumentation/opencensus-service/processor"
 )
 
 // Receiver is the type used to handle metrics from OpenCensus exporters.
 type Receiver struct {
-	metricSink         receiver.MetricsReceiverSink
+	nextProcessor      processor.MetricsDataProcessor
 	metricBufferPeriod time.Duration
 	metricBufferCount  int
 }
 
 // New creates a new ocmetrics.Receiver reference.
-func New(sr receiver.MetricsReceiverSink, opts ...Option) (*Receiver, error) {
-	if sr == nil {
-		return nil, errors.New("needs a non-nil receiver.MetricsReceiverSink")
+func New(nextProcessor processor.MetricsDataProcessor, opts ...Option) (*Receiver, error) {
+	if nextProcessor == nil {
+		return nil, errors.New("needs a non-nil processor.MetricsDataProcessor")
 	}
-	ocr := &Receiver{metricSink: sr}
+	ocr := &Receiver{nextProcessor: nextProcessor}
 	for _, opt := range opts {
 		opt.WithReceiver(ocr)
 	}
@@ -142,7 +142,7 @@ func (ocr *Receiver) batchMetricExporting(longLivedRPCCtx context.Context, paylo
 
 	nMetrics := int64(0)
 	for _, md := range mds {
-		ocr.metricSink.ReceiveMetricsData(ctx, *md)
+		ocr.nextProcessor.ProcessMetricsData(ctx, *md)
 		nMetrics += int64(len(md.Metrics))
 	}