From 0c665e23392b1726116fe7452550ae9c46c6d451 Mon Sep 17 00:00:00 2001 From: Henning Jacobs Date: Sun, 15 Jan 2017 14:14:32 +0100 Subject: [PATCH] #95 add --kubeconfig-path and KubeconfigDiscoverer --- kube_ops_view/cluster_discovery.py | 21 ++++++ kube_ops_view/kubernetes.py | 114 +++++++++++++++-------------- kube_ops_view/main.py | 7 +- 3 files changed, 87 insertions(+), 55 deletions(-) diff --git a/kube_ops_view/cluster_discovery.py b/kube_ops_view/cluster_discovery.py index c70f676..f9e6812 100644 --- a/kube_ops_view/cluster_discovery.py +++ b/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 diff --git a/kube_ops_view/kubernetes.py b/kube_ops_view/kubernetes.py index de2b4b7..4c6655d 100644 --- a/kube_ops_view/kubernetes.py +++ b/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') - 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'] + data = get_kubernetes_cluster(cluster) + yield data 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') - yield {'id': cluster_id, 'api_server_url': api_server_url, 'nodes': nodes, 'unassigned_pods': unassigned_pods} + logger.exception('Failed to query cluster {} ({})'.format(cluster.id, cluster.api_server_url)) diff --git a/kube_ops_view/main.py b/kube_ops_view/main.py index 6a21333..7655523 100644 --- a/kube_ops_view/main.py +++ b/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)