Add otlpjson as a serializer implementation (#3238)

Signed-off-by: Bogdan Drutu <[email protected]>

internal/otlp/decoder.go

@@ -0,0 +1,69 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package otlp
+
+import (
+	"bytes"
+
+	"github.com/gogo/protobuf/jsonpb"
+
+	otlpcollectorlogs "go.opentelemetry.io/collector/internal/data/protogen/collector/logs/v1"
+	otlpcollectormetrics "go.opentelemetry.io/collector/internal/data/protogen/collector/metrics/v1"
+	otlpcollectortrace "go.opentelemetry.io/collector/internal/data/protogen/collector/trace/v1"
+	"go.opentelemetry.io/collector/internal/model"
+)
+
+type decoder struct {
+	delegate jsonpb.Unmarshaler
+}
+
+// NewJSONTracesDecoder returns a serializer.TracesDecoder to decode from OTLP json bytes.
+func NewJSONTracesDecoder() model.TracesDecoder {
+	return &decoder{delegate: jsonpb.Unmarshaler{}}
+}
+
+// NewJSONMetricsDecoder returns a serializer.MetricsDecoder to decode from OTLP json bytes.
+func NewJSONMetricsDecoder() model.MetricsDecoder {
+	return &decoder{delegate: jsonpb.Unmarshaler{}}
+}
+
+// NewJSONLogsDecoder returns a serializer.LogsDecoder to decode from OTLP json bytes.
+func NewJSONLogsDecoder() model.LogsDecoder {
+	return &decoder{delegate: jsonpb.Unmarshaler{}}
+}
+
+func (d *decoder) DecodeLogs(buf []byte) (interface{}, error) {
+	ld := &otlpcollectorlogs.ExportLogsServiceRequest{}
+	if err := d.delegate.Unmarshal(bytes.NewReader(buf), ld); err != nil {
+		return nil, err
+	}
+	return ld, nil
+}
+
+func (d *decoder) DecodeMetrics(buf []byte) (interface{}, error) {
+	md := &otlpcollectormetrics.ExportMetricsServiceRequest{}
+	if err := d.delegate.Unmarshal(bytes.NewReader(buf), md); err != nil {
+		return nil, err
+	}
+	return md, nil
+}
+
+func (d *decoder) DecodeTraces(buf []byte) (interface{}, error) {
+	td := &otlpcollectortrace.ExportTraceServiceRequest{}
+	if err := d.delegate.Unmarshal(bytes.NewReader(buf), td); err != nil {
+		return nil, err
+	}
+	return td, nil
+}

internal/otlp/encoder.go

@@ -0,0 +1,81 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package otlp
+
+import (
+	"bytes"
+
+	"github.com/gogo/protobuf/jsonpb"
+
+	otlpcollectorlogs "go.opentelemetry.io/collector/internal/data/protogen/collector/logs/v1"
+	otlpcollectormetrics "go.opentelemetry.io/collector/internal/data/protogen/collector/metrics/v1"
+	otlpcollectortrace "go.opentelemetry.io/collector/internal/data/protogen/collector/trace/v1"
+	"go.opentelemetry.io/collector/internal/model"
+)
+
+type encoder struct {
+	delegate jsonpb.Marshaler
+}
+
+// NewJSONTracesEncoder returns a serializer.TracesUnmarshaler to encode to OTLP json bytes.
+func NewJSONTracesEncoder() model.TracesEncoder {
+	return &encoder{delegate: jsonpb.Marshaler{}}
+}
+
+// NewJSONMetricsEncoder returns a serializer.MetricsEncoder to encode to OTLP json bytes.
+func NewJSONMetricsEncoder() model.MetricsEncoder {
+	return &encoder{delegate: jsonpb.Marshaler{}}
+}
+
+// NewJSONLogsEncoder returns a serializer.LogsEncoder to encode to OTLP json bytes.
+func NewJSONLogsEncoder() model.LogsEncoder {
+	return &encoder{delegate: jsonpb.Marshaler{}}
+}
+
+func (e *encoder) EncodeLogs(modelData interface{}) ([]byte, error) {
+	ld, ok := modelData.(*otlpcollectorlogs.ExportLogsServiceRequest)
+	if !ok {
+		return nil, model.NewErrIncompatibleType(&otlpcollectorlogs.ExportLogsServiceRequest{}, modelData)
+	}
+	buf := bytes.Buffer{}
+	if err := e.delegate.Marshal(&buf, ld); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+func (e *encoder) EncodeMetrics(modelData interface{}) ([]byte, error) {
+	md, ok := modelData.(*otlpcollectormetrics.ExportMetricsServiceRequest)
+	if !ok {
+		return nil, model.NewErrIncompatibleType(&otlpcollectormetrics.ExportMetricsServiceRequest{}, modelData)
+	}
+	buf := bytes.Buffer{}
+	if err := e.delegate.Marshal(&buf, md); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+func (e *encoder) EncodeTraces(modelData interface{}) ([]byte, error) {
+	td, ok := modelData.(*otlpcollectortrace.ExportTraceServiceRequest)
+	if !ok {
+		return nil, model.NewErrIncompatibleType(&otlpcollectortrace.ExportTraceServiceRequest{}, modelData)
+	}
+	buf := bytes.Buffer{}
+	if err := e.delegate.Marshal(&buf, td); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}

internal/otlp/from_translator.go

@@ -0,0 +1,50 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package otlp
+
+import (
+	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/internal"
+	"go.opentelemetry.io/collector/internal/model"
+)
+
+type fromTranslator struct{}
+
+// NewFromTracesTranslator returns a model.FromTracesTranslator to convert from pdata to OTLP data model.
+func NewFromTracesTranslator() model.FromTracesTranslator {
+	return &fromTranslator{}
+}
+
+// NewFromMetricsTranslator returns a model.FromMetricsTranslator to convert from pdata to OTLP data model.
+func NewFromMetricsTranslator() model.FromMetricsTranslator {
+	return &fromTranslator{}
+}
+
+// NewFromLogsTranslator returns a model.FromLogsTranslator to convert from pdata to OTLP data model.
+func NewFromLogsTranslator() model.FromLogsTranslator {
+	return &fromTranslator{}
+}
+
+func (d *fromTranslator) FromLogs(ld pdata.Logs) (interface{}, error) {
+	return internal.LogsToOtlp(ld.InternalRep()), nil
+}
+
+func (d *fromTranslator) FromMetrics(md pdata.Metrics) (interface{}, error) {
+	return internal.MetricsToOtlp(md.InternalRep()), nil
+}
+
+func (d *fromTranslator) FromTraces(td pdata.Traces) (interface{}, error) {
+	return internal.TracesToOtlp(td.InternalRep()), nil
+}

internal/otlp/to_translator.go

@@ -0,0 +1,65 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package otlp
+
+import (
+	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/internal"
+	otlpcollectorlogs "go.opentelemetry.io/collector/internal/data/protogen/collector/logs/v1"
+	otlpcollectormetrics "go.opentelemetry.io/collector/internal/data/protogen/collector/metrics/v1"
+	otlpcollectortrace "go.opentelemetry.io/collector/internal/data/protogen/collector/trace/v1"
+	"go.opentelemetry.io/collector/internal/model"
+)
+
+type toTranslator struct{}
+
+// NewToTracesTranslator returns a model.ToTracesTranslator to convert from OTLP data model to pdata.
+func NewToTracesTranslator() model.ToTracesTranslator {
+	return &toTranslator{}
+}
+
+// NewToMetricsTranslator returns a model.ToMetricsTranslator to convert from OTLP data model to pdata.
+func NewToMetricsTranslator() model.ToMetricsTranslator {
+	return &toTranslator{}
+}
+
+// NewToLogsTranslator returns a model.ToLogsTranslator to convert from OTLP data model to pdata.
+func NewToLogsTranslator() model.ToLogsTranslator {
+	return &toTranslator{}
+}
+
+func (d *toTranslator) ToLogs(modelData interface{}) (pdata.Logs, error) {
+	ld, ok := modelData.(*otlpcollectorlogs.ExportLogsServiceRequest)
+	if !ok {
+		return pdata.Logs{}, model.NewErrIncompatibleType(&otlpcollectorlogs.ExportLogsServiceRequest{}, modelData)
+	}
+	return pdata.LogsFromInternalRep(internal.LogsFromOtlp(ld)), nil
+}
+
+func (d *toTranslator) ToMetrics(modelData interface{}) (pdata.Metrics, error) {
+	ld, ok := modelData.(*otlpcollectormetrics.ExportMetricsServiceRequest)
+	if !ok {
+		return pdata.Metrics{}, model.NewErrIncompatibleType(&otlpcollectormetrics.ExportMetricsServiceRequest{}, modelData)
+	}
+	return pdata.MetricsFromInternalRep(internal.MetricsFromOtlp(ld)), nil
+}
+
+func (d *toTranslator) ToTraces(modelData interface{}) (pdata.Traces, error) {
+	td, ok := modelData.(*otlpcollectortrace.ExportTraceServiceRequest)
+	if !ok {
+		return pdata.Traces{}, model.NewErrIncompatibleType(&otlpcollectortrace.ExportTraceServiceRequest{}, modelData)
+	}
+	return pdata.TracesFromInternalRep(internal.TracesFromOtlp(td)), nil
+}