[core] autoscale with resource availability #58619
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| import logging | ||
| import time | ||
| from typing import List, Dict | ||
|
|
||
| from ray.autoscaler.v2.instance_manager.common import InstanceUtil | ||
| from ray.autoscaler.v2.instance_manager.instance_manager import ( | ||
| InstanceUpdatedSubscriber | ||
| ) | ||
| from ray.autoscaler.v2.instance_manager.instance_storage import InstanceStorage | ||
| from ray.core.generated.instance_manager_pb2 import Instance, InstanceUpdateEvent | ||
| from ray.autoscaler.v2.schema import NodeType | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| class CloudResourceAvailability: | ||
| """CloudResourceAvailability indicates the availability of a type of | ||
| cloud resource. | ||
|
|
||
| During scaling up, if resource of a node type is requested but fail to | ||
| allocate, that type is considered unavailable at that timestamp. | ||
| This class records the last timestamp at which a node type is unavailable, | ||
| allowing the autoscaler to skip such node types when making future scaling | ||
| decisions. | ||
| """ | ||
|
|
||
| # The node type of cloud resource. | ||
| _node_type: NodeType | ||
| # Timestamp of the last failed resource allocation. | ||
| _last_unavailable_timestamp: int | ||
|
|
||
| def __init__( | ||
| self, | ||
| node_type: NodeType, | ||
| last_unavailability_timestamp: int | ||
| ): | ||
| self._node_type = node_type | ||
| self._last_unavailable_timestamp = last_unavailability_timestamp | ||
|
|
||
| def get_last_unavailability_timestamp(self) -> int: | ||
| return self._last_unavailable_timestamp | ||
|
|
||
| def set_last_unavailability_timestamp(self, timestamp: int): | ||
| self._last_unavailable_timestamp = timestamp | ||
|
|
||
|
|
||
| class CloudResourceMonitor(InstanceUpdatedSubscriber): | ||
| """CloudResourceMonitor records the availability of all node types. | ||
|
|
||
| In the Spot scenario, the resources in the cluster change dynamically. | ||
| When scaling up, it is necessary to know which node types are most | ||
| likely to have resources, in order to decide which type of node to request. | ||
| """ | ||
| def __init__( | ||
| self, | ||
| instance_storage: InstanceStorage, | ||
| ) -> None: | ||
| self._instance_storage = instance_storage | ||
| self._resource_availabilities: Dict[NodeType, CloudResourceAvailability] = {} | ||
|
|
||
| def allocation_failed(self, failed_event: InstanceUpdateEvent): | ||
| instances, _ = self._instance_storage.get_instances( | ||
| instance_ids=[failed_event.instance_id] | ||
| ) | ||
| for instance in instances.values(): | ||
| last_status = InstanceUtil.get_last_status_transition(instance) | ||
| if last_status: | ||
| last_unavailability_timestamp=(last_status.timestamp_ns) / 1000 | ||
| else: | ||
| last_unavailability_timestamp = time.time() | ||
|
There was a problem hiding this comment. Bug: Timestamp Unit Mismatch Causes Availability ErrorsUnit mismatch in timestamp handling. Line 67 converts |
||
| self._resource_availabilities[instance.instance_type] = ( | ||
| CloudResourceAvailability( | ||
| node_type=instance.instance_type, | ||
| last_unavailability_timestamp=last_unavailability_timestamp | ||
| ) | ||
| ) | ||
| logger.debug(f"Cloud Resource Type {instance.instance_type} is " | ||
| f"unavailable at timestamp={last_unavailability_timestamp}. " | ||
| f"We will lower its priority in feature schedules." | ||
| ) | ||
|
|
||
| def allocation_succeeded(self, succeeded_event: InstanceUpdateEvent): | ||
| if succeeded_event.instance_type in self._resource_availabilities: | ||
| self._resource_availabilities[ | ||
| succeeded_event.instance_type | ||
| ].set_last_unavailability_timestamp(0) | ||
| logger.debug(f"Cloud Resource Type {succeeded_event.instance_type} is " | ||
| f"available. We will prioritize scheduling this type " | ||
| f"in feature schedules." | ||
| ) | ||
|
|
||
| def notify(self, events: List[InstanceUpdateEvent]) -> None: | ||
| for event in events: | ||
| if event.new_instance_status == Instance.ALLOCATION_FAILED: | ||
| self.allocation_failed(event) | ||
| elif event.new_instance_status == Instance.ALLOCATED: | ||
| self.allocation_succeeded(event) | ||
|
|
||
| def get_resource_availabilities(self) -> Dict[NodeType, float]: | ||
| """Calculate the availability scores of node types. | ||
| Higher values indicate a higher likelihood of resource allocation. | ||
| """ | ||
| resource_availability_scores: Dict[NodeType, float] = {} | ||
| if self._resource_availabilities: | ||
| max_ts = max( | ||
| [ | ||
| r.get_last_unavailability_timestamp() | ||
| for r in self._resource_availabilities.values() | ||
| ] | ||
| ) | ||
| for node_type in self._resource_availabilities: | ||
| resource_availability_scores[node_type] = ( | ||
| 1 - | ||
| self._resource_availabilities[node_type].get_last_unavailability_timestamp() | ||
| / max_ts | ||
| ) if max_ts > 0 else 1 | ||
| return resource_availability_scores | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There's an inconsistency in the time units used for
last_unavailability_timestamp. Here,last_status.timestamp_ns(in nanoseconds) is divided by 1000, resulting in microseconds. However, in theelseblock on line 69,time.time()is used, which returns seconds.This will lead to incorrect availability scores because the timestamps are not comparable. All timestamps should be in the same unit. I suggest using seconds consistently.