icingabot/icingabot.py

366 rivejä
12 KiB
Python
Executable File

#!/usr/bin/python3
#
# Simple IRC Bot statusing icinga2 via its API.
#
# François Poulain <fpoulain@metrodore.fr>
#
#
# Based on example program using irc.bot.
# from
# Joel Rosdahl <joel@rosdahl.net>
# https://github.com/jaraco/irc/raw/master/scripts/testbot.py
import configparser
import json
import os.path
import re
import sys
from itertools import groupby
import requests
import irc.bot
"""A simple IRC Bot statusing icinga2.
It is based on TestBot example bot that uses the SingleServerIRCBot class from
irc.bot. The bot enters a channel and listens for commands in private messages
and channel traffic. Commands in channel messages are given by prefixing the
text by the bot name followed by a colon. Periodically, the bot read Icinga2
status and report all new KO services.
Requirements: python3-irc python3-requests
The known commands are:
"""
commands = {
"ack": {"help": "Acknowledge all services matching the query."},
"recheck": {
"help": "Recheck all services matching the query.",
"synonyms": [r"refresh"],
},
"list": {"help": "List all KO services.", "synonyms": [r"lsit", r"lits"]},
"leave": {
"help": "Disconnect the bot."
" The bot will try to reconnect after 300 seconds.",
},
"mute": {
"help": "Mute the bot (no more status report)."
" The bot will unmute after 1 hour or after receiving any command.",
"synonyms": [
r"fuck",
r"chut",
r"couché",
r"sieste",
r"t(a|o)\s*g(ueu|o)le",
],
},
"die": {"help": "Let the bot cease to exist."},
"help": {"help": "Print help."},
}
# Load configuration.
configurationFilename = "/etc/icingabot/icingabot.conf"
if os.path.isfile(configurationFilename):
config = configparser.RawConfigParser()
config.read(configurationFilename)
settings = {
"ircsrv": config.get("irc", "irc.server"),
"ircchan": config.get("irc", "irc.chan"),
"ircport": config.getint("irc", "irc.port"),
"ircnick": config.get("irc", "irc.nick"),
"irccmd": config.get("irc", "irc.cmd_prefix"),
# see /etc/icinga2/conf.d/api-users.conf
"icinga2user": config.get("icinga2", "icinga2.user"),
"icinga2pass": config.get("icinga2", "icinga2.password"),
"icinga2ca": config.get("icinga2", "icinga2.ca"),
"icinga2fqdn": config.get("icinga2", "icinga2.fqdn"),
"icinga2port": config.getint("icinga2", "icinga2.port"),
}
else:
print("Missing configuration file [/etc/icingabot/icingabot.conf].")
sys.exit()
class Icinga2ServiceManager:
ko_services = []
error_counter = 0
def build_request_url(self, uri, params={}):
# Since icinga2 wants « URL-encoded strings » but requests
# seems to makes « application/x-www-form-urlencoded »,
# so we munge params and shortcut urlencode.
# See
# https://www.icinga.com/docs/icinga2/latest/doc/12-icinga2-api/#parameters
url = "https://{}:{}{}".format(
settings["icinga2fqdn"], settings["icinga2port"], uri
)
return (
requests.Request("GET", url, params=params)
.prepare()
.url.replace("+", "%20")
)
def fetch_ko_services(self):
headers = {
"Accept": "application/json",
"X-HTTP-Method-Override": "GET",
}
data = {
"attrs": [
"last_check_result",
"display_name",
"host_name",
"acknowledgement",
],
"filter": "service.state!=ServiceOK",
}
try:
r = requests.post(
self.build_request_url("/v1/objects/services"),
headers=headers,
auth=(settings["icinga2user"], settings["icinga2pass"]),
data=json.dumps(data),
verify=settings["icinga2ca"],
)
self.error_counter = 0
if r.status_code == 200:
new_ko_services = [n for n in r.json()["results"] if n is not None]
news = [
n
for n in new_ko_services
if n["name"] not in [nn["name"] for nn in self.ko_services]
]
lost = [
n
for n in self.ko_services
if n["name"] not in [nn["name"] for nn in new_ko_services]
]
self.ko_services = new_ko_services
return (lost, news)
except Exception as e:
if self.error_counter < 10:
self.send("Unable to fetch from Icinga2: {}".format(e))
self.error_counter += 1
return (False, False)
def post_on_services(self, pattern, uri, data={}):
headers = {
"Accept": "application/json",
"X-HTTP-Method-Override": "POST",
}
params = {"type": "Service", "filter": "service.state!=ServiceOK"}
if pattern:
params['filter'] += '&& match("*{}*", service.__name)'.format(pattern)
try:
r = requests.post(
self.build_request_url(uri, params=params),
headers=headers,
auth=(settings["icinga2user"], settings["icinga2pass"]),
data=json.dumps(data),
verify=settings["icinga2ca"],
)
for a in r.json()["results"]:
self.send(a["status"])
if pattern is not None and not r.json()["results"]:
self.send("No matching service for « {} »".format(pattern))
except Exception as e:
self.send("Unable to post to Icinga2: {}".format(e))
def ack_service(self, pattern, comment, nick):
data = {
"author": nick,
"comment": comment or " ", # never "" !
}
self.post_on_services(pattern, '/v1/actions/acknowledge-problem', data)
def recheck_service(self, pattern):
self.post_on_services(pattern, '/v1/actions/reschedule-check')
class IcingaBot(Icinga2ServiceManager, irc.bot.SingleServerIRCBot):
args = ""
muted = False
nick_suffix = ""
def __init__(self, channel, nickname, server, port=6667):
irc.bot.SingleServerIRCBot.__init__(
self,
[(server, port)],
nickname,
nickname,
reconnection_interval=300,
)
self.nick = nickname
self.channel = channel
self.connection.execute_every(30, self.refresh_ko_services)
self.connection.execute_every(300, self.refresh_nick)
self.refresh_ko_services()
def refresh_nick(self):
if self.connection.is_connected():
self.connection.nick("{}{}".format(self.nick, self.nick_suffix))
def unmute(self):
self.muted = False
if self.ko_services:
self.nick_suffix = "[{}]".format(len(self.ko_services))
else:
self.nick_suffix = ""
def send(self, msg):
if not self.muted and self.connection.is_connected():
for line in msg.split("\n"):
self.connection.privmsg(self.channel, line)
def refresh_ko_services(self):
lost, news = self.fetch_ko_services()
if lost is False and news is False:
return
if self.ko_services:
self.nick_suffix = "[{}]".format(len(self.ko_services))
else:
self.nick_suffix = ""
return
def on_nicknameinuse(self, c, e):
c.nick(c.get_nickname() + "_")
def on_welcome(self, c, e):
c.join(self.channel)
def on_privmsg(self, c, e):
self.do_command(e, e.arguments[0])
def on_pubmsg(self, c, e):
if e.arguments[0].startswith(settings["irccmd"]):
self.do_command(e, e.arguments[0][1:])
return
a = e.arguments[0].split(":", 1)
if len(a) > 1 and a[0].lower() == self.connection.get_nickname().lower():
self.do_command(e, a[1].strip())
return
def do_cmd(self, s):
self.args = None
tokens = s.split(" ", 1)
cmd = tokens[0]
if len(tokens) > 1:
self.args = tokens[1].strip()
for k, v in commands.items():
if cmd == k and hasattr(self, "do_" + cmd):
return getattr(self, "do_" + cmd)
if "synonyms" in v:
for s in v["synonyms"]:
if re.match(s, cmd) and hasattr(self, "do_" + k):
return getattr(self, "do_" + k)
return False
def do_help(self, c, e):
for k, v in commands.items():
self.send("{} -- {}".format(k, v["help"]))
if "synonyms" in v:
self.send(" synonyms: {}".format(", ".join(v["synonyms"])))
def do_die(self, c, e):
print("Ok master, I die. Aaargh...")
self.die()
def do_leave(self, c, e):
self.disconnect()
def get_unack_ko_services(self):
return [s for s in self.ko_services if not s["attrs"]["acknowledgement"]]
def regrouped_ko_services(self):
def regroup_key(elem):
return elem["attrs"]["display_name"]
return [
(group, [service["attrs"]["host_name"] for service in services])
for group, services in groupby(
sorted(self.get_unack_ko_services(), key=regroup_key),
regroup_key,
)
]
def do_list(self, c, e):
if self.ko_services:
host_by_service = [
(service, [hostname.split(".")[0] for hostname in hosts])
for service, hosts in sorted(
self.regrouped_ko_services(), key=lambda x: -len(x[1])
)
]
self.send(
"\n".join(
[
"{} ({}): {}".format(
service, len(hostnames), ", ".join(hostnames)
)
for service, hostnames in host_by_service
]
)
)
acknowledged = [
s for s in self.ko_services if s["attrs"]["acknowledgement"]
]
if acknowledged:
self.send(
"Acknowledged ({}): {}".format(
len(acknowledged),
', '.join({s["attrs"]["display_name"] for s in acknowledged}),
)
)
else:
self.send("Nothing particularly exciting.")
def do_ack(self, c, e):
if self.args is None:
self.send(e.source.nick + ": usage: !ack <pattern or all> [: comment]")
return
tokens = self.args.split(":", 1)
pattern, comment = tokens[0].strip(), ""
if len(tokens) > 1:
comment = tokens[1].strip()
if pattern == "all":
pattern = None
self.ack_service(pattern, comment, e.source.nick)
def do_recheck(self, c, e):
if self.args == "all":
self.args = None
self.recheck_service(self.args)
self.connection.execute_delayed(15, self.refresh_ko_services)
def do_mute(self, c, e):
self.muted = True
self.nick_suffix = "[zZz]"
self.connection.execute_delayed(3600, self.refresh_ko_services)
def do_command(self, e, cmd):
nick = e.source.nick
c = self.connection
if self.do_cmd(cmd):
self.unmute()
(self.do_cmd(cmd))(c, e)
else:
self.send(nick + ": Not understood: " + cmd)
def main():
bot = IcingaBot(
settings["ircchan"],
settings["ircnick"],
settings["ircsrv"],
settings["ircport"],
)
bot.start()
if __name__ == "__main__":
main()