Skip to content

Instantly share code, notes, and snippets.

@stupoid

stupoid/lambda_canary.py

Last active Feb 23, 2021
Embed
What would you like to do?
Modified version of lambda-canary blueprint to monitor multiple sites and log via telegram_bot
import os
from urllib.parse import urlencode
from urllib.request import Request, urlopen
def getenvbool(key: str, default: bool) -> bool:
envvar = os.getenv(key, "").lower()
if envvar in {"true", "1"}:
return True
elif envvar in {"false", "0"}:
return False
else:
return default
CHECKS = os.environ[
"checks"
] # Multiple checks in the format "site1>expected_string_in_site1|site2>expected_string in site2"
TELEGRAM_BOT_TOKEN = os.environ[
"telegram_bot_token"
] # Token used by telegram bot to send message over telegram
TELEGRAM_CHANNEL_ID = os.environ[
"telegram_channel_id"
] # Used to identify channel to send message to (needed exact id for private channels)
NOTIFY_ON_SUCCESS = getenvbool(
"notify_on_success", False
) # Toggle send notification if nothing is wrong
def lambda_handler(event, context):
for check in CHECKS.split("|"):
site, expected = check.split(">", 1)
req = Request(site, headers={"User-Agent": "AWS Lambda"})
res = str(urlopen(req).read())
if expected not in res:
send_message(
f"Check <code>{expected}</code> at {site} <b>failed</b> at {event['time']}!",
disable_notification=False,
)
elif NOTIFY_ON_SUCCESS:
send_message(
f"Check <code>{expected}</code> at {site} <b>passed</b> at {event['time']}",
)
else:
print(f"Check {expected} at {site} passed at {event['time']}")
def send_message(text: str, disable_notification: bool = True):
params = urlencode(
{
"chat_id": TELEGRAM_CHANNEL_ID,
"text": text,
"parse_mode": "HTML",
"disable_notification": disable_notification,
}
)
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage?{params}"
urlopen(url)
@stupoid

This comment has been minimized.

Copy link
Owner Author

@stupoid stupoid commented Feb 23, 2021

Useful links:

Introduction to Telegram Bots
Telegram Bot API

Note to self:
As everything is sync and slow, a check that timed out may prevent subsequent checks from running.
Fix anything that that pops up ASAP.
Increase lambda timeout as needed.

TODO:

  • Make it async? (might avoid timeout issue)
  • Track check counts? (avoid silent failures) e.g. "1 out of 2 checks passed", "2 out of 2 checks passed"
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment