bundlewrap/bundles/icinga2/files/scripts/icinga_notification_wrapper

180 lines
4.3 KiB
Text
Raw Normal View History

2020-11-22 08:04:02 +00:00
#!/usr/bin/env python3
import email.mime.text
import smtplib
from argparse import ArgumentParser
from json import dumps
2020-11-22 08:04:02 +00:00
from subprocess import run
from sys import argv
2023-02-05 16:30:58 +00:00
from requests import post
SIPGATE_USER='${node.metadata['icinga2']['sipgate_user']}'
SIPGATE_PASS='${node.metadata['icinga2']['sipgate_pass']}'
STATUS_TO_EMOJI = {
'critical': '🔥',
'down': '🚨🚨🚨',
'ok': '🆗',
'up': '👌',
'warning': '⚡',
}
2020-11-22 08:04:02 +00:00
parser = ArgumentParser(
prog='icinga_notification_wrapper',
description='Icinga2 Notification Wrapper',
)
parser.add_argument(
'--host_name',
type=str,
required=True,
)
parser.add_argument(
'--service_name',
type=str,
)
parser.add_argument(
'--output',
type=str,
required=True,
)
parser.add_argument(
'--state',
type=str,
required=True,
)
parser.add_argument(
'--sms',
type=str,
)
parser.add_argument(
'--mail',
type=str,
)
parser.add_argument(
'--notification-type',
type=str,
required=True,
2020-11-22 08:04:02 +00:00
)
def log_to_syslog(message):
try:
msg = '{}/{}: {}'.format(args.host_name, args.service_name, message)
run(['logger', '-t', 'icinga_notification_wrapper', msg])
except:
# We don't expect this to fail. However, *if* it fails, we don't
# want it to get in the way of other notifications we may have
# to send.
pass
def notify_per_sms():
if args.notification_type.lower() == 'recovery':
output_text = ''
else:
output_text = '\n\n{}'.format(args.output)
if args.state.lower() in STATUS_TO_EMOJI:
message_text = '{emoji} {host}{service} {emoji}{output}'.format(
emoji=STATUS_TO_EMOJI[args.state.lower()],
host=args.host_name,
service=('/'+args.service_name if args.service_name else ''),
state=args.state.upper(),
output=output_text,
)
else:
message_text = 'ICINGA: {host}{service} is {state}{output}'.format(
host=args.host_name,
service=('/'+args.service_name if args.service_name else ''),
state=args.state.upper(),
output=output_text,
)
message = {
'message': message_text,
'smsId': 's0', # XXX what does this mean? Documentation is unclear
'recipient': args.sms
}
headers = {
'Content-type': 'application/json',
'Accept': 'application/json'
}
try:
r = post(
'https://api.sipgate.com/v2/sessions/sms',
json=message,
headers=headers,
auth=(SIPGATE_USER, SIPGATE_PASS),
)
if r.status_code == 204:
log_to_syslog('Sent a SMS to "{}"'.format(args.sms))
else:
r.raise_for_status()
except Exception as e:
log_to_syslog('Sending a SMS to "{}" failed: {}'.format(args.sms, repr(e)))
2020-11-22 08:04:02 +00:00
def notify_per_mail():
if args.notification_type.lower() == 'recovery':
# Do not send recovery emails.
return
# editorconfig-checker-disable
2020-11-22 08:04:02 +00:00
text = """
_ _
(_)____(_)___ ____ _____ _
/ / ___/ / __ \/ __ `/ __ `/
/ / /__/ / / / / /_/ / /_/ /
/_/\___/_/_/ /_/\__, /\__,_/
/____/
Host: {host}"""
if args.service_name:
text += """
Service: {service}"""
text += """
State: {state}
{output}"""
# editorconfig-checker-enable
2020-11-22 08:04:02 +00:00
mail = email.mime.text.MIMEText(text.format(
host=args.host_name,
service=args.service_name,
state=args.state.upper(),
2020-11-22 08:04:02 +00:00
output=args.output
),
'plain',
'utf-8',
)
if args.service_name:
mail['Subject'] = '[ICINGA] {}/{} is {}'.format(args.host_name, args.service_name, args.state)
else:
mail['Subject'] = '[ICINGA] {} is {}'.format(args.host_name, args.state)
mail['To'] = args.mail
mail['From'] = 'noreply+icinga@kunbox.net'
try:
s = smtplib.SMTP('localhost')
s.sendmail(mail['From'], [args.mail], mail.as_string())
s.quit()
log_to_syslog('Sent mail to "{}"'.format(args.mail))
except Exception as e:
log_to_syslog('Sending mail to "{}" failed: {}'.format(args.mail, repr(e)))
if __name__ == '__main__':
args = parser.parse_args()
log_to_syslog(' '.join(argv))
if args.mail:
2020-11-22 08:04:02 +00:00
notify_per_mail()
if args.sms:
notify_per_sms()