Add more context to aws config using Kafka.Provider parent

Author: myronjhicks

docs/config.md

@@ -73,11 +73,12 @@ ACHGateway:
       TLS: <boolean>
       AutoCommit: <boolean>
       [ SASLMechanism: <string> | default = "PLAIN" ]
-      AWS:
-        [ Region: <string> | default = "" ]
-        [ Profile: <string> | default = "" ]
-        [ RoleARN: <string> | default = "" ]
-        [ SessionName: <string> | default = "" ]
+      Provider:
+        AWS:
+          [ Region: <string> | default = "" ]
+          [ Profile: <string> | default = "" ]
+          [ RoleARN: <string> | default = "" ]
+          [ SessionName: <string> | default = "" ]
       Transform:
         Encoding:
           [ Base64: <boolean> | default = false ]
@@ -163,11 +164,12 @@ ACHGateway:
         [ TLS: <boolean> | default = false ]
         [ AutoCommit: <boolean> | default = false ]
         [ SASLMechanism: <string> | default = "PLAIN" ]
-        AWS:
-          [ Region: <string> | default = "" ]
-          [ Profile: <string> | default = "" ]
-          [ RoleARN: <string> | default = "" ]
-          [ SessionName: <string> | default = "" ]
+        Provider:
+          AWS:
+            [ Region: <string> | default = "" ]
+            [ Profile: <string> | default = "" ]
+            [ RoleARN: <string> | default = "" ]
+            [ SessionName: <string> | default = "" ]
     Webhook:
       [ Endpoint: <string> | default = "" ]
 ```

internal/kafka/kafka.go

@@ -54,12 +54,12 @@ func OpenTopic(logger log.Logger, cfg *service.KafkaConfig) (*pubsub.Topic, erro
 
 	switch cfg.SASLMechanism {
 	case "AWS_MSK_IAM":
-		if cfg.AWS != nil {
+		if cfg.Provider != nil && cfg.Provider.AWS != nil {
 			config.Net.SASL.TokenProvider = &MSKAccessTokenProvider{
-				Region:      cfg.AWS.Region,
-				Profile:     cfg.AWS.Profile,
-				RoleARN:     cfg.AWS.RoleARN,
-				SessionName: cfg.AWS.SessionName,
+				Region:      cfg.Provider.AWS.Region,
+				Profile:     cfg.Provider.AWS.Profile,
+				RoleARN:     cfg.Provider.AWS.RoleARN,
+				SessionName: cfg.Provider.AWS.SessionName,
 			}
 		}
 		config.Net.SASL.Mechanism = sarama.SASLTypeOAuth
@@ -98,12 +98,12 @@ func OpenSubscription(logger log.Logger, cfg *service.KafkaConfig) (*pubsub.Subs
 	// Default to PLAIN if no SASL mechanism is specified
 	switch cfg.SASLMechanism {
 	case "AWS_MSK_IAM":
-		if cfg.AWS != nil {
+		if cfg.Provider != nil && cfg.Provider.AWS != nil {
 			config.Net.SASL.TokenProvider = &MSKAccessTokenProvider{
-				Region:      cfg.AWS.Region,
-				Profile:     cfg.AWS.Profile,
-				RoleARN:     cfg.AWS.RoleARN,
-				SessionName: cfg.AWS.SessionName,
+				Region:      cfg.Provider.AWS.Region,
+				Profile:     cfg.Provider.AWS.Profile,
+				RoleARN:     cfg.Provider.AWS.RoleARN,
+				SessionName: cfg.Provider.AWS.SessionName,
 			}
 		}
 		config.Net.SASL.Mechanism = sarama.SASLTypeOAuth

internal/service/model_inbound.go

@@ -75,6 +75,11 @@ type AWSConfig struct {
 	SessionName string
 }
 
+// KafkaProvider represents the cloud provider settings for Kafka
+type KafkaProvider struct {
+	AWS *AWSConfig
+}
+
 type KafkaConfig struct {
 	Brokers []string
 	Key     string
@@ -89,7 +94,7 @@ type KafkaConfig struct {
 	AutoCommit    bool
 	SASLMechanism string
 
-	AWS *AWSConfig
+	Provider *KafkaProvider
 
 	Consumer KafkaConsumerConfig
 	Producer KafkaProducerConfig