Remove unused failure from the return values. (#434)

Author: Bogdan Drutu

cmd/occollector/app/sender/jaeger_thrift_http_sender.go

@@ -80,36 +80,35 @@ func NewJaegerThriftHTTPSender(
 }
 
 // ProcessSpans sends the received data to the configured Jaeger Thrift end-point.
-func (s *JaegerThriftHTTPSender) ProcessSpans(td data.TraceData, spanFormat string) (uint64, error) {
+func (s *JaegerThriftHTTPSender) ProcessSpans(td data.TraceData, spanFormat string) 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 {
-		return uint64(len(td.Spans)), err
+		return err
 	}
 
-	mSpans := tBatch.Spans
 	body, err := serializeThrift(tBatch)
 	if err != nil {
-		return uint64(len(mSpans)), err
+		return err
 	}
 	req, err := http.NewRequest("POST", s.url, body)
 	if err != nil {
-		return uint64(len(mSpans)), err
+		return err
 	}
 	req.Header.Set("Content-Type", "application/x-thrift")
 	for k, v := range s.headers {
 		req.Header.Set(k, v)
 	}
 	resp, err := s.client.Do(req)
 	if err != nil {
-		return uint64(len(mSpans)), err
+		return err
 	}
 	io.Copy(ioutil.Discard, resp.Body)
 	resp.Body.Close()
 	if resp.StatusCode >= http.StatusBadRequest {
-		return uint64(len(mSpans)), fmt.Errorf("Jaeger Thirft HTTP sender error: %d", resp.StatusCode)
+		return fmt.Errorf("Jaeger Thirft HTTP sender error: %d", resp.StatusCode)
 	}
-	return 0, nil
+	return nil
 }
 
 func serializeThrift(obj thrift.TStruct) (*bytes.Buffer, error) {

cmd/occollector/app/sender/jaeger_thrift_tchannel_sender.go

@@ -45,16 +45,16 @@ func NewJaegerThriftTChannelSender(
 }
 
 // ProcessSpans sends the received data to the configured Jaeger Thrift end-point.
-func (s *JaegerThriftTChannelSender) ProcessSpans(td data.TraceData, spanFormat string) (uint64, error) {
+func (s *JaegerThriftTChannelSender) ProcessSpans(td data.TraceData, spanFormat string) 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 {
-		return uint64(len(tBatch.Spans)), err
+		return err
 	}
 
 	if err := s.reporter.EmitBatch(tBatch); err != nil {
 		s.logger.Error("Reporter failed to report span batch", zap.Error(err))
-		return uint64(len(tBatch.Spans)), err
+		return err
 	}
-	return 0, nil
+	return nil
 }

internal/collector/processor/exporter_processor.go

@@ -32,11 +32,10 @@ func NewTraceExporterProcessor(traceExporters ...processor.TraceDataProcessor) S
 	return &exporterSpanProcessor{tdp: processor.NewMultiTraceDataProcessor(traceExporters)}
 }
 
-func (sp *exporterSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string) (uint64, error) {
+func (sp *exporterSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string) error {
 	err := sp.tdp.ProcessTraceData(context.Background(), td)
 	if err != nil {
-		// TODO: determine if the number of dropped spans is needed because it was wrong anyway.
-		return 0, err
+		return err
 	}
-	return 0, nil
+	return nil
 }

internal/collector/processor/multi_processor.go

@@ -107,21 +107,16 @@ func WithAddAttributes(attributes map[string]interface{}, overwrite bool) MultiP
 }
 
 // ProcessSpans implements the SpanProcessor interface
-func (msp *multiSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string) (uint64, error) {
+func (msp *multiSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string) error {
 	for _, preProcessFn := range msp.preProcessFns {
 		preProcessFn(td, spanFormat)
 	}
-	var maxFailures uint64
 	var errors []error
 	for _, sp := range msp.processors {
-		failures, err := sp.ProcessSpans(td, spanFormat)
+		err := sp.ProcessSpans(td, spanFormat)
 		if err != nil {
 			errors = append(errors, err)
 		}
-
-		if failures > maxFailures {
-			maxFailures = failures
-		}
 	}
-	return maxFailures, internal.CombineErrors(errors)
+	return internal.CombineErrors(errors)
 }

internal/collector/processor/multi_processor_test.go

@@ -49,45 +49,6 @@ func TestMultiSpanProcessorMultiplexing(t *testing.T) {
 	}
 }
 
-func TestMultiSpanProcessorSomeNotOk(t *testing.T) {
-	processors := make([]SpanProcessor, 3)
-	for i := range processors {
-		processors[i] = &mockSpanProcessor{}
-	}
-
-	// Make one processor return false for some spans
-	m := processors[1].(*mockSpanProcessor)
-	wantFailures := uint64(2)
-	m.Failures = wantFailures
-
-	tt := NewMultiSpanProcessor(processors)
-	spans := make([]*tracepb.Span, wantFailures+3)
-	for i := range spans {
-		spans[i] = &tracepb.Span{}
-	}
-	td := data.TraceData{
-		Spans: spans,
-	}
-
-	var wantSpansCount = 0
-	for i := 0; i < 2; i++ {
-		failures, _ := tt.ProcessSpans(td, "test")
-		batchSize := len(td.Spans)
-		wantSpansCount += batchSize
-		if wantFailures != failures {
-			t.Errorf("Wanted %d failures but got %d", wantFailures, failures)
-		}
-	}
-
-	for _, p := range processors {
-		m := p.(*mockSpanProcessor)
-		if m.TotalSpans != wantSpansCount {
-			t.Errorf("Wanted %d for every processor but got %d", wantSpansCount, m.TotalSpans)
-			return
-		}
-	}
-}
-
 func TestMultiSpanProcessorWhenOneErrors(t *testing.T) {
 	processors := make([]SpanProcessor, 3)
 	for i := range processors {
@@ -105,16 +66,13 @@ func TestMultiSpanProcessorWhenOneErrors(t *testing.T) {
 
 	var wantSpansCount = 0
 	for i := 0; i < 2; i++ {
-		failures, err := tt.ProcessSpans(td, "test")
+		err := tt.ProcessSpans(td, "test")
 		if err == nil {
 			t.Errorf("Wanted error got nil")
 			return
 		}
 		batchSize := len(td.Spans)
 		wantSpansCount += batchSize
-		if failures != uint64(batchSize) {
-			t.Errorf("Wanted all spans to fail, got a different value.")
-		}
 	}
 
 	for _, p := range processors {
@@ -233,19 +191,18 @@ func multiSpanProcessorWithAddAttributesTestHelper(t *testing.T, overwrite bool)
 }
 
 type mockSpanProcessor struct {
-	Failures   uint64
 	TotalSpans int
 	MustFail   bool
 }
 
 var _ SpanProcessor = &mockSpanProcessor{}
 
-func (p *mockSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string) (uint64, error) {
+func (p *mockSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string) error {
 	batchSize := len(td.Spans)
 	p.TotalSpans += batchSize
 	if p.MustFail {
-		return uint64(batchSize), fmt.Errorf("this processor must fail")
+		return fmt.Errorf("this processor must fail")
 	}
 
-	return p.Failures, nil
+	return nil
 }

internal/collector/processor/nodebatcher/node_batcher.go

@@ -107,11 +107,11 @@ func NewBatcher(name string, logger *zap.Logger, sender processor.SpanProcessor,
 
 // ProcessSpans implements batcher as a SpanProcessor and takes the provided spans and adds them to
 // batches
-func (b *batcher) ProcessSpans(td data.TraceData, spanFormat string) (uint64, error) {
+func (b *batcher) ProcessSpans(td data.TraceData, spanFormat string) error {
 	bucketID := b.genBucketID(td.Node, td.Resource, spanFormat)
 	bucket := b.getOrAddBucket(bucketID, td.Node, td.Resource, spanFormat)
 	bucket.add(td.Spans)
-	return 0, nil
+	return nil
 }
 
 func (b *batcher) genBucketID(node *commonpb.Node, resource *resourcepb.Resource, spanFormat string) string {
@@ -275,7 +275,7 @@ func (nb *nodeBatcher) sendBatch(batch *batch) {
 		Resource: nb.resource,
 		Spans:    spans,
 	}
-	_, err := nb.parent.sender.ProcessSpans(td, nb.spanFormat)
+	err := nb.parent.sender.ProcessSpans(td, nb.spanFormat)
 	// Assumed that the next processor always handles a batch, and doesn't error
 	if err != nil {
 		nb.logger.Error(

internal/collector/processor/nodebatcher/node_batcher_test.go

@@ -294,9 +294,9 @@ func newTestSender() *testSender {
 	}
 }
 
-func (ts *testSender) ProcessSpans(td data.TraceData, spanFormat string) (uint64, error) {
+func (ts *testSender) ProcessSpans(td data.TraceData, spanFormat string) error {
 	ts.reqChan <- td
-	return 0, nil
+	return nil
 }
 
 func (ts *testSender) waitFor(spans int, timeout time.Duration) error {

internal/collector/processor/processor.go

@@ -30,7 +30,7 @@ import (
 // SpanProcessor handles batches of spans converted to OpenCensus proto format.
 type SpanProcessor interface {
 	// ProcessSpans processes spans and return with the number of spans that failed and an error.
-	ProcessSpans(td data.TraceData, spanFormat string) (uint64, error)
+	ProcessSpans(td data.TraceData, spanFormat string) error
 	// TODO: (@pjanotti) For shutdown improvement, the interface needs a method to attempt that.
 }
 
@@ -39,7 +39,7 @@ type debugSpanProcessor struct{ logger *zap.Logger }
 
 var _ SpanProcessor = (*debugSpanProcessor)(nil)
 
-func (sp *debugSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string) (uint64, error) {
+func (sp *debugSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string) error {
 	if td.Node == nil {
 		sp.logger.Warn("Received batch with nil Node", zap.String("format", spanFormat))
 	}
@@ -49,7 +49,7 @@ func (sp *debugSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string)
 	stats.RecordWithTags(context.Background(), statsTags, StatReceivedSpanCount.M(int64(numSpans)))
 
 	sp.logger.Debug("debugSpanProcessor", zap.String("originalFormat", spanFormat), zap.Int("#spans", numSpans))
-	return 0, nil
+	return nil
 }
 
 // NewNoopSpanProcessor creates an OC SpanProcessor that just drops the received data.

internal/collector/processor/processor_to_sink.go

@@ -37,7 +37,5 @@ func WrapWithSpanSink(format string, p SpanProcessor) processor.TraceDataProcess
 }
 
 func (ps *protoProcessorSink) ProcessTraceData(ctx context.Context, td data.TraceData) error {
-	_, err := ps.protoProcessor.ProcessSpans(td, ps.sourceFormat)
-
-	return err
+	return ps.protoProcessor.ProcessSpans(td, ps.sourceFormat)
 }

internal/collector/processor/queued/queued_processor.go

@@ -111,15 +111,11 @@ func (sp *queuedSpanProcessor) Stop() {
 }
 
 // ProcessSpans implements the SpanProcessor interface
-func (sp *queuedSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string) (failures uint64, err error) {
-	allAdded := sp.enqueueSpanBatch(td, spanFormat)
-	if !allAdded {
-		failures = uint64(len(td.Spans))
-	}
-	return
+func (sp *queuedSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string) error {
+	return sp.enqueueSpanBatch(td, spanFormat)
 }
 
-func (sp *queuedSpanProcessor) enqueueSpanBatch(td data.TraceData, spanFormat string) bool {
+func (sp *queuedSpanProcessor) enqueueSpanBatch(td data.TraceData, spanFormat string) error {
 	item := &queueItem{
 		queuedTime: time.Now(),
 		td:         td,
@@ -134,12 +130,12 @@ func (sp *queuedSpanProcessor) enqueueSpanBatch(td data.TraceData, spanFormat st
 	if !addedToQueue {
 		sp.onItemDropped(item, statsTags)
 	}
-	return addedToQueue
+	return nil
 }
 
 func (sp *queuedSpanProcessor) processItemFromQueue(item *queueItem) {
 	startTime := time.Now()
-	_, err := sp.sender.ProcessSpans(item.td, item.spanFormat)
+	err := sp.sender.ProcessSpans(item.td, item.spanFormat)
 	if err == nil {
 		// Record latency metrics and return
 		sendLatencyMs := int64(time.Since(startTime) / time.Millisecond)