Support for AWS MSK services has been added

aws/policy/data-services.yaml

@@ -110,3 +110,41 @@ Statement:
       ForAnyValue:StringEquals:
         iam:AWSServiceName:
           - 'elasticache.amazonaws.com'
+  - Sid: KafkaConfiguration
+    Effect: Allow
+    Action:
+      - kafka:CreateConfiguration
+      - kafka:DescribeConfigurationRevision
+      - kafka:DeleteConfiguration
+      - kafka:ListConfigurationRevisions
+      - kafka:DescribeConfiguration
+      - kafka:ListConfigurations
+      - kafka:UpdateConfiguration
+    Resource: "*"
+  - Sid: KafkaCluster
+    Effect: Allow
+    Action:
+      - kafka:UpdateClusterKafkaVersion
+      - kafka:TagResource
+      - kafka:UpdateBrokerCount
+      - kafka:CreateCluster
+      - kafka:ListTagsForResource
+      - kafka:GetCompatibleKafkaVersions
+      - kafka:UpdateClusterConfiguration
+      - kafka:ListClusters
+      - kafka:ListScramSecrets
+      - kafka:DescribeCluster
+      - kafka:ListKafkaVersions
+      - kafka:GetBootstrapBrokers
+      - kafka:ListConfigurations
+      - kafka:UpdateBrokerStorage
+      - kafka:RebootBroker
+      - kafka:DescribeClusterOperation
+      - kafka:UpdateMonitoring
+      - kafka:ListConfigurationRevisions
+      - kafka:ListNodes
+      - kafka:DeleteCluster
+      - kafka:UpdateBrokerType
+      - kafka:UntagResource
+      - kafka:ListClusterOperations
+    Resource: "*"

aws/terminator/data_services.py

@@ -158,3 +158,45 @@ def created_time(self):
 
     def terminate(self):
         self.client.delete_cluster(ClusterIdentifier=self.id, SkipFinalClusterSnapshot=True)
+
+
+class KafkaConfiguration(Terminator):
+    @staticmethod
+    def create(credentials):
+        return Terminator._create(credentials, KafkaConfiguration, 'kafka', lambda client: client.list_configurations()['Configurations'])
+
+    @property
+    def id(self):
+        return self.instance['Arn']
+
+    @property
+    def name(self):
+        return self.instance['Name']
+
+    @property
+    def created_time(self):
+        return self.instance['CreationTime']
+
+    def terminate(self):
+        self.client.delete_configuration(Arn=self.id)
+
+
+class KafkaCluster(Terminator):
+    @staticmethod
+    def create(credentials):
+        return Terminator._create(credentials, KafkaCluster, 'kafka', lambda client: client.list_clusters()['ClusterInfoList'])
+
+    @property
+    def id(self):
+        return self.instance['ClusterArn']
+
+    @property
+    def name(self):
+        return self.instance['ClusterName']
+
+    @property
+    def created_time(self):
+        return self.instance['CreationTime']
+
+    def terminate(self):
+        self.client.delete_configuration(Arn=self.id)