from app.logcli.utils.logger import log_info, log_warning, log_error import time import duckdb import sys import importlib.util from pathlib import Path from typing import List, Optional from abc import ABC, abstractmethod from dataclasses import dataclass # Add the project root to the Python path project_root = str(Path(__file__).parent.parent.parent) if project_root not in sys.path: sys.path.append(project_root) @dataclass class TimeRange: start_ts: int end_ts: int class DeltaLakeStore: vendor: str region: str tenant_id: str o11y_id: str def __init__(self, vendor: str, region: str, tenant_id: str, o11y_id: str): self.vendor = vendor self.region = region self.tenant_id = tenant_id self.o11y_id = o11y_id self.db = duckdb.connect(database=":memory:", read_only=False) self.db.execute("INSTALL httpfs;") self.db.execute("LOAD httpfs;") self.db.execute("INSTALL delta;") self.db.execute("LOAD delta;") self.db.execute("SET enable_object_cache=true;") self.db.execute("SET enable_http_metadata_cache=true;") self.db.execute("SET enable_external_file_cache=true;") self.db.execute("SET parquet_metadata_cache=true;") self.db.execute("SET preserve_insertion_order=false;") if vendor == "aws": self.db.execute("INSTALL aws;") self.db.execute("LOAD aws;") self.db.execute( f"CREATE OR REPLACE SECRET secret (TYPE s3, PROVIDER credential_chain, ENDPOINT 's3.{region}.amazonaws.com');") slowlogsUrl = f"delta_scan('s3://o11y-prod-shared-{region}/deltalake/{tenant_id}/{o11y_id}/slowlogs')" elif vendor == "gcp": self.db.execute("INSTALL gcs;") self.db.execute("LOAD gcs;") self.db.execute( "CREATE OR REPLACE SECRET secret (TYPE gcs, PROVIDER credential_chain, ENDPOINT 'storage.googleapis.com');") slowlogsUrl = f"delta_scan('gs://o11y-prod-shared-{region}/deltalake/{tenant_id}/{o11y_id}/slowlogs')" elif vendor == "azure": self.db.execute("INSTALL azure;") self.db.execute("LOAD azure;") self.db.execute( "CREATE OR REPLACE SECRET secret (TYPE azure, PROVIDER credential_chain);") slowlogsUrl = f"delta_scan('az://o11y-prod-shared-{region}/deltalake/{tenant_id}/{o11y_id}/slowlogs')" elif vendor == "local": slowlogsUrl = f"delta_scan('{region}')" else: raise ValueError(f"Unsupported vendor: {vendor}") self.db.execute(f""" CREATE VIEW slowlogs AS SELECT * FROM {slowlogsUrl} """) def execute(self, query: str, parameters=None) -> duckdb.DuckDBPyConnection: return self.db.execute(query, parameters=parameters) def count_records_in_time_range(self, start_ts: int, end_ts: int) -> int: """Count the number of records in the specified time range.""" query = """ SELECT COUNT(*) as record_count FROM slowlogs WHERE time >= ? AND time < ? """ try: result = self.execute(query, (start_ts, end_ts)).fetchone() return result[0] if result else 0 except Exception as e: print(f"Error counting records: {e}") return 0 class Rule(ABC): @property @abstractmethod def name(self, data): pass @property @abstractmethod def description(self): pass @abstractmethod def check(self, store: DeltaLakeStore, window: TimeRange, comparison: TimeRange = None) -> any: pass class RuleNewSlowQuery(Rule): @dataclass class Result: rule: str message: str issues_count: int metadata: "RuleNewSlowQuery.MetaData" issues: "List[RuleNewSlowQuery.Issue]" @dataclass class MetaData: analysis_time: float time_window: TimeRange comparison_window: TimeRange @dataclass class Issue: type: str digest: str query: str query_time: int timestamp: float level: str # 'critical' or 'major' @property def name(self): return "new_slow_query" @property def description(self): return "New slow SQL that have not appeared in the past N hours." def check(self, store: DeltaLakeStore, window: TimeRange, comparison: TimeRange = None) -> Result: sql = f""" SELECT first(sl1.time) as time, sl1.digest, avg(sl1.query_time) as query_time, first(sl1.query) as query FROM slowlogs AS sl1 WHERE sl1.time >= {window.start_ts} AND sl1.time < {window.end_ts} AND sl1.query ILIKE 'SELECT%' AND NOT EXISTS ( SELECT sl2.time FROM slowlogs AS sl2 WHERE sl2.digest = sl1.digest AND sl2.time >= {comparison.start_ts} AND sl2.time < {comparison.end_ts} ) GROUP BY sl1.digest ORDER BY avg(sl1.query_time) DESC; """ rows = store.execute(sql).fetchall() issues = [] for row in rows: # Determine level based on query time # > 1 minute is critical level = 'critical' if row[2] > 60 else 'major' issues.append( RuleNewSlowQuery.Issue( type=self.name, digest=row[1], query=row[3], query_time=row[2], timestamp=row[0], level=level ) ) return RuleNewSlowQuery.Result( rule=RuleNewSlowQuery.__name__, message=self.description, issues_count=len(issues), issues=issues, metadata=RuleNewSlowQuery.MetaData( analysis_time=time.time(), time_window=window, comparison_window=comparison, ), ) class RuleHighTimeVariability(Rule): @dataclass class Result: rule: str message: str issues_count: int metadata: "RuleHighTimeVariability.MetaData" issues: "List[RuleHighTimeVariability.Issue]" @dataclass class MetaData: analysis_time: float time_window: TimeRange variability_threshold_pct: float min_stddev_seconds: float @dataclass class Issue: type: str digest: str query: str query_count: int severity: str avg_query_time_seconds: float stddev_seconds: float variability_pct: float level: str # 'critical' or 'major' @property def name(self): return "high_time_variability" @property def description(self): return "The execution time of a slow SQL has high variability in the past N hours." def check(self, store: DeltaLakeStore, window: TimeRange, comparison: TimeRange = None) -> Result: sql = f""" SELECT digest, first(query) AS query, avg(query_time) AS avg_query_time, stddev(query_time) AS stddev_query_time, count(*) AS query_count FROM slowlogs WHERE time >= {window.start_ts} AND time < {window.end_ts} GROUP BY digest HAVING query_count > 1 AND stddev_query_time > 1 AND (stddev_query_time / avg_query_time) > 0.2 ORDER BY (stddev_query_time / avg_query_time) DESC; """ rows = store.execute(sql).fetchall() issues = [] for row in rows: digest = row[0] query = row[1] avg_query_time = row[2] stddev_query_time = row[3] query_count = row[4] variability_pct = (stddev_query_time / avg_query_time) * \ 100 if avg_query_time > 0 else 0 level = 'critical' if ( variability_pct > 200 or avg_query_time > 10) else 'major' issues.append( RuleHighTimeVariability.Issue( type=self.name, digest=digest, query=query, query_count=query_count, severity=self._determine_severity( variability_pct, avg_query_time), avg_query_time_seconds=round(avg_query_time, 4), stddev_seconds=round(stddev_query_time, 4), variability_pct=round(variability_pct, 2), level=level ) ) return RuleHighTimeVariability.Result( rule=RuleHighTimeVariability.__name__, message=self.description, issues_count=len(issues), issues=issues, metadata=RuleHighTimeVariability.MetaData( analysis_time=time.time(), time_window=window, variability_threshold_pct="", min_stddev_seconds="", ), ) @staticmethod def _determine_severity(variability_pct: float, avg_query_time: float) -> str: if variability_pct > 100 or avg_query_time > 10: return "high" elif variability_pct > 50 or avg_query_time > 5: return "medium" return "low" class RuleMultipleExecutionPlans(Rule): @dataclass class Result: rule: str message: str issues_count: int metadata: "RuleMultipleExecutionPlans.MetaData" issues: "List[RuleMultipleExecutionPlans.Issue]" @dataclass class MetaData: analysis_time: float time_window: TimeRange @dataclass class Issue: type: str digest: str query: str plan_count: int avg_query_time: float max_query_time: float min_query_time: float total_query_time: float avg_memory_usage: float max_memory_usage: float plan_digests: List[str] plans: List[str] level: str # 'critical' or 'major' @property def name(self): return "multiple_execution_plans" @property def description(self): return "The execution plan of a slow SQL has changed in the past N hours." def check(self, store: DeltaLakeStore, window: TimeRange, comparison: TimeRange = None) -> Result: sql = f""" SELECT digest, first(query) as query, COUNT(DISTINCT plan_digest) AS plan_count, AVG(query_time) AS avg_query_time, MAX(query_time) AS max_query_time, AVG(mem_max) AS avg_memory_usage, MAX(mem_max) AS max_memory_usage, GROUP_CONCAT(DISTINCT plan_digest) AS plan_digests FROM slowlogs WHERE time >= {window.start_ts} AND time < {window.end_ts} GROUP BY digest HAVING plan_count > 1 ORDER BY plan_count DESC; """ rows = store.execute(sql).fetchall() issues = [] for row in rows: level = 'critical' if (row[2] > 5 or row[4] > 10) else 'major' issues.append( RuleMultipleExecutionPlans.Issue( type=self.name, digest=row[0], query=row[1], plan_count=row[2], avg_query_time=row[3], max_query_time=row[4], avg_memory_usage=row[5], max_memory_usage=row[6], plan_digests=row[7].split(",") if row[7] else [], level=level ) ) return RuleMultipleExecutionPlans.Result( rule=RuleMultipleExecutionPlans.__name__, message=self.description, issues_count=len(issues), issues=issues, metadata=RuleMultipleExecutionPlans.MetaData( analysis_time=time.time(), time_window=window, ), ) class RuleHighMemoryUsage(Rule): @dataclass class Result: rule: str message: str issues_count: int metadata: "RuleHighMemoryUsage.MetaData" issues: "List[RuleHighMemoryUsage.Issue]" @dataclass class MetaData: analysis_time: float time_window: TimeRange @dataclass class Issue: type: str digest: str query: str mem_max: int level: str # 'critical' or 'major' @property def name(self): return "high_memory_usage" @property def description(self): return "SQLs that consume more than 5GB of memory." def check(self, store: DeltaLakeStore, window: TimeRange, comparison: TimeRange = None) -> Result: sql = f""" SELECT digest, any_value(query) AS query, count(*) AS query_count, MAX(mem_max) AS max_memory FROM slowlogs WHERE time >= {window.start_ts} AND time < {window.end_ts} AND mem_max > 5368709120 -- 5GB in bytes (5 * 1024 * 1024 * 1024) GROUP BY digest ORDER BY max_memory DESC; """ rows = store.execute(sql).fetchall() issues = [] for row in rows: # > 10GB is critical level = 'critical' if row[3] > 10737418240 else 'major' issues.append( RuleHighMemoryUsage.Issue( type=self.name, digest=row[0], query=row[1], mem_max=row[3], level=level ) ) return RuleHighMemoryUsage.Result( rule=RuleHighMemoryUsage.__name__, message=f"Found {len(issues)} SQLs using more than 5GB of memory.", issues_count=len(issues), issues=issues, metadata=RuleHighMemoryUsage.MetaData( analysis_time=time.time(), time_window=window, ), ) class RuleMissingIndexes(Rule): @dataclass class Result: rule: str message: str issues_count: int metadata: "RuleMissingIndexes.MetaData" issues: "List[RuleMissingIndexes.Issue]" @dataclass class MetaData: analysis_time: float time_window: TimeRange @dataclass class Issue: type: str digest: str query: str query_time: float stats: str level: str # 'critical' or 'major' @property def name(self): return "missing_indexes" @property def description(self): return "SQLs that might benefit from additional indexes (contain ':missing]' in stats)." def check(self, store: DeltaLakeStore, window: TimeRange, comparison: TimeRange = None) -> Result: sql = f""" SELECT digest, any_value(query) AS query, MAX(query_time) AS max_query_time, any_value(stats) AS stats FROM slowlogs WHERE time >= {window.start_ts} AND time < {window.end_ts} AND stats LIKE '%:missing]%' GROUP BY digest ORDER BY max_query_time DESC; """ rows = store.execute(sql).fetchall() issues = [] for row in rows: # > 10 seconds is critical level = 'critical' if row[2] > 10 else 'major' issues.append( RuleMissingIndexes.Issue( type=self.name, digest=row[0], query=row[1], query_time=row[2], stats=row[3], level=level ) ) return RuleMissingIndexes.Result( rule=RuleMissingIndexes.__name__, message=f"Found {len(issues)} SQLs that might benefit from additional indexes.", issues_count=len(issues), issues=issues, metadata=RuleMissingIndexes.MetaData( analysis_time=time.time(), time_window=window, ), ) class RuleFailedQueries(Rule): @dataclass class Result: rule: str message: str issues_count: int metadata: "RuleFailedQueries.MetaData" issues: "List[RuleFailedQueries.Issue]" @dataclass class MetaData: analysis_time: float time_window: TimeRange @dataclass class Issue: type: str digest: str query: str count: int query_time: float level: str # 'critical' or 'major' @property def name(self): return "failed_queries" @property def description(self): return "SQLs that failed to execute (succ != 1)." def check(self, store: DeltaLakeStore, window: TimeRange, comparison: TimeRange = None) -> Result: sql = f""" SELECT digest, any_value(query) AS query, count(digest) AS count, avg(query_time) AS query_time FROM slowlogs WHERE time >= {window.start_ts} AND time < {window.end_ts} AND succ != 1 GROUP BY digest """ rows = store.execute(sql).fetchall() issues = [] for row in rows: level = 'critical' if row[2] > 10 or row[3] > 10 else 'major' issues.append( RuleFailedQueries.Issue( type=self.name, digest=row[0], query=row[1], count=row[2], query_time=row[3], level=level ) ) return RuleFailedQueries.Result( rule=RuleFailedQueries.__name__, message=f"Found {len(issues)} SQLs that failed to execute.", issues_count=len(issues), issues=issues, metadata=RuleFailedQueries.MetaData( analysis_time=time.time(), time_window=window, ), ) class RuleLargeFullScan(Rule): @dataclass class Result: rule: str message: str issues_count: int metadata: "RuleLargeFullScan.MetaData" issues: "List[RuleLargeFullScan.Issue]" @dataclass class MetaData: analysis_time: float time_window: TimeRange comparison_window: TimeRange = None @dataclass class Issue: type: str digest: str query: str plan_digest: str query_time: float est_rows: float store_type: str level: str # 'critical' or 'major' @property def name(self): return "large_table_full_scan" @property def description(self): return "Detect queries with large full table scans (estRows > 100,000 and not TiFlash)" def _has_large_full_scan(self, plan_node: dict) -> tuple[bool, float, str]: """Check if the execution plan node contains a large full table scan.""" if not isinstance(plan_node, dict): return False, 0, "" name = plan_node.get("name", "") if "TableFullScan_" in name: store_type = plan_node.get("storeType", "") est_rows = plan_node.get("estRows") if est_rows is not None: try: if isinstance(est_rows, str): est_rows = float(est_rows) if est_rows >= 10000 and store_type != "tiflash": level = 'critical' if est_rows >= 100000 else 'major' return True, est_rows, store_type, level except (ValueError, TypeError): pass # Check children recursively for child in plan_node.get("children", []): found, rows, st, level = self._has_large_full_scan(child) if found: return found, rows, st, level return False, 0, "", "" def check(self, store: DeltaLakeStore, window: TimeRange, comparison: TimeRange = None) -> Result: # First, get all slow queries with their plans query = """ SELECT digest, first(query) as query, plan_digest, max(query_time) as query_time, first(decoded_plan) as decoded_plan FROM slowlogs WHERE time >= ? AND time < ? AND decoded_plan IS NOT NULL AND decoded_plan like '%TableFullScan_%' GROUP BY digest, plan_digest """ try: import json rows = store.execute( query, (window.start_ts, window.end_ts)).fetchall() issues = [] for row in rows: plan_str = row[4] # Skip if plan is None or empty if not plan_str: log_info(f"Skipping empty plan for digest: {row[0]}") continue try: # Try to parse the plan JSON plan = json.loads(plan_str) # Skip if plan is not a dictionary if not isinstance(plan, dict): log_info( f"Plan is not a dictionary for digest: {row[0]}") continue plan = plan.get("main", plan) has_scan, est_rows, store_type, level = self._has_large_full_scan( plan) if has_scan: issues.append(RuleLargeFullScan.Issue( type=self.name, digest=row[0], query=row[1], plan_digest=row[2], query_time=row[3], est_rows=est_rows, store_type=store_type, level=level )) except json.JSONDecodeError as e: log_warning( f"Invalid JSON in plan for digest {row[0]}: {str(e)[:100]}") except Exception as e: log_warning( f"Unexpected error processing plan for digest {row[0]}: {str(e)}") return RuleLargeFullScan.Result( rule=RuleLargeFullScan.__name__, message=f"Found {len(issues)} queries with large full table scans (estRows > 100,000 and not TiFlash)", issues_count=len(issues), issues=issues, metadata=RuleLargeFullScan.MetaData( analysis_time=time.time(), time_window=window, comparison_window=comparison ) ) except Exception as e: log_error(f"Error checking for large full scans: {str(e)}") return None class RuleLongRunningQueries(Rule): @dataclass class Result: rule: str message: str issues_count: int metadata: "RuleLongRunningQueries.MetaData" issues: "List[RuleLongRunningQueries.Issue]" @dataclass class MetaData: analysis_time: float time_window: TimeRange @dataclass class Issue: type: str digest: str query: str level: str query_time: float @property def name(self): return "long_running_queries" @property def description(self): return "SQLs that run for more than 60 seconds." def check(self, store: DeltaLakeStore, window: TimeRange, comparison: TimeRange = None) -> Result: sql = f""" SELECT digest, any_value(query) AS query, MAX(query_time) AS max_query_time FROM slowlogs WHERE time >= {window.start_ts} AND time < {window.end_ts} AND query_time > 60 GROUP BY digest ORDER BY max_query_time DESC; """ rows = store.execute(sql).fetchall() issues = [] for row in rows: # Determine level based on query time # > 5 minutes is critical level = 'critical' if row[2] > 300 else 'major' issues.append( RuleLongRunningQueries.Issue( type=self.name, digest=row[0], query=row[1], query_time=row[2], level=level ) ) return RuleLongRunningQueries.Result( rule=RuleLongRunningQueries.__name__, message=f"Found {len(issues)} SQLs running for more than 60 seconds.", issues_count=len(issues), issues=issues, metadata=RuleLongRunningQueries.MetaData( analysis_time=time.time(), time_window=window, ), ) ALL_RULES = [ RuleNewSlowQuery(), RuleHighTimeVariability(), RuleMultipleExecutionPlans(), RuleHighMemoryUsage(), RuleLongRunningQueries(), RuleFailedQueries(), RuleMissingIndexes(), RuleLargeFullScan(), ] def load_custom_rules(directory: str) -> List[Rule]: """ Load custom rule classes from Python modules in the specified directory. Args: directory: Path to the directory containing rule modules Returns: List of instantiated rule objects """ rules = [] directory_path = Path(directory) if not directory_path.exists() or not directory_path.is_dir(): print( f"Warning: Directory {directory} does not exist or is not a directory") return [] # Import all Python files in the directory for file_path in directory_path.glob("*.py"): if file_path.name.startswith('_') or file_path.name.startswith('.'): continue module_name = file_path.stem spec = importlib.util.spec_from_file_location( module_name, str(file_path)) if spec is None or spec.loader is None: continue try: module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) # Find all Rule subclasses in the module for name, obj in module.__dict__.items(): if (isinstance(obj, type) and issubclass(obj, Rule) and obj != Rule and obj.__module__ == module_name): rule_instance = obj() # Add a flag to identify custom rules rule_instance.is_custom = True rules.append(rule_instance) print(f"Loaded custom rule: {name}") except Exception as e: print(f"Error loading module {file_path}: {e}") return rules def list_all_rules(custom_rules_dir: Optional[str] = None): """List all available rules with their names and descriptions.""" # Create a list of rule info dictionaries rules_info = [] def get_rule_info(rule, is_custom=False): # Handle both method and property access for name and description name = rule.name() if callable(rule.name) else rule.name description = rule.description() if callable( rule.description) else rule.description return { "name": name, "description": description, "is_custom": is_custom } # Add built-in rules for rule in ALL_RULES: rules_info.append(get_rule_info(rule, is_custom=False)) # Add custom rules if directory is provided if custom_rules_dir: custom_rules = load_custom_rules(custom_rules_dir) for rule in custom_rules: rules_info.append(get_rule_info(rule, is_custom=True)) return rules_info def check_all_rules(vendor: str, region: str, tenant_id: str, o11y_id: str, window: TimeRange, comparison: TimeRange = None, custom_rules_dir: Optional[str] = None, rules_filter: Optional[str] = None): store = DeltaLakeStore(vendor, region, tenant_id, o11y_id) results = {} # Log the number of records in the time window record_count = store.count_records_in_time_range( window.start_ts, window.end_ts) print( f"\nFound {record_count} records in time range {window.start_ts} to {window.end_ts}") # Get all rules including custom ones rules = ALL_RULES.copy() if custom_rules_dir: custom_rules = load_custom_rules(custom_rules_dir) rules.extend(custom_rules) # Filter rules if rules_filter is provided if rules_filter: rule_names = [name.strip() for name in rules_filter.split(',')] rules = [rule for rule in rules if rule.name in rule_names] if not rules: print(f"Warning: No rules matched the filter: {rules_filter}") return {} print(f"\nRunning {len(rules)} rules:") for i, rule in enumerate(rules, 1): rule_type = "[Custom] " if hasattr( rule, 'is_custom') and rule.is_custom else "[Built-in]" print(f" {i}. {rule_type} {rule.name}") print() # Execute filtered rules for rule in rules: try: print(f"Checking rule: {rule.name}...") result = rule.check(store, window, comparison) if result: # Only add if the rule returned a result results[rule.name] = result print( f" ✓ {rule.name}: Found {len(result.issues) if hasattr(result, 'issues') else 'No'} issues") else: print(f" ✓ {rule.name}: No issues found") except Exception as e: print(f" ✗ Error in {rule.name}: {e}") raise return results def main(): results = check_all_rules( region="us-east-1", tenant_id="1372813089209061633", o11y_id="ed869d81-91c5-4a87-a28d-20795c738d3f", window=TimeRange(start_ts=1749826800, end_ts=1749830400), comparison=TimeRange(start_ts=1749744000, end_ts=1749826800), ) print(results) if __name__ == "__main__": main()