Merge branch 'main' into 3318-RavenDB-state-store-new

Author: nmalocicvega

common/authentication/aws/aws.go

@@ -26,17 +26,6 @@ type EnvironmentSettings struct {
 	Metadata map[string]string
 }
 
-// TODO: Delete in Dapr 1.17 so we can move all IAM fields to use the defaults of:
-// accessKey and secretKey and region as noted in the docs, and Options struct above.
-type DeprecatedKafkaIAM struct {
-	Region         string `json:"awsRegion" mapstructure:"awsRegion"`
-	AccessKey      string `json:"awsAccessKey" mapstructure:"awsAccessKey"`
-	SecretKey      string `json:"awsSecretKey" mapstructure:"awsSecretKey"`
-	SessionToken   string `json:"awsSessionToken" mapstructure:"awsSessionToken"`
-	IamRoleArn     string `json:"awsIamRoleArn" mapstructure:"awsIamRoleArn"`
-	StsSessionName string `json:"awsStsSessionName" mapstructure:"awsStsSessionName"`
-}
-
 type Options struct {
 	Logger     logger.Logger
 	Properties map[string]string
@@ -89,7 +78,6 @@ type Provider interface {
 	ParameterStore() *ParameterStoreClients
 	Kinesis() *KinesisClients
 	Ses() *SesClients
-	Kafka(KafkaOptions) (*KafkaClients, error)
 
 	// Postgres is an outlier to the others in the sense that we can update only it's config,
 	// as we use a max connection time of 8 minutes.
@@ -115,14 +103,3 @@ func NewEnvironmentSettings(md map[string]string) (EnvironmentSettings, error) {
 
 	return es, nil
 }
-
-// Coalesce is a helper function to return the first non-empty string from the inputs
-// This helps us to migrate away from the deprecated duplicate aws auth profile metadata fields in Dapr 1.17.
-func Coalesce(values ...string) string {
-	for _, v := range values {
-		if v != "" {
-			return v
-		}
-	}
-	return ""
-}

common/authentication/aws/client.go

@@ -16,13 +16,8 @@ package aws
 import (
 	"context"
 	"errors"
-	"fmt"
 	"sync"
-	"time"
 
-	"github.com/IBM/sarama"
-	"github.com/aws/aws-msk-iam-sasl-signer-go/signer"
-	aws2 "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/session"
@@ -56,7 +51,6 @@ type Clients struct {
 	ParameterStore *ParameterStoreClients
 	kinesis        *KinesisClients
 	ses            *SesClients
-	kafka          *KafkaClients
 }
 
 func newClients() *Clients {
@@ -85,14 +79,6 @@ func (c *Clients) refresh(session *session.Session) error {
 		c.kinesis.New(session)
 	case c.ses != nil:
 		c.ses.New(session)
-	case c.kafka != nil:
-		// Note: we pass in nil for token provider
-		// as there are no special fields for x509 auth for it.
-		// Only static auth passes it in.
-		err := c.kafka.New(session, nil)
-		if err != nil {
-			return fmt.Errorf("failed to refresh Kafka AWS IAM Config: %w", err)
-		}
 	}
 	return nil
 }
@@ -139,16 +125,6 @@ type SesClients struct {
 	Ses *ses.SES
 }
 
-type KafkaClients struct {
-	config          *sarama.Config
-	consumerGroup   *string
-	brokers         *[]string
-	maxMessageBytes *int
-
-	ConsumerGroup sarama.ConsumerGroup
-	Producer      sarama.SyncProducer
-}
-
 func (c *S3Clients) New(session *session.Session) {
 	refreshedS3 := s3.New(session, session.Config)
 	c.S3 = refreshedS3
@@ -232,120 +208,3 @@ func (c *KinesisClients) WorkerCfg(ctx context.Context, stream, consumer, mode s
 func (c *SesClients) New(session *session.Session) {
 	c.Ses = ses.New(session, session.Config)
 }
-
-type KafkaOptions struct {
-	Config          *sarama.Config
-	ConsumerGroup   string
-	Brokers         []string
-	MaxMessageBytes int
-}
-
-func initKafkaClients(opts KafkaOptions) *KafkaClients {
-	return &KafkaClients{
-		config:          opts.Config,
-		consumerGroup:   &opts.ConsumerGroup,
-		brokers:         &opts.Brokers,
-		maxMessageBytes: &opts.MaxMessageBytes,
-	}
-}
-
-func (c *KafkaClients) New(session *session.Session, tokenProvider *mskTokenProvider) error {
-	const timeout = 10 * time.Second
-	creds, err := session.Config.Credentials.Get()
-	if err != nil {
-		return fmt.Errorf("failed to get credentials from session: %w", err)
-	}
-
-	// fill in token provider common fields across x509 and static auth
-	if tokenProvider == nil {
-		tokenProvider = &mskTokenProvider{}
-	}
-	tokenProvider.generateTokenTimeout = timeout
-	tokenProvider.region = *session.Config.Region
-	tokenProvider.accessKey = creds.AccessKeyID
-	tokenProvider.secretKey = creds.SecretAccessKey
-	tokenProvider.sessionToken = creds.SessionToken
-
-	c.config.Net.SASL.Enable = true
-	c.config.Net.SASL.Mechanism = sarama.SASLTypeOAuth
-	c.config.Net.SASL.TokenProvider = tokenProvider
-
-	_, err = c.config.Net.SASL.TokenProvider.Token()
-	if err != nil {
-		return fmt.Errorf("error validating iam credentials %v", err)
-	}
-
-	consumerGroup, err := sarama.NewConsumerGroup(*c.brokers, *c.consumerGroup, c.config)
-	if err != nil {
-		return err
-	}
-	c.ConsumerGroup = consumerGroup
-
-	producer, err := c.getSyncProducer()
-	if err != nil {
-		return err
-	}
-	c.Producer = producer
-
-	return nil
-}
-
-// Kafka specific
-type mskTokenProvider struct {
-	generateTokenTimeout time.Duration
-	accessKey            string
-	secretKey            string
-	sessionToken         string
-	awsIamRoleArn        string
-	awsStsSessionName    string
-	region               string
-}
-
-func (m *mskTokenProvider) Token() (*sarama.AccessToken, error) {
-	// this function can't use the context passed on Init because that context would be cancelled right after Init
-	ctx, cancel := context.WithTimeout(context.Background(), m.generateTokenTimeout)
-	defer cancel()
-
-	switch {
-	// we must first check if we are using the assume role auth profile
-	case m.awsIamRoleArn != "" && m.awsStsSessionName != "":
-		token, _, err := signer.GenerateAuthTokenFromRole(ctx, m.region, m.awsIamRoleArn, m.awsStsSessionName)
-		return &sarama.AccessToken{Token: token}, err
-	case m.accessKey != "" && m.secretKey != "":
-		token, _, err := signer.GenerateAuthTokenFromCredentialsProvider(ctx, m.region, aws2.CredentialsProviderFunc(func(ctx context.Context) (aws2.Credentials, error) {
-			return aws2.Credentials{
-				AccessKeyID:     m.accessKey,
-				SecretAccessKey: m.secretKey,
-				SessionToken:    m.sessionToken,
-			}, nil
-		}))
-		return &sarama.AccessToken{Token: token}, err
-
-	default: // load default aws creds
-		token, _, err := signer.GenerateAuthToken(ctx, m.region)
-		return &sarama.AccessToken{Token: token}, err
-	}
-}
-
-func (c *KafkaClients) getSyncProducer() (sarama.SyncProducer, error) {
-	// Add SyncProducer specific properties to copy of base config
-	c.config.Producer.RequiredAcks = sarama.WaitForAll
-	c.config.Producer.Retry.Max = 5
-	c.config.Producer.Return.Successes = true
-
-	if *c.maxMessageBytes > 0 {
-		c.config.Producer.MaxMessageBytes = *c.maxMessageBytes
-	}
-
-	saramaClient, err := sarama.NewClient(*c.brokers, c.config)
-	if err != nil {
-		return nil, err
-	}
-
-	producer, err := sarama.NewSyncProducerFromClient(saramaClient)
-	if err != nil {
-		return nil, err
-	}
-
-	return producer, nil
-}

common/authentication/aws/static.go

@@ -15,7 +15,6 @@ package aws
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"strconv"
 	"sync"
@@ -319,36 +318,6 @@ func (a *StaticAuth) getDatabaseToken(ctx context.Context, poolConfig *pgxpool.C
 	return authenticationToken, nil
 }
 
-func (a *StaticAuth) Kafka(opts KafkaOptions) (*KafkaClients, error) {
-	a.mu.Lock()
-	defer a.mu.Unlock()
-
-	// This means we've already set the config in our New function
-	// to use the SASL token provider.
-	if a.clients.kafka != nil {
-		return a.clients.kafka, nil
-	}
-
-	a.clients.kafka = initKafkaClients(opts)
-	// static auth has additional fields we need added,
-	// so we add those static auth specific fields here,
-	// and the rest of the token provider fields are added in New()
-	tokenProvider := mskTokenProvider{}
-	if a.assumeRoleARN != nil {
-		tokenProvider.awsIamRoleArn = *a.assumeRoleARN
-	}
-	if a.sessionName != "" {
-		tokenProvider.awsStsSessionName = a.sessionName
-	}
-
-	err := a.clients.kafka.New(a.session, &tokenProvider)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create AWS IAM Kafka config: %w", err)
-	}
-
-	return a.clients.kafka, nil
-}
-
 func (a *StaticAuth) createSession() (*session.Session, error) {
 	var awsConfig *aws.Config
 	if a.cfg == nil {
@@ -390,21 +359,7 @@ func (a *StaticAuth) createSession() (*session.Session, error) {
 }
 
 func (a *StaticAuth) Close() error {
-	a.mu.Lock()
-	defer a.mu.Unlock()
-
-	errs := make([]error, 2)
-	if a.clients.kafka != nil {
-		if a.clients.kafka.Producer != nil {
-			errs[0] = a.clients.kafka.Producer.Close()
-			a.clients.kafka.Producer = nil
-		}
-		if a.clients.kafka.ConsumerGroup != nil {
-			errs[1] = a.clients.kafka.ConsumerGroup.Close()
-			a.clients.kafka.ConsumerGroup = nil
-		}
-	}
-	return errors.Join(errs...)
+	return nil
 }
 
 func GetConfigV2(accessKey string, secretKey string, sessionToken string, region string, endpoint string) (awsv2.Config, error) {

common/authentication/aws/x509.go

@@ -138,23 +138,7 @@ func newX509(ctx context.Context, opts Options, cfg *aws.Config) (*x509, error)
 }
 
 func (a *x509) Close() error {
-	a.mu.Lock()
-	defer a.mu.Unlock()
-	close(a.closeCh)
-	a.wg.Wait()
-
-	errs := make([]error, 2)
-	if a.clients.kafka != nil {
-		if a.clients.kafka.Producer != nil {
-			errs[0] = a.clients.kafka.Producer.Close()
-			a.clients.kafka.Producer = nil
-		}
-		if a.clients.kafka.ConsumerGroup != nil {
-			errs[1] = a.clients.kafka.ConsumerGroup.Close()
-			a.clients.kafka.ConsumerGroup = nil
-		}
-	}
-	return errors.Join(errs...)
+	return nil
 }
 
 func (a *x509) getCertPEM(ctx context.Context) error {
@@ -409,26 +393,6 @@ func (a *x509) UpdatePostgres(ctx context.Context, poolConfig *pgxpool.Config) {
 	}
 }
 
-func (a *x509) Kafka(opts KafkaOptions) (*KafkaClients, error) {
-	a.mu.Lock()
-	defer a.mu.Unlock()
-
-	// This means we've already set the config in our New function
-	// to use the SASL token provider.
-	if a.clients.kafka != nil {
-		return a.clients.kafka, nil
-	}
-
-	a.clients.kafka = initKafkaClients(opts)
-	// Note: we pass in nil for token provider,
-	// as there are no special fields for x509 auth for it.
-	err := a.clients.kafka.New(a.session, nil)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create AWS IAM Kafka config: %w", err)
-	}
-	return a.clients.kafka, nil
-}
-
 func (a *x509) initializeTrustAnchors() error {
 	var (
 		trustAnchor arn.ARN

common/aws/auth/auth.go

@@ -22,13 +22,14 @@ type Options struct {
 	Logger     logger.Logger
 	Properties map[string]string
 
-	Region          string `json:"region" mapstructure:"region" mapstructurealiases:"awsRegion"`
-	AccessKey       string `json:"accessKey" mapstructure:"accessKey"`
-	SecretKey       string `json:"secretKey" mapstructure:"secretKey"`
-	SessionToken    string `json:"sessionToken" mapstructure:"sessionToken"`
-	AssumeRoleArn   string `json:"assumeRoleArn" mapstructure:"assumeRoleArn"`
-	TrustAnchorArn  string `json:"trustAnchorArn" mapstructure:"trustAnchorArn"`
-	TrustProfileArn string `json:"trustProfileArn" mapstructure:"trustProfileArn"`
+	Region                string `json:"region" mapstructure:"region" mapstructurealiases:"awsRegion"`
+	AccessKey             string `json:"accessKey" mapstructure:"accessKey"`
+	SecretKey             string `json:"secretKey" mapstructure:"secretKey"`
+	SessionToken          string `json:"sessionToken" mapstructure:"sessionToken"`
+	AssumeRoleArn         string `json:"assumeRoleArn" mapstructure:"assumeRoleArn"`
+	AssumeRoleSessionName string `json:"assumeRoleSessionName" mapstructure:"assumeRoleSessionName"`
+	TrustAnchorArn        string `json:"trustAnchorArn" mapstructure:"trustAnchorArn"`
+	TrustProfileArn       string `json:"trustProfileArn" mapstructure:"trustProfileArn"`
 
 	Endpoint string `json:"endpoint" mapstructure:"endpoint"`
 }
@@ -48,3 +49,14 @@ func NewCredentialProvider(ctx context.Context, opts Options, configOpts []func(
 	}
 	return newAuthStatic(ctx, opts, configOpts)
 }
+
+// Coalesce is a helper function to return the first non-empty string from the inputs
+// This helps us to migrate away from the deprecated duplicate aws auth profile metadata fields in Dapr 1.17.
+func Coalesce(values ...string) string {
+	for _, v := range values {
+		if v != "" {
+			return v
+		}
+	}
+	return ""
+}