aws - ec2 - Infracost integration, add cost filter

c7n/cli.py

@@ -328,9 +328,9 @@ def _setup_logger(options):
         external_log_level = logging.INFO
 
     logging.getLogger('botocore').setLevel(external_log_level)
-    logging.getLogger('urllib3').setLevel(external_log_level)
     logging.getLogger('s3transfer').setLevel(external_log_level)
     logging.getLogger('urllib3').setLevel(logging.ERROR)
+    logging.getLogger('gql').setLevel(logging.WARNING)
 
 
 def main():

c7n/filters/cost.py

@@ -0,0 +1,107 @@
+import os
+
+from c7n.cache import NullCache
+from gql import Client, gql
+from gql.transport.requests import RequestsHTTPTransport
+
+from .core import OPERATORS, Filter
+
+
+class Cost(Filter):
+    """Annotate resource monthly cost with Infracost pricing API.
+    It aims to provide an approximate cost for a generic case.
+    For example, it only grabs price of on-demand, no pre-installed softwares for EC2 instances.
+
+    Please use INFRACOST_API_ENDPOINT and INFRACOST_API_KEY to specify API config.
+
+    .. code-block:: yaml
+
+    policies:
+      - name: ec2-cost
+        resource: ec2
+        filters:
+          - type: cost
+            op: greater-than
+            value: 4
+            quantity: 730
+
+
+    reference: https://www.infracost.io/docs/cloud_pricing_api/overview/
+    """
+
+    schema = {
+        'type': 'object',
+        'additionalProperties': False,
+        'required': ['type'],
+        'properties': {
+            'api_endpoint': {'type': 'string'},
+            'api_key': {'type': 'string'},
+            # 'currency': {'type': 'number'},
+            'quantity': {'type': 'number'},
+            'op': {'$ref': '#/definitions/filters_common/comparison_operators'},
+            'type': {'enum': ['cost']},
+            'value': {'type': 'number'},
+        },
+    }
+    schema_alias = True
+    ANNOTATION_KEY = "c7n:Cost"
+
+    def __init__(self, data, manager=None):
+        super().__init__(data, manager)
+        self.cache = manager._cache or NullCache({})
+        self.api_endpoint = data.get(
+            "api_endpoint",
+            os.environ.get("INFRACOST_API_ENDPOINT", "https://pricing.api.infracost.io"),
+        )
+        self.api_key = data.get("api_key", os.environ.get("INFRACOST_API_KEY"))
+
+    def get_permissions(self):
+        return ("ec2:DescribeInstances",)
+
+    def validate(self):
+        name = self.__class__.__name__
+        if self.api_endpoint is None:
+            raise ValueError("%s Filter requires Infracost pricing_api_endpoint" % name)
+
+        if self.api_key is None:
+            raise ValueError("%s Filter requires Infracost api_key" % name)
+        return super(Cost, self).validate()
+
+    def process_resource(self, resource, client, query):
+        params = self.get_params(resource)
+        cache_key = str(params)
+
+        with self.cache:
+            price = self.cache.get(cache_key)
+            if not price:
+                price = self.get_infracost(client, query, params)
+                # TODO support configurable currency
+                price["USD"] = float(price["USD"]) * self.data.get("quantity", 1)
+                self.cache.save(cache_key, price)
+
+        resource[self.ANNOTATION_KEY] = price
+        op = self.data.get('operator', 'ge')
+        value = self.data.get('value', -1)
+        return OPERATORS[op](price["USD"], value)
+
+    def get_infracost(self, client, query, params):
+        result = client.execute(query, variable_values=params)
+        self.log.info(f"Infracost {params}: {result}")
+        return result["products"][0]["prices"][0]
+
+    def process(self, resources, event=None):
+        transport = RequestsHTTPTransport(
+            url=self.api_endpoint + "/graphql",
+            headers={'X-Api-Key': self.api_key},
+            verify=True,
+            retries=3,
+        )
+        client = Client(transport=transport, fetch_schema_from_transport=True)
+        query = gql(self.get_query())
+        return [r for r in resources if self.process_resource(r, client, query)]
+
+    def get_query(self):
+        raise NotImplementedError("use subclass")
+
+    def get_params(self, resource):
+        raise NotImplementedError("use subclass")

c7n/resources/ec2.py

@@ -23,6 +23,7 @@
 from c7n.filters import (
     FilterRegistry, AgeFilter, ValueFilter, Filter, DefaultVpcBase
 )
+from c7n.filters.cost import Cost
 from c7n.filters.offhours import OffHour, OnHour
 import c7n.filters.vpc as net_filters
 
@@ -177,6 +178,43 @@ class VpcFilter(net_filters.VpcFilter):
     RelatedIdsExpression = "VpcId"
 
 
+@filters.register('cost')
+class Ec2Cost(Cost):
+
+    def get_query(self):
+        # reference: https://gql.readthedocs.io/en/stable/usage/variables.html
+        return """
+            query ($region: String, $instanceType: String) {
+                products(
+                    filter: {
+                        vendorName: "aws",
+                        service: "AmazonEC2",
+                        productFamily: "Compute Instance",
+                        region: $region,
+                        attributeFilters: [
+                            { key: "instanceType", value: $instanceType }
+                            { key: "operatingSystem", value: "Linux" }
+                            { key: "tenancy", value: "Shared" }
+                            { key: "capacitystatus", value: "Used" }
+                            { key: "preInstalledSw", value: "NA" }
+                        ]
+                    },
+                ) {
+                    prices(
+                        filter: {purchaseOption: "on_demand"}
+                    ) { USD, unit, description, purchaseOption }
+                }
+            }
+        """
+
+    def get_params(self, resource):
+        params = {
+            "region": resource["Placement"]["AvailabilityZone"][:-1],
+            "instanceType": resource["InstanceType"],
+        }
+        return params
+
+
 @filters.register('check-permissions')
 class ComputePermissions(CheckPermissions):