metric instrumentation Tornado

CHANGELOG.md

@@ -53,6 +53,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#1197](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1197))
 - Add metric instumentation for flask
   ([#1186](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1186))
+- Add metric instrumentation for tornado
+  ([#1252](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1252))
 
 ## [1.12.0rc2-0.32b0](https://github.com/open-telemetry/opentelemetry-python/releases/tag/v1.12.0rc2-0.32b0) - 2022-07-01
 

instrumentation/README.md

@@ -38,7 +38,7 @@
 | [opentelemetry-instrumentation-sqlite3](./opentelemetry-instrumentation-sqlite3) | sqlite3 | No
 | [opentelemetry-instrumentation-starlette](./opentelemetry-instrumentation-starlette) | starlette ~= 0.13.0 | No
 | [opentelemetry-instrumentation-system-metrics](./opentelemetry-instrumentation-system-metrics) | psutil >= 5 | No
-| [opentelemetry-instrumentation-tornado](./opentelemetry-instrumentation-tornado) | tornado >= 5.1.1 | No
+| [opentelemetry-instrumentation-tornado](./opentelemetry-instrumentation-tornado) | tornado >= 5.1.1 | Yes
 | [opentelemetry-instrumentation-urllib](./opentelemetry-instrumentation-urllib) | urllib | No
 | [opentelemetry-instrumentation-urllib3](./opentelemetry-instrumentation-urllib3) | urllib3 >= 1.0.0, < 2.0.0 | Yes
 | [opentelemetry-instrumentation-wsgi](./opentelemetry-instrumentation-wsgi) | wsgi | Yes

instrumentation/opentelemetry-instrumentation-tornado/src/opentelemetry/instrumentation/tornado/__init__.py

@@ -157,6 +157,7 @@ def client_resposne_hook(span, future):
 from functools import partial
 from logging import getLogger
 from time import time_ns
+from timeit import default_timer
 from typing import Collection
 
 import tornado.web
@@ -177,6 +178,7 @@ def client_resposne_hook(span, future):
     http_status_to_status_code,
     unwrap,
 )
+from opentelemetry.metrics import get_meter
 from opentelemetry.propagators import textmap
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.trace.status import Status, StatusCode
@@ -231,15 +233,39 @@ def _instrument(self, **kwargs):
         process lifetime.
         """
         tracer_provider = kwargs.get("tracer_provider")
-        tracer = trace.get_tracer(__name__, __version__, tracer_provider)
+        self.tracer = trace.get_tracer(__name__, __version__, tracer_provider)
+
+        meter_provider = kwargs.get("meter_provider")
+        self.meter = get_meter(__name__, __version__, meter_provider)
+
+        self.duration_histogram = self.meter.create_histogram(
+            name="http.server.duration",
+            unit="ms",
+            description="measures the duration outbound HTTP requests",
+        )
+        self.request_size_histogram = self.meter.create_histogram(
+            name="http.server.request.size",
+            unit="By",
+            description="measures the size of HTTP request messages (compressed)",
+        )
+        self.response_size_histogram = self.meter.create_histogram(
+            name="http.server.response.size",
+            unit="By",
+            description="measures the size of HTTP response messages (compressed)",
+        )
+        self.active_requests_histogram = self.meter.create_up_down_counter(
+            name="http.server.active_requests",
+            unit="requests",
+            description="measures the number of concurrent HTTP requests that are currently in-flight",
+        )
 
         client_request_hook = kwargs.get("client_request_hook", None)
         client_response_hook = kwargs.get("client_response_hook", None)
         server_request_hook = kwargs.get("server_request_hook", None)
 
         def handler_init(init, handler, args, kwargs):
             cls = handler.__class__
-            if patch_handler_class(tracer, cls, server_request_hook):
+            if patch_handler_class(self, cls, server_request_hook):
                 self.patched_handlers.append(cls)
             return init(*args, **kwargs)
 
@@ -250,7 +276,11 @@ def handler_init(init, handler, args, kwargs):
             "tornado.httpclient",
             "AsyncHTTPClient.fetch",
             partial(
-                fetch_async, tracer, client_request_hook, client_response_hook
+                fetch_async,
+                self.tracer,
+                client_request_hook,
+                client_response_hook,
+                self.response_size_histogram,
             ),
         )
 
@@ -262,14 +292,14 @@ def _uninstrument(self, **kwargs):
         self.patched_handlers = []
 
 
-def patch_handler_class(tracer, cls, request_hook=None):
+def patch_handler_class(self, cls, request_hook=None):
     if getattr(cls, _OTEL_PATCHED_KEY, False):
         return False
 
     setattr(cls, _OTEL_PATCHED_KEY, True)
-    _wrap(cls, "prepare", partial(_prepare, tracer, request_hook))
-    _wrap(cls, "on_finish", partial(_on_finish, tracer))
-    _wrap(cls, "log_exception", partial(_log_exception, tracer))
+    _wrap(cls, "prepare", partial(_prepare, self, request_hook))
+    _wrap(cls, "on_finish", partial(_on_finish, self))
+    _wrap(cls, "log_exception", partial(_log_exception, self))
     return True
 
 
@@ -289,29 +319,28 @@ def _wrap(cls, method_name, wrapper):
     wrapt.apply_patch(cls, method_name, wrapper)
 
 
-def _prepare(tracer, request_hook, func, handler, args, kwargs):
-    start_time = time_ns()
+def _prepare(self, request_hook, func, handler, args, kwargs):
     request = handler.request
     if _excluded_urls.url_disabled(request.uri):
         return func(*args, **kwargs)
-    ctx = _start_span(tracer, handler, start_time)
+    ctx = _start_span(self, handler)
     if request_hook:
         request_hook(ctx.span, handler)
     return func(*args, **kwargs)
 
 
-def _on_finish(tracer, func, handler, args, kwargs):
+def _on_finish(self, func, handler, args, kwargs):
     response = func(*args, **kwargs)
-    _finish_span(tracer, handler)
+    _finish_span(self, handler)
     return response
 
 
-def _log_exception(tracer, func, handler, args, kwargs):
+def _log_exception(self, func, handler, args, kwargs):
     error = None
     if len(args) == 3:
         error = args[1]
 
-    _finish_span(tracer, handler, error)
+    _finish_span(self, handler, error)
     return func(*args, **kwargs)
 
 
@@ -377,11 +406,13 @@ def _get_full_handler_name(handler):
     return f"{klass.__module__}.{klass.__qualname__}"
 
 
-def _start_span(tracer, handler, start_time) -> _TraceContext:
+def _start_span(self, handler) -> _TraceContext:
+    start_time = default_timer()
+
     span, token = _start_internal_or_server_span(
-        tracer=tracer,
+        tracer=self.tracer,
         span_name=_get_operation_name(handler, handler.request),
-        start_time=start_time,
+        start_time=time_ns(),
         context_carrier=handler.request.headers,
         context_getter=textmap.default_getter,
     )
@@ -398,6 +429,14 @@ def _start_span(tracer, handler, start_time) -> _TraceContext:
             if len(custom_attributes) > 0:
                 span.set_attributes(custom_attributes)
 
+    metric_attributes = _create_metric_attributes(handler)
+    request_size = len(handler.request.body)
+
+    self.request_size_histogram.record(
+        request_size, attributes=metric_attributes
+    )
+    self.active_requests_histogram.add(1, attributes=metric_attributes)
+
     activation = trace.use_span(span, end_on_exit=True)
     activation.__enter__()  # pylint: disable=E1101
     ctx = _TraceContext(activation, span, token)
@@ -410,10 +449,15 @@ def _start_span(tracer, handler, start_time) -> _TraceContext:
     if propagator:
         propagator.inject(handler, setter=response_propagation_setter)
 
+    elapsed_time = round((default_timer() - start_time) * 1000)
+
+    self.duration_histogram.record(elapsed_time, attributes=metric_attributes)
+    self.active_requests_histogram.add(-1, attributes=metric_attributes)
+
     return ctx
 
 
-def _finish_span(tracer, handler, error=None):
+def _finish_span(self, handler, error=None):
     status_code = handler.get_status()
     reason = getattr(handler, "_reason")
     finish_args = (None, None, None)
@@ -423,7 +467,7 @@ def _finish_span(tracer, handler, error=None):
         if isinstance(error, tornado.web.HTTPError):
             status_code = error.status_code
             if not ctx and status_code == 404:
-                ctx = _start_span(tracer, handler, time_ns())
+                ctx = _start_span(self, handler)
         else:
             status_code = 500
             reason = None
@@ -462,3 +506,15 @@ def _finish_span(tracer, handler, error=None):
     if ctx.token:
         context.detach(ctx.token)
     delattr(handler, _HANDLER_CONTEXT_KEY)
+
+
+def _create_metric_attributes(handler):
+    metric_attributes = {
+        SpanAttributes.HTTP_METHOD: handler.request.method,
+        SpanAttributes.HTTP_SCHEME: handler.request.protocol,
+        SpanAttributes.HTTP_STATUS_CODE: handler.get_status(),
+        SpanAttributes.HTTP_FLAVOR: handler.request.version,
+        SpanAttributes.HTTP_HOST: handler.request.host,
+    }
+
+    return metric_attributes

instrumentation/opentelemetry-instrumentation-tornado/src/opentelemetry/instrumentation/tornado/client.py

@@ -41,7 +41,16 @@ def _normalize_request(args, kwargs):
     return (new_args, new_kwargs)
 
 
-def fetch_async(tracer, request_hook, response_hook, func, _, args, kwargs):
+def fetch_async(
+    tracer,
+    request_hook,
+    response_hook,
+    response_size_histogram,
+    func,
+    _,
+    args,
+    kwargs,
+):
     start_time = time_ns()
 
     # Return immediately if no args were provided (error)
@@ -78,12 +87,15 @@ def fetch_async(tracer, request_hook, response_hook, func, _, args, kwargs):
                 _finish_tracing_callback,
                 span=span,
                 response_hook=response_hook,
+                response_size_histogram=response_size_histogram,
             )
         )
         return future
 
 
-def _finish_tracing_callback(future, span, response_hook):
+def _finish_tracing_callback(
+    future, span, response_hook, response_size_histogram
+):
     status_code = None
     description = None
     exc = future.exception()
@@ -102,6 +114,22 @@ def _finish_tracing_callback(future, span, response_hook):
                 description=description,
             )
         )
+    response = future.result()
+    metric_attributes = _create_metric_attributes(response)
+    response_size = int(response.headers["Content-Length"])
+
+    response_size_histogram.record(response_size, attributes=metric_attributes)
+
     if response_hook:
         response_hook(span, future)
     span.end()
+
+
+def _create_metric_attributes(response):
+    metric_attributes = {
+        SpanAttributes.HTTP_STATUS_CODE: response.code,
+        SpanAttributes.HTTP_URL: remove_url_credentials(response.request.url),
+        SpanAttributes.HTTP_METHOD: response.request.method,
+    }
+
+    return metric_attributes

instrumentation/opentelemetry-instrumentation-tornado/src/opentelemetry/instrumentation/tornado/package.py

@@ -14,3 +14,5 @@
 
 
 _instruments = ("tornado >= 5.1.1",)
+
+_supports_metrics = True

instrumentation/opentelemetry-instrumentation-tornado/tests/test_instrumentation.py

@@ -55,12 +55,12 @@ def get_app(self):
         return app
 
     def setUp(self):
+        super().setUp()
         TornadoInstrumentor().instrument(
             server_request_hook=getattr(self, "server_request_hook", None),
             client_request_hook=getattr(self, "client_request_hook", None),
             client_response_hook=getattr(self, "client_response_hook", None),
         )
-        super().setUp()
         # pylint: disable=protected-access
         self.env_patch = patch.dict(
             "os.environ",