Refactor AWS Kinesis exporter to allow for extended types of encoding (…

…#3655)

* Adding support for future dynmaic encoding

In order to make the transition from only jaeger traces to otlp data
being written to kinesis, we've split up our approach and made sure it
does not introduce breaking changes.

* Adding in trivial tests for the jaeger encoder

Adding in tests for the encoder to ensure the expected errors are being
raised.

* Fixing linting issues

* Adding in license to files that are missing it

* Fixing typo in godoc

* Moving translate package to be internal only.

After some discussion on the PR, in order to keep things simple and
easy to follow, encoding is moved to be internal/translate.

* Fixing left over imports

* Addressing original TODO and removing log line.

* Fixing imports

exporter/awskinesisexporter/exporter.go

@@ -16,19 +16,21 @@ package awskinesisexporter
 
 import (
 	"context"
+	"fmt"
 
 	awskinesis "github.com/signalfx/opencensus-go-exporter-kinesis"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/consumer/pdata"
-	jaegertranslator "go.opentelemetry.io/collector/translator/trace/jaeger"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awskinesisexporter/internal/translate"
 )
 
 // Exporter implements an OpenTelemetry trace exporter that exports all spans to AWS Kinesis
 type Exporter struct {
 	awskinesis *awskinesis.Exporter
+	ew         translate.ExportWriter
 	logger     *zap.Logger
 }
 
@@ -55,24 +57,9 @@ func (e Exporter) Shutdown(context.Context) error {
 
 // ConsumeTraces receives a span batch and exports it to AWS Kinesis
 func (e Exporter) ConsumeTraces(_ context.Context, td pdata.Traces) error {
-	pBatches, err := jaegertranslator.InternalTracesToJaegerProto(td)
+	err := e.ew.WriteTraces(td)
 	if err != nil {
-		e.logger.Error("error translating span batch", zap.Error(err))
-		return consumererror.Permanent(err)
-	}
-	// TODO: Use a multi error type
-	var exportErr error
-	for _, pBatch := range pBatches {
-		for _, span := range pBatch.GetSpans() {
-			if span.Process == nil {
-				span.Process = pBatch.Process
-			}
-			err := e.awskinesis.ExportSpan(span)
-			if err != nil {
-				e.logger.Error("error exporting span to awskinesis", zap.Error(err))
-				exportErr = err
-			}
-		}
+		err = fmt.Errorf("issues writing traces to kinesis: %w", err)
 	}
-	return exportErr
+	return err
 }

exporter/awskinesisexporter/factory.go

@@ -21,6 +21,8 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awskinesisexporter/internal/translate"
 )
 
 const (
@@ -93,5 +95,9 @@ func createTracesExporter(
 		return nil, err
 	}
 
-	return Exporter{k, params.Logger}, nil
+	return Exporter{
+		awskinesis: k,
+		ew:         translate.JaegerExporter(k),
+		logger:     params.Logger,
+	}, nil
 }

exporter/awskinesisexporter/internal/translate/translate.go

@@ -0,0 +1,36 @@
+// Copyright  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 translate
+
+import (
+	"errors"
+
+	"go.opentelemetry.io/collector/consumer/pdata"
+)
+
+var (
+	// ErrUnsupportedEncodedType is used when the encoder type does not support the type of encoding
+	ErrUnsupportedEncodedType = errors.New("unsupported type to encode")
+)
+
+// ExportWriter wraps the kinesis exporter and transforms the data into
+// the desired output format.
+type ExportWriter interface {
+	WriteMetrics(md pdata.Metrics) error
+
+	WriteTraces(td pdata.Traces) error
+
+	WriteLogs(ld pdata.Logs) error
+}

exporter/awskinesisexporter/internal/translate/translate_jaeger.go

@@ -0,0 +1,57 @@
+// Copyright  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 translate
+
+import (
+	awskinesis "github.com/signalfx/opencensus-go-exporter-kinesis"
+	"go.opentelemetry.io/collector/consumer/consumererror"
+	"go.opentelemetry.io/collector/consumer/pdata"
+	jaegertranslator "go.opentelemetry.io/collector/translator/trace/jaeger"
+)
+
+type jaeger struct {
+	kinesis *awskinesis.Exporter
+}
+
+// Ensure the jaeger encoder meets the interface at compile time.
+var _ ExportWriter = (*jaeger)(nil)
+
+func JaegerExporter(kinesis *awskinesis.Exporter) ExportWriter {
+	return &jaeger{kinesis: kinesis}
+}
+
+func (j *jaeger) WriteTraces(td pdata.Traces) error {
+	traces, err := jaegertranslator.InternalTracesToJaegerProto(td)
+	if err != nil {
+		return err
+	}
+
+	var errs []error
+	for _, trace := range traces {
+		for _, span := range trace.GetSpans() {
+			if span.Process == nil {
+				span.Process = trace.Process
+			}
+			if err := j.kinesis.ExportSpan(span); err != nil {
+				errs = append(errs, err)
+			}
+		}
+	}
+
+	return consumererror.Combine(errs)
+}
+
+func (j *jaeger) WriteMetrics(_ pdata.Metrics) error { return ErrUnsupportedEncodedType }
+func (j *jaeger) WriteLogs(_ pdata.Logs) error       { return ErrUnsupportedEncodedType }

exporter/awskinesisexporter/internal/translate/translate_jaeger_test.go

@@ -0,0 +1,42 @@
+// Copyright  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 translate_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/consumer/pdata"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awskinesisexporter/internal/translate"
+)
+
+func TestEncodingTraceData(t *testing.T) {
+	t.Parallel()
+
+	assert.NoError(t, translate.JaegerExporter(nil).WriteTraces(pdata.NewTraces()), "Must not error when processing spans")
+}
+
+func TestEncodingMetricData(t *testing.T) {
+	t.Parallel()
+
+	assert.Error(t, translate.JaegerExporter(nil).WriteMetrics(pdata.NewMetrics()), "Must error when trying to encode unsupported type")
+}
+
+func TestEncodingLogData(t *testing.T) {
+	t.Parallel()
+
+	assert.Error(t, translate.JaegerExporter(nil).WriteLogs(pdata.NewLogs()), "Must error when trying to encode unsupported type")
+}