import os import time import tempfile from o11ydiagnosis.subtasks.slowlogconverter.task import Task from o11ydiagnosis.subtasks.slowlogconverter.metrics import metrics from o11ydiagnosis.subtasks.slowlogconverter.objstore import S3, GCS, AzureBlob class Config: task: Task cluster: str interval_secs: int def __init__(self, task, task_info, subtask_info): self.engine_task = task self.local_path = task.cache_prefix env: str = task_info["meta"]["env"] # prod/staging/dev provider: str = task_info["meta"]["provider"].lower() # aws/gcp region: str = task_info["meta"]["region"] # TODO: use env var ? tenant: str = task_info["meta"]["tenantID"] project: str = task_info["meta"]["projectID"] cluster: str = task_info["meta"]["clusterID"] o11y_cluster: str = task_info["meta"]["o11yClusterID"] cluster_type = task_info["meta"]["clusterType"].lower() serverless = cluster_type == "serverless" nextgen = cluster_type == "nextgen" or cluster_type == "premium" # interval_secs: int = subtask_info["interval"] if provider == "aws": s3 = S3() elif provider == "gcp": s3 = GCS() elif provider == "azure": s3 = AzureBlob() elif provider == "alicloud": s3 = S3() else: raise ValueError("invalid provider: " + provider) o11y_bucket = o11y_s3_bucket(env, region, serverless, nextgen) o11y_prefix = o11y_s3_prefix(tenant, project, cluster, serverless, nextgen) drhouse_bucket = drhouse_s3_bucket(region, env) drhouse_prefix = drhouse_s3_prefix(cluster) drhouse_local_path = self.local_path.rstrip("/") self.task = Task(s3, tenant, cluster, o11y_cluster, provider, region, o11y_bucket, o11y_prefix, drhouse_bucket, drhouse_prefix, drhouse_local_path) self.cluster = cluster metrics.register_backfill_handler(run_backfill_from_payload) # self.interval_secs = interval_secs self.interval_secs = 30 # temp hardcode def is_timing(self, ts: int) -> bool: # utc_time = datetime.utcfromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S") # log.info(f"[{self.cluster}] calling is_timing(ts={ts}) (utc time: {utc_time})") return ts % self.interval_secs == 0 def ready(self, ts: int) -> bool: return True def start(self, ts: int): # utc_time = datetime.utcfromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S") # log.info(f"[{self.cluster}] calling start(ts={ts}) (utc time: {utc_time})") now = int(time.time()) if ts + 3600 < now: # supplement mode if ts % 3600 == 0 or ts % 3600 == 3570: # Trigger at the hour and 30s before the next hour self.task.run(ts) else: # realtime mode hour_ts = ts - ts % 3600 if ts % 3600 <= 300: self.task.run(hour_ts - 3600) self.task.run(hour_ts) def close(self, ts: int): # utc_time = datetime.utcfromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S") # log.info(f"[{self.cluster}] calling close(ts={ts}) (utc time: {utc_time})") # paths = list(self.task.dirty_statsdb_paths.keys()) + list(self.task.dirty_merged_logs_paths.keys()) # paths = [path.removeprefix(self.local_path).lstrip("/") for path in paths] # if paths: # log.info(f"[{self.cluster}] uploading paths: {paths}") # self.engine_task.push(paths) # self.task.dirty_statsdb_paths.clear() # for merged_logs_path in self.task.dirty_merged_logs_paths.keys(): # os.remove(merged_logs_path) # self.task.dirty_merged_logs_paths.clear() pass # in o11y env, we use prod env to support staging environment. def o11y_s3_bucket(env: str, region: str, serverless: bool, nextgen: bool) -> str: bucket_name = os.environ['BUCKET_NAME'] if "BUCKET_NAME" in os.environ else None if bucket_name: return bucket_name if serverless or nextgen: return f"o11y-prod-shared-{region}" if env.lower() == "staging": return f"o11y-prod-shared-{region}" if env.lower() == "dev": return f"o11y-dev-shared-{region}" if env.lower() == "prod": return f"o11y-prod-shared-{region}" return f"o11y-test-shared-{region}" def o11y_s3_prefix(tenant: str, project: str, cluster: str, serverless: bool, nextgen: bool) -> str: if serverless: return f"0/prod-devtier/serverless/{tenant}/{project}/{cluster}" if nextgen: return f"0/next-gen/{cluster}" return f"0/{tenant}/{project}/{cluster}" def drhouse_s3_bucket(region: str, env: str) -> str: bucket_name = os.environ['BUCKET_NAME'] if "BUCKET_NAME" in os.environ else None if bucket_name: return bucket_name if env.lower() == "staging": return f"o11y-prod-shared-{region}" if env.lower() == "dev": return f"o11y-dev-shared-{region}" if env.lower() == "prod": return f"o11y-prod-shared-{region}" return f"o11y-test-shared-{region}" def drhouse_s3_prefix(cluster: str) -> str: return f"diagnosis/data/{cluster}" def _build_objstore(provider: str): provider = provider.lower() if provider == "aws" or provider == "alicloud": return S3() if provider == "gcp": return GCS() if provider == "azure": return AzureBlob() raise ValueError(f"invalid provider: {provider}") def run_backfill_from_payload(payload: dict): required = [ "provider", "region", "tenant", "project", "cluster", "o11y_cluster", "env", "cluster_type", "start_ts", "end_ts", ] missing = [k for k in required if payload.get(k) is None] if missing: raise ValueError(f"missing required fields: {', '.join(missing)}") provider = str(payload["provider"]).lower() region = str(payload["region"]) tenant = str(payload["tenant"]) project = str(payload["project"]) cluster = str(payload["cluster"]) o11y_cluster = str(payload["o11y_cluster"]) env = str(payload["env"]) cluster_type = str(payload["cluster_type"]).lower() start_ts = int(payload["start_ts"]) end_ts = int(payload["end_ts"]) if start_ts >= end_ts: raise ValueError("start_ts must be less than end_ts") serverless = cluster_type == "serverless" nextgen = cluster_type == "nextgen" or cluster_type == "premium" s3 = _build_objstore(provider) o11y_bucket = str(payload.get("o11y_bucket") or o11y_s3_bucket(env, region, serverless, nextgen)) o11y_prefix = str(payload.get("o11y_prefix") or o11y_s3_prefix(tenant, project, cluster, serverless, nextgen)) drhouse_bucket = str(payload.get("drhouse_bucket") or drhouse_s3_bucket(region, env)) drhouse_prefix = str(payload.get("drhouse_prefix") or drhouse_s3_prefix(cluster)) default_local_path = os.path.join(tempfile.gettempdir(), "slowlog-backfill", cluster) drhouse_local_path = str(payload.get("drhouse_local_path") or default_local_path) task = Task( s3=s3, tenant=tenant, cluster=cluster, o11y_cluster=o11y_cluster, vendor=provider, region=region, o11y_bucket=o11y_bucket, o11y_prefix=o11y_prefix, drhouse_bucket=drhouse_bucket, drhouse_prefix=drhouse_prefix, drhouse_local_path=drhouse_local_path, ) task.run_range(start_ts, end_ts)