Observe request summary, size, and duration with labels (#3461)

* Send metrics to count endpoints with method and status

* Report request size

* Observe request duration

* Add generic labels on metrics using request matchdict fields

* Reinitialize metrics on plugin initialize to ease testing

* Fix docstring

* Add missing method

* Fix crash in response callback when path is invalid unicode

kinto/__init__.py

@@ -38,6 +38,7 @@
     "record_id_generator": "kinto.views.RelaxedUUID",
     "project_name": "kinto",
     "admin_assets_path": None,
+    "metrics_matchdict_fields": ["bucket_id", "collection_id", "group_id", "record_id"],
 }
 
 

kinto/core/__init__.py

@@ -96,6 +96,7 @@
     "statsd_backend": "kinto.core.statsd",
     "statsd_prefix": "kinto.core",
     "statsd_url": None,
+    "metrics_matchdict_fields": [],
     "storage_backend": "",
     "storage_url": "",
     "storage_max_fetch_size": 10000,

kinto/core/initialization.py

@@ -472,13 +472,65 @@ def on_new_response(event):
         request = event.request
         metrics_service = config.registry.metrics
 
+        try:
+            endpoint = utils.strip_uri_prefix(request.path)
+        except UnicodeDecodeError as e:
+            # This `on_new_response` callback is also called when a HTTP 400
+            # is returned because of an invalid UTF-8 path. We still want metrics.
+            endpoint = str(e)
+
         # Count unique users.
         user_id = request.prefixed_userid
         if user_id:
             # Get rid of colons in metric packet (see #1282).
             auth, user_id = user_id.split(":")
             metrics_service.count("users", unique=[("auth", auth), ("userid", user_id)])
 
+        # Add extra labels to metrics, based on fields extracted from the request matchdict.
+        metrics_matchdict_fields = aslist(settings["metrics_matchdict_fields"])
+        # Turn the `id` field of object endpoints into `{resource}_id` (eg. `mushroom_id`, `bucket_id`)
+        enhanced_matchdict = dict(request.matchdict or {})
+        try:
+            enhanced_matchdict[request.current_resource_name + "_id"] = enhanced_matchdict.get(
+                "id", ""
+            )
+        except AttributeError:
+            # Not on a resource.
+            pass
+        metrics_matchdict_labels = [
+            (field, enhanced_matchdict.get(field, "")) for field in metrics_matchdict_fields
+        ]
+
+        # Count served requests.
+        metrics_service.count(
+            "request_summary",
+            unique=[
+                ("method", request.method.lower()),
+                ("endpoint", endpoint),
+                ("status", str(request.response.status_code)),
+            ]
+            + metrics_matchdict_labels,
+        )
+
+        try:
+            current = utils.msec_time()
+            duration = current - request._received_at
+            metrics_service.observe(
+                "request_duration",
+                duration,
+                labels=[("endpoint", endpoint)] + metrics_matchdict_labels,
+            )
+        except AttributeError:  # pragma: no cover
+            # Logging was not setup in this Kinto app (unlikely but possible)
+            pass
+
+        # Observe response size.
+        metrics_service.observe(
+            "request_size",
+            len(request.response.body or b""),
+            labels=[("endpoint", endpoint)] + metrics_matchdict_labels,
+        )
+
         # Count authentication verifications.
         if hasattr(request, "authn_type"):
             metrics_service.count(f"authn_type.{request.authn_type}")

kinto/core/metrics.py

@@ -16,6 +16,11 @@ def timer(key):
         Watch execution time.
         """
 
+    def observe(self, key, value, labels=[]):
+        """
+        Observe a give `value` for the specified `key`.
+        """
+
     def count(key, count=1, unique=None):
         """
         Count occurrences. If `unique` is set, overwrites the counter value
@@ -45,6 +50,9 @@ class NoOpMetricsService:
     def timer(self, key):
         return NoOpTimer()
 
+    def observe(self, key, value, labels=[]):
+        pass
+
     def count(self, key, count=1, unique=None):
         pass
 

kinto/plugins/prometheus.py

@@ -82,6 +82,28 @@ def timer(self, key):
 
         return Timer(_METRICS[key])
 
+    def observe(self, key, value, labels=[]):
+        global _METRICS
+
+        if key not in _METRICS:
+            _METRICS[key] = prometheus_module.Summary(
+                _fix_metric_name(key),
+                f"Summary of {key}",
+                labelnames=[label_name for label_name, _ in labels],
+                registry=get_registry(),
+            )
+
+        if not isinstance(_METRICS[key], prometheus_module.Summary):
+            raise RuntimeError(
+                f"Metric {key} already exists with different type ({_METRICS[key]})"
+            )
+
+        m = _METRICS[key]
+        if labels:
+            m = m.labels(*(label_value for _, label_value in labels))
+
+        m.observe(value)
+
     def count(self, key, count=1, unique=None):
         global _METRICS
 
@@ -150,4 +172,15 @@ def includeme(config):
     config.add_route("prometheus_metrics", "/__metrics__")
     config.add_view(metrics_view, route_name="prometheus_metrics")
 
+    # Reinitialize the registry on initialization.
+    # This is mainly useful in tests, where the plugin is included
+    # several times with different settings.
+    registry = get_registry()
+    for collector in _METRICS.values():
+        try:
+            registry.unregister(collector)
+        except KeyError:  # pragma: no cover
+            pass
+    _METRICS.clear()
+
     config.registry.registerUtility(PrometheusService(), metrics.IMetricsService)

kinto/plugins/statsd.py

@@ -21,6 +21,9 @@ def __init__(self, host, port, prefix):
     def timer(self, key):
         return self._client.timer(key)
 
+    def observe(self, key, value, labels=[]):
+        return self._client.gauge(key, value)
+
     def count(self, key, count=1, unique=None):
         if unique is None:
             return self._client.incr(key, count=count)

tests/core/test_initialization.py

@@ -411,6 +411,35 @@ def test_statsd_counts_nothing_on_anonymous_requests(self):
         app.get("/")
         self.assertFalse(self.mocked.count.called)
 
+    def test_statsd_counts_endpoints(self):
+        kinto.core.initialize(self.config, "0.0.1", "settings_prefix")
+        app = webtest.TestApp(self.config.make_wsgi_app())
+        app.get("/v0/__heartbeat__")
+        self.mocked().count.assert_any_call(
+            "request_summary",
+            unique=[("method", "get"), ("endpoint", "/__heartbeat__"), ("status", "200")],
+        )
+
+    def test_statsd_observe_request_size(self):
+        kinto.core.initialize(self.config, "0.0.1", "settings_prefix")
+        app = webtest.TestApp(self.config.make_wsgi_app())
+        app.get("/v0/__heartbeat__")
+        self.mocked().observe.assert_any_call(
+            "request_size",
+            len("{}"),
+            labels=[("endpoint", "/__heartbeat__")],
+        )
+
+    def test_statsd_observe_request_duration(self):
+        kinto.core.initialize(self.config, "0.0.1", "settings_prefix")
+        app = webtest.TestApp(self.config.make_wsgi_app())
+        app.get("/v0/__heartbeat__")
+        self.mocked().observe.assert_any_call(
+            "request_duration",
+            mock.ANY,
+            labels=[("endpoint", "/__heartbeat__")],
+        )
+
     def test_statsd_counts_views_and_methods(self):
         kinto.core.initialize(self.config, "0.0.1", "settings_prefix")
         app = webtest.TestApp(self.config.make_wsgi_app())

tests/plugins/test_prometheus.py

@@ -78,6 +78,18 @@ def test_timer_can_be_used_as_decorator_on_partial_function(self):
         resp = self.app.get("/__metrics__")
         self.assertIn("TYPE func_latency_partial summary", resp.text)
 
+    def test_observe_a_single_value(self):
+        self.app.app.registry.metrics.observe("price", 111)
+
+        resp = self.app.get("/__metrics__")
+        self.assertIn("price_sum 111", resp.text)
+
+    def test_observe_a_single_value_with_labels(self):
+        self.app.app.registry.metrics.observe("size", 3.14, labels=[("endpoint", "/buckets")])
+
+        resp = self.app.get("/__metrics__")
+        self.assertIn('size_sum{endpoint="/buckets"} 3.14', resp.text)
+
     def test_count_by_key(self):
         self.app.app.registry.metrics.count("key")
 
@@ -107,6 +119,10 @@ def test_metrics_cant_be_mixed(self):
         with self.assertRaises(RuntimeError):
             self.app.app.registry.metrics.count("timer")
 
+        self.app.app.registry.metrics.count("observer")
+        with self.assertRaises(RuntimeError):
+            self.app.app.registry.metrics.observe("observer", 42)
+
     def test_metrics_names_and_labels_are_transformed(self):
         self.app.app.registry.metrics.count("http.home.status", unique=[("code.get", "200")])
 

tests/plugins/test_statsd.py

@@ -31,6 +31,11 @@ def setUp(self):
         self.mocked_client = patch.start()
         self.addCleanup(patch.stop)
 
+    def test_observe_a_single_value(self):
+        with mock.patch.object(self.client, "_client") as mocked_client:
+            self.client.observe("size", 3.14)
+            mocked_client.gauge.assert_called_with("size", 3.14)
+
     def test_count_increments_the_counter_for_key(self):
         with mock.patch.object(self.client, "_client") as mocked_client:
             self.client.count("click")

tests/test_views_metrics.py

@@ -0,0 +1,45 @@
+import unittest
+from unittest import mock
+
+from kinto.core.testing import skip_if_no_prometheus
+
+from .support import BaseWebTest
+
+
+@skip_if_no_prometheus
+class ViewsMetricsTest(BaseWebTest, unittest.TestCase):
+    @classmethod
+    def get_app_settings(cls, extras=None):
+        settings = super().get_app_settings(extras)
+        settings["includes"] += "\nkinto.plugins.prometheus"
+        return settings
+
+    def setUp(self):
+        super().setUp()
+        patch = mock.patch("kinto.plugins.prometheus.PrometheusService")
+        self.mocked = patch.start()
+        self.addCleanup(patch.stop)
+
+    def test_metrics_have_matchdict_labels(self):
+        self.app.put("/buckets/beers", headers=self.headers)
+        self.app.put("/buckets/beers/groups/amateurs", headers=self.headers)
+        self.app.put("/buckets/beers/collections/barley", headers=self.headers)
+        self.app.put("/buckets/beers/collections/barley/records/abc", headers=self.headers)
+
+        resp = self.app.get("/__metrics__")
+        self.assertIn(
+            'request_size_sum{bucket_id="beers",collection_id="",endpoint="/buckets/beers",group_id="",record_id=""}',
+            resp.text,
+        )
+        self.assertIn(
+            'request_size_sum{bucket_id="beers",collection_id="",endpoint="/buckets/beers/groups/amateurs",group_id="amateurs",record_id=""}',
+            resp.text,
+        )
+        self.assertIn(
+            'request_summary_total{bucket_id="beers",collection_id="barley",endpoint="/buckets/beers/collections/barley",group_id="",method="put",record_id="",status="201"}',
+            resp.text,
+        )
+        self.assertIn(
+            'request_duration_sum{bucket_id="beers",collection_id="barley",endpoint="/buckets/beers/collections/barley/records/abc",group_id="",record_id="abc"}',
+            resp.text,
+        )