[extension/headers_setter] Add headers_setter extension.

.github/CODEOWNERS

@@ -73,6 +73,7 @@ extension/asapauthextension/                         @open-telemetry/collector-c
 extension/awsproxy/                                  @open-telemetry/collector-contrib-approvers @Aneurysm9 @mxiamxia
 extension/basicauthextension/                        @open-telemetry/collector-contrib-approvers @jpkrohling @svrakitin
 extension/bearertokenauthextension/                  @open-telemetry/collector-contrib-approvers @jpkrohling @pavankrish123
+extension/headerssetter/                             @open-telemetry/collector-contrib-approvers @jpkrohling
 extension/healthcheckextension/                      @open-telemetry/collector-contrib-approvers @jpkrohling
 extension/jaegerremotesampling/                      @open-telemetry/collector-contrib-approvers @jpkrohling
 extension/oauth2clientauthextension/                 @open-telemetry/collector-contrib-approvers @pavankrish123 @jpkrohling

cmd/configschema/go.mod

@@ -326,6 +326,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/basicauthextension v0.57.2 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/bearertokenauthextension v0.57.2 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/fluentbitextension v0.57.2 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/extension/headerssetter v0.0.0-00010101000000-000000000000 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/healthcheckextension v0.57.2 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/httpforwarder v0.57.2 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/oauth2clientauthextension v0.57.2 // indirect
@@ -726,6 +727,8 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/bear
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/fluentbitextension => ../../extension/fluentbitextension
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/headerssetter => ../../extension/headerssetter
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/healthcheckextension => ../../extension/healthcheckextension
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/httpforwarder => ../../extension/httpforwarder

extension/headerssetter/Makefile

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

extension/headerssetter/README.md

@@ -0,0 +1,81 @@
+# Headers Setter extension
+
+| Status        |                  |
+|---------------|------------------|
+| Stability     | [in development] |
+| Distributions | [contrib]        |
+
+The `headers_setter` extension implements `ClientAuthenticator` and is used to
+set requests headers in `gRPC` / `HTTP` exporters with values provided via
+extension configurations or requests metadata (context).
+
+Use cases include but are not limited to enabling multi-tenancy for observability
+backends such as [Tempo], [Mimir], [Loki] and others by setting the `X-Scope-OrgID`
+header to the value extracted from the context.
+
+## Configuration
+
+The following settings are required:
+
+- `headers`: a list of header configuration objects that specify headers and 
+   their value sources. Each configuration object has the following properties:
+    - `key`: the header name
+    - `value`: the header value is looked up from the `value` property of the
+       extension configuration
+    - `from_context`: the header value is looked up from the request metadata,
+       such as HTTP headers, using the property value as the key (likely a header name)
+
+The `value` and `from_context` properties are mutually exclusive.
+
+
+#### Configuration Example
+
+```yaml
+extensions:
+  headers_setter:
+    headers:
+      - key: X-Scope-OrgID
+        from_context: tenant_id
+      - key: User-ID
+        value: user_id
+
+receivers:
+  otlp:
+    protocols:
+      http:
+        include_metadata: true
+
+processors:
+  nop:
+
+exporters:
+  loki:
+    labels:
+      resource:
+        container_id: ""
+        container_name: ""
+    endpoint: https://localhost:<port>/loki/api/v1/push
+    auth:
+      authenticator: headers_setter
+
+service:
+  extensions: [ headers_setter ]
+  pipelines:
+    traces:
+      receivers: [ otlp ]
+      processors: [ nop ]
+      exporters: [ loki ]
+```
+
+## Limitations
+
+At the moment, it is not possible to use the `from_context` option to ge the
+header value if Collector's pipeline contains the batch processor. See [#4544].
+
+
+[alpha]: https://github.com/open-telemetry/opentelemetry-collector#alpha
+[contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
+[Mimir]: https://grafana.com/oss/mimir/
+[Tempo]: https://grafana.com/oss/tempo/
+[Loki]: https://grafana.com/oss/loki/
+[#4544]: https://github.com/open-telemetry/opentelemetry-collector/issues/4544

extension/headerssetter/config.go

@@ -0,0 +1,58 @@
+// 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 headerssetter // import "github.com/open-telemetry/opentelemetry-collector-contrib/extension/headerssetter"
+
+import (
+	"fmt"
+
+	"go.opentelemetry.io/collector/config"
+)
+
+var (
+	errMissingHeader        = fmt.Errorf("missing header name")
+	errMissingHeadersConfig = fmt.Errorf("missing headers configuration")
+	errMissingSource        = fmt.Errorf("missing header source, must be 'from_context' or 'value'")
+	errConflictingSources   = fmt.Errorf("invalid header source, must either 'from_context' or 'value'")
+)
+
+type Config struct {
+	config.ExtensionSettings `mapstructure:",squash"`
+	HeadersConfig            []HeaderConfig `mapstructure:"headers"`
+}
+
+type HeaderConfig struct {
+	Key         *string `mapstructure:"key"`
+	Value       *string `mapstructure:"value"`
+	FromContext *string `mapstructure:"from_context"`
+}
+
+// Validate checks if the extension configuration is valid
+func (cfg *Config) Validate() error {
+	if cfg.HeadersConfig == nil || len(cfg.HeadersConfig) == 0 {
+		return errMissingHeadersConfig
+	}
+	for _, header := range cfg.HeadersConfig {
+		if header.Key == nil || *header.Key == "" {
+			return errMissingHeader
+		}
+		if header.FromContext == nil && header.Value == nil {
+			return errMissingSource
+		}
+		if header.FromContext != nil && header.Value != nil {
+			return errConflictingSources
+		}
+	}
+	return nil
+}

extension/headerssetter/config_test.go

@@ -0,0 +1,138 @@
+// 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 headerssetter
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/service/servicetest"
+)
+
+func TestLoadConfig(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	assert.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Extensions[typeStr] = factory
+
+	cfg, err := servicetest.LoadConfigAndValidate(
+		filepath.Join("testdata", "config.yaml"),
+		factories,
+	)
+	require.Nil(t, err)
+	require.NotNil(t, cfg)
+
+	ext0 := cfg.Extensions[config.NewComponentID(typeStr)]
+
+	assert.Equal(t,
+		&Config{
+			ExtensionSettings: config.NewExtensionSettings(config.NewComponentID(typeStr)),
+			HeadersConfig: []HeaderConfig{
+				{
+					Key:         stringp("X-Scope-OrgID"),
+					FromContext: stringp("tenant_id"),
+					Value:       nil,
+				},
+				{
+					Key:         stringp("User-ID"),
+					FromContext: stringp("user_id"),
+					Value:       nil,
+				},
+			},
+		},
+		ext0)
+
+	assert.Equal(t, 1, len(cfg.Service.Extensions))
+	assert.Equal(t, config.NewComponentID(typeStr), cfg.Service.Extensions[0])
+}
+
+func TestValidateConfig(t *testing.T) {
+	tests := []struct {
+		name        string
+		header      []HeaderConfig
+		expectedErr error
+	}{
+		{
+			"header value from config property",
+			[]HeaderConfig{
+				{
+					Key:   stringp("name"),
+					Value: stringp("from config"),
+				},
+			},
+			nil,
+		},
+		{
+			"header value from context",
+			[]HeaderConfig{
+				{
+					Key:         stringp("name"),
+					FromContext: stringp("from config"),
+				},
+			},
+			nil,
+		},
+		{
+			"missing header name for from value",
+			[]HeaderConfig{
+				{Value: stringp("test")},
+			},
+			errMissingHeader,
+		},
+		{
+			"missing header name for from context",
+			[]HeaderConfig{
+				{FromContext: stringp("test")},
+			},
+			errMissingHeader,
+		},
+		{
+			"header value from context and value",
+			[]HeaderConfig{
+				{
+					Key:         stringp("name"),
+					Value:       stringp("from config"),
+					FromContext: stringp("from context"),
+				},
+			},
+			errConflictingSources,
+		},
+		{
+			"header value source is missing",
+			[]HeaderConfig{
+				{
+					Key: stringp("name"),
+				},
+			},
+			errMissingSource,
+		},
+		{
+			"headers configuration is missing",
+			nil,
+			errMissingHeadersConfig,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := Config{HeadersConfig: tt.header}
+			require.ErrorIs(t, cfg.Validate(), tt.expectedErr)
+		})
+	}
+}