fix: Updating credentials file to not use endpoint_url parameter (#10841)

config/aws/credentials.go

@@ -9,6 +9,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/sts"
 )
 
+// The endpoint_url supplied here is used for specific AWS service (Cloudwatch / Timestream / etc.)
 type CredentialConfig struct {
 	Region               string `toml:"region"`
 	AccessKey            string `toml:"access_key"`
@@ -24,27 +25,16 @@ type CredentialConfig struct {
 
 func (c *CredentialConfig) Credentials() (awsV2.Config, error) {
 	if c.RoleARN != "" {
-		return c.assumeCredentials()
+		return c.configWithAssumeCredentials()
 	}
-	return c.rootCredentials()
+	return c.configWithRootCredentials()
 }
 
-func (c *CredentialConfig) rootCredentials() (awsV2.Config, error) {
+func (c *CredentialConfig) configWithRootCredentials() (awsV2.Config, error) {
 	options := []func(*configV2.LoadOptions) error{
 		configV2.WithRegion(c.Region),
 	}
 
-	if c.EndpointURL != "" {
-		resolver := awsV2.EndpointResolverFunc(func(service, region string) (awsV2.Endpoint, error) {
-			return awsV2.Endpoint{
-				URL:               c.EndpointURL,
-				HostnameImmutable: true,
-				Source:            awsV2.EndpointSourceCustom,
-			}, nil
-		})
-		options = append(options, configV2.WithEndpointResolver(resolver))
-	}
-
 	if c.Profile != "" {
 		options = append(options, configV2.WithSharedConfigProfile(c.Profile))
 	}
@@ -60,14 +50,15 @@ func (c *CredentialConfig) rootCredentials() (awsV2.Config, error) {
 	return configV2.LoadDefaultConfig(context.Background(), options...)
 }
 
-func (c *CredentialConfig) assumeCredentials() (awsV2.Config, error) {
-	rootCredentials, err := c.rootCredentials()
+func (c *CredentialConfig) configWithAssumeCredentials() (awsV2.Config, error) {
+	// To generate credentials using assumeRole, we need to create AWS STS client with the default AWS endpoint,
+	defaultConfig, err := c.configWithRootCredentials()
 	if err != nil {
 		return awsV2.Config{}, err
 	}
 
 	var provider awsV2.CredentialsProvider
-	stsService := sts.NewFromConfig(rootCredentials)
+	stsService := sts.NewFromConfig(defaultConfig)
 	if c.WebIdentityTokenFile != "" {
 		provider = stscredsV2.NewWebIdentityRoleProvider(stsService, c.RoleARN, stscredsV2.IdentityTokenFile(c.WebIdentityTokenFile), func(opts *stscredsV2.WebIdentityRoleOptions) {
 			if c.RoleSessionName != "" {
@@ -82,6 +73,6 @@ func (c *CredentialConfig) assumeCredentials() (awsV2.Config, error) {
 		})
 	}
 
-	rootCredentials.Credentials = awsV2.NewCredentialsCache(provider)
-	return rootCredentials, nil
+	defaultConfig.Credentials = awsV2.NewCredentialsCache(provider)
+	return defaultConfig, nil
 }

plugins/outputs/cloudwatch_logs/README.md

@@ -7,10 +7,9 @@ This plugin will send logs to Amazon CloudWatch.
 This plugin uses a credential chain for Authentication with the CloudWatch Logs
 API endpoint. In the following order the plugin will attempt to authenticate.
 
-1. Web identity provider credentials via STS if `role_arn` and
-   `web_identity_token_file` are specified
-1. Assumed credentials via STS if `role_arn` attribute is specified (source
-   credentials are evaluated from subsequent rules)
+1. Web identity provider credentials via STS if `role_arn` and `web_identity_token_file` are specified
+1. Assumed credentials via STS if `role_arn` attribute is specified (source credentials are evaluated from subsequent rules).
+The `endpoint_url` attribute is used only for Cloudwatch Logs service. When fetching credentials, STS global endpoint will be used.
 1. Explicit credentials from `access_key`, `secret_key`, and `token` attributes
 1. Shared profile from `profile` attribute
 1. [Environment Variables][1]

plugins/outputs/cloudwatch_logs/cloudwatch_logs.go

@@ -9,6 +9,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
 	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types"
 
@@ -131,10 +133,31 @@ func (c *CloudWatchLogs) Connect() error {
 	var logGroupsOutput = &cloudwatchlogs.DescribeLogGroupsOutput{NextToken: &dummyToken}
 	var err error
 
-	cfg, err := c.CredentialConfig.Credentials()
+	awsCreds, awsErr := c.CredentialConfig.Credentials()
+	if awsErr != nil {
+		return awsErr
+	}
+
+	cfg, err := config.LoadDefaultConfig(context.TODO())
 	if err != nil {
 		return err
 	}
+	if c.CredentialConfig.EndpointURL != "" && c.CredentialConfig.Region != "" {
+		customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
+			return aws.Endpoint{
+				PartitionID:   "aws",
+				URL:           c.CredentialConfig.EndpointURL,
+				SigningRegion: c.CredentialConfig.Region,
+			}, nil
+		})
+
+		cfg, err = config.LoadDefaultConfig(context.TODO(), config.WithEndpointResolverWithOptions(customResolver))
+		if err != nil {
+			return err
+		}
+	}
+
+	cfg.Credentials = awsCreds.Credentials
 	c.svc = cloudwatchlogs.NewFromConfig(cfg)
 
 	//Find log group with name 'c.LogGroup'

plugins/outputs/cloudwatch_logs/cloudwatch_logs_test.go

@@ -207,6 +207,24 @@ func TestInit(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "valid config with EndpointURL",
+			plugin: &CloudWatchLogs{
+				CredentialConfig: internalaws.CredentialConfig{
+					Region:      "eu-central-1",
+					AccessKey:   "dummy",
+					SecretKey:   "dummy",
+					EndpointURL: "https://test.com",
+				},
+				LogGroup:     "TestLogGroup",
+				LogStream:    "tag:source",
+				LDMetricName: "docker_log",
+				LDSource:     "tag:location",
+				Log: testutil.Logger{
+					Name: "outputs.cloudwatch_logs",
+				},
+			},
+		},
 	}
 
 	for _, tt := range tests {

plugins/outputs/timestream/README.md

@@ -2,6 +2,19 @@
 
 The Timestream output plugin writes metrics to the [Amazon Timestream] service.
 
+## Authentication
+
+This plugin uses a credential chain for Authentication with Timestream
+API endpoint. In the following order the plugin will attempt to authenticate.
+
+1. Web identity provider credentials via STS if `role_arn` and `web_identity_token_file` are specified
+1. [Assumed credentials via STS] if `role_arn` attribute is specified (source credentials are evaluated from subsequent rules). The `endpoint_url` attribute is used only for Timestream service. When fetching credentials, STS global endpoint will be used.
+1. Explicit credentials from `access_key`, `secret_key`, and `token` attributes
+1. Shared profile from `profile` attribute
+1. [Environment Variables]
+1. [Shared Credentials]
+1. [EC2 Instance Profile]
+
 ## Configuration
 
 ```toml @sample.conf
@@ -169,3 +182,7 @@ go test -v ./plugins/outputs/timestream/...
 ```
 
 [Amazon Timestream]: https://aws.amazon.com/timestream/
+[Assumed credentials via STS]: https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/credentials/stscreds
+[Environment Variables]: https://github.com/aws/aws-sdk-go/wiki/configuring-sdk#environment-variables
+[Shared Credentials]: https://github.com/aws/aws-sdk-go/wiki/configuring-sdk#shared-credentials-file
+[EC2 Instance Profile]: http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html

plugins/outputs/timestream/timestream.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/timestreamwrite"
 	"github.com/aws/aws-sdk-go-v2/service/timestreamwrite/types"
 	"github.com/aws/smithy-go"
@@ -69,11 +70,45 @@ const MaxWriteRoutinesDefault = 1
 
 // WriteFactory function provides a way to mock the client instantiation for testing purposes.
 var WriteFactory = func(credentialConfig *internalaws.CredentialConfig) (WriteClient, error) {
-	cfg, err := credentialConfig.Credentials()
-	if err != nil {
-		return &timestreamwrite.Client{}, err
+
+	awsCreds, awsErr := credentialConfig.Credentials()
+	if awsErr != nil {
+		panic("Unable to load credentials config " + awsErr.Error())
 	}
-	return timestreamwrite.NewFromConfig(cfg), nil
+
+	cfg, cfgErr := config.LoadDefaultConfig(context.TODO())
+	if cfgErr != nil {
+		panic("Unable to load SDK config for Timestream " + cfgErr.Error())
+	}
+
+	if credentialConfig.EndpointURL != "" && credentialConfig.Region != "" {
+		customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
+			return aws.Endpoint{
+				PartitionID:   "aws",
+				URL:           credentialConfig.EndpointURL,
+				SigningRegion: credentialConfig.Region,
+			}, nil
+		})
+
+		cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithEndpointResolverWithOptions(customResolver))
+
+		if err != nil {
+			panic("unable to load SDK config for Timestream " + err.Error())
+		}
+
+		cfg.Credentials = awsCreds.Credentials
+
+		return timestreamwrite.NewFromConfig(cfg, func(o *timestreamwrite.Options) {
+			o.Region = credentialConfig.Region
+			o.EndpointDiscovery.EnableEndpointDiscovery = aws.EndpointDiscoveryDisabled
+		}), nil
+	}
+
+	cfg.Credentials = awsCreds.Credentials
+
+	return timestreamwrite.NewFromConfig(cfg, func(o *timestreamwrite.Options) {
+		o.Region = credentialConfig.Region
+	}), nil
 }
 
 func (*Timestream) SampleConfig() string {

plugins/outputs/timestream/timestream_test.go

@@ -693,6 +693,22 @@ func TestTransformMetricsUnsupportedFieldsAreSkipped(t *testing.T) {
 		[]*timestreamwrite.WriteRecordsInput{expectedResultMultiTable})
 }
 
+func TestCustomEndpoint(t *testing.T) {
+	customEndpoint := "http://test.custom.endpoint.com"
+	plugin := Timestream{
+		MappingMode:      MappingModeMultiTable,
+		DatabaseName:     tsDbName,
+		Log:              testutil.Logger{},
+		CredentialConfig: internalaws.CredentialConfig{EndpointURL: customEndpoint},
+	}
+
+	// validate config correctness
+	err := plugin.Connect()
+	require.Nil(t, err, "Invalid configuration")
+	// Check customURL is used
+	require.Equal(t, plugin.EndpointURL, customEndpoint)
+}
+
 func comparisonTest(t *testing.T,
 	mappingMode string,
 	telegrafMetrics []telegraf.Metric,