code: def test_autoscaling(
      printer: Callable,
      client: cx.Client,
      apis: Dict[str, Any],
      autoscaling_config: Dict[str, Union[int, float]],
      deploy_timeout: int = None,
      api_config_name: str = "cortex_cpu.yaml",
      node_groups: List[str] = [],
  ):
      max_replicas = autoscaling_config["max_replicas"]
      query_params = apis["query_params"]
  
      # increase the concurrency by 1 to ensure we get max_replicas replicas
      concurrency = max_replicas + 1
  
      all_apis = [apis["primary"]] + apis["dummy"]
      all_api_names = []
      for api in all_apis:
          api_dir = TEST_APIS_DIR / api
          with open(str(api_dir / api_config_name)) as f:
              api_specs = yaml.safe_load(f)
          assert len(api_specs) == 1
  
          if len(node_groups) > 0:
              api_specs[0]["node_groups"] = node_groups
          api_specs[0]["autoscaling"] = {
              "max_replicas": max_replicas,
              "downscale_stabilization_period": "1m",
          }
  
          all_api_names.append(api_specs[0]["name"])
          client.deploy(api_spec=api_specs[0])
  
      primary_api_name = all_api_names[0]
      autoscaling = client.get_api(primary_api_name)["spec"][
          "autoscaling"
      ]
  
      # controls the flow of requests
      request_stopper = td.Event()
  
      # determine upscale/downscale replica requests
      current_replicas = 1  # starting number of replicas
      test_timeout = 0  # measured in seconds
      while current_replicas < max_replicas:
          upscale_ceil = math.ceil(
              current_replicas * autoscaling["max_upscale_factor"]
          )
          if upscale_ceil > current_replicas + 1:
              current_replicas = upscale_ceil
          else:
              current_replicas += 1
          if current_replicas > max_replicas:
              current_replicas = max_replicas
          test_timeout += int(
              autoscaling["upscale_stabilization_period"] / (1000**3)
          )
      while current_replicas > 1:
          downscale_ceil = math.ceil(
              current_replicas * autoscaling["max_downscale_factor"]
          )
          if downscale_ceil < current_replicas - 1:
              current_replicas = downscale_ceil
          else:
              current_replicas -= 1
          test_timeout += int(
              autoscaling["downscale_stabilization_period"]
              / (1000**3)
          )
  
      # add overhead to the test timeout to account for the process of downloading images or adding nodes to the cluster
      test_timeout *= 2
  
      try:
          assert apis_ready(
              client=client,
              api_names=all_api_names,
              timeout=deploy_timeout,
          ), f"apis {all_api_names} not ready"
  
          threads_futures = make_requests_concurrently(
              client,
              primary_api_name,
              concurrency,
              request_stopper,
              query_params=query_params,
          )
  
          test_start_time = time.time()
  
          # upscale/downscale the api
          printer(f"scaling up to {max_replicas} replicas")
          while True:
              assert api_updated(
                  client, primary_api_name, timeout=deploy_timeout
              ), "api didn't scale up to the desired number of replicas in time"
              current_replicas = client.get_api(primary_api_name)[
                  "status"
              ]["requested"]
  
              # stop the requests from being made
              if (
                  current_replicas == max_replicas
                  and not request_stopper.is_set()
              ):
                  printer(f"scaling back down to 1 replica")
                  request_stopper.set()
  
              # check if the requesting threads are still healthy
              # if not, they'll raise an exception
              check_futures_healthy(threads_futures)
  
              # check if the test is taking too much time
              assert (
                  time.time() - test_start_time < test_timeout
              ), f"autoscaling test for api {primary_api_name} did not finish in {test_timeout}s; current number of replicas is {current_replicas}/{concurrency}"
  
              # stop the test if it has finished
              if current_replicas == 1 and request_stopper.is_set():
                  break
  
              # add some delay to reduce the number of gets
              time.sleep(1)
  
      except:
          # best effort
          try:
              api_info = client.get_api(primary_api_name)
              printer(json.dumps(api_info, indent=2))
          finally:
              raise
      finally:
          request_stopper.set()
          delete_apis(client, all_api_names)
  
imports: import json
  import math
  import os
  import re
  import threading
  import time
  from http import HTTPStatus
  from pathlib import Path
  from typing import Callable, Dict, Any, List, Union
  import boto3
  import cortex
  import requests
  import yaml
  from e2e.expectations import parse_expectations, assert_response_expectations, assert_json_expectations
  from e2e.generator import load_generator
  from e2e.utils import apis_ready, api_updated, wait_on_event, wait_on_futures, endpoint_ready, post_request, get_request, job_done, jobs_done, request_batch_prediction, request_task, retrieve_async_result, make_requests_concurrently, check_futures_healthy, retrieve_results_concurrently, stream_api_logs, stream_job_logs, wait_for
label_desc: Use image node_id 6 for calling a global function with example usage: test_autoscaling(printer, client, apis, autoscaling_config, deploy_timeout, api_config_name, node_groups) without return types