feat: Ingester RF-1

pkg/ingester-rf1/ingester.go

@@ -571,32 +571,43 @@ func (i *Ingester) loop() {
 	for {
 		select {
 		case <-flushTicker.C:
-			//i.logger.Log("msg", "starting periodic flush")
-			i.flushCtx.lock.Lock() // Stop new chunks being written while we swap destinations - we'll never unlock as this flushctx can no longer be used.
-			currentFlushCtx := i.flushCtx
-			// APIs become unblocked after resetting flushCtx
-			segmentWriter, err := wal.NewWalSegmentWriter()
-			if err != nil {
-				// TODO: handle this properly
-				panic(err)
-			}
-			i.flushCtx = &flushCtx{
-				lock:            &sync.RWMutex{},
-				flushDone:       make(chan struct{}),
-				newCtxAvailable: make(chan struct{}),
-				segmentWriter:   segmentWriter,
-			}
-			close(currentFlushCtx.newCtxAvailable) // Broadcast to all waiters that they can now fetch a new flushCtx. Small chance of a race but if they re-fetch the old one, they'll just check again immediately.
-			// Flush the finished context in the background & then notify watching API requests
-			// TODO: use multiple flush queues if required
-			i.flushQueues[0].Enqueue(currentFlushCtx)
-
+			i.doFlushTick()
 		case <-i.loopQuit:
 			return
 		}
 	}
 }
 
+func (i *Ingester) doFlushTick() {
+	i.flushCtx.lock.Lock()
+	defer i.flushCtx.lock.Unlock()
+
+	//i.logger.Log("msg", "starting periodic flush")
+	// Stop new chunks being written while we swap destinations - we'll never unlock as this flushctx can no longer be used.
+	currentFlushCtx := i.flushCtx
+	// Don't write empty files if there is nothing to write.
+	if currentFlushCtx.segmentWriter.InputSize() == 0 {
+		return
+	}
+
+	// APIs become unblocked after resetting flushCtx
+	segmentWriter, err := wal.NewWalSegmentWriter()
+	if err != nil {
+		// TODO: handle this properly
+		panic(err)
+	}
+	i.flushCtx = &flushCtx{
+		lock:            &sync.RWMutex{},
+		flushDone:       make(chan struct{}),
+		newCtxAvailable: make(chan struct{}),
+		segmentWriter:   segmentWriter,
+	}
+	close(currentFlushCtx.newCtxAvailable) // Broadcast to all waiters that they can now fetch a new flushCtx. Small chance of a race but if they re-fetch the old one, they'll just check again immediately.
+	// Flush the finished context in the background & then notify watching API requests
+	// TODO: use multiple flush queues if required
+	i.flushQueues[0].Enqueue(currentFlushCtx)
+}
+
 // PrepareShutdown will handle the /ingester/prepare_shutdown endpoint.
 //
 // Internally, when triggered, this handler will configure the ingester service to release their resources whenever a SIGTERM is received.