[v14] Dynamic Discovery Matchers for Databases

lib/auth/api.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/api/types/discoveryconfig"
 	"github.com/gravitational/teleport/lib/events"
 	"github.com/gravitational/teleport/lib/services"
 )
@@ -701,6 +702,9 @@ type ReadDiscoveryAccessPoint interface {
 	GetApps(context.Context) ([]types.Application, error)
 	// GetApp returns the specified application resource.
 	GetApp(ctx context.Context, name string) (types.Application, error)
+
+	// ListDiscoveryConfigs returns a paginated list of Discovery Config resources.
+	ListDiscoveryConfigs(ctx context.Context, pageSize int, nextKey string) ([]*discoveryconfig.DiscoveryConfig, string, error)
 }
 
 // DiscoveryAccessPoint is an API interface implemented by a certificate authority (CA) to be

lib/service/service.go

@@ -2282,17 +2282,28 @@ func (process *TeleportProcess) newLocalCacheForDatabase(clt auth.ClientI, cache
 	return auth.NewDatabaseWrapper(clt, cache), nil
 }
 
+// combinedDiscoveryClient is an auth.Client client with other, specific, services added to it.
+type combinedDiscoveryClient struct {
+	auth.ClientI
+	services.DiscoveryConfigsGetter
+}
+
 // newLocalCacheForDiscovery returns a new instance of access point for a discovery service.
 func (process *TeleportProcess) newLocalCacheForDiscovery(clt auth.ClientI, cacheName []string) (auth.DiscoveryAccessPoint, error) {
+	client := combinedDiscoveryClient{
+		ClientI:                clt,
+		DiscoveryConfigsGetter: clt.DiscoveryConfigClient(),
+	}
+
 	// if caching is disabled, return access point
 	if !process.Config.CachePolicy.Enabled {
-		return clt, nil
+		return client, nil
 	}
 	cache, err := process.NewLocalCache(clt, cache.ForDiscovery, cacheName)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-	return auth.NewDiscoveryWrapper(clt, cache), nil
+	return auth.NewDiscoveryWrapper(client, cache), nil
 }
 
 // newLocalCacheForProxy returns new instance of access point configured for a local proxy.

lib/service/servicecfg/discovery.go

@@ -43,8 +43,10 @@ type DiscoveryConfig struct {
 	PollInterval time.Duration
 }
 
-// IsEmpty validates if the Discovery Service config has no cloud matchers.
+// IsEmpty validates if the Discovery Service config has no matchers and no discovery group.
+// DiscoveryGroup is used to dynamically load Matchers when changing DiscoveryConfig resources.
 func (d DiscoveryConfig) IsEmpty() bool {
 	return len(d.AWSMatchers) == 0 && len(d.AzureMatchers) == 0 &&
-		len(d.GCPMatchers) == 0 && len(d.KubernetesMatchers) == 0
+		len(d.GCPMatchers) == 0 && len(d.KubernetesMatchers) == 0 &&
+		d.DiscoveryGroup == ""
 }

lib/service/servicecfg/discovery_test.go

@@ -0,0 +1,85 @@
+// Copyright 2023 Gravitational, Inc
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package servicecfg
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/gravitational/teleport/api/types"
+)
+
+func TestDiscoveryConfig_IsEmpty(t *testing.T) {
+	for _, tt := range []struct {
+		name     string
+		input    DiscoveryConfig
+		expected bool
+	}{
+		{
+			name: "returns false when discovery group is present",
+			input: DiscoveryConfig{
+				DiscoveryGroup: "my-discovery-group",
+			},
+			expected: false,
+		},
+		{
+			name: "returns false when has at least one aws matcher",
+			input: DiscoveryConfig{
+				AWSMatchers: []types.AWSMatcher{{
+					Types: []string{"ec2"},
+				}},
+			},
+			expected: false,
+		},
+		{
+			name: "returns false when has at least one azure matcher",
+			input: DiscoveryConfig{
+				AzureMatchers: []types.AzureMatcher{{
+					Types: []string{"aks"},
+				}},
+			},
+			expected: false,
+		},
+		{
+			name: "returns false when has at least one gcp matcher",
+			input: DiscoveryConfig{
+				GCPMatchers: []types.GCPMatcher{{
+					Types: []string{"gke"},
+				}},
+			},
+			expected: false,
+		},
+		{
+			name: "returns false when has at least one Kube matcher",
+			input: DiscoveryConfig{
+				KubernetesMatchers: []types.KubernetesMatcher{{
+					Types: []string{"app"},
+				}},
+			},
+			expected: false,
+		},
+		{
+			name:     "returns true when there are no matchers and no discovery group",
+			input:    DiscoveryConfig{},
+			expected: true,
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.input.IsEmpty()
+			require.Equal(t, tt.expected, got)
+		})
+	}
+}

lib/srv/db/watcher.go

@@ -114,16 +114,18 @@ func (s *Server) startCloudWatcher(ctx context.Context) error {
 		return trace.Wrap(err)
 	}
 
+	allFetchers := append(awsFetchers, azureFetchers...)
+	if len(allFetchers) == 0 {
+		s.log.Debugf("Not starting cloud database watcher: %v.", err)
+		return nil
+	}
+
 	watcher, err := discovery.NewWatcher(ctx, discovery.WatcherConfig{
-		Fetchers: append(awsFetchers, azureFetchers...),
-		Log:      logrus.WithField(trace.Component, "watcher:cloud"),
-		Origin:   types.OriginCloud,
+		FetchersFn: discovery.StaticFetchers(allFetchers),
+		Log:        logrus.WithField(trace.Component, "watcher:cloud"),
+		Origin:     types.OriginCloud,
 	})
 	if err != nil {
-		if trace.IsNotFound(err) {
-			s.log.Debugf("Not starting cloud database watcher: %v.", err)
-			return nil
-		}
 		return trace.Wrap(err)
 	}
 	go watcher.Start()

lib/srv/discovery/common/watcher.go

@@ -35,8 +35,8 @@ const (
 
 // WatcherConfig is the common discovery watcher configuration.
 type WatcherConfig struct {
-	// Fetchers holds fetchers used for this watcher.
-	Fetchers []Fetcher
+	// FetchersFn is a function that returns the fetchers used for this watcher.
+	FetchersFn func() []Fetcher
 	// Interval is the interval between fetches.
 	Interval time.Duration
 	// Log is the watcher logger.
@@ -66,7 +66,7 @@ func (c *WatcherConfig) CheckAndSetDefaults() error {
 	if c.Clock == nil {
 		c.Clock = clockwork.NewRealClock()
 	}
-	if len(c.Fetchers) == 0 {
+	if c.FetchersFn == nil {
 		return trace.NotFound("missing fetchers")
 	}
 	if c.Origin == "" {
@@ -122,7 +122,7 @@ func (w *Watcher) fetchAndSend() {
 		group, groupCtx     = errgroup.WithContext(w.ctx)
 	)
 	group.SetLimit(concurrencyLimit)
-	for _, fetcher := range w.cfg.Fetchers {
+	for _, fetcher := range w.cfg.FetchersFn() {
 		lFetcher := fetcher
 
 		group.Go(func() error {
@@ -179,3 +179,11 @@ func (w *Watcher) fetchAndSend() {
 func (w *Watcher) ResourcesC() <-chan types.ResourcesWithLabels {
 	return w.resourcesC
 }
+
+// StaticFetchers converts a list of Fetchers into a function that returns them.
+// Used to convert a static set of Fetchers into a FetchersFn generator.
+func StaticFetchers(fs []Fetcher) func() []Fetcher {
+	return func() []Fetcher {
+		return fs
+	}
+}

lib/srv/discovery/common/watcher_test.go

@@ -60,10 +60,10 @@ func TestWatcher(t *testing.T) {
 
 	clock := clockwork.NewFakeClock()
 	watcher, err := NewWatcher(ctx, WatcherConfig{
-		Fetchers: []Fetcher{appFetcher, noAuthFetcher, dbFetcher},
-		Interval: time.Hour,
-		Clock:    clock,
-		Origin:   types.OriginCloud,
+		FetchersFn: StaticFetchers([]Fetcher{appFetcher, noAuthFetcher, dbFetcher}),
+		Interval:   time.Hour,
+		Clock:      clock,
+		Origin:     types.OriginCloud,
 	})
 	require.NoError(t, err)
 	go watcher.Start()
@@ -77,6 +77,63 @@ func TestWatcher(t *testing.T) {
 	assertFetchResources(t, watcher, wantResources)
 }
 
+func TestWatcherWithDynamicFetchers(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	app1, err := types.NewAppV3(types.Metadata{Name: "app1"}, types.AppSpecV3{Cloud: types.CloudAWS})
+	require.NoError(t, err)
+	app2, err := types.NewAppV3(types.Metadata{Name: "app2"}, types.AppSpecV3{Cloud: types.CloudAWS})
+	require.NoError(t, err)
+
+	appFetcher := &mockFetcher{
+		resources:    types.ResourcesWithLabels{app1, app2},
+		resourceType: types.KindApp,
+		cloud:        types.CloudAWS,
+	}
+
+	noAuthFetcher := &mockFetcher{
+		noAuth:       true,
+		resourceType: types.KindKubeServer,
+		cloud:        types.CloudGCP,
+	}
+
+	// Start with two watchers.
+	fetchers := []Fetcher{appFetcher, noAuthFetcher}
+	fetchersFn := func() []Fetcher {
+		return fetchers
+	}
+
+	clock := clockwork.NewFakeClock()
+	watcher, err := NewWatcher(ctx, WatcherConfig{
+		FetchersFn: fetchersFn,
+		Interval:   time.Hour,
+		Clock:      clock,
+		Origin:     types.OriginCloud,
+	})
+	require.NoError(t, err)
+	go watcher.Start()
+
+	// Watcher should fetch once right away at watcher.Start.
+	assertFetchResources(t, watcher, types.ResourcesWithLabels{app1, app2})
+
+	// Add an extra fetcher during runtime.
+	db, err := types.NewDatabaseV3(types.Metadata{Name: "db"}, types.DatabaseSpecV3{Protocol: "mysql", URI: "db.mysql.database.azure.com:1234"})
+	require.NoError(t, err)
+	dbFetcher := &mockFetcher{
+		resources:    types.ResourcesWithLabels{db},
+		resourceType: types.KindDatabase,
+		cloud:        types.CloudAzure,
+	}
+	fetchers = append(fetchers, dbFetcher)
+
+	// During next iteration, the new fetcher must be used and a 3rd resource must appear.
+	clock.Advance(time.Hour + time.Minute)
+	assertFetchResources(t, watcher, types.ResourcesWithLabels{app1, app2, db})
+}
+
 func assertFetchResources(t *testing.T, watcher *Watcher, wantResources types.ResourcesWithLabels) {
 	select {
 	case fetchResources := <-watcher.ResourcesC():

lib/srv/discovery/database_watcher.go

@@ -31,7 +31,7 @@ import (
 const databaseEventPrefix = "db/"
 
 func (s *Server) startDatabaseWatchers() error {
-	if len(s.databaseFetchers) == 0 {
+	if len(s.databaseFetchers) == 0 && s.dynamicMatcherWatcher == nil {
 		return nil
 	}
 
@@ -60,11 +60,12 @@ func (s *Server) startDatabaseWatchers() error {
 	}
 
 	watcher, err := common.NewWatcher(s.ctx, common.WatcherConfig{
-		Fetchers:       s.databaseFetchers,
+		FetchersFn:     s.getAllDatabaseFetchers,
 		Log:            s.Log.WithField("kind", types.KindDatabase),
 		DiscoveryGroup: s.DiscoveryGroup,
 		Interval:       s.PollInterval,
 		Origin:         types.OriginCloud,
+		Clock:          s.clock,
 	})
 	if err != nil {
 		return trace.Wrap(err)
@@ -93,6 +94,18 @@ func (s *Server) startDatabaseWatchers() error {
 	return nil
 }
 
+func (s *Server) getAllDatabaseFetchers() []common.Fetcher {
+	allFetchers := make([]common.Fetcher, 0, len(s.databaseFetchers))
+
+	s.muDynamicFetchers.RLock()
+	for _, fetcherSet := range s.dynamicDatabaseFetchers {
+		allFetchers = append(allFetchers, fetcherSet...)
+	}
+	s.muDynamicFetchers.RUnlock()
+
+	return append(allFetchers, s.databaseFetchers...)
+}
+
 func (s *Server) getCurrentDatabases() types.ResourcesWithLabelsMap {
 	databases, err := s.AccessPoint.GetDatabases(s.ctx)
 	if err != nil {