diff --git a/plugins/rule-flood-guard/README.md b/plugins/rule-flood-guard/README.md index 27dafb077..e1ee6c422 100644 --- a/plugins/rule-flood-guard/README.md +++ b/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. | diff --git a/plugins/rule-flood-guard/aggregation.go b/plugins/rule-flood-guard/aggregation.go index 49ae03c5a..ca53eff69 100644 --- a/plugins/rule-flood-guard/aggregation.go +++ b/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,6 +37,9 @@ 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), @@ -37,23 +47,42 @@ func searchRuleBuckets(ctx context.Context, window time.Duration) ([]ruleBucket, } 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 } diff --git a/plugins/rule-flood-guard/backend.go b/plugins/rule-flood-guard/backend.go index f84e1eea8..c199d14e4 100644 --- a/plugins/rule-flood-guard/backend.go +++ b/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,12 +112,12 @@ 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) @@ -116,6 +125,7 @@ func (c *backendClient) deactivateOne(ctx context.Context, ruleName, relPath str 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) } diff --git a/plugins/rule-flood-guard/guard.go b/plugins/rule-flood-guard/guard.go index 447e9d9cd..e870facef 100644 --- a/plugins/rule-flood-guard/guard.go +++ b/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, }) } }