[awss3exporter]: add Sumo Logic marshaler for logs in Installed Colle…

…ctor format (#23649)

**Description:** This PR adds a new marshaller for `awss3` exporter. It
exports logs in the format of Sumo Logic Installed Collector. Metrics
and traces are not supported - creating an exporter for them will result
in an error.
Currently, nested typed (eg. map inside a map) might not be supported
correctly - I have to confirm the IC's behavior with them, but I wanted
to create the PR so that it can be reviewed early.

**Link to tracking Issue:** #23212

**Testing:** Unit tests and manual e2e tests. Some automatic e2e tests
will come later, but they will not be part of this repo, they will be a
test for integrating the ingest with Sumo Logic's backend.

**Documentation:** Readme updated.

---------

Signed-off-by: Katarzyna Kujawa <[email protected]>
Co-authored-by: Katarzyna Kujawa <[email protected]>

.chloggen/awss3-sumologic-ic-marshaller.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# 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 Sumo Logic Installed Collector marshaler
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [23212]
+
+# (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:

exporter/awss3exporter/README.md

@@ -28,9 +28,17 @@ The following exporter configuration parameters are supported.
 | `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`    | marshaler used to produce output data otlp_json                                                      |             |
+| `marshaler`    | marshaler used to produce output data                                                                | `otlp_json` |
 | `endpoint`     | overrides the endpoint used by the exporter instead of constructing it from `region` and `s3_bucket` |             |
 
+### Marshaler
+
+Marshaler determines the format of data sent to AWS S3. Currently, the following marshalers are implemented:
+
+- `otlp_json` (default): the [OpenTelemetry Protocol format](https://github.com/open-telemetry/opentelemetry-proto), represented as json.
+- `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.**
+
 # Example Configuration
 
 Following example configuration defines to store output in 'eu-central' region and bucket named 'databucket'.

exporter/awss3exporter/config.go

@@ -24,6 +24,7 @@ type MarshalerType string
 
 const (
 	OtlpJSON MarshalerType = "otlp_json"
+	SumoIC   MarshalerType = "sumo_ic"
 )
 
 // Config contains the main configuration options for the s3 exporter

exporter/awss3exporter/config_test.go

@@ -124,3 +124,29 @@ func TestConfig_Validate(t *testing.T) {
 		})
 	}
 }
+
+func TestMarshallerName(t *testing.T) {
+	factories, err := otelcoltest.NopFactories()
+	assert.Nil(t, err)
+
+	factory := NewFactory()
+	factories.Exporters[factory.Type()] = factory
+	cfg, err := otelcoltest.LoadConfigAndValidate(
+		filepath.Join("testdata", "marshaler.yaml"), factories)
+
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	e := cfg.Exporters[component.NewID("awss3")].(*Config)
+
+	assert.Equal(t, e,
+		&Config{
+			S3Uploader: S3UploaderConfig{
+				Region:      "us-east-1",
+				S3Bucket:    "foo",
+				S3Partition: "minute",
+			},
+			MarshalerName: "sumo_ic",
+		},
+	)
+}

exporter/awss3exporter/factory.go

@@ -5,6 +5,7 @@ package awss3exporter // import "github.com/open-telemetry/opentelemetry-collect
 
 import (
 	"context"
+	"fmt"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter"
@@ -57,6 +58,10 @@ func createMetricsExporter(ctx context.Context,
 		return nil, err
 	}
 
+	if config.(*Config).MarshalerName == SumoIC {
+		return nil, fmt.Errorf("metrics are not supported by sumo_ic output format")
+	}
+
 	return exporterhelper.NewMetricsExporter(ctx, params,
 		config,
 		s3Exporter.ConsumeMetrics)
@@ -71,6 +76,10 @@ func createTracesExporter(ctx context.Context,
 		return nil, err
 	}
 
+	if config.(*Config).MarshalerName == SumoIC {
+		return nil, fmt.Errorf("traces are not supported by sumo_ic output format")
+	}
+
 	return exporterhelper.NewTracesExporter(ctx,
 		params,
 		config,

exporter/awss3exporter/factory_test.go

@@ -48,3 +48,21 @@ func TestCreateLogsExporter(t *testing.T) {
 	assert.NoError(t, err)
 	require.NotNil(t, exp)
 }
+
+func TestUnsupportedMarshalerOptions(t *testing.T) {
+	cfg := createDefaultConfig()
+	cfg.(*Config).MarshalerName = SumoIC
+	exp, err := createMetricsExporter(
+		context.Background(),
+		exportertest.NewNopCreateSettings(),
+		cfg)
+	assert.Error(t, err)
+	require.Nil(t, exp)
+
+	exp2, err := createTracesExporter(
+		context.Background(),
+		exportertest.NewNopCreateSettings(),
+		cfg)
+	assert.Error(t, err)
+	require.Nil(t, exp2)
+}

exporter/awss3exporter/marshaler.go

@@ -31,6 +31,10 @@ func NewMarshaler(mType MarshalerType, logger *zap.Logger) (marshaler, error) {
 		marshaler.tracesMarshaler = &ptrace.JSONMarshaler{}
 		marshaler.metricsMarshaler = &pmetric.JSONMarshaler{}
 		marshaler.fileFormat = "json"
+	case SumoIC:
+		sumomarshaler := newSumoICMarshaler()
+		marshaler.logsMarshaler = &sumomarshaler
+		marshaler.fileFormat = "json.gz"
 	default:
 		return nil, ErrUnknownMarshaler
 	}

exporter/awss3exporter/marshaler_test.go

@@ -18,6 +18,12 @@ func TestMarshaler(t *testing.T) {
 		require.NotNil(t, m)
 		assert.Equal(t, m.format(), "json")
 	}
+	{
+		m, err := NewMarshaler("sumo_ic", zap.NewNop())
+		assert.NoError(t, err)
+		require.NotNil(t, m)
+		assert.Equal(t, m.format(), "json.gz")
+	}
 	{
 		m, err := NewMarshaler("unknown", zap.NewNop())
 		assert.Error(t, err)

exporter/awss3exporter/sumo_marshaler.go

@@ -0,0 +1,162 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package awss3exporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awss3exporter"
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+const (
+	logBodyKey = "log"
+)
+
+type sumoMarshaler struct{}
+
+func (*sumoMarshaler) format() string {
+	return string(SumoIC)
+}
+
+func newSumoICMarshaler() sumoMarshaler {
+	return sumoMarshaler{}
+}
+
+func logEntry(buf *bytes.Buffer, format string, a ...interface{}) {
+	buf.WriteString(fmt.Sprintf(format, a...))
+	buf.WriteString("\n")
+}
+
+func attributeValueToString(v pcommon.Value) (string, error) {
+	switch v.Type() {
+	case pcommon.ValueTypeStr:
+		return v.Str(), nil
+	case pcommon.ValueTypeBool:
+		return strconv.FormatBool(v.Bool()), nil
+	case pcommon.ValueTypeBytes:
+		return valueToJSON(v.Bytes().AsRaw())
+	case pcommon.ValueTypeDouble:
+		return strconv.FormatFloat(v.Double(), 'f', -1, 64), nil
+	case pcommon.ValueTypeInt:
+		return strconv.FormatInt(v.Int(), 10), nil
+	case pcommon.ValueTypeSlice:
+		return valueToJSON(v.Slice().AsRaw())
+	case pcommon.ValueTypeMap:
+		return valueToJSON(v.Map().AsRaw())
+	case pcommon.ValueTypeEmpty:
+		return "", nil
+	default:
+		return "", fmt.Errorf("unknown OpenTelemetry attribute value type: %q", v.Type())
+	}
+}
+
+func valueToJSON(m any) (string, error) {
+	jsonString := new(bytes.Buffer)
+	enc := json.NewEncoder(jsonString)
+	err := enc.Encode(m)
+
+	return strings.Trim(jsonString.String(), "\n"), err
+}
+
+const (
+	SourceCategoryKey = "_sourceCategory"
+	SourceHostKey     = "_sourceHost"
+	SourceNameKey     = "_sourceName"
+)
+
+func (sumoMarshaler) MarshalLogs(ld plog.Logs) ([]byte, error) {
+	buf := bytes.Buffer{}
+	rls := ld.ResourceLogs()
+	for i := 0; i < rls.Len(); i++ {
+		rl := rls.At(i)
+		ra := rl.Resource().Attributes()
+		sourceCategory, exists := ra.Get(SourceCategoryKey)
+		if !exists {
+			return nil, errors.New("_sourceCategory attribute does not exist")
+		}
+		sourceHost, exists := ra.Get(SourceHostKey)
+		if !exists {
+			return nil, errors.New("_sourceHost attribute does not exist")
+		}
+		sourceName, exists := ra.Get(SourceNameKey)
+		if !exists {
+			return nil, errors.New("_sourceName attribute does not exist")
+		}
+
+		sc, err := attributeValueToString(sourceCategory)
+		if err != nil {
+			return nil, err
+		}
+		sh, err := attributeValueToString(sourceHost)
+		if err != nil {
+			return nil, err
+		}
+		sn, err := attributeValueToString(sourceName)
+		if err != nil {
+			return nil, err
+		}
+		sc = strconv.Quote(sc)
+		sh = strconv.Quote(sh)
+		sn = strconv.Quote(sn)
+
+		// Remove the source attributes so that they won't be included in "fields" value.
+		ra.Remove(SourceCategoryKey)
+		ra.Remove(SourceHostKey)
+		ra.Remove(SourceNameKey)
+
+		fields, err := valueToJSON(ra.AsRaw())
+		if err != nil {
+			return nil, err
+		}
+
+		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)
+				dateVal := lr.ObservedTimestamp()
+
+				message, err := getMessageJSON(lr)
+				if err != nil {
+					return nil, err
+				}
+
+				logEntry(&buf, "{\"date\": \"%s\",\"sourceName\":%s,\"sourceHost\":%s,\"sourceCategory\":%s,\"fields\":%s,\"message\":%s}",
+					dateVal, sn, sh, sc, fields, message)
+			}
+		}
+	}
+	return buf.Bytes(), nil
+}
+
+func getMessageJSON(lr plog.LogRecord) (string, error) {
+	// The "message" fields is a JSON created from combining the actual log body and log-level attributes,
+	// where the log body is stored under "log" key.
+	// More info:
+	// https://help.sumologic.com/docs/send-data/opentelemetry-collector/data-source-configurations/additional-configurations-reference/#mapping-opentelemetry-concepts-to-sumo-logic
+	message := new(bytes.Buffer)
+	enc := json.NewEncoder(message)
+
+	lr.Body().CopyTo(lr.Attributes().PutEmpty(logBodyKey))
+	err := enc.Encode(lr.Attributes().AsRaw())
+
+	return strings.Trim(message.String(), "\n"), err
+}
+
+func (s sumoMarshaler) MarshalTraces(_ ptrace.Traces) ([]byte, error) {
+	return nil, fmt.Errorf("traces can't be marshaled into %s format", s.format())
+}
+
+func (s sumoMarshaler) MarshalMetrics(_ pmetric.Metrics) ([]byte, error) {
+	return nil, fmt.Errorf("metrics can't be marshaled into %s format", s.format())
+}