Use otlp request in wrapper, hide members in the wrapper

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
+	orig *otlpcollectorlog.ExportLogsServiceRequest
 }
 
 // NewLogs creates a new Logs.
 func NewLogs() Logs {
-	orig := []*otlplogs.ResourceLogs(nil)
-	return Logs{&orig}
+	return Logs{orig: &otlpcollectorlog.ExportLogsServiceRequest{}}
 }
 
 // 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{orig: internal.LogsToOtlp(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{orig: &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 internal.LogsFromOtlp(ld.orig)
 }
 
 // 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 ld.orig.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,11 @@ 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 ld.orig.Size()
 }
 
 func (ld Logs) ResourceLogs() ResourceLogsSlice {
-	return ResourceLogsSlice(ld)
+	return newResourceLogsSlice(&ld.orig.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,35 @@ 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)
+	wrapper := internal.LogsFromOtlp(&otlpcollectorlog.ExportLogsServiceRequest{})
+	ld := LogsFromInternalRep(wrapper)
+	assert.EqualValues(t, NewLogs(), ld)
+	assert.EqualValues(t, &otlpcollectorlog.ExportLogsServiceRequest{}, ld.orig)
 }
 
 func TestLogsToFromOtlpProtoBytes(t *testing.T) {
@@ -78,14 +86,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 +134,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)
 	}