mirror of
https://github.com/gravitational/teleport.git
synced 2026-10-12 02:15:01 +00:00
[api] Add ListSemaphores (#60672)
* [api] Add ListSemaphores This commits deperecates GetSemaphores, which should be replaced by a paginated counterpart: ListSemaphores Note that GetSemaphores is also used as a point getter when both Kind and Name are given. This commit only focuses on pagination and as such takes the existing pattern and adds paging support without adding a point getter for consistency with other RPCs. Ranging is not implemented on client side due to the compound key used. Towards: gravitational/teleport.e/issues/6759 Changelog: Add paginated API ListSemaphores, deprecate GetSemaphores * nil safe getters for filter, simplify match * only use new list for non point gets
This commit is contained in:
@@ -1424,7 +1424,9 @@ func (c *Client) CancelSemaphoreLease(ctx context.Context, lease types.Semaphore
|
||||
}
|
||||
|
||||
// GetSemaphores returns a list of all semaphores matching the supplied filter.
|
||||
// Deprecated: Prefer paginated variant such as [Client.ListSemaphores]
|
||||
func (c *Client) GetSemaphores(ctx context.Context, filter types.SemaphoreFilter) ([]types.Semaphore, error) {
|
||||
//nolint:staticcheck // TODO(okraport): deprecated, to be removed in v21
|
||||
rsp, err := c.grpc.GetSemaphores(ctx, &filter)
|
||||
if err != nil {
|
||||
return nil, trace.Wrap(err)
|
||||
@@ -1436,6 +1438,25 @@ func (c *Client) GetSemaphores(ctx context.Context, filter types.SemaphoreFilter
|
||||
return sems, nil
|
||||
}
|
||||
|
||||
// ListSemaphores returns a page of semaphores matching supplied filter.
|
||||
func (c *Client) ListSemaphores(ctx context.Context, limit int, start string, filter *types.SemaphoreFilter) ([]types.Semaphore, string, error) {
|
||||
resp, err := c.grpc.ListSemaphores(ctx, &proto.ListSemaphoresRequest{
|
||||
PageSize: int32(limit),
|
||||
PageToken: start,
|
||||
Filter: filter,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, "", trace.Wrap(err)
|
||||
}
|
||||
|
||||
sems := make([]types.Semaphore, 0, len(resp.Semaphores))
|
||||
for _, s := range resp.Semaphores {
|
||||
sems = append(sems, s)
|
||||
}
|
||||
|
||||
return sems, resp.NextPageToken, nil
|
||||
}
|
||||
|
||||
// DeleteSemaphore deletes a semaphore matching the supplied filter.
|
||||
func (c *Client) DeleteSemaphore(ctx context.Context, filter types.SemaphoreFilter) error {
|
||||
_, err := c.grpc.DeleteSemaphore(ctx, &filter)
|
||||
|
||||
+1785
-1255
File diff suppressed because it is too large
Load Diff
@@ -91,6 +91,7 @@ const (
|
||||
AuthService_KeepAliveSemaphoreLease_FullMethodName = "/proto.AuthService/KeepAliveSemaphoreLease"
|
||||
AuthService_CancelSemaphoreLease_FullMethodName = "/proto.AuthService/CancelSemaphoreLease"
|
||||
AuthService_GetSemaphores_FullMethodName = "/proto.AuthService/GetSemaphores"
|
||||
AuthService_ListSemaphores_FullMethodName = "/proto.AuthService/ListSemaphores"
|
||||
AuthService_DeleteSemaphore_FullMethodName = "/proto.AuthService/DeleteSemaphore"
|
||||
AuthService_EmitAuditEvent_FullMethodName = "/proto.AuthService/EmitAuditEvent"
|
||||
AuthService_CreateAuditStream_FullMethodName = "/proto.AuthService/CreateAuditStream"
|
||||
@@ -456,8 +457,11 @@ type AuthServiceClient interface {
|
||||
KeepAliveSemaphoreLease(ctx context.Context, in *types.SemaphoreLease, opts ...grpc.CallOption) (*emptypb.Empty, error)
|
||||
// CancelSemaphoreLease cancels semaphore lease early.
|
||||
CancelSemaphoreLease(ctx context.Context, in *types.SemaphoreLease, opts ...grpc.CallOption) (*emptypb.Empty, error)
|
||||
// Deprecated: Do not use.
|
||||
// GetSemaphores returns a list of all semaphores matching the supplied filter.
|
||||
GetSemaphores(ctx context.Context, in *types.SemaphoreFilter, opts ...grpc.CallOption) (*Semaphores, error)
|
||||
// ListSemaphores returns a page of all semaphores matching the supplied filter.
|
||||
ListSemaphores(ctx context.Context, in *ListSemaphoresRequest, opts ...grpc.CallOption) (*ListSemaphoresResponse, error)
|
||||
// DeleteSemaphore deletes a semaphore matching the supplied filter.
|
||||
DeleteSemaphore(ctx context.Context, in *types.SemaphoreFilter, opts ...grpc.CallOption) (*emptypb.Empty, error)
|
||||
// EmitAuditEvent emits audit event
|
||||
@@ -1698,6 +1702,7 @@ func (c *authServiceClient) CancelSemaphoreLease(ctx context.Context, in *types.
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Deprecated: Do not use.
|
||||
func (c *authServiceClient) GetSemaphores(ctx context.Context, in *types.SemaphoreFilter, opts ...grpc.CallOption) (*Semaphores, error) {
|
||||
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
|
||||
out := new(Semaphores)
|
||||
@@ -1708,6 +1713,16 @@ func (c *authServiceClient) GetSemaphores(ctx context.Context, in *types.Semapho
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (c *authServiceClient) ListSemaphores(ctx context.Context, in *ListSemaphoresRequest, opts ...grpc.CallOption) (*ListSemaphoresResponse, error) {
|
||||
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
|
||||
out := new(ListSemaphoresResponse)
|
||||
err := c.cc.Invoke(ctx, AuthService_ListSemaphores_FullMethodName, in, out, cOpts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (c *authServiceClient) DeleteSemaphore(ctx context.Context, in *types.SemaphoreFilter, opts ...grpc.CallOption) (*emptypb.Empty, error) {
|
||||
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
|
||||
out := new(emptypb.Empty)
|
||||
@@ -4112,8 +4127,11 @@ type AuthServiceServer interface {
|
||||
KeepAliveSemaphoreLease(context.Context, *types.SemaphoreLease) (*emptypb.Empty, error)
|
||||
// CancelSemaphoreLease cancels semaphore lease early.
|
||||
CancelSemaphoreLease(context.Context, *types.SemaphoreLease) (*emptypb.Empty, error)
|
||||
// Deprecated: Do not use.
|
||||
// GetSemaphores returns a list of all semaphores matching the supplied filter.
|
||||
GetSemaphores(context.Context, *types.SemaphoreFilter) (*Semaphores, error)
|
||||
// ListSemaphores returns a page of all semaphores matching the supplied filter.
|
||||
ListSemaphores(context.Context, *ListSemaphoresRequest) (*ListSemaphoresResponse, error)
|
||||
// DeleteSemaphore deletes a semaphore matching the supplied filter.
|
||||
DeleteSemaphore(context.Context, *types.SemaphoreFilter) (*emptypb.Empty, error)
|
||||
// EmitAuditEvent emits audit event
|
||||
@@ -4896,6 +4914,9 @@ func (UnimplementedAuthServiceServer) CancelSemaphoreLease(context.Context, *typ
|
||||
func (UnimplementedAuthServiceServer) GetSemaphores(context.Context, *types.SemaphoreFilter) (*Semaphores, error) {
|
||||
return nil, status.Errorf(codes.Unimplemented, "method GetSemaphores not implemented")
|
||||
}
|
||||
func (UnimplementedAuthServiceServer) ListSemaphores(context.Context, *ListSemaphoresRequest) (*ListSemaphoresResponse, error) {
|
||||
return nil, status.Errorf(codes.Unimplemented, "method ListSemaphores not implemented")
|
||||
}
|
||||
func (UnimplementedAuthServiceServer) DeleteSemaphore(context.Context, *types.SemaphoreFilter) (*emptypb.Empty, error) {
|
||||
return nil, status.Errorf(codes.Unimplemented, "method DeleteSemaphore not implemented")
|
||||
}
|
||||
@@ -6467,6 +6488,24 @@ func _AuthService_GetSemaphores_Handler(srv interface{}, ctx context.Context, de
|
||||
return interceptor(ctx, in, info, handler)
|
||||
}
|
||||
|
||||
func _AuthService_ListSemaphores_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
|
||||
in := new(ListSemaphoresRequest)
|
||||
if err := dec(in); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if interceptor == nil {
|
||||
return srv.(AuthServiceServer).ListSemaphores(ctx, in)
|
||||
}
|
||||
info := &grpc.UnaryServerInfo{
|
||||
Server: srv,
|
||||
FullMethod: AuthService_ListSemaphores_FullMethodName,
|
||||
}
|
||||
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
|
||||
return srv.(AuthServiceServer).ListSemaphores(ctx, req.(*ListSemaphoresRequest))
|
||||
}
|
||||
return interceptor(ctx, in, info, handler)
|
||||
}
|
||||
|
||||
func _AuthService_DeleteSemaphore_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
|
||||
in := new(types.SemaphoreFilter)
|
||||
if err := dec(in); err != nil {
|
||||
@@ -10513,6 +10552,10 @@ var AuthService_ServiceDesc = grpc.ServiceDesc{
|
||||
MethodName: "GetSemaphores",
|
||||
Handler: _AuthService_GetSemaphores_Handler,
|
||||
},
|
||||
{
|
||||
MethodName: "ListSemaphores",
|
||||
Handler: _AuthService_ListSemaphores_Handler,
|
||||
},
|
||||
{
|
||||
MethodName: "DeleteSemaphore",
|
||||
Handler: _AuthService_DeleteSemaphore_Handler,
|
||||
|
||||
@@ -372,6 +372,24 @@ message ChangePasswordRequest {
|
||||
webauthn.CredentialAssertionResponse Webauthn = 5 [(gogoproto.jsontag) = "webauthn"];
|
||||
}
|
||||
|
||||
message ListSemaphoresRequest {
|
||||
// The maximum number of items to return.
|
||||
// The server may impose a different page size at its discretion.
|
||||
int32 page_size = 1;
|
||||
// The next_page_token value returned from a previous List request, if any.
|
||||
string page_token = 2;
|
||||
// filter encodes semaphore filtering params.
|
||||
types.SemaphoreFilter filter = 3;
|
||||
}
|
||||
|
||||
message ListSemaphoresResponse {
|
||||
// a list of semaphores.
|
||||
repeated types.SemaphoreV3 semaphores = 1;
|
||||
// Token to retrieve the next page of results, or empty if there are no
|
||||
// more results in the list.
|
||||
string next_page_token = 2;
|
||||
}
|
||||
|
||||
// PluginDataSeq is a sequence of plugin data.
|
||||
message PluginDataSeq {
|
||||
repeated types.PluginDataV3 PluginData = 1 [(gogoproto.jsontag) = "plugin_data"];
|
||||
@@ -3083,7 +3101,11 @@ service AuthService {
|
||||
// CancelSemaphoreLease cancels semaphore lease early.
|
||||
rpc CancelSemaphoreLease(types.SemaphoreLease) returns (google.protobuf.Empty);
|
||||
// GetSemaphores returns a list of all semaphores matching the supplied filter.
|
||||
rpc GetSemaphores(types.SemaphoreFilter) returns (Semaphores);
|
||||
rpc GetSemaphores(types.SemaphoreFilter) returns (Semaphores) {
|
||||
option deprecated = true;
|
||||
}
|
||||
// ListSemaphores returns a page of all semaphores matching the supplied filter.
|
||||
rpc ListSemaphores(ListSemaphoresRequest) returns (ListSemaphoresResponse);
|
||||
// DeleteSemaphore deletes a semaphore matching the supplied filter.
|
||||
rpc DeleteSemaphore(types.SemaphoreFilter) returns (google.protobuf.Empty);
|
||||
|
||||
|
||||
+22
-2
@@ -327,17 +327,37 @@ type Semaphores interface {
|
||||
CancelSemaphoreLease(ctx context.Context, lease SemaphoreLease) error
|
||||
// GetSemaphores returns a list of semaphores matching supplied filter.
|
||||
GetSemaphores(ctx context.Context, filter SemaphoreFilter) ([]Semaphore, error)
|
||||
// ListSemaphores returns a page of semaphores matching supplied filter.
|
||||
ListSemaphores(ctx context.Context, limit int, start string, filter *SemaphoreFilter) ([]Semaphore, string, error)
|
||||
// DeleteSemaphore deletes a semaphore matching supplied filter.
|
||||
DeleteSemaphore(ctx context.Context, filter SemaphoreFilter) error
|
||||
}
|
||||
|
||||
// Match checks if the supplied semaphore matches this filter.
|
||||
func (f *SemaphoreFilter) Match(sem Semaphore) bool {
|
||||
if f.SemaphoreKind != "" && f.SemaphoreKind != sem.GetSubKind() {
|
||||
if f.GetSemaphoreKind() != "" && f.GetSemaphoreKind() != sem.GetSubKind() {
|
||||
return false
|
||||
}
|
||||
if f.SemaphoreName != "" && f.SemaphoreName != sem.GetName() {
|
||||
if f.GetSemaphoreName() != "" && f.GetSemaphoreName() != sem.GetName() {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// GetSemaphoreKind returns the semaphore kind to filter by if filter is non-nil
|
||||
func (f *SemaphoreFilter) GetSemaphoreKind() string {
|
||||
if f == nil {
|
||||
return ""
|
||||
}
|
||||
|
||||
return f.SemaphoreKind
|
||||
}
|
||||
|
||||
// GetSemaphoreName returns the semaphore name to filter by if filter is non-nil
|
||||
func (f *SemaphoreFilter) GetSemaphoreName() string {
|
||||
if f == nil {
|
||||
return ""
|
||||
}
|
||||
|
||||
return f.SemaphoreName
|
||||
}
|
||||
|
||||
@@ -133,7 +133,6 @@ func newDefaultConfig() *Config {
|
||||
"proto.AuthService.GetLocks": {},
|
||||
"proto.AuthService.GetMFADevices": {},
|
||||
"proto.AuthService.GetPluginData": {},
|
||||
"proto.AuthService.GetSemaphores": {},
|
||||
"proto.AuthService.GetSessionEvents": {},
|
||||
"proto.AuthService.GetSSHTargets": {},
|
||||
"proto.AuthService.GetSSODiagnosticInfo": {},
|
||||
|
||||
@@ -2033,11 +2033,18 @@ func testDisconnectScenarios(t *testing.T, suite *integrationTestSuite) {
|
||||
postFunc: func(ctx context.Context, t *testing.T, teleport *helpers.TeleInstance) {
|
||||
site := teleport.GetSiteAPI(helpers.Site)
|
||||
require.EventuallyWithT(t, func(t *assert.CollectT) {
|
||||
sems, err := site.GetSemaphores(ctx, types.SemaphoreFilter{
|
||||
filter := types.SemaphoreFilter{
|
||||
SemaphoreKind: types.SemaphoreKindConnection,
|
||||
})
|
||||
}
|
||||
sems, err := site.GetSemaphores(ctx, filter)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, sems, 1)
|
||||
|
||||
sems, next, err := site.ListSemaphores(ctx, 0, "", &filter)
|
||||
require.Empty(t, next)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, sems, 1)
|
||||
|
||||
}, 2*time.Second, 100*time.Millisecond)
|
||||
|
||||
timeoutCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
|
||||
@@ -5587,6 +5587,14 @@ func (a *ServerWithRoles) GetSemaphores(ctx context.Context, filter types.Semaph
|
||||
return a.authServer.GetSemaphores(ctx, filter)
|
||||
}
|
||||
|
||||
// ListSemaphores returns a page of semaphores matching supplied filter.
|
||||
func (a *ServerWithRoles) ListSemaphores(ctx context.Context, limit int, start string, filter *types.SemaphoreFilter) ([]types.Semaphore, string, error) {
|
||||
if err := a.authorizeAction(types.KindSemaphore, types.VerbReadNoSecrets, types.VerbList); err != nil {
|
||||
return nil, "", trace.Wrap(err)
|
||||
}
|
||||
return a.authServer.ListSemaphores(ctx, limit, start, filter)
|
||||
}
|
||||
|
||||
// DeleteSemaphore deletes a semaphore matching the supplied filter.
|
||||
func (a *ServerWithRoles) DeleteSemaphore(ctx context.Context, filter types.SemaphoreFilter) error {
|
||||
if err := a.authorizeAction(types.KindSemaphore, types.VerbDelete); err != nil {
|
||||
|
||||
@@ -1362,6 +1362,34 @@ func (g *GRPCServer) GetSemaphores(ctx context.Context, req *types.SemaphoreFilt
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ListSemaphores returns a page of semaphores matching supplied filter.
|
||||
func (g *GRPCServer) ListSemaphores(ctx context.Context, req *authpb.ListSemaphoresRequest) (*authpb.ListSemaphoresResponse, error) {
|
||||
auth, err := g.authenticate(ctx)
|
||||
if err != nil {
|
||||
return nil, trace.Wrap(err)
|
||||
}
|
||||
|
||||
semaphores, next, err := auth.ListSemaphores(ctx, int(req.GetPageSize()), req.GetPageToken(), req.GetFilter())
|
||||
if err != nil {
|
||||
return nil, trace.Wrap(err)
|
||||
}
|
||||
|
||||
resp := &authpb.ListSemaphoresResponse{
|
||||
Semaphores: make([]*types.SemaphoreV3, 0, len(semaphores)),
|
||||
NextPageToken: next,
|
||||
}
|
||||
|
||||
for _, sem := range semaphores {
|
||||
v3, ok := sem.(*types.SemaphoreV3)
|
||||
if !ok {
|
||||
return nil, trace.BadParameter("unexpected semaphore type: %T", sem)
|
||||
}
|
||||
resp.Semaphores = append(resp.Semaphores, v3)
|
||||
}
|
||||
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// DeleteSemaphore deletes a semaphore matching the supplied filter.
|
||||
func (g *GRPCServer) DeleteSemaphore(ctx context.Context, req *types.SemaphoreFilter) (*emptypb.Empty, error) {
|
||||
auth, err := g.authenticate(ctx)
|
||||
|
||||
@@ -20,6 +20,7 @@ package local
|
||||
|
||||
import (
|
||||
"context"
|
||||
"iter"
|
||||
"log/slog"
|
||||
"sort"
|
||||
"time"
|
||||
@@ -789,6 +790,60 @@ func (s *PresenceService) GetSemaphores(ctx context.Context, filter types.Semaph
|
||||
return sems, nil
|
||||
}
|
||||
|
||||
func (s *PresenceService) rangeSemaphores(ctx context.Context, start string, filter *types.SemaphoreFilter) iter.Seq2[types.Semaphore, error] {
|
||||
mapFn := func(item backend.Item) (types.Semaphore, bool) {
|
||||
sem, err := services.UnmarshalSemaphore(item.Value,
|
||||
services.WithExpires(item.Expires),
|
||||
services.WithRevision(item.Revision))
|
||||
if err != nil {
|
||||
s.logger.WarnContext(ctx, "Failed to unmarshal semaphore",
|
||||
"key", item.Key,
|
||||
"error", err,
|
||||
)
|
||||
return nil, false
|
||||
}
|
||||
|
||||
return sem, filter.Match(sem)
|
||||
}
|
||||
|
||||
startKey := backend.NewKey(semaphoresPrefix).AppendKey(backend.KeyFromString(start))
|
||||
endKey := backend.RangeEnd(backend.NewKey(semaphoresPrefix))
|
||||
|
||||
filterKind := filter.GetSemaphoreKind()
|
||||
filterName := filter.GetSemaphoreName()
|
||||
|
||||
// If the filter is set we optimize the ranges to avoid fetching items that will be filtered.
|
||||
if filterKind != "" && filterName != "" {
|
||||
// Special case when both kind and name are set the result should yield only a single item.
|
||||
startKey = backend.NewKey(semaphoresPrefix, filterKind, filterName)
|
||||
endKey = startKey
|
||||
} else if filterKind != "" {
|
||||
// Terminate range early as we know the only kind requested.
|
||||
endKey = backend.RangeEnd(backend.NewKey(semaphoresPrefix, filterKind))
|
||||
if start == "" {
|
||||
// We are not resuming previous list, skip to the start of the correct kind range
|
||||
startKey = backend.NewKey(semaphoresPrefix, filterKind)
|
||||
}
|
||||
}
|
||||
|
||||
return stream.FilterMap(
|
||||
s.Backend.Items(ctx, backend.ItemsParams{
|
||||
StartKey: startKey,
|
||||
EndKey: endKey,
|
||||
}),
|
||||
mapFn,
|
||||
)
|
||||
}
|
||||
|
||||
func semaphoreToPageToken(sem types.Semaphore) string {
|
||||
return sem.GetSubKind() + backend.SeparatorString + sem.GetName()
|
||||
}
|
||||
|
||||
// ListSemaphores returns a page of semaphores matching supplied filter.
|
||||
func (s *PresenceService) ListSemaphores(ctx context.Context, limit int, start string, filter *types.SemaphoreFilter) ([]types.Semaphore, string, error) {
|
||||
return generic.CollectPageAndCursor(s.rangeSemaphores(ctx, start, filter), limit, semaphoreToPageToken)
|
||||
}
|
||||
|
||||
// DeleteSemaphore deletes a semaphore matching the supplied filter
|
||||
func (s *PresenceService) DeleteSemaphore(ctx context.Context, filter types.SemaphoreFilter) error {
|
||||
if filter.SemaphoreKind == "" || filter.SemaphoreName == "" {
|
||||
|
||||
@@ -19,9 +19,11 @@
|
||||
package local
|
||||
|
||||
import (
|
||||
gocmp "cmp"
|
||||
"context"
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -1100,6 +1102,15 @@ func TestPresenceService_CancelSemaphoreLease(t *testing.T) {
|
||||
require.Len(t, semaphores, 1)
|
||||
require.Len(t, semaphores[0].LeaseRefs(), maxLeases)
|
||||
|
||||
semaphores, next, err := presence.ListSemaphores(ctx, 0, "", &types.SemaphoreFilter{
|
||||
SemaphoreKind: "test",
|
||||
SemaphoreName: "test",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, next)
|
||||
require.Len(t, semaphores, 1)
|
||||
require.Len(t, semaphores[0].LeaseRefs(), maxLeases)
|
||||
|
||||
// Cancel the leases concurrently and ensure that all
|
||||
// cancellations are honored
|
||||
errCh := make(chan error, maxLeases)
|
||||
@@ -1122,6 +1133,16 @@ func TestPresenceService_CancelSemaphoreLease(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
require.Len(t, semaphores, 1)
|
||||
require.Empty(t, semaphores[0].LeaseRefs())
|
||||
|
||||
semaphores, next, err = presence.ListSemaphores(ctx, 0, "", &types.SemaphoreFilter{
|
||||
SemaphoreKind: "test",
|
||||
SemaphoreName: "test",
|
||||
})
|
||||
|
||||
require.Empty(t, next)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, semaphores, 1)
|
||||
require.Empty(t, semaphores[0].LeaseRefs())
|
||||
}
|
||||
|
||||
// TestListResources_DuplicateResourceFilterByLabel tests that we can search for a specific label
|
||||
@@ -1493,3 +1514,171 @@ func TestPresenceService_RelayServer(t *testing.T) {
|
||||
require.Len(t, shortList, 1)
|
||||
require.Empty(t, cmp.Diff(relayB, shortList[0], diffOpts...))
|
||||
}
|
||||
|
||||
func TestPresenceService_ListSemaphores(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := t.Context()
|
||||
bk, err := memory.New(memory.Config{})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { bk.Close() })
|
||||
presence := NewPresenceService(bk)
|
||||
|
||||
newRequest := func(kind, name string) types.AcquireSemaphoreRequest {
|
||||
return types.AcquireSemaphoreRequest{
|
||||
SemaphoreKind: kind,
|
||||
SemaphoreName: name,
|
||||
MaxLeases: 5,
|
||||
Expires: time.Now().Add(time.Hour),
|
||||
Holder: "test",
|
||||
}
|
||||
}
|
||||
|
||||
expectedFromReq := func(req types.AcquireSemaphoreRequest) types.Semaphore {
|
||||
sem := &types.SemaphoreV3{
|
||||
SubKind: req.SemaphoreKind,
|
||||
Metadata: types.Metadata{
|
||||
Name: req.SemaphoreName,
|
||||
}}
|
||||
sem.CheckAndSetDefaults()
|
||||
return sem
|
||||
}
|
||||
|
||||
pageSize := 12
|
||||
totalItems := pageSize*2 + (pageSize / 2)
|
||||
expected := make([]types.Semaphore, 0, totalItems)
|
||||
|
||||
for i := range totalItems {
|
||||
kind := types.SemaphoreKindConnection
|
||||
name := fmt.Sprintf("sem-%05d", i)
|
||||
switch {
|
||||
case i%5 == 0:
|
||||
kind = types.SemaphoreKindUploadCompleter
|
||||
case i%3 == 0:
|
||||
// every third name is duplicated with previous item
|
||||
name = expected[i-1].GetName()
|
||||
kind = types.SemaphoreKindKubernetesConnection
|
||||
case i%2 == 0:
|
||||
kind = types.SemaphoreKindAccessListReminderLimiter
|
||||
}
|
||||
request := newRequest(kind, name)
|
||||
_, err := presence.AcquireSemaphore(ctx, request)
|
||||
require.NoError(t, err)
|
||||
expected = append(expected, expectedFromReq(request))
|
||||
|
||||
}
|
||||
|
||||
slices.SortFunc(expected, func(a, b types.Semaphore) int {
|
||||
if c := gocmp.Compare(a.GetSubKind(), b.GetSubKind()); c != 0 {
|
||||
return c
|
||||
}
|
||||
return gocmp.Compare(a.GetName(), b.GetName())
|
||||
})
|
||||
diffopts := cmp.Options{cmpopts.IgnoreFields(types.Metadata{}, "Revision"),
|
||||
cmpopts.IgnoreFields(types.SemaphoreSpecV3{}, "Leases"),
|
||||
cmpopts.IgnoreFields(types.Metadata{}, "Expires")}
|
||||
|
||||
t.Run("filter by name", func(t *testing.T) {
|
||||
wantName := "sem-00002"
|
||||
page, next, err := presence.ListSemaphores(ctx, 0, "", &types.SemaphoreFilter{
|
||||
SemaphoreName: wantName,
|
||||
})
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, next)
|
||||
require.Len(t, page, 2)
|
||||
for _, sem := range page {
|
||||
require.Equal(t, wantName, sem.GetName())
|
||||
}
|
||||
require.NotEqual(t, page[0].GetSubKind(), page[1].GetSubKind())
|
||||
})
|
||||
|
||||
t.Run("filter by name and kind returns single item", func(t *testing.T) {
|
||||
wantKind := types.SemaphoreKindUploadCompleter
|
||||
wantName := "sem-00000"
|
||||
page, next, err := presence.ListSemaphores(ctx, 0, "", &types.SemaphoreFilter{
|
||||
SemaphoreKind: wantKind,
|
||||
SemaphoreName: wantName,
|
||||
})
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, next)
|
||||
require.Len(t, page, 1)
|
||||
require.Equal(t, wantKind, page[0].GetSubKind())
|
||||
require.Equal(t, wantName, page[0].GetName())
|
||||
})
|
||||
|
||||
t.Run("unfiltered list all", func(t *testing.T) {
|
||||
got, gotNext, err := presence.ListSemaphores(ctx, 1000, "", nil)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, gotNext)
|
||||
require.Len(t, got, totalItems)
|
||||
require.Empty(t, cmp.Diff(expected, got, diffopts...))
|
||||
})
|
||||
|
||||
t.Run("pagination unfiltered", func(t *testing.T) {
|
||||
page1, page2Start, err := presence.ListSemaphores(ctx, pageSize, "", nil)
|
||||
require.NoError(t, err)
|
||||
require.NotEmpty(t, page2Start)
|
||||
require.Len(t, page1, pageSize)
|
||||
require.Empty(t, cmp.Diff(expected[:pageSize], page1, diffopts...))
|
||||
|
||||
page2, page3Start, err := presence.ListSemaphores(ctx, pageSize, page2Start, nil)
|
||||
require.NoError(t, err)
|
||||
require.NotEmpty(t, page3Start)
|
||||
require.Len(t, page2, pageSize)
|
||||
require.Empty(t, cmp.Diff(expected[pageSize:pageSize*2], page2, diffopts...))
|
||||
|
||||
page3, next, err := presence.ListSemaphores(ctx, pageSize, page3Start, nil)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, next)
|
||||
require.Len(t, page3, totalItems-2*pageSize)
|
||||
require.Empty(t, cmp.Diff(expected[pageSize*2:], page3, diffopts...))
|
||||
|
||||
allPages := append(page1, append(page2, page3...)...)
|
||||
require.Empty(t, cmp.Diff(expected, allPages, diffopts...))
|
||||
})
|
||||
|
||||
t.Run("pagination filtered by kind", func(t *testing.T) {
|
||||
filter := &types.SemaphoreFilter{SemaphoreKind: types.SemaphoreKindKubernetesConnection}
|
||||
page1, page2Start, err := presence.ListSemaphores(ctx, 2, "", filter)
|
||||
require.NoError(t, err)
|
||||
require.NotEmpty(t, page2Start)
|
||||
require.Len(t, page1, 2)
|
||||
|
||||
page2, page3Start, err := presence.ListSemaphores(ctx, 2, page2Start, filter)
|
||||
require.NoError(t, err)
|
||||
require.NotEmpty(t, page3Start)
|
||||
require.Len(t, page2, 2)
|
||||
|
||||
page3, next, err := presence.ListSemaphores(ctx, 0, page3Start, filter)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, next)
|
||||
require.Len(t, page3, 4)
|
||||
|
||||
allPages := append(page1, append(page2, page3...)...)
|
||||
|
||||
for _, sem := range allPages {
|
||||
require.Equal(t, types.SemaphoreKindKubernetesConnection, sem.GetSubKind())
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("pagination filtered by name", func(t *testing.T) {
|
||||
wantName := "sem-00002"
|
||||
filter := &types.SemaphoreFilter{SemaphoreName: wantName}
|
||||
page1, page2Start, err := presence.ListSemaphores(ctx, 1, "", filter)
|
||||
require.NoError(t, err)
|
||||
require.NotEmpty(t, page2Start)
|
||||
require.Len(t, page1, 1)
|
||||
|
||||
page2, next, err := presence.ListSemaphores(ctx, 0, page2Start, filter)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, next)
|
||||
require.Len(t, page2, 1)
|
||||
|
||||
allPages := append(page1, page2...)
|
||||
for _, sem := range allPages {
|
||||
require.Equal(t, wantName, sem.GetName())
|
||||
}
|
||||
})
|
||||
|
||||
}
|
||||
|
||||
@@ -1904,10 +1904,19 @@ func (rc *ResourceCommand) getCollection(ctx context.Context, client *authclient
|
||||
}
|
||||
return &remoteClusterCollection{remoteClusters: []types.RemoteCluster{remoteCluster}}, nil
|
||||
case types.KindSemaphore:
|
||||
sems, err := client.GetSemaphores(ctx, types.SemaphoreFilter{
|
||||
filter := types.SemaphoreFilter{
|
||||
SemaphoreKind: rc.ref.SubKind,
|
||||
SemaphoreName: rc.ref.Name,
|
||||
})
|
||||
}
|
||||
sems, err := clientutils.CollectWithFallback(ctx,
|
||||
func(ctx context.Context, pageSize int, pageToken string) ([]types.Semaphore, string, error) {
|
||||
return client.ListSemaphores(ctx, pageSize, pageToken, &filter)
|
||||
},
|
||||
func(ctx context.Context) ([]types.Semaphore, error) {
|
||||
return client.GetSemaphores(ctx, filter)
|
||||
},
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
return nil, trace.Wrap(err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user