From 5dde0ae2cc3c22be6e88f9c8a4c2389ee692d2e2 Mon Sep 17 00:00:00 2001 From: Nic Klaassen Date: Thu, 2 Jul 2026 13:56:12 -0700 Subject: [PATCH] foo added to the cache --- lib/auth/accesspoint/accesspoint.go | 2 + lib/auth/auth.go | 8 + lib/auth/authclient/api.go | 3 + lib/auth/authtest/authtest.go | 1 + lib/auth/grpcserver.go | 8 +- lib/auth/init.go | 3 + lib/auth/services.go | 1 + lib/cache/cache.go | 4 + lib/cache/cache_test.go | 7 + lib/cache/collections.go | 10 ++ lib/cache/foo.go | 110 ++++++++++++ lib/cache/foo_test.go | 175 ++++++++++++++++++++ lib/cache/inventory/inventory_cache_test.go | 4 + lib/service/service.go | 2 + lib/services/foo.go | 74 +++++++++ lib/services/local/events.go | 78 +++++++++ lib/services/local/foo.go | 8 + 17 files changed, 492 insertions(+), 6 deletions(-) create mode 100644 lib/cache/foo.go create mode 100644 lib/cache/foo_test.go create mode 100644 lib/services/foo.go diff --git a/lib/auth/accesspoint/accesspoint.go b/lib/auth/accesspoint/accesspoint.go index 6f2feadc358..be164173513 100644 --- a/lib/auth/accesspoint/accesspoint.go +++ b/lib/auth/accesspoint/accesspoint.go @@ -122,6 +122,7 @@ type Config struct { AppAuthConfig services.AppAuthConfigReader Summarizer services.Summarizer SubCAService services.SubCAServiceGetter + FooUpstream services.FooUpstream } func (c *Config) CheckAndSetDefaults() error { @@ -215,6 +216,7 @@ func NewCache(cfg Config) (*cache.Cache, error) { AppAuthConfig: cfg.AppAuthConfig, Summarizer: cfg.Summarizer, SubCAService: cfg.SubCAService, + FooUpstream: cfg.FooUpstream, } return cache.New(cfg.Setup(cacheCfg)) diff --git a/lib/auth/auth.go b/lib/auth/auth.go index 15e145b241b..f42b6a01d10 100644 --- a/lib/auth/auth.go +++ b/lib/auth/auth.go @@ -686,6 +686,13 @@ func NewServer(cfg *InitConfig, opts ...ServerOption) (as *Server, err error) { } } + if cfg.FooService == nil { + cfg.FooService, err = local.NewFooService(cfg.Backend) + if err != nil { + return nil, trace.Wrap(err, "creating FooService") + } + } + services := &Services{ TrustInternal: cfg.Trust, PresenceInternal: cfg.Presence, @@ -753,6 +760,7 @@ func NewServer(cfg *InitConfig, opts ...ServerOption) (as *Server, err error) { BeamsConfigService: cfg.BeamsConfigService, SubCAService: cfg.SubCAService, EnrollPairing: cfg.EnrollPairing, + FooService: cfg.FooService, } if cfg.FakePasswordHash == nil { diff --git a/lib/auth/authclient/api.go b/lib/auth/authclient/api.go index 1ea3824264c..548b07ca466 100644 --- a/lib/auth/authclient/api.go +++ b/lib/auth/authclient/api.go @@ -1587,6 +1587,9 @@ type Cache interface { // SubCAServiceGetter reads CertAuthorityOverride resources. services.SubCAServiceGetter + + // FooReader reads foo resources. + services.FooReader } type NodeWrapper struct { diff --git a/lib/auth/authtest/authtest.go b/lib/auth/authtest/authtest.go index 1bba41bfb22..4e64f77dad2 100644 --- a/lib/auth/authtest/authtest.go +++ b/lib/auth/authtest/authtest.go @@ -658,6 +658,7 @@ func InitAuthCache(p AuthCacheParams) error { StaticScopedToken: p.AuthServer.Services.ClusterConfigurationInternal, Summarizer: p.AuthServer.Services.Summarizer, SubCAService: p.AuthServer.Services.SubCAService, + FooUpstream: p.AuthServer.Services.FooService, }) if err != nil { return trace.Wrap(err) diff --git a/lib/auth/grpcserver.go b/lib/auth/grpcserver.go index 4619d74f102..3cb06f87ebf 100644 --- a/lib/auth/grpcserver.go +++ b/lib/auth/grpcserver.go @@ -6288,14 +6288,10 @@ func NewGRPCServer(cfg GRPCServerConfig) (*GRPCServer, error) { } scopedjoiningv1.RegisterScopedJoiningServiceServer(server, scopedJoining) - fooStorageService, err := local.NewFooService(cfg.AuthServer.bk) - if err != nil { - return nil, trace.Wrap(err, "creating foo service") - } fooService := foo.NewService(&foo.Config{ ScopedAuthorizer: cfg.ScopedAuthorizer, - Reader: fooStorageService, - Writer: fooStorageService, + Reader: cfg.AuthServer.Cache, + Writer: cfg.AuthServer.FooService, }) foov1.RegisterFooServiceServer(server, fooService) diff --git a/lib/auth/init.go b/lib/auth/init.go index 336088ca103..8e2a81d5ff8 100644 --- a/lib/auth/init.go +++ b/lib/auth/init.go @@ -482,6 +482,9 @@ type InitConfig struct { // EnrollPairing manages mobile device enrollment pairings. EnrollPairing services.EnrollPairing + + // FooService is a service for interacting with Foo resources. + FooService services.FooService } // Init instantiates and configures an instance of AuthServer diff --git a/lib/auth/services.go b/lib/auth/services.go index e0b9d304ebe..603030d3303 100644 --- a/lib/auth/services.go +++ b/lib/auth/services.go @@ -102,6 +102,7 @@ type Services struct { services.BeamsConfigService services.SubCAService services.EnrollPairing + services.FooService } // MFAService defines the interface for managing MFA resources in the backend. diff --git a/lib/cache/cache.go b/lib/cache/cache.go index 2d83294d0a2..238be93ee40 100644 --- a/lib/cache/cache.go +++ b/lib/cache/cache.go @@ -47,6 +47,7 @@ import ( "github.com/gravitational/teleport/lib/backend" "github.com/gravitational/teleport/lib/backend/backendmetrics" "github.com/gravitational/teleport/lib/defaults" + "github.com/gravitational/teleport/lib/foos" "github.com/gravitational/teleport/lib/observability/metrics" "github.com/gravitational/teleport/lib/observability/tracing" scopedaccess "github.com/gravitational/teleport/lib/scopes/access" @@ -228,6 +229,7 @@ func ForAuth(cfg Config) Config { {Kind: types.KindClassifier}, {Kind: types.KindRetrievalModel}, {Kind: types.KindValidatedMFAChallenge}, + {Kind: foos.Kind}, } cfg.QueueSize = defaults.AuthQueueSize // We don't want to enable partial health for auth cache because auth uses an event stream @@ -849,6 +851,8 @@ type Config struct { Summarizer services.Summarizer // SubCAService reads CertAuthorityOverride resources. SubCAService services.SubCAServiceGetter + // FooUpstream reads foos. + FooUpstream services.FooUpstream } // CheckAndSetDefaults checks parameters and sets default values diff --git a/lib/cache/cache_test.go b/lib/cache/cache_test.go index b5892eede45..43958e027ed 100644 --- a/lib/cache/cache_test.go +++ b/lib/cache/cache_test.go @@ -187,6 +187,7 @@ type testPack struct { appAuthConfigs *local.AppAuthConfigService summarizer *local.SummarizerService subCA *local.SubCAService + foos *local.FooService } // resourceOps contains helpers to modify the state of either types.Resource or types.Resource153 which @@ -563,6 +564,11 @@ func newPackWithoutCache(dir string, opts ...packOption) (*testPack, error) { return nil, trace.Wrap(err) } + p.foos, err = local.NewFooService(p.backend) + if err != nil { + return nil, trace.Wrap(err) + } + return p, nil } @@ -632,6 +638,7 @@ func newPack(t testing.TB, setupConfig func(c Config) Config, opts ...packOption StaticScopedToken: p.clusterConfigS, Summarizer: p.summarizer, SubCAService: p.subCA, + FooUpstream: p.foos, })) if err != nil { return nil, trace.Wrap(err) diff --git a/lib/cache/collections.go b/lib/cache/collections.go index 9887b776d33..00014e3529c 100644 --- a/lib/cache/collections.go +++ b/lib/cache/collections.go @@ -29,6 +29,7 @@ import ( clusterconfigv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/clusterconfig/v1" crownjewelv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/crownjewel/v1" dbobjectv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/dbobject/v1" + foov1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/foo/v1" healthcheckconfigv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/healthcheckconfig/v1" identitycenterv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/identitycenter/v1" kubewaitingcontainerv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1" @@ -49,6 +50,7 @@ import ( "github.com/gravitational/teleport/api/types/discoveryconfig" "github.com/gravitational/teleport/api/types/secreports" "github.com/gravitational/teleport/api/types/userloginstate" + "github.com/gravitational/teleport/lib/foos" scopedaccess "github.com/gravitational/teleport/lib/scopes/access" ) @@ -161,6 +163,7 @@ type collections struct { classifiers *collection[*summarizerv1.Classifier, classifierIndex] retrievalModels *collection[*summarizerv1.RetrievalModel, retrievalModelIndex] certAuthorityOverrides *collection[*subcav1.CertAuthorityOverride, certAuthorityOverrideIndex] + foos *collection[*foov1.Foo, fooIndex] } // isKnownUncollectedKind is true if a resource kind is not stored in @@ -871,6 +874,13 @@ func setupCollections(c Config) (*collections, error) { } out.certAuthorityOverrides = collect out.byKind[resourceKind] = out.certAuthorityOverrides + case foos.Kind: + collect, err := newFooCollection(c.FooUpstream, watch) + if err != nil { + return nil, trace.Wrap(err) + } + out.foos = collect + out.byKind[resourceKind] = out.foos default: if _, ok := out.byKind[resourceKind]; !ok { return nil, trace.BadParameter("resource %q is not supported", watch.Kind) diff --git a/lib/cache/foo.go b/lib/cache/foo.go new file mode 100644 index 00000000000..a8fad91fbc9 --- /dev/null +++ b/lib/cache/foo.go @@ -0,0 +1,110 @@ +package cache + +import ( + "context" + "iter" + + "github.com/gravitational/trace" + "google.golang.org/protobuf/proto" + + foov1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/foo/v1" + "github.com/gravitational/teleport/api/types" + "github.com/gravitational/teleport/api/utils/clientutils" + "github.com/gravitational/teleport/lib/foos" + "github.com/gravitational/teleport/lib/itertools/stream" + "github.com/gravitational/teleport/lib/scopes" + "github.com/gravitational/teleport/lib/services" +) + +type fooIndex string + +const ( + fooNameIndex fooIndex = "name" +) + +func fooNameIndexKey(foo *foov1.Foo) string { + cursor, err := foos.Cursor(foo) + if err != nil { + // TODO: figure out how to handle fallible index key creation. + panic(err) + } + return cursor +} + +func newFooCollection(upstream services.FooUpstream, w types.WatchKind) (*collection[*foov1.Foo, fooIndex], error) { + if upstream == nil { + return nil, trace.BadParameter("missing parameter FooUpstream") + } + + return &collection[*foov1.Foo, fooIndex]{ + store: newStore( + foos.Kind, + proto.CloneOf[*foov1.Foo], + map[fooIndex]func(*foov1.Foo) string{ + // sorted by name + fooNameIndex: fooNameIndexKey, + }), + fetcher: func(ctx context.Context, loadSecrets bool) ([]*foov1.Foo, error) { + return stream.Collect(clientutils.Resources(ctx, func(ctx context.Context, pageSize int, pageToken string) ([]*foov1.Foo, string, error) { + return upstream.ListFoos(ctx, foov1.ListFoosRequest_builder{ + PageSize: int32(pageSize), + PageToken: pageToken, + // TODO: propagate filter from WatchKind. + ScopeFilter: nil, + }.Build()) + })) + }, + watch: w, + }, nil +} + +func (c *Cache) GetFoo(ctx context.Context, req *foov1.GetFooRequest) (*foov1.Foo, error) { + ctx, span := c.Tracer.Start(ctx, "cache/GetFoo") + defer span.End() + + fooCursor, err := scopes.MakeResourceCursor(req.GetScope(), req.GetName()) + if err != nil { + return nil, trace.Wrap(err) + } + + getter := genericGetter[*foov1.Foo, fooIndex]{ + cache: c, + collection: c.collections.foos, + index: fooNameIndex, + upstreamGet: func(ctx context.Context, _ string) (*foov1.Foo, error) { + return c.FooUpstream.GetFoo(ctx, req) + }, + } + + out, err := getter.get(ctx, fooCursor) + return out, trace.Wrap(err) +} + +func (c *Cache) RangeFoos(ctx context.Context, req *foov1.ListFoosRequest, startKey, endKey string) iter.Seq2[*foov1.Foo, error] { + ctx, span := c.Tracer.Start(ctx, "cache/RangeFoos") + defer span.End() + + scopeFilter := req.GetScopeFilter() + if err := scopes.ValidateFilter(scopeFilter); err != nil { + return stream.Fail[*foov1.Foo](trace.Wrap(err)) + } + + lister := genericLister[*foov1.Foo, fooIndex]{ + cache: c, + collection: c.collections.foos, + index: fooNameIndex, + upstreamList: func(ctx context.Context, pageSize int, pageToken string) ([]*foov1.Foo, string, error) { + return c.FooUpstream.ListFoos(ctx, foov1.ListFoosRequest_builder{ + PageSize: int32(pageSize), + PageToken: pageToken, + ScopeFilter: scopeFilter, + }.Build()) + }, + filter: func(foo *foov1.Foo) bool { + return scopes.MatchScope(scopeFilter, foo.GetScope()) + }, + nextToken: fooNameIndexKey, + } + + return lister.Range(ctx, startKey, endKey) +} diff --git a/lib/cache/foo_test.go b/lib/cache/foo_test.go new file mode 100644 index 00000000000..3be8644987c --- /dev/null +++ b/lib/cache/foo_test.go @@ -0,0 +1,175 @@ +// Teleport +// Copyright (C) 2026 Gravitational, Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cache + +import ( + "context" + "iter" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + "github.com/gravitational/trace" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/testing/protocmp" + + foov1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/foo/v1" + headerv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/header/v1" + scopesv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/scopes/v1" + "github.com/gravitational/teleport/api/types" + "github.com/gravitational/teleport/lib/foos" + "github.com/gravitational/teleport/lib/itertools/stream" +) + +func TestFoos(t *testing.T) { + t.Parallel() + + p := newTestPack(t, ForAuth) + t.Cleanup(p.Close) + + ctx := t.Context() + unscoped := newFoo("foo-1", "", "unscoped") + scoped := newFoo("foo-1", "/security", "scoped") + + _, err := p.foos.CreateFoo(ctx, unscoped) + require.NoError(t, err) + _, err = p.foos.CreateFoo(ctx, scoped) + require.NoError(t, err) + + cmpOpts := []cmp.Option{ + protocmp.IgnoreFields(&headerv1.Metadata{}, "revision"), + protocmp.Transform(), + cmpopts.EquateEmpty(), + } + + assertCacheFoos := func(expected []*foov1.Foo) { + require.EventuallyWithT(t, func(t *assert.CollectT) { + got, err := stream.Collect(p.cache.RangeFoos(ctx, &foov1.ListFoosRequest{}, "", "")) + assert.NoError(t, err) + assert.Empty(t, cmp.Diff(expected, got, cmpOpts...)) + }, 2*time.Second, 10*time.Millisecond) + } + + assertCacheFoos([]*foov1.Foo{unscoped, scoped}) + + got, err := p.cache.GetFoo(ctx, foov1.GetFooRequest_builder{Name: "foo-1"}.Build()) + require.NoError(t, err) + require.Empty(t, cmp.Diff(unscoped, got, cmpOpts...)) + + got, err = p.cache.GetFoo(ctx, foov1.GetFooRequest_builder{Name: "foo-1", Scope: "/security"}.Build()) + require.NoError(t, err) + require.Empty(t, cmp.Diff(scoped, got, cmpOpts...)) + + scopedFoos, err := stream.Collect(p.cache.RangeFoos(ctx, foov1.ListFoosRequest_builder{ + ScopeFilter: scopesv1.Filter_builder{Scope: "/security", Mode: scopesv1.Mode_MODE_EXACT}.Build(), + }.Build(), "", "")) + require.NoError(t, err) + require.Empty(t, cmp.Diff([]*foov1.Foo{scoped}, scopedFoos, cmpOpts...)) + + require.NoError(t, p.foos.DeleteFoo(ctx, foov1.DeleteFooRequest_builder{Name: "foo-1", Scope: "/security"}.Build())) + assertCacheFoos([]*foov1.Foo{unscoped}) + + _, err = p.cache.GetFoo(ctx, foov1.GetFooRequest_builder{Name: "foo-1", Scope: "/security"}.Build()) + require.True(t, trace.IsNotFound(err), "expected NotFound after delete, got %v", err) +} + +func TestFoos153(t *testing.T) { + t.Parallel() + + p := newTestPack(t, ForAuth) + t.Cleanup(p.Close) + + testResources153(t, p, testFuncs[*foov1.Foo]{ + newResource: func(name string) (*foov1.Foo, error) { + return newFoo(name, "", name), nil + }, + create: func(ctx context.Context, foo *foov1.Foo) error { + _, err := p.foos.CreateFoo(ctx, foo) + return err + }, + list: func(ctx context.Context, pageSize int, pageToken string) ([]*foov1.Foo, string, error) { + return p.foos.ListFoos(ctx, foov1.ListFoosRequest_builder{ + PageSize: int32(pageSize), + PageToken: pageToken, + }.Build()) + }, + cacheGet: func(ctx context.Context, name string) (*foov1.Foo, error) { + return p.cache.GetFoo(ctx, foov1.GetFooRequest_builder{ + Name: name, + }.Build()) + }, + cacheList: func(ctx context.Context, pageSize int, pageToken string) ([]*foov1.Foo, string, error) { + var out []*foov1.Foo + for foo, err := range p.cache.RangeFoos(ctx, nil, pageToken, "") { + if err != nil { + return nil, "", trace.Wrap(err) + } + if len(out) == pageSize { + nextToken, err := foos.Cursor(foo) + if err != nil { + return nil, "", trace.Wrap(err) + } + return out, nextToken, nil + } + out = append(out, foo) + } + return out, "", nil + }, + cacheRange: func(ctx context.Context, startKey, endKey string) iter.Seq2[*foov1.Foo, error] { + return p.cache.RangeFoos(ctx, nil, startKey, endKey) + }, + update: func(ctx context.Context, foo *foov1.Foo) error { + _, err := p.foos.UpdateFoo(ctx, foo) + return err + }, + delete: func(ctx context.Context, name string) error { + return p.foos.DeleteFoo(ctx, foov1.DeleteFooRequest_builder{ + Name: name, + }.Build()) + }, + deleteAll: func(ctx context.Context) error { + for foo, err := range p.foos.RangeFoos(ctx, nil, "", "") { + if err != nil { + return trace.Wrap(err) + } + if err := p.foos.DeleteFoo(ctx, foov1.DeleteFooRequest_builder{ + Scope: foo.GetScope(), + Name: foo.GetMetadata().GetName(), + }.Build()); err != nil { + return trace.Wrap(err) + } + } + return nil + }, + }) +} + +func newFoo(name, scope, value string) *foov1.Foo { + return foov1.Foo_builder{ + Kind: foos.Kind, + Version: types.V1, + Metadata: headerv1.Metadata_builder{ + Name: name, + }.Build(), + Scope: scope, + Spec: foov1.FooSpec_builder{ + Value: value, + }.Build(), + }.Build() +} diff --git a/lib/cache/inventory/inventory_cache_test.go b/lib/cache/inventory/inventory_cache_test.go index 5bbeb701291..0511e58544e 100644 --- a/lib/cache/inventory/inventory_cache_test.go +++ b/lib/cache/inventory/inventory_cache_test.go @@ -232,6 +232,9 @@ func setupTestCache(t *testing.T, setupConfig cache.SetupConfigFn) (*testCache, }) require.NoError(t, err) + fooService, err := local.NewFooService(bkWrapper) + require.NoError(t, err) + c, err := cache.New(setupConfig(cache.Config{ Context: ctx, Events: eventsS, @@ -288,6 +291,7 @@ func setupTestCache(t *testing.T, setupConfig cache.SetupConfigFn) (*testCache, EventsC: eventsC, Summarizer: summaries, SubCAService: subCA, + FooUpstream: fooService, })) require.NoError(t, err) diff --git a/lib/service/service.go b/lib/service/service.go index 7c78eb245b4..14894fff839 100644 --- a/lib/service/service.go +++ b/lib/service/service.go @@ -3190,6 +3190,7 @@ func (process *TeleportProcess) newAccessCacheForServices(cfg accesspoint.Config cfg.AppAuthConfig = services.AppAuthConfig cfg.Summarizer = services.Summarizer cfg.SubCAService = services.SubCAService + cfg.FooUpstream = services.FooService return accesspoint.NewCache(cfg) } @@ -3241,6 +3242,7 @@ func (process *TeleportProcess) newAccessCacheForClient(cfg accesspoint.Config, cfg.HealthCheckConfig = client cfg.AppAuthConfig = client cfg.SubCAService = client + cfg.FooUpstream = services.NewFooClientAdapter(client.FooClient()) return accesspoint.NewCache(cfg) } diff --git a/lib/services/foo.go b/lib/services/foo.go new file mode 100644 index 00000000000..86b3b44f127 --- /dev/null +++ b/lib/services/foo.go @@ -0,0 +1,74 @@ +package services + +import ( + "context" + "iter" + + "github.com/gravitational/trace" + + foov1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/foo/v1" +) + +// FooService is a service for interacting with Foo resources, implemented only +// by the backend storage service. +// +// It should be included in [lib/auth.InitConfig] and embedded in [lib/auth.Services]. +type FooService interface { + CreateFoo(ctx context.Context, foo *foov1.Foo) (*foov1.Foo, error) + UpdateFoo(ctx context.Context, foo *foov1.Foo) (*foov1.Foo, error) + UpsertFoo(ctx context.Context, foo *foov1.Foo) (*foov1.Foo, error) + DeleteFoo(ctx context.Context, req *foov1.DeleteFooRequest) error + GetFoo(ctx context.Context, req *foov1.GetFooRequest) (*foov1.Foo, error) + ListFoos(ctx context.Context, req *foov1.ListFoosRequest) ([]*foov1.Foo, string, error) + RangeFoos(ctx context.Context, req *foov1.ListFoosRequest, startKey, endKey string) iter.Seq2[*foov1.Foo, error] +} + +// FooReader is a read interface for reading Foos from a backend storage +// service _or_ a cache. +// +// It should be embedded in [lib/auth/authclient.Cache] and consumed by the +// gRPC API layer. +type FooReader interface { + GetFoo(ctx context.Context, req *foov1.GetFooRequest) (*foov1.Foo, error) + RangeFoos(ctx context.Context, req *foov1.ListFoosRequest, startKey, endKey string) iter.Seq2[*foov1.Foo, error] +} + +// FooUpstream is a read interface for reading Foos from a backend storage +// service _or_ an API client. +// +// It should be included in [lib/cache.Config] to be consumed by the cache. +type FooUpstream interface { + GetFoo(ctx context.Context, req *foov1.GetFooRequest) (*foov1.Foo, error) + ListFoos(ctx context.Context, req *foov1.ListFoosRequest) ([]*foov1.Foo, string, error) +} + +// fooClientAdapter adapts a plain gRPC client to implement FooUpstream to be +// consumed by the cache. +// +// It is only necessary if the resource needs to be cached on proxies or agents. +type fooClientAdapter struct { + grpcClient foov1.FooServiceClient +} + +// NewFooClientAdapter adapts a plain gRPC client to implement FooUpstream. +func NewFooClientAdapter(grpcClient foov1.FooServiceClient) FooUpstream { + return fooClientAdapter{ + grpcClient: grpcClient, + } +} + +func (c fooClientAdapter) GetFoo(ctx context.Context, req *foov1.GetFooRequest) (*foov1.Foo, error) { + resp, err := c.grpcClient.GetFoo(ctx, req) + if err != nil { + return nil, trace.Wrap(err) + } + return resp.GetFoo(), nil +} + +func (c fooClientAdapter) ListFoos(ctx context.Context, req *foov1.ListFoosRequest) ([]*foov1.Foo, string, error) { + resp, err := c.grpcClient.ListFoos(ctx, req) + if err != nil { + return nil, "", trace.Wrap(err) + } + return resp.GetFoos(), resp.GetNextPageToken(), nil +} diff --git a/lib/services/local/events.go b/lib/services/local/events.go index 33aeae1adb8..adb8ee7f58f 100644 --- a/lib/services/local/events.go +++ b/lib/services/local/events.go @@ -33,6 +33,7 @@ import ( accessgraphsecretsv1pb "github.com/gravitational/teleport/api/gen/proto/go/teleport/accessgraph/v1" "github.com/gravitational/teleport/api/gen/proto/go/teleport/autoupdate/v1" dbobjectv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/dbobject/v1" + foov1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/foo/v1" headerv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/header/v1" kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1" machineidv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/machineid/v1" @@ -44,6 +45,8 @@ import ( "github.com/gravitational/teleport/lib/backend" "github.com/gravitational/teleport/lib/defaults" "github.com/gravitational/teleport/lib/devicetrust" + "github.com/gravitational/teleport/lib/foos" + "github.com/gravitational/teleport/lib/scopes" scopedaccess "github.com/gravitational/teleport/lib/scopes/access" "github.com/gravitational/teleport/lib/services" "github.com/gravitational/teleport/lib/services/local/generic" @@ -301,6 +304,8 @@ func (e *EventsService) NewWatcher(ctx context.Context, watch types.Watch) (type parser = newCertAuthorityOverrideParser() case types.KindValidatedMFAChallenge: parser = newValidatedMFAChallengeParser() + case foos.Kind: + parser = newFooParser() default: if watch.AllowPartialSuccess { continue @@ -1087,6 +1092,79 @@ func (p *roleParser) parse(event backend.Event) (types.Resource, error) { } } +func newFooParser() *fooParser { + return &fooParser{ + baseParser: newBaseParser( + fooUnscopedWatchPrefix(), + fooScopedWatchPrefix(), + ), + } +} + +type fooParser struct { + baseParser +} + +func (p *fooParser) parse(event backend.Event) (types.Resource, error) { + switch event.Type { + case types.OpDelete: + sqn, err := fooNameFromKey(event.Item.Key) + if err != nil { + return nil, trace.Wrap(err) + } + foo := foov1.Foo_builder{ + Kind: foos.Kind, + Version: types.V1, + Metadata: headerv1.Metadata_builder{ + Name: sqn.Name, + }.Build(), + Scope: sqn.Scope, + }.Build() + return types.Resource153ToLegacy(foo), nil + case types.OpPut: + foo, err := services.UnmarshalProtoResource[*foov1.Foo]( + event.Item.Value, + services.WithExpires(event.Item.Expires), + services.WithRevision(event.Item.Revision), + ) + if err != nil { + return nil, trace.Wrap(err) + } + return types.Resource153ToLegacy(foo), nil + default: + return nil, trace.BadParameter("event %v is not supported", event.Type) + } +} + +func fooNameFromKey(key backend.Key) (scopes.QualifiedName, error) { + switch { + case key.HasPrefix(fooScopedWatchPrefix()): + components := key.TrimPrefix(fooScopedWatchPrefix()).Components() + if len(components) != 2 { + return scopes.QualifiedName{}, trace.NotFound("failed parsing %v", key.String()) + } + encodedScope, name := components[0], components[1] + scope, err := scopes.DecodeFromKey(encodedScope) + if err != nil { + return scopes.QualifiedName{}, trace.Wrap(err) + } + return scopes.QualifiedName{ + Scope: scope, + Name: name, + }, nil + case key.HasPrefix(fooUnscopedWatchPrefix()): + components := key.TrimPrefix(fooUnscopedWatchPrefix()).Components() + if len(components) != 1 { + return scopes.QualifiedName{}, trace.NotFound("failed parsing %v", key.String()) + } + return scopes.QualifiedName{ + Name: components[0], + }, nil + default: + return scopes.QualifiedName{}, trace.NotFound("failed parsing %v", key.String()) + } +} + func newScopedRoleParser() *scopedRoleParser { return &scopedRoleParser{ baseParser: newBaseParser(scopedRoleWatchPrefix()), diff --git a/lib/services/local/foo.go b/lib/services/local/foo.go index 981f1b07c83..10e576298d2 100644 --- a/lib/services/local/foo.go +++ b/lib/services/local/foo.go @@ -31,6 +31,14 @@ import ( "github.com/gravitational/teleport/lib/services/local/generic" ) +func fooUnscopedWatchPrefix() backend.Key { + return backend.ExactKey("foo") +} + +func fooScopedWatchPrefix() backend.Key { + return backend.ExactKey("scoped", "foo") +} + // FooService is a storage service for Foos. type FooService struct { service *generic.ScopeAwareServiceWrapper[*foov1.Foo]