Support for AWS MSK services has been added

aws/policy/data-services.yaml

@@ -106,7 +106,47 @@ Statement:
       - iam:CreateServiceLinkedRole
     Resource:
       - 'arn:aws:iam::{{ aws_account_id }}:role/aws-service-role/elasticache.amazonaws.com/AWSServiceRoleForElastiCache'
+      - 'arn:aws:iam::{{ aws_account_id }}:role/aws-service-role/kafka.amazonaws.com/AWSServiceRoleForKafka'
     Condition:
       ForAnyValue:StringEquals:
         iam:AWSServiceName:
           - 'elasticache.amazonaws.com'
+          - 'kafka.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

@@ -1,3 +1,5 @@
+import datetime
+
 from . import DbTerminator, Terminator, get_tag_dict_from_tag_list
 
 
@@ -158,3 +160,53 @@ 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']
+
+    @property
+    def age_limit(self):
+        return datetime.timedelta(minutes=60)
+
+    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']
+
+    @property
+    def age_limit(self):
+        return datetime.timedelta(minutes=60)
+
+    def terminate(self):
+        self.client.delete_cluster(ClusterArn=self.id)