diff --git a/score/itf/plugins/docker.py b/score/itf/plugins/docker.py index ca1cd8f3..7c91a436 100644 --- a/score/itf/plugins/docker.py +++ b/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} diff --git a/test/integration/BUILD b/test/integration/BUILD index 4c9aa2dd..4b58cdd5 100644 --- a/test/integration/BUILD +++ b/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", diff --git a/test/integration/test_exec_output_not_lost.py b/test/integration/test_exec_output_not_lost.py new file mode 100644 index 00000000..8472dfb8 --- /dev/null +++ b/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()