aws s3 exporter second PR, implementation

.github/CODEOWNERS

@@ -29,6 +29,7 @@ exporter/awscloudwatchlogsexporter/                  @open-telemetry/collector-c
 exporter/awsemfexporter/                             @open-telemetry/collector-contrib-approvers @Aneurysm9 @shaochengwang @mxiamxia
 exporter/awskinesisexporter/                         @open-telemetry/collector-contrib-approvers @Aneurysm9 @MovieStoreGuy
 exporter/awsprometheusremotewriteexporter/           @open-telemetry/collector-contrib-approvers @Aneurysm9 @alolita
+exporter/awss3exporter/                              @open-telemetry/collector-contrib-approvers @pmm-sumo
 exporter/awsxrayexporter/                            @open-telemetry/collector-contrib-approvers @willarmiros
 exporter/azuremonitorexporter/                       @open-telemetry/collector-contrib-approvers @pcwiese
 exporter/carbonexporter/                             @open-telemetry/collector-contrib-approvers @pjanotti

CHANGELOG.md

@@ -211,6 +211,7 @@
 - `schemaprocessor`: Starting the initial work to allow from translating from semantic convention to another (#8371)
 - `saphanareceiver`: Added implementation of SAP HANA Metric Receiver (#8827)
 - `logstransformprocessor`: Add implementation of Logs Transform Processor (#9335)
+- `awss3exporter`: Add aws s3 exporter (#2835)
 
 ### 💡 Enhancements 💡
 

exporter/awss3exporter/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

exporter/awss3exporter/README.md

@@ -0,0 +1,51 @@
+# AWS S3 Exporter for OpenTelemetry Collector
+
+| Status                   |                       |
+| ------------------------ |-----------------------|
+| Stability                | [in development]      |
+| Supported pipeline types | traces, logs          |
+| Distributions            | [contrib]             |
+
+## Schema supported
+This exporter targets to support proto/json and proto/binary format
+
+## Exporter Configuration
+
+The following exporter configuration parameters are supported. 
+
+| Name                   | Description                                                                        | Default |
+| :--------------------- | :--------------------------------------------------------------------------------- | ------- |
+| `region`               | AWS region.                                                                        |         |
+| `s3_bucket`            | S3 bucket                                                                          |         |
+| `s3_prefix`            | prefix for the S3 key (root directory inside bucket).                              |         |
+| `s3_partition`         | time granularity of S3 key: hour or minute                                         |"minute" |
+| `file_prefix`          | file prefix defined by user                                                        |         |
+| `marshaler_name`       | marshaler used to produce output data otlp_json or otlp_proto                      |         |
+
+# Example Configuration
+
+Following example configuration defines to store output in 'eu-central' region and bucket named 'databucket'.
+
+```yaml
+exporters:
+  awss3:
+    s3uploader:
+        region: 'eu-central-1'
+        s3_bucket: 'databucket'
+        s3_prefix: 'metric'
+        s3_partition: 'minute'
+```
+
+Logs and traces will be stored inside 'databucket' in the following path format.
+
+```console
+metric/year=XXXX/month=XX/day=XX/hour=XX/minute=XX
+```
+
+## AWS Credential Configuration
+
+This exporter follows default credential resolution for the
+[aws-sdk-go](https://docs.aws.amazon.com/sdk-for-go/api/index.html).
+
+Follow the [guidelines](https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html) for the
+credential configuration.

exporter/awss3exporter/config.go

@@ -0,0 +1,45 @@
+// Copyright 2022 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 awss3exporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awss3exporter"
+
+import (
+	"go.opentelemetry.io/collector/config"
+	"go.uber.org/zap"
+)
+
+// S3UploaderConfig contains aws s3 uploader related config to controls things
+// like bucket, prefix, batching, connections, retries, etc.
+type S3UploaderConfig struct {
+	Region      string `mapstructure:"region"`
+	S3Bucket    string `mapstructure:"s3_bucket"`
+	S3Prefix    string `mapstructure:"s3_prefix"`
+	S3Partition string `mapstructure:"s3_partition"`
+	FilePrefix  string `mapstructure:"file_prefix"`
+}
+
+// Config contains the main configuration options for the awskinesis exporter
+type Config struct {
+	config.ExporterSettings `mapstructure:",squash"`
+
+	S3Uploader    S3UploaderConfig `mapstructure:"s3uploader"`
+	MarshalerName string           `mapstructure:"marshaler_name"`
+
+	// ResourceToTelemetrySettings is the option for converting resource attrihutes to telemetry attributes.
+	// "Enabled" - A boolean field to enable/disable this option. Default is `false`.
+	// If enabled, all the resource attributes will be converted to metric labels by default.
+	// exporterhelper.ResourceToTelemetrySettings `mapstructure:"resource_to_telemetry_conversion"`
+
+	logger *zap.Logger
+}

exporter/awss3exporter/config_test.go

@@ -0,0 +1,81 @@
+// Copyright 2021 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
+//
+//      shttp://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 awss3exporter
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/service/servicetest"
+)
+
+func TestLoadConfig(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	assert.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Exporters[typeStr] = factory
+	cfg, err := servicetest.LoadConfigAndValidate(filepath.Join("testdata", "default.yaml"), factories)
+
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	e := cfg.Exporters[config.NewComponentID(typeStr)]
+
+	assert.Equal(t, e,
+		&Config{
+			ExporterSettings: config.NewExporterSettings(config.NewComponentID(typeStr)),
+
+			S3Uploader: S3UploaderConfig{
+				Region:      "us-east-1",
+				S3Partition: "minute",
+			},
+			MarshalerName: "otlp_json",
+		},
+	)
+}
+
+func TestConfig(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	assert.Nil(t, err)
+
+	factory := NewFactory()
+	factories.Exporters[factory.Type()] = factory
+	cfg, err := servicetest.LoadConfigAndValidate(
+		filepath.Join("testdata", "config.yaml"), factories)
+
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	e := cfg.Exporters[config.NewComponentID(typeStr)]
+
+	assert.Equal(t, e,
+		&Config{
+			ExporterSettings: config.NewExporterSettings(config.NewComponentID(typeStr)),
+
+			S3Uploader: S3UploaderConfig{
+				Region:      "us-east-1",
+				S3Bucket:    "foo",
+				S3Prefix:    "bar",
+				S3Partition: "minute",
+			},
+			MarshalerName: "otlp_json",
+		},
+	)
+}

exporter/awss3exporter/data_writer.go

@@ -0,0 +1,21 @@
+// Copyright 2022 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 awss3exporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awss3exporter"
+
+import "context"
+
+type DataWriter interface {
+	WriteBuffer(ctx context.Context, buf []byte, config *Config, metadata string, format string) error
+}

exporter/awss3exporter/exporter.go

@@ -0,0 +1,98 @@
+// Copyright 2021 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 awss3exporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awss3exporter"
+
+import (
+	"context"
+	"errors"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.uber.org/zap"
+)
+
+type S3Exporter struct {
+	config     config.Exporter
+	dataWriter DataWriter
+	logger     *zap.Logger
+	marshaler  Marshaler
+}
+
+func NewS3Exporter(config config.Exporter,
+	params component.ExporterCreateSettings) (*S3Exporter, error) {
+
+	if config == nil {
+		return nil, errors.New("s3 exporter config is nil")
+	}
+
+	logger := params.Logger
+	expConfig := config.(*Config)
+	expConfig.logger = logger
+
+	validateConfig := expConfig.Validate()
+
+	if validateConfig != nil {
+		return nil, validateConfig
+	}
+
+	marshaler, err := NewMarshaler(expConfig.MarshalerName, logger)
+	if err != nil {
+		return nil, errors.New("unknown marshaler")
+	}
+
+	s3Exporter := &S3Exporter{
+		config:     config,
+		dataWriter: &S3Writer{},
+		logger:     logger,
+		marshaler:  marshaler,
+	}
+	return s3Exporter, nil
+}
+
+func (e *S3Exporter) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: false}
+}
+
+func (e *S3Exporter) Start(ctx context.Context, host component.Host) error {
+	return nil
+}
+
+func (e *S3Exporter) ConsumeLogs(ctx context.Context, logs plog.Logs) error {
+	buf, err := e.marshaler.MarshalLogs(logs)
+
+	if err != nil {
+		return err
+	}
+	expConfig := e.config.(*Config)
+
+	return e.dataWriter.WriteBuffer(ctx, buf, expConfig, "logs", e.marshaler.Format())
+}
+
+func (e *S3Exporter) ConsumeTraces(ctx context.Context, traces ptrace.Traces) error {
+	buf, err := e.marshaler.MarshalTraces(traces)
+	if err != nil {
+		return err
+	}
+	expConfig := e.config.(*Config)
+
+	return e.dataWriter.WriteBuffer(ctx, buf, expConfig, "traces", e.marshaler.Format())
+}
+
+func (e *S3Exporter) Shutdown(context.Context) error {
+	return nil
+}

exporter/awss3exporter/exporter_test.go

@@ -0,0 +1,59 @@
+// Copyright 2022 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 awss3exporter
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.uber.org/zap"
+)
+
+var testLogs = []byte(`{"resourceLogs":[{"resource":{"attributes":[{"key":"_sourceCategory","value":{"stringValue":"logfile"}},{"key":"_sourceHost","value":{"stringValue":"host"}}]},"scopeLogs":[{"scope":{},"logRecords":[{"observedTimeUnixNano":"1654257420681895000","body":{"stringValue":"2022-06-03 13:57:00.62739 +0200 CEST m=+14.018296742 log entry14"},"attributes":[{"key":"log.file.path_resolved","value":{"stringValue":"logwriter/data.log"}}],"traceId":"","spanId":""}]}],"schemaUrl":"https://opentelemetry.io/schemas/1.6.1"}]}`)
+
+type TestWriter struct {
+	t *testing.T
+}
+
+func (testWriter *TestWriter) WriteBuffer(ctx context.Context, buf []byte, config *Config, metadata string, format string) error {
+	assert.Equal(testWriter.t, testLogs, buf)
+	return nil
+}
+
+func getTestLogs(tb testing.TB) plog.Logs {
+	logsMarshaler := plog.NewJSONUnmarshaler()
+	logs, err := logsMarshaler.UnmarshalLogs(testLogs)
+	assert.NoError(tb, err, "Can't unmarshal testing logs data -> %s", err)
+	return logs
+}
+
+func getLogExporter(t *testing.T) *S3Exporter {
+	marshaler, _ := NewMarshaler("otlp_json", zap.NewNop())
+	exporter := &S3Exporter{
+		config:     createDefaultConfig(),
+		dataWriter: &TestWriter{t},
+		logger:     zap.NewNop(),
+		marshaler:  marshaler,
+	}
+	return exporter
+}
+
+func TestLog(t *testing.T) {
+	logs := getTestLogs(t)
+	exporter := getLogExporter(t)
+	assert.NoError(t, exporter.ConsumeLogs(context.Background(), logs))
+}