Repository: mercadona/rele

Issue:
Add message attributes to metrics log
## What?

Logging middleware logs next properties under `metrics`:

1. `agent`
1. `topic`
1. `status`
1. `subscription`

For example, `succeeded` message (`contrib.logging_middleware.LoggingMiddleware.post_process_message_success`).

```
    "levelname": "INFO",
	"message": "Successfully processed message for city-created - city-created-building-factory-spain",
    "module": "logging_middleware",
    "metrics": {
      "name": "subscriptions",
      "data": {
        "agent": "country-maker",
        "topic": "city-created",
        "status": "succeeded",
        "subscription": "city-created-building-factory-spain",
        "duration_seconds": 0.001
      }
    }
```

My proposal is to add a new more property called `attributes` which contains the message attributes.

For example, a `succeeded message would appear as follow:

```
    "levelname": "INFO",
	"message": "Successfully processed message for city-created - city-created-building-factory-spain",
    "module": "logging_middleware",
    "metrics": {
      "name": "subscriptions",
      "data": {
        "agent": "country-maker",
        "topic": "city-created",
        "status": "succeeded",
        "subscription": "city-created-building-factory-spain",
        "duration_seconds": 0.001,
		"attributes": {
			"country": "spain"
		}
      }
    }
```

It would apply for `received`, `failed` and previously mentioned `succeeded` messages.

## Why?

As a real scenario, for a topic, several subscriptions can be registered which only process a message if a message attribute has a particular value.

For example, a topic like `city-created` can have at least four subscriptions:

1. `city-created-building-factory-spain`
1. `city-created-roadway-factory-spain`
1. `city-created-building-factory-portugal`
1. `city-created-roadway-factory-portugal`

`city-created-building-factory-{country}` and `city-created-roadway-factory-{country}` are generic for each country and only process messages when an attribute called `country` matches a defined value like `spain` or `portugal`

The subscriptions run in different environments (`spain` and `portugal`), but logs are consumed in centralized system.

When filtering logs for `spain` environment, there will be logs for messages for all subscriptions (both countries). However, subscriptions for not related country will have a `succeeded` log message with a very short duration (skipped message) and subscriptions for related country will take a bit longer to execute.

If log message metrics contain `attributes`, it would be easy to implement a filter to skip logs from a different country to avoid any noise during log analysis.

Discussion:

Patch:
diff --git a/AUTHORS.md b/AUTHORS.md
index 35ad7a4..fda3f98 100644
--- a/AUTHORS.md
+++ b/AUTHORS.md
@@ -3,10 +3,11 @@ Thank you to all the contributors to Relé!
 * Andrew Graham-Yooll (@andrewgy8)
 * Miquel Torres Barcelo (@tobami)
 * Edgar Latorre (@edgarlatorre)
-* Lluis Guirado (@LluisGuiVi) 
-* Daniel Ruiz (@dani-ruiz) 
+* Lluis Guirado (@LluisGuiVi)
+* Daniel Ruiz (@dani-ruiz)
 * Juanjo Ponz (@jjponz)
 * Jose Antonio Navarro (@jonasae)
 * Antonio Bustos Rodriguez (@antoniobusrod)
 * David de la Iglesia (@ddelaiglesia) for the Logo!
 * Santiago Lanús (@sanntt)
+* Craig Mulligan (@hobochild)
diff --git a/rele/contrib/__init__.py b/rele/contrib/__init__.py
index f7a1a53..808ebea 100644
--- a/rele/contrib/__init__.py
+++ b/rele/contrib/__init__.py
@@ -1,4 +1,5 @@
 from .logging_middleware import LoggingMiddleware  # noqa
+from .flask_middleware import FlaskMiddleware  # noqa
 
 try:
     from .django_db_middleware import DjangoDBMiddleware  # noqa
diff --git a/rele/contrib/flask_middleware.py b/rele/contrib/flask_middleware.py
new file mode 100644
index 0000000..b5ae141
--- /dev/null
+++ b/rele/contrib/flask_middleware.py
@@ -0,0 +1,13 @@
+from rele.middleware import BaseMiddleware
+
+
+class FlaskMiddleware(BaseMiddleware):
+    def setup(self, config, **kwargs):
+        self.app = kwargs["flask_app"]
+
+    def pre_process_message(self, subscription, message):
+        self.ctx = self.app.app_context()
+        self.ctx.push()
+
+    def post_process_message(self):
+        self.ctx.pop()
diff --git a/rele/contrib/logging_middleware.py b/rele/contrib/logging_middleware.py
index c964371..e24a675 100644
--- a/rele/contrib/logging_middleware.py
+++ b/rele/contrib/logging_middleware.py
@@ -13,16 +13,19 @@ class LoggingMiddleware(BaseMiddleware):
     def __init__(self):
         self._logger = None
 
-    def setup(self, config):
+    def setup(self, config, **kwargs):
         self._logger = logging.getLogger(__name__)
         self._app_name = config.app_name
 
-    def _build_data_metrics(self, subscription, status, start_processing_time=None):
+    def _build_data_metrics(
+        self, subscription, message, status, start_processing_time=None
+    ):
         result = {
             "agent": self._app_name,
             "topic": subscription.topic,
             "status": status,
             "subscription": subscription.name,
+            "attributes": dict(message.attributes),
         }
 
         if start_processing_time is not None:
@@ -51,7 +54,7 @@ class LoggingMiddleware(BaseMiddleware):
             extra={
                 "metrics": {
                     "name": "subscriptions",
-                    "data": self._build_data_metrics(subscription, "received"),
+                    "data": self._build_data_metrics(subscription, message, "received"),
                 }
             },
         )
@@ -63,7 +66,7 @@ class LoggingMiddleware(BaseMiddleware):
                 "metrics": {
                     "name": "subscriptions",
                     "data": self._build_data_metrics(
-                        subscription, "succeeded", start_time
+                        subscription, message, "succeeded", start_time
                     ),
                 }
             },
@@ -80,7 +83,7 @@ class LoggingMiddleware(BaseMiddleware):
                 "metrics": {
                     "name": "subscriptions",
                     "data": self._build_data_metrics(
-                        subscription, "failed", start_time
+                        subscription, message, "failed", start_time
                     ),
                 },
                 "subscription_message": message,


Test patch:
diff --git a/tests/test_middleware.py b/tests/test_middleware.py
new file mode 100644
index 0000000..0b4c634
--- /dev/null
+++ b/tests/test_middleware.py
@@ -0,0 +1,26 @@
+import pytest
+import rele
+from unittest.mock import patch
+
+
+class TestMiddleware:
+    @pytest.fixture
+    def mock_init_global_publisher(self):
+        with patch("rele.config.init_global_publisher", autospec=True) as m:
+            yield m
+
+    @pytest.mark.usefixtures("mock_init_global_publisher")
+    @patch("rele.contrib.FlaskMiddleware.setup", autospec=True)
+    def test_setup_fn_is_called_with_kwargs(
+        self, mock_middleware_setup, project_id, credentials
+    ):
+
+        settings = {
+            "GC_PROJECT_ID": project_id,
+            "GC_CREDENTIALS": credentials,
+            "MIDDLEWARE": ["rele.contrib.FlaskMiddleware"],
+        }
+
+        rele.setup(settings, foo="bar")
+        assert mock_middleware_setup.called
+        mock_middleware_setup.call_args_list[0][-1] == {"foo": "bar"}
diff --git a/tests/test_subscription.py b/tests/test_subscription.py
index 09c2af5..368ca96 100644
--- a/tests/test_subscription.py
+++ b/tests/test_subscription.py
@@ -80,10 +80,14 @@ class TestCallback:
             yield mock_old_connections
 
     @pytest.fixture
-    def message_wrapper(self):
+    def published_at(self):
+        return time.time()
+
+    @pytest.fixture
+    def message_wrapper(self, published_at):
         rele_message = pubsub_v1.types.PubsubMessage(
             data=b'{"id": 123}',
-            attributes={"lang": "es", "published_at": str(time.time())},
+            attributes={"lang": "es", "published_at": str(published_at)},
             message_id="1",
         )
 
@@ -96,7 +100,9 @@ class TestCallback:
         )
         return pubsub_v1.subscriber.message.Message(rele_message, "ack-id", MagicMock())
 
-    def test_log_start_processing_when_callback_called(self, caplog, message_wrapper):
+    def test_log_start_processing_when_callback_called(
+        self, caplog, message_wrapper, published_at
+    ):
         with caplog.at_level(logging.DEBUG):
             callback = Callback(sub_stub)
             res = callback(message_wrapper)
@@ -113,6 +119,7 @@ class TestCallback:
                 "topic": "some-cool-topic",
                 "status": "received",
                 "subscription": "rele-some-cool-topic",
+                "attributes": {"lang": "es", "published_at": str(published_at),},
             },
         }
 
@@ -128,7 +135,9 @@ class TestCallback:
             "I am a task doing " "stuff with ID 123 (es)"
         )
 
-    def test_log_when_execution_is_succesful(self, message_wrapper, caplog):
+    def test_log_when_execution_is_succesful(
+        self, message_wrapper, caplog, published_at
+    ):
         callback = Callback(sub_stub)
         callback(message_wrapper)
 
@@ -144,11 +153,12 @@ class TestCallback:
                 "status": "succeeded",
                 "subscription": "rele-some-cool-topic",
                 "duration_seconds": pytest.approx(0.5, abs=0.5),
+                "attributes": {"lang": "es", "published_at": str(published_at),},
             },
         }
 
     def test_log_does_not_ack_called_message_when_execution_fails(
-        self, caplog, message_wrapper
+        self, caplog, message_wrapper, published_at
     ):
         @sub(topic="some-cool-topic", prefix="rele")
         def crashy_sub_stub(data, **kwargs):
@@ -172,6 +182,7 @@ class TestCallback:
                 "status": "failed",
                 "subscription": "rele-some-cool-topic",
                 "duration_seconds": pytest.approx(0.5, abs=0.5),
+                "attributes": {"lang": "es", "published_at": str(published_at),},
             },
         }
         assert failed_log.subscription_message == message_wrapper