feat: Add threshold support for Correlation Rule (#4793)

Author: VladimirFilonov

keep-ui/app/(keep)/rules/CorrelationSidebar/CorrelationForm.tsx

@@ -264,6 +264,41 @@ export const CorrelationForm = ({
             )}
           />
         </div>
+
+        <div>
+          <label
+            className="flex items-center text-tremor-default font-medium text-tremor-content-strong"
+            htmlFor="threshold"
+          >
+            Alerts threshold{" "}
+          </label>
+
+          <Controller
+            control={control}
+            name="threshold"
+            render={({ field: { value, onChange } }) => (
+              <TextInput
+              type="number"
+              placeholder="1"
+              className="mt-2"
+              {...register("threshold", {
+                required: {
+                  message: "Threshold is required",
+                  value: false,
+                },
+                validate: (value) => {
+                  if (value <= 0) {
+                    return "Threshold should be positive";
+                  }
+                  return true;
+                },
+              })}
+              error={isSubmitted && !!get(errors, "threshold.message")}
+              errorMessage={isSubmitted && get(errors, "threshold.message")}
+            />
+            )}
+          />
+        </div>
       </fieldset>
 
       <div className="flex items-center space-x-2">

keep-ui/app/(keep)/rules/CorrelationSidebar/CorrelationSidebarBody.tsx

@@ -77,6 +77,7 @@ export const CorrelationSidebarBody = ({
       incidentPrefix,
       multiLevel,
       multiLevelPropertyName,
+      threshold,
     } = correlationFormData;
 
     const body = {
@@ -94,6 +95,7 @@ export const CorrelationSidebarBody = ({
       incidentPrefix,
       multiLevel,
       multiLevelPropertyName,
+      threshold,
     };
 
     try {

keep-ui/app/(keep)/rules/CorrelationSidebar/index.tsx

@@ -27,6 +27,7 @@ export const DEFAULT_CORRELATION_FORM_VALUES: CorrelationFormType = {
   incidentPrefix: "",
   multiLevel: false,
   multiLevelPropertyName: "",
+  threshold: 1,
   query: {
     combinator: "or",
     rules: [
@@ -92,6 +93,7 @@ export const CorrelationSidebar = ({
         incidentPrefix: selectedRule.incident_prefix || "",
         multiLevel: selectedRule.multi_level,
         multiLevelPropertyName: selectedRule.multi_level_property_name || "",
+        threshold: selectedRule.threshold || 1,
       };
     }
 

keep-ui/app/(keep)/rules/CorrelationSidebar/types.ts

@@ -14,4 +14,5 @@ export type CorrelationFormType = {
   incidentPrefix: string;
   multiLevel: boolean;
   multiLevelPropertyName?: string;
+  threshold: number;
 };

keep-ui/utils/hooks/useRules.ts

@@ -27,6 +27,7 @@ export type Rule = {
   incident_prefix: string | null;
   multi_level: boolean;
   multi_level_property_name: string | null;
+  threshold: number;
 };
 
 export const useRules = (options?: SWRConfiguration) => {

keep/api/core/db.py

@@ -2156,6 +2156,7 @@ def create_rule(
     incident_prefix=None,
     multi_level=False,
     multi_level_property_name=None,
+    threshold=1,
 ):
     grouping_criteria = grouping_criteria or []
     with Session(engine) as session:
@@ -2177,6 +2178,7 @@ def create_rule(
             incident_prefix=incident_prefix,
             multi_level=multi_level,
             multi_level_property_name=multi_level_property_name,
+            threshold=threshold,
         )
         session.add(rule)
         session.commit()
@@ -2201,6 +2203,7 @@ def update_rule(
     incident_prefix,
     multi_level,
     multi_level_property_name,
+    threshold,
 ):
     rule_uuid = __convert_to_uuid(rule_id)
     if not rule_uuid:
@@ -2227,6 +2230,7 @@ def update_rule(
             rule.incident_prefix = incident_prefix
             rule.multi_level = multi_level
             rule.multi_level_property_name = multi_level_property_name
+            rule.threshold = threshold
             session.commit()
             session.refresh(rule)
             return rule
@@ -2338,7 +2342,7 @@ def create_incident_for_grouping_rule(
             rule_fingerprint=rule_fingerprint,
             is_predicted=True,
             is_candidate=rule.require_approve,
-            is_visible=rule.create_on == CreateIncidentOn.ANY.value,
+            is_visible=False,# rule.create_on == CreateIncidentOn.ANY.value,
             incident_type=IncidentType.RULE.value,
             same_incident_in_the_past_id=past_incident.id if past_incident else None,
             resolve_on=rule.resolve_on,

keep/api/models/db/migrations/versions/2025-05-15-00-34_fcef2c58b21c.py

@@ -0,0 +1,30 @@
+"""Add threshold field to Rule
+
+Revision ID: fcef2c58b21c
+Revises: 7b687c555318
+Create Date: 2025-05-15 00:34:31.753003
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "fcef2c58b21c"
+down_revision = "7b687c555318"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+
+    with op.batch_alter_table("rule", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("threshold", sa.Integer(), nullable=False, server_default="1"))
+        batch_op.create_check_constraint("rule_threshold_positive_int_constraint", "threshold>0")
+
+
+def downgrade() -> None:
+
+    with op.batch_alter_table("rule", schema=None) as batch_op:
+        batch_op.drop_constraint("rule_threshold_positive_int_constraint",  type_="check")
+        batch_op.drop_column("threshold")

keep/api/models/db/rule.py

@@ -2,6 +2,7 @@
 from enum import Enum
 from uuid import UUID, uuid4
 
+from sqlalchemy import CheckConstraint
 from sqlmodel import JSON, Column, Field, SQLModel
 
 # Currently a rule_definition is a list of SQL expressions
@@ -55,3 +56,4 @@ class Rule(SQLModel, table=True):
     incident_prefix: str | None = None
     multi_level: bool = False
     multi_level_property_name: str | None = None
+    threshold: int = Field(sa_column_args=(CheckConstraint("threshold>0"),), default=1)

keep/api/routes/rules.py

@@ -34,6 +34,7 @@ class RuleCreateDto(BaseModel):
     incidentPrefix: str = None
     multiLevel: bool = False
     multiLevelPropertyName: str = None
+    threshold: int = 1
 
 
 @router.get(
@@ -92,6 +93,7 @@ async def create_rule(
     incident_prefix = rule_create_request.incidentPrefix
     multi_level = rule_create_request.multiLevel
     multi_level_property_name = rule_create_request.multiLevelPropertyName
+    threshold = rule_create_request.threshold
 
     if not sql:
         raise HTTPException(status_code=400, detail="SQL is required")
@@ -118,6 +120,9 @@ async def create_rule(
     if not create_on:
         raise HTTPException(status_code=400, detail="createOn is required")
 
+    if not threshold:
+        raise HTTPException(status_code=400, detail="threshold is required")
+
     rule = create_rule_db(
         tenant_id=tenant_id,
         name=rule_name,
@@ -138,6 +143,7 @@ async def create_rule(
         incident_prefix=incident_prefix,
         multi_level=multi_level,
         multi_level_property_name=multi_level_property_name,
+        threshold=threshold,
     )
     logger.info("Rule created")
     return rule
@@ -193,6 +199,7 @@ async def update_rule(
         incident_prefix = body.get("incidentPrefix", None)
         multi_level = body.get("multiLevel", False)
         multi_level_property_name = body.get("multiLevelPropertyName", None)
+        threshold = body.get("threshold", 1)
     except Exception:
         raise HTTPException(status_code=400, detail="Invalid request body")
 
@@ -225,6 +232,9 @@ async def update_rule(
     if not create_on:
         raise HTTPException(status_code=400, detail="createOn is required")
 
+    if not threshold:
+        raise HTTPException(status_code=400, detail="threshold is required")
+
     rule = update_rule_db(
         tenant_id=tenant_id,
         rule_id=rule_id,
@@ -245,6 +255,7 @@ async def update_rule(
         incident_prefix=incident_prefix,
         multi_level=multi_level,
         multi_level_property_name=multi_level_property_name,
+        threshold=threshold,
     )
 
     if rule:

keep/rulesengine/rulesengine.py

@@ -138,22 +138,24 @@ def _run_cel_rules(
                                 rule_groups = self._extract_subrules(
                                     rule.definition_cel
                                 )
-
-                                if not rule.require_approve:
-                                    if rule.create_on == "any" or (
-                                        rule.create_on == "all"
-                                        and len(rule_groups) == len(matched_rules)
-                                    ):
-                                        self.logger.info(
-                                            "Single event is enough, so creating incident"
-                                        )
-                                        incident.is_visible = True
-                                    elif rule.create_on == "all":
-                                        incident = (
-                                            self._process_event_for_history_based_rule(
-                                                incident, rule, session
+                                firing_count = sum([alert.event.get("firingCounter", 1) for alert in incident.alerts])
+                                alerts_count = max(incident.alerts_count, firing_count)
+                                if alerts_count >= rule.threshold:
+                                    if not rule.require_approve:
+                                        if rule.create_on == "any" or (
+                                            rule.create_on == "all"
+                                            and len(rule_groups) == len(matched_rules)
+                                        ):
+                                            self.logger.info(
+                                                "Single event is enough, so creating incident"
+                                            )
+                                            incident.is_visible = True
+                                        elif rule.create_on == "all":
+                                            incident = (
+                                                self._process_event_for_history_based_rule(
+                                                    incident, rule, session
+                                                )
                                             )
-                                        )
 
                                 send_created_event = incident.is_visible
 
@@ -266,9 +268,9 @@ def _get_or_create_incident(
                 },
             )
             alerts = existed_incident.alerts
-            vairables = self.get_vaiables(rule.incident_name_template)
+            variables = self.get_vaiables(rule.incident_name_template)
             values = set()
-            for var in vairables:
+            for var in variables:
                 var_to_replace = ""
                 alerts_dtos = convert_db_alerts_to_dto_alerts(alerts)
                 for alert in alerts_dtos:
@@ -306,13 +308,13 @@ def _get_or_create_incident(
         if event.status == AlertStatus.FIRING.value:
             if rule.incident_name_template:
                 incident_name = copy.copy(rule.incident_name_template)
-                vairables = self.get_vaiables(rule.incident_name_template)
-                if not vairables:
+                variables = self.get_vaiables(rule.incident_name_template)
+                if not variables:
                     self.logger.warning(
                         f"Failed to fetch the appropriate labels from the event {event.id} and rule {rule.name}"
                     )
                     incident_name = None
-                for var in vairables:
+                for var in variables:
                     value = self.get_value_from_event(event, var)
                     pattern = r"\{\{\s*" + re.escape(var) + r"\s*\}\}"
                     incident_name = re.sub(pattern, value, incident_name)