add functions to support list resource names (#1352)

robusta-dev · Mar 19, 2024 · dfb4f28 · dfb4f28
1 parent 5d97f2a
commit dfb4f28
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 0 deletions.
diff --git a/src/robusta/core/discovery/resource_names.py b/src/robusta/core/discovery/resource_names.py
@@ -3,6 +3,7 @@
 from kubernetes import client
 from pydantic import BaseModel
 
+from robusta.integrations.kubernetes.custom_models import DeploymentConfig
 from robusta.utils.error_codes import ActionException, ErrorCodes
 
 
@@ -54,6 +55,10 @@ class ResourceLister(BaseModel):
         list_all=client.NetworkingV1Api().list_ingress_for_all_namespaces,
         list_namespaced=client.NetworkingV1Api().list_namespaced_ingress,
     ),
+    "deploymentconfig": ResourceLister(
+        list_all=DeploymentConfig.list_for_all_namespaces,
+        list_namespaced=DeploymentConfig.list_namespaced,
+    ),
 }
 
 

diff --git a/src/robusta/integrations/kubernetes/custom_models.py b/src/robusta/integrations/kubernetes/custom_models.py
@@ -586,6 +586,37 @@ class DeploymentConfig(HikaruDocumentBase, HikaruCRDDocumentMixin):
     def readNamespaced(self, name: str, namespace: str):
         obj = DeploymentConfig(metadata=ObjectMeta(name=name, namespace=namespace)).read()
         return type("", (object,), {"obj": obj})()
+
+    @classmethod
+    def list_namespaced(self, namespace: str):
+        deployconfigs_res = client.CustomObjectsApi().list_namespaced_custom_object(
+            group=DeploymentConfig.group,
+            version=DeploymentConfig.version,
+            namespace=namespace,
+            plural=DeploymentConfig.plural,
+        ) 
+        dc_list = type("", (object,), {"items": [
+            DeploymentConfig(metadata=ObjectMeta(**dc.get("metadata", {})), spec=DeploymentConfigSpec(**dc.get("spec", {})))
+            for dc in
+            deployconfigs_res.get("items", [])
+        ]})()
+
+        return dc_list
+
+    @classmethod
+    def list_for_all_namespaces(self):
+        deployconfigs_res = client.CustomObjectsApi().list_cluster_custom_object(
+            group=DeploymentConfig.group,
+            version=DeploymentConfig.version,
+            plural=DeploymentConfig.plural,
+        ) 
+        dc_list = type("", (object,), {"items": [
+            DeploymentConfig(metadata=ObjectMeta(**dc.get("metadata", {})), spec=DeploymentConfigSpec(**dc.get("spec", {})))
+            for dc in
+            deployconfigs_res.get("items", [])
+        ]})()
+
+        return dc_list
 
 
 def DictToK8sObj(obj: Dict, class_name):