Add ADR

Author: Tom-Willemsen

doc/architectural_decisions/009-kafka-streaming.md

@@ -0,0 +1,41 @@
+# 9. Kafka streaming
+
+## Status
+
+Current
+
+## Context
+
+Many facilities stream bluesky documents to an event-bus for consumption by out-of-process listeners.
+Event buses used for this purpose at other facilities include ZeroMQ, RabbitMQ, Kafka, Redis, NATS, and
+others.
+
+The capability this provides is that callbacks can be run in different processes or on other computers,
+without holding up or interfering with the local `RunEngine`. Other groups at ISIS have expressed some
+interest in being able to subscribe to bluesky documents.
+
+## Decision
+
+- We will stream our messages to Kafka, as opposed to some other message bus. This is because we already
+have Kafka infrastructure available for other purposes (e.g. event data & sample-environment data).
+- At the time of writing, we will not **depend** on Kafka for anything critical. This is because the 
+central Kafka instance is not currently considered "reliable" in an experiment controls context. However,
+streaming the documents will allow testing to be done. Kafka will eventually be deployed in a "reliable"
+way accessible to each instrument.
+- We will encode messages from bluesky using `msgpack` (with the `msgpack-numpy` extension), because:
+  - It is the default encoder used by the upstream `bluesky-kafka` integration
+  - It is a schema-less encoder, meaning we do not have to write/maintain fixed schemas for all the
+documents allowed by `event-model`
+  - It has reasonable performance in terms of encoding speed and message size
+  - `msgpack` is very widely supported in a range of programming languages
+- Kafka brokers will be configurable via an environment variable, `IBEX_BLUESKY_CORE_KAFKA_BROKER`
+
+```{note}
+Wherever Kafka is mentioned above, the actual implementation may be a Kafka-like (e.g. RedPanda).
+```
+
+## Justification & Consequences
+
+We will stream bluesky documents to Kafka, encoded using `msgpack-numpy`.
+
+At the time of writing this is purely to enable testing, and will not be used for "production" workflows.

pyproject.toml

@@ -43,12 +43,13 @@ dependencies = [
   "bluesky",  # Bluesky framework
   "bluesky-kafka",  # Bluesky-kafka integration
   "ophyd-async[ca] == 0.12.3",  # Device abstraction
-  "matplotlib",  # Plotting
   "lmfit",  # Fitting
-  "scipy",  # Definitions of erf/erfc functions
+  "matplotlib",  # Plotting
+  "msgpack-numpy",  # Encoding kafka messages
   "numpy",  # General array support
   "orjson", # json module which handles numpy arrays transparently
   "scipp",  # support for arrays with variances/units
+  "scipy",  # Definitions of erf/erfc functions
   "scippneutron",  # neutron-specific utilities for scipp
   "typing-extensions",  # TypeVar with default-arg support
   "tzdata",  # Windows timezone support

src/ibex_bluesky_core/run_engine/__init__.py

@@ -3,32 +3,28 @@
 import asyncio
 import functools
 import logging
+import os
+import socket
 from collections.abc import Generator
 from functools import cache
 from threading import Event, Lock
 from typing import Any, cast
 
 import bluesky.preprocessors as bpp
-import msgpack
+import msgpack_numpy
 from bluesky.run_engine import RunEngine, RunEngineResult
 from bluesky.utils import DuringTask, Msg, RunEngineControlException, RunEngineInterrupted
-
-from ibex_bluesky_core.callbacks import DocLoggingCallback
-from ibex_bluesky_core.preprocessors import add_rb_number_processor
-
-__all__ = ["get_kafka_topic_name", "get_run_engine", "run_plan"]
-
-
-import os
-import socket
-
 from bluesky_kafka import Publisher
 
+from ibex_bluesky_core.callbacks import DocLoggingCallback
 from ibex_bluesky_core.plan_stubs import CALL_QT_AWARE_MSG_KEY, CALL_SYNC_MSG_KEY
+from ibex_bluesky_core.preprocessors import add_rb_number_processor
 from ibex_bluesky_core.run_engine._msg_handlers import call_qt_aware_handler, call_sync_handler
 from ibex_bluesky_core.utils import is_matplotlib_backend_qt
 from ibex_bluesky_core.version import version
 
+__all__ = ["get_kafka_topic_name", "get_run_engine", "run_plan"]
+
 logger = logging.getLogger(__name__)
 
 
@@ -128,7 +124,7 @@ def get_run_engine() -> RunEngine:
         topic=get_kafka_topic_name(),
         bootstrap_servers=os.environ.get("IBEX_BLUESKY_CORE_KAFKA_BROKER", DEFAULT_KAFKA_BROKER),
         key="doc",
-        serializer=msgpack.dumps,
+        serializer=msgpack_numpy.dumps,
         producer_config={"enable.idempotence": True},
     )
     RE.subscribe(kafka_callback)