limit concurrent outstanding health checks
when a second request for a given health check wrapped in `at_most_every` is submitted,
return the already outstanding future instead of starting a new check.
This should ensure that the `at_most_every` condition is satisfied
import asyncio
+import json
import time
from functools import wraps
@@ -61,15 +62,22 @@ def at_most_every(_f=None, *, interval=60):
"""
last_time = time.monotonic() - interval - 1
last_result = None
+ outstanding = None
def caller(f):
@wraps(f)
async def wrapper(*args, **kwargs):
- nonlocal last_time, last_result
+ nonlocal last_time, last_result, outstanding
+ if outstanding is not None:
+ # do not allow multiple concurrent calls, return an existing future
+ return await outstanding
now = time.monotonic()
if now > last_time + interval:
- last_result = await f(*args, **kwargs)
- last_time = now
+ outstanding = asyncio.ensure_future(f(*args, **kwargs))
+ last_result = await outstanding
+ # complete, clear outstanding future and note the time
+ outstanding = None
+ last_time = time.monotonic()
return last_result
return wrapper
@@ -93,34 +101,35 @@ class HealthHandler(BaseHandler):
@at_most_every
async def _get_pods(self):
"""Get information about build and user pods"""
- app_log.info("Getting pod statistics")
+ namespace = self.settings["build_namespace"]
k8s = self.settings["kubernetes_client"]
pool = self.settings["executor"]
- get_user_pods = asyncio.wrap_future(
- pool.submit(
- k8s.list_namespaced_pod,
- self.settings["build_namespace"],
- label_selector="app=jupyterhub,component=singleuser-server",
- )
- )
+ app_log.info(f"Getting pod statistics for {namespace}")
- get_build_pods = asyncio.wrap_future(
+ label_selectors = [
+ "app=jupyterhub,component=singleuser-server",
+ "component=binderhub-build",
+ ]
+ requests = [
+ asyncio.wrap_future(
pool.submit(
k8s.list_namespaced_pod,
- self.settings["build_namespace"],
- label_selector="component=binderhub-build",
+ namespace,
+ label_selector=label_selector,
+ _preload_content=False,
)
)
-
- return await asyncio.gather(get_user_pods, get_build_pods)
+ for label_selector in label_selectors
+ ]
+ responses = await asyncio.gather(*requests)
+ return [json.loads(resp.read())["items"] for resp in responses]
@false_if_raises
@retry
async def check_jupyterhub_api(self, hub_url):
"""Check JupyterHub API health"""
await AsyncHTTPClient().fetch(hub_url + "hub/health", request_timeout=3)
-
return True
@false_if_raises
@@ -142,8 +151,8 @@ class HealthHandler(BaseHandler):
"""Compare number of active pods to available quota"""
user_pods, build_pods = await self._get_pods()
- n_user_pods = len(user_pods.items)
- n_build_pods = len(build_pods.items)
+ n_user_pods = len(user_pods)
+ n_build_pods = len(build_pods)
quota = self.settings["pod_quota"]
total_pods = n_user_pods + n_build_pods
