feat(session): Add repository method to delete terminated sessions
This will delete sessions that have been in the termianted state for longer than the given threshold.
This commit is contained in:
3 files changed
+114
No files matched your search
@@ -388,6 +388,17 @@ where
|
||||
and ss.end_time is null
|
||||
%s
|
||||
;
|
||||
`
|
||||
deleteTerminated = `
|
||||
delete from session
|
||||
using session_state
|
||||
where
|
||||
session.public_id = session_state.session_id
|
||||
and
|
||||
session_state.state = 'terminated'
|
||||
and
|
||||
session_state.start_time < wt_sub_seconds_from_now(@threshold_seconds)
|
||||
;
|
||||
`
|
||||
)
|
||||
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/boundary/internal/boundary"
|
||||
"github.com/hashicorp/boundary/internal/db"
|
||||
@@ -700,6 +701,20 @@ func (r *Repository) checkIfNoLongerActive(ctx context.Context, reportedSessions
|
||||
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
|
||||
|
||||
@@ -1503,3 +1503,91 @@ 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(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)
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user