Compare kubernetes manifest files

compare_charts.py

#!/usr/bin/env python3

import argparse
import difflib
import sys
from collections import defaultdict
from typing import Dict, List

import yaml

nested_dict = lambda: defaultdict(nested_dict)

def load_yaml(filename: str) -> List[Dict]:
    with open(filename) as f:
        result = yaml.load_all(f, Loader=yaml.Loader)
        return list(result)

def convert_to_dict(objects: List[Dict], default_ns: str) -> Dict:
    """
    Given a list of K8S objects, convert into a dict in the format:

    d[namespace][kind][name] = object_dict
    """
    result = nested_dict()
    for obj in objects:
        kind = obj["kind"]
        namespace = obj["metadata"].get("namespace", default_ns)
        name = obj["metadata"].get("name", "no_name")

        result[namespace][kind][name] = obj
    return result


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('before')
    parser.add_argument('after')
    parser.add_argument('-n', '--namespace', help="Default namespace")
    args = parser.parse_args()
    before = convert_to_dict(load_yaml(args.before), args.namespace)
    after = convert_to_dict(load_yaml(args.after), args.namespace)

    for namespace in before:
        print(f"Namespace {namespace} *************")
        for kind in before[namespace]:
            for name in before[namespace][kind]:
                if kind in after[namespace] and name in after[namespace][kind]:
                    print(f"Object {namespace}/{kind}/{name} ************")
                    before_obj = yaml.dump(before[namespace][kind][name], Dumper=yaml.Dumper).splitlines(keepends=True)
                    after_obj = yaml.dump(after[namespace][kind][name], Dumper=yaml.Dumper).splitlines(keepends=True)
                    diff = difflib.context_diff(before_obj, after_obj)
                    sys.stdout.writelines(diff)
                else:
                    print(f"Object {namespace}/{kind}/{name} deleted")

if __name__ == "__main__":
    main()