-
Notifications
You must be signed in to change notification settings - Fork 2.4k
/
get.py
120 lines (88 loc) · 3.7 KB
/
get.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
"""GET request handlers."""
from flask import request
from kubeflow.kubeflow.crud_backend import api, logging
from werkzeug.exceptions import NotFound
from .. import utils
from . import bp
log = logging.getLogger(__name__)
@bp.route("/api/config")
def get_config():
config = utils.load_spawner_ui_config()
return api.success_response("config", config)
@bp.route("/api/namespaces/<namespace>/pvcs")
def get_pvcs(namespace):
pvcs = api.list_pvcs(namespace).items
data = [{"name": pvc.metadata.name,
"size": pvc.spec.resources.requests["storage"],
"mode": pvc.spec.access_modes[0]} for pvc in pvcs]
return api.success_response("pvcs", data)
@bp.route("/api/namespaces/<namespace>/poddefaults")
def get_poddefaults(namespace):
pod_defaults = api.list_poddefaults(namespace)
# Return a list of pod defaults adding custom fields (label, desc) for forms
contents = []
for pd in pod_defaults["items"]:
label = list(pd["spec"]["selector"]["matchLabels"].keys())[0]
if "desc" in pd["spec"]:
desc = pd["spec"]["desc"]
else:
desc = pd["metadata"]["name"]
pd["label"] = label
pd["desc"] = desc
contents.append(pd)
log.info("Found poddefaults: %s", contents)
return api.success_response("poddefaults", contents)
@bp.route("/api/namespaces/<namespace>/notebooks")
def get_notebooks(namespace):
notebooks = api.list_notebooks(namespace)["items"]
contents = [utils.notebook_dict_from_k8s_obj(nb) for nb in notebooks]
return api.success_response("notebooks", contents)
@bp.route("/api/namespaces/<namespace>/notebooks/<name>")
def get_notebook(name, namespace):
notebook = api.get_notebook(name, namespace)
return api.success_response("notebook", notebook)
@bp.route("/api/namespaces/<namespace>/notebooks/<notebook_name>/pod")
def get_notebook_pod(notebook_name, namespace):
label_selector = "notebook-name=" + notebook_name
# There should be only one Pod for each Notebook,
# so we expect items to have length = 1
pods = api.list_pods(namespace = namespace, label_selector = label_selector)
if pods.items:
pod = pods.items[0]
return api.success_response(
"pod", api.serialize(pod),
)
else:
raise NotFound("No pod detected.")
@bp.route("/api/namespaces/<namespace>/notebooks/<notebook_name>/pod/<pod_name>/logs")
def get_pod_logs(namespace, notebook_name, pod_name):
container = notebook_name
logs = api.get_pod_logs(namespace, pod_name, container)
return api.success_response(
"logs", logs.split("\n"),
)
@bp.route("/api/namespaces/<namespace>/notebooks/<notebook_name>/events")
def get_notebook_events(notebook_name, namespace):
events = api.list_notebook_events(notebook_name, namespace).items
return api.success_response(
"events", api.serialize(events),
)
@bp.route("/api/gpus")
def get_gpu_vendors():
"""
Return a list of GPU vendors for which at least one node has the necessary
annotation required to schedule pods
"""
frontend_config = utils.load_spawner_ui_config()
gpus_value = frontend_config.get("gpus", {}).get("value", {})
config_vendor_keys = [
v.get("limitsKey", "") for v in gpus_value.get("vendors", [])
]
# Get all of the different resources installed in all nodes
installed_resources = set()
nodes = api.list_nodes().items
for node in nodes:
installed_resources.update(node.status.capacity.keys())
# Keep the vendors the key of which exists in at least one node
available_vendors = installed_resources.intersection(config_vendor_keys)
return api.success_response("vendors", list(available_vendors))