🐛 Activate tracing in aiohttp-based services (ITISFoundation#2559)

* updates servicelib.tracing

* doc director

* remove general constraint

* upgrades aiozipkin in servicelib

* upgrades storage

* upgrades webserver

* fixes storage validation

* cleanup datatypes in middlewares

* sets tracing sync to get middleware in the outermost layer

* sets tracing sync to get middleware in the outermost layer

* annotations on application-setup

* fixes test_login

* creates aiohttp.typing_extension

* restored aiozipkin to version 0.7.1

* removed old settings from storage

* sorted environs

* lower level of very verbose loggers

* fixes storage: tracing middleware must always go after envelope middleware

* removed tracing from director

Author: pcrespov

packages/service-library/requirements/_aiohttp.in

@@ -8,9 +8,10 @@
 
 openapi-core==0.12.0     # frozen until https://github.com/ITISFoundation/osparc-simcore/pull/1396 is CLOSED
 lazy-object-proxy~=1.4.3 # cannot upgrade due to contraints in openapi-core
+aiozipkin~=0.7.1         # 1.X has an issue with no content responses
+
 
 aiohttp
-aiozipkin
 aiopg[sa]
 jsonschema
 prometheus_client

packages/service-library/requirements/_aiohttp.txt

@@ -13,10 +13,7 @@ aiohttp==3.7.4.post0
 aiopg==1.3.1
     # via -r requirements/_aiohttp.in
 aiozipkin==0.7.1
-    # via
-    #   -c requirements/../../../requirements/constraints.txt
-    #   -c requirements/./../../../requirements/constraints.txt
-    #   -r requirements/_aiohttp.in
+    # via -r requirements/_aiohttp.in
 async-timeout==3.0.1
     # via
     #   aiohttp

packages/service-library/src/servicelib/aiohttp/application_setup.py

@@ -80,6 +80,7 @@ def setup(app: web.Application):
             ...
     """
     # TODO: resilience to failure. if this setup fails, then considering dependencies, is it fatal or app can start?
+    # TODO: enforce signature as def setup(app: web.Application, **kwargs) -> web.Application
 
     module_name = module_name.replace(".__init__", "")
     depends = depends or []
@@ -94,7 +95,7 @@ def setup(app: web.Application):
         # if passes config_enabled, invalidates info on section
         section = None
 
-    def decorate(setup_func):
+    def _decorate(setup_func: Callable):
 
         if "setup" not in setup_func.__name__:
             logger.warning("Rename '%s' to contain 'setup'", setup_func.__name__)
@@ -110,7 +111,7 @@ def setup_metadata() -> Dict:
 
         # wrapper
         @functools.wraps(setup_func)
-        def setup_wrapper(app: web.Application, *args, **kargs) -> bool:
+        def _wrapper(app: web.Application, *args, **kargs) -> bool:
             # pre-setup
             logger.debug(
                 "Setting up '%s' [%s; %s] ... ", module_name, category.name, depends
@@ -176,12 +177,12 @@ def setup_wrapper(app: web.Application, *args, **kargs) -> bool:
             )
             return completed
 
-        setup_wrapper.metadata = setup_metadata
-        setup_wrapper.MARK = "setup"
+        _wrapper.metadata = setup_metadata
+        _wrapper.MARK = "setup"
 
-        return setup_wrapper
+        return _wrapper
 
-    return decorate
+    return _decorate
 
 
 def is_setup_function(fun):

packages/service-library/src/servicelib/aiohttp/rest_middlewares.py

@@ -4,14 +4,18 @@
 """
 import json
 import logging
+from typing import Awaitable, Callable, Union
 
 from aiohttp import web
+from aiohttp.web_request import Request
+from aiohttp.web_response import StreamResponse
 from openapi_core.schema.exceptions import OpenAPIError
 
 from ..utils import is_production_environ
 from .rest_models import ErrorItemType, ErrorType, LogMessageType
 from .rest_responses import (
     JSON_CONTENT_TYPE,
+    _DataType,
     create_data_response,
     create_error_response,
     is_enveloped_from_map,
@@ -20,6 +24,7 @@
 )
 from .rest_utils import EnvelopeFactory
 from .rest_validators import OpenApiValidator
+from .typing_extension import Handler, Middleware
 
 DEFAULT_API_VERSION = "v0"
 
@@ -32,9 +37,8 @@ def is_api_request(request: web.Request, api_version: str) -> bool:
     return request.path.startswith(base_path)
 
 
-def error_middleware_factory(
-    api_version: str = DEFAULT_API_VERSION, log_exceptions=True
-):
+def error_middleware_factory(api_version: str, log_exceptions=True) -> Middleware:
+
     _is_prod: bool = is_production_environ()
 
     def _process_and_raise_unexpected_error(request: web.BaseRequest, err: Exception):
@@ -59,7 +63,7 @@ def _process_and_raise_unexpected_error(request: web.BaseRequest, err: Exception
         raise resp
 
     @web.middleware
-    async def _middleware_handler(request: web.Request, handler):
+    async def _middleware_handler(request: web.Request, handler: Handler):
         """
         Ensure all error raised are properly enveloped and json responses
         """
@@ -74,7 +78,7 @@ async def _middleware_handler(request: web.Request, handler):
         except web.HTTPError as err:
             # TODO: differenciate between server/client error
             if not err.reason:
-                err.reason = "Unexpected error"
+                err.set_status(err.status_code, reason="Unexpected error")
 
             err.content_type = JSON_CONTENT_TYPE
 
@@ -126,9 +130,9 @@ async def _middleware_handler(request: web.Request, handler):
     return _middleware_handler
 
 
-def validate_middleware_factory(api_version: str = DEFAULT_API_VERSION):
+def validate_middleware_factory(api_version: str) -> Middleware:
     @web.middleware
-    async def _middleware_handler(request: web.Request, handler):
+    async def _middleware_handler(request: web.Request, handler: Handler):
         """
         Validates requests against openapi specs and extracts body, params, etc ...
         Validate response against openapi specs
@@ -172,31 +176,42 @@ async def _middleware_handler(request: web.Request, handler):
     return _middleware_handler
 
 
-def envelope_middleware_factory(api_version: str = DEFAULT_API_VERSION):
+_ResponseOrBodyData = Union[StreamResponse, _DataType]
+HandlerFlexible = Callable[[Request], Awaitable[_ResponseOrBodyData]]
+MiddlewareFlexible = Callable[[Request, HandlerFlexible], Awaitable[StreamResponse]]
+
+
+def envelope_middleware_factory(api_version: str) -> MiddlewareFlexible:
+    # FIXME: This data conversion is very error-prone. Use decorators instead!
     _is_prod: bool = is_production_environ()
 
     @web.middleware
-    async def _middleware_handler(request: web.Request, handler):
+    async def _middleware_handler(
+        request: web.Request, handler: HandlerFlexible
+    ) -> StreamResponse:
         """
         Ensures all responses are enveloped as {'data': .. , 'error', ...} in json
+        ONLY for API-requests
         """
         if not is_api_request(request, api_version):
-            return await handler(request)
+            resp = await handler(request)
+            assert isinstance(resp, StreamResponse)  # nosec
+            return resp
 
-        resp = await handler(request)
+        # NOTE: the return values of this handler
+        resp: _ResponseOrBodyData = await handler(request)
 
-        if isinstance(resp, web.FileResponse):  # allows for files to be downloaded
+        if isinstance(resp, web.FileResponse):
             return resp
 
-        if not isinstance(resp, web.Response):
-            response = create_data_response(
+        if not isinstance(resp, StreamResponse):
+            resp = create_data_response(
                 data=resp,
                 skip_internal_error_details=_is_prod,
             )
-        else:
-            # Enforced by user. Should check it is json?
-            response = resp
-        return response
+
+        assert isinstance(resp, web.StreamResponse)  # nosec
+        return resp
 
     # adds identifier (mostly for debugging)
     _middleware_handler.__middleware_name__ = f"{__name__}.envelope_{api_version}"

packages/service-library/src/servicelib/aiohttp/rest_responses.py

@@ -17,6 +17,8 @@
 
 JsonLikeModel = Union[Dict[str, Any], List[Dict[str, Any]]]
 
+_DataType = Union[str, Dict[str, Any], List[Any]]
+
 
 def is_enveloped_from_map(payload: Mapping) -> bool:
     return all(k in ENVELOPE_KEYS for k in payload.keys() if not str(k).startswith("_"))
@@ -66,7 +68,7 @@ def unwrap_envelope(payload: Dict[str, Any]) -> Tuple:
 
 
 def create_data_response(
-    data: Union[Mapping, str], *, skip_internal_error_details=False
+    data: _DataType, *, skip_internal_error_details=False
 ) -> web.Response:
     response = None
     try:

packages/service-library/src/servicelib/aiohttp/tracing.py

@@ -3,37 +3,48 @@
 """
 import asyncio
 import logging
-from typing import Dict, Optional
+from typing import Iterable, Optional, Union
 
 import aiozipkin as az
-import trafaret as T
 from aiohttp import web
-from pydantic import AnyHttpUrl, BaseSettings
+from aiohttp.web import AbstractRoute
+from aiozipkin.tracer import Tracer
+from yarl import URL
 
 log = logging.getLogger(__name__)
 
 
 def setup_tracing(
-    app: web.Application, app_name: str, host: str, port: int, config: Dict
+    app: web.Application,
+    *,
+    service_name: str,
+    host: str,
+    port: int,
+    jaeger_base_url: Union[URL, str],
+    skip_routes: Optional[Iterable[AbstractRoute]] = None,
 ) -> bool:
-    zipkin_address = f"{config['zipkin_endpoint']}/api/v2/spans"
-    endpoint = az.create_endpoint(app_name, ipv4=host, port=port)
-    loop = asyncio.get_event_loop()
-    tracer = loop.run_until_complete(
-        az.create(zipkin_address, endpoint, sample_rate=1.0)
+    """
+    Sets up this service for a distributed tracing system
+    using zipkin (https://zipkin.io/) and Jaeger (https://www.jaegertracing.io/)
+    """
+    zipkin_address = URL(f"{jaeger_base_url}") / "api/v2/spans"
+
+    log.debug(
+        "Settings up tracing for %s at %s:%d -> %s",
+        service_name,
+        host,
+        port,
+        zipkin_address,
     )
-    az.setup(app, tracer)
-    return True
 
+    endpoint = az.create_endpoint(service_name, ipv4=host, port=port)
 
-schema = T.Dict(
-    {
-        T.Key("enabled", default=True, optional=True): T.Or(T.Bool(), T.ToInt),
-        T.Key("zipkin_endpoint", default="http://jaeger:9411"): T.String(),
-    }
-)
+    tracer: Tracer = asyncio.get_event_loop().run_until_complete(
+        az.create(f"{zipkin_address}", endpoint, sample_rate=1.0)
+    )
 
+    # WARNING: adds a middleware that should be the outermost since
+    # it expects stream responses while we allow data returns from a handler
+    az.setup(app, tracer, skip_routes=skip_routes)
 
-class TracingSettings(BaseSettings):
-    enabled: Optional[bool] = True
-    zipkin_endpoint: AnyHttpUrl = "http://jaeger:9411"
+    return True

packages/service-library/src/servicelib/aiohttp/typing_extension.py

@@ -0,0 +1,7 @@
+from typing import Awaitable, Callable
+
+from aiohttp.web import Request, StreamResponse
+
+# Taken from aiohttp.web_middlewares import _Handler, _Middleware
+Handler = Callable[[Request], Awaitable[StreamResponse]]
+Middleware = Callable[[Request, Handler], Awaitable[StreamResponse]]

packages/service-library/tests/aiohttp/test_tracing.py

@@ -0,0 +1,97 @@
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
+from asyncio import AbstractEventLoop
+from typing import Callable
+
+import pytest
+from aiohttp import web
+from aiohttp.client_reqrep import ClientResponse
+from aiohttp.test_utils import TestClient
+from servicelib.aiohttp.rest_responses import _collect_http_exceptions
+from servicelib.aiohttp.tracing import setup_tracing
+
+DEFAULT_JAEGER_BASE_URL = "http://jaeger:9411"
+
+
+@pytest.fixture()
+def client(
+    loop: AbstractEventLoop, aiohttp_client: Callable, aiohttp_unused_port: Callable
+) -> TestClient:
+    ports = [aiohttp_unused_port() for _ in range(2)]
+
+    async def redirect(request: web.Request) -> web.Response:
+        return web.HTTPFound(location="/return/200")
+
+    async def return_response(request: web.Request) -> web.Response:
+        code = int(request.match_info["code"])
+        return web.Response(status=code)
+
+    async def raise_response(request: web.Request):
+        status_code = int(request.match_info["code"])
+        status_to_http_exception = _collect_http_exceptions()
+        http_exception_cls = status_to_http_exception[status_code]
+        raise http_exception_cls(
+            reason=f"raised from raised_error with code {status_code}"
+        )
+
+    async def skip(request: web.Request):
+        return web.HTTPServiceUnavailable(reason="should not happen")
+
+    app = web.Application()
+    app.add_routes(
+        [
+            web.get("/redirect", redirect),
+            web.get("/return/{code}", return_response),
+            web.get("/raise/{code}", raise_response),
+            web.get("/skip", skip, name="skip"),
+        ]
+    )
+
+    print("Resources:")
+    for resource in app.router.resources():
+        print(resource)
+
+    # UNDER TEST ---
+    # SEE RoutesView to understand how resources can be iterated to get routes
+    resource = app.router["skip"]
+    routes_in_a_resource = list(resource)
+
+    setup_tracing(
+        app,
+        service_name=f"{__name__}.client",
+        host="127.0.0.1",
+        port=ports[0],
+        jaeger_base_url=DEFAULT_JAEGER_BASE_URL,
+        skip_routes=routes_in_a_resource,
+    )
+
+    return loop.run_until_complete(
+        aiohttp_client(app, server_kwargs={"port": ports[0]})
+    )
+
+
+async def test_setup_tracing(client: TestClient):
+    # NOTE: version aiozipkin 1.1.0 did not pass this test
+
+    res: ClientResponse
+
+    # on error
+    for code in (web.HTTPOk.status_code, web.HTTPBadRequest.status_code):
+        res = await client.get(f"/return/{code}")
+
+        assert res.status == code, await res.text()
+        res = await client.get(f"/raise/{code}")
+        assert res.status == code, await res.text()
+
+    res = await client.get("/redirect")
+    # TODO: check it was redirected
+    assert res.status == 200, await res.text()
+
+    res = await client.get("/skip")
+    assert res.status == web.HTTPServiceUnavailable.status_code
+
+    # using POST instead of GET ->  HTTPMethodNotAllowed
+    res = await client.post("/skip")
+    assert res.status == web.HTTPMethodNotAllowed.status_code, "GET and not POST"

requirements/constraints.txt

@@ -25,8 +25,6 @@ urllib3>=1.26.5                               # https://github.com/advisories/GH
 # TODO: https://aioredis.readthedocs.io/en/latest/migration/)
 aioredis<2.0.0
 # with new released version 1.0.0 (https://github.com/aio-libs/aiozipkin/releases).
-# TODO: Check again when moving to py3.8-3.9! SEE https://github.com/ITISFoundation/osparc-simcore/pull/2025
-aiozipkin==0.7.1
 # TODO: includes async features https://docs.sqlalchemy.org/en/14/changelog/migration_20.html
 sqlalchemy<2.0