feat(namespace): regex match (#336)

This MR will add ability in selecting some namespaces that will be
scanned by using regex pattern

README.md

@@ -307,6 +307,12 @@ List as many namespaces as you want with `-n` (in this case, `default` and `ingr
 krr simple -n default -n ingress-nginx
 ```
 
+The -n flag also supports regex matches like -n kube-.*. To use regexes, you must have permissions to list namespaces in the target cluster.
+
+```sh
+krr simple -n default -n 'ingress-.*'
+```
+
 See [example ServiceAccount and RBAC permissions](./tests/single_namespace_permissions.yaml)
 </details>
 

robusta_krr/core/integrations/kubernetes/__init__.py

@@ -1,8 +1,9 @@
 import asyncio
 import logging
+import re
 from collections import defaultdict
 from concurrent.futures import ThreadPoolExecutor
-from typing import Any, Awaitable, Callable, Iterable, Optional, Union
+from typing import Any, Awaitable, Callable, Iterable, Optional, Union, Literal
 
 from kubernetes import client, config  # type: ignore
 from kubernetes.client import ApiException
@@ -47,6 +48,43 @@ def __init__(self, cluster: Optional[str]=None):
 
         self.__jobs_for_cronjobs: dict[str, list[V1Job]] = {}
         self.__jobs_loading_locks: defaultdict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
+        self.__namespaces: Union[list[str, None]] = None
+
+    @property
+    def namespaces(self) -> Union[list[str], Literal["*"]]:
+        """wrapper for settings.namespaces, which will do expand namespace list if some regex pattern included
+
+        Returns:
+            A list of namespace that will be scanned
+        """
+        # just return the list if it's already initialized
+        if self.__namespaces != None:
+            return self.__namespaces
+
+        setting_ns = settings.namespaces
+        if setting_ns == "*":
+            self.__namespaces = setting_ns
+            return self.__namespaces
+
+        self.__namespaces = []
+        expand_list: list[re.Pattern] = []
+        ns_regex_chars = re.compile(r"[\\*|\(.*?\)|\[.*?\]|\^|\$]")
+        for ns in setting_ns:
+            if ns_regex_chars.search(ns):
+                logger.debug(f"{ns} is detected as regex pattern in expanding namespace list")
+                expand_list.append(re.compile(ns))
+            else:
+                self.__namespaces.append(ns)
+
+        if expand_list:
+            logger.info("found regex pattern in provided namespace argument, expanding namespace list")
+            all_ns = [ ns.metadata.name for ns in self.core.list_namespace().items ]
+            for expand_ns in expand_list:
+                for ns in all_ns:
+                    if expand_ns.fullmatch(ns) and ns not in self.__namespaces:
+                        self.__namespaces.append(ns)
+
+        return self.__namespaces
 
     async def list_scannable_objects(self) -> list[K8sObjectData]:
         """List all scannable objects.
@@ -56,7 +94,7 @@ async def list_scannable_objects(self) -> list[K8sObjectData]:
         """
 
         logger.info(f"Listing scannable objects in {self.cluster}")
-        logger.debug(f"Namespaces: {settings.namespaces}")
+        logger.debug(f"Namespaces: {self.namespaces}")
         logger.debug(f"Resources: {settings.resources}")
 
         self.__hpa_list = await self._try_list_hpa()
@@ -75,7 +113,7 @@ async def list_scannable_objects(self) -> list[K8sObjectData]:
             for workload_objects in workload_object_lists
             for object in workload_objects
             # NOTE: By default we will filter out kube-system namespace
-            if not (settings.namespaces == "*" and object.namespace == "kube-system")
+            if not (self.namespaces == "*" and object.namespace == "kube-system")
         ]
 
     async def _list_jobs_for_cronjobs(self, namespace: str) -> list[V1Job]:
@@ -189,7 +227,7 @@ async def _list_namespaced_or_global_objects(
         logger.debug(f"Listing {kind}s in {self.cluster}")
         loop = asyncio.get_running_loop()
 
-        if settings.namespaces == "*":
+        if self.namespaces == "*":
             requests = [
                 loop.run_in_executor(
                     self.executor,
@@ -209,7 +247,7 @@ async def _list_namespaced_or_global_objects(
                         label_selector=settings.selector,
                     ),
                 )
-                for namespace in settings.namespaces
+                for namespace in self.namespaces
             ]
 
         result = [

tests/test_krr.py

@@ -1,7 +1,11 @@
 import pytest
+from typing import Literal, Union
+from unittest.mock import patch, Mock, MagicMock
 from typer.testing import CliRunner
 
 from robusta_krr.main import app, load_commands
+from robusta_krr.core.integrations.kubernetes import ClusterLoader
+from robusta_krr.core.models.config import settings
 
 runner = CliRunner(mix_stderr=False)
 load_commands()
@@ -34,3 +38,42 @@ def test_output_formats(format: str, output: str):
         assert result.exit_code == 0, result.exc_info
     except AssertionError as e:
         raise e from result.exception
+
+@pytest.mark.parametrize(
+        "setting_namespaces,cluster_all_ns,expected",[
+            (
+                # default settings
+                "*",
+                ["kube-system", "robusta-frontend", "robusta-backend", "infra-grafana"],
+                "*"
+            ),
+            (
+                # list of namespace provided from arguments without regex pattern
+                ["robusta-krr", "kube-system"],
+                ["kube-system", "robusta-frontend", "robusta-backend", "robusta-krr"],
+                ["robusta-krr", "kube-system"]
+            ),
+            (
+                # list of namespace provided from arguments with regex pattern and will not duplicating in final result
+                ["robusta-.*", "robusta-frontend"],
+                ["kube-system", "robusta-frontend", "robusta-backend", "robusta-krr"],
+                ["robusta-frontend", "robusta-backend", "robusta-krr"]
+            ),
+            (
+                # namespace provided with regex pattern and will match for some namespaces
+                [".*end$"],
+                ["kube-system", "robusta-frontend", "robusta-backend", "robusta-krr"],
+                ["robusta-frontend", "robusta-backend"]
+            )
+        ]
+    )
+def test_cluster_namespace_list(
+        setting_namespaces: Union[Literal["*"], list[str]],
+        cluster_all_ns: list[str],
+        expected: Union[Literal["*"], list[str]],
+    ):
+    cluster = ClusterLoader()
+    with patch("robusta_krr.core.models.config.settings.namespaces", setting_namespaces):
+        with patch.object(cluster.core, "list_namespace", return_value=MagicMock(
+            items=[MagicMock(**{"metadata.name": m}) for m in cluster_all_ns])):
+            assert sorted(cluster.namespaces) == sorted(expected)