"""Lambda Functions for reporter_service_level_probe_cluster_total""" import json import os from datetime import datetime, timedelta, timezone from typing import Any import boto3 import duckdb import pymysql import sliutil.connection import sliutil.s3 SLI_TABLE_NAME = "sli_dedicated_cluster_downtime" SLI_METRIC_NAME = "reporter_service_level_probe_cluster_total" SLI_S3_BUCKET_NAME = os.environ["SLI_S3_BUCKET_NAME"] def find_last_timestamp(connection: pymysql.Connection, table: str) -> float: """Find the last timestamp 1. Try to find the last timestamp from `last_timestamp` table 2. Try to find the last timestamp in the last 3 days 3. If not found, return the timestamp of 30 minutes ago """ with connection.cursor() as cursor: sql = f""" SELECT timestamp FROM `last_timestamp` WHERE table_name="{table}" """ cursor.execute(sql) row = cursor.fetchone() if row is not None: return row["timestamp"] start_time = datetime.now(timezone.utc).replace( day=1, hour=0, minute=0, second=0 ) for _ in range(3): start_ts = start_time.timestamp() sql = f""" SELECT timestamp FROM `{table}` WHERE timestamp>{start_ts} ORDER BY timestamp DESC LIMIT 1 """ cursor.execute(sql) row = cursor.fetchone() if row is not None: return row["timestamp"] start_time = start_time - timedelta(days=1) return (datetime.now(timezone.utc) - timedelta(minutes=30)).timestamp() def update_last_timestamp( connection: pymysql.Connection, table: str, timestamp: float ) -> None: """Update the last timestamp""" with connection.cursor() as cursor: sql = f""" REPLACE INTO `last_timestamp` ( table_name, timestamp ) VALUES ('{table}', {timestamp}) """ cursor.execute(sql) connection.commit() 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 """ tidb_conn = sliutil.connection.get_tidb_connection() duckdb_conn = sliutil.connection.get_duckdb_connection() last_ts = float(find_last_timestamp(connection=tidb_conn, table=SLI_TABLE_NAME)) begin_ts = event.get("begin_ts", last_ts - 5 * 60) end_ts = event.get("end_ts", last_ts + 1 * 60 * 60) end_ts = min(end_ts, datetime.now(timezone.utc).timestamp()) handler = Handler(tidb_conn=tidb_conn, duckdb_conn=duckdb_conn) handler.run(begin_ts=begin_ts, end_ts=end_ts) result = { "statusCode": 200, "body": json.dumps( { "start_ts": begin_ts, "end_ts": end_ts, "downtime_count": len(handler.data), } ), } print(f"result: {result}") return result class Handler: """Handle metrics `reporter_service_level_probe_cluster_counter`""" def __init__( self, tidb_conn: pymysql.Connection, duckdb_conn: duckdb.DuckDBPyConnection ) -> None: self.data = [] self.tidb_conn = tidb_conn self.duckdb_conn = duckdb_conn self.s3_client = boto3.client("s3") def run(self, begin_ts: float, end_ts: float) -> None: """ 1. Search metrics `reporter_service_level_probe_cluster_total` 2. Parse the metrics 3. Save the data into table """ print( f"Start to reporter_service_level_probe_cluster_total, " f"begin_ts: {begin_ts}, end_ts: {end_ts}" ) rows = self.search(begin_ts=begin_ts, end_ts=end_ts) downtime_minute = 0 parser = None # the row is ordered by timestamp for row in rows: metric_timestamp = row[0] metric_group = json.loads(row[2]) metrics = metric_group["metrics"] metric_time = ( datetime.fromtimestamp(metric_timestamp, tz=timezone.utc) .replace(second=0) .timestamp() ) # renew DowntimeParser every minute if downtime_minute == 0: downtime_minute = metric_time parser = DowntimeParser(downtime=downtime_minute) elif metric_time != downtime_minute: for item in parser.get_dead_clusters(): # (timestamp, cluster_id, dead_count) self.data.append((downtime_minute, item[0], item[1])) downtime_minute = metric_time parser = DowntimeParser(downtime=downtime_minute) for metric in metrics: parser.parse(metric=metric) if parser is None: print("No data found") return for item in parser.get_dead_clusters(): # (timestamp, cluster_id, dead_count) self.data.append((downtime_minute, item[0], item[1])) self.save() update_last_timestamp( connection=self.tidb_conn, table=SLI_TABLE_NAME, timestamp=end_ts ) print("Finish reporter_service_level_probe_cluster_total") def search(self, begin_ts: float, end_ts: float) -> list: """List all metrics `reporter_service_level_probe_cluster_total` from begin_ts to end_ts""" file_names = sliutil.s3.ls_s3_objects( self.s3_client, begin_ts=begin_ts, end_ts=end_ts, bucket_name=SLI_S3_BUCKET_NAME, metric_name=SLI_METRIC_NAME, ) if len(file_names) == 0: print(f"search, no file found, begin_ts: {begin_ts}, end_ts: {end_ts}") return [] print(f"start to search from {len(file_names)} files") start = datetime.now(timezone.utc).timestamp() sql = f""" SELECT timestamp, instance, metrics FROM read_parquet({file_names}, hive_partitioning=false) order by timestamp """ # print(f"search, sql: {sql}") result = self.duckdb_conn.execute(sql).fetchall() end = datetime.now(timezone.utc).timestamp() print(f"search, take {end - start:.2f}s") return result def save(self) -> None: """Save data into table `reporter_service_level_probe_cluster_counter`""" print(f"save downtime, len: {len(self.data)}") with self.tidb_conn.cursor() as cursor: sql = f""" REPLACE INTO `{SLI_TABLE_NAME}` ( timestamp, cluster_id, dead_count ) VALUES (%s, %s, %s) """ cursor.executemany(sql, self.data) self.tidb_conn.commit() class DowntimeParser: """Downtime Parser""" def __init__(self, downtime: float) -> None: self.downtime = downtime self.cluster_probe_counter = {} self.active_clusters = {} self.cluster_dead_count = {} def get_dead_clusters(self) -> list: """return dead clusters and their dead count""" dead_clusters = [] for cluster_id in self.cluster_probe_counter: if ( cluster_id in self.cluster_dead_count and self.cluster_dead_count[cluster_id] > 0 ): dead_clusters.append((cluster_id, self.cluster_dead_count[cluster_id])) print( f"time:{self.downtime:.0f}, active clusters: {len(self.active_clusters)}, " f"dead clusters: {len(dead_clusters)}, {dead_clusters}" ) return dead_clusters def parse(self, metric: dict[str, Any]) -> None: """Parse the input metric""" cluster_id = metric["labels"]["cluster_id"] if cluster_id in self.active_clusters: # already active return status = metric["labels"]["status"] current_value = int(float(metric["value"])) if cluster_id not in self.cluster_probe_counter: self.cluster_probe_counter[cluster_id] = {} self.cluster_probe_counter[cluster_id][status] = current_value return if status not in self.cluster_probe_counter[cluster_id]: self.cluster_probe_counter[cluster_id][status] = current_value return last_value = self.cluster_probe_counter[cluster_id][status] if current_value > last_value: if status in ("alive", "skipped_by_not_ready", "skipped_by_suspend"): self.active_clusters[cluster_id] = 1 self.cluster_probe_counter.pop(cluster_id, None) self.cluster_dead_count.pop(cluster_id, None) return increment = current_value - last_value if cluster_id not in self.cluster_dead_count: self.cluster_dead_count[cluster_id] = increment else: self.cluster_dead_count[cluster_id] += increment self.cluster_probe_counter[cluster_id][status] = current_value