Split UI from event handling (#448)

* split UI from event handling

Signed-off-by: Alex Goodman <wagoodman@gmail.com>

* add event loop tests

Signed-off-by: Alex Goodman <wagoodman@gmail.com>

* use stereoscope cleanup function during signal handling

Signed-off-by: Alex Goodman <alex.goodman@anchore.com>

* correct error wrapping in packages cmd

Signed-off-by: Alex Goodman <alex.goodman@anchore.com>

* migrate ui event handlers to ui package

Signed-off-by: Alex Goodman <alex.goodman@anchore.com>

* clarify command worker input var + remove dead comments

Signed-off-by: Alex Goodman <alex.goodman@anchore.com>

cmd/event_loop.go

@@ -0,0 +1,90 @@
+package cmd
+
+import (
+	"errors"
+	"os"
+
+	"github.com/anchore/syft/internal/log"
+	"github.com/anchore/syft/internal/ui"
+	"github.com/hashicorp/go-multierror"
+	"github.com/wagoodman/go-partybus"
+)
+
+// eventLoop listens to worker errors (from execution path), worker events (from a partybus subscription), and
+// signal interrupts. Is responsible for handling each event relative to a given UI an to coordinate eventing until
+// an eventual graceful exit.
+// nolint:gocognit
+func eventLoop(workerErrs <-chan error, signals <-chan os.Signal, subscription *partybus.Subscription, ux ui.UI, cleanupFn func()) error {
+	defer cleanupFn()
+	events := subscription.Events()
+	var err error
+	if ux, err = setupUI(subscription.Unsubscribe, ux); err != nil {
+		return err
+	}
+
+	var retErr error
+
+	for {
+		if workerErrs == nil && events == nil {
+			break
+		}
+		select {
+		case err, isOpen := <-workerErrs:
+			if !isOpen {
+				workerErrs = nil
+				continue
+			}
+			if err != nil {
+				// capture the error from the worker and unsubscribe to complete a graceful shutdown
+				retErr = multierror.Append(retErr, err)
+				if err := subscription.Unsubscribe(); err != nil {
+					retErr = multierror.Append(retErr, err)
+				}
+			}
+		case e, isOpen := <-events:
+			if !isOpen {
+				events = nil
+				continue
+			}
+
+			if err := ux.Handle(e); err != nil {
+				if errors.Is(err, partybus.ErrUnsubscribe) {
+					log.Warnf("unable to unsubscribe from the event bus")
+					events = nil
+				} else {
+					retErr = multierror.Append(retErr, err)
+					// TODO: should we unsubscribe? should we try to halt execution? or continue?
+				}
+			}
+		case <-signals:
+			// ignore further results from any event source and exit ASAP, but ensure that all cache is cleaned up.
+			// we ignore further errors since cleaning up the tmp directories will affect running catalogers that are
+			// reading/writing from/to their nested temp dirs. This is acceptable since we are bailing without result.
+
+			// TODO: potential future improvement would be to pass context into workers with a cancel function that is
+			// to the event loop. In this way we can have a more controlled shutdown even at the most nested levels
+			// of processing.
+			events = nil
+			workerErrs = nil
+		}
+	}
+
+	if err := ux.Teardown(); err != nil {
+		retErr = multierror.Append(retErr, err)
+	}
+
+	return retErr
+}
+
+func setupUI(unsubscribe func() error, ux ui.UI) (ui.UI, error) {
+	if err := ux.Setup(unsubscribe); err != nil {
+		// replace the existing UI with a (simpler) logger UI
+		ux = ui.NewLoggerUI()
+		if err := ux.Setup(unsubscribe); err != nil {
+			// something is very wrong, bail.
+			return ux, err
+		}
+		log.Errorf("unable to setup given UI, falling back to logger: %+v", err)
+	}
+	return ux, nil
+}

cmd/event_loop_test.go

@@ -0,0 +1,455 @@
+package cmd
+
+import (
+	"fmt"
+	"os"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/anchore/syft/internal/ui"
+	"github.com/anchore/syft/syft/event"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/wagoodman/go-partybus"
+)
+
+var _ ui.UI = (*uiMock)(nil)
+
+type uiMock struct {
+	t           *testing.T
+	finalEvent  partybus.Event
+	unsubscribe func() error
+	mock.Mock
+}
+
+func (u *uiMock) Setup(unsubscribe func() error) error {
+	u.t.Logf("UI Setup called")
+	u.unsubscribe = unsubscribe
+	return u.Called(unsubscribe).Error(0)
+}
+
+func (u *uiMock) Handle(event partybus.Event) error {
+	u.t.Logf("UI Handle called: %+v", event.Type)
+	if event == u.finalEvent {
+		assert.NoError(u.t, u.unsubscribe())
+	}
+	return u.Called(event).Error(0)
+}
+
+func (u *uiMock) Teardown() error {
+	u.t.Logf("UI Teardown called")
+	return u.Called().Error(0)
+}
+
+func Test_eventLoop_gracefulExit(t *testing.T) {
+	test := func(t *testing.T) {
+
+		testBus := partybus.NewBus()
+		subscription := testBus.Subscribe()
+		t.Cleanup(testBus.Close)
+
+		finalEvent := partybus.Event{
+			Type: event.PresenterReady,
+		}
+
+		worker := func() <-chan error {
+			ret := make(chan error)
+			go func() {
+				t.Log("worker running")
+				// send an empty item (which is ignored) ensuring we've entered the select statement,
+				// then close (a partial shutdown).
+				ret <- nil
+				t.Log("worker sent nothing")
+				close(ret)
+				t.Log("worker closed")
+				// do the other half of the shutdown
+				testBus.Publish(finalEvent)
+				t.Log("worker published final event")
+			}()
+			return ret
+		}
+
+		signaler := func() <-chan os.Signal {
+			return nil
+		}
+
+		ux := &uiMock{
+			t:          t,
+			finalEvent: finalEvent,
+		}
+
+		// ensure the mock sees at least the final event
+		ux.On("Handle", finalEvent).Return(nil)
+		// ensure the mock sees basic setup/teardown events
+		ux.On("Setup", mock.AnythingOfType("func() error")).Return(nil)
+		ux.On("Teardown").Return(nil)
+
+		var cleanupCalled bool
+		cleanupFn := func() {
+			t.Log("cleanup called")
+			cleanupCalled = true
+		}
+
+		assert.NoError(t,
+			eventLoop(
+				worker(),
+				signaler(),
+				subscription,
+				ux,
+				cleanupFn,
+			),
+		)
+
+		assert.True(t, cleanupCalled, "cleanup function not called")
+		ux.AssertExpectations(t)
+	}
+
+	// if there is a bug, then there is a risk of the event loop never returning
+	testWithTimeout(t, 5*time.Second, test)
+}
+
+func Test_eventLoop_workerError(t *testing.T) {
+	test := func(t *testing.T) {
+
+		testBus := partybus.NewBus()
+		subscription := testBus.Subscribe()
+		t.Cleanup(testBus.Close)
+
+		workerErr := fmt.Errorf("worker error")
+
+		worker := func() <-chan error {
+			ret := make(chan error)
+			go func() {
+				t.Log("worker running")
+				// send an empty item (which is ignored) ensuring we've entered the select statement,
+				// then close (a partial shutdown).
+				ret <- nil
+				t.Log("worker sent nothing")
+				ret <- workerErr
+				t.Log("worker sent error")
+				close(ret)
+				t.Log("worker closed")
+				// note: NO final event is fired
+			}()
+			return ret
+		}
+
+		signaler := func() <-chan os.Signal {
+			return nil
+		}
+
+		ux := &uiMock{
+			t: t,
+		}
+
+		// ensure the mock sees basic setup/teardown events
+		ux.On("Setup", mock.AnythingOfType("func() error")).Return(nil)
+		ux.On("Teardown").Return(nil)
+
+		var cleanupCalled bool
+		cleanupFn := func() {
+			t.Log("cleanup called")
+			cleanupCalled = true
+		}
+
+		// ensure we see an error returned
+		assert.ErrorIs(t,
+			eventLoop(
+				worker(),
+				signaler(),
+				subscription,
+				ux,
+				cleanupFn,
+			),
+			workerErr,
+			"should have seen a worker error, but did not",
+		)
+
+		assert.True(t, cleanupCalled, "cleanup function not called")
+		ux.AssertExpectations(t)
+	}
+
+	// if there is a bug, then there is a risk of the event loop never returning
+	testWithTimeout(t, 5*time.Second, test)
+}
+
+func Test_eventLoop_unsubscribeError(t *testing.T) {
+	test := func(t *testing.T) {
+
+		testBus := partybus.NewBus()
+		subscription := testBus.Subscribe()
+		t.Cleanup(testBus.Close)
+
+		finalEvent := partybus.Event{
+			Type: event.PresenterReady,
+		}
+
+		worker := func() <-chan error {
+			ret := make(chan error)
+			go func() {
+				t.Log("worker running")
+				// send an empty item (which is ignored) ensuring we've entered the select statement,
+				// then close (a partial shutdown).
+				ret <- nil
+				t.Log("worker sent nothing")
+				close(ret)
+				t.Log("worker closed")
+				// do the other half of the shutdown
+				testBus.Publish(finalEvent)
+				t.Log("worker published final event")
+			}()
+			return ret
+		}
+
+		signaler := func() <-chan os.Signal {
+			return nil
+		}
+
+		ux := &uiMock{
+			t:          t,
+			finalEvent: finalEvent,
+		}
+
+		// ensure the mock sees at least the final event... note the unsubscribe error here
+		ux.On("Handle", finalEvent).Return(partybus.ErrUnsubscribe)
+		// ensure the mock sees basic setup/teardown events
+		ux.On("Setup", mock.AnythingOfType("func() error")).Return(nil)
+		ux.On("Teardown").Return(nil)
+
+		var cleanupCalled bool
+		cleanupFn := func() {
+			t.Log("cleanup called")
+			cleanupCalled = true
+		}
+
+		// unsubscribe errors should be handled and ignored, not propagated. We are additionally asserting that
+		// this case is handled as a controlled shutdown (this test should not timeout)
+		assert.NoError(t,
+			eventLoop(
+				worker(),
+				signaler(),
+				subscription,
+				ux,
+				cleanupFn,
+			),
+		)
+
+		assert.True(t, cleanupCalled, "cleanup function not called")
+		ux.AssertExpectations(t)
+	}
+
+	// if there is a bug, then there is a risk of the event loop never returning
+	testWithTimeout(t, 5*time.Second, test)
+}
+
+func Test_eventLoop_handlerError(t *testing.T) {
+	test := func(t *testing.T) {
+
+		testBus := partybus.NewBus()
+		subscription := testBus.Subscribe()
+		t.Cleanup(testBus.Close)
+
+		finalEvent := partybus.Event{
+			Type:  event.PresenterReady,
+			Error: fmt.Errorf("unable to create presenter"),
+		}
+
+		worker := func() <-chan error {
+			ret := make(chan error)
+			go func() {
+				t.Log("worker running")
+				// send an empty item (which is ignored) ensuring we've entered the select statement,
+				// then close (a partial shutdown).
+				ret <- nil
+				t.Log("worker sent nothing")
+				close(ret)
+				t.Log("worker closed")
+				// do the other half of the shutdown
+				testBus.Publish(finalEvent)
+				t.Log("worker published final event")
+			}()
+			return ret
+		}
+
+		signaler := func() <-chan os.Signal {
+			return nil
+		}
+
+		ux := &uiMock{
+			t:          t,
+			finalEvent: finalEvent,
+		}
+
+		// ensure the mock sees at least the final event... note the event error is propagated
+		ux.On("Handle", finalEvent).Return(finalEvent.Error)
+		// ensure the mock sees basic setup/teardown events
+		ux.On("Setup", mock.AnythingOfType("func() error")).Return(nil)
+		ux.On("Teardown").Return(nil)
+
+		var cleanupCalled bool
+		cleanupFn := func() {
+			t.Log("cleanup called")
+			cleanupCalled = true
+		}
+
+		// handle errors SHOULD propagate the event loop. We are additionally asserting that this case is
+		// handled as a controlled shutdown (this test should not timeout)
+		assert.ErrorIs(t,
+			eventLoop(
+				worker(),
+				signaler(),
+				subscription,
+				ux,
+				cleanupFn,
+			),
+			finalEvent.Error,
+			"should have seen a event error, but did not",
+		)
+
+		assert.True(t, cleanupCalled, "cleanup function not called")
+		ux.AssertExpectations(t)
+	}
+
+	// if there is a bug, then there is a risk of the event loop never returning
+	testWithTimeout(t, 5*time.Second, test)
+}
+
+func Test_eventLoop_signalsStopExecution(t *testing.T) {
+	test := func(t *testing.T) {
+
+		testBus := partybus.NewBus()
+		subscription := testBus.Subscribe()
+		t.Cleanup(testBus.Close)
+
+		worker := func() <-chan error {
+			// the worker will never return work and the event loop will always be waiting...
+			return make(chan error)
+		}
+
+		signaler := func() <-chan os.Signal {
+			ret := make(chan os.Signal)
+			go func() {
+				ret <- syscall.SIGINT
+				// note: we do NOT close the channel to ensure the event loop does not depend on that behavior to exit
+			}()
+			return ret
+		}
+
+		ux := &uiMock{
+			t: t,
+		}
+
+		// ensure the mock sees basic setup/teardown events
+		ux.On("Setup", mock.AnythingOfType("func() error")).Return(nil)
+		ux.On("Teardown").Return(nil)
+
+		var cleanupCalled bool
+		cleanupFn := func() {
+			t.Log("cleanup called")
+			cleanupCalled = true
+		}
+
+		assert.NoError(t,
+			eventLoop(
+				worker(),
+				signaler(),
+				subscription,
+				ux,
+				cleanupFn,
+			),
+		)
+
+		assert.True(t, cleanupCalled, "cleanup function not called")
+		ux.AssertExpectations(t)
+	}
+
+	// if there is a bug, then there is a risk of the event loop never returning
+	testWithTimeout(t, 5*time.Second, test)
+}
+
+func Test_eventLoop_uiTeardownError(t *testing.T) {
+	test := func(t *testing.T) {
+
+		testBus := partybus.NewBus()
+		subscription := testBus.Subscribe()
+		t.Cleanup(testBus.Close)
+
+		finalEvent := partybus.Event{
+			Type: event.PresenterReady,
+		}
+
+		worker := func() <-chan error {
+			ret := make(chan error)
+			go func() {
+				t.Log("worker running")
+				// send an empty item (which is ignored) ensuring we've entered the select statement,
+				// then close (a partial shutdown).
+				ret <- nil
+				t.Log("worker sent nothing")
+				close(ret)
+				t.Log("worker closed")
+				// do the other half of the shutdown
+				testBus.Publish(finalEvent)
+				t.Log("worker published final event")
+			}()
+			return ret
+		}
+
+		signaler := func() <-chan os.Signal {
+			return nil
+		}
+
+		ux := &uiMock{
+			t:          t,
+			finalEvent: finalEvent,
+		}
+
+		teardownError := fmt.Errorf("sorry, dave, the UI doesn't want to be torn down")
+
+		// ensure the mock sees at least the final event... note the event error is propagated
+		ux.On("Handle", finalEvent).Return(nil)
+		// ensure the mock sees basic setup/teardown events
+		ux.On("Setup", mock.AnythingOfType("func() error")).Return(nil)
+		ux.On("Teardown").Return(teardownError)
+
+		var cleanupCalled bool
+		cleanupFn := func() {
+			t.Log("cleanup called")
+			cleanupCalled = true
+		}
+
+		// ensure we see an error returned
+		assert.ErrorIs(t,
+			eventLoop(
+				worker(),
+				signaler(),
+				subscription,
+				ux,
+				cleanupFn,
+			),
+			teardownError,
+			"should have seen a UI teardown error, but did not",
+		)
+
+		assert.True(t, cleanupCalled, "cleanup function not called")
+		ux.AssertExpectations(t)
+	}
+
+	// if there is a bug, then there is a risk of the event loop never returning
+	testWithTimeout(t, 5*time.Second, test)
+}
+
+func testWithTimeout(t *testing.T, timeout time.Duration, test func(*testing.T)) {
+	done := make(chan bool)
+	go func() {
+		test(t)
+		done <- true
+	}()
+
+	select {
+	case <-time.After(timeout):
+		t.Fatal("test timed out")
+	case <-done:
+	}
+}

cmd/packages.go

@@ -6,10 +6,7 @@ import (
 	"io/ioutil"
 	"os"
 
-	"github.com/anchore/syft/syft/presenter/packages"
-
-	"github.com/spf13/viper"
-
+	"github.com/anchore/stereoscope"
 	"github.com/anchore/syft/internal"
 	"github.com/anchore/syft/internal/anchore"
 	"github.com/anchore/syft/internal/bus"
@@ -19,10 +16,12 @@ import (
 	"github.com/anchore/syft/syft/distro"
 	"github.com/anchore/syft/syft/event"
 	"github.com/anchore/syft/syft/pkg"
+	"github.com/anchore/syft/syft/presenter/packages"
 	"github.com/anchore/syft/syft/source"
 	"github.com/pkg/profile"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
+	"github.com/spf13/viper"
 	"github.com/wagoodman/go-partybus"
 )
 
@@ -186,9 +185,15 @@ func bindPackagesConfigOptions(flags *pflag.FlagSet) error {
 }
 
 func packagesExec(_ *cobra.Command, args []string) error {
-	errs := packagesExecWorker(args[0])
-	ux := ui.Select(appConfig.CliOptions.Verbosity > 0, appConfig.Quiet)
-	return ux(errs, eventSubscription)
+	// could be an image or a directory, with or without a scheme
+	userInput := args[0]
+	return eventLoop(
+		packagesExecWorker(userInput),
+		setupSignals(),
+		eventSubscription,
+		ui.Select(appConfig.CliOptions.Verbosity > 0, appConfig.Quiet),
+		stereoscope.Cleanup,
+	)
 }
 
 func packagesExecWorker(userInput string) <-chan error {
@@ -200,14 +205,14 @@ func packagesExecWorker(userInput string) <-chan error {
 
 		src, cleanup, err := source.New(userInput, appConfig.Registry.ToOptions())
 		if err != nil {
-			errs <- fmt.Errorf("failed to determine image source: %+v", err)
+			errs <- fmt.Errorf("failed to determine image source: %w", err)
 			return
 		}
 		defer cleanup()
 
 		catalog, d, err := syft.CatalogPackages(src, appConfig.Package.Cataloger.ScopeOpt)
 		if err != nil {
-			errs <- fmt.Errorf("failed to catalog input: %+v", err)
+			errs <- fmt.Errorf("failed to catalog input: %w", err)
 			return
 		}
 
@@ -261,7 +266,7 @@ func runPackageSbomUpload(src source.Source, s source.Metadata, catalog *pkg.Cat
 		Password: appConfig.Anchore.Password,
 	})
 	if err != nil {
-		return fmt.Errorf("failed to create anchore client: %+v", err)
+		return fmt.Errorf("failed to create anchore client: %w", err)
 	}
 
 	importCfg := anchore.ImportConfig{

cmd/power_user.go

@@ -4,8 +4,8 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/anchore/stereoscope"
 	"github.com/anchore/syft/internal"
-
 	"github.com/anchore/syft/internal/bus"
 	"github.com/anchore/syft/internal/presenter/poweruser"
 	"github.com/anchore/syft/internal/ui"
@@ -63,9 +63,15 @@ func init() {
 }
 
 func powerUserExec(_ *cobra.Command, args []string) error {
-	errs := powerUserExecWorker(args[0])
-	ux := ui.Select(appConfig.CliOptions.Verbosity > 0, appConfig.Quiet)
-	return ux(errs, eventSubscription)
+	// could be an image or a directory, with or without a scheme
+	userInput := args[0]
+	return eventLoop(
+		powerUserExecWorker(userInput),
+		setupSignals(),
+		eventSubscription,
+		ui.Select(appConfig.CliOptions.Verbosity > 0, appConfig.Quiet),
+		stereoscope.Cleanup,
+	)
 }
 
 func powerUserExecWorker(userInput string) <-chan error {

cmd/signals.go

@@ -0,0 +1,20 @@
+package cmd
+
+import (
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func setupSignals() <-chan os.Signal {
+	c := make(chan os.Signal, 1) // Note: A buffered channel is recommended for this; see https://golang.org/pkg/os/signal/#Notify
+
+	interruptions := []os.Signal{
+		syscall.SIGINT,
+		syscall.SIGTERM,
+	}
+
+	signal.Notify(c, interruptions...)
+
+	return c
+}