class_docstr: Executes the provided Airflow command on an ECS instance.
  
  The Airflow Scheduler creates a shell command, and passes it to the executor. This ECS Executor
  runs said Airflow command on a remote Amazon ECS Cluster with a task-definition configured to
  launch the same containers as the Scheduler. It then periodically checks in with the launched
  tasks (via task ARNs) to determine the status.
  
  This allows individual tasks to specify CPU, memory, GPU, env variables, etc. When initializing a task,
  there's an option for "executor config" which should be a dictionary with keys that match the
  ``ContainerOverride`` definition per AWS documentation (see link below).
  
  Prerequisite: proper configuration of Boto3 library
  .. seealso:: https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html for
  authentication and access-key management. You can store an environmental variable, setup aws config from
  console, or use IAM roles.
  
  .. seealso:: https://docs.aws.amazon.com/AmazonECS/latest/APIReference/API_ContainerOverride.html for an
   Airflow TaskInstance's executor_config.
class_docstr_tok: ["Executes","the","provided","Airflow","command","on","an","ECS","instance",".","The","Airflow","Scheduler","creates","a","shell","command",",","and","passes","it","to","the","executor",".","This","ECS","Executor","runs","said","Airflow","command","on","a","remote","Amazon","ECS","Cluster","with","a","task-definition","configured","to","launch","the","same","containers","as","the","Scheduler",".","It","then","periodically","checks","in","with","the","launched","tasks","(","via","task","ARNs",")","to","determine","the","status",".","This","allows","individual","tasks","to","specify","CPU",",","memory",",","GPU",",","env","variables",",","etc",".","When","initializing","a","task",",","there","'s","an","option","for","``","executor","config","''","which","should","be","a","dictionary","with","keys","that","match","the","``","ContainerOverride","``","definition","per","AWS","documentation","(","see","link","below",")",".","Prerequisite",":","proper","configuration","of","Boto3","library","..","seealso",":",":","https",":","\/\/boto3.amazonaws.com\/v1\/documentation\/api\/latest\/guide\/configuration.html","for","authentication","and","access-key","management",".","You","can","store","an","environmental","variable",",","setup","aws","config","from","console",",","or","use","IAM","roles",".","..","seealso",":",":","https",":","\/\/docs.aws.amazon.com\/AmazonECS\/latest\/APIReference\/API_ContainerOverride.html","for","an","Airflow","TaskInstance","'s","executor_config","."]
docstr: Takes tasks from the pending_tasks queue, and attempts to find an instance to run it on.
  
  If the launch type is EC2, this will attempt to place tasks on empty EC2 instances.  If
      there are no EC2 instances available, no task is placed and this function will be
      called again in the next heart-beat.
  
  If the launch type is FARGATE, this will run the tasks on new AWS Fargate instances.
docstr_tok: ["Takes","tasks","from","the","pending_tasks","queue",",","and","attempts","to","find","an","instance","to","run","it","on",".","If","the","launch","type","is","EC2",",","this","will","attempt","to","place","tasks","on","empty","EC2","instances",".","If","there","are","no","EC2","instances","available",",","no","task","is","placed","and","this","function","will","be","called","again","in","the","next","heart-beat",".","If","the","launch","type","is","FARGATE",",","this","will","run","the","tasks","on","new","AWS","Fargate","instances","."]
code: def attempt_task_runs(self):
      """
      Takes tasks from the pending_tasks queue, and attempts to find an instance to run it on.
  
      If the launch type is EC2, this will attempt to place tasks on empty EC2 instances.  If
          there are no EC2 instances available, no task is placed and this function will be
          called again in the next heart-beat.
  
      If the launch type is FARGATE, this will run the tasks on new AWS Fargate instances.
      """
      queue_len = len(self.pending_tasks)
      failure_reasons = defaultdict(int)
      for _ in range(queue_len):
          ecs_task = self.pending_tasks.popleft()
          task_key = ecs_task.key
          cmd = ecs_task.command
          queue = ecs_task.queue
          exec_config = ecs_task.executor_config
          attempt_number = ecs_task.attempt_number
          _failure_reasons = []
          try:
              run_task_response = self._run_task(
                  task_key, cmd, queue, exec_config
              )
          except NoCredentialsError:
              self.pending_tasks.appendleft(ecs_task)
              raise
          except ClientError as e:
              error_code = e.response["Error"]["Code"]
              if error_code in INVALID_CREDENTIALS_EXCEPTIONS:
                  self.pending_tasks.appendleft(ecs_task)
                  raise
              _failure_reasons.append(str(e))
          except Exception as e:
              # Failed to even get a response back from the Boto3 API or something else went
              # wrong.  For any possible failure we want to add the exception reasons to the
              # failure list so that it is logged to the user and most importantly the task is
              # added back to the pending list to be retried later.
              _failure_reasons.append(str(e))
          else:
              # We got a response back, check if there were failures. If so, add them to the
              # failures list so that it is logged to the user and most importantly the task
              # is added back to the pending list to be retried later.
              if run_task_response["failures"]:
                  _failure_reasons.extend(
                      [
                          f["reason"]
                          for f in run_task_response["failures"]
                      ]
                  )
  
          if _failure_reasons:
              for reason in _failure_reasons:
                  failure_reasons[reason] += 1
              # Make sure the number of attempts does not exceed MAX_RUN_TASK_ATTEMPTS
              if int(attempt_number) <= int(
                  self.__class__.MAX_RUN_TASK_ATTEMPTS
              ):
                  ecs_task.attempt_number += 1
                  self.pending_tasks.appendleft(ecs_task)
              else:
                  self.log.error(
                      "ECS task %s has failed a maximum of %s times. Marking as failed",
                      task_key,
                      attempt_number,
                  )
                  self.fail(task_key)
          elif not run_task_response["tasks"]:
              self.log.error(
                  "ECS RunTask Response: %s", run_task_response
              )
              raise EcsExecutorException(
                  "No failures and no ECS tasks provided in response. This should never happen."
              )
          else:
              task = run_task_response["tasks"][0]
              self.active_workers.add_task(
                  task,
                  task_key,
                  queue,
                  cmd,
                  exec_config,
                  attempt_number,
              )
      if failure_reasons:
          self.log.error(
              "Pending ECS tasks failed to launch for the following reasons: %s. Retrying later.",
              dict(failure_reasons),
          )
  
imports: from __future__ import annotations
  import time
  from collections import defaultdict, deque
  from copy import deepcopy
  from typing import TYPE_CHECKING
  from botocore.exceptions import ClientError, NoCredentialsError
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException
  from airflow.executors.base_executor import BaseExecutor
  from airflow.providers.amazon.aws.executors.ecs.boto_schema import BotoDescribeTasksSchema, BotoRunTaskSchema
  from airflow.providers.amazon.aws.executors.ecs.utils import CONFIG_DEFAULTS, CONFIG_GROUP_NAME, AllEcsConfigKeys, EcsExecutorException, EcsQueuedTask, EcsTaskCollection
  from airflow.providers.amazon.aws.executors.utils.exponential_backoff_retry import exponential_backoff_retry
  from airflow.providers.amazon.aws.hooks.ecs import EcsHook
  from airflow.utils import timezone
  from airflow.utils.state import State
label_desc: Use image node_id 11 for calling the AwsEcsExecutor obj's underlying member method code with example usage: obj.attempt_task_runs() without return types