Merge pull request #558 from nolar/log-responses

Enrich the K8s API errors with information from K8s API itself

kopf/clients/discovery.py

@@ -2,7 +2,7 @@
 
 import aiohttp
 
-from kopf.clients import auth
+from kopf.clients import auth, errors
 from kopf.structs import resources
 
 
@@ -25,7 +25,7 @@ async def discover(
                     response = await context.session.get(
                         url=resource.get_version_url(server=context.server),
                     )
-                    response.raise_for_status()
+                    await errors.check_response(response)
                     respdata = await response.json()
 
                     context._discovered_resources[resource.api_version].update({

kopf/clients/errors.py

@@ -0,0 +1,49 @@
+import collections.abc
+import json
+
+import aiohttp
+
+
+class APIClientResponseError(aiohttp.ClientResponseError):
+    """
+    Same as :class:`aiohttp.ClientResponseError`, but with information from K8s.
+    """
+
+
+async def check_response(
+        response: aiohttp.ClientResponse,
+) -> None:
+    """
+    Check for specialised K8s errors, and raise with extended information.
+
+    Built-in aiohttp's errors only provide the rudimentary titles of the status
+    codes, but not the explanation why that error happened. K8s API provides
+    this information in the bodies of non-2xx responses. That information can
+    replace aiohttp's error messages to be more helpful.
+
+    However, the same error classes are used for now, to meet the expectations
+    if some routines in their ``except:`` clauses analysing for specific HTTP
+    statuses: e.g. 401 for re-login activities, 404 for patching/deletion, etc.
+    """
+    if response.status >= 400:
+        try:
+            payload = await response.json()
+        except (json.JSONDecodeError, aiohttp.ContentTypeError, aiohttp.ClientConnectionError):
+            payload = None
+
+        # Better be safe: who knows which sensitive information can be dumped unless kind==Status.
+        if not isinstance(payload, collections.abc.Mapping) or payload.get('kind') != 'Status':
+            payload = None
+
+        # If no information can be retrieved, fall back to the original error.
+        if payload is None:
+            response.raise_for_status()
+        else:
+            details = payload.get('details')
+            message = payload.get('message') or f"{details}"
+            raise APIClientResponseError(
+                response.request_info,
+                response.history,
+                status=response.status,
+                headers=response.headers,
+                message=message)

kopf/clients/events.py

@@ -5,7 +5,7 @@
 
 import aiohttp
 
-from kopf.clients import auth
+from kopf.clients import auth, errors
 from kopf.structs import bodies, resources
 
 logger = logging.getLogger(__name__)
@@ -78,7 +78,7 @@ async def post_event(
             headers={'Content-Type': 'application/json'},
             json=body,
         )
-        response.raise_for_status()
+        await errors.check_response(response)
 
     # Events are helpful but auxiliary, they should not fail the handling cycle.
     # Yet we want to notice that something went wrong (in logs).
@@ -93,4 +93,3 @@ async def post_event(
     except aiohttp.ClientOSError as e:
         logger.warning(f"Failed to post an event. Ignoring and continuing. "
                        f"Event: type={type!r}, reason={reason!r}, message={message!r}.")
-

kopf/clients/fetching.py

@@ -3,7 +3,7 @@
 
 import aiohttp
 
-from kopf.clients import auth, discovery
+from kopf.clients import auth, discovery, errors
 from kopf.structs import bodies, resources
 
 _T = TypeVar('_T')
@@ -29,7 +29,7 @@ async def read_crd(
         response = await context.session.get(
             url=CRD_CRD.get_url(server=context.server, name=resource.name),
         )
-        response.raise_for_status()
+        await errors.check_response(response)
         respdata = await response.json()
         return cast(bodies.RawBody, respdata)
 
@@ -58,7 +58,7 @@ async def read_obj(
         response = await context.session.get(
             url=resource.get_url(server=context.server, namespace=namespace, name=name),
         )
-        response.raise_for_status()
+        await errors.check_response(response)
         respdata = await response.json()
         return cast(bodies.RawBody, respdata)
 
@@ -96,7 +96,7 @@ async def list_objs_rv(
     response = await context.session.get(
         url=resource.get_url(server=context.server, namespace=namespace),
     )
-    response.raise_for_status()
+    await errors.check_response(response)
     rsp = await response.json()
 
     items: List[bodies.RawBody] = []

kopf/clients/patching.py

@@ -2,7 +2,7 @@
 
 import aiohttp
 
-from kopf.clients import auth, discovery
+from kopf.clients import auth, discovery, errors
 from kopf.structs import bodies, patches, resources
 
 
@@ -63,8 +63,8 @@ async def patch_obj(
                 url=resource.get_url(server=context.server, namespace=namespace, name=name),
                 headers={'Content-Type': 'application/merge-patch+json'},
                 json=body_patch,
-                raise_for_status=True,
             )
+            await errors.check_response(response)
             patched_body = await response.json()
 
         if status_patch:
@@ -73,8 +73,8 @@ async def patch_obj(
                                      subresource='status' if as_subresource else None),
                 headers={'Content-Type': 'application/merge-patch+json'},
                 json={'status': status_patch},
-                raise_for_status=True,
             )
+            await errors.check_response(response)
             patched_body['status'] = await response.json()
 
     except aiohttp.ClientResponseError as e:

kopf/clients/watching.py

@@ -26,7 +26,7 @@
 
 import aiohttp
 
-from kopf.clients import auth, discovery, fetching
+from kopf.clients import auth, discovery, errors, fetching
 from kopf.structs import bodies, configuration, primitives, resources
 from kopf.utilities import backports
 
@@ -215,7 +215,7 @@ async def watch_objs(
                 sock_connect=settings.watching.connect_timeout,
             ),
         )
-        response.raise_for_status()
+        await errors.check_response(response)
 
         response_close_callback = lambda _: response.close()
         freeze_waiter.add_done_callback(response_close_callback)

tests/k8s/test_errors.py

@@ -0,0 +1,87 @@
+import aiohttp
+import pytest
+
+from kopf.clients.auth import APIContext, reauthenticated_request
+from kopf.clients.errors import APIClientResponseError, check_response
+
+
+@reauthenticated_request
+async def get_it(url: str, *, context: APIContext) -> None:
+    response = await context.session.get(url)
+    await check_response(response)
+    return await response.json()
+
+
+@pytest.mark.parametrize('status', [200, 202, 300, 304])
+async def test_no_error_on_success(
+        resp_mocker, aresponses, hostname, resource, status):
+
+    resp = aresponses.Response(
+        status=status,
+        reason="boo!",
+        headers={'Content-Type': 'application/json'},
+        text='{"kind": "Status", "code": "xxx", "message": "msg"}',
+    )
+    aresponses.add(hostname, '/', 'get', resp_mocker(return_value=resp))
+
+    await get_it(f"http://{hostname}/")
+
+
+@pytest.mark.parametrize('status', [400, 401, 403, 404, 500, 666])
+async def test_replaced_error_raised_with_payload(
+        resp_mocker, aresponses, hostname, resource, status):
+
+    resp = aresponses.Response(
+        status=status,
+        reason="boo!",
+        headers={'Content-Type': 'application/json'},
+        text='{"kind": "Status", "code": "xxx", "message": "msg"}',
+    )
+    aresponses.add(hostname, '/', 'get', resp_mocker(return_value=resp))
+
+    with pytest.raises(aiohttp.ClientResponseError) as err:
+        await get_it(f"http://{hostname}/")
+
+    assert isinstance(err.value, APIClientResponseError)
+    assert err.value.status == status
+    assert err.value.message == 'msg'
+
+
+@pytest.mark.parametrize('status', [400, 500, 666])
+async def test_original_error_raised_if_nonjson_payload(
+        resp_mocker, aresponses, hostname, resource, status):
+
+    resp = aresponses.Response(
+        status=status,
+        reason="boo!",
+        headers={'Content-Type': 'application/json'},
+        text='unparsable json',
+    )
+    aresponses.add(hostname, '/', 'get', resp_mocker(return_value=resp))
+
+    with pytest.raises(aiohttp.ClientResponseError) as err:
+        await get_it(f"http://{hostname}/")
+
+    assert not isinstance(err.value, APIClientResponseError)
+    assert err.value.status == status
+    assert err.value.message == 'boo!'
+
+
+@pytest.mark.parametrize('status', [400, 500, 666])
+async def test_original_error_raised_if_parseable_nonk8s_payload(
+        resp_mocker, aresponses, hostname, resource, status):
+
+    resp = aresponses.Response(
+        status=status,
+        reason="boo!",
+        headers={'Content-Type': 'application/json'},
+        text='{"kind": "NonStatus", "code": "xxx", "message": "msg"}',
+    )
+    aresponses.add(hostname, '/', 'get', resp_mocker(return_value=resp))
+
+    with pytest.raises(aiohttp.ClientResponseError) as err:
+        await get_it(f"http://{hostname}/")
+
+    assert not isinstance(err.value, APIClientResponseError)
+    assert err.value.status == status
+    assert err.value.message == 'boo!'