Add initial implementation of pdatagrcp

This is needed because if we split pdata, we will nolonger be able to depend on the InternalRep or generated proto classes (raw, grpc).

The pdatagrcp will eventually be public once we stabilize the API.

The API in the pdatagrcp package is inspired from the grpc generated classes and redirects all calls to the generated classes. Some simple renames:
* TraceService[Client|Server] -> Traces[Client|Server]
* MetricsService[Client|Server] -> Metrics[Client|Server]
* LogsService[Client|Server] -> Logs[Client|Server]

Other changes:
* Replace usages of grpc generated classes in otlpexporter to excercise this new package.
* Left some TODOs for the moment until this package is used more to determine the right API.

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>

Author: bogdandrutu

exporter/otlpexporter/otlp.go

@@ -30,10 +30,7 @@ import (
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/consumer/pdata"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
-	"go.opentelemetry.io/collector/internal"
-	otlplogs "go.opentelemetry.io/collector/internal/data/protogen/collector/logs/v1"
-	otlpmetrics "go.opentelemetry.io/collector/internal/data/protogen/collector/metrics/v1"
-	otlptrace "go.opentelemetry.io/collector/internal/data/protogen/collector/trace/v1"
+	"go.opentelemetry.io/collector/internal/pdatagrpc"
 )
 
 type exporter struct {
@@ -66,33 +63,31 @@ func (e *exporter) shutdown(context.Context) error {
 }
 
 func (e *exporter) pushTraceData(ctx context.Context, td pdata.Traces) error {
-	if err := e.w.exportTrace(ctx, internal.TracesToOtlp(td.InternalRep())); err != nil {
+	if err := e.w.exportTrace(ctx, td); err != nil {
 		return fmt.Errorf("failed to push trace data via OTLP exporter: %w", err)
 	}
 	return nil
 }
 
 func (e *exporter) pushMetricsData(ctx context.Context, md pdata.Metrics) error {
-	req := internal.MetricsToOtlp(md.InternalRep())
-	if err := e.w.exportMetrics(ctx, req); err != nil {
+	if err := e.w.exportMetrics(ctx, md); err != nil {
 		return fmt.Errorf("failed to push metrics data via OTLP exporter: %w", err)
 	}
 	return nil
 }
 
 func (e *exporter) pushLogData(ctx context.Context, ld pdata.Logs) error {
-	request := internal.LogsToOtlp(ld.InternalRep())
-	if err := e.w.exportLogs(ctx, request); err != nil {
+	if err := e.w.exportLogs(ctx, ld); err != nil {
 		return fmt.Errorf("failed to push log data via OTLP exporter: %w", err)
 	}
 	return nil
 }
 
 type grpcSender struct {
 	// gRPC clients and connection.
-	traceExporter  otlptrace.TraceServiceClient
-	metricExporter otlpmetrics.MetricsServiceClient
-	logExporter    otlplogs.LogsServiceClient
+	traceExporter  pdatagrpc.TracesClient
+	metricExporter pdatagrpc.MetricsClient
+	logExporter    pdatagrpc.LogsClient
 	clientConn     *grpc.ClientConn
 	metadata       metadata.MD
 	callOptions    []grpc.CallOption
@@ -110,9 +105,9 @@ func newGrpcSender(config *Config) (*grpcSender, error) {
 	}
 
 	gs := &grpcSender{
-		traceExporter:  otlptrace.NewTraceServiceClient(clientConn),
-		metricExporter: otlpmetrics.NewMetricsServiceClient(clientConn),
-		logExporter:    otlplogs.NewLogsServiceClient(clientConn),
+		traceExporter:  pdatagrpc.NewTracesClient(clientConn),
+		metricExporter: pdatagrpc.NewMetricsClient(clientConn),
+		logExporter:    pdatagrpc.NewLogsClient(clientConn),
 		clientConn:     clientConn,
 		metadata:       metadata.New(config.GRPCClientSettings.Headers),
 		callOptions: []grpc.CallOption{
@@ -126,18 +121,18 @@ func (gs *grpcSender) stop() error {
 	return gs.clientConn.Close()
 }
 
-func (gs *grpcSender) exportTrace(ctx context.Context, request *otlptrace.ExportTraceServiceRequest) error {
-	_, err := gs.traceExporter.Export(gs.enhanceContext(ctx), request, gs.callOptions...)
+func (gs *grpcSender) exportTrace(ctx context.Context, td pdata.Traces) error {
+	_, err := gs.traceExporter.Export(gs.enhanceContext(ctx), td, gs.callOptions...)
 	return processError(err)
 }
 
-func (gs *grpcSender) exportMetrics(ctx context.Context, request *otlpmetrics.ExportMetricsServiceRequest) error {
-	_, err := gs.metricExporter.Export(gs.enhanceContext(ctx), request, gs.callOptions...)
+func (gs *grpcSender) exportMetrics(ctx context.Context, md pdata.Metrics) error {
+	_, err := gs.metricExporter.Export(gs.enhanceContext(ctx), md, gs.callOptions...)
 	return processError(err)
 }
 
-func (gs *grpcSender) exportLogs(ctx context.Context, request *otlplogs.ExportLogsServiceRequest) error {
-	_, err := gs.logExporter.Export(gs.enhanceContext(ctx), request, gs.callOptions...)
+func (gs *grpcSender) exportLogs(ctx context.Context, ld pdata.Logs) error {
+	_, err := gs.logExporter.Export(gs.enhanceContext(ctx), ld, gs.callOptions...)
 	return processError(err)
 }
 

exporter/otlpexporter/otlp_test.go

@@ -33,15 +33,13 @@ import (
 	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/config/configtls"
 	"go.opentelemetry.io/collector/consumer/pdata"
-	"go.opentelemetry.io/collector/internal"
-	otlplogs "go.opentelemetry.io/collector/internal/data/protogen/collector/logs/v1"
-	otlpmetrics "go.opentelemetry.io/collector/internal/data/protogen/collector/metrics/v1"
-	otlptraces "go.opentelemetry.io/collector/internal/data/protogen/collector/trace/v1"
+	"go.opentelemetry.io/collector/internal/pdatagrpc"
 	"go.opentelemetry.io/collector/internal/testdata"
 	"go.opentelemetry.io/collector/obsreport"
 )
 
 type mockReceiver struct {
+	t            *testing.T
 	srv          *grpc.Server
 	requestCount int32
 	totalItems   int32
@@ -57,43 +55,35 @@ func (r *mockReceiver) GetMetadata() metadata.MD {
 
 type mockTracesReceiver struct {
 	mockReceiver
-	lastRequest *otlptraces.ExportTraceServiceRequest
+	lastRequest pdata.Traces
 }
 
-func (r *mockTracesReceiver) Export(
-	ctx context.Context,
-	req *otlptraces.ExportTraceServiceRequest,
-) (*otlptraces.ExportTraceServiceResponse, error) {
+func (r *mockTracesReceiver) Export(ctx context.Context, td pdata.Traces) (interface{}, error) {
 	atomic.AddInt32(&r.requestCount, 1)
-	spanCount := 0
-	for _, rs := range req.ResourceSpans {
-		for _, ils := range rs.InstrumentationLibrarySpans {
-			spanCount += len(ils.Spans)
-		}
-	}
-	atomic.AddInt32(&r.totalItems, int32(spanCount))
+	atomic.AddInt32(&r.totalItems, int32(td.SpanCount()))
 	r.mux.Lock()
 	defer r.mux.Unlock()
-	r.lastRequest = req
+	r.lastRequest = td
 	r.metadata, _ = metadata.FromIncomingContext(ctx)
-	return &otlptraces.ExportTraceServiceResponse{}, nil
+	return nil, nil
 }
 
-func (r *mockTracesReceiver) GetLastRequest() *otlptraces.ExportTraceServiceRequest {
+func (r *mockTracesReceiver) GetLastRequest() pdata.Traces {
 	r.mux.Lock()
 	defer r.mux.Unlock()
 	return r.lastRequest
 }
 
-func otlpTracesReceiverOnGRPCServer(ln net.Listener) *mockTracesReceiver {
+func otlpTracesReceiverOnGRPCServer(t *testing.T, ln net.Listener) *mockTracesReceiver {
 	rcv := &mockTracesReceiver{
 		mockReceiver: mockReceiver{
+			t:   t,
 			srv: obsreport.GRPCServerWithObservabilityEnabled(),
 		},
 	}
 
 	// Now run it as a gRPC server
-	otlptraces.RegisterTraceServiceServer(rcv.srv, rcv)
+	pdatagrpc.RegisterTracesServer(rcv.srv, rcv)
 	go func() {
 		_ = rcv.srv.Serve(ln)
 	}()
@@ -103,29 +93,20 @@ func otlpTracesReceiverOnGRPCServer(ln net.Listener) *mockTracesReceiver {
 
 type mockLogsReceiver struct {
 	mockReceiver
-	lastRequest *otlplogs.ExportLogsServiceRequest
+	lastRequest pdata.Logs
 }
 
-func (r *mockLogsReceiver) Export(
-	ctx context.Context,
-	req *otlplogs.ExportLogsServiceRequest,
-) (*otlplogs.ExportLogsServiceResponse, error) {
+func (r *mockLogsReceiver) Export(ctx context.Context, ld pdata.Logs) (interface{}, error) {
 	atomic.AddInt32(&r.requestCount, 1)
-	recordCount := 0
-	for _, rs := range req.ResourceLogs {
-		for _, il := range rs.InstrumentationLibraryLogs {
-			recordCount += len(il.Logs)
-		}
-	}
-	atomic.AddInt32(&r.totalItems, int32(recordCount))
+	atomic.AddInt32(&r.totalItems, int32(ld.LogRecordCount()))
 	r.mux.Lock()
 	defer r.mux.Unlock()
-	r.lastRequest = req
+	r.lastRequest = ld
 	r.metadata, _ = metadata.FromIncomingContext(ctx)
-	return &otlplogs.ExportLogsServiceResponse{}, nil
+	return nil, nil
 }
 
-func (r *mockLogsReceiver) GetLastRequest() *otlplogs.ExportLogsServiceRequest {
+func (r *mockLogsReceiver) GetLastRequest() pdata.Logs {
 	r.mux.Lock()
 	defer r.mux.Unlock()
 	return r.lastRequest
@@ -139,7 +120,7 @@ func otlpLogsReceiverOnGRPCServer(ln net.Listener) *mockLogsReceiver {
 	}
 
 	// Now run it as a gRPC server
-	otlplogs.RegisterLogsServiceServer(rcv.srv, rcv)
+	pdatagrpc.RegisterLogsServer(rcv.srv, rcv)
 	go func() {
 		_ = rcv.srv.Serve(ln)
 	}()
@@ -149,24 +130,21 @@ func otlpLogsReceiverOnGRPCServer(ln net.Listener) *mockLogsReceiver {
 
 type mockMetricsReceiver struct {
 	mockReceiver
-	lastRequest *otlpmetrics.ExportMetricsServiceRequest
+	lastRequest pdata.Metrics
 }
 
-func (r *mockMetricsReceiver) Export(
-	ctx context.Context,
-	req *otlpmetrics.ExportMetricsServiceRequest,
-) (*otlpmetrics.ExportMetricsServiceResponse, error) {
+func (r *mockMetricsReceiver) Export(ctx context.Context, md pdata.Metrics) (interface{}, error) {
 	atomic.AddInt32(&r.requestCount, 1)
-	_, recordCount := pdata.MetricsFromInternalRep(internal.MetricsFromOtlp(req)).MetricAndDataPointCount()
+	_, recordCount := md.MetricAndDataPointCount()
 	atomic.AddInt32(&r.totalItems, int32(recordCount))
 	r.mux.Lock()
 	defer r.mux.Unlock()
-	r.lastRequest = req
+	r.lastRequest = md
 	r.metadata, _ = metadata.FromIncomingContext(ctx)
-	return &otlpmetrics.ExportMetricsServiceResponse{}, nil
+	return nil, nil
 }
 
-func (r *mockMetricsReceiver) GetLastRequest() *otlpmetrics.ExportMetricsServiceRequest {
+func (r *mockMetricsReceiver) GetLastRequest() pdata.Metrics {
 	r.mux.Lock()
 	defer r.mux.Unlock()
 	return r.lastRequest
@@ -180,7 +158,7 @@ func otlpMetricsReceiverOnGRPCServer(ln net.Listener) *mockMetricsReceiver {
 	}
 
 	// Now run it as a gRPC server
-	otlpmetrics.RegisterMetricsServiceServer(rcv.srv, rcv)
+	pdatagrpc.RegisterMetricsServer(rcv.srv, rcv)
 	go func() {
 		_ = rcv.srv.Serve(ln)
 	}()
@@ -192,7 +170,7 @@ func TestSendTraces(t *testing.T) {
 	// Start an OTLP-compatible receiver.
 	ln, err := net.Listen("tcp", "localhost:")
 	require.NoError(t, err, "Failed to find an available address to run the gRPC server: %v", err)
-	rcv := otlpTracesReceiverOnGRPCServer(ln)
+	rcv := otlpTracesReceiverOnGRPCServer(t, ln)
 	// Also closes the connection.
 	defer rcv.srv.GracefulStop()
 
@@ -238,8 +216,6 @@ func TestSendTraces(t *testing.T) {
 	// A trace with 2 spans.
 	td = testdata.GenerateTracesTwoSpansSameResource()
 
-	expectedOTLPReq := internal.TracesToOtlp(td.Clone().InternalRep())
-
 	err = exp.ConsumeTraces(context.Background(), td)
 	assert.NoError(t, err)
 
@@ -253,7 +229,7 @@ func TestSendTraces(t *testing.T) {
 	// Verify received span.
 	assert.EqualValues(t, 2, atomic.LoadInt32(&rcv.totalItems))
 	assert.EqualValues(t, 2, atomic.LoadInt32(&rcv.requestCount))
-	assert.EqualValues(t, expectedOTLPReq, rcv.GetLastRequest())
+	assert.EqualValues(t, td, rcv.GetLastRequest())
 
 	require.EqualValues(t, rcv.GetMetadata().Get("header"), expectedHeader)
 }
@@ -308,8 +284,6 @@ func TestSendMetrics(t *testing.T) {
 	// A trace with 2 spans.
 	md = testdata.GenerateMetricsTwoMetrics()
 
-	expectedOTLPReq := internal.MetricsToOtlp(md.Clone().InternalRep())
-
 	err = exp.ConsumeMetrics(context.Background(), md)
 	assert.NoError(t, err)
 
@@ -323,7 +297,7 @@ func TestSendMetrics(t *testing.T) {
 	// Verify received metrics.
 	assert.EqualValues(t, 2, atomic.LoadInt32(&rcv.requestCount))
 	assert.EqualValues(t, 4, atomic.LoadInt32(&rcv.totalItems))
-	assert.EqualValues(t, expectedOTLPReq, rcv.GetLastRequest())
+	assert.EqualValues(t, md, rcv.GetLastRequest())
 
 	require.EqualValues(t, rcv.GetMetadata().Get("header"), expectedHeader)
 }
@@ -428,7 +402,7 @@ func TestSendTraceDataServerStartWhileRequest(t *testing.T) {
 	}()
 
 	time.Sleep(2 * time.Second)
-	rcv := otlpTracesReceiverOnGRPCServer(ln)
+	rcv := otlpTracesReceiverOnGRPCServer(t, ln)
 	defer rcv.srv.GracefulStop()
 	// Wait until one of the conditions below triggers.
 	select {
@@ -441,13 +415,13 @@ func TestSendTraceDataServerStartWhileRequest(t *testing.T) {
 }
 
 func startServerAndMakeRequest(t *testing.T, exp component.TracesExporter, td pdata.Traces, ln net.Listener) {
-	rcv := otlpTracesReceiverOnGRPCServer(ln)
+	rcv := otlpTracesReceiverOnGRPCServer(t, ln)
 	defer rcv.srv.GracefulStop()
 	// Ensure that initially there is no data in the receiver.
 	assert.EqualValues(t, 0, atomic.LoadInt32(&rcv.requestCount))
 
 	// Clone the request and store as expected.
-	expectedOTLPReq := internal.TracesToOtlp(td.Clone().InternalRep())
+	expectedData := td.Clone()
 
 	// Resend the request, this should succeed.
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -461,7 +435,7 @@ func startServerAndMakeRequest(t *testing.T, exp component.TracesExporter, td pd
 
 	// Verify received span.
 	assert.EqualValues(t, 2, atomic.LoadInt32(&rcv.totalItems))
-	assert.EqualValues(t, expectedOTLPReq, rcv.GetLastRequest())
+	assert.EqualValues(t, expectedData, rcv.GetLastRequest())
 }
 
 func TestSendLogData(t *testing.T) {
@@ -497,8 +471,8 @@ func TestSendLogData(t *testing.T) {
 	assert.EqualValues(t, 0, atomic.LoadInt32(&rcv.requestCount))
 
 	// Send empty request.
-	td := pdata.NewLogs()
-	assert.NoError(t, exp.ConsumeLogs(context.Background(), td))
+	ld := pdata.NewLogs()
+	assert.NoError(t, exp.ConsumeLogs(context.Background(), ld))
 
 	// Wait until it is received.
 	assert.Eventually(t, func() bool {
@@ -509,10 +483,9 @@ func TestSendLogData(t *testing.T) {
 	assert.EqualValues(t, 0, atomic.LoadInt32(&rcv.totalItems))
 
 	// A request with 2 log entries.
-	td = testdata.GenerateLogsTwoLogRecordsSameResource()
-	expectedOTLPReq := internal.LogsToOtlp(td.Clone().InternalRep())
+	ld = testdata.GenerateLogsTwoLogRecordsSameResource()
 
-	err = exp.ConsumeLogs(context.Background(), td)
+	err = exp.ConsumeLogs(context.Background(), ld)
 	assert.NoError(t, err)
 
 	// Wait until it is received.
@@ -523,5 +496,5 @@ func TestSendLogData(t *testing.T) {
 	// Verify received logs.
 	assert.EqualValues(t, 2, atomic.LoadInt32(&rcv.requestCount))
 	assert.EqualValues(t, 2, atomic.LoadInt32(&rcv.totalItems))
-	assert.EqualValues(t, expectedOTLPReq, rcv.GetLastRequest())
+	assert.EqualValues(t, ld, rcv.GetLastRequest())
 }