"""Lambda Functions for calculate SLA of TiDB Clusters""" import json import os from datetime import datetime, timezone import pymysql import requests import sliutil.alert import sliutil.connection import sliutil.secret import sliutil.time SLI_STAGE = os.environ["SLI_STAGE"] SLI_TABLE_NAME = "sli_dedicated_cluster_downtime" SLA_TABLE_NAME = "sla" ACTIVITY_HUB_ENDPOINT = os.environ["ACTIVITY_HUB_ENDPOINT"] ACTIVITY_HUB_SOURCE = os.environ["ACTIVITY_HUB_SOURCE"] ACTIVITY_HUB_KEY = sliutil.secret.get_secret("ACTIVITY_HUB_KEY") JIRA_PROJECT = os.environ["JIRA_PROJECT"] CLINIC_ENDPOINT = os.environ["CLINIC_ENDPOINT"] CLINIC_API_KEY = sliutil.secret.get_secret("CLINIC_API_KEY") CLUSTER_TYPE_DEDICATED = "dedicated" CLUSTER_TYPE_NEXT_GEN = "next-gen" INTERNAL_TENANT_IDS = [ "1372813089208521817", "1372813089187791289", "1372813089194211283", "1372813089209212300", "1372813089193131282", "1372813089200871284", ] JIRA_CUSTOM_FIELD_TENANT_ID = "customfield_10157" JIRA_CUSTOM_FIELD_CLUSTER_ID = "customfield_10149" 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() month_str = event.get("month", "") begin_ts, end_ts = sliutil.time.get_month_begin_end_ts(month=month_str) handler = Handler(connection=tidb_conn) handler.run(begin_ts=begin_ts, end_ts=end_ts) return { "statusCode": 200, "body": json.dumps( { "message": "hello world", # "location": ip.text.replace("\n", "") } ), } def list_alive_dedicated_clusters(clusters, cluster_type=CLUSTER_TYPE_DEDICATED): """List all alive clusters from clinic""" deploy_type = "deploy_type=dedicated" if cluster_type == CLUSTER_TYPE_NEXT_GEN: deploy_type = "deploy_type=premium&deploy_type=byoc" page = 1 per_page = 100 while True: resp = requests.get( ( f"{CLINIC_ENDPOINT}/clinic/api/v1/dashboard/clusters2?" f"{deploy_type}&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 clusters, cluster_type: {cluster_type}, status_code: {resp.status_code}" ) return data = resp.json() print(f"list_alive_clusters, cluster_type: {cluster_type}, page: {page}") for item in data["items"]: clusters[item["clusterID"]] = item if data["has_more"] is False: break page += 1 return def list_deleted_dedicated_clusters_ts( clusters, begin_ts: float, end_ts: float, cluster_type=CLUSTER_TYPE_DEDICATED ): """List all deleted clusters from clinic""" deploy_type = "deploy_type=dedicated" if cluster_type == CLUSTER_TYPE_NEXT_GEN: deploy_type = "deploy_type=premium&deploy_type=byoc" page = 1 per_page = 100 while True: resp = requests.get( ( f"{CLINIC_ENDPOINT}/clinic/api/v1/dashboard/clusters2?" f"{deploy_type}&show_deleted=true" f"&deleted_ts_from={begin_ts:.0f}&created_ts_to={end_ts:.0f}" 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 deleted clusters, cluster_type: {cluster_type}, status_code: {resp.status_code}" ) return data = resp.json() print(f"list_deleted_clusters_ts, cluster_type: {cluster_type}, page: {page}") for item in data["items"]: clusters[item["clusterID"]] = item if data["has_more"] is False: break page += 1 return def get_tidb_instance_count(cluster) -> int: """Safely read TiDB instance count from cluster topology data""" if not isinstance(cluster, dict): return 0 topology = cluster.get("topology") or {} if isinstance(topology, dict): tidb_count = topology.get("tidb", 0) try: return int(tidb_count) except (TypeError, ValueError): return 0 return 0 def is_enable_or_disable_tiproxy_event(event: dict) -> bool: """Check if the event is enable or disable tiproxy""" if event["name"] != "modifyclustersize": return False if event.get("payload", {}).get("detail", {}).get("extra_info", None) is None: return False extra_info = event["payload"]["detail"]["extra_info"] topology_before = extra_info.get("topology_before", {}).get("components", {}) topology_after = extra_info.get("topology_after", {}).get("components", {}) return ("tiproxy" in topology_before) != ("tiproxy" in topology_after) def get_cluster_event_window( cluster_id: str, begin_ts: float, end_ts: float, name: str ) -> list[tuple]: """Request activity hub to get cluster events""" windows = [] resp = requests.get( ( f"https://{ACTIVITY_HUB_ENDPOINT}/activities/{cluster_id}/from/{begin_ts:.0f}" f"/to/{end_ts:.0f}?name={name}&source=eventbus" ), timeout=15, headers={ "x-activityhub-source": ACTIVITY_HUB_SOURCE, "x-activityhub-key": ACTIVITY_HUB_KEY, }, ) if resp.status_code == 404: return windows data = resp.json() start_time = None for event in data["result"]: if "status" not in event["tags"]: continue # Skip cluster-size changes that don't enable or disable TiProxy. if event["name"] == "modifyclustersize" and not ( is_enable_or_disable_tiproxy_event(event) ): continue if event["tags"]["status"] == "started": record_start_time = datetime.fromtimestamp( event["record_time"], tz=timezone.utc ) start_time = record_start_time.replace(second=0).timestamp() elif event["tags"]["status"] == "succeeded" and start_time is not None: record_end_time = datetime.fromtimestamp( event["record_time"] + 60, tz=timezone.utc ) windows.append( ( event["name"], start_time, record_end_time.replace(second=0).timestamp(), ) ) start_time = None if start_time is not None: windows.append((name, start_time, end_ts)) return windows def generate_alert_content( cluster_id: str, cluster_name: str, cluster_deploy_type: str, tenant_id: str, tenant_name: str, month: str, ) -> dict: """Generate content of alert""" alert_name = "DedicatedTierSLABreak" if cluster_deploy_type == "premium": alert_name = "NextGenPremiumSLABreak" if cluster_deploy_type == "byoc": alert_name = "NextGenBYOCSLABreak" return { "project": {"key": JIRA_PROJECT}, "summary": f"[{SLI_STAGE.upper()}] {alert_name}", "description": f""" Labels: - cluster_id = {cluster_id} - cluster_name = {cluster_name} - tenant_id = {tenant_id} - tenant_name = {tenant_name} Annotations: - runbooks_url = https://pingcap.feishu.cn/wiki/XiUjw08iJiaElZkSnrNcgfu0nRe - ops_portal_link = https://ops.tidbcloud.com/sla/{cluster_id}/detail?month={month} """, "issuetype": {"name": "Task"}, "priority": {"name": "Major"}, "components": [{"name": "resilience"}], JIRA_CUSTOM_FIELD_TENANT_ID: tenant_id, JIRA_CUSTOM_FIELD_CLUSTER_ID: cluster_id, } def get_cluster_correction_windows( cluster_id: str, begin_ts: float, end_ts: float ) -> list[tuple]: """Get correction time windows of cluster""" windows = [] windows.extend( get_cluster_event_window(cluster_id, begin_ts, end_ts, "resumecluster") ) windows.extend( get_cluster_event_window(cluster_id, begin_ts, end_ts, "createcluster") ) windows.extend( get_cluster_event_window(cluster_id, begin_ts, end_ts, "deletecluster") ) windows.extend( get_cluster_event_window(cluster_id, begin_ts, end_ts, "modifyclustersize") ) return windows def set_fake_downtime( connection: pymysql.Connection, cluster_id: str, timestamp: float, comment: str ): """Set fake downtime for cluster""" with connection.cursor() as cursor: sql = f""" UPDATE `{SLI_TABLE_NAME}` SET fake=1, comment='{comment}' WHERE timestamp={timestamp:.0f} AND cluster_id='{cluster_id}' """ cursor.execute(sql) connection.commit() def summary_correction_windows( connection: pymysql.Connection, begin_ts: float, end_ts: float, downtimes: list ): """ Summary the downtime minutes of cluster, and skip the downtime minutes """ cluster_correction_windows = {} cluster_downtime_minute_count = {} skip_map = {} for row in downtimes: cluster_id = row["cluster_id"] downtime = row["downtime_time"] if cluster_id not in cluster_correction_windows: windows = get_cluster_correction_windows(cluster_id, begin_ts, end_ts) cluster_correction_windows[cluster_id] = windows skip_by_correction = False for window in cluster_correction_windows[cluster_id]: if window[1] <= downtime <= window[2]: if cluster_id not in skip_map: skip_map[cluster_id] = [(downtime, window[0])] else: skip_list = skip_map[cluster_id] skip_list.append((downtime, window[0])) skip_map[cluster_id] = skip_list set_fake_downtime( connection, cluster_id, downtime, f"skip by event: {window[0]}" ) skip_by_correction = True break if skip_by_correction: continue if cluster_id not in cluster_downtime_minute_count: cluster_downtime_minute_count[cluster_id] = 0 cluster_downtime_minute_count[cluster_id] += 1 print(f"cluster_skip_map: {skip_map}") print(f"cluster_downtime_minute_count: {cluster_downtime_minute_count}") return cluster_downtime_minute_count def get_date_from_timestamp(timestamp: float) -> str: """Get date from timestamp""" ts_datetime = datetime.fromtimestamp(timestamp, tz=timezone.utc) return f"{ts_datetime.year}-{ts_datetime.month}-{ts_datetime.day}" def list_jira_alerts(begin_ts: float, end_ts: float) -> dict: """List jira alerts from Jira""" issues_dedicated = sliutil.alert.search( f"project = {JIRA_PROJECT} AND summary ~ DedicatedTierSLABreak" f" AND created >= {get_date_from_timestamp(begin_ts)}" f" AND created <= {get_date_from_timestamp(end_ts)}" " ORDER BY created DESC" ) issues_premium = sliutil.alert.search( f"project = {JIRA_PROJECT} AND summary ~ NextGenPremiumSLABreak" f" AND created >= {get_date_from_timestamp(begin_ts)}" f" AND created <= {get_date_from_timestamp(end_ts)}" " ORDER BY created DESC" ) issues_byoc = sliutil.alert.search( f"project = {JIRA_PROJECT} AND summary ~ NextGenBYOCSLABreak" f" AND created >= {get_date_from_timestamp(begin_ts)}" f" AND created <= {get_date_from_timestamp(end_ts)}" " ORDER BY created DESC" ) issues = issues_dedicated + issues_premium + issues_byoc cluster_issue = {} for issue in issues: cluster_issue[issue["fields"]["customfield_10149"]] = { "key": issue["key"], "status": f"{issue['fields']['status']['name']}", } return cluster_issue class Handler: """Handle metrics `sli_dedicated_cluster_downtime`""" def __init__(self, connection: pymysql.Connection) -> None: self.connection = connection def run(self, begin_ts: float, end_ts: float) -> None: """Calculate SLA of TiDB Clusters from table `sli_dedicated_cluster_downtime` from begin_ts to end_ts and save the results into a new table """ print(f"Start to calculate SLA, begin_ts: {begin_ts}, end_ts: {end_ts}") # get dedicated clusters that existed between the begin_ts and end_ts dedicated_cluster_map = {} list_alive_dedicated_clusters(dedicated_cluster_map) list_deleted_dedicated_clusters_ts(dedicated_cluster_map, begin_ts, end_ts) print( f"number of dedicated clusters: {len(dedicated_cluster_map)}, msg: ", ",".join(dedicated_cluster_map.keys()), ) # get next-gen clusters that existed between the begin_ts and end_ts next_gen_cluster_map = {} list_alive_dedicated_clusters(next_gen_cluster_map, CLUSTER_TYPE_NEXT_GEN) list_deleted_dedicated_clusters_ts( next_gen_cluster_map, begin_ts, end_ts, CLUSTER_TYPE_NEXT_GEN ) print( f"number of next-gen clusters: {len(next_gen_cluster_map)}, msg: ", ",".join(next_gen_cluster_map.keys()), ) cluster_map = dedicated_cluster_map | next_gen_cluster_map month = sliutil.time.get_month_from_timestamp(timestamp=begin_ts) already_alerted = self.get_already_alerted_dict(month=month) downtime_cluster_sla = {} alerts = [] events = [] results = self.search(begin_ts=begin_ts, end_ts=end_ts) cluster_downtime_minute_count = summary_correction_windows( self.connection, begin_ts, end_ts, results ) for cluster_id, count in cluster_downtime_minute_count.items(): sla = 1 - count / sliutil.time.get_total_minutes_of_month( timestamp=begin_ts ) downtime_cluster_sla[cluster_id] = sla if cluster_id in already_alerted and already_alerted[cluster_id] is True: continue if cluster_id not in cluster_map: print(f"cluster {cluster_id} not found in cluster_map") continue if sla < 0.9999: cluster_deploy_type = cluster_map[cluster_id]["clusterDeployType"] if cluster_deploy_type == "dedicated": if ( get_tidb_instance_count(cluster_map[cluster_id]) == 1 and sla >= 0.999 ): print( f"cluster {cluster_id} has only one tidb, current sla: {sla}, skip alert" ) continue alerts.append( generate_alert_content( cluster_id=cluster_id, cluster_name=cluster_map[cluster_id]["clusterName"], cluster_deploy_type=cluster_deploy_type, tenant_id=cluster_map[cluster_id]["tenantID"], tenant_name=cluster_map[cluster_id]["tenantName"], month=month, ) ) # mark to alerted already_alerted[cluster_id] = True events.append(sliutil.alert.generate_event_content(name="SLABreak")) elif cluster_deploy_type == "premium": alerts.append( generate_alert_content( cluster_id=cluster_id, cluster_name=cluster_map[cluster_id]["clusterName"], cluster_deploy_type=cluster_deploy_type, tenant_id=cluster_map[cluster_id]["tenantID"], tenant_name=cluster_map[cluster_id]["tenantName"], month=month, ) ) # mark to alerted already_alerted[cluster_id] = True events.append(sliutil.alert.generate_event_content(name="SLABreak")) elif cluster_deploy_type == "byoc": # for byoc clusters, the threshold of SLABreak is tentatively set at 0.999 if sla < 0.999: alerts.append( generate_alert_content( cluster_id=cluster_id, cluster_name=cluster_map[cluster_id]["clusterName"], cluster_deploy_type=cluster_deploy_type, tenant_id=cluster_map[cluster_id]["tenantID"], tenant_name=cluster_map[cluster_id]["tenantName"], month=month, ) ) # mark to alerted already_alerted[cluster_id] = True events.append( sliutil.alert.generate_event_content(name="SLABreak") ) data = [] cluster_alerts = list_jira_alerts(begin_ts=begin_ts, end_ts=end_ts) for cluster in cluster_map.values(): alerted = False if ( cluster["clusterID"] in already_alerted and already_alerted[cluster["clusterID"]] is True ): alerted = True status = "perfect" tidb_instance_count = 1 if cluster["clusterDeployType"] == "dedicated": tidb_instance_count = get_tidb_instance_count(cluster) if cluster["clusterID"] in downtime_cluster_sla: if ( cluster_alerts.get(cluster["clusterID"], {}).get("status", None) == "Done" ): status = "breach" else: status = "warning" data.append( ( month, cluster["clusterID"], cluster["clusterName"], cluster["clusterDeployType"], status, downtime_cluster_sla.get(cluster["clusterID"], 1), alerted, cluster["tenantID"], cluster["tenantName"], "internal" if cluster["tenantID"] in INTERNAL_TENANT_IDS else "external", tidb_instance_count, cluster_alerts.get(cluster["clusterID"], {}).get("key", None), cluster_alerts.get(cluster["clusterID"], {}).get("status", None), ) ) self.save(data=data) sliutil.alert.create_alerts(alerts=alerts) if SLI_STAGE == "prod": sliutil.alert.send_events(events=events) print("Finish calculate SLA") def search(self, begin_ts: float, end_ts: float) -> list: """List downtime minutes from `sli_dedicated_cluster_downtime` from begin_ts to end_ts""" results = [] start = datetime.now(timezone.utc).timestamp() with self.connection.cursor() as cursor: sql = f""" SELECT cluster_id, timestamp as downtime_time FROM `{SLI_TABLE_NAME}` WHERE timestamp >= {begin_ts:.0f} AND timestamp <= {end_ts:.0f} AND fake=0 ORDER BY cluster_id ASC, downtime_time ASC """ # print("search, sql: ", sql) cursor.execute(sql) results.extend(cursor.fetchall()) end = datetime.now(timezone.utc).timestamp() print(f"search, take {end - start:.2f}s") return results def save(self, data: list) -> None: """Save data into table `sla`""" print(f"save data, len: {len(data)}") with self.connection.cursor() as cursor: sql = f""" INSERT INTO `{SLA_TABLE_NAME}` (month, cluster_id, cluster_name, cluster_deploy_type, status, value, alerted, tenant_id, tenant_name, tenant_kind, tidb_instance_count, issue_key, issue_status) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE status=VALUES(status), value=VALUES(value), alerted=VALUES(alerted), cluster_name=VALUES(cluster_name), cluster_deploy_type=VALUES(cluster_deploy_type), tenant_id=VALUES(tenant_id), tenant_name=VALUES(tenant_name), tenant_kind=VALUES(tenant_kind), tidb_instance_count=IF(VALUES(tidb_instance_count) != 0, VALUES(tidb_instance_count), tidb_instance_count), issue_key=VALUES(issue_key), issue_status=VALUES(issue_status) """ cursor.executemany(sql, data) self.connection.commit() def get_already_alerted_dict(self, month: str) -> dict[str, bool]: """Get dict for cluster id -> alerted""" with self.connection.cursor() as cursor: sql = f""" SELECT cluster_id, alerted FROM `{SLA_TABLE_NAME}` WHERE month = "{month}" AND alerted = true """ # print("get_already_alerted_dict, sql: ", sql) cursor.execute(sql) rows = cursor.fetchall() already_alerted = {} for row in rows: already_alerted[row["cluster_id"]] = bool(row["alerted"]) print( f"number of already alerted clusters: {len(already_alerted)}," f"msg: {already_alerted.keys()}" ) return already_alerted if __name__ == "__main__": # e = { # "type": "Event", # "name": "modifyclustersize", # "display_name": "ModifyClusterSize", # "catalog": "", # "severity": 0, # "activity_id": "eventbus-45f1d649-7cf4-4ed9-b56e-54cfeb6d2b1frewrite", # "application_id": "60178", # "concept_id": "3200385", # "target_id": "10628906340218045882", # "calibration_time": 1744838728, # "source": "eventbus", # "author": "system", # "context_id": "TW9kaWZ5Q2x1c3RlclNpemU-10628906340218045882-", # "context": None, # "tags": {"status": "started"}, # "payload": { # "author": "system", # "cluster_id": "10628906340218045882", # "detail": { # "cluster_id": "10628906340218045882", # "event_created_at_ms": "1744838728518", # "event_type": "ClusterScaling", # "extra_info": { # "topology_after": { # "components": { # "pd": { # "additional_iops": 0, # "component_type": "pd", # "quantity": 1, # "raft_log_iops": 0, # "raft_log_volume_type": "VOLUME_TYPE_UNSPECIFIED", # "storage_size_gi": 10, # "tier_key": "m6g.xlarge:DEPLOY_TYPE_DEDICATED:COMPONENT_TYPE_PD", # "tier_name": "m6g.xlarge", # "tier_unit_key": "aws:pd:m6g.xlarge:gp3:4000:400", # "volume_type": "gp3", # }, # "tidb": { # "additional_iops": 0, # "component_type": "tidb", # "quantity": 2, # "raft_log_iops": 0, # "raft_log_volume_type": "VOLUME_TYPE_UNSPECIFIED", # "storage_size_gi": 0, # "tier_key": "4C16G:DEPLOY_TYPE_DEDICATED:COMPONENT_TYPE_TIDB", # "tier_name": "4C16G", # "tier_unit_key": "aws:tidb:m6g.xlarge:local-storage:0:0", # "volume_type": "VOLUME_TYPE_UNSPECIFIED", # }, # "tikv": { # "additional_iops": 0, # "component_type": "tikv", # "quantity": 3, # "raft_log_iops": 3000, # "raft_log_volume_type": "gp3", # "storage_size_gi": 60, # "tier_key": "4C16G:DEPLOY_TYPE_DEDICATED:COMPONENT_TYPE_TIKV", # "tier_name": "4C16G", # "tier_unit_key": "aws:tikv:m6g.xlarge:gp3:4000:400:Standard", # "volume_type": "gp3", # }, # "tiproxy": { # "additional_iops": 0, # "component_type": "tiproxy", # "quantity": 2, # "raft_log_iops": 0, # "raft_log_volume_type": "VOLUME_TYPE_UNSPECIFIED", # "storage_size_gi": 0, # "tier_key": "4C8G:DEPLOY_TYPE_DEDICATED:COMPONENT_TYPE_TIPROXY", # "tier_name": "4C8G", # "tier_unit_key": "aws:tiproxy:c6g.xlarge:local-storage:0:0", # "volume_type": "VOLUME_TYPE_UNSPECIFIED", # }, # } # }, # "topology_before": { # "components": { # "pd": { # "additional_iops": 0, # "component_type": "pd", # "quantity": 1, # "raft_log_iops": 0, # "raft_log_volume_type": "VOLUME_TYPE_UNSPECIFIED", # "storage_size_gi": 10, # "tier_key": "m6g.xlarge:DEPLOY_TYPE_DEDICATED:COMPONENT_TYPE_PD", # "tier_name": "m6g.xlarge", # "tier_unit_key": "aws:pd:m6g.xlarge:gp3:4000:400", # "volume_type": "gp3", # }, # "tidb": { # "additional_iops": 0, # "component_type": "tidb", # "quantity": 2, # "raft_log_iops": 0, # "raft_log_volume_type": "VOLUME_TYPE_UNSPECIFIED", # "storage_size_gi": 0, # "tier_key": "4C16G:DEPLOY_TYPE_DEDICATED:COMPONENT_TYPE_TIDB", # "tier_name": "4C16G", # "tier_unit_key": "aws:tidb:m6g.xlarge:local-storage:0:0", # "volume_type": "VOLUME_TYPE_UNSPECIFIED", # }, # "tikv": { # "additional_iops": 0, # "component_type": "tikv", # "quantity": 3, # "raft_log_iops": 3000, # "raft_log_volume_type": "gp3", # "storage_size_gi": 60, # "tier_key": "4C16G:DEPLOY_TYPE_DEDICATED:COMPONENT_TYPE_TIKV", # "tier_name": "4C16G", # "tier_unit_key": "aws:tikv:m6g.xlarge:gp3:4000:400:Standard", # "volume_type": "gp3", # }, # } # }, # }, # "is_dev_tier": False, # "openapi_public_key": "", # "operation_id": "0", # "operator_id": "0", # "project_id": "3200385", # "status": "SCALING", # "tenant_id": "60178", # "user_email": "", # }, # "display_event_type": "ModifyClusterSize", # "display_status": "started", # "happened_at_ms": "1744838728518", # "project_id": "3200385", # "summary": "Start to modify cluster size.\nTopology before: TiDB x2 (4C16G) in DefaultGroup | TiKV x3 (4C16G,60GiB,Standard).\nTopology after: TiDB x2 (4C16G) and TiProxy x2 (4C8G) in DefaultGroup | TiKV x3 (4C16G,60GiB,Standard).", # "tenant_id": "60178", # }, # } # print(is_enable_or_disable_tiproxy_event(e)) clusters = {} list_alive_dedicated_clusters(clusters) print(clusters)