🧹 coordinator v2

policy/scan/local_scanner.go

@@ -52,7 +52,7 @@
 	// allows setting the upstream credentials from a job
 	allowJobCredentials bool
 	disableProgressBar  bool
-	scanMutex           sync.Mutex
+	// scanMutex           sync.Mutex
 }
 
 type ScannerOption func(*LocalScanner)
@@ -82,7 +82,7 @@
 }
 
 func NewLocalScanner(opts ...ScannerOption) *LocalScanner {
-	runtime := providers.Coordinator.NewRuntime()
+	runtime := providers.GlobalCoordinator.NewRuntime()
 
 	ls := &LocalScanner{
 		resolvedPolicyCache: inmemory.NewResolvedPolicyCache(ResolvedPolicyCacheSize),
@@ -134,8 +134,8 @@
 	// scan that is in progress, which will result in errors. We can address this later
 	// by attaching a provider to a scan and killing only the providers that are not
 	// not actively used.
-	s.scanMutex.Lock()
-	defer s.scanMutex.Unlock()
+	// s.scanMutex.Lock()
+	// defer s.scanMutex.Unlock()
 	if job == nil {
 		return nil, status.Errorf(codes.InvalidArgument, "missing scan job")
 	}
@@ -245,7 +245,7 @@
 
 func (s *LocalScanner) distributeJob(job *Job, ctx context.Context, upstream *upstream.UpstreamConfig) (*ScanResult, error) {
 	// Always shut down the coordinator, to make sure providers are killed
-	defer providers.Coordinator.Shutdown()
+	// defer providers.Coordinator.Shutdown()
 
 	reporter, err := createReporter(ctx, job, upstream)
 	if err != nil {
@@ -267,40 +267,28 @@
 	// Within this process, we set up a catch-all deferred function, that shuts
 	// down all runtimes, in case we exit early.
 	defer func() {
-		for _, asset := range discoveredAssets.Assets {
+		for _, asset := range discoveredAssets.RootAssets {
 			// we can call close multiple times and it will only execute once
-			if asset.Runtime != nil {
-				asset.Runtime.Close()
+			if asset.Coordinator != nil {
+				asset.Coordinator.Shutdown()
 			}
 		}
 	}()
 
-	if len(discoveredAssets.Assets) == 0 {
+	assets := discoveredAssets.GetFlattenedChildren()
+	if len(assets) == 0 {
 		return reporter.Reports(), nil
 	}
 
-	multiprogress, err := scan.CreateProgressBar(discoveredAssets, s.disableProgressBar)
-	if err != nil {
-		return nil, err
-	}
-
-	// start the progress bar
-	scanGroups := sync.WaitGroup{}
-	scanGroups.Add(1)
-	go func() {
-		defer scanGroups.Done()
-		if err := multiprogress.Open(); err != nil {
-			log.Error().Err(err).Msg("failed to open progress bar")
-		}
-	}()
+	log.Info().Msgf("discovered %d assets", len(assets))
 
-	assetBatches := slicesx.Batch(discoveredAssets.Assets, 100)
+	assetBatches := slicesx.Batch(assets, 100)
 	for i := range assetBatches {
 		batch := assetBatches[i]
 
 		// sync assets
 		if upstream != nil && upstream.ApiEndpoint != "" && !upstream.Incognito {
-			log.Info().Msg("synchronize assets")
+			log.Info().Msgf("synchronize %d assets", len(batch))
 			client, err := upstream.InitClient()
 			if err != nil {
 				return nil, err
@@ -353,20 +341,38 @@
 				}
 			}
 		}
+	}
+
+	multiprogress, err := scan.CreateProgressBar(discoveredAssets, s.disableProgressBar)
+	if err != nil {
+		return nil, err
+	}
+
+	// start the progress bar
+	scanGroups := sync.WaitGroup{}
+	scanGroups.Add(1)
+	go func() {
+		defer scanGroups.Done()
+		if err := multiprogress.Open(); err != nil {
+			log.Error().Err(err).Msg("failed to open progress bar")
+		}
+	}()
 
-		// // if a bundle was provided check that it matches the filter, bundles can also be downloaded
-		// // later therefore we do not want to stop execution here
-		// if job.Bundle != nil && job.Bundle.FilterPolicies(job.PolicyFilters) {
-		// 	return nil, false, errors.New("all available packs filtered out. nothing to do.")
-		// }
+	// // if a bundle was provided check that it matches the filter, bundles can also be downloaded
+	// // later therefore we do not want to stop execution here
+	// if job.Bundle != nil && job.Bundle.FilterPolicies(job.PolicyFilters) {
+	// 	return nil, false, errors.New("all available packs filtered out. nothing to do.")
+	// }
 
+	for k := range discoveredAssets.RootAssets {
+		root := discoveredAssets.RootAssets[k]
 		wg := sync.WaitGroup{}
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			for i := range batch {
-				asset := batch[i].Asset
-				runtime := batch[i].Runtime
+			for i := range root.Children {
+				asset := root.Children[i].Asset
+				runtime := root.Children[i].Runtime
 
 				log.Debug().Interface("asset", asset).Msg("start scan")
 
@@ -398,6 +404,7 @@
 				// shut down all ephemeral runtimes
 				runtime.Close()
 			}
+			root.Coordinator.Shutdown()
 		}()
 		wg.Wait()
 	}