incluster detection

kube_ops_view/cluster_discovery.py

@@ -0,0 +1,64 @@
+import kubernetes.client
+import kubernetes.config
+import tokens
+from requests.auth import AuthBase
+
+DEFAULT_CLUSTERS = 'http://localhost:8001/'
+
+tokens.configure(from_file_only=True)
+
+
+class StaticTokenAuth(AuthBase):
+    def __init__(self, token):
+        self.token = token
+
+    def __call__(self, request):
+        request.headers['Authorization'] = 'Bearer {}'.format(self.token)
+
+
+class OAuthTokenAuth(AuthBase):
+    def __init__(self, token_name):
+        self.token_name = token_name
+        tokens.manage(token_name)
+
+    def __call__(self, request):
+        token = tokens.get(self.token_name)
+        request.headers['Authorization'] = 'Bearer {}'.format(token)
+
+
+class Cluster:
+    def __init__(self, api_server_url, ssl_ca_cert=None, auth=None):
+        self.api_server_url = api_server_url
+        self.ssl_ca_cert = ssl_ca_cert
+        self.auth = auth
+
+
+class StaticClusterDiscoverer:
+
+    def __init__(self, api_server_urls):
+        self._clusters = []
+
+        if not api_server_urls:
+            try:
+                kubernetes.config.load_incluster_config()
+            except kubernetes.config.ConfigException:
+                cluster = Cluster('http://localhost:8001')
+            else:
+                config = kubernetes.client.configuration
+                cluster = Cluster(
+                    config.host,
+                    ssl_ca_cert=config.ssl_ca_cert,
+                    auth=StaticTokenAuth(config.api_key['authorization'].split(' ', 1)[-1]))
+            self._clusters.append(cluster)
+        else:
+            for api_server_url in api_server_urls:
+
+                if 'localhost' not in api_server_url:
+                    # TODO: hacky way of detecting whether we need a token or not
+                    auth = OAuthTokenAuth('read-only')
+                else:
+                    auth = None
+                self._clusters.append(Cluster(api_server_url, auth=auth))
+
+    def get_clusters(self):
+        return self._clusters

kube_ops_view/kubernetes.py

@@ -4,13 +4,9 @@ import re
 from urllib.parse import urljoin
 
 import requests
-import tokens
 
 CLUSTER_ID_INVALID_CHARS = re.compile('[^a-z0-9:-]')
 
-tokens.configure(from_file_only=True)
-tokens.manage('read-only')
-
 session = requests.Session()
 
 
@@ -57,13 +53,11 @@ def map_container(cont: dict, pod: dict):
     return obj
 
 
-def get_kubernetes_clusters(clusters):
-    for api_server_url in clusters:
+def get_kubernetes_clusters(cluster_discoverer):
+    for cluster in cluster_discoverer.get_clusters():
+        api_server_url = cluster.api_server_url
         cluster_id = generate_cluster_id(api_server_url)
-        if 'localhost' not in api_server_url:
-            # TODO: hacky way of detecting whether we need a token or not
-            session.headers['Authorization'] = 'Bearer {}'.format(tokens.get('read-only'))
-        response = session.get(urljoin(api_server_url, '/api/v1/nodes'), timeout=5)
+        response = session.get(urljoin(api_server_url, '/api/v1/nodes'), timeout=5, auth=cluster.auth, verify=cluster.ssl_ca_cert)
         response.raise_for_status()
         nodes = {}
         pods_by_namespace_name = {}

kube_ops_view/main.py

@@ -26,11 +26,11 @@ 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
 
 
 logger = logging.getLogger(__name__)
 
-DEFAULT_CLUSTERS = 'http://localhost:8001/'
 SERVER_STATUS = {'shutdown': False}
 AUTHORIZE_URL = os.getenv('AUTHORIZE_URL')
 APP_URL = os.getenv('APP_URL')
@@ -155,7 +155,7 @@ def authorized():
     return redirect(urljoin(APP_URL, '/'))
 
 
-def update(clusters: list, store, mock: bool):
+def update(cluster_discoverer, store, mock: bool):
     while True:
         lock = store.acquire_lock()
         if lock:
@@ -163,7 +163,7 @@ def update(clusters: list, store, mock: bool):
                 if mock:
                     _clusters = get_mock_clusters()
                 else:
-                    _clusters = get_kubernetes_clusters(clusters)
+                    _clusters = get_kubernetes_clusters(cluster_discoverer)
                 cluster_ids = []
                 for cluster in _clusters:
                     old_data = store.get(cluster['id'])
@@ -215,7 +215,7 @@ def print_version(ctx, param, value):
 @click.option('--secret-key', help='Secret key for session cookies', envvar='SECRET_KEY', default='development')
 @click.option('--redis-url', help='Redis URL to use for pub/sub and job locking', envvar='REDIS_URL')
 @click.option('--clusters', help='Comma separated list of Kubernetes API server URLs (default: {})'.format(DEFAULT_CLUSTERS),
-              envvar='CLUSTERS', default=DEFAULT_CLUSTERS)
+              envvar='CLUSTERS')
 def main(port, debug, mock, secret_key, redis_url, clusters):
     logging.basicConfig(level=logging.DEBUG if debug else logging.INFO)
 
@@ -225,8 +225,10 @@ def main(port, debug, mock, secret_key, redis_url, clusters):
     app.secret_key = secret_key
     app.store = store
 
+    api_server_urls = clusters.split(',') if clusters else []
+    gevent.spawn(update, cluster_discoverer=StaticClusterDiscoverer(api_server_urls), store=store, mock=mock)
+
     signal.signal(signal.SIGTERM, exit_gracefully)
     http_server = gevent.wsgi.WSGIServer(('0.0.0.0', port), app)
-    gevent.spawn(update, clusters=clusters.split(','), store=store, mock=mock)
     logger.info('Listening on :{}..'.format(port))
     http_server.serve_forever()

requirements.txt

@@ -6,3 +6,4 @@ requests
 stups-tokens>=1.1.19
 redlock-py
 json_delta>=2.0
+kubernetes