diff --git a/ocp_resources/application_aware_applied_cluster_resource_quota.py b/ocp_resources/application_aware_applied_cluster_resource_quota.py new file mode 100644 index 0000000000..c745cf85e8 --- /dev/null +++ b/ocp_resources/application_aware_applied_cluster_resource_quota.py @@ -0,0 +1,12 @@ +# API reference: https://github.com/kubevirt/application-aware-quota/tree/main/staging/src/kubevirt.io/application-aware-quota-api/pkg/apis/core/v1alpha1 +# TODO: update API reference when OCP doc is available + +from ocp_resources.resource import NamespacedResource + + +class ApplicationAwareAppliedClusterResourceQuota(NamespacedResource): + """ + ApplicationAwareAppliedClusterResourceQuota object. + """ + + api_group = NamespacedResource.ApiGroup.AAQ_KUBEVIRT_IO diff --git a/ocp_resources/application_aware_cluster_resource_quota.py b/ocp_resources/application_aware_cluster_resource_quota.py new file mode 100644 index 0000000000..a565039279 --- /dev/null +++ b/ocp_resources/application_aware_cluster_resource_quota.py @@ -0,0 +1,40 @@ +# API reference: https://github.com/kubevirt/application-aware-quota/tree/main/staging/src/kubevirt.io/application-aware-quota-api/pkg/apis/core/v1alpha1 +# TODO: update API reference when OCP doc is available +from typing import Dict, Any + +from ocp_resources.resource import MissingRequiredArgumentError, Resource + + +class ApplicationAwareClusterResourceQuota(Resource): + api_group = Resource.ApiGroup.AAQ_KUBEVIRT_IO + + def __init__( + self, + quota: Dict[str, Any] | None = None, + selector: Dict[str, Any] | None = None, + **kwargs: Any, + ) -> None: + """ + Create ApplicationAwareClusterResourceQuota object. + + Args: + quota (dict): defines the desired quota + example: {"hard": {"pod": 3, "requests.cpu": "500m"}} + selector (dict): Dict of Namespace labels/annotations to match + example: {"annotations": {"acrq-annotation": "true"}, "labels": {"matchLabels": {"acrq-label": "true"}}} + """ + super().__init__(**kwargs) + self.quota = quota + self.selector = selector + + def to_dict(self): + super().to_dict() + if not self.yaml_file: + if not (self.quota or self.selector): + raise MissingRequiredArgumentError(argument="'quota' or 'selector'") + + self.res["spec"] = {} + _spec = self.res["spec"] + + _spec["quota"] = self.quota + _spec["selector"] = self.selector diff --git a/ocp_resources/application_aware_resource_quota.py b/ocp_resources/application_aware_resource_quota.py new file mode 100644 index 0000000000..20e359b621 --- /dev/null +++ b/ocp_resources/application_aware_resource_quota.py @@ -0,0 +1,49 @@ +# API reference: https://github.com/kubevirt/application-aware-quota/tree/main/staging/src/kubevirt.io/application-aware-quota-api/pkg/apis/core/v1alpha1 +# TODO: update API reference when OCP doc is available +from typing import Dict, Any, List + +from ocp_resources.resource import MissingRequiredArgumentError, NamespacedResource + + +class ApplicationAwareResourceQuota(NamespacedResource): + api_group = NamespacedResource.ApiGroup.AAQ_KUBEVIRT_IO + + def __init__( + self, + hard: Dict[str, Any] | None = None, + scope_selector: Dict[str, Any] | None = None, + scopes: List[str] | None = None, + **kwargs: Any, + ) -> None: + """ + Create ApplicationAwareResourceQuota object. + + Args: + hard (dict): set of desired hard limits + example: {"pod": 3, "requests.cpu": "500m", "requests.memory/vmi": "4Gi", "requests.instances/vmi": 2} + scope_selector (dict, optional): collection of filters + example: {"matchExpressions": [{"operator": "In", "scopeName": "PriorityClass", "values": ["low"]}]} + scopes (list, optional): collection of filters + example: ["Terminating", "PriorityClass"] + """ + super().__init__(**kwargs) + self.hard = hard + self.scope_selector = scope_selector + self.scopes = scopes + + def to_dict(self): + super().to_dict() + if not self.yaml_file: + if not self.hard: + raise MissingRequiredArgumentError(argument="hard") + + self.res["spec"] = {} + _spec = self.res["spec"] + + _spec["hard"] = self.hard + + if self.scope_selector: + _spec["scopeSelector"] = self.scope_selector + + if self.scopes: + _spec["scopes"] = self.scopes