DiscoveryConfigStatus: update even when no resource is found (#50433)

* DiscoveryConfigStatus: update even when no resource is found During Auto Discover, when using a DiscoveryConfig, if no resources are found, the DiscoveryConfigStatus is not updated accordingly. This PR ensures that, even when no resources are found, the status will report so. * use comparable instead of any for generic method * remove useless un-named function on PreFetchHooks * prevent call to ssm:SendCommand with 0 instances * rename var from ok to found
gravitational · Jan 6, 2025 · b6ffcee · b6ffcee
1 parent 2182aa7
commit b6ffcee
Show file tree

Hide file tree

Showing 12 changed files with 222 additions and 57 deletions.
diff --git a/lib/srv/discovery/database_watcher.go b/lib/srv/discovery/database_watcher.go
@@ -74,17 +74,7 @@ func (s *Server) startDatabaseWatchers() error {
 			TriggerFetchC:  s.newDiscoveryConfigChangedSub(),
 			Origin:         types.OriginCloud,
 			Clock:          s.clock,
-			PreFetchHookFn: func() {
-				discoveryConfigs := slices.FilterMapUnique(
-					s.getAllDatabaseFetchers(),
-					func(f common.Fetcher) (s string, include bool) {
-						return f.GetDiscoveryConfigName(), f.GetDiscoveryConfigName() != ""
-					},
-				)
-				s.updateDiscoveryConfigStatus(discoveryConfigs...)
-
-				s.awsRDSResourcesStatus.reset()
-			},
+			PreFetchHookFn: s.databaseWatcherIterationStarted,
 		},
 	)
 	if err != nil {
@@ -151,6 +141,38 @@ func (s *Server) startDatabaseWatchers() error {
 	return nil
 }
 
+func (s *Server) databaseWatcherIterationStarted() {
+	allFetchers := s.getAllDatabaseFetchers()
+	if len(allFetchers) == 0 {
+		return
+	}
+
+	s.submitFetchersEvent(allFetchers)
+
+	awsResultGroups := slices.FilterMapUnique(
+		allFetchers,
+		func(f common.Fetcher) (awsResourceGroup, bool) {
+			include := f.GetDiscoveryConfigName() != "" && f.IntegrationName() != ""
+			resourceGroup := awsResourceGroup{
+				discoveryConfigName: f.GetDiscoveryConfigName(),
+				integration:         f.IntegrationName(),
+			}
+			return resourceGroup, include
+		},
+	)
+
+	for _, g := range awsResultGroups {
+		s.awsRDSResourcesStatus.iterationStarted(g)
+	}
+
+	discoveryConfigs := slices.FilterMapUnique(awsResultGroups, func(g awsResourceGroup) (s string, include bool) {
+		return g.discoveryConfigName, true
+	})
+	s.updateDiscoveryConfigStatus(discoveryConfigs...)
+
+	s.awsRDSResourcesStatus.reset()
+}
+
 func (s *Server) getAllDatabaseFetchers() []common.Fetcher {
 	allFetchers := make([]common.Fetcher, 0, len(s.databaseFetchers))
 
@@ -162,8 +184,6 @@ func (s *Server) getAllDatabaseFetchers() []common.Fetcher {
 
 	allFetchers = append(allFetchers, s.databaseFetchers...)
 
-	s.submitFetchersEvent(allFetchers)
-
 	return allFetchers
 }
 

diff --git a/lib/srv/discovery/discovery.go b/lib/srv/discovery/discovery.go
@@ -523,18 +523,7 @@ func (s *Server) initAWSWatchers(matchers []types.AWSMatcher) error {
 		s.ctx, s.getAllAWSServerFetchers, s.caRotationCh,
 		server.WithPollInterval(s.PollInterval),
 		server.WithTriggerFetchC(s.newDiscoveryConfigChangedSub()),
-		server.WithPreFetchHookFn(func() {
-			discoveryConfigs := libslices.FilterMapUnique(
-				s.getAllAWSServerFetchers(),
-				func(f server.Fetcher) (s string, include bool) {
-					return f.GetDiscoveryConfigName(), f.GetDiscoveryConfigName() != ""
-				},
-			)
-			s.updateDiscoveryConfigStatus(discoveryConfigs...)
-
-			s.awsEC2ResourcesStatus.reset()
-			s.awsEC2Tasks.reset()
-		}),
+		server.WithPreFetchHookFn(s.ec2WatcherIterationStarted),
 	)
 	if err != nil {
 		return trace.Wrap(err)
@@ -575,6 +564,38 @@ func (s *Server) initAWSWatchers(matchers []types.AWSMatcher) error {
 	return nil
 }
 
+func (s *Server) ec2WatcherIterationStarted() {
+	allFetchers := s.getAllAWSServerFetchers()
+	if len(allFetchers) == 0 {
+		return
+	}
+
+	s.submitFetchEvent(types.CloudAWS, types.AWSMatcherEC2)
+
+	awsResultGroups := libslices.FilterMapUnique(
+		allFetchers,
+		func(f server.Fetcher) (awsResourceGroup, bool) {
+			include := f.GetDiscoveryConfigName() != "" && f.IntegrationName() != ""
+			resourceGroup := awsResourceGroup{
+				discoveryConfigName: f.GetDiscoveryConfigName(),
+				integration:         f.IntegrationName(),
+			}
+			return resourceGroup, include
+		},
+	)
+	for _, g := range awsResultGroups {
+		s.awsEC2ResourcesStatus.iterationStarted(g)
+	}
+
+	discoveryConfigs := libslices.FilterMapUnique(awsResultGroups, func(g awsResourceGroup) (s string, include bool) {
+		return g.discoveryConfigName, true
+	})
+	s.updateDiscoveryConfigStatus(discoveryConfigs...)
+	s.awsEC2ResourcesStatus.reset()
+
+	s.awsEC2Tasks.reset()
+}
+
 func (s *Server) initKubeAppWatchers(matchers []types.KubernetesMatcher) error {
 	if len(matchers) == 0 {
 		return nil
@@ -1483,10 +1504,6 @@ func (s *Server) getAllAWSServerFetchers() []server.Fetcher {
 
 	allFetchers = append(allFetchers, s.staticServerAWSFetchers...)
 
-	if len(allFetchers) > 0 {
-		s.submitFetchEvent(types.CloudAWS, types.AWSMatcherEC2)
-	}
-
 	return allFetchers
 }
 

diff --git a/lib/srv/discovery/discovery_test.go b/lib/srv/discovery/discovery_test.go
@@ -285,6 +285,26 @@ func TestDiscoveryServer(t *testing.T) {
 	)
 	require.NoError(t, err)
 
+	dcForEC2StatusWithoutMatchName := uuid.NewString()
+	dcForEC2StatusWithoutMatch, err := discoveryconfig.NewDiscoveryConfig(
+		header.Metadata{Name: dcForEC2StatusWithoutMatchName},
+		discoveryconfig.Spec{
+			DiscoveryGroup: defaultDiscoveryGroup,
+			AWS: []types.AWSMatcher{{
+				Types:   []string{"ec2"},
+				Regions: []string{"eu-central-1"},
+				Tags:    map[string]utils.Strings{"teleport": {"yes"}},
+				SSM:     &types.AWSSSM{DocumentName: "document"},
+				Params: &types.InstallerParams{
+					InstallTeleport: true,
+					EnrollMode:      types.InstallParamEnrollMode_INSTALL_PARAM_ENROLL_MODE_SCRIPT,
+				},
+				Integration: "my-integration",
+			}},
+		},
+	)
+	require.NoError(t, err)
+
 	discoveryConfigForUserTaskEKSTestName := uuid.NewString()
 	discoveryConfigForUserTaskEKSTest, err := discoveryconfig.NewDiscoveryConfig(
 		header.Metadata{Name: discoveryConfigForUserTaskEKSTestName},
@@ -573,6 +593,31 @@ func TestDiscoveryServer(t *testing.T) {
 			},
 			wantInstalledInstances: []string{"instance-id-1"},
 		},
+		{
+			name:              "no nodes found using DiscoveryConfig and Integration, but DiscoveryConfig Status is still updated",
+			presentInstances:  []types.Server{},
+			foundEC2Instances: []ec2types.Instance{},
+			ssm:               &mockSSMClient{},
+			emitter:           &mockEmitter{},
+			staticMatchers:    Matchers{},
+			discoveryConfig:   dcForEC2StatusWithoutMatch,
+			wantDiscoveryConfigStatus: &discoveryconfig.Status{
+				State:               "DISCOVERY_CONFIG_STATE_SYNCING",
+				ErrorMessage:        nil,
+				DiscoveredResources: 0,
+				LastSyncTime:        fakeClock.Now().UTC(),
+				IntegrationDiscoveredResources: map[string]*discoveryconfigv1.IntegrationDiscoveredSummary{
+					"my-integration": {
+						AwsEc2: &discoveryconfigv1.ResourcesDiscoveredSummary{
+							Found:    0,
+							Enrolled: 0,
+							Failed:   0,
+						},
+					},
+				},
+			},
+			wantInstalledInstances: []string{},
+		},
 		{
 			name:             "one node found but SSM Run fails and DiscoverEC2 User Task is created",
 			presentInstances: []types.Server{},

diff --git a/lib/srv/discovery/kube_integration_watcher.go b/lib/srv/discovery/kube_integration_watcher.go
@@ -76,18 +76,7 @@ func (s *Server) startKubeIntegrationWatchers() error {
 		Interval:       s.PollInterval,
 		Origin:         types.OriginCloud,
 		TriggerFetchC:  s.newDiscoveryConfigChangedSub(),
-		PreFetchHookFn: func() {
-			discoveryConfigs := libslices.FilterMapUnique(
-				s.getKubeIntegrationFetchers(),
-				func(f common.Fetcher) (s string, include bool) {
-					return f.GetDiscoveryConfigName(), f.GetDiscoveryConfigName() != ""
-				},
-			)
-			s.updateDiscoveryConfigStatus(discoveryConfigs...)
-
-			s.awsEKSResourcesStatus.reset()
-			s.awsEKSTasks.reset()
-		},
+		PreFetchHookFn: s.kubernetesIntegrationWatcherIterationStarted,
 	})
 	if err != nil {
 		return trace.Wrap(err)
@@ -194,6 +183,38 @@ func (s *Server) startKubeIntegrationWatchers() error {
 	return nil
 }
 
+func (s *Server) kubernetesIntegrationWatcherIterationStarted() {
+	allFetchers := s.getKubeIntegrationFetchers()
+	if len(allFetchers) == 0 {
+		return
+	}
+
+	s.submitFetchersEvent(allFetchers)
+
+	awsResultGroups := libslices.FilterMapUnique(
+		allFetchers,
+		func(f common.Fetcher) (awsResourceGroup, bool) {
+			include := f.GetDiscoveryConfigName() != "" && f.IntegrationName() != ""
+			resourceGroup := awsResourceGroup{
+				discoveryConfigName: f.GetDiscoveryConfigName(),
+				integration:         f.IntegrationName(),
+			}
+			return resourceGroup, include
+		},
+	)
+	for _, g := range awsResultGroups {
+		s.awsEKSResourcesStatus.iterationStarted(g)
+	}
+
+	discoveryConfigs := libslices.FilterMapUnique(awsResultGroups, func(g awsResourceGroup) (s string, include bool) {
+		return g.discoveryConfigName, true
+	})
+	s.updateDiscoveryConfigStatus(discoveryConfigs...)
+
+	s.awsEKSResourcesStatus.reset()
+	s.awsEKSTasks.reset()
+}
+
 func (s *Server) enrollEKSClusters(region, integration, discoveryConfigName string, clusters []types.DiscoveredEKSCluster, agentVersion string, mu *sync.Mutex, enrollingClusters map[string]bool) {
 	mu.Lock()
 	for _, c := range clusters {

diff --git a/lib/srv/discovery/status.go b/lib/srv/discovery/status.go
@@ -296,6 +296,15 @@ func (ars *awsResourcesStatus) incrementFailed(g awsResourceGroup, count int) {
 	ars.awsResourcesResults[g] = groupStats
 }
 
+func (ars *awsResourcesStatus) iterationStarted(g awsResourceGroup) {
+	ars.mu.Lock()
+	defer ars.mu.Unlock()
+	if ars.awsResourcesResults == nil {
+		ars.awsResourcesResults = make(map[awsResourceGroup]awsResourceGroupResult)
+	}
+	ars.awsResourcesResults[g] = awsResourceGroupResult{}
+}
+
 func (ars *awsResourcesStatus) incrementFound(g awsResourceGroup, count int) {
 	ars.mu.Lock()
 	defer ars.mu.Unlock()

diff --git a/lib/srv/server/azure_watcher.go b/lib/srv/server/azure_watcher.go
@@ -121,6 +121,7 @@ type azureFetcherConfig struct {
 	ResourceGroup       string
 	AzureClientGetter   azureClientGetter
 	DiscoveryConfigName string
+	Integration         string
 }
 
 type azureInstanceFetcher struct {
@@ -132,6 +133,7 @@ type azureInstanceFetcher struct {
 	Parameters          map[string]string
 	ClientID            string
 	DiscoveryConfigName string
+	Integration         string
 }
 
 func newAzureInstanceFetcher(cfg azureFetcherConfig) *azureInstanceFetcher {
@@ -142,6 +144,7 @@ func newAzureInstanceFetcher(cfg azureFetcherConfig) *azureInstanceFetcher {
 		ResourceGroup:       cfg.ResourceGroup,
 		Labels:              cfg.Matcher.ResourceTags,
 		DiscoveryConfigName: cfg.DiscoveryConfigName,
+		Integration:         cfg.Integration,
 	}
 
 	if cfg.Matcher.Params != nil {
@@ -164,6 +167,12 @@ func (f *azureInstanceFetcher) GetDiscoveryConfigName() string {
 	return f.DiscoveryConfigName
 }
 
+// IntegrationName identifies the integration name whose credentials were used to fetch the resources.
+// Might be empty when the fetcher is using ambient credentials.
+func (f *azureInstanceFetcher) IntegrationName() string {
+	return f.Integration
+}
+
 // GetInstances fetches all Azure virtual machines matching configured filters.
 func (f *azureInstanceFetcher) GetInstances(ctx context.Context, _ bool) ([]Instances, error) {
 	client, err := f.AzureClientGetter.GetAzureVirtualMachinesClient(f.Subscription)

diff --git a/lib/srv/server/ec2_watcher.go b/lib/srv/server/ec2_watcher.go
@@ -461,3 +461,9 @@ func (f *ec2InstanceFetcher) GetInstances(ctx context.Context, rotation bool) ([
 func (f *ec2InstanceFetcher) GetDiscoveryConfigName() string {
 	return f.DiscoveryConfigName
 }
+
+// IntegrationName identifies the integration name whose credentials were used to fetch the resources.
+// Might be empty when the fetcher is using ambient credentials.
+func (f *ec2InstanceFetcher) IntegrationName() string {
+	return f.Integration
+}
diff --git a/lib/srv/server/gcp_watcher.go b/lib/srv/server/gcp_watcher.go
@@ -111,6 +111,7 @@ type gcpFetcherConfig struct {
 	GCPClient           gcp.InstancesClient
 	projectsClient      gcp.ProjectsClient
 	DiscoveryConfigName string
+	Integration         string
 }
 
 type gcpInstanceFetcher struct {
@@ -123,16 +124,19 @@ type gcpInstanceFetcher struct {
 	Parameters          map[string]string
 	projectsClient      gcp.ProjectsClient
 	DiscoveryConfigName string
+	Integration         string
 }
 
 func newGCPInstanceFetcher(cfg gcpFetcherConfig) *gcpInstanceFetcher {
 	fetcher := &gcpInstanceFetcher{
-		GCP:             cfg.GCPClient,
-		Zones:           cfg.Matcher.Locations,
-		ProjectIDs:      cfg.Matcher.ProjectIDs,
-		ServiceAccounts: cfg.Matcher.ServiceAccounts,
-		Labels:          cfg.Matcher.GetLabels(),
-		projectsClient:  cfg.projectsClient,
+		GCP:                 cfg.GCPClient,
+		Zones:               cfg.Matcher.Locations,
+		ProjectIDs:          cfg.Matcher.ProjectIDs,
+		ServiceAccounts:     cfg.Matcher.ServiceAccounts,
+		Labels:              cfg.Matcher.GetLabels(),
+		projectsClient:      cfg.projectsClient,
+		Integration:         cfg.Integration,
+		DiscoveryConfigName: cfg.DiscoveryConfigName,
 	}
 	if cfg.Matcher.Params != nil {
 		fetcher.Parameters = map[string]string{
@@ -152,6 +156,12 @@ func (f *gcpInstanceFetcher) GetDiscoveryConfigName() string {
 	return f.DiscoveryConfigName
 }
 
+// IntegrationName identifies the integration name whose credentials were used to fetch the resources.
+// Might be empty when the fetcher is using ambient credentials.
+func (f *gcpInstanceFetcher) IntegrationName() string {
+	return f.Integration
+}
+
 // GetInstances fetches all GCP virtual machines matching configured filters.
 func (f *gcpInstanceFetcher) GetInstances(ctx context.Context, _ bool) ([]Instances, error) {
 	// Key by project ID, then by zone.

diff --git a/lib/srv/server/ssm_install.go b/lib/srv/server/ssm_install.go
@@ -199,6 +199,10 @@ func (si *SSMInstaller) Run(ctx context.Context, req SSMRunRequest) error {
 		validInstances = instancesState.valid
 	}
 
+	if len(validInstances) == 0 {
+		return nil
+	}
+
 	validInstanceIDs := instanceIDsFrom(validInstances)
 	output, err := req.SSM.SendCommand(ctx, &ssm.SendCommandInput{
 		DocumentName: aws.String(req.DocumentName),

diff --git a/lib/srv/server/watcher.go b/lib/srv/server/watcher.go
@@ -45,6 +45,9 @@ type Fetcher interface {
 	// GetDiscoveryConfigName returns the DiscoveryConfig name that created this fetcher.
 	// Empty for Fetchers created from `teleport.yaml/discovery_service.aws.<Matcher>` matchers.
 	GetDiscoveryConfigName() string
+	// IntegrationName identifies the integration name whose credentials were used to fetch the resources.
+	// Might be empty when the fetcher is using ambient credentials.
+	IntegrationName() string
 }
 
 // WithTriggerFetchC sets a poll trigger to manual start a resource polling.