docs/cluster/cluster.html

<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1, minimum-scale=1" />
<meta name="generator" content="pdoc 0.10.0" />
<title>codeflare_sdk.cluster.cluster API documentation</title>
<meta name="description" content="The cluster sub-module contains the definition of the Cluster object, which represents
the resources requested by the user. It also contains functions …" />
<link rel="preload stylesheet" as="style" href="https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/11.0.1/sanitize.min.css" integrity="sha256-PK9q560IAAa6WVRRh76LtCaI8pjTJ2z11v0miyNNjrs=" crossorigin>
<link rel="preload stylesheet" as="style" href="https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/11.0.1/typography.min.css" integrity="sha256-7l/o7C8jubJiy74VsKTidCy1yBkRtiUGbVkYBylBqUg=" crossorigin>
<link rel="stylesheet preload" as="style" href="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/10.1.1/styles/github.min.css" crossorigin>
<style>:root{--highlight-color:#fe9}.flex{display:flex !important}body{line-height:1.5em}#content{padding:20px}#sidebar{padding:30px;overflow:hidden}#sidebar > *:last-child{margin-bottom:2cm}.http-server-breadcrumbs{font-size:130%;margin:0 0 15px 0}#footer{font-size:.75em;padding:5px 30px;border-top:1px solid #ddd;text-align:right}#footer p{margin:0 0 0 1em;display:inline-block}#footer p:last-child{margin-right:30px}h1,h2,h3,h4,h5{font-weight:300}h1{font-size:2.5em;line-height:1.1em}h2{font-size:1.75em;margin:1em 0 .50em 0}h3{font-size:1.4em;margin:25px 0 10px 0}h4{margin:0;font-size:105%}h1:target,h2:target,h3:target,h4:target,h5:target,h6:target{background:var(--highlight-color);padding:.2em 0}a{color:#058;text-decoration:none;transition:color .3s ease-in-out}a:hover{color:#e82}.title code{font-weight:bold}h2[id^="header-"]{margin-top:2em}.ident{color:#900}pre code{background:#f8f8f8;font-size:.8em;line-height:1.4em}code{background:#f2f2f1;padding:1px 4px;overflow-wrap:break-word}h1 code{background:transparent}pre{background:#f8f8f8;border:0;border-top:1px solid #ccc;border-bottom:1px solid #ccc;margin:1em 0;padding:1ex}#http-server-module-list{display:flex;flex-flow:column}#http-server-module-list div{display:flex}#http-server-module-list dt{min-width:10%}#http-server-module-list p{margin-top:0}.toc ul,#index{list-style-type:none;margin:0;padding:0}#index code{background:transparent}#index h3{border-bottom:1px solid #ddd}#index ul{padding:0}#index h4{margin-top:.6em;font-weight:bold}@media (min-width:200ex){#index .two-column{column-count:2}}@media (min-width:300ex){#index .two-column{column-count:3}}dl{margin-bottom:2em}dl dl:last-child{margin-bottom:4em}dd{margin:0 0 1em 3em}#header-classes + dl > dd{margin-bottom:3em}dd dd{margin-left:2em}dd p{margin:10px 0}.name{background:#eee;font-weight:bold;font-size:.85em;padding:5px 10px;display:inline-block;min-width:40%}.name:hover{background:#e0e0e0}dt:target .name{background:var(--highlight-color)}.name > span:first-child{white-space:nowrap}.name.class > span:nth-child(2){margin-left:.4em}.inherited{color:#999;border-left:5px solid #eee;padding-left:1em}.inheritance em{font-style:normal;font-weight:bold}.desc h2{font-weight:400;font-size:1.25em}.desc h3{font-size:1em}.desc dt code{background:inherit}.source summary,.git-link-div{color:#666;text-align:right;font-weight:400;font-size:.8em;text-transform:uppercase}.source summary > *{white-space:nowrap;cursor:pointer}.git-link{color:inherit;margin-left:1em}.source pre{max-height:500px;overflow:auto;margin:0}.source pre code{font-size:12px;overflow:visible}.hlist{list-style:none}.hlist li{display:inline}.hlist li:after{content:',\2002'}.hlist li:last-child:after{content:none}.hlist .hlist{display:inline;padding-left:1em}img{max-width:100%}td{padding:0 .5em}.admonition{padding:.1em .5em;margin-bottom:1em}.admonition-title{font-weight:bold}.admonition.note,.admonition.info,.admonition.important{background:#aef}.admonition.todo,.admonition.versionadded,.admonition.tip,.admonition.hint{background:#dfd}.admonition.warning,.admonition.versionchanged,.admonition.deprecated{background:#fd4}.admonition.error,.admonition.danger,.admonition.caution{background:lightpink}</style>
<style media="screen and (min-width: 700px)">@media screen and (min-width:700px){#sidebar{width:30%;height:100vh;overflow:auto;position:sticky;top:0}#content{width:70%;max-width:100ch;padding:3em 4em;border-left:1px solid #ddd}pre code{font-size:1em}.item .name{font-size:1em}main{display:flex;flex-direction:row-reverse;justify-content:flex-end}.toc ul ul,#index ul{padding-left:1.5em}.toc > ul > li{margin-top:.5em}}</style>
<style media="print">@media print{#sidebar h1{page-break-before:always}.source{display:none}}@media print{*{background:transparent !important;color:#000 !important;box-shadow:none !important;text-shadow:none !important}a[href]:after{content:" (" attr(href) ")";font-size:90%}a[href][title]:after{content:none}abbr[title]:after{content:" (" attr(title) ")"}.ir a:after,a[href^="javascript:"]:after,a[href^="#"]:after{content:""}pre,blockquote{border:1px solid #999;page-break-inside:avoid}thead{display:table-header-group}tr,img{page-break-inside:avoid}img{max-width:100% !important}@page{margin:0.5cm}p,h2,h3{orphans:3;widows:3}h1,h2,h3,h4,h5,h6{page-break-after:avoid}}</style>
<script defer src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/10.1.1/highlight.min.js" integrity="sha256-Uv3H6lx7dJmRfRvH8TH6kJD1TSK1aFcwgx+mdg3epi8=" crossorigin></script>
<script>window.addEventListener('DOMContentLoaded', () => hljs.initHighlighting())</script>
</head>
<body>
<main>
<article id="content">
<header>
<h1 class="title">Module <code>codeflare_sdk.cluster.cluster</code></h1>
</header>
<section id="section-intro">
<p>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.</p>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python"># Copyright 2022 IBM, Red Hat
#
# Licensed under the Apache License, Version 2.0 (the &#34;License&#34;);
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an &#34;AS IS&#34; BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

&#34;&#34;&#34;
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&#39;s working namespace.
&#34;&#34;&#34;

from time import sleep
from typing import List, Optional, Tuple, Dict

import openshift as oc
from kubernetes import config
from ray.job_submission import JobSubmissionClient
import urllib3

from .auth import config_check, api_config_handler
from ..utils import pretty_print
from ..utils.generate_yaml import (
    generate_appwrapper,
)
from ..utils.kube_api_helpers import _kube_api_error_handling
from ..utils.openshift_oauth import (
    create_openshift_oauth_objects,
    delete_openshift_oauth_objects,
)
from .config import ClusterConfiguration
from .model import (
    AppWrapper,
    AppWrapperStatus,
    CodeFlareClusterStatus,
    RayCluster,
    RayClusterStatus,
)
from kubernetes import client, config
import yaml
import os
import requests

from kubernetes import config


class Cluster:
    &#34;&#34;&#34;
    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.
    &#34;&#34;&#34;

    torchx_scheduler = &#34;ray&#34;

    def __init__(self, config: ClusterConfiguration):
        &#34;&#34;&#34;
        Create the resource cluster object by passing in a ClusterConfiguration
        (defined in the config sub-module). An AppWrapper will then be generated
        based off of the configured resources to represent the desired cluster
        request.
        &#34;&#34;&#34;
        self.config = config
        self.app_wrapper_yaml = self.create_app_wrapper()
        self._job_submission_client = None
        self.app_wrapper_name = self.app_wrapper_yaml.replace(&#34;.yaml&#34;, &#34;&#34;).split(&#34;/&#34;)[
            -1
        ]

    @property
    def _client_headers(self):
        k8_client = api_config_handler() or client.ApiClient()
        return {
            &#34;Authorization&#34;: k8_client.configuration.get_api_key_with_prefix(
                &#34;authorization&#34;
            )
        }

    @property
    def _client_verify_tls(self):
        return not self.config.openshift_oauth

    @property
    def job_client(self):
        if self._job_submission_client:
            return self._job_submission_client
        if self.config.openshift_oauth:
            print(
                api_config_handler().configuration.get_api_key_with_prefix(
                    &#34;authorization&#34;
                )
            )
            self._job_submission_client = JobSubmissionClient(
                self.cluster_dashboard_uri(),
                headers=self._client_headers,
                verify=self._client_verify_tls,
            )
        else:
            self._job_submission_client = JobSubmissionClient(
                self.cluster_dashboard_uri()
            )
        return self._job_submission_client

    def evaluate_dispatch_priority(self):
        priority_class = self.config.dispatch_priority

        try:
            config_check()
            api_instance = client.CustomObjectsApi(api_config_handler())
            priority_classes = api_instance.list_cluster_custom_object(
                group=&#34;scheduling.k8s.io&#34;,
                version=&#34;v1&#34;,
                plural=&#34;priorityclasses&#34;,
            )
        except Exception as e:  # pragma: no cover
            return _kube_api_error_handling(e)

        for pc in priority_classes[&#34;items&#34;]:
            if pc[&#34;metadata&#34;][&#34;name&#34;] == priority_class:
                return pc[&#34;value&#34;]
        print(f&#34;Priority class {priority_class} is not available in the cluster&#34;)
        return None

    def validate_image_config(self):
        &#34;&#34;&#34;
        Validates that the image configuration is not empty.

        :param image: The image string to validate
        :raises ValueError: If the image is not specified
        &#34;&#34;&#34;
        if self.config.image == &#34;&#34; or self.config.image == None:
            raise ValueError(&#34;Image must be specified in the ClusterConfiguration&#34;)

    def create_app_wrapper(self):
        &#34;&#34;&#34;
        Called upon cluster object creation, creates an AppWrapper yaml based on
        the specifications of the ClusterConfiguration.
        &#34;&#34;&#34;

        if self.config.namespace is None:
            self.config.namespace = get_current_namespace()
            if self.config.namespace is None:
                print(&#34;Please specify with namespace=&lt;your_current_namespace&gt;&#34;)
            elif type(self.config.namespace) is not str:
                raise TypeError(
                    f&#34;Namespace {self.config.namespace} is of type {type(self.config.namespace)}. Check your Kubernetes Authentication.&#34;
                )

        # Validate image configuration
        self.validate_image_config()

        # Before attempting to create the cluster AW, let&#39;s evaluate the ClusterConfig
        if self.config.dispatch_priority:
            if not self.config.mcad:
                raise ValueError(
                    &#34;Invalid Cluster Configuration, cannot have dispatch priority without MCAD&#34;
                )
            priority_val = self.evaluate_dispatch_priority()
            if priority_val == None:
                raise ValueError(
                    &#34;Invalid Cluster Configuration, AppWrapper not generated&#34;
                )
        else:
            priority_val = None

        name = self.config.name
        namespace = self.config.namespace
        head_cpus = self.config.head_cpus
        head_memory = self.config.head_memory
        head_gpus = self.config.head_gpus
        min_cpu = self.config.min_cpus
        max_cpu = self.config.max_cpus
        min_memory = self.config.min_memory
        max_memory = self.config.max_memory
        gpu = self.config.num_gpus
        workers = self.config.num_workers
        template = self.config.template
        image = self.config.image
        instascale = self.config.instascale
        mcad = self.config.mcad
        instance_types = self.config.machine_types
        env = self.config.envs
        local_interactive = self.config.local_interactive
        image_pull_secrets = self.config.image_pull_secrets
        dispatch_priority = self.config.dispatch_priority
        ingress_domain = self.config.ingress_domain
        ingress_options = self.config.ingress_options
        return generate_appwrapper(
            name=name,
            namespace=namespace,
            head_cpus=head_cpus,
            head_memory=head_memory,
            head_gpus=head_gpus,
            min_cpu=min_cpu,
            max_cpu=max_cpu,
            min_memory=min_memory,
            max_memory=max_memory,
            gpu=gpu,
            workers=workers,
            template=template,
            image=image,
            instascale=instascale,
            mcad=mcad,
            instance_types=instance_types,
            env=env,
            local_interactive=local_interactive,
            image_pull_secrets=image_pull_secrets,
            dispatch_priority=dispatch_priority,
            priority_val=priority_val,
            openshift_oauth=self.config.openshift_oauth,
            ingress_domain=ingress_domain,
            ingress_options=ingress_options,
        )

    # creates a new cluster with the provided or default spec
    def up(self):
        &#34;&#34;&#34;
        Applies the AppWrapper yaml, pushing the resource request onto
        the MCAD queue.
        &#34;&#34;&#34;
        namespace = self.config.namespace
        if self.config.openshift_oauth:
            create_openshift_oauth_objects(
                cluster_name=self.config.name, namespace=namespace
            )

        try:
            config_check()
            api_instance = client.CustomObjectsApi(api_config_handler())
            if self.config.mcad:
                with open(self.app_wrapper_yaml) as f:
                    aw = yaml.load(f, Loader=yaml.FullLoader)
                api_instance.create_namespaced_custom_object(
                    group=&#34;workload.codeflare.dev&#34;,
                    version=&#34;v1beta1&#34;,
                    namespace=namespace,
                    plural=&#34;appwrappers&#34;,
                    body=aw,
                )
            else:
                self._component_resources_up(namespace, api_instance)
        except Exception as e:  # pragma: no cover
            return _kube_api_error_handling(e)

    def down(self):
        &#34;&#34;&#34;
        Deletes the AppWrapper yaml, scaling-down and deleting all resources
        associated with the cluster.
        &#34;&#34;&#34;
        namespace = self.config.namespace
        try:
            config_check()
            api_instance = client.CustomObjectsApi(api_config_handler())
            if self.config.mcad:
                api_instance.delete_namespaced_custom_object(
                    group=&#34;workload.codeflare.dev&#34;,
                    version=&#34;v1beta1&#34;,
                    namespace=namespace,
                    plural=&#34;appwrappers&#34;,
                    name=self.app_wrapper_name,
                )
            else:
                self._component_resources_down(namespace, api_instance)
        except Exception as e:  # pragma: no cover
            return _kube_api_error_handling(e)

        if self.config.openshift_oauth:
            delete_openshift_oauth_objects(
                cluster_name=self.config.name, namespace=namespace
            )

    def status(
        self, print_to_console: bool = True
    ) -&gt; Tuple[CodeFlareClusterStatus, bool]:
        &#34;&#34;&#34;
        Returns the requested cluster&#39;s status, as well as whether or not
        it is ready for use.
        &#34;&#34;&#34;
        ready = False
        status = CodeFlareClusterStatus.UNKNOWN
        if self.config.mcad:
            # check the app wrapper status
            appwrapper = _app_wrapper_status(self.config.name, self.config.namespace)
            if appwrapper:
                if appwrapper.status in [
                    AppWrapperStatus.RUNNING,
                    AppWrapperStatus.COMPLETED,
                    AppWrapperStatus.RUNNING_HOLD_COMPLETION,
                ]:
                    ready = False
                    status = CodeFlareClusterStatus.STARTING
                elif appwrapper.status in [
                    AppWrapperStatus.FAILED,
                    AppWrapperStatus.DELETED,
                ]:
                    ready = False
                    status = CodeFlareClusterStatus.FAILED  # should deleted be separate
                    return status, ready  # exit early, no need to check ray status
                elif appwrapper.status in [
                    AppWrapperStatus.PENDING,
                    AppWrapperStatus.QUEUEING,
                ]:
                    ready = False
                    if appwrapper.status == AppWrapperStatus.PENDING:
                        status = CodeFlareClusterStatus.QUEUED
                    else:
                        status = CodeFlareClusterStatus.QUEUEING
                    if print_to_console:
                        pretty_print.print_app_wrappers_status([appwrapper])
                    return (
                        status,
                        ready,
                    )  # no need to check the ray status since still in queue

        # check the ray cluster status
        cluster = _ray_cluster_status(self.config.name, self.config.namespace)
        if cluster:
            if cluster.status == RayClusterStatus.UNKNOWN:
                ready = False
                status = CodeFlareClusterStatus.STARTING
            if cluster.status == RayClusterStatus.READY:
                ready = True
                status = CodeFlareClusterStatus.READY
            elif cluster.status in [
                RayClusterStatus.UNHEALTHY,
                RayClusterStatus.FAILED,
            ]:
                ready = False
                status = CodeFlareClusterStatus.FAILED

            if print_to_console:
                # overriding the number of gpus with requested
                cluster.worker_gpu = self.config.num_gpus
                pretty_print.print_cluster_status(cluster)
        elif print_to_console:
            if status == CodeFlareClusterStatus.UNKNOWN:
                pretty_print.print_no_resources_found()
            else:
                pretty_print.print_app_wrappers_status([appwrapper], starting=True)

        return status, ready

    def is_dashboard_ready(self) -&gt; bool:
        try:
            response = requests.get(
                self.cluster_dashboard_uri(),
                headers=self._client_headers,
                timeout=5,
                verify=self._client_verify_tls,
            )
        except requests.exceptions.SSLError:  # pragma no cover
            # SSL exception occurs when oauth ingress has been created but cluster is not up
            return False
        if response.status_code == 200:
            return True
        else:
            return False

    def wait_ready(self, timeout: Optional[int] = None, dashboard_check: bool = True):
        &#34;&#34;&#34;
        Waits for requested cluster to be ready, up to an optional timeout (s).
        Checks every five seconds.
        &#34;&#34;&#34;
        print(&#34;Waiting for requested resources to be set up...&#34;)
        ready = False
        dashboard_ready = False
        status = None
        time = 0
        while not ready:
            status, ready = self.status(print_to_console=False)
            if status == CodeFlareClusterStatus.UNKNOWN:
                print(
                    &#34;WARNING: Current cluster status is unknown, have you run cluster.up yet?&#34;
                )
            if not ready:
                if timeout and time &gt;= timeout:
                    raise TimeoutError(
                        f&#34;wait() timed out after waiting {timeout}s for cluster to be ready&#34;
                    )
                sleep(5)
                time += 5
        print(&#34;Requested cluster is up and running!&#34;)

        while dashboard_check and not dashboard_ready:
            dashboard_ready = self.is_dashboard_ready()
            if not dashboard_ready:
                if timeout and time &gt;= timeout:
                    raise TimeoutError(
                        f&#34;wait() timed out after waiting {timeout}s for dashboard to be ready&#34;
                    )
                sleep(5)
                time += 5
        if dashboard_ready:
            print(&#34;Dashboard is ready!&#34;)

    def details(self, print_to_console: bool = True) -&gt; RayCluster:
        cluster = _copy_to_ray(self)
        if print_to_console:
            pretty_print.print_clusters([cluster])
        return cluster

    def cluster_uri(self) -&gt; str:
        &#34;&#34;&#34;
        Returns a string containing the cluster&#39;s URI.
        &#34;&#34;&#34;
        return f&#34;ray://{self.config.name}-head-svc.{self.config.namespace}.svc:10001&#34;

    def cluster_dashboard_uri(self) -&gt; str:
        &#34;&#34;&#34;
        Returns a string containing the cluster&#39;s dashboard URI.
        &#34;&#34;&#34;
        try:
            config_check()
            api_instance = client.NetworkingV1Api(api_config_handler())
            ingresses = api_instance.list_namespaced_ingress(self.config.namespace)
        except Exception as e:  # pragma no cover
            return _kube_api_error_handling(e)

        for ingress in ingresses.items:
            annotations = ingress.metadata.annotations
            protocol = &#34;http&#34;
            if (
                ingress.metadata.name == f&#34;ray-dashboard-{self.config.name}&#34;
                or ingress.metadata.name.startswith(f&#34;{self.config.name}-ingress&#34;)
            ):
                if annotations == None:
                    protocol = &#34;http&#34;
                elif &#34;route.openshift.io/termination&#34; in annotations:
                    protocol = &#34;https&#34;
            return f&#34;{protocol}://{ingress.spec.rules[0].host}&#34;
        return &#34;Dashboard ingress not available yet, have you run cluster.up()?&#34;

    def list_jobs(self) -&gt; List:
        &#34;&#34;&#34;
        This method accesses the head ray node in your cluster and lists the running jobs.
        &#34;&#34;&#34;
        return self.job_client.list_jobs()

    def job_status(self, job_id: str) -&gt; str:
        &#34;&#34;&#34;
        This method accesses the head ray node in your cluster and returns the job status for the provided job id.
        &#34;&#34;&#34;
        return self.job_client.get_job_status(job_id)

    def job_logs(self, job_id: str) -&gt; str:
        &#34;&#34;&#34;
        This method accesses the head ray node in your cluster and returns the logs for the provided job id.
        &#34;&#34;&#34;
        return self.job_client.get_job_logs(job_id)

    def torchx_config(
        self, working_dir: str = None, requirements: str = None
    ) -&gt; Dict[str, str]:
        dashboard_address = urllib3.util.parse_url(self.cluster_dashboard_uri()).host
        to_return = {
            &#34;cluster_name&#34;: self.config.name,
            &#34;dashboard_address&#34;: dashboard_address,
        }
        if working_dir:
            to_return[&#34;working_dir&#34;] = working_dir
        if requirements:
            to_return[&#34;requirements&#34;] = requirements
        return to_return

    def from_k8_cluster_object(rc, mcad=True):
        machine_types = (
            rc[&#34;metadata&#34;][&#34;labels&#34;][&#34;orderedinstance&#34;].split(&#34;_&#34;)
            if &#34;orderedinstance&#34; in rc[&#34;metadata&#34;][&#34;labels&#34;]
            else []
        )
        local_interactive = (
            &#34;volumeMounts&#34;
            in rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0]
        )
        cluster_config = ClusterConfiguration(
            name=rc[&#34;metadata&#34;][&#34;name&#34;],
            namespace=rc[&#34;metadata&#34;][&#34;namespace&#34;],
            machine_types=machine_types,
            num_workers=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;minReplicas&#34;],
            min_cpus=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][
                &#34;containers&#34;
            ][0][&#34;resources&#34;][&#34;requests&#34;][&#34;cpu&#34;],
            max_cpus=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][
                &#34;containers&#34;
            ][0][&#34;resources&#34;][&#34;limits&#34;][&#34;cpu&#34;],
            min_memory=int(
                rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0][
                    &#34;resources&#34;
                ][&#34;requests&#34;][&#34;memory&#34;][:-1]
            ),
            max_memory=int(
                rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0][
                    &#34;resources&#34;
                ][&#34;limits&#34;][&#34;memory&#34;][:-1]
            ),
            num_gpus=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][
                &#34;containers&#34;
            ][0][&#34;resources&#34;][&#34;limits&#34;][&#34;nvidia.com/gpu&#34;],
            instascale=True if machine_types else False,
            image=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][
                0
            ][&#34;image&#34;],
            local_interactive=local_interactive,
            mcad=mcad,
        )
        return Cluster(cluster_config)

    def local_client_url(self):
        if self.config.local_interactive == True:
            ingress_domain = _get_ingress_domain(self)
            return f&#34;ray://{ingress_domain}&#34;
        else:
            return &#34;None&#34;

    def _component_resources_up(
        self, namespace: str, api_instance: client.CustomObjectsApi
    ):
        with open(self.app_wrapper_yaml) as f:
            yamls = yaml.load_all(f, Loader=yaml.FullLoader)
            for resource in yamls:
                if resource[&#34;kind&#34;] == &#34;RayCluster&#34;:
                    api_instance.create_namespaced_custom_object(
                        group=&#34;ray.io&#34;,
                        version=&#34;v1alpha1&#34;,
                        namespace=namespace,
                        plural=&#34;rayclusters&#34;,
                        body=resource,
                    )
                elif resource[&#34;kind&#34;] == &#34;Route&#34;:
                    api_instance.create_namespaced_custom_object(
                        group=&#34;route.openshift.io&#34;,
                        version=&#34;v1&#34;,
                        namespace=namespace,
                        plural=&#34;routes&#34;,
                        body=resource,
                    )
                elif resource[&#34;kind&#34;] == &#34;Secret&#34;:
                    secret_instance = client.CoreV1Api(api_config_handler())
                    secret_instance.create_namespaced_secret(
                        namespace=namespace,
                        body=resource,
                    )

    def _component_resources_down(
        self, namespace: str, api_instance: client.CustomObjectsApi
    ):
        with open(self.app_wrapper_yaml) as f:
            yamls = yaml.load_all(f, Loader=yaml.FullLoader)
            for resource in yamls:
                if resource[&#34;kind&#34;] == &#34;RayCluster&#34;:
                    api_instance.delete_namespaced_custom_object(
                        group=&#34;ray.io&#34;,
                        version=&#34;v1alpha1&#34;,
                        namespace=namespace,
                        plural=&#34;rayclusters&#34;,
                        name=self.app_wrapper_name,
                    )
                elif resource[&#34;kind&#34;] == &#34;Route&#34;:
                    name = resource[&#34;metadata&#34;][&#34;name&#34;]
                    api_instance.delete_namespaced_custom_object(
                        group=&#34;route.openshift.io&#34;,
                        version=&#34;v1&#34;,
                        namespace=namespace,
                        plural=&#34;routes&#34;,
                        name=name,
                    )
                elif resource[&#34;kind&#34;] == &#34;Secret&#34;:
                    name = resource[&#34;metadata&#34;][&#34;name&#34;]
                    secret_instance = client.CoreV1Api(api_config_handler())
                    secret_instance.delete_namespaced_secret(
                        namespace=namespace,
                        name=name,
                    )


def list_all_clusters(namespace: str, print_to_console: bool = True):
    &#34;&#34;&#34;
    Returns (and prints by default) a list of all clusters in a given namespace.
    &#34;&#34;&#34;
    clusters = _get_ray_clusters(namespace)
    if print_to_console:
        pretty_print.print_clusters(clusters)
    return clusters


def list_all_queued(namespace: str, print_to_console: bool = True):
    &#34;&#34;&#34;
    Returns (and prints by default) a list of all currently queued-up AppWrappers
    in a given namespace.
    &#34;&#34;&#34;
    app_wrappers = _get_app_wrappers(
        namespace, filter=[AppWrapperStatus.RUNNING, AppWrapperStatus.PENDING]
    )
    if print_to_console:
        pretty_print.print_app_wrappers_status(app_wrappers)
    return app_wrappers


def get_current_namespace():  # pragma: no cover
    if api_config_handler() != None:
        if os.path.isfile(&#34;/var/run/secrets/kubernetes.io/serviceaccount/namespace&#34;):
            try:
                file = open(
                    &#34;/var/run/secrets/kubernetes.io/serviceaccount/namespace&#34;, &#34;r&#34;
                )
                active_context = file.readline().strip(&#34;\n&#34;)
                return active_context
            except Exception as e:
                print(&#34;Unable to find current namespace&#34;)
                return None
        else:
            print(&#34;Unable to find current namespace&#34;)
            return None
    else:
        try:
            _, active_context = config.list_kube_config_contexts(config_check())
        except Exception as e:
            return _kube_api_error_handling(e)
        try:
            return active_context[&#34;context&#34;][&#34;namespace&#34;]
        except KeyError:
            return None


def get_cluster(cluster_name: str, namespace: str = &#34;default&#34;):
    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        rcs = api_instance.list_namespaced_custom_object(
            group=&#34;ray.io&#34;,
            version=&#34;v1alpha1&#34;,
            namespace=namespace,
            plural=&#34;rayclusters&#34;,
        )
    except Exception as e:
        return _kube_api_error_handling(e)

    for rc in rcs[&#34;items&#34;]:
        if rc[&#34;metadata&#34;][&#34;name&#34;] == cluster_name:
            mcad = _check_aw_exists(cluster_name, namespace)
            return Cluster.from_k8_cluster_object(rc, mcad=mcad)
    raise FileNotFoundError(
        f&#34;Cluster {cluster_name} is not found in {namespace} namespace&#34;
    )


# private methods
def _check_aw_exists(name: str, namespace: str) -&gt; bool:
    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        aws = api_instance.list_namespaced_custom_object(
            group=&#34;workload.codeflare.dev&#34;,
            version=&#34;v1beta1&#34;,
            namespace=namespace,
            plural=&#34;appwrappers&#34;,
        )
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e, print_error=False)

    for aw in aws[&#34;items&#34;]:
        if aw[&#34;metadata&#34;][&#34;name&#34;] == name:
            return True
    return False


# Cant test this until get_current_namespace is fixed
def _get_ingress_domain(self):  # pragma: no cover
    try:
        config_check()
        api_client = client.NetworkingV1Api(api_config_handler())
        if self.config.namespace != None:
            namespace = self.config.namespace
        else:
            namespace = get_current_namespace()
        ingresses = api_client.list_namespaced_ingress(namespace)
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)
    domain = None
    for ingress in ingresses.items:
        if ingress.spec.rules[0].http.paths[0].backend.service.port.number == 10001:
            domain = ingress.spec.rules[0].host
    return domain


def _app_wrapper_status(name, namespace=&#34;default&#34;) -&gt; Optional[AppWrapper]:
    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        aws = api_instance.list_namespaced_custom_object(
            group=&#34;workload.codeflare.dev&#34;,
            version=&#34;v1beta1&#34;,
            namespace=namespace,
            plural=&#34;appwrappers&#34;,
        )
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)

    for aw in aws[&#34;items&#34;]:
        if aw[&#34;metadata&#34;][&#34;name&#34;] == name:
            return _map_to_app_wrapper(aw)
    return None


def _ray_cluster_status(name, namespace=&#34;default&#34;) -&gt; Optional[RayCluster]:
    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        rcs = api_instance.list_namespaced_custom_object(
            group=&#34;ray.io&#34;,
            version=&#34;v1alpha1&#34;,
            namespace=namespace,
            plural=&#34;rayclusters&#34;,
        )
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)

    for rc in rcs[&#34;items&#34;]:
        if rc[&#34;metadata&#34;][&#34;name&#34;] == name:
            return _map_to_ray_cluster(rc)
    return None


def _get_ray_clusters(namespace=&#34;default&#34;) -&gt; List[RayCluster]:
    list_of_clusters = []
    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        rcs = api_instance.list_namespaced_custom_object(
            group=&#34;ray.io&#34;,
            version=&#34;v1alpha1&#34;,
            namespace=namespace,
            plural=&#34;rayclusters&#34;,
        )
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)

    for rc in rcs[&#34;items&#34;]:
        list_of_clusters.append(_map_to_ray_cluster(rc))
    return list_of_clusters


def _get_app_wrappers(
    namespace=&#34;default&#34;, filter=List[AppWrapperStatus]
) -&gt; List[AppWrapper]:
    list_of_app_wrappers = []

    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        aws = api_instance.list_namespaced_custom_object(
            group=&#34;workload.codeflare.dev&#34;,
            version=&#34;v1beta1&#34;,
            namespace=namespace,
            plural=&#34;appwrappers&#34;,
        )
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)

    for item in aws[&#34;items&#34;]:
        app_wrapper = _map_to_app_wrapper(item)
        if filter and app_wrapper.status in filter:
            list_of_app_wrappers.append(app_wrapper)
        else:
            # Unsure what the purpose of the filter is
            list_of_app_wrappers.append(app_wrapper)
    return list_of_app_wrappers


def _map_to_ray_cluster(rc) -&gt; Optional[RayCluster]:
    if &#34;state&#34; in rc[&#34;status&#34;]:
        status = RayClusterStatus(rc[&#34;status&#34;][&#34;state&#34;].lower())
    else:
        status = RayClusterStatus.UNKNOWN
    try:
        config_check()
        api_instance = client.NetworkingV1Api(api_config_handler())
        ingresses = api_instance.list_namespaced_ingress(rc[&#34;metadata&#34;][&#34;namespace&#34;])
    except Exception as e:  # pragma no cover
        return _kube_api_error_handling(e)
    ray_ingress = None
    for ingress in ingresses.items:
        annotations = ingress.metadata.annotations
        protocol = &#34;http&#34;
        if (
            ingress.metadata.name == f&#34;ray-dashboard-{rc[&#39;metadata&#39;][&#39;name&#39;]}&#34;
            or ingress.metadata.name.startswith(f&#34;{rc[&#39;metadata&#39;][&#39;name&#39;]}-ingress&#34;)
        ):
            if annotations == None:
                protocol = &#34;http&#34;
            elif &#34;route.openshift.io/termination&#34; in annotations:
                protocol = &#34;https&#34;
        ray_ingress = f&#34;{protocol}://{ingress.spec.rules[0].host}&#34;

    return RayCluster(
        name=rc[&#34;metadata&#34;][&#34;name&#34;],
        status=status,
        # for now we are not using autoscaling so same replicas is fine
        workers=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;replicas&#34;],
        worker_mem_max=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][
            &#34;containers&#34;
        ][0][&#34;resources&#34;][&#34;limits&#34;][&#34;memory&#34;],
        worker_mem_min=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][
            &#34;containers&#34;
        ][0][&#34;resources&#34;][&#34;requests&#34;][&#34;memory&#34;],
        worker_cpu=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][
            0
        ][&#34;resources&#34;][&#34;limits&#34;][&#34;cpu&#34;],
        worker_gpu=0,  # hard to detect currently how many gpus, can override it with what the user asked for
        namespace=rc[&#34;metadata&#34;][&#34;namespace&#34;],
        head_cpus=rc[&#34;spec&#34;][&#34;headGroupSpec&#34;][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0][
            &#34;resources&#34;
        ][&#34;limits&#34;][&#34;cpu&#34;],
        head_mem=rc[&#34;spec&#34;][&#34;headGroupSpec&#34;][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0][
            &#34;resources&#34;
        ][&#34;limits&#34;][&#34;memory&#34;],
        head_gpu=rc[&#34;spec&#34;][&#34;headGroupSpec&#34;][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0][
            &#34;resources&#34;
        ][&#34;limits&#34;][&#34;nvidia.com/gpu&#34;],
        dashboard=ray_ingress,
    )


def _map_to_app_wrapper(aw) -&gt; AppWrapper:
    if &#34;status&#34; in aw and &#34;canrun&#34; in aw[&#34;status&#34;]:
        return AppWrapper(
            name=aw[&#34;metadata&#34;][&#34;name&#34;],
            status=AppWrapperStatus(aw[&#34;status&#34;][&#34;state&#34;].lower()),
            can_run=aw[&#34;status&#34;][&#34;canrun&#34;],
            job_state=aw[&#34;status&#34;][&#34;queuejobstate&#34;],
        )
    return AppWrapper(
        name=aw[&#34;metadata&#34;][&#34;name&#34;],
        status=AppWrapperStatus(&#34;queueing&#34;),
        can_run=False,
        job_state=&#34;Still adding to queue&#34;,
    )


def _copy_to_ray(cluster: Cluster) -&gt; RayCluster:
    ray = RayCluster(
        name=cluster.config.name,
        status=cluster.status(print_to_console=False)[0],
        workers=cluster.config.num_workers,
        worker_mem_min=cluster.config.min_memory,
        worker_mem_max=cluster.config.max_memory,
        worker_cpu=cluster.config.min_cpus,
        worker_gpu=cluster.config.num_gpus,
        namespace=cluster.config.namespace,
        dashboard=cluster.cluster_dashboard_uri(),
        head_cpus=cluster.config.head_cpus,
        head_mem=cluster.config.head_memory,
        head_gpu=cluster.config.head_gpus,
    )
    if ray.status == CodeFlareClusterStatus.READY:
        ray.status = RayClusterStatus.READY
    return ray</code></pre>
</details>
</section>
<section>
</section>
<section>
</section>
<section>
<h2 class="section-title" id="header-functions">Functions</h2>
<dl>
<dt id="codeflare_sdk.cluster.cluster.get_cluster"><code class="name flex">
<span>def <span class="ident">get_cluster</span></span>(<span>cluster_name: str, namespace: str = 'default')</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def get_cluster(cluster_name: str, namespace: str = &#34;default&#34;):
    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        rcs = api_instance.list_namespaced_custom_object(
            group=&#34;ray.io&#34;,
            version=&#34;v1alpha1&#34;,
            namespace=namespace,
            plural=&#34;rayclusters&#34;,
        )
    except Exception as e:
        return _kube_api_error_handling(e)

    for rc in rcs[&#34;items&#34;]:
        if rc[&#34;metadata&#34;][&#34;name&#34;] == cluster_name:
            mcad = _check_aw_exists(cluster_name, namespace)
            return Cluster.from_k8_cluster_object(rc, mcad=mcad)
    raise FileNotFoundError(
        f&#34;Cluster {cluster_name} is not found in {namespace} namespace&#34;
    )</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.get_current_namespace"><code class="name flex">
<span>def <span class="ident">get_current_namespace</span></span>(<span>)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def get_current_namespace():  # pragma: no cover
    if api_config_handler() != None:
        if os.path.isfile(&#34;/var/run/secrets/kubernetes.io/serviceaccount/namespace&#34;):
            try:
                file = open(
                    &#34;/var/run/secrets/kubernetes.io/serviceaccount/namespace&#34;, &#34;r&#34;
                )
                active_context = file.readline().strip(&#34;\n&#34;)
                return active_context
            except Exception as e:
                print(&#34;Unable to find current namespace&#34;)
                return None
        else:
            print(&#34;Unable to find current namespace&#34;)
            return None
    else:
        try:
            _, active_context = config.list_kube_config_contexts(config_check())
        except Exception as e:
            return _kube_api_error_handling(e)
        try:
            return active_context[&#34;context&#34;][&#34;namespace&#34;]
        except KeyError:
            return None</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.list_all_clusters"><code class="name flex">
<span>def <span class="ident">list_all_clusters</span></span>(<span>namespace: str, print_to_console: bool = True)</span>
</code></dt>
<dd>
<div class="desc"><p>Returns (and prints by default) a list of all clusters in a given namespace.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def list_all_clusters(namespace: str, print_to_console: bool = True):
    &#34;&#34;&#34;
    Returns (and prints by default) a list of all clusters in a given namespace.
    &#34;&#34;&#34;
    clusters = _get_ray_clusters(namespace)
    if print_to_console:
        pretty_print.print_clusters(clusters)
    return clusters</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.list_all_queued"><code class="name flex">
<span>def <span class="ident">list_all_queued</span></span>(<span>namespace: str, print_to_console: bool = True)</span>
</code></dt>
<dd>
<div class="desc"><p>Returns (and prints by default) a list of all currently queued-up AppWrappers
in a given namespace.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def list_all_queued(namespace: str, print_to_console: bool = True):
    &#34;&#34;&#34;
    Returns (and prints by default) a list of all currently queued-up AppWrappers
    in a given namespace.
    &#34;&#34;&#34;
    app_wrappers = _get_app_wrappers(
        namespace, filter=[AppWrapperStatus.RUNNING, AppWrapperStatus.PENDING]
    )
    if print_to_console:
        pretty_print.print_app_wrappers_status(app_wrappers)
    return app_wrappers</code></pre>
</details>
</dd>
</dl>
</section>
<section>
<h2 class="section-title" id="header-classes">Classes</h2>
<dl>
<dt id="codeflare_sdk.cluster.cluster.Cluster"><code class="flex name class">
<span>class <span class="ident">Cluster</span></span>
<span>(</span><span>config: <a title="codeflare_sdk.cluster.config.ClusterConfiguration" href="config.html#codeflare_sdk.cluster.config.ClusterConfiguration">ClusterConfiguration</a>)</span>
</code></dt>
<dd>
<div class="desc"><p>An object for requesting, bringing up, and taking down resources.
Can also be used for seeing the resource cluster status and details.</p>
<p>Note that currently, the underlying implementation is a Ray cluster.</p>
<p>Create the resource cluster object by passing in a ClusterConfiguration
(defined in the config sub-module). An AppWrapper will then be generated
based off of the configured resources to represent the desired cluster
request.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">class Cluster:
    &#34;&#34;&#34;
    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.
    &#34;&#34;&#34;

    torchx_scheduler = &#34;ray&#34;

    def __init__(self, config: ClusterConfiguration):
        &#34;&#34;&#34;
        Create the resource cluster object by passing in a ClusterConfiguration
        (defined in the config sub-module). An AppWrapper will then be generated
        based off of the configured resources to represent the desired cluster
        request.
        &#34;&#34;&#34;
        self.config = config
        self.app_wrapper_yaml = self.create_app_wrapper()
        self._job_submission_client = None
        self.app_wrapper_name = self.app_wrapper_yaml.replace(&#34;.yaml&#34;, &#34;&#34;).split(&#34;/&#34;)[
            -1
        ]

    @property
    def _client_headers(self):
        k8_client = api_config_handler() or client.ApiClient()
        return {
            &#34;Authorization&#34;: k8_client.configuration.get_api_key_with_prefix(
                &#34;authorization&#34;
            )
        }

    @property
    def _client_verify_tls(self):
        return not self.config.openshift_oauth

    @property
    def job_client(self):
        if self._job_submission_client:
            return self._job_submission_client
        if self.config.openshift_oauth:
            print(
                api_config_handler().configuration.get_api_key_with_prefix(
                    &#34;authorization&#34;
                )
            )
            self._job_submission_client = JobSubmissionClient(
                self.cluster_dashboard_uri(),
                headers=self._client_headers,
                verify=self._client_verify_tls,
            )
        else:
            self._job_submission_client = JobSubmissionClient(
                self.cluster_dashboard_uri()
            )
        return self._job_submission_client

    def evaluate_dispatch_priority(self):
        priority_class = self.config.dispatch_priority

        try:
            config_check()
            api_instance = client.CustomObjectsApi(api_config_handler())
            priority_classes = api_instance.list_cluster_custom_object(
                group=&#34;scheduling.k8s.io&#34;,
                version=&#34;v1&#34;,
                plural=&#34;priorityclasses&#34;,
            )
        except Exception as e:  # pragma: no cover
            return _kube_api_error_handling(e)

        for pc in priority_classes[&#34;items&#34;]:
            if pc[&#34;metadata&#34;][&#34;name&#34;] == priority_class:
                return pc[&#34;value&#34;]
        print(f&#34;Priority class {priority_class} is not available in the cluster&#34;)
        return None

    def validate_image_config(self):
        &#34;&#34;&#34;
        Validates that the image configuration is not empty.

        :param image: The image string to validate
        :raises ValueError: If the image is not specified
        &#34;&#34;&#34;
        if self.config.image == &#34;&#34; or self.config.image == None:
            raise ValueError(&#34;Image must be specified in the ClusterConfiguration&#34;)

    def create_app_wrapper(self):
        &#34;&#34;&#34;
        Called upon cluster object creation, creates an AppWrapper yaml based on
        the specifications of the ClusterConfiguration.
        &#34;&#34;&#34;

        if self.config.namespace is None:
            self.config.namespace = get_current_namespace()
            if self.config.namespace is None:
                print(&#34;Please specify with namespace=&lt;your_current_namespace&gt;&#34;)
            elif type(self.config.namespace) is not str:
                raise TypeError(
                    f&#34;Namespace {self.config.namespace} is of type {type(self.config.namespace)}. Check your Kubernetes Authentication.&#34;
                )

        # Validate image configuration
        self.validate_image_config()

        # Before attempting to create the cluster AW, let&#39;s evaluate the ClusterConfig
        if self.config.dispatch_priority:
            if not self.config.mcad:
                raise ValueError(
                    &#34;Invalid Cluster Configuration, cannot have dispatch priority without MCAD&#34;
                )
            priority_val = self.evaluate_dispatch_priority()
            if priority_val == None:
                raise ValueError(
                    &#34;Invalid Cluster Configuration, AppWrapper not generated&#34;
                )
        else:
            priority_val = None

        name = self.config.name
        namespace = self.config.namespace
        head_cpus = self.config.head_cpus
        head_memory = self.config.head_memory
        head_gpus = self.config.head_gpus
        min_cpu = self.config.min_cpus
        max_cpu = self.config.max_cpus
        min_memory = self.config.min_memory
        max_memory = self.config.max_memory
        gpu = self.config.num_gpus
        workers = self.config.num_workers
        template = self.config.template
        image = self.config.image
        instascale = self.config.instascale
        mcad = self.config.mcad
        instance_types = self.config.machine_types
        env = self.config.envs
        local_interactive = self.config.local_interactive
        image_pull_secrets = self.config.image_pull_secrets
        dispatch_priority = self.config.dispatch_priority
        ingress_domain = self.config.ingress_domain
        ingress_options = self.config.ingress_options
        return generate_appwrapper(
            name=name,
            namespace=namespace,
            head_cpus=head_cpus,
            head_memory=head_memory,
            head_gpus=head_gpus,
            min_cpu=min_cpu,
            max_cpu=max_cpu,
            min_memory=min_memory,
            max_memory=max_memory,
            gpu=gpu,
            workers=workers,
            template=template,
            image=image,
            instascale=instascale,
            mcad=mcad,
            instance_types=instance_types,
            env=env,
            local_interactive=local_interactive,
            image_pull_secrets=image_pull_secrets,
            dispatch_priority=dispatch_priority,
            priority_val=priority_val,
            openshift_oauth=self.config.openshift_oauth,
            ingress_domain=ingress_domain,
            ingress_options=ingress_options,
        )

    # creates a new cluster with the provided or default spec
    def up(self):
        &#34;&#34;&#34;
        Applies the AppWrapper yaml, pushing the resource request onto
        the MCAD queue.
        &#34;&#34;&#34;
        namespace = self.config.namespace
        if self.config.openshift_oauth:
            create_openshift_oauth_objects(
                cluster_name=self.config.name, namespace=namespace
            )

        try:
            config_check()
            api_instance = client.CustomObjectsApi(api_config_handler())
            if self.config.mcad:
                with open(self.app_wrapper_yaml) as f:
                    aw = yaml.load(f, Loader=yaml.FullLoader)
                api_instance.create_namespaced_custom_object(
                    group=&#34;workload.codeflare.dev&#34;,
                    version=&#34;v1beta1&#34;,
                    namespace=namespace,
                    plural=&#34;appwrappers&#34;,
                    body=aw,
                )
            else:
                self._component_resources_up(namespace, api_instance)
        except Exception as e:  # pragma: no cover
            return _kube_api_error_handling(e)

    def down(self):
        &#34;&#34;&#34;
        Deletes the AppWrapper yaml, scaling-down and deleting all resources
        associated with the cluster.
        &#34;&#34;&#34;
        namespace = self.config.namespace
        try:
            config_check()
            api_instance = client.CustomObjectsApi(api_config_handler())
            if self.config.mcad:
                api_instance.delete_namespaced_custom_object(
                    group=&#34;workload.codeflare.dev&#34;,
                    version=&#34;v1beta1&#34;,
                    namespace=namespace,
                    plural=&#34;appwrappers&#34;,
                    name=self.app_wrapper_name,
                )
            else:
                self._component_resources_down(namespace, api_instance)
        except Exception as e:  # pragma: no cover
            return _kube_api_error_handling(e)

        if self.config.openshift_oauth:
            delete_openshift_oauth_objects(
                cluster_name=self.config.name, namespace=namespace
            )

    def status(
        self, print_to_console: bool = True
    ) -&gt; Tuple[CodeFlareClusterStatus, bool]:
        &#34;&#34;&#34;
        Returns the requested cluster&#39;s status, as well as whether or not
        it is ready for use.
        &#34;&#34;&#34;
        ready = False
        status = CodeFlareClusterStatus.UNKNOWN
        if self.config.mcad:
            # check the app wrapper status
            appwrapper = _app_wrapper_status(self.config.name, self.config.namespace)
            if appwrapper:
                if appwrapper.status in [
                    AppWrapperStatus.RUNNING,
                    AppWrapperStatus.COMPLETED,
                    AppWrapperStatus.RUNNING_HOLD_COMPLETION,
                ]:
                    ready = False
                    status = CodeFlareClusterStatus.STARTING
                elif appwrapper.status in [
                    AppWrapperStatus.FAILED,
                    AppWrapperStatus.DELETED,
                ]:
                    ready = False
                    status = CodeFlareClusterStatus.FAILED  # should deleted be separate
                    return status, ready  # exit early, no need to check ray status
                elif appwrapper.status in [
                    AppWrapperStatus.PENDING,
                    AppWrapperStatus.QUEUEING,
                ]:
                    ready = False
                    if appwrapper.status == AppWrapperStatus.PENDING:
                        status = CodeFlareClusterStatus.QUEUED
                    else:
                        status = CodeFlareClusterStatus.QUEUEING
                    if print_to_console:
                        pretty_print.print_app_wrappers_status([appwrapper])
                    return (
                        status,
                        ready,
                    )  # no need to check the ray status since still in queue

        # check the ray cluster status
        cluster = _ray_cluster_status(self.config.name, self.config.namespace)
        if cluster:
            if cluster.status == RayClusterStatus.UNKNOWN:
                ready = False
                status = CodeFlareClusterStatus.STARTING
            if cluster.status == RayClusterStatus.READY:
                ready = True
                status = CodeFlareClusterStatus.READY
            elif cluster.status in [
                RayClusterStatus.UNHEALTHY,
                RayClusterStatus.FAILED,
            ]:
                ready = False
                status = CodeFlareClusterStatus.FAILED

            if print_to_console:
                # overriding the number of gpus with requested
                cluster.worker_gpu = self.config.num_gpus
                pretty_print.print_cluster_status(cluster)
        elif print_to_console:
            if status == CodeFlareClusterStatus.UNKNOWN:
                pretty_print.print_no_resources_found()
            else:
                pretty_print.print_app_wrappers_status([appwrapper], starting=True)

        return status, ready

    def is_dashboard_ready(self) -&gt; bool:
        try:
            response = requests.get(
                self.cluster_dashboard_uri(),
                headers=self._client_headers,
                timeout=5,
                verify=self._client_verify_tls,
            )
        except requests.exceptions.SSLError:  # pragma no cover
            # SSL exception occurs when oauth ingress has been created but cluster is not up
            return False
        if response.status_code == 200:
            return True
        else:
            return False

    def wait_ready(self, timeout: Optional[int] = None, dashboard_check: bool = True):
        &#34;&#34;&#34;
        Waits for requested cluster to be ready, up to an optional timeout (s).
        Checks every five seconds.
        &#34;&#34;&#34;
        print(&#34;Waiting for requested resources to be set up...&#34;)
        ready = False
        dashboard_ready = False
        status = None
        time = 0
        while not ready:
            status, ready = self.status(print_to_console=False)
            if status == CodeFlareClusterStatus.UNKNOWN:
                print(
                    &#34;WARNING: Current cluster status is unknown, have you run cluster.up yet?&#34;
                )
            if not ready:
                if timeout and time &gt;= timeout:
                    raise TimeoutError(
                        f&#34;wait() timed out after waiting {timeout}s for cluster to be ready&#34;
                    )
                sleep(5)
                time += 5
        print(&#34;Requested cluster is up and running!&#34;)

        while dashboard_check and not dashboard_ready:
            dashboard_ready = self.is_dashboard_ready()
            if not dashboard_ready:
                if timeout and time &gt;= timeout:
                    raise TimeoutError(
                        f&#34;wait() timed out after waiting {timeout}s for dashboard to be ready&#34;
                    )
                sleep(5)
                time += 5
        if dashboard_ready:
            print(&#34;Dashboard is ready!&#34;)

    def details(self, print_to_console: bool = True) -&gt; RayCluster:
        cluster = _copy_to_ray(self)
        if print_to_console:
            pretty_print.print_clusters([cluster])
        return cluster

    def cluster_uri(self) -&gt; str:
        &#34;&#34;&#34;
        Returns a string containing the cluster&#39;s URI.
        &#34;&#34;&#34;
        return f&#34;ray://{self.config.name}-head-svc.{self.config.namespace}.svc:10001&#34;

    def cluster_dashboard_uri(self) -&gt; str:
        &#34;&#34;&#34;
        Returns a string containing the cluster&#39;s dashboard URI.
        &#34;&#34;&#34;
        try:
            config_check()
            api_instance = client.NetworkingV1Api(api_config_handler())
            ingresses = api_instance.list_namespaced_ingress(self.config.namespace)
        except Exception as e:  # pragma no cover
            return _kube_api_error_handling(e)

        for ingress in ingresses.items:
            annotations = ingress.metadata.annotations
            protocol = &#34;http&#34;
            if (
                ingress.metadata.name == f&#34;ray-dashboard-{self.config.name}&#34;
                or ingress.metadata.name.startswith(f&#34;{self.config.name}-ingress&#34;)
            ):
                if annotations == None:
                    protocol = &#34;http&#34;
                elif &#34;route.openshift.io/termination&#34; in annotations:
                    protocol = &#34;https&#34;
            return f&#34;{protocol}://{ingress.spec.rules[0].host}&#34;
        return &#34;Dashboard ingress not available yet, have you run cluster.up()?&#34;

    def list_jobs(self) -&gt; List:
        &#34;&#34;&#34;
        This method accesses the head ray node in your cluster and lists the running jobs.
        &#34;&#34;&#34;
        return self.job_client.list_jobs()

    def job_status(self, job_id: str) -&gt; str:
        &#34;&#34;&#34;
        This method accesses the head ray node in your cluster and returns the job status for the provided job id.
        &#34;&#34;&#34;
        return self.job_client.get_job_status(job_id)

    def job_logs(self, job_id: str) -&gt; str:
        &#34;&#34;&#34;
        This method accesses the head ray node in your cluster and returns the logs for the provided job id.
        &#34;&#34;&#34;
        return self.job_client.get_job_logs(job_id)

    def torchx_config(
        self, working_dir: str = None, requirements: str = None
    ) -&gt; Dict[str, str]:
        dashboard_address = urllib3.util.parse_url(self.cluster_dashboard_uri()).host
        to_return = {
            &#34;cluster_name&#34;: self.config.name,
            &#34;dashboard_address&#34;: dashboard_address,
        }
        if working_dir:
            to_return[&#34;working_dir&#34;] = working_dir
        if requirements:
            to_return[&#34;requirements&#34;] = requirements
        return to_return

    def from_k8_cluster_object(rc, mcad=True):
        machine_types = (
            rc[&#34;metadata&#34;][&#34;labels&#34;][&#34;orderedinstance&#34;].split(&#34;_&#34;)
            if &#34;orderedinstance&#34; in rc[&#34;metadata&#34;][&#34;labels&#34;]
            else []
        )
        local_interactive = (
            &#34;volumeMounts&#34;
            in rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0]
        )
        cluster_config = ClusterConfiguration(
            name=rc[&#34;metadata&#34;][&#34;name&#34;],
            namespace=rc[&#34;metadata&#34;][&#34;namespace&#34;],
            machine_types=machine_types,
            num_workers=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;minReplicas&#34;],
            min_cpus=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][
                &#34;containers&#34;
            ][0][&#34;resources&#34;][&#34;requests&#34;][&#34;cpu&#34;],
            max_cpus=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][
                &#34;containers&#34;
            ][0][&#34;resources&#34;][&#34;limits&#34;][&#34;cpu&#34;],
            min_memory=int(
                rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0][
                    &#34;resources&#34;
                ][&#34;requests&#34;][&#34;memory&#34;][:-1]
            ),
            max_memory=int(
                rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0][
                    &#34;resources&#34;
                ][&#34;limits&#34;][&#34;memory&#34;][:-1]
            ),
            num_gpus=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][
                &#34;containers&#34;
            ][0][&#34;resources&#34;][&#34;limits&#34;][&#34;nvidia.com/gpu&#34;],
            instascale=True if machine_types else False,
            image=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][
                0
            ][&#34;image&#34;],
            local_interactive=local_interactive,
            mcad=mcad,
        )
        return Cluster(cluster_config)

    def local_client_url(self):
        if self.config.local_interactive == True:
            ingress_domain = _get_ingress_domain(self)
            return f&#34;ray://{ingress_domain}&#34;
        else:
            return &#34;None&#34;

    def _component_resources_up(
        self, namespace: str, api_instance: client.CustomObjectsApi
    ):
        with open(self.app_wrapper_yaml) as f:
            yamls = yaml.load_all(f, Loader=yaml.FullLoader)
            for resource in yamls:
                if resource[&#34;kind&#34;] == &#34;RayCluster&#34;:
                    api_instance.create_namespaced_custom_object(
                        group=&#34;ray.io&#34;,
                        version=&#34;v1alpha1&#34;,
                        namespace=namespace,
                        plural=&#34;rayclusters&#34;,
                        body=resource,
                    )
                elif resource[&#34;kind&#34;] == &#34;Route&#34;:
                    api_instance.create_namespaced_custom_object(
                        group=&#34;route.openshift.io&#34;,
                        version=&#34;v1&#34;,
                        namespace=namespace,
                        plural=&#34;routes&#34;,
                        body=resource,
                    )
                elif resource[&#34;kind&#34;] == &#34;Secret&#34;:
                    secret_instance = client.CoreV1Api(api_config_handler())
                    secret_instance.create_namespaced_secret(
                        namespace=namespace,
                        body=resource,
                    )

    def _component_resources_down(
        self, namespace: str, api_instance: client.CustomObjectsApi
    ):
        with open(self.app_wrapper_yaml) as f:
            yamls = yaml.load_all(f, Loader=yaml.FullLoader)
            for resource in yamls:
                if resource[&#34;kind&#34;] == &#34;RayCluster&#34;:
                    api_instance.delete_namespaced_custom_object(
                        group=&#34;ray.io&#34;,
                        version=&#34;v1alpha1&#34;,
                        namespace=namespace,
                        plural=&#34;rayclusters&#34;,
                        name=self.app_wrapper_name,
                    )
                elif resource[&#34;kind&#34;] == &#34;Route&#34;:
                    name = resource[&#34;metadata&#34;][&#34;name&#34;]
                    api_instance.delete_namespaced_custom_object(
                        group=&#34;route.openshift.io&#34;,
                        version=&#34;v1&#34;,
                        namespace=namespace,
                        plural=&#34;routes&#34;,
                        name=name,
                    )
                elif resource[&#34;kind&#34;] == &#34;Secret&#34;:
                    name = resource[&#34;metadata&#34;][&#34;name&#34;]
                    secret_instance = client.CoreV1Api(api_config_handler())
                    secret_instance.delete_namespaced_secret(
                        namespace=namespace,
                        name=name,
                    )</code></pre>
</details>
<h3>Class variables</h3>
<dl>
<dt id="codeflare_sdk.cluster.cluster.Cluster.torchx_scheduler"><code class="name">var <span class="ident">torchx_scheduler</span></code></dt>
<dd>
<div class="desc"></div>
</dd>
</dl>
<h3>Instance variables</h3>
<dl>
<dt id="codeflare_sdk.cluster.cluster.Cluster.job_client"><code class="name">var <span class="ident">job_client</span></code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">@property
def job_client(self):
    if self._job_submission_client:
        return self._job_submission_client
    if self.config.openshift_oauth:
        print(
            api_config_handler().configuration.get_api_key_with_prefix(
                &#34;authorization&#34;
            )
        )
        self._job_submission_client = JobSubmissionClient(
            self.cluster_dashboard_uri(),
            headers=self._client_headers,
            verify=self._client_verify_tls,
        )
    else:
        self._job_submission_client = JobSubmissionClient(
            self.cluster_dashboard_uri()
        )
    return self._job_submission_client</code></pre>
</details>
</dd>
</dl>
<h3>Methods</h3>
<dl>
<dt id="codeflare_sdk.cluster.cluster.Cluster.cluster_dashboard_uri"><code class="name flex">
<span>def <span class="ident">cluster_dashboard_uri</span></span>(<span>self) ‑> str</span>
</code></dt>
<dd>
<div class="desc"><p>Returns a string containing the cluster's dashboard URI.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def cluster_dashboard_uri(self) -&gt; str:
    &#34;&#34;&#34;
    Returns a string containing the cluster&#39;s dashboard URI.
    &#34;&#34;&#34;
    try:
        config_check()
        api_instance = client.NetworkingV1Api(api_config_handler())
        ingresses = api_instance.list_namespaced_ingress(self.config.namespace)
    except Exception as e:  # pragma no cover
        return _kube_api_error_handling(e)

    for ingress in ingresses.items:
        annotations = ingress.metadata.annotations
        protocol = &#34;http&#34;
        if (
            ingress.metadata.name == f&#34;ray-dashboard-{self.config.name}&#34;
            or ingress.metadata.name.startswith(f&#34;{self.config.name}-ingress&#34;)
        ):
            if annotations == None:
                protocol = &#34;http&#34;
            elif &#34;route.openshift.io/termination&#34; in annotations:
                protocol = &#34;https&#34;
        return f&#34;{protocol}://{ingress.spec.rules[0].host}&#34;
    return &#34;Dashboard ingress not available yet, have you run cluster.up()?&#34;</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.cluster_uri"><code class="name flex">
<span>def <span class="ident">cluster_uri</span></span>(<span>self) ‑> str</span>
</code></dt>
<dd>
<div class="desc"><p>Returns a string containing the cluster's URI.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def cluster_uri(self) -&gt; str:
    &#34;&#34;&#34;
    Returns a string containing the cluster&#39;s URI.
    &#34;&#34;&#34;
    return f&#34;ray://{self.config.name}-head-svc.{self.config.namespace}.svc:10001&#34;</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.create_app_wrapper"><code class="name flex">
<span>def <span class="ident">create_app_wrapper</span></span>(<span>self)</span>
</code></dt>
<dd>
<div class="desc"><p>Called upon cluster object creation, creates an AppWrapper yaml based on
the specifications of the ClusterConfiguration.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def create_app_wrapper(self):
    &#34;&#34;&#34;
    Called upon cluster object creation, creates an AppWrapper yaml based on
    the specifications of the ClusterConfiguration.
    &#34;&#34;&#34;

    if self.config.namespace is None:
        self.config.namespace = get_current_namespace()
        if self.config.namespace is None:
            print(&#34;Please specify with namespace=&lt;your_current_namespace&gt;&#34;)
        elif type(self.config.namespace) is not str:
            raise TypeError(
                f&#34;Namespace {self.config.namespace} is of type {type(self.config.namespace)}. Check your Kubernetes Authentication.&#34;
            )

    # Validate image configuration
    self.validate_image_config()

    # Before attempting to create the cluster AW, let&#39;s evaluate the ClusterConfig
    if self.config.dispatch_priority:
        if not self.config.mcad:
            raise ValueError(
                &#34;Invalid Cluster Configuration, cannot have dispatch priority without MCAD&#34;
            )
        priority_val = self.evaluate_dispatch_priority()
        if priority_val == None:
            raise ValueError(
                &#34;Invalid Cluster Configuration, AppWrapper not generated&#34;
            )
    else:
        priority_val = None

    name = self.config.name
    namespace = self.config.namespace
    head_cpus = self.config.head_cpus
    head_memory = self.config.head_memory
    head_gpus = self.config.head_gpus
    min_cpu = self.config.min_cpus
    max_cpu = self.config.max_cpus
    min_memory = self.config.min_memory
    max_memory = self.config.max_memory
    gpu = self.config.num_gpus
    workers = self.config.num_workers
    template = self.config.template
    image = self.config.image
    instascale = self.config.instascale
    mcad = self.config.mcad
    instance_types = self.config.machine_types
    env = self.config.envs
    local_interactive = self.config.local_interactive
    image_pull_secrets = self.config.image_pull_secrets
    dispatch_priority = self.config.dispatch_priority
    ingress_domain = self.config.ingress_domain
    ingress_options = self.config.ingress_options
    return generate_appwrapper(
        name=name,
        namespace=namespace,
        head_cpus=head_cpus,
        head_memory=head_memory,
        head_gpus=head_gpus,
        min_cpu=min_cpu,
        max_cpu=max_cpu,
        min_memory=min_memory,
        max_memory=max_memory,
        gpu=gpu,
        workers=workers,
        template=template,
        image=image,
        instascale=instascale,
        mcad=mcad,
        instance_types=instance_types,
        env=env,
        local_interactive=local_interactive,
        image_pull_secrets=image_pull_secrets,
        dispatch_priority=dispatch_priority,
        priority_val=priority_val,
        openshift_oauth=self.config.openshift_oauth,
        ingress_domain=ingress_domain,
        ingress_options=ingress_options,
    )</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.details"><code class="name flex">
<span>def <span class="ident">details</span></span>(<span>self, print_to_console: bool = True) ‑> <a title="codeflare_sdk.cluster.model.RayCluster" href="model.html#codeflare_sdk.cluster.model.RayCluster">RayCluster</a></span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def details(self, print_to_console: bool = True) -&gt; RayCluster:
    cluster = _copy_to_ray(self)
    if print_to_console:
        pretty_print.print_clusters([cluster])
    return cluster</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.down"><code class="name flex">
<span>def <span class="ident">down</span></span>(<span>self)</span>
</code></dt>
<dd>
<div class="desc"><p>Deletes the AppWrapper yaml, scaling-down and deleting all resources
associated with the cluster.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def down(self):
    &#34;&#34;&#34;
    Deletes the AppWrapper yaml, scaling-down and deleting all resources
    associated with the cluster.
    &#34;&#34;&#34;
    namespace = self.config.namespace
    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        if self.config.mcad:
            api_instance.delete_namespaced_custom_object(
                group=&#34;workload.codeflare.dev&#34;,
                version=&#34;v1beta1&#34;,
                namespace=namespace,
                plural=&#34;appwrappers&#34;,
                name=self.app_wrapper_name,
            )
        else:
            self._component_resources_down(namespace, api_instance)
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)

    if self.config.openshift_oauth:
        delete_openshift_oauth_objects(
            cluster_name=self.config.name, namespace=namespace
        )</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.evaluate_dispatch_priority"><code class="name flex">
<span>def <span class="ident">evaluate_dispatch_priority</span></span>(<span>self)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def evaluate_dispatch_priority(self):
    priority_class = self.config.dispatch_priority

    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        priority_classes = api_instance.list_cluster_custom_object(
            group=&#34;scheduling.k8s.io&#34;,
            version=&#34;v1&#34;,
            plural=&#34;priorityclasses&#34;,
        )
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)

    for pc in priority_classes[&#34;items&#34;]:
        if pc[&#34;metadata&#34;][&#34;name&#34;] == priority_class:
            return pc[&#34;value&#34;]
    print(f&#34;Priority class {priority_class} is not available in the cluster&#34;)
    return None</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.from_k8_cluster_object"><code class="name flex">
<span>def <span class="ident">from_k8_cluster_object</span></span>(<span>rc, mcad=True)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def from_k8_cluster_object(rc, mcad=True):
    machine_types = (
        rc[&#34;metadata&#34;][&#34;labels&#34;][&#34;orderedinstance&#34;].split(&#34;_&#34;)
        if &#34;orderedinstance&#34; in rc[&#34;metadata&#34;][&#34;labels&#34;]
        else []
    )
    local_interactive = (
        &#34;volumeMounts&#34;
        in rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0]
    )
    cluster_config = ClusterConfiguration(
        name=rc[&#34;metadata&#34;][&#34;name&#34;],
        namespace=rc[&#34;metadata&#34;][&#34;namespace&#34;],
        machine_types=machine_types,
        num_workers=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;minReplicas&#34;],
        min_cpus=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][
            &#34;containers&#34;
        ][0][&#34;resources&#34;][&#34;requests&#34;][&#34;cpu&#34;],
        max_cpus=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][
            &#34;containers&#34;
        ][0][&#34;resources&#34;][&#34;limits&#34;][&#34;cpu&#34;],
        min_memory=int(
            rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0][
                &#34;resources&#34;
            ][&#34;requests&#34;][&#34;memory&#34;][:-1]
        ),
        max_memory=int(
            rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][0][
                &#34;resources&#34;
            ][&#34;limits&#34;][&#34;memory&#34;][:-1]
        ),
        num_gpus=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][
            &#34;containers&#34;
        ][0][&#34;resources&#34;][&#34;limits&#34;][&#34;nvidia.com/gpu&#34;],
        instascale=True if machine_types else False,
        image=rc[&#34;spec&#34;][&#34;workerGroupSpecs&#34;][0][&#34;template&#34;][&#34;spec&#34;][&#34;containers&#34;][
            0
        ][&#34;image&#34;],
        local_interactive=local_interactive,
        mcad=mcad,
    )
    return Cluster(cluster_config)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.is_dashboard_ready"><code class="name flex">
<span>def <span class="ident">is_dashboard_ready</span></span>(<span>self) ‑> bool</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def is_dashboard_ready(self) -&gt; bool:
    try:
        response = requests.get(
            self.cluster_dashboard_uri(),
            headers=self._client_headers,
            timeout=5,
            verify=self._client_verify_tls,
        )
    except requests.exceptions.SSLError:  # pragma no cover
        # SSL exception occurs when oauth ingress has been created but cluster is not up
        return False
    if response.status_code == 200:
        return True
    else:
        return False</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.job_logs"><code class="name flex">
<span>def <span class="ident">job_logs</span></span>(<span>self, job_id: str) ‑> str</span>
</code></dt>
<dd>
<div class="desc"><p>This method accesses the head ray node in your cluster and returns the logs for the provided job id.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def job_logs(self, job_id: str) -&gt; str:
    &#34;&#34;&#34;
    This method accesses the head ray node in your cluster and returns the logs for the provided job id.
    &#34;&#34;&#34;
    return self.job_client.get_job_logs(job_id)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.job_status"><code class="name flex">
<span>def <span class="ident">job_status</span></span>(<span>self, job_id: str) ‑> str</span>
</code></dt>
<dd>
<div class="desc"><p>This method accesses the head ray node in your cluster and returns the job status for the provided job id.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def job_status(self, job_id: str) -&gt; str:
    &#34;&#34;&#34;
    This method accesses the head ray node in your cluster and returns the job status for the provided job id.
    &#34;&#34;&#34;
    return self.job_client.get_job_status(job_id)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.list_jobs"><code class="name flex">
<span>def <span class="ident">list_jobs</span></span>(<span>self) ‑> List[~T]</span>
</code></dt>
<dd>
<div class="desc"><p>This method accesses the head ray node in your cluster and lists the running jobs.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def list_jobs(self) -&gt; List:
    &#34;&#34;&#34;
    This method accesses the head ray node in your cluster and lists the running jobs.
    &#34;&#34;&#34;
    return self.job_client.list_jobs()</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.local_client_url"><code class="name flex">
<span>def <span class="ident">local_client_url</span></span>(<span>self)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def local_client_url(self):
    if self.config.local_interactive == True:
        ingress_domain = _get_ingress_domain(self)
        return f&#34;ray://{ingress_domain}&#34;
    else:
        return &#34;None&#34;</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.status"><code class="name flex">
<span>def <span class="ident">status</span></span>(<span>self, print_to_console: bool = True) ‑> Tuple[<a title="codeflare_sdk.cluster.model.CodeFlareClusterStatus" href="model.html#codeflare_sdk.cluster.model.CodeFlareClusterStatus">CodeFlareClusterStatus</a>, bool]</span>
</code></dt>
<dd>
<div class="desc"><p>Returns the requested cluster's status, as well as whether or not
it is ready for use.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def status(
    self, print_to_console: bool = True
) -&gt; Tuple[CodeFlareClusterStatus, bool]:
    &#34;&#34;&#34;
    Returns the requested cluster&#39;s status, as well as whether or not
    it is ready for use.
    &#34;&#34;&#34;
    ready = False
    status = CodeFlareClusterStatus.UNKNOWN
    if self.config.mcad:
        # check the app wrapper status
        appwrapper = _app_wrapper_status(self.config.name, self.config.namespace)
        if appwrapper:
            if appwrapper.status in [
                AppWrapperStatus.RUNNING,
                AppWrapperStatus.COMPLETED,
                AppWrapperStatus.RUNNING_HOLD_COMPLETION,
            ]:
                ready = False
                status = CodeFlareClusterStatus.STARTING
            elif appwrapper.status in [
                AppWrapperStatus.FAILED,
                AppWrapperStatus.DELETED,
            ]:
                ready = False
                status = CodeFlareClusterStatus.FAILED  # should deleted be separate
                return status, ready  # exit early, no need to check ray status
            elif appwrapper.status in [
                AppWrapperStatus.PENDING,
                AppWrapperStatus.QUEUEING,
            ]:
                ready = False
                if appwrapper.status == AppWrapperStatus.PENDING:
                    status = CodeFlareClusterStatus.QUEUED
                else:
                    status = CodeFlareClusterStatus.QUEUEING
                if print_to_console:
                    pretty_print.print_app_wrappers_status([appwrapper])
                return (
                    status,
                    ready,
                )  # no need to check the ray status since still in queue

    # check the ray cluster status
    cluster = _ray_cluster_status(self.config.name, self.config.namespace)
    if cluster:
        if cluster.status == RayClusterStatus.UNKNOWN:
            ready = False
            status = CodeFlareClusterStatus.STARTING
        if cluster.status == RayClusterStatus.READY:
            ready = True
            status = CodeFlareClusterStatus.READY
        elif cluster.status in [
            RayClusterStatus.UNHEALTHY,
            RayClusterStatus.FAILED,
        ]:
            ready = False
            status = CodeFlareClusterStatus.FAILED

        if print_to_console:
            # overriding the number of gpus with requested
            cluster.worker_gpu = self.config.num_gpus
            pretty_print.print_cluster_status(cluster)
    elif print_to_console:
        if status == CodeFlareClusterStatus.UNKNOWN:
            pretty_print.print_no_resources_found()
        else:
            pretty_print.print_app_wrappers_status([appwrapper], starting=True)

    return status, ready</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.torchx_config"><code class="name flex">
<span>def <span class="ident">torchx_config</span></span>(<span>self, working_dir: str = None, requirements: str = None) ‑> Dict[str, str]</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def torchx_config(
    self, working_dir: str = None, requirements: str = None
) -&gt; Dict[str, str]:
    dashboard_address = urllib3.util.parse_url(self.cluster_dashboard_uri()).host
    to_return = {
        &#34;cluster_name&#34;: self.config.name,
        &#34;dashboard_address&#34;: dashboard_address,
    }
    if working_dir:
        to_return[&#34;working_dir&#34;] = working_dir
    if requirements:
        to_return[&#34;requirements&#34;] = requirements
    return to_return</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.up"><code class="name flex">
<span>def <span class="ident">up</span></span>(<span>self)</span>
</code></dt>
<dd>
<div class="desc"><p>Applies the AppWrapper yaml, pushing the resource request onto
the MCAD queue.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def up(self):
    &#34;&#34;&#34;
    Applies the AppWrapper yaml, pushing the resource request onto
    the MCAD queue.
    &#34;&#34;&#34;
    namespace = self.config.namespace
    if self.config.openshift_oauth:
        create_openshift_oauth_objects(
            cluster_name=self.config.name, namespace=namespace
        )

    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        if self.config.mcad:
            with open(self.app_wrapper_yaml) as f:
                aw = yaml.load(f, Loader=yaml.FullLoader)
            api_instance.create_namespaced_custom_object(
                group=&#34;workload.codeflare.dev&#34;,
                version=&#34;v1beta1&#34;,
                namespace=namespace,
                plural=&#34;appwrappers&#34;,
                body=aw,
            )
        else:
            self._component_resources_up(namespace, api_instance)
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.validate_image_config"><code class="name flex">
<span>def <span class="ident">validate_image_config</span></span>(<span>self)</span>
</code></dt>
<dd>
<div class="desc"><p>Validates that the image configuration is not empty.</p>
<p>:param image: The image string to validate
:raises ValueError: If the image is not specified</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def validate_image_config(self):
    &#34;&#34;&#34;
    Validates that the image configuration is not empty.

    :param image: The image string to validate
    :raises ValueError: If the image is not specified
    &#34;&#34;&#34;
    if self.config.image == &#34;&#34; or self.config.image == None:
        raise ValueError(&#34;Image must be specified in the ClusterConfiguration&#34;)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.cluster.cluster.Cluster.wait_ready"><code class="name flex">
<span>def <span class="ident">wait_ready</span></span>(<span>self, timeout: Optional[int] = None, dashboard_check: bool = True)</span>
</code></dt>
<dd>
<div class="desc"><p>Waits for requested cluster to be ready, up to an optional timeout (s).
Checks every five seconds.</p></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def wait_ready(self, timeout: Optional[int] = None, dashboard_check: bool = True):
    &#34;&#34;&#34;
    Waits for requested cluster to be ready, up to an optional timeout (s).
    Checks every five seconds.
    &#34;&#34;&#34;
    print(&#34;Waiting for requested resources to be set up...&#34;)
    ready = False
    dashboard_ready = False
    status = None
    time = 0
    while not ready:
        status, ready = self.status(print_to_console=False)
        if status == CodeFlareClusterStatus.UNKNOWN:
            print(
                &#34;WARNING: Current cluster status is unknown, have you run cluster.up yet?&#34;
            )
        if not ready:
            if timeout and time &gt;= timeout:
                raise TimeoutError(
                    f&#34;wait() timed out after waiting {timeout}s for cluster to be ready&#34;
                )
            sleep(5)
            time += 5
    print(&#34;Requested cluster is up and running!&#34;)

    while dashboard_check and not dashboard_ready:
        dashboard_ready = self.is_dashboard_ready()
        if not dashboard_ready:
            if timeout and time &gt;= timeout:
                raise TimeoutError(
                    f&#34;wait() timed out after waiting {timeout}s for dashboard to be ready&#34;
                )
            sleep(5)
            time += 5
    if dashboard_ready:
        print(&#34;Dashboard is ready!&#34;)</code></pre>
</details>
</dd>
</dl>
</dd>
</dl>
</section>
</article>
<nav id="sidebar">
<h1>Index</h1>
<div class="toc">
<ul></ul>
</div>
<ul id="index">
<li><h3>Super-module</h3>
<ul>
<li><code><a title="codeflare_sdk.cluster" href="index.html">codeflare_sdk.cluster</a></code></li>
</ul>
</li>
<li><h3><a href="#header-functions">Functions</a></h3>
<ul class="">
<li><code><a title="codeflare_sdk.cluster.cluster.get_cluster" href="#codeflare_sdk.cluster.cluster.get_cluster">get_cluster</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.get_current_namespace" href="#codeflare_sdk.cluster.cluster.get_current_namespace">get_current_namespace</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.list_all_clusters" href="#codeflare_sdk.cluster.cluster.list_all_clusters">list_all_clusters</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.list_all_queued" href="#codeflare_sdk.cluster.cluster.list_all_queued">list_all_queued</a></code></li>
</ul>
</li>
<li><h3><a href="#header-classes">Classes</a></h3>
<ul>
<li>
<h4><code><a title="codeflare_sdk.cluster.cluster.Cluster" href="#codeflare_sdk.cluster.cluster.Cluster">Cluster</a></code></h4>
<ul class="">
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.cluster_dashboard_uri" href="#codeflare_sdk.cluster.cluster.Cluster.cluster_dashboard_uri">cluster_dashboard_uri</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.cluster_uri" href="#codeflare_sdk.cluster.cluster.Cluster.cluster_uri">cluster_uri</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.create_app_wrapper" href="#codeflare_sdk.cluster.cluster.Cluster.create_app_wrapper">create_app_wrapper</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.details" href="#codeflare_sdk.cluster.cluster.Cluster.details">details</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.down" href="#codeflare_sdk.cluster.cluster.Cluster.down">down</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.evaluate_dispatch_priority" href="#codeflare_sdk.cluster.cluster.Cluster.evaluate_dispatch_priority">evaluate_dispatch_priority</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.from_k8_cluster_object" href="#codeflare_sdk.cluster.cluster.Cluster.from_k8_cluster_object">from_k8_cluster_object</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.is_dashboard_ready" href="#codeflare_sdk.cluster.cluster.Cluster.is_dashboard_ready">is_dashboard_ready</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.job_client" href="#codeflare_sdk.cluster.cluster.Cluster.job_client">job_client</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.job_logs" href="#codeflare_sdk.cluster.cluster.Cluster.job_logs">job_logs</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.job_status" href="#codeflare_sdk.cluster.cluster.Cluster.job_status">job_status</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.list_jobs" href="#codeflare_sdk.cluster.cluster.Cluster.list_jobs">list_jobs</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.local_client_url" href="#codeflare_sdk.cluster.cluster.Cluster.local_client_url">local_client_url</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.status" href="#codeflare_sdk.cluster.cluster.Cluster.status">status</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.torchx_config" href="#codeflare_sdk.cluster.cluster.Cluster.torchx_config">torchx_config</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.torchx_scheduler" href="#codeflare_sdk.cluster.cluster.Cluster.torchx_scheduler">torchx_scheduler</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.up" href="#codeflare_sdk.cluster.cluster.Cluster.up">up</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.validate_image_config" href="#codeflare_sdk.cluster.cluster.Cluster.validate_image_config">validate_image_config</a></code></li>
<li><code><a title="codeflare_sdk.cluster.cluster.Cluster.wait_ready" href="#codeflare_sdk.cluster.cluster.Cluster.wait_ready">wait_ready</a></code></li>
</ul>
</li>
</ul>
</li>
</ul>
</nav>
</main>
<footer id="footer">
<p>Generated by <a href="https://pdoc3.github.io/pdoc" title="pdoc: Python API documentation generator"><cite>pdoc</cite> 0.10.0</a>.</p>
</footer>
</body>
</html>