Emit an alert on exception

src/nextline_alert/emitter.py

@@ -0,0 +1,44 @@
+import traceback
+from logging import getLogger
+
+import httpx
+from nextline.plugin.spec import Context, hookimpl
+
+
+class Emitter:
+    def __init__(self, url: str):
+        self._url = url
+        self._logger = getLogger(__name__)
+        self._logger.info(f'Campana endpoint: {url}')
+
+    @hookimpl
+    async def on_end_run(self, context: Context) -> None:
+        run_arg = context.run_arg
+        nextline = context.nextline
+        if e := nextline.exception():
+            run_no_str = 'unknown' if run_arg is None else f'{run_arg.run_no}'
+            alertname = f'Run {run_no_str} failed'
+            desc = ''.join(traceback.format_exception(type(e), e, e.__traceback__))
+            self._logger.info(f"Emitting alert: '{alertname}'")
+            try:
+                await emit(self._url, alertname, desc)
+            except BaseException:
+                self._logger.exception(f"Failed to emit alert: '{alertname}'")
+                self._logger.debug(f'Alert description: {desc!r}')
+
+
+async def emit(url: str, alertname: str, description: str) -> None:
+    data = {
+        'status': 'firing',
+        'alerts': [
+            {
+                'status': 'firing',
+                'labels': {'alertname': alertname},
+                'annotations': {'description': description, 'groups': 'nextline'},
+            }
+        ],
+    }
+
+    async with httpx.AsyncClient() as client:
+        response = await client.post(url, json=data)
+        response.raise_for_status()

src/nextline_alert/plugin.py

@@ -1,11 +1,13 @@
 from collections.abc import Mapping
 from pathlib import Path
-from typing import Optional
+from typing import Optional, cast
 
 from apluggy import asynccontextmanager
-from nextlinegraphql.hook import spec
 from dynaconf import Dynaconf, Validator
+from nextline import Nextline
+from nextlinegraphql.hook import spec
 
+from .emitter import Emitter
 
 HERE = Path(__file__).resolve().parent
 DEFAULT_CONFIG_PATH = HERE / 'default.toml'
@@ -30,8 +32,13 @@ def dynaconf_settings_files(self) -> Optional[tuple[str, ...]]:
     def dynaconf_validators(self) -> Optional[tuple[Validator, ...]]:
         return VALIDATORS
 
+    @spec.hookimpl
+    def configure(self, settings: Dynaconf):
+        self._url = settings.alert.campana_url
 
     @spec.hookimpl
     @asynccontextmanager
     async def lifespan(self, context: Mapping):
+        nextline = cast(Nextline, context['nextline'])
+        nextline.register(Emitter(url=self._url))
         yield