mirror of
https://github.com/gravitational/teleport.git
synced 2026-10-11 22:49:54 +00:00
Add ScopedOnly mode to the generic scope-aware service (#68096)
The ScopeAwareService and ScopeAwareServiceWrapper no have an option ScopedOnly flag. When set, the service ignores all items in the unscoped key range. This is useful for _new_ scoped items that will never have any resources exist in an unscoped key range, i.e. ScopedTokens, ScopedRoles, and ScopedRoleAssignments.
This commit is contained in:
@@ -43,10 +43,16 @@ type ScopedResource interface {
|
||||
// separate key range from resources with an empty scope, and namespaced by
|
||||
// their scope. The ScopeAwareService transparently handles listing all scoped
|
||||
// and unscoped resources, as well as creating and querying individual resources
|
||||
// from the correct key range.
|
||||
// from the correct key range. A ScopeAwareService can also be configured to only
|
||||
// consume scoped resources. This is useful for situations in which a resource
|
||||
// has never had an unscoped variant.
|
||||
type ScopeAwareService[T ScopedResource] struct {
|
||||
// scopedOnly indicates that the service will only operate on scoped resources.
|
||||
// The unscoped fallback path will be ignored in all cases.
|
||||
scopedOnly bool
|
||||
// UnscopedService is the underlying service for resources with an empty scope.
|
||||
// Resources will be keyed at <backend_prefix>/<name>
|
||||
// Resources will be keyed at <backend_prefix>/<name>. It will be nil if the
|
||||
// service was configured to be scoped only.
|
||||
UnscopedService *Service[T]
|
||||
// ScopedService is the underlying service for resources with a scope.
|
||||
// Resources will be keyed at <scoped_prefix>/<backend_prefix>/<encoded_scope>/<name>
|
||||
@@ -55,6 +61,9 @@ type ScopeAwareService[T ScopedResource] struct {
|
||||
|
||||
// ScopeAwareServiceConfig holds configuration options for ScopeAwareService.
|
||||
type ScopeAwareServiceConfig[T ScopedResource] struct {
|
||||
// ScopedOnly indicates that the service will only operate on scoped resources.
|
||||
// The unscoped fallback path will be ignored in all cases.
|
||||
ScopedOnly bool
|
||||
// Backend used to persist the resource.
|
||||
Backend backend.Backend
|
||||
// ResourceKind is the friendly name of the resource.
|
||||
@@ -80,11 +89,11 @@ type ScopeAwareServiceConfig[T ScopedResource] struct {
|
||||
|
||||
// NewScopeAwareService returns a new scope-aware service.
|
||||
func NewScopeAwareService[T ScopedResource](cfg *ScopeAwareServiceConfig[T]) (*ScopeAwareService[T], error) {
|
||||
unscopedService, err := NewService(&ServiceConfig[T]{
|
||||
scopedService, err := NewService(&ServiceConfig[T]{
|
||||
Backend: cfg.Backend,
|
||||
ResourceKind: cfg.ResourceKind,
|
||||
PageLimit: cfg.PageLimit,
|
||||
BackendPrefix: cfg.UnscopedBackendPrefix,
|
||||
BackendPrefix: cfg.ScopedBackendPrefix,
|
||||
MarshalFunc: cfg.MarshalFunc,
|
||||
UnmarshalFunc: cfg.UnmarshalFunc,
|
||||
ValidateFunc: cfg.ValidateFunc,
|
||||
@@ -94,11 +103,18 @@ func NewScopeAwareService[T ScopedResource](cfg *ScopeAwareServiceConfig[T]) (*S
|
||||
return nil, trace.Wrap(err)
|
||||
}
|
||||
|
||||
scopedService, err := NewService(&ServiceConfig[T]{
|
||||
if cfg.ScopedOnly {
|
||||
return &ScopeAwareService[T]{
|
||||
scopedOnly: true,
|
||||
ScopedService: scopedService,
|
||||
}, nil
|
||||
}
|
||||
|
||||
unscopedService, err := NewService(&ServiceConfig[T]{
|
||||
Backend: cfg.Backend,
|
||||
ResourceKind: cfg.ResourceKind,
|
||||
PageLimit: cfg.PageLimit,
|
||||
BackendPrefix: cfg.ScopedBackendPrefix,
|
||||
BackendPrefix: cfg.UnscopedBackendPrefix,
|
||||
MarshalFunc: cfg.MarshalFunc,
|
||||
UnmarshalFunc: cfg.UnmarshalFunc,
|
||||
ValidateFunc: cfg.ValidateFunc,
|
||||
@@ -123,6 +139,22 @@ func NewScopeAwareService[T ScopedResource](cfg *ScopeAwareServiceConfig[T]) (*S
|
||||
// service's relative key. [scopes.ResourceCursorScopedStart] is the boundary
|
||||
// between unscoped and scoped resources.
|
||||
func (s *ScopeAwareService[T]) Resources(ctx context.Context, startKey, endKey string) iter.Seq2[T, error] {
|
||||
if s.scopedOnly {
|
||||
if endKey != "" && !scopes.IsScopedResourceCursor(endKey) {
|
||||
return stream.Empty[T]()
|
||||
}
|
||||
if endKey == scopes.ResourceCursorScopedStart() {
|
||||
return stream.Empty[T]()
|
||||
}
|
||||
|
||||
scopedStartKey := ""
|
||||
if scopes.IsScopedResourceCursor(startKey) {
|
||||
scopedStartKey = strings.TrimPrefix(startKey, scopes.ResourceCursorPrefix)
|
||||
}
|
||||
scopedEndKey := strings.TrimPrefix(endKey, scopes.ResourceCursorPrefix)
|
||||
return s.ScopedService.Resources(ctx, scopedStartKey, scopedEndKey)
|
||||
}
|
||||
|
||||
var streams []stream.Stream[T]
|
||||
|
||||
if !scopes.IsScopedResourceCursor(startKey) {
|
||||
@@ -166,14 +198,18 @@ func (s *ScopeAwareService[T]) ListResources(ctx context.Context, pageSize int,
|
||||
// unified scoped and unscoped collection. It always returns all matching
|
||||
// unscoped resources before matching scoped resources.
|
||||
func (s *ScopeAwareService[T]) ListResourcesWithFilter(ctx context.Context, pageSize int, nextToken string, matcher func(T) bool) ([]T, string, error) {
|
||||
if pageSize <= 0 || pageSize > int(s.UnscopedService.pageLimit) {
|
||||
pageSize = int(s.UnscopedService.pageLimit)
|
||||
if pageSize <= 0 || pageSize > int(s.ScopedService.pageLimit) {
|
||||
pageSize = int(s.ScopedService.pageLimit)
|
||||
}
|
||||
|
||||
// Check if the token was scoped, if so the caller has already paged over
|
||||
// all unscoped resources and we should return the next page of scoped
|
||||
// resources.
|
||||
if scopes.IsScopedResourceCursor(nextToken) {
|
||||
if s.scopedOnly && nextToken != "" && !scopes.IsScopedResourceCursor(nextToken) {
|
||||
return nil, "", trace.BadParameter("scoped-only storage service received invalid pagination token")
|
||||
}
|
||||
|
||||
// If in scoped only mode we don't care about unscoped resources. If the
|
||||
// token was scoped the caller has already paged over all unscoped
|
||||
// resources and we should return the next page of scoped resources.
|
||||
if s.scopedOnly || scopes.IsScopedResourceCursor(nextToken) {
|
||||
nextToken = strings.TrimPrefix(nextToken, scopes.ResourceCursorPrefix)
|
||||
resources, nextToken, err := s.ScopedService.ListResourcesWithFilter(ctx, pageSize, nextToken, matcher)
|
||||
if nextToken != "" {
|
||||
@@ -232,8 +268,12 @@ func (s *ScopeAwareService[T]) DeleteResource(ctx context.Context, scopedName sc
|
||||
return svc.DeleteResource(ctx, scopedName.Name)
|
||||
}
|
||||
|
||||
// DeleteAllResources deletes all scoped and unscoped resources.
|
||||
// DeleteAllResources deletes all scoped and unscoped resources (if not in scoped only mode).
|
||||
func (s *ScopeAwareService[T]) DeleteAllResources(ctx context.Context) error {
|
||||
if s.scopedOnly {
|
||||
return trace.Wrap(s.ScopedService.DeleteAllResources(ctx))
|
||||
}
|
||||
|
||||
return trace.NewAggregate(
|
||||
s.UnscopedService.DeleteAllResources(ctx),
|
||||
s.ScopedService.DeleteAllResources(ctx),
|
||||
@@ -292,6 +332,9 @@ func (s *ScopeAwareService[T]) ConditionalUpdateResource(ctx context.Context, re
|
||||
// returns the scoped service with the encoded scope appended to its backend prefix.
|
||||
func (s *ScopeAwareService[T]) WithScopePrefix(scope string) (*Service[T], error) {
|
||||
if scope == "" {
|
||||
if s.scopedOnly {
|
||||
return nil, trace.BadParameter("scoped-only storage service received an empty scope")
|
||||
}
|
||||
return s.UnscopedService, nil
|
||||
}
|
||||
encodedScope, err := scopes.EncodeForKey(scope)
|
||||
@@ -314,6 +357,9 @@ func (s *ScopeAwareService[T]) WithScopePrefix(scope string) (*Service[T], error
|
||||
// resource, i.e. members of a scoped access list.
|
||||
func (s *ScopeAwareService[T]) WithScopedResourcePrefix(scopedName scopes.QualifiedName) (*Service[T], error) {
|
||||
if scopedName.Scope == "" {
|
||||
if s.scopedOnly {
|
||||
return nil, trace.BadParameter("scoped-only storage service received an empty scope")
|
||||
}
|
||||
return s.UnscopedService.WithPrefix(scopedName.Name), nil
|
||||
}
|
||||
encodedScope, err := scopes.EncodeForKey(scopedName.Scope)
|
||||
|
||||
@@ -24,6 +24,7 @@ import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/gravitational/trace"
|
||||
"github.com/jonboulle/clockwork"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
@@ -263,3 +264,224 @@ func TestScopeAwareService(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, page)
|
||||
}
|
||||
|
||||
func newScopedOnlyServiceForTest(t *testing.T) *ScopeAwareService[*testResource] {
|
||||
t.Helper()
|
||||
|
||||
memBackend, err := memory.New(memory.Config{
|
||||
Context: t.Context(),
|
||||
Clock: clockwork.NewFakeClock(),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
service, err := NewScopeAwareService(&ScopeAwareServiceConfig[*testResource]{
|
||||
ScopedOnly: true,
|
||||
Backend: memBackend,
|
||||
ResourceKind: "generic resource",
|
||||
ScopedBackendPrefix: backend.NewKey(testScopedTopPrefix, testUnscopedPrefix),
|
||||
PageLimit: 200,
|
||||
MarshalFunc: marshalResource,
|
||||
UnmarshalFunc: unmarshalResource,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
return service
|
||||
}
|
||||
|
||||
func newScopedTestResourceWithBody(sqn scopes.QualifiedName) *testResource {
|
||||
r := newScopedTestResource(sqn)
|
||||
r.Spec.PropA = specDataFor(sqn)
|
||||
return r
|
||||
}
|
||||
|
||||
func requireTestResourceBody(t *testing.T, want scopes.QualifiedName, got *testResource) {
|
||||
t.Helper()
|
||||
require.Equal(t, want.Name, got.GetName())
|
||||
require.Equal(t, want.Scope, got.GetScope())
|
||||
require.Equal(t, specDataFor(want), got.Spec.PropA)
|
||||
}
|
||||
|
||||
func scopedNames(resources []*testResource) []scopes.QualifiedName {
|
||||
out := make([]scopes.QualifiedName, 0, len(resources))
|
||||
for _, r := range resources {
|
||||
out = append(out, scopes.QualifiedName{Scope: r.GetScope(), Name: r.GetName()})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func collectScopedStream(t *testing.T, seq iter.Seq2[*testResource, error]) []*testResource {
|
||||
t.Helper()
|
||||
var out []*testResource
|
||||
for r, err := range seq {
|
||||
require.NoError(t, err)
|
||||
out = append(out, r)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestScopeAwareService_ScopedOnly(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
t.Run("rejects empty scope on every operation", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
svc := newScopedOnlyServiceForTest(t)
|
||||
ctx := t.Context()
|
||||
unscoped := scopes.QualifiedName{Name: "foo"}
|
||||
|
||||
_, err := svc.CreateResource(ctx, newScopedTestResource(unscoped))
|
||||
require.True(t, trace.IsBadParameter(err), "create: %v", err)
|
||||
|
||||
_, err = svc.UpsertResource(ctx, newScopedTestResource(unscoped))
|
||||
require.True(t, trace.IsBadParameter(err), "upsert: %v", err)
|
||||
|
||||
_, err = svc.ConditionalUpdateResource(ctx, newScopedTestResource(unscoped))
|
||||
require.True(t, trace.IsBadParameter(err), "update: %v", err)
|
||||
|
||||
_, err = svc.GetResource(ctx, unscoped)
|
||||
require.True(t, trace.IsBadParameter(err), "get: %v", err)
|
||||
|
||||
err = svc.DeleteResource(ctx, unscoped)
|
||||
require.True(t, trace.IsBadParameter(err), "delete: %v", err)
|
||||
|
||||
_, err = svc.WithScopedResourcePrefix(unscoped)
|
||||
require.True(t, trace.IsBadParameter(err), "with scoped resource prefix: %v", err)
|
||||
|
||||
_, err = svc.WithScopePrefix("")
|
||||
require.True(t, trace.IsBadParameter(err), "with scope prefix: %v", err)
|
||||
})
|
||||
|
||||
t.Run("CRUD", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
svc := newScopedOnlyServiceForTest(t)
|
||||
ctx := t.Context()
|
||||
qn := scopes.QualifiedName{Scope: "/security", Name: "foo"}
|
||||
|
||||
created, err := svc.CreateResource(ctx, newScopedTestResourceWithBody(qn))
|
||||
require.NoError(t, err)
|
||||
|
||||
got, err := svc.GetResource(ctx, qn)
|
||||
require.NoError(t, err)
|
||||
requireTestResourceBody(t, qn, got)
|
||||
|
||||
updated := newScopedTestResource(qn)
|
||||
updated.Spec.PropA = "updated!"
|
||||
updated.Metadata.Revision = created.Metadata.Revision
|
||||
updated, err = svc.ConditionalUpdateResource(ctx, updated)
|
||||
require.NoError(t, err)
|
||||
require.NotEqual(t, created.Metadata.Revision, updated.Metadata.Revision)
|
||||
|
||||
got, err = svc.GetResource(ctx, qn)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "updated!", got.Spec.PropA)
|
||||
|
||||
_, err = svc.ConditionalUpdateResource(ctx, created)
|
||||
require.True(t, trace.IsCompareFailed(err))
|
||||
|
||||
upserted, err := svc.UpsertResource(ctx, created)
|
||||
require.NoError(t, err)
|
||||
requireTestResourceBody(t, qn, upserted)
|
||||
|
||||
require.NoError(t, svc.DeleteResource(ctx, qn))
|
||||
_, err = svc.GetResource(ctx, qn)
|
||||
require.True(t, trace.IsNotFound(err), "expected not found, got %v", err)
|
||||
})
|
||||
|
||||
t.Run("same name across scopes does not conflict", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
svc := newScopedOnlyServiceForTest(t)
|
||||
ctx := t.Context()
|
||||
a := scopes.QualifiedName{Scope: "/a", Name: "shared"}
|
||||
b := scopes.QualifiedName{Scope: "/b", Name: "shared"}
|
||||
_, err := svc.CreateResource(ctx, newScopedTestResourceWithBody(a))
|
||||
require.NoError(t, err)
|
||||
_, err = svc.CreateResource(ctx, newScopedTestResourceWithBody(b))
|
||||
require.NoError(t, err)
|
||||
|
||||
gotA, err := svc.GetResource(ctx, a)
|
||||
require.NoError(t, err)
|
||||
requireTestResourceBody(t, a, gotA)
|
||||
gotB, err := svc.GetResource(ctx, b)
|
||||
require.NoError(t, err)
|
||||
requireTestResourceBody(t, b, gotB)
|
||||
})
|
||||
|
||||
t.Run("Resources bounded ranges", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
svc := newScopedOnlyServiceForTest(t)
|
||||
ctx := t.Context()
|
||||
resources := []scopes.QualifiedName{
|
||||
{Scope: "/a", Name: "1"},
|
||||
{Scope: "/b", Name: "2"},
|
||||
{Scope: "/c", Name: "3"},
|
||||
}
|
||||
for _, qn := range resources {
|
||||
_, err := svc.CreateResource(ctx, newScopedTestResource(qn))
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// A scoped-only service has no unscoped resources, so ranges ending
|
||||
// before the scoped resource range must be empty.
|
||||
require.Empty(t, collectScopedStream(t, svc.Resources(ctx, "", "name5")))
|
||||
require.Empty(t, collectScopedStream(t, svc.Resources(ctx, "", scopes.ResourceCursorScopedStart())))
|
||||
|
||||
start, err := scopes.MakeResourceCursor("/b", "2")
|
||||
require.NoError(t, err)
|
||||
end, err := scopes.MakeResourceCursor("/c", "3")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t,
|
||||
[]scopes.QualifiedName{{Scope: "/b", Name: "2"}},
|
||||
scopedNames(collectScopedStream(t, svc.Resources(ctx, start, end))),
|
||||
)
|
||||
})
|
||||
|
||||
t.Run("ListResourcesWithFilter pagination", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
svc := newScopedOnlyServiceForTest(t)
|
||||
ctx := t.Context()
|
||||
want := []scopes.QualifiedName{
|
||||
{Scope: "/a", Name: "1"},
|
||||
{Scope: "/b", Name: "2"},
|
||||
{Scope: "/c", Name: "3"},
|
||||
}
|
||||
for _, qn := range want {
|
||||
_, err := svc.CreateResource(ctx, newScopedTestResource(qn))
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// Single full page.
|
||||
got, next, err := svc.ListResourcesWithFilter(ctx, 100, "", func(*testResource) bool { return true })
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, next)
|
||||
require.Equal(t, want, scopedNames(got))
|
||||
|
||||
// One item per page.
|
||||
var paged []*testResource
|
||||
token := ""
|
||||
for {
|
||||
page, nextToken, err := svc.ListResourcesWithFilter(ctx, 1, token, func(*testResource) bool { return true })
|
||||
require.NoError(t, err)
|
||||
if nextToken != "" {
|
||||
require.True(t, scopes.IsScopedResourceCursor(nextToken))
|
||||
}
|
||||
paged = append(paged, page...)
|
||||
if nextToken == "" {
|
||||
break
|
||||
}
|
||||
token = nextToken
|
||||
}
|
||||
require.Equal(t, want, scopedNames(paged))
|
||||
|
||||
// Default page size.
|
||||
got, _, err = svc.ListResourcesWithFilter(ctx, 0, "", func(*testResource) bool { return true })
|
||||
require.NoError(t, err)
|
||||
require.Len(t, got, 3)
|
||||
|
||||
// A scoped-only service must reject non-empty unscoped tokens. Such a
|
||||
// token could only come from a different service mode and would otherwise
|
||||
// be silently interpreted as a scoped service relative key.
|
||||
_, _, err = svc.ListResourcesWithFilter(ctx, 100, "legacy", func(*testResource) bool { return true })
|
||||
require.True(t, trace.IsBadParameter(err), "expected bad parameter, got %v", err)
|
||||
|
||||
require.NoError(t, svc.DeleteAllResources(ctx))
|
||||
require.Empty(t, collectScopedStream(t, svc.Resources(ctx, "", "")))
|
||||
})
|
||||
}
|
||||
|
||||
@@ -45,6 +45,7 @@ type ScopedResourceMetadata interface {
|
||||
// wrapper transparently routes reads and writes to the correct range.
|
||||
func NewScopeAwareServiceWrapper[T ScopedResourceMetadata](cfg ScopeAwareServiceWrapperConfig[T]) (*ScopeAwareServiceWrapper[T], error) {
|
||||
serviceConfig := &ScopeAwareServiceConfig[scopedResourceMetadataAdapter[T]]{
|
||||
ScopedOnly: cfg.ScopedOnly,
|
||||
Backend: cfg.Backend,
|
||||
ResourceKind: cfg.ResourceKind,
|
||||
PageLimit: cfg.PageLimit,
|
||||
@@ -77,6 +78,9 @@ func NewScopeAwareServiceWrapper[T ScopedResourceMetadata](cfg ScopeAwareService
|
||||
// ScopeAwareServiceWrapper. It mirrors ScopeAwareServiceConfig but operates on
|
||||
// the bare RFD 153 resource type T rather than an adapter.
|
||||
type ScopeAwareServiceWrapperConfig[T ScopedResourceMetadata] struct {
|
||||
// ScopedOnly indicates that the service will only operate on scoped resources.
|
||||
// The unscoped fallback path will be ignored in all cases.
|
||||
ScopedOnly bool
|
||||
// Backend used to persist the resource.
|
||||
Backend backend.Backend
|
||||
// ResourceKind is the friendly name of the resource.
|
||||
|
||||
Reference in New Issue
Block a user