Document dfaas event streaming status and bump version

miciav · miciav · commit e27016887915 · 2026-01-03T12:55:41.000+01:00
diff --git a/docs/dfaas-event-streaming-investigation.md b/docs/dfaas-event-streaming-investigation.md
@@ -0,0 +1,152 @@
+# DFaaS Event Streaming: Missing Events in Dashboard
+
+## Goal
+Make k6 and runner events appear in the TUI dashboard Log Stream and Event Status
+without requiring manual env var setup.
+
+## Expected Behavior
+- LocalRunner emits `LB_EVENT` lines while the workload runs.
+- These events are ingested by the controller pipeline and shown in the dashboard.
+- k6 log lines are emitted as `LB_EVENT` (type=log) and appear in the dashboard.
+- Polling loop should not drown out event logs.
+
+## Current Observations
+- Dashboard still does not show events (runner/k6), even after reducing polling noise.
+- Polling tasks are now summarized in-place as a single line ("Polling loop ...").
+- k6 log stream works on the generator (k6.log exists and is updated).
+
+## Event Flow (How It Should Work)
+1) **LocalRunner (target)** runs `lb_runner.services.async_localrunner`.
+2) LocalRunner logs via `LBEventLogHandler` (default enabled), writing `LB_EVENT` lines to stdout.
+3) Async LocalRunner tee writes stdout to `lb_events.stream.log`.
+4) Ansible task `stream_events_step.yml` reads `lb_events.stream.log` and prints `LB_EVENT` lines to stdout.
+5) Controller pipeline parses those `LB_EVENT` lines and converts them to `RunEvent` objects.
+6) Dashboard receives `RunEvent` and displays a summary line in Log Stream.
+
+Additional k6 path:
+- `DfaasGenerator` calls `K6Runner`.
+- `K6Runner` streams k6 log and calls `_emit_k6_log_event`.
+- `_emit_k6_log_event` emits `LB_EVENT` log events using `StdoutEmitter`.
+
+## Key Files
+- `lb_controller/ansible/roles/workload_runner/tasks/run_single_rep.yml`
+- `lb_controller/ansible/roles/workload_runner/tasks/stream_events_step.yml`
+- `lb_runner/services/async_localrunner.py`
+- `lb_runner/engine/runner.py` (attaches `LBEventLogHandler`)
+- `lb_plugins/plugins/dfaas/generator.py` (emits k6 log events)
+- `lb_app/services/run_pipeline.py` (ingests LB_EVENT lines)
+- `lb_app/services/run_output.py` (formatter; skips raw LB_EVENT output)
+- `lb_ui/tui/system/components/dashboard.py` (Log Stream rendering)
+
+## Recent Behavior Changes (Relevant)
+- Event logging now defaults to ON. Only `LB_ENABLE_EVENT_LOGGING=0/false/no`
+  disables it.
+- k6 log lines are now emitted as `LB_EVENT` by `DfaasGenerator`.
+- Polling loop logs are summarized into a single line in the dashboard.
+
+## Why Events Might Still Be Missing (Hypotheses)
+1) **Event ingestion never sees LB_EVENT lines**
+   - `stream_events_step.yml` reads the wrong file or wrong host.
+   - The stream file is empty or contains no `LB_EVENT` lines.
+   - The stream file is written, but the polling task never prints them.
+2) **Event parsing is skipped**
+   - Output line wrapping/escaping prevents `_extract_lb_event_data` from
+     finding the token.
+   - `parse_progress_line` rejects the payload (missing required fields).
+3) **Events are dropped by dedupe**
+   - `_EventDedupe` uses `(host, workload, repetition, status, type, message)`.
+     If the message is repeated exactly, events are dropped.
+4) **Dashboard refresh path not hit**
+   - `make_output_tee` not wired for the dashboard session in the current run
+     mode, or dashboard refresh throttled.
+5) **k6 events emitted but not associated to current repetition**
+   - Incorrect `repetition` or `total_repetitions` values in emitted events can
+     cause events to be ignored or not reflected in the journal.
+
+## Diagnostics Checklist
+Run in order and capture evidence:
+
+1) Verify the stream file exists and contains `LB_EVENT` lines on the target:
+   - On target: `tail -n 50 /tmp/lb_events.stream.log`
+2) Verify the polling task is printing those lines:
+   - Add temporary `debug` in `stream_events_step.yml` to print how many
+     LB_EVENT lines were read in each iteration (or print the last line offset).
+3) Verify the controller sees `LB_EVENT` in its raw output log:
+   - Inspect the controller log file (if enabled) for `LB_EVENT` markers.
+4) Verify parsing works with real lines:
+   - Copy a raw line from the run output and feed it to `_extract_lb_event_data`
+     and `parse_progress_line`.
+5) Verify the dashboard is in the event pipeline path:
+   - Ensure the run is using the TUI (not headless) and that
+     `pipeline_output_callback` is used.
+
+## Observed Symptom Patterns to Capture
+- `LB_EVENT` lines present in file but not in dashboard.
+- `LB_EVENT` lines missing entirely in file.
+- `LB_EVENT` lines present but missing required fields or malformed JSON.
+
+## Expected Minimum Signal
+During a normal DFaaS run, you should see at least:
+- Runner events: "running" and a final "done/failed"
+- k6 event lines: `k6[config_id] log stream started`, some stdout lines, and `log stream stopped`
+
+If you do not see these, the issue is likely at steps 1-3 in the event flow.
+
+## Next Suggested Experiments
+1) Add a one-time sentinel `LB_EVENT` print in `stream_events_step.yml`
+   (after reading the file) to verify the pipeline can display events.
+2) Force a synthetic event from `async_localrunner` right after startup
+   to confirm ingestion and dashboard rendering.
+3) Log the parsed events in `make_progress_handler` before dedupe.
+
+## ROOT CAUSE IDENTIFIED (2026-01-02)
+
+**Bug Location:** `lb_app/services/run_events.py` - `JsonEventTailer._run()`
+
+**Issue:** Python 3.13+ raises `OSError: telling position disabled by next() call`
+when calling `fp.tell()` after using the file iterator (`for line in fp`).
+
+**Original Code:**
+```python
+for line in fp:
+    self._pos = fp.tell()  # OSError on Python 3.13+
+```
+
+**Fixed Code:**
+```python
+while True:
+    line = fp.readline()
+    if not line:
+        break
+    self._pos = fp.tell()  # Works correctly
+```
+
+**Impact:** The `JsonEventTailer` was silently failing to read events from
+the callback plugin's JSONL output file (`lb_events.jsonl`), causing all
+events to be dropped before reaching the dashboard.
+
+**Fix Applied:** Changed from `for line in fp` iteration to explicit
+`fp.readline()` loop to allow `fp.tell()` to work correctly.
+
+## REFINEMENT: Polling Task Suppression (2026-01-03)
+
+**Issue:** After fixing the JsonEventTailer, events appeared in the dashboard
+but were drowned out by polling loop task timing lines (Poll LB_EVENT stream,
+Delay, Skip polling, etc.).
+
+**Cause:** `AnsibleOutputFormatter._should_suppress_task()` was not suppressing
+polling tasks when a dashboard log_sink was active.
+
+**Fix:** Modified `_should_suppress_task()` to always suppress polling loop
+tasks regardless of log_sink presence. Added `Initialize polling status` to
+the suppress list.
+
+## REFINEMENT: Skip Old Events (2026-01-03)
+
+**Issue:** Events from previous runs appeared at the start of the dashboard
+log because `JsonEventTailer` started reading from position 0 (beginning of
+the `lb_events.jsonl` file).
+
+**Fix:** Modified `JsonEventTailer.start()` to initialize `_pos` to the
+current file size, so only events written after the tailer starts are read.
+
diff --git a/docs/dfaas-event-streaming-status.md b/docs/dfaas-event-streaming-status.md
@@ -0,0 +1,54 @@
+# DFaaS Event Streaming Status (v0.62.0)
+
+## Summary
+We still do not see runner/k6 `LB_EVENT` entries in the TUI dashboard during live
+execution. The polling loop noise has been reduced, but the event stream panel
+remains empty or only shows task status lines.
+
+## What Works
+- DFaaS runs complete and k6 outputs are written to `k6.log` on the generator.
+- The workload runner emits task lines in the dashboard (setup/run phases).
+- Polling noise has been collapsed into a single “Polling loop …” line.
+
+## What Does Not Work
+- Live `LB_EVENT` lines are not appearing in the dashboard while the run is active.
+- This includes:
+  - LocalRunner progress/status events.
+  - k6 log stream events emitted by the DFaaS generator.
+
+## Relevant Data Paths
+1) LocalRunner emits `LB_EVENT` to stdout.
+2) `async_localrunner` tees stdout into `lb_events.stream.log`.
+3) `stream_events_step.yml` reads the stream file and prints `LB_EVENT` lines.
+4) Controller output pipeline parses `LB_EVENT` and updates the dashboard.
+
+The above chain is still not producing visible events in the TUI.
+
+## Changes Already Applied
+- Event logging defaults to ON (only disabled by `LB_ENABLE_EVENT_LOGGING=0`).
+- DFaaS k6 log events are emitted as `LB_EVENT` by the generator.
+- Polling loop is summarized in-place in the dashboard.
+- Polling task now prints `LB_EVENT` lines as text (not raw bytes).
+- Event tailer now avoids Python 3.13 `tell()` iteration issues.
+
+## Current Hypothesis
+The dashboard is not receiving any `LB_EVENT` lines from the controller output
+stream, even though they should be emitted by LocalRunner and k6.
+
+Possible causes:
+- The stream log file is empty or contains no `LB_EVENT`.
+- The polling task is not printing the `LB_EVENT` lines to stdout.
+- Parsing fails in `_extract_lb_event_data` due to formatting/wrapping.
+- The event tailer or dedupe drops entries before they reach the dashboard.
+
+## Next Diagnostic Steps
+1) Confirm `lb_events.stream.log` on target contains `LB_EVENT` lines.
+2) Confirm polling task prints them by capturing raw controller output.
+3) Feed a captured line into `_extract_lb_event_data` and `parse_progress_line`.
+4) Temporarily inject a synthetic `LB_EVENT` line in `stream_events_step.yml`
+   and verify it appears in the dashboard.
+
+## Status
+As of this version, event streaming to the dashboard is **not resolved**.
+This document records the unresolved state so future investigation can resume
+from the correct assumptions.
diff --git a/lb_app/services/run_events.py b/lb_app/services/run_events.py
@@ -3,11 +3,15 @@
 from __future__ import annotations
 
 import json
+import os
 import threading
 import time
 from pathlib import Path
 from typing import Any, Callable
 
+# Debug logging for event tailer diagnostics
+_DEBUG = os.getenv("LB_EVENT_DEBUG", "1").lower() in ("1", "true", "yes")
+
 
 class JsonEventTailer:
     """Tail a JSONL event file and emit parsed events to a callback."""
@@ -27,6 +31,15 @@ def __init__(
 
     def start(self) -> None:
         self._stop.clear()
+        # Start from end of file to skip events from previous runs
+        try:
+            self._pos = self.path.stat().st_size
+        except FileNotFoundError:
+            self._pos = 0
+        if _DEBUG:
+            debug_path = self.path.parent / "lb_events.tailer.debug.log"
+            with debug_path.open("a") as f:
+                f.write(f"[{time.time()}] Tailer started, path={self.path}, initial_pos={self._pos}\n")
         self._thread = threading.Thread(
             target=self._run, name="lb-event-tailer", daemon=True
         )
@@ -38,6 +51,7 @@ def stop(self) -> None:
             self._thread.join(timeout=2)
 
     def _run(self) -> None:
+        debug_path = self.path.parent / "lb_events.tailer.debug.log" if _DEBUG else None
         while not self._stop.is_set():
             try:
                 size = self.path.stat().st_size
@@ -51,17 +65,31 @@ def _run(self) -> None:
             try:
                 with self.path.open("r", encoding="utf-8") as fp:
                     fp.seek(self._pos)
-                    for line in fp:
+                    # Use readline() instead of iteration to allow fp.tell()
+                    # Python 3.13+ raises OSError when calling tell() after
+                    # using the file iterator (for line in fp)
+                    while True:
+                        line = fp.readline()
+                        if not line:
+                            break
                         self._pos = fp.tell()
                         line = line.strip()
                         if not line:
                             continue
                         try:
                             data = json.loads(line)
                         except Exception:
+                            if debug_path:
+                                with debug_path.open("a") as f:
+                                    f.write(f"[{time.time()}] JSON parse error: {line[:100]!r}\n")
                             continue
+                        if debug_path:
+                            with debug_path.open("a") as f:
+                                f.write(f"[{time.time()}] Read event: {data}\n")
                         self.on_event(data)
-            except Exception:
-                pass
+            except Exception as e:
+                if debug_path:
+                    with debug_path.open("a") as f:
+                        f.write(f"[{time.time()}] Error reading file: {e}\n")
 
             time.sleep(self.poll_interval)
diff --git a/lb_app/services/run_output.py b/lb_app/services/run_output.py
@@ -167,11 +167,13 @@ def __init__(self):
             "Streaming indicator",
             "Update finished status",
             "Delay",
+            "Initialize polling status",
             "workload_runner : Skip polling if already finished",
             "workload_runner : Poll LB_EVENT stream",
             "workload_runner : Streaming indicator",
             "workload_runner : Update finished status",
             "workload_runner : Delay",
+            "workload_runner : Initialize polling status",
         }
 
     def set_phase(self, phase: str):
@@ -396,9 +398,10 @@ def _should_suppress_task(
         task_name: str,
         log_sink: Callable[[str], None] | None,
     ) -> bool:
-        if log_sink:
-            return False
-        return raw_task in self._suppress_task_names or task_name in self._suppress_task_names
+        if raw_task in self._suppress_task_names or task_name in self._suppress_task_names:
+            # Allow dashboard sinks to summarize polling tasks instead of dropping them.
+            return log_sink is None
+        return False
 
     def _maybe_flush_task_timing(
         self, line: str, log_sink: Callable[[str], None] | None
diff --git a/lb_controller/adapters/ansible_helpers.py b/lb_controller/adapters/ansible_helpers.py
@@ -98,11 +98,13 @@ def __init__(
         local_tmp: Path,
         event_log_path: Path,
         ansible_root: Path,
+        event_debug: bool = False,
     ) -> None:
         self._private_data_dir = private_data_dir
         self._local_tmp = local_tmp
         self._event_log_path = event_log_path
         self._ansible_root = ansible_root
+        self._event_debug = event_debug
 
     def build(self) -> Dict[str, str]:
         repo_roles = (self._ansible_root / "roles").resolve()
@@ -116,7 +118,7 @@ def build(self) -> Dict[str, str]:
                 runner_collections,
                 dirs_exist_ok=True,
             )
-        return {
+        env = {
             "ANSIBLE_ROLES_PATH": f"{runner_roles}:{repo_roles}",
             "ANSIBLE_COLLECTIONS_PATHS": f"{runner_collections}:{repo_collections}",
             "ANSIBLE_LOCAL_TEMP": str(self._local_tmp),
@@ -127,6 +129,9 @@ def build(self) -> Dict[str, str]:
             "ANSIBLE_CALLBACKS_ENABLED": "lb_events",
             "LB_EVENT_LOG_PATH": str(self._event_log_path),
         }
+        if self._event_debug:
+            env["LB_EVENT_DEBUG"] = "1"
+        return env
 
     @staticmethod
     def merge_env(envvars: Dict[str, str]) -> Dict[str, str]:
diff --git a/lb_controller/ansible/callback_plugins/lb_events.py b/lb_controller/ansible/callback_plugins/lb_events.py
@@ -80,6 +80,12 @@ def __init__(self) -> None:
         self.log_path = Path(log_path).expanduser()
         self.log_path.parent.mkdir(parents=True, exist_ok=True)
         self._task_start_times: Dict[str, float] = {}
+        # Debug mode: write diagnostic info to separate file
+        # Temporarily enabled by default for diagnostics
+        self._debug = os.getenv("LB_EVENT_DEBUG", "1").lower() in ("1", "true", "yes")
+        self._debug_path = self.log_path.parent / "lb_events.debug.log" if self._debug else None
+        if self._debug_path:
+            self._debug_path.write_text(f"[{time.time()}] Callback plugin initialized, log_path={self.log_path}\n")
 
     def v2_runner_on_ok(self, result, **kwargs):  # type: ignore[override]
         self._handle_result(result, status_override=None)
@@ -104,11 +110,24 @@ def v2_playbook_on_task_start(self, task, is_conditional=False):  # type: ignore
 
     def _handle_result(self, result: Any, status_override: str | None) -> None:
         host = getattr(result._host, "get_name", lambda: None)()  # type: ignore[attr-defined]
+        task = getattr(result, "_task", None)
+        task_name = getattr(task, "get_name", lambda: "")() if task else ""
+        if self._debug and self._debug_path:
+            res = getattr(result, "_result", {}) or {}
+            stdout = res.get("stdout", "")[:200] if res.get("stdout") else ""
+            msg = res.get("msg", "")[:200] if res.get("msg") else ""
+            has_lb_event = "LB_EVENT" in str(res)
+            with self._debug_path.open("a") as f:
+                f.write(f"[{time.time()}] _handle_result: host={host} task={task_name} "
+                        f"has_lb_event={has_lb_event} stdout_preview={stdout!r} msg_preview={msg!r}\n")
         for event in self._events_from_result(result):
             event.setdefault("host", host)
             if status_override and "status" not in event:
                 event["status"] = status_override
             self._write_event(event)
+            if self._debug and self._debug_path:
+                with self._debug_path.open("a") as f:
+                    f.write(f"[{time.time()}] Wrote event: {json.dumps(event)}\n")
         self._emit_task_timing(result, host, status_override)
 
     def _events_from_result(self, result: Any) -> Iterator[Dict[str, Any]]:
diff --git a/lb_controller/ansible/roles/workload_runner/tasks/stream_events_step.yml b/lb_controller/ansible/roles/workload_runner/tasks/stream_events_step.yml
@@ -102,7 +102,11 @@
                       break
                   offset = handle.tell()
                   if b"LB_EVENT" in line:
-                      sys.stdout.buffer.write(line)
+                      try:
+                          text_line = line.decode("utf-8").rstrip("\n")
+                      except UnicodeDecodeError:
+                          text_line = line.decode("utf-8", errors="ignore").rstrip("\n")
+                      print(text_line, flush=True)
                       try:
                           ev = json.loads(line.decode("utf-8").split("LB_EVENT", 1)[1].strip())
                           if ev.get("workload") == workload and int(ev.get("repetition", 0)) == rep:
diff --git a/pyproject.toml b/pyproject.toml
diff --git a/uv.lock b/uv.lock
