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: Make a test API call to check the health of the ECS Executor.
  
  Deliberately use an invalid task ID, some potential outcomes in order:
    1. `AccessDeniedException` is raised if there are insufficient permissions.
    2. `ClusterNotFoundException` is raised if permissions exist but the cluster does not.
    3. The API responds with a failure message if the cluster is found and there
       are permissions, but the cluster itself has issues.
    4. `InvalidParameterException` is raised if the permissions and cluster exist but the task does not.
  
  The last one is considered a success state for the purposes of this check.
docstr_tok: ["Make","a","test","API","call","to","check","the","health","of","the","ECS","Executor",".","Deliberately","use","an","invalid","task","ID",",","some","potential","outcomes","in","order",":","1",".","`","AccessDeniedException","`","is","raised","if","there","are","insufficient","permissions",".","2",".","`","ClusterNotFoundException","`","is","raised","if","permissions","exist","but","the","cluster","does","not",".","3",".","The","API","responds","with","a","failure","message","if","the","cluster","is","found","and","there","are","permissions",",","but","the","cluster","itself","has","issues",".","4",".","`","InvalidParameterException","`","is","raised","if","the","permissions","and","cluster","exist","but","the","task","does","not",".","The","last","one","is","considered","a","success","state","for","the","purposes","of","this","check","."]
code: def check_health(self):
      """
      Make a test API call to check the health of the ECS Executor.
  
      Deliberately use an invalid task ID, some potential outcomes in order:
        1. `AccessDeniedException` is raised if there are insufficient permissions.
        2. `ClusterNotFoundException` is raised if permissions exist but the cluster does not.
        3. The API responds with a failure message if the cluster is found and there
           are permissions, but the cluster itself has issues.
        4. `InvalidParameterException` is raised if the permissions and cluster exist but the task does not.
  
      The last one is considered a success state for the purposes of this check.
      """
      success_status = "succeeded."
      status = success_status
  
      try:
          invalid_task_id = "a" * 32
          self.ecs.stop_task(cluster=self.cluster, task=invalid_task_id)
  
          # If it got this far, something is wrong.  stop_task() called with an
          # invalid taskID should have thrown a ClientError.  All known reasons are
          # covered in the ``except`` block below, and this should never be reached.
          status = "failed for an unknown reason. "
      except ClientError as ex:
          error_code = ex.response["Error"]["Code"]
          error_message = ex.response["Error"]["Message"]
  
          if ("InvalidParameterException" in error_code) and (
              "task was not found" in error_message
          ):
              # This failure is expected, and means we're healthy
              pass
          else:
              # Catch all for unexpected failures
              status = f"failed because: {error_message}. "
      except Exception as e:
          # Any non-ClientError exceptions. This can include Botocore exceptions for example
          status = f"failed because: {e}. "
      finally:
          msg_prefix = "ECS Executor health check has %s"
          if status == success_status:
              self.IS_BOTO_CONNECTION_HEALTHY = True
              self.log.info(msg_prefix, status)
          else:
              msg_error_suffix = "The ECS executor will not be able to run Airflow tasks until the issue is addressed."
              raise AirflowException(
                  msg_prefix % status + msg_error_suffix
              )
  
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 3 for calling the AwsEcsExecutor obj's underlying member method code with example usage: obj.check_health() without return types