Files
trainer/sdk/python/test/e2e/utils.py
T
Andrey VelichkevichandGitHub a44b9d0b93 [SDK] Add resources per worker for Create Job API (#1990)
* [SDK] Add resources for create Job API

* Fix unbound var

* Assign values in get pod template

* Add torchrun issue

* Test to create PyTorchJob from Image

* Fix e2e to create from image

* Fix condition

* Modify check test conditions
2024-01-18 16:38:59 +00:00

89 lines
3.1 KiB
Python

import logging
import time
from kubeflow.training import TrainingClient
from kubeflow.training.constants import constants
from test.e2e.constants import TEST_GANG_SCHEDULER_NAME_SCHEDULER_PLUGINS
from test.e2e.constants import DEFAULT_SCHEDULER_PLUGINS_NAME
from test.e2e.constants import TEST_GANG_SCHEDULER_NAME_VOLCANO
logging.basicConfig(format="%(message)s")
logging.getLogger().setLevel(logging.INFO)
def verify_unschedulable_job_e2e(client: TrainingClient, name: str, namespace: str):
"""Verify unschedulable Training Job e2e test."""
logging.info(f"\n\n\n{client.job_kind} is creating")
job = client.wait_for_job_conditions(
name, namespace, expected_conditions={constants.JOB_CONDITION_CREATED}
)
logging.info("Checking 3 times that pods are not scheduled")
for num in range(3):
logging.info(f"Number of attempts: {int(num)+1}/3")
# Job should have correct conditions
if not client.is_job_created(job=job) or client.is_job_running(job=job):
raise Exception(
f"{client.job_kind} should be in Created condition. "
f"{client.job_kind} should not be in Running condition."
)
logging.info("Sleeping 5 seconds...")
time.sleep(5)
def verify_job_e2e(
client: TrainingClient,
name: str,
namespace: str,
wait_timeout: int = 600,
):
"""Verify Training Job e2e test."""
# Wait until Job is Succeeded.
logging.info(f"\n\n\n{client.job_kind} is running")
job = client.wait_for_job_conditions(name, namespace, wait_timeout=wait_timeout)
# Job should have Created, Running, and Succeeded conditions.
conditions = client.get_job_conditions(job=job)
# If Job is complete fast, it has 2 conditions: Created and Succeeded.
if len(conditions) < 2:
raise Exception(f"{client.job_kind} conditions are invalid: {conditions}")
# Job should have correct conditions.
if (
not client.is_job_created(job=job)
or not client.is_job_succeeded(job=job)
or client.is_job_running(job=job)
or client.is_job_restarting(job=job)
or client.is_job_failed(job=job)
):
raise Exception(
f"{client.job_kind} should be in Succeeded and Created conditions. "
f"{client.job_kind} should not be in Running, Restarting, or Failed conditions."
)
def get_pod_spec_scheduler_name(gang_scheduler_name: str) -> str:
if gang_scheduler_name == TEST_GANG_SCHEDULER_NAME_SCHEDULER_PLUGINS:
return DEFAULT_SCHEDULER_PLUGINS_NAME
elif gang_scheduler_name == TEST_GANG_SCHEDULER_NAME_VOLCANO:
return TEST_GANG_SCHEDULER_NAME_VOLCANO
return ""
def print_job_results(client: TrainingClient, name: str, namespace: str):
# Print Job.
logging.info(f"\n\n\n{client.job_kind} info")
logging.info(client.get_job(name, namespace))
# Print Job pod names.
logging.info(f"\n\n\n{client.job_kind} pod names")
logging.info(client.get_job_pod_names(name, namespace))
# Print Job logs.
logging.info(f"\n\n\n{client.job_kind} logs")
client.get_job_logs(name, namespace)