Add decoupleafterbatch converter to ensure decouple processor follows batch processor

collector/go.mod

@@ -20,6 +20,7 @@ replace cloud.google.com/go => cloud.google.com/go v0.107.0
 
 require (
 	github.com/golang-collections/go-datastructures v0.0.0-20150211160725-59788d5eb259
+	github.com/google/go-cmp v0.6.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/confmap/provider/s3provider v0.92.0
 	github.com/open-telemetry/opentelemetry-lambda/collector/lambdacomponents v0.91.0
 	github.com/open-telemetry/opentelemetry-lambda/collector/lambdalifecycle v0.0.0-00010101000000-000000000000

collector/internal/collector/collector.go

@@ -32,6 +32,7 @@ import (
 	"go.uber.org/zap/zapcore"
 
 	"github.com/open-telemetry/opentelemetry-lambda/collector/internal/confmap/converter/disablequeuedretryconverter"
+	"github.com/open-telemetry/opentelemetry-lambda/collector/internal/confmap/converter/decoupleafterbatchconverter"
 )
 
 // Collector runs a single otelcol as a go routine within the
@@ -68,7 +69,7 @@ func NewCollector(logger *zap.Logger, factories otelcol.Factories, version strin
 		ResolverSettings: confmap.ResolverSettings{
 			URIs:       []string{getConfig(l)},
 			Providers:  mapProvider,
-			Converters: []confmap.Converter{expandconverter.New(), disablequeuedretryconverter.New()},
+			Converters: []confmap.Converter{expandconverter.New(), disablequeuedretryconverter.New(), decoupleafterbatchconverter.New()},
 		},
 	}
 	cfgProvider, err := otelcol.NewConfigProvider(cfgSet)

collector/internal/confmap/converter/decoupleafterbatchconverter/README.md

@@ -0,0 +1,11 @@
+# DecoupleAfterBatch Converter
+
+The `DecoupleAfterBatch` converter automatically modifies the collector's configuration for the Lambda distribution. Its purpose is to ensure that a decouple processor is always present after a batch processor in a pipeline, in order to prevent potential data loss due to the Lambda environment being frozen.
+
+## Behavior
+
+The converter scans the collector's configuration and makes the following adjustments:
+
+1. If a pipeline contains a batch processor with no decouple processor defined after it, the converter will automatically add a decouple processor to the pipeline immediately after the batch processor.
+
+2. If a pipeline contains a batch processor with a decouple processor already defined after it or there is no batch processor defined, the converter will not make any changes to the pipeline configuration.

collector/internal/confmap/converter/decoupleafterbatchconverter/converter.go

@@ -0,0 +1,118 @@
+// Copyright The OpenTelemetry Authors
+//
+// 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.
+
+// The decoupleafterbatchconverter implements the Converter for mutating Collector
+// configurations to ensure the decouple processor is placed after the batch processor.
+// This is logically implemented by appending the decouple processor to the end of
+// processor chains where a batch processor is found unless another decouple processor
+// was seen.
+package decoupleafterbatchconverter
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"go.opentelemetry.io/collector/confmap"
+)
+
+const (
+	serviceKey        = "service"
+	pipelinesKey      = "pipelines"
+	processorsKey     = "processors"
+	batchProcessor    = "batch"
+	decoupleProcessor = "decouple"
+)
+
+type converter struct{}
+
+// New returns a confmap.Converter that ensures the decoupleprocessor is placed first in the pipeline.
+func New() confmap.Converter {
+	return &converter{}
+}
+
+func (c converter) Convert(_ context.Context, conf *confmap.Conf) error {
+	serviceVal := conf.Get(serviceKey)
+	service, ok := serviceVal.(map[string]interface{})
+	if !ok {
+		return nil
+	}
+
+	pipelinesVal, ok := service[pipelinesKey]
+	if !ok {
+		return nil
+	}
+
+	pipelines, ok := pipelinesVal.(map[string]interface{})
+	if !ok {
+		return nil
+	}
+
+	// accumulates updates over the pipelines and applies them
+	// once all pipeline configs are processed
+	updates := make(map[string]interface{})
+	for telemetryType, pipelineVal := range pipelines {
+		pipeline, ok := pipelineVal.(map[string]interface{})
+		if !ok {
+			continue
+		}
+
+		processorsVal, ok := pipeline[processorsKey]
+		if !ok {
+			continue
+		}
+
+		processors, ok := processorsVal.([]interface{})
+		if !ok {
+			continue
+		}
+
+		// accumulate config updates
+		if shouldAppendDecouple(processors) {
+			processors = append(processors, decoupleProcessor)
+			updates[fmt.Sprintf("%s::%s::%s::%s", serviceKey, pipelinesKey, telemetryType, processorsKey)] = processors
+			break
+		}
+
+	}
+
+	// apply all updates
+	if len(updates) > 0 {
+		if err := conf.Merge(confmap.NewFromStringMap(updates)); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// The shouldAppendDecouple is the filter predicate for the Convert function action. It tells whether
+// (bool) there was a decouple processor after the last
+// batch processor, which Convert uses to decide whether to append the decouple processor.
+func shouldAppendDecouple(processors []interface{}) bool {
+	var shouldAppendDecouple bool
+	for _, processorVal := range processors {
+		processor, ok := processorVal.(string)
+		if !ok {
+			continue
+		}
+		processorBaseName := strings.Split(processor, "/")[0]
+		if processorBaseName == batchProcessor {
+			shouldAppendDecouple = true
+		} else if processorBaseName == decoupleProcessor {
+			shouldAppendDecouple = false
+		}
+	}
+	return shouldAppendDecouple
+}

collector/internal/confmap/converter/decoupleafterbatchconverter/converter_test.go

@@ -0,0 +1,153 @@
+// Copyright The OpenTelemetry Authors
+//
+// 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 decoupleafterbatchconverter
+
+import (
+	"context"
+	"testing"
+
+	"go.opentelemetry.io/collector/confmap"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+func TestConvert(t *testing.T) {
+	// Since this really tests differences in input, it's easier to read cases
+	// without the repeated definition of other fields in the config.
+	baseConf := func(input []interface{}) *confmap.Conf {
+		return confmap.NewFromStringMap(map[string]interface{}{
+			"service": map[string]interface{}{
+				"pipelines": map[string]interface{}{
+					"traces": map[string]interface{}{
+						"processors": input,
+					},
+				},
+			},
+		})
+	}
+
+	testCases := []struct {
+		name     string
+		input    *confmap.Conf
+		expected *confmap.Conf
+		err      error
+	}{
+		// This test is first, because it illustrates the difference in making the rule that when
+		// batch is present the converter appends decouple processor to the end of chain versus
+		// the approach of this code which is to do this only when the last instance of batch
+		// is not followed by decouple processor.
+		{
+			name:     "batch then decouple in middle of chain",
+			input:    baseConf([]interface{}{"processor1", "batch", "decouple", "processor2"}),
+			expected: baseConf([]interface{}{"processor1", "batch", "decouple", "processor2"}),
+		},
+		{
+			name:     "no service",
+			input:    confmap.New(),
+			expected: confmap.New(),
+		},
+		{
+			name: "no pipelines",
+			input: confmap.NewFromStringMap(
+				map[string]interface{}{
+					"service": map[string]interface{}{
+						"extensions": map[string]interface{}{},
+					},
+				},
+			),
+			expected: confmap.NewFromStringMap(
+				map[string]interface{}{
+					"service": map[string]interface{}{
+						"extensions": map[string]interface{}{},
+					},
+				},
+			),
+		},
+		{
+			name: "no processors in chain",
+			input: confmap.NewFromStringMap(
+				map[string]interface{}{
+					"service": map[string]interface{}{
+						"extensions": map[string]interface{}{},
+						"pipelines": map[string]interface{}{
+							"traces": map[string]interface{}{},
+						},
+					},
+				},
+			),
+			expected: confmap.NewFromStringMap(map[string]interface{}{
+				"service": map[string]interface{}{
+					"extensions": map[string]interface{}{},
+					"pipelines": map[string]interface{}{
+						"traces": map[string]interface{}{},
+					},
+				},
+			},
+			),
+		},
+		{
+			name:     "batch processor in singleton chain",
+			input:    baseConf([]interface{}{"batch"}),
+			expected: baseConf([]interface{}{"batch", "decouple"}),
+		},
+		{
+			name:     "batch processor present twice",
+			input:    baseConf([]interface{}{"batch", "processor1", "batch"}),
+			expected: baseConf([]interface{}{"batch", "processor1", "batch", "decouple"}),
+		},
+
+		{
+			name:     "batch processor not present",
+			input:    baseConf([]interface{}{"processor1", "processor2"}),
+			expected: baseConf([]interface{}{"processor1", "processor2"}),
+		},
+		{
+			name:     "batch sandwiched between input no decouple",
+			input:    baseConf([]interface{}{"processor1", "batch", "processor2"}),
+			expected: baseConf([]interface{}{"processor1", "batch", "processor2", "decouple"}),
+		},
+
+		{
+			name:     "batch and decouple input already present in correct position",
+			input:    baseConf([]interface{}{"processor1", "batch", "processor2", "decouple"}),
+			expected: baseConf([]interface{}{"processor1", "batch", "processor2", "decouple"}),
+		},
+		{
+			name:     "decouple and batch",
+			input:    baseConf([]interface{}{"decouple", "batch"}),
+			expected: baseConf([]interface{}{"decouple", "batch", "decouple"}),
+		},
+		{
+			name:     "decouple then batch mixed with others in the pipelinefirst then batch somewhere",
+			input:    baseConf([]interface{}{"processor1", "decouple", "processor2", "batch", "processor3"}),
+			expected: baseConf([]interface{}{"processor1", "decouple", "processor2", "batch", "processor3", "decouple"}),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			conf := tc.input
+			expected := tc.expected
+
+			c := New()
+			err := c.Convert(context.Background(), conf)
+			if err != tc.err {
+				t.Errorf("unexpected error converting: %v", err)
+			}
+			if diff := cmp.Diff(expected.ToStringMap(), conf.ToStringMap()); diff != "" {
+				t.Errorf("Convert() mismatch: (-want +got):\n%s", diff)
+			}
+		})
+	}
+}

collector/internal/confmap/converter/disablequeuedretryconverter/converter.go

@@ -11,7 +11,6 @@
 // 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 disablequeuedretryconverter // import "github.com/open-telemetry/opentelemetry-lambda/collector/internal/confmap/converter/disablequeuedretryconverter"
 
 import (
@@ -58,6 +57,8 @@ var exporters = map[string]struct{}{
 type converter struct {
 }
 
+
+
 // New returns a confmap.Converter, that ensures queued retry is disabled for all configured exporters.
 func New() confmap.Converter {
 	return &converter{}

collector/lambdacomponents/go.mod

@@ -148,7 +148,7 @@ require (
 	go.opentelemetry.io/otel/sdk/metric v1.21.0 // indirect
 	go.opentelemetry.io/otel/trace v1.21.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
-	go.uber.org/zap v1.26.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/exp v0.0.0-20240103183307-be819d1f06fc // indirect
 	golang.org/x/net v0.20.0 // indirect
 	golang.org/x/sys v0.16.0 // indirect
@@ -157,7 +157,7 @@ require (
 	google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20231127180814-3a041ad873d4 // indirect
 	google.golang.org/grpc v1.60.1 // indirect
-	google.golang.org/protobuf v1.32.0 // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

collector/lambdacomponents/go.sum

@@ -555,8 +555,8 @@ github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
-github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stvp/go-udp-testing v0.0.0-20201019212854-469649b16807/go.mod h1:7jxmlfBCDBXRzr0eAQJ48XC1hBu1np4CS5+cHEYfwpc=
 github.com/tidwall/gjson v1.10.2 h1:APbLGOM0rrEkd8WBw9C24nllro4ajFuJu0Sc9hRz8Bo=
 github.com/tidwall/gjson v1.10.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
@@ -687,8 +687,8 @@ go.opentelemetry.io/proto/otlp v1.0.0/go.mod h1:Sy6pihPLfYHkr3NkUbEhGHFhINUSI/v8
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
-go.uber.org/zap v1.26.0 h1:sI7k6L95XOKS281NhVKOFCUNIvv9e0w4BF8N3u+tCRo=
-go.uber.org/zap v1.26.0/go.mod h1:dtElttAiwGvoJ/vj4IwHBS/gXsEu/pZ50mUIRWuG0so=
+go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
+go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
@@ -969,8 +969,8 @@ google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQ
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
-google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7I=
-google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=