codeflare_sdk.ray.cluster package

Submodules

codeflare_sdk.ray.cluster.build_ray_cluster module

This sub-module exists primarily to be used internally by the Cluster object

(in the cluster sub-module) for RayCluster generation.

codeflare_sdk.ray.cluster.build_ray_cluster.add_queue_label(cluster: Cluster, labels: dict)

The add_queue_label() function updates the given base labels with the local queue label if Kueue exists on the Cluster

codeflare_sdk.ray.cluster.build_ray_cluster.build_ray_cluster(cluster: Cluster)

build_ray_cluster is used for creating a Ray Cluster dict

The resource is a dict template which uses Kubernetes Objects for creating metadata, resource requests, specs and containers. The result is sanitised and returned either as a dict or written as a yaml file.

codeflare_sdk.ray.cluster.build_ray_cluster.gen_names(name)

Generates a unique name for the Ray Cluster

codeflare_sdk.ray.cluster.build_ray_cluster.generate_env_vars(cluster)

User-supplied environment variables.

codeflare_sdk.ray.cluster.build_ray_cluster.generate_image_pull_secrets(cluster)

Secret references for pulling the Ray image.

codeflare_sdk.ray.cluster.build_ray_cluster.get_default_local_queue(cluster: Cluster, labels: dict)

The get_default_local_queue() function attempts to find a local queue with the default label == true, if that is the case the labels variable is updated with that local queue

codeflare_sdk.ray.cluster.build_ray_cluster.get_head_container_spec(cluster)

The ray-head container.

codeflare_sdk.ray.cluster.build_ray_cluster.get_labels(cluster: Cluster)

The get_labels() function generates a dict “labels” which includes the base label, local queue label and user defined labels

codeflare_sdk.ray.cluster.build_ray_cluster.get_metadata(cluster: Cluster)

The get_metadata() function builds and returns a V1ObjectMeta Object using cluster configuration parameters

codeflare_sdk.ray.cluster.build_ray_cluster.get_pod_spec(cluster: Cluster, containers: List, tolerations: List[V1Toleration]) → V1PodSpec

The get_pod_spec() function generates a V1PodSpec for the head/worker containers

codeflare_sdk.ray.cluster.build_ray_cluster.get_worker_container_spec(cluster)

The machine-learning worker container.

codeflare_sdk.ray.cluster.build_ray_cluster.head_worker_extended_resources_from_cluster(cluster) → Tuple[dict, dict]

Non-GPU/CPU/memory resources for the head and worker rayStartParams.

codeflare_sdk.ray.cluster.build_ray_cluster.head_worker_gpu_count_from_cluster(cluster) → Tuple[int, int]

Total GPUs requested for the head and for a worker, respectively.

codeflare_sdk.ray.cluster.build_ray_cluster.local_queue_exists(cluster: Cluster)

The local_queue_exists() checks if the user inputted local_queue exists in the given namespace and returns a bool

codeflare_sdk.ray.cluster.build_ray_cluster.with_nb_annotations(annotations: dict)

The with_nb_annotations() function generates the annotation for NB Prefix if the SDK is running in a notebook and appends any user set annotations

codeflare_sdk.ray.cluster.build_ray_cluster.write_to_file(cluster: Cluster, resource: dict)

The write_to_file function writes the built Ray Cluster dict as a yaml file in the .codeflare folder

codeflare_sdk.ray.cluster.cluster module

The cluster sub-module contains the definition of the Cluster object, which represents the resources requested by the user. It also contains functions for checking the cluster setup queue, a list of all existing clusters, and the user’s working namespace.

class codeflare_sdk.ray.cluster.cluster.Cluster(config: ClusterConfiguration, api_client: ApiClient | None = None)

Bases: object

An object for requesting, bringing up, and taking down resources. Can also be used for seeing the resource cluster status and details.

Note that currently, the underlying implementation is a Ray cluster.

apply(force: bool = False, timeout: int = 300) → None

Applies the Cluster yaml using server-side apply.

After applying the cluster resources, this method waits for the CA secret to be created by KubeRay and generates TLS certificates for mTLS connections.

Args:

force (bool):

If True, conflicts will be forced during server-side apply.

timeout (int):

Maximum time in seconds to wait for post-apply operations such as TLS certificate generation. Default is 300 seconds (5 minutes).

cluster_dashboard_uri() → str

Returns a string containing the cluster’s dashboard URI. Tries HTTPRoute first (RHOAI v3.0+), then falls back to OpenShift Routes or Ingresses.

cluster_uri() → str

Returns a string containing the cluster’s URI.

Note: If connecting to a cluster with mTLS enabled, ensure you have called cluster.wait_ready() first to automatically generate TLS certificates.

config_check() → str

create_resource() → None

Called upon cluster object creation, creates a RayCluster yaml based on the specifications of the ClusterConfiguration.

details(print_to_console: bool = True) → RayCluster

Retrieves details about the Ray Cluster.

This method returns a copy of the Ray Cluster information and optionally prints the details to the console.

Args:

print_to_console (bool):

Flag to determine if the cluster details should be printed to the console. Defaults to True.

Returns:

RayCluster:

A copy of the Ray Cluster details.

down() → None

Deletes the RayCluster, scaling-down and deleting all resources associated with the cluster.

Also removes the TLS certificates generated for this cluster to prevent accumulation of sensitive key material.

get_dynamic_client() → DynamicClient

is_dashboard_ready() → bool

Checks if the cluster’s dashboard is ready and accessible.

This method attempts to send a GET request to the cluster dashboard URI. If the request is successful (HTTP status code 200), it returns True. For OAuth-protected dashboards, a 302 redirect to the OAuth login page also indicates the dashboard is ready (the OAuth proxy is working). If an SSL error occurs, it returns False, indicating the dashboard is not ready.

Returns:

bool:

True if the dashboard is ready, False otherwise.

property job_client

job_logs(job_id: str) → str

This method accesses the head ray node in your cluster and returns the logs for the provided job id.

job_status(job_id: str) → str

This method accesses the head ray node in your cluster and returns the job status for the provided job id.

list_jobs() → List[Dict]

This method accesses the head ray node in your cluster and lists the running jobs.

local_client_url() → str

Constructs the URL for the local Ray client.

Returns:

str:

The Ray client URL based on the ingress domain.

refresh_certificates() → None

Refreshes TLS certificates by removing old ones and generating new ones.

This is useful when: - The server CA secret has been rotated - Certificates have expired - You encounter TLS handshake failures - You need to regenerate certificates for any reason

The method will: 1. Remove existing client certificates 2. Fetch the latest CA from Kubernetes 3. Generate new client certificates 4. Update environment variables for Ray

Example:

>>> # If you get TLS errors after CA rotation
>>> cluster.refresh_certificates()
>>> # Now you can reconnect
>>> ray.init(address=cluster.cluster_uri())

status(print_to_console: bool = True) → Tuple[CodeFlareClusterStatus, bool]

Returns the requested cluster’s status, as well as whether or not it is ready for use.

up() → None

Applies the Cluster yaml, pushing the resource request onto the Kueue localqueue.

wait_ready(timeout: int | None = None, dashboard_check: bool = True) → None

Waits for the requested cluster to be ready, up to an optional timeout.

This method checks the status of the cluster every five seconds until it is ready or the timeout is reached. If dashboard_check is enabled, it will also check for the readiness of the dashboard.

TLS certificates are ensured to exist for mTLS connections. If certificates were already generated by apply(), this is a no-op. Otherwise, certificates will be generated when the cluster becomes ready.

Args:

timeout (Optional[int]):

The maximum time to wait for the cluster to be ready in seconds. If None, waits indefinitely.

dashboard_check (bool):

Flag to determine if the dashboard readiness should be checked. Defaults to True.

Raises:

TimeoutError:

If the timeout is reached before the cluster or dashboard is ready.

codeflare_sdk.ray.cluster.cluster.get_cluster(cluster_name: str, namespace: str = 'default', verify_tls: bool = True, write_to_file: bool = False, api_client: ApiClient | None = None) → Cluster

Retrieves an existing Ray Cluster as a Cluster object.

This function fetches an existing Ray Cluster from the Kubernetes cluster and returns it as a Cluster object, including its YAML configuration under Cluster.resource_yaml.

Args:

cluster_name (str):

The name of the Ray Cluster.

namespace (str, optional):

The Kubernetes namespace where the Ray Cluster is located. Default is “default”.

verify_tls (bool, optional):

Whether to verify TLS when connecting to the cluster. Default is True.

write_to_file (bool, optional):

If True, writes the resource configuration to a YAML file. Default is False.

Returns:

Cluster:

A Cluster object representing the retrieved Ray Cluster.

Raises:

Exception:

If the Ray Cluster cannot be found or does not exist.

codeflare_sdk.ray.cluster.cluster.list_all_clusters(namespace: str, print_to_console: bool = True, api_client: ApiClient | None = None) → List[RayCluster]

Returns (and prints by default) a list of all clusters in a given namespace.

Args:

namespace: The namespace to list clusters from. print_to_console: Whether to pretty-print the result. api_client: Kubernetes API client to use. None resolves the

process-wide client, preserving legacy behaviour.

codeflare_sdk.ray.cluster.cluster.list_all_queued(namespace: str, print_to_console: bool = True, api_client: ApiClient | None = None) → List[RayCluster]

Returns (and prints by default) a list of all currently queued-up Ray Clusters in a given namespace.

A cluster is considered queued if it has an associated Kueue Workload that has not been admitted yet (workload.status.admission is None or empty).

codeflare_sdk.ray.cluster.cluster.remove_autogenerated_fields(resource)

Recursively remove autogenerated fields from a dictionary.

codeflare_sdk.ray.cluster.config module

The config sub-module contains the definition of the ClusterConfiguration dataclass, which is used to specify resource requirements and other details when creating a Cluster object. It also defines WorkerGroup for multi-worker-group clusters.

class codeflare_sdk.ray.cluster.config.ClusterConfiguration(name: str | None = None, namespace: str | None = None, head_cpu_requests: int | str = 1, head_cpu_limits: int | str = 2, head_memory_requests: int | str = 5, head_memory_limits: int | str = 8, head_extended_resource_requests: ~typing.Dict[str, str | int] = <factory>, head_tolerations: ~typing.List[~kubernetes.client.models.v1_toleration.V1Toleration] | None = None, worker_cpu_requests: int | str = 1, worker_cpu_limits: int | str = 1, num_workers: int = 1, worker_memory_requests: int | str = 3, worker_memory_limits: int | str = 6, worker_tolerations: ~typing.List[~kubernetes.client.models.v1_toleration.V1Toleration] | None = None, envs: ~typing.Dict[str, str] = <factory>, image: str = '', image_pull_secrets: ~typing.List[str] = <factory>, write_to_file: bool = False, verify_tls: bool = True, labels: ~typing.Dict[str, str] = <factory>, worker_extended_resource_requests: ~typing.Dict[str, str | int] = <factory>, extended_resource_mapping: ~typing.Dict[str, str] = <factory>, overwrite_default_resource_mapping: bool = False, local_queue: str | None = None, enable_autoscaling: bool = False, min_workers: int | None = None, max_workers: int | None = None, annotations: ~typing.Dict[str, str] = <factory>, volumes: list[~kubernetes.client.models.v1_volume.V1Volume] = <factory>, volume_mounts: list[~kubernetes.client.models.v1_volume_mount.V1VolumeMount] = <factory>, additional_worker_groups: ~typing.List[~codeflare_sdk.ray.cluster.config.WorkerGroup] = <factory>, enable_gcs_ft: bool = False, enable_usage_stats: bool = False, redis_address: str | None = None, redis_password_secret: ~typing.Dict[str, str] | None = None, external_storage_namespace: str | None = None)

Bases: object

This dataclass is used to specify resource requirements and other details, and is passed in as an argument when creating a Cluster object.

Args:

name:

The name of the cluster.

namespace:

The namespace in which the cluster should be created.

head_extended_resource_requests:

A dictionary of extended resource requests for the head node. ex: {“nvidia.com/gpu”: 1}

head_tolerations:

List of tolerations for head nodes.

num_workers:

The number of workers to create.

worker_tolerations:

List of tolerations for worker nodes.

envs:

A dictionary of environment variables to set for the cluster.

image:

The image to use for the cluster.

image_pull_secrets:

A list of image pull secrets to use for the cluster.

write_to_file:

A boolean indicating whether to write the cluster configuration to a file.

verify_tls:

A boolean indicating whether to verify TLS when connecting to the cluster.

labels:

A dictionary of labels to apply to the cluster.

worker_extended_resource_requests:

A dictionary of extended resource requests for each worker. ex: {“nvidia.com/gpu”: 1}

extended_resource_mapping:

A dictionary of custom resource mappings to map extended resource requests to RayCluster resource names

overwrite_default_resource_mapping:

A boolean indicating whether to overwrite the default resource mapping.

annotations:

A dictionary of annotations to apply to the cluster.

volumes:

A list of V1Volume objects to add to the Cluster

volume_mounts:

A list of V1VolumeMount objects to add to the Cluster

enable_autoscaling:

A boolean indicating whether to enable Ray in-tree autoscaling. When True, min_workers and max_workers must also be set. With Kueue (local_queue or a default LocalQueue), autoscaling is allowed only when Red Hat Build of Kueue (Unmanaged) is >= 1.4; RHOAI-managed Kueue still blocks autoscaling.

min_workers:

Minimum number of workers when autoscaling is enabled. Maps to workerGroupSpecs.replicas and workerGroupSpecs.minReplicas. Required when enable_autoscaling is True.

max_workers:

Maximum number of workers when autoscaling is enabled. Maps to workerGroupSpecs.maxReplicas. Required when enable_autoscaling is True.

enable_gcs_ft:

A boolean indicating whether to enable GCS fault tolerance.

enable_usage_stats:

A boolean indicating whether to capture and send Ray usage stats externally.

redis_address:

The address of the Redis server to use for GCS fault tolerance, required when enable_gcs_ft is True.

redis_password_secret:

Kubernetes secret reference containing Redis password. ex: {“name”: “secret-name”, “key”: “password-key”}

external_storage_namespace:

The storage namespace to use for GCS fault tolerance. By default, KubeRay sets it to the UID of RayCluster.

additional_worker_groups: List[WorkerGroup]

annotations: Dict[str, str]

enable_autoscaling: bool = False

enable_gcs_ft: bool = False

enable_usage_stats: bool = False

envs: Dict[str, str]

extended_resource_mapping: Dict[str, str]

external_storage_namespace: str | None = None

head_cpu_limits: int | str = 2

head_cpu_requests: int | str = 1

head_extended_resource_requests: Dict[str, str | int]

head_memory_limits: int | str = 8

head_memory_requests: int | str = 5

head_tolerations: List[V1Toleration] | None = None

image: str = ''

image_pull_secrets: List[str]

labels: Dict[str, str]

local_queue: str | None = None

max_workers: int | None = None

min_workers: int | None = None

name: str | None = None

namespace: str | None = None

num_workers: int = 1

overwrite_default_resource_mapping: bool = False

redis_address: str | None = None

redis_password_secret: Dict[str, str] | None = None

verify_tls: bool = True

volume_mounts: list[V1VolumeMount]

volumes: list[V1Volume]

worker_cpu_limits: int | str = 1

worker_cpu_requests: int | str = 1

worker_extended_resource_requests: Dict[str, str | int]

worker_memory_limits: int | str = 6

worker_memory_requests: int | str = 3

worker_tolerations: List[V1Toleration] | None = None

write_to_file: bool = False

class codeflare_sdk.ray.cluster.config.WorkerGroup(group_name: str, replicas: int = 1, min_replicas: int | None = None, max_replicas: int | None = None, cpu_requests: int | str = 1, cpu_limits: int | str = 1, memory_requests: int | str = 3, memory_limits: int | str = 6, gpu_type: str | None = None, gpu_count: int | None = None, extended_resource_requests: ~typing.Dict[str, str | int] = <factory>, image: str | None = None, envs: ~typing.Dict[str, str] = <factory>, labels: ~typing.Dict[str, str] = <factory>, tolerations: ~typing.List[~kubernetes.client.models.v1_toleration.V1Toleration] | None = None)

Bases: object

Defines an additional worker group with its own resource profile.

Used in ClusterConfiguration.additional_worker_groups to create heterogeneous clusters with multiple workerGroupSpecs in the KubeRay RayCluster CR.

Args:

group_name:

Required. Unique name for this worker group. Maps to groupName in the CRD.

replicas:

Number of worker pods in this group.

min_replicas:

Minimum replicas for autoscaling. If None, equals replicas.

max_replicas:

Maximum replicas for autoscaling. If None, equals replicas.

cpu_requests:

CPU requests per worker pod.

cpu_limits:

CPU limits per worker pod.

memory_requests:

Memory requests per worker pod. Int values treated as GB.

memory_limits:

Memory limits per worker pod. Int values treated as GB.

gpu_type:

Extended resource key, e.g. “nvidia.com/gpu”. Requires gpu_count.

gpu_count:

Number of GPUs per worker. Requires gpu_type.

extended_resource_requests:

Additional Kubernetes extended resources requested by each worker pod. GPU resources may also be represented with gpu_type and gpu_count.

image:

Container image. None inherits from ClusterConfiguration.image.

envs:

Per-group environment variables. Merged over cluster-level (group wins on conflict).

labels:

Pod labels for this worker group. Merged over cluster-level (group wins on conflict).

tolerations:

Pod tolerations. None inherits from ClusterConfiguration.worker_tolerations.

cpu_limits: int | str = 1

cpu_requests: int | str = 1

envs: Dict[str, str]

extended_resource_requests: Dict[str, str | int]

gpu_count: int | None = None

gpu_type: str | None = None

group_name: str

image: str | None = None

labels: Dict[str, str]

max_replicas: int | None = None

memory_limits: int | str = 6

memory_requests: int | str = 3

min_replicas: int | None = None

replicas: int = 1

tolerations: List[V1Toleration] | None = None

codeflare_sdk.ray.cluster.pretty_print module

This sub-module exists primarily to be used internally by the Cluster object (in the cluster sub-module) for pretty-printing cluster status and details.

codeflare_sdk.ray.cluster.pretty_print.print_cluster_status(cluster: RayCluster)

Pretty prints the status of a passed-in cluster

codeflare_sdk.ray.cluster.pretty_print.print_clusters(clusters: List[RayCluster])

codeflare_sdk.ray.cluster.pretty_print.print_no_resources_found()

codeflare_sdk.ray.cluster.pretty_print.print_ray_clusters_status(clusters: List[RayCluster], starting: bool = False)

codeflare_sdk.ray.cluster.raycluster_spec module

Canonical, side-effect-free pieces of a RayCluster spec (RHOAIENG-98942).

Both spec builders consume ClusterConfiguration and used to carry their own copy of everything below — build_ray_cluster.py for a standalone RayCluster, rayjobs/config.py for the rayClusterSpec embedded in a RayJob. The copies had already drifted, which is what RHOAIENG-98942 is for, so the shared pieces live here and both builders delegate.

Nothing in this module touches the Kubernetes API. That matters: the standalone builder resolves Kueue LocalQueues while rendering and the RayJob builder does not, so anything that needs the cluster has to stay in the caller rather than move in here.

codeflare_sdk.ray.cluster.raycluster_spec.build_head_container(config: ClusterConfiguration) → V1Container

The ray-head container.

codeflare_sdk.ray.cluster.raycluster_spec.build_resource_requirements(cpu_requests: int | str, cpu_limits: int | str, memory_requests: int | str, memory_limits: int | str, extended: Dict[str, int] | None = None) → V1ResourceRequirements

CPU/memory requests and limits, plus any extended resources.

Memory is stringified so the Kubernetes client serialises quantities such as 16 as "16G" rather than a bare number; CPU is then written back unstringified, because Ray accepts the numeric form. Both builders carried this shape (RHOAIENG-59978) before it moved here.

codeflare_sdk.ray.cluster.raycluster_spec.build_worker_container(config: ClusterConfiguration) → V1Container

The machine-learning worker container.

codeflare_sdk.ray.cluster.raycluster_spec.cpu_limit_to_num_cpus(cpu_limit: int | str) → str

Render a CPU limit as the integer string Ray’s num-cpus expects.

codeflare_sdk.ray.cluster.raycluster_spec.env_vars(config: ClusterConfiguration) → List[V1EnvVar]

User-supplied environment variables, for the head and worker alike.

codeflare_sdk.ray.cluster.raycluster_spec.extended_resources(config: ClusterConfiguration) → Tuple[dict, dict]

Non-GPU/CPU/memory resources for the head and worker rayStartParams.

Keyed by the mapped resource type, so two accelerators sharing a type are summed. __post_init__ has already rejected an unmapped key, so the fallback to the raw key below never fires in practice.

codeflare_sdk.ray.cluster.raycluster_spec.format_resources_param(resources: dict) → str

Serialise a resources dict the way rayStartParams expects it.

codeflare_sdk.ray.cluster.raycluster_spec.gcs_fault_tolerance_options(config: ClusterConfiguration) → dict | None

gcsFaultToleranceOptions, or None when GCS FT is off.

Shared so the RayJob path stops dropping it (RHOAIENG-98943).

codeflare_sdk.ray.cluster.raycluster_spec.gpu_counts(config: ClusterConfiguration) → Tuple[int, int]

Total GPUs requested for the head and for a worker, respectively.

codeflare_sdk.ray.cluster.raycluster_spec.image_pull_secrets(config: ClusterConfiguration) → List[V1LocalObjectReference]

Secret references for pulling the Ray image.

codeflare_sdk.ray.cluster.raycluster_spec.merge_storage(provided: list, defaults: list) → list

Append the default volumes/mounts to whatever the user supplied.

codeflare_sdk.ray.cluster.raycluster_spec.worker_replica_counts(config: ClusterConfiguration) → Tuple[int, int, int]

(replicas, minReplicas, maxReplicas) for the default worker group.

Autoscaling hands the range to the autoscaler; otherwise all three are num_workers.

codeflare_sdk.ray.cluster.status module

The status sub-module defines Enums containing information for Ray cluster states states, and CodeFlare cluster states, as well as dataclasses to store information for Ray clusters.

class codeflare_sdk.ray.cluster.status.CodeFlareClusterStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Defines the possible reportable states of a Codeflare cluster.

FAILED = 5

QUEUED = 3

QUEUEING = 4

READY = 1

STARTING = 2

SUSPENDED = 7

UNKNOWN = 6

class codeflare_sdk.ray.cluster.status.RayCluster(name: str, status: ~codeflare_sdk.ray.cluster.status.RayClusterStatus, head_cpu_requests: int, head_cpu_limits: int, head_mem_requests: str, head_mem_limits: str, num_workers: int, worker_mem_requests: str, worker_mem_limits: str, worker_cpu_requests: int | str, worker_cpu_limits: int | str, namespace: str, dashboard: str, worker_extended_resources: ~typing.Dict[str, int] = <factory>, head_extended_resources: ~typing.Dict[str, int] = <factory>, worker_groups: ~typing.List[~codeflare_sdk.ray.cluster.status.WorkerGroupStatus] = <factory>)

Bases: object

For storing information about a Ray cluster.

dashboard: str

head_cpu_limits: int

head_cpu_requests: int

head_extended_resources: Dict[str, int]

head_mem_limits: str

head_mem_requests: str

name: str

namespace: str

num_workers: int

status: RayClusterStatus

worker_cpu_limits: int | str

worker_cpu_requests: int | str

worker_extended_resources: Dict[str, int]

worker_groups: List[WorkerGroupStatus]

worker_mem_limits: str

worker_mem_requests: str

class codeflare_sdk.ray.cluster.status.RayClusterStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Defines the possible reportable states of a Ray cluster.

FAILED = 'failed'

READY = 'ready'

SUSPENDED = 'suspended'

UNHEALTHY = 'unhealthy'

UNKNOWN = 'unknown'

class codeflare_sdk.ray.cluster.status.WorkerGroupStatus(group_name: str, replicas: int, min_replicas: int | None, max_replicas: int | None, cpu_requests: int | str, cpu_limits: int | str, memory_requests: int | str, memory_limits: int | str, extended_resource_limits: ~typing.Dict[str, str | int] = <factory>)

Bases: object

Observed state for a worker group returned by KubeRay.

cpu_limits: int | str

cpu_requests: int | str

extended_resource_limits: Dict[str, str | int]

group_name: str

max_replicas: int | None

memory_limits: int | str

memory_requests: int | str

min_replicas: int | None

replicas: int

Module contents