diff --git a/internal/db/schema/migrations/oss/postgres/92/05_delete_terminated_session_job.up.sql b/internal/db/schema/migrations/oss/postgres/92/05_delete_terminated_session_job.up.sql new file mode 100644 index 000000000..9a07fd162 --- /dev/null +++ b/internal/db/schema/migrations/oss/postgres/92/05_delete_terminated_session_job.up.sql @@ -0,0 +1,34 @@ +-- Copyright (c) HashiCorp, Inc. +-- SPDX-License-Identifier: BUSL-1.1 + +begin; + + create table session_delete_terminated_job ( + batch_size int not null + constraint batch_size_must_be_greater_than_0 + check(batch_size > 0), + create_time wt_timestamp, + update_time wt_timestamp + ); + comment on table session_delete_terminated_job is + 'session_delete_terminated_job is a single row table that contains settings for the delete terminated sessions job.'; + + -- this index ensures that there will only ever be one row in the + -- table. see: + -- https://www.postgresql.org/docs/current/indexes-expressional.html + create unique index session_delete_terminated_job_one_row + on session_delete_terminated_job((batch_size is not null)); + + create trigger immutable_columns before update on session_delete_terminated_job + for each row execute procedure immutable_columns('create_time'); + + create trigger default_create_time_column before insert on session_delete_terminated_job + for each row execute procedure default_create_time(); + + create trigger update_time_column before update on session_delete_terminated_job + for each row execute procedure update_time_column(); + + insert into session_delete_terminated_job(batch_size) values(5000); + +commit; + diff --git a/internal/session/job_delete_terminated_sessions.go b/internal/session/job_delete_terminated_sessions.go index a5d2e9584..e0f9c4cda 100644 --- a/internal/session/job_delete_terminated_sessions.go +++ b/internal/session/job_delete_terminated_sessions.go @@ -5,10 +5,12 @@ package session import ( "context" + "sync" "time" "github.com/hashicorp/boundary/internal/errors" "github.com/hashicorp/boundary/internal/scheduler" + "github.com/hashicorp/boundary/internal/scheduler/batch" ) type deleteTerminatedJob struct { @@ -18,8 +20,8 @@ type deleteTerminatedJob struct { // state for it to be deleted. threshold time.Duration - // the number of sessions deleted in the most recent run - deletedInRun int + mu sync.Mutex + batch *batch.Batch } func newDeleteTerminatedJob(ctx context.Context, repo *Repository, threshold time.Duration) (*deleteTerminatedJob, error) { @@ -38,24 +40,49 @@ func newDeleteTerminatedJob(ctx context.Context, repo *Repository, threshold tim // Status reports the job’s current status. The status is periodically persisted by // the scheduler when a job is running, and will be used to verify a job is making progress. func (d *deleteTerminatedJob) Status() scheduler.JobStatus { - return scheduler.JobStatus{ - Completed: d.deletedInRun, - Total: d.deletedInRun, + d.mu.Lock() + defer d.mu.Unlock() + if d.batch != nil { + return d.batch.Status() } + return scheduler.JobStatus{} } // Run performs the required work depending on the implementation. // The context is used to notify the job that it should exit early. -func (d *deleteTerminatedJob) Run(ctx context.Context, _ time.Duration) error { +func (d *deleteTerminatedJob) Run(ctx context.Context, statusThreshold time.Duration) error { const op = "session.(deleteTerminatedJob).Run" - d.deletedInRun = 0 - var err error - d.deletedInRun, err = d.repo.deleteSessionsTerminatedBefore(ctx, d.threshold) + params, err := d.repo.getDeleteJobParams(ctx, d.threshold) + switch { + case err != nil: + return errors.Wrap(ctx, err, op) + case params.TotalToDelete == 0: + return nil + } + + exec := func() batch.Exec { + return func(ctx context.Context, batchSize int) (int, error) { + return d.repo.deleteTerminatedSessionsBatch(ctx, params.WindowStartTime, batchSize) + } + } + + config := &batch.Config{ + Size: params.BatchSize, + TotalToComplete: params.TotalToDelete, + StatusThreshold: statusThreshold, + Exec: exec(), + Store: d.repo.setDeleteJobBatchSize, + } + + batch, err := batch.New(ctx, config) if err != nil { return errors.Wrap(ctx, err, op) } - return nil + d.mu.Lock() + d.batch = batch + d.mu.Unlock() + return batch.Run(ctx) } // NextRunIn returns the duration until the next job run should be scheduled. This diff --git a/internal/session/job_delete_terminated_sessions_test.go b/internal/session/job_delete_terminated_sessions_test.go index 8c5d49bc4..9aa4cefb2 100644 --- a/internal/session/job_delete_terminated_sessions_test.go +++ b/internal/session/job_delete_terminated_sessions_test.go @@ -34,48 +34,13 @@ func TestDeleteTermiantedSessionsJob(t *testing.T) { threshold time.Duration expected int }{ - { - 0, - 0, - time.Nanosecond, - 0, - }, - { - 1, - 1, - time.Nanosecond, - 1, - }, - { - 1, - 1, - time.Hour, - 0, - }, - { - 10, - 10, - time.Nanosecond, - 10, - }, - { - 10, - 4, - time.Nanosecond, - 4, - }, - { - 10, - 0, - time.Nanosecond, - 0, - }, - { - 10, - 10, - time.Hour, - 0, - }, + {0, 0, time.Nanosecond, 0}, + {1, 1, time.Nanosecond, 1}, + {1, 1, time.Hour, 0}, + {10, 10, time.Nanosecond, 10}, + {10, 4, time.Nanosecond, 4}, + {10, 0, time.Nanosecond, 0}, + {10, 10, time.Hour, 0}, } for _, tc := range cases { @@ -101,9 +66,9 @@ func TestDeleteTermiantedSessionsJob(t *testing.T) { job, err := newDeleteTerminatedJob(ctx, repo, tc.threshold) require.NoError(t, err) - err = job.Run(ctx, 0) + err = job.Run(ctx, 1*time.Second) require.NoError(t, err) - assert.Equal(t, tc.expected, job.deletedInRun) + assert.Equal(t, tc.expected, job.Status().Completed) }) } } diff --git a/internal/session/query.go b/internal/session/query.go index 491c7555a..5d512dc31 100644 --- a/internal/session/query.go +++ b/internal/session/query.go @@ -341,17 +341,6 @@ update session_connection ) %s returning public_id; -` - deleteTerminated = ` -delete from session -using session_state -where - session.public_id = session_state.session_id -and - session_state.state = 'terminated' -and - lower(session_state.active_time_range) < wt_sub_seconds_from_now(@threshold_seconds) -; ` sessionCredentialRewrapQuery = ` select distinct @@ -457,6 +446,42 @@ values ` ) +// queries for the delete terminated sessions job +const ( + getDeleteJobParams = ` +with total (to_delete) as ( + select count(session_id) + from session_state + where session_state.state = 'terminated' + and lower(session_state.active_time_range) < wt_sub_seconds_from_now(@threshold_seconds) +), +params (batch_size) as ( + select batch_size + from session_delete_terminated_job +) +select total.to_delete as total_to_delete, + params.batch_size as batch_size, + wt_sub_seconds_from_now(@threshold_seconds) as window_start_time + from total, params; +` + setDeleteJobBatchSize = ` +update session_delete_terminated_job + set batch_size = @batch_size; +` + deleteTerminatedInBatch = ` +with batch (session_id) as ( + select session_id + from session_state + where state = 'terminated' + and lower(session_state.active_time_range) < @terminated_before + limit @batch_size +) +delete + from session + where public_id in (select session_id from batch); +` +) + func batchInsertSessionCredentialDynamic(creds []*DynamicCredential) (string, []any, error) { if len(creds) <= 0 { return "", nil, fmt.Errorf("empty slice of DynamicCredential, cannot build query") diff --git a/internal/session/repository_jobs.go b/internal/session/repository_jobs.go new file mode 100644 index 000000000..250705c4f --- /dev/null +++ b/internal/session/repository_jobs.go @@ -0,0 +1,72 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package session + +import ( + "context" + "database/sql" + "time" + + "github.com/hashicorp/boundary/internal/db/timestamp" + "github.com/hashicorp/boundary/internal/errors" +) + +type deleteJobParams struct { + TotalToDelete int + BatchSize int + WindowStartTime *timestamp.Timestamp +} + +func (r *Repository) getDeleteJobParams(ctx context.Context, threshold time.Duration) (deleteJobParams, error) { + const op = "session.(Repository).getDeleteJobParams" + + args := []any{ + sql.Named("threshold_seconds", threshold.Seconds()), + } + rows, err := r.reader.Query(ctx, getDeleteJobParams, args) + if err != nil { + return deleteJobParams{}, errors.Wrap(ctx, err, op, errors.WithMsg("error getting parameters for delete terminated sessions job")) + } + defer rows.Close() + + var jobParams deleteJobParams + for rows.Next() { + if err := r.reader.ScanRows(ctx, rows, &jobParams); err != nil { + return deleteJobParams{}, errors.Wrap(ctx, err, op, errors.WithMsg("scan row failed")) + } + } + if err := rows.Err(); err != nil { + return deleteJobParams{}, errors.Wrap(ctx, err, op, errors.WithMsg("next row failed")) + } + return jobParams, nil +} + +func (r *Repository) setDeleteJobBatchSize(ctx context.Context, batchSize int) error { + const op = "session.(Repository).setDeleteJobBatchSize" + + args := []any{ + sql.Named("batch_size", batchSize), + } + + _, err := r.writer.Exec(ctx, setDeleteJobBatchSize, args) + if err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("error setting delete job batch size")) + } + return nil +} + +func (r *Repository) deleteTerminatedSessionsBatch(ctx context.Context, terminatedBefore *timestamp.Timestamp, batchSize int) (int, error) { + const op = "session.(Repository).deleteTerminatedSessionsBatch" + + args := []any{ + sql.Named("terminated_before", terminatedBefore), + sql.Named("batch_size", batchSize), + } + + c, err := r.writer.Exec(ctx, deleteTerminatedInBatch, args) + if err != nil { + return 0, errors.Wrap(ctx, err, op, errors.WithMsg("error deleting terminated sessions")) + } + return c, nil +} diff --git a/internal/session/repository_jobs_test.go b/internal/session/repository_jobs_test.go new file mode 100644 index 000000000..514e89f29 --- /dev/null +++ b/internal/session/repository_jobs_test.go @@ -0,0 +1,185 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package session + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/iam" + "github.com/hashicorp/boundary/internal/kms" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRepository_getDeleteJobParams(t *testing.T) { + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + wrapper := db.TestWrapper(t) + iamRepo := iam.TestRepo(t, conn, wrapper) + kms := kms.TestKms(t, conn, wrapper) + repo, err := NewRepository(ctx, rw, rw, kms) + composedOf := TestSessionParams(t, conn, wrapper, iamRepo) + + const defaultBatchSize = 5000 + + cases := []struct { + sessionCount int + terminateCount int + threshold time.Duration + expectedBatchSize int + expectedTotalToDelete int + }{ + {0, 0, time.Nanosecond, defaultBatchSize, 0}, + {1, 1, time.Nanosecond, defaultBatchSize, 1}, + {1, 1, time.Hour, defaultBatchSize, 0}, + {10, 10, time.Nanosecond, defaultBatchSize, 10}, + {10, 4, time.Nanosecond, defaultBatchSize, 4}, + {10, 0, time.Nanosecond, defaultBatchSize, 0}, + {10, 10, time.Hour, defaultBatchSize, 0}, + } + + for _, tc := range cases { + t.Run(fmt.Sprintf("%d_%d_%s_%d", tc.sessionCount, tc.terminateCount, tc.threshold, tc.expectedBatchSize), func(t *testing.T) { + t.Cleanup(func() { + sdb, err := conn.SqlDB(ctx) + require.NoError(t, err) + _, err = sdb.Exec(`delete from session;`) + require.NoError(t, err) + }) + + for i := 0; i < tc.sessionCount; i++ { + s := TestSession(t, conn, wrapper, composedOf) + if i < tc.terminateCount { + _, err = repo.CancelSession(ctx, s.PublicId, s.Version) + require.NoError(t, err) + } + } + + c, err := repo.TerminateCompletedSessions(ctx) + require.NoError(t, err) + assert.Equal(t, tc.terminateCount, c) + + p, err := repo.getDeleteJobParams(ctx, tc.threshold) + require.NoError(t, err) + assert.Equal(t, tc.expectedBatchSize, p.BatchSize) + assert.Equal(t, tc.expectedTotalToDelete, p.TotalToDelete) + }) + } +} + +func TestRepository_deleteTerminatedSessionsBatch(t *testing.T) { + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + wrapper := db.TestWrapper(t) + iamRepo := iam.TestRepo(t, conn, wrapper) + kms := kms.TestKms(t, conn, wrapper) + repo, err := NewRepository(ctx, rw, rw, kms) + composedOf := TestSessionParams(t, conn, wrapper, iamRepo) + + cases := []struct { + sessionCount int + terminateCount int + batchSize int + expected int + }{ + {0, 0, 10, 0}, + {1, 1, 10, 1}, + {10, 10, 5, 5}, + {10, 2, 5, 2}, + {10, 0, 10, 0}, + } + + for _, tc := range cases { + t.Run(fmt.Sprintf("%d_%d_%d", tc.sessionCount, tc.terminateCount, tc.batchSize), func(t *testing.T) { + t.Cleanup(func() { + sdb, err := conn.SqlDB(ctx) + require.NoError(t, err) + _, err = sdb.Exec(`delete from session;`) + require.NoError(t, err) + }) + + // add initial group of sessions and terminate some of them + { + for i := 0; i < tc.sessionCount; i++ { + s := TestSession(t, conn, wrapper, composedOf) + if i < tc.terminateCount { + _, err = repo.CancelSession(ctx, s.PublicId, s.Version) + require.NoError(t, err) + } + } + + c, err := repo.TerminateCompletedSessions(ctx) + require.NoError(t, err) + assert.Equal(t, tc.terminateCount, c) + } + + // get the job parameters + p, err := repo.getDeleteJobParams(ctx, time.Nanosecond) + require.NoError(t, err) + assert.Equal(t, tc.terminateCount, p.TotalToDelete) + + // add more sessions to test the WindowStartTime + { + for i := 0; i < tc.sessionCount; i++ { + s := TestSession(t, conn, wrapper, composedOf) + if i < tc.terminateCount { + _, err = repo.CancelSession(ctx, s.PublicId, s.Version) + require.NoError(t, err) + } + } + + c, err := repo.TerminateCompletedSessions(ctx) + require.NoError(t, err) + assert.Equal(t, tc.terminateCount, c) + } + + c, err := repo.deleteTerminatedSessionsBatch(ctx, p.WindowStartTime, tc.batchSize) + require.NoError(t, err) + assert.Equal(t, tc.expected, c) + }) + } +} + +func TestRepository_setDeleteJobBatchSize(t *testing.T) { + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + wrapper := db.TestWrapper(t) + kms := kms.TestKms(t, conn, wrapper) + repo, err := NewRepository(ctx, rw, rw, kms) + + cases := []struct { + batchSize int + expectErr bool + }{ + {-1, true}, + {0, true}, + {1, false}, + {9, false}, + {10, false}, + {10000, false}, + {10001, false}, + } + + for _, tc := range cases { + t.Run(fmt.Sprintf("%d_%t", tc.batchSize, tc.expectErr), func(t *testing.T) { + err = repo.setDeleteJobBatchSize(ctx, tc.batchSize) + if tc.expectErr { + assert.Error(t, err) + return + } + assert.NoError(t, err) + + p, err := repo.getDeleteJobParams(ctx, time.Nanosecond) + require.NoError(t, err) + assert.Equal(t, tc.batchSize, p.BatchSize) + }) + } +} diff --git a/internal/session/repository_session.go b/internal/session/repository_session.go index 864db5c26..0b82821a6 100644 --- a/internal/session/repository_session.go +++ b/internal/session/repository_session.go @@ -909,20 +909,6 @@ func (r *Repository) CheckIfNotActive(ctx context.Context, reportedSessions []st return notActive, nil } -func (r *Repository) deleteSessionsTerminatedBefore(ctx context.Context, threshold time.Duration) (int, error) { - const op = "session.(Repository).deleteTerminated" - - args := []any{ - sql.Named("threshold_seconds", threshold.Seconds()), - } - - c, err := r.writer.Exec(ctx, deleteTerminated, args) - if err != nil { - return 0, errors.Wrap(ctx, err, op, errors.WithMsg("error deleting terminated sessions")) - } - return c, nil -} - func fetchStates(ctx context.Context, r db.Reader, sessionId string, opt ...db.Option) ([]*State, error) { const op = "session.fetchStates" var states []*State diff --git a/internal/session/repository_session_test.go b/internal/session/repository_session_test.go index 9e28a9081..a3e03d6bd 100644 --- a/internal/session/repository_session_test.go +++ b/internal/session/repository_session_test.go @@ -5,7 +5,6 @@ package session import ( "context" - "fmt" "testing" "time" @@ -1827,94 +1826,6 @@ func TestRepository_deleteTargetFKey(t *testing.T) { } } -func TestRepository_deleteTerminated(t *testing.T) { - ctx := context.Background() - conn, _ := db.TestSetup(t, "postgres") - rw := db.New(conn) - wrapper := db.TestWrapper(t) - iamRepo := iam.TestRepo(t, conn, wrapper) - kms := kms.TestKms(t, conn, wrapper) - repo, err := NewRepository(ctx, rw, rw, kms) - composedOf := TestSessionParams(t, conn, wrapper, iamRepo) - - cases := []struct { - sessionCount int - terminateCount int - threshold time.Duration - expected int - }{ - { - 0, - 0, - time.Nanosecond, - 0, - }, - { - 1, - 1, - time.Nanosecond, - 1, - }, - { - 1, - 1, - time.Hour, - 0, - }, - { - 10, - 10, - time.Nanosecond, - 10, - }, - { - 10, - 4, - time.Nanosecond, - 4, - }, - { - 10, - 0, - time.Nanosecond, - 0, - }, - { - 10, - 10, - time.Hour, - 0, - }, - } - - for _, tc := range cases { - t.Run(fmt.Sprintf("%d_%d_%s", tc.sessionCount, tc.terminateCount, tc.threshold), func(t *testing.T) { - t.Cleanup(func() { - sdb, err := conn.SqlDB(ctx) - require.NoError(t, err) - _, err = sdb.Exec(`delete from session;`) - require.NoError(t, err) - }) - - for i := 0; i < tc.sessionCount; i++ { - s := TestSession(t, conn, wrapper, composedOf) - if i < tc.terminateCount { - _, err = repo.CancelSession(ctx, s.PublicId, s.Version) - require.NoError(t, err) - } - - } - c, err := repo.TerminateCompletedSessions(ctx) - require.NoError(t, err) - assert.Equal(t, tc.terminateCount, c) - - c, err = repo.deleteSessionsTerminatedBefore(ctx, tc.threshold) - require.NoError(t, err) - assert.Equal(t, tc.expected, c) - }) - } -} - func Test_decrypt(t *testing.T) { conn, _ := db.TestSetup(t, "postgres") wrapper := db.TestWrapper(t)