wehub-resource-sync
322 lines
11 KiB
Python
322 lines
11 KiB
Python
import time
|
|
|
|
from ray_release.anyscale_util import create_cluster_env_from_image
|
|
from ray_release.cluster_manager.cluster_manager import ClusterManager
|
|
from ray_release.config import COMPUTE_CONFIG_MODEL_FIELDS
|
|
from ray_release.exception import (
|
|
ClusterComputeCreateError,
|
|
ClusterEnvBuildError,
|
|
ClusterEnvBuildTimeout,
|
|
ClusterEnvCreateError,
|
|
)
|
|
from ray_release.logger import logger
|
|
from ray_release.retry import retry
|
|
from ray_release.util import (
|
|
anyscale_cluster_env_build_url,
|
|
format_link,
|
|
)
|
|
|
|
REPORT_S = 30.0
|
|
|
|
|
|
class MinimalClusterManager(ClusterManager):
|
|
"""Minimal manager.
|
|
|
|
Builds app config and compute template but does not start or stop session.
|
|
"""
|
|
|
|
@retry(
|
|
init_delay_sec=10,
|
|
jitter_sec=5,
|
|
max_retry_count=2,
|
|
exceptions=(ClusterEnvCreateError,),
|
|
)
|
|
def create_cluster_env(self):
|
|
assert self.cluster_env_id is None
|
|
assert self.cluster_env_name
|
|
|
|
logger.info(
|
|
f"Test uses a cluster env with name "
|
|
f"{self.cluster_env_name}. Looking up existing "
|
|
f"cluster envs with this name."
|
|
)
|
|
|
|
self.cluster_env_id = create_cluster_env_from_image(
|
|
image=self.test.get_anyscale_byod_image(),
|
|
test_name=self.cluster_env_name,
|
|
runtime_env=self.test.get_byod_runtime_env(),
|
|
sdk=self.sdk,
|
|
cluster_env_id=self.cluster_env_id,
|
|
cluster_env_name=self.cluster_env_name,
|
|
)
|
|
|
|
def build_cluster_env(self, timeout: float = 600.0):
|
|
assert self.cluster_env_id
|
|
assert self.cluster_env_build_id is None
|
|
|
|
# Fetch build
|
|
build_id = None
|
|
last_status = None
|
|
error_message = None
|
|
config_json = None
|
|
result = self.sdk.list_cluster_environment_builds(self.cluster_env_id)
|
|
if not result or not result.results:
|
|
raise ClusterEnvBuildError(f"No build found for cluster env: {result}")
|
|
|
|
build = sorted(result.results, key=lambda b: b.created_at)[-1]
|
|
build_id = build.id
|
|
last_status = build.status
|
|
error_message = build.error_message
|
|
config_json = build.config_json
|
|
|
|
if last_status == "succeeded":
|
|
logger.info(
|
|
f"Link to succeeded cluster env build: "
|
|
f"{format_link(anyscale_cluster_env_build_url(build_id))}"
|
|
)
|
|
self.cluster_env_build_id = build_id
|
|
return
|
|
|
|
if last_status == "failed":
|
|
logger.info(f"Previous cluster env build failed: {error_message}")
|
|
logger.info("Starting new cluster env build...")
|
|
|
|
# Retry build
|
|
result = self.sdk.create_cluster_environment_build(
|
|
dict(
|
|
cluster_environment_id=self.cluster_env_id, config_json=config_json
|
|
)
|
|
)
|
|
build_id = result.result.id
|
|
|
|
logger.info(
|
|
f"Link to created cluster env build: "
|
|
f"{format_link(anyscale_cluster_env_build_url(build_id))}"
|
|
)
|
|
|
|
# Build found but not failed/finished yet
|
|
completed = False
|
|
start_wait = time.time()
|
|
next_report = start_wait + REPORT_S
|
|
timeout_at = time.monotonic() + timeout
|
|
logger.info(f"Waiting for build {build_id} to finish...")
|
|
logger.info(
|
|
f"Track progress here: "
|
|
f"{format_link(anyscale_cluster_env_build_url(build_id))}"
|
|
)
|
|
while not completed:
|
|
now = time.time()
|
|
if now > next_report:
|
|
logger.info(
|
|
f"... still waiting for build {build_id} to finish "
|
|
f"({int(now - start_wait)} seconds) ..."
|
|
)
|
|
next_report = next_report + REPORT_S
|
|
|
|
result = self.sdk.get_build(build_id)
|
|
build = result.result
|
|
|
|
if build.status == "failed":
|
|
raise ClusterEnvBuildError(
|
|
f"Cluster env build failed. Please see "
|
|
f"{anyscale_cluster_env_build_url(build_id)} for details. "
|
|
f"Error message: {build.error_message}"
|
|
)
|
|
|
|
if build.status == "succeeded":
|
|
logger.info("Build succeeded.")
|
|
self.cluster_env_build_id = build_id
|
|
return
|
|
|
|
completed = build.status not in ["in_progress", "pending"]
|
|
|
|
if completed:
|
|
raise ClusterEnvBuildError(
|
|
f"Unknown build status: {build.status}. Please see "
|
|
f"{anyscale_cluster_env_build_url(build_id)} for details"
|
|
)
|
|
|
|
if time.monotonic() > timeout_at:
|
|
raise ClusterEnvBuildTimeout(
|
|
f"Time out when building cluster env {self.cluster_env_name}"
|
|
)
|
|
|
|
time.sleep(1)
|
|
|
|
self.cluster_env_build_id = build_id
|
|
|
|
def create_cluster_compute(self, _repeat: bool = True):
|
|
assert self.cluster_compute_id is None
|
|
|
|
if self.cluster_compute and self.test.uses_anyscale_sdk_2026():
|
|
return self._create_cluster_compute_new_sdk(_repeat=_repeat)
|
|
|
|
if self.cluster_compute:
|
|
assert self.cluster_compute
|
|
|
|
logger.info(
|
|
f"Tests uses compute template "
|
|
f"with name {self.cluster_compute_name}. "
|
|
f"Looking up existing cluster computes."
|
|
)
|
|
|
|
paging_token = None
|
|
while not self.cluster_compute_id:
|
|
result = self.sdk.search_cluster_computes(
|
|
dict(
|
|
project_id=self.project_id,
|
|
name=dict(equals=self.cluster_compute_name),
|
|
include_anonymous=True,
|
|
paging=dict(paging_token=paging_token),
|
|
)
|
|
)
|
|
paging_token = result.metadata.next_paging_token
|
|
|
|
for res in result.results:
|
|
if res.name == self.cluster_compute_name:
|
|
self.cluster_compute_id = res.id
|
|
logger.info(
|
|
f"Cluster compute already exists "
|
|
f"with ID {self.cluster_compute_id}"
|
|
)
|
|
break
|
|
|
|
if not paging_token:
|
|
break
|
|
|
|
if not self.cluster_compute_id:
|
|
logger.info(
|
|
f"Cluster compute not found. "
|
|
f"Creating with name {self.cluster_compute_name}."
|
|
)
|
|
try:
|
|
result = self.sdk.create_cluster_compute(
|
|
dict(
|
|
name=self.cluster_compute_name,
|
|
project_id=self.project_id,
|
|
config=self.cluster_compute,
|
|
)
|
|
)
|
|
self.cluster_compute_id = result.result.id
|
|
except Exception as e:
|
|
if _repeat:
|
|
logger.warning(
|
|
f"Got exception when trying to create cluster "
|
|
f"compute: {e}. Sleeping for 10 seconds and then "
|
|
f"try again once..."
|
|
)
|
|
time.sleep(10)
|
|
return self.create_cluster_compute(_repeat=False)
|
|
|
|
raise ClusterComputeCreateError(
|
|
"Could not create cluster compute"
|
|
) from e
|
|
|
|
logger.info(
|
|
f"Cluster compute template created with "
|
|
f"name {self.cluster_compute_name} and "
|
|
f"ID {self.cluster_compute_id}"
|
|
)
|
|
|
|
def _create_cluster_compute_new_sdk(self, _repeat: bool = True):
|
|
"""Create cluster compute using the anyscale.compute_config API (2026 SDK)."""
|
|
assert self.cluster_compute
|
|
|
|
logger.info(
|
|
f"Test uses a compute config (2026 SDK) with name "
|
|
f"{self.cluster_compute_name}. Looking up existing "
|
|
f"compute configs with this name."
|
|
)
|
|
|
|
# Check if compute config already exists by name
|
|
anyscale_sdk = self.test.anyscale
|
|
try:
|
|
compute_config_version = anyscale_sdk.compute_config.get(
|
|
self.cluster_compute_name
|
|
)
|
|
self.cluster_compute_id = compute_config_version.id
|
|
logger.info(
|
|
f"Compute config already exists " f"with ID {self.cluster_compute_id}"
|
|
)
|
|
return
|
|
except RuntimeError as e:
|
|
if "not found" not in str(e):
|
|
raise
|
|
logger.info(
|
|
f"Compute config not found. "
|
|
f"Creating with name {self.cluster_compute_name}."
|
|
)
|
|
|
|
# Build ComputeConfig from the cluster compute dict, excluding
|
|
# keys like idle_termination_minutes/maximum_uptime_minutes that
|
|
# were added by set_cluster_compute() and are not part of the
|
|
# ComputeConfig model.
|
|
ComputeConfig = anyscale_sdk.compute_config.ComputeConfig
|
|
config_dict = {
|
|
k: v
|
|
for k, v in self.cluster_compute.items()
|
|
if k in COMPUTE_CONFIG_MODEL_FIELDS
|
|
}
|
|
config = ComputeConfig.from_dict(config_dict)
|
|
|
|
try:
|
|
full_name = anyscale_sdk.compute_config.create(
|
|
config, name=self.cluster_compute_name
|
|
)
|
|
compute_config_version = anyscale_sdk.compute_config.get(full_name)
|
|
self.cluster_compute_id = compute_config_version.id
|
|
except Exception as e:
|
|
if _repeat:
|
|
logger.warning(
|
|
f"Got exception when trying to create compute "
|
|
f"config: {e}. Sleeping for 10 seconds and then "
|
|
f"try again once..."
|
|
)
|
|
time.sleep(10)
|
|
return self._create_cluster_compute_new_sdk(_repeat=False)
|
|
|
|
raise ClusterComputeCreateError("Could not create cluster compute") from e
|
|
|
|
logger.info(
|
|
f"Compute config created with "
|
|
f"name {full_name} and "
|
|
f"ID {self.cluster_compute_id}"
|
|
)
|
|
|
|
def build_configs(self, timeout: float = 30.0):
|
|
try:
|
|
self.create_cluster_compute()
|
|
except AssertionError as e:
|
|
# If already exists, ignore
|
|
logger.warning(str(e))
|
|
except ClusterComputeCreateError as e:
|
|
raise e
|
|
except Exception as e:
|
|
raise ClusterComputeCreateError(
|
|
f"Unexpected cluster compute build error: {e}"
|
|
) from e
|
|
|
|
try:
|
|
self.create_cluster_env()
|
|
except AssertionError as e:
|
|
# If already exists, ignore
|
|
logger.warning(str(e))
|
|
except ClusterEnvCreateError as e:
|
|
raise e
|
|
except Exception as e:
|
|
raise ClusterEnvCreateError(
|
|
f"Unexpected cluster env create error: {e}"
|
|
) from e
|
|
|
|
try:
|
|
self.build_cluster_env(timeout=timeout)
|
|
except AssertionError as e:
|
|
# If already exists, ignore
|
|
logger.warning(str(e))
|
|
except (ClusterEnvBuildError, ClusterEnvBuildTimeout) as e:
|
|
raise e
|
|
except Exception as e:
|
|
raise ClusterEnvBuildError(
|
|
f"Unexpected cluster env build error: {e}"
|
|
) from e
|