131 lines
3.7 KiB
Python
Executable File
131 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())
|