[ROB-1605] Custom Namespaced resource monitoring (#1869)

* partially working

* working version

* updating namespace cache

* changed names

* changes

* refactored

* moved namespace discovery to discovery thread

* removed unneeded discovery

* changed error message

* moved namespace discovery

* cached namespaces

* fixed issues

* loaded metadata from cache

* removed unused imports

* bugfix, namespace metada being removed

* made safer, if added new namespace

Author: Avi-Robusta

src/robusta/core/discovery/discovery.py

@@ -19,6 +19,7 @@
     Volume,
 )
 from kubernetes import client
+from kubernetes.client.exceptions import ApiException
 from kubernetes.client import (
     V1Container,
     V1DaemonSet,
@@ -87,7 +88,24 @@ class Config:
 
 DISCOVERY_STACKTRACE_FILE = "/tmp/make_discovery_stacktrace"
 DISCOVERY_STACKTRACE_TIMEOUT_S = int(os.environ.get("DISCOVERY_STACKTRACE_TIMEOUT_S", 10))
-
+KIND_TO_COREV1_METHOD = {
+    "pods": "list_pod_for_all_namespaces",
+    "configmaps": "list_config_map_for_all_namespaces",
+    "endpoints": "list_endpoints_for_all_namespaces",
+    "services": "list_service_for_all_namespaces",
+    "secrets": "list_secret_for_all_namespaces",
+    "persistentvolumeclaims": "list_persistent_volume_claim_for_all_namespaces",
+    "serviceaccounts": "list_service_account_for_all_namespaces",
+    "replicationcontrollers": "list_replication_controller_for_all_namespaces",
+    "limitranges": "list_limit_range_for_all_namespaces",
+    "resourcequotas": "list_resource_quota_for_all_namespaces",
+    "events": "list_event_for_all_namespaces",
+    "podtemplates": "list_pod_template_for_all_namespaces",
+}
+
+class ResourceAccessForbiddenError(Exception):
+    """Raised when access to a Kubernetes resource is forbidden (HTTP 403)."""
+    pass
 
 class Discovery:
     executor = ProcessPoolExecutor(max_workers=1)  # always 1 discovery process
@@ -184,6 +202,91 @@ def create_service_info_from_hikaru(obj: Union[Deployment, DaemonSet, StatefulSe
             ),
         )
 
+    @staticmethod
+    def count_resources(kind, api_group, version):
+        if not api_group:
+            items = Discovery._fetch_corev1_resources(kind, version)
+        else:
+            items = Discovery._fetch_crd_resources(kind, api_group, version)
+
+        return Discovery._count_items_by_namespace(items, kind)
+
+
+    @staticmethod
+    def _fetch_corev1_resources(kind, version):
+        if version != "v1":
+            logging.error(f"Unsupported CoreV1 resource version '{version}' for kind '{kind}'")
+            return []
+
+        method_name = KIND_TO_COREV1_METHOD.get(kind.lower())
+        if not method_name:
+            logging.warning(f"No CoreV1Api mapping for kind: '{kind}'")
+            return []
+
+        core_v1 = client.CoreV1Api()
+        method = getattr(core_v1, method_name, None)
+        if not method:
+            logging.warning(f"CoreV1Api does not have method '{method_name}' for kind '{kind}'")
+            return []
+
+        try:
+            response = method()
+            return getattr(response, 'items', response)
+        except ApiException as e:
+            if e.status == 403:
+                raise ResourceAccessForbiddenError(f"Access forbidden to CoreV1 resource '{kind}': {e.body}")
+            raise
+
+
+    @staticmethod
+    def _fetch_crd_resources(kind, api_group, version):
+        items = []
+        continue_ref = None
+        for _ in range(DISCOVERY_MAX_BATCHES):
+            try:
+                crd_res = client.CustomObjectsApi().list_cluster_custom_object(
+                    group=api_group,
+                    version=version,
+                    plural=kind.lower(),
+                    limit=DISCOVERY_BATCH_SIZE,
+                    _continue=continue_ref,
+                )
+                items.extend(crd_res.get("items", []))
+                continue_ref = crd_res.get("metadata", {}).get("continue")
+                if not continue_ref:
+                    break
+            except ApiException as e:
+                if e.status == 403:
+                    raise ResourceAccessForbiddenError(
+                        f"Access forbidden to resource '{kind}' in group '{api_group}': {e.body}"
+                    )
+                logging.exception(f"Failed to list {kind} from api group '{api_group}'.")
+                break
+        return items
+
+
+    @staticmethod
+    def _count_items_by_namespace(items, kind):
+        namespace_counts = defaultdict(int)
+        for item in items:
+            metadata = getattr(item, 'metadata', None)
+            namespace = None
+
+            if metadata:
+                namespace = getattr(metadata, 'namespace', None)
+            elif isinstance(item, dict):
+                metadata = item.get('metadata', {})
+                namespace = metadata.get('namespace')
+
+            if not namespace:
+                logging.warning(f"Missing namespace for resource '{kind}': metadata={metadata}")
+                continue
+
+            namespace_counts[namespace] += 1
+
+        return dict(namespace_counts)
+
+
     @staticmethod
     def discovery_process() -> DiscoveryResults:
         create_monkey_patches()

src/robusta/core/model/namespaces.py

@@ -1,9 +1,21 @@
+from typing import Optional, List
 from pydantic import BaseModel
 from kubernetes.client import V1Namespace
+import json
 
+class ResourceCount(BaseModel):
+    kind: str
+    apiVersion: str
+    apiGroup: str
+    count: int
+
+
+class NamespaceMetadata(BaseModel):
+    resources: Optional[List[ResourceCount]]
 
 class NamespaceInfo(BaseModel):
     name: str
+    metadata: Optional[NamespaceMetadata] = None
     deleted: bool = False
 
     @classmethod
@@ -14,7 +26,17 @@ def from_api_server(cls, namespace: V1Namespace) -> "NamespaceInfo":
 
     @classmethod
     def from_db_row(cls, namespace: dict) -> "NamespaceInfo":
+        metadata_raw = namespace.get("metadata")
+
+        if isinstance(metadata_raw, str):
+            metadata_dict = json.loads(metadata_raw)
+        elif isinstance(metadata_raw, dict):
+            metadata_dict = metadata_raw
+        else:
+            metadata_dict = None
+
         return cls(
-            name=namespace["name"],
-            deleted=namespace["deleted"],
+            name=namespace.get("name"),
+            deleted=namespace.get("deleted", False),
+            metadata=NamespaceMetadata(**metadata_dict) if metadata_dict else None
         )

src/robusta/core/sinks/robusta/robusta_sink.py

@@ -9,8 +9,8 @@
 
 import requests
 from hikaru.model.rel_1_26 import DaemonSet, Deployment, Job, Node, Pod, ReplicaSet, StatefulSet
-
-from robusta.core.discovery.discovery import DISCOVERY_STACKTRACE_TIMEOUT_S, Discovery, DiscoveryResults
+from robusta.core.model.namespaces import NamespaceMetadata, ResourceCount
+from robusta.core.discovery.discovery import DISCOVERY_STACKTRACE_TIMEOUT_S, Discovery, DiscoveryResults, ResourceAccessForbiddenError
 from robusta.core.discovery.top_service_resolver import TopLevelResource, TopServiceResolver
 from robusta.core.discovery.utils import from_api_server_node
 from robusta.core.model.base_params import HolmesParams
@@ -51,6 +51,7 @@
 # Define the cache with a single slot and the configured TTL
 _holmes_slackbot_cache = TTLCache(maxsize=1, ttl=HOLMES_SLACKBOT_CACHE_TTL)
 
+
 class RobustaSink(SinkBase, EventHandler):
     services_publish_lock = threading.Lock()
 
@@ -61,6 +62,9 @@ def __init__(self, sink_config: RobustaSinkConfigWrapper, registry):
         self.token = sink_config.robusta_sink.token
         self.ttl_hours = sink_config.robusta_sink.ttl_hours
         self.persist_events = sink_config.robusta_sink.persist_events
+        self.namespace_monitored_resources = sink_config.robusta_sink.namespaceMonitoredResources
+        self.namespace_discovery_seconds = sink_config.robusta_sink.namespace_discovery_seconds
+
         robusta_token = RobustaToken(**json.loads(base64.b64decode(self.token)))
         if self.account_id != robusta_token.account_id:
             logging.error(
@@ -134,7 +138,7 @@ def _on_config_reload(self) -> None:
         if not hasattr(self, '_thread'):
             self._thread = threading.Thread(target=self.__discover_cluster, daemon=True)
             self._thread.start()
-
+        
     def set_cluster_active(self, active: bool):
         self.dal.set_cluster_active(active)
 
@@ -324,6 +328,63 @@ def __send_helm_release_events(self, release_data: List[HelmRelease]):
         except Exception:
             logging.error("Error occurred while sending `helm release` trigger event", exc_info=True)
 
+    def __discover_custom_namespaced_resources(self, namespaces: List[NamespaceInfo]):
+        if not self.namespace_monitored_resources:
+            return
+
+        try:
+            # Step 1: Collect counts in a temporary map
+            resource_map = {}  # type: Dict[str, List[ResourceCount]]
+
+            for resource in self.namespace_monitored_resources:
+                try:
+                    results = Discovery.count_resources(
+                        kind=resource.kind,
+                        api_group=resource.apiGroup,
+                        version=resource.apiVersion
+                    )
+
+                    for namespace_name, count in results.items():
+                        if namespace_name not in resource_map:
+                            resource_map[namespace_name] = []
+
+                        resource_map[namespace_name].append(ResourceCount(
+                            kind=resource.kind,
+                            apiVersion=resource.apiVersion,
+                            apiGroup=resource.apiGroup,
+                            count=count
+                        ))
+
+                except ResourceAccessForbiddenError as e:
+                    logging.warning(f"Skipping resource {resource.kind} due to insufficient permissions: {e}")
+                except Exception as e:
+                    logging.exception(f"Unexpected error counting resource {resource.kind}: {e}")
+
+            # Step 2: Apply metadata to matching NamespaceInfo entries
+            for ns in namespaces:
+                if ns.name in resource_map:
+                    if not ns.metadata:
+                        ns.metadata = NamespaceMetadata(resources=[])
+                    ns.metadata.resources.extend(resource_map[ns.name])
+                    
+            logging.info("Discovered Namespaced custom resources")
+            return namespaces
+
+        except Exception as e:
+            logging.exception(f"Namespace discovery failed: {e}")
+
+    def __add_cached_namespace_metadata(self, namespaces: List[NamespaceInfo]):
+        discovered_namespaces = {namespace.name: namespace for namespace in namespaces}
+        updated_namespaces: List[NamespaceInfo] = []
+
+        for namespace_name, namespace in discovered_namespaces.items():
+            cached_namespace = self.__namespaces_cache.get(namespace_name)
+            if cached_namespace:
+                namespace.metadata = cached_namespace.metadata
+            updated_namespaces.append(namespace)
+        
+        return updated_namespaces
+
     def __discover_resources(self) -> DiscoveryResults:
         # discovery is using the k8s python API and not Hikaru, since it's performance is 10 times better
         try:
@@ -342,7 +403,13 @@ def __discover_resources(self) -> DiscoveryResults:
             self.__publish_new_helm_releases(results.helm_releases)
 
             self.__assert_namespaces_cache_initialized()
-            self.__publish_new_namespaces(results.namespaces)
+            namespaces = results.namespaces
+            if self.namespace_monitored_resources and (time.time() - self.last_namespace_discovery) >= self.namespace_discovery_seconds:
+                namespaces = self.__discover_custom_namespaced_resources(namespaces)
+                self.last_namespace_discovery = time.time()
+            elif self.namespace_monitored_resources:
+                namespaces = self.__add_cached_namespace_metadata(namespaces)
+            self.__publish_new_namespaces(namespaces)
 
             self.__pods_running_count = results.pods_running_count
 
@@ -559,25 +626,26 @@ def __discovery_watchdog(self):
     def __discover_cluster(self):
         logging.info("Cluster discovery initialized")
         get_history = self.__should_run_history()
+        self.last_namespace_discovery = 0
         while self.__active:
             start_t = time.time()
             self.__periodic_cluster_status()
             discovery_results = self.__discover_resources()
+
             if get_history:
                 self.__get_events_history()
                 get_history = False
-
             if discovery_results and discovery_results.helm_releases:
                 self.__send_helm_release_events(release_data=discovery_results.helm_releases)
-
             duration = round(time.time() - start_t)
-            # for small cluster duration is discovery_period_sec. For bigger clusters, up to 5 min
             sleep_dur = min(max(self.__discovery_period_sec, 3 * duration), 300)
+
             logging.debug(f"Discovery duration: {duration} next discovery in {sleep_dur}")
             time.sleep(sleep_dur)
 
         logging.info(f"Service discovery for sink {self.sink_name} ended.")
 
+
     def __periodic_cluster_status(self):
         first_alert = False
 

src/robusta/core/sinks/robusta/robusta_sink_params.py

@@ -1,5 +1,5 @@
 from pydantic.main import BaseModel
-
+from typing import List, Optional
 from robusta.core.sinks.sink_base_params import SinkBaseParams
 from robusta.core.sinks.sink_config import SinkConfigBase
 
@@ -11,12 +11,18 @@ class RobustaToken(BaseModel):
     email: str
     password: str
 
+class NamespaceMonitoredResources(BaseModel):
+    apiGroup: Optional[str] # no group in V1 core resources
+    apiVersion: str
+    kind: str
 
 class RobustaSinkParams(SinkBaseParams):
     token: str
     ttl_hours: int = 4380  # Time before unactive cluster data is deleted. 6 Months default.
     persist_events: bool = False
-
+    namespaceMonitoredResources: Optional[List[NamespaceMonitoredResources]]
+    namespace_discovery_seconds: int = 3600
+    
     @classmethod
     def _get_sink_type(cls):
         return "robusta"