docstr: Spin up or down resources as necessary.
  - For every queue in self.queues do the following:
      1. Check if there are tasks waiting in the queue.
      2. Check if there are enough idle workers available for those tasks.
      3. In case more instances are required, and we haven't reached max instances allowed,
         create the required instances with regards to the maximum number defined in self.queues
         Choose which instance to create according to their order in self.queues. Won't create more instances
         if maximum number defined has already reached.
  - spin down instances according to their idle time. instance which is idle for more than self.max_idle_time_min
  minutes would be removed.
docstr_tok: ["Spin","up","or","down","resources","as","necessary",".","-","For","every","queue","in","self.queues","do","the","following",":","1",".","Check","if","there","are","tasks","waiting","in","the","queue",".","2",".","Check","if","there","are","enough","idle","workers","available","for","those","tasks",".","3",".","In","case","more","instances","are","required",",","and","we","have","n't","reached","max","instances","allowed",",","create","the","required","instances","with","regards","to","the","maximum","number","defined","in","self.queues","Choose","which","instance","to","create","according","to","their","order","in","self.queues",".","Wo","n't","create","more","instances","if","maximum","number","defined","has","already","reached",".","-","spin","down","instances","according","to","their","idle","time",".","instance","which","is","idle","for","more","than","self.max_idle_time_min","minutes","would","be","removed","."]
code: def supervisor(self):
      """
      Spin up or down resources as necessary.
      - For every queue in self.queues do the following:
          1. Check if there are tasks waiting in the queue.
          2. Check if there are enough idle workers available for those tasks.
          3. In case more instances are required, and we haven't reached max instances allowed,
             create the required instances with regards to the maximum number defined in self.queues
             Choose which instance to create according to their order in self.queues. Won't create more instances
             if maximum number defined has already reached.
      - spin down instances according to their idle time. instance which is idle for more than self.max_idle_time_min
      minutes would be removed.
      """
      self.ensure_queues()
  
      idle_workers = {}
      # Workers that we spun but have not yet reported back to the API
      spun_workers = {}  # worker_id -> (resource type, spin time)
      previous_workers = set()
      unknown_workers = deque(maxlen=256)
      task_logger = get_task_logger()
      up_machines = defaultdict(int)
  
      while self._running():
          queue_id_to_name, queue_name_to_id = self.queue_mapping()
          all_workers = self.get_workers()
  
          # update spun_workers (remove instances that are fully registered)
          for worker in all_workers:
              if worker.id not in previous_workers:
                  if not spun_workers.pop(worker.id, None):
                      if worker.id not in unknown_workers:
                          self.logger.info(
                              "Removed unknown worker from spun_workers: %s",
                              worker.id,
                          )
                          unknown_workers.append(worker.id)
                  else:
                      previous_workers.add(worker.id)
  
          for worker_id in self.stale_workers(spun_workers):
              out = spun_workers.pop(worker_id, None)
              if out is None:
                  self.logger.warning(
                      "Ignoring unknown stale worker: %r", worker_id
                  )
                  continue
              resource = out[0]
              try:
                  self.logger.info(
                      "Spinning down stuck worker: %r", worker_id
                  )
                  self.driver.spin_down_worker(
                      WorkerId(worker_id).cloud_id
                  )
                  up_machines[resource] -= 1
              except Exception as err:
                  self.logger.info(
                      "Cannot spin down %r: %r", worker_id, err
                  )
  
          self.update_idle_workers(all_workers, idle_workers)
          required_idle_resources = (
              []
          )  # idle resources we'll need to keep running
          allocate_new_resources = self.extra_allocations()
  
          # Check if we have tasks waiting on one of the designated queues
          for queue in self.queues:
              entries = self.api_client.queues.get_by_id(
                  queue_name_to_id[queue]
              ).entries
              self.logger.info(
                  "Found %d tasks in queue %r", len(entries), queue
              )
              if entries and len(entries) > 0:
                  queue_resources = self.queues[queue]
  
                  # If we have an idle worker matching the required resource,
                  # remove it from the required allocation resources
                  free_queue_resources = [
                      resource_name
                      for _, resource_name, _ in idle_workers.values()
                      if any(
                          q_r
                          for q_r in queue_resources
                          if resource_name == q_r[0]
                      )
                  ]
                  # if we have an instance waiting to be spun
                  # remove it from the required allocation resources
                  for resource, _ in spun_workers.values():
                      if resource in [qr[0] for qr in queue_resources]:
                          free_queue_resources.append(resource)
  
                  required_idle_resources.extend(free_queue_resources)
                  spin_up_count = len(entries) - len(
                      free_queue_resources
                  )
                  spin_up_resources = []
  
                  # Add as many resources as possible to handle this queue's entries
                  for resource, max_instances in queue_resources:
                      if len(spin_up_resources) >= spin_up_count:
                          break
                      # check if we can add instances to `resource`
                      currently_running_workers = len(
                          [
                              worker
                              for worker in all_workers
                              if WorkerId(worker.id).name == resource
                          ]
                      )
                      spun_up_workers = sum(
                          1
                          for r, _ in spun_workers.values()
                          if r == resource
                      )
                      max_allowed = (
                          int(max_instances)
                          - currently_running_workers
                          - spun_up_workers
                      )
                      if max_allowed > 0:
                          spin_up_resources.extend(
                              [resource]
                              * min(spin_up_count, max_allowed)
                          )
                  allocate_new_resources.extend(spin_up_resources)
  
          # Now we actually spin the new machines
          for resource in allocate_new_resources:
              task_id = None
              try:
                  if isinstance(resource, tuple):
                      worker_id, task_id = resource
                      resource = WorkerId(worker_id).name
  
                  queue = self.resource_to_queue[resource]
                  suffix = (
                      ", task_id={!r}".format(task_id)
                      if task_id
                      else ""
                  )
                  self.logger.info(
                      "Spinning new instance resource=%r, prefix=%r, queue=%r%s",
                      resource,
                      self.workers_prefix,
                      queue,
                      suffix,
                  )
                  resource_conf = self.resource_configurations[resource]
                  worker_prefix = self.gen_worker_prefix(
                      resource, resource_conf
                  )
                  instance_id = self.driver.spin_up_worker(
                      resource_conf,
                      worker_prefix,
                      queue,
                      task_id=task_id,
                  )
                  self.monitor_startup(instance_id)
                  worker_id = "{}:{}".format(worker_prefix, instance_id)
                  self.logger.info("New instance ID: %s", instance_id)
                  spun_workers[worker_id] = (resource, time())
                  up_machines[resource] += 1
              except Exception as ex:
                  self.logger.exception(
                      "Failed to start new instance (resource %r), Error: %s",
                      resource,
                      ex,
                  )
  
          # Go over the idle workers list, and spin down idle workers
          for worker_id in list(idle_workers):
              timestamp, resource_name, worker = idle_workers[worker_id]
              # skip resource types that might be needed
              if resource_name in required_idle_resources:
                  continue
              # Remove from both cloud and clearml all instances that are idle for longer than MAX_IDLE_TIME_MIN
              if time() - timestamp > self.max_idle_time_min * MINUTE:
                  wid = WorkerId(worker_id)
                  cloud_id = wid.cloud_id
                  self.driver.spin_down_worker(cloud_id)
                  up_machines[wid.name] -= 1
                  self.logger.info(
                      "Spin down instance cloud id %r", cloud_id
                  )
                  idle_workers.pop(worker_id, None)
  
          if task_logger:
              self.report_app_stats(
                  task_logger,
                  queue_id_to_name,
                  up_machines,
                  idle_workers,
              )
  
          # Nothing else to do
          self.logger.info(
              "Idle for %.2f seconds",
              self.polling_interval_time_min * MINUTE,
          )
          sleep(self.polling_interval_time_min * MINUTE)
  
imports: import re
  from collections import defaultdict, deque
  from enum import Enum
  from itertools import chain
  from threading import Event, Thread
  from time import sleep, time
  import attr
  from attr.validators import instance_of
  from .cloud_driver import CloudDriver
  from ..None import Task
  from ..backend_api import Session
  from ..backend_api.session import defs
  from ..backend_api.session.client import APIClient
  from ..debugging import get_logger
label_desc: Use image node_id 12 for calling the AutoScaler obj's underlying member method code with example usage: obj.supervisor() without return types