From 48d49f6dc3548991b87d1fd6904fdb25d5a4cfe4 Mon Sep 17 00:00:00 2001
From: Markus Opolka <markus@martialblog.de>
Date: Thu, 20 Jul 2023 14:05:00 +0200
Subject: [PATCH] Add example script to notify alertmanager

---
 .gitignore                                  |   3 +-
 contrib/notify-alertmanager-example.py      | 118 ++++++++++++++++++++
 contrib/test_notify-alertmanager-example.py |  17 +++
 3 files changed, 137 insertions(+), 1 deletion(-)
 create mode 100644 contrib/notify-alertmanager-example.py
 create mode 100644 contrib/test_notify-alertmanager-example.py

diff --git a/.gitignore b/.gitignore
index 2c5c956..befc39e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,4 +3,5 @@ check_prometheus
 vendor/
 coverage.html
 coverage.out
-dist/
\ No newline at end of file
+dist/
+__pycache__/
\ No newline at end of file
diff --git a/contrib/notify-alertmanager-example.py b/contrib/notify-alertmanager-example.py
new file mode 100644
index 0000000..f57d01f
--- /dev/null
+++ b/contrib/notify-alertmanager-example.py
@@ -0,0 +1,118 @@
+#!/usr/bin/python
+
+
+from argparse import ArgumentParser
+from urllib import request
+import json
+import socket
+import sys
+
+
+DEBUG = False
+
+
+def cli(args):
+  """
+  Commmand Line Arguments
+  """
+
+  parser = ArgumentParser(description='Send notifications to the Prometheus Alertmanager')
+
+  parser.add_argument('--hostname', type=str, required=True)
+  parser.add_argument('--service', type=str, default='hostalive', required=True)
+  parser.add_argument('--output', type=str, default='')
+  parser.add_argument('--state', type=int, required=True, choices=[0,1,2,3])
+  parser.add_argument('--alert-api-url', type=str, default='http://localhost:9093/api/v1/alerts')
+
+  parser.add_argument('--debug', action='store_true')
+  parser.set_defaults(debug=False)
+
+  return parser.parse_args(args)
+
+
+def post_alert(url, data):
+  """
+  HTTP Post the data to URL
+  """
+  if DEBUG:
+    print('DEBUG: Posting Alert')
+
+  content = None
+  try:
+    req = request.Request(url, method="POST")
+    req.add_header('Content-Type', 'application/json')
+    r = request.urlopen(req, data=data)
+    content = r.read()
+  except Exception as e:
+    print('ERROR:', e)
+    sys.exit(1)
+
+  if DEBUG:
+    print('DEBUG: Server Return:')
+    print(content)
+
+
+def generate_alert(args):
+  if DEBUG:
+    print('DEBUG: Generating Alert')
+
+  # Default Value
+  state_name_mapping = {
+    "0": "Up",
+    "1": "Up",
+    "2": "Down",
+    "3": "Down"
+  }
+
+  # Change mapping if hostalive
+  if args.service != "hostalive":
+    state_name_mapping = {
+      "0": "OK",
+      "1": "Warning",
+      "2": "Critical",
+      "3": "Unknown"
+    }
+
+  alert_state_name_mapping = {
+    "0": "resolved",
+    "1": "firing",
+    "2": "firing",
+    "3": "firing"
+  }
+
+  status = state_name_mapping[str(args.state)]
+  alert_status = alert_state_name_mapping[str(args.state)]
+
+  return [{
+    "status": alert_status,
+    "generatorURL": "foo",
+    "labels": {
+      "alertname": f"{args.service}_{args.hostname}",
+      "instance": args.hostname,
+      "service": args.service,
+    },
+    "annotations": {
+      "summary": f"Service {args.service} on {args.hostname} is {status}",
+    }
+  }]
+
+def main(args):
+
+  DEBUG = args.debug
+
+  if DEBUG:
+    print('DEBUG: CLI Arguments')
+    print('DEBUG:', args)
+
+  alert = generate_alert(args)
+  alert_json = json.dumps(alert).encode()
+
+  if DEBUG:
+    print('DEBUG: Alert to be posted')
+    print('DEBUG:', alert)
+
+  post_alert(args.alert_api_url, alert_json)
+
+if __name__ == "__main__":
+  ARGS = cli(sys.argv[1:])
+  main(ARGS)
diff --git a/contrib/test_notify-alertmanager-example.py b/contrib/test_notify-alertmanager-example.py
new file mode 100644
index 0000000..4a3b8d8
--- /dev/null
+++ b/contrib/test_notify-alertmanager-example.py
@@ -0,0 +1,17 @@
+import unittest
+import importlib
+
+loader = importlib.machinery.SourceFileLoader('notify', 'notify-alertmanager-example.py')
+mod = loader.load_module()
+
+class TestNotify(unittest.TestCase):
+
+    def test_generate_alert(self):
+        args = mod.cli(['--debug', '--alert-api-url', 'https://localhost', '--hostname', 'barfoo', '--state', '1', '--service', 'foobar'])
+        actual = mod.generate_alert(args)[0]
+        self.assertEqual(actual['labels']['service'], 'foobar')
+        self.assertEqual(actual['labels']['instance'], 'barfoo')
+        self.assertEqual(actual['labels']['alertname'], 'foobar_barfoo')
+
+if __name__ == '__main__':
+    unittest.main()