mirror of
https://github.com/coder/coder.git
synced 2026-09-24 15:04:27 +08:00
feat: purge boundary logs past retention (#24815)
Add a periodic purge job for `boundary_logs` rows past their retention threshold, following the same pattern as the existing audit log and connection log purge jobs in `dbpurge`. Expose a `--boundary-log-retention` deployment flag (env `CODER_BOUNDARY_LOG_RETENTION`, YAML `retention.boundary_logs`). Default is `0` (keep indefinitely). When set to a positive duration, `purgeTick` deletes rows where `captured_at` is older than the threshold in batches of 10,000, matching other log purge operations. The `boundary_logs` label is added to the `records_purged_total` Prometheus counter. Also removes the random-UUID fallback for `OwnerID` in `dbgen.BoundarySession`. The previous fallback generated a UUID that could never satisfy the `boundary_sessions_owner_id_fkey` FK constraint, masking test setup bugs. Callers must now provide a valid user ID or accept NULL (the legitimate "user deleted" state).
This commit is contained in:
Sas Swart
@@ -29,6 +29,10 @@ const (
|
||||
connectionLogsBatchSize = 10000
|
||||
// Batch size for audit log deletion.
|
||||
auditLogsBatchSize = 10000
|
||||
// Batch size for boundary log deletion.
|
||||
boundaryLogsBatchSize = 10000
|
||||
// Batch size for boundary session deletion.
|
||||
boundarySessionsBatchSize = 10000
|
||||
// Telemetry heartbeats are used to deduplicate events across replicas. We
|
||||
// don't need to persist heartbeat rows for longer than 24 hours, as they
|
||||
// are only used for deduplication across replicas. The time needs to be
|
||||
@@ -251,6 +255,26 @@ func (i *instance) purgeTick(ctx context.Context, db database.Store, start time.
|
||||
}
|
||||
}
|
||||
|
||||
var purgedBoundaryLogs, purgedBoundarySessions int64
|
||||
boundaryLogsRetention := i.vals.Retention.BoundaryLogs.Value()
|
||||
if boundaryLogsRetention > 0 {
|
||||
deleteBoundaryLogsBefore := start.Add(-boundaryLogsRetention)
|
||||
purgedBoundaryLogs, err = tx.DeleteOldBoundaryLogs(ctx, database.DeleteOldBoundaryLogsParams{
|
||||
BeforeTime: deleteBoundaryLogsBefore,
|
||||
LimitCount: boundaryLogsBatchSize,
|
||||
})
|
||||
if err != nil {
|
||||
return xerrors.Errorf("failed to delete old boundary logs: %w", err)
|
||||
}
|
||||
purgedBoundarySessions, err = tx.DeleteOldBoundarySessions(ctx, database.DeleteOldBoundarySessionsParams{
|
||||
BeforeTime: deleteBoundaryLogsBefore,
|
||||
LimitCount: boundarySessionsBatchSize,
|
||||
})
|
||||
if err != nil {
|
||||
return xerrors.Errorf("failed to delete old boundary sessions: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
var purgedChats, purgedChatFiles, purgedChatDebugRuns int64
|
||||
if purgeChats {
|
||||
purgedChats, purgedChatFiles, err = i.purgeChatsInTx(ctx, tx, start, chatRetentionDays)
|
||||
@@ -278,6 +302,8 @@ func (i *instance) purgeTick(ctx context.Context, db database.Store, start time.
|
||||
slog.F("aibridge_records", purgedAIBridgeRecords),
|
||||
slog.F("connection_logs", purgedConnectionLogs),
|
||||
slog.F("audit_logs", purgedAuditLogs),
|
||||
slog.F("boundary_logs", purgedBoundaryLogs),
|
||||
slog.F("boundary_sessions", purgedBoundarySessions),
|
||||
slog.F("chats", purgedChats),
|
||||
slog.F("chat_files", purgedChatFiles),
|
||||
slog.F("chat_debug_runs", purgedChatDebugRuns),
|
||||
@@ -290,6 +316,8 @@ func (i *instance) purgeTick(ctx context.Context, db database.Store, start time.
|
||||
i.recordsPurged.WithLabelValues("aibridge_records").Add(float64(purgedAIBridgeRecords))
|
||||
i.recordsPurged.WithLabelValues("connection_logs").Add(float64(purgedConnectionLogs))
|
||||
i.recordsPurged.WithLabelValues("audit_logs").Add(float64(purgedAuditLogs))
|
||||
i.recordsPurged.WithLabelValues("boundary_logs").Add(float64(purgedBoundaryLogs))
|
||||
i.recordsPurged.WithLabelValues("boundary_sessions").Add(float64(purgedBoundarySessions))
|
||||
i.recordsPurged.WithLabelValues("chats").Add(float64(purgedChats))
|
||||
i.recordsPurged.WithLabelValues("chat_debug_runs").Add(float64(purgedChatDebugRuns))
|
||||
i.recordsPurged.WithLabelValues("chat_files").Add(float64(purgedChatFiles))
|
||||
|
||||
@@ -1671,6 +1671,265 @@ func TestDeleteOldAuditLogs(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestDeleteOldBoundaryLogs(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
now := time.Date(2025, 1, 15, 7, 30, 0, 0, time.UTC)
|
||||
retentionPeriod := 90 * 24 * time.Hour
|
||||
beforeThreshold := now.Add(-retentionPeriod).Add(-24 * time.Hour) // 91 days ago (older than threshold, before the cutoff)
|
||||
afterThreshold := now.Add(-15 * 24 * time.Hour) // 15 days ago (newer than threshold, after the cutoff)
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
retentionConfig codersdk.RetentionConfig
|
||||
oldLogTime time.Time
|
||||
recentLogTime *time.Time // nil means no recent log created
|
||||
expectOldDeleted bool
|
||||
expectedLogsRemaining int
|
||||
}{
|
||||
{
|
||||
name: "RetentionEnabled",
|
||||
retentionConfig: codersdk.RetentionConfig{
|
||||
BoundaryLogs: serpent.Duration(retentionPeriod),
|
||||
},
|
||||
oldLogTime: beforeThreshold,
|
||||
recentLogTime: &afterThreshold,
|
||||
expectOldDeleted: true,
|
||||
expectedLogsRemaining: 1, // only recent log remains
|
||||
},
|
||||
{
|
||||
name: "RetentionDisabled",
|
||||
retentionConfig: codersdk.RetentionConfig{
|
||||
BoundaryLogs: serpent.Duration(0),
|
||||
},
|
||||
oldLogTime: now.Add(-365 * 24 * time.Hour), // 1 year ago
|
||||
recentLogTime: nil,
|
||||
expectOldDeleted: false,
|
||||
expectedLogsRemaining: 1, // old log is kept
|
||||
},
|
||||
{
|
||||
name: "RetentionNegative",
|
||||
retentionConfig: codersdk.RetentionConfig{
|
||||
BoundaryLogs: serpent.Duration(-retentionPeriod),
|
||||
},
|
||||
oldLogTime: now.Add(-365 * 24 * time.Hour), // 1 year ago
|
||||
recentLogTime: nil,
|
||||
expectOldDeleted: false,
|
||||
expectedLogsRemaining: 1, // old log is kept
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := testutil.Context(t, testutil.WaitShort)
|
||||
clk := quartz.NewMock(t)
|
||||
clk.Set(now).MustWait(ctx)
|
||||
|
||||
db, _ := dbtestutil.NewDB(t, dbtestutil.WithDumpOnFailure())
|
||||
logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
|
||||
|
||||
// Create the prerequisite rows (user, org, template, workspace,
|
||||
// build, agent) needed to satisfy boundary_sessions foreign keys.
|
||||
user := dbgen.User(t, db, database.User{})
|
||||
org := dbgen.Organization(t, db, database.Organization{})
|
||||
_ = dbgen.OrganizationMember(t, db, database.OrganizationMember{UserID: user.ID, OrganizationID: org.ID})
|
||||
tv := dbgen.TemplateVersion(t, db, database.TemplateVersion{OrganizationID: org.ID, CreatedBy: user.ID})
|
||||
tmpl := dbgen.Template(t, db, database.Template{OrganizationID: org.ID, ActiveVersionID: tv.ID, CreatedBy: user.ID})
|
||||
ws := dbgen.Workspace(t, db, database.WorkspaceTable{
|
||||
OwnerID: user.ID,
|
||||
OrganizationID: org.ID,
|
||||
TemplateID: tmpl.ID,
|
||||
})
|
||||
wb := mustCreateWorkspaceBuild(t, db, org, tv, ws.ID, now, 1)
|
||||
agent := mustCreateAgent(t, db, wb)
|
||||
|
||||
session := dbgen.BoundarySession(t, db, database.BoundarySession{
|
||||
WorkspaceAgentID: agent.ID,
|
||||
OwnerID: uuid.NullUUID{UUID: user.ID, Valid: true},
|
||||
})
|
||||
|
||||
// Create old boundary log.
|
||||
oldLogs := dbgen.BoundaryLogs(t, db, []database.BoundaryLog{{
|
||||
SessionID: session.ID,
|
||||
SequenceNumber: 0,
|
||||
CapturedAt: tc.oldLogTime,
|
||||
CreatedAt: tc.oldLogTime,
|
||||
}})
|
||||
oldLog := oldLogs[0]
|
||||
|
||||
// Create recent boundary log if specified.
|
||||
var recentLog database.BoundaryLog
|
||||
if tc.recentLogTime != nil {
|
||||
recentLogs := dbgen.BoundaryLogs(t, db, []database.BoundaryLog{{
|
||||
SessionID: session.ID,
|
||||
SequenceNumber: 1,
|
||||
CapturedAt: *tc.recentLogTime,
|
||||
CreatedAt: *tc.recentLogTime,
|
||||
}})
|
||||
recentLog = recentLogs[0]
|
||||
}
|
||||
|
||||
// Run the purge.
|
||||
done := awaitDoTick(ctx, t, clk)
|
||||
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{
|
||||
Retention: tc.retentionConfig,
|
||||
}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
|
||||
defer closer.Close()
|
||||
testutil.TryReceive(ctx, t, done)
|
||||
|
||||
// Verify results.
|
||||
logs, err := db.ListBoundaryLogsBySessionID(ctx, database.ListBoundaryLogsBySessionIDParams{
|
||||
SessionID: session.ID,
|
||||
LimitOpt: 100,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Len(t, logs, tc.expectedLogsRemaining, "unexpected number of boundary logs remaining")
|
||||
|
||||
logIDs := make([]uuid.UUID, len(logs))
|
||||
for i, l := range logs {
|
||||
logIDs[i] = l.ID
|
||||
}
|
||||
|
||||
if tc.expectOldDeleted {
|
||||
require.NotContains(t, logIDs, oldLog.ID, "old boundary log should be deleted")
|
||||
} else {
|
||||
require.Contains(t, logIDs, oldLog.ID, "old boundary log should NOT be deleted")
|
||||
}
|
||||
|
||||
if tc.recentLogTime != nil {
|
||||
require.Contains(t, logIDs, recentLog.ID, "recent boundary log should be kept")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeleteOldBoundarySessions(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
now := time.Date(2025, 1, 15, 7, 30, 0, 0, time.UTC)
|
||||
retentionPeriod := 90 * 24 * time.Hour
|
||||
// oldTime is 91 days ago (past threshold).
|
||||
oldTime := now.Add(-retentionPeriod).Add(-24 * time.Hour)
|
||||
// recentTime is 15 days ago (within threshold).
|
||||
recentTime := now.Add(-15 * 24 * time.Hour)
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
retentionConfig codersdk.RetentionConfig
|
||||
sessionUpdatedAt time.Time
|
||||
// logTime is the captured_at for the single log inserted with the session.
|
||||
// Set to nil to create a session with no logs.
|
||||
logTime *time.Time
|
||||
expectSessionDeleted bool
|
||||
}{
|
||||
{
|
||||
name: "SessionDeletedWhenAllLogsExpired",
|
||||
retentionConfig: codersdk.RetentionConfig{
|
||||
BoundaryLogs: serpent.Duration(retentionPeriod),
|
||||
},
|
||||
sessionUpdatedAt: oldTime,
|
||||
logTime: &oldTime, // log is old; will be purged first, leaving session empty
|
||||
expectSessionDeleted: true,
|
||||
},
|
||||
{
|
||||
name: "SessionKeptWhenRecentLogExists",
|
||||
retentionConfig: codersdk.RetentionConfig{
|
||||
BoundaryLogs: serpent.Duration(retentionPeriod),
|
||||
},
|
||||
sessionUpdatedAt: oldTime,
|
||||
logTime: &recentTime, // recent log survives log purge, so session kept
|
||||
expectSessionDeleted: false,
|
||||
},
|
||||
{
|
||||
name: "SessionKeptWhenRetentionDisabled",
|
||||
retentionConfig: codersdk.RetentionConfig{
|
||||
BoundaryLogs: serpent.Duration(0),
|
||||
},
|
||||
sessionUpdatedAt: oldTime,
|
||||
logTime: &oldTime,
|
||||
expectSessionDeleted: false,
|
||||
},
|
||||
{
|
||||
name: "SessionKeptWhenRetentionNegative",
|
||||
retentionConfig: codersdk.RetentionConfig{
|
||||
BoundaryLogs: serpent.Duration(-retentionPeriod),
|
||||
},
|
||||
sessionUpdatedAt: oldTime,
|
||||
logTime: &oldTime,
|
||||
expectSessionDeleted: false,
|
||||
},
|
||||
{
|
||||
name: "SessionKeptWhenUpdatedAtRecent",
|
||||
retentionConfig: codersdk.RetentionConfig{
|
||||
BoundaryLogs: serpent.Duration(retentionPeriod),
|
||||
},
|
||||
sessionUpdatedAt: recentTime, // session itself is recent. NOT eligible for session purge
|
||||
logTime: nil, // no logs; but updated_at guard keeps it
|
||||
expectSessionDeleted: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := testutil.Context(t, testutil.WaitShort)
|
||||
clk := quartz.NewMock(t)
|
||||
clk.Set(now).MustWait(ctx)
|
||||
|
||||
db, _ := dbtestutil.NewDB(t, dbtestutil.WithDumpOnFailure())
|
||||
logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
|
||||
|
||||
// Create the prerequisite rows needed to satisfy boundary_sessions FKs.
|
||||
user := dbgen.User(t, db, database.User{})
|
||||
org := dbgen.Organization(t, db, database.Organization{})
|
||||
_ = dbgen.OrganizationMember(t, db, database.OrganizationMember{UserID: user.ID, OrganizationID: org.ID})
|
||||
tv := dbgen.TemplateVersion(t, db, database.TemplateVersion{OrganizationID: org.ID, CreatedBy: user.ID})
|
||||
tmpl := dbgen.Template(t, db, database.Template{OrganizationID: org.ID, ActiveVersionID: tv.ID, CreatedBy: user.ID})
|
||||
ws := dbgen.Workspace(t, db, database.WorkspaceTable{
|
||||
OwnerID: user.ID,
|
||||
OrganizationID: org.ID,
|
||||
TemplateID: tmpl.ID,
|
||||
})
|
||||
wb := mustCreateWorkspaceBuild(t, db, org, tv, ws.ID, now, 1)
|
||||
agent := mustCreateAgent(t, db, wb)
|
||||
|
||||
session := dbgen.BoundarySession(t, db, database.BoundarySession{
|
||||
WorkspaceAgentID: agent.ID,
|
||||
OwnerID: uuid.NullUUID{UUID: user.ID, Valid: true},
|
||||
UpdatedAt: tc.sessionUpdatedAt,
|
||||
})
|
||||
|
||||
if tc.logTime != nil {
|
||||
dbgen.BoundaryLogs(t, db, []database.BoundaryLog{{
|
||||
SessionID: session.ID,
|
||||
SequenceNumber: 0,
|
||||
CapturedAt: *tc.logTime,
|
||||
CreatedAt: *tc.logTime,
|
||||
}})
|
||||
}
|
||||
|
||||
// Run the purge.
|
||||
done := awaitDoTick(ctx, t, clk)
|
||||
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{
|
||||
Retention: tc.retentionConfig,
|
||||
}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
|
||||
defer closer.Close()
|
||||
testutil.TryReceive(ctx, t, done)
|
||||
|
||||
// Verify session presence/absence.
|
||||
_, err := db.GetBoundarySessionByID(ctx, session.ID)
|
||||
if tc.expectSessionDeleted {
|
||||
require.ErrorIs(t, err, sql.ErrNoRows, "session should have been deleted")
|
||||
} else {
|
||||
require.NoError(t, err, "session should still exist")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeleteExpiredAPIKeys(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user