From 9acc08b211e17cc6723402d30f4e883a4458dc8d Mon Sep 17 00:00:00 2001 From: badra001 Date: Fri, 17 Jul 2026 15:10:14 -0400 Subject: [PATCH] add metrics --- .../purge_check_s3_buckets.py | 192 ++++++++++++------ 1 file changed, 131 insertions(+), 61 deletions(-) diff --git a/local-app/python-tools/cross-organization/purge_check_s3_buckets.py b/local-app/python-tools/cross-organization/purge_check_s3_buckets.py index 8776aa70..f61fddec 100755 --- a/local-app/python-tools/cross-organization/purge_check_s3_buckets.py +++ b/local-app/python-tools/cross-organization/purge_check_s3_buckets.py @@ -3,10 +3,11 @@ import sys import re import csv +import json import queue import time import argparse -import traceback +from datetime import datetime, timezone from collections import defaultdict from concurrent.futures import ThreadPoolExecutor @@ -17,7 +18,7 @@ from rich.progress import Progress, BarColumn, TextColumn, TimeElapsedColumn, SpinnerColumn # --- VERSIONING --- -__version__ = "2.0.0" +__version__ = "2.1.0" console = Console() @@ -56,11 +57,20 @@ def build_assumed_role_session(target_account_id, role_name, region_name, profil raise RuntimeError(f"Cross-Account Delegation Failure for role '{role_name}' on account {target_account_id}: {str(e)}") def purge_bucket_worker_task(target_block, role_name, primary_region, dry_run, aws_profile, progress_queue, task_index): - """Worker lifecycle executing object purgation over live queue updates.""" - start_time = time.time() + """Worker lifecycle executing object purgation over live queue updates with fine-grained tracking metrics.""" + time_format = "%Y-%m-%dT%H:%M:%S.%f" + start_epoch = time.time() + start_timestamp_str = datetime.now(timezone.utc).strftime(time_format)[:-3] + "Z" + account_id = target_block["account_id"] + account_name = target_block["alias"] bucket_name = target_block["bucket_name"] + source_size = target_block["source_size"] + source_objects = target_block["source_objects"] + deleted_count = 0 + error_count = 0 + status_message = "INITIALIZED" def update_lane_ui(current_msg, total_weight): if progress_queue: @@ -79,45 +89,62 @@ def update_lane_ui(current_msg, total_weight): if dry_run: update_lane_ui("Evaluating_Size", "100") - # Safe object counting simulation loop total_targets = sum(1 for _ in bucket.object_versions.all()) update_lane_ui(f"Dry_Run_Count_{total_targets}", str(total_targets)) - elapsed = time.time() - start_time - return {"bucket": bucket_name, "status": "DRY_RUN_APPROVED", "time": elapsed, "count": total_targets} + + end_epoch = time.time() + end_timestamp_str = datetime.now(timezone.utc).strftime(time_format)[:-3] + "Z" + return { + "account_id": account_id, "account_name": account_name, "region": primary_region, + "bucket_name": bucket_name, "size_bytes": source_size, "object_count": source_objects, + "start_time": start_timestamp_str, "end_time": end_timestamp_str, + "elapsed_time_seconds": round(end_epoch - start_epoch, 3), "error_count": 0, "status": "DRY_RUN_APPROVED" + } - # Armed Destructive Execution Logic Block update_lane_ui("Listing_Objects", "Evaluating") - - # Pull down versions collection references version_paginator = bucket.object_versions batch = [] - # First phase evaluation count - update_lane_ui("Purging_Objects_0", "100") - for version in version_paginator.all(): - batch.append({"Key": version.object_key, "VersionId": version.id}) - deleted_count += 1 - - if len(batch) == 1000: - bucket.delete_objects(Delete={"Objects": batch, "Quiet": True}) - update_lane_ui(f"Purging_Objects_{deleted_count}", f"{deleted_count + 1000}") - batch = [] + try: + batch.append({"Key": version.object_key, "VersionId": version.id}) + deleted_count += 1 + if len(batch) == 1000: + bucket.delete_objects(Delete={"Objects": batch, "Quiet": True}) + update_lane_ui(f"Purging_Objects_{deleted_count}", f"{deleted_count + 1000}") + batch = [] + except Exception: + error_count += 1 + if batch: - bucket.delete_objects(Delete={"Objects": batch, "Quiet": True}) + try: + bucket.delete_objects(Delete={"Objects": batch, "Quiet": True}) + except Exception: + error_count += 1 - elapsed = time.time() - start_time - update_lane_ui(f"Purged_{deleted_count}", str(deleted_count)) - return {"bucket": bucket_name, "status": "PURGED_SUCCESSFULLY", "time": elapsed, "count": deleted_count} + status_message = "PURGED_SUCCESSFULLY" if error_count == 0 else "PURGED_WITH_ERRORS" except Exception as err: - elapsed = time.time() - start_time + error_count += 1 + status_message = f"FAILED: {str(err)}" update_lane_ui("PURGE_FAILED", "0") - return {"bucket": bucket_name, "status": f"FAILED: {str(err)}", "time": elapsed, "count": deleted_count} + + end_epoch = time.time() + end_timestamp_str = datetime.now(timezone.utc).strftime(time_format)[:-3] + "Z" + + return { + "account_id": account_id, "account_name": account_name, "region": primary_region, + "bucket_name": bucket_name, "size_bytes": source_size, "object_count": source_objects, + "start_time": start_timestamp_str, "end_time": end_timestamp_str, + "elapsed_time_seconds": round(end_epoch - start_epoch, 3), "error_count": error_count, "status": status_message + } def main(): - global_start = time.time() + global_start_epoch = time.time() + time_format = "%Y-%m-%dT%H:%M:%S.%f" + global_start_timestamp = datetime.now(timezone.utc).strftime(time_format)[:-3] + "Z" + parser = argparse.ArgumentParser(description=f"Enterprise Multi-Threaded S3 Purge Engine Framework v{__version__}") parser.add_argument("--csv", required=True, help="Path to generated s3_global_inventory_*.csv asset file") @@ -127,12 +154,12 @@ def main(): parser.add_argument("--workers", type=int, default=4, help="Maximum concurrent worker slots channels") parser.add_argument("--region", default="us-east-1", help="Execution regional fallback context") parser.add_argument("--profile", help="AWS configuration context configuration handle target") + parser.add_argument("--output", default=".", help="Folder destination directory path to save logs (Default: current directory)") parser.add_argument("--execute", action="store_true", help="Arm structural execution destructive deletions") args = parser.parse_args() dry_run = not args.execute - # Compile Regular Expression filters try: bucket_pattern = re.compile(args.filter_bucket, re.IGNORECASE) account_pattern = re.compile(args.filter_account, re.IGNORECASE) if args.filter_account else None @@ -142,38 +169,46 @@ def main(): targets = [] - # Process dataset file if not os.path.exists(args.csv): console.print(f"[bold red]Error: Target inventory asset file path '{args.csv}' does not exist.[/bold red]") sys.exit(1) + # Ingest the inventory database file mapping source metrics properties with open(args.csv, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: b_name = row["Bucket Name"] acct_id = row["Account ID"] acct_alias = row["Account Alias"] + b_region = row.get("Region", args.region) + + # Map clean size and object names regardless of dynamic column labels (B, MB, GB, etc.) + raw_size = 0 + raw_objects = 0 + for k, v in row.items(): + if "size" in k.lower(): + try: raw_size = int(float(v)) + except Exception: pass + if "object" in k.lower(): + try: raw_objects = int(v.replace(",", "")) + except Exception: pass - # Filter out resource bucket profiles if not bucket_pattern.search(b_name): continue - # Account subset layer pattern evaluations if account_pattern: if not (account_pattern.search(acct_id) or account_pattern.search(acct_alias)): continue targets.append({ - "account_id": acct_id, - "alias": acct_alias, - "bucket_name": b_name + "account_id": acct_id, "alias": acct_alias, "bucket_name": b_name, "region": b_region, + "source_size": raw_size, "source_objects": raw_objects }) console.print(f"[bold blue]======================================================================[/bold blue]") console.print(f"Info: Purge Worker Engine Lifecycle v{__version__} Active") console.print(f"[bold blue]======================================================================[/bold blue]") console.print(f"Target Bucket Filter Regex : [cyan]{args.filter_bucket}[/cyan]") - console.print(f"Target Account Filter Regex: [cyan]{args.filter_account if args.filter_account else 'None (All Accounts Eligible)'}[/cyan]") console.print(f"Discovered Filter Targets : [green]{len(targets)} Buckets[/green]") console.print(f"Safe Execution Dry-Run Mode: [white]{dry_run}[/white]") console.print(f"[bold blue]----------------------------------------------------------------------[/bold blue]") @@ -184,11 +219,9 @@ def main(): progress_queue = queue.Queue() account_timers = defaultdict(float) - summary_results = [] + operation_tracking_results = [] - # Dynamic Layout Rendering Interface progress_table = Table.grid(expand=True) - overall_progress = Progress( SpinnerColumn(), TextColumn("[bold yellow]Overall Purge Progress:[/bold yellow]"), @@ -211,9 +244,7 @@ def main(): render_lanes_count = min(args.workers, len(targets)) lane_task_ids = {} for i in range(render_lanes_count): - tid = lane_progress.add_task( - "Awaiting batch assignment", total=100, lane_idx=i, bucket_label="IDLE", current_test="START", total_tests="0" - ) + tid = lane_progress.add_task("Awaiting batch assignment", total=100, lane_idx=i, bucket_label="IDLE", current_test="START", total_tests="0") lane_task_ids[i] = tid progress_table.add_row(overall_progress) @@ -223,12 +254,11 @@ def main(): with Live(progress_table, refresh_per_second=10, console=console): with ThreadPoolExecutor(max_workers=args.workers) as executor: futures_map = {} - for idx, item in enumerate(targets): lane_idx = idx % args.workers fut = executor.submit( purge_bucket_worker_task, - item, args.role_name, args.region, dry_run, args.profile, progress_queue, lane_idx + item, args.role_name, item["region"], dry_run, args.profile, progress_queue, lane_idx ) futures_map[fut] = (item, lane_idx) @@ -246,14 +276,10 @@ def main(): pct = int((current_val / total_w) * 100) if total_w > 0 else 0 if pct > 100: pct = 100 - # Squeeze long bucket designations to make sure text columns do not clip terminal rows b_display = signal["bucket_name"] - if len(b_display) > 30: - b_display = f"...{b_display[-27:]}" + if len(b_display) > 30: b_display = f"...{b_display[-27:]}" - lane_progress.update( - t_id, bucket_label=b_display, completed=pct, current_test=signal["current_test"], total_tests=signal["total_tests"] - ) + lane_progress.update(t_id, bucket_label=b_display, completed=pct, current_test=signal["current_test"], total_tests=signal["total_tests"]) except queue.Empty: break @@ -262,25 +288,69 @@ def main(): meta, assigned_lane = futures_map.pop(fut) try: res = fut.result() - account_timers[meta["alias"]] += res["time"] - summary_results.append(res) + account_timers[meta["alias"]] += res["elapsed_time_seconds"] + operation_tracking_results.append(res) except Exception as fatal_fut_err: - summary_results.append({"bucket": meta["bucket_name"], "status": f"CRITICAL_EXCEPTION: {fatal_fut_err}", "time": 0.0, "count": 0}) + time_now_str = datetime.now(timezone.utc).strftime(time_format)[:-3] + "Z" + operation_tracking_results.append({ + "account_id": meta["account_id"], "account_name": meta["alias"], "region": meta["region"], + "bucket_name": meta["bucket_name"], "size_bytes": meta["source_size"], "object_count": meta["source_objects"], + "start_time": time_now_str, "end_time": time_now_str, "elapsed_time_seconds": 0.0, + "error_count": 1, "status": f"CRITICAL_EXCEPTION: {str(fatal_fut_err)}" + }) overall_progress.update(macro_task_id, advance=1) - lane_progress.update( - lane_task_ids[assigned_lane], bucket_label="IDLE", completed=0, current_test="IDLE", total_tests="0" - ) + lane_progress.update(lane_task_ids[assigned_lane], bucket_label="IDLE", completed=0, current_test="IDLE", total_tests="0") time.sleep(0.05) - global_duration = time.time() - global_start + global_end_epoch = time.time() + global_end_timestamp = datetime.now(timezone.utc).strftime(time_format)[:-3] + "Z" + global_total_elapsed = round(global_end_epoch - global_start_epoch, 3) + + # --- ARTIFACT PERSISTENCE PHASE --- + if args.output != "." and not os.path.exists(args.output): + os.makedirs(args.output, exist_ok=True) + + timestamp_suffix = datetime.now().strftime("%Y%m%dT%H%M%S") + csv_out_path = os.path.join(args.output, f"s3_purge_summary_{timestamp_suffix}.csv") + json_out_path = os.path.join(args.output, f"s3_purge_summary_{timestamp_suffix}.json") + + # 1. Export CSV Log + csv_fields = ["Account ID", "Account Name", "Region", "Bucket Name", "Source Size Bytes", "Source Object Count", "Operation Start", "Operation End", "Elapsed Time (s)", "Error Count", "Status"] + with open(csv_out_path, "w", newline="", encoding="utf-8") as cf: + writer = csv.writer(cf) + writer.writerow(csv_fields) + for r in operation_tracking_results: + writer.writerow([ + r["account_id"], r["account_name"], r["region"], r["bucket_name"], + r["size_bytes"], r["object_count"], r["start_time"], r["end_time"], + r["elapsed_time_seconds"], r["error_count"], r["status"] + ]) + + # 2. Export Master JSON Document with Global Engine Telemetry + json_telemetry_payload = { + "orchestrator_metadata": { + "engine_version": __version__, + "command_args": sys.argv, + "global_start_time": global_start_timestamp, + "global_end_time": global_end_timestamp, + "global_elapsed_seconds": global_total_elapsed, + "dry_run_active": dry_run, + "filters_applied": { + "bucket_regex": args.filter_bucket, + "account_regex": args.filter_account if args.filter_account else "all" + } + }, + "operation_results": operation_tracking_results + } + with open(json_out_path, "w", encoding="utf-8") as jf: + json.dump(json_telemetry_payload, jf, indent=2) + console.print(f"\n[bold blue]======================================================================[/bold blue]") - console.print(f"[*] PURGE SEQUENCE LIFECYCLE CONCLUDED | Total Run Time: {global_duration:.2f}s") + console.print(f"[*] PURGE SEQUENCE LIFECYCLE CONCLUDED | Total Run Time: {global_total_elapsed:.2f}s") console.print(f"[bold blue]======================================================================[/bold blue]") - - # Calculate performance times grouped per target profile sector - for alias_name, duration_seconds in sorted(account_timers.items()): - console.print(f" • Account Sector Group Focus: [white]{alias_name:<25}[/white] | Cumulative Active Operations Duration: [green]{duration_seconds:.2f}s[/green]") + console.print(f"Artifact CSV Export : [bold white]{csv_out_path}[/bold white]") + console.print(f"Artifact JSON Export : [bold white]{json_out_path}[/bold white]") console.print(f"[bold blue]======================================================================[/bold blue]\n") if __name__ == "__main__":