add converter to update batch processor

when sapm exporter enables token_access_passthrough
use this converter to add include_metadata with the
right header key

internal/configconverter/testdata/include_metadata_on_sapm_token_passthrough/agent_config_w_include_metadata.yaml

@@ -0,0 +1,6 @@
+receivers:
+  sapm:
+    include_metadata: true
+
+processors:
+  batch:

internal/configconverter/testdata/include_metadata_on_sapm_token_passthrough/agent_config_wo_include_metadata.yaml

@@ -0,0 +1,9 @@
+receivers:
+  sapm:
+
+
+processors:
+  batch:
+    send_batch_size: 10000
+    timeout: 10s
+

internal/configconverter/testdata/include_metadata_on_sapm_token_passthrough/expected_agent_config_w_include_metadata.yaml

@@ -0,0 +1,9 @@
+receivers:
+  sapm:
+    include_metadata: true
+
+processors:
+  batch:
+    metadata_keys:
+    - X-SF-Token
+

internal/configconverter/update_batchproc_on_token_passthrough.go

@@ -0,0 +1,52 @@
+// 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 configconverter
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/collector/confmap"
+)
+
+func UpdateBatchProcOnTokenPassthrough(_ context.Context, in *confmap.Conf) error {
+	if in == nil {
+		return nil
+	}
+
+	out := map[string]any{}
+
+	// Check for sapm receiver with include_metadata set to true
+	if in.IsSet("receivers::sapm::include_metadata") {
+		if _, ok := in.Get("receivers::sapm::include_metadata").(bool); ok {
+			// Add metadata_keys to batch processor
+			switch batchProcessor := in.Get("processors::batch").(type) {
+			case nil:
+				out["processors::batch"] = map[string]any{
+					"metadata_keys": []interface{}{"X-SF-Token"},
+				}
+			case map[string]interface{}:
+				batchProcessor["metadata_keys"] = []any{"X-SF-Token"}
+				out["processors::batch"] = batchProcessor
+			default:
+				return fmt.Errorf("unexpected type for processors::batch: %T", batchProcessor)
+			}
+		}
+	}
+
+	// Merge the modified configuration back into the original Conf
+	modifiedConf := confmap.NewFromStringMap(out)
+	return in.Merge(modifiedConf)
+}

internal/configconverter/update_batchproc_on_token_passthrough_test.go

@@ -0,0 +1,54 @@
+// 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 configconverter
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/confmap/confmaptest"
+)
+
+func TestUpdateBatchProcOnTokenPassthrough(t *testing.T) {
+	cfgMap, err := confmaptest.LoadConf("testdata/include_metadata_on_sapm_token_passthrough/agent_config_w_include_metadata.yaml")
+	require.NotNil(t, cfgMap)
+	require.NoError(t, err)
+
+	expectedCfgMap, err := confmaptest.LoadConf("testdata/include_metadata_on_sapm_token_passthrough/expected_agent_config_w_include_metadata.yaml")
+	require.NoError(t, err)
+	require.NotNil(t, cfgMap)
+
+	err = UpdateBatchProcOnTokenPassthrough(context.Background(), cfgMap)
+	require.NoError(t, err)
+
+	assert.Equal(t, expectedCfgMap, cfgMap)
+}
+
+func TestNoTokenAccessPassthrough(t *testing.T) {
+	cfgMap, err := confmaptest.LoadConf("testdata/include_metadata_on_sapm_token_passthrough/agent_config_wo_include_metadata.yaml")
+	require.NotNil(t, cfgMap)
+	require.NoError(t, err)
+
+	expectedCfgMap, err := confmaptest.LoadConf("testdata/include_metadata_on_sapm_token_passthrough/agent_config_wo_include_metadata.yaml")
+	require.NoError(t, err)
+	require.NotNil(t, cfgMap)
+
+	err = UpdateBatchProcOnTokenPassthrough(context.Background(), cfgMap)
+	require.NoError(t, err)
+
+	assert.Equal(t, expectedCfgMap, cfgMap)
+}

internal/settings/settings.go

@@ -206,6 +206,7 @@ func (s *Settings) ConfMapConverterFactories() []confmap.ConverterFactory {
 	if !s.noConvertConfig {
 		confMapConverterFactories = append(
 			confMapConverterFactories,
+			configconverter.ConverterFactoryFromFunc(configconverter.UpdateBatchProcOnTokenPassthrough),
 			configconverter.ConverterFactoryFromFunc(configconverter.NormalizeGcp),
 			configconverter.ConverterFactoryFromFunc(configconverter.DisableKubeletUtilizationMetrics),
 			configconverter.ConverterFactoryFromFunc(configconverter.DisableExcessiveInternalMetrics),

internal/settings/settings_test.go

@@ -156,7 +156,7 @@ func TestNewSettingsConvertConfig(t *testing.T) {
 	require.Equal(t, []string(nil), settings.discoveryProperties)
 
 	require.Equal(t, []string{configPath, anotherConfigPath}, settings.ResolverURIs())
-	require.Equal(t, 6, len(settings.ConfMapConverterFactories()))
+	require.Equal(t, 7, len(settings.ConfMapConverterFactories()))
 	require.Equal(t, []string{"--feature-gates", "foo", "--feature-gates", "-bar"}, settings.ColCoreArgs())
 }