"""Authenticated Cloud Monitoring Pub/Sub bridge to a signed Feishu bot.""" from __future__ import annotations import base64 import hashlib import hmac import json import os import re import time from datetime import UTC, datetime from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import urlencode import httpx from google.api_core.exceptions import PreconditionFailed from google.cloud import storage MAX_BODY = 262144 POLICY = "pingcap-datapipelines job failure" class DeliveryError(RuntimeError): """A safe error category; never includes the webhook or response body.""" def signature(timestamp: str, secret: str) -> str: key = f"{timestamp}\n{secret}".encode() return base64.b64encode(hmac.new(key, b"", hashlib.sha256).digest()).decode() def send_message(text: str, webhook: str, secret: str) -> None: if not re.fullmatch(r"https://open\.feishu\.cn/open-apis/bot/v2/hook/[A-Za-z0-9-]+", webhook) or not secret: raise DeliveryError("invalid_bot_configuration") timestamp = str(int(time.time())) payload = { "timestamp": timestamp, "sign": signature(timestamp, secret), "msg_type": "text", "content": {"text": text}, } try: response = httpx.post(webhook, json=payload, timeout=10, follow_redirects=False) if response.status_code != 200: raise DeliveryError(f"feishu_http_{response.status_code}") result = response.json() if not isinstance(result, dict): raise DeliveryError("invalid_feishu_response") code = result.get("code", result.get("StatusCode")) if type(code) is not int or code != 0: raise DeliveryError(f"feishu_code_{code}" if type(code) is int else "invalid_feishu_response") except DeliveryError: raise except Exception: raise DeliveryError("feishu_request_failed") from None def decode_event(envelope: dict, project: str) -> dict: message = envelope["message"] payload = json.loads(base64.b64decode(message["data"], validate=True)) incident = payload["incident"] labels = incident.get("resource", {}).get("labels", {}) if incident.get("policy_name") != POLICY: raise ValueError("unexpected_policy") if incident.get("resource", {}).get("type") != "cloud_run_job": raise ValueError("unexpected_resource") if labels.get("project_id", incident.get("scoping_project_id")) != project: raise ValueError("unexpected_project") if not re.fullmatch(r"pingcap-[a-z0-9-]+", labels.get("job_name", "")): raise ValueError("unexpected_job") if incident.get("state") not in {"open", "closed"} or not incident.get("incident_id"): raise ValueError("invalid_incident") return incident def message_text(incident: dict, project: str) -> str: labels = incident["resource"]["labels"] job = labels["job_name"] region = labels.get("location", "us-west2") if not re.fullmatch(r"[a-z0-9-]+", region): raise ValueError("invalid_region") started = datetime.fromtimestamp(int(incident["started_at"]), UTC).isoformat() test = "【测试告警】" if job.startswith("pingcap-alert-smoke-") else "【Pipeline 任务失败】" query = f'resource.type="cloud_run_job" AND resource.labels.job_name="{job}"' logs = "https://console.cloud.google.com/logs/query?" + urlencode({"project": project, "query": query}) job_url = f"https://console.cloud.google.com/run/jobs/details/{region}/{job}/executions?project={project}" # Never forward free-form incident summaries, documentation, logs or request payloads. return ( f"PingCAP-Data-Pipelines {test}\n任务：{job}\n项目：{project}\n" f"告警开始时间（UTC）：{started}\n状态：Cloud Run Job 执行失败\n" f"执行记录：{job_url}\n排查日志：{logs}\n" "请确认失败原因及 extract → manifest → load → final 的恢复结果。" ) def relay(envelope: dict, *, project: str, bucket, webhook: str, secret: str, sender=send_message) -> dict: incident = decode_event(envelope, project) if incident["state"] != "open": # A closed metric window is not evidence of business/data recovery. return {"event": "alert_ignored", "reason": "closed_incident"} incident_id = str(incident["incident_id"]) key = hashlib.sha256(incident_id.encode()).hexdigest() receipt = bucket.blob(f"sent/{key}.json") if receipt.exists(timeout=5, retry=None): return {"event": "alert_duplicate", "receipt_id": key} text = message_text(incident, project) sender(text, webhook, secret) safe_receipt = { "event": "feishu_alert_sent", "receipt_id": key, "job": incident["resource"]["labels"]["job_name"], "feishu_code": 0, "sent_at": datetime.now(UTC).isoformat(), } try: receipt.upload_from_string( json.dumps(safe_receipt), content_type="application/json", if_generation_match=0, timeout=5, retry=None ) except PreconditionFailed: pass return safe_receipt def main() -> None: project = os.environ["ALERT_PROJECT"] bucket = storage.Client(project=project).bucket(os.environ["ALERT_RECEIPT_BUCKET"]) webhook = os.environ["FEISHU_FAILURE_WEBHOOK_URL"] secret = os.environ["FEISHU_FAILURE_SIGNING_SECRET"] if not webhook or not secret: raise RuntimeError("missing_bot_configuration") class Handler(BaseHTTPRequestHandler): def log_message(self, *_args) -> None: pass def do_GET(self) -> None: self.send_response(200 if self.path == "/ready" else 404) self.end_headers() def do_POST(self) -> None: status = 500 try: if self.path != "/pubsub": raise ValueError("invalid_path") size = int(self.headers.get("Content-Length", "0")) if not 0 < size <= MAX_BODY: raise ValueError("invalid_body_length") result = relay( json.loads(self.rfile.read(size)), project=project, bucket=bucket, webhook=webhook, secret=secret ) print(json.dumps(result), flush=True) status = 204 except (ValueError, KeyError, TypeError): print(json.dumps({"event": "alert_invalid"}), flush=True) status = 400 except Exception as error: print( json.dumps( { "event": "alert_delivery_failed", "error_type": type(error).__name__, "reason": str(error) if isinstance(error, DeliveryError) else "delivery_failed", } ), flush=True, ) self.send_response(status) self.end_headers() HTTPServer(("0.0.0.0", int(os.environ.get("PORT", "8080"))), Handler).serve_forever() if __name__ == "__main__": main()