fix(rule-flood-guard): scope flood detection and rule disabling per tenant (#2495)

Author: yllada

plugins/rule-flood-guard/README.md

@@ -8,6 +8,23 @@ the notification bell, suggesting they mark the alerts as false positives
 
 It runs as its own independent plugin, separate from `plugins/alerts`.
 
+## Multitenancy
+
+Everything is scoped per tenant:
+
+- **Counting.** Alerts are grouped by tenant, rule name and data source. The
+  threshold is compared against one tenant's own volume, so two tenants that
+  are each below it never add up to a flood between them.
+- **Disabling.** The rule is disabled only for the tenant that flooded. Every
+  other tenant keeps it running.
+- **Notifying.** The offending tenant is notified, since the disable applies to
+  them and the remediation is theirs to apply. The platform tenant receives a
+  copy so the operator keeps instance-wide visibility — unless it is already
+  the offending tenant, in which case a single notification is sent.
+
+Alerts that carry no tenant are dropped rather than attributed to a default
+tenant.
+
 ## Configuration
 
 Settings live in `system_plugins_rule-flood-guard.yaml`, in the same pipeline
@@ -26,7 +43,7 @@ plugins:
 | Field | Default | Meaning |
 |---|---|---|
 | `enabled` | `true` | Turns the guard on or off. |
-| `threshold` | `50` | How many alerts from the same rule and data source trigger the auto-disable. |
+| `threshold` | `50` | How many alerts from the same rule and data source, within a single tenant, trigger the auto-disable. |
 | `windowHours` | `24` | Time window used to count alerts. |
 | `intervalSeconds` | `300` | How often the guard checks. |
 

plugins/rule-flood-guard/aggregation.go

@@ -14,11 +14,18 @@ const (
 )
 
 type ruleBucket struct {
+	TenantID   string
 	RuleName   string
 	DataSource string
 	Count      int64
 }
 
+// floodGroupFields is the grouping key. tenantId is not a grouping detail, it
+// is the isolation: the driver renders these as a literal GROUP BY, so dropping
+// it sums every tenant's alerts into one bucket and disables a rule on volume
+// no single customer produced.
+var floodGroupFields = []string{"tenantId", "name", "dataSource"}
+
 func floodFilters() []store.Filter {
 	return []store.Filter{
 		{Field: "status", Op: store.OpEq, Value: statusOpen},
@@ -30,30 +37,52 @@ func floodFilters() []store.Filter {
 func searchRuleBuckets(ctx context.Context, window time.Duration) ([]ruleBucket, error) {
 	now := time.Now().UTC()
 	scope := store.Scope{
+		// One pass has to see the whole instance. Isolation comes from grouping
+		// on tenantId below, not from narrowing this read — do not "fix" this
+		// to a single tenant.
 		Tenant:  store.AllTenants,
 		Dataset: datasetAlerts,
 		From:    now.Add(-window),
 		To:      now,
 	}
 
 	groups, err := alertStore.GroupBy(ctx, scope,
-		[]string{"name", "dataSource"},
+		floodGroupFields,
 		floodFilters(),
 		store.GroupOpts{Limit: maxCombinations},
 	)
 	if err != nil {
 		return nil, err
 	}
+
+	return bucketsFromGroups(groups), nil
+}
+
+func bucketsFromGroups(groups []store.Group) []ruleBucket {
 	buckets := make([]ruleBucket, 0, len(groups))
-	for _, byName := range groups {
-		for _, bySource := range byName.Children {
-			buckets = append(buckets, ruleBucket{
-				RuleName:   byName.Key,
-				DataSource: bySource.Key,
-				Count:      bySource.Count,
-			})
+
+	var walk func(level []store.Group, path map[string]string)
+	walk = func(level []store.Group, path map[string]string) {
+		for _, g := range level {
+			next := make(map[string]string, len(path)+1)
+			for k, v := range path {
+				next[k] = v
+			}
+			next[g.Field] = g.Key
+
+			if len(g.Children) == 0 {
+				buckets = append(buckets, ruleBucket{
+					TenantID:   next["tenantId"],
+					RuleName:   next["name"],
+					DataSource: next["dataSource"],
+					Count:      g.Count,
+				})
+				continue
+			}
+			walk(g.Children, next)
 		}
 	}
+	walk(groups, map[string]string{})
 
-	return buckets, nil
+	return buckets
 }

plugins/rule-flood-guard/backend.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -14,7 +15,14 @@ import (
 	"github.com/threatwinds/go-sdk/catcher"
 )
 
-const notificationMessageTemplate = "Correlation rule '%s' generated over %d open, un-deduplicated alerts from data source '%s' in the last 24h and was automatically disabled to prevent alert flooding. If this volume is expected, use an Alert Tag Rule to mark it 'False positive', or add deduplicateBy/groupBy to the rule, then re-enable it."
+// The window is interpolated rather than hardcoded: windowHours is
+// configurable, so a fixed "24h" would misreport the period the count covers.
+const notificationMessageTemplate = "Correlation rule '%s' generated %d open, un-deduplicated alerts from data source '%s' for tenant '%s' in the last %dh and was automatically disabled for that tenant to prevent alert flooding. If this volume is expected, use an Alert Tag Rule to mark it 'False positive', or add deduplicateBy/groupBy to the rule, then re-enable it."
+
+// tenantHeader scopes every backend call. Without it the middleware treats an
+// internal caller as tenantless and the backend falls back to the platform
+// tenant, so the disable would land on the wrong tenant's rule list.
+const tenantHeader = "X-Tenant-Id"
 
 type backendClient struct {
 	baseURL     string
@@ -36,13 +44,14 @@ type ruleSearchResult struct {
 	RuleActive bool   `json:"ruleActive"`
 }
 
-func (c *backendClient) resolveRule(ctx context.Context, ruleName string) ([]ruleSearchResult, error) {
+func (c *backendClient) resolveRule(ctx context.Context, tenantID, ruleName string) ([]ruleSearchResult, error) {
 	endpoint := c.baseURL + "/api/v1/eventprocessing/correlation-rule/search-by-filters?ruleName=" + url.QueryEscape(ruleName)
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
 	if err != nil {
 		return nil, err
 	}
 	req.Header.Set("X-Internal-Key", c.internalKey)
+	req.Header.Set(tenantHeader, tenantID)
 
 	resp, err := c.httpClient.Do(req)
 	if err != nil {
@@ -53,7 +62,7 @@ func (c *backendClient) resolveRule(ctx context.Context, ruleName string) ([]rul
 	if resp.StatusCode >= 400 {
 		body, _ := io.ReadAll(resp.Body)
 		return nil, catcher.Error("search-by-filters call returned error status", nil, map[string]any{
-			"status": resp.StatusCode, "body": string(body), "ruleName": ruleName,
+			"status": resp.StatusCode, "body": string(body), "ruleName": ruleName, "tenantId": tenantID,
 		})
 	}
 
@@ -70,7 +79,7 @@ func (c *backendClient) resolveRule(ctx context.Context, ruleName string) ([]rul
 	}
 	if len(matches) > 1 {
 		catcher.Warn("rule-flood-guard: ambiguous rule name collision, disabling every exact match", map[string]any{
-			"ruleName": ruleName, "matches": len(matches),
+			"ruleName": ruleName, "matches": len(matches), "tenantId": tenantID,
 		})
 	}
 	return matches, nil
@@ -80,8 +89,8 @@ type activateDeactivateResponse struct {
 	Changed bool `json:"changed"`
 }
 
-func (c *backendClient) Deactivate(ctx context.Context, ruleName string) (bool, error) {
-	matches, err := c.resolveRule(ctx, ruleName)
+func (c *backendClient) Deactivate(ctx context.Context, tenantID, ruleName string) (bool, error) {
+	matches, err := c.resolveRule(ctx, tenantID, ruleName)
 	if err != nil {
 		return false, err
 	}
@@ -94,7 +103,7 @@ func (c *backendClient) Deactivate(ctx context.Context, ruleName string) (bool,
 		if !matches[i].RuleActive {
 			continue
 		}
-		ruleChanged, err := c.deactivateOne(ctx, ruleName, matches[i].RelPath)
+		ruleChanged, err := c.deactivateOne(ctx, tenantID, ruleName, matches[i].RelPath)
 		if err != nil {
 			return changed, err
 		}
@@ -103,19 +112,20 @@ func (c *backendClient) Deactivate(ctx context.Context, ruleName string) (bool,
 		}
 	}
 	catcher.Info("rule-flood-guard: exact-match rules processed for deactivation", map[string]any{
-		"ruleName": ruleName, "matches": len(matches), "changed": changed,
+		"ruleName": ruleName, "matches": len(matches), "changed": changed, "tenantId": tenantID,
 	})
 	return changed, nil
 }
 
-func (c *backendClient) deactivateOne(ctx context.Context, ruleName, relPath string) (bool, error) {
+func (c *backendClient) deactivateOne(ctx context.Context, tenantID, ruleName, relPath string) (bool, error) {
 	endpoint := fmt.Sprintf("%s/api/v1/eventprocessing/correlation-rule/activate-deactivate?relPath=%s&active=false",
 		c.baseURL, url.QueryEscape(relPath))
 	req, err := http.NewRequestWithContext(ctx, http.MethodPut, endpoint, nil)
 	if err != nil {
 		return false, err
 	}
 	req.Header.Set("X-Internal-Key", c.internalKey)
+	req.Header.Set(tenantHeader, tenantID)
 
 	resp, err := c.httpClient.Do(req)
 	if err != nil {
@@ -126,7 +136,8 @@ func (c *backendClient) deactivateOne(ctx context.Context, ruleName, relPath str
 	if resp.StatusCode >= 400 {
 		body, _ := io.ReadAll(resp.Body)
 		return false, catcher.Error("activate-deactivate call returned error status", nil, map[string]any{
-			"status": resp.StatusCode, "body": string(body), "ruleName": ruleName, "relPath": relPath,
+			"status": resp.StatusCode, "body": string(body), "ruleName": ruleName,
+			"relPath": relPath, "tenantId": tenantID,
 		})
 	}
 
@@ -143,12 +154,20 @@ type notifyRequest struct {
 	Message string `json:"message"`
 }
 
-// platformTenant receives the flood notifications. A rule that floods is
-// deactivated for the whole instance and the rules are not a tenant's to
-// re-enable, so the operator is the one who has to know.
+// platformTenant is the operator's tenant. It gets a copy of every flood
+// notification so the operator keeps the instance-wide visibility they had
+// before the disable became per-tenant.
 const platformTenant = "ce66672c-e36d-4761-a8c8-90058fee1a24"
 
-func (c *backendClient) Notify(ctx context.Context, message string) error {
+func (c *backendClient) Notify(ctx context.Context, tenantID, message string) error {
+	err := c.notifyTenant(ctx, tenantID, message)
+	if tenantID != platformTenant {
+		err = errors.Join(err, c.notifyTenant(ctx, platformTenant, message))
+	}
+	return err
+}
+
+func (c *backendClient) notifyTenant(ctx context.Context, tenantID, message string) error {
 	payload, err := json.Marshal(notifyRequest{Source: "SYSTEM", Type: "WARNING", Message: message})
 	if err != nil {
 		return err
@@ -159,7 +178,7 @@ func (c *backendClient) Notify(ctx context.Context, message string) error {
 		return err
 	}
 	req.Header.Set("X-Internal-Key", c.internalKey)
-	req.Header.Set("X-Tenant-Id", platformTenant)
+	req.Header.Set(tenantHeader, tenantID)
 	req.Header.Set("Content-Type", "application/json")
 
 	resp, err := c.httpClient.Do(req)
@@ -171,12 +190,12 @@ func (c *backendClient) Notify(ctx context.Context, message string) error {
 	if resp.StatusCode >= 400 {
 		body, _ := io.ReadAll(resp.Body)
 		return catcher.Error("notify call returned error status", nil, map[string]any{
-			"status": resp.StatusCode, "body": string(body),
+			"status": resp.StatusCode, "body": string(body), "tenantId": tenantID,
 		})
 	}
 	return nil
 }
 
-func floodNotificationMessage(ruleName string, threshold int64, dataSource string) string {
-	return fmt.Sprintf(notificationMessageTemplate, ruleName, threshold, dataSource)
+func floodNotificationMessage(tenantID, ruleName string, count int64, dataSource string, windowHours int) string {
+	return fmt.Sprintf(notificationMessageTemplate, ruleName, count, dataSource, tenantID, windowHours)
 }

plugins/rule-flood-guard/guard.go

@@ -10,8 +10,8 @@ import (
 type searchFunc func(ctx context.Context, window time.Duration) ([]ruleBucket, error)
 
 type disableNotifier interface {
-	Deactivate(ctx context.Context, ruleName string) (bool, error)
-	Notify(ctx context.Context, message string) error
+	Deactivate(ctx context.Context, tenantID, ruleName string) (bool, error)
+	Notify(ctx context.Context, tenantID, message string) error
 }
 
 type getConfig func() Config
@@ -32,21 +32,28 @@ func evaluateOnce(ctx context.Context, search searchFunc, client disableNotifier
 		if b.Count <= cfg.Threshold {
 			continue
 		}
+		if b.TenantID == "" {
+			catcher.Warn("rule-flood-guard: dropping a flooding bucket with no tenant", map[string]any{
+				"ruleName": b.RuleName, "dataSource": b.DataSource, "count": b.Count,
+			})
+			continue
+		}
 
-		changed, err := client.Deactivate(ctx, b.RuleName)
+		changed, err := client.Deactivate(ctx, b.TenantID, b.RuleName)
 		if err != nil {
 			_ = catcher.Error("rule-flood-guard: failed to deactivate rule", err, map[string]any{
-				"ruleName": b.RuleName, "dataSource": b.DataSource, "count": b.Count,
+				"tenantId": b.TenantID, "ruleName": b.RuleName,
+				"dataSource": b.DataSource, "count": b.Count,
 			})
 		}
 		if !changed {
 			continue
 		}
 
-		msg := floodNotificationMessage(b.RuleName, cfg.Threshold, b.DataSource)
-		if err := client.Notify(ctx, msg); err != nil {
+		msg := floodNotificationMessage(b.TenantID, b.RuleName, b.Count, b.DataSource, cfg.WindowHours)
+		if err := client.Notify(ctx, b.TenantID, msg); err != nil {
 			_ = catcher.Error("rule-flood-guard: failed to send notification", err, map[string]any{
-				"ruleName": b.RuleName, "dataSource": b.DataSource,
+				"tenantId": b.TenantID, "ruleName": b.RuleName, "dataSource": b.DataSource,
 			})
 		}
 	}