You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2023/09/04 19:00:33 UTC
[airflow] branch main updated: Cleanup Docker operator logging (#33914)
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 3d27504a62 Cleanup Docker operator logging (#33914)
3d27504a62 is described below
commit 3d27504a6232cacb12a9e3dc5837513e558bd52b
Author: Daniel Wolf <95...@users.noreply.github.com>
AuthorDate: Mon Sep 4 21:00:26 2023 +0200
Cleanup Docker operator logging (#33914)
]
Co-authored-by: Tzu-ping Chung <ur...@gmail.com>
---
airflow/providers/docker/exceptions.py | 45 +++++++++++++++++++++++++
airflow/providers/docker/operators/docker.py | 13 ++++---
tests/providers/docker/operators/test_docker.py | 7 ++--
3 files changed, 57 insertions(+), 8 deletions(-)
diff --git a/airflow/providers/docker/exceptions.py b/airflow/providers/docker/exceptions.py
new file mode 100644
index 0000000000..5162e180f8
--- /dev/null
+++ b/airflow/providers/docker/exceptions.py
@@ -0,0 +1,45 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Exceptions used by Docker provider."""
+from __future__ import annotations
+
+from airflow.exceptions import AirflowException, AirflowSkipException
+
+
+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) -> 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) -> None:
+ super().__init__(message)
+ self.logs = logs
diff --git a/airflow/providers/docker/operators/docker.py b/airflow/providers/docker/operators/docker.py
index 3908ae54d3..327e8e7968 100644
--- a/airflow/providers/docker/operators/docker.py
+++ b/airflow/providers/docker/operators/docker.py
@@ -33,8 +33,12 @@ from docker.errors import APIError
from docker.types import LogConfig, Mount
from dotenv import dotenv_values
-from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning, AirflowSkipException
+from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.models import BaseOperator
+from airflow.providers.docker.exceptions import (
+ DockerContainerFailedException,
+ DockerContainerFailedSkipException,
+)
from airflow.providers.docker.hooks.docker import DockerHook
if TYPE_CHECKING:
@@ -403,12 +407,11 @@ class DockerOperator(BaseOperator):
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()
diff --git a/tests/providers/docker/operators/test_docker.py b/tests/providers/docker/operators/test_docker.py
index abe8bb4be2..f042e32c1d 100644
--- a/tests/providers/docker/operators/test_docker.py
+++ b/tests/providers/docker/operators/test_docker.py
@@ -27,6 +27,7 @@ from docker.errors import APIError
from docker.types import DeviceRequest, LogConfig, Mount
from airflow.exceptions import AirflowException, AirflowSkipException
+from airflow.providers.docker.exceptions import DockerContainerFailedException
from airflow.providers.docker.operators.docker import DockerOperator
TEST_CONN_ID = "docker_test_connection"
@@ -553,19 +554,19 @@ class TestDockerOperator:
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 raised_exception.value.logs == [log_line[0].strip(), log_line[1].decode("utf-8")]
def test_auto_remove_container_fails(self):
self.client_mock.wait.return_value = {"StatusCode": 1}