Merge branch 'main' into chore/flag-autoscaling-v2beta2

.chloggen/feature-oauth2clientauth-read-from-file.yaml

@@ -0,0 +1,29 @@
+# Use this changelog template to create an entry for release notes.
+
+# 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: oauth2clientauthextension
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Enable dynamically reading ClientID and ClientSecret from files
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [26117]
+
+# (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: |
+  - Read the client ID and/or secret from a file by specifying the file path to the ClientIDFile (`client_id_file`) and ClientSecretFile (`client_secret_file`) fields respectively.
+  - The file is read every time the client issues a new token. This means that the corresponding value can change dynamically during the execution by modifying the file contents.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user, api]

.chloggen/pkg-stanza-flush.yaml

@@ -0,0 +1,30 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: breaking
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: pkg/stanza
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Remove Flusher from tokenize.SplitterConfig
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [26517]
+
+# (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: Removes the following in favor of flush.WithPeriod
+ - tokenize.DefaultFlushPeriod
+ - tokenize.FlusherConfig
+ - tokenize.NewFlusherConfig
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [api]

extension/oauth2clientauthextension/README.md

@@ -74,7 +74,13 @@ Following are the configuration fields
 
 - [**token_url**](https://datatracker.ietf.org/doc/html/rfc6749#section-3.2) - The resource server's token endpoint URLs.
 - [**client_id**](https://datatracker.ietf.org/doc/html/rfc6749#section-2.2) - The client identifier issued to the client.
+- **client_id_file** - The file path to retrieve the client identifier issued to the client.
+  The extension reads this file and updates the client ID used whenever it needs to issue a new token. This enables dynamically changing the client credentials by modifying the file contents when, for example, they need to rotate. <!-- Intended whitespace for compact new line -->  
+  This setting takes precedence over `client_id`.
 - [**client_secret**](https://datatracker.ietf.org/doc/html/rfc6749#section-2.3.1) - The secret string associated with above identifier.
+- **client_secret_file** - The file path to retrieve the secret string associated with above identifier.
+  The extension reads this file and updates the client secret used whenever it needs to issue a new token. This enables dynamically changing the client credentials by modifying the file contents when, for example, they need to rotate. <!-- Intended whitespace for compact new line -->  
+  This setting takes precedence over `client_secret`.
 - [**endpoint_params**](https://github.com/golang/oauth2/blob/master/clientcredentials/clientcredentials.go#L44) - Additional parameters that are sent to the token endpoint.
 - [**scopes**](https://datatracker.ietf.org/doc/html/rfc6749#section-3.3) - **Optional** optional requested permissions associated for the client.
 - [**timeout**](https://golang.org/src/net/http/client.go#L90) -  **Optional** specifies the timeout on the underlying client to authorization server for fetching the tokens (initial and while refreshing).

extension/oauth2clientauthextension/clientcredentialsconfig.go

@@ -0,0 +1,102 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package oauth2clientauthextension // import "github.com/open-telemetry/opentelemetry-collector-contrib/extension/oauth2clientauthextension"
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+
+	"go.uber.org/multierr"
+	"golang.org/x/oauth2"
+	"golang.org/x/oauth2/clientcredentials"
+)
+
+// clientCredentialsConfig is a clientcredentials.Config wrapper to allow
+// values read from files in the ClientID and ClientSecret fields.
+//
+// Values from files can be retrieved by populating the ClientIDFile or
+// the ClientSecretFile fields with the path to the file.
+//
+// Priority: File > Raw value
+//
+// Example - Retrieve secret from file:
+//
+//	cfg := clientCredentialsConfig{
+//		Config: clientcredentials.Config{
+//			ClientID:     "clientId",
+//			...
+//		},
+//		ClientSecretFile: "/path/to/client/secret",
+//	}
+type clientCredentialsConfig struct {
+	clientcredentials.Config
+
+	ClientIDFile     string
+	ClientSecretFile string
+}
+
+type clientCredentialsTokenSource struct {
+	ctx    context.Context
+	config *clientCredentialsConfig
+}
+
+// clientCredentialsTokenSource implements TokenSource
+var _ oauth2.TokenSource = (*clientCredentialsTokenSource)(nil)
+
+func readCredentialsFile(path string) (string, error) {
+	f, err := os.ReadFile(path)
+	if err != nil {
+		return "", fmt.Errorf("failed to read credentials file %q: %w", path, err)
+	}
+
+	credential := strings.TrimSpace(string(f))
+	if credential == "" {
+		return "", fmt.Errorf("empty credentials file %q", path)
+	}
+	return credential, nil
+}
+
+func getActualValue(value, filepath string) (string, error) {
+	if len(filepath) > 0 {
+		return readCredentialsFile(filepath)
+	}
+
+	return value, nil
+}
+
+// createConfig creates a proper clientcredentials.Config with values retrieved
+// from files, if the user has specified '*_file' values
+func (c *clientCredentialsConfig) createConfig() (*clientcredentials.Config, error) {
+	clientID, err := getActualValue(c.ClientID, c.ClientIDFile)
+	if err != nil {
+		return nil, multierr.Combine(errNoClientIDProvided, err)
+	}
+
+	clientSecret, err := getActualValue(c.ClientSecret, c.ClientSecretFile)
+	if err != nil {
+		return nil, multierr.Combine(errNoClientSecretProvided, err)
+	}
+
+	return &clientcredentials.Config{
+		ClientID:       clientID,
+		ClientSecret:   clientSecret,
+		TokenURL:       c.TokenURL,
+		Scopes:         c.Scopes,
+		EndpointParams: c.EndpointParams,
+	}, nil
+}
+
+func (c *clientCredentialsConfig) TokenSource(ctx context.Context) oauth2.TokenSource {
+	return oauth2.ReuseTokenSource(nil, clientCredentialsTokenSource{ctx: ctx, config: c})
+}
+
+func (ts clientCredentialsTokenSource) Token() (*oauth2.Token, error) {
+	cfg, err := ts.config.createConfig()
+	if err != nil {
+		return nil, err
+	}
+	return cfg.TokenSource(ts.ctx).Token()
+}

extension/oauth2clientauthextension/config.go

@@ -26,10 +26,16 @@ type Config struct {
 	// See https://datatracker.ietf.org/doc/html/rfc6749#section-2.2
 	ClientID string `mapstructure:"client_id"`
 
+	// ClientIDFile is the file path to read the application's ID from.
+	ClientIDFile string `mapstructure:"client_id_file"`
+
 	// ClientSecret is the application's secret.
 	// See https://datatracker.ietf.org/doc/html/rfc6749#section-2.3.1
 	ClientSecret configopaque.String `mapstructure:"client_secret"`
 
+	// ClientSecretFile is the file pathg to read the application's secret from.
+	ClientSecretFile string `mapstructure:"client_secret_file"`
+
 	// EndpointParams specifies additional parameters for requests to the token endpoint.
 	EndpointParams url.Values `mapstructure:"endpoint_params"`
 
@@ -54,10 +60,10 @@ var _ component.Config = (*Config)(nil)
 
 // Validate checks if the extension configuration is valid
 func (cfg *Config) Validate() error {
-	if cfg.ClientID == "" {
+	if cfg.ClientID == "" && cfg.ClientIDFile == "" {
 		return errNoClientIDProvided
 	}
-	if cfg.ClientSecret == "" {
+	if cfg.ClientSecret == "" && cfg.ClientSecretFile == "" {
 		return errNoClientSecretProvided
 	}
 	if cfg.TokenURL == "" {

extension/oauth2clientauthextension/extension.go

@@ -19,7 +19,7 @@ import (
 // clientAuthenticator provides implementation for providing client authentication using OAuth2 client credentials
 // workflow for both gRPC and HTTP clients.
 type clientAuthenticator struct {
-	clientCredentials *clientcredentials.Config
+	clientCredentials *clientCredentialsConfig
 	logger            *zap.Logger
 	client            *http.Client
 }
@@ -36,10 +36,10 @@ var _ oauth2.TokenSource = (*errorWrappingTokenSource)(nil)
 var errFailedToGetSecurityToken = fmt.Errorf("failed to get security token from token endpoint")
 
 func newClientAuthenticator(cfg *Config, logger *zap.Logger) (*clientAuthenticator, error) {
-	if cfg.ClientID == "" {
+	if cfg.ClientID == "" && cfg.ClientIDFile == "" {
 		return nil, errNoClientIDProvided
 	}
-	if cfg.ClientSecret == "" {
+	if cfg.ClientSecret == "" && cfg.ClientSecretFile == "" {
 		return nil, errNoClientSecretProvided
 	}
 	if cfg.TokenURL == "" {
@@ -55,12 +55,16 @@ func newClientAuthenticator(cfg *Config, logger *zap.Logger) (*clientAuthenticat
 	transport.TLSClientConfig = tlsCfg
 
 	return &clientAuthenticator{
-		clientCredentials: &clientcredentials.Config{
-			ClientID:       cfg.ClientID,
-			ClientSecret:   string(cfg.ClientSecret),
-			TokenURL:       cfg.TokenURL,
-			Scopes:         cfg.Scopes,
-			EndpointParams: cfg.EndpointParams,
+		clientCredentials: &clientCredentialsConfig{
+			Config: clientcredentials.Config{
+				ClientID:       cfg.ClientID,
+				ClientSecret:   string(cfg.ClientSecret),
+				TokenURL:       cfg.TokenURL,
+				Scopes:         cfg.Scopes,
+				EndpointParams: cfg.EndpointParams,
+			},
+			ClientIDFile:     cfg.ClientIDFile,
+			ClientSecretFile: cfg.ClientSecretFile,
 		},
 		logger: logger,
 		client: &http.Client{

extension/oauth2clientauthextension/extension_test.go

@@ -16,6 +16,7 @@ import (
 	"go.opentelemetry.io/collector/config/configtls"
 	"go.uber.org/zap"
 	"golang.org/x/oauth2"
+	"golang.org/x/oauth2/clientcredentials"
 	grpcOAuth "google.golang.org/grpc/credentials/oauth"
 )
 
@@ -134,6 +135,98 @@ func TestOAuthClientSettings(t *testing.T) {
 	}
 }
 
+func TestOAuthClientSettingsCredsConfig(t *testing.T) {
+	// test files for TLS testing
+	var (
+		testCredsFile        = "testdata/test-cred.txt"
+		testCredsEmptyFile   = "testdata/test-cred-empty.txt"
+		testCredsMissingFile = "testdata/test-cred-missing.txt"
+	)
+
+	tests := []struct {
+		name                 string
+		settings             *Config
+		expectedClientConfig *clientcredentials.Config
+		shouldError          bool
+		expectedError        error
+	}{
+		{
+			name: "client_id_file",
+			settings: &Config{
+				ClientIDFile: testCredsFile,
+				ClientSecret: "testsecret",
+				TokenURL:     "https://example.com/v1/token",
+				Scopes:       []string{"resource.read"},
+			},
+			expectedClientConfig: &clientcredentials.Config{
+				ClientID:     "testcreds",
+				ClientSecret: "testsecret",
+			},
+			shouldError:   false,
+			expectedError: nil,
+		},
+		{
+			name: "client_secret_file",
+			settings: &Config{
+				ClientID:         "testclientid",
+				ClientSecretFile: testCredsFile,
+				TokenURL:         "https://example.com/v1/token",
+				Scopes:           []string{"resource.read"},
+			},
+			expectedClientConfig: &clientcredentials.Config{
+				ClientID:     "testclientid",
+				ClientSecret: "testcreds",
+			},
+			shouldError:   false,
+			expectedError: nil,
+		},
+		{
+			name: "empty_client_creds_file",
+			settings: &Config{
+				ClientIDFile: testCredsEmptyFile,
+				ClientSecret: "testsecret",
+				TokenURL:     "https://example.com/v1/token",
+				Scopes:       []string{"resource.read"},
+			},
+			shouldError:   true,
+			expectedError: errNoClientIDProvided,
+		},
+		{
+			name: "missing_client_creds_file",
+			settings: &Config{
+				ClientID:         "testclientid",
+				ClientSecretFile: testCredsMissingFile,
+				TokenURL:         "https://example.com/v1/token",
+				Scopes:           []string{"resource.read"},
+			},
+			shouldError:   true,
+			expectedError: errNoClientSecretProvided,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			rc, _ := newClientAuthenticator(test.settings, zap.NewNop())
+			cfg, err := rc.clientCredentials.createConfig()
+			if test.shouldError {
+				assert.NotNil(t, err)
+				assert.ErrorAs(t, err, &test.expectedError)
+				return
+			}
+			assert.NoError(t, err)
+			assert.Equal(t, test.expectedClientConfig.ClientID, cfg.ClientID)
+			assert.Equal(t, test.expectedClientConfig.ClientSecret, cfg.ClientSecret)
+
+			// test tls settings
+			transport := rc.client.Transport.(*http.Transport)
+			tlsClientConfig := transport.TLSClientConfig
+			tlsTestSettingConfig, err := test.settings.TLSSetting.LoadTLSConfig()
+			assert.Nil(t, err)
+			assert.Equal(t, tlsClientConfig.Certificates, tlsTestSettingConfig.Certificates)
+		})
+	}
+}
+
 type testRoundTripper struct {
 	testString string
 }

extension/oauth2clientauthextension/testdata/test-cred.txt

@@ -0,0 +1 @@
+testcreds

pkg/stanza/fileconsumer/config.go

@@ -28,6 +28,7 @@ const (
 	defaultMaxLogSize         = 1024 * 1024
 	defaultMaxConcurrentFiles = 1024
 	defaultEncoding           = "utf-8"
+	defaultFlushPeriod        = 500 * time.Millisecond
 )
 
 var allowFileDeletion = featuregate.GlobalRegistry().MustRegister(
@@ -79,6 +80,7 @@ type Config struct {
 	Splitter                tokenize.SplitterConfig `mapstructure:",squash,omitempty"`
 	TrimConfig              trim.Config             `mapstructure:",squash,omitempty"`
 	Encoding                string                  `mapstructure:"encoding,omitempty"`
+	FlushPeriod             time.Duration           `mapstructure:"force_flush_period,omitempty"`
 	Header                  *HeaderConfig           `mapstructure:"header,omitempty"`
 }
 
@@ -99,7 +101,7 @@ func (c Config) Build(logger *zap.SugaredLogger, emit emit.Callback) (*Manager,
 	}
 
 	// Ensure that splitter is buildable
-	factory := splitter.NewMultilineFactory(c.Splitter, enc, int(c.MaxLogSize), c.TrimConfig.Func())
+	factory := splitter.NewMultilineFactory(c.Splitter, enc, int(c.MaxLogSize), c.TrimConfig.Func(), c.FlushPeriod)
 	if _, err := factory.Build(); err != nil {
 		return nil, err
 	}
@@ -118,7 +120,7 @@ func (c Config) BuildWithSplitFunc(logger *zap.SugaredLogger, emit emit.Callback
 	}
 
 	// Ensure that splitter is buildable
-	factory := splitter.NewCustomFactory(c.Splitter.Flusher, splitFunc)
+	factory := splitter.NewCustomFactory(splitFunc, c.FlushPeriod)
 	if _, err := factory.Build(); err != nil {
 		return nil, err
 	}