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

…gs (#30318)

**Description:** <Describe what has changed.>
This pull request introduces a new marshaller for the awss3 exporter. It
disregards resources, labels, and selectively extracts only the body
content of the logs, facilitating simplified AWS Athena connection and
querying. While straightforward, this enhancement accommodates various
use cases, enabling the extraction of specific logs for statistical
analysis and long-term archival reporting.

Example when write only body to S3:
```
EXTERNAL_REQUEST 2023-07-18T10:05:01.976Z GET 200 localhost /organization/v0/properties 0:0:0:0:0:0:0:1 4264 348 wUN0NeTMByHVmIOjo79ImwylAjjkRtkr
EXTERNAL_REQUEST 2023-07-18T10:05:01.976Z GET 200 localhost /organization/v0/properties 0:0:0:0:0:0:0:1 4264 348 wUN0NeTMByHVmIOjo79ImwylAjjkRtkr
EXTERNAL_REQUEST 2023-07-18T10:05:01.976Z GET 200 localhost /organization/v0/properties 0:0:0:0:0:0:0:1 4264 348 wUN0NeTMByHVmIOjo79ImwylAjjkRtkr
EXTERNAL_REQUEST 2023-07-18T10:05:01.976Z GET 200 localhost /organization/v0/properties 0:0:0:0:0:0:0:1 4264 348 wUN0NeTMByHVmIOjo79ImwylAjjkRtkr
```

**Testing:**
Work as expected when used Athena to Query:
<img width="962" alt="athena"
src="https://github.com/open-telemetry/opentelemetry-collector-contrib/assets/85854989/8f682024-df4e-402b-9de1-ee8268ea1936">

.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")
+ }
 }