icinga2-checks/check_matrix_msg.py

91 lines
3.2 KiB
Python
Raw Normal View History

import argparse
import asyncio
import sys
import time
from nio import JoinResponse, RoomSendResponse
from checker import nagios, print_icinga2_check_status
from checker.synapse_client import login_and_cache
async def main(args):
start_time = time.time()
client = await login_and_cache(args.username, args.password, args.homeserver, auth_file=args.auth_file)
join_response = await client.join(args.room)
if not isinstance(join_response, JoinResponse):
print_icinga2_check_status(f'failed to join room!\n{join_response}', nagios.STATE_CRIT)
sys.exit(nagios.STATE_CRIT)
time.sleep(1)
test_msg = f'TEST MESSAGE AT {time.time()}'
send_response = await client.room_send(
room_id=args.room,
message_type="m.room.message",
content={"msgtype": "m.text", "body": test_msg},
)
if not isinstance(send_response, RoomSendResponse):
print_icinga2_check_status(f'failed to send message!\n{send_response}', nagios.STATE_CRIT)
sys.exit(nagios.STATE_CRIT)
time.sleep(1)
event_id = send_response.event_id
found_msg = False
for i in range(args.tries):
sync_response = await client.sync(30000)
room_timeline = sync_response.rooms.join[args.room].timeline
if any(event.event_id == event_id for event in room_timeline.events):
found_msg = True
break
await client.close()
if not found_msg:
print_icinga2_check_status(f'failed to find our message in the room timeline', nagios.STATE_CRIT)
sys.exit(nagios.STATE_CRIT)
elapsed_time = round((time.time() - start_time) - 2, 2)
perfdata = {
'elapsed': {
'value': elapsed_time,
'warn': args.warn,
'crit': args.critical,
'min': 0,
'unit': 's'
}
}
if elapsed_time >= args.critical:
return_code = nagios.STATE_CRIT
return_msg = f'message send time was {elapsed_time} seconds'
elif elapsed_time >= args.warn:
return_code = nagios.STATE_WARN
return_msg = f'message send time was {elapsed_time} seconds'
else:
return_code = nagios.STATE_OK
return_msg = 'message send successful'
print_icinga2_check_status(return_msg, return_code, perfdata=perfdata)
sys.exit(return_code)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='Test sending message to a Matrix room.')
parser.add_argument('-s', '--homeserver', required=True, help='Your Matrix server. Example: https://your-homeserver.org')
parser.add_argument('-u', '--username', required=True, help='Matrix username.')
parser.add_argument('-p', '--password', required=True, help='Matrix password.')
parser.add_argument('-r', '--room', required=True, help='Room to send a message to.')
parser.add_argument('-a', '--auth-file', help="File to cache the bot's login details to.")
parser.add_argument('-t', '--tries', type=int, default=10, help="How many times to check for the test message.")
parser.add_argument('--warn', type=int, default=20, help='Warn level for message send time.')
parser.add_argument('--critical', type=int, default=30, help='Critical level for message send time.')
args = parser.parse_args()
asyncio.run(main(args))