"""Lambda Functions for reporter_service_level_probe_cluster_total""" import json import os import time from datetime import datetime, timedelta, timezone from typing import Any import boto3 import duckdb import pymysql import requests import sliutil.alert import sliutil.connection import sliutil.s3 import sliutil.secret import sliutil.signature SLI_BYOC_TABLE_NAME = "sli_byoc_cluster_downtime" # used for saving last timestamp in table `last_timestamp` SLI_TABLE_NAME = "sli_dedicated_cluster_downtime" SLI_METRIC_NAME = "reporter_service_level_probe_cluster_total" SLI_STAGE = os.environ["SLI_STAGE"] SLI_ASSUME_ROLE = os.environ["SLI_ASSUME_ROLE"] DEV_OPS_API_KEY = sliutil.secret.get_secret("DEV_OPS_API_KEY") DEV_OPS_URL = os.environ["DEV_OPS_URL"] CLINIC_ENDPOINT = os.environ["CLINIC_ENDPOINT"] CLINIC_API_KEY = sliutil.secret.get_secret("CLINIC_API_KEY") JIRA_PROJECT = os.environ["JIRA_PROJECT"] JIRA_CUSTOM_FIELD_TENANT_ID = "customfield_10157" JIRA_CUSTOM_FIELD_CLUSTER_ID = "customfield_10149" ALERT_COMMENT = "signature_incorrect" ALERT_RECORDS_TABLE = "alert_records" ALERT_TYPE_READ_FILE_FAILED = "read_file_failed" 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"] 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 get_iam_info(cluster_id: str) -> dict[str, str] | None: url = f"{DEV_OPS_URL}/api/v1/devops/clusters/{cluster_id}/iam-info" headers = { "X-API-KEY": DEV_OPS_API_KEY, "Accept": "application/json", "Content-Type": "application/json", } try: response = requests.get(url, headers=headers, timeout=10) response.raise_for_status() response_data = response.json() return response_data 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}") except requests.exceptions.RequestException as e: print(f"request error: {e}") except json.JSONDecodeError: print("response is not in valid JSON format") print(f"response text: {response.text}") except ValueError as e: print(f"response structure error: {e}") return None def list_alive_byoc_clusters(clusters): """List all alive byoc clusters from clinic""" page = 1 per_page = 100 while True: resp = requests.get( ( f"{CLINIC_ENDPOINT}/clinic/api/v1/dashboard/clusters2?" f"deploy_type=byoc&show_deleted=false" f"&page={page}&limit={per_page}" ), timeout=30, headers={"Authorization": f"Bearer {CLINIC_API_KEY}"}, ) if resp.status_code != 200: print( f"Failed to list alive byoc clusters, status_code: {resp.status_code}" ) return data = resp.json() print(f"list_alive_byoc_clusters, page: {page}") for item in data["items"]: clusters[item["clusterID"]] = item if data["has_more"] is False: break page += 1 return def integrity_verification(row: Any) -> str: timestamp = row[0] instance = row[1] metrics_json_str = row[2] signature = row[3] if not signature: return True try: metrics_dict = json.loads(metrics_json_str) except json.JSONDecodeError as e: raise ValueError(f"failed to load metrics: {e}") from e target_dict = { "timestamp": timestamp, "instance": instance, "metrics": metrics_dict, } target_json = json.dumps(target_dict, indent=2, ensure_ascii=False) generated_sig = sliutil.signature.generate_signature_hash_for_sli_data(target_json) return generated_sig def get_assume_role_session(role_arn, session_name): sts_client = boto3.client("sts") response = sts_client.assume_role( RoleArn=role_arn, RoleSessionName=session_name, DurationSeconds=3600 ) credentials = response["Credentials"] return credentials def already_send_read_file_failed_alert_in_1h( connection: pymysql.Connection, cluster_id: str, timestamp: float ) -> bool: ONE_HOUR_SECONDS = 3600 with connection.cursor() as cursor: try: sql = f""" SELECT MAX(timestamp) AS latest_alert_ts FROM `{ALERT_RECORDS_TABLE}` WHERE cluster_id = %s AND alert_type = %s """ cursor.execute(sql, (cluster_id, ALERT_TYPE_READ_FILE_FAILED)) result = cursor.fetchone() latest_alert_ts = result["latest_alert_ts"] if result else None if latest_alert_ts is None: return False if not isinstance(latest_alert_ts, (int, float)): raise TypeError(f"Invalid timestamp type: {type(latest_alert_ts)}") time_diff = timestamp - latest_alert_ts return time_diff < ONE_HOUR_SECONDS except Exception as e: print(f"Error checking alert history: {e!s}") return False def save_alert_records( connection: pymysql.Connection, cluster_id: str, timestamp: float, alert_type: str ) -> None: with connection.cursor() as cursor: sql = f""" REPLACE INTO {ALERT_RECORDS_TABLE} ( timestamp, cluster_id, alert_type ) VALUES (%s, %s, %s) """ cursor.execute(sql, (int(timestamp), cluster_id, alert_type)) connection.commit() def already_send_signature_incorrect_alert( connection: pymysql.Connection, cluster_id: str, timestamp: float ) -> bool: with connection.cursor() as cursor: sql = f""" SELECT * FROM `{SLI_TABLE_NAME}` WHERE timestamp={timestamp:.0f} AND cluster_id='{cluster_id}' AND dead_count=0 AND fake=1 AND comment='{ALERT_COMMENT}' """ cursor.execute(sql) row = cursor.fetchone() return row is not None def generate_read_file_failed_alert_content( cluster_id: str, tenant_id: str, begin_ts: float, end_ts: float ) -> dict: """Generate content of alert""" return { "project": {"key": JIRA_PROJECT}, "summary": f"[{SLI_STAGE.upper()}] BYOCSliReadFileFailed", "description": f""" Detail: - cluster_id: {cluster_id} - tenant_id: {tenant_id} - begin_ts: {begin_ts} - end_ts: {end_ts} """, "issuetype": {"name": "Task"}, "priority": {"name": "Critical"}, "components": [{"name": "resilience"}], JIRA_CUSTOM_FIELD_TENANT_ID: tenant_id, JIRA_CUSTOM_FIELD_CLUSTER_ID: cluster_id, } def generate_signature_incorrect_alert_content( cluster_id: str, tenant_id: str, incorrect_minutes: [] ) -> dict: """Generate content of alert""" return { "project": {"key": JIRA_PROJECT}, "summary": f"[{SLI_STAGE.upper()}] BYOCSliSignatureIncorrect", "description": f""" Detail: - cluster_id: {cluster_id} - tenant_id: {tenant_id} - incorrect_minutes: {", ".join(map(str, incorrect_minutes))} """, "issuetype": {"name": "Task"}, "priority": {"name": "Critical"}, "components": [{"name": "resilience"}], JIRA_CUSTOM_FIELD_TENANT_ID: tenant_id, JIRA_CUSTOM_FIELD_CLUSTER_ID: cluster_id, } def update_duckdb_credentials( duckdb_conn: duckdb.DuckDBPyConnection, aws_access_key_id: str, aws_secret_access_key: str, aws_session_token: str, region: str, ) -> duckdb.DuckDBPyConnection: """Update duckdb s3 credentials""" duckdb_conn.execute( f""" SET s3_access_key_id='{aws_access_key_id}'; SET s3_secret_access_key='{aws_secret_access_key}'; SET s3_session_token='{aws_session_token}'; SET s3_region='{region}'; """ ) return duckdb_conn 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 """ """ 1. list byoc clusters 2. for each byoc cluster: 2.1 get IAM info 2.2 assume role and get credential 2.3 init duckdb_conn and s3 client by using the credential 2.4 init Handler """ tidb_conn = sliutil.connection.get_tidb_connection() duckdb_conn = sliutil.connection.get_duckdb_connection() byoc_alive_clusters = {} list_alive_byoc_clusters(byoc_alive_clusters) for cluster in byoc_alive_clusters.values(): if cluster["clusterDeployType"] != "byoc": continue if cluster["clusterStatus"] != "available": continue print(f"cluster: {cluster}") cluster_id = cluster["clusterID"] tenant_id = cluster["tenantID"] iam_info = get_iam_info(cluster_id=cluster_id) aws_account = iam_info["tidb_cloud_account_id"] region = iam_info["region"] bucket_name = iam_info["bucket_name"] try: role_arn = f"arn:aws:iam::{aws_account}:role/{SLI_ASSUME_ROLE}" credentials = sliutil.s3.get_assume_role_session( role_arn=role_arn, session_name="sli-analyzer-session" ) except Exception as e: print(f"assume role failed, cluster_id: {cluster_id}, error: {e}") continue update_duckdb_credentials( duckdb_conn=duckdb_conn, aws_access_key_id=credentials["AccessKeyId"], aws_secret_access_key=credentials["SecretAccessKey"], aws_session_token=credentials["SessionToken"], region=region, ) last_ts = float( find_last_timestamp(connection=tidb_conn, table=SLI_BYOC_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, credentials=credentials, bucket_name=bucket_name, cluster_id=cluster_id, tenant_id=tenant_id, ) try: print( f"start to analysis byoc cluster downtime, " f"begin_ts: {begin_ts}, end_ts: {end_ts}, cluster_id: {cluster_id}" ) handler.run(begin_ts=begin_ts, end_ts=end_ts) print(f"finish analysis byoc cluster downtime, cluster_id: {cluster_id}") except Exception as e: print( f"failed to analysis byoc cluster downtime, cluster_id: {cluster_id}, error: {e}" ) result = { "statusCode": 200, "body": json.dumps( { "message": "success", } ), } 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, credentials: Any, bucket_name: str, cluster_id: str, tenant_id: str, ) -> None: self.data = [] self.tidb_conn = tidb_conn self.duckdb_conn = duckdb_conn self.credentials = credentials self.s3_client = self._init_s3_client() self.bucket_name = bucket_name self.cluster_id = cluster_id self.tenant_id = tenant_id def _init_s3_client(self): if self.credentials: return boto3.client( "s3", aws_access_key_id=self.credentials["AccessKeyId"], aws_secret_access_key=self.credentials["SecretAccessKey"], aws_session_token=self.credentials["SessionToken"], ) else: return 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 """ rows = [] try: rows = self.search(begin_ts=begin_ts, end_ts=end_ts) except Exception as e: print(f"search failed: {e}") send_read_file_failed_alert = already_send_read_file_failed_alert_in_1h( connection=self.tidb_conn, cluster_id=self.cluster_id, timestamp=time.time(), ) print(f"already send alert in past 1 hour? {send_read_file_failed_alert}") if not send_read_file_failed_alert: alert = generate_read_file_failed_alert_content( cluster_id=self.cluster_id, tenant_id=self.tenant_id, begin_ts=begin_ts, end_ts=end_ts, ) sliutil.alert.create_alerts(alerts=[alert]) save_alert_records( connection=self.tidb_conn, cluster_id=self.cluster_id, timestamp=time.time(), alert_type=ALERT_TYPE_READ_FILE_FAILED, ) downtime_minute = 0 verify_success = 0 parser = None incorrect_minutes = {} # the row is ordered by timestamp for row in rows: metric_timestamp = row[0] metric_group = json.loads(row[2]) metrics = metric_group["metrics"] # data integrity verification signature = integrity_verification(row) correct = signature == row[3] metric_time = ( datetime.fromtimestamp(metric_timestamp, tz=timezone.utc) .replace(second=0) .timestamp() ) # data of this minute is invalid if metric_time in incorrect_minutes: continue if not correct: incorrect_minutes[metric_time] = True continue else: verify_success += 1 # renew DowntimeParser every minute if downtime_minute == 0: downtime_minute = metric_time parser = DowntimeParser(downtime=downtime_minute) elif ( metric_time != downtime_minute ): # if metric_time changes, we need save the data of the last minute # but if downtime_minute is in the incorrect_minutes, we don't need to save it if downtime_minute not in incorrect_minutes: 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(f"No data found, cluster_id: {self.cluster_id}") return # if downtime_minute is in the incorrect_minutes dict, data of this minute is invalid if downtime_minute not in incorrect_minutes: for item in parser.get_dead_clusters(): # (timestamp, cluster_id, dead_count) self.data.append((downtime_minute, item[0], item[1])) print( f"verify_success_count: {verify_success}, incorrect_minutes: {incorrect_minutes}, cluster_id: {self.cluster_id}" ) self.save() update_last_timestamp( connection=self.tidb_conn, table=SLI_BYOC_TABLE_NAME, timestamp=end_ts ) no_alert_minutes = [] for m in incorrect_minutes: send_alert = already_send_signature_incorrect_alert( connection=self.tidb_conn, cluster_id=self.cluster_id, timestamp=m ) if not send_alert: no_alert_minutes.append(m) if len(no_alert_minutes) > 0: alert = generate_signature_incorrect_alert_content( cluster_id=self.cluster_id, tenant_id=self.tenant_id, incorrect_minutes=incorrect_minutes, ) sliutil.alert.create_alerts(alerts=[alert]) self.save_send_alert_records(no_alert_minutes) 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=self.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, signature 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() def save_send_alert_records(self, minutes: []): print(f"save signature incorrect alert records, len: {len(minutes)}") with self.tidb_conn.cursor() as cursor: sql = f""" REPLACE INTO `{SLI_TABLE_NAME}` ( timestamp, cluster_id, dead_count, fake, comment ) VALUES (%s, %s, 0, 1, %s) """ params = [(minute, self.cluster_id, ALERT_COMMENT) for minute in minutes] cursor.executemany(sql, params) 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