From 9c3eb1604618854e4c63cdbc250e97c3d91d41b2 Mon Sep 17 00:00:00 2001 From: Michael Milton Date: Fri, 1 Dec 2023 16:02:43 -0500 Subject: [PATCH] Sessions list pagination (#4079) * Session list pagination refactor * address pr comments, add explanatory comments, clean up sql queries, move tests into proper files * fix build error * fix test error --- .../common/scopeids/scope_ids_test.go | 2 +- internal/daemon/controller/handler.go | 2 +- .../sessions/session_list_benchmarks_test.go | 2 +- .../handlers/sessions/session_service.go | 181 +++-- .../handlers/sessions/session_service_test.go | 500 +++++++++++- .../79/03_session_base_table_updates.up.sql | 14 + .../db/sqltest/tests/pagination/session.sql | 12 + internal/gen/controller.swagger.json | 43 ++ .../api/services/session_service.pb.go | 209 +++-- .../api/services/v1/session_service.proto | 29 + internal/session/options.go | 10 + internal/session/query.go | 63 ++ internal/session/repository_session.go | 188 +++-- internal/session/repository_session_test.go | 260 +++++-- internal/session/service_list.go | 56 ++ internal/session/service_list_ext_test.go | 723 ++++++++++++++++++ internal/session/service_list_page.go | 70 ++ internal/session/service_list_refresh.go | 76 ++ internal/session/service_list_refresh_page.go | 83 ++ internal/session/session.go | 38 + 20 files changed, 2339 insertions(+), 222 deletions(-) create mode 100644 internal/db/schema/migrations/oss/postgres/79/03_session_base_table_updates.up.sql create mode 100644 internal/db/sqltest/tests/pagination/session.sql create mode 100644 internal/session/service_list.go create mode 100644 internal/session/service_list_ext_test.go create mode 100644 internal/session/service_list_page.go create mode 100644 internal/session/service_list_refresh.go create mode 100644 internal/session/service_list_refresh_page.go diff --git a/internal/daemon/controller/common/scopeids/scope_ids_test.go b/internal/daemon/controller/common/scopeids/scope_ids_test.go index 6590ba827..a439b4a28 100644 --- a/internal/daemon/controller/common/scopeids/scope_ids_test.go +++ b/internal/daemon/controller/common/scopeids/scope_ids_test.go @@ -51,7 +51,7 @@ func TestListingScopeIds(t *testing.T) { sessionsRepoFn := func(opt ...session.Option) (*session.Repository, error) { return session.NewRepository(ctx, rw, rw, kms, opt...) } - sess, err := sessions.NewService(ctx, sessionsRepoFn, iamRepoFn) + sess, err := sessions.NewService(ctx, sessionsRepoFn, iamRepoFn, 1000) require.NoError(t, err) tcs := []struct { diff --git a/internal/daemon/controller/handler.go b/internal/daemon/controller/handler.go index 7f945db3d..8273ff90b 100644 --- a/internal/daemon/controller/handler.go +++ b/internal/daemon/controller/handler.go @@ -252,7 +252,7 @@ func (c *Controller) registerGrpcServices(s *grpc.Server) error { services.RegisterRoleServiceServer(s, rs) } if _, ok := currentServices[services.SessionService_ServiceDesc.ServiceName]; !ok { - ss, err := sessions.NewService(c.baseContext, c.SessionRepoFn, c.IamRepoFn) + ss, err := sessions.NewService(c.baseContext, c.SessionRepoFn, c.IamRepoFn, c.conf.RawConfig.Controller.MaxPageSize) if err != nil { return fmt.Errorf("failed to create session handler service: %w", err) } diff --git a/internal/daemon/controller/handlers/sessions/session_list_benchmarks_test.go b/internal/daemon/controller/handlers/sessions/session_list_benchmarks_test.go index 4269e421e..d7b05dbce 100644 --- a/internal/daemon/controller/handlers/sessions/session_list_benchmarks_test.go +++ b/internal/daemon/controller/handlers/sessions/session_list_benchmarks_test.go @@ -109,7 +109,7 @@ func BenchmarkSessionList(b *testing.B) { return serversRepo, nil } - s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn) + s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000) require.NoError(b, err) var users []*userWithToken diff --git a/internal/daemon/controller/handlers/sessions/session_service.go b/internal/daemon/controller/handlers/sessions/session_service.go index 4c24a8878..7df9fed10 100644 --- a/internal/daemon/controller/handlers/sessions/session_service.go +++ b/internal/daemon/controller/handlers/sessions/session_service.go @@ -14,6 +14,8 @@ import ( "github.com/hashicorp/boundary/internal/daemon/controller/handlers" "github.com/hashicorp/boundary/internal/errors" pbs "github.com/hashicorp/boundary/internal/gen/controller/api/services" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/pagination" "github.com/hashicorp/boundary/internal/perms" "github.com/hashicorp/boundary/internal/requests" "github.com/hashicorp/boundary/internal/session" @@ -53,14 +55,15 @@ func init() { type Service struct { pbs.UnsafeSessionServiceServer - repoFn session.RepositoryFactory - iamRepoFn common.IamRepoFactory + repoFn session.RepositoryFactory + iamRepoFn common.IamRepoFactory + maxPageSize uint } var _ pbs.SessionServiceServer = (*Service)(nil) // NewService returns a session service which handles session related requests to boundary. -func NewService(ctx context.Context, repoFn session.RepositoryFactory, iamRepoFn common.IamRepoFactory) (Service, error) { +func NewService(ctx context.Context, repoFn session.RepositoryFactory, iamRepoFn common.IamRepoFactory, maxPageSize uint) (Service, error) { const op = "sessions.NewService" if repoFn == nil { return Service{}, errors.New(ctx, errors.InvalidParameter, op, "missing session repository") @@ -68,7 +71,10 @@ func NewService(ctx context.Context, repoFn session.RepositoryFactory, iamRepoFn if iamRepoFn == nil { return Service{}, errors.New(ctx, errors.InvalidParameter, op, "missing iam repository") } - return Service{repoFn: repoFn, iamRepoFn: iamRepoFn}, nil + if maxPageSize == 0 { + maxPageSize = uint(globals.DefaultMaxPageSize) + } + return Service{repoFn: repoFn, iamRepoFn: iamRepoFn, maxPageSize: maxPageSize}, nil } // GetSessions implements the interface pbs.SessionServiceServer. @@ -130,7 +136,7 @@ func (s Service) ListSessions(ctx context.Context, req *pbs.ListSessionsRequest) const op = "session.(Service).ListSessions" if err := validateListRequest(ctx, req); err != nil { - return nil, err + return nil, errors.Wrap(ctx, err, op) } authResults := s.authResult(ctx, req.GetScopeId(), action.List, false) @@ -143,13 +149,12 @@ func (s Service) ListSessions(ctx context.Context, req *pbs.ListSessionsRequest) req.GetRecursive() && authResults.AuthenticationFinished { } else { - return nil, authResults.Error + return nil, errors.Wrap(ctx, authResults.Error, op) } } var scopeIds map[string]*scopes.ScopeInfo var err error - if !req.GetRecursive() { scopeIds = map[string]*scopes.ScopeInfo{authResults.Scope.Id: authResults.Scope} } else { @@ -159,6 +164,42 @@ func (s Service) ListSessions(ctx context.Context, req *pbs.ListSessionsRequest) } } + pageSize := int(s.maxPageSize) + // Use the requested page size only if it is smaller than + // the configured max. + if req.GetPageSize() != 0 && uint(req.GetPageSize()) < s.maxPageSize { + pageSize = int(req.GetPageSize()) + } + var filterItemFn func(ctx context.Context, item *session.Session) (bool, error) + switch { + case req.GetFilter() != "": + // Only use a filter if we need to + filter, err := handlers.NewFilter(ctx, req.GetFilter()) + if err != nil { + return nil, err + } + filterItemFn = func(ctx context.Context, item *session.Session) (bool, error) { + outputOpts, ok := newOutputOpts(ctx, item, scopeIds, authResults) + if !ok { + return false, nil + } + pbItem, err := toProto(ctx, item, outputOpts...) + if err != nil { + return false, err + } + return filter.Match(pbItem), nil + } + default: + filterItemFn = func(ctx context.Context, item *session.Session) (bool, error) { + return true, nil + } + } + + grantsHash, err := authResults.GrantsHash(ctx) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + listPerms := authResults.ACL().ListPermissions(scopeIds, resource.Session, IdActions, authResults.UserId) repo, err := s.repoFn(session.WithPermissions(&perms.UserPermissions{ @@ -169,51 +210,76 @@ func (s Service) ListSessions(ctx context.Context, req *pbs.ListSessionsRequest) return nil, errors.Wrap(ctx, err, op) } - sesList, err := repo.ListSessions(ctx, session.WithTerminated(req.GetIncludeTerminated()), session.WithLimit(-1)) - if err != nil { - return nil, err - } - if len(sesList) == 0 { - return &pbs.ListSessionsResponse{}, nil - } - - filter, err := handlers.NewFilter(ctx, req.GetFilter()) - if err != nil { - return nil, err - } - finalItems := make([]*pb.Session, 0, len(sesList)) - res := perms.Resource{ - Type: resource.Session, - } - for _, item := range sesList { - res.Id = item.GetPublicId() - res.ScopeId = item.GetProjectId() - authorizedActions := authResults.FetchActionSetForId(ctx, item.GetPublicId(), IdActions, auth.WithResource(&res)).Strings() - if len(authorizedActions) == 0 { - continue - } - - outputFields := authResults.FetchOutputFields(res, action.List).SelfOrDefaults(authResults.UserId) - outputOpts := make([]handlers.Option, 0, 3) - outputOpts = append(outputOpts, handlers.WithOutputFields(outputFields)) - if outputFields.Has(globals.ScopeField) { - outputOpts = append(outputOpts, handlers.WithScope(scopeIds[item.ProjectId])) - } - if outputFields.Has(globals.AuthorizedActionsField) { - outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authorizedActions)) - } - - item, err := toProto(ctx, item, outputOpts...) + var listResp *pagination.ListResponse[*session.Session] + var sortBy string + includeTerminated := req.GetIncludeTerminated() + if req.GetListToken() == "" { + sortBy = "created_time" + listResp, err = session.List(ctx, grantsHash, pageSize, filterItemFn, repo, includeTerminated) if err != nil { return nil, err } - - if filter.Match(item) { - finalItems = append(finalItems, item) + } else { + listToken, err := handlers.ParseListToken(ctx, req.GetListToken(), resource.Session, grantsHash) + if err != nil { + return nil, err + } + switch st := listToken.Subtype.(type) { + case *listtoken.PaginationToken: + sortBy = "created_time" + listResp, err = session.ListPage(ctx, grantsHash, pageSize, filterItemFn, listToken, repo, includeTerminated) + if err != nil { + return nil, err + } + case *listtoken.StartRefreshToken: + sortBy = "updated_time" + listResp, err = session.ListRefresh(ctx, grantsHash, pageSize, filterItemFn, listToken, repo, includeTerminated) + if err != nil { + return nil, err + } + case *listtoken.RefreshToken: + sortBy = "updated_time" + listResp, err = session.ListRefreshPage(ctx, grantsHash, pageSize, filterItemFn, listToken, repo, includeTerminated) + if err != nil { + return nil, err + } + default: + return nil, handlers.ApiErrorWithCodeAndMessage(codes.InvalidArgument, "unexpected list token subtype: %T", st) } } - return &pbs.ListSessionsResponse{Items: finalItems}, nil + finalItems := make([]*pb.Session, 0, len(listResp.Items)) + for _, item := range listResp.Items { + outputOpts, ok := newOutputOpts(ctx, item, scopeIds, authResults) + if !ok { + continue + } + item, err := toProto(ctx, item, outputOpts...) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + finalItems = append(finalItems, item) + } + respType := "delta" + if listResp.CompleteListing { + respType = "complete" + } + resp := &pbs.ListSessionsResponse{ + Items: finalItems, + EstItemCount: uint32(listResp.EstimatedItemCount), + RemovedIds: listResp.DeletedIds, + ResponseType: respType, + SortBy: sortBy, + SortDir: "desc", + } + + if listResp.ListToken != nil { + resp.ListToken, err = handlers.MarshalListToken(ctx, listResp.ListToken, pbs.ResourceType_RESOURCE_TYPE_SESSION) + if err != nil { + return nil, err + } + } + return resp, nil } // CancelSession implements the interface pbs.SessionServiceServer. @@ -506,3 +572,26 @@ func validateCancelRequest(req *pbs.CancelSessionRequest) error { } return nil } + +func newOutputOpts(ctx context.Context, item *session.Session, scopeIds map[string]*scopes.ScopeInfo, authResults auth.VerifyResults) ([]handlers.Option, bool) { + res := perms.Resource{ + Type: resource.Session, + Id: item.GetPublicId(), + ScopeId: item.GetProjectId(), + } + authorizedActions := authResults.FetchActionSetForId(ctx, item.GetPublicId(), IdActions, auth.WithResource(&res)).Strings() + if len(authorizedActions) == 0 { + return nil, false + } + + outputFields := authResults.FetchOutputFields(res, action.List).SelfOrDefaults(authResults.UserId) + outputOpts := make([]handlers.Option, 0, 3) + outputOpts = append(outputOpts, handlers.WithOutputFields(outputFields)) + if outputFields.Has(globals.ScopeField) { + outputOpts = append(outputOpts, handlers.WithScope(scopeIds[item.ProjectId])) + } + if outputFields.Has(globals.AuthorizedActionsField) { + outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authorizedActions)) + } + return outputOpts, true +} diff --git a/internal/daemon/controller/handlers/sessions/session_service_test.go b/internal/daemon/controller/handlers/sessions/session_service_test.go index f121285a8..898a1008d 100644 --- a/internal/daemon/controller/handlers/sessions/session_service_test.go +++ b/internal/daemon/controller/handlers/sessions/session_service_test.go @@ -34,6 +34,7 @@ import ( pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/sessions" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "golang.org/x/exp/slices" "google.golang.org/grpc/codes" "google.golang.org/protobuf/testing/protocmp" ) @@ -139,7 +140,7 @@ func TestGetSession(t *testing.T) { t.Run(tc.name, func(t *testing.T) { assert, require := assert.New(t), require.New(t) - s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn) + s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000) require.NoError(err, "Couldn't create new session service.") requestInfo := authpb.RequestInfo{ @@ -233,7 +234,7 @@ func TestList_Self(t *testing.T) { Endpoint: "tcp://127.0.0.1:22", }) - s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn) + s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000) require.NoError(t, err, "Couldn't create new session service.") cases := []struct { @@ -283,6 +284,9 @@ func TestList(t *testing.T) { kms := kms.TestKms(t, conn, wrap) ctx := context.Background() + sqlDB, err := conn.SqlDB(ctx) + require.NoError(t, err) + iamRepo := iam.TestRepo(t, conn, wrap) rw := db.New(conn) @@ -456,6 +460,16 @@ func TestList(t *testing.T) { wantIncludeTerminatedSessions = append(wantIncludeTerminatedSessions, expected) } + // Run analyze to update postgres estimates + _, err = sqlDB.ExecContext(ctx, "analyze") + require.NoError(t, err) + + // Reverse slices since response is ordered by created_time descending (newest first) + slices.Reverse(wantSession) + slices.Reverse(wantOtherSession) + slices.Reverse(wantAllSessions) + slices.Reverse(wantIncludeTerminatedSessions) + cases := []struct { name string req *pbs.ListSessionsRequest @@ -465,38 +479,92 @@ func TestList(t *testing.T) { err error }{ { - name: "List Many Sessions", - req: &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId()}, - res: &pbs.ListSessionsResponse{Items: wantSession}, - otherRes: &pbs.ListSessionsResponse{Items: []*pb.Session{}}, + name: "List Many Sessions", + req: &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId()}, + res: &pbs.ListSessionsResponse{ + Items: wantSession, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + EstItemCount: 10, + }, + otherRes: &pbs.ListSessionsResponse{ + Items: []*pb.Session{}, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + }, allSessionRes: &pbs.ListSessionsResponse{Items: wantSession}, }, { - name: "List Many Include Terminated", - req: &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), IncludeTerminated: true}, - res: &pbs.ListSessionsResponse{Items: wantIncludeTerminatedSessions}, - otherRes: &pbs.ListSessionsResponse{Items: []*pb.Session{}}, + name: "List Many Include Terminated", + req: &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), IncludeTerminated: true}, + res: &pbs.ListSessionsResponse{ + Items: wantIncludeTerminatedSessions, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + EstItemCount: 11, + }, + otherRes: &pbs.ListSessionsResponse{ + Items: []*pb.Session{}, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + }, allSessionRes: &pbs.ListSessionsResponse{Items: wantIncludeTerminatedSessions}, }, { - name: "List No Sessions", - req: &pbs.ListSessionsRequest{ScopeId: pNoSessions.GetPublicId()}, - res: &pbs.ListSessionsResponse{}, - otherRes: &pbs.ListSessionsResponse{Items: []*pb.Session{}}, + name: "List No Sessions", + req: &pbs.ListSessionsRequest{ScopeId: pNoSessions.GetPublicId()}, + res: &pbs.ListSessionsResponse{ + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + }, + otherRes: &pbs.ListSessionsResponse{ + Items: []*pb.Session{}, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + }, allSessionRes: &pbs.ListSessionsResponse{}, }, { - name: "List Sessions Recursively", - req: &pbs.ListSessionsRequest{ScopeId: scope.Global.String(), Recursive: true}, - res: &pbs.ListSessionsResponse{Items: wantSession}, - otherRes: &pbs.ListSessionsResponse{Items: wantOtherSession}, + name: "List Sessions Recursively", + req: &pbs.ListSessionsRequest{ScopeId: scope.Global.String(), Recursive: true}, + res: &pbs.ListSessionsResponse{ + Items: wantSession, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + EstItemCount: 10, + }, + otherRes: &pbs.ListSessionsResponse{ + Items: wantOtherSession, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + EstItemCount: 10, + }, allSessionRes: &pbs.ListSessionsResponse{Items: wantAllSessions}, }, { - name: "Filter To Single Sessions", - req: &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Filter: fmt.Sprintf(`"/item/id"==%q`, wantSession[4].Id)}, - res: &pbs.ListSessionsResponse{Items: wantSession[4:5]}, - otherRes: &pbs.ListSessionsResponse{Items: []*pb.Session{}}, + name: "Filter To Single Sessions", + req: &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Filter: fmt.Sprintf(`"/item/id"==%q`, wantSession[4].Id)}, + res: &pbs.ListSessionsResponse{ + Items: wantSession[4:5], + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + EstItemCount: 1, + }, + otherRes: &pbs.ListSessionsResponse{ + Items: []*pb.Session{}, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + }, allSessionRes: &pbs.ListSessionsResponse{Items: wantSession[4:5]}, }, { @@ -505,17 +573,55 @@ func TestList(t *testing.T) { ScopeId: scope.Global.String(), Recursive: true, Filter: fmt.Sprintf(`"/item/scope/id" matches "^%s"`, pWithSessions.GetPublicId()[:8]), }, - res: &pbs.ListSessionsResponse{Items: wantSession}, - otherRes: &pbs.ListSessionsResponse{Items: []*pb.Session{}}, + res: &pbs.ListSessionsResponse{ + Items: wantSession, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + EstItemCount: 10, + }, + otherRes: &pbs.ListSessionsResponse{ + Items: []*pb.Session{}, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + }, allSessionRes: &pbs.ListSessionsResponse{Items: wantSession}, }, { - name: "Filter To Nothing", - req: &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Filter: `"/item/id" == ""`}, - res: &pbs.ListSessionsResponse{}, - otherRes: &pbs.ListSessionsResponse{Items: []*pb.Session{}}, + name: "Filter To Nothing", + req: &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Filter: `"/item/id" == ""`}, + res: &pbs.ListSessionsResponse{ + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + }, + otherRes: &pbs.ListSessionsResponse{ + Items: []*pb.Session{}, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + }, allSessionRes: &pbs.ListSessionsResponse{}, }, + { + name: "Paginate listing", + req: &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Recursive: true, PageSize: 2}, + res: &pbs.ListSessionsResponse{ + Items: wantSession[:2], + ResponseType: "delta", + SortBy: "created_time", + SortDir: "desc", + EstItemCount: 21, + }, + otherRes: &pbs.ListSessionsResponse{ + Items: []*pb.Session{}, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + }, + allSessionRes: &pbs.ListSessionsResponse{Items: wantSession[:2]}, + }, { name: "Filter Bad Format", req: &pbs.ListSessionsRequest{ScopeId: pWithSessions.GetPublicId(), Filter: `//badformat/`}, @@ -526,7 +632,7 @@ func TestList(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { require, assert := require.New(t), assert.New(t) - s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn) + s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000) require.NoError(err, "Couldn't create new session service.") // Test without anon user @@ -545,7 +651,7 @@ func TestList(t *testing.T) { } require.NoError(gErr) if tc.res != nil { - require.Equal(len(tc.res.GetItems()), len(got.GetItems()), "Didn't get expected number of sessions: %v", got.GetItems()) + require.Equal(len(tc.res.GetItems()), len(got.GetItems()), "Didn't get expected number of sessions: %v", len(got.GetItems())) for i, wantSess := range tc.res.GetItems() { assert.True(got.GetItems()[i].GetExpirationTime().AsTime().Sub(wantSess.GetExpirationTime().AsTime()) < 10*time.Millisecond) assert.Equal(0, len(wantSess.GetConnections())) // no connections on list @@ -605,6 +711,338 @@ func TestList(t *testing.T) { } } +func TestListPagination(t *testing.T) { + // Set database read timeout to avoid duplicates in response + oldReadTimeout := globals.RefreshReadLookbackDuration + globals.RefreshReadLookbackDuration = 0 + t.Cleanup(func() { + globals.RefreshReadLookbackDuration = oldReadTimeout + }) + conn, _ := db.TestSetup(t, "postgres") + wrap := db.TestWrapper(t) + kms := kms.TestKms(t, conn, wrap) + ctx := context.Background() + + sqlDB, err := conn.SqlDB(ctx) + require.NoError(t, err) + + iamRepo := iam.TestRepo(t, conn, wrap) + + rw := db.New(conn) + sessRepo, err := session.NewRepository(ctx, rw, rw, kms) + require.NoError(t, err) + + iamRepoFn := func() (*iam.Repository, error) { + return iamRepo, nil + } + sessRepoFn := func(opt ...session.Option) (*session.Repository, error) { + return session.NewRepository(ctx, rw, rw, kms, opt...) + } + tokenRepoFn := func() (*authtoken.Repository, error) { + return authtoken.NewRepository(ctx, rw, rw, kms) + } + serversRepoFn := func() (*server.Repository, error) { + return server.NewRepository(ctx, rw, rw, kms) + } + + o, pWithSessions := iam.TestScopes(t, iamRepo) + + at := authtoken.TestAuthToken(t, conn, kms, o.GetPublicId()) + uId := at.GetIamUserId() + + hc := static.TestCatalogs(t, conn, pWithSessions.GetPublicId(), 1)[0] + hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0] + h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0] + static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h}) + tar := tcp.TestTarget(ctx, t, conn, pWithSessions.GetPublicId(), "test", target.WithHostSources([]string{hs.GetPublicId()})) + + var allSessions []*pb.Session + for i := 0; i < 10; i++ { + sess := session.TestSession(t, conn, wrap, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ProjectId: pWithSessions.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + }) + + session.TestConnection(t, conn, sess.PublicId, "127.0.0.1", 22, "127.0.0.2", 23, "127.0.0.1") + + status, states := convertStates(sess.States) + + firstOrgSession := &pb.Session{ + Id: sess.GetPublicId(), + ScopeId: pWithSessions.GetPublicId(), + AuthTokenId: at.GetPublicId(), + UserId: at.GetIamUserId(), + TargetId: sess.TargetId, + Endpoint: sess.Endpoint, + HostSetId: sess.HostSetId, + HostId: sess.HostId, + Version: sess.Version, + UpdatedTime: sess.UpdateTime.GetTimestamp(), + CreatedTime: sess.CreateTime.GetTimestamp(), + ExpirationTime: sess.ExpirationTime.GetTimestamp(), + Scope: &scopes.ScopeInfo{Id: pWithSessions.GetPublicId(), Type: scope.Project.String(), ParentScopeId: o.GetPublicId()}, + Status: status, + States: states, + Certificate: sess.Certificate, + Type: tcp.Subtype.String(), + AuthorizedActions: testAuthorizedActions, + Connections: []*pb.Connection{}, // connections should not be returned for list + } + allSessions = append(allSessions, firstOrgSession) + } + // Reverse slices since response is ordered by created_time descending (newest first) + slices.Reverse(allSessions) + + // Run analyze to update postgres estimates + _, err = sqlDB.ExecContext(ctx, "analyze") + require.NoError(t, err) + + requestInfo := authpb.RequestInfo{ + TokenFormat: uint32(auth.AuthTokenTypeBearer), + PublicId: at.GetPublicId(), + Token: at.GetToken(), + } + requestContext := context.WithValue(context.Background(), requests.ContextRequestInformationKey, &requests.RequestContext{}) + ctx = auth.NewVerifierContext(requestContext, iamRepoFn, tokenRepoFn, serversRepoFn, kms, &requestInfo) + + s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000) + require.NoError(t, err, "Couldn't create new session service.") + + // Start paginating, recursively + req := &pbs.ListSessionsRequest{ + ScopeId: "global", + Recursive: true, + Filter: "", + ListToken: "", + PageSize: 2, + } + got, err := s.ListSessions(ctx, req) + require.NoError(t, err) + require.Len(t, got.GetItems(), 2) + // Compare without comparing the list token + assert.Empty(t, + cmp.Diff( + got, + &pbs.ListSessionsResponse{ + Items: allSessions[0:2], + ResponseType: "delta", + SortBy: "created_time", + SortDir: "desc", + RemovedIds: nil, + EstItemCount: 10, + }, + protocmp.Transform(), + protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"), + ), + ) + + // Request second page + req.ListToken = got.ListToken + got, err = s.ListSessions(ctx, req) + require.NoError(t, err) + require.Len(t, got.GetItems(), 2) + // Compare without comparing the list token + assert.Empty(t, + cmp.Diff( + got, + &pbs.ListSessionsResponse{ + Items: allSessions[2:4], + ResponseType: "delta", + SortBy: "created_time", + SortDir: "desc", + RemovedIds: nil, + EstItemCount: 10, + }, + protocmp.Transform(), + protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"), + ), + ) + + // Request rest of results + req.ListToken = got.ListToken + req.PageSize = 10 + got, err = s.ListSessions(ctx, req) + require.NoError(t, err) + require.Len(t, got.GetItems(), 6) + // Compare without comparing the list token + assert.Empty(t, + cmp.Diff( + got, + &pbs.ListSessionsResponse{ + Items: allSessions[4:], + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + RemovedIds: nil, + EstItemCount: 10, + }, + protocmp.Transform(), + protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"), + ), + ) + + // Create another session + sess := session.TestSession(t, conn, wrap, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ProjectId: pWithSessions.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + }) + + session.TestConnection(t, conn, sess.PublicId, "127.0.0.1", 22, "127.0.0.2", 23, "127.0.0.1") + + status, states := convertStates(sess.States) + + newSession := &pb.Session{ + Id: sess.GetPublicId(), + ScopeId: pWithSessions.GetPublicId(), + AuthTokenId: at.GetPublicId(), + UserId: at.GetIamUserId(), + TargetId: sess.TargetId, + Endpoint: sess.Endpoint, + HostSetId: sess.HostSetId, + HostId: sess.HostId, + Version: sess.Version, + UpdatedTime: sess.UpdateTime.GetTimestamp(), + CreatedTime: sess.CreateTime.GetTimestamp(), + ExpirationTime: sess.ExpirationTime.GetTimestamp(), + Scope: &scopes.ScopeInfo{Id: pWithSessions.GetPublicId(), Type: scope.Project.String(), ParentScopeId: o.GetPublicId()}, + Status: status, + States: states, + Certificate: sess.Certificate, + Type: tcp.Subtype.String(), + AuthorizedActions: testAuthorizedActions, + Connections: []*pb.Connection{}, // connections should not be returned for list + } + // Add to the front since it's most recently updated + allSessions = append([]*pb.Session{newSession}, allSessions...) + + // Delete one of the other sessions + _, err = sessRepo.DeleteSession(ctx, allSessions[len(allSessions)-1].Id) + require.NoError(t, err) + deletedSession := allSessions[len(allSessions)-1] + allSessions = allSessions[:len(allSessions)-1] + + // cancel session in lieu of updating it + canceledSession, err := sessRepo.CancelSession(ctx, allSessions[1].Id, allSessions[1].Version) + require.NoError(t, err) + canceledStatus, canceledStates := convertStates(canceledSession.States) + allSessions[1].Status = canceledStatus + allSessions[1].States = canceledStates + allSessions[1].UpdatedTime = canceledSession.UpdateTime.GetTimestamp() + allSessions[1].Version = 2 + allSessions = append( + []*pb.Session{allSessions[1]}, + append( + []*pb.Session{allSessions[0]}, + allSessions[2:]..., + )..., + ) + + // Run analyze to update postgres estimates + _, err = sqlDB.ExecContext(ctx, "analyze") + require.NoError(t, err) + + // Request updated results + req.ListToken = got.ListToken + req.PageSize = 1 + got, err = s.ListSessions(ctx, req) + require.NoError(t, err) + require.Len(t, got.GetItems(), 1) + // Compare without comparing the list token + assert.Empty(t, + cmp.Diff( + got, + &pbs.ListSessionsResponse{ + Items: []*pb.Session{allSessions[0]}, + ResponseType: "delta", + SortBy: "updated_time", + SortDir: "desc", + // Should contain the deleted session + RemovedIds: []string{deletedSession.Id}, + EstItemCount: 10, + }, + protocmp.Transform(), + protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"), + ), + ) + + // Get next page + req.ListToken = got.ListToken + got, err = s.ListSessions(ctx, req) + require.NoError(t, err) + require.Len(t, got.GetItems(), 1) + // Compare without comparing the list token + assert.Empty(t, + cmp.Diff( + got, + &pbs.ListSessionsResponse{ + Items: []*pb.Session{allSessions[1]}, + ResponseType: "complete", + SortBy: "updated_time", + SortDir: "desc", + RemovedIds: nil, + EstItemCount: 10, + }, + protocmp.Transform(), + protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"), + ), + ) + + // Request new page with filter requiring looping + // to fill the page. + req.ListToken = "" + req.PageSize = 1 + req.Filter = fmt.Sprintf(`"/item/id"==%q or "/item/id"==%q`, allSessions[len(allSessions)-2].Id, allSessions[len(allSessions)-1].Id) + got, err = s.ListSessions(ctx, req) + require.NoError(t, err) + require.Len(t, got.GetItems(), 1) + assert.Empty(t, + cmp.Diff( + got, + &pbs.ListSessionsResponse{ + Items: []*pb.Session{allSessions[len(allSessions)-2]}, + ResponseType: "delta", + SortBy: "created_time", + SortDir: "desc", + // Should be empty again + RemovedIds: nil, + EstItemCount: 10, + }, + protocmp.Transform(), + protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"), + ), + ) + req.ListToken = got.ListToken + // Get the second page + got, err = s.ListSessions(ctx, req) + require.NoError(t, err) + require.Len(t, got.GetItems(), 1) + assert.Empty(t, + cmp.Diff( + got, + &pbs.ListSessionsResponse{ + Items: []*pb.Session{allSessions[len(allSessions)-1]}, + ResponseType: "complete", + SortBy: "created_time", + SortDir: "desc", + RemovedIds: nil, + EstItemCount: 10, + }, + protocmp.Transform(), + protocmp.IgnoreFields(&pbs.ListSessionsResponse{}, "list_token"), + ), + ) +} + func convertStates(in []*session.State) (string, []*pb.SessionState) { out := make([]*pb.SessionState, 0, len(in)) for _, s := range in { @@ -727,7 +1165,7 @@ func TestCancel(t *testing.T) { t.Run(tc.name, func(t *testing.T) { assert, require := assert.New(t), require.New(t) - s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn) + s, err := sessions.NewService(ctx, sessRepoFn, iamRepoFn, 1000) require.NoError(err, "Couldn't create new session service.") tc.req.Version = version diff --git a/internal/db/schema/migrations/oss/postgres/79/03_session_base_table_updates.up.sql b/internal/db/schema/migrations/oss/postgres/79/03_session_base_table_updates.up.sql new file mode 100644 index 000000000..1bffedf3f --- /dev/null +++ b/internal/db/schema/migrations/oss/postgres/79/03_session_base_table_updates.up.sql @@ -0,0 +1,14 @@ +-- Copyright (c) HashiCorp, Inc. +-- SPDX-License-Identifier: BUSL-1.1 + +begin; + + -- Add new indexes for the create time and update time queries. + create index session_create_time_public_id_idx + on session (create_time desc, public_id asc); + create index session_update_time_public_id_idx + on session (update_time desc, public_id asc); + + analyze session; + +commit; \ No newline at end of file diff --git a/internal/db/sqltest/tests/pagination/session.sql b/internal/db/sqltest/tests/pagination/session.sql new file mode 100644 index 000000000..f1ed46374 --- /dev/null +++ b/internal/db/sqltest/tests/pagination/session.sql @@ -0,0 +1,12 @@ +-- Copyright (c) HashiCorp, Inc. +-- SPDX-License-Identifier: BUSL-1.1 + +begin; + select plan(2); + + select has_index('session', 'session_create_time_public_id_idx', array['create_time', 'public_id']); + select has_index('session', 'session_update_time_public_id_idx', array['update_time', 'public_id']); + + select * from finish(); + +rollback; \ No newline at end of file diff --git a/internal/gen/controller.swagger.json b/internal/gen/controller.swagger.json index 2d37a0ebd..b0ef8bd20 100644 --- a/internal/gen/controller.swagger.json +++ b/internal/gen/controller.swagger.json @@ -2915,6 +2915,21 @@ "in": "query", "required": false, "type": "boolean" + }, + { + "name": "list_token", + "description": "An opaque token used to continue an existing iteration or\nrequest updated items. If not specified, pagination\nwill start from the beginning.", + "in": "query", + "required": false, + "type": "string" + }, + { + "name": "page_size", + "description": "The maximum size of a page in this iteration.\nIf unset, the default page size configured will be used.\nIf the page_size is greater than the default page configured,\nan error will be returned.", + "in": "query", + "required": false, + "type": "integer", + "format": "int64" } ], "tags": [ @@ -7675,6 +7690,34 @@ "type": "object", "$ref": "#/definitions/controller.api.resources.sessions.v1.Session" } + }, + "response_type": { + "type": "string", + "description": "The type of response, either \"delta\" or \"complete\".\nDelta signifies that this is part of a paginated result\nor an update to a previously completed pagination.\nComplete signifies that it is the last page." + }, + "list_token": { + "type": "string", + "description": "An opaque token used to continue an existing pagination or\nrequest updated items. Use this token in the next list request\nto request the next page." + }, + "sort_by": { + "type": "string", + "description": "The name of the field which the items are sorted by." + }, + "sort_dir": { + "type": "string", + "description": "The direction of the sort, either \"asc\" or \"desc\"." + }, + "removed_ids": { + "type": "array", + "items": { + "type": "string" + }, + "description": "A list of item IDs that have been removed since they were returned\nas part of a pagination. They should be dropped from any client cache.\nThis may contain items that are not known to the cache, if they were\ncreated and deleted between listings." + }, + "est_item_count": { + "type": "integer", + "format": "int64", + "description": "An estimate at the total items available. This may change during pagination." } } }, diff --git a/internal/gen/controller/api/services/session_service.pb.go b/internal/gen/controller/api/services/session_service.pb.go index 7767b7f77..77da310ac 100644 --- a/internal/gen/controller/api/services/session_service.pb.go +++ b/internal/gen/controller/api/services/session_service.pb.go @@ -131,6 +131,15 @@ type ListSessionsRequest struct { // Experimental. By default only non-terminated (i.e. pending, active, canceling) are returned. // Set this option to include terminated sessions as well. IncludeTerminated bool `protobuf:"varint,40,opt,name=include_terminated,proto3" json:"include_terminated,omitempty" class:"public" eventstream:"observation"` // @gotags: `class:"public" eventstream:"observation"` + // An opaque token used to continue an existing iteration or + // request updated items. If not specified, pagination + // will start from the beginning. + ListToken string `protobuf:"bytes,50,opt,name=list_token,proto3" json:"list_token,omitempty" class:"public"` // @gotags: `class:"public"` + // The maximum size of a page in this iteration. + // If unset, the default page size configured will be used. + // If the page_size is greater than the default page configured, + // an error will be returned. + PageSize uint32 `protobuf:"varint,60,opt,name=page_size,proto3" json:"page_size,omitempty" class:"public"` // @gotags: `class:"public"` } func (x *ListSessionsRequest) Reset() { @@ -193,12 +202,46 @@ func (x *ListSessionsRequest) GetIncludeTerminated() bool { return false } +func (x *ListSessionsRequest) GetListToken() string { + if x != nil { + return x.ListToken + } + return "" +} + +func (x *ListSessionsRequest) GetPageSize() uint32 { + if x != nil { + return x.PageSize + } + return 0 +} + type ListSessionsResponse struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields Items []*sessions.Session `protobuf:"bytes,1,rep,name=items,proto3" json:"items,omitempty"` + // The type of response, either "delta" or "complete". + // Delta signifies that this is part of a paginated result + // or an update to a previously completed pagination. + // Complete signifies that it is the last page. + ResponseType string `protobuf:"bytes,2,opt,name=response_type,proto3" json:"response_type,omitempty" class:"public"` // @gotags: `class:"public"` + // An opaque token used to continue an existing pagination or + // request updated items. Use this token in the next list request + // to request the next page. + ListToken string `protobuf:"bytes,3,opt,name=list_token,proto3" json:"list_token,omitempty" class:"public"` // @gotags: `class:"public"` + // The name of the field which the items are sorted by. + SortBy string `protobuf:"bytes,4,opt,name=sort_by,proto3" json:"sort_by,omitempty" class:"public"` // @gotags: `class:"public"` + // The direction of the sort, either "asc" or "desc". + SortDir string `protobuf:"bytes,5,opt,name=sort_dir,proto3" json:"sort_dir,omitempty" class:"public"` // @gotags: `class:"public"` + // A list of item IDs that have been removed since they were returned + // as part of a pagination. They should be dropped from any client cache. + // This may contain items that are not known to the cache, if they were + // created and deleted between listings. + RemovedIds []string `protobuf:"bytes,6,rep,name=removed_ids,proto3" json:"removed_ids,omitempty" class:"public"` // @gotags: `class:"public"` + // An estimate at the total items available. This may change during pagination. + EstItemCount uint32 `protobuf:"varint,7,opt,name=est_item_count,proto3" json:"est_item_count,omitempty" class:"public"` // @gotags: `class:"public"` } func (x *ListSessionsResponse) Reset() { @@ -240,6 +283,48 @@ func (x *ListSessionsResponse) GetItems() []*sessions.Session { return nil } +func (x *ListSessionsResponse) GetResponseType() string { + if x != nil { + return x.ResponseType + } + return "" +} + +func (x *ListSessionsResponse) GetListToken() string { + if x != nil { + return x.ListToken + } + return "" +} + +func (x *ListSessionsResponse) GetSortBy() string { + if x != nil { + return x.SortBy + } + return "" +} + +func (x *ListSessionsResponse) GetSortDir() string { + if x != nil { + return x.SortDir + } + return "" +} + +func (x *ListSessionsResponse) GetRemovedIds() []string { + if x != nil { + return x.RemovedIds + } + return nil +} + +func (x *ListSessionsResponse) GetEstItemCount() uint32 { + if x != nil { + return x.EstItemCount + } + return 0 +} + type CancelSessionRequest struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache @@ -365,7 +450,7 @@ var file_controller_api_services_v1_session_service_proto_rawDesc = []byte{ 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x2e, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x2e, 0x76, 0x31, - 0x2e, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x22, 0x96, + 0x2e, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x22, 0xd4, 0x01, 0x0a, 0x13, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x19, 0x0a, 0x08, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x49, @@ -375,62 +460,78 @@ var file_controller_api_services_v1_session_service_proto_rawDesc = []byte{ 0x06, 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x12, 0x2e, 0x0a, 0x12, 0x69, 0x6e, 0x63, 0x6c, 0x75, 0x64, 0x65, 0x5f, 0x74, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x65, 0x64, 0x18, 0x28, 0x20, 0x01, 0x28, 0x08, 0x52, 0x12, 0x69, 0x6e, 0x63, 0x6c, 0x75, 0x64, 0x65, 0x5f, 0x74, 0x65, 0x72, - 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x65, 0x64, 0x22, 0x5b, 0x0a, 0x14, 0x4c, 0x69, 0x73, 0x74, 0x53, - 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, - 0x43, 0x0a, 0x05, 0x69, 0x74, 0x65, 0x6d, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x2d, - 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, - 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x2e, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, - 0x6e, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x05, 0x69, - 0x74, 0x65, 0x6d, 0x73, 0x22, 0x40, 0x0a, 0x14, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x53, 0x65, - 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, - 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x18, 0x0a, 0x07, - 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x07, 0x76, - 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x22, 0x5a, 0x0a, 0x15, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, - 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, - 0x41, 0x0a, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2d, 0x2e, + 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x65, 0x64, 0x12, 0x1e, 0x0a, 0x0a, 0x6c, 0x69, 0x73, 0x74, 0x5f, + 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x32, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x6c, 0x69, 0x73, + 0x74, 0x5f, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x1c, 0x0a, 0x09, 0x70, 0x61, 0x67, 0x65, 0x5f, + 0x73, 0x69, 0x7a, 0x65, 0x18, 0x3c, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x09, 0x70, 0x61, 0x67, 0x65, + 0x5f, 0x73, 0x69, 0x7a, 0x65, 0x22, 0xa1, 0x02, 0x0a, 0x14, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x65, + 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x43, + 0x0a, 0x05, 0x69, 0x74, 0x65, 0x6d, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x2d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x2e, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, - 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x04, 0x69, 0x74, - 0x65, 0x6d, 0x32, 0x95, 0x04, 0x0a, 0x0e, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x53, 0x65, - 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0xa7, 0x01, 0x0a, 0x0a, 0x47, 0x65, 0x74, 0x53, 0x65, 0x73, - 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x2d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, - 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, - 0x31, 0x2e, 0x47, 0x65, 0x74, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, - 0x65, 0x73, 0x74, 0x1a, 0x2e, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, - 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, - 0x2e, 0x47, 0x65, 0x74, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, - 0x6e, 0x73, 0x65, 0x22, 0x3a, 0x92, 0x41, 0x18, 0x12, 0x16, 0x47, 0x65, 0x74, 0x73, 0x20, 0x61, - 0x20, 0x73, 0x69, 0x6e, 0x67, 0x6c, 0x65, 0x20, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x2e, - 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x19, 0x62, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x12, 0x11, 0x2f, 0x76, - 0x31, 0x2f, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x2f, 0x7b, 0x69, 0x64, 0x7d, 0x12, - 0x9f, 0x01, 0x0a, 0x0c, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, - 0x12, 0x2f, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, - 0x69, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x4c, 0x69, - 0x73, 0x74, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, - 0x74, 0x1a, 0x30, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, - 0x70, 0x69, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x4c, - 0x69, 0x73, 0x74, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, - 0x6e, 0x73, 0x65, 0x22, 0x2c, 0x92, 0x41, 0x15, 0x12, 0x13, 0x4c, 0x69, 0x73, 0x74, 0x73, 0x20, - 0x61, 0x6c, 0x6c, 0x20, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x2e, 0x82, 0xd3, 0xe4, - 0x93, 0x02, 0x0e, 0x12, 0x0c, 0x2f, 0x76, 0x31, 0x2f, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, - 0x73, 0x12, 0xb6, 0x01, 0x0a, 0x0d, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x53, 0x65, 0x73, 0x73, - 0x69, 0x6f, 0x6e, 0x12, 0x30, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, - 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, - 0x2e, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, - 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x31, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, + 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x05, 0x69, 0x74, + 0x65, 0x6d, 0x73, 0x12, 0x24, 0x0a, 0x0d, 0x72, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x5f, + 0x74, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x72, 0x65, 0x73, 0x70, + 0x6f, 0x6e, 0x73, 0x65, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x12, 0x1e, 0x0a, 0x0a, 0x6c, 0x69, 0x73, + 0x74, 0x5f, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x6c, + 0x69, 0x73, 0x74, 0x5f, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x18, 0x0a, 0x07, 0x73, 0x6f, 0x72, + 0x74, 0x5f, 0x62, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x73, 0x6f, 0x72, 0x74, + 0x5f, 0x62, 0x79, 0x12, 0x1a, 0x0a, 0x08, 0x73, 0x6f, 0x72, 0x74, 0x5f, 0x64, 0x69, 0x72, 0x18, + 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x73, 0x6f, 0x72, 0x74, 0x5f, 0x64, 0x69, 0x72, 0x12, + 0x20, 0x0a, 0x0b, 0x72, 0x65, 0x6d, 0x6f, 0x76, 0x65, 0x64, 0x5f, 0x69, 0x64, 0x73, 0x18, 0x06, + 0x20, 0x03, 0x28, 0x09, 0x52, 0x0b, 0x72, 0x65, 0x6d, 0x6f, 0x76, 0x65, 0x64, 0x5f, 0x69, 0x64, + 0x73, 0x12, 0x26, 0x0a, 0x0e, 0x65, 0x73, 0x74, 0x5f, 0x69, 0x74, 0x65, 0x6d, 0x5f, 0x63, 0x6f, + 0x75, 0x6e, 0x74, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0e, 0x65, 0x73, 0x74, 0x5f, 0x69, + 0x74, 0x65, 0x6d, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x22, 0x40, 0x0a, 0x14, 0x43, 0x61, 0x6e, + 0x63, 0x65, 0x6c, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, + 0x64, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, + 0x28, 0x0d, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x22, 0x5a, 0x0a, 0x15, 0x43, + 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, + 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x41, 0x0a, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, + 0x28, 0x0b, 0x32, 0x2d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, + 0x61, 0x70, 0x69, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x2e, 0x73, 0x65, + 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, + 0x6e, 0x52, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x32, 0x95, 0x04, 0x0a, 0x0e, 0x53, 0x65, 0x73, 0x73, + 0x69, 0x6f, 0x6e, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0xa7, 0x01, 0x0a, 0x0a, 0x47, + 0x65, 0x74, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x2d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, + 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, + 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x47, 0x65, 0x74, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, + 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2e, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, + 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, + 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x47, 0x65, 0x74, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, + 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x3a, 0x92, 0x41, 0x18, 0x12, 0x16, 0x47, + 0x65, 0x74, 0x73, 0x20, 0x61, 0x20, 0x73, 0x69, 0x6e, 0x67, 0x6c, 0x65, 0x20, 0x53, 0x65, 0x73, + 0x73, 0x69, 0x6f, 0x6e, 0x2e, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x19, 0x62, 0x04, 0x69, 0x74, 0x65, + 0x6d, 0x12, 0x11, 0x2f, 0x76, 0x31, 0x2f, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x2f, + 0x7b, 0x69, 0x64, 0x7d, 0x12, 0x9f, 0x01, 0x0a, 0x0c, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x65, 0x73, + 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x2f, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, - 0x76, 0x31, 0x2e, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, - 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x40, 0x92, 0x41, 0x14, 0x12, 0x12, 0x43, - 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x73, 0x20, 0x61, 0x20, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, - 0x2e, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x23, 0x3a, 0x01, 0x2a, 0x62, 0x04, 0x69, 0x74, 0x65, 0x6d, - 0x22, 0x18, 0x2f, 0x76, 0x31, 0x2f, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x2f, 0x7b, - 0x69, 0x64, 0x7d, 0x3a, 0x63, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x42, 0x4d, 0x5a, 0x4b, 0x67, 0x69, - 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, - 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, - 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x67, 0x65, 0x6e, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, - 0x6c, 0x65, 0x72, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, - 0x3b, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, - 0x33, + 0x76, 0x31, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, + 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x30, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, + 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, + 0x2e, 0x76, 0x31, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, + 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x2c, 0x92, 0x41, 0x15, 0x12, 0x13, 0x4c, + 0x69, 0x73, 0x74, 0x73, 0x20, 0x61, 0x6c, 0x6c, 0x20, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, + 0x73, 0x2e, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x0e, 0x12, 0x0c, 0x2f, 0x76, 0x31, 0x2f, 0x73, 0x65, + 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0xb6, 0x01, 0x0a, 0x0d, 0x43, 0x61, 0x6e, 0x63, 0x65, + 0x6c, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x30, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, + 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, + 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x53, 0x65, 0x73, 0x73, + 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x31, 0x2e, 0x63, 0x6f, 0x6e, + 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x73, 0x65, 0x72, 0x76, + 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x53, 0x65, + 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x40, 0x92, + 0x41, 0x14, 0x12, 0x12, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x73, 0x20, 0x61, 0x20, 0x53, 0x65, + 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x2e, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x23, 0x3a, 0x01, 0x2a, 0x62, + 0x04, 0x69, 0x74, 0x65, 0x6d, 0x22, 0x18, 0x2f, 0x76, 0x31, 0x2f, 0x73, 0x65, 0x73, 0x73, 0x69, + 0x6f, 0x6e, 0x73, 0x2f, 0x7b, 0x69, 0x64, 0x7d, 0x3a, 0x63, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x42, + 0x4d, 0x5a, 0x4b, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, + 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, + 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x67, 0x65, 0x6e, 0x2f, 0x63, 0x6f, + 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x73, 0x65, 0x72, + 0x76, 0x69, 0x63, 0x65, 0x73, 0x3b, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x62, 0x06, + 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/internal/proto/controller/api/services/v1/session_service.proto b/internal/proto/controller/api/services/v1/session_service.proto index 63d806f09..96d61d496 100644 --- a/internal/proto/controller/api/services/v1/session_service.proto +++ b/internal/proto/controller/api/services/v1/session_service.proto @@ -61,10 +61,39 @@ message ListSessionsRequest { // Experimental. By default only non-terminated (i.e. pending, active, canceling) are returned. // Set this option to include terminated sessions as well. bool include_terminated = 40 [json_name = "include_terminated"]; // @gotags: `class:"public" eventstream:"observation"` + // An opaque token used to continue an existing iteration or + // request updated items. If not specified, pagination + // will start from the beginning. + string list_token = 50 [json_name = "list_token"]; // @gotags: `class:"public"` + // The maximum size of a page in this iteration. + // If unset, the default page size configured will be used. + // If the page_size is greater than the default page configured, + // an error will be returned. + uint32 page_size = 60 [json_name = "page_size"]; // @gotags: `class:"public"` } message ListSessionsResponse { repeated resources.sessions.v1.Session items = 1; + // The type of response, either "delta" or "complete". + // Delta signifies that this is part of a paginated result + // or an update to a previously completed pagination. + // Complete signifies that it is the last page. + string response_type = 2 [json_name = "response_type"]; // @gotags: `class:"public"` + // An opaque token used to continue an existing pagination or + // request updated items. Use this token in the next list request + // to request the next page. + string list_token = 3 [json_name = "list_token"]; // @gotags: `class:"public"` + // The name of the field which the items are sorted by. + string sort_by = 4 [json_name = "sort_by"]; // @gotags: `class:"public"` + // The direction of the sort, either "asc" or "desc". + string sort_dir = 5 [json_name = "sort_dir"]; // @gotags: `class:"public"` + // A list of item IDs that have been removed since they were returned + // as part of a pagination. They should be dropped from any client cache. + // This may contain items that are not known to the cache, if they were + // created and deleted between listings. + repeated string removed_ids = 6 [json_name = "removed_ids"]; // @gotags: `class:"public"` + // An estimate at the total items available. This may change during pagination. + uint32 est_item_count = 7 [json_name = "est_item_count"]; // @gotags: `class:"public"` } message CancelSessionRequest { diff --git a/internal/session/options.go b/internal/session/options.go index 622469d28..6ddc9f3a5 100644 --- a/internal/session/options.go +++ b/internal/session/options.go @@ -10,6 +10,7 @@ import ( "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/db/timestamp" + "github.com/hashicorp/boundary/internal/pagination" "github.com/hashicorp/boundary/internal/perms" ) @@ -40,6 +41,7 @@ type options struct { withPermissions *perms.UserPermissions withIgnoreDecryptionFailures bool withRandomReader io.Reader + withStartPageAfterItem pagination.Item } func getDefaultOptions() options { @@ -149,3 +151,11 @@ func WithRandomReader(rand io.Reader) Option { o.withRandomReader = rand } } + +// WithStartPageAfterItem is used to paginate over the results. +// The next page will start after the provided item. +func WithStartPageAfterItem(item pagination.Item) Option { + return func(o *options) { + o.withStartPageAfterItem = item + } +} diff --git a/internal/session/query.go b/internal/session/query.go index 59a20c799..e9c26b1b0 100644 --- a/internal/session/query.go +++ b/internal/session/query.go @@ -422,6 +422,69 @@ update session_credential key_id = ? where session_id = ? and credential_sha256 = ?; +` + listSessionsTemplate = ` +with session_ids as ( + select public_id + from session + -- search condition for applying permissions is constructed + where %s + order by create_time desc, public_id asc + limit %d +) + select * + from session_list + where session_list.public_id in (select * from session_ids) + order by create_time desc, public_id asc; +` + listSessionsPageTemplate = ` +with session_ids as ( + select public_id + from session + where (create_time, public_id) < (@last_item_create_time, @last_item_id) + -- search condition for applying permissions is constructed + and %s + order by create_time desc, public_id asc + limit %d +) + select * + from session_list + where session_list.public_id in (select * from session_ids) + order by create_time desc, public_id asc; +` + refreshSessionsTemplate = ` +with session_ids as ( + select public_id + from session + where update_time > @updated_after_time + -- search condition for applying permissions is constructed + and %s + order by update_time desc, public_id asc + limit %d +) + select * + from session_list + where session_list.public_id in (select * from session_ids) +order by update_time desc, public_id asc; +` + refreshSessionsPageTemplate = ` +with session_ids as ( + select public_id + from session + where update_time > @updated_after_time + and (update_time, public_id) < (@last_item_update_time, @last_item_id) + -- search condition for applying permissions is constructed + and %s + order by update_time desc, public_id asc + limit %d +) + select * + from session_list + where session_list.public_id in (select * from session_ids) +order by update_time desc, public_id asc; +` + estimateCountSessions = ` + select reltuples::bigint as estimate from pg_class where oid in ('session'::regclass) ` ) diff --git a/internal/session/repository_session.go b/internal/session/repository_session.go index 2b8b63ae4..e9804ccb1 100644 --- a/internal/session/repository_session.go +++ b/internal/session/repository_session.go @@ -274,71 +274,173 @@ func (r *Repository) LookupSession(ctx context.Context, sessionId string, opt .. return &session, authzSummary, nil } -// ListSessions lists sessions. Sessions returned will be limited by the list -// permissions of the repository. Supports the WithTerminated, WithLimit, -// WithOrderByCreateTime options. -func (r *Repository) ListSessions(ctx context.Context, opt ...Option) ([]*Session, error) { +// listSessions lists sessions. Sessions returned will be limited by the list +// permissions of the repository. +// Supported options: +// - withTerminated +// - withLimit +// - withStartPageAfterItem +func (r *Repository) listSessions(ctx context.Context, opt ...Option) ([]*Session, time.Time, error) { const op = "session.(Repository).ListSessions" - opts := getOpts(opt...) where, args := r.listPermissionWhereClauses() if len(where) == 0 { - return nil, nil + return nil, time.Time{}, nil } - var whereClause string + opts := getOpts(opt...) + + var permissionWhereClause string if len(where) > 0 { - whereClause = " where (" + strings.Join(where, " or ") + ")" + permissionWhereClause = "(" + strings.Join(where, " or ") + ")" if !opts.withTerminated { - whereClause += "and termination_reason is null" + permissionWhereClause += " and termination_reason is null" } } else { if !opts.withTerminated { - whereClause = "where termination_reason is null" + permissionWhereClause = " where termination_reason is null" } } - var limit string - switch { - case opts.withLimit < 0: // any negative number signals unlimited results - case opts.withLimit == 0: // zero signals the default value and default limits - limit = fmt.Sprintf("limit %d", r.defaultLimit) - default: - // non-zero signals an override of the default limit for the repo. - limit = fmt.Sprintf("limit %d", opts.withLimit) - } - var withOrder string - switch opts.withOrderByCreateTime { - case db.AscendingOrderBy: - withOrder = "order by create_time asc" - case db.DescendingOrderBy: - fallthrough - default: - withOrder = "order by create_time" + limit := r.defaultLimit + if opts.withLimit > 0 { + limit = opts.withLimit } - q := sessionList - query := fmt.Sprintf(q, whereClause, withOrder, limit, withOrder) + query := fmt.Sprintf(listSessionsTemplate, permissionWhereClause, limit) + if opts.withStartPageAfterItem != nil { + query = fmt.Sprintf(listSessionsPageTemplate, permissionWhereClause, limit) + args = append(args, + sql.Named("last_item_create_time", opts.withStartPageAfterItem.GetCreateTime()), + sql.Named("last_item_id", opts.withStartPageAfterItem.GetPublicId()), + ) + } - rows, err := r.reader.Query(ctx, query, args) + return r.querySessions(ctx, query, args, limit) +} + +// listSessionsRefresh lists sessions limited by the list +// permissions of the repository. +// Supported options: +// - withTerminated +// - withLimit +// - withStartPageAfterItem +func (r *Repository) listSessionsRefresh(ctx context.Context, updatedAfter time.Time, opt ...Option) ([]*Session, time.Time, error) { + const op = "session.(Repository).ListSessionsRefresh" + + if updatedAfter.IsZero() { + return nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "missing updated after time") + } + + where, args := r.listPermissionWhereClauses() + if len(where) == 0 { + return nil, time.Time{}, nil + } + + opts := getOpts(opt...) + + var permissionWhereClause string + if len(where) > 0 { + permissionWhereClause = "(" + strings.Join(where, " or ") + ")" + if !opts.withTerminated { + permissionWhereClause += " and termination_reason is null" + } + } else { + if !opts.withTerminated { + permissionWhereClause = " where termination_reason is null" + } + } + + limit := r.defaultLimit + if opts.withLimit > 0 { + limit = opts.withLimit + } + + query := fmt.Sprintf(refreshSessionsTemplate, permissionWhereClause, limit) + args = append(args, + sql.Named("updated_after_time", timestamp.New(updatedAfter)), + ) + if opts.withStartPageAfterItem != nil { + query = fmt.Sprintf(refreshSessionsPageTemplate, permissionWhereClause, limit) + args = append(args, + sql.Named("last_item_update_time", opts.withStartPageAfterItem.GetUpdateTime()), + sql.Named("last_item_id", opts.withStartPageAfterItem.GetPublicId()), + ) + } + + return r.querySessions(ctx, query, args, limit) +} + +func (r *Repository) querySessions(ctx context.Context, query string, args []any, limit int) ([]*Session, time.Time, error) { + const op = "session.(Repository).querySessions" + + var sessions []*Session + var transactionTimestamp time.Time + if _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(rd db.Reader, w db.Writer) error { + rows, err := rd.Query(ctx, query, args) + if err != nil { + return err + } + defer rows.Close() + var sessionsList []*sessionListView + for rows.Next() { + var s sessionListView + if err := rd.ScanRows(ctx, rows, &s); err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("scan row failed")) + } + sessionsList = append(sessionsList, &s) + } + sessions, err = r.convertToSessions(ctx, sessionsList) + if err != nil { + return errors.Wrap(ctx, err, op) + } + transactionTimestamp, err = rd.Now(ctx) + return err + }); err != nil { + return nil, time.Time{}, err + } + return sessions, transactionTimestamp, nil +} + +// listDeletedIds lists the public IDs of any sessions deleted since the timestamp provided. +func (r *Repository) listDeletedIds(ctx context.Context, since time.Time) ([]string, time.Time, error) { + const op = "session.(Repository).listDeletedIds" + var deletedSessions []*deletedSession + var transactionTimestamp time.Time + if _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, _ db.Writer) error { + if err := r.SearchWhere(ctx, &deletedSessions, "delete_time >= ?", []any{since}); err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("failed to query deleted auth tokens")) + } + var err error + transactionTimestamp, err = r.Now(ctx) + if err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("failed to get transaction timestamp")) + } + return nil + }); err != nil { + return nil, time.Time{}, err + } + var sessionIds []string + for _, sess := range deletedSessions { + sessionIds = append(sessionIds, sess.PublicId) + } + return sessionIds, transactionTimestamp, nil +} + +// estimatedCount returns an estimate of the total number of items in the session table. +func (r *Repository) estimatedCount(ctx context.Context) (int, error) { + const op = "session.(Repository).estimatedCount" + rows, err := r.reader.Query(ctx, estimateCountSessions, nil) if err != nil { - return nil, errors.Wrap(ctx, err, op) + return 0, errors.Wrap(ctx, err, op, errors.WithMsg("failed to query total sessions")) } - defer rows.Close() - - var sessionsList []*sessionListView + var count int for rows.Next() { - var s sessionListView - if err := r.reader.ScanRows(ctx, rows, &s); err != nil { - return nil, errors.Wrap(ctx, err, op, errors.WithMsg("scan row failed")) + if err := r.reader.ScanRows(ctx, rows, &count); err != nil { + return 0, errors.Wrap(ctx, err, op, errors.WithMsg("failed to query total sessions")) } - sessionsList = append(sessionsList, &s) } - sessions, err := r.convertToSessions(ctx, sessionsList) - if err != nil { - return nil, errors.Wrap(ctx, err, op) - } - return sessions, nil + return count, nil } // DeleteSession will delete a session from the repository. diff --git a/internal/session/repository_session_test.go b/internal/session/repository_session_test.go index 48488bba0..befb4d81b 100644 --- a/internal/session/repository_session_test.go +++ b/internal/session/repository_session_test.go @@ -69,18 +69,9 @@ func TestRepository_ListSession(t *testing.T) { perms *perms.UserPermissions wantCnt int wantErr bool + wantTTime time.Time withConnections int }{ - { - name: "no-limit", - createCnt: testLimit + 1, - args: args{ - opt: []Option{WithLimit(-1)}, - }, - perms: listPerms, - wantCnt: testLimit + 1, - wantErr: false, - }, { name: "default-limit", createCnt: testLimit + 1, @@ -88,6 +79,7 @@ func TestRepository_ListSession(t *testing.T) { perms: listPerms, wantCnt: testLimit, wantErr: false, + wantTTime: time.Now(), }, { name: "custom-limit", @@ -95,9 +87,10 @@ func TestRepository_ListSession(t *testing.T) { args: args{ opt: []Option{WithLimit(3)}, }, - perms: listPerms, - wantCnt: 3, - wantErr: false, + perms: listPerms, + wantCnt: 3, + wantErr: false, + wantTTime: time.Now(), }, { name: "withNoPerms", @@ -106,6 +99,7 @@ func TestRepository_ListSession(t *testing.T) { perms: &perms.UserPermissions{}, wantCnt: 0, wantErr: false, + wantTTime: time.Time{}, }, { name: "withPermsDifferentScopeId", @@ -120,8 +114,9 @@ func TestRepository_ListSession(t *testing.T) { }, }, }, - wantCnt: 0, - wantErr: false, + wantCnt: 0, + wantErr: false, + wantTTime: time.Now(), }, { name: "withPermsNonListAction", @@ -136,8 +131,9 @@ func TestRepository_ListSession(t *testing.T) { }, }, }, - wantCnt: 0, - wantErr: false, + wantCnt: 0, + wantErr: false, + wantTTime: time.Time{}, }, { name: "multiple-connections", @@ -147,6 +143,7 @@ func TestRepository_ListSession(t *testing.T) { wantCnt: testLimit, wantErr: false, withConnections: 3, + wantTTime: time.Now(), }, } for _, tt := range tests { @@ -167,13 +164,12 @@ func TestRepository_ListSession(t *testing.T) { } } assert.Equal(tt.createCnt, len(testSessions)) - got, err := repo.ListSessions(context.Background(), tt.args.opt...) - if tt.wantErr { - require.Error(err) - return - } + got, ttime, err := repo.listSessions(ctx, tt.args.opt...) require.NoError(err) assert.Equal(tt.wantCnt, len(got)) + // Transaction timestamp should be within ~10 seconds of now + assert.True(tt.wantTTime.Before(ttime.Add(10 * time.Second))) + assert.True(tt.wantTTime.After(ttime.Add(-10 * time.Second))) for i := 0; i < len(got); i++ { // connections should not be returned for list requests assert.Equal(0, len(got[i].Connections)) @@ -190,27 +186,6 @@ func TestRepository_ListSession(t *testing.T) { } }) } - t.Run("withOrder", func(t *testing.T) { - assert, require := assert.New(t), require.New(t) - db.TestDeleteWhere(t, conn, func() any { i := AllocSession(); return &i }(), "1=1") - wantCnt := 5 - for i := 0; i < wantCnt; i++ { - _ = TestSession(t, conn, wrapper, composedOf) - } - - repo, err := NewRepository(ctx, rw, rw, kms, WithLimit(testLimit), WithPermissions(listPerms)) - require.NoError(err) - - got, err := repo.ListSessions(context.Background(), WithOrderByCreateTime(db.AscendingOrderBy)) - require.NoError(err) - assert.Equal(wantCnt, len(got)) - - for i := 0; i < len(got)-1; i++ { - first := got[i].CreateTime.Timestamp.AsTime() - second := got[i+1].CreateTime.Timestamp.AsTime() - assert.True(first.Before(second)) - } - }) t.Run("onlySelf", func(t *testing.T) { assert, require := assert.New(t), require.New(t) db.TestDeleteWhere(t, conn, func() any { i := AllocSession(); return &i }(), "1=1") @@ -233,10 +208,114 @@ func TestRepository_ListSession(t *testing.T) { } repo, err := NewRepository(ctx, rw, rw, kms, WithLimit(testLimit), WithPermissions(p)) require.NoError(err) - got, err := repo.ListSessions(context.Background(), WithUserId(s.UserId)) + got, ttime, err := repo.listSessions(ctx, WithUserId(s.UserId)) require.NoError(err) assert.Equal(1, len(got)) assert.Equal(s.UserId, got[0].UserId) + // Transaction timestamp should be within ~10 seconds of now + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + }) + t.Run("withStartPageAfter", func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + db.TestDeleteWhere(t, conn, func() any { i := AllocSession(); return &i }(), "1=1") + + composedOf := TestSessionParams(t, conn, wrapper, iamRepo) + + listPerms := &perms.UserPermissions{ + UserId: composedOf.UserId, + Permissions: []perms.Permission{ + { + ScopeId: composedOf.ProjectId, + Resource: resource.Session, + Action: action.List, + }, + }, + } + + for i := 0; i < 10; i++ { + _ = TestSession(t, conn, wrapper, composedOf) + } + + repo, err := NewRepository(ctx, rw, rw, kms, WithPermissions(listPerms)) + require.NoError(err) + page1, ttime, err := repo.listSessions(ctx, WithLimit(2)) + require.NoError(err) + require.Len(page1, 2) + // Transaction timestamp should be within ~10 seconds of now + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + page2, ttime, err := repo.listSessions(ctx, WithLimit(2), WithStartPageAfterItem(page1[1])) + require.NoError(err) + require.Len(page2, 2) + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + for _, item := range page1 { + assert.NotEqual(item.GetPublicId(), page2[0].GetPublicId()) + assert.NotEqual(item.GetPublicId(), page2[1].GetPublicId()) + } + page3, ttime, err := repo.listSessions(ctx, WithLimit(2), WithStartPageAfterItem(page2[1])) + require.NoError(err) + require.Len(page3, 2) + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + for _, item := range page2 { + assert.NotEqual(item.GetPublicId(), page3[0].GetPublicId()) + assert.NotEqual(item.GetPublicId(), page3[1].GetPublicId()) + } + page4, ttime, err := repo.listSessions(ctx, WithLimit(2), WithStartPageAfterItem(page3[1])) + require.NoError(err) + require.Len(page4, 2) + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + for _, item := range page3 { + assert.NotEqual(item.GetPublicId(), page4[0].GetPublicId()) + assert.NotEqual(item.GetPublicId(), page4[1].GetPublicId()) + } + page5, ttime, err := repo.listSessions(ctx, WithLimit(2), WithStartPageAfterItem(page4[1])) + require.NoError(err) + require.Len(page5, 2) + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + for _, item := range page4 { + assert.NotEqual(item.GetPublicId(), page5[0].GetPublicId()) + assert.NotEqual(item.GetPublicId(), page5[1].GetPublicId()) + } + page6, ttime, err := repo.listSessions(ctx, WithLimit(2), WithStartPageAfterItem(page5[1])) + require.NoError(err) + require.Empty(page6) + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + + // Cancel the first two sessions in lieu of updating + _, err = repo.CancelSession(ctx, page1[0].PublicId, page1[0].Version) + require.NoError(err) + _, err = repo.CancelSession(ctx, page1[1].PublicId, page1[1].Version) + require.NoError(err) + + // since it will return newest to oldest, we get page1[1] first + page7, ttime, err := repo.listSessionsRefresh( + ctx, + time.Now().Add(-1*time.Second), + WithLimit(1), + ) + require.NoError(err) + require.Len(page7, 1) + require.Equal(page7[0].GetPublicId(), page1[1].GetPublicId()) + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + + page8, ttime, err := repo.listSessionsRefresh( + context.Background(), + time.Now().Add(-1*time.Second), + WithLimit(1), + WithStartPageAfterItem(page7[0]), + ) + require.NoError(err) + require.Len(page8, 1) + require.Equal(page8[0].GetPublicId(), page1[0].GetPublicId()) + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) }) } @@ -268,9 +347,100 @@ func TestRepository_ListSessions_Multiple_Scopes(t *testing.T) { Permissions: p, })) require.NoError(t, err) - got, err := repo.ListSessions(context.Background()) + got, ttime, err := repo.listSessions(ctx) require.NoError(t, err) assert.Equal(t, len(p), len(got)) + // Transaction timestamp should be within ~10 seconds of now + assert.True(t, time.Now().Before(ttime.Add(10*time.Second))) + assert.True(t, time.Now().After(ttime.Add(-10*time.Second))) +} + +func Test_listDeletedIds(t *testing.T) { + t.Parallel() + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + wrapper := db.TestWrapper(t) + testKms := kms.TestKms(t, conn, wrapper) + iamRepo := iam.TestRepo(t, conn, wrapper) + + rw := db.New(conn) + repo, err := NewRepository(ctx, rw, rw, testKms) + require.NoError(t, err) + + composedOf := TestSessionParams(t, conn, wrapper, iamRepo) + + s := TestSession(t, conn, wrapper, composedOf) + _ = TestState(t, conn, s.PublicId, StatusActive) + + // Expect no entries at the start + deletedIds, ttime, err := repo.listDeletedIds(ctx, time.Now().AddDate(-1, 0, 0)) + require.NoError(t, err) + require.Empty(t, deletedIds) + // Transaction timestamp should be within ~10 seconds of now + assert.True(t, time.Now().Before(ttime.Add(10*time.Second))) + assert.True(t, time.Now().After(ttime.Add(-10*time.Second))) + + // Delete a session + _, err = repo.DeleteSession(ctx, s.GetPublicId()) + require.NoError(t, err) + + // Expect a single entry + deletedIds, ttime, err = repo.listDeletedIds(ctx, time.Now().AddDate(-1, 0, 0)) + require.NoError(t, err) + require.Equal(t, []string{s.GetPublicId()}, deletedIds) + // Transaction timestamp should be within ~10 seconds of now + assert.True(t, time.Now().Before(ttime.Add(10*time.Second))) + assert.True(t, time.Now().After(ttime.Add(-10*time.Second))) + + // Try again with the time set to now, expect no entries + deletedIds, ttime, err = repo.listDeletedIds(ctx, time.Now()) + require.NoError(t, err) + require.Empty(t, deletedIds) + // Transaction timestamp should be within ~10 seconds of now + assert.True(t, time.Now().Before(ttime.Add(10*time.Second))) + assert.True(t, time.Now().After(ttime.Add(-10*time.Second))) +} + +func Test_estimatedCount(t *testing.T) { + t.Parallel() + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + sqlDb, err := conn.SqlDB(ctx) + require.NoError(t, err) + wrapper := db.TestWrapper(t) + testKms := kms.TestKms(t, conn, wrapper) + iamRepo := iam.TestRepo(t, conn, wrapper) + + rw := db.New(conn) + repo, err := NewRepository(ctx, rw, rw, testKms) + require.NoError(t, err) + + composedOf := TestSessionParams(t, conn, wrapper, iamRepo) + + // Check total entries at start, expect 0 + numItems, err := repo.estimatedCount(ctx) + require.NoError(t, err) + assert.Equal(t, 0, numItems) + + // Create a session, expect 1 entries + s := TestSession(t, conn, wrapper, composedOf) + _ = TestState(t, conn, s.PublicId, StatusActive) + // Run analyze to update estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + numItems, err = repo.estimatedCount(ctx) + require.NoError(t, err) + assert.Equal(t, 1, numItems) + + // Delete the target, expect 0 again + _, err = repo.DeleteSession(ctx, s.GetPublicId()) + require.NoError(t, err) + // Run analyze to update estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + numItems, err = repo.estimatedCount(ctx) + require.NoError(t, err) + assert.Equal(t, 0, numItems) } func TestRepository_CreateSession(t *testing.T) { diff --git a/internal/session/service_list.go b/internal/session/service_list.go new file mode 100644 index 000000000..956444549 --- /dev/null +++ b/internal/session/service_list.go @@ -0,0 +1,56 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package session + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/pagination" +) + +// List lists up to page size sessions, filtering out entries that +// do not pass the filter item function. It will automatically request +// more sessions from the database, at page size chunks, to fill the page. +// It returns a new list token used to continue pagination or refresh items. +// Sessions are ordered by create time descending (most recently created first). +func List( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn pagination.ListFilterFunc[*Session], + repo *Repository, + includeTerminated bool, +) (*pagination.ListResponse[*Session], error) { + const op = "session.List" + + switch { + case len(grantsHash) == 0: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case repo == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + } + + listItemsFn := func(ctx context.Context, lastPageItem *Session, limit int) ([]*Session, time.Time, error) { + opts := []Option{ + WithLimit(limit), + } + if lastPageItem != nil { + opts = append(opts, WithStartPageAfterItem(lastPageItem)) + } + if includeTerminated { + opts = append(opts, + WithTerminated(includeTerminated), + ) + } + return repo.listSessions(ctx, opts...) + } + + return pagination.List(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedCount) +} diff --git a/internal/session/service_list_ext_test.go b/internal/session/service_list_ext_test.go new file mode 100644 index 000000000..0d3e6d23e --- /dev/null +++ b/internal/session/service_list_ext_test.go @@ -0,0 +1,723 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package session_test + +import ( + "context" + "slices" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + "github.com/hashicorp/boundary/globals" + "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/db/timestamp" + "github.com/hashicorp/boundary/internal/iam" + "github.com/hashicorp/boundary/internal/kms" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/perms" + "github.com/hashicorp/boundary/internal/session" + "github.com/hashicorp/boundary/internal/types/action" + "github.com/hashicorp/boundary/internal/types/resource" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestService_List(t *testing.T) { + fiveDaysAgo := time.Now() + // Set database read timeout to avoid duplicates in response + oldReadTimeout := globals.RefreshReadLookbackDuration + globals.RefreshReadLookbackDuration = 0 + t.Cleanup(func() { + globals.RefreshReadLookbackDuration = oldReadTimeout + }) + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + sqlDb, err := conn.SqlDB(ctx) + require.NoError(t, err) + wrapper := db.TestWrapper(t) + iamRepo := iam.TestRepo(t, conn, wrapper) + rw := db.New(conn) + kms := kms.TestKms(t, conn, wrapper) + composedOf := session.TestSessionParams(t, conn, wrapper, iamRepo) + + listPerms := &perms.UserPermissions{ + UserId: composedOf.UserId, + Permissions: []perms.Permission{ + { + ScopeId: composedOf.ProjectId, + Resource: resource.Session, + Action: action.List, + }, + }, + } + var allSessions []*session.Session + for i := 0; i < 5; i++ { + s := session.TestSession(t, conn, wrapper, composedOf) + allSessions = append(allSessions, s) + } + + repo, err := session.NewRepository(ctx, rw, rw, kms, session.WithPermissions(listPerms)) + require.NoError(t, err) + // Reverse since we read items in descending order (newest first) + slices.Reverse(allSessions) + + // Run analyze to update postgres estimates + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + cmpIgnoreUnexportedOpts := cmpopts.IgnoreUnexported(session.Session{}, session.State{}, timestamp.Timestamp{}, timestamppb.Timestamp{}) + cmpIgnoreFieldsOpts := cmpopts.IgnoreFields(session.Session{}, "CtCertificatePrivateKey", "CertificatePrivateKey", "KeyId") + + t.Run("List validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + _, err := session.List(ctx, nil, 1, filterFunc, repo, true) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + _, err := session.List(ctx, []byte("some hash"), 0, filterFunc, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("negative page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + _, err := session.List(ctx, []byte("some hash"), -1, filterFunc, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("nil filter func", func(t *testing.T) { + t.Parallel() + _, err := session.List(ctx, []byte("some hash"), 1, nil, repo, true) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + _, err := session.List(ctx, []byte("some hash"), 1, filterFunc, nil, true) + require.ErrorContains(t, err, "missing repo") + }) + }) + t.Run("ListPage validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, nil, 1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("negative page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("nil filter func", func(t *testing.T) { + t.Parallel() + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), 1, nil, tok, repo, true) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil token", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + _, err = session.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, true) + require.ErrorContains(t, err, "missing token") + }) + t.Run("wrong token type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "token did not have a pagination token component") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil, true) + require.ErrorContains(t, err, "missing repo") + }) + }) + t.Run("ListRefresh validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, nil, 1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("negative page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("nil filter func", func(t *testing.T) { + t.Parallel() + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), 1, nil, tok, repo, true) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil token", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + _, err = session.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, true) + require.ErrorContains(t, err, "missing token") + }) + t.Run("wrong token type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "token did not have a start-refresh token component") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil, true) + require.ErrorContains(t, err, "missing repo") + }) + }) + t.Run("ListRefresh validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, nil, 1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("negative page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("nil filter func", func(t *testing.T) { + t.Parallel() + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), 1, nil, tok, repo, true) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil token", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + _, err = session.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, true) + require.ErrorContains(t, err, "missing token") + }) + t.Run("wrong token type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "token did not have a start-refresh token component") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil, true) + require.ErrorContains(t, err, "missing repo") + }) + }) + t.Run("ListRefreshPage validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, nil, 1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("negative page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("nil filter func", func(t *testing.T) { + t.Parallel() + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, []byte("some hash"), 1, nil, tok, repo, true) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil token", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + _, err = session.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, true) + require.ErrorContains(t, err, "missing token") + }) + t.Run("wrong token type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "token did not have a refresh token component") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil, true) + require.ErrorContains(t, err, "missing repo") + }) + }) + t.Run("ListRefreshPage validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, nil, 1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("negative page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("nil filter func", func(t *testing.T) { + t.Parallel() + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, []byte("some hash"), 1, nil, tok, repo, true) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil token", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + _, err = session.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, true) + require.ErrorContains(t, err, "missing token") + }) + t.Run("wrong token type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, true) + require.ErrorContains(t, err, "token did not have a refresh token component") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = session.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil, true) + require.ErrorContains(t, err, "missing repo") + }) + }) + + t.Run("simple pagination", func(t *testing.T) { + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + resp, err := session.List(ctx, []byte("some hash"), 1, filterFunc, repo, true) + require.NoError(t, err) + require.NotNil(t, resp.ListToken) + require.Equal(t, resp.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp.CompleteListing) + require.Equal(t, resp.EstimatedItemCount, 5) + require.Empty(t, resp.DeletedIds) + require.Len(t, resp.Items, 1) + require.Empty(t, cmp.Diff(resp.Items[0], allSessions[0], cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + resp2, err := session.ListPage(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp2.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp2.CompleteListing) + require.Equal(t, resp2.EstimatedItemCount, 5) + require.Empty(t, resp2.DeletedIds) + require.Len(t, resp2.Items, 1) + require.Empty(t, cmp.Diff(resp2.Items[0], allSessions[1], cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + resp3, err := session.ListPage(ctx, []byte("some hash"), 1, filterFunc, resp2.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp3.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp3.CompleteListing) + require.Equal(t, resp3.EstimatedItemCount, 5) + require.Empty(t, resp3.DeletedIds) + require.Len(t, resp3.Items, 1) + require.Empty(t, cmp.Diff(resp3.Items[0], allSessions[2], cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + resp4, err := session.ListPage(ctx, []byte("some hash"), 1, filterFunc, resp3.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp4.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp4.CompleteListing) + require.Equal(t, resp4.EstimatedItemCount, 5) + require.Empty(t, resp4.DeletedIds) + require.Len(t, resp4.Items, 1) + require.Empty(t, cmp.Diff(resp4.Items[0], allSessions[3], cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + resp5, err := session.ListPage(ctx, []byte("some hash"), 1, filterFunc, resp4.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp5.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp5.CompleteListing) + require.Equal(t, resp5.EstimatedItemCount, 5) + require.Empty(t, resp5.DeletedIds) + require.Len(t, resp5.Items, 1) + require.Empty(t, cmp.Diff(resp5.Items[0], allSessions[4], cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + // Finished initial pagination phase, request refresh + // Expect no results. + resp6, err := session.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp5.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp6.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp6.CompleteListing) + require.Equal(t, resp6.EstimatedItemCount, 5) + require.Empty(t, resp6.DeletedIds) + require.Empty(t, resp6.Items) + + // Create some new sessions + newS1 := session.TestSession(t, conn, wrapper, composedOf) + newS2 := session.TestSession(t, conn, wrapper, composedOf) + t.Cleanup(func() { + _, err = repo.DeleteSession(ctx, newS1.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSession(ctx, newS2.GetPublicId()) + require.NoError(t, err) + // Run analyze to update target estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + }) + // Run analyze to update target estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + // Refresh again, should get newS2 + resp7, err := session.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp6.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp7.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp7.CompleteListing) + require.Equal(t, resp7.EstimatedItemCount, 7) + require.Empty(t, resp7.DeletedIds) + require.Len(t, resp7.Items, 1) + require.Empty(t, cmp.Diff(resp7.Items[0], newS2, cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + // Refresh again, should get newS1 + resp8, err := session.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, resp6.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp8.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp8.CompleteListing) + require.Equal(t, resp8.EstimatedItemCount, 7) + require.Empty(t, resp8.DeletedIds) + require.Len(t, resp8.Items, 1) + require.Empty(t, cmp.Diff(resp8.Items[0], newS1, cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + // Refresh again, should get no results + resp9, err := session.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp6.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp9.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp9.CompleteListing) + require.Equal(t, resp9.EstimatedItemCount, 7) + require.Empty(t, resp9.DeletedIds) + require.Empty(t, resp9.Items) + }) + + t.Run("simple pagination with aggressive filtering", func(t *testing.T) { + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return s.GetPublicId() == allSessions[1].GetPublicId() || + s.GetPublicId() == allSessions[len(allSessions)-1].GetPublicId(), nil + } + resp, err := session.List(ctx, []byte("some hash"), 1, filterFunc, repo, true) + require.NoError(t, err) + require.NotNil(t, resp.ListToken) + require.Equal(t, resp.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp.CompleteListing) + require.Equal(t, resp.EstimatedItemCount, 5) + require.Empty(t, resp.DeletedIds) + require.Len(t, resp.Items, 1) + require.Empty(t, cmp.Diff(resp.Items[0], allSessions[1], cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + resp2, err := session.ListPage(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo, true) + require.NoError(t, err) + require.NotNil(t, resp2.ListToken) + require.Equal(t, resp2.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp2.CompleteListing) + require.Equal(t, resp2.EstimatedItemCount, 5) + require.Empty(t, resp2.DeletedIds) + require.Len(t, resp2.Items, 1) + require.Empty(t, cmp.Diff(resp2.Items[0], allSessions[len(allSessions)-1], cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + // request a refresh, nothing should be returned + resp3, err := session.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp3.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp3.CompleteListing) + require.Equal(t, resp3.EstimatedItemCount, 5) + require.Empty(t, resp3.DeletedIds) + require.Empty(t, resp3.Items) + + // Create some new allSessions + newS1 := session.TestSession(t, conn, wrapper, composedOf) + newS2 := session.TestSession(t, conn, wrapper, composedOf) + newS3 := session.TestSession(t, conn, wrapper, composedOf) + newS4 := session.TestSession(t, conn, wrapper, composedOf) + // Run analyze to update target estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + t.Cleanup(func() { + _, err = repo.DeleteSession(ctx, newS1.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSession(ctx, newS2.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSession(ctx, newS3.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSession(ctx, newS4.GetPublicId()) + require.NoError(t, err) + // Run analyze to update target estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + }) + + filterFunc = func(_ context.Context, s *session.Session) (bool, error) { + return s.GetPublicId() == newS3.GetPublicId() || + s.GetPublicId() == newS1.GetPublicId(), nil + } + // Refresh again, should get newS3 + resp4, err := session.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp3.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp4.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp4.CompleteListing) + require.Equal(t, resp4.EstimatedItemCount, 9) + require.Empty(t, resp4.DeletedIds) + require.Len(t, resp4.Items, 1) + require.Empty(t, cmp.Diff(resp4.Items[0], newS3, cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + // Refresh again, should get newS1 + resp5, err := session.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, resp4.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp5.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp5.CompleteListing) + require.Equal(t, resp5.EstimatedItemCount, 9) + require.Empty(t, resp5.DeletedIds) + require.Len(t, resp5.Items, 1) + require.Empty(t, cmp.Diff(resp5.Items[0], newS1, cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + }) + + t.Run("simple pagination with deletion", func(t *testing.T) { + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + deletedSessionId := allSessions[0].GetPublicId() + _, err := repo.DeleteSession(ctx, deletedSessionId) + require.NoError(t, err) + allSessions = allSessions[1:] + + // Run analyze to update target estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + resp, err := session.List(ctx, []byte("some hash"), 1, filterFunc, repo, true) + require.NoError(t, err) + require.NotNil(t, resp.ListToken) + require.Equal(t, resp.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp.CompleteListing) + require.Equal(t, resp.EstimatedItemCount, 4) + require.Empty(t, resp.DeletedIds) + require.Len(t, resp.Items, 1) + require.Empty(t, cmp.Diff(resp.Items[0], allSessions[0], cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + // request remaining results + resp2, err := session.ListPage(ctx, []byte("some hash"), 3, filterFunc, resp.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp2.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp2.CompleteListing) + require.Equal(t, resp2.EstimatedItemCount, 4) + require.Empty(t, resp2.DeletedIds) + require.Len(t, resp2.Items, 3) + require.Empty(t, cmp.Diff(resp2.Items, allSessions[1:], cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + deletedSessionId = allSessions[0].GetPublicId() + _, err = repo.DeleteSession(ctx, deletedSessionId) + require.NoError(t, err) + allSessions = allSessions[1:] + + // Run analyze to update target estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + // request a refresh, nothing should be returned except the deleted id + resp3, err := session.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp2.ListToken, repo, true) + require.NoError(t, err) + require.Equal(t, resp3.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp3.CompleteListing) + require.Equal(t, resp3.EstimatedItemCount, 3) + require.Contains(t, resp3.DeletedIds, deletedSessionId) + require.Empty(t, resp3.Items) + }) + + t.Run("simple pagination with updated items", func(t *testing.T) { + filterFunc := func(_ context.Context, s *session.Session) (bool, error) { + return true, nil + } + resp, err := session.List(ctx, []byte("some hash"), 3, filterFunc, repo, true) + require.NoError(t, err) + require.NotNil(t, resp.ListToken) + require.Equal(t, resp.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp.CompleteListing) + require.Equal(t, resp.EstimatedItemCount, 3) + require.Empty(t, resp.DeletedIds) + require.Len(t, resp.Items, 3) + require.Empty(t, cmp.Diff(resp.Items, allSessions, cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + + // create a new session and run analyze to update session estimate + newSession := session.TestSession(t, conn, wrapper, composedOf) + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + resp2, err := session.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo, true) + require.NoError(t, err) + require.NotNil(t, resp2.ListToken) + require.Equal(t, resp2.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp2.CompleteListing) + require.Equal(t, resp2.EstimatedItemCount, 4) + require.Empty(t, resp2.DeletedIds) + require.Len(t, resp2.Items, 1) + require.Empty(t, cmp.Diff(resp2.Items[0], newSession, cmpIgnoreUnexportedOpts, cmpIgnoreFieldsOpts)) + }) +} diff --git a/internal/session/service_list_page.go b/internal/session/service_list_page.go new file mode 100644 index 000000000..b1ea12060 --- /dev/null +++ b/internal/session/service_list_page.go @@ -0,0 +1,70 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package session + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/pagination" +) + +// ListPage lists up to page size sessions, filtering out entries that +// do not pass the filter item function. It will automatically request +// more sessions from the database, at page size chunks, to fill the page. +// It will start its paging based on the information in the token. +// It returns a new list token used to continue pagination or refresh items. +// Sessions are ordered by create time descending (most recently created first). +func ListPage( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn pagination.ListFilterFunc[*Session], + tok *listtoken.Token, + repo *Repository, + includeTerminated bool, +) (*pagination.ListResponse[*Session], error) { + const op = "session.ListPage" + + switch { + case len(grantsHash) == 0: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case tok == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + } + if _, ok := tok.Subtype.(*listtoken.PaginationToken); !ok { + return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a pagination token component") + } + + listItemsFn := func(ctx context.Context, lastPageItem *Session, limit int) ([]*Session, time.Time, error) { + opts := []Option{ + WithLimit(limit), + } + if lastPageItem != nil { + opts = append(opts, WithStartPageAfterItem(lastPageItem)) + } else { + lastItem, err := tok.LastItem(ctx) + if err != nil { + return nil, time.Time{}, err + } + opts = append(opts, WithStartPageAfterItem(lastItem)) + } + if includeTerminated { + opts = append(opts, + WithTerminated(includeTerminated), + ) + } + return repo.listSessions(ctx, opts...) + } + + return pagination.ListPage(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedCount, tok) +} diff --git a/internal/session/service_list_refresh.go b/internal/session/service_list_refresh.go new file mode 100644 index 000000000..09d2dc5eb --- /dev/null +++ b/internal/session/service_list_refresh.go @@ -0,0 +1,76 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package session + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/globals" + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/pagination" +) + +// ListRefresh lists up to page size sessions, filtering out entries that +// do not pass the filter item function. It will automatically request +// more sessions from the database, at page size chunks, to fill the page. +// It will start its paging based on the information in the token. +// It returns a new list token used to continue pagination or refresh items. +// Sessions are ordered by update time descending (most recently updated first). +// Sessions may contain items that were already returned during the initial +// pagination phase. It also returns a list of any sessions deleted since the +// start of the initial pagination phase or last response. +func ListRefresh( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn pagination.ListFilterFunc[*Session], + tok *listtoken.Token, + repo *Repository, + includeTerminated bool, +) (*pagination.ListResponse[*Session], error) { + const op = "session.ListRefresh" + + switch { + case len(grantsHash) == 0: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case tok == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + } + rt, ok := tok.Subtype.(*listtoken.StartRefreshToken) + if !ok { + return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a start-refresh token component") + } + + listItemsFn := func(ctx context.Context, lastPageItem *Session, limit int) ([]*Session, time.Time, error) { + opts := []Option{ + WithLimit(limit), + } + if lastPageItem != nil { + opts = append(opts, WithStartPageAfterItem(lastPageItem)) + } + if includeTerminated { + opts = append(opts, + WithTerminated(includeTerminated), + ) + } + // Add the database read timeout to account for any creations missed due to concurrent + // transactions in the initial pagination phase. + return repo.listSessionsRefresh(ctx, rt.PreviousPhaseUpperBound.Add(-globals.RefreshReadLookbackDuration), opts...) + } + listDeletedIdsFn := func(ctx context.Context, since time.Time) ([]string, time.Time, error) { + // Add the database read timeout to account for any deletions missed due to concurrent + // transactions in previous requests. + return repo.listDeletedIds(ctx, since.Add(-globals.RefreshReadLookbackDuration)) + } + + return pagination.ListRefresh(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedCount, listDeletedIdsFn, tok) +} diff --git a/internal/session/service_list_refresh_page.go b/internal/session/service_list_refresh_page.go new file mode 100644 index 000000000..2def76609 --- /dev/null +++ b/internal/session/service_list_refresh_page.go @@ -0,0 +1,83 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package session + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/globals" + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/pagination" +) + +// ListRefreshPage lists up to page size sessions, filtering out entries that +// do not pass the filter item function. It will automatically request +// more sessions from the database, at page size chunks, to fill the page. +// It will start its paging based on the information in the token. +// It returns a new list token used to continue pagination or refresh items. +// Sessions are ordered by update time descending (most recently updated first). +// Sessions may contain items that were already returned during the initial +// pagination phase. It also returns a list of any sessions deleted since the +// last response. +func ListRefreshPage( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn pagination.ListFilterFunc[*Session], + tok *listtoken.Token, + repo *Repository, + includeTerminated bool, +) (*pagination.ListResponse[*Session], error) { + const op = "session.ListRefreshPage" + + switch { + case len(grantsHash) == 0: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case tok == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + } + rt, ok := tok.Subtype.(*listtoken.RefreshToken) + if !ok { + return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a refresh token component") + } + + listItemsFn := func(ctx context.Context, lastPageItem *Session, limit int) ([]*Session, time.Time, error) { + opts := []Option{ + WithLimit(limit), + } + if lastPageItem != nil { + opts = append(opts, WithStartPageAfterItem(lastPageItem)) + } else { + lastItem, err := tok.LastItem(ctx) + if err != nil { + return nil, time.Time{}, err + } + opts = append(opts, WithStartPageAfterItem(lastItem)) + } + if includeTerminated { + opts = append(opts, + WithTerminated(includeTerminated), + ) + } + // Add the database read timeout to account for any creations missed due to concurrent + // transactions in the original list pagination phase. + return repo.listSessionsRefresh(ctx, rt.PhaseLowerBound.Add(-globals.RefreshReadLookbackDuration), opts...) + } + + listDeletedIdsFn := func(ctx context.Context, since time.Time) ([]string, time.Time, error) { + // Add the database read timeout to account for any deletes missed due to concurrent + // transactions in the original list pagination phase. + return repo.listDeletedIds(ctx, since.Add(-globals.RefreshReadLookbackDuration)) + } + + return pagination.ListRefreshPage(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedCount, listDeletedIdsFn, tok) +} diff --git a/internal/session/session.go b/internal/session/session.go index 565a56f3d..37f329c83 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -18,6 +18,7 @@ import ( "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/db/timestamp" "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/types/resource" "github.com/hashicorp/boundary/internal/util" wrapping "github.com/hashicorp/go-kms-wrapping/v2" "github.com/hashicorp/go-kms-wrapping/v2/extras/structwrapping" @@ -152,6 +153,33 @@ func (s Session) GetUserId() string { return s.UserId } +// GetResourceType returns the resource type of the Session +func (s Session) GetResourceType() resource.Type { + return resource.Session +} + +func (s Session) GetUpdateTime() *timestamp.Timestamp { + return s.UpdateTime +} + +func (s Session) GetCreateTime() *timestamp.Timestamp { + return s.CreateTime +} + +// GetDescription returns an empty string so that +// Session will satisfy resource requirements +func (s Session) GetDescription() string { + return "" +} + +func (s Session) GetName() string { + return "" +} + +func (s Session) GetVersion() uint32 { + return 0 +} + var ( _ Cloneable = (*Session)(nil) _ db.VetForWriter = (*Session)(nil) @@ -571,3 +599,13 @@ type sessionListView struct { func (s *sessionListView) TableName() string { return "session_list" } + +type deletedSession struct { + PublicId string `gorm:"primary_key"` + DeleteTime *timestamp.Timestamp +} + +// TableName returns the tablename to override the default gorm table name +func (s *deletedSession) TableName() string { + return "session_deleted" +}