nozzle: crash on MaxRetriesReached

src/stackdriver-nozzle/mocks/logger.go

@@ -52,27 +52,35 @@ func (m *MockLogger) Debug(action string, data ...lager.Data) {
 
 func (m *MockLogger) Info(action string, data ...lager.Data) {
 	m.mutex.Lock()
+	defer m.mutex.Unlock()
 	m.logs = append(m.logs, Log{
 		Level:  lager.INFO,
 		Action: action,
 		Datas:  data,
 	})
-	m.mutex.Unlock()
+
 }
 
 func (m *MockLogger) Error(action string, err error, data ...lager.Data) {
 	m.mutex.Lock()
+	defer m.mutex.Unlock()
 	m.logs = append(m.logs, Log{
 		Level:  lager.ERROR,
 		Action: action,
 		Err:    err,
 		Datas:  data,
 	})
-	m.mutex.Unlock()
 }
 
 func (m *MockLogger) Fatal(action string, err error, data ...lager.Data) {
-	panic("NYI")
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	m.logs = append(m.logs, Log{
+		Level:  lager.FATAL,
+		Action: action,
+		Err:    err,
+		Datas:  data,
+	})
 }
 
 func (m *MockLogger) WithData(lager.Data) lager.Logger {

src/stackdriver-nozzle/nozzle/nozzle.go

@@ -24,6 +24,7 @@ import (
 	"github.com/cloudfoundry-community/stackdriver-tools/src/stackdriver-nozzle/cloudfoundry"
 	"github.com/cloudfoundry-community/stackdriver-tools/src/stackdriver-nozzle/heartbeat"
 	"github.com/cloudfoundry/lager"
+	"github.com/cloudfoundry/noaa/consumer"
 	"github.com/cloudfoundry/sonde-go/events"
 	"github.com/gorilla/websocket"
 )
@@ -134,7 +135,11 @@ func (n *nozzle) handleEvent(envelope *events.Envelope) {
 }
 
 func (n *nozzle) handleFirehoseError(err error) {
-	n.logger.Error("firehose", err)
+	if err == consumer.ErrMaxRetriesReached {
+		n.logger.Fatal("firehose", err)
+	} else {
+		n.logger.Error("firehose", err)
+	}
 
 	closeErr, ok := err.(*websocket.CloseError)
 	if !ok {

src/stackdriver-nozzle/nozzle/nozzle_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cloudfoundry-community/stackdriver-tools/src/stackdriver-nozzle/mocks"
 	"github.com/cloudfoundry-community/stackdriver-tools/src/stackdriver-nozzle/nozzle"
 	"github.com/cloudfoundry/lager"
+	"github.com/cloudfoundry/noaa/consumer"
 	"github.com/cloudfoundry/sonde-go/events"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
@@ -142,6 +143,16 @@ var _ = Describe("Nozzle", func() {
 		}))
 	})
 
+	It("crashes on unrecoverable firehose errors", func() {
+		err := consumer.ErrMaxRetriesReached
+		go func() { firehose.Errs <- err }()
+		Eventually(logger.Logs).Should(ContainElement(mocks.Log{
+			Level:  lager.FATAL,
+			Err:    err,
+			Action: "firehose",
+		}))
+	})
+
 	It("is resilient to multiple exists", func(done Done) {
 		Expect(subject.Stop()).NotTo(HaveOccurred())
 		Expect(subject.Stop()).To(HaveOccurred())