class_docstr: Execute a command as an ephemeral docker swarm service.
  
  Example use-case - Using Docker Swarm orchestration to make one-time
  scripts highly available.
  
  A temporary directory is created on the host and
  mounted into a container to allow storing files
  that together exceed the default disk size of 10GB in a container.
  The path to the mounted directory can be accessed
  via the environment variable ``AIRFLOW_TMP_DIR``.
  
  If a login to a private registry is required prior to pulling the image, a
  Docker connection needs to be configured in Airflow and the connection ID
  be provided with the parameter ``docker_conn_id``.
  
  :param image: Docker image from which to create the container.
      If image tag is omitted, "latest" will be used.
  :param api_version: Remote API version. Set to ``auto`` to automatically
      detect the server's version.
  :param auto_remove: Auto-removal of the container on daemon side when the
      container's process exits.
      The default is False.
  :param command: Command to be run in the container. (templated)
  :param docker_url: URL of the host running the docker daemon.
      Default is unix://var/run/docker.sock
  :param environment: Environment variables to set in the container. (templated)
  :param force_pull: Pull the docker image on every run. Default is False.
  :param mem_limit: Maximum amount of memory the container can use.
      Either a float value, which represents the limit in bytes,
      or a string like ``128m`` or ``1g``.
  :param tls_ca_cert: Path to a PEM-encoded certificate authority
      to secure the docker connection.
  :param tls_client_cert: Path to the PEM-encoded certificate
      used to authenticate docker client.
  :param tls_client_key: Path to the PEM-encoded key used to authenticate docker client.
  :param tls_hostname: Hostname to match against
      the docker server certificate or False to disable the check.
  :param tls_ssl_version: Version of SSL to use when communicating with docker daemon.
  :param tmp_dir: Mount point inside the container to
      a temporary directory created on the host by the operator.
      The path is also made available via the environment variable
      ``AIRFLOW_TMP_DIR`` inside the container.
  :param user: Default user inside the docker container.
  :param docker_conn_id: The :ref:`Docker connection id <howto/connection:docker>`
  :param tty: Allocate pseudo-TTY to the container of this service
      This needs to be set see logs of the Docker container / service.
  :param enable_logging: Show the application's logs in operator's logs.
      Supported only if the Docker engine is using json-file or journald logging drivers.
      The `tty` parameter should be set to use this with Python applications.
  :param configs: List of docker configs to be exposed to the containers of the swarm service.
      The configs are ConfigReference objects as per the docker api
      [https://docker-py.readthedocs.io/en/stable/services.html#docker.models.services.ServiceCollection.create]_
  :param secrets: List of docker secrets to be exposed to the containers of the swarm service.
      The secrets are SecretReference objects as per the docker create_service api.
      [https://docker-py.readthedocs.io/en/stable/services.html#docker.models.services.ServiceCollection.create]_
  :param mode: Indicate whether a service should be deployed as a replicated or global service,
      and associated parameters
  :param networks: List of network names or IDs or NetworkAttachmentConfig to attach the service to.
  :param placement: Placement instructions for the scheduler. If a list is passed instead,
      it is assumed to be a list of constraints as part of a Placement object.
class_docstr_tok: ["Execute","a","command","as","an","ephemeral","docker","swarm","service",".","Example","use-case","-","Using","Docker","Swarm","orchestration","to","make","one-time","scripts","highly","available",".","A","temporary","directory","is","created","on","the","host","and","mounted","into","a","container","to","allow","storing","files","that","together","exceed","the","default","disk","size","of","10GB","in","a","container",".","The","path","to","the","mounted","directory","can","be","accessed","via","the","environment","variable","``","AIRFLOW_TMP_DIR","``",".","If","a","login","to","a","private","registry","is","required","prior","to","pulling","the","image",",","a","Docker","connection","needs","to","be","configured","in","Airflow","and","the","connection","ID","be","provided","with","the","parameter","``","docker_conn_id","``",".",":","param","image",":","Docker","image","from","which","to","create","the","container",".","If","image","tag","is","omitted",",","``","latest","''","will","be","used",".",":","param","api_version",":","Remote","API","version",".","Set","to","``","auto","``","to","automatically","detect","the","server","'s","version",".",":","param","auto_remove",":","Auto-removal","of","the","container","on","daemon","side","when","the","container","'s","process","exits",".","The","default","is","False",".",":","param","command",":","Command","to","be","run","in","the","container",".","(","templated",")",":","param","docker_url",":","URL","of","the","host","running","the","docker","daemon",".","Default","is","unix",":","\/\/var\/run\/docker.sock",":","param","environment",":","Environment","variables","to","set","in","the","container",".","(","templated",")",":","param","force_pull",":","Pull","the","docker","image","on","every","run",".","Default","is","False",".",":","param","mem_limit",":","Maximum","amount","of","memory","the","container","can","use",".","Either","a","float","value",",","which","represents","the","limit","in","bytes",",","or","a","string","like","``","128m","``","or","``","1g","``",".",":","param","tls_ca_cert",":","Path","to","a","PEM-encoded","certificate","authority","to","secure","the","docker","connection",".",":","param","tls_client_cert",":","Path","to","the","PEM-encoded","certificate","used","to","authenticate","docker","client",".",":","param","tls_client_key",":","Path","to","the","PEM-encoded","key","used","to","authenticate","docker","client",".",":","param","tls_hostname",":","Hostname","to","match","against","the","docker","server","certificate","or","False","to","disable","the","check",".",":","param","tls_ssl_version",":","Version","of","SSL","to","use","when","communicating","with","docker","daemon",".",":","param","tmp_dir",":","Mount","point","inside","the","container","to","a","temporary","directory","created","on","the","host","by","the","operator",".","The","path","is","also","made","available","via","the","environment","variable","``","AIRFLOW_TMP_DIR","``","inside","the","container",".",":","param","user",":","Default","user","inside","the","docker","container",".",":","param","docker_conn_id",":","The",":","ref",":","`","Docker","connection","id","<","howto\/connection",":","docker",">","`",":","param","tty",":","Allocate","pseudo-TTY","to","the","container","of","this","service","This","needs","to","be","set","see","logs","of","the","Docker","container","\/","service",".",":","param","enable_logging",":","Show","the","application","'s","logs","in","operator","'s","logs",".","Supported","only","if","the","Docker","engine","is","using","json-file","or","journald","logging","drivers",".","The","`","tty","`","parameter","should","be","set","to","use","this","with","Python","applications",".",":","param","configs",":","List","of","docker","configs","to","be","exposed","to","the","containers","of","the","swarm","service",".","The","configs","are","ConfigReference","objects","as","per","the","docker","api","[","https",":","\/\/docker-py.readthedocs.io\/en\/stable\/services.html","#","docker.models.services.ServiceCollection.create","]","_",":","param","secrets",":","List","of","docker","secrets","to","be","exposed","to","the","containers","of","the","swarm","service",".","The","secrets","are","SecretReference","objects","as","per","the","docker","create_service","api",".","[","https",":","\/\/docker-py.readthedocs.io\/en\/stable\/services.html","#","docker.models.services.ServiceCollection.create","]","_",":","param","mode",":","Indicate","whether","a","service","should","be","deployed","as","a","replicated","or","global","service",",","and","associated","parameters",":","param","networks",":","List","of","network","names","or","IDs","or","NetworkAttachmentConfig","to","attach","the","service","to",".",":","param","placement",":","Placement","instructions","for","the","scheduler",".","If","a","list","is","passed","instead",",","it","is","assumed","to","be","a","list","of","constraints","as","part","of","a","Placement","object","."]
code: def _stream_logs_to_output(self) -> None:
      if not self.service:
          raise Exception("The 'service' should be initialized before!")
      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)
  
          if len(logs) == 0:
              return last_line_logged, since
  
          last_line_logged = line
  
          # 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
          )
  
file_path: airflow/airflow/providers/docker/operators/docker_swarm.py
imports: from __future__ import annotations
  import re
  from datetime import datetime
  from time import sleep
  from typing import TYPE_CHECKING
  from docker import types
  from airflow.exceptions import AirflowException
  from airflow.providers.docker.operators.docker import DockerOperator
  from airflow.utils.strings import get_random_string
label_desc: Use image node_id 6 for calling the DockerSwarmOperator obj's underlying member method code with example usage: obj._stream_logs_to_output() and returns: last_line_logged,  last_timestamp, last_line_logged,  since