Make metrics server asyncio

Get server functional

Finish config, get test done

Reduce initial metrics we are reporting

Update the README

Try and fix mypy check

Fix integration tests because coverage was added

Add test dep

Update schema again

Use conditional import to protect users without extras installed

Author: AlanCoding

.github/workflows/ci.yml

@@ -24,9 +24,9 @@ jobs:
           python-version: ${{ matrix.py_version }}
 
       - name: Install dispatcherd
-        run: pip install -e .[pg_notify]
+        run: pip install -e .[pg_notify,metrics]
       - run: make postgres
-      - run: pip install pytest pytest-asyncio
+      - run: pip install pytest pytest-asyncio httpx
       - run: pytest tests/unit tests/integration -vv -s
 
   black:
@@ -60,7 +60,7 @@ jobs:
           show-progress: false
 
       - run: pip install mypy
-      - run: pip install -e .[pg_notify]
+      - run: pip install -e .[pg_notify,metrics]
       - run: python3 -m pip install types-PyYAML
       - run: mypy dispatcherd
 

README.md

@@ -235,144 +235,32 @@ Dispatcherd is sponsored by [Red Hat, Inc](https://www.redhat.com).
 
 ## Metrics
 
-Start the dispatcher.
+You can run a demo of the metrics server. In your first terminal tab, run:
 
 ```
-pip install prometheus_client
+pip install .[pg_notify,metrics]
+dispatcherd
+```
 
-$ curl http://localhost:8070
+In another tab run:
 
+```
 curl http://localhost:8070
+```
+
+This should report metrics in the following general format:
+
+```
+$ curl http://localhost:8070
 # HELP dispatcher_messages_received_total Number of messages received by dispatchermain
 # TYPE dispatcher_messages_received_total counter
-dispatcher_messages_received_total 88.0
+dispatcher_messages_received_total 263.0
 # HELP dispatcher_control_messages_count_total Number of control messages received.
 # TYPE dispatcher_control_messages_count_total counter
-dispatcher_control_messages_count_total 10.0
-# HELP dispatcher_worker_created_at Creation time of worker
-# TYPE dispatcher_worker_created_at gauge
-dispatcher_worker_created_at{worker_index="0"} 286576.365272104
-dispatcher_worker_created_at{worker_index="1"} 286576.365368035
-dispatcher_worker_created_at{worker_index="2"} 286578.814706941
-dispatcher_worker_created_at{worker_index="3"} 286578.817639637
-dispatcher_worker_created_at{worker_index="4"} 286578.820265243
-dispatcher_worker_created_at{worker_index="5"} 286578.822075874
-dispatcher_worker_created_at{worker_index="6"} 286578.824725725
-dispatcher_worker_created_at{worker_index="7"} 286578.837810563
-dispatcher_worker_created_at{worker_index="8"} 286578.844329095
-dispatcher_worker_created_at{worker_index="9"} 286578.863277972
-dispatcher_worker_created_at{worker_index="10"} 286578.878555905
-dispatcher_worker_created_at{worker_index="11"} 286579.36656921
-# HELP dispatcher_worker_finished_count Finished count of tasks by the worker
-# TYPE dispatcher_worker_finished_count gauge
-dispatcher_worker_finished_count{worker_index="0"} 2.0
-dispatcher_worker_finished_count{worker_index="1"} 1.0
-dispatcher_worker_finished_count{worker_index="2"} 1.0
-dispatcher_worker_finished_count{worker_index="3"} 7.0
-dispatcher_worker_finished_count{worker_index="4"} 8.0
-dispatcher_worker_finished_count{worker_index="5"} 1.0
-dispatcher_worker_finished_count{worker_index="6"} 0.0
-dispatcher_worker_finished_count{worker_index="7"} 0.0
-dispatcher_worker_finished_count{worker_index="8"} 0.0
-dispatcher_worker_finished_count{worker_index="9"} 2.0
-dispatcher_worker_finished_count{worker_index="10"} 1.0
-dispatcher_worker_finished_count{worker_index="11"} 1.0
-# HELP dispatcher_worker_status Status of worker.
-# TYPE dispatcher_worker_status gauge
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="0"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="0"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="0"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="0"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="0"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="0"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="0"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="0"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="1"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="1"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="1"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="1"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="1"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="1"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="1"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="1"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="2"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="2"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="2"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="2"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="2"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="2"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="2"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="2"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="3"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="3"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="3"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="3"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="3"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="3"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="3"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="3"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="4"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="4"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="4"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="4"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="4"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="4"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="4"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="4"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="5"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="5"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="5"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="5"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="5"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="5"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="5"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="5"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="6"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="6"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="6"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="6"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="6"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="6"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="6"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="6"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="7"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="7"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="7"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="7"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="7"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="7"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="7"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="7"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="8"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="8"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="8"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="8"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="8"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="8"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="8"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="8"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="9"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="9"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="9"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="9"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="9"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="9"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="9"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="9"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="10"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="10"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="10"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="10"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="10"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="10"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="10"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="10"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="error",worker_index="11"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="exited",worker_index="11"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="initialized",worker_index="11"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="ready",worker_index="11"} 1.0
-dispatcher_worker_status{dispatcher_worker_status="retired",worker_index="11"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="spawned",worker_index="11"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="starting",worker_index="11"} 0.0
-dispatcher_worker_status{dispatcher_worker_status="stopping",worker_index="11"} 0.0
+dispatcher_control_messages_count_total 0.0
+# HELP dispatcher_worker_count_total Number of workers running.
+# TYPE dispatcher_worker_count_total counter
+dispatcher_worker_count_total 3.0
 ```
+
+We expect to add more metrics in the future.

dispatcher.yml

@@ -8,6 +8,8 @@ service:
     scaledown_wait: 15  # seconds
   main_kwargs:
     node_id: demo-server-a
+  metrics_kwargs:
+    log_level: debug
 brokers:
   pg_notify:
     config:

dispatcherd/factories.py

@@ -62,6 +62,13 @@ def from_settings(settings: LazySettings = global_settings) -> DispatcherMain:
     producers = producers_from_settings(settings=settings)
     pool = pool_from_settings(settings=settings)
     extra_kwargs = settings.service.get('main_kwargs', {})
+
+    metrics_kwargs = settings.service.get('metrics_kwargs')
+    if metrics_kwargs:
+        from .service.metrics import DispatcherMetricsServer
+
+        extra_kwargs['metrics'] = DispatcherMetricsServer(**metrics_kwargs)
+
     return DispatcherMain(producers, pool, **extra_kwargs)
 
 
@@ -105,7 +112,7 @@ def get_control_from_settings(publish_broker: Optional[str] = None, settings: La
 
 # ---- Schema generation ----
 
-SERIALIZED_TYPES = (int, str, dict, type(None), tuple, list, float)
+SERIALIZED_TYPES = (int, str, dict, type(None), tuple, list, float, bool)
 
 
 def is_valid_annotation(annotation):
@@ -134,6 +141,9 @@ def generate_settings_schema(settings: LazySettings = global_settings) -> dict:
 
     ret['service']['pool_kwargs'] = schema_for_cls(WorkerPool)
     ret['service']['main_kwargs'] = schema_for_cls(DispatcherMain)
+    from .service.metrics import DispatcherMetricsServer
+
+    ret['service']['metrics_kwargs'] = schema_for_cls(DispatcherMetricsServer)
     ret['service']['process_manager_kwargs'] = {}
     pm_classes = (process.ProcessManager, process.ForkServerManager)
     for pm_cls in pm_classes:

dispatcherd/protocols.py

@@ -193,6 +193,17 @@ def get_status_data(self) -> dict:
     async def shutdown(self) -> None: ...
 
 
+class DispatcherMetricsServer(Protocol):
+    """
+    Metrics object will be created by factories.
+
+    If the extra is not installed, we may not have needed python dependencies.
+    This only needs to capture the interace of the initialized object with DispatcherMain.
+    """
+
+    async def start_server(self, dispatcher: 'DispatcherMain') -> None: ...
+
+
 class DispatcherMain(Protocol):
     """
     Describes the primary dispatcherd interface.
@@ -207,6 +218,9 @@ class DispatcherMain(Protocol):
     fd_lock: asyncio.Lock  # Forking and locking may need to be serialized, which this does
     producers: Iterable[Producer]
 
+    received_count: int
+    control_count: int
+
     async def main(self) -> None:
         """This is the method that runs the service, bring your own event loop"""
         ...