#95 add --kubeconfig-path and KubeconfigDiscoverer

kube_ops_view/cluster_discovery.py

@@ -1,4 +1,5 @@
 import time
+from pathlib import Path
 from urllib.parse import urljoin
 
 import kubernetes.client
@@ -116,3 +117,23 @@ class ClusterRegistryDiscoverer:
         if now - self._last_cache_refresh > self._cache_lifetime:
             self.refresh()
         return self._clusters
+
+
+class KubeconfigDiscoverer:
+
+    def __init__(self, kubeconfig_path: Path):
+        self._path = kubeconfig_path
+
+    def get_clusters(self):
+        # Kubernetes Python client expects "vintage" string path
+        config_file = str(self._path)
+        contexts, current_context = kubernetes.config.list_kube_config_contexts(config_file)
+        for context in contexts:
+            config = kubernetes.client.ConfigurationObject()
+            kubernetes.config.load_kube_config(config_file, context=context['name'], client_configuration=config)
+            cluster = Cluster(
+                context['name'],
+                config.host,
+                ssl_ca_cert=config.ssl_ca_cert,
+                auth=StaticTokenAuth(config.api_key['authorization'].split(' ', 1)[-1]))
+            yield cluster

kube_ops_view/kubernetes.py

@@ -51,60 +51,68 @@ def request(cluster, path, **kwargs):
     return session.get(urljoin(cluster.api_server_url, path), auth=cluster.auth, verify=cluster.ssl_ca_cert, **kwargs)
 
 
+def get_kubernetes_cluster(cluster):
+    cluster_id = cluster.id
+    api_server_url = cluster.api_server_url
+    response = request(cluster, '/api/v1/nodes')
+    response.raise_for_status()
+    nodes = {}
+    pods_by_namespace_name = {}
+    unassigned_pods = {}
+    for node in response.json()['items']:
+        obj = map_node(node)
+        nodes[obj['name']] = obj
+    response = request(cluster, '/api/v1/pods')
+    response.raise_for_status()
+    for pod in response.json()['items']:
+        obj = map_pod(pod)
+        if 'deletionTimestamp' in pod['metadata']:
+            obj['deleted'] = datetime.datetime.strptime(pod['metadata']['deletionTimestamp'],
+                                                        '%Y-%m-%dT%H:%M:%SZ').replace(
+                tzinfo=datetime.timezone.utc).timestamp()
+        for cont in pod['spec']['containers']:
+            obj['containers'].append(map_container(cont, pod))
+        pods_by_namespace_name[(obj['namespace'], obj['name'])] = obj
+        pod_key = '{}/{}'.format(obj['namespace'], obj['name'])
+        if 'nodeName' in pod['spec'] and pod['spec']['nodeName'] in nodes:
+            nodes[pod['spec']['nodeName']]['pods'][pod_key] = obj
+        else:
+            unassigned_pods[pod_key] = obj
+
+    try:
+        response = request(cluster, '/api/v1/namespaces/kube-system/services/heapster/proxy/apis/metrics/v1alpha1/nodes')
+        response.raise_for_status()
+        data = response.json()
+        if not data.get('items'):
+            logger.info('Heapster node metrics not available (yet)')
+        else:
+            for metrics in data['items']:
+                nodes[metrics['metadata']['name']]['usage'] = metrics['usage']
+    except:
+        logger.exception('Failed to get node metrics')
+    try:
+        response = request(cluster, '/api/v1/namespaces/kube-system/services/heapster/proxy/apis/metrics/v1alpha1/pods')
+        response.raise_for_status()
+        data = response.json()
+        if not data.get('items'):
+            logger.info('Heapster pod metrics not available (yet)')
+        else:
+            for metrics in data['items']:
+                pod = pods_by_namespace_name.get((metrics['metadata']['namespace'], metrics['metadata']['name']))
+                if pod:
+                    for container in pod['containers']:
+                        for container_metrics in metrics['containers']:
+                            if container['name'] == container_metrics['name']:
+                                container['resources']['usage'] = container_metrics['usage']
+    except:
+        logger.exception('Failed to get pod metrics')
+    return {'id': cluster_id, 'api_server_url': api_server_url, 'nodes': nodes, 'unassigned_pods': unassigned_pods}
+
+
 def get_kubernetes_clusters(cluster_discoverer):
     for cluster in cluster_discoverer.get_clusters():
-        cluster_id = cluster.id
-        api_server_url = cluster.api_server_url
-        response = request(cluster, '/api/v1/nodes')
-        response.raise_for_status()
-        nodes = {}
-        pods_by_namespace_name = {}
-        unassigned_pods = {}
-        for node in response.json()['items']:
-            obj = map_node(node)
-            nodes[obj['name']] = obj
-        response = request(cluster, '/api/v1/pods')
-        response.raise_for_status()
-        for pod in response.json()['items']:
-            obj = map_pod(pod)
-            if 'deletionTimestamp' in pod['metadata']:
-                obj['deleted'] = datetime.datetime.strptime(pod['metadata']['deletionTimestamp'],
-                                                            '%Y-%m-%dT%H:%M:%SZ').replace(
-                    tzinfo=datetime.timezone.utc).timestamp()
-            for cont in pod['spec']['containers']:
-                obj['containers'].append(map_container(cont, pod))
-            pods_by_namespace_name[(obj['namespace'], obj['name'])] = obj
-            pod_key = '{}/{}'.format(obj['namespace'], obj['name'])
-            if 'nodeName' in pod['spec'] and pod['spec']['nodeName'] in nodes:
-                nodes[pod['spec']['nodeName']]['pods'][pod_key] = obj
-            else:
-                unassigned_pods[pod_key] = obj
-
         try:
-            response = request(cluster, '/api/v1/namespaces/kube-system/services/heapster/proxy/apis/metrics/v1alpha1/nodes')
+            data = get_kubernetes_cluster(cluster)
-            response.raise_for_status()
+            yield data
-            data = response.json()
-            if not data.get('items'):
-                logger.info('Heapster node metrics not available (yet)')
-            else:
-                for metrics in data['items']:
-                    nodes[metrics['metadata']['name']]['usage'] = metrics['usage']
         except:
-            logger.exception('Failed to get node metrics')
+            logger.exception('Failed to query cluster {} ({})'.format(cluster.id, cluster.api_server_url))
-        try:
-            response = request(cluster, '/api/v1/namespaces/kube-system/services/heapster/proxy/apis/metrics/v1alpha1/pods')
-            response.raise_for_status()
-            data = response.json()
-            if not data.get('items'):
-                logger.info('Heapster pod metrics not available (yet)')
-            else:
-                for metrics in data['items']:
-                    pod = pods_by_namespace_name.get((metrics['metadata']['namespace'], metrics['metadata']['name']))
-                    if pod:
-                        for container in pod['containers']:
-                            for container_metrics in metrics['containers']:
-                                if container['name'] == container_metrics['name']:
-                                    container['resources']['usage'] = container_metrics['usage']
-        except:
-            logger.exception('Failed to get pod metrics')
-        yield {'id': cluster_id, 'api_server_url': api_server_url, 'nodes': nodes, 'unassigned_pods': unassigned_pods}

kube_ops_view/main.py

@@ -26,7 +26,7 @@ from urllib.parse import urljoin
 from .mock import get_mock_clusters
 from .kubernetes import get_kubernetes_clusters
 from .stores import MemoryStore, RedisStore
-from .cluster_discovery import DEFAULT_CLUSTERS, StaticClusterDiscoverer, ClusterRegistryDiscoverer
+from .cluster_discovery import DEFAULT_CLUSTERS, StaticClusterDiscoverer, ClusterRegistryDiscoverer, KubeconfigDiscoverer
 
 
 logger = logging.getLogger(__name__)
@@ -217,7 +217,8 @@ def print_version(ctx, param, value):
 @click.option('--clusters', help='Comma separated list of Kubernetes API server URLs (default: {})'.format(DEFAULT_CLUSTERS),
               envvar='CLUSTERS')
 @click.option('--cluster-registry-url', help='URL to cluster registry', envvar='CLUSTER_REGISTRY_URL')
-def main(port, debug, mock, secret_key, redis_url, clusters, cluster_registry_url):
+@click.option('--kubeconfig-path', type=click.Path(exists=True), help='Path to kubeconfig file', envvar='KUBECONFIG_PATH')
+def main(port, debug, mock, secret_key, redis_url, clusters, cluster_registry_url, kubeconfig_path):
     logging.basicConfig(level=logging.DEBUG if debug else logging.INFO)
 
     store = RedisStore(redis_url) if redis_url else MemoryStore()
@@ -228,6 +229,8 @@ def main(port, debug, mock, secret_key, redis_url, clusters, cluster_registry_ur
 
     if cluster_registry_url:
         discoverer = ClusterRegistryDiscoverer(cluster_registry_url)
+    elif kubeconfig_path:
+        discoverer = KubeconfigDiscoverer(Path(kubeconfig_path))
     else:
         api_server_urls = clusters.split(',') if clusters else []
         discoverer = StaticClusterDiscoverer(api_server_urls)