Merge pull request #47 from mihaitodor/service-draining-status

Add service DRAINING status and a handler for the `drain` HTTP API endpoint

Author: mihaitodor

catalog/services_state.go

@@ -30,6 +30,7 @@ const (
 	TOMBSTONE_SLEEP_INTERVAL = 2 * time.Second                // Sleep between local service checks
 	TOMBSTONE_RETRANSMIT     = 1 * time.Second                // Time between tombstone retranmission
 	ALIVE_LIFESPAN           = 1*time.Minute + 20*time.Second // Down if not heard from in 80 seconds
+	DRAINING_LIFESPAN        = 10 * time.Minute               // Down if not heard from in 10 minutes
 	ALIVE_SLEEP_INTERVAL     = 1 * time.Second                // Sleep between local service checks
 	ALIVE_BROADCAST_INTERVAL = 1 * time.Minute                // Broadcast Alive messages every minute
 )
@@ -132,6 +133,11 @@ func (state *ServicesState) ProcessServiceMsgs(looper director.Looper) {
 	})
 }
 
+// UpdateService enqueues a state update for a given service
+func (state *ServicesState) UpdateService(svc service.Service) {
+	state.ServiceMsgs <- svc
+}
+
 // Shortcut for checking if the Servers map has an entry for this
 // hostname.
 func (state *ServicesState) HasServer(hostname string) bool {
@@ -304,6 +310,11 @@ func (state *ServicesState) AddServiceEntry(newSvc service.Service) {
 		// Store the previous newSvc so we can compare it
 		oldEntry := server.Services[newSvc.ID]
 
+		// Make sure we preserve the DRAINING status for services
+		if oldEntry.Status == service.DRAINING && newSvc.Status == service.ALIVE {
+			newSvc.Status = oldEntry.Status
+		}
+
 		// Update the new one
 		server.Services[newSvc.ID] = &newSvc
 
@@ -320,12 +331,28 @@ func (state *ServicesState) AddServiceEntry(newSvc service.Service) {
 	}
 }
 
+// GetLocalServiceByID returns a service for a given ID if it
+// happens to exist on the current host. Returns an error otherwise.
+func (state *ServicesState) GetLocalServiceByID(id string) (service.Service, error) {
+	state.RLock()
+	defer state.RUnlock()
+
+	if server, ok := state.Servers[state.Hostname]; ok {
+		if svc, ok := server.Services[id]; ok {
+			return *svc, nil
+		}
+	}
+
+	return service.Service{},
+		fmt.Errorf("service with ID %q not found on host %q", id, state.Hostname)
+}
+
 // Merge a complete state struct into this one. Usually used on
 // node startup and during anti-entropy operations.
 func (state *ServicesState) Merge(otherState *ServicesState) {
 	for _, server := range otherState.Servers {
-		for _, service := range server.Services {
-			state.ServiceMsgs <- *service
+		for _, svc := range server.Services {
+			state.UpdateService(*svc)
 		}
 	}
 }
@@ -403,8 +430,8 @@ func (state *ServicesState) Print(list *memberlist.Memberlist) {
 // don't already know about.
 func (state *ServicesState) TrackNewServices(fn func() []service.Service, looper director.Looper) {
 	looper.Loop(func() error {
-		for _, container := range fn() {
-			state.ServiceMsgs <- container
+		for _, svc := range fn() {
+			state.UpdateService(svc)
 		}
 		return nil
 	})
@@ -610,12 +637,16 @@ func (state *ServicesState) TombstoneOthersServices() []service.Service {
 			}
 		}
 
+		svcLifespan := ALIVE_LIFESPAN
+		if svc.IsDraining() {
+			svcLifespan = DRAINING_LIFESPAN
+		}
 		// Everything that is not tombstoned needs to be considered for
 		// removal if it exceeds the allowed ALIVE_TIMESPAN
 		if !svc.IsTombstone() &&
-			svc.Updated.Before(time.Now().UTC().Add(0-ALIVE_LIFESPAN)) {
-			log.Warnf("Found expired service %s from %s, tombstoning",
-				svc.Name, svc.Hostname,
+			svc.Updated.Before(time.Now().UTC().Add(0-svcLifespan)) {
+			log.Warnf("Found expired service %s ID %s from %s, tombstoning",
+				svc.Name, svc.ID, svc.Hostname,
 			)
 
 			// Because we don't know that other hosts haven't gotten a newer

catalog/services_state_test.go

@@ -219,6 +219,59 @@ func Test_ServicesStateWithData(t *testing.T) {
 				}
 				So(pendingBroadcast, ShouldBeFalse)
 			})
+
+			Convey("Sets a service's status to DRAINING", func() {
+				state.AddServiceEntry(svc)
+
+				svc.Status = service.DRAINING
+				svc.Updated = time.Now().UTC()
+
+				state.AddServiceEntry(svc)
+
+				So(state.HasServer(anotherHostname), ShouldBeTrue)
+				So(state.Servers[anotherHostname].Services[svc.ID].Status,
+					ShouldEqual, service.DRAINING)
+			})
+
+			Convey("Doesn't mark a DRAINING service as ALIVE", func() {
+				svc.Status = service.DRAINING
+				state.AddServiceEntry(svc)
+
+				svc.Status = service.ALIVE
+				svc.Updated = time.Now().UTC()
+
+				state.AddServiceEntry(svc)
+
+				So(state.HasServer(anotherHostname), ShouldBeTrue)
+				So(state.Servers[anotherHostname].Services[svc.ID].Status,
+					ShouldEqual, service.DRAINING)
+			})
+		})
+
+		Convey("GetLocalServiceByID()", func() {
+			Convey("Returns an existing service on the current host", func() {
+				state.Hostname = anotherHostname
+				state.AddServiceEntry(svc)
+
+				returnedSvc, err := state.GetLocalServiceByID(svc.ID)
+				So(err, ShouldBeNil)
+				So(returnedSvc.ID, ShouldEqual, svc.ID)
+			})
+
+			Convey("Doesn't return a service running on other hosts", func() {
+				state.AddServiceEntry(svc)
+
+				_, err := state.GetLocalServiceByID(svc.ID)
+				So(err, ShouldNotBeNil)
+			})
+
+			Convey("Returns an error for a non-existent service ID", func() {
+				state.AddServiceEntry(svc)
+
+				_, err := state.GetLocalServiceByID("missing")
+				So(err, ShouldNotBeNil)
+				So(err.Error(), ShouldContainSubstring, "not found")
+			})
 		})
 
 		Convey("Merge() merges state we care about from other state structs", func() {
@@ -416,6 +469,36 @@ func Test_TrackingAndBroadcasting(t *testing.T) {
 			So(state.Servers[hostname].LastChanged.After(lastChanged), ShouldBeTrue)
 		})
 
+		Convey("Draining services have a lifespan and then are tombstoned", func() {
+			lastChanged := state.Servers[hostname].LastChanged
+			service1.Status = service.DRAINING
+			state.AddServiceEntry(service1)
+			svc := state.Servers[hostname].Services[service1.ID]
+			stamp := service1.Updated.Add(0 - DRAINING_LIFESPAN - 5*time.Second)
+			svc.Updated = stamp
+
+			state.TombstoneOthersServices()
+
+			So(svc.Status, ShouldEqual, service.TOMBSTONE)
+			So(svc.Updated, ShouldBeTheSameTimeAs, stamp.Add(time.Second))
+			So(state.Servers[hostname].LastChanged.After(lastChanged), ShouldBeTrue)
+		})
+
+		Convey("Draining services are not tombstoned before their lifespan expires", func() {
+			lastChanged := state.Servers[hostname].LastChanged
+			service1.Status = service.DRAINING
+			state.AddServiceEntry(service1)
+			svc := state.Servers[hostname].Services[service1.ID]
+			stamp := service1.Updated.Add(0 - ALIVE_LIFESPAN - 5*time.Second)
+			svc.Updated = stamp
+
+			state.TombstoneOthersServices()
+
+			So(svc.Status, ShouldEqual, service.DRAINING)
+			So(svc.Updated, ShouldBeTheSameTimeAs, stamp)
+			So(state.Servers[hostname].LastChanged.After(lastChanged), ShouldBeTrue)
+		})
+
 		Convey("Unhealthy/Unknown services have a lifespan and then are tombstoned", func() {
 			unhealthyService := service.Service{ID: "unhealthy_shakespeare", Hostname: hostname, Updated: baseTime, Status: service.UNHEALTHY}
 			state.AddServiceEntry(unhealthyService)

healthy/healthy.go

@@ -11,8 +11,8 @@ import (
 	"time"
 
 	"github.com/Nitro/sidecar/service"
-	log "github.com/sirupsen/logrus"
 	"github.com/relistan/go-director"
+	log "github.com/sirupsen/logrus"
 )
 
 const (

receiver/http_test.go

@@ -105,7 +105,7 @@ func Test_updateHandler(t *testing.T) {
 			So(received, ShouldBeTrue)
 		})
 
-		Convey("enqueus all updates if no Subscriptions are provided", func() {
+		Convey("enqueues all updates if no Subscriptions are provided", func() {
 			evtState := deepcopy.Copy(state).(*catalog.ServicesState)
 			evtState.LastChanged = time.Now().UTC()
 
@@ -223,5 +223,33 @@ func Test_updateHandler(t *testing.T) {
 			So(lastReceivedState.LastChanged.Before(state.LastChanged), ShouldBeTrue)
 			So(len(lastReceivedState.Servers["chaucer"].Services), ShouldEqual, 2)
 		})
+
+		Convey("enqueues an update to mark a service as DRAINING", func() {
+			evtState := deepcopy.Copy(state).(*catalog.ServicesState)
+			evtState.LastChanged = time.Now().UTC()
+
+			change := catalog.StateChangedEvent{
+				State: evtState,
+				ChangeEvent: catalog.ChangeEvent{
+					Service: service.Service{
+						Name:    "nobody-wants-this",
+						ID:      "10101010101",
+						Updated: time.Now().UTC(),
+						Created: time.Now().UTC(),
+						Status:  service.DRAINING,
+					},
+					PreviousStatus: service.ALIVE,
+				},
+			}
+
+			encoded, _ := json.Marshal(change)
+			req := httptest.NewRequest("POST", "/update", bytes.NewBuffer(encoded))
+
+			UpdateHandler(recorder, req, rcvr)
+			resp := recorder.Result()
+
+			So(resp.StatusCode, ShouldEqual, 200)
+			So(len(rcvr.ReloadChan), ShouldEqual, 1)
+		})
 	})
 }

receiver/receiver.go

@@ -57,6 +57,8 @@ func ShouldNotify(oldStatus int, newStatus int) bool {
 		if oldStatus == service.ALIVE {
 			return true
 		}
+	case service.DRAINING:
+		return true
 	default:
 		log.Errorf("Got unknown service change status: %d", newStatus)
 		return false

service/service.go

@@ -19,6 +19,7 @@ const (
 	TOMBSTONE = iota
 	UNHEALTHY = iota
 	UNKNOWN   = iota
+	DRAINING  = iota
 )
 
 type Port struct {
@@ -56,6 +57,10 @@ func (svc *Service) IsTombstone() bool {
 	return svc.Status == TOMBSTONE
 }
 
+func (svc *Service) IsDraining() bool {
+	return svc.Status == DRAINING
+}
+
 func (svc *Service) Invalidates(otherSvc *Service) bool {
 	return otherSvc != nil && svc.Updated.After(otherSvc.Updated)
 }
@@ -162,6 +167,8 @@ func StatusString(status int) string {
 		return "Unhealthy"
 	case UNKNOWN:
 		return "Unknown"
+	case DRAINING:
+		return "Draining"
 	default:
 		return "Tombstone"
 	}

services_delegate.go

@@ -50,7 +50,7 @@ func (d *servicesDelegate) Start() {
 				log.Errorf("Start(): error decoding message: %s", err)
 				continue
 			}
-			d.state.ServiceMsgs <- *entry
+			d.state.UpdateService(*entry)
 		}
 	}()
 

sidecarhttp/envoy_api_test.go

@@ -2,8 +2,6 @@ package sidecarhttp
 
 import (
 	"encoding/json"
-	"io/ioutil"
-	"net/http"
 	"net/http/httptest"
 	"testing"
 	"time"
@@ -290,12 +288,3 @@ func Test_listenersHandler(t *testing.T) {
 		})
 	})
 }
-
-// getResult fetchs the status code, headers, and body from a recorder
-func getResult(recorder *httptest.ResponseRecorder) (code int, headers *http.Header, body string) {
-	resp := recorder.Result()
-	bodyBytes, _ := ioutil.ReadAll(resp.Body)
-	body = string(bodyBytes)
-
-	return resp.StatusCode, &resp.Header, body
-}

sidecarhttp/http_api.go

@@ -35,6 +35,7 @@ type SidecarApi struct {
 func (s *SidecarApi) HttpMux() http.Handler {
 	router := mux.NewRouter()
 	router.HandleFunc("/services/{name}.{extension}", wrap(s.oneServiceHandler)).Methods("GET")
+	router.HandleFunc("/services/{id}/drain", wrap(s.drainServiceHandler)).Methods("POST")
 	router.HandleFunc("/services.{extension}", wrap(s.servicesHandler)).Methods("GET")
 	router.HandleFunc("/state.{extension}", wrap(s.stateHandler)).Methods("GET")
 	router.HandleFunc("/watch", wrap(s.watchHandler)).Methods("GET")
@@ -289,6 +290,58 @@ func (s *SidecarApi) stateHandler(response http.ResponseWriter, req *http.Reques
 	}
 }
 
+// drainServiceHandler instructs Sidecar to set the status of a given service
+// instance to DRAINING. This allows us to decomission the given service
+// instance and let it sit around for a short amount of time, so it can finish
+// processing the requests that are still in flight.
+func (s *SidecarApi) drainServiceHandler(response http.ResponseWriter, req *http.Request, params map[string]string) {
+	defer req.Body.Close()
+
+	if req.Method != http.MethodPost {
+		sendJsonError(response, 400, fmt.Sprintf("Bad request - Method %q not allowed", req.Method))
+		return
+	}
+
+	serviceID, ok := params["id"]
+	if !ok {
+		sendJsonError(response, 404, "Not Found - No service ID provided")
+		return
+	}
+
+	if s.state == nil {
+		sendJsonError(response, 500, "Internal Server Error - Something went terribly wrong")
+		return
+	}
+
+	svc, err := s.state.GetLocalServiceByID(serviceID)
+	if err != nil {
+		sendJsonError(response, 404, fmt.Sprintf("Not Found - Service ID %q not found", serviceID))
+		return
+	}
+
+	svc.Updated = time.Now()
+	svc.Status = service.DRAINING
+	s.state.UpdateService(svc)
+
+	result := struct {
+		Message string
+	}{
+		Message: fmt.Sprintf("Service %q instance %q set to DRAINING", svc.Name, svc.ID),
+	}
+	jsonBytes, err := json.MarshalIndent(&result, "", "  ")
+	if err != nil {
+		sendJsonError(response, 500, "Internal Server Error - Something went terribly wrong")
+		return
+	}
+
+	response.Header().Set("Content-Type", "application/json")
+	response.WriteHeader(202)
+	_, err = response.Write(jsonBytes)
+	if err != nil {
+		log.Errorf("Error writing drain service response to client: %s", err)
+	}
+}
+
 // Send back a JSON encoded error and message
 func sendJsonError(response http.ResponseWriter, status int, message string) {
 	output := map[string]string{