asgi: Add ASGI middleware (#716)

Adding an ASGI extension.

Co-authored-by: Emil Madsen <[email protected]>
Co-authored-by: alrex <[email protected]>
Co-authored-by: Florimond Manca <[email protected]>

Author: 4 people

docs-requirements.txt

@@ -3,6 +3,7 @@ sphinx-rtd-theme~=0.4
 sphinx-autodoc-typehints~=1.10.2
 
 # Required by ext packages
+asgiref~=3.0
 ddtrace>=0.34.0
 aiohttp ~= 3.0
 Deprecated>=1.2.6

docs/ext/asgi/asgi.rst

@@ -0,0 +1,10 @@
+opentelemetry.ext.asgi package
+==============================
+
+Module contents
+---------------
+
+.. automodule:: opentelemetry.ext.asgi
+    :members:
+    :undoc-members:
+    :show-inheritance:

ext/opentelemetry-ext-asgi/CHANGELOG.md

@@ -0,0 +1,5 @@
+# Changelog
+
+## Unreleased
+
+- Add ASGI middleware ([#716](https://github.com/open-telemetry/opentelemetry-python/pull/716))

ext/opentelemetry-ext-asgi/README.rst

@@ -0,0 +1,60 @@
+OpenTelemetry ASGI Middleware
+=============================
+
+|pypi|
+
+.. |pypi| image:: https://badge.fury.io/py/opentelemetry-ext-asgi.svg
+   :target: https://pypi.org/project/opentelemetry-ext-asgi/
+
+
+This library provides a ASGI middleware that can be used on any ASGI framework
+(such as Django, Starlette, FastAPI or Quart) to track requests timing through OpenTelemetry.
+
+Installation
+------------
+
+::
+
+    pip install opentelemetry-ext-asgi
+
+
+Usage (Quart)
+-------------
+
+.. code-block:: python
+
+    from quart import Quart
+    from opentelemetry.ext.asgi import OpenTelemetryMiddleware
+
+    app = Quart(__name__)
+    app.asgi_app = OpenTelemetryMiddleware(app.asgi_app)
+
+    @app.route("/")
+    async def hello():
+        return "Hello!"
+
+    if __name__ == "__main__":
+        app.run(debug=True)
+
+
+Usage (Django 3.0)
+------------------
+
+Modify the application's ``asgi.py`` file as shown below.
+
+.. code-block:: python
+
+    import os
+    from django.core.asgi import get_asgi_application
+    from opentelemetry.ext.asgi import OpenTelemetryMiddleware
+
+    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'asgi_example.settings')
+
+    application = get_asgi_application()
+    application = OpenTelemetryMiddleware(application)
+
+
+References
+----------
+
+* `OpenTelemetry Project <https://opentelemetry.io/>`_

ext/opentelemetry-ext-asgi/setup.cfg

@@ -0,0 +1,50 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+[metadata]
+name = opentelemetry-ext-asgi
+description = ASGI Middleware for OpenTelemetry
+long_description = file: README.rst
+long_description_content_type = text/x-rst
+author = OpenTelemetry Authors
+author_email = [email protected]
+url = https://github.com/open-telemetry/opentelemetry-python/ext/opentelemetry-ext-asgi
+platforms = any
+license = Apache-2.0
+classifiers =
+    Development Status :: 4 - Beta
+    Intended Audience :: Developers
+    License :: OSI Approved :: Apache Software License
+    Programming Language :: Python
+    Programming Language :: Python :: 3
+    Programming Language :: Python :: 3.5
+    Programming Language :: Python :: 3.6
+    Programming Language :: Python :: 3.7
+    Programming Language :: Python :: 3.8
+
+[options]
+python_requires = >=3.5
+package_dir=
+    =src
+packages=find_namespace:
+install_requires =
+    opentelemetry-api == 0.8.dev0
+    asgiref ~= 3.0
+
+[options.extras_require]
+test =
+    opentelemetry-ext-testutil
+
+[options.packages.find]
+where = src

ext/opentelemetry-ext-asgi/setup.py

@@ -0,0 +1,26 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import os
+
+import setuptools
+
+BASE_DIR = os.path.dirname(__file__)
+VERSION_FILENAME = os.path.join(
+    BASE_DIR, "src", "opentelemetry", "ext", "asgi", "version.py"
+)
+PACKAGE_INFO = {}
+with open(VERSION_FILENAME) as f:
+    exec(f.read(), PACKAGE_INFO)
+
+setuptools.setup(version=PACKAGE_INFO["__version__"])

ext/opentelemetry-ext-asgi/src/opentelemetry/ext/asgi/__init__.py

@@ -0,0 +1,211 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+The opentelemetry-ext-asgi package provides an ASGI middleware that can be used
+on any ASGI framework (such as Django-channels / Quart) to track requests
+timing through OpenTelemetry.
+"""
+
+import operator
+import typing
+import urllib
+from functools import wraps
+
+from asgiref.compatibility import guarantee_single_callable
+
+from opentelemetry import context, propagators, trace
+from opentelemetry.ext.asgi.version import __version__  # noqa
+from opentelemetry.trace.status import Status, StatusCanonicalCode
+
+
+def get_header_from_scope(scope: dict, header_name: str) -> typing.List[str]:
+    """Retrieve a HTTP header value from the ASGI scope.
+
+    Returns:
+        A list with a single string with the header value if it exists, else an empty list.
+    """
+    headers = scope.get("headers")
+    return [
+        value.decode("utf8")
+        for (key, value) in headers
+        if key.decode("utf8") == header_name
+    ]
+
+
+def http_status_to_canonical_code(code: int, allow_redirect: bool = True):
+    # pylint:disable=too-many-branches,too-many-return-statements
+    if code < 100:
+        return StatusCanonicalCode.UNKNOWN
+    if code <= 299:
+        return StatusCanonicalCode.OK
+    if code <= 399:
+        if allow_redirect:
+            return StatusCanonicalCode.OK
+        return StatusCanonicalCode.DEADLINE_EXCEEDED
+    if code <= 499:
+        if code == 401:  # HTTPStatus.UNAUTHORIZED:
+            return StatusCanonicalCode.UNAUTHENTICATED
+        if code == 403:  # HTTPStatus.FORBIDDEN:
+            return StatusCanonicalCode.PERMISSION_DENIED
+        if code == 404:  # HTTPStatus.NOT_FOUND:
+            return StatusCanonicalCode.NOT_FOUND
+        if code == 429:  # HTTPStatus.TOO_MANY_REQUESTS:
+            return StatusCanonicalCode.RESOURCE_EXHAUSTED
+        return StatusCanonicalCode.INVALID_ARGUMENT
+    if code <= 599:
+        if code == 501:  # HTTPStatus.NOT_IMPLEMENTED:
+            return StatusCanonicalCode.UNIMPLEMENTED
+        if code == 503:  # HTTPStatus.SERVICE_UNAVAILABLE:
+            return StatusCanonicalCode.UNAVAILABLE
+        if code == 504:  # HTTPStatus.GATEWAY_TIMEOUT:
+            return StatusCanonicalCode.DEADLINE_EXCEEDED
+        return StatusCanonicalCode.INTERNAL
+    return StatusCanonicalCode.UNKNOWN
+
+
+def collect_request_attributes(scope):
+    """Collects HTTP request attributes from the ASGI scope and returns a
+    dictionary to be used as span creation attributes."""
+    server = scope.get("server") or ["0.0.0.0", 80]
+    port = server[1]
+    server_host = server[0] + (":" + str(port) if port != 80 else "")
+    full_path = scope.get("root_path", "") + scope.get("path", "")
+    http_url = scope.get("scheme", "http") + "://" + server_host + full_path
+    query_string = scope.get("query_string")
+    if query_string and http_url:
+        if isinstance(query_string, bytes):
+            query_string = query_string.decode("utf8")
+        http_url = http_url + ("?" + urllib.parse.unquote(query_string))
+
+    result = {
+        "component": scope["type"],
+        "http.scheme": scope.get("scheme"),
+        "http.host": server_host,
+        "host.port": port,
+        "http.flavor": scope.get("http_version"),
+        "http.target": scope.get("path"),
+        "http.url": http_url,
+    }
+    http_method = scope.get("method")
+    if http_method:
+        result["http.method"] = http_method
+    http_host_value = ",".join(get_header_from_scope(scope, "host"))
+    if http_host_value:
+        result["http.server_name"] = http_host_value
+    http_user_agent = get_header_from_scope(scope, "user-agent")
+    if len(http_user_agent) > 0:
+        result["http.user_agent"] = http_user_agent[0]
+
+    if "client" in scope and scope["client"] is not None:
+        result["net.peer.ip"] = scope.get("client")[0]
+        result["net.peer.port"] = scope.get("client")[1]
+
+    # remove None values
+    result = {k: v for k, v in result.items() if v is not None}
+
+    return result
+
+
+def set_status_code(span, status_code):
+    """Adds HTTP response attributes to span using the status_code argument."""
+    try:
+        status_code = int(status_code)
+    except ValueError:
+        span.set_status(
+            Status(
+                StatusCanonicalCode.UNKNOWN,
+                "Non-integer HTTP status: " + repr(status_code),
+            )
+        )
+    else:
+        span.set_attribute("http.status_code", status_code)
+        span.set_status(Status(http_status_to_canonical_code(status_code)))
+
+
+def get_default_span_name(scope):
+    """Default implementation for name_callback"""
+    method_or_path = scope.get("method") or scope.get("path")
+
+    return method_or_path
+
+
+class OpenTelemetryMiddleware:
+    """The ASGI application middleware.
+
+    This class is an ASGI middleware that starts and annotates spans for any
+    requests it is invoked with.
+
+    Args:
+        app: The ASGI application callable to forward requests to.
+        name_callback: Callback which calculates a generic span name for an
+                       incoming HTTP request based on the ASGI scope.
+                       Optional: Defaults to get_default_span_name.
+    """
+
+    def __init__(self, app, name_callback=None):
+        self.app = guarantee_single_callable(app)
+        self.tracer = trace.get_tracer(__name__, __version__)
+        self.name_callback = name_callback or get_default_span_name
+
+    async def __call__(self, scope, receive, send):
+        """The ASGI application
+
+        Args:
+            scope: A ASGI environment.
+            receive: An awaitable callable yielding dictionaries
+            send: An awaitable callable taking a single dictionary as argument.
+        """
+        if scope["type"] not in ("http", "websocket"):
+            return await self.app(scope, receive, send)
+
+        token = context.attach(
+            propagators.extract(get_header_from_scope, scope)
+        )
+        span_name = self.name_callback(scope)
+
+        try:
+            with self.tracer.start_as_current_span(
+                span_name + " asgi",
+                kind=trace.SpanKind.SERVER,
+                attributes=collect_request_attributes(scope),
+            ):
+
+                @wraps(receive)
+                async def wrapped_receive():
+                    with self.tracer.start_as_current_span(
+                        span_name + " asgi." + scope["type"] + ".receive"
+                    ) as receive_span:
+                        message = await receive()
+                        if message["type"] == "websocket.receive":
+                            set_status_code(receive_span, 200)
+                        receive_span.set_attribute("type", message["type"])
+                    return message
+
+                @wraps(send)
+                async def wrapped_send(message):
+                    with self.tracer.start_as_current_span(
+                        span_name + " asgi." + scope["type"] + ".send"
+                    ) as send_span:
+                        if message["type"] == "http.response.start":
+                            status_code = message["status"]
+                            set_status_code(send_span, status_code)
+                        elif message["type"] == "websocket.send":
+                            set_status_code(send_span, 200)
+                        send_span.set_attribute("type", message["type"])
+                        await send(message)
+
+                await self.app(scope, wrapped_receive, wrapped_send)
+        finally:
+            context.detach(token)

ext/opentelemetry-ext-asgi/src/opentelemetry/ext/asgi/version.py

@@ -0,0 +1,15 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+__version__ = "0.8.dev0"