docker: do not lose early exec output

score/itf/plugins/docker.py

@@ -15,11 +15,13 @@
 import io
 import os
 import shlex
+import struct
 import tarfile
 import threading
 import time
 import docker as pypi_docker
 import pytest
+from docker.utils.socket import consume_socket_output, demux_adaptor
 
 from score.itf.core.com.ssh import Ssh
 from score.itf.core.process.async_process import AsyncProcess
@@ -64,6 +66,65 @@ def pytest_addoption(parser):
     )
 
 
+_STDOUT = 1
+_FRAME_HEADER_SIZE = 8
+
+
+def _read_frames(buffered_reader, tty):
+    """Yield ``(stream_id, data)`` frames from the exec response body.
+
+    docker-py reads exec output from the raw socket, but http.client has already
+    pulled everything that arrived together with the HTTP response headers into
+    its buffered reader. Output produced right after the exec started (e.g. the
+    first line a command prints) is therefore silently dropped whenever the client
+    reads the reply late. Reading through the buffered reader keeps that data.
+    """
+    if tty:
+        while True:
+            data = buffered_reader.read1(65536)
+            if not data:
+                return
+            yield _STDOUT, data
+    while True:
+        header = buffered_reader.read(_FRAME_HEADER_SIZE)
+        if len(header) < _FRAME_HEADER_SIZE:
+            return
+        stream_id, size = struct.unpack(">BxxxL", header)
+        data = buffered_reader.read(size) if size else b""
+        if data:
+            yield stream_id, data
+        if len(data) < size:
+            return
+
+
+class _ExecAPIClient(pypi_docker.APIClient):
+    """APIClient that does not lose early exec output (see :func:`_read_frames`)."""
+
+    def _read_from_socket(self, response, stream, tty=True, demux=False):
+        sock = self._get_raw_response_socket(response)
+        self._disable_socket_timeout(sock)
+        frames = _read_frames(response.raw._fp.fp, tty)
+        if demux:
+            frames = (demux_adaptor(*frame) for frame in frames)
+        else:
+            frames = (data for (_, data) in frames)
+        if stream:
+            return frames
+        try:
+            return consume_socket_output(frames, demux=demux)
+        finally:
+            response.close()
+
+
+# Workaround for docker-py losing exec output (reimplements `_read_from_socket` without the timeout
+# and without dropping early frames). See https://github.com/docker/docker-py/issues/3332 and
+# https://github.com/docker/docker-py/issues/2042; open fixes: docker-py PRs #3333 and #3438.
+# Remove this once the upstream issues/PRs are resolved.
+class _ExecDockerClient(pypi_docker.DockerClient):
+    def __init__(self, *args, **kwargs):  # pylint: disable=super-init-not-called
+        self.api = _ExecAPIClient(*args, **kwargs)
+
+
 class DockerAsyncProcess(AsyncProcess):
     """Handle for a non-blocking command execution inside a Docker container."""
 
@@ -138,7 +199,7 @@ def __init__(self, container, network=None):
         super().__init__()
         self.container = container
         self.network = network
-        self._client = pypi_docker.from_env(timeout=DOCKER_CLIENT_TIMEOUT)
+        self._client = _ExecDockerClient.from_env(timeout=DOCKER_CLIENT_TIMEOUT)
 
     def __getattr__(self, name):
         return getattr(self.container, name)
@@ -352,7 +413,7 @@ def target_init(request, _docker_configuration):
             raise subprocess.CalledProcessError(result.returncode, docker_image_bootstrap)
 
     docker_image = request.config.getoption("docker_image")
-    client = pypi_docker.from_env(timeout=DOCKER_CLIENT_TIMEOUT)
+    client = _ExecDockerClient.from_env(timeout=DOCKER_CLIENT_TIMEOUT)
 
     known_keys = {"command", "init", "environment", "volumes", "shm_size", "detach", "auto_remove"}
     reserved_overrides = {k for k in ("detach", "auto_remove") if k in _docker_configuration}

test/integration/BUILD

@@ -44,6 +44,7 @@ py_itf_test(
     srcs = [
         "test_async_exec.py",
         "test_docker.py",
+        "test_exec_output_not_lost.py",
     ],
     args = [
         "--docker-image=ubuntu:24.04",

test/integration/test_exec_output_not_lost.py

@@ -0,0 +1,47 @@
+# *******************************************************************************
+# Copyright (c) 2026 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License Version 2.0 which is available at
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+# *******************************************************************************
+import http.client
+import time
+
+import pytest
+
+
+@pytest.fixture
+def slow_reply_reader(monkeypatch):
+    """Emulate a loaded host: the client reads the reply of a request late.
+
+    The daemon then sends the HTTP response headers and the first exec output in
+    one burst, which http.client swallows into its read buffer.
+    """
+    original = http.client.HTTPConnection.getresponse
+
+    def delayed_getresponse(self, *args, **kwargs):
+        time.sleep(0.5)
+        return original(self, *args, **kwargs)
+
+    monkeypatch.setattr(http.client.HTTPConnection, "getresponse", delayed_getresponse)
+
+
+def test_execute_returns_output_with_slow_reader(target, slow_reply_reader):
+    exit_code, output = target.execute("echo Hello World!")
+    assert exit_code == 0
+    assert output.decode() == "Hello World!\n"
+
+
+def test_execute_async_pid_not_lost_with_slow_reader(target, slow_reply_reader):
+    proc = target.execute_async("sleep 0.3; echo first_application_log; sleep 5")
+    try:
+        assert proc.pid() > 0
+        assert proc.is_running()
+    finally:
+        proc.stop()