Add PREPARING_TO_LEAVE state to ingester handover

Author: Marcus Cobden

pkg/ingester/ingester_lifecycle.go

@@ -198,9 +198,7 @@ loop:
 			break loop
 		}
 	}
-
-	// Mark ourselved as Leaving so no more samples are send to us.
-	i.changeState(ring.LEAVING)
+	i.changeState(ring.PREPARING_TO_LEAVE)
 
 	// Do the transferring / flushing on a background goroutine so we can continue
 	// to heartbeat to consul.
@@ -318,14 +316,16 @@ func (i *Ingester) updateConsul() error {
 }
 
 // changeState updates consul with state transitions for us.  NB this must be
-// called from loop()!  Use ChangeState for calls from outside of loop().
+// called from loop()!
+// Use ChangeState for calls from outside of loop() (unless the loop has shut down)
 func (i *Ingester) changeState(state ring.IngesterState) error {
 	// Only the following state transitions can be triggered externally
 	if !((i.state == ring.PENDING && state == ring.JOINING) || // triggered by TransferChunks at the beginning
 		(i.state == ring.JOINING && state == ring.PENDING) || // triggered by TransferChunks on failure
 		(i.state == ring.JOINING && state == ring.ACTIVE) || // triggered by TransferChunks on success
 		(i.state == ring.PENDING && state == ring.ACTIVE) || // triggered by autoJoin
-		(i.state == ring.ACTIVE && state == ring.LEAVING)) { // triggered by shutdown
+		(i.state == ring.ACTIVE && state == ring.PREPARING_TO_LEAVE) || // triggered by shutdown
+		(i.state == ring.PREPARING_TO_LEAVE && state == ring.LEAVING)) { // triggered by shutdown
 		return fmt.Errorf("Changing ingester state from %v -> %v is disallowed", i.state, state)
 	}
 
@@ -343,6 +343,10 @@ func (i *Ingester) processShutdown() {
 			flushRequired = false
 		}
 	}
+	if i.state != ring.LEAVING {
+		// Mark ourselved as Leaving so no more samples are send to us.
+		i.changeState(ring.LEAVING)
+	}
 
 	if flushRequired {
 		i.flushAllChunks()
@@ -411,6 +415,12 @@ func (i *Ingester) transferChunks() error {
 			}
 
 			sentChunks.Add(float64(len(chunks)))
+			if i.state != ring.LEAVING {
+				// Mark ourselved as Leaving so no more samples are send to us.
+				// We wait until we have sent the first item through the stream, so that the remote
+				// side has a chance to mark all the tokens for transfer.
+				i.changeState(ring.LEAVING)
+			}
 		}
 	}
 

pkg/ring/replication_strategy.go

@@ -48,7 +48,7 @@ func (r *Ring) replicationStrategy(ingesters []*IngesterDesc, op Operation) (
 
 // IsHealthy checks whether an ingester appears to be alive and heartbeating
 func (r *Ring) IsHealthy(ingester *IngesterDesc, op Operation) bool {
-	if op == Write && ingester.State != ACTIVE {
+	if op == Write && (ingester.State != ACTIVE && ingester.State != PREPARING_TO_LEAVE) {
 		return false
 	} else if op == Read && ingester.State == JOINING {
 		return false

pkg/ring/ring.go

@@ -213,20 +213,22 @@ func (r *Ring) getInternal(key uint32, op Operation) (ReplicationSet, error) {
 		distinctHosts[token.Ingester] = struct{}{}
 		ingester := r.ringDesc.Ingesters[token.Ingester]
 
-		// TODO update comment
-		// We do not want to Write to Ingesters that are not ACTIVE, but we do want
-		// to write the extra replica somewhere.  So we increase the size of the set
-		// of replicas for the key. This means we have to also increase the
-		// size of the replica set for read, but we can read from Leaving ingesters,
-		// so don't skip it in this case.
+		// We want to only Write to Ingesters that are ACTIVE (or PREPARING_TO_LEAVE), however
+		// we still write to the same number of replicas.  So we increase the size of the set of
+		// replicas for the key.
+		// To match this, we have to also increase the size of the replica set for read, but we
+		// can still read from ingesters which are PREPARING_TO_LEAVE or LEAVING so don't skip it
+		// in this case.
 		// NB dead ingester will be filtered later (by replication_strategy.go).
-		if op == Write && ingester.State != ACTIVE {
-			if nextIngester := r.ringDesc.Ingesters[token.NextIngester]; token.NextIngester != "" && ingester.State == JOINING {
+		if op == Write && !(ingester.State == ACTIVE || ingester.State == PREPARING_TO_LEAVE) {
+			// If a token is being transferred to a new ingester, we can send writes to that ingester instead
+			// where they will be queued until the transfer is complete
+			if nextIngester := r.ringDesc.Ingesters[token.NextIngester]; nextIngester != nil && nextIngester.State == JOINING {
 				ingester = nextIngester
 			} else {
 				n++
 			}
-		} else if op == Read && (ingester.State != ACTIVE && ingester.State != LEAVING) {
+		} else if op == Read && !(ingester.State == ACTIVE || ingester.State == PREPARING_TO_LEAVE || ingester.State == LEAVING) {
 			n++
 		}
 

pkg/ring/ring.proto

@@ -27,4 +27,5 @@ enum IngesterState {
 
 	PENDING = 2;
 	JOINING = 3;
+	PREPARING_TO_LEAVE = 4;
 }