fixes flaky test

This commit adds a dedicated set of workers for processing only
in-memory part of vmagent persistentqueue. It should help to mitigate
an issue when stale data at file-based queue prevents from ingestion
recent data.

 There is a downside -in case of remote storage is not reachable,
it's possible to get only a part of data ingested and other part queued
into file-based queue. But it should be acceptable, because there is no
strong guarantees for the data ingestion order.

Fixes https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8833

app/vmagent/remotewrite/client.go

@@ -207,9 +207,13 @@ func (c *client) init(argIdx, concurrency int, sanitizedURL string) {
 	metrics.GetOrCreateGauge(fmt.Sprintf(`vmagent_remotewrite_queues{url=%q}`, c.sanitizedURL), func() float64 {
 		return float64(concurrency)
 	})
-	for range concurrency {
+	perWorkersConcurrency := max(concurrency/2, 1)
+	for range perWorkersConcurrency {
 		c.wg.Go(c.runWorker)
 	}
+	for range perWorkersConcurrency {
+		c.wg.Go(c.runWorkerForInmemoryQueue)
+	}
 	logger.Infof("initialized client for -remoteWrite.url=%q", c.sanitizedURL)
 }
 
@@ -348,6 +352,52 @@ func (c *client) runWorker() {
 	}
 }
 
+func (c *client) runWorkerForInmemoryQueue() {
+	var ok bool
+	var block []byte
+	ch := make(chan bool, 1)
+	for {
+		block, ok = c.fq.MustReadInMemoryBlockBlocking(block[:0])
+		if !ok {
+			return
+		}
+		go func() {
+			startTime := time.Now()
+			ch <- c.sendBlock(block)
+			c.sendDuration.Add(time.Since(startTime).Seconds())
+		}()
+		select {
+		case ok := <-ch:
+			if ok {
+				// The block has been sent successfully
+				continue
+			}
+			// Return unsent block to the queue.
+			c.fq.MustWriteBlockIgnoreDisabledPQ(block)
+			return
+		case <-c.stopCh:
+			// c must be stopped. Wait up to 5 seconds for the in-flight request to complete.
+			// If it succeeds, drain the remaining in-memory queue before returning.
+			stopCtx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+			defer cancel()
+
+			select {
+			case ok := <-ch:
+				if !ok {
+					// Return unsent block to the queue.
+					c.fq.MustWriteBlockIgnoreDisabledPQ(block)
+				} else {
+					c.drainInMemoryQueue(stopCtx, block[:0])
+				}
+			case <-stopCtx.Done():
+				// Return unsent block to the queue.
+				c.fq.MustWriteBlockIgnoreDisabledPQ(block)
+			}
+			return
+		}
+	}
+}
+
 func (c *client) doRequest(url string, body []byte) (*http.Response, error) {
 	req, err := c.newRequest(url, body)
 	if err != nil {

apptest/tests/vmagent_remotewrite_test.go

@@ -296,6 +296,7 @@ func TestSingleVMAgentDropOnOverload(t *testing.T) {
 		"-remoteWrite.disableOnDiskQueue=true",
 		// use only 1 worker to get a full queue faster
 		"-remoteWrite.queues=1",
+		"-remoteWrite.flushInterval=1ms",
 		// fastqueue size is roughly memory.Allowed() / len(urls) / *maxRowsPerBlock / 100
 		// Use very large maxRowsPerBlock to get fastqueue of minimal length(2).
 		// See initRemoteWriteCtxs function in remotewrite.go for details.
@@ -332,13 +333,21 @@ func TestSingleVMAgentDropOnOverload(t *testing.T) {
 	vmagent.APIV1ImportPrometheusNoWaitFlush(t, []string{
 		"foo_bar 1 1652169600000", // 2022-05-10T08:00:00Z
 	}, apptest.QueryOpts{})
-
 	waitFor(
 		func() bool {
 			return vmagent.RemoteWriteRequests(t, url1) == 1 && vmagent.RemoteWriteRequests(t, url2) == 1
 		},
 	)
-
+	// Wait until second request got flushed to remote write server
+	// since there are 2 indepent queues (general and in-memory) with minimal capacity of 1
+	vmagent.APIV1ImportPrometheusNoWaitFlush(t, []string{
+		"foo_bar 1 1652169600000", // 2022-05-10T08:00:00Z
+	}, apptest.QueryOpts{})
+	waitFor(
+		func() bool {
+			return vmagent.RemoteWriteRequests(t, url1) == 2 && vmagent.RemoteWriteRequests(t, url2) == 2
+		},
+	)
 	// Send 2 more requests, the first RW endpoint should receive everything, the second should add them to the queue
 	// since worker is busy with the first request.
 	for i := range 2 {
@@ -348,7 +357,7 @@ func TestSingleVMAgentDropOnOverload(t *testing.T) {
 
 		waitFor(
 			func() bool {
-				return vmagent.RemoteWriteRequests(t, url1) == 2+i && vmagent.RemoteWritePendingInmemoryBlocks(t, url2) == 1+i
+				return vmagent.RemoteWriteRequests(t, url1) == 3+i && vmagent.RemoteWritePendingInmemoryBlocks(t, url2) == 1+i
 			},
 		)
 	}
@@ -360,7 +369,7 @@ func TestSingleVMAgentDropOnOverload(t *testing.T) {
 
 	waitFor(
 		func() bool {
-			return vmagent.RemoteWriteRequests(t, url1) == 4 && vmagent.RemoteWriteSamplesDropped(t, url2) > 0
+			return vmagent.RemoteWriteRequests(t, url1) == 5 && vmagent.RemoteWriteSamplesDropped(t, url2) > 0
 		},
 	)
 }

docs/victoriametrics/changelog/CHANGELOG.md

@@ -29,6 +29,7 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 * SECURITY: upgrade Go builder from Go1.26.3 to Go1.26.4. See [the list of issues addressed in Go1.26.4](https://github.com/golang/go/issues?q=milestone%3AGo1.26.4%20label%3ACherryPickApproved).
 
 * FEATURE: [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/): support `match[]=<label_selector>` query parameters in `/api/v1/rules` and `/api/v1/alerts` APIs to return only the rules that have configured labels satisfying the provided label selectors. See [11020](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11020).
+* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): prioritise recently ingested data over historical data stored at file-based [persistent queue](https://docs.victoriametrics.com/victoriametrics/vmagent/#on-disk-persistence). See [#8833](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8833)
 * FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/), [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): add `-opentelemetry.promoteAllResourceAttributes` and `-opentelemetry.promoteScopeMetadata` command-line flags to allow managing label promotion for resource attributes and OTel scope metadata. See [OpenTelemetry](https://docs.victoriametrics.com/victoriametrics/integrations/opentelemetry/) docs and [#10931](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10931).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/) : introduce `vmagent_remotewrite_kafka_outbuf_latency_seconds` and  `vmagent_remotewrite_kafka_rtt_seconds` metrics for [kafka integration](https://docs.victoriametrics.com/victoriametrics/integrations/kafka/). The metrics could help identify throughput bottlenecks. See [#10730](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10730).
 * FEATURE: [vmauth](https://docs.victoriametrics.com/victoriametrics/vmauth/): properly log user information when a missing route error occurs. See [#11052](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11052).

lib/persistentqueue/fastqueue.go

@@ -194,19 +194,12 @@ func (fq *FastQueue) tryWriteBlock(block []byte, ignoreDisabledPQ bool) bool {
 
 	isPQWriteAllowed := !fq.isPQDisabled || ignoreDisabledPQ
 
-	fq.flushInmemoryBlocksToFileIfNeededLocked()
-	if n := fq.pq.GetPendingBytes(); n > 0 {
-		// The file-based queue isn't drained yet. This means that in-memory queue cannot be used yet.
-		// So put the block to file-based queue.
-		if len(fq.ch) > 0 {
-			logger.Panicf("BUG: the in-memory queue must be empty when the file-based queue is non-empty; it contains %d pending bytes", n)
-		}
-		if !isPQWriteAllowed {
-			return false
-		}
-		fq.pq.MustWriteBlock(block)
-		return true
+	if !isPQWriteAllowed && fq.pq.GetPendingBytes() > 0 {
+		// fast path there is pending data at file-based queue,
+		// it must be drained before in-memory queue could be used.
+		return false
 	}
+	fq.flushInmemoryBlocksToFileIfNeededLocked()
 	if len(fq.ch) == cap(fq.ch) {
 		// There is no space left in the in-memory queue. Put the data to file-based queue.
 		if !isPQWriteAllowed {
@@ -216,7 +209,7 @@ func (fq *FastQueue) tryWriteBlock(block []byte, ignoreDisabledPQ bool) bool {
 		fq.pq.MustWriteBlock(block)
 		return true
 	}
-	// Fast path - put the block to in-memory queue.
+
 	bb := blockBufPool.Get()
 	bb.B = append(bb.B[:0], block...)
 	fq.ch <- bb
@@ -239,16 +232,15 @@ func (fq *FastQueue) MustReadBlock(dst []byte) ([]byte, bool) {
 		if fq.stopDeadline > 0 && fasttime.UnixTimestamp() > fq.stopDeadline {
 			return dst, false
 		}
-		if len(fq.ch) > 0 {
-			return fq.mustReadInMemoryBlockLocked(dst), true
-		}
 		if n := fq.pq.GetPendingBytes(); n > 0 {
 			data, ok := fq.pq.MustReadBlockNonblocking(dst)
 			if ok {
 				return data, true
 			}
 			dst = data
-			continue
+		}
+		if len(fq.ch) > 0 {
+			return fq.mustReadInMemoryBlockLocked(dst), true
 		}
 		if fq.stopDeadline > 0 {
 			return dst, false
@@ -259,6 +251,28 @@ func (fq *FastQueue) MustReadBlock(dst []byte) ([]byte, bool) {
 	}
 }
 
+// MustReadInMemoryBlockBlocking reads the next block from the in-memory queue into dst and returns it.
+// It returns (dst, true) if a block was available, or (nil, false) if the in-memory queue is empty.
+// It blocks until a block is available or the stop deadline is exceeded, in which case it returns (dst, false).
+func (fq *FastQueue) MustReadInMemoryBlockBlocking(dst []byte) ([]byte, bool) {
+	fq.mu.Lock()
+	defer fq.mu.Unlock()
+
+	for {
+		if fq.stopDeadline > 0 && fasttime.UnixTimestamp() > fq.stopDeadline {
+			return dst, false
+		}
+		if len(fq.ch) > 0 {
+			return fq.mustReadInMemoryBlockLocked(dst), true
+		}
+		if fq.stopDeadline > 0 {
+			return dst, false
+		}
+		// There are no blocks. Wait for new block.
+		fq.cond.Wait()
+	}
+}
+
 // MustReadInMemoryBlock reads the next block from the in-memory queue into dst and returns it.
 // It returns (dst, true) if a block was available, or (nil, false) if the in-memory queue is empty.
 // It does not block waiting for new blocks.
@@ -277,9 +291,6 @@ func (fq *FastQueue) mustReadInMemoryBlockLocked(dst []byte) []byte {
 	if len(fq.ch) == 0 {
 		logger.Panicf("BUG: the function must not be called when in-memory queue is empty. Caller should verify the queue len upfront")
 	}
-	if n := fq.pq.GetPendingBytes(); n > 0 {
-		logger.Panicf("BUG: the file-based queue must be empty when the in-memory queue is non-empty; it contains %d pending bytes", n)
-	}
 	bb := <-fq.ch
 	fq.pendingInmemoryBytes -= uint64(len(bb.B))
 	fq.lastInmemoryBlockReadTime = fasttime.UnixTimestamp()