mirror of
https://github.com/zulip/zulip.git
synced 2025-11-16 03:41:58 +00:00
107 lines
3.0 KiB
Python
Executable File
107 lines
3.0 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
"""
|
|
Nagios plugin to check that messages take no longer than 15 seconds to
|
|
be received after being sent on Humbug.
|
|
|
|
It must be run on a machine that is using the live database for the
|
|
Django ORM.
|
|
"""
|
|
|
|
import datetime
|
|
import os
|
|
import sys
|
|
import optparse
|
|
|
|
usage = "usage: check_send_receive_time [options]"
|
|
parser = optparse.OptionParser(usage=usage)
|
|
parser.add_option('--site',
|
|
dest='site',
|
|
default="https://humbughq.com",
|
|
action='store')
|
|
|
|
(options, args) = parser.parse_args()
|
|
|
|
sys.path.append('/home/humbug/humbug/api')
|
|
import humbug
|
|
# Until we run scripts like this with a packaged and installed API, or
|
|
# rename the top-level humbug directory to something that doesn't
|
|
# conflict with an API module, we have to use hacks like this to be
|
|
# able to import both api.humbug and the top level humbug for
|
|
# humbug.settings.
|
|
del sys.modules["humbug"]
|
|
|
|
sys.path.insert(0, '/home/humbug/humbug')
|
|
os.environ['DJANGO_SETTINGS_MODULE'] = "humbug.settings"
|
|
from zephyr.models import UserProfile
|
|
|
|
states = {
|
|
"OK": 0,
|
|
"WARNING": 1,
|
|
"CRITICAL": 2,
|
|
"UNKNOWN": 3
|
|
}
|
|
|
|
def report(state, time, msg=None):
|
|
if msg:
|
|
print "%s: %s" % (state, msg)
|
|
else:
|
|
print "%s: send time was %s" % (state, time)
|
|
exit(states[state])
|
|
|
|
def send_humbug(sender, message):
|
|
result = sender.send_message(message)
|
|
if result["result"] != "success":
|
|
report("CRITICAL", "Error sending Humbug, args were: %s, %s" % (message, result))
|
|
|
|
def get_humbug(recipient, max_message_id):
|
|
return recipient.get_messages({'last': str(max_message_id)})['messages']
|
|
|
|
# Hamlet and Othello are default users
|
|
sender = UserProfile.objects.get(user__email="hamlet@humbughq.com")
|
|
recipient = UserProfile.objects.get(user__email="othello@humbughq.com")
|
|
|
|
humbug_sender = humbug.Client(
|
|
email=sender.user.email,
|
|
api_key=sender.api_key,
|
|
verbose=True,
|
|
client="test: Humbug API",
|
|
site=options.site)
|
|
|
|
humbug_recipient = humbug.Client(
|
|
email=recipient.user.email,
|
|
api_key=recipient.api_key,
|
|
verbose=True,
|
|
client="test: Humbug API",
|
|
site=options.site)
|
|
|
|
max_message_id = humbug_recipient.get_profile().get('max_message_id')
|
|
# This msg could be randomly generated everytime, may be better to do so.
|
|
msg_to_send = "Testing time to send and receive a message."
|
|
time_start = datetime.datetime.now()
|
|
|
|
send_humbug(humbug_sender, {
|
|
"type": 'private',
|
|
"content": msg_to_send,
|
|
"subject": "time to send",
|
|
"to": recipient.user.email,
|
|
})
|
|
|
|
msg_content = []
|
|
|
|
while msg_to_send not in msg_content:
|
|
messages = get_humbug(humbug_recipient, max_message_id)
|
|
time_diff = datetime.datetime.now() - time_start
|
|
|
|
# Prevents checking the same messages everytime in the conditional
|
|
# statement of the while loop
|
|
max_message_id = max([msg['id'] for msg in messages])
|
|
msg_content = [m['content'] for m in messages]
|
|
|
|
if time_diff.seconds > 3:
|
|
report('WARNING', time_diff)
|
|
if time_diff.seconds > 6:
|
|
report('CRITICAL', time_diff)
|
|
|
|
report('OK', time_diff)
|