import json import os from datetime import datetime, timezone import pymysql import requests import sliutil.connection import sliutil.secret SERVERLESS_DB_HOST = os.getenv("SERVERLESS_DB_HOST") SERVERLESS_DB_PORT = 4000 SERVERLESS_DB_USER = os.getenv("SERVERLESS_DB_USER") SERVERLESS_DB_PASSWD = sliutil.secret.get_secret("SERVERLESS_DB_PASSWD") SERVERLESS_DB_NAME = "serverless_prod" SERVERLESS_TABLE_NAME = "downtimes" SERVERLESS_REGION_STATUS_TABLE_NAME = "serverless_region_status" SYNC_STATUSPAGE_ENDPOINT = os.getenv("SYNC_STATUSPAGE_ENDPOINT") DEV_OPS_API_KEY = sliutil.secret.get_secret("DEV_OPS_API_KEY") DEV_OPS_URL = os.environ["DEV_OPS_URL"] SERVICE_PLAN_STARTER = "Starter" SERVICE_PLAN_ESSENTIAL = "Essential" def lambda_handler(event, context): # pylint:disable=unused-argument """AWS Lambda Function Handler Parameters ---------- event: dict, required API Gateway Lambda Proxy Input Format Event doc: https://docs.aws.amazon.com/apigateway/latest/developerguide/set-up-lambda-proxy-integrations.html#api-gateway-simple-proxy-for-lambda-input-format # pylint:disable=line-too-long # noqa context: object, required Lambda Context runtime methods and attributes Context doc: https://docs.aws.amazon.com/lambda/latest/dg/python-context-object.html Returns ------ API Gateway Lambda Proxy Output Format: dict Return doc: https://docs.aws.amazon.com/apigateway/latest/developerguide/set-up-lambda-proxy-integrations.html # pylint:disable=line-too-long """ now_ts = datetime.now(timezone.utc).timestamp() end_ts = event.get("end_ts", now_ts) end_ts = min(end_ts, now_ts) end_ts = int(end_ts) begin_ts = end_ts - 5 * 60 dry_run = event.get("dry_run", False) serverless_conn = sliutil.connection.get_tidb_connection( host=SERVERLESS_DB_HOST, port=SERVERLESS_DB_PORT, user=SERVERLESS_DB_USER, password=SERVERLESS_DB_PASSWD, database=SERVERLESS_DB_NAME, ) sla_conn = sliutil.connection.get_tidb_connection() handler = Handler( serverless_conn=serverless_conn, sla_conn=sla_conn, dry_run=dry_run ) starter_events = handler.run( service_plan=SERVICE_PLAN_STARTER, begin_ts=begin_ts, end_ts=end_ts ) essential_events = handler.run( service_plan=SERVICE_PLAN_ESSENTIAL, begin_ts=begin_ts, end_ts=end_ts ) result = { "statusCode": 200, "body": json.dumps( { "start_ts": begin_ts, "end_ts": end_ts, "starter_events": starter_events, "essential_events": essential_events, } ), } print(f"result: {result}") return result def get_essential_cluster_ids() -> list[str]: page = 1 per_page = 200 headers = { "X-API-KEY": DEV_OPS_API_KEY, "Accept": "application/json", "Content-Type": "application/json", } result = [] while True: url = f"{DEV_OPS_URL}/api/v1/devops/clusters?deploy_type=DEPLOY_TYPE_SERVERLESS&service_plan=Essential&page={page}&per_page={per_page}" try: response = requests.get(url, headers=headers, timeout=20) response.raise_for_status() response_data = response.json() result.extend(x["id"] for x in response_data["items"]) if response_data["total"] <= page * per_page: print(f"list essential cluster ids, total count: {len(result)}") return result except requests.exceptions.HTTPError as e: print(f"HTTP request error: {e}") print(f"status code: {response.status_code}") print(f"response text: {response.text}") return result except requests.exceptions.RequestException as e: print(f"request error: {e}") return result except json.JSONDecodeError: print("response is not in valid JSON format") print(f"response text: {response.text}") return result except ValueError as e: print(f"response structure error: {e}") return result class Handler: """Handle sync serverless status to statuspage""" def __init__( self, serverless_conn: pymysql.Connection, sla_conn: pymysql.Connection, dry_run: bool = False, ) -> None: self.data = [] self.sla_conn = sla_conn self.serverless_conn = serverless_conn self.dry_run = dry_run def run(self, service_plan: str, begin_ts: int, end_ts: int) -> None: down_regions = self.get_down_regions(service_plan, begin_ts, end_ts) region_status = self.get_current_region_status(service_plan) event_time = datetime.fromtimestamp(end_ts, tz=timezone.utc) events = [] # Handle DOWN events: currently UP but had downtime for region, downtime_cnt in down_regions.items(): if downtime_cnt < 3: continue current = region_status.get(region, "UP") if current == "UP": events.append( { "region": region, "status": "DOWN", "timestamp": event_time.isoformat(), } ) region_status[region] = "DOWN" # Handle UP events: currently DOWN but no downtime in window for region, current in region_status.items(): if current == "DOWN" and region not in down_regions: events.append( { "region": region, "status": "UP", "timestamp": event_time.isoformat(), } ) region_status[region] = "UP" if not self.dry_run: self.update_statuspage(service_plan, events) self.update_region_status(service_plan, region_status, event_time) return events def get_down_regions( self, service_plan: str, begin_ts: int, end_ts: int, ) -> dict[str, int]: """Get down regions from serverless db""" essential_cluster_placeholder = "" cluster_cnt_threshold = 5 if service_plan == SERVICE_PLAN_ESSENTIAL: cluster_cnt_threshold = 2 essential_cluster_ids = get_essential_cluster_ids() if not essential_cluster_ids or len(essential_cluster_ids) == 0: return {} quoted = ", ".join(f"'{cid}'" for cid in essential_cluster_ids) essential_cluster_placeholder = f"cluster_id IN ({quoted}) AND " sql = f""" SELECT region, COUNT(1) AS downtime_cnt FROM ( SELECT region, down_time, COUNT(1) AS cluster_cnt FROM {SERVERLESS_TABLE_NAME} WHERE {essential_cluster_placeholder}down_time >= FROM_UNIXTIME({begin_ts}) AND down_time <= FROM_UNIXTIME({end_ts}) AND down_rate >= 1 GROUP BY region, down_time HAVING cluster_cnt >= {cluster_cnt_threshold} ORDER BY region, down_time ) AS t GROUP BY region ORDER BY region; """ # print(f"get_down_regions sql: {sql}") with self.serverless_conn.cursor() as cur: cur.execute(sql) down_regions = { row["region"]: row["downtime_cnt"] for row in cur.fetchall() } print(f"get_down_regions: {down_regions}, service_plan: {service_plan}") return down_regions def get_current_region_status(self, service_plan: str) -> dict[str, str]: sql = f""" SELECT region, status FROM {SERVERLESS_REGION_STATUS_TABLE_NAME} WHERE service_plan = '{service_plan}' """ with self.sla_conn.cursor() as cur: cur.execute(sql) region_status = {row["region"]: row["status"] for row in cur.fetchall()} print( f"get_current_region_status: {region_status}, service_plan: {service_plan}" ) return region_status def update_region_status( self, service_plan: str, region_status: dict, event_time: datetime ) -> None: sql = f""" INSERT INTO {SERVERLESS_REGION_STATUS_TABLE_NAME} (service_plan, region, status, last_updated) VALUES (%s, %s, %s, %s) ON DUPLICATE KEY UPDATE status = VALUES(status), last_updated = VALUES(last_updated) """ print(f"update_region_status: {region_status}, service_plan: {service_plan}") with self.sla_conn.cursor() as cur: for region, status in region_status.items(): cur.execute(sql, (service_plan, region, status, event_time)) self.sla_conn.commit() def update_statuspage(self, service_plan: str, events) -> None: for event in events: try: resp = requests.post( url=SYNC_STATUSPAGE_ENDPOINT, json={ "website_name": f"TiDB Cloud {service_plan}({event['region']})", "alert_type": event["status"], }, timeout=5, ) resp.raise_for_status() print(f"update statuspage, event: {event}, resp: {resp.content}") except requests.exceptions.HTTPError as e: print(f"update statuspage failed, event: {event}, error: {e}") if __name__ == "__main__": # Test the lambda function locally ev = { "end_ts": 1760950800, "dry_run": True, } lambda_handler(ev, None)