Merge pull request #2113 from leggedrobotics/feat/improve-action-log-following

Feat/improve action log following

Author: wp99cp

backend/src/services/action.service.ts

@@ -207,7 +207,7 @@ export class ActionService {
             (action.executionEndedAt?.getTime() ?? Date.now()) * 1_000_000 +
             60_000_000_000;
 
-        let logQl = `{job="queue-consumer", action_run_uuid="${actionUuid}"}`;
+        let logQl = `{job="queue-consumer"} | json action_run_uuid="labels.action_run_uuid" | action_run_uuid="${actionUuid}"`;
         if (query.search) {
             logQl += ` |= "${query.search}"`;
         }
@@ -285,7 +285,6 @@ export class ActionService {
                             message: string;
                             type?: LogType;
                         };
-
                         return {
                             timestamp:
                                 validParsed.timestamp ??
@@ -294,6 +293,7 @@ export class ActionService {
                                 ).toISOString(),
                             message: validParsed.message,
                             type: validParsed.type ?? ('stdout' as LogType),
+                            _tsNs: tsNs,
                         };
                     } catch {
                         return {
@@ -305,17 +305,43 @@ export class ActionService {
                                     ? lineJson
                                     : JSON.stringify(lineJson),
                             type: 'stdout' as LogType,
+                            _tsNs: tsNs,
                         };
                     }
                 }),
             );
 
-            // Sort by timestamp (Loki might return multiple streams interleaved)
-            allLogs.sort(
-                (a, b) =>
-                    new Date(a.timestamp).getTime() -
-                    new Date(b.timestamp).getTime(),
-            );
+            // First strictly compare exact ISO strings (Docker log timestamp has nanosec encoding format).
+            // Then fallback to tsNs Loki timestamps.
+            allLogs.sort((a, b) => {
+                const cmp = a.timestamp.localeCompare(b.timestamp);
+                if (cmp !== 0) {
+                    return cmp;
+                }
+
+                const tsNsA =
+                    '_tsNs' in a && typeof a._tsNs === 'string'
+                        ? BigInt(a._tsNs)
+                        : BigInt(new Date(a.timestamp).getTime()) * 1_000_000n;
+                const tsNsB =
+                    '_tsNs' in b && typeof b._tsNs === 'string'
+                        ? BigInt(b._tsNs)
+                        : BigInt(new Date(b.timestamp).getTime()) * 1_000_000n;
+
+                if (tsNsA < tsNsB) {
+                    return -1;
+                }
+                if (tsNsA > tsNsB) {
+                    return 1;
+                }
+                return 0;
+            });
+
+            for (const log of allLogs) {
+                if ('_tsNs' in log) {
+                    delete (log as Record<string, unknown>)._tsNs;
+                }
+            }
 
             const count = allLogs.length;
             const data = allLogs.slice(query.skip, query.skip + query.take);

cli/kleinkram/api/routes.py

@@ -202,7 +202,16 @@ def get_run(
     if resp.status_code == 404:
         raise kleinkram.errors.RunNotFound(f"Run not found: {run_id}")
     resp.raise_for_status()
-    return _parse_run(RunObject(resp.json()))
+    run_object = resp.json()
+
+    try:
+        logs_resp = client.get(f"{ACTION_ENDPOINT}s/{run_id}/logs")
+        if logs_resp.status_code == 200:
+            run_object["logs"] = logs_resp.json().get("data", [])
+    except Exception:
+        pass
+
+    return _parse_run(RunObject(run_object))
 
 
 def get_action_templates(

cli/kleinkram/cli/app.py

@@ -219,7 +219,7 @@ def check_version_compatibility() -> None:
 
 @app.callback()
 def cli(
-    verbose: bool = typer.Option(True, help="Enable verbose mode."),
+    verbose: bool = typer.Option(False, help="Enable verbose mode."),
     debug: bool = typer.Option(False, help="Enable debug mode."),
     version: Optional[bool] = typer.Option(None, "--version", "-v", callback=_version_callback),
     log_level: Optional[LogLevel] = typer.Option(None, help="Set log level."),

cli/kleinkram/config.py

@@ -245,7 +245,7 @@ def endpoint_table(config: Config) -> Table:
 @dataclass
 class SharedState:
     log_file: Optional[Path] = None
-    verbose: bool = True
+    verbose: bool = False
     debug: bool = False
     max_table_size: int = MAX_TABLE_SIZE
 

cli/kleinkram/printing.py

@@ -5,6 +5,7 @@
 import time
 from dataclasses import asdict
 from datetime import datetime
+from datetime import timezone
 from pathlib import Path
 from typing import List
 from typing import Mapping
@@ -17,6 +18,10 @@
 import httpx
 import typer
 from rich.console import Console
+from rich.console import Group
+from rich.live import Live
+from rich.panel import Panel
+from rich.spinner import Spinner
 from rich.table import Table
 from rich.text import Text
 
@@ -539,6 +544,71 @@ def print_action_templates_table(templates: Sequence[ActionTemplate], *, pprint:
             print(template.uuid)
 
 
+def generate_live_layout(run_details: Run) -> Group:
+    verbose = get_shared_state().verbose
+
+    # Calculate elapsed time
+    if run_details.updated_at:
+        elapsed = run_details.updated_at - run_details.created_at
+    else:
+        elapsed = datetime.now(timezone.utc) - run_details.created_at
+
+    elapsed_seconds = int(elapsed.total_seconds())
+    hours, remainder = divmod(elapsed_seconds, 3600)
+    minutes, seconds = divmod(remainder, 60)
+    elapsed_str = f"{hours}:{minutes:02}:{seconds:02}"
+
+    state_upper = run_details.state.upper()
+
+    if state_upper == "DONE":
+        status_color = "green"
+        icon = Text("[✔]", style="green")
+        state_text = "DONE"
+    elif state_upper in {"FAILED", "UNPROCESSABLE"}:
+        status_color = "red"
+        icon = Text("[✘]", style="red")
+        state_text = f"{run_details.state}"
+    else:
+        status_color = "yellow"
+        icon = Spinner("dots", style="blue")
+        state_text = f"{run_details.state}"
+
+    header_text = Text()
+    header_text.append(f"Running Action {run_details.uuid} ", style="bold")
+    header_text.append(f"{elapsed_str} ", style="dim")
+    header_text.append(f"({state_text})", style=status_color)
+
+    header_table = Table.grid(padding=(0, 1))
+    header_table.add_column()
+    header_table.add_column()
+    header_table.add_row(icon, header_text)
+
+    logs_text = Text()
+    last_logs = run_details.logs[-20:]
+    if not last_logs:
+        logs_text.append(" => ", style="blue")
+        if state_upper in {"DONE", "FAILED", "UNPROCESSABLE"}:
+            logs_text.append("No logs produced.", style="dim")
+        else:
+            logs_text.append("Waiting for logs...", style="dim")
+    else:
+        for i, log in enumerate(last_logs):
+            logs_text.append(" => ", style="blue")
+
+            if verbose:
+                level = log.level.upper().strip()
+                color = LOG_LEVEL_COLORS.get(level, "white")
+                timestamp_str = log.timestamp.strftime("%Y-%m-%d %H:%M:%S")
+                level_str = f"[{level.ljust(8)}]"
+                logs_text.append(f"[{timestamp_str}] {level_str} ", style=str(color))
+
+            logs_text.append(log.message.strip())
+            if i < len(last_logs) - 1:
+                logs_text.append("\n")
+
+    return Group(header_table, logs_text)
+
+
 def follow_run_logs(client: AuthenticatedClient, run_uuid: str) -> int:
     """
     Polls the API for run details and prints new logs as they arrive.
@@ -549,40 +619,67 @@ def follow_run_logs(client: AuthenticatedClient, run_uuid: str) -> int:
     typer.echo(f"Following logs for run {run_uuid}...")
 
     TERMINAL_STATES = {"DONE", "FAILED", "UNPROCESSABLE"}
-    printed_log_count = 0
     current_run_state = None
     exit_code = 0  # Assume success
 
     try:
-        while current_run_state not in TERMINAL_STATES:
-            try:
-                run_details: Run = kleinkram.api.routes.get_run(client, run_uuid)
-                current_run_state = run_details.state.upper()
-
-                # Print only new logs
-                new_logs = run_details.logs[printed_log_count:]
-                if new_logs:
-                    # Always pretty-print when following
-                    print_run_logs(new_logs, pprint=True)
-                    printed_log_count = len(run_details.logs)
-
-                if current_run_state in TERMINAL_STATES:
-                    color = typer.colors.GREEN if run_details.state.upper() == "DONE" else typer.colors.RED
-                    typer.secho(
-                        f"\nRun finished with state: {run_details.state} ({run_details.state_cause})",
-                        fg=color,
-                    )
-                    if run_details.state.upper() != "DONE":
-                        exit_code = 1  # Set failure exit code
+        with Live(refresh_per_second=10) as live:
+            while current_run_state not in TERMINAL_STATES:
+                try:
+                    run_details: Run = kleinkram.api.routes.get_run(client, run_uuid)
+                    current_run_state = run_details.state.upper()
+
+                    live.update(generate_live_layout(run_details), refresh=True)
+
+                    if current_run_state in TERMINAL_STATES:
+                        if current_run_state != "DONE":
+                            exit_code = 1  # Set failure exit code
+                        break
+
+                    time.sleep(1)  # Poll every 1 seconds for a more responsive timer
+
+                except kleinkram.errors.RunNotFound:
+                    time.sleep(1)
+                except httpx.HTTPStatusError:
+                    time.sleep(3)  # Wait longer on API errors
+                except httpx.RequestError as e:
+                    # Catch raw socket errors (like Connection reset by peer)
+                    err_msg = Text(f"\nConnection lost. Please check your network and try again. ({e})", style="red")
+                    live.console.print(err_msg)
+                    time.sleep(3)
+
+        # After the Live block finishes, print the final state clearly
+        try:
+            run_details: Run = kleinkram.api.routes.get_run(client, run_uuid)
+            state_upper = run_details.state.upper()
+
+            has_warnings = False
+            for log in run_details.logs:
+                if "CORRUPTED" in log.message.upper() or log.level.upper() in {"WARN", "WARNING", "ERROR"}:
+                    has_warnings = True
                     break
 
-                time.sleep(2)  # Poll every 2 seconds
-
-            except kleinkram.errors.RunNotFound:
-                time.sleep(1)
-            except httpx.HTTPStatusError as e:
-                typer.secho(f"Error fetching run status: {e}", fg=typer.colors.RED)
-                time.sleep(5)  # Wait longer on API errors
+            if state_upper == "DONE" and has_warnings:
+                color = typer.colors.YELLOW
+                state_display = "DONE (Completed with Warnings)"
+            elif state_upper == "DONE":
+                color = typer.colors.GREEN
+                state_display = "DONE"
+            else:
+                color = typer.colors.RED
+                state_display = run_details.state
+
+            state_cause_str = f" ({run_details.state_cause})" if run_details.state_cause else ""
+            typer.secho(
+                f"\nRun finished with state: {state_display}{state_cause_str}",
+                fg=color,
+            )
+        except httpx.HTTPStatusError as e:
+            typer.secho(f"\nFailed to fetch final run details (API error): {e}", fg=typer.colors.RED)
+        except httpx.RequestError as e:
+            typer.secho(f"\nFailed to fetch final run details (network error): {e}", fg=typer.colors.RED)
+        except Exception as e:
+            typer.secho(f"\nFailed to fetch final run details: {e}", fg=typer.colors.RED)
 
     except KeyboardInterrupt:
         typer.secho(

cli/setup.cfg

@@ -38,6 +38,7 @@ warn_unused_configs = True
 exclude = (setup.py|build/|tests/)
 
 [tool:pytest]
+norecursedirs = test-venv .venv venv build dist
 python_files = tests/*.py tests/**/*.py tests.py test_*.py *_tests.py
 markers =
 	slow: marks tests as slow (deselect with '-m "not slow"')

observability/loki/loki-config.yml

@@ -45,11 +45,11 @@ limits_config:
     max_entries_limit_per_query: 50000
     retention_period: 720h # Default 30 days
     retention_stream:
-        - selector: '{action_run_uuid=~".+"}'
+        - selector: '{job="queue-consumer"}'
           priority: 1
           period: 2160h # 90 days for action logs
 
 compactor:
     working_directory: /tmp/loki/boltdb-shipper-compactor
-    shared_store: filesystem
     retention_enabled: true
+    delete_request_store: filesystem

packages/backend-common/src/modules/action-dispatcher/action-dispatcher.service.ts

@@ -107,6 +107,17 @@ export class ActionDispatcherService implements OnModuleInit {
             );
         }
 
+        try {
+            const lokiUrl = process.env.LOKI_URL ?? 'http://loki:3100';
+            const { default: axios } = await import('axios');
+            await axios.get(`${lokiUrl}/ready`, { timeout: 2000 });
+        } catch {
+            this.logger.error('Loki logging system is down or unreachable');
+            throw new ConflictException(
+                'Logging system (Loki) is not available. Please try again later.',
+            );
+        }
+
         let action = this.actionRepository.create({
             mission,
             creator,