-
Notifications
You must be signed in to change notification settings - Fork 289
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(session): Delete terminated sessions in batches
This changes the periodic job that deletes terminated sessions to delete the those sessions in batches. Prior to this, a single SQL DELETE command was used to delete all qualifying terminated sessions. In systems under heavy load, the statement could timeout leading to a situation where no qualifying terminated sessions were ever being deleted.
- Loading branch information
Showing
8 changed files
with
373 additions
and
168 deletions.
There are no files selected for viewing
34 changes: 34 additions & 0 deletions
34
internal/db/schema/migrations/oss/postgres/92/05_delete_terminated_session_job.up.sql
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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; | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| } |
Oops, something went wrong.