ac/ac-deploy
1
0
Fork 0
mirror of git://f0xx.org/ac/ac-deploy synced 2026-07-29 02:57:38 +03:00
Code Issues Packages Projects Releases Wiki Activity

added 'monitoring' path - ask devs why it's there

Browse Source
This commit is contained in:
Anton Afanasyeu
2026-06-28 22:01:59 +02:00
parent 3c5233c903
commit 2b2c32cff5
13 changed files with 1146 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

371
monitoring/cast04/alert-notifier.py Normal file
View File

@@ -0,0 +1,371 @@
#!/usr/bin/env python3
"""
Alertmanager webhook receiver — alert-notifier.py
Runs on cast04:9099. Receives JSON from Alertmanager, then for each batch:
1. Shortens the alert URL via the project URL shortener (s.f0xx.org)
2. Fetches QR PNG from the shortener's built-in QR endpoint
3. Sends HTML e-mail with the QR embedded as inline image
4. Sends Telegram message with the QR image attached
Config is read from alert-notifier.env in the same directory as this script.
"""
import http.server
import json
import smtplib
import ssl
import subprocess
import urllib.request
import urllib.parse
import email.mime.multipart
import email.mime.text
import email.mime.image
import email.utils
import logging
import os
import sys
import pathlib
# ── Config defaults (overridden by .env file) ───────────────────────────────
SHORTENER_INTERNAL = os.environ.get('SHORTENER_INTERNAL', 'http://10.7.16.128')
SHORTENER_HOST_HDR = os.environ.get('SHORTENER_HOST_HDR', 's.f0xx.org')
SHORTENER_BEARER = os.environ.get('SHORTENER_BEARER', '')
SHORTENER_TTL = int(os.environ.get('SHORTENER_TTL', '86400')) # 1 day default
SMTP_HOST = os.environ.get('SMTP_HOST', 'smtp.gmail.com')
SMTP_PORT = int(os.environ.get('SMTP_PORT', '587'))
SMTP_USER = os.environ.get('SMTP_USER', 'bestcastr@gmail.com')
SMTP_PASS = os.environ.get('SMTP_PASS', '')
EMAIL_FROM = os.environ.get('EMAIL_FROM', 'Android Cast Alerts <bestcastr@gmail.com>')
EMAIL_TO = os.environ.get('EMAIL_TO', 'a.afanasieff@gmail.com')
TELEGRAM_BOT = os.environ.get('TELEGRAM_BOT', '')
TELEGRAM_CHAT = int(os.environ.get('TELEGRAM_CHAT', '0'))
LISTEN_PORT = int(os.environ.get('LISTEN_PORT', '9099'))
# Default URL to shorten when no alert-specific URL is available
FALLBACK_URL = os.environ.get(
'FALLBACK_URL',
'https://apps.f0xx.org/app/androidcast_project/alertmanager/'
)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s',
stream=sys.stdout,
)
log = logging.getLogger('alert-notifier')
# ── URL shortener ────────────────────────────────────────────────────────────
def shorten(url: str) -> dict:
"""
POST the destination URL (no ?src=qr) to the project URL shortener.
Returns dict with keys:
short_url – plain https://s.f0xx.org/SLUG (used as click-link in email/TG)
qr_url – https://s.f0xx.org/SLUG?src=qr (encoded into the QR image)
slug – bare slug
Raises on failure.
"""
payload = json.dumps({
'url': url,
'bearer': SHORTENER_BEARER,
'ttl': SHORTENER_TTL,
}).encode()
req = urllib.request.Request(
f'{SHORTENER_INTERNAL}/api/v1/shorten',
data=payload,
headers={
'Content-Type': 'application/json',
'Host': SHORTENER_HOST_HDR,
},
method='POST',
)
with urllib.request.urlopen(req, timeout=8) as resp:
data = json.loads(resp.read())
if data.get('code', '') != '0':
raise RuntimeError(f"shortener error: {data}")
short_url = data['u']
slug = short_url.rsplit('/', 1)[-1]
# The QR image encodes the short URL WITH ?src=qr so every mobile scan
# is counted separately from normal link clicks on the backend.
qr_url = short_url + ('&src=qr' if '?' in short_url else '?src=qr')
return {
'short_url': short_url,
'qr_url': qr_url,
'slug': slug,
}
def make_qr_png(url: str) -> bytes:
"""
Generate a QR code PNG for *url* using the system qrencode binary.
Returns raw PNG bytes. Raises on failure.
"""
result = subprocess.run(
['qrencode', '-t', 'PNG', '-o', '-', '-s', '6', '--', url],
capture_output=True,
check=True,
)
return result.stdout
# ── E-mail ───────────────────────────────────────────────────────────────────
SEVERITY_EMOJI = {
'critical': '🔴',
'warning': '🟡',
'info': '🔵',
}
STATUS_EMOJI = {
'firing': '🔥',
'resolved': '✅',
}
def _alert_html_row(alert: dict) -> str:
status = alert.get('status', 'firing')
labels = alert.get('labels', {})
anns = alert.get('annotations', {})
name = labels.get('alertname', '(unknown)')
instance = labels.get('instance', '')
severity = labels.get('severity', '')
summary = anns.get('summary', name)
desc = anns.get('description', '')
starts = alert.get('startsAt', '')[:19].replace('T', ' ')
s_emoji = STATUS_EMOJI.get(status, '')
sv_emoji = SEVERITY_EMOJI.get(severity, '')
color = '#c0392b' if severity == 'critical' else '#e67e22' if severity == 'warning' else '#2980b9'
return f"""
<tr>
<td style="padding:8px;border-bottom:1px solid #eee;">
<b style="color:{color};">{s_emoji} {sv_emoji} {status.upper()}</b>
&mdash; <b>{name}</b>
{f'<span style="color:#666;font-size:0.9em"> @ {instance}</span>' if instance else ''}
</td>
</tr>
<tr>
<td style="padding:4px 8px 12px 8px;border-bottom:1px solid #eee;color:#333;">
{f'<b>Summary:</b> {summary}<br>' if summary else ''}
{f'<b>Detail:</b> {desc}<br>' if desc else ''}
{f'<b>Fired:</b> {starts} UTC' if starts else ''}
</td>
</tr>"""
def send_email(subject: str, alerts: list, short_url: str, qr_png: bytes) -> None:
msg = email.mime.multipart.MIMEMultipart('related')
msg['From'] = EMAIL_FROM
msg['To'] = EMAIL_TO
msg['Subject'] = subject
msg['Date'] = email.utils.formatdate(localtime=False)
msg['Message-ID'] = email.utils.make_msgid(domain='cast04.intra.raptor.org')
rows = ''.join(_alert_html_row(a) for a in alerts)
html = f"""<!DOCTYPE html>
<html><body style="font-family:Arial,sans-serif;max-width:640px;margin:auto;">
<div style="background:#1a1a2e;padding:16px;border-radius:6px 6px 0 0;">
<h2 style="color:#fff;margin:0;">Android Cast — Monitoring Alert</h2>
</div>
<div style="border:1px solid #ddd;border-top:none;padding:16px;border-radius:0 0 6px 6px;">
<table width="100%" cellpadding="0" cellspacing="0">{rows}</table>
<hr style="margin:16px 0;">
<p style="margin:0;">
<b>View in Alertmanager:</b>
<a href="{short_url}">{short_url}</a>
</p>
<p style="margin:8px 0 4px 0;color:#555;font-size:0.85em;">
Scan QR to open on mobile (tracked via url-shortener):
</p>
<img src="cid:alert_qr" alt="QR code" width="200" height="200"
style="display:block;border:1px solid #ccc;padding:4px;border-radius:4px;">
</div>
<p style="color:#999;font-size:0.75em;text-align:center;margin-top:8px;">
cast04 monitoring · <a href="https://apps.f0xx.org/app/androidcast_project/monitor/">Grafana</a>
</p>
</body></html>"""
alternative = email.mime.multipart.MIMEMultipart('alternative')
alternative.attach(email.mime.text.MIMEText(html, 'html', 'utf-8'))
msg.attach(alternative)
qr_img = email.mime.image.MIMEImage(qr_png, 'png')
qr_img.add_header('Content-ID', '<alert_qr>')
qr_img.add_header('Content-Disposition', 'inline', filename='alert-qr.png')
msg.attach(qr_img)
ctx = ssl.create_default_context()
with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=15) as s:
s.ehlo()
s.starttls(context=ctx)
s.ehlo()
s.login(SMTP_USER, SMTP_PASS)
s.sendmail(SMTP_USER, EMAIL_TO, msg.as_bytes())
log.info('email sent to %s subject=%r', EMAIL_TO, subject)
# ── Telegram ─────────────────────────────────────────────────────────────────
def send_telegram(text: str, qr_png: bytes) -> None:
if not TELEGRAM_BOT or not TELEGRAM_CHAT:
return
import io, email.generator
boundary = '----WebKitFormBoundary7MA4YWxkTrZu0gW'
# Build multipart/form-data body manually
body = (
f'--{boundary}\r\n'
f'Content-Disposition: form-data; name="chat_id"\r\n\r\n'
f'{TELEGRAM_CHAT}\r\n'
f'--{boundary}\r\n'
f'Content-Disposition: form-data; name="caption"\r\n\r\n'
f'{text}\r\n'
f'--{boundary}\r\n'
f'Content-Disposition: form-data; name="parse_mode"\r\n\r\n'
f'HTML\r\n'
f'--{boundary}\r\n'
f'Content-Disposition: form-data; name="photo"; filename="alert-qr.png"\r\n'
f'Content-Type: image/png\r\n\r\n'
).encode() + qr_png + f'\r\n--{boundary}--\r\n'.encode()
req = urllib.request.Request(
f'https://api.telegram.org/bot{TELEGRAM_BOT}/sendPhoto',
data=body,
headers={'Content-Type': f'multipart/form-data; boundary={boundary}'},
)
with urllib.request.urlopen(req, timeout=15) as resp:
result = json.loads(resp.read())
if not result.get('ok'):
raise RuntimeError(f"Telegram error: {result}")
log.info('telegram photo sent to chat_id=%s', TELEGRAM_CHAT)
# ── HTTP handler ─────────────────────────────────────────────────────────────
class WebhookHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args): # suppress default access log
log.debug(fmt, *args)
def do_POST(self):
if self.path != '/webhook':
self.send_response(404)
self.end_headers()
return
length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(length)
try:
payload = json.loads(body)
except json.JSONDecodeError:
self.send_response(400)
self.end_headers()
return
self.send_response(200)
self.end_headers()
# Process asynchronously (we already returned 200)
try:
self._handle(payload)
except Exception as exc:
log.error('webhook handler error: %s', exc, exc_info=True)
def _handle(self, payload: dict) -> None:
alerts = payload.get('alerts', [])
status = payload.get('status', 'firing')
ext_url = payload.get('externalURL', FALLBACK_URL)
if not alerts:
return
# ── shorten & QR ────────────────────────────────────────────────────
try:
short = shorten(ext_url)
# QR image encodes short_url?src=qr — generated locally via qrencode
qr_png = make_qr_png(short['qr_url'])
log.info('shortened %s → %s qr_url=%s (%d B)',
ext_url, short['short_url'], short['qr_url'], len(qr_png))
except Exception as exc:
log.error('shorten/qr failed: %s', exc)
return
# ── subject / Telegram text ──────────────────────────────────────────
firing_names = [a['labels'].get('alertname', '?') for a in alerts if a.get('status') == 'firing']
resolved_names = [a['labels'].get('alertname', '?') for a in alerts if a.get('status') == 'resolved']
if status == 'resolved':
subject = f'[RESOLVED] {", ".join(resolved_names or firing_names)}'
else:
subject = f'[ALERT] {", ".join(firing_names or [a["labels"].get("alertname","?") for a in alerts])}'
tg_lines = []
for a in alerts:
s = a.get('status', 'firing')
lbl = a.get('labels', {})
ann = a.get('annotations', {})
emoji = STATUS_EMOJI.get(s, '') + ' ' + SEVERITY_EMOJI.get(lbl.get('severity',''), '')
tg_lines.append(
f'{emoji} <b>{s.upper()}</b> {lbl.get("alertname","?")} @ {lbl.get("instance","")}\n'
f'{ann.get("summary","")}\n'
f'{ann.get("description","")}'
)
tg_text = '\n\n'.join(tg_lines) + f'\n\n🔗 {short["short_url"]}'
# ── send ─────────────────────────────────────────────────────────────
try:
send_email(subject, alerts, short['short_url'], qr_png)
except Exception as exc:
log.error('email failed: %s', exc)
try:
send_telegram(tg_text, qr_png)
except Exception as exc:
log.error('telegram failed: %s', exc)
# ── Entry point ───────────────────────────────────────────────────────────────
def load_env(path: str) -> None:
"""Load KEY=VALUE pairs from an .env file into os.environ."""
try:
with open(path) as f:
for line in f:
line = line.strip()
if not line or line.startswith('#') or '=' not in line:
continue
k, _, v = line.partition('=')
os.environ.setdefault(k.strip(), v.strip())
except FileNotFoundError:
pass
if __name__ == '__main__':
env_file = pathlib.Path(__file__).with_suffix('.env')
load_env(str(env_file))
# Re-read globals that may have been overridden by .env
SHORTENER_INTERNAL = os.environ.get('SHORTENER_INTERNAL', SHORTENER_INTERNAL)
SHORTENER_HOST_HDR = os.environ.get('SHORTENER_HOST_HDR', SHORTENER_HOST_HDR)
SHORTENER_BEARER = os.environ.get('SHORTENER_BEARER', SHORTENER_BEARER)
SHORTENER_TTL = int(os.environ.get('SHORTENER_TTL', str(SHORTENER_TTL)))
SMTP_HOST = os.environ.get('SMTP_HOST', SMTP_HOST)
SMTP_PORT = int(os.environ.get('SMTP_PORT', str(SMTP_PORT)))
SMTP_USER = os.environ.get('SMTP_USER', SMTP_USER)
SMTP_PASS = os.environ.get('SMTP_PASS', SMTP_PASS)
EMAIL_FROM = os.environ.get('EMAIL_FROM', EMAIL_FROM)
EMAIL_TO = os.environ.get('EMAIL_TO', EMAIL_TO)
TELEGRAM_BOT = os.environ.get('TELEGRAM_BOT', TELEGRAM_BOT)
TELEGRAM_CHAT = int(os.environ.get('TELEGRAM_CHAT', str(TELEGRAM_CHAT)))
LISTEN_PORT = int(os.environ.get('LISTEN_PORT', str(LISTEN_PORT)))
FALLBACK_URL = os.environ.get('FALLBACK_URL', FALLBACK_URL)
server = http.server.HTTPServer(('127.0.0.1', LISTEN_PORT), WebhookHandler)
log.info('alert-notifier listening on 127.0.0.1:%d', LISTEN_PORT)
server.serve_forever()