Generalize UI events for cataloging tasks (#2369)

* generalize ui events for cataloging tasks Signed-off-by: Alex Goodman <wagoodman@users.noreply.github.com> * moderate review comments Signed-off-by: Alex Goodman <wagoodman@users.noreply.github.com> * incorporate review comments Signed-off-by: Alex Goodman <wagoodman@users.noreply.github.com> * rename cataloger task progress object Signed-off-by: Alex Goodman <wagoodman@users.noreply.github.com> * migrate cataloger task fn to bus helper Signed-off-by: Alex Goodman <wagoodman@users.noreply.github.com> --------- Signed-off-by: Alex Goodman <wagoodman@users.noreply.github.com>
2026-11-04 16:36:40 +01:00 · 2023-11-30 16:25:50 +00:00
parent b943da6433
commit 4adfbeb5f0
28 changed files with 427 additions and 807 deletions
@@ -6,14 +6,14 @@ import (
 	"runtime/debug"
 	"sync"

+	"github.com/dustin/go-humanize"
 	"github.com/hashicorp/go-multierror"
-	"github.com/wagoodman/go-partybus"
 	"github.com/wagoodman/go-progress"

 	"github.com/anchore/syft/internal/bus"
 	"github.com/anchore/syft/internal/log"
 	"github.com/anchore/syft/syft/artifact"
-	"github.com/anchore/syft/syft/event"
+	"github.com/anchore/syft/syft/event/monitor"
 	"github.com/anchore/syft/syft/file"
 	"github.com/anchore/syft/syft/linux"
 	"github.com/anchore/syft/syft/pkg"
@@ -35,21 +35,6 @@ type catalogResult struct {
 	Error      error
 }

-// newMonitor creates a new Monitor object and publishes the object on the bus as a PackageCatalogerStarted event.
-func newMonitor() (*progress.Manual, *progress.Manual) {
-	filesProcessed := progress.Manual{}
-	packagesDiscovered := progress.Manual{}
-
-	bus.Publish(partybus.Event{
-		Type: event.PackageCatalogerStarted,
-		Value: Monitor{
-			FilesProcessed:     progress.Monitorable(&filesProcessed),
-			PackagesDiscovered: progress.Monitorable(&packagesDiscovered),
-		},
-	})
-	return &filesProcessed, &packagesDiscovered
-}
-
 func runCataloger(cataloger pkg.Cataloger, resolver file.Resolver) (catalogerResult *catalogResult, err error) {
 	// handle individual cataloger panics
 	defer func() {
@@ -101,6 +86,7 @@ func runCataloger(cataloger pkg.Cataloger, resolver file.Resolver) (catalogerRes
 		}
 		catalogerResult.Packages = append(catalogerResult.Packages, p)
 	}
+
 	catalogerResult.Relationships = append(catalogerResult.Relationships, relationships...)
 	log.WithFields("cataloger", cataloger.Name()).Trace("cataloging complete")
 	return catalogerResult, err
@@ -116,9 +102,7 @@ func Catalog(resolver file.Resolver, _ *linux.Release, parallelism int, cataloge
 	catalog := pkg.NewCollection()
 	var allRelationships []artifact.Relationship

-	filesProcessed, packagesDiscovered := newMonitor()
-	defer filesProcessed.SetCompleted()
-	defer packagesDiscovered.SetCompleted()
+	prog := monitorPackageCatalogingTask()

 	// perform analysis, accumulating errors for each failed analysis
 	var errs error
@@ -131,10 +115,11 @@ func Catalog(resolver file.Resolver, _ *linux.Release, parallelism int, cataloge

 	jobs := make(chan pkg.Cataloger, nCatalogers)
 	results := make(chan *catalogResult, nCatalogers)
-	discoveredPackages := make(chan int64, nCatalogers)

 	waitGroup := sync.WaitGroup{}

+	var totalPackagesDiscovered int64
+
 	for i := 0; i < parallelism; i++ {
 		waitGroup.Add(1)

@@ -148,20 +133,16 @@ func Catalog(resolver file.Resolver, _ *linux.Release, parallelism int, cataloge
 				// ensure we set the error to be aggregated
 				result.Error = err

-				discoveredPackages <- result.Discovered
+				prog.Add(result.Discovered)
+				totalPackagesDiscovered += result.Discovered
+				count := humanize.Comma(totalPackagesDiscovered)
+				prog.AtomicStage.Set(fmt.Sprintf("%s packages", count))

 				results <- result
 			}
 		}()
 	}

-	// dynamically show updated discovered package status
-	go func() {
-		for discovered := range discoveredPackages {
-			packagesDiscovered.Add(discovered)
-		}
-	}()
-
 	// Enqueue the jobs
 	for _, cataloger := range catalogers {
 		jobs <- cataloger
@@ -171,7 +152,6 @@ func Catalog(resolver file.Resolver, _ *linux.Release, parallelism int, cataloge
 	// Wait for the jobs to finish
 	waitGroup.Wait()
 	close(results)
-	close(discoveredPackages)

 	// collect the results
 	for result := range results {
@@ -186,6 +166,12 @@ func Catalog(resolver file.Resolver, _ *linux.Release, parallelism int, cataloge

 	allRelationships = append(allRelationships, pkg.NewRelationships(catalog)...)

+	if errs != nil {
+		prog.SetError(errs)
+	} else {
+		prog.SetCompleted()
+	}
+
 	return catalog, allRelationships, errs
 }

@@ -228,3 +214,16 @@ func packageFileOwnershipRelationships(p pkg.Package, resolver file.PathResolver
 	}
 	return relationships, nil
 }
+
+func monitorPackageCatalogingTask() *monitor.CatalogerTaskProgress {
+	info := monitor.GenericTask{
+		Title: monitor.Title{
+			Default:      "Catalog packages",
+			WhileRunning: "Cataloging packages",
+			OnSuccess:    "Cataloged packages",
+		},
+		HideOnSuccess: false,
+	}
+
+	return bus.StartCatalogerTask(info, -1, "")
+}
@@ -20,6 +20,7 @@ import (
 	"github.com/go-git/go-git/v5/storage/memory"
 	"github.com/scylladb/go-set/strset"

+	"github.com/anchore/syft/internal/bus"
 	"github.com/anchore/syft/internal/licenses"
 	"github.com/anchore/syft/internal/log"
 	"github.com/anchore/syft/syft/event/monitor"
@@ -31,7 +32,6 @@ import (
 type goLicenses struct {
 	opts                  CatalogerConfig
 	localModCacheResolver file.WritableResolver
-	progress              *monitor.CatalogerTask
 	lowerLicenseFileNames *strset.Set
 }

@@ -39,11 +39,6 @@ func newGoLicenses(opts CatalogerConfig) goLicenses {
 	return goLicenses{
 		opts:                  opts,
 		localModCacheResolver: modCacheResolver(opts.LocalModCacheDir),
-		progress: &monitor.CatalogerTask{
-			SubStatus:          true,
-			RemoveOnCompletion: true,
-			Title:              "Downloading go mod",
-		},
 		lowerLicenseFileNames: strset.New(lowercaseLicenseFiles()...),
 	}
 }
@@ -123,7 +118,16 @@ func (c *goLicenses) getLicensesFromRemote(moduleName, moduleVersion string) ([]

 	proxies := remotesForModule(c.opts.Proxies, c.opts.NoProxy, moduleName)

-	fsys, err := getModule(c.progress, proxies, moduleName, moduleVersion)
+	prog := bus.StartCatalogerTask(monitor.GenericTask{
+		Title: monitor.Title{
+			Default:      "Download go mod",
+			WhileRunning: "Downloading go mod",
+			OnSuccess:    "Downloaded go mod",
+		},
+		HideOnSuccess: true,
+	}, -1, "")
+
+	fsys, err := getModule(prog, proxies, moduleName, moduleVersion)
 	if err != nil {
 		return nil, err
 	}
@@ -206,7 +210,7 @@ func processCaps(s string) string {
 	})
 }

-func getModule(progress *monitor.CatalogerTask, proxies []string, moduleName, moduleVersion string) (fsys fs.FS, err error) {
+func getModule(progress *monitor.CatalogerTaskProgress, proxies []string, moduleName, moduleVersion string) (fsys fs.FS, err error) {
 	for _, proxy := range proxies {
 		u, _ := url.Parse(proxy)
 		if proxy == "direct" {
@@ -218,7 +222,7 @@ func getModule(progress *monitor.CatalogerTask, proxies []string, moduleName, mo
 			fsys, err = getModuleProxy(progress, proxy, moduleName, moduleVersion)
 		case "file":
 			p := filepath.Join(u.Path, moduleName, "@v", moduleVersion)
-			progress.SetValue(fmt.Sprintf("file: %s", p))
+			progress.AtomicStage.Set(fmt.Sprintf("file: %s", p))
 			fsys = os.DirFS(p)
 		}
 		if fsys != nil {
@@ -228,9 +232,9 @@ func getModule(progress *monitor.CatalogerTask, proxies []string, moduleName, mo
 	return
 }

-func getModuleProxy(progress *monitor.CatalogerTask, proxy string, moduleName string, moduleVersion string) (out fs.FS, _ error) {
+func getModuleProxy(progress *monitor.CatalogerTaskProgress, proxy string, moduleName string, moduleVersion string) (out fs.FS, _ error) {
 	u := fmt.Sprintf("%s/%s/@v/%s.zip", proxy, moduleName, moduleVersion)
-	progress.SetValue(u)
+	progress.AtomicStage.Set(u)

 	// get the module zip
 	resp, err := http.Get(u) //nolint:gosec
@@ -241,7 +245,7 @@ func getModuleProxy(progress *monitor.CatalogerTask, proxy string, moduleName st

 	if resp.StatusCode != http.StatusOK {
 		u = fmt.Sprintf("%s/%s/@v/%s.zip", proxy, strings.ToLower(moduleName), moduleVersion)
-		progress.SetValue(u)
+		progress.AtomicStage.Set(u)

 		// try lowercasing it; some packages have mixed casing that really messes up the proxy
 		resp, err = http.Get(u) //nolint:gosec
@@ -284,14 +288,14 @@ func findVersionPath(f fs.FS, dir string) string {
 	return ""
 }

-func getModuleRepository(progress *monitor.CatalogerTask, moduleName string, moduleVersion string) (fs.FS, error) {
+func getModuleRepository(progress *monitor.CatalogerTaskProgress, moduleName string, moduleVersion string) (fs.FS, error) {
 	repoName := moduleName
 	parts := strings.Split(moduleName, "/")
 	if len(parts) > 2 {
 		repoName = fmt.Sprintf("%s/%s/%s", parts[0], parts[1], parts[2])
 	}

-	progress.SetValue(fmt.Sprintf("git: %s", repoName))
+	progress.AtomicStage.Set(fmt.Sprintf("git: %s", repoName))

 	f := memfs.New()
 	buf := &bytes.Buffer{}