Initial refactoring to include Azure status reporting

api/proto/teleport/usageevents/v1/usageevents.proto

@@ -775,6 +775,19 @@ message AccessGraphAWSScanEvent {
   uint64 total_accounts = 11;
 }
 
+// AccessGraphAzureScanEvent is emitted when the Access Graph
+// Azure scan is enabled.
+message AccessGraphAzureScanEvent {
+  // total_vms is the total amount of virtual machines found in the Azure scan.
+  uint64 total_vms = 1;
+  // total_principals is the total amount of users found in the Azure scan.
+  uint64 total_principals = 2;
+  // total_role_definitions is the total amount of roles definitions found in the Azure scan.
+  uint64 total_role_definitions = 3;
+  // total_role_assignments is the total amount of roles assignments found in the Azure scan.
+  uint64 total_role_assignments = 4;
+}
+
 // UIAccessGraphCrownJewelDiffViewEvent is emitted when a user reviews the output of a Crown Jewel access path dff.
 message UIAccessGraphCrownJewelDiffViewEvent {
   // affected_resource_source is the source of the affected resource.
@@ -899,6 +912,7 @@ message UsageEventOneOf {
     UIAccessGraphCrownJewelDiffViewEvent ui_access_graph_crown_jewel_diff_view = 59;
     UserTaskStateEvent user_task_state_event = 60;
     UIIntegrationEnrollStepEvent ui_integration_enroll_step_event = 61;
+    AccessGraphAzureScanEvent access_graph_azure_scan_event = 62;
   }
   reserved 2; //UIOnboardGetStartedClickEvent
   reserved "ui_onboard_get_started_click";

lib/srv/discovery/access_graph_aws.go

@@ -82,7 +82,7 @@ func (s *Server) reconcileAccessGraph(ctx context.Context, currentTAGResources *
 		return trace.Wrap(errNoAccessGraphFetchers)
 	}
 
-	s.awsSyncStatus.iterationStarted(allFetchers, s.clock.Now())
+	s.awsSyncStatus.awsIterationStarted(allFetchers, s.clock.Now())
 	for _, discoveryConfigName := range s.awsSyncStatus.discoveryConfigs() {
 		s.updateDiscoveryConfigStatus(discoveryConfigName)
 	}
@@ -127,7 +127,7 @@ func (s *Server) reconcileAccessGraph(ctx context.Context, currentTAGResources *
 	upsert, toDel := aws_sync.ReconcileResults(currentTAGResources, result)
 	pushErr := push(stream, upsert, toDel)
 
-	s.awsSyncStatus.iterationFinished(allFetchers, pushErr, s.clock.Now())
+	s.awsSyncStatus.awsIterationFinished(allFetchers, pushErr, s.clock.Now())
 	for _, discoveryConfigName := range s.awsSyncStatus.discoveryConfigs() {
 		s.updateDiscoveryConfigStatus(discoveryConfigName)
 	}

lib/srv/discovery/access_graph_test.go

@@ -247,13 +247,13 @@ func TestServer_updateDiscoveryConfigStatus(t *testing.T) {
 					AccessPoint: accessPoint,
 					clock:       clock,
 				},
-				awsSyncStatus: awsSyncStatus{},
+				awsSyncStatus: tagSyncStatus{},
 			}
 
 			if tt.args.preRun {
-				s.awsSyncStatus.iterationStarted(tt.args.fetchers, s.clock.Now())
+				s.awsSyncStatus.awsIterationStarted(tt.args.fetchers, s.clock.Now())
 			} else {
-				s.awsSyncStatus.iterationFinished(tt.args.fetchers, tt.args.pushErr, s.clock.Now())
+				s.awsSyncStatus.awsIterationFinished(tt.args.fetchers, tt.args.pushErr, s.clock.Now())
 			}
 
 			for _, discoveryConfigName := range s.awsSyncStatus.discoveryConfigs() {

lib/srv/discovery/discovery.go

@@ -440,7 +440,7 @@ type Server struct {
 
 	dynamicDiscoveryConfig map[string]*discoveryconfig.DiscoveryConfig
 
-	awsSyncStatus         awsSyncStatus
+	awsSyncStatus         tagSyncStatus
 	awsEC2ResourcesStatus awsResourcesStatus
 	awsRDSResourcesStatus awsResourcesStatus
 	awsEKSResourcesStatus awsResourcesStatus
@@ -480,7 +480,7 @@ func New(ctx context.Context, cfg *Config) (*Server, error) {
 		dynamicTAGAWSFetchers:      make(map[string][]aws_sync.AWSSync),
 		dynamicTAGAzureFetchers:    make(map[string][]*azure_sync.Fetcher),
 		dynamicDiscoveryConfig:     make(map[string]*discoveryconfig.DiscoveryConfig),
-		awsSyncStatus:              awsSyncStatus{},
+		awsSyncStatus:              tagSyncStatus{},
 		awsEC2ResourcesStatus:      newAWSResourceStatusCollector(types.AWSMatcherEC2),
 		awsRDSResourcesStatus:      newAWSResourceStatusCollector(types.AWSMatcherRDS),
 		awsEKSResourcesStatus:      newAWSResourceStatusCollector(types.AWSMatcherEKS),

lib/srv/discovery/status.go

@@ -62,6 +62,8 @@ func (s *Server) updateDiscoveryConfigStatus(discoveryConfigNames ...string) {
 		// Merge AWS Sync (TAG) status
 		discoveryConfigStatus = s.awsSyncStatus.mergeIntoGlobalStatus(discoveryConfigName, discoveryConfigStatus)
 
+		// Merge Azure Sync (TAG) status
+
 		// Merge AWS EC2 Instances (auto discovery) status
 		discoveryConfigStatus = s.awsEC2ResourcesStatus.mergeIntoGlobalStatus(discoveryConfigName, discoveryConfigStatus)
 
@@ -84,16 +86,16 @@ func (s *Server) updateDiscoveryConfigStatus(discoveryConfigNames ...string) {
 	}
 }
 
-// awsSyncStatus contains all the status for aws_sync Fetchers grouped by DiscoveryConfig.
-type awsSyncStatus struct {
+// tagSyncStatus contains all the status for both AWS and Azure fetchers grouped by DiscoveryConfig.
+type tagSyncStatus struct {
 	mu sync.RWMutex
-	// awsSyncResults maps the DiscoveryConfig name to a aws_sync result.
-	// Each DiscoveryConfig might have multiple `aws_sync` matchers.
-	awsSyncResults map[string][]awsSyncResult
+	// syncResults maps the DiscoveryConfig name to a AWS or Azure fetcher result.
+	// Each DiscoveryConfig might have multiple AWS or Azure matchers.
+	syncResults map[string][]tagSyncResult
 }
 
-// awsSyncResult stores the result of the aws_sync Matchers for a given DiscoveryConfig.
-type awsSyncResult struct {
+// tagSyncResult stores the result of the aws_sync Matchers for a given DiscoveryConfig.
+type tagSyncResult struct {
 	// state is the State for the DiscoveryConfigStatus.
 	// Allowed values are:
 	// - DISCOVERY_CONFIG_STATE_SYNCING
@@ -105,11 +107,11 @@ type awsSyncResult struct {
 	discoveredResources uint64
 }
 
-func (d *awsSyncStatus) iterationFinished(fetchers []aws_sync.AWSSync, pushErr error, lastUpdate time.Time) {
+func (d *tagSyncStatus) awsIterationFinished(fetchers []aws_sync.AWSSync, pushErr error, lastUpdate time.Time) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
-	d.awsSyncResults = make(map[string][]awsSyncResult)
+	d.syncResults = make(map[string][]tagSyncResult)
 	for _, fetcher := range fetchers {
 		// Only update the status for fetchers that are from the discovery config.
 		if !fetcher.IsFromDiscoveryConfig() {
@@ -119,7 +121,7 @@ func (d *awsSyncStatus) iterationFinished(fetchers []aws_sync.AWSSync, pushErr e
 		count, statusErr := fetcher.Status()
 		statusAndPushErr := trace.NewAggregate(statusErr, pushErr)
 
-		fetcherResult := awsSyncResult{
+		fetcherResult := tagSyncResult{
 			state:               discoveryconfigv1.DiscoveryConfigState_DISCOVERY_CONFIG_STATE_RUNNING.String(),
 			lastSyncTime:        lastUpdate,
 			discoveredResources: count,
@@ -131,46 +133,46 @@ func (d *awsSyncStatus) iterationFinished(fetchers []aws_sync.AWSSync, pushErr e
 			fetcherResult.state = discoveryconfigv1.DiscoveryConfigState_DISCOVERY_CONFIG_STATE_ERROR.String()
 		}
 
-		d.awsSyncResults[fetcher.DiscoveryConfigName()] = append(d.awsSyncResults[fetcher.DiscoveryConfigName()], fetcherResult)
+		d.syncResults[fetcher.DiscoveryConfigName()] = append(d.syncResults[fetcher.DiscoveryConfigName()], fetcherResult)
 	}
 }
 
-func (d *awsSyncStatus) discoveryConfigs() []string {
+func (d *tagSyncStatus) discoveryConfigs() []string {
 	d.mu.RLock()
 	defer d.mu.RUnlock()
 
-	ret := make([]string, 0, len(d.awsSyncResults))
-	for k := range d.awsSyncResults {
+	ret := make([]string, 0, len(d.syncResults))
+	for k := range d.syncResults {
 		ret = append(ret, k)
 	}
 	return ret
 }
 
-func (d *awsSyncStatus) iterationStarted(fetchers []aws_sync.AWSSync, lastUpdate time.Time) {
+func (d *tagSyncStatus) awsIterationStarted(fetchers []aws_sync.AWSSync, lastUpdate time.Time) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
-	d.awsSyncResults = make(map[string][]awsSyncResult)
+	d.syncResults = make(map[string][]tagSyncResult)
 	for _, fetcher := range fetchers {
 		// Only update the status for fetchers that are from the discovery config.
 		if !fetcher.IsFromDiscoveryConfig() {
 			continue
 		}
 
-		fetcherResult := awsSyncResult{
+		fetcherResult := tagSyncResult{
 			state:        discoveryconfigv1.DiscoveryConfigState_DISCOVERY_CONFIG_STATE_SYNCING.String(),
 			lastSyncTime: lastUpdate,
 		}
 
-		d.awsSyncResults[fetcher.DiscoveryConfigName()] = append(d.awsSyncResults[fetcher.DiscoveryConfigName()], fetcherResult)
+		d.syncResults[fetcher.DiscoveryConfigName()] = append(d.syncResults[fetcher.DiscoveryConfigName()], fetcherResult)
 	}
 }
 
-func (d *awsSyncStatus) mergeIntoGlobalStatus(discoveryConfigName string, existingStatus discoveryconfig.Status) discoveryconfig.Status {
+func (d *tagSyncStatus) mergeIntoGlobalStatus(discoveryConfigName string, existingStatus discoveryconfig.Status) discoveryconfig.Status {
 	d.mu.RLock()
 	defer d.mu.RUnlock()
 
-	awsStatusFetchers, found := d.awsSyncResults[discoveryConfigName]
+	awsStatusFetchers, found := d.syncResults[discoveryConfigName]
 	if !found {
 		return existingStatus
 	}