out_kafka: msk iam role arn

include/fluent-bit/aws/flb_aws_msk_iam.h

@@ -36,11 +36,16 @@ struct flb_msk_iam_cb {
 
 /*
  * Register the oauthbearer refresh callback for MSK IAM authentication.
+ * Optionally accepts role_arn, sts_endpoint, and external_id for STS
+ * Assume Role support. Pass NULL for any unused optional parameter.
  * Returns context pointer on success or NULL on failure.
  */
 struct flb_aws_msk_iam *flb_aws_msk_iam_register_oauth_cb(struct flb_config *config,
                                                           rd_kafka_conf_t *kconf,
                                                           const char *cluster_arn,
+                                                          const char *role_arn,
+                                                          const char *sts_endpoint,
+                                                          const char *external_id,
                                                           struct flb_kafka_opaque *opaque);
 void flb_aws_msk_iam_destroy(struct flb_aws_msk_iam *ctx);
 

plugins/in_kafka/in_kafka.c

@@ -359,6 +359,7 @@ static int in_kafka_init(struct flb_input_instance *ins,
         ctx->msk_iam = flb_aws_msk_iam_register_oauth_cb(config,
                                                          kafka_conf,
                                                          ctx->aws_msk_iam_cluster_arn,
+                                                         NULL, NULL, NULL,
                                                          ctx->opaque);
         if (!ctx->msk_iam) {
             flb_plg_error(ins, "failed to setup MSK IAM authentication");

plugins/out_kafka/kafka.c

@@ -1619,6 +1619,23 @@ static struct flb_config_map config_map[] = {
     0, FLB_TRUE, offsetof(struct flb_out_kafka, aws_msk_iam),
     "Enable AWS MSK IAM authentication"
    },
+   {
+    FLB_CONFIG_MAP_STR, "aws_msk_iam_role_arn", NULL,
+    0, FLB_TRUE, offsetof(struct flb_out_kafka, aws_msk_iam_role_arn),
+    "ARN of an IAM role to assume for MSK IAM authentication "
+    "(e.g. for cross-account access)"
+   },
+   {
+    FLB_CONFIG_MAP_STR, "aws_msk_iam_sts_endpoint", NULL,
+    0, FLB_TRUE, offsetof(struct flb_out_kafka, aws_msk_iam_sts_endpoint),
+    "Custom endpoint for the AWS STS API, used with aws_msk_iam_role_arn"
+   },
+   {
+    FLB_CONFIG_MAP_STR, "aws_msk_iam_external_id", NULL,
+    0, FLB_TRUE, offsetof(struct flb_out_kafka, aws_msk_iam_external_id),
+    "External ID for the STS AssumeRole API, used with aws_msk_iam_role_arn "
+    "when the target role requires an external ID"
+   },
 #endif
 
    /* EOF */

plugins/out_kafka/kafka_config.c

@@ -233,6 +233,9 @@ struct flb_out_kafka *flb_out_kafka_create(struct flb_output_instance *ins,
         ctx->msk_iam = flb_aws_msk_iam_register_oauth_cb(config,
                                                          ctx->conf,
                                                          ctx->aws_msk_iam_cluster_arn,
+                                                         ctx->aws_msk_iam_role_arn,
+                                                         ctx->aws_msk_iam_sts_endpoint,
+                                                         ctx->aws_msk_iam_external_id,
                                                          ctx->opaque);
         if (!ctx->msk_iam) {
             flb_plg_error(ctx->ins, "failed to setup MSK IAM authentication");

plugins/out_kafka/kafka_config.h

@@ -154,6 +154,9 @@ struct flb_out_kafka {
 
 #ifdef FLB_HAVE_AWS_MSK_IAM
     flb_sds_t aws_msk_iam_cluster_arn;
+    flb_sds_t aws_msk_iam_role_arn;
+    flb_sds_t aws_msk_iam_sts_endpoint;
+    flb_sds_t aws_msk_iam_external_id;
     struct flb_aws_msk_iam *msk_iam;
 #endif