scripts: begining of the log filter preparation

Author: lev-starkware

scripts/consensus_report/generate_logs_and_report.py

@@ -5,6 +5,8 @@
 import os
 import sys
 from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Optional, Tuple
 
 import yaml
 
@@ -43,6 +45,132 @@ def load_env_map() -> dict[str, EnvConfig]:
     return env_map
 
 
+# ------------------------------
+# Time utilities
+# ------------------------------
+
+
+def fmt_utc(dt: datetime) -> str:
+    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def get_24_hours_window() -> Tuple[datetime, datetime]:
+    now = datetime.now(timezone.utc)
+    start = now - timedelta(hours=24)
+    return start, now
+
+
+# ------------------------------
+# Filter builders
+# ------------------------------
+
+
+def common_prefix(ns_re: str) -> str:
+    return (
+        'resource.type="k8s_container" '
+        f'AND resource.labels.namespace_name=~"{ns_re}" '
+        'AND (logName:"/logs/stdout" OR logName:"/logs/stderr")'
+    )
+
+
+def consensus_height_filter(common: str, height: int) -> str:
+    return f'{common} AND jsonPayload.message:"Running consensus for height {height}"'
+
+
+def wide_search_filter(common: str, height: int) -> str:
+    return f"""{common} AND (
+      (
+        -- Consensus logs (keyed by height)
+        (
+          jsonPayload.spans.height="{height}"
+          OR jsonPayload.message:"{height}"
+          OR textPayload:"{height}"
+        )
+        AND
+        resource.labels.container_name="sequencer-core"
+        AND
+        (
+          jsonPayload.message=~"^START_ROUND_(PROPOSER|VALIDATOR):"
+          OR jsonPayload.message:"DECISION_REACHED"
+          OR jsonPayload.message:"PROPOSAL_FAILED"
+          OR jsonPayload.message=~"(?i)prevote|precommit|propose"
+
+        )
+      )
+      OR
+      (
+        -- Batcher logs (keyed by propose_block_input)
+        jsonPayload.filename:"apollo_batcher" AND "BlockNumber({height})"
+        AND
+        (
+          jsonPayload.message:"finishing block building"
+          OR jsonPayload.message:"Received final number of transactions in block proposal:"
+          OR jsonPayload.message:"Finished building block as proposer"
+        )
+      )
+      OR
+      (
+        -- Blockifier logs
+        jsonPayload.filename:"blockifier"
+        AND
+        jsonPayload.message:"Block {height} final weights"
+      )
+    )"""
+
+
+def add_time_bounds(flt: str, start: datetime, end: datetime) -> str:
+    return f'{flt} AND timestamp>="{fmt_utc(start)}" AND timestamp<"{fmt_utc(end)}"'
+
+
+# ------------------------------
+# Timestamp discovery + windowing
+# ------------------------------
+
+
+def compute_window(
+    args: argparse.Namespace,
+    environment: Optional[EnvConfig] = None,
+    common_filter_prefix: Optional[str] = None,
+) -> Tuple[datetime, datetime]:
+    """Compute the time window based on provided arguments.
+
+    Priority/validation:
+      - --auto, --near, --start/--end, --last-24-hours are mutually exclusive
+      - --start/--end must be provided together
+      - --last-24-hours (or no args) uses (current_time - 24 hours) to current_time window
+      - --auto requires environment and common_filter_prefix parameters
+    """
+
+    # TODO(lev): Implement all the logic for the different time options
+
+    # Default: last 24 hours window
+    return get_24_hours_window()
+
+
+def prepare_filter(args, environment) -> Tuple[str, str, str]:
+    """Prepare log filter and time bounds. Returns (log_filter, start_time, end_time)."""
+    common_filter_prefix = common_prefix(environment.namespace_re)
+    wide_filter = wide_search_filter(common_filter_prefix, args.height)
+    start_time, end_time = compute_window(args, environment, common_filter_prefix)
+    log_filter = add_time_bounds(wide_filter, start_time, end_time)
+
+    if args.print_filters:
+        print(
+            "START_MARKER_FILTER:\n"
+            + consensus_height_filter(common_filter_prefix, args.height)
+            + "\n"
+        )
+        print(
+            "END_MARKER_FILTER:\n"
+            + consensus_height_filter(common_filter_prefix, args.height + 1)
+            + "\n"
+        )
+        print("FINAL_FILTER:\n" + log_filter + "\n")
+        raise SystemExit(0)
+
+    return log_filter, start_time, end_time
+
+
 # ------------------------------
 # Main
 # ------------------------------
@@ -94,9 +222,14 @@ def main() -> int:
     env_map = load_env_map()
     args = get_args(env_map)
 
-    env_map[args.env]
+    environment = env_map[args.env]
+
+    try:
+        log_filter, start_time, end_time = prepare_filter(args, environment)
+    except Exception as e:
+        print(f"Error: {e}", file=sys.stderr)
+        return 2
 
-    # TODO(lev): Add filter preparation
     # TODO(lev): Add log downloading
     # TODO(lev): Add report generation