Skip to content

Commit

Permalink
Use otlp request in wrapper, hide members in the wrapper (#2692)
Browse files Browse the repository at this point in the history 
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 <[email protected]>
  • Loading branch information
@bogdandrutu
bogdandrutu authored Mar 17, 2021
1 parent 0215107 commit df9f565
Show file tree
Hide file tree
Showing 12 changed files with 228 additions and 218 deletions.
57 changes: 24 additions & 33 deletions consumer/pdata/log.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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.
Expand Down
45 changes: 26 additions & 19 deletions consumer/pdata/log_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)

Expand Down Expand Up @@ -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) {
Expand All @@ -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")
}

Expand Down Expand Up @@ -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())
}
}
7 changes: 2 additions & 5 deletions exporter/fileexporter/file_exporter.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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 {
Expand Down
134 changes: 69 additions & 65 deletions exporter/fileexporter/file_exporter_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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",
},
},
},
},
Expand Down Expand Up @@ -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()))
})
}
Expand Down
6 changes: 2 additions & 4 deletions exporter/otlpexporter/otlp.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down
Loading

0 comments on commit df9f565

Please sign in to comment.