Merge pull request #1052 from unpoller/fix-1030-input-panic-crash

Recover input plugin panics in poller goroutines (fixes #1030)

pkg/poller/inputs.go

@@ -65,6 +65,21 @@ func NewInput(i *InputPlugin) {
 	inputs = append(inputs, i)
 }
 
+// recoverInitialize runs input.Initialize and converts a panic into an error.
+// Each input plugin's Initialize call runs in its own goroutine, so a panic
+// there cannot be caught by a recover() in the caller; left unrecovered, it
+// crashes the whole process before the app even starts.
+// See https://github.com/unpoller/unpoller/issues/1030
+func recoverInitialize(input *InputPlugin, l Logger) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("input plugin %s panicked initializing: %v", input.Name, r) //nolint:err113
+		}
+	}()
+
+	return input.Initialize(l)
+}
+
 // InitializeInputs runs the passed-in initializer method for each input plugin.
 func (u *UnifiPoller) InitializeInputs() error {
 	inputSync.RLock()
@@ -81,10 +96,11 @@ func (u *UnifiPoller) InitializeInputs() error {
 
 		go func(input *InputPlugin) {
 			defer wg.Done()
+
 			// This must return, or the app locks up here.
 			u.LogDebugf("inititalizing input... %s", input.Name)
 
-			if err := input.Initialize(u); err != nil {
+			if err := recoverInitialize(input, u); err != nil {
 				u.LogDebugf("error initializing input ... %s", input.Name)
 
 				errChan <- err
@@ -130,6 +146,18 @@ type eventInputResult struct {
 	err  error
 }
 
+// recoverEvents runs input.Events and converts a panic into an error. See
+// recoverInitialize for why this is necessary.
+func recoverEvents(input *InputPlugin, filter *Filter) (e *Events, err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("input plugin %s panicked collecting events: %v", input.Name, r) //nolint:err113
+		}
+	}()
+
+	return input.Events(filter)
+}
+
 func collectEvents(filter *Filter, inputs []*InputPlugin) (*Events, error) {
 	resultChan := make(chan eventInputResult, len(inputs))
 	wg := &sync.WaitGroup{}
@@ -148,13 +176,19 @@ func collectEvents(filter *Filter, inputs []*InputPlugin) (*Events, error) {
 				return
 			}
 
-			e, err := input.Events(filter)
+			e, err := recoverEvents(input, filter)
 			if err != nil {
 				resultChan <- eventInputResult{err: err}
 
 				return
 			}
 
+			if e == nil {
+				resultChan <- eventInputResult{}
+
+				return
+			}
+
 			resultChan <- eventInputResult{logs: e.Logs}
 		}(input)
 	}
@@ -199,6 +233,20 @@ type metricInputResult struct {
 	err    error
 }
 
+// recoverMetrics runs input.Metrics and converts a panic into an error (e.g.
+// from a malformed controller response, such as an unexpected Site Speed
+// Test aggregated-dashboard payload). See recoverInitialize for why this is
+// necessary.
+func recoverMetrics(input *InputPlugin, filter *Filter) (m *Metrics, err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("input plugin %s panicked collecting metrics: %v", input.Name, r) //nolint:err113
+		}
+	}()
+
+	return input.Metrics(filter)
+}
+
 func collectMetrics(filter *Filter, inputs []*InputPlugin) (*Metrics, error) {
 	resultChan := make(chan metricInputResult, len(inputs))
 	wg := &sync.WaitGroup{}
@@ -217,7 +265,7 @@ func collectMetrics(filter *Filter, inputs []*InputPlugin) (*Metrics, error) {
 				return
 			}
 
-			m, err := input.Metrics(filter)
+			m, err := recoverMetrics(input, filter)
 			resultChan <- metricInputResult{metric: m, err: err}
 		}(input)
 	}

pkg/poller/inputs_test.go

@@ -0,0 +1,98 @@
+package poller_test
+
+import (
+	"testing"
+
+	"github.com/unpoller/unpoller/pkg/poller"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// panicInput is an Input that panics on Metrics and Events, simulating a
+// malformed controller response crashing an input plugin. See issue #1030.
+type panicInput struct{}
+
+func (panicInput) Initialize(poller.Logger) error { return nil }
+
+func (panicInput) Metrics(*poller.Filter) (*poller.Metrics, error) {
+	panic("simulated aggregated-dashboard panic")
+}
+
+func (panicInput) Events(*poller.Filter) (*poller.Events, error) {
+	panic("simulated aggregated-dashboard panic")
+}
+
+func (panicInput) RawMetrics(*poller.Filter) ([]byte, error) { return nil, nil }
+
+func (panicInput) DebugInput() (bool, error) { return false, nil }
+
+func TestCollectMetricsRecoversPanickingInput(t *testing.T) {
+	t.Parallel()
+
+	collector := poller.NewTestCollector(t)
+	collector.AddInput(&poller.InputPlugin{Name: "panic-input", Input: panicInput{}})
+
+	var metrics *poller.Metrics
+
+	var err error
+
+	require.NotPanics(t, func() {
+		metrics, err = collector.Metrics(nil)
+	})
+
+	assert.NotNil(t, metrics)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "panic-input")
+}
+
+// nilEventsInput simulates a disabled input plugin, which returns (nil, nil)
+// from Events. See https://github.com/unpoller/unpoller/issues/1030.
+type nilEventsInput struct{}
+
+func (nilEventsInput) Initialize(poller.Logger) error { return nil }
+
+func (nilEventsInput) Metrics(*poller.Filter) (*poller.Metrics, error) { return nil, nil }
+
+func (nilEventsInput) Events(*poller.Filter) (*poller.Events, error) { return nil, nil }
+
+func (nilEventsInput) RawMetrics(*poller.Filter) ([]byte, error) { return nil, nil }
+
+func (nilEventsInput) DebugInput() (bool, error) { return false, nil }
+
+func TestCollectEventsHandlesNilEventsResult(t *testing.T) {
+	t.Parallel()
+
+	collector := poller.NewTestCollector(t)
+	collector.AddInput(&poller.InputPlugin{Name: "nil-events-input", Input: nilEventsInput{}})
+
+	var events *poller.Events
+
+	var err error
+
+	require.NotPanics(t, func() {
+		events, err = collector.Events(nil)
+	})
+
+	assert.NotNil(t, events)
+	require.NoError(t, err)
+}
+
+func TestCollectEventsRecoversPanickingInput(t *testing.T) {
+	t.Parallel()
+
+	collector := poller.NewTestCollector(t)
+	collector.AddInput(&poller.InputPlugin{Name: "panic-input", Input: panicInput{}})
+
+	var events *poller.Events
+
+	var err error
+
+	require.NotPanics(t, func() {
+		events, err = collector.Events(nil)
+	})
+
+	assert.NotNil(t, events)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "panic-input")
+}