Generate helm values for Kubecost network cost allocations

kubecost-subnets.py

import socket

import requests
import yaml


AWS_IP_RANGES_URL = "https://ip-ranges.amazonaws.com/ip-ranges.json"


def kubecost_direct_classification(kops_subnet):
    # https://github.com/kubecost/cost-analyzer-helm-chart/blob/5eedab0433445a5b8e134113beb95f4598cd5e2d/cost-analyzer/values.yaml#L506-L513
    network = {
        "region": kops_subnet["zone"][:-1],
        "ips": [
            kops_subnet["cidr"],
        ],
    }
    zone = kops_subnet.get("zone")
    if zone:
        network["zone"] = zone
    return network


def network_sort_key(network):
    # ignores netmask
    return socket.inet_aton(network.split("/")[0])


class KubecostNetworks(object):
    def __init__(self):
        self.direct_classification = {}
        self.in_zone = set(
            [
                # Loopback
                "127.0.0.1",
                # IPv4 Link Local Address Space, AWS EC2 metadata API etc.
                "169.254.0.0/16",
            ]
        )

    def from_aws(self, match_services=["DYNAMODB", "S3"], region="us-east-1"):
        """Downloads the AWS IP ranges and classifies any matching service
        prefix in the same region as "in-zone" (free) or directly classifies
        for cross-region."""
        r = requests.get(AWS_IP_RANGES_URL)
        ip_ranges = r.json()
        for prefix in ip_ranges["prefixes"]:
            if match_services and prefix["service"] not in match_services:
                continue
            if prefix["region"] == region:
                self.in_zone.add(prefix["ip_prefix"])
                continue
            dc = {
                "region": prefix["region"],
                "ips": [
                    prefix["ip_prefix"],
                ],
            }
            if prefix["region"] in self.direct_classification:
                self.direct_classification[prefix["region"]]["ips"].append(*dc["ips"])
            else:
                self.direct_classification[prefix["region"]] = dc

    def from_kops(self, manifests):
        zones = self.direct_classification
        for manifest in manifests:
            subnets = yaml.safe_load(manifest).get("subnets")
            if not subnets:
                continue
            for subnet in subnets:
                dc = kubecost_direct_classification(subnet)
                zone = dc["zone"]
                if zone in zones:
                    zones[zone]["ips"].append(*dc["ips"])
                else:
                    zones[zone] = dc

    def _remove_duplicates(self):
        for network in self.direct_classification.values():
            network["ips"] = sorted(list(set(network["ips"])), key=network_sort_key)

    def as_kubecost_values(self):
        self._remove_duplicates()
        out = {"networkCosts": {"config": {"destinations": {}}}}
        direct_classification = list(self.direct_classification.values())
        if direct_classification:
            out["networkCosts"]["config"]["destinations"]["direct-classification"] = direct_classification
        if self.in_zone:
            out["networkCosts"]["config"]["destinations"]["in-zone"] = sorted(list(self.in_zone), key=network_sort_key)
        return yaml.dump(out)


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(
        description="Converts kops subnets to kubecost network cost classifications values.",
        epilog="""\
Typical usage:
  python scripts/kubecost-subnets.py ../kops/cluster-provisioning/env/*/*/*/manifest.yaml >> envs/kubecost-values.yaml
""",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "kops_manifests",
        help="the path to the kops manifests that define the list of subnets",
        nargs="*",
        type=argparse.FileType("r"),
    )
    args = parser.parse_args()
    kc = KubecostNetworks()
    kc.from_aws()
    if args.kops_manifests:
        kc.from_kops(args.kops_manifests)
    print(kc.as_kubecost_values(), end="")