diff --git a/.gitignore b/.gitignore index dfc750fc439..759d1b202a6 100644 --- a/.gitignore +++ b/.gitignore @@ -70,3 +70,4 @@ __pycache__ # local agent/tooling dirs (not part of the repo) .agents/ .claude/ +.worktrees/ diff --git a/cmd/crowdsec/metrics.go b/cmd/crowdsec/metrics.go index 320b89bc9bc..06c86c8cdcf 100644 --- a/cmd/crowdsec/metrics.go +++ b/cmd/crowdsec/metrics.go @@ -46,7 +46,7 @@ func computeDynamicMetrics(next http.Handler, dbClient *database.Client) http.Ha metrics.GlobalActiveDecisions.Reset() for _, d := range decisions { - metrics.GlobalActiveDecisions.With(prometheus.Labels{"reason": d.Scenario, "origin": d.Origin, "action": d.Type}).Set(float64(d.Count)) + metrics.GlobalActiveDecisions.With(prometheus.Labels{"reason": d.Scenario, "origin": d.Origin, "action": d.Type, "machine": d.Machine}).Set(float64(d.Count)) } metrics.GlobalAlerts.Reset() diff --git a/pkg/database/decisions.go b/pkg/database/decisions.go index 2f2534ed235..a423c907505 100644 --- a/pkg/database/decisions.go +++ b/pkg/database/decisions.go @@ -12,16 +12,19 @@ import ( "github.com/crowdsecurity/crowdsec/pkg/csnet" "github.com/crowdsecurity/crowdsec/pkg/database/ent" + "github.com/crowdsecurity/crowdsec/pkg/database/ent/alert" "github.com/crowdsecurity/crowdsec/pkg/database/ent/decision" ) const decisionDeleteBulkSize = 256 // scientifically proven to be the best value for bulk delete type DecisionsByScenario struct { - Scenario string - Count int - Origin string - Type string + Scenario string + Count int + Origin string + Type string + Machine string + AlertDecisions int `sql:"alert_decisions"` } func (c *Client) QueryAllDecisionsWithFilters(ctx context.Context, now time.Time, filter map[string][]string) ([]*ent.Decision, error) { @@ -83,9 +86,7 @@ func (c *Client) QueryExpiredDecisionsWithFilters(ctx context.Context, now time. } func (c *Client) QueryDecisionCountByScenario(ctx context.Context) ([]*DecisionsByScenario, error) { - query := c.Ent.Decision.Query().Where( - decision.UntilGT(time.Now().UTC()), - ) + query := c.Ent.Decision.Query().Where(decision.UntilGT(time.Now().UTC())) query, err := applyDecisionFilter(query, make(map[string][]string)) if err != nil { @@ -93,15 +94,92 @@ func (c *Client) QueryDecisionCountByScenario(ctx context.Context) ([]*Decisions return nil, fmt.Errorf("count all decisions with filters: %w", QueryFail) } - var r []*DecisionsByScenario + var raw []*DecisionsByScenario - err = query.GroupBy(decision.FieldScenario, decision.FieldOrigin, decision.FieldType).Aggregate(ent.Count()).Scan(ctx, &r) + err = query.GroupBy(decision.FieldScenario, decision.FieldOrigin, decision.FieldType, decision.FieldAlertDecisions). + Aggregate(ent.Count()).Scan(ctx, &raw) if err != nil { c.Log.Warningf("QueryDecisionCountByScenario : %s", err) return nil, fmt.Errorf("count all decisions with filters: %w", QueryFail) } - return r, nil + machineByAlertID, err := c.resolveMachinesByAlertID(ctx, raw) + if err != nil { + c.Log.Warningf("QueryDecisionCountByScenario : %s", err) + return nil, fmt.Errorf("count all decisions with filters: %w", QueryFail) + } + + return foldByMachine(raw, machineByAlertID), nil +} + +// resolveMachinesByAlertID looks up, for each distinct alert referenced in rows, +// the MachineId of its owning machine. Alerts with no owning machine (Edges.Owner +// == nil) are simply absent from the returned map — callers must fall back to "N/A". +func (c *Client) resolveMachinesByAlertID(ctx context.Context, rows []*DecisionsByScenario) (map[int]string, error) { + seen := make(map[int]struct{}) + alertIDs := make([]int, 0, len(rows)) + + for _, row := range rows { + if row.AlertDecisions == 0 { + continue + } + if _, ok := seen[row.AlertDecisions]; !ok { + seen[row.AlertDecisions] = struct{}{} + alertIDs = append(alertIDs, row.AlertDecisions) + } + } + + machineByAlertID := make(map[int]string, len(alertIDs)) + if len(alertIDs) == 0 { + return machineByAlertID, nil + } + + alerts, err := c.Ent.Alert.Query().Where(alert.IDIn(alertIDs...)).WithOwner().All(ctx) + if err != nil { + return nil, err + } + + for _, a := range alerts { + if a.Edges.Owner != nil { + machineByAlertID[a.ID] = a.Edges.Owner.MachineId + } + } + + return machineByAlertID, nil +} + +// foldByMachine re-aggregates rows (grouped by scenario/origin/type/alert) into rows +// grouped by scenario/origin/type/machine, summing counts for alerts owned by the +// same machine. Falls back to "N/A" for alerts with no owning machine, matching the +// convention in pkg/apiserver/controllers/v1/alerts.go's FormatOneAlert. +func foldByMachine(rows []*DecisionsByScenario, machineByAlertID map[int]string) []*DecisionsByScenario { + type key struct{ scenario, origin, typ, machine string } + + folded := make(map[key]*DecisionsByScenario, len(rows)) + order := make([]key, 0, len(rows)) + + for _, row := range rows { + machineID, ok := machineByAlertID[row.AlertDecisions] + if !ok { + machineID = "N/A" + } + + k := key{row.Scenario, row.Origin, row.Type, machineID} + if existing, ok := folded[k]; ok { + existing.Count += row.Count + continue + } + + folded[k] = &DecisionsByScenario{Scenario: row.Scenario, Origin: row.Origin, Type: row.Type, Machine: machineID, Count: row.Count} + order = append(order, k) + } + + result := make([]*DecisionsByScenario, 0, len(order)) + for _, k := range order { + result = append(result, folded[k]) + } + + return result } func (c *Client) QueryDecisionWithFilter(ctx context.Context, filter map[string][]string) ([]*ent.Decision, error) { diff --git a/pkg/database/decisions_test.go b/pkg/database/decisions_test.go new file mode 100644 index 00000000000..e107e1d34b8 --- /dev/null +++ b/pkg/database/decisions_test.go @@ -0,0 +1,157 @@ +package database + +import ( + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/crowdsecurity/crowdsec/pkg/models" +) + +// makeDecisionsScenarioAlert builds a single alert carrying one active decision, for +// the QueryDecisionCountByScenario tests. Reuses the "8760h" duration convention from +// flush_test.go's makeFlushAlert so the decision stays active for the test's lifetime. +func makeDecisionsScenarioAlert(value string) *models.Alert { + now := time.Now().UTC().Format(time.RFC3339) + + scenario := "test/decisions-by-scenario" + scenarioVersion := "0.1" + scenarioHash := "deadbeef" + leakSpeed := "10s" + scope := "Ip" + simulated := false + message := "decisions by scenario test" + capacity := int32(1) + eventsCount := int32(1) + uuid := fmt.Sprintf("uuid-%s-%d", value, time.Now().UnixNano()) + duration := "8760h" + decisionType := "ban" + origin := "test" + + return &models.Alert{ + Capacity: &capacity, + Scenario: &scenario, + ScenarioVersion: &scenarioVersion, + ScenarioHash: &scenarioHash, + Leakspeed: &leakSpeed, + Message: &message, + EventsCount: &eventsCount, + Simulated: &simulated, + StartAt: &now, + StopAt: &now, + UUID: uuid, + Source: &models.Source{ + Scope: &scope, + Value: &value, + IP: value, + }, + Decisions: []*models.Decision{ + { + Duration: &duration, + Type: &decisionType, + Scope: &scope, + Value: &value, + Origin: &origin, + Scenario: &scenario, + Simulated: &simulated, + }, + }, + } +} + +// TestQueryDecisionCountByScenario_MachineOwnedAlert ensures a decision belonging to +// an alert owned by a machine resolves to that machine's ID in the Machine field. +func TestQueryDecisionCountByScenario_MachineOwnedAlert(t *testing.T) { + ctx := t.Context() + c := getDBClient(t, ctx) + + machineID := "decisions-test-machine" + registerFlushTestMachine(t, ctx, c, machineID) + + _, err := c.CreateAlert(ctx, machineID, []*models.Alert{ + makeDecisionsScenarioAlert("1.2.3.4"), + }) + require.NoError(t, err) + + rows, err := c.QueryDecisionCountByScenario(ctx) + require.NoError(t, err) + require.Len(t, rows, 1) + require.Equal(t, machineID, rows[0].Machine) + require.Equal(t, 1, rows[0].Count) +} + +// TestQueryDecisionCountByScenario_NoOwningMachine ensures a decision belonging to an +// alert with no owning machine falls back to "N/A" in the Machine field. +func TestQueryDecisionCountByScenario_NoOwningMachine(t *testing.T) { + ctx := t.Context() + c := getDBClient(t, ctx) + + _, err := c.CreateAlert(ctx, "", []*models.Alert{ + makeDecisionsScenarioAlert("1.2.3.4"), + }) + require.NoError(t, err) + + rows, err := c.QueryDecisionCountByScenario(ctx) + require.NoError(t, err) + require.Len(t, rows, 1) + require.Equal(t, "N/A", rows[0].Machine) +} + +// TestQueryDecisionCountByScenario_FoldsSameMachineAcrossAlerts ensures two separate +// alerts owned by the same machine, sharing scenario/origin/type on their decisions, +// fold into a single row with the summed count rather than staying split per alert. +func TestQueryDecisionCountByScenario_FoldsSameMachineAcrossAlerts(t *testing.T) { + ctx := t.Context() + c := getDBClient(t, ctx) + + machineID := "decisions-test-machine" + registerFlushTestMachine(t, ctx, c, machineID) + + _, err := c.CreateAlert(ctx, machineID, []*models.Alert{makeDecisionsScenarioAlert("1.2.3.4")}) + require.NoError(t, err) + _, err = c.CreateAlert(ctx, machineID, []*models.Alert{makeDecisionsScenarioAlert("5.6.7.8")}) + require.NoError(t, err) + + rows, err := c.QueryDecisionCountByScenario(ctx) + require.NoError(t, err) + require.Len(t, rows, 1) + require.Equal(t, machineID, rows[0].Machine) + require.Equal(t, 2, rows[0].Count) +} + +// TestQueryDecisionCountByScenario_SplitsAcrossDifferentMachines ensures alerts owned +// by distinct machines, sharing scenario/origin/type on their decisions, are reported +// as separate rows rather than merged - while the sum of their counts still matches +// what a pre-change single (scenario, origin, type) row would have reported. +func TestQueryDecisionCountByScenario_SplitsAcrossDifferentMachines(t *testing.T) { + ctx := t.Context() + c := getDBClient(t, ctx) + + machine1 := "decisions-test-machine-1" + machine2 := "decisions-test-machine-2" + registerFlushTestMachine(t, ctx, c, machine1) + registerFlushTestMachine(t, ctx, c, machine2) + + _, err := c.CreateAlert(ctx, machine1, []*models.Alert{makeDecisionsScenarioAlert("1.2.3.4")}) + require.NoError(t, err) + _, err = c.CreateAlert(ctx, machine2, []*models.Alert{makeDecisionsScenarioAlert("5.6.7.8")}) + require.NoError(t, err) + + rows, err := c.QueryDecisionCountByScenario(ctx) + require.NoError(t, err) + require.Len(t, rows, 2) + + countByMachine := make(map[string]int, len(rows)) + total := 0 + + for _, row := range rows { + countByMachine[row.Machine] = row.Count + total += row.Count + } + + require.Equal(t, 1, countByMachine[machine1]) + require.Equal(t, 1, countByMachine[machine2]) + require.Equal(t, 2, total) +} diff --git a/pkg/metrics/global.go b/pkg/metrics/global.go index c1a60b5958a..b952489eb4e 100644 --- a/pkg/metrics/global.go +++ b/pkg/metrics/global.go @@ -70,7 +70,7 @@ var GlobalActiveDecisions = prometheus.NewGaugeVec( Name: GlobalActiveDecisionsMetricName, Help: "Number of active decisions.", }, - []string{"reason", "origin", "action"}, + []string{"reason", "origin", "action", "machine"}, ) const GlobalAlertsMetricName = "cs_alerts"