alerts: create metas only if alert was not discarded (#4633)

Author: blotus

pkg/database/alerts.go

@@ -653,11 +653,6 @@ func (c *Client) createAlertBatch(ctx context.Context, machineID string, owner *
 			return nil, rollbackOnError(tx, err, fmt.Sprintf("building events for alert %s", alertItem.UUID))
 		}
 
-		metas, err := buildMetaCreates(ctx, c.Log, txEnt, alertItem)
-		if err != nil {
-			c.Log.Warningf("error creating alert meta: %s", err)
-		}
-
 		decisions, discardCount, err := c.buildDecisions(ctx, c.Log, txEnt, alertItem, stopAtTime)
 		if err != nil {
 			return nil, rollbackOnError(tx, err, fmt.Sprintf("building decisions for alert %s", alertItem.UUID))
@@ -669,6 +664,13 @@ func (c *Client) createAlertBatch(ctx context.Context, machineID string, owner *
 			continue
 		}
 
+		// after the discard check: metas are linked to the alert only when it is built,
+		// so inserting them earlier leaves orphan rows behind when the alert is dropped
+		metas, err := buildMetaCreates(ctx, c.Log, txEnt, alertItem)
+		if err != nil {
+			c.Log.Warningf("error creating alert meta: %s", err)
+		}
+
 		builder := txEnt.Alert.
 			Create().
 			SetScenario(*alertItem.Scenario).

pkg/database/alerts_pg_race_test.go

@@ -147,11 +147,17 @@ func TestAlertCreateVsFlushOrphansRace(t *testing.T) {
 	require.Zero(t, edgeErrCount.Load(), "got %d edge-constraint errors (events deleted out from under alert insert)", edgeErrCount.Load())
 	require.Zero(t, otherErrCount.Load(), "got %d unexpected errors", otherErrCount.Load())
 	require.Positive(t, successCount.Load(), "no alerts were inserted at all — test setup is wrong")
+
+	// every meta of a committed alert must still be there: the orphan flush must not
+	// see rows that are still in flight
+	metaCount, err := dbClient.Ent.Meta.Query().Count(ctx)
+	require.NoError(t, err)
+	require.Equal(t, int(successCount.Load())*2, metaCount, "orphan flush deleted metas of in-flight alerts")
 }
 
-// makeRaceAlerts builds a batch of distinct alerts with several events and a
-// decision each, so the create path actually exercises Event.CreateBulk plus
-// the alert↔events O2M edge update plus decision attachment.
+// makeRaceAlerts builds a batch of distinct alerts with several events, metas and
+// a decision each, so the create path actually exercises Event.CreateBulk and
+// Meta.CreateBulk plus the alert↔events O2M edge update plus decision attachment.
 func makeRaceAlerts(workerID, batchID, eventsPerAlert int) []*models.Alert {
 	now := time.Now().UTC().Format(time.RFC3339)
 
@@ -201,6 +207,10 @@ func makeRaceAlerts(workerID, batchID, eventsPerAlert int) []*models.Alert {
 				IP:    value,
 			},
 			Events: events,
+			Meta: models.Meta{
+				{Key: "datasource_path", Value: "/var/log/test.log"},
+				{Key: "datasource_type", Value: "file"},
+			},
 			Decisions: []*models.Decision{
 				{
 					Duration:  &duration,

pkg/database/flush.go

@@ -17,6 +17,7 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/database/ent/decision"
 	"github.com/crowdsecurity/crowdsec/pkg/database/ent/event"
 	"github.com/crowdsecurity/crowdsec/pkg/database/ent/machine"
+	"github.com/crowdsecurity/crowdsec/pkg/database/ent/meta"
 	"github.com/crowdsecurity/crowdsec/pkg/database/ent/metric"
 	"github.com/crowdsecurity/crowdsec/pkg/database/ent/predicate"
 	"github.com/crowdsecurity/crowdsec/pkg/logging"
@@ -27,6 +28,11 @@ const (
 	// how long to keep metrics in the local database
 	defaultMetricsMaxAge = 7 * 24 * time.Hour
 	flushInterval        = 1 * time.Minute
+	// orphan metas are deleted in bounded batches: an install upgrading with a
+	// backlog can carry hundreds of millions of them, and a single unbounded
+	// DELETE would hold the table for minutes.
+	orphanMetaBatchSize = 5_000
+	orphanMetaMaxPerRun = 500_000
 )
 
 func (c *Client) StartFlushScheduler(ctx context.Context, config *csconfig.FlushDBCfg) (gocron.Scheduler, error) {
@@ -183,6 +189,54 @@ func (c *Client) FlushOrphans(ctx context.Context) {
 	if eventsCount > 0 {
 		c.Log.Infof("%d deleted orphan decisions", eventsCount)
 	}
+
+	metasCount, err := c.flushOrphanMetas(ctx)
+	if err != nil {
+		c.Log.Warningf("error while deleting orphan metas: %s", err)
+		return
+	}
+
+	if metasCount > 0 {
+		c.Log.Infof("%d deleted orphan metas", metasCount)
+	}
+}
+
+// flushOrphanMetas deletes meta rows that were never linked to an alert, up to
+// orphanMetaMaxPerRun per call. Versions before 1.8.0 inserted metas outside the
+// alert transaction, so an upgraded database can hold a large backlog: it is
+// drained over several runs instead of in one statement.
+func (c *Client) flushOrphanMetas(ctx context.Context) (int, error) {
+	deleted := 0
+
+	for deleted < orphanMetaMaxPerRun {
+		ids, err := c.Ent.Meta.Query().Where(meta.Not(meta.HasOwner())).Limit(orphanMetaBatchSize).IDs(ctx)
+		if err != nil {
+			return deleted, err
+		}
+
+		if len(ids) == 0 {
+			break
+		}
+
+		count, err := c.Ent.Meta.Delete().Where(meta.IDIn(ids...)).Exec(ctx)
+		if err != nil {
+			return deleted, err
+		}
+
+		// nothing deleted for a full batch means someone else got there first: stop
+		// rather than spin on the same rows
+		if count == 0 {
+			break
+		}
+
+		deleted += count
+
+		if len(ids) < orphanMetaBatchSize {
+			break
+		}
+	}
+
+	return deleted, nil
 }
 
 func (c *Client) flushBouncers(ctx context.Context, authType string, duration *time.Duration) {

pkg/database/flush_test.go

@@ -3,12 +3,15 @@ package database
 import (
 	"context"
 	"fmt"
+	"slices"
 	"testing"
 	"time"
 
 	"github.com/go-openapi/strfmt"
 	"github.com/stretchr/testify/require"
 
+	"github.com/crowdsecurity/crowdsec/pkg/database/ent"
+	"github.com/crowdsecurity/crowdsec/pkg/database/ent/meta"
 	"github.com/crowdsecurity/crowdsec/pkg/models"
 	"github.com/crowdsecurity/crowdsec/pkg/types"
 )
@@ -156,3 +159,91 @@ func TestFlushAlerts_MaxItemsKeepsActiveDecisions(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, 1, decCount, "active decision must not be cascade-deleted")
 }
+
+func countMetas(t *testing.T, ctx context.Context, c *Client) (total int, orphans int) {
+	t.Helper()
+
+	total, err := c.Ent.Meta.Query().Count(ctx)
+	require.NoError(t, err)
+
+	orphans, err = c.Ent.Meta.Query().Where(meta.Not(meta.HasOwner())).Count(ctx)
+	require.NoError(t, err)
+
+	return total, orphans
+}
+
+// A dropped alert must not leave its metas behind: they are inserted before the
+// alert row, and only the alert insert links them.
+func TestCreateAlert_DroppedAlertLeavesNoMetas(t *testing.T) {
+	ctx := t.Context()
+	c := getDBClient(t, ctx)
+
+	machineID := "test-dropped-alert-metas"
+	registerFlushTestMachine(t, ctx, c, machineID)
+
+	// the decision value is not a valid address, so the decision is discarded and
+	// the alert with it
+	dropped := makeFlushAlert("not-an-ip", true)
+	dropped.Meta = models.Meta{{Key: "k1", Value: "v1"}, {Key: "k2", Value: "v2"}}
+
+	kept := makeFlushAlert("1.2.3.4", true)
+	kept.Meta = models.Meta{{Key: "k1", Value: "v1"}}
+
+	_, err := c.CreateAlert(ctx, machineID, []*models.Alert{dropped, kept})
+	require.NoError(t, err)
+
+	alerts, err := c.Ent.Alert.Query().Count(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 1, alerts)
+
+	total, orphans := countMetas(t, ctx, c)
+	require.Equal(t, 1, total)
+	require.Zero(t, orphans)
+}
+
+// Metas orphaned by an older version are reaped, linked ones are not.
+func TestFlushOrphans_DeletesOrphanMetas(t *testing.T) {
+	tests := []struct {
+		name    string
+		orphans int
+	}{
+		{name: "single batch", orphans: 3},
+		{name: "several batches", orphans: orphanMetaBatchSize + 1},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := t.Context()
+			c := getDBClient(t, ctx)
+
+			machineID := "test-orphan-metas"
+			registerFlushTestMachine(t, ctx, c, machineID)
+
+			alert := makeFlushAlert("1.2.3.4", true)
+			alert.Meta = models.Meta{{Key: "k1", Value: "v1"}}
+
+			_, err := c.CreateAlert(ctx, machineID, []*models.Alert{alert})
+			require.NoError(t, err)
+
+			for chunk := range slices.Chunk(make([]struct{}, tc.orphans), 500) {
+				builders := make([]*ent.MetaCreate, len(chunk))
+				for i := range chunk {
+					builders[i] = c.Ent.Meta.Create().SetKey("orphan").SetValue("v")
+				}
+
+				_, err := c.Ent.Meta.CreateBulk(builders...).Save(ctx)
+				require.NoError(t, err)
+			}
+
+			total, orphans := countMetas(t, ctx, c)
+			require.Equal(t, tc.orphans+1, total)
+			require.Equal(t, tc.orphans, orphans)
+
+			c.FlushOrphans(ctx)
+
+			total, orphans = countMetas(t, ctx, c)
+			require.Equal(t, 1, total)
+			require.Zero(t, orphans)
+		})
+	}
+}