from collections import defaultdict
from typing import Any
from chariot import _apis
from chariot.config import getLogger
from chariot.projects.projects import _get_project_id_from_project_args
log = getLogger(__name__)
[docs]
class Resource:
"""Base class for Chariot Resources.
Attributes
----------
project_id
id of the project that the resource belongs to
id
id of the resource
_meta
metainformation of the resource. this is what is returned by
the get-by-id endpoint of the corresponding microservice.
"""
def __init__(
self,
project_id: str | None = None,
id: str | None = None,
project_name: str | None = None,
subproject_name: str | None = None,
organization_id: str | None = None,
name: str | None = None,
version: str | None = None,
metadata: Any | None = None,
):
if metadata:
if hasattr(metadata, "id"):
id = metadata.id
if hasattr(metadata, "projects"):
if isinstance(metadata.projects, list):
project_id = metadata.projects[0].id
else:
project_id = metadata.projects.id
elif hasattr(metadata, "project_id"):
project_id = metadata.project_id
if not (project_id and id) and hasattr(metadata, "name"):
name = metadata.name
project_id = _get_project_id_from_project_args(
project_id=project_id,
project_name=project_name,
subproject_name=subproject_name,
organization_id=organization_id,
)
if name is None and id is None:
raise ValueError("Both `id` and `name` were not specified.")
if name is not None and id is not None:
raise ValueError("Both `id` and `name` cannot both be specified.")
if not id and name is not None:
id = self._get_id_by_name_and_version(project_id=project_id, name=name, version=version)
if id is None:
raise ValueError("`id` cannot be None")
self.id = id
self.project_id = project_id
self._meta = metadata or self._get_resource_meta(project_id=project_id, id=id)
@staticmethod
def _get_resource_meta(project_id: str, id: str):
raise NotImplementedError
@staticmethod
def _get_id_by_name_and_version(project_id: str, name: str, version: str) -> str:
raise NotImplementedError
# TODO: move this into sdk/chariot/system_resources/resource.py
def _parse_value(value):
if value.endswith("m"):
return int(value[:-1])
if value.endswith("Ki"):
return int(value[:-2])
if value.endswith("Gi"):
return int(value[:-2]) * 1024
return int(value)
# TODO: move this into sdk/chariot/system_resources/resource.py
def _update_total(total, capacity):
for key, value in capacity.items():
if key in ["node"] or not value:
continue
try:
value = _parse_value(value)
total[key] += value
except Exception as e:
log.warn(f"Could not parse {value} for {key} when updating totals, {e}")
# TODO: move this into sdk/chariot/system_resources/resource.py
[docs]
@_apis.login_required
def get_capacity():
"""Returns the current total allocated, total allocatable, and single node allocatable resource capacity of the cluster"""
capacities = _apis.resources.capacities_api.v2_nodes_capacities_get().data.capacities
allocatable = defaultdict(int)
allocated = defaultdict(int)
capacities = (
[
{
"node": k,
"allocatable": {"node": k, **v.allocatable.to_dict()},
"allocated": {"node": k, **v.allocated.to_dict()},
}
for k, v in capacities.items()
]
if isinstance(capacities, dict)
else [cap.to_dict() for cap in capacities]
)
for capacity in capacities:
_update_total(allocatable, capacity["allocatable"])
_update_total(allocated, capacity["allocated"])
result = {
"total_allocated": allocated,
"total_allocatable": allocatable,
}
for key in allocatable.keys():
if key in ["pods"]:
continue
result[f"largest_free_{key}"] = sorted(
capacities, key=lambda x: _parse_value(x["allocatable"][key])
)[-1]["allocatable"]
return result