[vpc] enhancements

aws_quota/check/quota_check.py

@@ -17,6 +17,8 @@ class QuotaCheck:
     scope: QuotaScope = None
     service_code: str = None
     quota_code: str = None
+    warning_threshold: float = None
+    error_threshold: float = None
 
     def __init__(self, boto_session: boto3.Session) -> None:
         super().__init__()

aws_quota/check/vpc.py

@@ -27,6 +27,13 @@ def get_vpc_by_id(session: boto3.Session, vpc_id: str) -> dict:
     except StopIteration:
         raise KeyError
 
+@cachetools.cached(cache=cachetools.TTLCache(1, 60))
+def get_vpc_peering_connections(session: boto3.Session) -> typing.List[dict]:
+    return session.client('ec2').describe_vpc_peering_connections(
+        Filters=[
+            {'Name': 'status-code', 'Values': ['active']},
+        ]
+    )['VpcPeeringConnections']
 
 @cachetools.cached(cache=cachetools.TTLCache(1, 60))
 def get_all_sgs(session: boto3.Session) -> typing.List[dict]:
@@ -80,6 +87,16 @@ class InternetGatewayCountCheck(QuotaCheck):
     def current(self):
         return len(self.boto_session.client('ec2').describe_internet_gateways()['InternetGateways'])
 
+class VpcEndpointCountCheck(QuotaCheck):
+    key = "vpc_endpoint"
+    description = "Gateway VPC endpoints per Region"
+    scope = QuotaScope.REGION
+    service_code = 'vpc'
+    quota_code = 'L-1B52E74A'
+
+    @property
+    def current(self):
+        return len(self.boto_session.client('ec2').describe_vpc_endpoints()['VpcEndpoints'])
 
 class NetworkInterfaceCountCheck(QuotaCheck):
     key = "ni_count"
@@ -104,6 +121,16 @@ class SecurityGroupCountCheck(QuotaCheck):
     def current(self):
         return len(self.boto_session.client('ec2').describe_security_groups()['SecurityGroups'])
 
+class NatGatewayCountCheck(QuotaCheck):
+    key = "nat_count"
+    description = "NAT gateways per Region"
+    scope = QuotaScope.REGION
+    service_code = 'vpc'
+    quota_code = 'L-FE5A380F'
+
+    @property
+    def current(self):
+        return len(self.boto_session.client('ec2').describe_nat_gateways()['NatGateways'])
 
 class RulesPerSecurityGroupCheck(InstanceQuotaCheck):
     key = "vpc_rules_per_sg"
@@ -275,3 +302,29 @@ def current(self) -> int:
             return len(list(filter(lambda cbas: cbas['Ipv6CidrBlockState']['State'] == 'associated', vpc['Ipv6CidrBlockAssociationSet'])))
         except KeyError:
             raise InstanceWithIdentifierNotFound(self)
+
+class ActiveVpcPeeringConnectionsPerVpcCheck(InstanceQuotaCheck):
+    key = "vpc_peering_connections_per_vpc"
+    description = "Active VPC peering connections per VPC"
+    service_code = 'vpc'
+    quota_code = 'L-7E9ECCDB'
+    instance_id = 'VPC ID'
+
+    @staticmethod
+    def get_all_identifiers(session: boto3.Session) -> typing.List[str]:
+        return [vpc['VpcId'] for vpc in get_all_vpcs(session)]
+
+    @property
+    def current(self) -> int:
+        peering_connections_per_vpc = 0
+        try:
+            vpc = get_vpc_by_id(self.boto_session, self.instance_id)
+            vpc_peering_connections = get_vpc_peering_connections(self.boto_session)
+            for peering_connection in vpc_peering_connections:
+                for vpc_info in [peering_connection['AccepterVpcInfo'], peering_connection['RequesterVpcInfo']]:
+                    if vpc_info['VpcId'] == vpc['VpcId'] and self.boto_session.region_name == vpc_info['Region']:
+                        peering_connections_per_vpc += 1
+
+            return peering_connections_per_vpc
+        except KeyError:
+            raise InstanceWithIdentifierNotFound(self)