Checkmk-rocketchat-notifica.../notifications/rocketchat
DeltaLima (Marcus Hanisch) 9736ce3a20 initial commit
2024-03-06 11:36:40 +01:00

130 lines
3.7 KiB
Python
Executable file

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Rocketchat
# SPDX-FileCopyrightText: 2013 Mathias Kettner <mk@mathias-kettner.de>
# SPDX-FileCopyrightText: 2021 Stefan Gehn <stefan+cmk@srcbox.net>
# SPDX-FileCopyrightText: 2023 Mark Junghanns <mark@markjunghanns.de>
# SPDX-FileCopyrightText: 2024 DeltaLima (Marcus Hanisch) <dontsendmespam@deltalima.org>
#
# SPDX-License-Identifier: GPL-2.0-only
# Rocket.Chat notification based on asciimail notification from check_mk 1.2.6p16.
from __future__ import unicode_literals
import sys
import requests
from cmk.notification_plugins import utils
tmpl_host_text = """$HOSTSTATE_EMOJI$ `$HOSTNAME$`
```
Host: $HOSTNAME$
Event: $EVENT_TXT$
Output: $HOSTOUTPUT$
```
$PARAMETER_RC_CUSTOMTEXT$
"""
tmpl_service_text = """$SERVICESTATE_EMOJI$ `$HOSTNAME$/$SERVICEDESC$`
```
Host: $HOSTNAME$
Service: $SERVICEDESC$
Event: $EVENT_TXT$
Output: $SERVICEOUTPUT$
```
$PARAMETER_RC_CUSTOMTEXT$
"""
def hoststate_as_emoji(hoststate):
if hoststate == "UP":
return "\ud83d\udd35" # large blue circle
elif hoststate == "DOWN":
return "\ud83d\udd34" # large red circle
elif hoststate == "UNREACH":
return "\u26aa\ufe0f" # medium white circle
return hoststate
def servicestate_as_emoji(servicestate):
if servicestate == "OK":
return "\ud83c\udd97" # squared ok
elif servicestate == "WARN":
return "\u26a0\ufe0f" # warning sign
elif servicestate == "CRIT":
return "\u2757\ufe0f" # heavy exclamation mark symbol
elif servicestate == "UNKN":
return "\u2754" # white question mark ornament
return servicestate
def construct_message_text(context):
context["HOSTSTATE_EMOJI"] = hoststate_as_emoji(context.get("HOSTSHORTSTATE", ""))
context["SERVICESTATE_EMOJI"] = servicestate_as_emoji(
context.get("SERVICESHORTSTATE", "")
)
notification_type = context["NOTIFICATIONTYPE"]
if notification_type in ["PROBLEM", "RECOVERY"]:
txt_info = "$PREVIOUS@HARDSHORTSTATE$ -> $@SHORTSTATE$"
elif notification_type.startswith("FLAP"):
if "START" in notification_type:
txt_info = "Started Flapping"
else:
txt_info = "Stopped Flapping ($@SHORTSTATE$)"
elif notification_type.startswith("DOWNTIME"):
what = notification_type[8:].title()
txt_info = "Downtime " + what + " ($@SHORTSTATE$)"
elif notification_type == "ACKNOWLEDGEMENT":
txt_info = "Acknowledged ($@SHORTSTATE$)"
elif notification_type == "CUSTOM":
txt_info = "Custom Notification ($@SHORTSTATE$)"
else:
txt_info = notification_type # Should never happen
context["EVENT_TXT"] = utils.substitute_context(
txt_info.replace("@", context["WHAT"]), context
)
if context["WHAT"] == "HOST":
tmpl_text = tmpl_host_text
else:
tmpl_text = tmpl_service_text
return utils.substitute_context(tmpl_text, context)
def send_rocketchat_message(webhook_url, text):
url = format(webhook_url)
json = {
"text": text,
}
r = requests.post(url=url, json=json)
if r.status_code != 200:
sys.stderr.write(
"Failed to send Rocket.Chat message. Status: {}, Response: {}\n".format(
r.status_code, r.text
)
)
return 1 # Temporary error to make Checkmk retry
sys.stdout.write(
"Sent message to Rocket.Chat {}\n"
)
return 0
def main():
context = utils.collect_context()
webhook_url = context["PARAMETER_WEBHOOK_URL"]
rc_customtext = context["PARAMETER_RC_CUSTOMTEXT"]
text = construct_message_text(context)
return send_rocketchat_message(webhook_url, text)
if __name__ == "__main__":
sys.exit(main())