add metrics

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__":