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