import os import time import sqlite3 import threading import pickle import tempfile from datetime import datetime, timezone from concurrent.futures import ThreadPoolExecutor from threading import Timer from typing import List from pympler import muppy, summary from o11ydiagnosis.subtasks.slowlogconverter.log import log from o11ydiagnosis.subtasks.slowlogconverter.metrics import metrics from o11ydiagnosis.subtasks.slowlogconverter.objstore import ObjectStore from o11ydiagnosis.subtasks.slowlogconverter.slowlog import StreamParser from o11ydiagnosis.subtasks.slowlogconverter.write_deltalake import write_slowlogs_to_deltalake, try_to_optimize_table OPTIMIZE_WORKERS = int(os.getenv("SLOWLOG_OPTIMIZE_WORKERS", "4")) OPTIMIZE_EXECUTOR = ThreadPoolExecutor(max_workers=max(1, OPTIMIZE_WORKERS)) _OPTIMIZE_INFLIGHT_CLUSTERS = set() _OPTIMIZE_PENDING_CLUSTERS = set() _OPTIMIZE_LOCK = threading.Lock() _TASK_REGISTRY = {} _TASK_REGISTRY_LOCK = threading.Lock() def log_metrics_periodically(): """Log the oldest processed hour and memory usage every 5 minutes, and clean up stale cluster metrics.""" try: current_time = time.time() # Collect clusters to clean up clusters_to_clean = [] for cluster, hour_ts in metrics._cluster_hours.items(): last_update_ts = metrics.last_update_timestamp_for_cluster(cluster) if last_update_ts is not None and current_time - last_update_ts > 900: # 15 minutes threshold clusters_to_clean.append(cluster) # Clean up collected clusters for cluster in clusters_to_clean: log.info(f"[Metrics] Cleaning up stale metrics for cluster {cluster} (inactive for >15 minutes)") # Remove all metrics for this cluster utc_hour = datetime.fromtimestamp(2147483647).strftime("%Y%m%d%H") metrics.record_logs_written(0, cluster, utc_hour, 2147483647, 2147483647, 0, 2147483647) # Log remaining active cluster metrics oldest = metrics.get_oldest_processed_hour() if oldest: cluster, hour_ts = oldest hour_str = datetime.fromtimestamp(hour_ts, tz=timezone.utc).strftime('%Y%m%d%H') log.info(f"[Metrics] Oldest processed hour - Cluster: {cluster}, UTC Hour: {hour_str}") # Log memory usage all_objects = muppy.get_objects() sum_obj = summary.summarize(all_objects) sum_obj.sort(key=lambda x: x[2], reverse=True) # Sort by size log.info("[Memory] Top 10 memory-consuming objects:") for obj in sum_obj[:10]: log.info(f"[Memory] {obj[0]}: {obj[1]} objects, {obj[2]/1024/1024:.2f} MB") # Log 1-minute metrics logs_count = metrics.total_logs_written_1m() log.info(f"[Metrics] Logs written in last 1 minute: {logs_count}") metrics.reset_logs_written_1m() # Schedule next run global _metrics_timer _metrics_timer = Timer(60, log_metrics_periodically) _metrics_timer.daemon = True # Allow program to exit even if timer is still running _metrics_timer.start() except Exception as e: log.error(f"[Metrics] Error logging metrics: {e}") # Global timer variable to allow cleanup _metrics_timer = None def start_metrics_logging(): """Start the periodic metrics logging""" global _metrics_timer if _metrics_timer is not None: _metrics_timer.cancel() log_metrics_periodically() def stop_metrics_logging(): """Stop the periodic metrics logging""" global _metrics_timer if _metrics_timer is not None: _metrics_timer.cancel() _metrics_timer = None # Start the metrics logging when the module is imported start_metrics_logging() class Task: s3: ObjectStore tenant: str cluster: str o11y_cluster: str o11y_bucket: str o11y_prefix: str drhouse_bucket: str drhouse_prefix: str drhouse_local_path: str def __init__( self, s3: ObjectStore, tenant: str, cluster: str, o11y_cluster: str, vendor: str, region: str, o11y_bucket: str, o11y_prefix: str, drhouse_bucket: str, drhouse_prefix: str, drhouse_local_path: str, ): self.s3 = s3 self.tenant = tenant self.cluster = cluster self.o11y_cluster = o11y_cluster self.vendor = vendor self.region = region self.o11y_bucket = o11y_bucket self.o11y_prefix = o11y_prefix.rstrip("/") self.drhouse_bucket = drhouse_bucket self.drhouse_prefix = drhouse_prefix.rstrip("/") self.drhouse_local_path = drhouse_local_path.rstrip("/") # 创建临时目录 self.temp_dir = os.path.join(self.drhouse_local_path, "temp") self._ensure_temp_dir() with _TASK_REGISTRY_LOCK: _TASK_REGISTRY[self.cluster] = self @classmethod def trigger_backfill(cls, cluster: str, start_ts: int, end_ts: int): with _TASK_REGISTRY_LOCK: task = _TASK_REGISTRY.get(cluster) if task is None: raise ValueError(f"cluster {cluster} not found") task.run_range(start_ts, end_ts) def run_range(self, start_ts: int, end_ts: int): if start_ts >= end_ts: raise ValueError("start_ts must be less than end_ts") cursor = int(start_ts) end_ts = int(end_ts) while cursor < end_ts: window_end = min(cursor + 3600, end_ts) self.run(cursor, window_begin=cursor, window_end=window_end, schedule_optimize=False) cursor = window_end def _ensure_temp_dir(self): """确保临时目录存在且为空""" try: # 删除旧的临时目录（如果存在） if os.path.exists(self.temp_dir): import shutil shutil.rmtree(self.temp_dir) # 创建新的临时目录 os.makedirs(self.temp_dir, exist_ok=True) log.info(f"[{self.cluster}] created clean temp directory: {self.temp_dir}") except Exception as e: log.warning(f"[{self.cluster}] failed to ensure temp directory: {e}") raise def _resolved_last_optimize_timestamp(self, fallback_path: str = None) -> float: """Return a safe last optimize timestamp. - If metric is missing/zero, optionally use a staggered fallback derived from path hash. - If metric is unexpectedly in the future, treat it as invalid and use the same fallback. """ last_optimize_timestamp = metrics.last_optimize_timestamp.labels(cluster=self.cluster)._value.get() now = time.time() use_fallback = False if last_optimize_timestamp is None or last_optimize_timestamp <= 0: use_fallback = True elif last_optimize_timestamp > now + 300: log.warning( f"[{self.cluster}] last_optimize_timestamp in future " f"({last_optimize_timestamp}), fallback to safe value" ) use_fallback = True if use_fallback: if fallback_path is None: return 0 return now - hash(fallback_path) % 3600 return last_optimize_timestamp def _need_optimize(self) -> bool: last_optimize_timestamp = self._resolved_last_optimize_timestamp() if last_optimize_timestamp <= 0: return True return time.time() - last_optimize_timestamp > 3600 def _update_optimize_metrics(self): with _OPTIMIZE_LOCK: pending_global = len(_OPTIMIZE_PENDING_CLUSTERS) inflight_global = len(_OPTIMIZE_INFLIGHT_CLUSTERS) running_global = max(0, inflight_global - pending_global) pending = 1 if self.cluster in _OPTIMIZE_PENDING_CLUSTERS else 0 running = 1 if (self.cluster in _OPTIMIZE_INFLIGHT_CLUSTERS and self.cluster not in _OPTIMIZE_PENDING_CLUSTERS) else 0 metrics.update_optimize_state( cluster=self.cluster, pending=pending, running=running, pending_global=pending_global, running_global=running_global ) def _schedule_background_optimize(self, begin: int, deltalake_path: str): with _OPTIMIZE_LOCK: if self.cluster in _OPTIMIZE_INFLIGHT_CLUSTERS: log.info(f"[{self.cluster}] optimize job already running, skip scheduling") metrics.record_optimize_skipped(self.cluster, "already_inflight") return _OPTIMIZE_INFLIGHT_CLUSTERS.add(self.cluster) _OPTIMIZE_PENDING_CLUSTERS.add(self.cluster) metrics.record_optimize_scheduled(self.cluster) self._update_optimize_metrics() def _background_job(): optimize_start = time.time() with _OPTIMIZE_LOCK: _OPTIMIZE_PENDING_CLUSTERS.discard(self.cluster) self._update_optimize_metrics() try: self._run_sqlstatement_and_topsql_optimize(begin, deltalake_path) optimize_end = time.time() duration = optimize_end - optimize_start metrics.record_optimize_completed(self.cluster, "success", duration) log.info(f"[{self.cluster}] background optimize finished, cost {optimize_end - optimize_start:.2f} seconds") except Exception as e: duration = time.time() - optimize_start metrics.record_optimize_completed(self.cluster, "failed", duration) log.error(f"[{self.cluster}] background optimize failed: {e}") finally: with _OPTIMIZE_LOCK: _OPTIMIZE_PENDING_CLUSTERS.discard(self.cluster) _OPTIMIZE_INFLIGHT_CLUSTERS.discard(self.cluster) self._update_optimize_metrics() try: OPTIMIZE_EXECUTOR.submit(_background_job) log.info(f"[{self.cluster}] scheduled background optimize job") except Exception: with _OPTIMIZE_LOCK: _OPTIMIZE_PENDING_CLUSTERS.discard(self.cluster) _OPTIMIZE_INFLIGHT_CLUSTERS.discard(self.cluster) metrics.record_optimize_skipped(self.cluster, "submit_failed") self._update_optimize_metrics() raise def _run_sqlstatement_and_topsql_optimize(self, begin: int, deltalake_path: str): # optimize sql statement table sqlstatementpath = f"deltalake/{self.tenant}/{self.o11y_cluster}/persisted_statements_summary" sqlstatementcheckpoint = f"{sqlstatementpath}/_delta_log/_last_checkpoint" sqlstatementorignal = f"{sqlstatementpath}/_delta_log/00000000000000000000.json" if self.s3.existed(self.o11y_bucket, sqlstatementcheckpoint) or self.s3.existed(self.o11y_bucket, sqlstatementorignal): optimize_sqlstatement_start = time.time() try_to_optimize_table(self.region, deltalake_path + "/persisted_statements_summary", datetime.utcfromtimestamp(begin).strftime("%Y%m%d")) optimize_sqlstatement_end = time.time() log.info(f"[{self.cluster}] {sqlstatementpath} optimize_sqlstatement_end, cost {optimize_sqlstatement_end - optimize_sqlstatement_start:.2f} seconds") else: log.info(f"[{self.cluster}] {sqlstatementpath} not existed, skip optimize") # optimize topsql table topsql_root_key = f"deltalake/org={self.tenant}/cluster={self.cluster}/type=topsql" def topsql_table_uri(relative_table_key: str) -> str: # Convert a bucket-relative key to a full URI for DeltaTable. if self.vendor == "aws" or self.vendor == "alicloud": return f"s3://{self.o11y_bucket}/{relative_table_key}" if self.vendor == "azure": storage_account = os.getenv("AZURE_STORAGE_ACCOUNT") return f"abfss://{self.o11y_bucket}@{storage_account}.dfs.core.windows.net/{relative_table_key}" if self.vendor == "gcp": return f"gs://{self.o11y_bucket}/{relative_table_key}" if self.vendor == "local": return relative_table_key raise ValueError(f"unsupported vendor: {self.vendor}") def optimize_if_exists(relative_table_key: str): checkpoint_key = f"{relative_table_key}/_delta_log/_last_checkpoint" original_key = f"{relative_table_key}/_delta_log/00000000000000000000.json" if self.s3.existed(self.o11y_bucket, checkpoint_key) or self.s3.existed(self.o11y_bucket, original_key): optimize_topsql_start = time.time() try_to_optimize_table( self.region, topsql_table_uri(relative_table_key), datetime.utcfromtimestamp(begin).strftime("%Y%m%d"), ) optimize_topsql_end = time.time() log.info( f"[{self.cluster}] topsql table optimized: {relative_table_key}, cost {optimize_topsql_end - optimize_topsql_start:.2f} seconds" ) # 1) components without instance layer for component in ["topsql_plan_meta", "topsql_sql_meta"]: optimize_if_exists(f"{topsql_root_key}/component={component}") # 2) components with instance layer for component in ["tikv", "tidb", "topru"]: instance_prefix = f"{topsql_root_key}/component={component}/instance=" listed_keys = self.s3.list(self.o11y_bucket, instance_prefix) instance_ids = set() for k in listed_keys: idx = k.find("/instance=") if idx == -1: continue after = k[idx + len("/instance="):] instance_id = after.split("/", 1)[0] if instance_id: instance_ids.add(instance_id) for instance_id in instance_ids: optimize_if_exists(f"{topsql_root_key}/component={component}/instance={instance_id}") def run(self, hour_ts: int, window_begin: int = None, window_end: int = None, schedule_optimize: bool = True): begin = hour_ts - hour_ts % 3600 end = begin + 3600 if window_begin is not None and window_end is not None: begin = int(window_begin) end = int(window_end) if begin >= end: raise ValueError("window_begin must be less than window_end") utc_hour = datetime.utcfromtimestamp(begin).strftime("%Y%m%d%H") log.info(f"[{self.cluster}] calling run(hour_ts={hour_ts}, begin={begin}, end={end}) (utc hour: {utc_hour})") deltalake_path = deltalake_base_path(self.vendor, self.drhouse_bucket, self.tenant, self.o11y_cluster) # optimize in background, do not block slowlogs processing if schedule_optimize and self._need_optimize(): self._schedule_background_optimize(begin, deltalake_path) # make sure temp dir exists self._ensure_temp_dir() metadb_path = f"{self.drhouse_local_path}/meta.db" with sqlite3.connect(metadb_path) as metadb: metadb.isolation_level = None metadb.execute(""" CREATE TABLE IF NOT EXISTS "processed_objects" ( "timestamp" INTEGER NOT NULL, "object_key" TEXT NOT NULL, PRIMARY KEY ("timestamp", "object_key") ); """) slowlog_path = f"{self.o11y_prefix}/logs/slowlog" matched_keys = find_matched_keys(self.s3, self.o11y_bucket, slowlog_path, begin, end) log.info(f"[{self.cluster}] matched objects: {len(matched_keys)}") if not matched_keys: last_optimize_timestamp = self._resolved_last_optimize_timestamp(fallback_path=slowlog_path) metrics.record_logs_written(0, self.cluster, utc_hour, time.time(), last_optimize_timestamp, 0) return cursor = metadb.cursor() try: cursor.execute( """ SELECT "object_key" FROM "processed_objects" WHERE "timestamp" >= ?; """, (begin - 3600,), ) rows = cursor.fetchall() existed_keys = [row[0] for row in rows] finally: cursor.close() log.info(f"[{self.cluster}] loaded existed keys: {len(existed_keys)}") matched_keys = set(matched_keys) added_keys = list(matched_keys - set(existed_keys)) log.info(f"[{self.cluster}] new objects: {len(added_keys)}") if not added_keys: last_optimize_timestamp = self._resolved_last_optimize_timestamp(fallback_path=slowlog_path) metrics.record_logs_written(0, self.cluster, utc_hour, time.time(), last_optimize_timestamp, 0) log.info(f"[{self.cluster}] no new object, skipped") return # Read, parse and merge added slow logs. FILE_BATCH_SIZE = 100 total_files = added_keys download_time = 0 insert_time = 0 for i in range(0, len(total_files), FILE_BATCH_SIZE): added_keys = total_files[i:i + FILE_BATCH_SIZE] download_start = time.time() log.info(f"[{self.cluster}] downloading {i}/{len(total_files)} batch objects: {len(added_keys)}") # Clean up any existing .pkl files in the directory for filename in os.listdir(self.drhouse_local_path): if filename.endswith('.pkl'): file_path = os.path.join(self.drhouse_local_path, filename) try: os.remove(file_path) log.info(f"Removed existing .pkl file: {file_path}") except OSError as e: log.warning(f"Failed to remove {file_path}: {e}") # Create a single temporary file for all logs temp_file = tempfile.NamedTemporaryFile('wb', delete=False, dir=self.drhouse_local_path, suffix=".pkl") temp_filename = temp_file.name temp_file.close() file_lock = threading.Lock() lastest_record_timestamp = 0 parse_error_count = 0 def _thread_task(key: str) -> int: """Process a single log file and append logs to the shared temp file. Returns the number of logs processed.""" log_count = 0 nonlocal lastest_record_timestamp, parse_error_count parse_start = time.time() try: # 处理S3 key，去除.gz后缀（因为S3会自动解压） s3_key = os.path.basename(key) if s3_key.endswith('.gz'): s3_key = s3_key[:-3] # 去掉.gz后缀 local_file_path = os.path.join(self.temp_dir, s3_key) self.s3.download(self.o11y_bucket, key, local_file_path) # 使用本地文件进行解析 with open(local_file_path, 'rb') as f: parser = StreamParser(f) sequence = 0 batch = [] current_record_timestamp = 0 current_parse_error_count = 0 while True: try: sl = parser.read_next() except ValueError as e: log.error(f"[{self.cluster}] failed to parse log in {s3_key}: {e}") current_parse_error_count += 1 continue if not sl: break if not sl.time: current_parse_error_count += 1 continue if not sl.digest: current_parse_error_count += 1 continue if sl.time > current_record_timestamp: current_record_timestamp = sl.time dt_object = datetime.fromtimestamp(sl.time, tz=timezone.utc) sl.id = generate_id(dt_object, sequence) sequence += 1 batch.append(sl) log_count += 1 # Write batch to file when it reaches a certain size if len(batch) >= 100: # Adjust batch size as needed with file_lock: with open(temp_filename, 'ab') as f: pickle.dump(batch, f) if current_record_timestamp > lastest_record_timestamp: lastest_record_timestamp = current_record_timestamp parse_error_count += current_parse_error_count batch = [] # Write any remaining logs in the last batch if batch: with file_lock: with open(temp_filename, 'ab') as f: pickle.dump(batch, f) if current_record_timestamp > lastest_record_timestamp: lastest_record_timestamp = current_record_timestamp parse_error_count += current_parse_error_count except Exception as e: log.exception(f"[{self.cluster}] failed to process {key}: {str(e)}") finally: # 清理临时文件 try: os.remove(local_file_path) except OSError as e: log.warning(f"Failed to remove {local_file_path}: {e}") parse_end = time.time() log.info(f"[{self.cluster}] processed {key}: {log_count} logs, cost {parse_end-parse_start:.2f} seconds") return log_count # Process all files in parallel _THREAD_COUNT = min(len(added_keys), 100) # Reduced thread count total_logs = 0 with ThreadPoolExecutor(_THREAD_COUNT) as executor: log_counts = list(executor.map(_thread_task, added_keys)) total_logs = sum(log_counts) download_end = time.time() log.info(f"[{self.cluster}] finished processing {len(added_keys)} files, total logs: {total_logs}, cost {download_end-download_start:.2f} seconds") download_time += download_end - download_start insert_start = time.time() if total_logs == 0: os.remove(temp_filename) last_optimize_timestamp = self._resolved_last_optimize_timestamp(fallback_path=slowlog_path) metrics.record_logs_written(0, self.cluster, utc_hour, time.time(), last_optimize_timestamp, parse_error_count) continue def read_logs_in_batches(filename, batch_size): """Read logs from the temporary file in batches. Each item in the file is a list of logs from a single batch.""" with open(filename, 'rb') as f: while True: try: batch = pickle.load(f) for log_item in batch: yield log_item except EOFError: break LOG_BATCH_SIZE = int(os.getenv("LOG_BATCH_SIZE", 10000)) total_logs_written = 0 current_batch = [] # Generate partition date from the hour timestamp partition_date = datetime.utcfromtimestamp(begin).strftime("%Y%m%d") # Process logs in batches optimize_count = 0 for log_item in read_logs_in_batches(temp_filename, LOG_BATCH_SIZE): current_batch.append(log_item) if len(current_batch) >= LOG_BATCH_SIZE: log.info(f"[{self.cluster}] writing batch with {len(current_batch)} logs to deltalake") deltalake_start_time = time.time() optimize_count += 1 last_optimize_timestamp = self._resolved_last_optimize_timestamp(fallback_path=deltalake_path) optimize = time.time() - last_optimize_timestamp > 3600 write_slowlogs_to_deltalake( current_batch, self.region, self.cluster, deltalake_path, optimize=optimize or self.vendor == "local", partition_date=partition_date, merge_write=False ) deltalake_end_time = time.time() log_count = len(current_batch) total_logs_written += log_count # Record metrics with the current UTC hour metrics.record_logs_written(log_count, self.cluster, utc_hour, lastest_record_timestamp, last_optimize_timestamp if not optimize else time.time(), parse_error_count) log.info(f"[{self.cluster}] batch written, cost {deltalake_end_time - deltalake_start_time:.2f} seconds") current_batch = [] # Write the last batch if it's not empty if current_batch: log.info(f"[{self.cluster}] writing final batch with {len(current_batch)} logs to deltalake") deltalake_start_time = time.time() last_optimize_timestamp = self._resolved_last_optimize_timestamp(fallback_path=deltalake_path) optimize = time.time() - last_optimize_timestamp > 3600 write_slowlogs_to_deltalake( current_batch, self.region, self.cluster, deltalake_path, optimize=optimize or self.vendor == "local", partition_date=partition_date, merge_write=False ) deltalake_end_time = time.time() log_count = len(current_batch) total_logs_written += log_count # Record metrics with the current UTC hour metrics.record_logs_written(log_count, self.cluster, utc_hour, lastest_record_timestamp, last_optimize_timestamp if not optimize else time.time(), parse_error_count) log.info(f"[{self.cluster}] final batch written, cost {deltalake_end_time - deltalake_start_time:.2f} seconds") # Clean up try: os.remove(temp_filename) except Exception as e: log.error(f"[{self.cluster}] failed to remove temp file {temp_filename}: {e}") insert_end = time.time() log.info(f"[{self.cluster}] finished writing {total_logs_written} logs to deltalake, cost {insert_end-insert_start:.2f} seconds") insert_time += insert_end - insert_start # Mark all processed keys in the database data_to_insert = [(begin, key) for key in added_keys] if data_to_insert: # Only execute if there are items to insert metadb.executemany( """ INSERT OR REPLACE INTO "processed_objects" ( "timestamp", "object_key" ) VALUES ( ?, ? ); """, data_to_insert, ) log.info(f"[{self.cluster}] marked {len(data_to_insert)} keys as processed in the database") # Delete expired records expired_ts = begin - 24 * 3600 metadb.execute( """DELETE FROM "processed_objects" WHERE "timestamp" < ?;""", (expired_ts,), ) log.info(f"[{self.cluster}] finished processing {len(added_keys)} files, total logs: {total_logs}, download time: {download_time:.2f} seconds, insert time: {insert_time:.2f} seconds") # [begin, end) def find_common_ts_prefixes_limit_1h(begin: int, end: int) -> List[str]: prefixes = [] base = begin // 1000 for n in range(0, 5): # 0 ~ 4 MUST covered 1h (+5min) if (base + n) * 1000 > end: break prefixes.append(str(base + n)) return prefixes def find_matched_keys(s3: ObjectStore, bucket: str, path: str, begin: int, end: int) -> List[str]: prefixes = [f"{path}/{p}" for p in find_common_ts_prefixes_limit_1h(begin, end + 300)] # Assume 5 minutes max upload latency keys = s3.lists(bucket, prefixes) matched_keys = [] for k in keys: filename = os.path.basename(k) upload_ts = int(filename[:10]) if upload_ts < begin or upload_ts > end + 300: # Assume 5 minutes max upload latency continue matched_keys.append(k) return matched_keys def read_keys_from_merged_logs(filepath: str) -> List[str]: with open(filepath) as f: lines = f.readlines() keys = [] for line in lines: if line.startswith("# "): keys.append(line.lstrip("# ").strip()) return keys def generate_id(dt: datetime, sequence_number: int) -> int: timestamp = int(dt.timestamp()) if not 0 <= timestamp < (1 << 34): raise ValueError(f"Timestamp {timestamp} exceeds 34 bits limit") if not 0 <= sequence_number < (1 << 28): raise ValueError(f"Sequence {sequence_number} exceeds 28 bits limit") return (timestamp << 28) | sequence_number def parse_id(id_val: int) -> tuple[datetime, int]: timestamp = id_val >> 28 sequence_number = id_val & ((1 << 28) - 1) dt = datetime.fromtimestamp(timestamp) return dt, sequence_number def deltalake_base_path(vendor, bucket, tenant, o11y_cluster): if vendor == "aws" or vendor == "alicloud": return f"s3://{bucket}/deltalake/{tenant}/{o11y_cluster}" if vendor == "azure": storage_account = os.getenv("AZURE_STORAGE_ACCOUNT") return f"abfss://{bucket}@{storage_account}.dfs.core.windows.net/deltalake/{tenant}/{o11y_cluster}" if vendor == "gcp": return f"gs://{bucket}/deltalake/{tenant}/{o11y_cluster}" if vendor == "local": return "./output/leo"