diff --git a/backend/routes/tasks.py b/backend/routes/tasks.py index e3acf0b0..ab117741 100644 --- a/backend/routes/tasks.py +++ b/backend/routes/tasks.py @@ -13,7 +13,7 @@ from fastapi import APIRouter, Depends, Query, WebSocket, WebSocketDisconnect from pydantic import BaseModel -from sqlalchemy import desc +from sqlalchemy import desc, func from sqlalchemy.orm import Session, joinedload @@ -92,14 +92,22 @@ def get_tasks( # Order by creation time (newest first) and paginate # Eager load relationships to avoid N+1 queries # B3: load_only() — task list only needs project.name and library_module.name - tasks = query.options( + # + # #195: expose the log size on the list too. ``logs`` is a deferred Text + # column, so we compute its length in SQL (func.length) rather than loading + # every task's full log body just to measure it — this keeps the list cheap + # while giving clients the same ``logs_full_size`` the detail endpoint + # reports. The full ``logs`` body stays detail-only by design. + rows = query.options( joinedload(Task.project).load_only(Project.id, Project.name), joinedload(Task.module).joinedload(ProjectModule.library_module).load_only(ModuleLibrary.id, ModuleLibrary.name) + ).add_columns( + func.length(Task.logs).label("logs_full_size") ).order_by(desc(Task.created_at)).limit(limit).offset(offset).all() # Format response tasks_data = [] - for task in tasks: + for task, logs_full_size in rows: task_dict = { "id": task.id, "celery_task_id": task.celery_task_id, @@ -118,6 +126,12 @@ def get_tasks( "exit_code": task.exit_code, "error": task.error, "archived": task.archived, + # NULL logs → length is NULL → report 0, matching the detail + # endpoint's ``len(logs) if logs else 0``. ``logs_truncated`` is + # always False here: the list never returns a truncated body (it + # returns no body at all — fetch the detail endpoint for logs). + "logs_full_size": logs_full_size or 0, + "logs_truncated": False, } tasks_data.append(task_dict) diff --git a/backend/services/execution/opentofu_runtime.py b/backend/services/execution/opentofu_runtime.py index 904fb63d..a9e3267f 100644 --- a/backend/services/execution/opentofu_runtime.py +++ b/backend/services/execution/opentofu_runtime.py @@ -15,15 +15,18 @@ - Dependency checking → variable_assembler.can_execute() """ +import codecs import hashlib import json import logging import os import re +import select import shutil import subprocess import tempfile import time +from collections.abc import Callable from pathlib import Path from sqlalchemy.orm import Session @@ -36,6 +39,173 @@ logger = logging.getLogger(__name__) +def _stream_subprocess( + cmd: list[str], + *, + cwd: str, + env: dict, + timeout: int, + on_output: Callable[[str], None], +) -> tuple[int, str]: + """Run ``cmd`` streaming stdout+stderr line-by-line to ``on_output``. + + Behaves like ``subprocess.run(..., capture_output=True, text=True, + timeout=...)`` from the caller's point of view: it returns + ``(returncode, combined_output)`` and raises ``subprocess.TimeoutExpired`` + (with ``output`` populated) on timeout, so the existing timeout-handling + branches in each ``run_*`` method work unchanged. + + The difference — and the whole point (issue #195) — is that output is + delivered incrementally: each line is handed to ``on_output`` the moment + OpenTofu emits it, letting the caller persist progress (``task.logs`` / + ``logs_full_size``) *during* a long run instead of only at completion. + stderr is merged into stdout (``STDOUT``) so lines interleave in the order + they were produced. + + This is a faithful, safe wrapper of ``subprocess.run``'s guarantees, not a + partial reimplementation. Two properties matter for correctness and are + handled exactly as CPython's ``subprocess.run`` handles them (its POSIX + ``_communicate`` reads the pipes with a ``selectors`` loop — which is what we + do here — never a plain ``for line in proc.stdout`` that a descendant can + wedge): + + * **The timeout is really enforced.** The deadline cannot be honoured by the + naive "kill the direct child ⇒ the pipe reaches EOF" implication, because + that implication is *false* whenever any descendant (e.g. a ``local-exec`` + / ``null_resource`` / ``data "external"`` grandchild) inherited the write + end of the stdout pipe: a blocking read would then hang forever and + ``TimeoutExpired`` would never fire, permanently locking the module. We + instead ``select`` on the pipe with the *remaining* time budget, so the + loop always terminates at the deadline regardless of who holds the pipe; + we then ``kill()`` the child and raise ``TimeoutExpired`` with the partial + output. Reading on this (the caller's) thread — rather than a helper + thread blocked in ``read()`` — is also what lets ``with Popen`` close the + pipe on exit without deadlocking on that thread's buffer lock. + * **The child is killed on ANY abrupt exit.** ``subprocess.run`` wraps its + read in ``except BaseException: process.kill(); raise`` precisely so that + a ``SoftTimeLimitExceeded`` (Celery raises it in this very thread, while it + is blocked in ``select``) / ``KeyboardInterrupt`` / ``UnicodeDecodeError`` + (non-UTF-8 provider output) / ``OSError`` cannot leave a live ``tofu + apply`` orphaned — still mutating cloud state and ``.tfstate`` after the + task is marked failed and the workspace lock released. We do the same. + """ + deadline = time.monotonic() + timeout + chunks: list[str] = [] # decoded+newline-normalized pieces == combined output + # Incremental UTF-8 decoder: correctly reassembles multibyte characters that + # straddle two reads, and (strict) raises UnicodeDecodeError on genuinely + # invalid bytes — the same failure ``text=True`` would surface. + decoder = codecs.getincrementaldecoder("utf-8")("strict") + pending = "" # current partial line, not yet newline-terminated + carry_cr = False # a trailing '\r' held back — maybe the first half of a CRLF + + def _deliver(line: str) -> None: + try: + on_output(line) + except Exception: # noqa: BLE001 — a log sink must never break the run + logger.exception("on_output callback raised while streaming subprocess output") + + def _emit(text: str, *, flush: bool = False) -> None: + # MINOR 3: match ``subprocess.run(text=True)`` universal-newline semantics + # (the no-callback path uses it) so streamed logs don't diverge — translate + # CRLF and lone CR to LF. Normalize on the *accumulated* stream (via + # ``carry_cr``), never per-chunk, so a CRLF split across two reads + # (``…\r`` | ``\n…``) is not mistaken for two newlines. ``chunks`` (the + # returned combined output) is fed here too, so return value and delivered + # lines stay identical to the ``text=True`` path. + nonlocal pending, carry_cr + if carry_cr: + text = "\r" + text + carry_cr = False + text = text.replace("\r\n", "\n") + if not flush and text.endswith("\r"): + # Hold the trailing CR: the next read may bring the LF of a CRLF. + carry_cr = True + text = text[:-1] + text = text.replace("\r", "\n") + if text: + chunks.append(text) + pending += text + newline = pending.find("\n") + while newline != -1: + _deliver(pending[:newline]) + pending = pending[newline + 1:] + newline = pending.find("\n") + if flush and pending: + _deliver(pending) + pending = "" + + # ``with Popen(...)`` guarantees the pipes are closed on every exit path, + # exactly like the ``with Popen`` inside ``subprocess.run``. bufsize=0 keeps + # the parent side unbuffered so our ``os.read`` on the fd sees bytes as soon + # as the child writes them (line streaming, issue #195). + with subprocess.Popen( + cmd, + cwd=cwd, + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + bufsize=0, + ) as proc: + assert proc.stdout is not None + fd = proc.stdout.fileno() + timed_out = False + try: + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + timed_out = True + break + # PEP 475: select retries automatically on EINTR, but a signal + # handler that *raises* (Celery's SoftTimeLimitExceeded) surfaces + # its exception here — handled by the kill-and-reraise below. + readable, _, _ = select.select([fd], [], [], remaining) + if not readable: + timed_out = True # deadline reached with no more output + break + data = os.read(fd, 65536) + if not data: + break # EOF: all write ends (incl. any descendant's) closed + text = decoder.decode(data) # may raise UnicodeDecodeError → killed below + _emit(text) + + # MINOR 2: honour the deadline BEFORE the final decoder flush. If the + # deadline expired with a partial multibyte sequence buffered, + # ``decoder.decode(b"", final=True)`` would raise UnicodeDecodeError, + # which would surface IN PLACE OF TimeoutExpired and bypass the + # callers' graceful ``except subprocess.TimeoutExpired`` branch. Raise + # TimeoutExpired first, with the partial output accumulated so far. + if timed_out: + proc.kill() + raise subprocess.TimeoutExpired(cmd, timeout, output="".join(chunks)) + + # Genuine EOF on all pipe write ends. Flush any bytes buffered inside + # the incremental decoder plus the trailing line that had no newline. + _emit(decoder.decode(b"", final=True), flush=True) + + # MINOR 1: EOF does NOT imply the child has exited — a descendant may + # have closed the inherited stdout pipe while the child keeps running + # (real EOF, child alive). A bare ``proc.wait()`` here would block + # forever, defeating the timeout the docstring promises. Bound the wait + # by the remaining deadline; on expiry, kill and raise TimeoutExpired + # with the accumulated partial output (same shape as the in-loop path). + try: + proc.wait(timeout=max(0, deadline - time.monotonic())) + except subprocess.TimeoutExpired: + proc.kill() + raise subprocess.TimeoutExpired(cmd, timeout, output="".join(chunks)) + except BaseException: + # Any abrupt exit — SoftTimeLimitExceeded, KeyboardInterrupt, + # UnicodeDecodeError, OSError, … — must not orphan a live child. + # Mirror CPython's ``subprocess.run``: kill, then re-raise. + try: + proc.kill() + except Exception: # noqa: BLE001 — process may have already exited + pass + raise + + return proc.returncode, "".join(chunks) + + def _add_provider_lock_timeout_hint(output: str) -> str: """Annotate known provider install stalls with a concrete runtime hint.""" normalized_output = (output or "").lower() @@ -1516,7 +1686,43 @@ def write_provider_config(self, work_dir: str, module: ProjectModule, variables: APPLY_TIMEOUT = 90 * 60 # 90 minutes - long-running resource creation REFRESH_TIMEOUT = 30 * 60 # 30 minutes - state refresh - def run_init(self, work_dir: str, env: dict, timeout: int | None = None) -> tuple[int, str]: + @staticmethod + def _run_tofu( + cmd: list[str], + work_dir: str, + tofu_env: dict, + timeout: int, + on_output: Callable[[str], None] | None, + ) -> tuple[int, str]: + """Execute a tofu command, returning ``(returncode, stdout+stderr)``. + + When ``on_output`` is provided the output is streamed line-by-line via + :func:`_stream_subprocess` (issue #195); otherwise it falls back to the + classic blocking ``subprocess.run`` capture. Both paths raise + ``subprocess.TimeoutExpired`` on timeout so each caller's existing + timeout branch is unchanged. + """ + if on_output is not None: + return _stream_subprocess( + cmd, cwd=work_dir, env=tofu_env, timeout=timeout, on_output=on_output, + ) + result = subprocess.run( + cmd, + cwd=work_dir, + env=tofu_env, + capture_output=True, + text=True, + timeout=timeout, + ) + return result.returncode, result.stdout + result.stderr + + def run_init( + self, + work_dir: str, + env: dict, + timeout: int | None = None, + on_output: Callable[[str], None] | None = None, + ) -> tuple[int, str]: """ Run tofu init. @@ -1524,6 +1730,9 @@ def run_init(self, work_dir: str, env: dict, timeout: int | None = None) -> tupl work_dir: Workspace directory env: Environment variables timeout: Optional timeout in seconds (default: INIT_TIMEOUT) + on_output: Optional line callback. When provided, output is streamed + line-by-line (issue #195) so callers can persist progress during + the run; when None, behaviour is the classic blocking capture. Returns: Tuple of (exit_code, output) @@ -1535,19 +1744,15 @@ def run_init(self, work_dir: str, env: dict, timeout: int | None = None) -> tupl logger.info(f"Running tofu init in {work_dir} (timeout: {timeout}s)") try: - result = subprocess.run( + returncode, raw_output = self._run_tofu( ["tofu", "init", "-no-color", "-input=false"], - cwd=work_dir, - env=tofu_env, - capture_output=True, - text=True, - timeout=timeout + work_dir, tofu_env, timeout, on_output, ) - output = _add_provider_lock_timeout_hint(result.stdout + result.stderr) - logger.info(f"tofu init completed with exit code {result.returncode}") + output = _add_provider_lock_timeout_hint(raw_output) + logger.info(f"tofu init completed with exit code {returncode}") - return result.returncode, output + return returncode, output except subprocess.TimeoutExpired as e: # S14-021: Handle timeout for init @@ -1565,7 +1770,13 @@ def run_init(self, work_dir: str, env: dict, timeout: int | None = None) -> tupl output = _add_provider_lock_timeout_hint((stdout or "") + (stderr or "") + timeout_msg) return 1, output - def run_plan(self, work_dir: str, env: dict, timeout: int | None = None) -> tuple[int, str]: + def run_plan( + self, + work_dir: str, + env: dict, + timeout: int | None = None, + on_output: Callable[[str], None] | None = None, + ) -> tuple[int, str]: """ Run tofu plan. @@ -1573,6 +1784,7 @@ def run_plan(self, work_dir: str, env: dict, timeout: int | None = None) -> tupl work_dir: Workspace directory env: Environment variables timeout: Optional timeout in seconds (default: PLAN_TIMEOUT) + on_output: Optional line callback for incremental log streaming (#195). Returns: Tuple of (exit_code, output) @@ -1584,19 +1796,14 @@ def run_plan(self, work_dir: str, env: dict, timeout: int | None = None) -> tupl logger.info(f"Running tofu plan in {work_dir} (timeout: {timeout}s)") try: - result = subprocess.run( + returncode, output = self._run_tofu( ["tofu", "plan", "-no-color", "-input=false", "-out=plan.out"], - cwd=work_dir, - env=tofu_env, - capture_output=True, - text=True, - timeout=timeout + work_dir, tofu_env, timeout, on_output, ) - output = result.stdout + result.stderr - logger.info(f"tofu plan completed with exit code {result.returncode}") + logger.info(f"tofu plan completed with exit code {returncode}") - return result.returncode, output + return returncode, output except subprocess.TimeoutExpired as e: # S14-021: Handle timeout for plan @@ -1673,6 +1880,7 @@ def run_apply( env: dict, timeout: int | None = None, module: ProjectModule | None = None, + on_output: Callable[[str], None] | None = None, ) -> tuple[int, str, dict]: """ Run tofu apply and capture outputs. @@ -1681,6 +1889,7 @@ def run_apply( work_dir: Workspace directory env: Environment variables timeout: Optional timeout in seconds (default: APPLY_TIMEOUT) + on_output: Optional line callback for incremental log streaming (#195). Returns: Tuple of (exit_code, output, captured_outputs) @@ -1693,21 +1902,16 @@ def run_apply( try: # Apply the plan - result = subprocess.run( + returncode, output = self._run_tofu( ["tofu", "apply", "-no-color", "-input=false", "-auto-approve", "plan.out"], - cwd=work_dir, - env=tofu_env, - capture_output=True, - text=True, - timeout=timeout + work_dir, tofu_env, timeout, on_output, ) - output = result.stdout + result.stderr - logger.info(f"tofu apply completed with exit code {result.returncode}") + logger.info(f"tofu apply completed with exit code {returncode}") # Capture outputs if apply succeeded outputs = {} - if result.returncode == 0: + if returncode == 0: outputs = self._capture_outputs(work_dir, tofu_env) if module is not None: normalized = normalize_infrastructure_access_outputs( @@ -1717,7 +1921,7 @@ def run_apply( ) outputs = normalized.outputs - return result.returncode, output, outputs + return returncode, output, outputs except subprocess.TimeoutExpired as e: # S14-021: Handle timeout for apply @@ -1841,7 +2045,13 @@ def normalize_outputs(self, outputs: dict, module: ProjectModule) -> dict: ) return normalized.outputs - def run_destroy(self, work_dir: str, env: dict, timeout: int | None = None) -> tuple[int, str]: + def run_destroy( + self, + work_dir: str, + env: dict, + timeout: int | None = None, + on_output: Callable[[str], None] | None = None, + ) -> tuple[int, str]: """ Run tofu destroy. @@ -1849,6 +2059,7 @@ def run_destroy(self, work_dir: str, env: dict, timeout: int | None = None) -> t work_dir: Workspace directory env: Environment variables timeout: Timeout in seconds (default: 30 minutes) + on_output: Optional line callback for incremental log streaming (#195). Returns: Tuple of (exit_code, output) @@ -1860,19 +2071,14 @@ def run_destroy(self, work_dir: str, env: dict, timeout: int | None = None) -> t logger.info(f"Running tofu destroy in {work_dir} (timeout: {timeout}s)") try: - result = subprocess.run( + returncode, output = self._run_tofu( ["tofu", "destroy", "-no-color", "-input=false", "-auto-approve"], - cwd=work_dir, - env=tofu_env, - capture_output=True, - text=True, - timeout=timeout + work_dir, tofu_env, timeout, on_output, ) - output = result.stdout + result.stderr - logger.info(f"tofu destroy completed with exit code {result.returncode}") + logger.info(f"tofu destroy completed with exit code {returncode}") - return result.returncode, output + return returncode, output except subprocess.TimeoutExpired as e: timeout_msg = ( @@ -1895,7 +2101,8 @@ def run_destroy_with_retry( module: ProjectModule | None = None, max_retries: int | None = None, initial_delay: float | None = None, - timeout: int | None = None + timeout: int | None = None, + on_output: Callable[[str], None] | None = None, ) -> tuple[int, str]: """ Run tofu destroy with retry logic for dependency violations. @@ -1941,7 +2148,9 @@ def run_destroy_with_retry( time.sleep(delay) # Run destroy - exit_code, output = self.run_destroy(work_dir, env, timeout=timeout) + exit_code, output = self.run_destroy( + work_dir, env, timeout=timeout, on_output=on_output, + ) all_output += output # Success - return immediately diff --git a/backend/tasks/_tofu_helpers.py b/backend/tasks/_tofu_helpers.py index aee67b37..bda55ba4 100644 --- a/backend/tasks/_tofu_helpers.py +++ b/backend/tasks/_tofu_helpers.py @@ -11,6 +11,8 @@ import logging import re +import time +from collections.abc import Callable from datetime import UTC, datetime from celery import Task @@ -23,6 +25,81 @@ logger = logging.getLogger(__name__) +class TofuLogStreamer: + """Incrementally flush ``task.logs`` to the DB as a tofu step streams output. + + Fixes the first defect in #195: OpenTofu task logs were buffered until the + task completed, so ``logs_full_size`` sat at 0 for the whole run and only + jumped to its final value at the end — a running module was opaque. The + ``run_*`` runtime methods now stream stdout line-by-line through an + ``on_output`` callback; this helper turns that callback into throttled + writes of ``task.logs`` so the task-detail endpoint's ``logs_full_size`` + grows *during* the run. + + Usage:: + + streamer = TofuLogStreamer(task, db) + ... + all_logs += header + code, logs = engine.run_plan(work_dir, env, on_output=streamer.begin(all_logs)) + all_logs += logs # unchanged: final source of truth + task.logs = all_logs # written + committed by the task as before + + ``begin(base)`` snapshots the log accumulated before the step (section + headers + earlier steps) and returns the sink. During the step the sink + persists ``base + ``. The task still writes the + complete ``all_logs`` (built from each step's returned output) at the end, + so the final content and size are exactly what they were before — only the + *timing* of visibility changes. The sink runs on the task's own thread (the + runtime reads the pipe synchronously), so touching ``task``/``db`` here is + safe. Both the write and commit are best-effort: a log flush must never + fail the operation. + """ + + def __init__(self, task: TaskModel, db, *, interval: float = 2.0): + self._task = task + self._db = db + self._interval = interval + self._base = "" + self._buf: list[str] = [] + self._last = 0.0 + # High-water mark of persisted log length. task.logs must only ever grow + # during a run (the strict-growth invariant behind logs_full_size); a + # step whose base is shorter than what we already persisted — e.g. a + # stale-plan retry whose begin() base omits the first apply's already- + # streamed output (#195 F1) — must NOT rewind it. + self._persisted_len = 0 + + def begin(self, base: str) -> Callable[[str], None]: + """Start streaming a step whose output extends ``base``; return the sink.""" + self._base = base + self._buf = [] + self._last = 0.0 # force the first line to flush immediately + return self._sink + + def _sink(self, line: str) -> None: + self._buf.append(line + "\n") + now = time.monotonic() + if now - self._last >= self._interval: + self._last = now + self._flush() + + def _flush(self) -> None: + content = self._base + "".join(self._buf) + # Never shrink the persisted log: a retry whose base predates output + # already streamed and committed would otherwise truncate task.logs + # mid-run, violating strict growth (#195 F1). The task's own final + # ``task.logs = all_logs`` write remains the source of truth. + if len(content) < self._persisted_len: + return + try: + self._task.logs = content + self._db.commit() + self._persisted_len = len(content) + except Exception: # noqa: BLE001 — a log flush must never fail the step + self._db.rollback() + + # ============================================================================ # DRY Helper Functions # ============================================================================ diff --git a/backend/tasks/opentofu_tasks.py b/backend/tasks/opentofu_tasks.py index dc797110..d93f1a28 100644 --- a/backend/tasks/opentofu_tasks.py +++ b/backend/tasks/opentofu_tasks.py @@ -38,6 +38,7 @@ from tasks._task_lookup import fetch_task_or_raise from tasks._tofu_helpers import ( CallbackTask, + TofuLogStreamer, _cleanup_stuck_finalizers, _create_notification, _is_namespace_finalizer_issue, @@ -244,8 +245,12 @@ def run_opentofu_init(self, task_db_id: int, module_id: int, keep_workspace: boo # Get credentials env = get_cloud_credentials_env(project, db) - # Run init - exit_code, logs = engine.run_init(work_dir, env) + # Run init — stream output so logs_full_size grows during the + # run instead of only at completion (#195). + streamer = TofuLogStreamer(task, db) + exit_code, logs = engine.run_init( + work_dir, env, on_output=streamer.begin(""), + ) # Update task task.exit_code = exit_code @@ -424,6 +429,8 @@ def run_opentofu_plan(self, task_db_id: int, module_id: int, keep_workspace: boo db.commit() all_logs = "" + # Stream tofu output into task.logs during the run (#195). + streamer = TofuLogStreamer(task, db) # Check if workspace is initialized - run init if needed needs_reinit, reinit_reason = workspace.needs_reinit(module) @@ -438,7 +445,9 @@ def run_opentofu_plan(self, task_db_id: int, module_id: int, keep_workspace: boo # Need to run init first task.command = "tofu init && tofu plan" all_logs += f"[{_ts()}] --- INIT ---\n" - init_code, init_logs = engine.run_init(work_dir, env) + init_code, init_logs = engine.run_init( + work_dir, env, on_output=streamer.begin(all_logs), + ) all_logs += init_logs if init_code != 0: @@ -458,7 +467,9 @@ def run_opentofu_plan(self, task_db_id: int, module_id: int, keep_workspace: boo # Run plan (saves plan.out to workspace) all_logs += f"[{_ts()}] --- PLAN ---\n" - exit_code, plan_logs = engine.run_plan(work_dir, env) + exit_code, plan_logs = engine.run_plan( + work_dir, env, on_output=streamer.begin(all_logs), + ) all_logs += plan_logs task.exit_code = exit_code @@ -603,6 +614,8 @@ def run_opentofu_apply(self, task_db_id: int, module_id: int, keep_workspace: bo all_logs = "" used_saved_plan = False + # Stream tofu output into task.logs during the run (#195). + streamer = TofuLogStreamer(task, db) def _ensure_workspace_initialized() -> bool: """Ensure providers/modules are installed before reconcile/plan/apply.""" @@ -616,7 +629,9 @@ def _ensure_workspace_initialized() -> bool: all_logs += f"[{_ts()}] === INIT REQUIRED: {reinit_reason} ===\n" all_logs += f"[{_ts()}] --- INIT ---\n" - init_code, init_logs = engine.run_init(work_dir, env) + init_code, init_logs = engine.run_init( + work_dir, env, on_output=streamer.begin(all_logs), + ) all_logs += init_logs if init_code != 0: task.exit_code = init_code @@ -705,7 +720,9 @@ def _ensure_workspace_initialized() -> bool: # Run plan all_logs += f"[{_ts()}] --- PLAN ---\n" - plan_code, plan_logs = engine.run_plan(work_dir, env) + plan_code, plan_logs = engine.run_plan( + work_dir, env, on_output=streamer.begin(all_logs), + ) all_logs += plan_logs if plan_code != 0: task.exit_code = plan_code @@ -725,7 +742,15 @@ def _ensure_workspace_initialized() -> bool: # Apply (uses plan.out which exists either from saved plan or just-created plan) all_logs += f"[{_ts()}] --- APPLY ---\n" - apply_code, apply_logs, outputs = engine.run_apply(work_dir, env, module=module) + apply_code, apply_logs, outputs = engine.run_apply( + work_dir, env, module=module, on_output=streamer.begin(all_logs), + ) + # #195 F1: fold the first apply's output into all_logs *now*, before + # the retry block. This both preserves it (it was previously dropped + # when the retry reassigned apply_logs) and keeps every retry + # streamer.begin(all_logs) base from rewinding task.logs behind the + # output that first apply already streamed and committed. + all_logs += apply_logs # Bounded stale-plan recovery: clear stale plan, re-plan once, then retry apply. # This prevents repeated failures when remote state changed after plan creation. @@ -746,7 +771,9 @@ def _ensure_workspace_initialized() -> bool: else: if reinit_reason: all_logs += f"[{_ts()}] === INIT REQUIRED: {reinit_reason} ===\n" - init_code, init_logs = engine.run_init(work_dir, env) + init_code, init_logs = engine.run_init( + work_dir, env, on_output=streamer.begin(all_logs), + ) all_logs += init_logs if init_code != 0: task.exit_code = init_code @@ -763,7 +790,9 @@ def _ensure_workspace_initialized() -> bool: all_logs += "\n" all_logs += f"[{_ts()}] --- PLAN (RETRY AFTER STALE PLAN) ---\n" - plan_code, plan_logs = engine.run_plan(work_dir, env) + plan_code, plan_logs = engine.run_plan( + work_dir, env, on_output=streamer.begin(all_logs), + ) all_logs += plan_logs if plan_code != 0: task.exit_code = plan_code @@ -781,9 +810,12 @@ def _ensure_workspace_initialized() -> bool: all_logs += "\n" all_logs += f"[{_ts()}] --- APPLY (RETRY) ---\n" - apply_code, apply_logs, outputs = engine.run_apply(work_dir, env, module=module) - - all_logs += apply_logs + apply_code, apply_logs, outputs = engine.run_apply( + work_dir, env, module=module, on_output=streamer.begin(all_logs), + ) + # Fold the retry apply's output in here (the first apply's was + # already folded in above, #195 F1). + all_logs += apply_logs task.exit_code = apply_code task.logs = all_logs @@ -1041,6 +1073,8 @@ def run_opentofu_destroy(self, task_db_id: int, module_id: int, keep_workspace: db.commit() all_logs = "" + # Stream tofu output into task.logs during the run (#195). + streamer = TofuLogStreamer(task, db) # Check if workspace is initialized needs_reinit, reinit_reason = workspace.needs_reinit(module) @@ -1053,7 +1087,9 @@ def run_opentofu_destroy(self, task_db_id: int, module_id: int, keep_workspace: task.command = "tofu init && tofu destroy" # Init all_logs += f"[{_ts()}] --- INIT ---\n" - init_code, init_logs = engine.run_init(work_dir, env) + init_code, init_logs = engine.run_init( + work_dir, env, on_output=streamer.begin(all_logs), + ) all_logs += init_logs if init_code != 0: task.exit_code = init_code @@ -1073,7 +1109,8 @@ def run_opentofu_destroy(self, task_db_id: int, module_id: int, keep_workspace: all_logs += f"[{_ts()}] --- DESTROY ---\n" timeout = engine.get_destroy_timeout(module) destroy_code, destroy_logs = engine.run_destroy_with_retry( - work_dir, env, module=module, timeout=timeout + work_dir, env, module=module, timeout=timeout, + on_output=streamer.begin(all_logs), ) all_logs += destroy_logs @@ -1090,7 +1127,8 @@ def run_opentofu_destroy(self, task_db_id: int, module_id: int, keep_workspace: logger.info("Retrying destroy after finalizer cleanup") all_logs += f"\n[{_ts()}] --- DESTROY RETRY ---\n" retry_code, retry_logs = engine.run_destroy_with_retry( - work_dir, env, module=module, timeout=300 # Shorter timeout for retry + work_dir, env, module=module, timeout=300, # Shorter timeout for retry + on_output=streamer.begin(all_logs), ) all_logs += f"{retry_logs}\n" diff --git a/backend/tests/component/test_opentofu_runtime_streaming.py b/backend/tests/component/test_opentofu_runtime_streaming.py new file mode 100644 index 00000000..f82af25d --- /dev/null +++ b/backend/tests/component/test_opentofu_runtime_streaming.py @@ -0,0 +1,470 @@ +"""#195: OpenTofu runtime streams subprocess output incrementally. + +Before this fix ``run_init``/``run_plan``/``run_apply``/``run_destroy`` used a +blocking ``subprocess.run(capture_output=True)`` that returned the entire log +only when the process exited, so a caller had nothing to persist until the very +end. These tests prove the streaming path delivers each line *while the process +is still running*, and that the ``run_*`` methods route through it only when an +``on_output`` callback is supplied (the classic blocking capture is preserved +otherwise, which the existing test_opentofu_runtime.py suite locks). +""" + +import os +import subprocess +import sys + +import pytest + +import services.execution.opentofu_runtime as otr +from services.execution.opentofu_runtime import OpenTofuRuntime, _stream_subprocess + + +@pytest.mark.component +class TestStreamSubprocess: + def test_delivers_each_line_before_process_finishes(self, tmp_path): + """A handshake proves lines arrive live, not buffered until exit. + + The child prints ``line-1`` then blocks on a sentinel that the test's + ``on_output`` writes only when it *receives* ``line-1``; only then does + the child print ``line-2``. The blocking loop runs FAR longer than the + watchdog (``timeout`` below), so a buffered implementation — the #195 bug, + where ``on_output`` fires only after the process exits — cannot + self-release: the sentinel never appears, the child deadlocks, and the + watchdog kills it → ``_stream_subprocess`` raises ``TimeoutExpired`` and + this test ERRORS. Genuine streaming releases the child within + milliseconds, so the call returns well under the watchdog. Both the raise + AND the timing bound below make the distinction non-vacuous — a fully + buffered impl fails, proven by mutation. + """ + import time + + go = tmp_path / "go" + # ~300s of blocking: >> the 8s watchdog, so a buffered impl MUST deadlock + # rather than self-release before the watchdog fires. + script = ( + "import sys, time, pathlib\n" + "print('line-1', flush=True)\n" + "sentinel = pathlib.Path(sys.argv[1])\n" + "for _ in range(6000):\n" + " if sentinel.exists():\n" + " break\n" + " time.sleep(0.05)\n" + "print('line-2', flush=True)\n" + ) + + received: list[str] = [] + + def on_output(line: str) -> None: + received.append(line) + if line == "line-1": + go.write_text("go") # unblock the child only after we SEE line-1 + + started = time.monotonic() + code, output = _stream_subprocess( + [sys.executable, "-c", script, str(go)], + cwd=str(tmp_path), + env=dict(os.environ), + timeout=8, + on_output=on_output, + ) + elapsed = time.monotonic() - started + + assert code == 0 + # line-2 was printed *only* because on_output saw line-1 and released it. + assert received == ["line-1", "line-2"] + assert "line-1" in output and "line-2" in output + # Live streaming releases the child in ms; a buffered impl would deadlock + # and blow the 8s watchdog. The timing bound makes that explicit. + assert elapsed < 5, f"took {elapsed:.1f}s — output looks buffered, not streamed" + + def test_returns_combined_output_and_exit_code(self, tmp_path): + script = "import sys; print('out'); print('err', file=sys.stderr); sys.exit(3)" + seen: list[str] = [] + code, output = _stream_subprocess( + [sys.executable, "-c", script], + cwd=str(tmp_path), + env=dict(os.environ), + timeout=20, + on_output=seen.append, + ) + assert code == 3 + # stderr is merged into stdout so both surface in the log and the sink. + assert "out" in output and "err" in output + assert "out" in seen and "err" in seen + + def test_timeout_kills_and_raises_with_partial_output(self, tmp_path): + script = ( + "import sys, time\n" + "print('before-hang', flush=True)\n" + "time.sleep(30)\n" + ) + seen: list[str] = [] + with pytest.raises(subprocess.TimeoutExpired) as exc: + _stream_subprocess( + [sys.executable, "-c", script], + cwd=str(tmp_path), + env=dict(os.environ), + timeout=1, + on_output=seen.append, + ) + # The partial output produced before the hang is carried on the + # exception, matching subprocess.run(...).stdout semantics the run_* + # timeout branches rely on. + assert "before-hang" in (exc.value.output or "") + assert seen == ["before-hang"] + + def test_timeout_enforced_when_grandchild_holds_the_pipe(self, tmp_path): + """B-1 reproduction: a descendant inheriting stdout must not defeat the timeout. + + The child spawns a grandchild that INHERITS the stdout pipe (no + ``stdout=`` redirect) and then the child exits. The grandchild lives far + longer than the timeout, so the stdout pipe never reaches EOF on the + "kill the direct child" path — ``for line in proc.stdout`` blocks on a + write end still held open by the grandchild. A correct implementation + must still enforce the deadline (kill + close the read end) and raise + ``TimeoutExpired`` at ~the deadline, NOT hang for the grandchild's whole + lifetime. Before the fix this returned ~6x over the deadline (or never). + """ + import time + + # Child exits immediately after spawning a grandchild that inherits fd 1 + # (stdout) and sleeps FAR longer than the 2s timeout below, so the pipe + # stays open. The grandchild lifetime is bounded so a broken impl (which + # would block on it) still eventually frees CI rather than hanging. + script = ( + "import subprocess, sys, time\n" + "subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'])\n" + "print('child-exiting', flush=True)\n" + # child returns here; grandchild keeps the stdout write end open + ) + seen: list[str] = [] + started = time.monotonic() + with pytest.raises(subprocess.TimeoutExpired) as exc: + _stream_subprocess( + [sys.executable, "-c", script], + cwd=str(tmp_path), + env=dict(os.environ), + timeout=2, + on_output=seen.append, + ) + elapsed = time.monotonic() - started + + # Enforced at ~the deadline, not at the grandchild's 30s lifetime. + assert elapsed < 6, f"took {elapsed:.1f}s — timeout not enforced past a grandchild-held pipe" + # Partial output streamed before the deadline is preserved on the exception. + assert "child-exiting" in (exc.value.output or "") + assert "child-exiting" in seen + + def test_exception_in_read_path_kills_the_child(self, tmp_path, monkeypatch): + """B-2 reproduction: an exception raised in the read path must kill the child. + + Celery delivers ``SoftTimeLimitExceeded`` by raising it in the worker's + main thread — which, mid-run, is blocked inside the pipe read (here, + ``select``). That is NOT an ``on_output`` error (those are guarded and + intentionally swallowed), so it escapes the read loop. If the helper does + not kill the child on that path, a live ``tofu apply`` is orphaned and + keeps mutating cloud state after the task is marked failed. + + We reproduce it faithfully: a real long-lived child, and the exception + raised from ``select`` on the *second* wait — exactly where the signal + lands while the reader blocks waiting for more output — after the first + line has already streamed. The child must be *killed* (SIGKILL, promptly) + when it surfaces — not merely reaped 30s later by ``with Popen`` waiting + for it to finish sleeping, which is exactly the orphaned-tofu bug. + """ + import time + + from celery.exceptions import SoftTimeLimitExceeded + + real_select = otr.select.select + state = {"n": 0} + + def flaky_select(rlist, wlist, xlist, timeout=None): + state["n"] += 1 + if state["n"] == 1: + return real_select(rlist, wlist, xlist, timeout) # deliver 'alive' + # Second wait: the child is now sleeping and the reader is blocked — + # exactly when Celery's soft-time-limit signal fires in this thread. + raise SoftTimeLimitExceeded("soft time limit exceeded") + + monkeypatch.setattr(otr.select, "select", flaky_select) + + procs: list[subprocess.Popen] = [] + real_popen = subprocess.Popen + + def spy_popen(*args, **kwargs): + proc = real_popen(*args, **kwargs) + procs.append(proc) + return proc + + monkeypatch.setattr(otr.subprocess, "Popen", spy_popen) + + # A real child that would live for 30s if left orphaned. + script = ( + "import sys, time\n" + "print('alive', flush=True)\n" + "time.sleep(30)\n" + ) + seen: list[str] = [] + started = time.monotonic() + with pytest.raises(SoftTimeLimitExceeded): + _stream_subprocess( + [sys.executable, "-c", script], + cwd=str(tmp_path), + env=dict(os.environ), + timeout=30, # long; the exception fires long before the deadline + on_output=seen.append, + ) + elapsed = time.monotonic() - started + + assert seen == ["alive"] # first line streamed before the raise + assert procs, "Popen was not invoked" + proc = procs[0] + # The child must have been SIGKILLed, not left to finish its 30s sleep. + # A negative returncode is death-by-signal; without the kill the child + # would exit 0 (and only after ~30s), which the timing bound also catches. + assert proc.returncode is not None and proc.returncode < 0, ( + f"child not killed by signal (returncode={proc.returncode}) — orphaned tofu" + ) + assert elapsed < 5, f"took {elapsed:.1f}s — child was reaped, not killed" + + def test_eof_with_child_still_alive_enforces_timeout(self, tmp_path, monkeypatch): + """MINOR 1 reproduction: real pipe EOF while the child is still alive must + still honour the deadline — not fall into an UNBOUNDED ``proc.wait()``. + + The child prints a line then closes BOTH fd 1 and fd 2 (the merged + stdout/stderr write end ⇒ genuine EOF, ``os.read`` returns ``b""``) and + then sleeps far longer than the timeout while still alive. The read loop + breaks on EOF with ``timed_out=False``; a bare ``proc.wait()`` there would + block for the child's whole 60s lifetime, silently defeating the timeout + the docstring promises. The fix bounds that wait by the remaining deadline, + kills the child, and raises ``TimeoutExpired`` at ~the deadline. + """ + import time + + script = ( + "import os, sys, time\n" + "sys.stdout.write('bye\\n'); sys.stdout.flush()\n" + "os.close(1)\n" # close stdout write end + "os.close(2)\n" # close the merged stderr write end ⇒ real EOF + "time.sleep(60)\n" # child stays alive far past the 2s timeout + ) + procs: list[subprocess.Popen] = [] + real_popen = subprocess.Popen + + def spy_popen(*args, **kwargs): + proc = real_popen(*args, **kwargs) + procs.append(proc) + return proc + + monkeypatch.setattr(otr.subprocess, "Popen", spy_popen) + + seen: list[str] = [] + started = time.monotonic() + with pytest.raises(subprocess.TimeoutExpired) as exc: + _stream_subprocess( + [sys.executable, "-c", script], + cwd=str(tmp_path), + env=dict(os.environ), + timeout=2, + on_output=seen.append, + ) + elapsed = time.monotonic() - started + + # Enforced at ~the deadline, NOT at the child's 60s lifetime. + assert elapsed < 6, f"took {elapsed:.1f}s — unbounded proc.wait() after EOF, timeout not enforced" + # Partial output streamed before EOF is preserved on the exception. + assert "bye" in (exc.value.output or "") + assert "bye" in seen + # The still-alive child must be SIGKILLed, not left to finish its sleep. + proc = procs[0] + assert proc.returncode is not None and proc.returncode < 0, ( + f"child not killed after EOF-alive timeout (returncode={proc.returncode})" + ) + + def test_timeout_with_partial_multibyte_raises_timeout_not_unicode(self, tmp_path): + """MINOR 2 reproduction: a deadline hit while a partial multibyte char is + buffered must raise ``TimeoutExpired`` — NOT ``UnicodeDecodeError``. + + The child emits a full line, then the first 2 of the 3 UTF-8 bytes of + ``€`` (U+20AC = ``e2 82 ac``), then hangs with the incomplete sequence + buffered inside the incremental decoder. If the ``decoder.decode(b"", + final=True)`` flush runs BEFORE the timeout check, ``final=True`` raises + ``UnicodeDecodeError`` on the truncated sequence — which surfaces in place + of ``TimeoutExpired`` and bypasses ``run_apply``/``run_destroy``'s graceful + ``except subprocess.TimeoutExpired`` branch. The fix raises TimeoutExpired + before that final flush. + """ + script = ( + "import os, time\n" + "os.write(1, b'line1\\n')\n" + "os.write(1, b'\\xe2\\x82')\n" # first 2 of 3 bytes of U+20AC EURO SIGN + "time.sleep(30)\n" + ) + seen: list[str] = [] + with pytest.raises(subprocess.TimeoutExpired) as exc: + _stream_subprocess( + [sys.executable, "-c", script], + cwd=str(tmp_path), + env=dict(os.environ), + timeout=2, + on_output=seen.append, + ) + # The complete line streamed before the hang is preserved; the dangling + # partial byte pair is simply not flushed (would raise on final decode). + assert "line1" in (exc.value.output or "") + assert seen == ["line1"] + + def test_streamed_newlines_match_text_mode(self, tmp_path): + """MINOR 3: streamed decode must apply universal-newline translation like + the no-callback ``subprocess.run(text=True)`` path. + + ``a\\r\\nb\\rc\\nd`` (CRLF, lone CR, LF) must normalize to ``a\\nb\\nc\\nd`` + in BOTH the returned combined output and the delivered lines — otherwise + streamed logs diverge from non-streamed runs. + """ + script = ( + "import os\n" + "os.write(1, b'a\\r\\nb\\rc\\nd')\n" + ) + seen: list[str] = [] + code, output = _stream_subprocess( + [sys.executable, "-c", script], + cwd=str(tmp_path), + env=dict(os.environ), + timeout=20, + on_output=seen.append, + ) + assert code == 0 + assert output == "a\nb\nc\nd" + assert seen == ["a", "b", "c", "d"] + + def test_crlf_split_across_reads_normalizes_to_single_lf(self, tmp_path, monkeypatch): + """MINOR 3 edge: a CRLF split across two reads (``…\\r`` | ``\\n…``) must + become ONE ``\\n``, not two. + + Normalizing per-chunk would turn the split ``\\r``+``\\n`` into ``\\n\\n``. + The fix normalizes on the accumulated stream (holding a trailing ``\\r`` + until the next read arrives). Reads are injected deterministically so the + boundary is exactly on the CRLF. + """ + reads = [b"a\r", b"\nb\rc\n", b""] # CRLF straddles read #1/#2; lone CR mid-#2 + it = iter(reads) + + class _FakeStdout: + def fileno(self): + return 4321 + + class _FakeProc: + returncode = 0 + stdout = _FakeStdout() + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def wait(self, timeout=None): + return 0 + + def kill(self): + pass + + monkeypatch.setattr(otr.subprocess, "Popen", lambda *a, **k: _FakeProc()) + monkeypatch.setattr(otr.select, "select", lambda r, w, x, t: (list(r), [], [])) + monkeypatch.setattr(otr.os, "read", lambda fd, n: next(it)) + + seen: list[str] = [] + code, output = _stream_subprocess( + ["x"], + cwd=str(tmp_path), + env={}, + timeout=10, + on_output=seen.append, + ) + assert code == 0 + # Split CRLF → single '\n'; lone '\r' → '\n'. No spurious blank line. + assert output == "a\nb\nc\n" + assert seen == ["a", "b", "c"] + + +@pytest.mark.component +class TestRunMethodsRouteThroughStreaming: + @staticmethod + def _runtime(monkeypatch): + # OpenTofuRuntime only needs a DB session for workspace ops, not for the + # subprocess-running methods under test here. Stub the system-defaults + # gate so construction doesn't need a real DB. + monkeypatch.setattr( + otr, "check_required_configured", + lambda _db: {"all_configured": True, "missing": []}, + ) + return OpenTofuRuntime(db=None) + + def _patch_stream(self, monkeypatch): + calls: dict = {} + + def fake_stream(cmd, *, cwd, env, timeout, on_output): + calls["cmd"] = cmd + calls["on_output"] = on_output + on_output("streamed-1") + on_output("streamed-2") + return 0, "streamed-1\nstreamed-2\n" + + monkeypatch.setattr(otr, "_stream_subprocess", fake_stream) + return calls + + def test_run_plan_streams_when_on_output_given(self, monkeypatch): + calls = self._patch_stream(monkeypatch) + runtime = self._runtime(monkeypatch) + seen: list[str] = [] + + code, output = runtime.run_plan("/tmp/w", {}, on_output=seen.append) + + assert code == 0 + assert seen == ["streamed-1", "streamed-2"] + assert "streamed-1" in output and "streamed-2" in output + assert "plan" in calls["cmd"] + + def test_run_init_streams_when_on_output_given(self, monkeypatch): + calls = self._patch_stream(monkeypatch) + runtime = self._runtime(monkeypatch) + seen: list[str] = [] + + code, output = runtime.run_init("/tmp/w", {}, on_output=seen.append) + + assert code == 0 + assert seen == ["streamed-1", "streamed-2"] + assert "init" in calls["cmd"] + + def test_run_apply_streams_when_on_output_given(self, monkeypatch): + self._patch_stream(monkeypatch) + runtime = self._runtime(monkeypatch) + # apply captures outputs on success; stub that out. + monkeypatch.setattr(runtime, "_capture_outputs", lambda *a, **k: {}) + seen: list[str] = [] + + code, output, _outputs = runtime.run_apply("/tmp/w", {}, on_output=seen.append) + + assert code == 0 + assert seen == ["streamed-1", "streamed-2"] + + def test_blocking_path_used_when_no_on_output(self, monkeypatch): + """Without on_output the classic subprocess.run capture is used, NOT the + streaming helper — preserving legacy behaviour and the existing suite.""" + def _boom(*a, **k): + raise AssertionError("_stream_subprocess must not run without on_output") + + monkeypatch.setattr(otr, "_stream_subprocess", _boom) + + completed = subprocess.CompletedProcess( + args=["tofu", "plan"], returncode=0, stdout="blocking-out", stderr="", + ) + monkeypatch.setattr(otr.subprocess, "run", lambda *a, **k: completed) + + runtime = self._runtime(monkeypatch) + code, output = runtime.run_plan("/tmp/w", {}) + + assert code == 0 + assert output == "blocking-out" diff --git a/backend/tests/component/test_opentofu_tasks.py b/backend/tests/component/test_opentofu_tasks.py index 81e89fd0..c29fe5bb 100644 --- a/backend/tests/component/test_opentofu_tasks.py +++ b/backend/tests/component/test_opentofu_tasks.py @@ -543,7 +543,12 @@ def test_apply_reconcile_imports_clear_saved_plan_before_plan_reuse( assert result["success"] is True assert mock_workspace.clear_plan.call_count == 2 assert all(call.args == (module,) for call in mock_workspace.clear_plan.call_args_list) - mock_engine.run_plan.assert_called_once_with("/tmp/workspace", {}) + # run_plan is now called with an additive on_output= streaming callback + # (#195); assert on the positional args and ignore the sink kwarg. + mock_engine.run_plan.assert_called_once() + plan_args, plan_kwargs = mock_engine.run_plan.call_args + assert plan_args == ("/tmp/workspace", {}) + assert set(plan_kwargs) <= {"on_output"} db.refresh(task) assert "=== RECONCILIATION INVALIDATED SAVED PLAN ===" in (task.logs or "") @@ -719,3 +724,148 @@ def test_roks_register_enqueues_scan(self, db): enqueue_cluster_scan(cluster.id) mock_scan_task.delay.assert_called_once_with(55) + + +# ── #195: Incremental log streaming during a run ───────────────────────────── + +class TestOpenTofuTaskLogStreaming: + """logs_full_size must grow while the task is in_progress, not jump from 0 + to its final value only at completion (issue #195).""" + + @patch(f"{_MOD}.update_project_counts") + @patch(f"{_MOD}.create_deployment_record") + @patch(f"{_MOD}._notify_task_started") + @patch(f"{_MOD}.module_lock") + @patch(f"{_MOD}.get_cloud_credentials_env", return_value={}) + @patch(f"{_MOD}.check_dependencies", return_value=(True, [])) + @patch(f"{_MOD}.OpenTofuRuntime") + @patch(f"{_MOD}.get_db_context") + @patch(f"{_MOD}.datetime") + def test_plan_persists_growing_logs_before_completion( + self, mock_dt, mock_db_ctx, mock_runtime_cls, _mock_deps, _mock_creds, + mock_lock, _mock_notify, _mock_deploy, _mock_counts, db, + ): + from sqlalchemy import func + + from tasks._tofu_helpers import TofuLogStreamer + + naive_now = datetime.utcnow() + mock_dt.now.return_value = naive_now + mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw) + + project = _make_project(db) + lib = _make_library(db) + module = _make_module(db, project, lib, status="initialized") + task = _make_task(db, project, module, "plan") + db.commit() + + mock_db_ctx.return_value.__enter__ = MagicMock(return_value=db) + mock_db_ctx.return_value.__exit__ = MagicMock(return_value=False) + mock_lock.return_value.__enter__ = MagicMock( + return_value=ModuleLock(module_id=module.id, task_id=task.id, fence_token=0), + ) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + # As each line streams, record what a concurrent GET /api/tasks/{id} + # would observe: the persisted logs_full_size (SQL length, exactly the + # detail endpoint's computation) and the task status. This is measured + # from the DB, not the in-memory buffer, so it proves the flush. + observed_sizes: list[int] = [] + observed_status: list[str] = [] + + def streaming_run_plan(work_dir, env, on_output=None, timeout=None): + assert on_output is not None, "plan task must pass an on_output sink (#195)" + for i in range(1, 6): + on_output(f"tofu progress line {i}") + size = db.query(func.length(TaskModel.logs)).filter( + TaskModel.id == task.id + ).scalar() + status = db.query(TaskModel.status).filter( + TaskModel.id == task.id + ).scalar() + observed_sizes.append(size or 0) + observed_status.append(status) + return (0, "".join(f"tofu progress line {i}\n" for i in range(1, 6))) + + mock_engine = MagicMock() + mock_engine.prepare_persistent_workspace.return_value = "/tmp/workspace" + mock_engine.run_plan.side_effect = streaming_run_plan + mock_runtime_cls.return_value = mock_engine + + mock_workspace = MagicMock() + mock_workspace.is_initialized.return_value = True + mock_workspace.needs_reinit.return_value = (False, None) + + # Force interval=0 so every streamed line flushes — otherwise the 2s + # throttle would coalesce this fast test's lines into a single flush. + def _fast_streamer(t, d, **_kw): + return TofuLogStreamer(t, d, interval=0) + + with patch("services.workspace_manager.WorkspaceManager", return_value=mock_workspace), \ + patch(f"{_MOD}.TofuLogStreamer", _fast_streamer): + from tasks.opentofu_tasks import run_opentofu_plan + result = run_opentofu_plan(task.id, module.id) + + assert result["success"] is True + + # (a) The bug reproduced: without streaming every one of these would be + # 0. Instead the size is non-zero from the first line and grows. + assert len(observed_sizes) == 5 + assert all(s > 0 for s in observed_sizes), observed_sizes + assert observed_sizes == sorted(observed_sizes) + assert len(set(observed_sizes)) == 5, f"not strictly growing: {observed_sizes}" + # (b) …and this all happened while the task was still running. + assert observed_status == ["in_progress"] * 5 + + # Final state intact: complete log persisted, size == detail's len(). + db.refresh(task) + assert task.status == "completed" + assert "tofu progress line 5" in task.logs + assert len(task.logs) >= observed_sizes[-1] + + @patch(f"{_MOD}.update_project_counts") + @patch(f"{_MOD}.create_deployment_record") + @patch(f"{_MOD}._notify_task_started") + @patch(f"{_MOD}.module_lock") + @patch(f"{_MOD}.get_cloud_credentials_env", return_value={}) + @patch(f"{_MOD}.check_dependencies", return_value=(True, [])) + @patch(f"{_MOD}.OpenTofuRuntime") + @patch(f"{_MOD}.get_db_context") + @patch(f"{_MOD}.datetime") + def test_plan_passes_on_output_sink_to_runtime( + self, mock_dt, mock_db_ctx, mock_runtime_cls, _mock_deps, _mock_creds, + mock_lock, _mock_notify, _mock_deploy, _mock_counts, db, + ): + """Guard: the plan task must hand run_plan a callable on_output sink.""" + naive_now = datetime.utcnow() + mock_dt.now.return_value = naive_now + mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw) + + project = _make_project(db) + lib = _make_library(db) + module = _make_module(db, project, lib, status="initialized") + task = _make_task(db, project, module, "plan") + db.commit() + + mock_db_ctx.return_value.__enter__ = MagicMock(return_value=db) + mock_db_ctx.return_value.__exit__ = MagicMock(return_value=False) + mock_lock.return_value.__enter__ = MagicMock( + return_value=ModuleLock(module_id=module.id, task_id=task.id, fence_token=0), + ) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + mock_engine = MagicMock() + mock_engine.prepare_persistent_workspace.return_value = "/tmp/workspace" + mock_engine.run_plan.return_value = (0, "Plan: 1 to add") + mock_runtime_cls.return_value = mock_engine + + mock_workspace = MagicMock() + mock_workspace.is_initialized.return_value = True + mock_workspace.needs_reinit.return_value = (False, None) + + with patch("services.workspace_manager.WorkspaceManager", return_value=mock_workspace): + from tasks.opentofu_tasks import run_opentofu_plan + run_opentofu_plan(task.id, module.id) + + _args, kwargs = mock_engine.run_plan.call_args + assert callable(kwargs.get("on_output")) diff --git a/backend/tests/integration/test_routes_tasks.py b/backend/tests/integration/test_routes_tasks.py index 69b85d9a..ca1a4c1b 100644 --- a/backend/tests/integration/test_routes_tasks.py +++ b/backend/tests/integration/test_routes_tasks.py @@ -49,6 +49,58 @@ def test_list_tasks_with_status_filter(self, client, admin_headers, sample_user, assert task["status"] == "completed" +class TestTaskListLogFields: + """#195 (second defect): the list endpoint must expose logs_full_size so a + client enumerating a module's tasks can tell empty from populated. It used + to be omitted entirely (surfaced as null), while the detail endpoint had it.""" + + def test_list_includes_logs_full_size_matching_detail( + self, client, admin_headers, sample_user, sample_project, db + ): + from tests.factories import TaskFactory + + log_body = "\n".join(f"[04:42:{i:02d}] line {i}" for i in range(20)) + task = TaskFactory( + db, project=sample_project, task_type="plan", + status="completed", logs=log_body, + ) + db.commit() + + listed = client.get( + f"/api/tasks?project_id={sample_project.id}", headers=admin_headers + ).json() + row = next(t for t in listed["tasks"] if t["id"] == task.id) + + # The field is present (not omitted) and equals the true log size. + assert "logs_full_size" in row + assert row["logs_full_size"] == len(log_body) + assert "logs_truncated" in row + + # …and it matches what the detail endpoint reports for the same task. + detail = client.get(f"/api/tasks/{task.id}", headers=admin_headers).json() + assert detail["logs_full_size"] == row["logs_full_size"] + + def test_list_reports_zero_for_task_without_logs( + self, client, admin_headers, sample_user, sample_project, db + ): + from tests.factories import TaskFactory + + task = TaskFactory( + db, project=sample_project, task_type="plan", status="queued", logs=None + ) + db.commit() + + listed = client.get( + f"/api/tasks?project_id={sample_project.id}", headers=admin_headers + ).json() + row = next(t for t in listed["tasks"] if t["id"] == task.id) + + # NULL logs → 0 (matching the detail endpoint), not null/omitted. + assert row["logs_full_size"] == 0 + detail = client.get(f"/api/tasks/{task.id}", headers=admin_headers).json() + assert detail["logs_full_size"] == 0 + + class TestTaskDetail: """GET /api/tasks/{id}.""" diff --git a/backend/tests/unit/test_tofu_log_streamer.py b/backend/tests/unit/test_tofu_log_streamer.py new file mode 100644 index 00000000..ae4a4185 --- /dev/null +++ b/backend/tests/unit/test_tofu_log_streamer.py @@ -0,0 +1,113 @@ +"""Unit tests for TofuLogStreamer — the incremental task.logs flusher (#195). + +The streamer turns the runtime's per-line ``on_output`` callback into throttled +writes of ``task.logs`` so ``logs_full_size`` grows during a run. These tests +lock: base+buffer composition, growth across lines, interval throttling, the +force-flush of the first line, and that a failed commit never propagates. +""" + +from unittest.mock import MagicMock + +import pytest + +from tasks._tofu_helpers import TofuLogStreamer + + +@pytest.mark.unit +class TestTofuLogStreamer: + def test_first_line_flushes_and_prepends_base(self): + task = MagicMock() + db = MagicMock() + streamer = TofuLogStreamer(task, db, interval=0) + + sink = streamer.begin("HEADER\n") + sink("first line") + + assert task.logs == "HEADER\nfirst line\n" + db.commit.assert_called() + + def test_logs_grow_across_lines(self): + task = MagicMock() + db = MagicMock() + streamer = TofuLogStreamer(task, db, interval=0) # every line flushes + + sink = streamer.begin("BASE\n") + sizes = [] + for i in range(1, 6): + sink(f"line {i}") + sizes.append(len(task.logs)) + + # Strictly increasing — the whole point of #195: not 0 until the end. + assert sizes == sorted(sizes) + assert len(set(sizes)) == len(sizes) + assert db.commit.call_count == 5 + assert task.logs.startswith("BASE\n") + assert "line 5" in task.logs + + def test_interval_throttles_intermediate_lines(self, monkeypatch): + task = MagicMock() + db = MagicMock() + # begin() sets last=0.0 so the first line always flushes; a large + # interval then suppresses the second line arriving within the window. + monkeypatch.setattr("time.monotonic", MagicMock(side_effect=[1000.0, 1000.1])) + streamer = TofuLogStreamer(task, db, interval=100) + + sink = streamer.begin("") + sink("first") # 1000.0 - 0.0 >= 100 -> flush + sink("second") # 1000.1 - 1000.0 < 100 -> throttled + + assert db.commit.call_count == 1 + # Persisted content still reflects only the flushed line; the throttled + # line is folded into task.logs by the task's final all_logs write. + assert task.logs == "first\n" + + def test_begin_resets_base_and_buffer(self): + task = MagicMock() + db = MagicMock() + streamer = TofuLogStreamer(task, db, interval=0) + + streamer.begin("STEP-A\n")("a-line") + assert task.logs == "STEP-A\na-line\n" + + # A new step starts from a fresh base; the previous step's buffer is gone. + streamer.begin("STEP-A\na-line\nSTEP-B\n")("b-line") + assert task.logs == "STEP-A\na-line\nSTEP-B\nb-line\n" + + def test_commit_failure_is_swallowed_and_rolls_back(self): + task = MagicMock() + db = MagicMock() + db.commit.side_effect = RuntimeError("db gone") + streamer = TofuLogStreamer(task, db, interval=0) + + sink = streamer.begin("H\n") + sink("a line") # must not raise + + db.rollback.assert_called_once() + + def test_retry_with_shorter_base_never_rewinds_persisted_logs(self): + """#195 F1: a step whose base is shorter than what was already persisted + must NOT truncate task.logs mid-run (strict-growth invariant). + + Reproduces the stale-plan retry path shape: the first apply streams and + commits output onto ``task.logs``; a later ``begin(base)`` is then handed + a base that omits that output. The persisted log must hold at its + high-water mark instead of shrinking. + """ + task = MagicMock() + db = MagicMock() + streamer = TofuLogStreamer(task, db, interval=0) + + # First step persists a long log. + sink = streamer.begin("HEADER\n--- APPLY ---\n") + sink("apply-line-1") + sink("apply-line-2") + high_water = len(task.logs) + assert high_water > 0 + + # A retry begins from a base that DROPS the first apply's streamed output + # (the F1 bug: begin(all_logs) before all_logs += apply_logs). + sink2 = streamer.begin("HEADER\n") + sink2("retry-line") + + # The persisted log must never have shrunk below the high-water mark. + assert len(task.logs) >= high_water, "task.logs rewound on the retry path"