[v18] Database authorization resource consumption enhancements

lib/reversetunnel/localsite.go

@@ -191,6 +191,11 @@ func (s *localSite) GitServerWatcher() (*services.GenericWatcher[types.Server, r
 	return s.srv.GitServerWatcher, nil
 }
 
+// DatabaseServerWatcher returns a Database server watcher for this cluster.
+func (s *localSite) DatabaseServerWatcher() (*services.GenericWatcher[types.DatabaseServer, readonly.DatabaseServer], error) {
+	return s.srv.DatabaseServerWatcher, nil
+}
+
 // GetClient returns a client to the full Auth Server API.
 func (s *localSite) GetClient() (authclient.ClientI, error) {
 	return s.client, nil

lib/reversetunnel/peer.go

@@ -114,8 +114,13 @@ func (p *clusterPeers) GetClient() (authclient.ClientI, error) {
 	return peer.GetClient()
 }
 
-func (p *clusterPeers) String() string {
-	return fmt.Sprintf("clusterPeer(%v)", p.clusterName)
+// DatabaseServerWatcher returns a watcher for database servers in the leaf cluster.
+func (p *clusterPeers) DatabaseServerWatcher() (*services.GenericWatcher[types.DatabaseServer, readonly.DatabaseServer], error) {
+	peer, err := p.pickPeer()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return peer.DatabaseServerWatcher()
 }
 
 func (p *clusterPeers) GetStatus() string {
@@ -216,6 +221,10 @@ func (s *clusterPeer) GitServerWatcher() (*services.GenericWatcher[types.Server,
 	return nil, s.discoveryError("unable to fetch git server watcher for leaf cluster")
 }
 
+func (s *clusterPeer) DatabaseServerWatcher() (*services.GenericWatcher[types.DatabaseServer, readonly.DatabaseServer], error) {
+	return nil, s.discoveryError("unable to fetch database server watcher for leaf cluster")
+}
+
 func (s *clusterPeer) GetClient() (authclient.ClientI, error) {
 	return nil, s.discoveryError("unable to fetch auth client for leaf cluster")
 }

lib/reversetunnel/remotesite.go

@@ -89,6 +89,9 @@ type remoteSite struct {
 	// nodeWatcher provides access the node set for the remote site
 	nodeWatcher *services.GenericWatcher[types.Server, readonly.Server]
 
+	// databaseServerWatcher is a database server watcher.
+	databaseServerWatcher *services.GenericWatcher[types.DatabaseServer, readonly.DatabaseServer]
+
 	// remoteCA is the last remote certificate authority recorded by the client.
 	// It is used to detect CA rotation status changes. If the rotation
 	// state has been changed, the tunnel will reconnect to re-create the client
@@ -179,8 +182,9 @@ func (s *remoteSite) GetClient() (authclient.ClientI, error) {
 	return s.remoteClient, nil
 }
 
-func (s *remoteSite) String() string {
-	return fmt.Sprintf("remoteSite(%v)", s.domainName)
+// DatabaseServerWatcher returns the Database server watcher for the leaf cluster.
+func (s *remoteSite) DatabaseServerWatcher() (*services.GenericWatcher[types.DatabaseServer, readonly.DatabaseServer], error) {
+	return s.databaseServerWatcher, nil
 }
 
 func (s *remoteSite) connectionCount() int {

lib/reversetunnel/srv.go

@@ -215,6 +215,9 @@ type Config struct {
 	// CertAuthorityWatcher is a cert authority watcher.
 	CertAuthorityWatcher *services.CertAuthorityWatcher
 
+	// DatabaseServerWatcher is a database server watcher.
+	DatabaseServerWatcher *services.GenericWatcher[types.DatabaseServer, readonly.DatabaseServer]
+
 	// CircuitBreakerConfig configures the auth client circuit breaker
 	CircuitBreakerConfig breaker.Config
 
@@ -286,6 +289,9 @@ func (cfg *Config) CheckAndSetDefaults() error {
 	if cfg.CertAuthorityWatcher == nil {
 		return trace.BadParameter("missing parameter CertAuthorityWatcher")
 	}
+	if cfg.DatabaseServerWatcher == nil {
+		return trace.BadParameter("missing parameter DatabaseServerWatcher")
+	}
 	return nil
 }
 
@@ -1254,6 +1260,19 @@ func newRemoteSite(srv *server, domainName string, sconn ssh.Conn) (*remoteSite,
 		return nil, trace.Wrap(err)
 	}
 	remoteSite.nodeWatcher = nodeWatcher
+
+	databaseServerWatcher, err := services.NewDatabaseServerWatcher(closeContext, services.DatabaseServerWatcherConfig{
+		ResourceWatcherConfig: services.ResourceWatcherConfig{
+			Component: srv.Component,
+			Logger:    srv.Logger,
+			Client:    accessPoint,
+		},
+	})
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	remoteSite.databaseServerWatcher = databaseServerWatcher
+
 	// instantiate a cache of host certificates for the forwarding server. the
 	// certificate cache is created in each site (instead of creating it in
 	// reversetunnel.server and passing it along) so that the host certificate

lib/reversetunnelclient/api.go

@@ -124,6 +124,8 @@ type Cluster interface {
 	NodeWatcher() (*services.GenericWatcher[types.Server, readonly.Server], error)
 	// GitServerWatcher returns the Git server watcher for the cluster
 	GitServerWatcher() (*services.GenericWatcher[types.Server, readonly.Server], error)
+	// DatabaseServerWatcher returns the watcher that maintains the database server set for the cluster
+	DatabaseServerWatcher() (*services.GenericWatcher[types.DatabaseServer, readonly.DatabaseServer], error)
 	// GetTunnelsCount returns the amount of active inbound tunnels
 	// from the remote cluster
 	GetTunnelsCount() int

lib/reversetunnelclient/fake.go

@@ -26,7 +26,10 @@ import (
 
 	"github.com/gravitational/trace"
 
+	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/lib/auth/authclient"
+	"github.com/gravitational/teleport/lib/services"
+	"github.com/gravitational/teleport/lib/services/readonly"
 )
 
 // FakeServer is a fake Server implementation used in tests.
@@ -68,17 +71,32 @@ type FakeCluster struct {
 	closedMtx sync.Mutex
 	// closed is set to true after the cluster is being closed.
 	closed bool
+	// databaseServerWatcher is a database server watcher to speed up database server look up.
+	databaseServerWatcher *services.GenericWatcher[types.DatabaseServer, readonly.DatabaseServer]
 }
 
 // NewFakeCluster is a FakeCluster constructor.
 func NewFakeCluster(clusterName string, accessPoint authclient.RemoteProxyAccessPoint) *FakeCluster {
+	databaseServerWatcher, _ := services.NewDatabaseServerWatcher(context.TODO(), services.DatabaseServerWatcherConfig{
+		ResourceWatcherConfig: services.ResourceWatcherConfig{
+			Component: "FakeCluster",
+			Client:    accessPoint,
+		},
+	})
+
 	return &FakeCluster{
-		Name:        clusterName,
-		connCh:      make(chan net.Conn),
-		AccessPoint: accessPoint,
+		Name:                  clusterName,
+		connCh:                make(chan net.Conn),
+		AccessPoint:           accessPoint,
+		databaseServerWatcher: databaseServerWatcher,
 	}
 }
 
+// DatabaseServerWatcher returns the watcher that maintains the database server set for the cluster
+func (s *FakeCluster) DatabaseServerWatcher() (*services.GenericWatcher[types.DatabaseServer, readonly.DatabaseServer], error) {
+	return s.databaseServerWatcher, nil
+}
+
 // CachingAccessPoint returns caching auth server client.
 func (s *FakeCluster) CachingAccessPoint() (authclient.RemoteProxyAccessPoint, error) {
 	return s.AccessPoint, nil

lib/service/service.go

@@ -4925,6 +4925,18 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
 		return trace.Wrap(err)
 	}
 
+	databaseServerWatcher, err := services.NewDatabaseServerWatcher(process.ExitContext(), services.DatabaseServerWatcherConfig{
+		ResourceWatcherConfig: services.ResourceWatcherConfig{
+			Component: teleport.ComponentProxy,
+			Logger:    process.logger.With(teleport.ComponentKey, teleport.ComponentProxy),
+			Client:    accessPoint,
+		},
+		DatabaseServersGetter: accessPoint,
+	})
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
 	serverTLSConfig, err := conn.ServerTLSConfig(cfg.CipherSuites)
 	if err != nil {
 		return trace.Wrap(err)
@@ -5173,35 +5185,35 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
 			reversetunnel.Config{
 				ClientTLSCipherSuites:   process.Config.CipherSuites,
 				GetClientTLSCertificate: conn.ClientGetCertificate,
-
-				Context:               process.ExitContext(),
-				Component:             teleport.Component(teleport.ComponentProxy, process.id),
-				ID:                    conn.HostUUID(),
-				ClusterName:           clusterName,
-				Listener:              rtListener,
-				GetHostSigners:        conn.ServerGetHostSigners,
-				LocalAuthClient:       conn.Client,
-				LocalAccessPoint:      accessPoint,
-				NewCachingAccessPoint: process.newLocalCacheForRemoteProxy,
-				Limiter:               reverseTunnelLimiter,
-				KeyGen:                cfg.Keygen,
-				Ciphers:               cfg.Ciphers,
-				KEXAlgorithms:         cfg.KEXAlgorithms,
-				MACAlgorithms:         cfg.MACAlgorithms,
-				DataDir:               process.Config.DataDir,
-				PollingPeriod:         process.Config.PollingPeriod,
-				FIPS:                  cfg.FIPS,
-				Emitter:               streamEmitter,
-				Logger:                process.logger,
-				LockWatcher:           lockWatcher,
-				PeerClient:            peerClient,
-				NodeWatcher:           nodeWatcher,
-				GitServerWatcher:      gitServerWatcher,
-				CertAuthorityWatcher:  caWatcher,
-				CircuitBreakerConfig:  process.Config.CircuitBreakerConfig,
-				LocalAuthAddresses:    utils.NetAddrsToStrings(process.Config.AuthServerAddresses()),
-				IngressReporter:       ingressReporter,
-				PROXYSigner:           proxySigner,
+				Context:                 process.ExitContext(),
+				Component:               teleport.Component(teleport.ComponentProxy, process.id),
+				ID:                      conn.HostUUID(),
+				ClusterName:             clusterName,
+				Listener:                rtListener,
+				GetHostSigners:          conn.ServerGetHostSigners,
+				LocalAuthClient:         conn.Client,
+				LocalAccessPoint:        accessPoint,
+				NewCachingAccessPoint:   process.newLocalCacheForRemoteProxy,
+				Limiter:                 reverseTunnelLimiter,
+				KeyGen:                  cfg.Keygen,
+				Ciphers:                 cfg.Ciphers,
+				KEXAlgorithms:           cfg.KEXAlgorithms,
+				MACAlgorithms:           cfg.MACAlgorithms,
+				DataDir:                 process.Config.DataDir,
+				PollingPeriod:           process.Config.PollingPeriod,
+				FIPS:                    cfg.FIPS,
+				Emitter:                 streamEmitter,
+				Logger:                  process.logger,
+				LockWatcher:             lockWatcher,
+				PeerClient:              peerClient,
+				NodeWatcher:             nodeWatcher,
+				GitServerWatcher:        gitServerWatcher,
+				DatabaseServerWatcher:   databaseServerWatcher,
+				CertAuthorityWatcher:    caWatcher,
+				CircuitBreakerConfig:    process.Config.CircuitBreakerConfig,
+				LocalAuthAddresses:      utils.NetAddrsToStrings(process.Config.AuthServerAddresses()),
+				IngressReporter:         ingressReporter,
+				PROXYSigner:             proxySigner,
 			})
 		if err != nil {
 			return trace.Wrap(err)

lib/services/presence.go

@@ -48,7 +48,8 @@ type NodesGetter interface {
 
 // DatabaseServersGetter is a service that gets database servers.
 type DatabaseServersGetter interface {
-	GetDatabaseServers(context.Context, string, ...MarshalOption) ([]types.DatabaseServer, error)
+	// GetDatabaseServers returns all registered database proxy servers.
+	GetDatabaseServers(ctx context.Context, namespace string, opts ...MarshalOption) ([]types.DatabaseServer, error)
 }
 
 // AppServersGetter is a service that gets application servers.

lib/services/readonly/readonly.go

@@ -248,6 +248,28 @@ type Application interface {
 
 var _ Application = types.Application(nil)
 
+// DatabaseServer is a read only variant of [types.DatabaseServer]
+type DatabaseServer struct {
+	inner types.DatabaseServer
+}
+
+// GetDatabaseName returns the name of the database this server is proxying.
+func (d DatabaseServer) GetDatabaseName() string {
+	if d.inner == nil {
+		return ""
+	}
+	db := d.inner.GetDatabase()
+	if db == nil {
+		return ""
+	}
+	return db.GetName()
+}
+
+// NewDatabaseServer returns a new read-only DatabaseServer.
+func NewDatabaseServer(server types.DatabaseServer) DatabaseServer {
+	return DatabaseServer{inner: server}
+}
+
 // KubeServer is a read only variant of [types.KubeServer].
 type KubeServer interface {
 	// ResourceWithLabels provides common resource methods.

lib/services/watcher.go

@@ -490,6 +490,60 @@ func NewAppWatcher(ctx context.Context, cfg AppWatcherConfig) (*GenericWatcher[t
 	return w, trace.Wrap(err)
 }
 
+type DatabaseServerWatcherConfig struct {
+	DatabaseServersGetter
+	ResourceWatcherConfig
+}
+
+// CheckAndSetDefaults checks parameters and sets default values.
+func (cfg *DatabaseServerWatcherConfig) CheckAndSetDefaults() error {
+	if err := cfg.ResourceWatcherConfig.CheckAndSetDefaults(); err != nil {
+		return trace.Wrap(err)
+	}
+
+	if cfg.MaxStaleness == 0 {
+		const databaseServerMaxStaleness = time.Minute
+		cfg.MaxStaleness = databaseServerMaxStaleness
+	}
+
+	if cfg.DatabaseServersGetter == nil {
+		getter, ok := cfg.Client.(DatabaseServersGetter)
+		if !ok {
+			return trace.BadParameter("missing parameter DatabaseServersGetter and Client not usable as DatabaseServersGetter")
+		}
+		cfg.DatabaseServersGetter = getter
+	}
+
+	return nil
+}
+
+func NewDatabaseServerWatcher(ctx context.Context, cfg DatabaseServerWatcherConfig) (*GenericWatcher[types.DatabaseServer, readonly.DatabaseServer], error) {
+	if err := cfg.CheckAndSetDefaults(); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	w, err := NewGenericResourceWatcher(ctx, GenericWatcherConfig[types.DatabaseServer, readonly.DatabaseServer]{
+		ResourceWatcherConfig: cfg.ResourceWatcherConfig,
+		ResourceKind:          types.KindDatabaseServer,
+		ResourceKey: func(r types.DatabaseServer) string {
+			// the host ID is guaranteed not to contain "/"
+			return r.GetHostID() + "/" + r.GetName()
+		},
+		DeleteKey: func(r types.Resource) string {
+			// database servers put the host ID in the description in delete events
+			return r.GetMetadata().Description + "/" + r.GetName()
+		},
+		ResourceGetter: func(ctx context.Context) ([]types.DatabaseServer, error) {
+			return cfg.DatabaseServersGetter.GetDatabaseServers(ctx, apidefaults.Namespace)
+		},
+		DisableUpdateBroadcast: true,
+		CloneFunc:              types.DatabaseServer.Copy,
+		ReadOnlyFunc:           readonly.NewDatabaseServer,
+	})
+
+	return w, trace.Wrap(err)
+}
+
 // KubeServerWatcherConfig is an KubeServerWatcher configuration.
 type KubeServerWatcherConfig struct {
 	// KubernetesServerGetter is responsible for fetching kube_server resources.