class_docstr: e2e verification of autoscaling following the steps in the Ray documentation.
  kubectl is used throughout, as that reflects the instructions in the docs.
class_docstr_tok: ["e2e","verification","of","autoscaling","following","the","steps","in","the","Ray","documentation",".","kubectl","is","used","throughout",",","as","that","reflects","the","instructions","in","the","docs","."]
docstr: Test the following behaviors:
  
  1. Spinning up a Ray cluster
  2. Scaling up Ray workers via autoscaler.sdk.request_resources()
  3. Scaling up by updating the CRD's minReplicas
  4. Scaling down by removing the resource request and reducing maxReplicas
  5. Autoscaler recognizes GPU annotations and Ray custom resources.
  6. Autoscaler and operator ignore pods marked for deletion.
  7. Autoscaler logs work. Autoscaler events are piped to the driver.
  8. Ray utils show correct resource limits in the head container.
  
  Tests the following modes of interaction with a Ray cluster on K8s:
  1. kubectl exec
  2. Ray Client
  3. Ray Job Submission
  
  TODO (Dmitri): Split up the test logic.
  Too much is stuffed into this one test case.
  
  Resources requested by this test are safely within the bounds of an m5.xlarge
  instance.
  
  The resource REQUESTS are:
  - One Ray head pod
      - Autoscaler: .25 CPU, .5 Gi memory
      - Ray node: .5 CPU, .5 Gi memeory
  - Three Worker pods
      - Ray node: .5 CPU, .5 Gi memory
  Total: 2.25 CPU, 2.5 Gi memory.
  
  Including operator and system pods, the total CPU requested is around 3.
  
  The cpu LIMIT of each Ray container is 1.
  The `num-cpus` arg to Ray start is 1 for each Ray container; thus Ray accounts
  1 CPU for each Ray node in the test.
docstr_tok: ["Test","the","following","behaviors",":","1",".","Spinning","up","a","Ray","cluster","2",".","Scaling","up","Ray","workers","via","autoscaler.sdk.request_resources","(",")","3",".","Scaling","up","by","updating","the","CRD","'s","minReplicas","4",".","Scaling","down","by","removing","the","resource","request","and","reducing","maxReplicas","5",".","Autoscaler","recognizes","GPU","annotations","and","Ray","custom","resources",".","6",".","Autoscaler","and","operator","ignore","pods","marked","for","deletion",".","7",".","Autoscaler","logs","work",".","Autoscaler","events","are","piped","to","the","driver",".","8",".","Ray","utils","show","correct","resource","limits","in","the","head","container",".","Tests","the","following","modes","of","interaction","with","a","Ray","cluster","on","K8s",":","1.","kubectl","exec","2",".","Ray","Client","3",".","Ray","Job","Submission","TODO","(","Dmitri",")",":","Split","up","the","test","logic",".","Too","much","is","stuffed","into","this","one","test","case",".","Resources","requested","by","this","test","are","safely","within","the","bounds","of","an","m5.xlarge","instance",".","The","resource","REQUESTS","are",":","-","One","Ray","head","pod","-","Autoscaler",":",".25","CPU",",",".5","Gi","memory","-","Ray","node",":",".5","CPU",",",".5","Gi","memeory","-","Three","Worker","pods","-","Ray","node",":",".5","CPU",",",".5","Gi","memory","Total",":","2.25","CPU",",","2.5","Gi","memory",".","Including","operator","and","system","pods",",","the","total","CPU","requested","is","around","3",".","The","cpu","LIMIT","of","each","Ray","container","is","1",".","The","`","num-cpus","`","arg","to","Ray","start","is","1","for","each","Ray","container",";","thus","Ray","accounts","1","CPU","for","each","Ray","node","in","the","test","."]
code: def testAutoscaling(self):
      """Test the following behaviors:
  
      1. Spinning up a Ray cluster
      2. Scaling up Ray workers via autoscaler.sdk.request_resources()
      3. Scaling up by updating the CRD's minReplicas
      4. Scaling down by removing the resource request and reducing maxReplicas
      5. Autoscaler recognizes GPU annotations and Ray custom resources.
      6. Autoscaler and operator ignore pods marked for deletion.
      7. Autoscaler logs work. Autoscaler events are piped to the driver.
      8. Ray utils show correct resource limits in the head container.
  
      Tests the following modes of interaction with a Ray cluster on K8s:
      1. kubectl exec
      2. Ray Client
      3. Ray Job Submission
  
      TODO (Dmitri): Split up the test logic.
      Too much is stuffed into this one test case.
  
      Resources requested by this test are safely within the bounds of an m5.xlarge
      instance.
  
      The resource REQUESTS are:
      - One Ray head pod
          - Autoscaler: .25 CPU, .5 Gi memory
          - Ray node: .5 CPU, .5 Gi memeory
      - Three Worker pods
          - Ray node: .5 CPU, .5 Gi memory
      Total: 2.25 CPU, 2.5 Gi memory.
  
      Including operator and system pods, the total CPU requested is around 3.
  
      The cpu LIMIT of each Ray container is 1.
      The `num-cpus` arg to Ray start is 1 for each Ray container; thus Ray accounts
      1 CPU for each Ray node in the test.
      """
      switch_to_ray_parent_dir()
  
      # Cluster creation
      logger.info("Creating a RayCluster with no worker pods.")
      self._apply_ray_cr(min_replicas=0, cpu_replicas=0, gpu_replicas=0)
  
      logger.info("Confirming presence of head.")
      wait_for_pods(goal_num_pods=1, namespace=RAY_CLUSTER_NAMESPACE)
  
      logger.info("Waiting for head pod to start Running.")
      wait_for_pod_to_start(
          pod_name_filter=HEAD_POD_PREFIX,
          namespace=RAY_CLUSTER_NAMESPACE,
      )
      logger.info("Confirming Ray is up on the head pod.")
      wait_for_ray_health(
          pod_name_filter=HEAD_POD_PREFIX,
          namespace=RAY_CLUSTER_NAMESPACE,
      )
  
      head_pod = get_pod(
          pod_name_filter=HEAD_POD_PREFIX,
          namespace=RAY_CLUSTER_NAMESPACE,
      )
      assert head_pod, "Could not find the Ray head pod."
  
      # Confirm head pod resource allocation.
      # (This is a misplaced test of Ray's resource detection in containers.
      # See the TODO in the docstring.)
      logger.info("Confirming head pod resource allocation.")
      out = kubectl_exec_python_script(  # Interaction mode #1: `kubectl exec`
          script_name="check_cpu_and_memory.py",
          pod=head_pod,
          container="ray-head",
          namespace="default",
      )
  
      # Scale-up
      logger.info("Scaling up to one worker via Ray resource request.")
      # The request for 2 cpus should give us a 1-cpu head (already present) and a
      # 1-cpu worker (will await scale-up).
      kubectl_exec_python_script(  # Interaction mode #1: `kubectl exec`
          script_name="scale_up.py",
          pod=head_pod,
          container="ray-head",
          namespace="default",
      )
      # Check that stdout autoscaler logging is working.
      logs = kubectl_logs(
          head_pod, namespace="default", container="autoscaler"
      )
      assert "Adding 1 node(s) of type small-group." in logs
      logger.info("Confirming number of workers.")
      wait_for_pods(goal_num_pods=2, namespace=RAY_CLUSTER_NAMESPACE)
  
      # Ray CR updates.
      logger.info("Scaling up to two workers by editing minReplicas.")
      # replicas=1 reflects the current number of workers
      # (which is what we expect to be already present in the Ray CR)
      self._apply_ray_cr(
          min_replicas=2,
          cpu_replicas=1,
          gpu_replicas=0,
          # Confirm CPU, GPU replicas set on the Ray CR by the autoscaler are 1, 0:
          validate_replicas=True,
      )
      logger.info("Confirming number of workers.")
      wait_for_pods(goal_num_pods=3, namespace=RAY_CLUSTER_NAMESPACE)
  
      # GPU upscaling.
      # 1. Check we haven't spuriously already started a fake GPU node.
      assert not any(
          "gpu" in pod_name
          for pod_name in get_pod_names(namespace=RAY_CLUSTER_NAMESPACE)
      )
      # 2. Trigger GPU upscaling by requesting placement of a GPU actor.
      logger.info("Scheduling an Actor with GPU demands.")
      # Use Ray Client to validate that it works against KubeRay.
      with ray_client_port_forward(  # Interaction mode #2: Ray Client
          head_service=HEAD_SERVICE, ray_namespace="gpu-test"
      ):
          gpu_actor_placement.main()
      # 3. Confirm new pod number and presence of fake GPU worker.
      logger.info("Confirming fake GPU worker up-scaling.")
      wait_for_pods(goal_num_pods=4, namespace=RAY_CLUSTER_NAMESPACE)
  
      gpu_workers = [
          pod_name
          for pod_name in get_pod_names(namespace=RAY_CLUSTER_NAMESPACE)
          if "gpu" in pod_name
      ]
      assert len(gpu_workers) == 1
      # 4. Confirm that the GPU actor is up and that Ray believes
      # the node the actor is on has a GPU.
      logger.info("Confirming GPU actor placement.")
      with ray_client_port_forward(
          head_service=HEAD_SERVICE, ray_namespace="gpu-test"
      ):
          out = gpu_actor_validation.main()
      # Confirms the actor was placed on a GPU-annotated node.
      # (See gpu_actor_validation.py for details.)
      assert "on-a-gpu-node" in out
  
      # Scale-down
      logger.info("Reducing min workers to 0.")
      # Max workers remains 300.
      self._apply_ray_cr(
          min_replicas=0,
          cpu_replicas=2,
          gpu_replicas=1,
          # Confirm CPU, GPU replicas set on the Ray CR by the autoscaler are 2, 1:
          validate_replicas=True,
      )
      logger.info("Removing resource demands.")
      kubectl_exec_python_script(
          script_name="scale_down.py",
          pod=head_pod,
          container="ray-head",
          namespace="default",
      )
      # Autoscaler should trigger scale-down after resource demands are removed.
      logger.info("Confirming workers are gone.")
      # Check that stdout autoscaler logging is working.
      logs = kubectl_logs(
          head_pod, namespace="default", container="autoscaler"
      )
      assert "Removing 1 nodes of type fake-gpu-group (idle)." in logs
      wait_for_pods(goal_num_pods=1, namespace=RAY_CLUSTER_NAMESPACE)
  
      # Check custom resource upscaling.
  
      # Submit two {"Custom2": 3} bundles to upscale two workers with 5
      # Custom2 capacity each.
      logger.info(
          "Scaling up workers with request for custom resources."
      )
      job_logs = (
          ray_job_submit(  # Interaction mode #3: Ray Job Submission
              script_name="scale_up_custom.py",
              head_service=HEAD_SERVICE,
          )
      )
      assert "Submitted custom scale request!" in job_logs, job_logs
  
      logger.info("Confirming two workers have scaled up.")
      wait_for_pods(goal_num_pods=3, namespace=RAY_CLUSTER_NAMESPACE)
  
      # Cluster deletion
      logger.info("Deleting Ray cluster.")
      kubectl_delete(
          kind="raycluster",
          name=RAY_CLUSTER_NAME,
          namespace=RAY_CLUSTER_NAMESPACE,
      )
      logger.info("Confirming Ray pods are gone.")
      wait_for_pods(goal_num_pods=0, namespace=RAY_CLUSTER_NAMESPACE)
  
file_path: ray/python/ray/tests/kuberay/test_autoscaling_e2e.py
imports: import base64
  import copy
  import logging
  import os
  import pytest
  import subprocess
  import sys
  import tempfile
  import unittest
  from typing import Any, Dict
  import yaml
  from ray.tests.kuberay.utils import get_pod, get_pod_names, get_raycluster, ray_client_port_forward, ray_job_submit, switch_to_ray_parent_dir, kubectl_exec_python_script, kubectl_logs, kubectl_delete, wait_for_pods, wait_for_pod_to_start, wait_for_ray_health
  from ray.tests.kuberay.scripts import gpu_actor_placement, gpu_actor_validation, non_terminated_nodes_count
label_desc: Use image node_id 4 for calling the KubeRayAutoscalingTest obj's underlying member method code with example usage: obj.testAutoscaling() without return types