import json import os import re from datetime import datetime, timedelta, timezone from uuid import uuid4 import aiohttp import boto3 import tuotuo from sinks import FullActivityClient, jira_alerts from tuotuo.activities import Activity, ActivityType, Sevirity from tuotuo.logs import LogEntity from tuotuo.metrics import QueryParam, QueryRange ACTIVITY_BUCKET = os.environ["ACTIVITY_BUCKET"] PATH = "tikv_oom_detector" CE_ENDPOINT = os.environ["CE_ENDPOINT"] ACTIVITY_HUB_ENDPOINT = os.environ["ACTIVITY_HUB_ENDPOINT"] SOURCE_KEY = os.environ["SOURCE_KEY"] SOURCE = os.environ["SOURCE"] JIRA_ENDPOINT = os.environ.get("JIRA_ENDPOINT") JIRA_EMAIL = os.environ.get("JIRA_EMAIL") JIRA_API_TOKEN = os.environ.get("JIRA_API_TOKEN") async def write_activity_to_hub(activity: Activity): async with FullActivityClient( source=SOURCE, source_key=SOURCE_KEY, endpoint=ACTIVITY_HUB_ENDPOINT ) as client: return await client.write_activity(activity) def alert_to_jira(activity: Activity, clinic_org_id: str): if not all((JIRA_ENDPOINT, JIRA_EMAIL, JIRA_API_TOKEN)): print("[Warning][Jira alert skipped] Jira Cloud credentials are not fully configured") return alerts = [] occurrence_time = activity.calibration_time content = f'StartsAt: {datetime.fromtimestamp(occurrence_time, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S")} UTC\n' content += "Labels:\n" content += f"- cluster_id = {activity.target_id}\n" content += f"- instance = {activity.payload['instance']}\n" alert = { "project": {"key": "O11Y"}, "summary": "[PROD] TiKV OOM", "description": content, "issuetype": {"name": "Task"}, "priority": {"name": "Major"}, "components": [{"name": "tikv"}], # "customfield_10615": activity.application_id, # tenant_id # "customfield_10616": activity.target_id, # cluster_id "customfield_10157": activity.application_id, # jira cloud tenant_id "customfield_10149": activity.target_id, # jira cloud cluster_id } alerts.append(alert) try: jira_res = jira_alerts( jira_url=JIRA_ENDPOINT, email=JIRA_EMAIL, token=JIRA_API_TOKEN, alerts=alerts ) except Exception as error: print(f"[Warning][Jira alert failed] {type(error).__name__}") return print(jira_res) def dict_to_s3(dict, bucket, key): s3 = boto3.resource("s3") res = s3.Bucket(bucket).put_object(Key=key, Body=json.dumps(dict)) return res def generate_activity(cluster_id, instance, entity, tenant_id, project_id): activity_id = str(uuid4()) occurrence_time = int(entity.time.timestamp()) detect_time = int(datetime.now().timestamp()) summary = f"{instance} of cluster {cluster_id}: an OOM occurred." details = { "cluster_id": cluster_id, "instance": instance, "tenant_id": tenant_id, "project_id": project_id, "occurrence_time": occurrence_time, "detect_time": detect_time, "evidence": { "source": "Resource Overview Grafana Dashboard", "content": [entity.content], }, } res = dict_to_s3(details, ACTIVITY_BUCKET, f"{PATH}/{activity_id}.json") print(res) payload = { "summary": summary, "cluster_id": cluster_id, "instance": instance, "occurrence_time": occurrence_time, "detect_time": detect_time, "details": { "type": "s3", "bucket": ACTIVITY_BUCKET, "key": f"{PATH}/{activity_id}.json", }, } oom_event = Activity( activity_type=ActivityType.EVENT, activity_name="TiKV OOM", target_id=cluster_id, calibration_time=occurrence_time, source=SOURCE, source_key=SOURCE_KEY, catalog="OOM", severity=Sevirity.WARNING, activity_id=activity_id, application_id=tenant_id, concept_id=project_id, report_time=detect_time, author="tikv-oom-detector", context_id=activity_id, # context=[], payload=payload, occurrence_time=occurrence_time, detect_time=detect_time, ) return oom_event async def find_oom_evicted( cluster_id: str, start: datetime, end: datetime, tenant_id: str = None, project_id: str = None, region_name: str = None, clinic_org_id: str = None, provider: str = None, ): oom_events = [] kube_pod_info_promql = QueryParam( time_arg=float(end.timestamp()), query_template='kube_pod_info{namespace=~".*$tidb_cluster", pod=~"db-tikv-([0-9]{1,4})"}', cluster_id=cluster_id, region_name=region_name, clinic_org_id=clinic_org_id, provider=provider, query_name="kube_pod_info", query_labels=["pod"], ) metrics_client = tuotuo.client("metrics") async with metrics_client: pod_info_res = await metrics_client.query(kube_pod_info_promql) if pod_info_res.exception is None and len(pod_info_res) > 0: pod_info = { "cluster": pod_info_res[0]["metric"]["cluster"], "namespace": pod_info_res[0]["metric"]["namespace"], } query_range = QueryRange(start, end, "15s") last_terminated_reason_promql = QueryParam( time_arg=query_range, cluster_id=cluster_id, query_template='sum(kube_pod_container_status_last_terminated_reason{cluster="$cluster", namespace="$namespace", reason="OOMKilled", container="tikv"}) by (pod)', cluster=pod_info["cluster"], namespace=pod_info["namespace"], query_name="terminated_reason", query_labels=["pod"], ) last_terminated_reason_res = await metrics_client.query(last_terminated_reason_promql) if last_terminated_reason_res.exception is None: last_terminated_reason = last_terminated_reason_res.frame else: last_terminated_reason = None restart_count_promql = QueryParam( time_arg=query_range, cluster_id=cluster_id, query_template='increase(kube_pod_container_status_restarts_total{cluster="$cluster", namespace="$namespace", container="tikv"}[15s])', cluster=pod_info["cluster"], namespace=pod_info["namespace"], query_name="restart_count", query_labels=["pod"], ) restart_count_res = await metrics_client.query(restart_count_promql) if restart_count_res.exception is None: restart_count = restart_count_res.frame else: restart_count = None if last_terminated_reason is not None and restart_count is not None: oom_reboot = last_terminated_reason.mul(restart_count) for col_name, col in oom_reboot.items(): for index, item in col.items(): if ( item > 0.5 ): # if the pod is oom killed and rebooted, the value should be 1.0 fake_entity = LogEntity( index, cluster_id, "tikv", col_name, "fake", 'The conclusion comes from 2 metrics which are similar to the first 2 panels in Resource Overview Dashboard:\nsum(kube_pod_container_status_last_terminated_reason{cluster="$cluster", namespace="$namespace", reason="OOMKilled", container="tikv"}) by (pod)\nincrease(kube_pod_container_status_restarts_total{cluster="$cluster", namespace="$namespace", container="tikv"}[15s])', ) oom_event = generate_activity( cluster_id, col_name, fake_entity, tenant_id, project_id ) oom_events.append(oom_event) if len(oom_events) > 0: for activity in oom_events: res = await write_activity_to_hub(activity) alert_to_jira(activity, clinic_org_id) if res.get("status") == "ok": print(f"[Info][Write event Successful] {cluster_id} OOM activity: {activity}") else: print(f"[Error][Write event Failed] {cluster_id} OOM activity: {activity}") return oom_events if __name__ == "__main__": import asyncio import time """ input: cluster_id, start, end, tenant_id, project_id output: oom_events excute: write details to s3 and write oom_events to activityhub non-goal: alert, invoke consumer non-goals are implemented in agent """ cluster_id = "1379661944636804070" tenant_id = "1372813089208281359" project_id = "1372813089206481407" start = datetime.fromtimestamp(1675756800) # 2023-02-07 16:00:00 end = datetime.fromtimestamp(1675760400) # 2023-02-07 17:00:00 proc_start = time.perf_counter() oom_events = asyncio.run(find_oom_evicted(cluster_id, start, end, tenant_id, project_id)) proc_end = time.perf_counter() print(f"Elapsed time: {proc_end - proc_start} seconds") for oom_event in oom_events: print( datetime.fromtimestamp(oom_event.calibration_time) .replace(tzinfo=timezone.utc) .astimezone(timezone(timedelta(hours=8))), ": ", oom_event.payload.get("summary"), )