Don't flush chunks after a successful transfer (#449)

* Don't flush chunks after a successful transfer

* Add log message for successful chunk transfer.

* Remove old log line.

Author: tomwilkie

pkg/ingester/ingester_claim.go

@@ -4,6 +4,7 @@ import (
 	"io"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/common/log"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/storage/local/chunk"
 
@@ -52,7 +53,10 @@ func (i *Ingester) TransferChunks(stream client.Ingester_TransferChunksServer) e
 		// We can't send "extra" fields with a streaming call, so we repeat
 		// wireSeries.FromIngesterId and assume it is the same every time
 		// round this loop.
-		fromIngesterID = wireSeries.FromIngesterId
+		if fromIngesterID == "" {
+			fromIngesterID = wireSeries.FromIngesterId
+			log.Infof("Processing TransferChunks request from ingester '%s'.")
+		}
 		metric := util.FromLabelPairs(wireSeries.Labels)
 		userCtx := user.Inject(stream.Context(), wireSeries.UserId)
 		descs, err := fromWireChunks(wireSeries.Chunks)

pkg/ingester/ingester_lifecycle.go

@@ -325,13 +325,20 @@ func (i *Ingester) processShutdown() {
 			flushRequired = false
 		}
 	}
+
 	if flushRequired {
 		i.flushAllChunks()
-	}
 
-	// Close the flush queues, to unblock waiting workers.
-	for _, flushQueue := range i.flushQueues {
-		flushQueue.Close()
+		// Close the flush queues, to unblock waiting workers.
+		for _, flushQueue := range i.flushQueues {
+			flushQueue.Close()
+		}
+	} else {
+
+		// Close & empty all the flush queues, to unblock waiting workers.
+		for _, flushQueue := range i.flushQueues {
+			flushQueue.DrainAndClose()
+		}
 	}
 
 	// Wait for chunks to be flushed.
@@ -389,6 +396,7 @@ func (i *Ingester) transferChunks() error {
 		return err
 	}
 
+	log.Infof("Successfully sent chunks to '%s'", targetIngester.Addr)
 	return nil
 }
 

pkg/util/priority_queue.go

@@ -66,6 +66,16 @@ func (pq *PriorityQueue) Close() {
 	pq.cond.Broadcast()
 }
 
+// DrainAndClose closed the queue and removes all the items from it.
+func (pq *PriorityQueue) DrainAndClose() {
+	pq.lock.Lock()
+	defer pq.lock.Unlock()
+	pq.closed = true
+	pq.queue = nil
+	pq.hit = map[string]struct{}{}
+	pq.cond.Broadcast()
+}
+
 // Enqueue adds an operation to the queue in priority order. If the operation
 // is already on the queue, it will be ignored.
 func (pq *PriorityQueue) Enqueue(op Op) {