get-k8s-resources/get-k8s-resources.py

#!/usr/bin/env python

# Note: This license has also been called the "Simplified BSD License"
# and the "FreeBSD License". See also the 3-clause BSD License.
#
# Copyright 2019 Elijah Lazkani
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# 1. Redistributions of source code must retain the above copyright
#    notice, this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above
#    copyright notice, this list of conditions and the following
#    disclaimer in the documentation and/or other materials provided
#    with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

import sys
import json
import argparse
from kubernetes import client, config

from kubernetes.config.config_exception import ConfigException


def get_deployments(api, namespace):
    if namespace:
        deployments = api.list_namespaced_deployment(namespace=namespace)
    else:
        deployments = api.list_deployment_for_all_namespaces()

    return deployments


def get_pods(api, namespace):
    if namespace:
        pods = api.list_namespaced_pod(namespace=namespace)
    else:
        pods = api.list_pod_for_all_namespaces()

    return pods


def get_services(api, namespace):
    if namespace:
        services = api.list_namespaced_service(namespace=namespace)
    else:
        services = api.list_service_for_all_namespaces()

    return services


def get_ingresses(api, namespace):
    if namespace:
        ingresses = api.list_namespaced_ingress(namespace=namespace)
    else:
        ingresses = api.list_ingress_for_all_namespaces()

    return ingresses


def json_dump(object_list):
    return json.dumps(object_list)


def print_list(k8s_object_list):
    _k8s_object_list = []
    for k8s_object in k8s_object_list.items:
        _k8s_object_list.append(k8s_object.metadata.name)

    print(json_dump(_k8s_object_list))


def get_extensions(api):
    return client.ExtensionsV1beta1Api(api)


def get_core_api_v1(api):
    return client.CoreV1Api(api)


def get_client(kubeconfig, context):
    try:
        return config.new_client_from_config(
            config_file=kubeconfig, context=context
        )
    except ConfigException as e:
        print(f"Error: {e}")
        sys.exit(1)


k8s_types = {
    "deployments": {"function": get_deployments, "api": get_extensions},
    "services": {"function": get_services, "api": get_core_api_v1},
    "pods": {"function": get_pods, "api": get_core_api_v1},
    "ingresses": {"function": get_ingresses, "api": get_extensions},
}


def main():
    parser = argumentparser()
    _client = get_client(parser.kubeconfig, parser.context)
    api = None
    if parser.type in k8s_types.keys():
        api = k8s_types[parser.type]["api"](_client)
        k8s_object_list = k8s_types[parser.type]["function"](
            api, parser.namespace
        )
        print_list(k8s_object_list)


def argumentparser():
    parser = argparse.ArgumentParser(
        description="Returns a json list of kubernetes deployments."
    )
    parser.add_argument(
        "-n", "--namespace", default=None, help="filter by namespace"
    )
    parser.add_argument("-c", "--context", default=None, help="context to use")
    parser.add_argument(
        "-t",
        "--type",
        required=True,
        choices=k8s_types.keys(),
        help="type of kubernetes resource to get",
    )
    parser.add_argument(
        "--kubeconfig", default=None, help="path to the kubeconfig"
    )
    return parser.parse_args()


if __name__ == "__main__":
    main()