feat: add support for StrimziPodSet (#388)

fixes #372

---------

Signed-off-by: Saketh kappala <[email protected]>
Co-authored-by: Avi-Robusta <[email protected]>

Author: sappusaketh

robusta_krr/core/integrations/kubernetes/__init__.py

@@ -101,6 +101,7 @@ async def list_scannable_objects(self) -> list[K8sObjectData]:
         workload_object_lists = await asyncio.gather(
             self._list_deployments(),
             self._list_rollouts(),
+            self._list_strimzipodsets(),
             self._list_deploymentconfig(),
             self._list_all_statefulsets(),
             self._list_all_daemon_set(),
@@ -204,7 +205,10 @@ def __build_scannable_object(
         labels = {}
         annotations = {}
         if item.metadata.labels:
-            labels = item.metadata.labels
+            if type(item.metadata.labels) is ObjectLikeDict:
+                labels = item.metadata.labels.__dict__
+            else:  
+                labels = item.metadata.labels
 
         if item.metadata.annotations:
             if type(item.metadata.annotations) is ObjectLikeDict:
@@ -299,7 +303,7 @@ async def _list_scannable_objects(
 
                 result.extend(self.__build_scannable_object(item, container, kind) for container in containers)
         except ApiException as e:
-            if kind in ("Rollout", "DeploymentConfig") and e.status in [400, 401, 403, 404]:
+            if kind in ("Rollout", "DeploymentConfig", "StrimziPodSet") and e.status in [400, 401, 403, 404]:
                 if self.__kind_available[kind]:
                     logger.debug(f"{kind} API not available in {self.cluster}")
                 self.__kind_available[kind] = False
@@ -364,6 +368,30 @@ async def _extract_containers(item: Any) -> list[V1Container]:
             extract_containers=_extract_containers,
         )
 
+    def _list_strimzipodsets(self) -> list[K8sObjectData]:
+        # NOTE: Using custom objects API returns dicts, but all other APIs return objects
+        # We need to handle this difference using a small wrapper
+        return self._list_scannable_objects(
+            kind="StrimziPodSet",
+            all_namespaces_request=lambda **kwargs: ObjectLikeDict(
+                self.custom_objects.list_cluster_custom_object(
+                    group="core.strimzi.io",
+                    version="v1beta2",
+                    plural="strimzipodsets",
+                    **kwargs,
+                )
+            ),
+            namespaced_request=lambda **kwargs: ObjectLikeDict(
+                self.custom_objects.list_namespaced_custom_object(
+                    group="core.strimzi.io",
+                    version="v1beta2",
+                    plural="strimzipodsets",
+                    **kwargs,
+                )
+            ),
+            extract_containers=lambda item: item.spec.pods[0].spec.containers,
+        )
+
     def _list_deploymentconfig(self) -> list[K8sObjectData]:
         # NOTE: Using custom objects API returns dicts, but all other APIs return objects
         # We need to handle this difference using a small wrapper

robusta_krr/core/models/objects.py

@@ -8,7 +8,7 @@
 from robusta_krr.utils.batched import batched
 from kubernetes.client.models import V1LabelSelector
 
-KindLiteral = Literal["Deployment", "DaemonSet", "StatefulSet", "Job", "CronJob", "Rollout", "DeploymentConfig"]
+KindLiteral = Literal["Deployment", "DaemonSet", "StatefulSet", "Job", "CronJob", "Rollout", "DeploymentConfig", "StrimziPodSet"]
 
 
 class PodData(pd.BaseModel):

robusta_krr/main.py

@@ -98,7 +98,7 @@ def run_strategy(
                     None,
                     "--resource",
                     "-r",
-                    help="List of resources to run on (Deployment, StatefulSet, DaemonSet, Job, Rollout). By default, will run on all resources. Case insensitive.",
+                    help="List of resources to run on (Deployment, StatefulSet, DaemonSet, Job, Rollout, StrimziPodSet). By default, will run on all resources. Case insensitive.",
                     rich_help_panel="Kubernetes Settings",
                 ),
                 selector: Optional[str] = typer.Option(