from objstore.objstore import pick_object_store from logs.download import DEFAULT_COMPONENTS from logs.stats import _generate_download_tasks, _execute_downloads, merge_stats from datetime import datetime import textwrap def download_stats_and_compare( cluster_id: str, vendor: str, region: str, components: list, levels: list, start_time_1: str, end_time_1: str, start_time_2: str, end_time_2: str, destination_path: str, ) -> None: """ Downloads and merges log statistics for two specified time periods, then compares them. 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_1: The start time for the first log data period in 'YYYYMMDDHH' format. end_time_1: The end time for the first log data period in 'YYYYMMDDHH' format. start_time_2: The start time for the second log data period in 'YYYYMMDDHH' format. end_time_2: The end time for the second log data period 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}" if not components: components = DEFAULT_COMPONENTS print( f"No components specified, using default components: {components}") # --- Process Period 1 --- print("\n" + "=" * 80) print(f"Processing Period 1: {start_time_1} to {end_time_1}") print("=" * 80) try: start_dt_1 = datetime.strptime(start_time_1, '%Y%m%d%H') end_dt_1 = datetime.strptime(end_time_1, '%Y%m%d%H') except ValueError as e: print( f"Error parsing start_time_1 or end_time_1. Please use 'YYYYMMDDHH' format: {e}") return download_tasks_1 = _generate_download_tasks( start_dt_1, end_dt_1, store_prefix, local_base_dir) if download_tasks_1: _execute_downloads(obj_store, bucket_name, download_tasks_1) else: print("No log files to download for Period 1.") period_a_data = {} # Initialize empty if no files local_files_1 = [task["local_file"] for task in download_tasks_1] period_a_data = merge_stats( log_stats_file_paths=local_files_1, levels=levels, components=components) # --- Process Period 2 --- print("\n" + "=" * 80) print(f"Processing Period 2: {start_time_2} to {end_time_2}") print("=" * 80) try: start_dt_2 = datetime.strptime(start_time_2, '%Y%m%d%H') end_dt_2 = datetime.strptime(end_time_2, '%Y%m%d%H') except ValueError as e: print( f"Error parsing start_time_2 or end_time_2. Please use 'YYYYMMDDHH' format: {e}") return download_tasks_2 = _generate_download_tasks( start_dt_2, end_dt_2, store_prefix, local_base_dir) if download_tasks_2: _execute_downloads(obj_store, bucket_name, download_tasks_2) else: print("No log files to download for Period 2.") period_b_data = {} # Initialize empty if no files local_files_2 = [task["local_file"] for task in download_tasks_2] period_b_data = merge_stats( log_stats_file_paths=local_files_2, levels=levels, components=components) # --- Compare the two periods --- compare_log_stats(period_a_data, period_b_data) def compare_log_stats( period_a_data: dict, period_b_data: dict, comparison_threshold: int = 5 ) -> None: """ Compares two sets of structured log statistics, highlighting: 1. Log entries present in period_b but not in period_a. 2. Log entries where the count in period_b is significantly higher than in period_a. Args: period_a_data: Structured log data for the first time period. period_b_data: Structured log data for the second time period. comparison_threshold: Multiplier to consider a count "significantly higher". E.g., if 5, count in period_b must be > 5 * count in period_a. """ print("\n" + "#" * 80) print(" LOG STATISTICS COMPARISON") print("#" * 80) # Flatten the structured data for easier comparison flat_period_a = _flatten_structured_data(period_a_data) flat_period_b = _flatten_structured_data(period_b_data) new_entries = {} increased_count_entries = {} # Identify new entries in period_b for key, entry_b in flat_period_b.items(): if key not in flat_period_a: _add_to_nested_dict(new_entries, entry_b) else: # Identify significantly increased counts entry_a = flat_period_a[key] if entry_b['count'] > entry_a['count'] * comparison_threshold: _add_to_nested_dict(increased_count_entries, { "component": entry_b["component"], "level": entry_b["level"], "file": entry_b["file"], "line": entry_b["line"], "message": entry_b["message"], "count_period_a": entry_a["count"], "count_period_b": entry_b["count"], "sample": entry_b["sample"] }) if not new_entries and not increased_count_entries: print("\nNo significant differences found between the two periods.") return _print_comparison_section( new_entries, "NEW LOG ENTRIES (Present in Period B, Not in Period A)") _print_comparison_section( increased_count_entries, f"SIGNIFICANTLY INCREASED LOG COUNTS (Count in Period B > {comparison_threshold}x Period A)") def _flatten_structured_data(structured_data: dict) -> dict: """Helper to flatten the structured data for easier key-based comparison.""" flat_data = {} for component, levels_data in structured_data.items(): for level, entries in levels_data.items(): for entry in entries: # Create a unique key (component, level, file, message digest for message based comparison) # Using msg_digest for more robust message comparison key = (entry["component"], entry["level"], entry["file"], entry["line"], entry["message"]) flat_data[key] = entry return flat_data def _add_to_nested_dict(target_dict: dict, entry: dict): """Helper to add an entry to a nested dictionary by component and level.""" component = entry["component"] level = entry["level"] target_dict.setdefault(component, {}).setdefault(level, []).append(entry) def _print_comparison_section(data: dict, title: str) -> None: """Helper to print a section of the comparison report.""" log_level_order = {"ERROR": 4, "WARN": 3, "INFO": 2, "DEBUG": 1} if not data: return print(f"\n### {title}") print("-" * (len(title) + 6)) # Adjust underline length for component in sorted(data.keys()): print(f"\n#### Component: {component}") component_data = data[component] 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 entries by count (for increased counts) or message (for new entries) if "count_period_b" in log_entries[0]: # For increased counts sorted_entries = sorted( log_entries, key=lambda x: x["count_period_b"], reverse=True) else: # For new entries sorted_entries = sorted( log_entries, key=lambda x: x["count"], reverse=True) print(f"\n ##### Level: {level.upper()}") # Define headers dynamically based on the type of comparison if "count_period_a" in sorted_entries[0]: headers = ["COUNT (A)", "COUNT (B)", "MESSAGE", "LOCATION", "SAMPLE"] else: headers = ["COUNT", "MESSAGE", "LOCATION", "SAMPLE"] col_widths = {header: len(header) for header in headers} # Pre-process entries to determine maximum widths processed_entries = [] for entry in sorted_entries: message_display = textwrap.shorten( entry['message'], width=60, placeholder="...") location_full = f"{entry['file']}:{entry['line']}" sample_display = textwrap.shorten( entry['sample'] if entry['sample'] else "", width=40, placeholder="...") processed_entry = { "MESSAGE_DISPLAY": message_display, "LOCATION_DISPLAY": location_full, "SAMPLE_DISPLAY": sample_display, "RAW_SAMPLE": entry['sample'] if entry['sample'] else "" } if "count_period_a" in entry: processed_entry["COUNT (A)"] = str(entry["count_period_a"]) processed_entry["COUNT (B)"] = str(entry["count_period_b"]) col_widths["COUNT (A)"] = max( col_widths["COUNT (A)"], len(processed_entry["COUNT (A)"])) col_widths["COUNT (B)"] = max( col_widths["COUNT (B)"], len(processed_entry["COUNT (B)"])) else: processed_entry["COUNT"] = str(entry["count"]) col_widths["COUNT"] = max( col_widths["COUNT"], len(processed_entry["COUNT"])) 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)) processed_entries.append(processed_entry) # 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) - 4)) print(header_line) print(" " + "-" * (len(header_line) - 4)) # Print table rows for row_data in processed_entries: row_parts = [] if "COUNT (A)" in row_data: row_parts.append( f"{row_data['COUNT (A)']:<{col_widths['COUNT (A)']}}") row_parts.append( f"{row_data['COUNT (B)']:<{col_widths['COUNT (B)']}}") else: row_parts.append( f"{row_data['COUNT']:<{col_widths['COUNT']}}") row_parts.append( f"{row_data['MESSAGE_DISPLAY']:<{col_widths['MESSAGE']}}") row_parts.append( f"{row_data['LOCATION_DISPLAY']:<{col_widths['LOCATION']}}") row_parts.append( f"{row_data['SAMPLE_DISPLAY']:<{col_widths['SAMPLE']}}") row_line = f" | {' | '.join(row_parts)} |" print(row_line) 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'], width=len(header_line) - 8, initial_indent=" ", subsequent_indent=" " ) print(wrapped_sample) print(" " + "-" * (len(header_line) - 4))