mirror of
https://github.com/jupyterhub/jupyterhub.git
synced 2025-10-18 07:23:00 +00:00
Move exponential backoff into a function
Also use the 'Full Jitter' jitter algorithm from https://www.awsarchitectureblog.com/2015/03/backoff.html
This commit is contained in:
@@ -20,7 +20,7 @@ from .. import __version__
|
|||||||
from .. import orm
|
from .. import orm
|
||||||
from ..objects import Server
|
from ..objects import Server
|
||||||
from ..spawner import LocalProcessSpawner
|
from ..spawner import LocalProcessSpawner
|
||||||
from ..utils import url_path_join, DT_SCALE
|
from ..utils import url_path_join, exponential_backoff
|
||||||
|
|
||||||
# pattern for the authentication token header
|
# pattern for the authentication token header
|
||||||
auth_header_pat = re.compile(r'^(?:token|bearer)\s+([^\s]+)$', flags=re.IGNORECASE)
|
auth_header_pat = re.compile(r'^(?:token|bearer)\s+([^\s]+)$', flags=re.IGNORECASE)
|
||||||
@@ -597,7 +597,7 @@ class UserSpawnHandler(BaseHandler):
|
|||||||
# record redirect count in query parameter
|
# record redirect count in query parameter
|
||||||
if redirects:
|
if redirects:
|
||||||
self.log.warning("Redirect loop detected on %s", self.request.uri)
|
self.log.warning("Redirect loop detected on %s", self.request.uri)
|
||||||
yield gen.sleep(min(1 * (DT_SCALE ** redirects), 10))
|
yield gen.sleep(min(1 * (2 ** redirects), 10))
|
||||||
# rewrite target url with new `redirects` query value
|
# rewrite target url with new `redirects` query value
|
||||||
url_parts = urlparse(target)
|
url_parts = urlparse(target)
|
||||||
query_parts = parse_qs(url_parts.query)
|
query_parts = parse_qs(url_parts.query)
|
||||||
|
@@ -26,7 +26,7 @@ from traitlets import (
|
|||||||
|
|
||||||
from .objects import Server
|
from .objects import Server
|
||||||
from .traitlets import Command, ByteSpecification
|
from .traitlets import Command, ByteSpecification
|
||||||
from .utils import random_port, url_path_join, DT_MIN, DT_MAX, DT_SCALE
|
from .utils import random_port, url_path_join, exponential_backoff
|
||||||
|
|
||||||
|
|
||||||
class Spawner(LoggingConfigurable):
|
class Spawner(LoggingConfigurable):
|
||||||
@@ -666,21 +666,19 @@ class Spawner(LoggingConfigurable):
|
|||||||
self.log.exception("Unhandled error in poll callback for %s", self)
|
self.log.exception("Unhandled error in poll callback for %s", self)
|
||||||
return status
|
return status
|
||||||
|
|
||||||
death_interval = Float(DT_MIN)
|
|
||||||
|
|
||||||
@gen.coroutine
|
@gen.coroutine
|
||||||
def wait_for_death(self, timeout=10):
|
def wait_for_death(self, timeout=10):
|
||||||
"""Wait for the single-user server to die, up to timeout seconds"""
|
"""Wait for the single-user server to die, up to timeout seconds"""
|
||||||
loop = IOLoop.current()
|
@gen.coroutine
|
||||||
tic = loop.time()
|
def _wait_for_death():
|
||||||
dt = self.death_interval
|
|
||||||
while dt > 0:
|
|
||||||
status = yield self.poll()
|
status = yield self.poll()
|
||||||
if status is not None:
|
return status is not None
|
||||||
break
|
|
||||||
else:
|
yield exponential_backoff(
|
||||||
yield gen.sleep(dt)
|
_wait_for_death,
|
||||||
dt = min(dt * DT_SCALE, DT_MAX, timeout - (loop.time() - tic))
|
'Process did not die in {timeout} seconds'.format(timeout=timeout),
|
||||||
|
timeout=timeout
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def _try_setcwd(path):
|
def _try_setcwd(path):
|
||||||
|
@@ -4,6 +4,7 @@
|
|||||||
# Distributed under the terms of the Modified BSD License.
|
# Distributed under the terms of the Modified BSD License.
|
||||||
|
|
||||||
from binascii import b2a_hex
|
from binascii import b2a_hex
|
||||||
|
import random
|
||||||
import errno
|
import errno
|
||||||
import hashlib
|
import hashlib
|
||||||
from hmac import compare_digest
|
from hmac import compare_digest
|
||||||
@@ -48,29 +49,62 @@ def can_connect(ip, port):
|
|||||||
else:
|
else:
|
||||||
return True
|
return True
|
||||||
|
|
||||||
# exponential falloff factors:
|
@gen.coroutine
|
||||||
# start at 100ms, falloff by 2x
|
def exponential_backoff(
|
||||||
# never longer than 5s
|
pass_func,
|
||||||
DT_MIN = 0.1
|
fail_message,
|
||||||
DT_SCALE = 2
|
start_wait=0.1,
|
||||||
DT_MAX = 5
|
scale_factor=2,
|
||||||
|
max_wait=5,
|
||||||
|
timeout=10,
|
||||||
|
*args, **kwargs):
|
||||||
|
"""
|
||||||
|
Exponentially backoff until pass_func is true.
|
||||||
|
|
||||||
|
This function will wait with exponential backoff + random jitter for as
|
||||||
|
many iterations as needed, with maximum timeout timeout. If pass_func is
|
||||||
|
still returning false at the end of timeout, a TimeoutError will be raised.
|
||||||
|
|
||||||
|
It'll start waiting at start_wait, scaling up by continuously multiplying itself
|
||||||
|
by scale_factor until pass_func returns true. It'll never wait for more than
|
||||||
|
max_wait seconds per iteration.
|
||||||
|
|
||||||
|
*args and **kwargs are passed to pass_func. pass_func maybe a future, although
|
||||||
|
that is not entirely recommended.
|
||||||
|
|
||||||
|
It'll return the value of pass_func when it's truthy!
|
||||||
|
"""
|
||||||
|
loop = ioloop.IOLoop.current()
|
||||||
|
start_tic = loop.time()
|
||||||
|
dt = start_wait
|
||||||
|
while True:
|
||||||
|
if (loop.time() - start_tic) > timeout:
|
||||||
|
# We time out!
|
||||||
|
break
|
||||||
|
ret = yield gen.maybe_future(pass_func(*args, **kwargs))
|
||||||
|
# Truthy!
|
||||||
|
if ret:
|
||||||
|
return ret
|
||||||
|
else:
|
||||||
|
yield gen.sleep(dt)
|
||||||
|
# Add some random jitter to improve performance
|
||||||
|
# This makes sure that we don't overload any single iteration
|
||||||
|
# of the tornado loop with too many things
|
||||||
|
# See https://www.awsarchitectureblog.com/2015/03/backoff.html
|
||||||
|
# for a good example of why and how this helps. We're using their
|
||||||
|
# full Jitter implementation equivalent.
|
||||||
|
dt = min(max_wait, random.uniform(0, dt * scale_factor))
|
||||||
|
raise TimeoutError(fail_message)
|
||||||
|
|
||||||
|
|
||||||
@gen.coroutine
|
@gen.coroutine
|
||||||
def wait_for_server(ip, port, timeout=10):
|
def wait_for_server(ip, port, timeout=10):
|
||||||
"""Wait for any server to show up at ip:port."""
|
"""Wait for any server to show up at ip:port."""
|
||||||
if ip in {'', '0.0.0.0'}:
|
if ip in {'', '0.0.0.0'}:
|
||||||
ip = '127.0.0.1'
|
ip = '127.0.0.1'
|
||||||
loop = ioloop.IOLoop.current()
|
yield exponential_backoff(
|
||||||
tic = loop.time()
|
lambda: can_connect(ip, port),
|
||||||
dt = DT_MIN
|
"Server at {ip}:{port} didn't respond in {timeout} seconds".format(ip=ip, port=port, timeout=timeout)
|
||||||
while dt > 0:
|
|
||||||
if can_connect(ip, port):
|
|
||||||
return
|
|
||||||
else:
|
|
||||||
yield gen.sleep(dt)
|
|
||||||
dt = min(dt * DT_SCALE, DT_MAX, timeout - (loop.time() - tic))
|
|
||||||
raise TimeoutError(
|
|
||||||
"Server at {ip}:{port} didn't respond in {timeout} seconds".format(**locals())
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -80,13 +114,12 @@ def wait_for_http_server(url, timeout=10):
|
|||||||
|
|
||||||
Any non-5XX response code will do, even 404.
|
Any non-5XX response code will do, even 404.
|
||||||
"""
|
"""
|
||||||
loop = ioloop.IOLoop.current()
|
|
||||||
tic = loop.time()
|
|
||||||
client = AsyncHTTPClient()
|
client = AsyncHTTPClient()
|
||||||
dt = DT_MIN
|
@gen.coroutine
|
||||||
while dt > 0:
|
def is_reachable():
|
||||||
try:
|
try:
|
||||||
r = yield client.fetch(url, follow_redirects=False)
|
r = yield client.fetch(url, follow_redirects=False)
|
||||||
|
return r
|
||||||
except HTTPError as e:
|
except HTTPError as e:
|
||||||
if e.code >= 500:
|
if e.code >= 500:
|
||||||
# failed to respond properly, wait and try again
|
# failed to respond properly, wait and try again
|
||||||
@@ -95,25 +128,21 @@ def wait_for_http_server(url, timeout=10):
|
|||||||
# but 502 or other proxy error is conceivable
|
# but 502 or other proxy error is conceivable
|
||||||
app_log.warning(
|
app_log.warning(
|
||||||
"Server at %s responded with error: %s", url, e.code)
|
"Server at %s responded with error: %s", url, e.code)
|
||||||
yield gen.sleep(dt)
|
|
||||||
else:
|
else:
|
||||||
app_log.debug("Server at %s responded with %s", url, e.code)
|
app_log.debug("Server at %s responded with %s", url, e.code)
|
||||||
return e.response
|
return e.response
|
||||||
except (OSError, socket.error) as e:
|
except (OSError, socket.error) as e:
|
||||||
if e.errno not in {errno.ECONNABORTED, errno.ECONNREFUSED, errno.ECONNRESET}:
|
if e.errno not in {errno.ECONNABORTED, errno.ECONNREFUSED, errno.ECONNRESET}:
|
||||||
app_log.warning("Failed to connect to %s (%s)", url, e)
|
app_log.warning("Failed to connect to %s (%s)", url, e)
|
||||||
yield gen.sleep(dt)
|
return False
|
||||||
else:
|
re = yield exponential_backoff(
|
||||||
return r
|
is_reachable,
|
||||||
dt = min(dt * DT_SCALE, DT_MAX, timeout - (loop.time() - tic))
|
"Server at {url} didn't respond in {timeout} seconds".format(url=url, timeout=timeout)
|
||||||
|
|
||||||
raise TimeoutError(
|
|
||||||
"Server at {url} didn't respond in {timeout} seconds".format(**locals())
|
|
||||||
)
|
)
|
||||||
|
return re
|
||||||
|
|
||||||
|
|
||||||
# Decorators for authenticated Handlers
|
# Decorators for authenticated Handlers
|
||||||
|
|
||||||
def auth_decorator(check_auth):
|
def auth_decorator(check_auth):
|
||||||
"""Make an authentication decorator.
|
"""Make an authentication decorator.
|
||||||
|
|
||||||
|
Reference in New Issue
Block a user