Use otlp request in wrapper, hide members in the wrapper

Tested in contrib and was able to access InternalWrapper.Orig,
with this change this is no longer possible.

Also removes one extra allocation for the request object when
using otlp receiver/exporter since we keep the initial pointer around.

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

consumer/pdata/log.go

@@ -29,56 +29,51 @@ import (
 // Must use NewLogs functions to create new instances.
 // Important: zero-initialized instance is not valid for use.
 type Logs struct {
-	orig *[]*otlplogs.ResourceLogs
+	wrapper internal.LogsWrapper
 }
 
 // NewLogs creates a new Logs.
 func NewLogs() Logs {
-	orig := []*otlplogs.ResourceLogs(nil)
-	return Logs{&orig}
+	return Logs{wrapper: internal.NewLogsWrapper()}
 }
 
 // LogsFromInternalRep creates the internal Logs representation from the ProtoBuf. Should
 // not be used outside this module. This is intended to be used only by OTLP exporter and
 // File exporter, which legitimately need to work with OTLP Protobuf structs.
-func LogsFromInternalRep(logs internal.OtlpLogsWrapper) Logs {
-	return Logs{logs.Orig}
+func LogsFromInternalRep(logs internal.LogsWrapper) Logs {
+	return Logs{wrapper: logs}
+}
+
+// LogsFromOtlpProtoBytes converts OTLP Collector ExportLogsServiceRequest
+// ProtoBuf bytes to the internal Logs.
+//
+// Returns an invalid Logs instance if error is not nil.
+func LogsFromOtlpProtoBytes(data []byte) (Logs, error) {
+	req := otlpcollectorlog.ExportLogsServiceRequest{}
+	if err := req.Unmarshal(data); err != nil {
+		return Logs{}, err
+	}
+	return Logs{wrapper: internal.LogsFromOtlp(&req)}, nil
 }
 
 // InternalRep returns internal representation of the logs. Should not be used outside
 // this module. This is intended to be used only by OTLP exporter and File exporter,
 // which legitimately need to work with OTLP Protobuf structs.
-func (ld Logs) InternalRep() internal.OtlpLogsWrapper {
-	return internal.OtlpLogsWrapper{Orig: ld.orig}
+func (ld Logs) InternalRep() internal.LogsWrapper {
+	return ld.wrapper
 }
 
 // ToOtlpProtoBytes returns the internal Logs to OTLP Collector ExportTraceServiceRequest
 // ProtoBuf bytes. This is intended to export OTLP Protobuf bytes for OTLP/HTTP transports.
 func (ld Logs) ToOtlpProtoBytes() ([]byte, error) {
-	logs := otlpcollectorlog.ExportLogsServiceRequest{
-		ResourceLogs: *ld.orig,
-	}
-	return logs.Marshal()
-}
-
-// FromOtlpProtoBytes converts OTLP Collector ExportLogsServiceRequest
-// ProtoBuf bytes to the internal Logs. Overrides current data.
-// Calling this function on zero-initialized structure causes panic.
-// Use it with NewLogs or on existing initialized Logs.
-func (ld Logs) FromOtlpProtoBytes(data []byte) error {
-	logs := otlpcollectorlog.ExportLogsServiceRequest{}
-	if err := logs.Unmarshal(data); err != nil {
-		return err
-	}
-	*ld.orig = logs.ResourceLogs
-	return nil
+	return internal.LogsToOtlp(ld.wrapper).Marshal()
 }
 
 // Clone returns a copy of Logs.
 func (ld Logs) Clone() Logs {
-	rls := NewResourceLogsSlice()
-	ld.ResourceLogs().CopyTo(rls)
-	return Logs(rls)
+	cloneLd := NewLogs()
+	ld.ResourceLogs().CopyTo(cloneLd.ResourceLogs())
+	return cloneLd
 }
 
 // LogRecordCount calculates the total number of log records.
@@ -99,15 +94,12 @@ func (ld Logs) LogRecordCount() int {
 // SizeBytes returns the number of bytes in the internal representation of the
 // logs.
 func (ld Logs) SizeBytes() int {
-	size := 0
-	for i := range *ld.orig {
-		size += (*ld.orig)[i].Size()
-	}
-	return size
+	return internal.LogsToOtlp(ld.wrapper).Size()
 }
 
 func (ld Logs) ResourceLogs() ResourceLogsSlice {
-	return ResourceLogsSlice(ld)
+	req := internal.LogsToOtlp(ld.wrapper)
+	return newResourceLogsSlice(&req.ResourceLogs)
 }
 
 // SeverityNumber is the public alias of otlplogs.SeverityNumber from internal package.

consumer/pdata/log_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"go.opentelemetry.io/collector/internal"
+	otlpcollectorlog "go.opentelemetry.io/collector/internal/data/protogen/collector/logs/v1"
 	otlplogs "go.opentelemetry.io/collector/internal/data/protogen/logs/v1"
 )
 
@@ -48,28 +49,34 @@ func TestLogRecordCount(t *testing.T) {
 }
 
 func TestLogRecordCountWithEmpty(t *testing.T) {
-	assert.EqualValues(t, 0, LogsFromInternalRep(internal.LogsFromOtlp([]*otlplogs.ResourceLogs{{}})).LogRecordCount())
-	assert.EqualValues(t, 0, LogsFromInternalRep(internal.LogsFromOtlp([]*otlplogs.ResourceLogs{
-		{
-			InstrumentationLibraryLogs: []*otlplogs.InstrumentationLibraryLogs{{}},
+	assert.Zero(t, NewLogs().LogRecordCount())
+	assert.Zero(t, LogsFromInternalRep(internal.LogsFromOtlp(&otlpcollectorlog.ExportLogsServiceRequest{
+		ResourceLogs: []*otlplogs.ResourceLogs{{}},
+	})).LogRecordCount())
+	assert.Zero(t, LogsFromInternalRep(internal.LogsFromOtlp(&otlpcollectorlog.ExportLogsServiceRequest{
+		ResourceLogs: []*otlplogs.ResourceLogs{
+			{
+				InstrumentationLibraryLogs: []*otlplogs.InstrumentationLibraryLogs{{}},
+			},
 		},
 	})).LogRecordCount())
-	assert.EqualValues(t, 1, LogsFromInternalRep(internal.LogsFromOtlp([]*otlplogs.ResourceLogs{
-		{
-			InstrumentationLibraryLogs: []*otlplogs.InstrumentationLibraryLogs{
-				{
-					Logs: []*otlplogs.LogRecord{{}},
+	assert.Equal(t, 1, LogsFromInternalRep(internal.LogsFromOtlp(&otlpcollectorlog.ExportLogsServiceRequest{
+		ResourceLogs: []*otlplogs.ResourceLogs{
+			{
+				InstrumentationLibraryLogs: []*otlplogs.InstrumentationLibraryLogs{
+					{
+						Logs: []*otlplogs.LogRecord{{}},
+					},
 				},
 			},
 		},
 	})).LogRecordCount())
 }
 
 func TestToFromLogProto(t *testing.T) {
-	otlp := []*otlplogs.ResourceLogs(nil)
-	td := LogsFromInternalRep(internal.LogsFromOtlp(otlp))
-	assert.EqualValues(t, NewLogs(), td)
-	assert.EqualValues(t, otlp, *td.orig)
+	ld := LogsFromInternalRep(internal.NewLogsWrapper())
+	assert.EqualValues(t, NewLogs(), ld)
+	assert.EqualValues(t, &otlpcollectorlog.ExportLogsServiceRequest{}, internal.LogsToOtlp(ld.wrapper))
 }
 
 func TestLogsToFromOtlpProtoBytes(t *testing.T) {
@@ -78,14 +85,13 @@ func TestLogsToFromOtlpProtoBytes(t *testing.T) {
 	bytes, err := send.ToOtlpProtoBytes()
 	assert.NoError(t, err)
 
-	recv := NewLogs()
-	err = recv.FromOtlpProtoBytes(bytes)
+	recv, err := LogsFromOtlpProtoBytes(bytes)
 	assert.NoError(t, err)
 	assert.EqualValues(t, send, recv)
 }
 
 func TestLogsFromInvalidOtlpProtoBytes(t *testing.T) {
-	err := NewLogs().FromOtlpProtoBytes([]byte{0xFF})
+	_, err := LogsFromOtlpProtoBytes([]byte{0xFF})
 	assert.EqualError(t, err, "unexpected EOF")
 }
 
@@ -127,8 +133,8 @@ func BenchmarkLogsFromOtlp(b *testing.B) {
 	b.ResetTimer()
 	b.ReportAllocs()
 	for n := 0; n < b.N; n++ {
-		traces := NewLogs()
-		require.NoError(b, traces.FromOtlpProtoBytes(buf))
-		assert.Equal(b, baseLogs.ResourceLogs().Len(), traces.ResourceLogs().Len())
+		logs, err := LogsFromOtlpProtoBytes(buf)
+		require.NoError(b, err)
+		assert.Equal(b, baseLogs.ResourceLogs().Len(), logs.ResourceLogs().Len())
 	}
 }

exporter/fileexporter/file_exporter.go

@@ -25,7 +25,6 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"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"
 	otlptrace "go.opentelemetry.io/collector/internal/data/protogen/collector/trace/v1"
 )
@@ -55,10 +54,8 @@ func (e *fileExporter) ConsumeMetrics(_ context.Context, md pdata.Metrics) error
 }
 
 func (e *fileExporter) ConsumeLogs(_ context.Context, ld pdata.Logs) error {
-	request := otlplogs.ExportLogsServiceRequest{
-		ResourceLogs: internal.LogsToOtlp(ld.InternalRep()),
-	}
-	return exportMessageAsLine(e, &request)
+	request := internal.LogsToOtlp(ld.InternalRep())
+	return exportMessageAsLine(e, request)
 }
 
 func exportMessageAsLine(e *fileExporter, message proto.Message) error {

exporter/fileexporter/file_exporter_test.go

@@ -73,105 +73,109 @@ func TestFileLogsExporterNoErrors(t *testing.T) {
 	require.NotNil(t, exporter)
 
 	now := time.Now()
-	ld := []*logspb.ResourceLogs{
-		{
-			Resource: otresourcepb.Resource{
-				Attributes: []otlpcommon.KeyValue{
-					{
-						Key:   "attr1",
-						Value: otlpcommon.AnyValue{Value: &otlpcommon.AnyValue_StringValue{StringValue: "value1"}},
-					},
-				},
-			},
-			InstrumentationLibraryLogs: []*logspb.InstrumentationLibraryLogs{
-				{
-					Logs: []*logspb.LogRecord{
-						{
-							TimeUnixNano: uint64(now.UnixNano()),
-							Name:         "logA",
-						},
+	otlp := &collectorlogs.ExportLogsServiceRequest{
+		ResourceLogs: []*logspb.ResourceLogs{
+			{
+				Resource: otresourcepb.Resource{
+					Attributes: []otlpcommon.KeyValue{
 						{
-							TimeUnixNano: uint64(now.UnixNano()),
-							Name:         "logB",
+							Key:   "attr1",
+							Value: otlpcommon.AnyValue{Value: &otlpcommon.AnyValue_StringValue{StringValue: "value1"}},
 						},
 					},
 				},
-			},
-		},
-		{
-			Resource: otresourcepb.Resource{
-				Attributes: []otlpcommon.KeyValue{
+				InstrumentationLibraryLogs: []*logspb.InstrumentationLibraryLogs{
 					{
-						Key:   "attr2",
-						Value: otlpcommon.AnyValue{Value: &otlpcommon.AnyValue_StringValue{StringValue: "value2"}},
+						Logs: []*logspb.LogRecord{
+							{
+								TimeUnixNano: uint64(now.UnixNano()),
+								Name:         "logA",
+							},
+							{
+								TimeUnixNano: uint64(now.UnixNano()),
+								Name:         "logB",
+							},
+						},
 					},
 				},
 			},
-			InstrumentationLibraryLogs: []*logspb.InstrumentationLibraryLogs{
-				{
-					Logs: []*logspb.LogRecord{
+			{
+				Resource: otresourcepb.Resource{
+					Attributes: []otlpcommon.KeyValue{
 						{
-							TimeUnixNano: uint64(now.UnixNano()),
-							Name:         "logC",
+							Key:   "attr2",
+							Value: otlpcommon.AnyValue{Value: &otlpcommon.AnyValue_StringValue{StringValue: "value2"}},
+						},
+					},
+				},
+				InstrumentationLibraryLogs: []*logspb.InstrumentationLibraryLogs{
+					{
+						Logs: []*logspb.LogRecord{
+							{
+								TimeUnixNano: uint64(now.UnixNano()),
+								Name:         "logC",
+							},
 						},
 					},
 				},
 			},
 		},
 	}
-	assert.NoError(t, exporter.ConsumeLogs(context.Background(), pdata.LogsFromInternalRep(internal.LogsFromOtlp(ld))))
+	assert.NoError(t, exporter.ConsumeLogs(context.Background(), pdata.LogsFromInternalRep(internal.LogsFromOtlp(otlp))))
 	assert.NoError(t, exporter.Shutdown(context.Background()))
 
 	var unmarshaler = &jsonpb.Unmarshaler{}
 	var j collectorlogs.ExportLogsServiceRequest
 
 	assert.NoError(t, unmarshaler.Unmarshal(mf, &j))
-	assert.EqualValues(t, ld, j.ResourceLogs)
+	assert.EqualValues(t, otlp.ResourceLogs, j.ResourceLogs)
 }
 
 func TestFileLogsExporterErrors(t *testing.T) {
 
 	now := time.Now()
-	ld := []*logspb.ResourceLogs{
-		{
-			Resource: otresourcepb.Resource{
-				Attributes: []otlpcommon.KeyValue{
-					{
-						Key:   "attr1",
-						Value: otlpcommon.AnyValue{Value: &otlpcommon.AnyValue_StringValue{StringValue: "value1"}},
-					},
-				},
-			},
-			InstrumentationLibraryLogs: []*logspb.InstrumentationLibraryLogs{
-				{
-					Logs: []*logspb.LogRecord{
-						{
-							TimeUnixNano: uint64(now.UnixNano()),
-							Name:         "logA",
-						},
+	otlp := &collectorlogs.ExportLogsServiceRequest{
+		ResourceLogs: []*logspb.ResourceLogs{
+			{
+				Resource: otresourcepb.Resource{
+					Attributes: []otlpcommon.KeyValue{
 						{
-							TimeUnixNano: uint64(now.UnixNano()),
-							Name:         "logB",
+							Key:   "attr1",
+							Value: otlpcommon.AnyValue{Value: &otlpcommon.AnyValue_StringValue{StringValue: "value1"}},
 						},
 					},
 				},
-			},
-		},
-		{
-			Resource: otresourcepb.Resource{
-				Attributes: []otlpcommon.KeyValue{
+				InstrumentationLibraryLogs: []*logspb.InstrumentationLibraryLogs{
 					{
-						Key:   "attr2",
-						Value: otlpcommon.AnyValue{Value: &otlpcommon.AnyValue_StringValue{StringValue: "value2"}},
+						Logs: []*logspb.LogRecord{
+							{
+								TimeUnixNano: uint64(now.UnixNano()),
+								Name:         "logA",
+							},
+							{
+								TimeUnixNano: uint64(now.UnixNano()),
+								Name:         "logB",
+							},
+						},
 					},
 				},
 			},
-			InstrumentationLibraryLogs: []*logspb.InstrumentationLibraryLogs{
-				{
-					Logs: []*logspb.LogRecord{
+			{
+				Resource: otresourcepb.Resource{
+					Attributes: []otlpcommon.KeyValue{
 						{
-							TimeUnixNano: uint64(now.UnixNano()),
-							Name:         "logC",
+							Key:   "attr2",
+							Value: otlpcommon.AnyValue{Value: &otlpcommon.AnyValue_StringValue{StringValue: "value2"}},
+						},
+					},
+				},
+				InstrumentationLibraryLogs: []*logspb.InstrumentationLibraryLogs{
+					{
+						Logs: []*logspb.LogRecord{
+							{
+								TimeUnixNano: uint64(now.UnixNano()),
+								Name:         "logC",
+							},
 						},
 					},
 				},
@@ -210,7 +214,7 @@ func TestFileLogsExporterErrors(t *testing.T) {
 			exporter := &fileExporter{file: mf}
 			require.NotNil(t, exporter)
 
-			assert.Error(t, exporter.ConsumeLogs(context.Background(), pdata.LogsFromInternalRep(internal.LogsFromOtlp(ld))))
+			assert.Error(t, exporter.ConsumeLogs(context.Background(), pdata.LogsFromInternalRep(internal.LogsFromOtlp(otlp))))
 			assert.NoError(t, exporter.Shutdown(context.Background()))
 		})
 	}

exporter/otlpexporter/otlp.go

@@ -85,10 +85,8 @@ func (e *exporterImp) pushMetricsData(ctx context.Context, md pdata.Metrics) err
 	return nil
 }
 
-func (e *exporterImp) pushLogData(ctx context.Context, logs pdata.Logs) error {
-	request := &otlplogs.ExportLogsServiceRequest{
-		ResourceLogs: internal.LogsToOtlp(logs.InternalRep()),
-	}
+func (e *exporterImp) pushLogData(ctx context.Context, ld pdata.Logs) error {
+	request := internal.LogsToOtlp(ld.InternalRep())
 	if err := e.w.exportLogs(ctx, request); err != nil {
 		return fmt.Errorf("failed to push log data via OTLP exporter: %w", err)
 	}