wehub-resource-sync
1776 lines
72 KiB
Python
1776 lines
72 KiB
Python
import logging
|
|
import math
|
|
import time
|
|
import uuid
|
|
from collections import defaultdict
|
|
from typing import Dict, List, Optional, Set, Tuple
|
|
|
|
from ray._common.utils import binary_to_hex
|
|
from ray.autoscaler.v2.instance_manager.cloud_providers.kuberay.cloud_provider import (
|
|
KubeRayProvider,
|
|
)
|
|
from ray.autoscaler.v2.instance_manager.common import InstanceUtil
|
|
from ray.autoscaler.v2.instance_manager.config import (
|
|
AutoscalingConfig,
|
|
InstanceReconcileConfig,
|
|
Provider,
|
|
)
|
|
from ray.autoscaler.v2.instance_manager.instance_manager import InstanceManager
|
|
from ray.autoscaler.v2.instance_manager.node_provider import (
|
|
CloudInstance,
|
|
CloudInstanceId,
|
|
CloudInstanceProviderError,
|
|
ICloudInstanceProvider,
|
|
LaunchNodeError,
|
|
TerminateNodeError,
|
|
)
|
|
from ray.autoscaler.v2.instance_manager.subscribers.cloud_resource_monitor import (
|
|
CloudResourceMonitor,
|
|
)
|
|
from ray.autoscaler.v2.instance_manager.subscribers.ray_stopper import RayStopError
|
|
from ray.autoscaler.v2.instance_manager.subscribers.threaded_ray_installer import (
|
|
RayInstallError,
|
|
)
|
|
from ray.autoscaler.v2.metrics_reporter import AutoscalerMetricsReporter
|
|
from ray.autoscaler.v2.scheduler import IResourceScheduler, SchedulingRequest
|
|
from ray.autoscaler.v2.schema import AutoscalerInstance, NodeType
|
|
from ray.autoscaler.v2.utils import is_head_node
|
|
from ray.core.generated.autoscaler_pb2 import (
|
|
AutoscalingState,
|
|
ClusterResourceState,
|
|
FailedInstanceRequest,
|
|
NodeState,
|
|
NodeStatus,
|
|
PendingInstance,
|
|
PendingInstanceRequest,
|
|
)
|
|
from ray.core.generated.instance_manager_pb2 import (
|
|
GetInstanceManagerStateRequest,
|
|
Instance as IMInstance,
|
|
InstanceUpdateEvent as IMInstanceUpdateEvent,
|
|
NodeKind,
|
|
StatusCode,
|
|
UpdateInstanceManagerStateRequest,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Reconciler:
|
|
"""
|
|
A singleton class that reconciles the instance states of the instance manager
|
|
for autoscaler.
|
|
|
|
"""
|
|
|
|
@staticmethod
|
|
def reconcile(
|
|
instance_manager: InstanceManager,
|
|
scheduler: IResourceScheduler,
|
|
cloud_provider: ICloudInstanceProvider,
|
|
cloud_resource_monitor: CloudResourceMonitor,
|
|
ray_cluster_resource_state: ClusterResourceState,
|
|
non_terminated_cloud_instances: Dict[CloudInstanceId, CloudInstance],
|
|
autoscaling_config: AutoscalingConfig,
|
|
cloud_provider_errors: Optional[List[CloudInstanceProviderError]] = None,
|
|
ray_install_errors: Optional[List[RayInstallError]] = None,
|
|
ray_stop_errors: Optional[List[RayStopError]] = None,
|
|
metrics_reporter: Optional[AutoscalerMetricsReporter] = None,
|
|
_logger: Optional[logging.Logger] = None,
|
|
) -> AutoscalingState:
|
|
"""
|
|
The reconcile method computes InstanceUpdateEvents for the instance manager
|
|
by:
|
|
|
|
1. Reconciling the instance manager's instances with external states like
|
|
the cloud provider's, the ray cluster's states, the ray installer's results.
|
|
It performs "passive" status transitions for the instances (where the status
|
|
transition should only be reflecting the external states of the cloud provider
|
|
and the ray cluster, and should not be actively changing them)
|
|
|
|
2. Stepping the instances to the active states by computing instance status
|
|
transitions that are needed and updating the instance manager's state.
|
|
These transitions should be "active" where the transitions have side effects
|
|
(through InstanceStatusSubscriber) to the cloud provider and the ray cluster.
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
scheduler: The resource scheduler to make scaling decisions.
|
|
cloud_provider: The cloud instance provider used to launch and
|
|
terminate nodes.
|
|
cloud_resource_monitor: The cloud resource monitor for monitoring
|
|
resource availability of all node types.
|
|
ray_cluster_resource_state: The ray cluster's resource state.
|
|
non_terminated_cloud_instances: The non-terminated cloud instances
|
|
from the cloud provider.
|
|
autoscaling_config: The autoscaling config.
|
|
cloud_provider_errors: The errors from the cloud provider.
|
|
ray_install_errors: The errors from RayInstaller.
|
|
ray_stop_errors: The errors from RayStopper.
|
|
metrics_reporter: The metric reporter to report the autoscaler
|
|
metrics.
|
|
_logger: The logger (for testing).
|
|
|
|
Returns:
|
|
The updated autoscaling state after reconciling.
|
|
"""
|
|
cloud_provider_errors = cloud_provider_errors or []
|
|
ray_install_errors = ray_install_errors or []
|
|
ray_stop_errors = ray_stop_errors or []
|
|
|
|
autoscaling_state = AutoscalingState()
|
|
autoscaling_state.last_seen_cluster_resource_state_version = (
|
|
ray_cluster_resource_state.cluster_resource_state_version
|
|
)
|
|
|
|
Reconciler._sync_from(
|
|
instance_manager=instance_manager,
|
|
ray_nodes=ray_cluster_resource_state.node_states,
|
|
non_terminated_cloud_instances=non_terminated_cloud_instances,
|
|
cloud_provider_errors=cloud_provider_errors,
|
|
ray_install_errors=ray_install_errors,
|
|
ray_stop_errors=ray_stop_errors,
|
|
autoscaling_config=autoscaling_config,
|
|
)
|
|
|
|
Reconciler._step_next(
|
|
autoscaling_state=autoscaling_state,
|
|
instance_manager=instance_manager,
|
|
scheduler=scheduler,
|
|
cloud_provider=cloud_provider,
|
|
cloud_resource_monitor=cloud_resource_monitor,
|
|
ray_cluster_resource_state=ray_cluster_resource_state,
|
|
non_terminated_cloud_instances=non_terminated_cloud_instances,
|
|
autoscaling_config=autoscaling_config,
|
|
_logger=_logger,
|
|
)
|
|
|
|
Reconciler._report_metrics(
|
|
instance_manager=instance_manager,
|
|
autoscaling_config=autoscaling_config,
|
|
metrics_reporter=metrics_reporter,
|
|
)
|
|
|
|
return autoscaling_state
|
|
|
|
@staticmethod
|
|
def _sync_from(
|
|
instance_manager: InstanceManager,
|
|
ray_nodes: List[NodeState],
|
|
non_terminated_cloud_instances: Dict[CloudInstanceId, CloudInstance],
|
|
cloud_provider_errors: List[CloudInstanceProviderError],
|
|
ray_install_errors: List[RayInstallError],
|
|
ray_stop_errors: List[RayStopError],
|
|
autoscaling_config: AutoscalingConfig,
|
|
):
|
|
"""
|
|
Reconcile the instance states of the instance manager from external states like
|
|
the cloud provider's, the ray cluster's states, the ray installer's results,
|
|
etc.
|
|
|
|
For each instance, we try to figure out if we need to transition the instance
|
|
status to a new status, and if so, what the new status should be.
|
|
|
|
These transitions should be purely "passive", meaning they should only be
|
|
reflecting the external states of the cloud provider and the ray cluster,
|
|
and should not be actively changing the states of the cloud provider or the ray
|
|
cluster.
|
|
|
|
More specifically, we will reconcile status transitions for:
|
|
1. QUEUED/REQUESTED -> ALLOCATED:
|
|
When an instance with launch request id (indicating a previous launch
|
|
request was made) could be assigned to an unassigned cloud instance
|
|
of the same instance type.
|
|
2. REQUESTED -> ALLOCATION_FAILED:
|
|
When there's an error from the cloud provider for launch failure so
|
|
that the instance becomes ALLOCATION_FAILED.
|
|
3. ALLOCATED -> ALLOCATION_TIMEOUT:
|
|
When an instance has been allocated to a cloud instance, but is stuck in
|
|
this state. For example, a kubernetes pod remains pending due to
|
|
insufficient resources.
|
|
4. * -> RAY_RUNNING:
|
|
When a ray node on a cloud instance joins the ray cluster, we will
|
|
transition the instance to RAY_RUNNING.
|
|
5. * -> TERMINATED:
|
|
When the cloud instance is already terminated, we will transition the
|
|
instance to TERMINATED.
|
|
6. TERMINATING -> TERMINATION_FAILED:
|
|
When there's an error from the cloud provider for termination failure.
|
|
7. * -> RAY_STOPPED:
|
|
When ray was stopped on the cloud instance, we will transition the
|
|
instance to RAY_STOPPED.
|
|
8. * -> RAY_INSTALL_FAILED:
|
|
When there's an error from RayInstaller.
|
|
9. RAY_STOP_REQUESTED -> RAY_RUNNING:
|
|
When requested to stop ray, but failed to stop/drain the ray node
|
|
(e.g. idle termination drain rejected by the node).
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
ray_nodes: The ray cluster's states of ray nodes.
|
|
non_terminated_cloud_instances: The non-terminated cloud instances from
|
|
the cloud provider.
|
|
cloud_provider_errors: The errors from the cloud provider.
|
|
ray_install_errors: The errors from RayInstaller.
|
|
ray_stop_errors: The errors from RayStopper.
|
|
autoscaling_config: The autoscaling config.
|
|
"""
|
|
|
|
# Handle 1 & 2 for cloud instance allocation.
|
|
Reconciler._handle_cloud_instance_allocation(
|
|
instance_manager,
|
|
non_terminated_cloud_instances,
|
|
cloud_provider_errors,
|
|
)
|
|
Reconciler._handle_cloud_instance_terminated(
|
|
instance_manager, non_terminated_cloud_instances
|
|
)
|
|
|
|
Reconciler._handle_cloud_instance_termination_errors(
|
|
instance_manager, cloud_provider_errors
|
|
)
|
|
|
|
Reconciler._handle_extra_cloud_instances(
|
|
instance_manager, non_terminated_cloud_instances, ray_nodes
|
|
)
|
|
|
|
Reconciler._handle_ray_status_transition(
|
|
instance_manager, ray_nodes, autoscaling_config
|
|
)
|
|
|
|
Reconciler._handle_ray_install_failed(instance_manager, ray_install_errors)
|
|
|
|
Reconciler._handle_ray_stop_failed(instance_manager, ray_stop_errors, ray_nodes)
|
|
|
|
@staticmethod
|
|
def _step_next(
|
|
autoscaling_state: AutoscalingState,
|
|
instance_manager: InstanceManager,
|
|
scheduler: IResourceScheduler,
|
|
cloud_provider: ICloudInstanceProvider,
|
|
cloud_resource_monitor: CloudResourceMonitor,
|
|
ray_cluster_resource_state: ClusterResourceState,
|
|
non_terminated_cloud_instances: Dict[CloudInstanceId, CloudInstance],
|
|
autoscaling_config: AutoscalingConfig,
|
|
_logger: Optional[logging.Logger] = None,
|
|
):
|
|
"""
|
|
Step the reconciler to the next state by computing instance status transitions
|
|
that are needed and updating the instance manager's state.
|
|
|
|
Specifically, we will:
|
|
1. Shut down leak cloud instances
|
|
Leaked cloud instances that are not managed by the instance manager.
|
|
2. Terminating instances with ray stopped or ray install failure.
|
|
3. Scale down the cluster:
|
|
(* -> RAY_STOP_REQUESTED/TERMINATING)
|
|
b. Extra cloud due to max nodes config.
|
|
c. Cloud instances with outdated configs.
|
|
4. Scale up the cluster:
|
|
(new QUEUED)
|
|
Create new instances based on the IResourceScheduler's decision for
|
|
scaling up.
|
|
5. Request cloud provider to launch new instances.
|
|
(QUEUED -> REQUESTED)
|
|
6. Install ray
|
|
(ALLOCATED -> RAY_INSTALLING)
|
|
When ray could be installed and launched.
|
|
7. Reconcile IM instances whose ray nodes are missing from GCS.
|
|
8. Handle any stuck instances with timeouts.
|
|
|
|
Args:
|
|
autoscaling_state: The autoscaling state populated by this reconcile loop.
|
|
instance_manager: The instance manager to reconcile.
|
|
scheduler: The resource scheduler to make scaling decisions.
|
|
cloud_provider: The cloud instance provider used to launch and
|
|
terminate nodes.
|
|
cloud_resource_monitor: The cloud resource monitor for monitoring
|
|
resource availability of all node types.
|
|
ray_cluster_resource_state: The ray cluster's resource state.
|
|
non_terminated_cloud_instances: The non-terminated cloud instances
|
|
from the cloud provider.
|
|
autoscaling_config: The autoscaling config.
|
|
_logger: The logger (for testing).
|
|
"""
|
|
|
|
Reconciler._handle_missing_ray_nodes(
|
|
instance_manager=instance_manager,
|
|
ray_nodes=ray_cluster_resource_state.node_states,
|
|
)
|
|
|
|
Reconciler._handle_stuck_instances(
|
|
instance_manager=instance_manager,
|
|
reconcile_config=autoscaling_config.get_instance_reconcile_config(),
|
|
_logger=_logger or logger,
|
|
)
|
|
|
|
Reconciler._scale_cluster(
|
|
autoscaling_state=autoscaling_state,
|
|
instance_manager=instance_manager,
|
|
cloud_resource_monitor=cloud_resource_monitor,
|
|
ray_state=ray_cluster_resource_state,
|
|
scheduler=scheduler,
|
|
autoscaling_config=autoscaling_config,
|
|
cloud_provider=cloud_provider,
|
|
)
|
|
|
|
# Fix: Terminate instances before launching new ones
|
|
# This ensures that when launching new instances, the replica count
|
|
# does not include instances that are about to be terminated,
|
|
# preventing the maxReplicas limit from being incorrectly triggered.
|
|
Reconciler._terminate_instances(instance_manager=instance_manager)
|
|
|
|
Reconciler._handle_instances_launch(
|
|
instance_manager=instance_manager, autoscaling_config=autoscaling_config
|
|
)
|
|
|
|
if not autoscaling_config.disable_node_updaters():
|
|
Reconciler._install_ray(
|
|
instance_manager=instance_manager,
|
|
non_terminated_cloud_instances=non_terminated_cloud_instances,
|
|
)
|
|
|
|
Reconciler._fill_autoscaling_state(
|
|
instance_manager=instance_manager,
|
|
autoscaling_state=autoscaling_state,
|
|
autoscaling_config=autoscaling_config,
|
|
)
|
|
|
|
#######################################################
|
|
# Utility methods for reconciling instance states.
|
|
#######################################################
|
|
|
|
@staticmethod
|
|
def _handle_cloud_instance_allocation(
|
|
instance_manager: InstanceManager,
|
|
non_terminated_cloud_instances: Dict[CloudInstanceId, CloudInstance],
|
|
cloud_provider_errors: List[CloudInstanceProviderError],
|
|
):
|
|
im_instances, version = Reconciler._get_im_instances(instance_manager)
|
|
updates = {}
|
|
|
|
# Compute intermediate states.
|
|
|
|
instances_with_launch_requests: List[IMInstance] = []
|
|
for instance in im_instances:
|
|
if instance.status != IMInstance.REQUESTED:
|
|
continue
|
|
|
|
assert (
|
|
instance.launch_request_id
|
|
), "Instance in REQUESTED status should have launch_request_id set."
|
|
instances_with_launch_requests.append(instance)
|
|
|
|
assigned_cloud_instance_ids: Set[CloudInstanceId] = {
|
|
instance.cloud_instance_id
|
|
for instance in im_instances
|
|
if instance.cloud_instance_id
|
|
and instance.status
|
|
not in [IMInstance.TERMINATED, IMInstance.ALLOCATION_FAILED]
|
|
}
|
|
launch_errors: Dict[str, LaunchNodeError] = {
|
|
error.request_id: error
|
|
for error in cloud_provider_errors
|
|
if isinstance(error, LaunchNodeError)
|
|
}
|
|
unassigned_cloud_instances_by_type: Dict[
|
|
str, List[CloudInstance]
|
|
] = defaultdict(list)
|
|
|
|
for cloud_instance_id, cloud_instance in non_terminated_cloud_instances.items():
|
|
if cloud_instance_id not in assigned_cloud_instance_ids:
|
|
unassigned_cloud_instances_by_type[cloud_instance.node_type].append(
|
|
cloud_instance
|
|
)
|
|
|
|
# Sort the request instance by the increasing request time.
|
|
instances_with_launch_requests.sort(
|
|
key=lambda instance: InstanceUtil.get_status_transition_times_ns(
|
|
instance, IMInstance.REQUESTED
|
|
)
|
|
)
|
|
|
|
# For each instance, try to allocate or fail the allocation.
|
|
for instance in instances_with_launch_requests:
|
|
# Try allocate or fail with errors.
|
|
update_event = Reconciler._try_resolve_pending_allocation(
|
|
instance, unassigned_cloud_instances_by_type, launch_errors
|
|
)
|
|
if not update_event:
|
|
continue
|
|
|
|
updates[instance.instance_id] = update_event
|
|
|
|
# Update the instance manager for the events.
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _try_resolve_pending_allocation(
|
|
im_instance: IMInstance,
|
|
unassigned_cloud_instances_by_type: Dict[str, List[CloudInstance]],
|
|
launch_errors: Dict[str, LaunchNodeError],
|
|
) -> Optional[IMInstanceUpdateEvent]:
|
|
"""
|
|
Allocate, or fail the cloud instance allocation for the instance.
|
|
|
|
Args:
|
|
im_instance: The instance to allocate or fail.
|
|
unassigned_cloud_instances_by_type: The unassigned cloud instances by type.
|
|
launch_errors: The launch errors from the cloud provider.
|
|
|
|
Returns:
|
|
Instance update to ALLOCATED: if there's a matching unassigned cloud
|
|
instance with the same type.
|
|
Instance update to ALLOCATION_FAILED: if the instance allocation failed
|
|
with errors.
|
|
None: if there's no update.
|
|
|
|
"""
|
|
unassigned_cloud_instance = None
|
|
|
|
# Try to allocate an unassigned cloud instance.
|
|
# TODO(rickyx): We could also look at the launch request id
|
|
# on the cloud node and the im instance later once all node providers
|
|
# support request id. For now, we only look at the instance type.
|
|
if len(unassigned_cloud_instances_by_type.get(im_instance.instance_type, [])):
|
|
unassigned_cloud_instance = unassigned_cloud_instances_by_type[
|
|
im_instance.instance_type
|
|
].pop()
|
|
|
|
if unassigned_cloud_instance:
|
|
return IMInstanceUpdateEvent(
|
|
instance_id=im_instance.instance_id,
|
|
new_instance_status=IMInstance.ALLOCATED,
|
|
cloud_instance_id=unassigned_cloud_instance.cloud_instance_id,
|
|
node_kind=unassigned_cloud_instance.node_kind,
|
|
instance_type=unassigned_cloud_instance.node_type,
|
|
details=(
|
|
"allocated unassigned cloud instance "
|
|
f"{unassigned_cloud_instance.cloud_instance_id}"
|
|
),
|
|
)
|
|
|
|
# If there's a launch error, transition to ALLOCATION_FAILED.
|
|
launch_error = launch_errors.get(im_instance.launch_request_id)
|
|
if launch_error and launch_error.node_type == im_instance.instance_type:
|
|
return IMInstanceUpdateEvent(
|
|
instance_id=im_instance.instance_id,
|
|
new_instance_status=IMInstance.ALLOCATION_FAILED,
|
|
details=f"launch failed with {str(launch_error)}",
|
|
)
|
|
# No update.
|
|
return None
|
|
|
|
@staticmethod
|
|
def _handle_ray_stop_failed(
|
|
instance_manager: InstanceManager,
|
|
ray_stop_errors: List[RayStopError],
|
|
ray_nodes: List[NodeState],
|
|
):
|
|
"""
|
|
The instance requested to stop ray, but failed to stop/drain the ray node.
|
|
E.g. connection errors, idle termination drain rejected by the node.
|
|
|
|
We will transition the instance back to RAY_RUNNING.
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
ray_stop_errors: The errors from RayStopper.
|
|
ray_nodes: The ray cluster's states of ray nodes.
|
|
"""
|
|
instances, version = Reconciler._get_im_instances(instance_manager)
|
|
updates = {}
|
|
|
|
ray_stop_errors_by_instance_id = {
|
|
error.im_instance_id: error for error in ray_stop_errors
|
|
}
|
|
|
|
ray_nodes_by_ray_node_id = {binary_to_hex(n.node_id): n for n in ray_nodes}
|
|
|
|
ray_stop_requested_instances = {
|
|
instance.instance_id: instance
|
|
for instance in instances
|
|
if instance.status == IMInstance.RAY_STOP_REQUESTED
|
|
}
|
|
|
|
for instance_id, instance in ray_stop_requested_instances.items():
|
|
stop_error = ray_stop_errors_by_instance_id.get(instance_id)
|
|
if not stop_error:
|
|
continue
|
|
|
|
assert instance.node_id
|
|
ray_node = ray_nodes_by_ray_node_id.get(instance.node_id)
|
|
assert ray_node is not None and ray_node.status in [
|
|
NodeStatus.RUNNING,
|
|
NodeStatus.IDLE,
|
|
], (
|
|
"There should be a running ray node for instance with ray stop "
|
|
"requested failed."
|
|
)
|
|
|
|
updates[instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance_id,
|
|
new_instance_status=IMInstance.RAY_RUNNING,
|
|
details="failed to stop/drain ray",
|
|
ray_node_id=instance.node_id,
|
|
)
|
|
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _handle_ray_install_failed(
|
|
instance_manager: InstanceManager, ray_install_errors: List[RayInstallError]
|
|
):
|
|
|
|
instances, version = Reconciler._get_im_instances(instance_manager)
|
|
updates = {}
|
|
|
|
# Get all instances with RAY_INSTALLING status.
|
|
instances_with_ray_installing = {
|
|
instance.instance_id: instance
|
|
for instance in instances
|
|
if instance.status == IMInstance.RAY_INSTALLING
|
|
}
|
|
|
|
install_errors = {error.im_instance_id: error for error in ray_install_errors}
|
|
|
|
# For each instance with RAY_INSTALLING status, check if there's any
|
|
# install error.
|
|
for instance_id, instance in instances_with_ray_installing.items():
|
|
install_error = install_errors.get(instance_id)
|
|
if install_error:
|
|
updates[instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance_id,
|
|
new_instance_status=IMInstance.RAY_INSTALL_FAILED,
|
|
details=(
|
|
f"failed to install ray with errors: {install_error.details}"
|
|
),
|
|
)
|
|
|
|
# Update the instance manager for the events.
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _handle_cloud_instance_terminated(
|
|
instance_manager: InstanceManager,
|
|
non_terminated_cloud_instances: Dict[CloudInstanceId, CloudInstance],
|
|
):
|
|
"""
|
|
For any IM (instance manager) instance with a cloud node id, if the mapped
|
|
cloud instance is no longer running, transition the instance to TERMINATED.
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
non_terminated_cloud_instances: The non-terminated cloud instances from
|
|
the cloud provider.
|
|
"""
|
|
updates = {}
|
|
instances, version = Reconciler._get_im_instances(instance_manager)
|
|
|
|
non_terminated_instances_with_cloud_instance_assigned = {
|
|
instance.cloud_instance_id: instance
|
|
for instance in instances
|
|
if instance.cloud_instance_id and instance.status != IMInstance.TERMINATED
|
|
}
|
|
|
|
for (
|
|
cloud_instance_id,
|
|
instance,
|
|
) in non_terminated_instances_with_cloud_instance_assigned.items():
|
|
if cloud_instance_id in non_terminated_cloud_instances.keys():
|
|
# The cloud instance is still running.
|
|
continue
|
|
|
|
# The cloud instance is terminated.
|
|
updates[instance.instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance.instance_id,
|
|
new_instance_status=IMInstance.TERMINATED,
|
|
cloud_instance_id=cloud_instance_id,
|
|
details=f"cloud instance {cloud_instance_id} no longer found",
|
|
)
|
|
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _handle_cloud_instance_termination_errors(
|
|
instance_manager: InstanceManager,
|
|
cloud_provider_errors: List[CloudInstanceProviderError],
|
|
):
|
|
"""
|
|
If any TERMINATING instances have termination errors, transition the instance to
|
|
TERMINATION_FAILED.
|
|
|
|
We will retry the termination for the TERMINATION_FAILED instances in the next
|
|
reconciler step.
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
cloud_provider_errors: The errors from the cloud provider.
|
|
|
|
"""
|
|
instances, version = Reconciler._get_im_instances(instance_manager)
|
|
updates = {}
|
|
|
|
termination_errors = {
|
|
error.cloud_instance_id: error
|
|
for error in cloud_provider_errors
|
|
if isinstance(error, TerminateNodeError)
|
|
}
|
|
|
|
terminating_instances_by_cloud_instance_id = {
|
|
instance.cloud_instance_id: instance
|
|
for instance in instances
|
|
if instance.status == IMInstance.TERMINATING
|
|
}
|
|
|
|
for cloud_instance_id, failure in termination_errors.items():
|
|
instance = terminating_instances_by_cloud_instance_id.get(cloud_instance_id)
|
|
if not instance:
|
|
# The instance is no longer in TERMINATING status.
|
|
continue
|
|
|
|
updates[instance.instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance.instance_id,
|
|
new_instance_status=IMInstance.TERMINATION_FAILED,
|
|
details=f"termination failed: {str(failure)}",
|
|
)
|
|
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _get_im_instances(
|
|
instance_manager: InstanceManager,
|
|
) -> Tuple[List[IMInstance], int]:
|
|
reply = instance_manager.get_instance_manager_state(
|
|
request=GetInstanceManagerStateRequest()
|
|
)
|
|
assert reply.status.code == StatusCode.OK
|
|
im_state = reply.state
|
|
return im_state.instances, im_state.version
|
|
|
|
@staticmethod
|
|
def _update_instance_manager(
|
|
instance_manager: InstanceManager,
|
|
version: int,
|
|
updates: Dict[str, IMInstanceUpdateEvent],
|
|
) -> None:
|
|
if not updates:
|
|
return
|
|
|
|
updates = list(updates.values()) or []
|
|
|
|
reply = instance_manager.update_instance_manager_state(
|
|
request=UpdateInstanceManagerStateRequest(
|
|
expected_version=version,
|
|
updates=updates,
|
|
)
|
|
)
|
|
# TODO: While it's possible that a version mismatch
|
|
# happens, or some other failures could happen. But given
|
|
# the current implementation:
|
|
# 1. There's only 1 writer (the reconciler) for updating the instance
|
|
# manager states, so there shouldn't be version mismatch.
|
|
# 2. Any failures in one reconciler step should be caught at a higher
|
|
# level and be retried in the next reconciler step. If the IM
|
|
# fails to be updated, we don't have sufficient info to handle it
|
|
# here.
|
|
assert (
|
|
reply.status.code == StatusCode.OK
|
|
), f"Failed to update instance manager: {reply}"
|
|
|
|
@staticmethod
|
|
def _handle_ray_status_transition(
|
|
instance_manager: InstanceManager,
|
|
ray_nodes: List[NodeState],
|
|
autoscaling_config: AutoscalingConfig,
|
|
):
|
|
"""
|
|
Handle the ray status transition for the instance manager.
|
|
|
|
If a new ray node running on the instance, transition it to RAY_RUNNING.
|
|
If a ray node stopped, transition it to RAY_STOPPED.
|
|
If a ray node is draining, transition it to RAY_STOPPING.
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
ray_nodes: The ray cluster's states of ray nodes.
|
|
autoscaling_config: The autoscaling config.
|
|
"""
|
|
instances, version = Reconciler._get_im_instances(instance_manager)
|
|
updates = {}
|
|
|
|
im_instances_by_cloud_instance_id = {
|
|
instance.cloud_instance_id: instance
|
|
for instance in instances
|
|
if instance.cloud_instance_id
|
|
and instance.status
|
|
not in [IMInstance.TERMINATED, IMInstance.ALLOCATION_FAILED]
|
|
}
|
|
im_instances_by_ray_node_id = {
|
|
instance.node_id: instance for instance in instances if instance.node_id
|
|
}
|
|
|
|
for ray_node in ray_nodes:
|
|
im_instance = None
|
|
ray_node_id = binary_to_hex(ray_node.node_id)
|
|
if ray_node_id in im_instances_by_ray_node_id:
|
|
im_instance = im_instances_by_ray_node_id[ray_node_id]
|
|
else:
|
|
if autoscaling_config.provider == Provider.READ_ONLY:
|
|
# We will use the node id as the cloud instance id for read-only
|
|
# provider.
|
|
im_instance = im_instances_by_cloud_instance_id[ray_node_id]
|
|
elif ray_node.instance_id:
|
|
im_instance = im_instances_by_cloud_instance_id[
|
|
ray_node.instance_id
|
|
]
|
|
else:
|
|
# This should only happen to a ray node that's not managed by us.
|
|
logger.warning(
|
|
f"Ray node {ray_node_id} has no instance id. "
|
|
"This only happens to a ray node not managed by autoscaler. "
|
|
"If not, please file a bug at "
|
|
"https://github.com/ray-project/ray"
|
|
)
|
|
continue
|
|
|
|
assert im_instance is not None, (
|
|
f"Ray node {ray_node_id} has no matching "
|
|
f"instance with cloud instance id={ray_node.instance_id}. We should "
|
|
"not see a ray node with cloud instance id not found in IM since "
|
|
"we have reconciled all cloud instances, and ray nodes by now."
|
|
)
|
|
|
|
reconciled_im_status = Reconciler._reconciled_im_status_from_ray_status(
|
|
ray_node.status, im_instance.status
|
|
)
|
|
|
|
if reconciled_im_status != im_instance.status:
|
|
updates[ray_node_id] = IMInstanceUpdateEvent(
|
|
instance_id=im_instance.instance_id,
|
|
new_instance_status=reconciled_im_status,
|
|
details=(
|
|
f"ray node {ray_node_id} is "
|
|
f"{NodeStatus.Name(ray_node.status)}"
|
|
),
|
|
ray_node_id=ray_node_id,
|
|
instance_type=im_instance.instance_type,
|
|
)
|
|
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _reconciled_im_status_from_ray_status(
|
|
ray_status: NodeStatus, cur_im_status: IMInstance.InstanceStatus
|
|
) -> "IMInstance.InstanceStatus":
|
|
"""
|
|
Reconcile the instance status from the ray node status.
|
|
Args:
|
|
ray_status: the current ray node status.
|
|
cur_im_status: the current IM instance status.
|
|
Returns:
|
|
The reconciled IM instance status
|
|
|
|
Raises:
|
|
ValueError: If the ray status is unknown.
|
|
"""
|
|
reconciled_im_status = None
|
|
if ray_status in [NodeStatus.RUNNING, NodeStatus.IDLE]:
|
|
reconciled_im_status = IMInstance.RAY_RUNNING
|
|
elif ray_status == NodeStatus.DEAD:
|
|
reconciled_im_status = IMInstance.RAY_STOPPED
|
|
elif ray_status == NodeStatus.DRAINING:
|
|
reconciled_im_status = IMInstance.RAY_STOPPING
|
|
else:
|
|
raise ValueError(f"Unknown ray status: {ray_status}")
|
|
|
|
if (
|
|
cur_im_status == reconciled_im_status
|
|
or cur_im_status
|
|
in InstanceUtil.get_reachable_statuses(reconciled_im_status)
|
|
):
|
|
# No need to reconcile if the instance is already in the reconciled status
|
|
# or has already transitioned beyond it.
|
|
return cur_im_status
|
|
|
|
return reconciled_im_status
|
|
|
|
@staticmethod
|
|
def _handle_instances_launch(
|
|
instance_manager: InstanceManager, autoscaling_config: AutoscalingConfig
|
|
):
|
|
|
|
instances, version = Reconciler._get_im_instances(instance_manager)
|
|
|
|
queued_instances = []
|
|
requested_instances = []
|
|
running_instances = []
|
|
|
|
for instance in instances:
|
|
if instance.status == IMInstance.QUEUED:
|
|
queued_instances.append(instance)
|
|
elif instance.status == IMInstance.REQUESTED:
|
|
requested_instances.append(instance)
|
|
elif instance.status == IMInstance.RAY_RUNNING:
|
|
running_instances.append(instance)
|
|
|
|
if not queued_instances:
|
|
# No QUEUED instances
|
|
return
|
|
|
|
to_launch = Reconciler._compute_to_launch(
|
|
queued_instances,
|
|
requested_instances,
|
|
running_instances,
|
|
autoscaling_config.get_upscaling_speed(),
|
|
autoscaling_config.get_max_concurrent_launches(),
|
|
)
|
|
|
|
# Transition the instances to REQUESTED for instance launcher to
|
|
# launch them.
|
|
updates = {}
|
|
new_launch_request_id = str(uuid.uuid4())
|
|
for instance_type, instances in to_launch.items():
|
|
for instance in instances:
|
|
# Reuse launch request id for any QUEUED instances that have been
|
|
# requested before due to retry.
|
|
launch_request_id = (
|
|
new_launch_request_id
|
|
if len(instance.launch_request_id) == 0
|
|
else instance.launch_request_id
|
|
)
|
|
updates[instance.instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance.instance_id,
|
|
new_instance_status=IMInstance.REQUESTED,
|
|
launch_request_id=launch_request_id,
|
|
instance_type=instance_type,
|
|
details=(
|
|
f"requested to launch {instance_type} with request id "
|
|
f"{launch_request_id}"
|
|
),
|
|
)
|
|
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _compute_to_launch(
|
|
queued_instances: List[IMInstance],
|
|
requested_instances: List[IMInstance],
|
|
running_instances: List[IMInstance],
|
|
upscaling_speed: float,
|
|
max_concurrent_launches: int,
|
|
) -> Dict[NodeType, List[IMInstance]]:
|
|
def _group_by_type(instances):
|
|
instances_by_type = defaultdict(list)
|
|
for instance in instances:
|
|
instances_by_type[instance.instance_type].append(instance)
|
|
return instances_by_type
|
|
|
|
# Sort the instances by the time they were queued.
|
|
def _sort_by_earliest_queued(instance: IMInstance) -> List[int]:
|
|
queue_times = InstanceUtil.get_status_transition_times_ns(
|
|
instance, IMInstance.QUEUED
|
|
)
|
|
return sorted(queue_times)
|
|
|
|
queued_instances_by_type = _group_by_type(queued_instances)
|
|
running_instances_by_type = _group_by_type(running_instances)
|
|
|
|
total_num_requested_to_launch = len(requested_instances)
|
|
all_to_launch: Dict[NodeType : List[IMInstance]] = defaultdict(list)
|
|
|
|
for (
|
|
instance_type,
|
|
queued_instances_for_type,
|
|
) in queued_instances_by_type.items():
|
|
running_instances_for_type = running_instances_by_type.get(
|
|
instance_type, []
|
|
)
|
|
|
|
# Enforce the max allowed pending nodes based on current running nodes
|
|
num_desired_to_upscale = max(
|
|
1,
|
|
math.ceil(upscaling_speed * max(len(running_instances_for_type), 1)),
|
|
)
|
|
|
|
# Enforce global limit, at most we can launch `max_concurrent_launches`
|
|
num_to_launch = min(
|
|
max_concurrent_launches - total_num_requested_to_launch,
|
|
num_desired_to_upscale,
|
|
)
|
|
|
|
# Cap both ends 0 <= num_to_launch <= num_queued
|
|
num_to_launch = max(0, num_to_launch)
|
|
num_to_launch = min(len(queued_instances_for_type), num_to_launch)
|
|
|
|
to_launch = sorted(queued_instances_for_type, key=_sort_by_earliest_queued)[
|
|
:num_to_launch
|
|
]
|
|
|
|
all_to_launch[instance_type].extend(to_launch)
|
|
total_num_requested_to_launch += num_to_launch
|
|
|
|
return all_to_launch
|
|
|
|
@staticmethod
|
|
def _handle_missing_ray_nodes(
|
|
instance_manager: InstanceManager,
|
|
ray_nodes: List[NodeState],
|
|
) -> None:
|
|
"""
|
|
Reconcile IM instances that still point at ray nodes missing from GCS.
|
|
|
|
GCS is the source of truth for Ray node liveness. A worker instance in
|
|
RAY_RUNNING / RAY_STOP_REQUESTED / RAY_STOPPING with a node_id should
|
|
have a corresponding entry in the GCS node snapshot. If the node is
|
|
absent entirely, GCS has either GC'd its DEAD entry or otherwise no
|
|
longer considers the ray node alive. Treat this the same as a stopped ray
|
|
node so the instance can continue through the existing RAY_STOPPED ->
|
|
TERMINATING path instead of lingering as RAY_RUNNING with a stale
|
|
node_id.
|
|
|
|
We intentionally skip the head node here. During head startup/restart,
|
|
GCS can be reachable while the head raylet is not present in
|
|
ClusterResourceState yet; head-node recovery is outside this worker
|
|
cleanup path.
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
ray_nodes: The ray cluster's view of ray nodes from GCS.
|
|
"""
|
|
statuses_requiring_ray_node = {
|
|
IMInstance.RAY_RUNNING,
|
|
IMInstance.RAY_STOP_REQUESTED,
|
|
IMInstance.RAY_STOPPING,
|
|
}
|
|
im_instances, version = Reconciler._get_im_instances(instance_manager)
|
|
candidates = [
|
|
instance
|
|
for instance in im_instances
|
|
if instance.status in statuses_requiring_ray_node
|
|
and instance.node_kind != NodeKind.HEAD
|
|
and instance.node_id
|
|
]
|
|
if not candidates:
|
|
return
|
|
|
|
ray_node_ids = {binary_to_hex(n.node_id) for n in ray_nodes}
|
|
|
|
updates = {}
|
|
for instance in candidates:
|
|
if instance.node_id in ray_node_ids:
|
|
continue
|
|
|
|
updates[instance.instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance.instance_id,
|
|
new_instance_status=IMInstance.RAY_STOPPED,
|
|
details=f"ray node {instance.node_id} no longer found in GCS",
|
|
ray_node_id=instance.node_id,
|
|
instance_type=instance.instance_type,
|
|
)
|
|
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _handle_stuck_instances(
|
|
instance_manager: InstanceManager,
|
|
reconcile_config: InstanceReconcileConfig,
|
|
_logger: logging.Logger,
|
|
):
|
|
"""
|
|
Handle stuck instances with timeouts.
|
|
|
|
Instances could be stuck in the following status and needs to be updated:
|
|
- REQUESTED: cloud provider is slow/fails to launch instances.
|
|
- ALLOCATED: ray fails to be started on the instance.
|
|
- RAY_INSTALLING: ray fails to be installed on the instance.
|
|
- TERMINATING: cloud provider is slow/fails to terminate instances.
|
|
|
|
Instances could be in the following status which could be unbounded or
|
|
transient, and we don't have a timeout mechanism to handle them. We would
|
|
warn if they are stuck for too long:
|
|
- RAY_STOPPING: ray taking time to drain.
|
|
- QUEUED: cloud provider is slow to launch instances, resulting in long
|
|
queue.
|
|
|
|
Reconciler should handle below statuses, if not, could be slow
|
|
reconcilation loop or a bug:
|
|
- RAY_INSTALL_FAILED
|
|
- RAY_STOPPED
|
|
- TERMINATION_FAILED
|
|
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
reconcile_config: The instance reconcile config.
|
|
_logger: The logger to log the warning messages. It's used for testing.
|
|
|
|
"""
|
|
instances, version = Reconciler._get_im_instances(instance_manager)
|
|
|
|
instances_by_status = defaultdict(list)
|
|
for instance in instances:
|
|
instances_by_status[instance.status].append(instance)
|
|
|
|
im_updates = {}
|
|
|
|
# Fail or retry the cloud instance allocation if it's stuck
|
|
# in the REQUESTED state.
|
|
for instance in instances_by_status[IMInstance.REQUESTED]:
|
|
update = Reconciler._handle_stuck_requested_instance(
|
|
instance,
|
|
reconcile_config.request_status_timeout_s,
|
|
reconcile_config.max_num_retry_request_to_allocate,
|
|
)
|
|
if update:
|
|
im_updates[instance.instance_id] = update
|
|
|
|
# Leaked ALLOCATED instances should be terminated.
|
|
# This usually happens when ray fails to be started on the instance, so
|
|
# it's unable to be RAY_RUNNING after a long time.
|
|
for instance in instances_by_status[IMInstance.ALLOCATED]:
|
|
assert (
|
|
instance.cloud_instance_id
|
|
), "cloud instance id should be set on ALLOCATED instance"
|
|
update = Reconciler._handle_stuck_instance(
|
|
instance,
|
|
reconcile_config.allocate_status_timeout_s,
|
|
new_status=IMInstance.ALLOCATION_TIMEOUT,
|
|
cloud_instance_id=instance.cloud_instance_id,
|
|
instance_type=instance.instance_type,
|
|
)
|
|
if update:
|
|
im_updates[instance.instance_id] = update
|
|
|
|
# Fail the installation if it's stuck in RAY_INSTALLING for too long.
|
|
# If RAY_INSTALLING is stuck for too long, it's likely that the instance
|
|
# is not able to install ray, so we should also fail the installation.
|
|
for instance in instances_by_status[IMInstance.RAY_INSTALLING]:
|
|
update = Reconciler._handle_stuck_instance(
|
|
instance,
|
|
reconcile_config.ray_install_status_timeout_s,
|
|
new_status=IMInstance.RAY_INSTALL_FAILED,
|
|
)
|
|
if update:
|
|
im_updates[instance.instance_id] = update
|
|
|
|
# If we tried to terminate the instance, but it doesn't terminate (disappear
|
|
# from the cloud provider) after a long time, we fail the termination.
|
|
# This will trigger another attempt to terminate the instance.
|
|
for instance in instances_by_status[IMInstance.TERMINATING]:
|
|
update = Reconciler._handle_stuck_instance(
|
|
instance,
|
|
reconcile_config.terminating_status_timeout_s,
|
|
new_status=IMInstance.TERMINATION_FAILED,
|
|
)
|
|
if update:
|
|
im_updates[instance.instance_id] = update
|
|
|
|
# If we tried to stop ray on the instance, but it doesn't stop after a long
|
|
# time, we will transition it back to RAY_RUNNING as the stop/drain somehow
|
|
# failed. Instances whose ray node has already disappeared from GCS are moved
|
|
# to RAY_STOPPED earlier by _handle_missing_ray_nodes, so anything still in
|
|
# RAY_STOP_REQUESTED here genuinely failed to drain.
|
|
for instance in instances_by_status[IMInstance.RAY_STOP_REQUESTED]:
|
|
update = Reconciler._handle_stuck_instance(
|
|
instance,
|
|
reconcile_config.ray_stop_requested_status_timeout_s,
|
|
new_status=IMInstance.RAY_RUNNING,
|
|
ray_node_id=instance.node_id,
|
|
)
|
|
if update:
|
|
im_updates[instance.instance_id] = update
|
|
|
|
# These statues could be unbounded or transient, and we don't have a timeout
|
|
# mechanism to handle them. We only warn if they are stuck for too long.
|
|
for status in [
|
|
# Ray taking time to drain. We could also have a timeout when Drain protocol
|
|
# supports timeout.
|
|
IMInstance.RAY_STOPPING,
|
|
# These should just be transient, we will terminate instances with this
|
|
# status in the next reconciler step.
|
|
IMInstance.RAY_INSTALL_FAILED,
|
|
IMInstance.RAY_STOPPED,
|
|
IMInstance.TERMINATION_FAILED,
|
|
# Instances could be in the QUEUED status for a long time if the cloud
|
|
# provider is slow to launch instances.
|
|
IMInstance.QUEUED,
|
|
]:
|
|
Reconciler._warn_stuck_instances(
|
|
instances_by_status[status],
|
|
status=status,
|
|
warn_interval_s=reconcile_config.transient_status_warn_interval_s,
|
|
logger=_logger,
|
|
)
|
|
|
|
Reconciler._update_instance_manager(instance_manager, version, im_updates)
|
|
|
|
@staticmethod
|
|
def _warn_stuck_instances(
|
|
instances: List[IMInstance],
|
|
status: IMInstance.InstanceStatus,
|
|
warn_interval_s: int,
|
|
logger: logging.Logger,
|
|
):
|
|
"""Warn if any instance is stuck in a transient/unbounded status for too
|
|
long.
|
|
"""
|
|
for instance in instances:
|
|
status_times_ns = InstanceUtil.get_status_transition_times_ns(
|
|
instance, select_instance_status=status
|
|
)
|
|
assert len(status_times_ns) >= 1
|
|
status_time_ns = sorted(status_times_ns)[-1]
|
|
|
|
if time.time_ns() - status_time_ns > warn_interval_s * 1e9:
|
|
logger.warning(
|
|
"Instance {}({}) is stuck in {} for {} seconds.".format(
|
|
instance.instance_id,
|
|
IMInstance.InstanceStatus.Name(instance.status),
|
|
IMInstance.InstanceStatus.Name(status),
|
|
(time.time_ns() - status_time_ns) // 1e9,
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def _is_head_node_running(instance_manager: InstanceManager) -> bool:
|
|
"""
|
|
Check if the head node is running and ready.
|
|
|
|
If we scale up the cluster before head node is running,
|
|
it would cause issues when launching the worker nodes.
|
|
|
|
There are corner cases when the GCS is up (so the ray cluster resource
|
|
state is retrievable from the GCS), but the head node's raylet is not
|
|
running so the head node is missing from the reported nodes. This happens
|
|
when the head node is still starting up, or the raylet is not running
|
|
due to some issues, and this would yield false.
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
|
|
Returns:
|
|
True if the head node is running and ready, False otherwise.
|
|
"""
|
|
|
|
im_instances, _ = Reconciler._get_im_instances(instance_manager)
|
|
|
|
for instance in im_instances:
|
|
if instance.node_kind == NodeKind.HEAD:
|
|
if instance.status == IMInstance.RAY_RUNNING:
|
|
return True
|
|
return False
|
|
|
|
@staticmethod
|
|
def _scale_cluster(
|
|
autoscaling_state: AutoscalingState,
|
|
instance_manager: InstanceManager,
|
|
cloud_resource_monitor: CloudResourceMonitor,
|
|
ray_state: ClusterResourceState,
|
|
scheduler: IResourceScheduler,
|
|
autoscaling_config: AutoscalingConfig,
|
|
cloud_provider: ICloudInstanceProvider,
|
|
) -> None:
|
|
"""
|
|
Scale the cluster based on the resource state and the resource scheduler's
|
|
decision:
|
|
|
|
- It launches new instances if needed.
|
|
- It terminates extra ray nodes if they should be shut down (preemption
|
|
or idle termination)
|
|
|
|
Args:
|
|
autoscaling_state: The autoscaling state to reconcile.
|
|
instance_manager: The instance manager to reconcile.
|
|
cloud_resource_monitor: The cloud resource monitor for monitoring resource
|
|
availability of all node types.
|
|
ray_state: The ray cluster's resource state.
|
|
scheduler: The resource scheduler to make scaling decisions.
|
|
autoscaling_config: The autoscaling config.
|
|
cloud_provider: The cloud provider interface.
|
|
"""
|
|
|
|
# Get the current instance states.
|
|
im_instances, version = Reconciler._get_im_instances(instance_manager)
|
|
|
|
im_instances_by_instance_id = {
|
|
i.instance_id: i for i in im_instances if i.instance_id
|
|
}
|
|
autoscaler_instances = []
|
|
ray_nodes_by_id = {
|
|
binary_to_hex(node.node_id): node for node in ray_state.node_states
|
|
}
|
|
|
|
for im_instance in im_instances:
|
|
ray_node = ray_nodes_by_id.get(im_instance.node_id)
|
|
autoscaler_instances.append(
|
|
AutoscalerInstance(
|
|
ray_node=ray_node,
|
|
im_instance=im_instance,
|
|
cloud_instance_id=(
|
|
im_instance.cloud_instance_id
|
|
if im_instance.cloud_instance_id
|
|
else None
|
|
),
|
|
)
|
|
)
|
|
|
|
# TODO(rickyx): We should probably name it as "Planner" or "Scaler"
|
|
# or "ClusterScaler"
|
|
sched_request = SchedulingRequest(
|
|
node_type_configs=autoscaling_config.get_node_type_configs(),
|
|
max_num_nodes=autoscaling_config.get_max_num_nodes(),
|
|
resource_requests=ray_state.pending_resource_requests,
|
|
gang_resource_requests=ray_state.pending_gang_resource_requests,
|
|
cluster_resource_constraints=ray_state.cluster_resource_constraints,
|
|
current_instances=autoscaler_instances,
|
|
idle_timeout_s=autoscaling_config.get_idle_timeout_s(),
|
|
disable_launch_config_check=(
|
|
autoscaling_config.disable_launch_config_check()
|
|
),
|
|
cloud_resource_availabilities=(
|
|
cloud_resource_monitor.get_resource_availabilities()
|
|
),
|
|
recoverable_resource_availabilities=(
|
|
cloud_resource_monitor.get_recoverable_resource_availabilities()
|
|
),
|
|
)
|
|
|
|
if isinstance(cloud_provider, KubeRayProvider):
|
|
sched_request.ippr_specs = cloud_provider.get_ippr_specs()
|
|
sched_request.ippr_statuses = cloud_provider.get_ippr_statuses()
|
|
|
|
# Ask scheduler for updates to the cluster shape.
|
|
reply = scheduler.schedule(sched_request)
|
|
|
|
# Populate the autoscaling state.
|
|
autoscaling_state.infeasible_resource_requests.extend(
|
|
reply.infeasible_resource_requests
|
|
)
|
|
autoscaling_state.infeasible_gang_resource_requests.extend(
|
|
reply.infeasible_gang_resource_requests
|
|
)
|
|
autoscaling_state.infeasible_cluster_resource_constraints.extend(
|
|
reply.infeasible_cluster_resource_constraints
|
|
)
|
|
|
|
if not Reconciler._is_head_node_running(instance_manager):
|
|
# We shouldn't be scaling the cluster until the head node is ready.
|
|
# This could happen when the head node (i.e. the raylet) is still
|
|
# pending registration even though GCS is up.
|
|
# We will wait until the head node is running and ready to avoid
|
|
# scaling the cluster from min worker nodes constraint.
|
|
return
|
|
|
|
if autoscaling_config.provider == Provider.READ_ONLY:
|
|
# We shouldn't be scaling the cluster if the provider is read-only.
|
|
return
|
|
|
|
# Scale the clusters if needed.
|
|
to_launch = reply.to_launch
|
|
to_terminate = reply.to_terminate
|
|
updates = {}
|
|
# Add terminating instances.
|
|
for terminate_request in to_terminate:
|
|
instance_id = terminate_request.instance_id
|
|
if terminate_request.instance_status == IMInstance.QUEUED:
|
|
# QUEUED instances have no cloud resources allocated yet.
|
|
# Cancel the allocation request by transitioning directly to TERMINATED.
|
|
updates[terminate_request.instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance_id,
|
|
new_instance_status=IMInstance.TERMINATED,
|
|
termination_request=terminate_request,
|
|
details=f"allocation canceled: {terminate_request.details}",
|
|
)
|
|
elif terminate_request.instance_status in (
|
|
IMInstance.ALLOCATED,
|
|
IMInstance.RAY_INSTALLING,
|
|
):
|
|
# The instance is not yet running, so we can't request to stop/drain Ray.
|
|
# Therefore, we can skip the RAY_STOP_REQUESTED state and directly terminate the node.
|
|
im_instance_to_terminate = im_instances_by_instance_id[instance_id]
|
|
updates[terminate_request.instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance_id,
|
|
new_instance_status=IMInstance.TERMINATING,
|
|
cloud_instance_id=im_instance_to_terminate.cloud_instance_id,
|
|
termination_request=terminate_request,
|
|
details=f"terminating ray: {terminate_request.details}",
|
|
)
|
|
else:
|
|
updates[terminate_request.instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance_id,
|
|
new_instance_status=IMInstance.RAY_STOP_REQUESTED,
|
|
termination_request=terminate_request,
|
|
details=f"draining ray: {terminate_request.details}",
|
|
)
|
|
|
|
# Add new instances.
|
|
for launch_request in to_launch:
|
|
for _ in range(launch_request.count):
|
|
instance_id = InstanceUtil.random_instance_id()
|
|
updates[instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance_id,
|
|
new_instance_status=IMInstance.QUEUED,
|
|
instance_type=launch_request.instance_type,
|
|
upsert=True,
|
|
details=(
|
|
f"queuing new instance of {launch_request.instance_type} "
|
|
"from scheduler"
|
|
),
|
|
)
|
|
|
|
if isinstance(cloud_provider, KubeRayProvider):
|
|
cloud_provider.do_ippr_requests(reply.to_ippr)
|
|
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _terminate_instances(instance_manager: InstanceManager):
|
|
"""
|
|
Terminate instances with the below statuses:
|
|
- RAY_STOPPED: ray was stopped on the cloud instance.
|
|
- ALLOCATION_TIMEOUT: cloud provider timed out to allocate a running cloud instance.
|
|
- RAY_INSTALL_FAILED: ray installation failed on the cloud instance,
|
|
we will not retry.
|
|
- TERMINATION_FAILED: cloud provider failed to terminate the instance
|
|
or timeout for termination happened, we will retry again.
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
"""
|
|
|
|
im_instances, version = Reconciler._get_im_instances(instance_manager)
|
|
updates = {}
|
|
statuses_to_terminate = {
|
|
IMInstance.RAY_STOPPED,
|
|
IMInstance.ALLOCATION_TIMEOUT,
|
|
IMInstance.RAY_INSTALL_FAILED,
|
|
IMInstance.TERMINATION_FAILED,
|
|
}
|
|
inactive_statuses = statuses_to_terminate | {
|
|
IMInstance.TERMINATED,
|
|
IMInstance.ALLOCATION_FAILED,
|
|
}
|
|
# A RAY_STOPPED row can be stale after a raylet restart on the same
|
|
# cloud instance. If another active IM row still owns that cloud
|
|
# instance, clean up only the stale row without terminating the cloud.
|
|
active_cloud_instance_ids = {
|
|
instance.cloud_instance_id
|
|
for instance in im_instances
|
|
if instance.cloud_instance_id and instance.status not in inactive_statuses
|
|
}
|
|
for instance in im_instances:
|
|
if instance.status not in statuses_to_terminate:
|
|
continue
|
|
|
|
if (
|
|
instance.status == IMInstance.RAY_STOPPED
|
|
and instance.cloud_instance_id in active_cloud_instance_ids
|
|
):
|
|
updates[instance.instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance.instance_id,
|
|
new_instance_status=IMInstance.TERMINATED,
|
|
details=(
|
|
"marking stale instance record as terminated from "
|
|
f"{IMInstance.InstanceStatus.Name(instance.status)} "
|
|
"without terminating its cloud instance"
|
|
),
|
|
)
|
|
continue
|
|
|
|
# Terminate the instance.
|
|
updates[instance.instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance.instance_id,
|
|
new_instance_status=IMInstance.TERMINATING,
|
|
cloud_instance_id=instance.cloud_instance_id,
|
|
details="terminating instance from "
|
|
f"{IMInstance.InstanceStatus.Name(instance.status)}",
|
|
)
|
|
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _install_ray(
|
|
instance_manager: InstanceManager,
|
|
non_terminated_cloud_instances: Dict[CloudInstanceId, CloudInstance],
|
|
) -> None:
|
|
"""
|
|
Install ray on the allocated instances when it's ready (cloud instance
|
|
should be running)
|
|
|
|
This is needed if ray installation needs to be performed by
|
|
the instance manager.
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
non_terminated_cloud_instances: The non-terminated cloud instances
|
|
from the cloud provider.
|
|
"""
|
|
im_instances, version = Reconciler._get_im_instances(instance_manager)
|
|
updates = {}
|
|
for instance in im_instances:
|
|
if instance.status != IMInstance.ALLOCATED:
|
|
continue
|
|
|
|
if instance.node_kind == NodeKind.HEAD:
|
|
# Skip head node.
|
|
continue
|
|
|
|
cloud_instance = non_terminated_cloud_instances.get(
|
|
instance.cloud_instance_id
|
|
)
|
|
|
|
assert cloud_instance, (
|
|
f"Cloud instance {instance.cloud_instance_id} is not found "
|
|
"in non_terminated_cloud_instances."
|
|
)
|
|
|
|
if not cloud_instance.is_running:
|
|
# It might still be pending (e.g. setting up ssh)
|
|
continue
|
|
|
|
# Install ray on the running cloud instance
|
|
updates[instance.instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=instance.instance_id,
|
|
new_instance_status=IMInstance.RAY_INSTALLING,
|
|
details="installing ray",
|
|
)
|
|
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _fill_autoscaling_state(
|
|
instance_manager: InstanceManager,
|
|
autoscaling_state: AutoscalingState,
|
|
autoscaling_config: AutoscalingConfig,
|
|
) -> None:
|
|
def get_provider_instance_type(instance_type: str) -> str:
|
|
return autoscaling_config.get_provider_instance_type(instance_type)
|
|
|
|
# Use the IM instance version for the autoscaler_state_version
|
|
instances, version = Reconciler._get_im_instances(instance_manager)
|
|
autoscaling_state.autoscaler_state_version = version
|
|
|
|
# Group instances by status
|
|
instances_by_status = defaultdict(list)
|
|
for instance in instances:
|
|
instances_by_status[instance.status].append(instance)
|
|
|
|
# Pending instance requests
|
|
instances_by_launch_request = defaultdict(list)
|
|
queued_instances = []
|
|
for instance in (
|
|
instances_by_status[IMInstance.REQUESTED]
|
|
+ instances_by_status[IMInstance.QUEUED]
|
|
):
|
|
if instance.launch_request_id:
|
|
instances_by_launch_request[instance.launch_request_id].append(instance)
|
|
else:
|
|
queued_instances.append(instance)
|
|
|
|
for _, instances in instances_by_launch_request.items():
|
|
num_instances_by_type = defaultdict(int)
|
|
for instance in instances:
|
|
num_instances_by_type[instance.instance_type] += 1
|
|
|
|
# All instances with same request id should have the same
|
|
# request time.
|
|
request_update = InstanceUtil.get_last_status_transition(
|
|
instances[0], IMInstance.REQUESTED
|
|
)
|
|
request_time_ns = request_update.timestamp_ns if request_update else 0
|
|
|
|
for instance_type, count in num_instances_by_type.items():
|
|
autoscaling_state.pending_instance_requests.append(
|
|
PendingInstanceRequest(
|
|
instance_type_name=get_provider_instance_type(instance_type),
|
|
ray_node_type_name=instance_type,
|
|
count=int(count),
|
|
request_ts=int(request_time_ns // 1e9),
|
|
)
|
|
)
|
|
|
|
# Pending instances
|
|
for instance in (
|
|
instances_by_status[IMInstance.ALLOCATED]
|
|
+ instances_by_status[IMInstance.RAY_INSTALLING]
|
|
):
|
|
status_history = sorted(
|
|
instance.status_history, key=lambda x: x.timestamp_ns, reverse=True
|
|
)
|
|
autoscaling_state.pending_instances.append(
|
|
PendingInstance(
|
|
instance_id=instance.instance_id,
|
|
instance_type_name=get_provider_instance_type(
|
|
instance.instance_type
|
|
),
|
|
ray_node_type_name=instance.instance_type,
|
|
details=status_history[0].details,
|
|
)
|
|
)
|
|
|
|
# Failed instance requests
|
|
for instance in instances_by_status[IMInstance.ALLOCATION_FAILED]:
|
|
request_status_update = InstanceUtil.get_last_status_transition(
|
|
instance, IMInstance.REQUESTED
|
|
)
|
|
failed_status_update = InstanceUtil.get_last_status_transition(
|
|
instance, IMInstance.ALLOCATION_FAILED
|
|
)
|
|
failed_time = (
|
|
failed_status_update.timestamp_ns if failed_status_update else 0
|
|
)
|
|
request_time = (
|
|
request_status_update.timestamp_ns if request_status_update else 0
|
|
)
|
|
autoscaling_state.failed_instance_requests.append(
|
|
FailedInstanceRequest(
|
|
instance_type_name=get_provider_instance_type(
|
|
instance.instance_type
|
|
),
|
|
ray_node_type_name=instance.instance_type,
|
|
start_ts=int(request_time // 1e9),
|
|
failed_ts=int(
|
|
failed_time // 1e9,
|
|
),
|
|
reason=failed_status_update.details,
|
|
count=1,
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def _handle_stuck_requested_instance(
|
|
instance: IMInstance, timeout_s: int, max_num_retry_request_to_allocate: int
|
|
) -> Optional[IMInstanceUpdateEvent]:
|
|
"""
|
|
Fail the cloud instance allocation if it's stuck in the REQUESTED state.
|
|
|
|
Args:
|
|
instance: The instance to handle.
|
|
timeout_s: The timeout in seconds.
|
|
max_num_retry_request_to_allocate: The maximum number of times an instance
|
|
could be requested to allocate.
|
|
|
|
Returns:
|
|
Instance update to ALLOCATION_FAILED: if the instance allocation failed
|
|
with errors.
|
|
None: if there's no update.
|
|
|
|
"""
|
|
if not InstanceUtil.has_timeout(instance, timeout_s):
|
|
# Not timeout yet, be patient.
|
|
return None
|
|
|
|
all_request_times_ns = sorted(
|
|
InstanceUtil.get_status_transition_times_ns(
|
|
instance, select_instance_status=IMInstance.REQUESTED
|
|
)
|
|
)
|
|
# Fail the allocation if we have tried too many times.
|
|
if len(all_request_times_ns) > max_num_retry_request_to_allocate:
|
|
return IMInstanceUpdateEvent(
|
|
instance_id=instance.instance_id,
|
|
new_instance_status=IMInstance.ALLOCATION_FAILED,
|
|
details=(
|
|
"failed to allocate cloud instance after "
|
|
f"{len(all_request_times_ns)} attempts > "
|
|
f"max_num_retry_request_to_allocate={max_num_retry_request_to_allocate}" # noqa
|
|
),
|
|
)
|
|
|
|
# Retry the allocation if we could by transitioning to QUEUED again.
|
|
return IMInstanceUpdateEvent(
|
|
instance_id=instance.instance_id,
|
|
new_instance_status=IMInstance.QUEUED,
|
|
details=f"queue again to launch after timeout={timeout_s}s",
|
|
)
|
|
|
|
@staticmethod
|
|
def _handle_stuck_instance(
|
|
instance: IMInstance,
|
|
timeout_s: int,
|
|
new_status: IMInstance.InstanceStatus,
|
|
**update_kwargs: Dict,
|
|
) -> Optional[IMInstanceUpdateEvent]:
|
|
"""
|
|
Fail the instance if it's stuck in the status for too long.
|
|
|
|
Args:
|
|
instance: The instance to handle.
|
|
timeout_s: The timeout in seconds.
|
|
new_status: The new status to transition to.
|
|
**update_kwargs: Keyword arguments for InstanceUpdateEvent.
|
|
|
|
Returns:
|
|
Instance update to the new status: if the instance is stuck in the status
|
|
for too long.
|
|
None: if there's no update.
|
|
|
|
"""
|
|
if not InstanceUtil.has_timeout(instance, timeout_s):
|
|
# Not timeout yet, be patient.
|
|
return None
|
|
|
|
return IMInstanceUpdateEvent(
|
|
instance_id=instance.instance_id,
|
|
new_instance_status=new_status,
|
|
details=f"timeout={timeout_s}s at status "
|
|
f"{IMInstance.InstanceStatus.Name(instance.status)}",
|
|
**update_kwargs,
|
|
)
|
|
|
|
@staticmethod
|
|
def _handle_extra_cloud_instances(
|
|
instance_manager: InstanceManager,
|
|
non_terminated_cloud_instances: Dict[CloudInstanceId, CloudInstance],
|
|
ray_nodes: List[NodeState],
|
|
):
|
|
"""
|
|
For extra cloud instances (i.e. cloud instances that are non terminated as
|
|
returned by cloud provider, but not managed by the instance manager), we
|
|
will create new IM instances with ALLOCATED status.
|
|
|
|
Such instances could either be:
|
|
1. Leaked instances that are incorrectly started by the cloud instance
|
|
provider, and they would be terminated eventually if they fail to
|
|
transition to RAY_RUNNING by stuck instances reconciliation, or they
|
|
would join the ray cluster and be terminated when the cluster scales down.
|
|
2. Instances that are started by the cloud instance provider intentionally
|
|
but not yet discovered by the instance manager. This could happen for
|
|
a. Head node that's started before the autoscaler.
|
|
b. Worker nodes that's started by the cloud provider upon users'
|
|
actions: i.e. KubeRay scaling up the cluster with ray cluster config
|
|
change.
|
|
3. Ray nodes with cloud instance id not in the cloud provider. This could
|
|
happen if there's delay in the Ray's state (i.e. cloud instance already
|
|
terminated, but the ray node is still not dead yet).
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
non_terminated_cloud_instances: The non-terminated cloud instances from
|
|
the cloud provider.
|
|
ray_nodes: The ray cluster's states of ray nodes.
|
|
"""
|
|
Reconciler._handle_extra_cloud_instances_from_ray_nodes(
|
|
instance_manager, ray_nodes
|
|
)
|
|
Reconciler._handle_extra_cloud_instances_from_cloud_provider(
|
|
instance_manager, non_terminated_cloud_instances
|
|
)
|
|
|
|
@staticmethod
|
|
def _handle_extra_cloud_instances_from_cloud_provider(
|
|
instance_manager: InstanceManager,
|
|
non_terminated_cloud_instances: Dict[CloudInstanceId, CloudInstance],
|
|
):
|
|
"""
|
|
For extra cloud instances that are not managed by the instance manager but
|
|
are running in the cloud provider, we will create new IM instances with
|
|
ALLOCATED status.
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
non_terminated_cloud_instances: The non-terminated cloud instances from
|
|
the cloud provider.
|
|
"""
|
|
updates = {}
|
|
|
|
instances, version = Reconciler._get_im_instances(instance_manager)
|
|
cloud_instance_ids_managed_by_im = {
|
|
instance.cloud_instance_id
|
|
for instance in instances
|
|
if instance.cloud_instance_id
|
|
and instance.status
|
|
not in [IMInstance.TERMINATED, IMInstance.ALLOCATION_FAILED]
|
|
}
|
|
|
|
# Find the extra cloud instances that are not managed by the instance manager.
|
|
for cloud_instance_id, cloud_instance in non_terminated_cloud_instances.items():
|
|
if cloud_instance_id in cloud_instance_ids_managed_by_im:
|
|
continue
|
|
updates[cloud_instance_id] = IMInstanceUpdateEvent(
|
|
instance_id=InstanceUtil.random_instance_id(), # Assign a new id.
|
|
cloud_instance_id=cloud_instance_id,
|
|
new_instance_status=IMInstance.ALLOCATED,
|
|
node_kind=cloud_instance.node_kind,
|
|
instance_type=cloud_instance.node_type,
|
|
details=(
|
|
"allocated unmanaged cloud instance :"
|
|
f"{cloud_instance.cloud_instance_id} "
|
|
f"({NodeKind.Name(cloud_instance.node_kind)}) from cloud provider"
|
|
),
|
|
upsert=True,
|
|
)
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _handle_extra_cloud_instances_from_ray_nodes(
|
|
instance_manager: InstanceManager, ray_nodes: List[NodeState]
|
|
):
|
|
"""
|
|
For extra cloud instances reported by Ray but not managed by the instance
|
|
manager, we will create new IM instances with ALLOCATED status.
|
|
|
|
Args:
|
|
instance_manager: The instance manager to reconcile.
|
|
ray_nodes: The ray cluster's states of ray nodes.
|
|
"""
|
|
updates = {}
|
|
|
|
instances, version = Reconciler._get_im_instances(instance_manager)
|
|
cloud_instance_ids_managed_by_im = {
|
|
instance.cloud_instance_id
|
|
for instance in instances
|
|
if instance.cloud_instance_id
|
|
and not instance.node_id
|
|
and instance.status
|
|
not in [IMInstance.TERMINATED, IMInstance.ALLOCATION_FAILED]
|
|
}
|
|
ray_node_ids_managed_by_im = {
|
|
instance.node_id for instance in instances if instance.node_id
|
|
}
|
|
|
|
for ray_node in ray_nodes:
|
|
if not ray_node.instance_id:
|
|
continue
|
|
|
|
ray_node_id = binary_to_hex(ray_node.node_id)
|
|
if ray_node_id in ray_node_ids_managed_by_im:
|
|
continue
|
|
|
|
cloud_instance_id = ray_node.instance_id
|
|
if cloud_instance_id in cloud_instance_ids_managed_by_im:
|
|
continue
|
|
|
|
is_head = is_head_node(ray_node)
|
|
updates[ray_node_id] = IMInstanceUpdateEvent(
|
|
instance_id=InstanceUtil.random_instance_id(), # Assign a new id.
|
|
cloud_instance_id=cloud_instance_id,
|
|
new_instance_status=IMInstance.ALLOCATED,
|
|
node_kind=NodeKind.HEAD if is_head else NodeKind.WORKER,
|
|
ray_node_id=ray_node_id,
|
|
instance_type=ray_node.ray_node_type_name,
|
|
details=(
|
|
"allocated unmanaged worker cloud instance from ray node: "
|
|
f"{ray_node_id}"
|
|
),
|
|
upsert=True,
|
|
)
|
|
|
|
Reconciler._update_instance_manager(instance_manager, version, updates)
|
|
|
|
@staticmethod
|
|
def _report_metrics(
|
|
instance_manager: InstanceManager,
|
|
autoscaling_config: AutoscalingConfig,
|
|
metrics_reporter: Optional[AutoscalerMetricsReporter] = None,
|
|
):
|
|
if not metrics_reporter:
|
|
return
|
|
|
|
instances, _ = Reconciler._get_im_instances(instance_manager)
|
|
node_type_configs = autoscaling_config.get_node_type_configs()
|
|
|
|
metrics_reporter.report_instances(instances, node_type_configs)
|
|
metrics_reporter.report_resources(instances, node_type_configs)
|