Feat: stream live apply logs to audit API

CHANGELOG.md

@@ -5,6 +5,17 @@ All notable changes to this project will be documented in this file.
 The format follows [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## \[0.11.7\] - 2026-09-03
+
+### Added
+
+- Stream `terraform apply` output to the audit API as log chunks in real time,
+  enabling the tfaudit frontend to poll and display live apply output.
+  Chunks are flushed every 10 lines or 5 seconds (whichever comes first) via
+  a new `POST /log_chunk` IAM-authenticated endpoint on `audit_api_url`. Chunks
+  are posted to every URL in `audit_api_url`, matching the multi-URL behavior
+  added in `0.11.4`.
+
 ## \[0.11.6\] - 2026-08-31
 
 ### Fixed

terrawrap/utils/cli.py

@@ -11,7 +11,7 @@
 import tempfile
 import time
 from enum import Enum
-from typing import List, Optional, Tuple, Union
+from typing import Callable, List, Optional, Tuple, Union
 from urllib.parse import urlparse
 
 import requests
@@ -46,7 +46,10 @@
 ]
 AUDIT_POST_PATH = "/audit_info"
 AUDIT_UPDATE_PATH = "/update_audit_info"
+LOG_CHUNK_POST_PATH = "/log_chunk"
 OUTPUT_COMPRESSION_THRESHOLD = 5 * 1024 * 1024
+CHUNK_LINE_COUNT = 10
+CHUNK_FLUSH_INTERVAL = 5.0
 
 
 class Status(str, Enum):
@@ -109,6 +112,24 @@ def execute_command(
     else:
         logger.info("No audit_api_url provided")
 
+    should_stream = bool(audit_api_urls and kwargs["cwd"] and ("apply" in args or "destroy" in args))
+    chunk_seq = 0
+
+    def _chunk_callback(content: str) -> None:
+        nonlocal chunk_seq
+        for url in audit_api_urls:
+            try:
+                _post_log_chunk(
+                    audit_api_url=url,
+                    path=kwargs["cwd"],
+                    start_time=start_time,
+                    sequence=chunk_seq,
+                    content=content,
+                )
+            except Exception as exc:  # pylint: disable=broad-except
+                logger.warning("Failed to post log chunk %d to %s: %s", chunk_seq, url, exc)
+        chunk_seq += 1
+
     jitter = Jitter()
     time_passed = 0
     exit_code = 0
@@ -120,6 +141,7 @@ def execute_command(
             capture_stderr,
             print_command,
             *pargs,
+            on_chunk=_chunk_callback if should_stream else None,
             **kwargs,
         )
 
@@ -170,6 +192,7 @@ def _execute_command(
     capture_stderr: bool,
     print_command: bool,
     *pargs,
+    on_chunk: Optional[Callable[[str], None]] = None,
     **kwargs,
 ) -> Tuple[int, List[str]]:
     """
@@ -179,6 +202,7 @@ def _execute_command(
     :param capture_stderr: True if stderr should be captured. Defaults to True.
     :param print_command: True if the command should be printed before executing. Defaults to False.
     :param pargs: Any additional positional arguments to Popen.
+    :param on_chunk: Optional callback invoked with buffered output chunks for live streaming.
     :param kwargs: Any additional keyword arguments to Popen.
     :return: A tuple of the exit code and output of the command.
     """
@@ -193,6 +217,10 @@ def _execute_command(
         # pylint: disable=consider-using-with
         process = subprocess.Popen(args, *pargs, **kwargs)
 
+        buf: List[str] = []
+        line_count = 0
+        last_flush = time.time()
+
         # Terraform's output (e.g. the box-drawing characters in its error
         # formatting) contains multi-byte UTF-8 sequences. Decoding one raw
         # byte at a time would mangle those into replacement characters, so
@@ -209,9 +237,23 @@ def _execute_command(
             if print_output and output:
                 print(output, end="", flush=True)
 
+            if on_chunk and output:
+                buf.append(output)
+                if output == "\n":
+                    line_count += 1
+                now = time.time()
+                if line_count >= CHUNK_LINE_COUNT or now - last_flush >= CHUNK_FLUSH_INTERVAL:
+                    on_chunk("".join(buf))
+                    buf = []
+                    line_count = 0
+                    last_flush = now
+
             if is_eof:
                 break
 
+        if on_chunk and buf:
+            on_chunk("".join(buf))
+
         exit_code = process.poll()
 
         stdout_read.seek(0)
@@ -294,3 +336,34 @@ def _post_audit_info(
         logger.info("Successfully posted data to provided url: %s", audit_api_url)
     except requests.exceptions.RequestException:
         logger.error("Unable to post data to provided url: %s", audit_api_url)
+
+
+def _post_log_chunk(
+    audit_api_url: str,
+    path: str,
+    start_time: int,
+    sequence: int,
+    content: str,
+) -> None:
+    """POST a single log chunk to the audit API during an apply."""
+    root = get_git_root(path)
+    directory = path.replace(root, "")
+
+    auth = BotoAWSRequestsAuth(
+        aws_host=urlparse(audit_api_url).hostname,
+        aws_region="us-west-2",
+        aws_service="execute-api",
+    )
+
+    response = requests.post(
+        url=audit_api_url + LOG_CHUNK_POST_PATH,
+        auth=auth,
+        json={
+            "directory": directory,
+            "start_time": start_time,
+            "sequence": sequence,
+            "content": content,
+        },
+        timeout=10,
+    )
+    response.raise_for_status()

terrawrap/version.py

@@ -1,4 +1,4 @@
 """Place of record for the package version"""
 
-__version__ = "0.11.6"
+__version__ = "0.11.7"
 __git_hash__ = "GIT_HASH"

test/unit/test_cli.py

@@ -12,6 +12,7 @@
     Status,
     _get_retriable_errors,
     _post_audit_info,
+    _post_log_chunk,
     execute_command,
 )
 
@@ -308,3 +309,144 @@ def test_post_audit_info_signs_for_url_host(self, _, mock_auth):
             aws_region="us-west-2",
             aws_service="execute-api",
         )
+
+
+class TestPostLogChunk(TestCase):
+    """Test the log-chunk POST helper"""
+
+    @patch("terrawrap.utils.cli.BotoAWSRequestsAuth")
+    @patch("requests.post")
+    def test_post_log_chunk_payload(self, mock_post, _):
+        """Posts directory/start_time/sequence/content to the log_chunk endpoint"""
+        os.chdir(os.path.normpath(os.path.dirname(__file__) + "/../helpers"))
+
+        _post_log_chunk(
+            audit_api_url="https://foo.bar",
+            path=os.path.join(os.getcwd(), "mock_directory/config/.tf_wrapper"),
+            start_time=12345,
+            sequence=2,
+            content="hello\n",
+        )
+
+        mock_post.assert_called_once_with(
+            url="https://foo.bar/log_chunk",
+            auth=ANY,
+            json={
+                "directory": "/test/helpers/mock_directory/config/.tf_wrapper",
+                "start_time": 12345,
+                "sequence": 2,
+                "content": "hello\n",
+            },
+            timeout=10,
+        )
+
+    @patch("terrawrap.utils.cli.BotoAWSRequestsAuth")
+    @patch("requests.post")
+    def test_post_log_chunk_signs_for_url_host(self, _, mock_auth):
+        """SigV4 host must come from the audit_api_url, not a hardcoded value."""
+        os.chdir(os.path.normpath(os.path.dirname(__file__) + "/../helpers"))
+
+        _post_log_chunk(
+            audit_api_url="https://terraform-audit-api.devops-testing.amplify.com",
+            path=os.path.join(os.getcwd(), "mock_directory/config/.tf_wrapper"),
+            start_time=12345,
+            sequence=0,
+            content="x",
+        )
+
+        mock_auth.assert_called_with(
+            aws_host="terraform-audit-api.devops-testing.amplify.com",
+            aws_region="us-west-2",
+            aws_service="execute-api",
+        )
+
+    @patch("terrawrap.utils.cli.BotoAWSRequestsAuth")
+    @patch("requests.post")
+    def test_post_log_chunk_raises_on_http_error(self, mock_post, _):
+        """A non-2xx response raises, so the caller decides how to handle it."""
+        mock_post.return_value.raise_for_status.side_effect = MOCK_ERROR
+        os.chdir(os.path.normpath(os.path.dirname(__file__) + "/../helpers"))
+
+        with self.assertRaises(HTTPError):
+            _post_log_chunk(
+                audit_api_url="https://foo.bar",
+                path=os.path.join(os.getcwd(), "mock_directory/config/.tf_wrapper"),
+                start_time=12345,
+                sequence=0,
+                content="x",
+            )
+
+
+class TestChunkStreaming(TestCase):
+    """Test that execute_command streams log chunks via on_chunk during apply/destroy.
+
+    Runs a real short-lived subprocess (no Popen mock) so the byte-by-byte
+    read/flush logic in _execute_command is exercised end to end.
+    """
+
+    def setUp(self):
+        self.audit_info_patcher = patch("terrawrap.utils.cli._post_audit_info")
+        self.audit_info_patcher.start()
+
+    def tearDown(self):
+        self.audit_info_patcher.stop()
+
+    @patch("terrawrap.utils.cli._post_log_chunk")
+    def test_streams_chunks_when_applying(self, mock_post_chunk):
+        """Output is flushed once at the CHUNK_LINE_COUNT threshold and once more at EOF"""
+        code = "\n".join(f"print({i})" for i in range(15))
+        exit_code, _ = execute_command(
+            ["python3", "-c", code, "apply"],
+            audit_api_url="https://foo.bar",
+            cwd=os.getcwd(),
+            print_output=False,
+        )
+
+        self.assertEqual(exit_code, 0)
+        self.assertEqual(mock_post_chunk.call_count, 2)
+        sequences = [c.kwargs["sequence"] for c in mock_post_chunk.call_args_list]
+        self.assertEqual(sequences, [0, 1])
+
+        first_chunk, second_chunk = (c.kwargs["content"] for c in mock_post_chunk.call_args_list)
+        self.assertEqual(first_chunk.count("\n"), 10)
+        self.assertEqual(second_chunk, "10\n11\n12\n13\n14\n")
+
+    @patch("terrawrap.utils.cli._post_log_chunk")
+    def test_no_streaming_without_audit_api_url(self, mock_post_chunk):
+        """No audit_api_url means no chunk streaming, even for an apply command"""
+        execute_command(
+            ["python3", "-c", "print('x')", "apply"],
+            cwd=os.getcwd(),
+            print_output=False,
+        )
+
+        mock_post_chunk.assert_not_called()
+
+    @patch("terrawrap.utils.cli._post_log_chunk")
+    def test_no_streaming_for_non_apply_commands(self, mock_post_chunk):
+        """Streaming only triggers for apply/destroy commands"""
+        execute_command(
+            ["python3", "-c", "print('x')"],
+            audit_api_url="https://foo.bar",
+            cwd=os.getcwd(),
+            print_output=False,
+        )
+
+        mock_post_chunk.assert_not_called()
+
+    @patch("terrawrap.utils.cli._post_log_chunk")
+    def test_chunk_post_failure_swallowed(self, mock_post_chunk):
+        """A log-chunk POST failure is swallowed by design — this is opt-in telemetry
+        riding alongside the real apply, so any failure here (network, git, auth) must
+        never take down the apply itself. Raising a generic Exception (not just a
+        requests error) proves the broad except in _chunk_callback is intentional."""
+        mock_post_chunk.side_effect = Exception("boom")
+
+        exit_code, _ = execute_command(
+            ["python3", "-c", "print('x')", "apply"],
+            audit_api_url="https://foo.bar",
+            cwd=os.getcwd(),
+            print_output=False,
+        )
+
+        self.assertEqual(exit_code, 0)