from objstore.objstore import pick_object_store from logs.download import DEFAULT_COMPONENTS, _download_single_file from datetime import datetime, timedelta from concurrent.futures import ThreadPoolExecutor, as_completed import textwrap import sqlite3 def download_stats_and_merge( cluster_id: str, vendor: str, region: str, components: list, levels: list, start_time: str, end_time: str, destination_path: str, ) -> None: """ Downloads log statistics files from object storage and merges them into a structured format. Args: cluster_id: The ID of the cluster. vendor: The object storage vendor (e.g., 's3', 'gcs'). region: The geographical region of the object storage bucket. components: A list of log components to filter by. If empty, all components are used. levels: A list of log levels to filter by. start_time: The start time for log data in 'YYYYMMDDHH' format. end_time: The end time for log data in 'YYYYMMDDHH' format. destination_path: The base local path to save downloaded files. """ obj_store = pick_object_store(vendor) bucket_name = f"o11y-prod-shared-{region}" store_prefix = f"diagnosis/data/{cluster_id}" local_base_dir = f"{destination_path}/stats/{cluster_id}" # Use default components if none are specified. if not components: components = DEFAULT_COMPONENTS print( f"No components specified, using default components: {components}") try: start_dt = datetime.strptime(start_time, '%Y%m%d%H') end_dt = datetime.strptime(end_time, '%Y%m%d%H') except ValueError as e: print( f"Error parsing start_time or end_time. Please use 'YYYYMMDDHH' format: {e}") return download_tasks = _generate_download_tasks( start_dt, end_dt, store_prefix, local_base_dir) if download_tasks: _execute_downloads(obj_store, bucket_name, download_tasks) else: print("No log files to download for the specified criteria.") local_files = [task["local_file"] for task in download_tasks] merged_data = merge_stats( log_stats_file_paths=local_files, levels=levels, components=components) print_formatted_log_stats(merged_data) def _generate_download_tasks( start_dt: datetime, end_dt: datetime, store_prefix: str, local_base_dir: str ) -> list: """ Generates a list of download tasks, each containing an S3 key and local file path. """ tasks = [] current_dt = start_dt while current_dt <= end_dt: current_hour_str = current_dt.strftime('%Y%m%d%H') s3_key = f"{store_prefix}/log-stats/1h/{current_hour_str}.db" # Corrected this line to use local_base_dir local_file = f"{local_base_dir}/{current_hour_str}.db" tasks.append({"s3_key": s3_key, "local_file": local_file}) current_dt += timedelta(hours=1) return tasks def _execute_downloads(obj_store: str, bucket_name: str, download_tasks: list) -> None: """ Executes file downloads concurrently using a thread pool. """ max_workers = min(len(download_tasks), 30) print( f"Initiating download of {len(download_tasks)} log files with {max_workers} concurrent threads.") with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [ executor.submit( _download_single_file, obj_store, bucket_name, task["s3_key"], task["local_file"], ) for task in download_tasks ] for future in as_completed(futures): try: success = future.result() if not success: print("One or more downloads failed.") except Exception as exc: print(f"A download task generated an exception: {exc}") def merge_stats(log_stats_file_paths: list, levels: list, components: list) -> dict: """ Merges log statistics from multiple SQLite database files into a single structured dictionary. Args: log_stats_file_paths: A list of paths to the SQLite database files. levels: A list of log levels to filter the merged data. components: A list of log components to filter the merged data. Returns: A dictionary containing the merged and structured log statistics. """ merged_data = {} for file_path in log_stats_file_paths: try: with sqlite3.connect(f'file:{file_path}?mode=ro', uri=True) as src_conn: src_cur = src_conn.cursor() src_cur.execute( 'SELECT component, level, position, msg_digest, count, sample FROM log_stats') for row in src_cur: component, level, position, message, count, sample = row if levels and level not in levels: continue if components and component not in components: continue # Use a tuple as key for uniqueness (component, level, position) key = (component, level, position) if key in merged_data: # Aggregate the 'count' for existing entries current_component, current_level, current_position, \ current_message, current_count, current_sample = merged_data[key] merged_data[key] = ( current_component, current_level, current_position, current_message, # Assuming message is consistent for same key current_count + count, current_sample, # Assuming sample is consistent or last one is fine ) else: merged_data[key] = row except sqlite3.Error as e: print(f"Skip processing {file_path} due to SQLite error: {e}") continue return _structure_merged_data(merged_data) def _structure_merged_data(merged_raw_data: dict) -> dict: """ Restructures the raw merged data into a nested dictionary by component and level. """ structured_data = {} for row in merged_raw_data.values(): component, level, position, message, count, sample = row position_parts = position.rsplit(':', 1) file = position_parts[0] line = position_parts[1] if len(position_parts) > 1 else None entry = { "component": component, "level": level, "file": file, "line": line, "message": message, "count": count, "sample": sample, } structured_data.setdefault(component, {}).setdefault( level, []).append(entry) return structured_data def print_formatted_log_stats(structured_data: dict) -> None: """ Formats and prints the structured log statistics to the console in a table format, prioritizing high-occurrence logs and organizing by component and level. Args: structured_data: A dictionary containing the structured log statistics. """ # Define the desired order of log levels log_level_order = {"ERROR": 4, "WARN": 3, "INFO": 2, "DEBUG": 1} if not structured_data: print("\nNo log statistics available for the specified criteria.") return print("\n" + "=" * 80) print(" LOG STATISTICS SUMMARY") print("=" * 80) # Sort components alphabetically for consistent output for component in sorted(structured_data.keys()): print(f"\n### Component: {component}") print("-" * (len(component) + 14)) # Adjust underline length component_data = structured_data[component] # Sort levels based on the defined order (ERROR, WARN, INFO, DEBUG) sorted_levels = sorted( component_data.keys(), key=lambda x: log_level_order.get(x.upper(), 0), reverse=True ) for level in sorted_levels: log_entries = component_data[level] # Sort log entries by count in descending order sorted_entries = sorted( log_entries, key=lambda x: x["count"], reverse=True) if not sorted_entries: print(f" No {level.upper()} logs found for this component.") continue print(f"\n #### Level: {level.upper()}") # Define table headers headers = ["COUNT", "MESSAGE", "LOCATION", "SAMPLE"] # Initialize column widths with header lengths col_widths = {header: len(header) for header in headers} # Pre-process entries to determine actual content and maximum widths processed_entries = [] limit = 1000 # Display top N entries for i, entry in enumerate(sorted_entries): if i >= limit: break # Only process up to the limit count_str = str(entry['count']) location_full = f"{entry['file']}:{entry['line']}" # Apply initial truncation limits for MESSAGE and SAMPLE # These are soft limits for display, full content can be printed later message_display = textwrap.shorten( entry['message'], width=60, placeholder="...") sample_display = textwrap.shorten( entry['sample'] if entry['sample'] else "", width=40, placeholder="...") processed_entry = { "COUNT": count_str, "MESSAGE_DISPLAY": message_display, "LOCATION_DISPLAY": location_full, "SAMPLE_DISPLAY": sample_display, # Keep full sample "RAW_SAMPLE": entry['sample'] if entry['sample'] else "" } processed_entries.append(processed_entry) # Update column widths based on the displayed (potentially truncated) content col_widths["COUNT"] = max(col_widths["COUNT"], len(count_str)) col_widths["MESSAGE"] = max( col_widths["MESSAGE"], len(message_display)) col_widths["LOCATION"] = max( col_widths["LOCATION"], len(location_full)) col_widths["SAMPLE"] = max( col_widths["SAMPLE"], len(sample_display)) # If there are more entries than the limit, add a "truncated" row if len(sorted_entries) > limit: truncated_row = { "COUNT": "...", "MESSAGE_DISPLAY": "... (more entries)", "LOCATION_DISPLAY": "...", "SAMPLE_DISPLAY": "...", "RAW_SAMPLE": "" } processed_entries.append(truncated_row) # Ensure column widths accommodate the "truncated" indicators col_widths["COUNT"] = max( col_widths["COUNT"], len(truncated_row["COUNT"])) col_widths["MESSAGE"] = max( col_widths["MESSAGE"], len(truncated_row["MESSAGE_DISPLAY"])) col_widths["LOCATION"] = max( col_widths["LOCATION"], len(truncated_row["LOCATION_DISPLAY"])) col_widths["SAMPLE"] = max( col_widths["SAMPLE"], len(truncated_row["SAMPLE_DISPLAY"])) # Print table header header_line_parts = [f"{h:<{col_widths[h]}}" for h in headers] header_line = f" | {' | '.join(header_line_parts)} |" print(" " + "-" * (len(header_line) - 2)) print(header_line) print(" " + "-" * (len(header_line) - 2)) # Print table rows for row_data in processed_entries: row_line = ( f" | {row_data['COUNT']:<{col_widths['COUNT']}} " f"| {row_data['MESSAGE_DISPLAY']:<{col_widths['MESSAGE']}} " f"| {row_data['LOCATION_DISPLAY']:<{col_widths['LOCATION']}} " f"| {row_data['SAMPLE_DISPLAY']:<{col_widths['SAMPLE']}} |" ) print(row_line) # Print full raw sample if it was originally longer than its display limit if row_data['RAW_SAMPLE'] and len(row_data['RAW_SAMPLE']) > col_widths["SAMPLE"]: print(" Full Sample:") wrapped_sample = textwrap.fill( row_data['RAW_SAMPLE'], # Roughly match table width, subtract borders and padding width=len(header_line) - 6, initial_indent=" ", # Deeper indent for clarity subsequent_indent=" " ) print(wrapped_sample) # Add separator after each entry (including full sample if printed) print(" " + "-" * (len(header_line) - 2)) print("\n" + "=" * 80)