Add panic recovers

cmd/conduit/main.go

@@ -50,6 +50,7 @@ func init() {
 
 // runConduitCmdWithConfig run the main logic with a supplied conduit config
 func runConduitCmdWithConfig(cfg *conduit.Config) error {
+	defer conduit.HandlePanic(logger)
 
 	// From docs:
 	// BindEnv takes one or more parameters. The first parameter is the key name, the rest are the name of the

cmd/conduit/main_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// TestInitDataDirectory tests the initialization of the data directory
 func TestInitDataDirectory(t *testing.T) {
 	verifyFile := func(file string) {
 		require.FileExists(t, file)

conduit/common.go

@@ -0,0 +1,10 @@
+package conduit
+
+import log "github.com/sirupsen/logrus"
+
+// HandlePanic function to log panics in a common way
+func HandlePanic(logger *log.Logger) {
+	if r := recover(); r != nil {
+		logger.Panicf("conduit pipeline experienced a panic: %v", r)
+	}
+}

conduit/pipeline.go

@@ -292,6 +292,8 @@ func (p *pipelineImpl) Start() {
 	p.wg.Add(1)
 	go func() {
 		defer p.wg.Done()
+		// We need to add a separate recover function here since it launches its own go-routine
+		defer HandlePanic(p.logger)
 		for {
 		pipelineRun:
 			select {