Filter out duplicated status

pull/22028/head
Lidi Zheng 5 years ago
parent a729bd1828
commit 750b602118
  1. 17
      src/python/grpcio_health_checking/grpc_health/v1/_async.py
  2. 2
      src/python/grpcio_tests/tests_aio/health_check/BUILD.bazel
  3. 28
      src/python/grpcio_tests/tests_aio/health_check/health_servicer_test.py

@ -40,6 +40,7 @@ class AsyncHealthServicer(_health_pb2_grpc.HealthServicer):
async def Watch(self, request: _health_pb2.HealthCheckRequest, context):
condition = self._server_watchers[request.service]
last_status = None
try:
async with condition:
while True:
@ -47,14 +48,22 @@ class AsyncHealthServicer(_health_pb2_grpc.HealthServicer):
request.service,
_health_pb2.HealthCheckResponse.SERVICE_UNKNOWN)
# Responds with current health state
await context.write(
_health_pb2.HealthCheckResponse(status=status))
# NOTE(lidiz) If the observed status is the same, it means
# intermediate statuses has been discarded. It's consider
# acceptable since peer only interested in eventual status.
if status != last_status:
# Responds with current health state
await context.write(
_health_pb2.HealthCheckResponse(status=status))
# Records the last sent status
last_status = status
# Polling on health state changes
await condition.wait()
finally:
del self._server_watchers[request.service]
if request.service in self._server_watchers:
del self._server_watchers[request.service]
async def _set(self, service: str,
status: _health_pb2.HealthCheckResponse.ServingStatus):

@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
package(default_visibility = ["//visibility:public"])
package(default_testonly = 1)
py_test(
name = "health_servicer_test",

@ -16,6 +16,7 @@
import asyncio
import logging
import time
import random
import unittest
import grpc
@ -34,6 +35,8 @@ _UNKNOWN_SERVICE = 'grpc.test.TestServiceUnknown'
_NOT_SERVING_SERVICE = 'grpc.test.TestServiceNotServing'
_WATCH_SERVICE = 'grpc.test.WatchService'
_LARGE_NUMBER_OF_STATUS_CHANGE = 1000
async def _pipe_to_queue(call, queue):
async for response in call:
@ -226,6 +229,31 @@ class HealthServicerTest(AioTestBase):
await task
self.assertTrue(queue.empty())
async def test_no_duplicate_status(self):
request = health_pb2.HealthCheckRequest(service=_WATCH_SERVICE)
call = self._stub.Watch(request)
queue = asyncio.Queue()
task = self.loop.create_task(_pipe_to_queue(call, queue))
self.assertEqual(health_pb2.HealthCheckResponse.SERVICE_UNKNOWN,
(await queue.get()).status)
last_status = health_pb2.HealthCheckResponse.SERVICE_UNKNOWN
for _ in range(_LARGE_NUMBER_OF_STATUS_CHANGE):
if random.randint(0, 1) == 0:
status = health_pb2.HealthCheckResponse.SERVING
else:
status = health_pb2.HealthCheckResponse.NOT_SERVING
await self._servicer.set(_WATCH_SERVICE, status)
if status != last_status:
self.assertEqual(status, (await queue.get()).status)
last_status = status
call.cancel()
await task
self.assertTrue(queue.empty())
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG)

Loading…
Cancel
Save