Cleanup Docker operator logging

airflow/providers/docker/operators/docker.py

@@ -44,6 +44,30 @@
     from airflow.utils.context import Context
 
 
+class DockerContainerFailedException(AirflowException):
+    """
+    Raised when a Docker container returns an error.
+
+    :param logs: The log output of the failed Docker container
+    """
+
+    def __init__(self, message: str | None = None, logs: list[str | bytes] | None = None):
+        super().__init__(message)
+        self.logs = logs
+
+
+class DockerContainerFailedSkipException(AirflowSkipException):
+    """
+    Raised when a Docker container returns an error and task should be skipped.
+
+    :param logs: The log output of the failed Docker container
+    """
+
+    def __init__(self, message: str | None = None, logs: list[str | bytes] | None = None):
+        super().__init__(message)
+        self.logs = logs
+
+
 def stringify(line: str | bytes):
     """Make sure string is returned even if bytes are passed. Docker stream can return bytes."""
     decode_method = getattr(line, "decode", None)
@@ -403,12 +427,11 @@ def _run_image_with_mounts(self, target_mounts, add_tmp_variable: bool) -> list[
 
             result = self.cli.wait(self.container["Id"])
             if result["StatusCode"] in self.skip_on_exit_code:
-                raise AirflowSkipException(
-                    f"Docker container returned exit code {self.skip_on_exit_code}. Skipping."
+                raise DockerContainerFailedSkipException(
+                    f"Docker container returned exit code {self.skip_on_exit_code}. Skipping.", logs=log_lines
                 )
             elif result["StatusCode"] != 0:
-                joined_log_lines = "\n".join(log_lines)
-                raise AirflowException(f"Docker container failed: {result!r} lines {joined_log_lines}")
+                raise DockerContainerFailedException(f"Docker container failed: {result!r}", logs=log_lines)
 
             if self.retrieve_output:
                 return self._attempt_to_retrieve_result()

tests/providers/docker/operators/test_docker.py

@@ -27,7 +27,7 @@
 from docker.types import DeviceRequest, LogConfig, Mount
 
 from airflow.exceptions import AirflowException, AirflowSkipException
-from airflow.providers.docker.operators.docker import DockerOperator
+from airflow.providers.docker.operators.docker import DockerContainerFailedException, DockerOperator
 
 TEST_CONN_ID = "docker_test_connection"
 TEST_DOCKER_URL = "unix://var/run/docker.test.sock"
@@ -553,19 +553,19 @@ def test_skip(self, extra_kwargs, actual_exit_code, expected_exc):
     def test_execute_container_fails(self):
         failed_msg = {"StatusCode": 1}
         log_line = ["unicode container log 😁   ", b"byte string container log"]
-        expected_message = "Docker container failed: {failed_msg} lines {expected_log_output}"
+        expected_message = "Docker container failed: {failed_msg}"
         self.client_mock.attach.return_value = log_line
         self.client_mock.wait.return_value = failed_msg
 
         operator = DockerOperator(image="ubuntu", owner="unittest", task_id="unittest")
 
-        with pytest.raises(AirflowException) as raised_exception:
+        with pytest.raises(DockerContainerFailedException) as raised_exception:
             operator.execute(None)
 
-        assert str(raised_exception.value) == expected_message.format(
-            failed_msg=failed_msg,
-            expected_log_output=f'{log_line[0].strip()}\n{log_line[1].decode("utf-8")}',
-        )
+            assert str(raised_exception.value) == expected_message.format(
+                failed_msg=failed_msg,
+            )
+            assert raised_exception.logs == f'{log_line[0].strip()}\n{log_line[1].decode("utf-8")}'
 
     def test_auto_remove_container_fails(self):
         self.client_mock.wait.return_value = {"StatusCode": 1}