Adding Readiness Endpoint

core/internal/consumer/coordinator.go

@@ -106,6 +106,9 @@ func (cc *Coordinator) Start() error {
 	if err != nil {
 		return errors.New("Error starting consumer module: " + err.Error())
 	}
+	// All consumers started, Burrow is ready to serve requests
+	// set the readiness probe
+	cc.App.AppReady = true
 	return nil
 }
 

core/internal/httpserver/coordinator.go

@@ -126,8 +126,9 @@ func (hc *Coordinator) Configure() {
 	// This is a catchall for undefined URLs
 	hc.router.NotFound = &defaultHandler{}
 
-	// This is a healthcheck URL. Please don't change it
+	// This is a healthcheck and readiness URLs. Please don't change it
 	hc.router.GET("/burrow/admin", hc.handleAdmin)
+	hc.router.GET("/burrow/admin/ready", hc.handleReady)
 
 	hc.router.Handler(http.MethodGet, "/metrics", hc.handlePrometheusMetrics())
 
@@ -296,6 +297,24 @@ func (hc *Coordinator) handleAdmin(w http.ResponseWriter, r *http.Request, _ htt
 	w.Write([]byte("GOOD"))
 }
 
+// handleReady will use the AppReady bool from  the ApplicationContext to determine
+// whether Burrow is ready to serve requests
+func (hc *Coordinator) handleReady(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
+	// Add CORS header, if configured
+	corsHeader := viper.GetString("general.access-control-allow-origin")
+	if corsHeader != "" {
+		w.Header().Set("Access-Control-Allow-Origin", corsHeader)
+	}
+
+	if hc.App.AppReady {
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte("READY"))
+	} else {
+		w.WriteHeader(http.StatusServiceUnavailable)
+		w.Write([]byte("STARTING"))
+	}
+}
+
 func (hc *Coordinator) getLogLevel(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
 	requestInfo := makeRequestInfo(r)
 	hc.writeResponse(w, r, http.StatusOK, httpResponseLogLevel{

core/internal/httpserver/coordinator_test.go

@@ -35,6 +35,7 @@ func fixtureConfiguredCoordinator() *Coordinator {
 			LogLevel:         &logLevel,
 			StorageChannel:   make(chan *protocol.StorageRequest),
 			EvaluatorChannel: make(chan *protocol.EvaluatorRequest),
+			AppReady:         false,
 		},
 	}
 
@@ -58,6 +59,27 @@ func TestHttpServer_handleAdmin(t *testing.T) {
 	assert.Equalf(t, "GOOD", rr.Body.String(), "Expected response body to be 'GOOD', not '%v'", rr.Body.String())
 }
 
+func TestHttpServer_handleReady(t *testing.T) {
+	coordinator := fixtureConfiguredCoordinator()
+
+	// Set up a request
+	req, err := http.NewRequest("GET", "/burrow/admin/ready", nil)
+	assert.NoError(t, err, "Expected request setup to return no error")
+
+	// Call the handler via httprouter, the app is not ready so we expect "STARTING" and HTTP 503
+	rr := httptest.NewRecorder()
+	coordinator.router.ServeHTTP(rr, req)
+	assert.Equalf(t, http.StatusServiceUnavailable, rr.Code, "Expected response code to be 503, not %v", rr.Code)
+	assert.Equalf(t, "STARTING", rr.Body.String(), "Expected response body to be 'STARTING', not '%v'", rr.Body.String())
+
+	// Change the AppReady, and try again
+	coordinator.App.AppReady = true
+	rr = httptest.NewRecorder()
+	coordinator.router.ServeHTTP(rr, req)
+	assert.Equalf(t, http.StatusOK, rr.Code, "Expected response code to be 200, not %v", rr.Code)
+	assert.Equalf(t, "READY", rr.Body.String(), "Expected response body to be 'READY', not '%v'", rr.Body.String())
+}
+
 func TestHttpServer_getClusterList(t *testing.T) {
 	coordinator := fixtureConfiguredCoordinator()
 

core/protocol/protocol.go

@@ -60,6 +60,9 @@ type ApplicationContext struct {
 	// This is the channel over which any module should send storage requests for storage of offsets and group
 	// information, or to fetch the same information. It is serviced by the storage Coordinator.
 	StorageChannel chan *StorageRequest
+
+	// This is a boolean flag which is set by the last subsystem, the consumer, in order to signal when Burrow is ready
+	AppReady bool
 }
 
 // Module is a common interface for all modules so that they can be manipulated by the coordinators in the same way.