Analysis Progress Monitoring Improvement

analyze/analyze.go

@@ -96,6 +96,14 @@ type Analyzer struct {
 	Formatter Formatter
 	Config    *models.Config
 	Opa       *opa.Opa
+	Observer  ProgressObserver
+}
+
+func (a *Analyzer) observer() ProgressObserver {
+	if a.Observer != nil {
+		return a.Observer
+	}
+	return noopObserver{}
 }
 
 func (a *Analyzer) AnalyzeOrg(ctx context.Context, org string, numberOfGoroutines *int) ([]*models.PackageInsights, error) {
@@ -115,7 +123,7 @@ func (a *Analyzer) AnalyzeOrg(ctx context.Context, org string, numberOfGoroutine
 	inventory := scanner.NewInventory(a.Opa, pkgsupplyClient, provider, providerVersion)
 
 	log.Debug().Msgf("Starting repository analysis for organization: %s on %s", org, provider)
-	bar := a.ProgressBar(0, "Analyzing repositories")
+	obs := a.observer()
 
 	var reposWg sync.WaitGroup
 	errChan := make(chan error, 1)
@@ -143,17 +151,16 @@ func (a *Analyzer) AnalyzeOrg(ctx context.Context, org string, numberOfGoroutine
 			continue
 		}
 		if repoBatch.TotalCount != 0 {
-			bar.ChangeMax(repoBatch.TotalCount)
+			obs.OnDiscoveryCompleted(org, repoBatch.TotalCount)
 		}
 
 		for _, repo := range repoBatch.Repositories {
 			if a.Config.IgnoreForks && repo.GetIsFork() {
-				bar.ChangeMax(repoBatch.TotalCount - 1)
+				obs.OnRepoSkipped(repo.GetRepoIdentifier(), "fork")
 				continue
 			}
 			if repo.GetSize() == 0 {
-				bar.ChangeMax(repoBatch.TotalCount - 1)
-				log.Info().Str("repo", repo.GetRepoIdentifier()).Msg("Skipping empty repository")
+				obs.OnRepoSkipped(repo.GetRepoIdentifier(), "empty")
 				continue
 			}
 			if err := goRoutineLimitSem.Acquire(ctx, 1); err != nil {
@@ -166,22 +173,26 @@ func (a *Analyzer) AnalyzeOrg(ctx context.Context, org string, numberOfGoroutine
 				defer goRoutineLimitSem.Release(1)
 				defer reposWg.Done()
 				repoNameWithOwner := repo.GetRepoIdentifier()
+				obs.OnRepoStarted(repoNameWithOwner)
 				repoKey, err := a.cloneRepo(ctx, repo.BuildGitURL(a.ScmClient.GetProviderBaseURL()), a.ScmClient.GetToken(), "HEAD")
 				if err != nil {
 					log.Error().Err(err).Str("repo", repoNameWithOwner).Msg("failed to clone repo")
+					obs.OnRepoError(repoNameWithOwner, err)
 					return
 				}
 				defer a.GitClient.Cleanup(repoKey)
 
 				pkg, err := a.GeneratePackageInsights(ctx, repoKey, repo, "HEAD")
 				if err != nil {
 					log.Error().Err(err).Str("repo", repoNameWithOwner).Msg("failed to generate package insights")
+					obs.OnRepoError(repoNameWithOwner, err)
 					return
 				}
 
 				files, err := a.GitClient.ListFiles(repoKey, []string{".yml", ".yaml"})
 				if err != nil {
 					log.Error().Err(err).Str("repo", repoNameWithOwner).Msg("failed to list files")
+					obs.OnRepoError(repoNameWithOwner, err)
 					return
 				}
 
@@ -199,6 +210,7 @@ func (a *Analyzer) AnalyzeOrg(ctx context.Context, org string, numberOfGoroutine
 				scannedPkg, err := inventory.ScanPackageScanner(ctx, *pkg, memScanner)
 				if err != nil {
 					log.Error().Err(err).Str("repo", repoNameWithOwner).Msg("failed to scan package")
+					obs.OnRepoError(repoNameWithOwner, err)
 					return
 				}
 
@@ -208,7 +220,7 @@ func (a *Analyzer) AnalyzeOrg(ctx context.Context, org string, numberOfGoroutine
 					log.Error().Msg("Context canceled while sending package to channel")
 					return
 				}
-				_ = bar.Add(1)
+				obs.OnRepoCompleted(repoNameWithOwner, scannedPkg)
 			}(repo)
 		}
 	}
@@ -227,12 +239,13 @@ func (a *Analyzer) AnalyzeOrg(ctx context.Context, org string, numberOfGoroutine
 		}
 	}
 
-	_ = bar.Finish()
-
+	obs.OnFinalizeStarted(len(scannedPackages))
 	err = a.finalizeAnalysis(ctx, scannedPackages)
 	if err != nil {
+		obs.OnFinalizeCompleted()
 		return scannedPackages, err
 	}
+	obs.OnFinalizeCompleted()
 
 	return scannedPackages, nil
 }
@@ -259,10 +272,12 @@ func (a *Analyzer) AnalyzeStaleBranches(ctx context.Context, repoString string,
 
 	inventory := scanner.NewInventory(a.Opa, pkgsupplyClient, provider, providerVersion)
 
+	obs := a.observer()
 	log.Debug().Msgf("Starting repository analysis for: %s/%s on %s", org, repoName, provider)
 	bar := a.ProgressBar(3, "Cloning repository")
 	_ = bar.RenderBlank()
 
+	obs.OnRepoStarted(repoString)
 	repoUrl := repo.BuildGitURL(a.ScmClient.GetProviderBaseURL())
 	repoKey, err := a.fetchCone(ctx, repoUrl, a.ScmClient.GetToken(), "refs/heads/*:refs/remotes/origin/*", ".github/workflows")
 	if err != nil {
@@ -363,6 +378,9 @@ func (a *Analyzer) AnalyzeStaleBranches(ctx context.Context, repoString string,
 	}
 
 	_ = bar.Finish()
+	obs.OnRepoCompleted(repoString, scannedPackage)
+
+	obs.OnFinalizeStarted(1)
 	if *expand {
 		expanded := []results.Finding{}
 		for _, finding := range scannedPackage.FindingsResults.Findings {
@@ -385,6 +403,7 @@ func (a *Analyzer) AnalyzeStaleBranches(ctx context.Context, repoString string,
 		scannedPackage.FindingsResults.Findings = expanded
 
 		if err := a.Formatter.Format(ctx, []*models.PackageInsights{scannedPackage}); err != nil {
+			obs.OnFinalizeCompleted()
 			return nil, fmt.Errorf("failed to finalize analysis of package: %w", err)
 		}
 	} else {
@@ -398,10 +417,11 @@ func (a *Analyzer) AnalyzeStaleBranches(ctx context.Context, repoString string,
 		}
 
 		if err := a.Formatter.FormatWithPath(ctx, []*models.PackageInsights{scannedPackage}, results); err != nil {
+			obs.OnFinalizeCompleted()
 			return nil, fmt.Errorf("failed to finalize analysis of package: %w", err)
 		}
-
 	}
+	obs.OnFinalizeCompleted()
 
 	return scannedPackage, nil
 }
@@ -427,10 +447,12 @@ func (a *Analyzer) AnalyzeRepo(ctx context.Context, repoString string, ref strin
 	pkgsupplyClient := pkgsupply.NewStaticClient()
 	inventory := scanner.NewInventory(a.Opa, pkgsupplyClient, provider, providerVersion)
 
+	obs := a.observer()
 	log.Debug().Msgf("Starting repository analysis for: %s/%s on %s", org, repoName, provider)
 	bar := a.ProgressBar(2, "Cloning repository")
 	_ = bar.RenderBlank()
 
+	obs.OnRepoStarted(repoString)
 	repoKey, err := a.cloneRepo(ctx, repo.BuildGitURL(a.ScmClient.GetProviderBaseURL()), a.ScmClient.GetToken(), ref)
 	if err != nil {
 		return nil, err
@@ -466,11 +488,15 @@ func (a *Analyzer) AnalyzeRepo(ctx context.Context, repoString string, ref strin
 		return nil, err
 	}
 	_ = bar.Finish()
+	obs.OnRepoCompleted(repoString, scannedPackage)
 
+	obs.OnFinalizeStarted(1)
 	err = a.finalizeAnalysis(ctx, []*models.PackageInsights{scannedPackage})
 	if err != nil {
+		obs.OnFinalizeCompleted()
 		return nil, err
 	}
+	obs.OnFinalizeCompleted()
 
 	return scannedPackage, nil
 }

analyze/analyze_test.go

@@ -2,8 +2,10 @@ package analyze
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/boostsecurityio/poutine/formatters/noop"
@@ -188,3 +190,119 @@ jobs:
 		require.NotNil(t, result)
 	})
 }
+
+// mockObserver records observer events for testing.
+type mockObserver struct {
+	mu     sync.Mutex
+	events []string
+}
+
+func (m *mockObserver) OnDiscoveryCompleted(org string, totalCount int) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.events = append(m.events, fmt.Sprintf("discovery_completed:%s:%d", org, totalCount))
+}
+func (m *mockObserver) OnRepoStarted(repo string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.events = append(m.events, "repo_started:"+repo)
+}
+func (m *mockObserver) OnRepoCompleted(repo string, _ *models.PackageInsights) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.events = append(m.events, "repo_completed:"+repo)
+}
+func (m *mockObserver) OnRepoError(repo string, _ error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.events = append(m.events, "repo_error:"+repo)
+}
+func (m *mockObserver) OnRepoSkipped(repo string, reason string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.events = append(m.events, fmt.Sprintf("repo_skipped:%s:%s", repo, reason))
+}
+func (m *mockObserver) OnFinalizeStarted(total int) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.events = append(m.events, fmt.Sprintf("finalize_started:%d", total))
+}
+func (m *mockObserver) OnFinalizeCompleted() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.events = append(m.events, "finalize_completed")
+}
+
+func TestProgressObserverNilSafe(t *testing.T) {
+	ctx := context.Background()
+	opaClient, err := newTestOpa(ctx)
+	require.NoError(t, err)
+
+	// Observer is nil — should not panic
+	analyzer := NewAnalyzer(nil, nil, &noop.Format{}, models.DefaultConfig(), opaClient)
+	assert.Nil(t, analyzer.Observer)
+
+	// AnalyzeManifest doesn't use observer, but this ensures nil Observer doesn't crash
+	_, err = analyzer.AnalyzeManifest(ctx, strings.NewReader("on: push\njobs:\n  test:\n    runs-on: ubuntu-latest\n    steps:\n      - uses: actions/checkout@v4"), "github-actions")
+	require.NoError(t, err)
+}
+
+func TestProgressObserverInterface(t *testing.T) {
+	obs := &mockObserver{}
+
+	// Verify the interface is implemented
+	var _ ProgressObserver = obs
+	var _ ProgressObserver = &noopObserver{}
+	var _ ProgressObserver = &ProgressBarObserver{}
+
+	// Test mock records events
+	obs.OnDiscoveryCompleted("test-org", 10)
+	obs.OnRepoStarted("test-org/repo1")
+	obs.OnRepoCompleted("test-org/repo1", nil)
+	obs.OnRepoSkipped("test-org/repo2", "fork")
+	obs.OnRepoError("test-org/repo3", errors.New("clone failed"))
+	obs.OnFinalizeStarted(1)
+	obs.OnFinalizeCompleted()
+
+	assert.Equal(t, []string{
+		"discovery_completed:test-org:10",
+		"repo_started:test-org/repo1",
+		"repo_completed:test-org/repo1",
+		"repo_skipped:test-org/repo2:fork",
+		"repo_error:test-org/repo3",
+		"finalize_started:1",
+		"finalize_completed",
+	}, obs.events)
+}
+
+func TestProgressObserverConcurrency(t *testing.T) {
+	// Exercise the concurrent methods on both observer implementations
+	// to verify no data races. Run with: go test -race
+	observers := []ProgressObserver{
+		&mockObserver{},
+		NewProgressBarObserver(true),
+	}
+
+	for _, obs := range observers {
+		obs.OnDiscoveryCompleted("org", 100)
+
+		var wg sync.WaitGroup
+		for i := 0; i < 50; i++ {
+			wg.Add(1)
+			go func(n int) {
+				defer wg.Done()
+				repo := fmt.Sprintf("org/repo-%d", n)
+				obs.OnRepoStarted(repo)
+				if n%5 == 0 {
+					obs.OnRepoError(repo, errors.New("error"))
+				} else {
+					obs.OnRepoCompleted(repo, nil)
+				}
+			}(i)
+		}
+		wg.Wait()
+
+		obs.OnFinalizeStarted(40)
+		obs.OnFinalizeCompleted()
+	}
+}

analyze/observer.go

@@ -0,0 +1,68 @@
+package analyze
+
+import "github.com/boostsecurityio/poutine/models"
+
+// ProgressObserver receives structured progress events during analysis.
+// All methods must be non-blocking — implementations should not perform
+// expensive I/O or hold locks for extended periods.
+//
+// # Concurrency
+//
+// During AnalyzeOrg, methods are called from two contexts:
+//
+// Main goroutine (sequential, never concurrent with each other):
+//   - OnDiscoveryCompleted
+//   - OnRepoSkipped
+//   - OnFinalizeStarted
+//   - OnFinalizeCompleted
+//
+// Worker goroutines (called concurrently from multiple goroutines):
+//   - OnRepoStarted
+//   - OnRepoCompleted
+//   - OnRepoError
+//
+// Implementations MUST ensure that the worker-goroutine methods are
+// goroutine-safe. Note that worker methods may also run concurrently
+// with the main-goroutine methods above.
+//
+// During AnalyzeRepo and AnalyzeStaleBranches all methods are called
+// from a single goroutine.
+type ProgressObserver interface {
+	// OnDiscoveryCompleted is called from the main goroutine when the
+	// total repo count is known (first batch with TotalCount > 0).
+	OnDiscoveryCompleted(org string, totalCount int)
+
+	// OnRepoStarted is called from a worker goroutine when analysis
+	// of a repo begins. Must be goroutine-safe.
+	OnRepoStarted(repo string)
+
+	// OnRepoCompleted is called from a worker goroutine when a repo
+	// finishes successfully. Must be goroutine-safe.
+	OnRepoCompleted(repo string, pkg *models.PackageInsights)
+
+	// OnRepoError is called from a worker goroutine when a repo
+	// analysis fails (non-fatal). Must be goroutine-safe.
+	OnRepoError(repo string, err error)
+
+	// OnRepoSkipped is called from the main goroutine when a repo is
+	// skipped (fork, empty, etc.).
+	OnRepoSkipped(repo string, reason string)
+
+	// OnFinalizeStarted is called from the main goroutine when the
+	// formatting/output phase begins, after all repos are processed.
+	OnFinalizeStarted(totalPackages int)
+
+	// OnFinalizeCompleted is called from the main goroutine when
+	// formatting is done.
+	OnFinalizeCompleted()
+}
+
+type noopObserver struct{}
+
+func (noopObserver) OnDiscoveryCompleted(string, int)                {}
+func (noopObserver) OnRepoStarted(string)                            {}
+func (noopObserver) OnRepoCompleted(string, *models.PackageInsights) {}
+func (noopObserver) OnRepoError(string, error)                       {}
+func (noopObserver) OnRepoSkipped(string, string)                    {}
+func (noopObserver) OnFinalizeStarted(int)                           {}
+func (noopObserver) OnFinalizeCompleted()                            {}