feat: unroll processor

factories/processors.go

@@ -27,6 +27,7 @@ import (
 	"github.com/observiq/bindplane-agent/processor/samplingprocessor"
 	"github.com/observiq/bindplane-agent/processor/spancountprocessor"
 	"github.com/observiq/bindplane-agent/processor/throughputmeasurementprocessor"
+	"github.com/observiq/bindplane-agent/processor/unrollprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/cumulativetodeltaprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatorateprocessor"
@@ -87,4 +88,5 @@ var defaultProcessors = []processor.Factory{
 	throughputmeasurementprocessor.NewFactory(),
 	tailsamplingprocessor.NewFactory(),
 	transformprocessor.NewFactory(),
+	unrollprocessor.NewFactory(),
 }

go.mod

@@ -1,6 +1,6 @@
 module github.com/observiq/bindplane-agent
 
-go 1.22.7
+go 1.23.3
 
 require (
 	github.com/google/uuid v1.6.0
@@ -25,6 +25,7 @@ require (
 	github.com/observiq/bindplane-agent/processor/samplingprocessor v1.66.0
 	github.com/observiq/bindplane-agent/processor/spancountprocessor v1.66.0
 	github.com/observiq/bindplane-agent/processor/throughputmeasurementprocessor v1.66.0
+	github.com/observiq/bindplane-agent/processor/unrollprocessor v1.66.0
 	github.com/observiq/bindplane-agent/receiver/awss3rehydrationreceiver v1.66.0
 	github.com/observiq/bindplane-agent/receiver/azureblobrehydrationreceiver v1.66.0
 	github.com/observiq/bindplane-agent/receiver/httpreceiver v1.66.0
@@ -853,6 +854,8 @@ replace github.com/observiq/bindplane-agent/processor/datapointcountprocessor =>
 
 replace github.com/observiq/bindplane-agent/processor/lookupprocessor => ./processor/lookupprocessor
 
+replace github.com/observiq/bindplane-agent/processor/unrollprocessor => ./processor/unrollprocessor
+
 replace github.com/observiq/bindplane-agent/expr => ./expr
 
 replace github.com/observiq/bindplane-agent/counter => ./counter

processor/unrollprocessor/Makefile

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

processor/unrollprocessor/config.go

@@ -0,0 +1,42 @@
+// Copyright  observIQ, Inc.
+//
+// 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 unrollprocessor
+
+import (
+	"errors"
+)
+
+// Config is the configuration for the unroll processor.
+type Config struct {
+	UnrollKey string      `mapstructure:"unroll_key"`
+	Field     UnrollField `mapstructure:"field"`
+}
+
+// UnrollField is the field to unroll.
+type UnrollField string
+
+const (
+	// UnrollFieldBody is the only supported field for unrolling logs.
+	UnrollFieldBody UnrollField = "body"
+)
+
+// Validate checks the configuration for any issues.
+func (c *Config) Validate() error {
+	if c.Field != UnrollFieldBody {
+		return errors.New("only unrolling logs from a body slice is currently supported")
+	}
+
+	return nil
+}

processor/unrollprocessor/doc.go

@@ -0,0 +1,19 @@
+// Copyright  observIQ, Inc.
+//
+// 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.
+
+//go:generate mdatagen metadata.yaml
+
+// Package transformprocessor contains the logic to execute telemetry transform based
+// on the OpenTelemetry Transformation Language.
+package unrollprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/unrollprocessor"

processor/unrollprocessor/factory.go

@@ -0,0 +1,68 @@
+// Copyright  observIQ, Inc.
+//
+// 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 unrollprocessor // import "github.com/observiq/bindplane-agent/processor/unrollprocessor"
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/collector/processor/processorhelper"
+
+	"github.com/observiq/bindplane-agent/processor/unrollprocessor/internal/metadata"
+)
+
+var processorCapabilities = consumer.Capabilities{MutatesData: true}
+
+const typeStr = "unroll"
+
+// NewFactory returns a new factory for the Transform processor.
+func NewFactory() processor.Factory {
+	return processor.NewFactory(
+		metadata.Type,
+		createDefaultConfig,
+		processor.WithLogs(createLogsProcessor, metadata.LogsStability),
+	)
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{
+		UnrollKey: "message",
+		Field:     UnrollFieldBody,
+	}
+}
+
+func createLogsProcessor(
+	ctx context.Context,
+	set processor.Settings,
+	cfg component.Config,
+	nextConsumer consumer.Logs,
+) (processor.Logs, error) {
+	oCfg := cfg.(*Config)
+
+	proc, err := newUnrollProcessor(oCfg)
+	if err != nil {
+		return nil, fmt.Errorf("invalid config for \"unroll\" processor %w", err)
+	}
+	return processorhelper.NewLogs(
+		ctx,
+		set,
+		cfg,
+		nextConsumer,
+		proc.ProcessLogs,
+		processorhelper.WithCapabilities(processorCapabilities))
+}

processor/unrollprocessor/go.mod

@@ -0,0 +1,56 @@
+module github.com/observiq/bindplane-agent/processor/unrollprocessor
+
+go 1.23.3
+
+require (
+	github.com/stretchr/testify v1.9.0
+	go.opentelemetry.io/collector/component v0.114.0
+	go.opentelemetry.io/collector/component/componenttest v0.114.0
+	go.opentelemetry.io/collector/confmap v1.20.0
+	go.opentelemetry.io/collector/consumer v0.114.0
+	go.opentelemetry.io/collector/consumer/consumertest v0.114.0
+	go.opentelemetry.io/collector/pdata v1.20.0
+	go.opentelemetry.io/collector/processor v0.114.0
+	go.opentelemetry.io/collector/processor/processortest v0.114.0
+	go.uber.org/goleak v1.3.0
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/go-logr/logr v1.4.2 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/knadh/koanf/maps v0.1.1 // indirect
+	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect
+	github.com/knadh/koanf/v2 v2.1.2 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/rogpeppe/go-internal v1.12.0 // indirect
+	go.opentelemetry.io/collector/component/componentstatus v0.114.0 // indirect
+	go.opentelemetry.io/collector/config/configtelemetry v0.114.0 // indirect
+	go.opentelemetry.io/collector/consumer/consumerprofiles v0.114.0 // indirect
+	go.opentelemetry.io/collector/pdata/pprofile v0.114.0 // indirect
+	go.opentelemetry.io/collector/pdata/testdata v0.114.0 // indirect
+	go.opentelemetry.io/collector/pipeline v0.114.0 // indirect
+	go.opentelemetry.io/collector/processor/processorprofiles v0.114.0 // indirect
+	go.opentelemetry.io/otel v1.32.0 // indirect
+	go.opentelemetry.io/otel/metric v1.32.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.32.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.32.0 // indirect
+	go.opentelemetry.io/otel/trace v1.32.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/net v0.30.0 // indirect
+	golang.org/x/sys v0.27.0 // indirect
+	golang.org/x/text v0.19.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240822170219-fc7c04adadcd // indirect
+	google.golang.org/grpc v1.67.1 // indirect
+	google.golang.org/protobuf v1.35.1 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)