mirror of https://github.com/Icinga/icinga2.git
72 lines
2.4 KiB
Python
Executable File
72 lines
2.4 KiB
Python
Executable File
#!/usr/bin/env python
|
|
from __future__ import unicode_literals
|
|
|
|
import sys
|
|
|
|
import utils
|
|
import ido_tests
|
|
|
|
|
|
def main():
|
|
failures = False
|
|
|
|
run_query = lambda q: utils.run_pgsql_query(q, b'/usr/bin/psql')
|
|
|
|
if not ido_tests.validate_tables([d['Name'] for d in run_query('\\dt')
|
|
if d['Type'] == 'table']):
|
|
return 1 # Bail out as we cannot proceed without any data
|
|
|
|
host_info = run_query('select * from icinga_hosts')
|
|
if not ido_tests.verify_host_config(host_info):
|
|
return 1 # Bail out as we cannot proceed without any data
|
|
|
|
service_info = run_query(
|
|
'select c2.alias, c1.* from icinga_services as c1 '
|
|
'inner join icinga_hosts as c2'
|
|
' on c1.host_object_id = c2.host_object_id'
|
|
)
|
|
if not ido_tests.verify_service_config(service_info):
|
|
return 1 # Bail out as we cannot proceed without any data
|
|
|
|
hostchecks_data = run_query(
|
|
'select c.alias, unix_timestamp(s.last_check) as last_check'
|
|
' from icinga_hoststatus as s '
|
|
'inner join icinga_hosts as c'
|
|
' on s.host_object_id = c.host_object_id'
|
|
)
|
|
if not ido_tests.check_last_host_status_update(hostchecks_data):
|
|
failures = True
|
|
|
|
servicechecks_data = run_query(
|
|
'select c2.alias, c1.display_name, unix_timestamp(s.last_check) as last_check'
|
|
' from icinga_servicestatus as s '
|
|
'inner join icinga_services as c1'
|
|
' on s.service_object_id = c1.service_object_id '
|
|
'inner join icinga_hosts as c2'
|
|
' on c1.host_object_id = c2.host_object_id'
|
|
)
|
|
if not ido_tests.check_last_service_status_update(servicechecks_data):
|
|
failures = True
|
|
|
|
logentry_info = run_query(
|
|
'select hosts.alias,'
|
|
' max(unix_timestamp(logs.entry_time)) as entry_time,'
|
|
' max(unix_timestamp(hist.state_time)) as state_time'
|
|
' from icinga_logentries as logs '
|
|
'inner join icinga_hosts as hosts'
|
|
' on logs.object_id = hosts.host_object_id '
|
|
'inner join icinga_statehistory as hist'
|
|
' on hist.object_id = hosts.host_object_id '
|
|
"where hosts.alias = 'localhost' and hist.state_type = 1 "
|
|
'group by hosts.alias'
|
|
)
|
|
if not ido_tests.check_logentries(logentry_info):
|
|
failures = True
|
|
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|
|
|