Added prometheus alert RRM

helm/robusta/templates/runner-service-account.yaml

@@ -44,6 +44,20 @@ rules:
       - watch
       - patch
 
+  {{- if .Values.monitorHelmReleases }}
+  - apiGroups:
+      - "monitoring.coreos.com"
+    resources:
+      - prometheusrules
+    verbs:
+      - get
+      - list
+      - delete
+      - create
+      - patch
+      - update
+  {{end}}
+
   - apiGroups:
       - ""
     resources:

helm/robusta/templates/runner.yaml

@@ -52,6 +52,8 @@ spec:
             value: {{ .Release.Name | quote }}
           - name: PROMETHEUS_ENABLED
             value: {{ .Values.enablePrometheusStack | quote}}
+          - name: MANAGED_PROMETHEUS_ALERTS_ENABLED
+            value: {{ .Values.enableManagedPrometheusAlerts | quote}}
           - name: SEND_ADDITIONAL_TELEMETRY
             value: {{ .Values.runner.sendAdditionalTelemetry | quote }}
           - name: LOG_LEVEL

helm/robusta/values.yaml

@@ -74,6 +74,7 @@ lightActions:
 
 # install prometheus, alert-manager, and grafana along with Robusta?
 enablePrometheusStack: false
+enableManagedPrometheusAlerts: false
 enableServiceMonitors: true
 monitorHelmReleases: true
 

playbooks/robusta_playbooks/k8s_resource_enrichments.py

@@ -90,7 +90,7 @@ def to_pod_row(pod: Pod, cluster_name: str) -> List:
     ]
 
 
-def get_related_pods(resource) -> list[Pod]:
+def get_related_pods(resource) -> List[Pod]:
     kind: str = resource.kind or ""
     if kind not in supported_resources:
         raise ActionException(ErrorCodes.RESOURCE_NOT_SUPPORTED, f"Related pods is not supported for resource {kind}")

src/robusta/core/discovery/discovery.py

@@ -362,7 +362,7 @@ def discovery_process() -> DiscoveryResults:
                             # we use map here to deduplicate and pick only the latest release data
                             helm_releases_map[decoded_release_row.get_service_key()] = decoded_release_row
                         except Exception as e:
-                            logging.error(f"an error occured while decoding helm releases: {e}")
+                            logging.error(f"an error occurred while decoding helm releases: {e}")
 
                     continue_ref = secrets.metadata._continue
                     if not continue_ref:

src/robusta/core/model/cluster_status.py

@@ -1,3 +1,5 @@
+from datetime import datetime
+
 from pydantic.main import BaseModel, List, Optional
 
 
@@ -18,6 +20,7 @@ class ActivityStats(BaseModel):
     alertManagerConnection: bool
     prometheusConnection: bool
     prometheusRetentionTime: str
+    managedPrometheusAlerts: bool
 
 
 class ClusterStatus(BaseModel):
@@ -29,3 +32,11 @@ class ClusterStatus(BaseModel):
     ttl_hours: int
     stats: ClusterStats
     activity_stats: ActivityStats
+
+
+class Account(BaseModel):
+    id: str
+    account_type: int
+    is_test_account: bool
+    name: Optional[str]
+    creation_date: datetime

src/robusta/core/model/env_vars.py

@@ -41,6 +41,7 @@ def load_bool(env_var, default: bool):
 
 PROMETHEUS_REQUEST_TIMEOUT_SECONDS = float(os.environ.get("PROMETHEUS_REQUEST_TIMEOUT_SECONDS", 90.0))
 PROMETHEUS_ENABLED = os.environ.get("PROMETHEUS_ENABLED", "false").lower() == "true"
+MANAGED_PROMETHEUS_ALERTS_ENABLED = os.environ.get("MANAGED_PROMETHEUS_ALERTS_ENABLED", "false").lower() == "true"
 PROMETHEUS_SSL_ENABLED = os.environ.get("PROMETHEUS_SSL_ENABLED", "false").lower() == "true"
 
 INCOMING_REQUEST_TIME_WINDOW_SECONDS = int(os.environ.get("INCOMING_REQUEST_TIME_WINDOW_SECONDS", 3600))
@@ -95,4 +96,8 @@ def load_bool(env_var, default: bool):
 
 PROMETHEUS_ERROR_LOG_PERIOD_SEC = int(os.environ.get("DISCOVERY_MAX_BATCHES", 14400))
 
+RRM_PERIOD_SEC = int(os.environ.get("RRM_PERIOD_SEC", 90))
+
+MAX_ALLOWED_RULES_PER_CRD_ALERT = int(os.environ.get("MAX_ALLOWED_RULES_PER_CRD_ALERT", 600))
+
 IMAGE_REGISTRY = os.environ.get("IMAGE_REGISTRY", "us-central1-docker.pkg.dev/genuine-flight-317411/devel")

src/robusta/core/sinks/pagerduty/pagerduty_sink.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Dict, Any, Optional
+from typing import Dict, Any, Optional, List
 
 import requests
 
@@ -233,7 +233,7 @@ def __to_unformatted_text_for_alerts(block: BaseBlock) -> str:
         return ""
 
     @staticmethod
-    def __to_unformatted_text_for_changes(block: KubernetesDiffBlock) -> Optional[list[str]]:
+    def __to_unformatted_text_for_changes(block: KubernetesDiffBlock) -> Optional[List[str]]:
         return list(map(
             lambda diff: diff.formatted_path,
             block.diffs,

src/robusta/core/sinks/robusta/dal/supabase_dal.py

@@ -1,14 +1,16 @@
 import json
 import logging
 import time
-from typing import Any, Dict, List
-
+from datetime import datetime
+from typing import Any, Dict, List, Optional
 import requests
+from postgrest.base_request_builder import BaseFilterRequestBuilder
+from postgrest.utils import sanitize_param
 from postgrest.types import ReturnMethod
 from supabase import create_client
 from supabase.lib.client_options import ClientOptions
 
-from robusta.core.model.cluster_status import ClusterStatus
+from robusta.core.model.cluster_status import ClusterStatus, Account
 from robusta.core.model.env_vars import SUPABASE_LOGIN_RATE_LIMIT_SEC, SUPABASE_TIMEOUT_SECONDS
 from robusta.core.model.helm_release import HelmRelease
 from robusta.core.model.jobs import JobInfo
@@ -19,8 +21,10 @@
 from robusta.core.reporting.base import Finding
 from robusta.core.reporting.blocks import EventsBlock, EventsRef, ScanReportBlock, ScanReportRow
 from robusta.core.reporting.consts import EnrichmentAnnotation
-from robusta.core.sinks.robusta import RobustaSinkParams
 from robusta.core.sinks.robusta.dal.model_conversion import ModelConversion
+from robusta.core.sinks.robusta.rrm.account_resource_fetcher import AccountResourceFetcher
+from robusta.core.sinks.robusta.rrm.types import AccountResource, ResourceKind, \
+    AccountResourceStatusType, AccountResourceStatusInfo
 
 SERVICES_TABLE = "Services"
 NODES_TABLE = "Nodes"
@@ -33,19 +37,23 @@
 UPDATE_CLUSTER_NODE_COUNT = "update_cluster_node_count"
 SCANS_RESULT_TABLE = "ScansResults"
 RESOURCE_EVENTS = "ResourceEvents"
+ACCOUNT_RESOURCE_TABLE = "AccountResource"
+ACCOUNT_RESOURCE_STATUS_TABLE = "AccountResourceStatus"
+ACCOUNTS_TABLE = "Accounts"
 
 
-class SupabaseDal:
+class SupabaseDal(AccountResourceFetcher):
     def __init__(
-        self,
-        url: str,
-        key: str,
-        account_id: str,
-        email: str,
-        password: str,
-        sink_params: RobustaSinkParams,
-        cluster_name: str,
-        signing_key: str,
+            self,
+            url: str,
+            key: str,
+            account_id: str,
+            email: str,
+            password: str,
+            sink_name: str,
+            persist_events: bool,
+            cluster_name: str,
+            signing_key: str,
     ):
         self.url = url
         self.key = key
@@ -58,7 +66,8 @@ def __init__(
         self.sign_in_time = 0
         self.sign_in()
         self.client.auth.on_auth_state_change(self.__update_token_patch)
-        self.sink_params = sink_params
+        self.sink_name = sink_name
+        self.persist_events = persist_events
         self.signing_key = signing_key
 
     def __to_db_scanResult(self, scanResult: ScanReportRow) -> Dict[Any, Any]:
@@ -111,7 +120,7 @@ def persist_finding(self, finding: Finding):
             evidence = ModelConversion.to_evidence_json(
                 account_id=self.account_id,
                 cluster_id=self.cluster,
-                sink_name=self.sink_params.name,
+                sink_name=self.sink_name,
                 signing_key=self.signing_key,
                 finding_id=finding.id,
                 enrichment=enrichment,
@@ -172,15 +181,8 @@ def get_active_services(self) -> List[ServiceInfo]:
             res = (
                 self.client.table(SERVICES_TABLE)
                 .select(
-                    "name",
-                    "type",
-                    "namespace",
-                    "classification",
-                    "config",
-                    "ready_pods",
-                    "total_pods",
-                    "is_helm_release",
-                )
+                    "name", "type", "namespace", "classification", "config", "ready_pods", "total_pods",
+                    "is_helm_release")
                 .filter("account_id", "eq", self.account_id)
                 .filter("cluster", "eq", self.cluster)
                 .filter("deleted", "eq", False)
@@ -280,6 +282,14 @@ def publish_nodes(self, nodes: List[NodeInfo]):
             self.handle_supabase_error()
             raise
 
+    @staticmethod
+    def custom_filter_request_builder(frq: BaseFilterRequestBuilder, operator: str,
+                                      criteria: str) -> BaseFilterRequestBuilder:
+        key, val = sanitize_param(operator), f"{criteria}"
+        frq.params = frq.params.set(key, val)
+
+        return frq
+
     def get_active_jobs(self) -> List[JobInfo]:
         try:
             res = (
@@ -495,13 +505,93 @@ def persist_events_block(self, block: EventsBlock):
     def persist_platform_blocks(self, enrichment: Enrichment, finding_id):
         blocks = enrichment.blocks
         for i, block in enumerate(blocks):
-            if isinstance(block, EventsBlock) and self.sink_params.persist_events and block.events:
+            if isinstance(block, EventsBlock) and self.persist_events and block.events:
                 self.persist_events_block(block)
                 event = block.events[0]
                 blocks[i] = EventsRef(name=event.name, namespace=event.namespace, kind=event.kind.lower())
             if isinstance(block, ScanReportBlock):
                 self.persist_scan(block)
 
+    def get_account_resources(
+            self, resource_kind: ResourceKind, updated_at: Optional[datetime]
+    ) -> List[AccountResource]:
+        try:
+            query_builder = (
+                self.client.table(ACCOUNT_RESOURCE_TABLE)
+                .select("entity_id", "resource_kind", "clusters_target_set", "resource_state", "deleted", "enabled",
+                        "updated_at")
+                .filter("resource_kind", "eq", resource_kind)
+                .filter("account_id", "eq", self.account_id)
+            )
+            if updated_at:
+                query_builder.gt("updated_at", updated_at.isoformat())
+            else:
+                # in the initial db fetch don't include the deleted records.
+                # in the subsequent db fetch allow even the deleted records so that they can be removed from the cluster
+                query_builder.filter("deleted", "eq", False)
+                query_builder.filter("enabled", "eq", True)
+
+                query_builder = SupabaseDal.custom_filter_request_builder(
+                    query_builder,
+                    operator="or",
+                    criteria=f'(clusters_target_set.cs.["*"], clusters_target_set.cs.["{self.cluster}"])',
+                )
+            query_builder = query_builder.order(column="updated_at", desc=False)
+
+            res = query_builder.execute()
+        except Exception as e:
+            msg = f"Failed to get existing account resources (supabase) error: {e}"
+            logging.error(msg)
+            self.handle_supabase_error()
+            raise Exception(msg)
+
+        account_resources: List[AccountResource] = []
+        for data in res.data:
+            resource_state = data["resource_state"]
+            resource = AccountResource(
+                entity_id=data["entity_id"],
+                resource_kind=data["resource_kind"],
+                clusters_target_set=data["clusters_target_set"],
+                resource_state=resource_state,
+                deleted=data["deleted"],
+                enabled=data["enabled"],
+                updated_at=data["updated_at"],
+            )
+
+            account_resources.append(resource)
+
+        return account_resources
+
+    def __to_db_account_resource_status(self,
+                                        status_type: Optional[AccountResourceStatusType],
+                                        info: Optional[AccountResourceStatusInfo]) -> Dict[Any, Any]:
+
+        data = {
+            "account_id": self.account_id,
+            "cluster_id": self.cluster,
+            "status": status_type,
+            "info": info.dict() if info else None,
+            "updated_at": "now()",
+            "latest_revision": "now()"}
+
+        if status_type != AccountResourceStatusType.error:
+            data["synced_revision"] = "now()"
+
+        return data
+
+    def set_account_resource_status(
+            self, status_type: Optional[AccountResourceStatusType],
+            info: Optional[AccountResourceStatusInfo]
+    ):
+        try:
+            data = self.__to_db_account_resource_status(status_type=status_type, info=info)
+
+            self.client.table(ACCOUNT_RESOURCE_STATUS_TABLE).upsert(data).execute()
+        except Exception as e:
+            logging.error(f"Failed to persist resource events error: {e}")
+            self.handle_supabase_error()
+            raise
+
     def __rpc_patch(self, func_name: str, params: dict) -> Dict[str, Any]:
         """
         Supabase client is async. Sync impl of rpc call