Fix left trailing space logs for docker operator (apache#33692)

providers/src/airflow/providers/docker/operators/docker.py

@@ -47,6 +47,8 @@
 from airflow.utils.types import NOTSET, ArgNotSet
 
 if TYPE_CHECKING:
+    from logging import Logger
+
     from docker import APIClient
     from docker.types import DeviceRequest
 
@@ -62,6 +64,16 @@ def stringify(line: str | bytes):
         return line
 
 
+def fetch_logs(logstream, log: Logger):
+    log_lines = []
+    for log_chunk in logstream:
+        log_chunk = stringify(log_chunk).rstrip()
+        log_lines.append(log_chunk)
+        for log_chunk_line in log_chunk.split("\n"):
+            log.info("%s", log_chunk_line)
+    return log_lines
+
+
 class DockerOperator(BaseOperator):
     """
     Execute a command inside a docker container.
@@ -430,12 +442,7 @@ def _run_image_with_mounts(self, target_mounts, add_tmp_variable: bool) -> list[
         try:
             self.cli.start(self.container["Id"])
 
-            log_lines = []
-            for log_chunk in logstream:
-                log_chunk = stringify(log_chunk).strip()
-                log_lines.append(log_chunk)
-                for log_chunk_line in log_chunk.split("\n"):
-                    self.log.info("%s", log_chunk_line)
+            log_lines = fetch_logs(logstream, self.log)
 
             result = self.cli.wait(self.container["Id"])
             if result["StatusCode"] in self.skip_on_exit_code:

providers/tests/docker/operators/test_docker.py

@@ -29,6 +29,7 @@
 from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning, AirflowSkipException
 from airflow.providers.docker.exceptions import DockerContainerFailedException
 from airflow.providers.docker.operators.docker import DockerOperator
+from airflow.providers.docker.operators.docker import fetch_logs
 from airflow.utils.task_instance_session import set_current_task_instance_session
 
 TEST_CONN_ID = "docker_test_connection"
@@ -865,3 +866,39 @@ def test_partial_deprecated_skip_exit_code_ambiguous(
                     pytest.raises(ValueError, match="Conflicting `skip_on_exit_code` provided"),
                 ):
                     ti.render_templates()
+
+    @pytest.mark.parametrize(
+        "log_lines, expected_lines",
+        [
+            pytest.param(
+                [
+                    "return self.main(*args, **kwargs)",
+                    "                 ^^^^^^^^^^^^^^^^",
+                ],
+                [
+                    "return self.main(*args, **kwargs)",
+                    "                 ^^^^^^^^^^^^^^^^",
+                ],
+                id="should-not-remove-leading-spaces"
+            ),
+            pytest.param(
+                [
+                    "   ^^^^^^^^^^^^^^^^   ",
+                ],
+                [
+                    "   ^^^^^^^^^^^^^^^^",
+                ],
+                id="should-remove-trailing-spaces"
+            ),
+        ],
+    )
+    @mock.patch("logging.Logger")
+    def test_fetch_logs(self, logger_mock, log_lines, expected_lines):
+        fetch_logs(
+            log_lines,
+            logger_mock
+        )
+        assert logger_mock.info.call_args_list == [
+            call("%s", line)
+            for line in expected_lines
+        ]