Merge pull request #821 from philipp-sontag-by/handle-list-objs-errors

Handle connection errors during list_obj calls gracefully

Author: nolar

kopf/_cogs/clients/watching.py

@@ -155,14 +155,18 @@ async def continuous_watch(
 
     # First, list the resources regularly, and get the list's resource version.
     # Simulate the events with type "None" event - used in detection of causes.
-    objs, resource_version = await fetching.list_objs(
-        logger=logger,
-        settings=settings,
-        resource=resource,
-        namespace=namespace,
-    )
-    for obj in objs:
-        yield {'type': None, 'object': obj}
+    try:
+        objs, resource_version = await fetching.list_objs(
+            logger=logger,
+            settings=settings,
+            resource=resource,
+            namespace=namespace,
+        )
+        for obj in objs:
+            yield {'type': None, 'object': obj}
+
+    except (aiohttp.ClientConnectionError, aiohttp.ClientPayloadError, asyncio.TimeoutError):
+        return
 
     # Notify the watcher that the initial listing is over, even if there was nothing yielded.
     yield Bookmark.LISTED

tests/k8s/test_watching_continuously.py

@@ -11,6 +11,7 @@
 import asyncio
 import logging
 
+import aiohttp
 import pytest
 
 from kopf._cogs.clients.watching import Bookmark, WatchingError, continuous_watch
@@ -180,3 +181,26 @@ async def test_long_line_parsing(
     assert len(events[1]['object']['spec']['field']) == 1
     assert len(events[2]['object']['spec']['field']) == 2 * 1024 * 1024
     assert len(events[3]['object']['spec']['field']) == 4 * 1024 * 1024
+
+@pytest.mark.parametrize("connection_error",
+    [
+        aiohttp.ClientConnectionError,
+        aiohttp.ClientPayloadError,
+        asyncio.TimeoutError
+    ]
+)
+async def test_list_objs_connection_errors_are_caught(
+        settings, resource, stream, namespace, enforced_session, mocker, connection_error):
+
+    enforced_session.request = mocker.Mock(side_effect=connection_error())
+    stream.feed([], namespace=namespace)
+    stream.close(namespace=namespace)
+
+    events = []
+    async for event in continuous_watch(settings=settings,
+                                            resource=resource,
+                                            namespace=namespace,
+                                            operator_pause_waiter=asyncio.Future()):
+        events.append(event)
+
+    assert len(events) == 0