Fix batchProcessor to set correct span format flags

cmd/collector/app/handler/grpc_handler.go

@@ -86,8 +86,8 @@ func (c *batchConsumer) consume(ctx context.Context, batch *model.Batch) error {
 		}
 	}
 	_, err = c.spanProcessor.ProcessSpans(batch.Spans, processor.SpansOptions{
-		InboundTransport: processor.GRPCTransport,
-		SpanFormat:       processor.ProtoSpanFormat,
+		InboundTransport: c.spanOptions.InboundTransport,
+		SpanFormat:       c.spanOptions.SpanFormat,
 		Tenant:           tenant,
 	})
 	if err != nil {

cmd/collector/app/handler/grpc_handler_test.go

@@ -40,6 +40,8 @@
 	mux           sync.Mutex
 	spans         []*model.Span
 	tenants       map[string]bool
+	transport     processor.InboundTransport
+	spanFormat    processor.SpanFormat
 }
 
 func (p *mockSpanProcessor) ProcessSpans(spans []*model.Span, opts processor.SpansOptions) ([]bool, error) {
@@ -51,6 +53,8 @@
 		p.tenants = make(map[string]bool)
 	}
 	p.tenants[opts.Tenant] = true
+	p.transport = opts.InboundTransport
+	p.spanFormat = opts.SpanFormat
 	return oks, p.expectedError
 }
 
@@ -66,17 +70,35 @@
 	return p.tenants
 }
 
+func (p *mockSpanProcessor) getTransport() processor.InboundTransport {
+	p.mux.Lock()
+	defer p.mux.Unlock()
+	return p.transport
+}
+
+func (p *mockSpanProcessor) getSpanFormat() processor.SpanFormat {
+	p.mux.Lock()
+	defer p.mux.Unlock()
+	return p.spanFormat
+}
+
 func (p *mockSpanProcessor) reset() {
 	p.mux.Lock()
 	defer p.mux.Unlock()
 	p.spans = nil
 	p.tenants = nil
+	p.transport = ""
+	p.spanFormat = ""
 }
 
 func (p *mockSpanProcessor) Close() error {
 	return nil
 }
 
+type mockSpanWriter struct{}
+
+func (w *mockSpanWriter) WriteSpan(ctx context.Context, span *model.Span) error { return nil }
+
 func initializeGRPCTestServer(t *testing.T, beforeServe func(s *grpc.Server)) (*grpc.Server, net.Addr) {
 	server := grpc.NewServer()
 	beforeServe(server)
@@ -363,3 +385,47 @@
 		})
 	}
 }
+
+func TestBatchConsumer(t *testing.T) {
+	tests := []struct {
+		name               string
+		batch              model.Batch
+		transport          processor.InboundTransport
+		spanFormat         processor.SpanFormat
+		expectedTransport  processor.InboundTransport
+		expectedSpanFormat processor.SpanFormat
+	}{
+		{
+			name: "batchconsumer passes provided span options to processor",
+			batch: model.Batch{
+				Process: &model.Process{ServiceName: "testservice"},
+				Spans: []*model.Span{
+					{OperationName: "test-op", Process: &model.Process{ServiceName: "foo"}},
+				},
+			},
+			transport:          processor.GRPCTransport,
+			spanFormat:         processor.OTLPSpanFormat,
+			expectedTransport:  processor.GRPCTransport,
+			expectedSpanFormat: processor.OTLPSpanFormat,
+		},
+	}
+
+	logger, _ := testutils.NewLogger()
+	for _, tc := range tests {
+		tc := tc
+		t.Parallel()
+		t.Run(tc.name, func(t *testing.T) {
+			processor := mockSpanProcessor{}
+			batchConsumer := newBatchConsumer(logger, &processor, tc.transport, tc.spanFormat, tenancy.NewManager(&tenancy.Options{}))
+			err := batchConsumer.consume(context.Background(), &model.Batch{
+				Process: &model.Process{ServiceName: "testservice"},
+				Spans: []*model.Span{
+					{OperationName: "test-op", Process: &model.Process{ServiceName: "foo"}},
+				},
+			})
+			assert.NoError(t, err)
+			assert.Equal(t, tc.transport, processor.getTransport())
+			assert.Equal(t, tc.expectedSpanFormat, processor.getSpanFormat())
+		})
+	}
+}