Add ApplicationAwareQuota resources (#1821)

Configurable resources: ApplicationAwareResourceQuota, ApplicationAwareClusterResourceQuota Non-configurable: ApplicationAwareClusterResourceQuota Signed-off-by: dshchedr <[email protected]> Co-authored-by: Meni Yakove <[email protected]>
RedHatQE · May 30, 2024 · a8d9633 · a8d9633
1 parent 83bc748
commit a8d9633
Show file tree

Hide file tree

Showing 3 changed files with 101 additions and 0 deletions.
diff --git a/ocp_resources/application_aware_applied_cluster_resource_quota.py 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
@@ -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
@@ -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