Merge pull request #1 from InformaticsMatters/sc-3299-rstream

Initial rstream logic

Author: alanbchristie

.pylintrc

@@ -3,4 +3,7 @@
 # Disable some distracting checks.
 # See http://pylint.pycqa.org/en/latest/user_guide/message-control.html
 disable = R0801,
-          too-few-public-methods
+          too-few-public-methods,
+          too-many-arguments,
+          too-many-locals,
+          too-many-statements

README.md

@@ -173,6 +173,69 @@ by the client using the command: -
     the one used by RabbitMQ in the docker-compose file. If you need to change this
     you can provide a different AMPQ URL as a 2nd argument on the command line.
 
+## Version 1
+Version 1 uses the `pika` package and relies on a classic exchange-based
+RabbitMQ Topic Queue.
+
+## Version 2
+Version 2 uses the `rstream` package and relies on a RabbitMQ stream.
+
+Version 2 also extends messages prior to forwarding them to their socket clients.
+It does this by appending the messages's **offset** in the stream (an `integer`,
+which we refer to as an **ordinal**)), and the message's **timestamp** (also an `integer`),
+both of which are provided by the backing RabbitMQ stream as the messages are received
+by the Event Stream's consumer. The **offset** can be used as a unique message identifier.
+
+When received as a protobuf string the values are appended to the end of the original
+message using `|` as a delimiter. Here is an example, with the message split at the
+delimiter for clarity: -
+
+    accountserver.MerchantProcessingCharge
+    |timestamp: "2025-04-30T19:20:37.926+00:00" merchant_kind: "DATA_MANAGER" [...]
+    |offset: 2
+    |timestamp: 1746042171620
+
+JSON strings will have these values added to the received dictionary using the
+keys `ess_offset` and `ess_timestamp`, again, displayed here over several lines
+for clarity: -
+
+    {
+        "ess_offset": 2,
+        "ess_timestamp": 1746042171620,
+        "message_type": "accountserver.MerchantProcessingCharge",
+        "message_body": {
+            "timestamp": "2025-04-30T19:20:37.926+00:00",
+            "merchant_kind": "DATA_MANAGER"
+        }
+    }
+
+## Connecting to sockets (historical events)
+The streaming service keeps historical events based on a maximum age and file size.
+Consequently you can connect to your websocket and retrieve these historical events
+as long as they remain in the backend streaming queue. You identify the
+start-time of your events by using **headers** in the websocket request for your
+stream's **LOCATION**.
+
+If you do not provide any header value your socket will only deliver new events.
+
+You can select the start of your events buy providing either an **ordinal**,
+**timestamp**, or **datetime** string.
+
+To stream from a specific **ordinal** (**offset**), provide it as the numerical value
+of the header property `X-StreamOffsetFromOrdinal`.
+
+To stream from a specific **timestamp**, provide it as the numerical value
+of the header property `X-StreamOffsetFromTimestamp`.
+
+To stream from a specific **datetime**, provide the date/time string as the value
+of the header property `X-StreamOffsetFromDatetime`. The datetime string is extremely
+flexible and is interpreted by the **python-dateutil** package's `parse` function.
+UTC is used as the reference for messages and the string will be interpreted as a
+UTC value if it has no timezone specification. For example, if you are in CEST and
+it is `13:33` and you want to retrieve times from 13:33 (local time) then you will need
+to provide a **datetime** string value that has the time set to
+`11:33` (the UTC time for 13:33 CEST) or specify `13:33+02:00`
+
 ---
 
 [black]: https://black.readthedocs.io/en/stable

ansible/parameters-template.yaml

@@ -11,3 +11,6 @@ ess_image_tag: SetMe
 ess_service_account: SetMe
 ess_ws_hostname: SetMe
 ess_ampq_url: SetMe
+
+ess_shared_volume_volume_storageclass: SetMe
+ess_log_volume_volume_storageclass: SetMe

ansible/roles/app/templates/deployment.yaml.j2

@@ -111,6 +111,32 @@ spec:
 {% if ess_api_mem_limit %}
             memory: {{ ess_api_mem_limit }}
 {% endif %}
+{% endif %}
+
+      - name: memcached
+        image: memcached:{{ ess_memcached_tag }}
+        ports:
+        - containerPort: 11211
+          name: memcached
+        env:
+        - name: MEMCACHED_MAX_CONNECTIONS
+          value: '{{ ess_memcached_max_connections }}'
+        - name: MEMCACHED_MAX_ITEM_SIZE
+          value: '36'
+        - name: MEMCACHED_MEMORY_LIMIT
+          value: '{{ ess_memcached_memory_limit }}'
+        resources:
+          requests:
+            cpu: {{ ess_memcached_cpu_request }}
+            memory: {{ ess_memcached_mem_request }}
+{% if ess_memcached_cpu_limit or ess_memcached_mem_limit %}
+          limits:
+{% if ess_memcached_cpu_limit %}
+            cpu: {{ ess_memcached_cpu_limit }}
+{% endif %}
+{% if ess_memcached_mem_limit %}
+            memory: {{ ess_memcached_mem_limit }}
+{% endif %}
 {% endif %}
 
       volumes:

ansible/roles/app/vars/main.yaml

@@ -11,5 +11,20 @@ ess_cert_manager_issuer_base: letsencrypt-{{ ess_ingress_class }}
 # One of 'production', 'staging' or blank (for no SSL).
 ess_cert_issuer:
 
+# memcached image tag
+ess_memcached_tag: 1.6.38-alpine3.21
+ess_memcached_memory_limit: 200m
+# memcached resources
+ess_memcached_cpu_request: 100m
+ess_memcached_cpu_limit:
+ess_memcached_mem_request: 256Mi
+ess_memcached_mem_limit: 256Mi
+
+# Environment
+# Number of connections
+ess_memcached_max_connections: 4
+# Max item size - we use uuid4() strings
+ess_memcached_max_item_size: 36
+
 ess_priority_class:
 ess_api_termination_grace_period_seconds: 30