Merge pull request #18112 from ericgribkoff/enter_graceful_shutdown

add enter_graceful_shutdown() to health service
pull/18098/head
Eric Gribkoff 6 years ago committed by GitHub
commit bba348c302
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 34
      src/python/grpcio_health_checking/grpc_health/v1/health.py
  2. 26
      src/python/grpcio_tests/tests/health_check/_health_servicer_test.py

@ -82,6 +82,7 @@ class HealthServicer(_health_pb2_grpc.HealthServicer):
self._send_response_callbacks = {}
self.Watch.__func__.experimental_non_blocking = experimental_non_blocking
self.Watch.__func__.experimental_thread_pool = experimental_thread_pool
self._gracefully_shutting_down = False
def _on_close_callback(self, send_response_callback, service):
@ -135,9 +136,30 @@ class HealthServicer(_health_pb2_grpc.HealthServicer):
the service
"""
with self._lock:
self._server_status[service] = status
if service in self._send_response_callbacks:
for send_response_callback in self._send_response_callbacks[
service]:
send_response_callback(
_health_pb2.HealthCheckResponse(status=status))
if self._gracefully_shutting_down:
return
else:
self._server_status[service] = status
if service in self._send_response_callbacks:
for send_response_callback in self._send_response_callbacks[
service]:
send_response_callback(
_health_pb2.HealthCheckResponse(status=status))
def enter_graceful_shutdown(self):
"""Permanently sets the status of all services to NOT_SERVING.
This should be invoked when the server is entering a graceful shutdown
period. After this method is invoked, future attempts to set the status
of a service will be ignored.
This is an EXPERIMENTAL API.
"""
with self._lock:
if self._gracefully_shutting_down:
return
else:
for service in self._server_status:
self.set(service,
_health_pb2.HealthCheckResponse.NOT_SERVING) # pylint: disable=no-member
self._gracefully_shutting_down = True

@ -194,7 +194,7 @@ class BaseWatchTests(object):
thread.join()
# Wait, if necessary, for serving thread to process client cancellation
timeout = time.time() + test_constants.SHORT_TIMEOUT
timeout = time.time() + test_constants.TIME_ALLOWANCE
while time.time(
) < timeout and self._servicer._send_response_callbacks[_WATCH_SERVICE]:
time.sleep(1)
@ -203,6 +203,30 @@ class BaseWatchTests(object):
'watch set should be empty')
self.assertTrue(response_queue.empty())
def test_graceful_shutdown(self):
request = health_pb2.HealthCheckRequest(service='')
response_queue = queue.Queue()
rendezvous = self._stub.Watch(request)
thread = threading.Thread(
target=_consume_responses, args=(rendezvous, response_queue))
thread.start()
response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT)
self.assertEqual(health_pb2.HealthCheckResponse.SERVING,
response.status)
self._servicer.enter_graceful_shutdown()
response = response_queue.get(timeout=test_constants.SHORT_TIMEOUT)
self.assertEqual(health_pb2.HealthCheckResponse.NOT_SERVING,
response.status)
# This should be a no-op.
self._servicer.set('', health_pb2.HealthCheckResponse.SERVING)
rendezvous.cancel()
thread.join()
self.assertTrue(response_queue.empty())
class HealthServicerTest(BaseWatchTests.WatchTests):

Loading…
Cancel
Save