Add Alertmanager notification

promgen/notification/alertmanager.py

@@ -0,0 +1,46 @@
+# Copyright (c) 2026 LINE Corporation
+# These sources are released under the terms of the MIT license: see LICENSE
+
+import logging
+
+from django import forms
+
+from promgen import util
+from promgen.notification import NotificationBase
+
+logger = logging.getLogger(__name__)
+
+
+class FormAlertmanager(forms.Form):
+    value = forms.URLField(
+        required=True,
+        label="URL",
+        help_text="URL to Alertmanager's Alerts API end point.",
+    )
+    alias = forms.CharField(
+        required=False,
+        help_text="Optional description to be displayed instead of the URL.",
+    )
+
+
+class NotificationAlertmanager(NotificationBase):
+    """
+    Send Promgen-managed alerts directly to user's Alertmanager instances.
+    """
+
+    form = FormAlertmanager
+
+    def _send(self, url, data):
+        alerts = data.get("alerts", [])
+        if alerts:
+            # Convert to Alertmanager format
+            # https://github.com/prometheus/alertmanager/blob/main/api/v2/openapi.yaml#L459
+            alertmanager_json = [
+                {
+                    k: alert.get(k)
+                    for k in ("startsAt", "endsAt", "annotations", "labels", "generatorURL")
+                    if k in alert
+                }
+                for alert in alerts
+            ]
+            util.post(url, json=alertmanager_json).raise_for_status()

promgen/tests/notification/alertmanager.json

@@ -0,0 +1,34 @@
+[
+  {
+    "annotations": {
+      "grafana": "http://grafana.example.com/dashboard/db/overview?var-service=test-service&var-project=test-project",
+      "summary": "Error on a.example"
+    },
+    "endsAt": "0001-01-01T00:00:00Z",
+    "generatorURL": "http://prometheus.example",
+    "labels": {
+      "alertname": "test-alert",
+      "instance": "a.example",
+      "project": "test-project",
+      "service": "test-service",
+      "severity": "major"
+    },
+    "startsAt": "2017-10-06T16:21:28.214665326+09:00"
+  },
+  {
+    "annotations": {
+      "grafana": "http://grafana.example.com/dashboard/db/overview?var-service=test-service&var-project=test-project",
+      "summary": "Error on b.example"
+    },
+    "endsAt": "0001-01-01T00:00:00Z",
+    "generatorURL": "http://prometheus.example",
+    "labels": {
+      "alertname": "test-alert",
+      "instance": "b.example",
+      "project": "test-project",
+      "service": "test-service",
+      "severity": "major"
+    },
+    "startsAt": "2017-10-06T16:21:28.214665326+09:00"
+  }
+]

promgen/tests/notification/test_alertmanager.py

@@ -0,0 +1,50 @@
+# Copyright (c) 2026 LINE Corporation
+# These sources are released under the terms of the MIT license: see LICENSE
+
+from unittest import mock
+
+from django.contrib.auth.models import Permission
+from django.test import override_settings
+
+from promgen import models, rest, tests
+from promgen.notification.alertmanager import NotificationAlertmanager
+
+
+class AlertmanagerTest(tests.PromgenTest):
+    def setUp(self):
+        one = models.Project.objects.get(pk=1)
+        two = models.Service.objects.get(pk=1)
+
+        self.senderA = NotificationAlertmanager.create(
+            obj=one, value="http://alertmanager/api/v2/alerts", owner_id=1
+        )
+        self.senderB = NotificationAlertmanager.create(
+            obj=two, value="http://alertmanager_2/api/v2/alerts", owner_id=1
+        )
+
+        self.user = self.force_login(username="demo")
+        permission = Permission.objects.get(codename="process_alert")
+        self.user.user_permissions.add(permission)
+
+    @override_settings(PROMGEN=tests.SETTINGS)
+    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
+    @override_settings(CELERY_TASK_EAGER_PROPAGATES=True)
+    @mock.patch("promgen.util.post")
+    def test_alertmanager(self, mock_post):
+        response = self.fireAlert()
+        self.assertRoute(response, rest.AlertReceiver, 202)
+        self.assertCount(models.AlertError, 0, "No failed alerts")
+
+        self.assertCount(models.Alert, 1, "Alert should be queued")
+        self.assertEqual(mock_post.call_count, 2, "Two alerts should be sent")
+
+        # Our sample is the same as the original's "alerts" field.
+        _SAMPLE = tests.Data("notification", "alertmanager.json").json()
+
+        mock_post.assert_has_calls(
+            [
+                mock.call("http://alertmanager/api/v2/alerts", json=_SAMPLE),
+                mock.call("http://alertmanager_2/api/v2/alerts", json=_SAMPLE),
+            ],
+            any_order=True,
+        )

pyproject.toml

@@ -56,6 +56,7 @@ promgen = "promgen.manage:main"
 promgen = "promgen.discovery.default:DiscoveryPromgen"
 
 [project.entry-points."promgen.notification"]
+alertmanager = "promgen.notification.alertmanager:NotificationAlertmanager"
 email = "promgen.notification.email:NotificationEmail"
 linenotify = "promgen.notification.linenotify:NotificationLineNotify"
 pagerduty = "promgen.notification.pagerduty:NotificationPagerDuty"