[exporter/awss3] add marsheler config to pick only body content of logs

.chloggen/feat_allow_pick_only_log_body.yaml

@@ -0,0 +1,26 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: awss3exporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add a marshaler that stores the body of log records in s3.
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [30318]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/awss3exporter/README.md

@@ -43,6 +43,8 @@ Marshaler determines the format of data sent to AWS S3. Currently, the following
 - `otlp_proto`: the [OpenTelemetry Protocol format](https://github.com/open-telemetry/opentelemetry-proto), represented as Protocol Buffers. A single protobuf message is written into each object.
 - `sumo_ic`: the [Sumo Logic Installed Collector Archive format](https://help.sumologic.com/docs/manage/data-archiving/archive/).
   **This format is supported only for logs.**
+- `body`: export the log body as string.
+  **This format is supported only for logs.**
 
 # Example Configuration
 

exporter/awss3exporter/body_marshaler.go

@@ -0,0 +1,52 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package awss3exporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awss3exporter"
+
+import (
+	"bytes"
+	"fmt"
+
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+type bodyMarshaler struct{}
+
+func (*bodyMarshaler) format() string {
+	return "txt"
+}
+
+func newbodyMarshaler() bodyMarshaler {
+	return bodyMarshaler{}
+}
+
+func (bodyMarshaler) MarshalLogs(ld plog.Logs) ([]byte, error) {
+	buf := bytes.Buffer{}
+	rls := ld.ResourceLogs()
+	for i := 0; i < rls.Len(); i++ {
+		rl := rls.At(i)
+
+		ills := rl.ScopeLogs()
+		for j := 0; j < ills.Len(); j++ {
+			ils := ills.At(j)
+			logs := ils.LogRecords()
+			for k := 0; k < logs.Len(); k++ {
+				lr := logs.At(k)
+				body := lr.Body()
+				buf.WriteString(body.AsString())
+				buf.WriteString("\n")
+			}
+		}
+	}
+	return buf.Bytes(), nil
+}
+
+func (s bodyMarshaler) MarshalTraces(_ ptrace.Traces) ([]byte, error) {
+	return nil, fmt.Errorf("traces can't be marshaled into %s format", s.format())
+}
+
+func (s bodyMarshaler) MarshalMetrics(_ pmetric.Metrics) ([]byte, error) {
+	return nil, fmt.Errorf("metrics can't be marshaled into %s format", s.format())
+}

exporter/awss3exporter/body_marshaler_test.go

@@ -0,0 +1,81 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package awss3exporter
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+)
+
+func TestBodyMarshalerWithBooleanType(t *testing.T) {
+	logs := plog.NewLogs()
+	rls := logs.ResourceLogs().AppendEmpty()
+	sl := rls.ScopeLogs().AppendEmpty()
+
+	const recordNum = 0
+	ts := pcommon.Timestamp(int64(recordNum) * time.Millisecond.Nanoseconds())
+	logRecord := sl.LogRecords().AppendEmpty()
+	logRecord.SetTimestamp(ts)
+
+	// Boolean
+	logRecord.Body().SetBool(true)
+
+	marshaler := &bodyMarshaler{}
+	require.NotNil(t, marshaler)
+	body, err := marshaler.MarshalLogs(logs)
+	assert.NoError(t, err)
+	assert.Equal(t, body, []byte("true\n"))
+}
+
+func TestBodyMarshalerWithNumberType(t *testing.T) {
+	logs := plog.NewLogs()
+	rls := logs.ResourceLogs().AppendEmpty()
+	sl := rls.ScopeLogs().AppendEmpty()
+
+	const recordNum = 0
+	ts := pcommon.Timestamp(int64(recordNum) * time.Millisecond.Nanoseconds())
+	logRecord := sl.LogRecords().AppendEmpty()
+	logRecord.SetTimestamp(ts)
+
+	// Number
+	logRecord.Body().SetDouble(0.05)
+
+	marshaler := &bodyMarshaler{}
+	require.NotNil(t, marshaler)
+	body, err := marshaler.MarshalLogs(logs)
+	assert.NoError(t, err)
+	assert.Equal(t, body, []byte("0.05\n"))
+}
+
+func TestBodyMarshalerWithMapType(t *testing.T) {
+	logs := plog.NewLogs()
+	rls := logs.ResourceLogs().AppendEmpty()
+	sl := rls.ScopeLogs().AppendEmpty()
+
+	const recordNum = 0
+	ts := pcommon.Timestamp(int64(recordNum) * time.Millisecond.Nanoseconds())
+	logRecord := sl.LogRecords().AppendEmpty()
+	logRecord.SetTimestamp(ts)
+
+	// Map
+	m := logRecord.Body().SetEmptyMap()
+	m.PutStr("foo", "foo")
+	m.PutStr("bar", "bar")
+	m.PutBool("foobar", false)
+	m.PutDouble("foobardouble", 0.006)
+	m.PutInt("foobarint", 1)
+
+	var expect = `{"bar":"bar","foo":"foo","foobar":false,"foobardouble":0.006,"foobarint":1}`
+
+	marshaler := &bodyMarshaler{}
+	require.NotNil(t, marshaler)
+	body, err := marshaler.MarshalLogs(logs)
+	assert.NoError(t, err)
+	assert.Equal(t, body, []byte(expect+"\n"))
+}

exporter/awss3exporter/config.go

@@ -29,6 +29,7 @@ const (
 	OtlpProtobuf MarshalerType = "otlp_proto"
 	OtlpJSON     MarshalerType = "otlp_json"
 	SumoIC       MarshalerType = "sumo_ic"
+	Body         MarshalerType = "body"
 )
 
 // Config contains the main configuration options for the s3 exporter

exporter/awss3exporter/marshaler.go

@@ -40,6 +40,10 @@ func newMarshaler(mType MarshalerType, logger *zap.Logger) (marshaler, error) {
 		sumomarshaler := newSumoICMarshaler()
 		marshaler.logsMarshaler = &sumomarshaler
 		marshaler.fileFormat = "json.gz"
+	case Body:
+		exportbodyMarshaler := newbodyMarshaler()
+		marshaler.logsMarshaler = &exportbodyMarshaler
+		marshaler.fileFormat = exportbodyMarshaler.format()
 	default:
 		return nil, ErrUnknownMarshaler
 	}

exporter/awss3exporter/marshaler_test.go

@@ -35,4 +35,10 @@ func TestMarshaler(t *testing.T) {
 		assert.Error(t, err)
 		require.Nil(t, m)
 	}
+	{
+		m, err := newMarshaler("body", zap.NewNop())
+		assert.NoError(t, err)
+		require.NotNil(t, m)
+		assert.Equal(t, m.format(), "txt")
+	}
 }