Skip to content

Commit 2d3d627

Browse files
karolszmndykarolszmndy
authored
Merge pull request #1 from mondaycom/feat/add-aws-msk-topic-table
feat: add aws_msk_topic table
2 parents 55cd246 + c9cf4fb commit 2d3d627

5 files changed

Lines changed: 441 additions & 15 deletions

File tree

‎aws/plugin.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -571,6 +571,7 @@ func Plugin(ctx context.Context) *plugin.Plugin {
571571
"aws_mgn_application": tableAwsMGNApplication(ctx),
572572
"aws_mq_broker": tableAwsMQBroker(ctx),
573573
"aws_msk_cluster": tableAwsMSKCluster(ctx),
574+
"aws_msk_topic": tableAwsMSKTopic(ctx),
574575
"aws_msk_serverless_cluster": tableAwsMSKServerlessCluster(ctx),
575576
"aws_mskconnect_connector": tableAwsMSKConnectConnector(ctx),
576577
"aws_neptune_db_cluster_snapshot": tableAwsNeptuneDBClusterSnapshot(ctx),

‎aws/table_aws_msk_topic.go‎

Lines changed: 261 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,261 @@
1+
package aws
2+
3+
import (
4+
"context"
5+
6+
"github.com/aws/aws-sdk-go-v2/aws"
7+
"github.com/aws/aws-sdk-go-v2/service/kafka"
8+
"github.com/aws/aws-sdk-go-v2/service/kafka/types"
9+
10+
"github.com/turbot/steampipe-plugin-sdk/v5/grpc/proto"
11+
"github.com/turbot/steampipe-plugin-sdk/v5/plugin"
12+
"github.com/turbot/steampipe-plugin-sdk/v5/plugin/transform"
13+
)
14+
15+
//// TABLE DEFINITION
16+
17+
func tableAwsMSKTopic(_ context.Context) *plugin.Table {
18+
return &plugin.Table{
19+
Name: "aws_msk_topic",
20+
Description: "AWS MSK Topic",
21+
Get: &plugin.GetConfig{
22+
KeyColumns: plugin.AllColumns([]string{"cluster_arn", "topic_name"}),
23+
Hydrate: getMSKTopic,
24+
Tags: map[string]string{"service": "kafka", "action": "DescribeTopic"},
25+
IgnoreConfig: &plugin.IgnoreConfig{
26+
ShouldIgnoreErrorFunc: shouldIgnoreErrors([]string{"NotFoundException"}),
27+
},
28+
},
29+
List: &plugin.ListConfig{
30+
Hydrate: listMSKTopics,
31+
ParentHydrate: listKafkaClusters(string(types.ClusterTypeProvisioned)),
32+
Tags: map[string]string{"service": "kafka", "action": "ListTopics"},
33+
KeyColumns: plugin.KeyColumnSlice{
34+
{Name: "cluster_arn", Require: plugin.Optional},
35+
{Name: "topic_name", Require: plugin.Optional},
36+
},
37+
},
38+
HydrateConfig: []plugin.HydrateConfig{
39+
{
40+
Func: describeMSKTopic,
41+
Tags: map[string]string{"service": "kafka", "action": "DescribeTopic"},
42+
},
43+
},
44+
GetMatrixItemFunc: SupportedRegionMatrix(AWS_KAFKA_SERVICE_ID),
45+
Columns: awsRegionalColumns([]*plugin.Column{
46+
{
47+
Name: "cluster_arn",
48+
Description: "The Amazon Resource Name (ARN) of the MSK cluster.",
49+
Type: proto.ColumnType_STRING,
50+
},
51+
{
52+
Name: "topic_name",
53+
Description: "The name of the Kafka topic.",
54+
Type: proto.ColumnType_STRING,
55+
},
56+
{
57+
Name: "topic_arn",
58+
Description: "The Amazon Resource Name (ARN) of the topic.",
59+
Type: proto.ColumnType_STRING,
60+
},
61+
{
62+
Name: "partition_count",
63+
Description: "The number of partitions for the topic.",
64+
Type: proto.ColumnType_INT,
65+
},
66+
{
67+
Name: "replication_factor",
68+
Description: "The replication factor for the topic.",
69+
Type: proto.ColumnType_INT,
70+
},
71+
{
72+
Name: "out_of_sync_replica_count",
73+
Description: "The number of out-of-sync replicas for the topic.",
74+
Type: proto.ColumnType_INT,
75+
},
76+
{
77+
Name: "status",
78+
Description: "The status of the topic.",
79+
Type: proto.ColumnType_STRING,
80+
Hydrate: describeMSKTopic,
81+
Transform: transform.FromField("Status"),
82+
},
83+
{
84+
Name: "configs",
85+
Description: "Topic configurations encoded as a Base64 string.",
86+
Type: proto.ColumnType_STRING,
87+
Hydrate: describeMSKTopic,
88+
Transform: transform.FromField("Configs"),
89+
},
90+
91+
// Standard columns
92+
{
93+
Name: "title",
94+
Description: resourceInterfaceDescription("title"),
95+
Type: proto.ColumnType_STRING,
96+
Transform: transform.FromField("TopicName"),
97+
},
98+
{
99+
Name: "akas",
100+
Description: resourceInterfaceDescription("akas"),
101+
Type: proto.ColumnType_JSON,
102+
Transform: transform.FromField("TopicArn").Transform(transform.EnsureStringArray),
103+
},
104+
}),
105+
}
106+
}
107+
108+
type mskTopicRow struct {
109+
ClusterArn *string
110+
TopicName *string
111+
TopicArn *string
112+
PartitionCount *int32
113+
ReplicationFactor *int32
114+
OutOfSyncReplicaCount *int32
115+
}
116+
117+
//// LIST FUNCTION
118+
119+
func listMSKTopics(ctx context.Context, d *plugin.QueryData, h *plugin.HydrateData) (interface{}, error) {
120+
logger := plugin.Logger(ctx)
121+
122+
cluster := h.Item.(types.Cluster)
123+
clusterArn := cluster.ClusterArn
124+
125+
// If cluster_arn qual is provided, skip clusters that don't match
126+
if d.EqualsQuals["cluster_arn"] != nil {
127+
qualClusterArn := d.EqualsQuals["cluster_arn"].GetStringValue()
128+
if qualClusterArn != *clusterArn {
129+
return nil, nil
130+
}
131+
}
132+
133+
svc, err := KafkaClient(ctx, d)
134+
if err != nil {
135+
logger.Error("aws_msk_topic.listMSKTopics", "service_creation_error", err)
136+
return nil, err
137+
}
138+
if svc == nil {
139+
return nil, nil
140+
}
141+
142+
maxLimit := int32(100)
143+
if d.QueryContext.Limit != nil {
144+
limit := int32(*d.QueryContext.Limit)
145+
if limit < maxLimit {
146+
if limit < 20 {
147+
maxLimit = 20
148+
} else {
149+
maxLimit = limit
150+
}
151+
}
152+
}
153+
154+
input := kafka.ListTopicsInput{
155+
ClusterArn: clusterArn,
156+
MaxResults: aws.Int32(maxLimit),
157+
}
158+
159+
if d.EqualsQuals["topic_name"] != nil {
160+
input.TopicNameFilter = aws.String(d.EqualsQuals["topic_name"].GetStringValue())
161+
}
162+
163+
paginator := kafka.NewListTopicsPaginator(svc, &input, func(o *kafka.ListTopicsPaginatorOptions) {
164+
o.Limit = maxLimit
165+
o.StopOnDuplicateToken = true
166+
})
167+
168+
for paginator.HasMorePages() {
169+
d.WaitForListRateLimit(ctx)
170+
171+
output, err := paginator.NextPage(ctx)
172+
if err != nil {
173+
logger.Error("aws_msk_topic.listMSKTopics", "api_error", err)
174+
return nil, err
175+
}
176+
177+
for _, topic := range output.Topics {
178+
d.StreamListItem(ctx, mskTopicRow{
179+
ClusterArn: clusterArn,
180+
TopicName: topic.TopicName,
181+
TopicArn: topic.TopicArn,
182+
PartitionCount: topic.PartitionCount,
183+
ReplicationFactor: topic.ReplicationFactor,
184+
OutOfSyncReplicaCount: topic.OutOfSyncReplicaCount,
185+
})
186+
187+
if d.RowsRemaining(ctx) == 0 {
188+
return nil, nil
189+
}
190+
}
191+
}
192+
193+
return nil, nil
194+
}
195+
196+
//// HYDRATE FUNCTIONS
197+
198+
func getMSKTopic(ctx context.Context, d *plugin.QueryData, _ *plugin.HydrateData) (interface{}, error) {
199+
logger := plugin.Logger(ctx)
200+
201+
clusterArn := d.EqualsQuals["cluster_arn"].GetStringValue()
202+
topicName := d.EqualsQuals["topic_name"].GetStringValue()
203+
if clusterArn == "" || topicName == "" {
204+
return nil, nil
205+
}
206+
207+
svc, err := KafkaClient(ctx, d)
208+
if err != nil {
209+
logger.Error("aws_msk_topic.getMSKTopic", "service_creation_error", err)
210+
return nil, err
211+
}
212+
if svc == nil {
213+
return nil, nil
214+
}
215+
216+
output, err := svc.DescribeTopic(ctx, &kafka.DescribeTopicInput{
217+
ClusterArn: aws.String(clusterArn),
218+
TopicName: aws.String(topicName),
219+
})
220+
if err != nil {
221+
logger.Error("aws_msk_topic.getMSKTopic", "api_error", err)
222+
return nil, err
223+
}
224+
225+
return mskTopicRow{
226+
ClusterArn: aws.String(clusterArn),
227+
TopicName: output.TopicName,
228+
TopicArn: output.TopicArn,
229+
PartitionCount: output.PartitionCount,
230+
ReplicationFactor: output.ReplicationFactor,
231+
}, nil
232+
}
233+
234+
func describeMSKTopic(ctx context.Context, d *plugin.QueryData, h *plugin.HydrateData) (interface{}, error) {
235+
logger := plugin.Logger(ctx)
236+
row := h.Item.(mskTopicRow)
237+
238+
if row.ClusterArn == nil || row.TopicName == nil {
239+
return nil, nil
240+
}
241+
242+
svc, err := KafkaClient(ctx, d)
243+
if err != nil {
244+
logger.Error("aws_msk_topic.describeMSKTopic", "service_creation_error", err)
245+
return nil, err
246+
}
247+
if svc == nil {
248+
return nil, nil
249+
}
250+
251+
output, err := svc.DescribeTopic(ctx, &kafka.DescribeTopicInput{
252+
ClusterArn: row.ClusterArn,
253+
TopicName: row.TopicName,
254+
})
255+
if err != nil {
256+
logger.Error("aws_msk_topic.describeMSKTopic", "api_error", err)
257+
return nil, err
258+
}
259+
260+
return output, nil
261+
}

0 commit comments

Comments
 (0)