pylib/nagios
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

nagios status: added possibilty to report recent problems of the last -l hours

Browse Source
This commit is contained in:
Dirk Alders 2024-08-16 07:33:05 +02:00
parent fd556d92ad
commit 930f4d13e1
1 changed files with 42 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
status.py
View File

@ -36,17 +36,20 @@ class nagios_service(dict):
SID_ERROR = 16
#
SNAME_OK = "OKAY"
SNAME_WARNING_LAST = "WARNING_LAST"
SNAME_ERROR_LAST = "ERROR_LAST"
SNAME_FLAPPING = "FLAPPING"
SNAME_WARNING = "WARNING"
SNAME_ERROR = "ERROR"
SNAME_UNKNOWN = "UNKNOWN"
TIMEFORMAT = "%d.%m.%Y %H:%M"
def __init__(self, service_dict) -> None:
def __init__(self, service_dict, last) -> None:
super().__init__(service_dict)
self.__last__ = last
def is_problem(self):
return self["status"] != self.SID_OK or self.flapping()
return self.status() != self.SNAME_OK
def host_name(self):
return self['host_name']
@ -57,8 +60,14 @@ class nagios_service(dict):
def status(self):
status = self["status"]
default = self.SNAME_UNKNOWN
if status == self.SID_OK and self.flapping():
return self.SNAME_FLAPPING
tm_interrest = time.localtime(time.time() - 60 * 60 * self.__last__)
if status == self.SID_OK:
if self.flapping():
return self.SNAME_FLAPPING
elif self.__last_critical__() > tm_interrest:
return self.SNAME_ERROR_LAST
elif self.__last_warning__() > tm_interrest:
return self.SNAME_WARNING_LAST
return {
self.SID_OK: self.SNAME_OK,
self.SID_WARNING: self.SNAME_WARNING,
@ -68,6 +77,21 @@ class nagios_service(dict):
def last_okay(self):
return time.strftime(self.TIMEFORMAT, time.localtime(self["last_time_ok"] / 1000))
def __last_warning__(self):
return time.localtime(self["last_time_warning"] / 1000)
def last_warning(self):
return time.strftime(self.TIMEFORMAT, self.__last_warning__())
def __last_critical__(self):
return time.localtime(self["last_time_critical"] / 1000)
def last_critical(self):
return time.strftime(self.TIMEFORMAT, self.__last_critical__())
def last_unknown(self):
return time.strftime(self.TIMEFORMAT, time.localtime(self["last_time_unknown"] / 1000))
def info(self):
return self.get("plugin_output") or self.get("long_plugin_output")
@ -77,29 +101,29 @@ class nagios_service(dict):
def __color__(self):
return {
self.SNAME_OK: colors.OKGREEN,
self.SNAME_WARNING_LAST: colors.OKBLUE,
self.SNAME_ERROR_LAST: colors.OKBLUE,
self.SNAME_FLAPPING: colors.OKBLUE,
self.SNAME_WARNING: colors.WARNING,
self.SNAME_ERROR: colors.FAIL,
self.SNAME_UNKNOWN: colors.OKCYAN
}.get(self.status())
return colors.OKBLUE
def __head__(self, color=False):
rv = headline(color)
rv += "+-----------------------------------------------------------------------------------------\n"
rv += "| Host | Service Name | State | Last time Okay |\n"
rv += "+---------------------------+-----------------+----------+------------------+-------------\n"
rv += "+---------------------------------------------------------------------------------------------\n"
rv += "| Host | Service Name | State | Last time Okay |\n"
rv += "+---------------------------+-----------------+--------------+------------------+-------------\n"
return rv
def __foot__(self):
return "+---------------------------+-----------------+----------+------------------+-------------\n"
return "+---------------------------+-----------------+--------------+------------------+-------------\n"
def __str__(self, color=False):
cols = []
cols.append((self.host_name, 25, False))
cols.append((self.name, 15, color))
cols.append((self.status, 8, color))
cols.append((self.status, 12, color))
cols.append((self.last_okay, 16, color))
cols.append((self.info, 0, False))
#
@ -136,13 +160,13 @@ class __nagios_status__(dict):
for host in self:
for service in self[host]:
if service.is_problem():
rv.__add_service__(host, service)
rv.__add_service__(host, service, service.__last__)
return rv
def __add_service__(self, host, service):
def __add_service__(self, host, service, last):
if host not in self:
self[host] = []
self[host].append(nagios_service(service))
self[host].append(nagios_service(service, last))
def __str__(self, color=False) -> str:
rv = ""
@ -164,7 +188,7 @@ class nagios_status(__nagios_status__):
KEY_DATA = 'data'
KEY_SERVICELIST = 'servicelist'
def __init__(self, host, secure) -> None:
def __init__(self, host, secure, last) -> None:
super().__init__()
#
if secure:
@ -177,7 +201,7 @@ class nagios_status(__nagios_status__):
hosts = data.get(self.KEY_DATA, {}).get(self.KEY_SERVICELIST, {})
for host in hosts:
for service in hosts.get(host, {}):
self.__add_service__(host, hosts.get(host).get(service))
self.__add_service__(host, hosts.get(host).get(service), last)
if __name__ == "__main__":
@ -185,12 +209,13 @@ if __name__ == "__main__":
parser.add_argument("hostname", help="The hostname and port of the nagios server (e.g. nagios:8080)")
parser.add_argument("-a", "--all", action="store_true", default=False,
help="print the status of all nagios monitorings. Default is all problems.")
parser.add_argument("-last", "--last", type=int, default=0, help="Report problems of the last l hours.")
parser.add_argument("-s", "--secure", action="store_true", default=False, help="Enables secure connection (https)")
parser.add_argument("-m", "--monochrome", action="store_true", default=False, help="No colored output")
args = parser.parse_args()
#
try:
s = nagios_status(args.hostname, args.secure)
s = nagios_status(args.hostname, args.secure, args.last)
except requests.ConnectionError:
print(headline(not args.monochrome) + "Can not connect to given host.")
else: