Merge pull request #128 from minrk/share-apiclient

share kubernetes client instances

kubespawner/clients.py

@@ -0,0 +1,29 @@
+import weakref
+
+import kubernetes.client
+
+_client_cache = {}
+
+
+def shared_client(ClientType, *args, **kwargs):
+    """Return a single shared kubernetes client instance
+
+    A weak reference to the instance is cached,
+    so that concurrent calls to shared_client
+    will all return the same instance until
+    all references to the client are cleared.
+    """
+    kwarg_key = tuple((key, kwargs[key]) for key in sorted(kwargs))
+    cache_key = (ClientType, args, kwarg_key)
+    client = None
+    if cache_key in _client_cache:
+        # resolve cached weakref
+        # client can still be None after this!
+        client = _client_cache[cache_key]()
+
+    if client is None:
+        Client = getattr(kubernetes.client, ClientType)
+        client = Client(*args, **kwargs)
+        # cache weakref so that clients can be garbage collected
+        _client_cache[cache_key] = weakref.ref(client)
+    return client

kubespawner/proxy.py

@@ -1,3 +1,4 @@
+from concurrent.futures import ThreadPoolExecutor
 import os
 import string
 import escapism
@@ -10,7 +11,7 @@
 from kubespawner.objects import make_ingress
 from kubespawner.utils import generate_hashed_slug
 from kubespawner.reflector import NamespacedResourceReflector
-from concurrent.futures import ThreadPoolExecutor
+from .clients import shared_client
 from traitlets import Unicode
 from tornado import gen
 from tornado.concurrent import run_on_executor
@@ -96,8 +97,8 @@ def __init__(self, *args, **kwargs):
         self.service_reflector = ServiceReflector(parent=self, namespace=self.namespace)
         self.endpoint_reflector = EndpointsReflector(parent=self, namespace=self.namespace)
 
-        self.core_api = client.CoreV1Api()
-        self.extension_api = client.ExtensionsV1beta1Api()
+        self.core_api = shared_client('CoreV1Api')
+        self.extension_api = shared_client('ExtensionsV1beta1Api')
 
     @run_on_executor
     def asynchronize(self, method, *args, **kwargs):

kubespawner/spawner.py

@@ -6,12 +6,8 @@
 """
 import os
 import json
-import time
 import string
-import threading
-import sys
 from urllib.parse import urlparse, urlunparse
-import json
 import multiprocessing
 from concurrent.futures import ThreadPoolExecutor
 
@@ -26,6 +22,7 @@
 from kubernetes import client
 import escapism
 
+from .clients import shared_client
 from kubespawner.traitlets import Callable
 from kubespawner.utils import Callable
 from kubespawner.objects import make_pod, make_pvc
@@ -80,7 +77,7 @@ def on_reflector_failure():
                 on_failure=on_reflector_failure
             )
 
-        self.api = client.CoreV1Api()
+        self.api = shared_client('CoreV1Api')
 
         self.pod_name = self._expand_user_properties(self.pod_name_template)
         self.pvc_name = self._expand_user_properties(self.pvc_name_template)

setup.py

@@ -6,7 +6,7 @@
     install_requires=[
         'jupyterhub>=0.8',
         'pyYAML',
-        'kubernetes==3.*',
+        'kubernetes==4.*',
         'escapism',
     ],
     setup_requires=['pytest-runner'],