Add column BootstrapBrokers(bootstrap_broker_string, bootstrap_broker…

…_string_tls) in table aws_msk_cluster (#2390) (#2399)



Co-authored-by: KOH JUHO <[email protected]>

aws/table_aws_msk_cluster.go

@@ -41,6 +41,10 @@ func tableAwsMSKCluster(_ context.Context) *plugin.Table {
 				Func: getKafkaClusterOperation,
 				Tags: map[string]string{"service": "kafka", "action": "DescribeClusterOperation"},
 			},
+			{
+				Func: getKafkaClusterBootstrapBrokers,
+				Tags: map[string]string{"service": "kafka", "action": "GetBootstrapBrokers"},
+			},
 		},
 		GetMatrixItemFunc: SupportedRegionMatrix(kafkav1.EndpointsID),
 		Columns: awsRegionalColumns([]*plugin.Column{
@@ -94,6 +98,18 @@ func tableAwsMSKCluster(_ context.Context) *plugin.Table {
 				Hydrate:     getKafkaClusterOperation,
 				Transform:   transform.FromValue(),
 			},
+			{
+				Name:        "bootstrap_broker_string",
+				Description: "A string containing one or more hostname:port pairs of Kafka brokers suitable for use with Apache Kafka clients.",
+				Type:        proto.ColumnType_STRING,
+				Hydrate:     getKafkaClusterBootstrapBrokers,
+			},
+			{
+				Name:        "bootstrap_broker_string_tls",
+				Description: "A string containing one or more hostname:port pairs of Kafka brokers suitable for TLS authentication.",
+				Type:        proto.ColumnType_STRING,
+				Hydrate:     getKafkaClusterBootstrapBrokers,
+			},
 			{
 				Name:        "provisioned",
 				Description: "Information about the provisioned cluster.",
@@ -306,3 +322,37 @@ func getKafkaClusterConfiguration(ctx context.Context, d *plugin.QueryData, h *p
 
 	return op, nil
 }
+
+func getKafkaClusterBootstrapBrokers(ctx context.Context, d *plugin.QueryData, h *plugin.HydrateData) (interface{}, error) {
+	logger := plugin.Logger(ctx)
+	cluster := h.Item.(types.Cluster)
+
+	clusterArn := aws.ToString(cluster.ClusterArn)
+	// Empty check
+	if clusterArn == "" {
+		return nil, nil
+	}
+
+	svc, err := KafkaClient(ctx, d)
+	if err != nil {
+		logger.Error("aws_msk_cluster.getKafkaClusterBootstrapBrokers", "service_creation_error", err)
+		return nil, err
+	}
+
+	// Unsupported region, return no data
+	if svc == nil {
+		return nil, nil
+	}
+
+	params := &kafka.GetBootstrapBrokersInput{
+		ClusterArn: &clusterArn,
+	}
+
+	op, err := svc.GetBootstrapBrokers(ctx, params)
+	if err != nil {
+		logger.Error("aws_msk_cluster.getKafkaClusterBootstrapBrokers", "api_error", err)
+		return nil, err
+	}
+
+	return op, nil
+}