We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
TaskInstances do not succeed when using enable_logging=True option in DockerSwarmOperator
### Apache Airflow Provider(s)

docker

### Versions of Apache Airflow Providers

apache-airflow-providers-celery==3.1.0
apache-airflow-providers-docker==3.3.0

### Apache Airflow version

2.5.0

### Operating System

centos 7

### Deployment

Other Docker-based deployment

### Deployment details

Running an a docker-swarm cluster deployed locally. 

### What happened

Same issue as https://github.com/apache/airflow/issues/13675

With logging_enabled=True the DAG never completes and stays in running. 

When using DockerSwarmOperator together with the default enable_logging=True option, tasks do not succeed and stay in state running. When checking the docker service logs I can clearly see that the container ran and ended successfully. Airflow however does not recognize that the container finished and keeps the tasks in state running.

### What you think should happen instead

DAG should complete.

### How to reproduce

Docker-compose deployment:
```console
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.5.0/docker-compose.yaml'
docker compose up airflow-init
docker compose up -d
```

DAG code:

```python
from airflow import DAG
from docker.types import Mount, SecretReference
from airflow.providers.docker.operators.docker_swarm import DockerSwarmOperator
from datetime import timedelta
from airflow.utils.dates import days_ago
from airflow.models import Variable


# Setup default args for the job
default_args = {
	'owner': 'airflow',
	'start_date': days_ago(2),
	'retries': 0
}
 
# Create the DAG
dag = DAG(
    'test_dag',         # DAG ID
    default_args=default_args,
    schedule_interval='0 0 * * *', 
    catchup=False
)

# # Create the DAG object
with dag as dag:
	  docker_swarm_task = DockerSwarmOperator(
			task_id="job_run",
			image="<any image>",
			execution_timeout=timedelta(minutes=5),
			command="<specific code>",
			api_version='auto',
			tty=True,
			enable_logging=True
		)
```

### Anything else

_No response_

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/providers/docker/operators/docker_swarm.py`
Content:
```
1 # Licensed to the Apache Software Foundation (ASF) under one
2 # or more contributor license agreements.  See the NOTICE file
3 # distributed with this work for additional information
4 # regarding copyright ownership.  The ASF licenses this file
5 # to you under the Apache License, Version 2.0 (the
6 # "License"); you may not use this file except in compliance
7 # with the License.  You may obtain a copy of the License at
8 #
9 #   http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing,
12 # software distributed under the License is distributed on an
13 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14 # KIND, either express or implied.  See the License for the
15 # specific language governing permissions and limitations
16 # under the License.
17 """Run ephemeral Docker Swarm services."""
18 from __future__ import annotations
19 
20 from typing import TYPE_CHECKING
21 
22 from docker import types
23 
24 from airflow.exceptions import AirflowException
25 from airflow.providers.docker.operators.docker import DockerOperator
26 from airflow.utils.strings import get_random_string
27 
28 if TYPE_CHECKING:
29     from airflow.utils.context import Context
30 
31 
32 class DockerSwarmOperator(DockerOperator):
33     """
34     Execute a command as an ephemeral docker swarm service.
35 
36     Example use-case - Using Docker Swarm orchestration to make one-time
37     scripts highly available.
38 
39     A temporary directory is created on the host and
40     mounted into a container to allow storing files
41     that together exceed the default disk size of 10GB in a container.
42     The path to the mounted directory can be accessed
43     via the environment variable ``AIRFLOW_TMP_DIR``.
44 
45     If a login to a private registry is required prior to pulling the image, a
46     Docker connection needs to be configured in Airflow and the connection ID
47     be provided with the parameter ``docker_conn_id``.
48 
49     :param image: Docker image from which to create the container.
50         If image tag is omitted, "latest" will be used.
51     :param api_version: Remote API version. Set to ``auto`` to automatically
52         detect the server's version.
53     :param auto_remove: Auto-removal of the container on daemon side when the
54         container's process exits.
55         The default is False.
56     :param command: Command to be run in the container. (templated)
57     :param docker_url: URL of the host running the docker daemon.
58         Default is unix://var/run/docker.sock
59     :param environment: Environment variables to set in the container. (templated)
60     :param force_pull: Pull the docker image on every run. Default is False.
61     :param mem_limit: Maximum amount of memory the container can use.
62         Either a float value, which represents the limit in bytes,
63         or a string like ``128m`` or ``1g``.
64     :param tls_ca_cert: Path to a PEM-encoded certificate authority
65         to secure the docker connection.
66     :param tls_client_cert: Path to the PEM-encoded certificate
67         used to authenticate docker client.
68     :param tls_client_key: Path to the PEM-encoded key used to authenticate docker client.
69     :param tls_hostname: Hostname to match against
70         the docker server certificate or False to disable the check.
71     :param tls_ssl_version: Version of SSL to use when communicating with docker daemon.
72     :param tmp_dir: Mount point inside the container to
73         a temporary directory created on the host by the operator.
74         The path is also made available via the environment variable
75         ``AIRFLOW_TMP_DIR`` inside the container.
76     :param user: Default user inside the docker container.
77     :param docker_conn_id: The :ref:`Docker connection id <howto/connection:docker>`
78     :param tty: Allocate pseudo-TTY to the container of this service
79         This needs to be set see logs of the Docker container / service.
80     :param enable_logging: Show the application's logs in operator's logs.
81         Supported only if the Docker engine is using json-file or journald logging drivers.
82         The `tty` parameter should be set to use this with Python applications.
83     :param configs: List of docker configs to be exposed to the containers of the swarm service.
84         The configs are ConfigReference objects as per the docker api
85         [https://docker-py.readthedocs.io/en/stable/services.html#docker.models.services.ServiceCollection.create]_
86     :param secrets: List of docker secrets to be exposed to the containers of the swarm service.
87         The secrets are SecretReference objects as per the docker create_service api.
88         [https://docker-py.readthedocs.io/en/stable/services.html#docker.models.services.ServiceCollection.create]_
89     :param mode: Indicate whether a service should be deployed as a replicated or global service,
90         and associated parameters
91     :param networks: List of network names or IDs or NetworkAttachmentConfig to attach the service to.
92     :param placement: Placement instructions for the scheduler. If a list is passed instead,
93         it is assumed to be a list of constraints as part of a Placement object.
94     """
95 
96     def __init__(
97         self,
98         *,
99         image: str,
100         enable_logging: bool = True,
101         configs: list[types.ConfigReference] | None = None,
102         secrets: list[types.SecretReference] | None = None,
103         mode: types.ServiceMode | None = None,
104         networks: list[str | types.NetworkAttachmentConfig] | None = None,
105         placement: types.Placement | list[types.Placement] | None = None,
106         **kwargs,
107     ) -> None:
108         super().__init__(image=image, **kwargs)
109         self.enable_logging = enable_logging
110         self.service = None
111         self.configs = configs
112         self.secrets = secrets
113         self.mode = mode
114         self.networks = networks
115         self.placement = placement
116 
117     def execute(self, context: Context) -> None:
118         self.environment["AIRFLOW_TMP_DIR"] = self.tmp_dir
119         return self._run_service()
120 
121     def _run_service(self) -> None:
122         self.log.info("Starting docker service from image %s", self.image)
123         self.service = self.cli.create_service(
124             types.TaskTemplate(
125                 container_spec=types.ContainerSpec(
126                     image=self.image,
127                     command=self.format_command(self.command),
128                     mounts=self.mounts,
129                     env=self.environment,
130                     user=self.user,
131                     tty=self.tty,
132                     configs=self.configs,
133                     secrets=self.secrets,
134                 ),
135                 restart_policy=types.RestartPolicy(condition="none"),
136                 resources=types.Resources(mem_limit=self.mem_limit),
137                 networks=self.networks,
138                 placement=self.placement,
139             ),
140             name=f"airflow-{get_random_string()}",
141             labels={"name": f"airflow__{self.dag_id}__{self.task_id}"},
142             mode=self.mode,
143         )
144         if self.service is None:
145             raise Exception("Service should be set here")
146         self.log.info("Service started: %s", self.service)
147 
148         # wait for the service to start the task
149         while not self.cli.tasks(filters={"service": self.service["ID"]}):
150             continue
151 
152         if self.enable_logging:
153             self._stream_logs_to_output()
154 
155         while True:
156             if self._has_service_terminated():
157                 self.log.info("Service status before exiting: %s", self._service_status())
158                 break
159 
160         self.log.info("auto_removeauto_removeauto_removeauto_removeauto_remove : %s", str(self.auto_remove))
161         if self.service and self._service_status() != "complete":
162             if self.auto_remove == "success":
163                 self.cli.remove_service(self.service["ID"])
164             raise AirflowException(f"Service did not complete: {self.service!r}")
165         elif self.auto_remove == "success":
166             if not self.service:
167                 raise Exception("The 'service' should be initialized before!")
168             self.cli.remove_service(self.service["ID"])
169 
170     def _service_status(self) -> str | None:
171         if not self.service:
172             raise Exception("The 'service' should be initialized before!")
173         return self.cli.tasks(filters={"service": self.service["ID"]})[0]["Status"]["State"]
174 
175     def _has_service_terminated(self) -> bool:
176         status = self._service_status()
177         return status in ["complete", "failed", "shutdown", "rejected", "orphaned", "remove"]
178 
179     def _stream_logs_to_output(self) -> None:
180         if not self.service:
181             raise Exception("The 'service' should be initialized before!")
182         logs = self.cli.service_logs(
183             self.service["ID"], follow=True, stdout=True, stderr=True, is_tty=self.tty
184         )
185         line = ""
186         for log in logs:
187             try:
188                 log = log.decode()
189             except UnicodeDecodeError:
190                 continue
191             if log == "\n":
192                 self.log.info(line)
193                 line = ""
194             else:
195                 line += log
196         # flush any remaining log stream
197         if line:
198             self.log.info(line)
199 
200     def on_kill(self) -> None:
201         if self.hook.client_created and self.service is not None:
202             self.log.info("Removing docker service: %s", self.service["ID"])
203             self.cli.remove_service(self.service["ID"])
204
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/airflow/providers/docker/operators/docker_swarm.py b/airflow/providers/docker/operators/docker_swarm.py
--- a/airflow/providers/docker/operators/docker_swarm.py
+++ b/airflow/providers/docker/operators/docker_swarm.py
@@ -17,6 +17,9 @@
 """Run ephemeral Docker Swarm services."""
 from __future__ import annotations
 
+import re
+from datetime import datetime
+from time import sleep
 from typing import TYPE_CHECKING
 
 from docker import types
@@ -179,23 +182,34 @@
     def _stream_logs_to_output(self) -> None:
         if not self.service:
             raise Exception("The 'service' should be initialized before!")
-        logs = self.cli.service_logs(
-            self.service["ID"], follow=True, stdout=True, stderr=True, is_tty=self.tty
-        )
-        line = ""
-        for log in logs:
-            try:
-                log = log.decode()
-            except UnicodeDecodeError:
-                continue
-            if log == "\n":
-                self.log.info(line)
-                line = ""
-            else:
-                line += log
-        # flush any remaining log stream
-        if line:
-            self.log.info(line)
+        last_line_logged, last_timestamp = "", 0
+
+        def stream_new_logs(last_line_logged, since=0):
+            logs = self.cli.service_logs(
+                self.service["ID"],
+                follow=False,
+                stdout=True,
+                stderr=True,
+                is_tty=self.tty,
+                since=since,
+                timestamps=True,
+            )
+            logs = b"".join(logs).decode().splitlines()
+            if last_line_logged in logs:
+                logs = logs[logs.index(last_line_logged) + 1 :]
+            for line in logs:
+                match = re.match(r"(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d{6,}Z) (.*)", line)
+                timestamp, message = match.groups()
+                self.log.info(message)
+            # Floor nanoseconds to microseconds
+            last_timestamp = re.sub(r"(\.\d{6})\d+Z", r"\1Z", timestamp)
+            last_timestamp = datetime.strptime(last_timestamp, "%Y-%m-%dT%H:%M:%S.%fZ")
+            last_timestamp = last_timestamp.timestamp()
+            return last_line_logged, last_timestamp
+
+        while not self._has_service_terminated():
+            sleep(2)
+            last_line_logged, last_timestamp = stream_new_logs(last_line_logged, since=last_timestamp)
 
     def on_kill(self) -> None:
         if self.hook.client_created and self.service is not None:

{"golden_diff": "diff --git a/airflow/providers/docker/operators/docker_swarm.py b/airflow/providers/docker/operators/docker_swarm.py\n--- a/airflow/providers/docker/operators/docker_swarm.py\n+++ b/airflow/providers/docker/operators/docker_swarm.py\n@@ -17,6 +17,9 @@\n \"\"\"Run ephemeral Docker Swarm services.\"\"\"\n from __future__ import annotations\n \n+import re\n+from datetime import datetime\n+from time import sleep\n from typing import TYPE_CHECKING\n \n from docker import types\n@@ -179,23 +182,34 @@\n     def _stream_logs_to_output(self) -> None:\n         if not self.service:\n             raise Exception(\"The 'service' should be initialized before!\")\n-        logs = self.cli.service_logs(\n-            self.service[\"ID\"], follow=True, stdout=True, stderr=True, is_tty=self.tty\n-        )\n-        line = \"\"\n-        for log in logs:\n-            try:\n-                log = log.decode()\n-            except UnicodeDecodeError:\n-                continue\n-            if log == \"\\n\":\n-                self.log.info(line)\n-                line = \"\"\n-            else:\n-                line += log\n-        # flush any remaining log stream\n-        if line:\n-            self.log.info(line)\n+        last_line_logged, last_timestamp = \"\", 0\n+\n+        def stream_new_logs(last_line_logged, since=0):\n+            logs = self.cli.service_logs(\n+                self.service[\"ID\"],\n+                follow=False,\n+                stdout=True,\n+                stderr=True,\n+                is_tty=self.tty,\n+                since=since,\n+                timestamps=True,\n+            )\n+            logs = b\"\".join(logs).decode().splitlines()\n+            if last_line_logged in logs:\n+                logs = logs[logs.index(last_line_logged) + 1 :]\n+            for line in logs:\n+                match = re.match(r\"(\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}.\\d{6,}Z) (.*)\", line)\n+                timestamp, message = match.groups()\n+                self.log.info(message)\n+            # Floor nanoseconds to microseconds\n+            last_timestamp = re.sub(r\"(\\.\\d{6})\\d+Z\", r\"\\1Z\", timestamp)\n+            last_timestamp = datetime.strptime(last_timestamp, \"%Y-%m-%dT%H:%M:%S.%fZ\")\n+            last_timestamp = last_timestamp.timestamp()\n+            return last_line_logged, last_timestamp\n+\n+        while not self._has_service_terminated():\n+            sleep(2)\n+            last_line_logged, last_timestamp = stream_new_logs(last_line_logged, since=last_timestamp)\n \n     def on_kill(self) -> None:\n         if self.hook.client_created and self.service is not None:\n", "issue": "TaskInstances do not succeed when using enable_logging=True option in DockerSwarmOperator\n### Apache Airflow Provider(s)\r\n\r\ndocker\r\n\r\n### Versions of Apache Airflow Providers\r\n\r\napache-airflow-providers-celery==3.1.0\r\napache-airflow-providers-docker==3.3.0\r\n\r\n### Apache Airflow version\r\n\r\n2.5.0\r\n\r\n### Operating System\r\n\r\ncentos 7\r\n\r\n### Deployment\r\n\r\nOther Docker-based deployment\r\n\r\n### Deployment details\r\n\r\nRunning an a docker-swarm cluster deployed locally. \r\n\r\n### What happened\r\n\r\nSame issue as https://github.com/apache/airflow/issues/13675\r\n\r\nWith logging_enabled=True the DAG never completes and stays in running. \r\n\r\nWhen using DockerSwarmOperator together with the default enable_logging=True option, tasks do not succeed and stay in state running. When checking the docker service logs I can clearly see that the container ran and ended successfully. Airflow however does not recognize that the container finished and keeps the tasks in state running.\r\n\r\n### What you think should happen instead\r\n\r\nDAG should complete.\r\n\r\n### How to reproduce\r\n\r\nDocker-compose deployment:\r\n```console\r\ncurl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.5.0/docker-compose.yaml'\r\ndocker compose up airflow-init\r\ndocker compose up -d\r\n```\r\n\r\nDAG code:\r\n\r\n```python\r\nfrom airflow import DAG\r\nfrom docker.types import Mount, SecretReference\r\nfrom airflow.providers.docker.operators.docker_swarm import DockerSwarmOperator\r\nfrom datetime import timedelta\r\nfrom airflow.utils.dates import days_ago\r\nfrom airflow.models import Variable\r\n\r\n\r\n# Setup default args for the job\r\ndefault_args = {\r\n\t'owner': 'airflow',\r\n\t'start_date': days_ago(2),\r\n\t'retries': 0\r\n}\r\n \r\n# Create the DAG\r\ndag = DAG(\r\n    'test_dag',         # DAG ID\r\n    default_args=default_args,\r\n    schedule_interval='0 0 * * *', \r\n    catchup=False\r\n)\r\n\r\n# # Create the DAG object\r\nwith dag as dag:\r\n\t  docker_swarm_task = DockerSwarmOperator(\r\n\t\t\ttask_id=\"job_run\",\r\n\t\t\timage=\"<any image>\",\r\n\t\t\texecution_timeout=timedelta(minutes=5),\r\n\t\t\tcommand=\"<specific code>\",\r\n\t\t\tapi_version='auto',\r\n\t\t\ttty=True,\r\n\t\t\tenable_logging=True\r\n\t\t)\r\n```\r\n\r\n### Anything else\r\n\r\n_No response_\r\n\r\n### Are you willing to submit PR?\r\n\r\n- [X] Yes I am willing to submit a PR!\r\n\r\n### Code of Conduct\r\n\r\n- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)\r\n\n", "before_files": [{"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"Run ephemeral Docker Swarm services.\"\"\"\nfrom __future__ import annotations\n\nfrom typing import TYPE_CHECKING\n\nfrom docker import types\n\nfrom airflow.exceptions import AirflowException\nfrom airflow.providers.docker.operators.docker import DockerOperator\nfrom airflow.utils.strings import get_random_string\n\nif TYPE_CHECKING:\n    from airflow.utils.context import Context\n\n\nclass DockerSwarmOperator(DockerOperator):\n    \"\"\"\n    Execute a command as an ephemeral docker swarm service.\n\n    Example use-case - Using Docker Swarm orchestration to make one-time\n    scripts highly available.\n\n    A temporary directory is created on the host and\n    mounted into a container to allow storing files\n    that together exceed the default disk size of 10GB in a container.\n    The path to the mounted directory can be accessed\n    via the environment variable ``AIRFLOW_TMP_DIR``.\n\n    If a login to a private registry is required prior to pulling the image, a\n    Docker connection needs to be configured in Airflow and the connection ID\n    be provided with the parameter ``docker_conn_id``.\n\n    :param image: Docker image from which to create the container.\n        If image tag is omitted, \"latest\" will be used.\n    :param api_version: Remote API version. Set to ``auto`` to automatically\n        detect the server's version.\n    :param auto_remove: Auto-removal of the container on daemon side when the\n        container's process exits.\n        The default is False.\n    :param command: Command to be run in the container. (templated)\n    :param docker_url: URL of the host running the docker daemon.\n        Default is unix://var/run/docker.sock\n    :param environment: Environment variables to set in the container. (templated)\n    :param force_pull: Pull the docker image on every run. Default is False.\n    :param mem_limit: Maximum amount of memory the container can use.\n        Either a float value, which represents the limit in bytes,\n        or a string like ``128m`` or ``1g``.\n    :param tls_ca_cert: Path to a PEM-encoded certificate authority\n        to secure the docker connection.\n    :param tls_client_cert: Path to the PEM-encoded certificate\n        used to authenticate docker client.\n    :param tls_client_key: Path to the PEM-encoded key used to authenticate docker client.\n    :param tls_hostname: Hostname to match against\n        the docker server certificate or False to disable the check.\n    :param tls_ssl_version: Version of SSL to use when communicating with docker daemon.\n    :param tmp_dir: Mount point inside the container to\n        a temporary directory created on the host by the operator.\n        The path is also made available via the environment variable\n        ``AIRFLOW_TMP_DIR`` inside the container.\n    :param user: Default user inside the docker container.\n    :param docker_conn_id: The :ref:`Docker connection id <howto/connection:docker>`\n    :param tty: Allocate pseudo-TTY to the container of this service\n        This needs to be set see logs of the Docker container / service.\n    :param enable_logging: Show the application's logs in operator's logs.\n        Supported only if the Docker engine is using json-file or journald logging drivers.\n        The `tty` parameter should be set to use this with Python applications.\n    :param configs: List of docker configs to be exposed to the containers of the swarm service.\n        The configs are ConfigReference objects as per the docker api\n        [https://docker-py.readthedocs.io/en/stable/services.html#docker.models.services.ServiceCollection.create]_\n    :param secrets: List of docker secrets to be exposed to the containers of the swarm service.\n        The secrets are SecretReference objects as per the docker create_service api.\n        [https://docker-py.readthedocs.io/en/stable/services.html#docker.models.services.ServiceCollection.create]_\n    :param mode: Indicate whether a service should be deployed as a replicated or global service,\n        and associated parameters\n    :param networks: List of network names or IDs or NetworkAttachmentConfig to attach the service to.\n    :param placement: Placement instructions for the scheduler. If a list is passed instead,\n        it is assumed to be a list of constraints as part of a Placement object.\n    \"\"\"\n\n    def __init__(\n        self,\n        *,\n        image: str,\n        enable_logging: bool = True,\n        configs: list[types.ConfigReference] | None = None,\n        secrets: list[types.SecretReference] | None = None,\n        mode: types.ServiceMode | None = None,\n        networks: list[str | types.NetworkAttachmentConfig] | None = None,\n        placement: types.Placement | list[types.Placement] | None = None,\n        **kwargs,\n    ) -> None:\n        super().__init__(image=image, **kwargs)\n        self.enable_logging = enable_logging\n        self.service = None\n        self.configs = configs\n        self.secrets = secrets\n        self.mode = mode\n        self.networks = networks\n        self.placement = placement\n\n    def execute(self, context: Context) -> None:\n        self.environment[\"AIRFLOW_TMP_DIR\"] = self.tmp_dir\n        return self._run_service()\n\n    def _run_service(self) -> None:\n        self.log.info(\"Starting docker service from image %s\", self.image)\n        self.service = self.cli.create_service(\n            types.TaskTemplate(\n                container_spec=types.ContainerSpec(\n                    image=self.image,\n                    command=self.format_command(self.command),\n                    mounts=self.mounts,\n                    env=self.environment,\n                    user=self.user,\n                    tty=self.tty,\n                    configs=self.configs,\n                    secrets=self.secrets,\n                ),\n                restart_policy=types.RestartPolicy(condition=\"none\"),\n                resources=types.Resources(mem_limit=self.mem_limit),\n                networks=self.networks,\n                placement=self.placement,\n            ),\n            name=f\"airflow-{get_random_string()}\",\n            labels={\"name\": f\"airflow__{self.dag_id}__{self.task_id}\"},\n            mode=self.mode,\n        )\n        if self.service is None:\n            raise Exception(\"Service should be set here\")\n        self.log.info(\"Service started: %s\", self.service)\n\n        # wait for the service to start the task\n        while not self.cli.tasks(filters={\"service\": self.service[\"ID\"]}):\n            continue\n\n        if self.enable_logging:\n            self._stream_logs_to_output()\n\n        while True:\n            if self._has_service_terminated():\n                self.log.info(\"Service status before exiting: %s\", self._service_status())\n                break\n\n        self.log.info(\"auto_removeauto_removeauto_removeauto_removeauto_remove : %s\", str(self.auto_remove))\n        if self.service and self._service_status() != \"complete\":\n            if self.auto_remove == \"success\":\n                self.cli.remove_service(self.service[\"ID\"])\n            raise AirflowException(f\"Service did not complete: {self.service!r}\")\n        elif self.auto_remove == \"success\":\n            if not self.service:\n                raise Exception(\"The 'service' should be initialized before!\")\n            self.cli.remove_service(self.service[\"ID\"])\n\n    def _service_status(self) -> str | None:\n        if not self.service:\n            raise Exception(\"The 'service' should be initialized before!\")\n        return self.cli.tasks(filters={\"service\": self.service[\"ID\"]})[0][\"Status\"][\"State\"]\n\n    def _has_service_terminated(self) -> bool:\n        status = self._service_status()\n        return status in [\"complete\", \"failed\", \"shutdown\", \"rejected\", \"orphaned\", \"remove\"]\n\n    def _stream_logs_to_output(self) -> None:\n        if not self.service:\n            raise Exception(\"The 'service' should be initialized before!\")\n        logs = self.cli.service_logs(\n            self.service[\"ID\"], follow=True, stdout=True, stderr=True, is_tty=self.tty\n        )\n        line = \"\"\n        for log in logs:\n            try:\n                log = log.decode()\n            except UnicodeDecodeError:\n                continue\n            if log == \"\\n\":\n                self.log.info(line)\n                line = \"\"\n            else:\n                line += log\n        # flush any remaining log stream\n        if line:\n            self.log.info(line)\n\n    def on_kill(self) -> None:\n        if self.hook.client_created and self.service is not None:\n            self.log.info(\"Removing docker service: %s\", self.service[\"ID\"])\n            self.cli.remove_service(self.service[\"ID\"])\n", "path": "airflow/providers/docker/operators/docker_swarm.py"}], "after_files": [{"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"Run ephemeral Docker Swarm services.\"\"\"\nfrom __future__ import annotations\n\nimport re\nfrom datetime import datetime\nfrom time import sleep\nfrom typing import TYPE_CHECKING\n\nfrom docker import types\n\nfrom airflow.exceptions import AirflowException\nfrom airflow.providers.docker.operators.docker import DockerOperator\nfrom airflow.utils.strings import get_random_string\n\nif TYPE_CHECKING:\n    from airflow.utils.context import Context\n\n\nclass DockerSwarmOperator(DockerOperator):\n    \"\"\"\n    Execute a command as an ephemeral docker swarm service.\n\n    Example use-case - Using Docker Swarm orchestration to make one-time\n    scripts highly available.\n\n    A temporary directory is created on the host and\n    mounted into a container to allow storing files\n    that together exceed the default disk size of 10GB in a container.\n    The path to the mounted directory can be accessed\n    via the environment variable ``AIRFLOW_TMP_DIR``.\n\n    If a login to a private registry is required prior to pulling the image, a\n    Docker connection needs to be configured in Airflow and the connection ID\n    be provided with the parameter ``docker_conn_id``.\n\n    :param image: Docker image from which to create the container.\n        If image tag is omitted, \"latest\" will be used.\n    :param api_version: Remote API version. Set to ``auto`` to automatically\n        detect the server's version.\n    :param auto_remove: Auto-removal of the container on daemon side when the\n        container's process exits.\n        The default is False.\n    :param command: Command to be run in the container. (templated)\n    :param docker_url: URL of the host running the docker daemon.\n        Default is unix://var/run/docker.sock\n    :param environment: Environment variables to set in the container. (templated)\n    :param force_pull: Pull the docker image on every run. Default is False.\n    :param mem_limit: Maximum amount of memory the container can use.\n        Either a float value, which represents the limit in bytes,\n        or a string like ``128m`` or ``1g``.\n    :param tls_ca_cert: Path to a PEM-encoded certificate authority\n        to secure the docker connection.\n    :param tls_client_cert: Path to the PEM-encoded certificate\n        used to authenticate docker client.\n    :param tls_client_key: Path to the PEM-encoded key used to authenticate docker client.\n    :param tls_hostname: Hostname to match against\n        the docker server certificate or False to disable the check.\n    :param tls_ssl_version: Version of SSL to use when communicating with docker daemon.\n    :param tmp_dir: Mount point inside the container to\n        a temporary directory created on the host by the operator.\n        The path is also made available via the environment variable\n        ``AIRFLOW_TMP_DIR`` inside the container.\n    :param user: Default user inside the docker container.\n    :param docker_conn_id: The :ref:`Docker connection id <howto/connection:docker>`\n    :param tty: Allocate pseudo-TTY to the container of this service\n        This needs to be set see logs of the Docker container / service.\n    :param enable_logging: Show the application's logs in operator's logs.\n        Supported only if the Docker engine is using json-file or journald logging drivers.\n        The `tty` parameter should be set to use this with Python applications.\n    :param configs: List of docker configs to be exposed to the containers of the swarm service.\n        The configs are ConfigReference objects as per the docker api\n        [https://docker-py.readthedocs.io/en/stable/services.html#docker.models.services.ServiceCollection.create]_\n    :param secrets: List of docker secrets to be exposed to the containers of the swarm service.\n        The secrets are SecretReference objects as per the docker create_service api.\n        [https://docker-py.readthedocs.io/en/stable/services.html#docker.models.services.ServiceCollection.create]_\n    :param mode: Indicate whether a service should be deployed as a replicated or global service,\n        and associated parameters\n    :param networks: List of network names or IDs or NetworkAttachmentConfig to attach the service to.\n    :param placement: Placement instructions for the scheduler. If a list is passed instead,\n        it is assumed to be a list of constraints as part of a Placement object.\n    \"\"\"\n\n    def __init__(\n        self,\n        *,\n        image: str,\n        enable_logging: bool = True,\n        configs: list[types.ConfigReference] | None = None,\n        secrets: list[types.SecretReference] | None = None,\n        mode: types.ServiceMode | None = None,\n        networks: list[str | types.NetworkAttachmentConfig] | None = None,\n        placement: types.Placement | list[types.Placement] | None = None,\n        **kwargs,\n    ) -> None:\n        super().__init__(image=image, **kwargs)\n        self.enable_logging = enable_logging\n        self.service = None\n        self.configs = configs\n        self.secrets = secrets\n        self.mode = mode\n        self.networks = networks\n        self.placement = placement\n\n    def execute(self, context: Context) -> None:\n        self.environment[\"AIRFLOW_TMP_DIR\"] = self.tmp_dir\n        return self._run_service()\n\n    def _run_service(self) -> None:\n        self.log.info(\"Starting docker service from image %s\", self.image)\n        self.service = self.cli.create_service(\n            types.TaskTemplate(\n                container_spec=types.ContainerSpec(\n                    image=self.image,\n                    command=self.format_command(self.command),\n                    mounts=self.mounts,\n                    env=self.environment,\n                    user=self.user,\n                    tty=self.tty,\n                    configs=self.configs,\n                    secrets=self.secrets,\n                ),\n                restart_policy=types.RestartPolicy(condition=\"none\"),\n                resources=types.Resources(mem_limit=self.mem_limit),\n                networks=self.networks,\n                placement=self.placement,\n            ),\n            name=f\"airflow-{get_random_string()}\",\n            labels={\"name\": f\"airflow__{self.dag_id}__{self.task_id}\"},\n            mode=self.mode,\n        )\n        if self.service is None:\n            raise Exception(\"Service should be set here\")\n        self.log.info(\"Service started: %s\", self.service)\n\n        # wait for the service to start the task\n        while not self.cli.tasks(filters={\"service\": self.service[\"ID\"]}):\n            continue\n\n        if self.enable_logging:\n            self._stream_logs_to_output()\n\n        while True:\n            if self._has_service_terminated():\n                self.log.info(\"Service status before exiting: %s\", self._service_status())\n                break\n\n        self.log.info(\"auto_removeauto_removeauto_removeauto_removeauto_remove : %s\", str(self.auto_remove))\n        if self.service and self._service_status() != \"complete\":\n            if self.auto_remove == \"success\":\n                self.cli.remove_service(self.service[\"ID\"])\n            raise AirflowException(f\"Service did not complete: {self.service!r}\")\n        elif self.auto_remove == \"success\":\n            if not self.service:\n                raise Exception(\"The 'service' should be initialized before!\")\n            self.cli.remove_service(self.service[\"ID\"])\n\n    def _service_status(self) -> str | None:\n        if not self.service:\n            raise Exception(\"The 'service' should be initialized before!\")\n        return self.cli.tasks(filters={\"service\": self.service[\"ID\"]})[0][\"Status\"][\"State\"]\n\n    def _has_service_terminated(self) -> bool:\n        status = self._service_status()\n        return status in [\"complete\", \"failed\", \"shutdown\", \"rejected\", \"orphaned\", \"remove\"]\n\n    def _stream_logs_to_output(self) -> None:\n        if not self.service:\n            raise Exception(\"The 'service' should be initialized before!\")\n        last_line_logged, last_timestamp = \"\", 0\n\n        def stream_new_logs(last_line_logged, since=0):\n            logs = self.cli.service_logs(\n                self.service[\"ID\"],\n                follow=False,\n                stdout=True,\n                stderr=True,\n                is_tty=self.tty,\n                since=since,\n                timestamps=True,\n            )\n            logs = b\"\".join(logs).decode().splitlines()\n            if last_line_logged in logs:\n                logs = logs[logs.index(last_line_logged) + 1 :]\n            for line in logs:\n                match = re.match(r\"(\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}.\\d{6,}Z) (.*)\", line)\n                timestamp, message = match.groups()\n                self.log.info(message)\n            # Floor nanoseconds to microseconds\n            last_timestamp = re.sub(r\"(\\.\\d{6})\\d+Z\", r\"\\1Z\", timestamp)\n            last_timestamp = datetime.strptime(last_timestamp, \"%Y-%m-%dT%H:%M:%S.%fZ\")\n            last_timestamp = last_timestamp.timestamp()\n            return last_line_logged, last_timestamp\n\n        while not self._has_service_terminated():\n            sleep(2)\n            last_line_logged, last_timestamp = stream_new_logs(last_line_logged, since=last_timestamp)\n\n    def on_kill(self) -> None:\n        if self.hook.client_created and self.service is not None:\n            self.log.info(\"Removing docker service: %s\", self.service[\"ID\"])\n            self.cli.remove_service(self.service[\"ID\"])\n", "path": "airflow/providers/docker/operators/docker_swarm.py"}]}