Repository: cloud-custodian/cloud-custodian

Issue:
Add support for Kinesis Analytics V2
**Is your feature request related to a problem? Please describe.**
Make sure all kinesis analytics services are in a private subnet

**Describe the solution you'd like**
Add a new resource type for KinesisAnalyticsV2 including tagging, delete and subnet filter. Link to boto3 docs: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kinesisanalyticsv2.html#KinesisAnalyticsV2.Client.describe_application

**Describe alternatives you've considered**
NA

**Additional context**
NA


Discussion:
PratMis: Looks like CI is not able to import docker

Patch:
diff --git a/c7n/resources/kinesis.py b/c7n/resources/kinesis.py
index e54333e6e..cdaf8e0b4 100644
--- a/c7n/resources/kinesis.py
+++ b/c7n/resources/kinesis.py
@@ -7,6 +7,7 @@ from c7n.manager import resources
 from c7n.filters.kms import KmsRelatedFilter
 from c7n.query import ConfigSource, DescribeSource, QueryResourceManager, TypeInfo
 from c7n.tags import universal_augment
+from c7n.filters.vpc import SubnetFilter
 from c7n.utils import local_session, type_schema, get_retry
 
 
@@ -278,6 +279,49 @@ class DescribeVideoStream(DescribeSource):
         return universal_augment(self.manager, super().augment(resources))
 
 
+@resources.register('kinesis-analyticsv2')
+class KinesisAnalyticsAppV2(QueryResourceManager):
+
+    class resource_type(TypeInfo):
+        service = "kinesisanalyticsv2"
+        enum_spec = ('list_applications', 'ApplicationSummaries', None)
+        detail_spec = ('describe_application', 'ApplicationName',
+                       'ApplicationName', 'ApplicationDetail')
+        name = "ApplicationName"
+        arn = id = "ApplicationARN"
+        arn_type = 'application'
+        universal_taggable = object()
+        cfn_type = 'AWS::KinesisAnalyticsV2::Application'
+        permission_prefix = "kinesisanalytics"
+
+    permissions = ("kinesisanalytics:DescribeApplication",)
+
+    def augment(self, resources):
+        return universal_augment(self, super().augment(resources))
+
+
+@KinesisAnalyticsAppV2.action_registry.register('delete')
+class KinesisAnalyticsAppV2Delete(Action):
+
+    schema = type_schema('delete')
+    permissions = ("kinesisanalytics:DeleteApplication",)
+
+    def process(self, resources):
+        client = local_session(
+            self.manager.session_factory).client('kinesisanalyticsv2')
+        for r in resources:
+            client.delete_application(
+                ApplicationName=r['ApplicationName'],
+                CreateTimestamp=r['CreateTimestamp'])
+
+
+@KinesisAnalyticsAppV2.filter_registry.register('subnet')
+class KinesisAnalyticsSubnetFilter(SubnetFilter):
+
+    RelatedIdsExpression = 'ApplicationConfigurationDescription.' \
+        'VpcConfigurationDescriptions[].SubnetIds[]'
+
+
 @resources.register('kinesis-video')
 class KinesisVideoStream(QueryResourceManager):
     retry = staticmethod(
diff --git a/c7n/resources/resource_map.py b/c7n/resources/resource_map.py
index 7d62f33ff..445f2ca9e 100644
--- a/c7n/resources/resource_map.py
+++ b/c7n/resources/resource_map.py
@@ -110,6 +110,7 @@ ResourceMap = {
     "aws.key-pair": "c7n.resources.vpc.KeyPair",
     "aws.kinesis": "c7n.resources.kinesis.KinesisStream",
     "aws.kinesis-analytics": "c7n.resources.kinesis.AnalyticsApp",
+    "aws.kinesis-analyticsv2": "c7n.resources.kinesis.KinesisAnalyticsAppV2",
     "aws.kinesis-video": "c7n.resources.kinesis.KinesisVideoStream",
     "aws.kms": "c7n.resources.kms.KeyAlias",
     "aws.kms-key": "c7n.resources.kms.Key",


Test patch:
diff --git a/tests/data/placebo/test_kinesis_analyticsv2_app_delete/ec2.DescribeSubnets_1.json b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/ec2.DescribeSubnets_1.json
new file mode 100644
index 000000000..a59d1bcbd
--- /dev/null
+++ b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/ec2.DescribeSubnets_1.json
@@ -0,0 +1,30 @@
+{
+    "status_code": 200,
+    "data": {
+        "Subnets": [
+            {
+                "AvailabilityZone": "us-east-1b",
+                "AvailabilityZoneId": "use1-az6",
+                "AvailableIpAddressCount": 251,
+                "CidrBlock": "10.0.0.0/24",
+                "DefaultForAz": false,
+                "MapPublicIpOnLaunch": false,
+                "MapCustomerOwnedIpOnLaunch": false,
+                "State": "available",
+                "SubnetId": "subnet-05b58b4afe5124322",
+                "VpcId": "vpc-0a59f94e3fca9fbc0",
+                "OwnerId": "644160558196",
+                "AssignIpv6AddressOnCreation": false,
+                "Ipv6CidrBlockAssociationSet": [],
+                "Tags": [
+                    {
+                        "Key": "Name",
+                        "Value": "implied"
+                    }
+                ],
+                "SubnetArn": "arn:aws:ec2:us-east-1:644160558196:subnet/subnet-05b58b4afe5124322"
+            }
+        ],
+        "ResponseMetadata": {}
+    }
+}
\ No newline at end of file
diff --git a/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DeleteApplication_1.json b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DeleteApplication_1.json
new file mode 100644
index 000000000..5b2170a07
--- /dev/null
+++ b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DeleteApplication_1.json
@@ -0,0 +1,6 @@
+{
+    "status_code": 200,
+    "data": {
+        "ResponseMetadata": {}
+    }
+}
\ No newline at end of file
diff --git a/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DescribeApplication_1.json b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DescribeApplication_1.json
new file mode 100644
index 000000000..ffb7bce54
--- /dev/null
+++ b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DescribeApplication_1.json
@@ -0,0 +1,59 @@
+{
+    "status_code": 200,
+    "data": {
+        "ApplicationDetail": {
+            "ApplicationARN": "arn:aws:kinesisanalytics:us-east-1:644160558196:application/test1",
+            "ApplicationName": "test1",
+            "RuntimeEnvironment": "FLINK-1_6",
+            "ServiceExecutionRole": "arn:aws:iam::644160558196:role/service-role/kinesis-analytics-test2-us-east-1",
+            "ApplicationStatus": "READY",
+            "ApplicationVersionId": 1,
+            "CreateTimestamp": {
+                "__class__": "datetime",
+                "year": 2020,
+                "month": 8,
+                "day": 19,
+                "hour": 17,
+                "minute": 43,
+                "second": 36,
+                "microsecond": 0
+            },
+            "LastUpdateTimestamp": {
+                "__class__": "datetime",
+                "year": 2020,
+                "month": 8,
+                "day": 19,
+                "hour": 17,
+                "minute": 43,
+                "second": 36,
+                "microsecond": 0
+            },
+            "ApplicationConfigurationDescription": {
+                "FlinkApplicationConfigurationDescription": {
+                    "CheckpointConfigurationDescription": {
+                        "ConfigurationType": "DEFAULT",
+                        "CheckpointingEnabled": true,
+                        "CheckpointInterval": 60000,
+                        "MinPauseBetweenCheckpoints": 5000
+                    },
+                    "MonitoringConfigurationDescription": {
+                        "ConfigurationType": "DEFAULT",
+                        "MetricsLevel": "APPLICATION",
+                        "LogLevel": "INFO"
+                    },
+                    "ParallelismConfigurationDescription": {
+                        "ConfigurationType": "DEFAULT",
+                        "Parallelism": 1,
+                        "ParallelismPerKPU": 1,
+                        "CurrentParallelism": 1,
+                        "AutoScalingEnabled": true
+                    }
+                },
+                "ApplicationSnapshotConfigurationDescription": {
+                    "SnapshotsEnabled": true
+                }
+            }
+        },
+        "ResponseMetadata": {}
+    }
+}
\ No newline at end of file
diff --git a/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DescribeApplication_2.json b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DescribeApplication_2.json
new file mode 100644
index 000000000..34cb25104
--- /dev/null
+++ b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DescribeApplication_2.json
@@ -0,0 +1,82 @@
+{
+    "status_code": 200,
+    "data": {
+        "ApplicationDetail": {
+            "ApplicationARN": "arn:aws:kinesisanalytics:us-east-1:644160558196:application/test2",
+            "ApplicationName": "test2",
+            "RuntimeEnvironment": "FLINK-1_8",
+            "ServiceExecutionRole": "arn:aws:iam::644160558196:role/service-role/kinesis-analytics-test2-us-east-1",
+            "ApplicationStatus": "READY",
+            "ApplicationVersionId": 3,
+            "CreateTimestamp": {
+                "__class__": "datetime",
+                "year": 2020,
+                "month": 8,
+                "day": 18,
+                "hour": 12,
+                "minute": 58,
+                "second": 47,
+                "microsecond": 0
+            },
+            "LastUpdateTimestamp": {
+                "__class__": "datetime",
+                "year": 2020,
+                "month": 8,
+                "day": 18,
+                "hour": 13,
+                "minute": 4,
+                "second": 14,
+                "microsecond": 0
+            },
+            "ApplicationConfigurationDescription": {
+                "ApplicationCodeConfigurationDescription": {
+                    "CodeContentType": "ZIPFILE",
+                    "CodeContentDescription": {
+                        "CodeMD5": "c4031c4997af1d0a77ac5623c3c0ee4e",
+                        "CodeSize": 22218709,
+                        "S3ApplicationCodeLocationDescription": {
+                            "BucketARN": "arn:aws:s3:::repela",
+                            "FileKey": "kd.jar"
+                        }
+                    }
+                },
+                "FlinkApplicationConfigurationDescription": {
+                    "CheckpointConfigurationDescription": {
+                        "ConfigurationType": "DEFAULT",
+                        "CheckpointingEnabled": true,
+                        "CheckpointInterval": 60000,
+                        "MinPauseBetweenCheckpoints": 5000
+                    },
+                    "MonitoringConfigurationDescription": {
+                        "ConfigurationType": "DEFAULT",
+                        "MetricsLevel": "APPLICATION",
+                        "LogLevel": "INFO"
+                    },
+                    "ParallelismConfigurationDescription": {
+                        "ConfigurationType": "DEFAULT",
+                        "Parallelism": 1,
+                        "ParallelismPerKPU": 1,
+                        "CurrentParallelism": 1,
+                        "AutoScalingEnabled": true
+                    }
+                },
+                "ApplicationSnapshotConfigurationDescription": {
+                    "SnapshotsEnabled": true
+                },
+                "VpcConfigurationDescriptions": [
+                    {
+                        "VpcConfigurationId": "3.1",
+                        "VpcId": "vpc-0a59f94e3fca9fbc0",
+                        "SubnetIds": [
+                            "subnet-05b58b4afe5124322"
+                        ],
+                        "SecurityGroupIds": [
+                            "sg-05b111f88a2aefa2c"
+                        ]
+                    }
+                ]
+            }
+        },
+        "ResponseMetadata": {}
+    }
+}
\ No newline at end of file
diff --git a/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DescribeApplication_3.json b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DescribeApplication_3.json
new file mode 100644
index 000000000..6d8d2725c
--- /dev/null
+++ b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.DescribeApplication_3.json
@@ -0,0 +1,82 @@
+{
+    "status_code": 200,
+    "data": {
+        "ApplicationDetail": {
+            "ApplicationARN": "arn:aws:kinesisanalytics:us-east-1:644160558196:application/test2",
+            "ApplicationName": "test2",
+            "RuntimeEnvironment": "FLINK-1_8",
+            "ServiceExecutionRole": "arn:aws:iam::644160558196:role/service-role/kinesis-analytics-test2-us-east-1",
+            "ApplicationStatus": "DELETING",
+            "ApplicationVersionId": 3,
+            "CreateTimestamp": {
+                "__class__": "datetime",
+                "year": 2020,
+                "month": 8,
+                "day": 18,
+                "hour": 12,
+                "minute": 58,
+                "second": 47,
+                "microsecond": 0
+            },
+            "LastUpdateTimestamp": {
+                "__class__": "datetime",
+                "year": 2021,
+                "month": 5,
+                "day": 18,
+                "hour": 0,
+                "minute": 55,
+                "second": 30,
+                "microsecond": 0
+            },
+            "ApplicationConfigurationDescription": {
+                "ApplicationCodeConfigurationDescription": {
+                    "CodeContentType": "ZIPFILE",
+                    "CodeContentDescription": {
+                        "CodeMD5": "c4031c4997af1d0a77ac5623c3c0ee4e",
+                        "CodeSize": 22218709,
+                        "S3ApplicationCodeLocationDescription": {
+                            "BucketARN": "arn:aws:s3:::repela",
+                            "FileKey": "kd.jar"
+                        }
+                    }
+                },
+                "FlinkApplicationConfigurationDescription": {
+                    "CheckpointConfigurationDescription": {
+                        "ConfigurationType": "DEFAULT",
+                        "CheckpointingEnabled": true,
+                        "CheckpointInterval": 60000,
+                        "MinPauseBetweenCheckpoints": 5000
+                    },
+                    "MonitoringConfigurationDescription": {
+                        "ConfigurationType": "DEFAULT",
+                        "MetricsLevel": "APPLICATION",
+                        "LogLevel": "INFO"
+                    },
+                    "ParallelismConfigurationDescription": {
+                        "ConfigurationType": "DEFAULT",
+                        "Parallelism": 1,
+                        "ParallelismPerKPU": 1,
+                        "CurrentParallelism": 1,
+                        "AutoScalingEnabled": true
+                    }
+                },
+                "ApplicationSnapshotConfigurationDescription": {
+                    "SnapshotsEnabled": true
+                },
+                "VpcConfigurationDescriptions": [
+                    {
+                        "VpcConfigurationId": "3.1",
+                        "VpcId": "vpc-0a59f94e3fca9fbc0",
+                        "SubnetIds": [
+                            "subnet-05b58b4afe5124322"
+                        ],
+                        "SecurityGroupIds": [
+                            "sg-05b111f88a2aefa2c"
+                        ]
+                    }
+                ]
+            }
+        },
+        "ResponseMetadata": {}
+    }
+}
\ No newline at end of file
diff --git a/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.ListApplications_1.json b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.ListApplications_1.json
new file mode 100644
index 000000000..8621d9423
--- /dev/null
+++ b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/kinesisanalytics.ListApplications_1.json
@@ -0,0 +1,22 @@
+{
+    "status_code": 200,
+    "data": {
+        "ApplicationSummaries": [
+            {
+                "ApplicationName": "test1",
+                "ApplicationARN": "arn:aws:kinesisanalytics:us-east-1:644160558196:application/test1",
+                "ApplicationStatus": "READY",
+                "ApplicationVersionId": 1,
+                "RuntimeEnvironment": "FLINK-1_6"
+            },
+            {
+                "ApplicationName": "test2",
+                "ApplicationARN": "arn:aws:kinesisanalytics:us-east-1:644160558196:application/test2",
+                "ApplicationStatus": "READY",
+                "ApplicationVersionId": 3,
+                "RuntimeEnvironment": "FLINK-1_8"
+            }
+        ],
+        "ResponseMetadata": {}
+    }
+}
\ No newline at end of file
diff --git a/tests/data/placebo/test_kinesis_analyticsv2_app_delete/tagging.GetResources_1.json b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/tagging.GetResources_1.json
new file mode 100644
index 000000000..c4ffa3fc6
--- /dev/null
+++ b/tests/data/placebo/test_kinesis_analyticsv2_app_delete/tagging.GetResources_1.json
@@ -0,0 +1,18 @@
+{
+    "status_code": 200,
+    "data": {
+        "PaginationToken": "",
+        "ResourceTagMappingList": [
+            {
+                "ResourceARN": "arn:aws:kinesisanalytics:us-east-1:644160558196:application/test2",
+                "Tags": [
+                    {
+                        "Key": "hello",
+                        "Value": "pratyush"
+                    }
+                ]
+            }
+        ],
+        "ResponseMetadata": {}
+    }
+}
\ No newline at end of file
diff --git a/tests/test_kinesis.py b/tests/test_kinesis.py
index 61863990e..d11e57ddf 100644
--- a/tests/test_kinesis.py
+++ b/tests/test_kinesis.py
@@ -259,3 +259,30 @@ class Kinesis(BaseTest):
         self.assertTrue(len(resources), 1)
         self.assertEqual(resources[0]['KmsKeyId'],
             'arn:aws:kms:us-east-1:123456789012:key/0d543df5-915c-42a1-afa1-c9c5f1f97955')
+
+
+class KinesisAnalyticsAppV2(BaseTest):
+
+    def test_kinesis_analyticsv2_app_delete(self):
+        factory = self.replay_flight_data("test_kinesis_analyticsv2_app_delete")
+        p = self.load_policy(
+            {
+                "name": "kapp",
+                "resource": "kinesis-analyticsv2",
+                "filters": [{"type": "subnet", "key": "tag:Name", "value": "implied", "op": "eq"}],
+                "actions": ["delete"],
+            },
+            session_factory=factory,
+        )
+        resources = p.run()
+        self.assertEqual(len(resources), 1)
+        self.assertEqual(
+            factory().client("kinesisanalyticsv2").describe_application(
+                ApplicationName=resources[0]['ApplicationName']
+            )[
+                "ApplicationDetail"
+            ][
+                "ApplicationStatus"
+            ],
+            "DELETING",
+        )