diff --git a/src/robusta/core/discovery/resource_names.py b/src/robusta/core/discovery/resource_names.py index 86a77cce1..c16a651c1 100644 --- 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 index 7fed38fb4..d929b9c36 100644 --- 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):