maintenance: make VictoriaMetrics flush recovery durable (#4289)

Co-authored-by: Duansg <siguoduan@gmail.com>
2026-09-17 09:40:58 +00:00 · 2026-08-26 10:58:09 +08:00
parent db73a43f9e
commit c884f11995
11 changed files with 794 additions and 125 deletions
@@ -26,6 +26,7 @@ import org.apache.hertzbeat.common.entity.dto.Message;
 import org.apache.hertzbeat.common.entity.dto.MetricsData;
 import org.apache.hertzbeat.common.entity.dto.MetricsHistoryData;
 import org.apache.hertzbeat.warehouse.service.MetricsDataService;
+import org.apache.hertzbeat.warehouse.service.WarehouseStorageStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
@@ -51,14 +52,14 @@ public class MetricsDataController {

    @GetMapping("/api/warehouse/storage/status")
    @Operation(summary = "Query Warehouse Storage Server Status", description = "Query the availability status of the storage service under the warehouse")
-    public ResponseEntity<Message<Void>> getWarehouseStorageServerStatus() {
-        Boolean status = metricsDataService.getWarehouseStorageServerStatus();
-        if (Boolean.TRUE.equals(status)) {
-            return ResponseEntity.ok(Message.success());
+    public ResponseEntity<Message<WarehouseStorageStatus>> getWarehouseStorageServerStatus() {
+        WarehouseStorageStatus status = metricsDataService.getWarehouseStorageStatus();
+        Message<WarehouseStorageStatus> message = Message.success(status);
+        if (!status.available()) {
+            message.setCode(FAIL_CODE);
+            message.setMsg("Service not available!");
        }
-
-        // historyDataReader does not exist or is not available
-        return ResponseEntity.ok(Message.fail(FAIL_CODE, "Service not available!"));
+        return ResponseEntity.ok(message);
    }

    @GetMapping("/api/monitor/{monitorId}/metrics/{metrics}")
@@ -31,6 +31,13 @@ public interface MetricsDataService {
     */
    Boolean getWarehouseStorageServerStatus();

+    /**
+     * Query storage availability and write-path loss/backlog counters.
+     *
+     * @return warehouse storage status
+     */
+    WarehouseStorageStatus getWarehouseStorageStatus();
+
    /**
     * Query Real Time Metrics Data
     * @param monitorId Monitor Id
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hertzbeat.warehouse.service;
+
+/**
+ * Operator-visible status for the configured warehouse history storage.
+ *
+ * @param available whether the storage's read-side health check is available
+ * @param droppedMetrics cumulative metric samples discarded by bounded writer backpressure
+ * @param pendingMetrics metric samples currently buffered or retained for retry
+ */
+public record WarehouseStorageStatus(boolean available, long droppedMetrics, int pendingMetrics) {
+}
@@ -35,6 +35,7 @@ import org.apache.hertzbeat.common.entity.dto.ValueRow;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
 import org.apache.hertzbeat.common.support.exception.CommonException;
 import org.apache.hertzbeat.warehouse.service.MetricsDataService;
+import org.apache.hertzbeat.warehouse.service.WarehouseStorageStatus;
 import org.apache.hertzbeat.warehouse.store.history.tsdb.HistoryDataReader;
 import org.apache.hertzbeat.warehouse.store.realtime.RealTimeDataReader;
 import org.springframework.stereotype.Service;
@@ -89,7 +90,17 @@ public class MetricsDataServiceImpl implements MetricsDataService {

    @Override
    public Boolean getWarehouseStorageServerStatus() {
-        return historyDataReader.isPresent() && historyDataReader.get().isServerAvailable();
+        return getWarehouseStorageStatus().available();
+    }
+
+    @Override
+    public WarehouseStorageStatus getWarehouseStorageStatus() {
+        return historyDataReader
+            .map(reader -> new WarehouseStorageStatus(
+                reader.isServerAvailable(),
+                reader.getDroppedMetricCount(),
+                reader.getPendingMetricCount()))
+            .orElseGet(() -> new WarehouseStorageStatus(false, 0, 0));
    }

    @Override
@@ -33,6 +33,20 @@ public interface HistoryDataReader {
     */
    boolean isServerAvailable();

+    /**
+     * @return cumulative metric samples dropped by the history writer
+     */
+    default long getDroppedMetricCount() {
+        return 0;
+    }
+
+    /**
+     * @return metric samples currently waiting for history persistence
+     */
+    default int getPendingMetricCount() {
+        return 0;
+    }
+
    /**
     * @return whether this storage supports observability log query
     */
@@ -37,6 +37,7 @@ import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -101,17 +102,25 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag
    private static final String MONITOR_METRICS_KEY = "__metrics__";
    private static final String MONITOR_METRIC_KEY = "__metric__";
    private static final long MAX_WAIT_MS = 500L;
-    private static final int MAX_RETRIES = 3;
+    private static final int MAX_BUFFER_OFFER_ATTEMPTS = 3;
+    private static final int MAX_BATCHES_PER_FLUSH_TASK = 16;
+    private static final long FAILED_FLUSH_RETRY_SECONDS = 1L;

    private final VictoriaMetricsClusterProperties vmClusterProps;
    private final VictoriaMetricsInsertProperties vmInsertProps;
    private final VictoriaMetricsSelectProperties vmSelectProps;
    private final RestTemplate restTemplate;
    private final BlockingQueue<VictoriaMetricsDataStorage.VictoriaMetricsContent> metricsBufferQueue;
+    private final Object metricsFlushLock = new Object();
    private final AtomicLong ignoredLabelCollisionCount = new AtomicLong();

    private HashedWheelTimer metricsFlushTimer = null;
    private MetricsFlushTask metricsFlushtask = null;
+    private final AtomicBoolean immediateFlushPending = new AtomicBoolean(false);
+    private final AtomicBoolean flushInFlight = new AtomicBoolean(false);
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+    private final AtomicLong droppedMetricCount = new AtomicLong();
+    private List<VictoriaMetricsDataStorage.VictoriaMetricsContent> retryBatch = Collections.emptyList();
    private boolean isBatchImportEnabled = false;


@@ -138,7 +147,7 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag
            Thread thread = new Thread(r, "victoria-metrics-flush-timer");
            thread.setDaemon(true);
            return thread;
-        }, 1, TimeUnit.SECONDS, 512);
+        }, 100, TimeUnit.MILLISECONDS, 512);
        metricsFlushtask = new MetricsFlushTask();
        this.metricsFlushTimer.newTimeout(metricsFlushtask, 0, TimeUnit.SECONDS);
    }
@@ -177,6 +186,10 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag

    @Override
    public void saveData(CollectRep.MetricsData metricsData) {
+        if (closed.get()) {
+            log.warn("[Victoria Metrics] Rejecting metrics after storage shutdown");
+            return;
+        }
        if (!isServerAvailable()) {
            serverAvailable = checkVictoriaMetricsDatasourceAvailable();
        }
@@ -298,9 +311,22 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag

    @Override
    public void destroy() {
+        synchronized (metricsFlushLock) {
+            if (!closed.compareAndSet(false, true)) {
+                return;
+            }
+        }
        if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) {
            metricsFlushTimer.stop();
        }
+        immediateFlushPending.set(false);
+        while (hasPendingMetrics()) {
+            if (!flushBufferedMetrics()) {
+                log.error("[Victoria Metrics] Unable to flush {} buffered metrics during shutdown",
+                        getPendingMetricCount());
+                break;
+            }
+        }
    }

    @Override
@@ -600,6 +626,10 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag
     * Save metric data to victoria-metric via HTTP call
     */
    public void doSaveData(List<VictoriaMetricsDataStorage.VictoriaMetricsContent> contentList) {
+        trySaveData(contentList);
+    }
+
+    private boolean trySaveData(List<VictoriaMetricsDataStorage.VictoriaMetricsContent> contentList) {
        try {
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
@@ -619,12 +649,15 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag
                httpEntity, String.class);
            if (responseEntity.getStatusCode().is2xxSuccessful()) {
                log.debug("insert metrics data to victoria-metrics success.");
+                return true;
            } else {
-                log.error("insert metrics data to victoria-metrics failed. {}", responseEntity.getBody());
+                log.error("insert metrics data to victoria-metrics failed with status {}",
+                        responseEntity.getStatusCode());
            }
        } catch (Exception e){
            log.error("flush metrics data to victoria-metrics error: {}.", e.getMessage(), e);
        }
+        return false;
    }

    /**
@@ -632,49 +665,123 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag
     * @param contentList victoriaMetricsContent List
     */
    private void sendVictoriaMetrics(List<VictoriaMetricsDataStorage.VictoriaMetricsContent> contentList) {
-        for (VictoriaMetricsDataStorage.VictoriaMetricsContent content : contentList) {
-            boolean offered = false;
-            int retryCount = 0;
-            while (!offered && retryCount < MAX_RETRIES) {
-                try {
-                    // Attempt to add to the queue for a limited time
-                    offered = metricsBufferQueue.offer(content, MAX_WAIT_MS, TimeUnit.MILLISECONDS);
-                    if (!offered) {
-                        // If the queue is still full, trigger an immediate refresh to free up space
-                        if (retryCount == 0) {
-                            log.debug("victoria metrics buffer queue is full, triggering immediate flush");
+        for (int index = 0; index < contentList.size(); index++) {
+            VictoriaMetricsDataStorage.VictoriaMetricsContent content = contentList.get(index);
+            boolean offered = metricsBufferQueue.offer(content);
+            for (int attempt = 1; attempt <= MAX_BUFFER_OFFER_ATTEMPTS && !offered; attempt++) {
+                if (closed.get()) {
+                    return;
+                }
                triggerImmediateFlush();
-                        }
-                        retryCount++;
-                        // The short sleep allows the queue to clear out
-                        if (retryCount < MAX_RETRIES) {
-                            Thread.sleep(100L * retryCount);
-                        }
-                    }
+                try {
+                    offered = metricsBufferQueue.offer(content, MAX_WAIT_MS, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
-                    log.error("[Victoria Metrics] Interrupted while offering metrics to buffer queue", e);
-                    break;
+                    recordDroppedMetrics(contentList.size() - index, "producer interrupted");
+                    return;
                }
            }
-            // When the maximum number of retries is reached, if it still cannot be added to the queue, the data is saved directly
            if (!offered) {
-                log.warn("[Victoria Metrics] Failed to add metrics to buffer after {} retries, saving directly", MAX_RETRIES);
-                try {
-                    doSaveData(contentList);
-                } catch (Exception e) {
-                    log.error("[Victoria Metrics] Failed to save metrics directly: {}", e.getMessage(), e);
+                recordDroppedMetrics(contentList.size() - index, "buffer remained full");
+                return;
            }
-            }
-            // Refresh in advance to avoid waiting
            if (metricsBufferQueue.size() >= vmInsertProps.bufferSize() * 0.8) {
                triggerImmediateFlush();
            }
        }
    }

+    private void recordDroppedMetrics(int count, String reason) {
+        long total = droppedMetricCount.addAndGet(count);
+        if (total == count || (total & (total - 1)) == 0 || total % 100 == 0) {
+            log.error("[Victoria Metrics] Dropped {} metrics because {}; cumulative dropped metrics: {}",
+                    count, reason, total);
+        }
+    }
+
+    @Override
+    public long getDroppedMetricCount() {
+        return droppedMetricCount.get();
+    }
+
    private void triggerImmediateFlush() {
-        metricsFlushTimer.newTimeout(metricsFlushtask, 0, TimeUnit.MILLISECONDS);
+        scheduleImmediateFlush(0, TimeUnit.MILLISECONDS);
+    }
+
+    private void scheduleImmediateFlush(long delay, TimeUnit unit) {
+        if (closed.get() || metricsFlushTimer == null || metricsFlushTimer.isStop()) {
+            return;
+        }
+        if (immediateFlushPending.compareAndSet(false, true)) {
+            try {
+                metricsFlushTimer.newTimeout(new ImmediateMetricsFlushTask(), delay, unit);
+            } catch (RuntimeException e) {
+                immediateFlushPending.set(false);
+                if (!closed.get()) {
+                    log.warn("[Victoria Metrics] Unable to schedule immediate flush: {}", e.getMessage());
+                }
+            }
+        }
+    }
+
+    private boolean flushBufferedMetrics() {
+        if (!flushInFlight.compareAndSet(false, true)) {
+            return false;
+        }
+        try {
+            List<VictoriaMetricsDataStorage.VictoriaMetricsContent> batch;
+            synchronized (metricsFlushLock) {
+                if (retryBatch.isEmpty()) {
+                    List<VictoriaMetricsDataStorage.VictoriaMetricsContent> nextBatch =
+                            new ArrayList<>(vmInsertProps.bufferSize());
+                    metricsBufferQueue.drainTo(nextBatch, vmInsertProps.bufferSize());
+                    retryBatch = nextBatch;
+                }
+                batch = retryBatch;
+            }
+            if (batch.isEmpty()) {
+                return true;
+            }
+            if (!trySaveData(batch)) {
+                log.warn("[Victoria Metrics] Retaining {} metrics items for retry", batch.size());
+                return false;
+            }
+            synchronized (metricsFlushLock) {
+                if (retryBatch == batch) {
+                    retryBatch = Collections.emptyList();
+                }
+            }
+            log.debug("[Victoria Metrics] Flushed {} metrics items", batch.size());
+            return true;
+        } finally {
+            flushInFlight.set(false);
+        }
+    }
+
+    private boolean hasPendingMetrics() {
+        synchronized (metricsFlushLock) {
+            return !retryBatch.isEmpty() || !metricsBufferQueue.isEmpty();
+        }
+    }
+
+    @Override
+    public int getPendingMetricCount() {
+        synchronized (metricsFlushLock) {
+            return retryBatch.size() + metricsBufferQueue.size();
+        }
+    }
+
+    private void schedulePeriodicFlush() {
+        if (closed.get() || metricsFlushTimer == null || metricsFlushTimer.isStop()) {
+            return;
+        }
+        try {
+            metricsFlushTimer.newTimeout(metricsFlushtask, vmInsertProps.flushInterval(), TimeUnit.SECONDS);
+        } catch (RuntimeException e) {
+            if (!closed.get()) {
+                log.warn("[Victoria Metrics] Unable to schedule periodic flush: {}", e.getMessage());
+            }
+        }
    }

    /**
@@ -683,18 +790,47 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag
    private class MetricsFlushTask implements TimerTask {
        @Override
        public void run(Timeout timeout) {
+            boolean flushSucceeded = false;
            try {
-                List<VictoriaMetricsDataStorage.VictoriaMetricsContent> batch = new ArrayList<>(vmInsertProps.bufferSize());
-                metricsBufferQueue.drainTo(batch, vmInsertProps.bufferSize());
-                if (!batch.isEmpty()) {
-                    doSaveData(batch);
-                    log.debug("[Victoria Metrics] Flushed {} metrics items", batch.size());
-                }
-                if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) {
-                    metricsFlushTimer.newTimeout(this, vmInsertProps.flushInterval(), TimeUnit.SECONDS);
-                }
+                flushSucceeded = flushBufferedMetrics();
            } catch (Exception e) {
                log.error("[VictoriaMetrics] flush task error: {}", e.getMessage(), e);
+            } finally {
+                if (!flushSucceeded && hasPendingMetrics() && !closed.get()) {
+                    scheduleImmediateFlush(FAILED_FLUSH_RETRY_SECONDS, TimeUnit.SECONDS);
+                }
+                schedulePeriodicFlush();
+            }
+        }
+    }
+
+    /**
+     * Executes an immediate flush without creating another periodic chain.
+     */
+    private class ImmediateMetricsFlushTask implements TimerTask {
+        @Override
+        public void run(Timeout timeout) {
+            boolean flushSucceeded = false;
+            try {
+                int flushedBatches = 0;
+                do {
+                    flushSucceeded = flushBufferedMetrics();
+                    flushedBatches++;
+                } while (flushSucceeded
+                        && hasPendingMetrics()
+                        && !closed.get()
+                        && flushedBatches < MAX_BATCHES_PER_FLUSH_TASK);
+            } catch (Exception e) {
+                log.error("[VictoriaMetrics] immediate flush task error: {}", e.getMessage(), e);
+            } finally {
+                immediateFlushPending.set(false);
+                if (hasPendingMetrics() && !closed.get()) {
+                    if (flushSucceeded) {
+                        scheduleImmediateFlush(0, TimeUnit.MILLISECONDS);
+                    } else {
+                        scheduleImmediateFlush(FAILED_FLUSH_RETRY_SECONDS, TimeUnit.SECONDS);
+                    }
+                }
            }
        }
    }
@@ -108,16 +108,24 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {
        MONITOR_METRIC_KEY,
        LABEL_KEY_INSTANCE);
    private static final long MAX_WAIT_MS = 500L;
-    private static final int MAX_RETRIES = 3;
+    private static final int MAX_BUFFER_OFFER_ATTEMPTS = 3;
+    private static final int MAX_BATCHES_PER_FLUSH_TASK = 16;
+    private static final long FAILED_FLUSH_RETRY_SECONDS = 1L;

    private final VictoriaMetricsProperties victoriaMetricsProp;
    private final RestTemplate restTemplate;
    private final BlockingQueue<VictoriaMetricsDataStorage.VictoriaMetricsContent> metricsBufferQueue;
+    private final Object metricsFlushLock = new Object();
    private final AtomicLong ignoredLabelCollisionCount = new AtomicLong();

    private HashedWheelTimer metricsFlushTimer = null;
+    private MetricsFlushTask metricsFlushTask = null;
    private final VictoriaMetricsProperties.InsertConfig insertConfig;
-    private final AtomicBoolean draining = new AtomicBoolean(false);
+    private final AtomicBoolean immediateFlushPending = new AtomicBoolean(false);
+    private final AtomicBoolean flushInFlight = new AtomicBoolean(false);
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+    private final AtomicLong droppedMetricCount = new AtomicLong();
+    private List<VictoriaMetricsContent> retryBatch = Collections.emptyList();

    public VictoriaMetricsDataStorage(VictoriaMetricsProperties victoriaMetricsProperties, RestTemplate restTemplate) {
        if (victoriaMetricsProperties == null) {
@@ -135,12 +143,12 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {

    private void initializeFlushTimer() {
        this.metricsFlushTimer = new HashedWheelTimer(r -> {
-            Thread thread = new Thread(r, "victoria-metrics-cluster-flush-timer");
+            Thread thread = new Thread(r, "victoria-metrics-flush-timer");
            thread.setDaemon(true);
            return thread;
-        }, 1, TimeUnit.SECONDS, 512);
-        // start flush interval timer
-        this.metricsFlushTimer.newTimeout(new MetricsFlushTask(null), insertConfig.flushInterval(), TimeUnit.SECONDS);
+        }, 100, TimeUnit.MILLISECONDS, 512);
+        metricsFlushTask = new MetricsFlushTask();
+        this.metricsFlushTimer.newTimeout(metricsFlushTask, insertConfig.flushInterval(), TimeUnit.SECONDS);
    }

    private boolean checkVictoriaMetricsDatasourceAvailable() {
@@ -169,6 +177,10 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {

    @Override
    public void saveData(CollectRep.MetricsData metricsData) {
+        if (closed.get()) {
+            log.warn("[Victoria Metrics] Rejecting metrics after storage shutdown");
+            return;
+        }
        if (!isServerAvailable()) {
            serverAvailable = checkVictoriaMetricsDatasourceAvailable();
        }
@@ -317,9 +329,22 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {

    @Override
    public void destroy() {
+        synchronized (metricsFlushLock) {
+            if (!closed.compareAndSet(false, true)) {
+                return;
+            }
+        }
        if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) {
            metricsFlushTimer.stop();
        }
+        immediateFlushPending.set(false);
+        while (hasPendingMetrics()) {
+            if (!flushBufferedMetrics()) {
+                log.error("[Victoria Metrics] Unable to flush {} buffered metrics during shutdown",
+                        getPendingMetricCount());
+                break;
+            }
+        }
    }

    @Override
@@ -630,54 +655,121 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {
     * @param contentList victoriaMetricsContent List
     */
    private void sendVictoriaMetrics(List<VictoriaMetricsDataStorage.VictoriaMetricsContent> contentList) {
-        for (VictoriaMetricsDataStorage.VictoriaMetricsContent content : contentList) {
-            boolean offered = false;
-            int retryCount = 0;
-            while (!offered && retryCount < MAX_RETRIES) {
-                try {
-                    // Attempt to add to the queue for a limited time
-                    offered = metricsBufferQueue.offer(content, MAX_WAIT_MS, TimeUnit.MILLISECONDS);
-                    if (!offered) {
-                        // If the queue is still full, trigger an immediate refresh to free up space
-                        if (retryCount == 0) {
-                            log.debug("victoria metrics buffer queue is full, triggering immediate flush");
+        for (int index = 0; index < contentList.size(); index++) {
+            VictoriaMetricsDataStorage.VictoriaMetricsContent content = contentList.get(index);
+            boolean offered = metricsBufferQueue.offer(content);
+            for (int attempt = 1; attempt <= MAX_BUFFER_OFFER_ATTEMPTS && !offered; attempt++) {
+                if (closed.get()) {
+                    return;
+                }
                triggerImmediateFlush();
-                        }
-                        retryCount++;
-                        // The short sleep allows the queue to clear out
-                        if (retryCount < MAX_RETRIES) {
-                            Thread.sleep(100L * retryCount);
-                        }
-                    }
+                try {
+                    offered = metricsBufferQueue.offer(content, MAX_WAIT_MS, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
-                    log.error("[Victoria Metrics] Interrupted while offering metrics to buffer queue", e);
-                    break;
+                    recordDroppedMetrics(contentList.size() - index, "producer interrupted");
+                    return;
                }
            }
-            // When the maximum number of retries is reached, if it still cannot be added to the queue, the data is saved directly
            if (!offered) {
-                log.warn("[Victoria Metrics] Failed to add metrics to buffer after {} retries, saving directly", MAX_RETRIES);
-                try {
-                    doSaveData(contentList);
-                } catch (Exception e) {
-                    log.error("[Victoria Metrics] Failed to save metrics directly: {}", e.getMessage(), e);
+                recordDroppedMetrics(contentList.size() - index, "buffer remained full");
+                return;
            }
-            }
-        }
-        // Refresh in advance to avoid waiting
-        if (metricsBufferQueue.size() >= insertConfig.bufferSize() * 0.8
-            && draining.compareAndSet(false, true)) {
+            if (metricsBufferQueue.size() >= insertConfig.bufferSize() * 0.8) {
                triggerImmediateFlush();
            }
        }
+    }
+
+    private void recordDroppedMetrics(int count, String reason) {
+        long total = droppedMetricCount.addAndGet(count);
+        if (total == count || (total & (total - 1)) == 0 || total % 100 == 0) {
+            log.error("[Victoria Metrics] Dropped {} metrics because {}; cumulative dropped metrics: {}",
+                    count, reason, total);
+        }
+    }
+
+    @Override
+    public long getDroppedMetricCount() {
+        return droppedMetricCount.get();
+    }

    private void triggerImmediateFlush() {
-        List<VictoriaMetricsDataStorage.VictoriaMetricsContent> batch = new ArrayList<>(insertConfig.bufferSize());
-        metricsBufferQueue.drainTo(batch, insertConfig.bufferSize());
-        draining.set(false);
-        if (!batch.isEmpty()) {
-            metricsFlushTimer.newTimeout(new MetricsFlushTask(batch), 0, TimeUnit.MILLISECONDS);
+        scheduleImmediateFlush(0, TimeUnit.MILLISECONDS);
+    }
+
+    private void scheduleImmediateFlush(long delay, TimeUnit unit) {
+        if (closed.get() || metricsFlushTimer == null || metricsFlushTimer.isStop()) {
+            return;
+        }
+        if (immediateFlushPending.compareAndSet(false, true)) {
+            try {
+                metricsFlushTimer.newTimeout(new ImmediateMetricsFlushTask(), delay, unit);
+            } catch (RuntimeException e) {
+                immediateFlushPending.set(false);
+                if (!closed.get()) {
+                    log.warn("[Victoria Metrics] Unable to schedule immediate flush: {}", e.getMessage());
+                }
+            }
+        }
+    }
+
+    private boolean flushBufferedMetrics() {
+        if (!flushInFlight.compareAndSet(false, true)) {
+            return false;
+        }
+        try {
+            List<VictoriaMetricsContent> batch;
+            synchronized (metricsFlushLock) {
+                if (retryBatch.isEmpty()) {
+                    List<VictoriaMetricsContent> nextBatch = new ArrayList<>(insertConfig.bufferSize());
+                    metricsBufferQueue.drainTo(nextBatch, insertConfig.bufferSize());
+                    retryBatch = nextBatch;
+                }
+                batch = retryBatch;
+            }
+            if (batch.isEmpty()) {
+                return true;
+            }
+            if (!trySaveData(batch)) {
+                log.warn("[Victoria Metrics] Retaining {} metrics items for retry", batch.size());
+                return false;
+            }
+            synchronized (metricsFlushLock) {
+                if (retryBatch == batch) {
+                    retryBatch = Collections.emptyList();
+                }
+            }
+            log.debug("[Victoria Metrics] Flushed {} metrics items", batch.size());
+            return true;
+        } finally {
+            flushInFlight.set(false);
+        }
+    }
+
+    private boolean hasPendingMetrics() {
+        synchronized (metricsFlushLock) {
+            return !retryBatch.isEmpty() || !metricsBufferQueue.isEmpty();
+        }
+    }
+
+    @Override
+    public int getPendingMetricCount() {
+        synchronized (metricsFlushLock) {
+            return retryBatch.size() + metricsBufferQueue.size();
+        }
+    }
+
+    private void schedulePeriodicFlush() {
+        if (closed.get() || metricsFlushTimer == null || metricsFlushTimer.isStop()) {
+            return;
+        }
+        try {
+            metricsFlushTimer.newTimeout(metricsFlushTask, insertConfig.flushInterval(), TimeUnit.SECONDS);
+        } catch (RuntimeException e) {
+            if (!closed.get()) {
+                log.warn("[Victoria Metrics] Unable to schedule periodic flush: {}", e.getMessage());
+            }
        }
    }

@@ -685,42 +777,46 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {
     * Regularly refresh the buffer queue to the vm
     */
    private class MetricsFlushTask implements TimerTask {
-        private final List<VictoriaMetricsDataStorage.VictoriaMetricsContent> batch;
-
-        public MetricsFlushTask(List<VictoriaMetricsDataStorage.VictoriaMetricsContent> batch) {
-            this.batch = batch;
-        }
-
        @Override
        public void run(Timeout timeout) {
+            boolean flushSucceeded = false;
            try {
-                if (batch == null) {
-                    // If the batch is null, it means that the timer is triggered by flush interval timer
-                    List<VictoriaMetricsDataStorage.VictoriaMetricsContent> batchT = new ArrayList<>(insertConfig.bufferSize());
-                    metricsBufferQueue.drainTo(batchT, insertConfig.bufferSize());
-                    triggerDoSaveData(batchT);
-                    // Reschedule the next flush task
-                    triggerIntervalFlushTimer();
-                } else {
-                    // If the batch is not null, it means that the timer is triggered by the immediate flush
-                    triggerDoSaveData(batch);
-                }
+                flushSucceeded = flushBufferedMetrics();
            } catch (Exception e) {
                log.error("[VictoriaMetrics] flush task error: {}", e.getMessage(), e);
+            } finally {
+                if (!flushSucceeded && hasPendingMetrics() && !closed.get()) {
+                    scheduleImmediateFlush(FAILED_FLUSH_RETRY_SECONDS, TimeUnit.SECONDS);
+                }
+                schedulePeriodicFlush();
+            }
        }
    }

-        private void triggerDoSaveData(List<VictoriaMetricsContent> batch) {
-            if (!batch.isEmpty()) {
-                doSaveData(batch);
-                log.debug("[Victoria Metrics] Flushed {} metrics items", batch.size());
+    private class ImmediateMetricsFlushTask implements TimerTask {
+        @Override
+        public void run(Timeout timeout) {
+            boolean flushSucceeded = false;
+            try {
+                int flushedBatches = 0;
+                do {
+                    flushSucceeded = flushBufferedMetrics();
+                    flushedBatches++;
+                } while (flushSucceeded
+                        && hasPendingMetrics()
+                        && !closed.get()
+                        && flushedBatches < MAX_BATCHES_PER_FLUSH_TASK);
+            } catch (Exception e) {
+                log.error("[VictoriaMetrics] immediate flush task error: {}", e.getMessage(), e);
+            } finally {
+                immediateFlushPending.set(false);
+                if (hasPendingMetrics() && !closed.get()) {
+                    if (flushSucceeded) {
+                        scheduleImmediateFlush(0, TimeUnit.MILLISECONDS);
+                    } else {
+                        scheduleImmediateFlush(FAILED_FLUSH_RETRY_SECONDS, TimeUnit.SECONDS);
                    }
                }
-
-        private void triggerIntervalFlushTimer() {
-            if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) {
-                metricsFlushTimer.newTimeout(new MetricsFlushTask(null), insertConfig.flushInterval(), TimeUnit.SECONDS);
-                log.debug("[Victoria Metrics] Rescheduled next flush task in {} seconds.", insertConfig.flushInterval());
            }
        }
    }
@@ -728,7 +824,7 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {
    /**
     * Save metric data to victoria-metric via HTTP call
     */
-    private void doSaveData(List<VictoriaMetricsContent> contentList) {
+    private boolean trySaveData(List<VictoriaMetricsContent> contentList) {
        try {
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
@@ -765,12 +861,15 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {
                httpEntity, String.class);
            if (responseEntity.getStatusCode().is2xxSuccessful()) {
                log.debug("insert metrics data to victoria-metrics success.");
+                return true;
            } else {
-                log.error("insert metrics data to victoria-metrics failed. {}", responseEntity.getBody());
+                log.error("insert metrics data to victoria-metrics failed with status {}",
+                        responseEntity.getStatusCode());
            }
        } catch (Exception e){
            log.error("flush metrics data to victoria-metrics error: {}.", e.getMessage(), e);
        }
+        return false;
    }

 }
@@ -30,6 +30,7 @@ import org.apache.hertzbeat.common.entity.dto.Field;
 import org.apache.hertzbeat.common.entity.dto.MetricsData;
 import org.apache.hertzbeat.common.entity.dto.MetricsHistoryData;
 import org.apache.hertzbeat.warehouse.service.MetricsDataService;
+import org.apache.hertzbeat.warehouse.service.WarehouseStorageStatus;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -64,18 +65,24 @@ class MetricsDataControllerTest {

    @Test
    void getWarehouseStorageServerStatus() throws Exception {
-        when(metricsDataService.getWarehouseStorageServerStatus()).thenReturn(true);
+        when(metricsDataService.getWarehouseStorageStatus())
+                .thenReturn(new WarehouseStorageStatus(true, 0, 0));
        this.mockMvc.perform(MockMvcRequestBuilders.get("/api/warehouse/storage/status"))
                .andExpect(status().isOk())
                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
-                .andExpect(jsonPath("$.data").isEmpty())
+                .andExpect(jsonPath("$.data.available").value(true))
+                .andExpect(jsonPath("$.data.droppedMetrics").value(0))
+                .andExpect(jsonPath("$.data.pendingMetrics").value(0))
                .andExpect(jsonPath("$.msg").isEmpty())
                .andReturn();
-        when(metricsDataService.getWarehouseStorageServerStatus()).thenReturn(false);
+        when(metricsDataService.getWarehouseStorageStatus())
+                .thenReturn(new WarehouseStorageStatus(false, 7, 3));
        this.mockMvc.perform(MockMvcRequestBuilders.get("/api/warehouse/storage/status"))
                .andExpect(status().isOk())
                .andExpect(jsonPath("$.code").value((int) CommonConstants.FAIL_CODE))
-                .andExpect(jsonPath("$.data").isEmpty())
+                .andExpect(jsonPath("$.data.available").value(false))
+                .andExpect(jsonPath("$.data.droppedMetrics").value(7))
+                .andExpect(jsonPath("$.data.pendingMetrics").value(3))
                .andReturn();
    }

@@ -17,6 +17,7 @@

 package org.apache.hertzbeat.warehouse.service;

+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -68,7 +69,10 @@ public class MetricsDataServiceTest {
        assertFalse(metricsDataService.getWarehouseStorageServerStatus());

        when(historyDataReader.isServerAvailable()).thenReturn(true);
+        when(historyDataReader.getDroppedMetricCount()).thenReturn(7L);
+        when(historyDataReader.getPendingMetricCount()).thenReturn(3);
        assertTrue(metricsDataService.getWarehouseStorageServerStatus());
+        assertEquals(new WarehouseStorageStatus(true, 7, 3), metricsDataService.getWarehouseStorageStatus());
    }

    @Test
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.warehouse.store.history.tsdb.vm;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hertzbeat.common.timer.TimerTask;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.web.client.RestTemplate;
+
+/**
+ * Test case for {@link VictoriaMetricsClusterDataStorage}.
+ */
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class VictoriaMetricsClusterDataStorageTest {
+
+    @Mock
+    private RestTemplate restTemplate;
+
+    @Test
+    void flushesDataAddedWhileAnImmediateFlushIsRunning() throws Exception {
+        mockHealthCheck();
+        CountDownLatch firstWriteStarted = new CountDownLatch(1);
+        CountDownLatch releaseFirstWrite = new CountDownLatch(1);
+        List<String> successfulBodies = new CopyOnWriteArrayList<>();
+        AtomicInteger writes = new AtomicInteger();
+        when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
+                .thenAnswer(invocation -> {
+                    HttpEntity<String> request = invocation.getArgument(1);
+                    if (writes.getAndIncrement() == 0) {
+                        firstWriteStarted.countDown();
+                        assertThat(releaseFirstWrite.await(5, TimeUnit.SECONDS)).isTrue();
+                    }
+                    successfulBodies.add(request.getBody());
+                    return ResponseEntity.noContent().build();
+                });
+        VictoriaMetricsClusterDataStorage storage = createStorage(2, 3600);
+
+        try {
+            // Allow the constructor's initial empty periodic run to settle.
+            Thread.sleep(1200);
+            saveOneMetric(storage);
+            saveOneMetric(storage);
+            assertThat(firstWriteStarted.await(5, TimeUnit.SECONDS)).isTrue();
+
+            saveOneMetric(storage);
+            saveOneMetric(storage);
+            releaseFirstWrite.countDown();
+
+            await().atMost(5, TimeUnit.SECONDS)
+                    .untilAsserted(() -> assertThat(successfulBodies).hasSize(2));
+            assertThat(successfulBodies.stream().mapToLong(VictoriaMetricsClusterDataStorageTest::lineCount).sum())
+                    .isEqualTo(4);
+        } finally {
+            releaseFirstWrite.countDown();
+            storage.destroy();
+        }
+    }
+
+    @Test
+    void clusterFlushClaimsTheRetryBatchWhileTheHttpWriteIsInFlight() throws Exception {
+        mockHealthCheck();
+        CountDownLatch firstWriteStarted = new CountDownLatch(1);
+        CountDownLatch releaseFirstWrite = new CountDownLatch(1);
+        AtomicInteger writes = new AtomicInteger();
+        when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
+                .thenAnswer(invocation -> {
+                    if (writes.incrementAndGet() == 1) {
+                        firstWriteStarted.countDown();
+                        assertThat(releaseFirstWrite.await(5, TimeUnit.SECONDS)).isTrue();
+                    }
+                    return ResponseEntity.noContent().build();
+                });
+        VictoriaMetricsClusterDataStorage storage = createStorage(10, 3600);
+        ExecutorService flushers = Executors.newFixedThreadPool(2);
+
+        try {
+            Thread.sleep(1200);
+            saveOneMetric(storage);
+            Future<Boolean> first = flushers.submit(() ->
+                    ReflectionTestUtils.invokeMethod(storage, "flushBufferedMetrics"));
+            assertThat(firstWriteStarted.await(5, TimeUnit.SECONDS)).isTrue();
+            Future<Boolean> second = flushers.submit(() ->
+                    ReflectionTestUtils.invokeMethod(storage, "flushBufferedMetrics"));
+
+            assertThat(second.get(2, TimeUnit.SECONDS)).isFalse();
+            assertThat(writes).hasValue(1);
+            releaseFirstWrite.countDown();
+            assertThat(first.get(2, TimeUnit.SECONDS)).isTrue();
+            assertThat(writes).hasValue(1);
+        } finally {
+            releaseFirstWrite.countDown();
+            storage.destroy();
+            flushers.shutdownNow();
+        }
+    }
+
+    @Test
+    void retriesPeriodicFlushFailuresQuicklyWhenTheConfiguredIntervalIsLong() throws Exception {
+        mockHealthCheck();
+        List<String> attemptedBodies = new CopyOnWriteArrayList<>();
+        AtomicInteger writes = new AtomicInteger();
+        when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
+                .thenAnswer(invocation -> {
+                    HttpEntity<String> request = invocation.getArgument(1);
+                    attemptedBodies.add(request.getBody());
+                    if (writes.getAndIncrement() == 0) {
+                        return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).build();
+                    }
+                    return ResponseEntity.noContent().build();
+                });
+        VictoriaMetricsClusterDataStorage storage = createStorage(10, 3600);
+
+        try {
+            // Let the constructor's initial empty periodic run schedule the
+            // production-length interval, then invoke that periodic path.
+            Thread.sleep(1200);
+            saveOneMetric(storage);
+            TimerTask periodicTask = (TimerTask) ReflectionTestUtils.getField(storage, "metricsFlushtask");
+            assertThat(periodicTask).isNotNull();
+            periodicTask.run(null);
+
+            await().atMost(4, TimeUnit.SECONDS)
+                    .untilAsserted(() -> assertThat(attemptedBodies).hasSizeGreaterThanOrEqualTo(2));
+            assertThat(attemptedBodies.get(1)).isEqualTo(attemptedBodies.get(0));
+            assertThat(lineCount(attemptedBodies.get(1))).isEqualTo(1);
+        } finally {
+            storage.destroy();
+        }
+    }
+
+    @Test
+    void destroyFlushesBufferedMetricsAndRejectsLaterWrites() {
+        mockHealthCheck();
+        List<String> successfulBodies = new CopyOnWriteArrayList<>();
+        when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
+                .thenAnswer(invocation -> {
+                    HttpEntity<String> request = invocation.getArgument(1);
+                    successfulBodies.add(request.getBody());
+                    return ResponseEntity.noContent().build();
+                });
+        VictoriaMetricsClusterDataStorage storage = createStorage(10, 3600);
+
+        saveOneMetric(storage);
+        storage.destroy();
+        saveOneMetric(storage);
+
+        assertThat(successfulBodies).hasSize(1);
+        assertThat(lineCount(successfulBodies.get(0))).isEqualTo(1);
+    }
+
+    @Test
+    void persistentWriteFailureDoesNotBlockTheWarehouseProducerIndefinitely() throws Exception {
+        mockHealthCheck();
+        AtomicInteger writes = new AtomicInteger();
+        when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
+                .thenAnswer(invocation -> {
+                    writes.incrementAndGet();
+                    return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).build();
+                });
+        VictoriaMetricsClusterDataStorage storage = createStorage(1, 3600);
+        ExecutorService producer = Executors.newSingleThreadExecutor();
+        Future<?> pendingWrite = null;
+
+        try {
+            Thread.sleep(1200);
+            saveOneMetric(storage);
+            await().atMost(3, TimeUnit.SECONDS).until(() -> writes.get() > 0);
+            saveOneMetric(storage);
+
+            pendingWrite = producer.submit(() -> saveOneMetric(storage));
+
+            pendingWrite.get(3, TimeUnit.SECONDS);
+            assertThat(storage.getDroppedMetricCount()).isGreaterThanOrEqualTo(1);
+        } finally {
+            if (pendingWrite != null) {
+                pendingWrite.cancel(true);
+            }
+            storage.destroy();
+            producer.shutdownNow();
+        }
+    }
+
+    private void mockHealthCheck() {
+        when(restTemplate.exchange(
+                anyString(),
+                eq(HttpMethod.GET),
+                any(HttpEntity.class),
+                eq(String.class)))
+                .thenReturn(ResponseEntity.ok("{\"status\":\"success\"}"));
+    }
+
+    private VictoriaMetricsClusterDataStorage createStorage(int bufferSize, int flushInterval) {
+        VictoriaMetricsInsertProperties insert =
+                new VictoriaMetricsInsertProperties("http://localhost:8480", null, null, bufferSize, flushInterval);
+        VictoriaMetricsSelectProperties select =
+                new VictoriaMetricsSelectProperties("http://localhost:8481", null, null);
+        VictoriaMetricsClusterProperties properties =
+                new VictoriaMetricsClusterProperties(true, "0", insert, select);
+        return new VictoriaMetricsClusterDataStorage(properties, restTemplate);
+    }
+
+    private static void saveOneMetric(VictoriaMetricsClusterDataStorage storage) {
+        storage.saveData(VictoriaMetricsDataStorageTest.generateMockedMetricsData());
+    }
+
+    private static long lineCount(String body) {
+        return body.lines().filter(line -> !line.isBlank()).count();
+    }
+}
@@ -55,10 +55,17 @@ import org.springframework.web.client.RestTemplate;

 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;

+import org.springframework.test.util.ReflectionTestUtils;
+
 /**
 * Test case for {@link VictoriaMetricsDataStorage}
 */
@@ -194,6 +201,110 @@ class VictoriaMetricsDataStorageTest {
                                .isGreaterThanOrEqualTo(threadCount * writeSize / bufferSize));
    }

+    @Test
+    void failedSingleNodeFlushRetainsTheBatchAndRetriesQuickly() {
+        when(victoriaMetricsProperties.insert()).thenReturn(new VictoriaMetricsProperties.InsertConfig(
+                10, 1, new VictoriaMetricsProperties.Compression(false)));
+        List<String> attemptedBodies = new CopyOnWriteArrayList<>();
+        AtomicInteger writes = new AtomicInteger();
+        when(restTemplate.postForEntity(
+                startsWith(victoriaMetricsProperties.url()),
+                any(HttpEntity.class),
+                eq(String.class)
+        )).thenAnswer(invocation -> {
+            HttpEntity<String> request = invocation.getArgument(1);
+            attemptedBodies.add(request.getBody());
+            if (writes.getAndIncrement() == 0) {
+                return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).build();
+            }
+            return ResponseEntity.noContent().build();
+        });
+        victoriaMetricsDataStorage = new VictoriaMetricsDataStorage(victoriaMetricsProperties, restTemplate);
+
+        victoriaMetricsDataStorage.saveData(generateMockedMetricsData());
+
+        Awaitility.await()
+                .pollInterval(250, TimeUnit.MILLISECONDS)
+                .atMost(7, TimeUnit.SECONDS)
+                .untilAsserted(() -> assertThat(attemptedBodies).hasSizeGreaterThanOrEqualTo(2));
+        assertThat(attemptedBodies.get(1)).isEqualTo(attemptedBodies.get(0));
+    }
+
+    @Test
+    void singleNodeFlushClaimsTheRetryBatchWhileTheHttpWriteIsInFlight() throws Exception {
+        when(victoriaMetricsProperties.insert()).thenReturn(new VictoriaMetricsProperties.InsertConfig(
+                10, 3600, new VictoriaMetricsProperties.Compression(false)));
+        CountDownLatch firstWriteStarted = new CountDownLatch(1);
+        CountDownLatch releaseFirstWrite = new CountDownLatch(1);
+        AtomicInteger writes = new AtomicInteger();
+        when(restTemplate.postForEntity(
+                startsWith(victoriaMetricsProperties.url()),
+                any(HttpEntity.class),
+                eq(String.class)
+        )).thenAnswer(invocation -> {
+            if (writes.incrementAndGet() == 1) {
+                firstWriteStarted.countDown();
+                assertThat(releaseFirstWrite.await(5, TimeUnit.SECONDS)).isTrue();
+            }
+            return ResponseEntity.noContent().build();
+        });
+        victoriaMetricsDataStorage = new VictoriaMetricsDataStorage(victoriaMetricsProperties, restTemplate);
+        victoriaMetricsDataStorage.saveData(generateMockedMetricsData());
+        ExecutorService flushers = Executors.newFixedThreadPool(2);
+
+        try {
+            Future<Boolean> first = flushers.submit(() ->
+                    ReflectionTestUtils.invokeMethod(victoriaMetricsDataStorage, "flushBufferedMetrics"));
+            assertThat(firstWriteStarted.await(5, TimeUnit.SECONDS)).isTrue();
+            Future<Boolean> second = flushers.submit(() ->
+                    ReflectionTestUtils.invokeMethod(victoriaMetricsDataStorage, "flushBufferedMetrics"));
+
+            assertThat(second.get(2, TimeUnit.SECONDS)).isFalse();
+            assertThat(writes).hasValue(1);
+            releaseFirstWrite.countDown();
+            assertThat(first.get(2, TimeUnit.SECONDS)).isTrue();
+            assertThat(writes).hasValue(1);
+        } finally {
+            releaseFirstWrite.countDown();
+            flushers.shutdownNow();
+        }
+    }
+
+    @Test
+    void persistentSingleNodeFailureDoesNotBlockTheWarehouseProducerIndefinitely() throws Exception {
+        when(victoriaMetricsProperties.insert()).thenReturn(new VictoriaMetricsProperties.InsertConfig(
+                1, 3600, new VictoriaMetricsProperties.Compression(false)));
+        AtomicInteger writes = new AtomicInteger();
+        when(restTemplate.postForEntity(
+                startsWith(victoriaMetricsProperties.url()),
+                any(HttpEntity.class),
+                eq(String.class)
+        )).thenAnswer(invocation -> {
+            writes.incrementAndGet();
+            return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).build();
+        });
+        victoriaMetricsDataStorage = new VictoriaMetricsDataStorage(victoriaMetricsProperties, restTemplate);
+        ExecutorService producer = Executors.newSingleThreadExecutor();
+        Future<?> pendingWrite = null;
+
+        try {
+            victoriaMetricsDataStorage.saveData(generateMockedMetricsData());
+            Awaitility.await().atMost(3, TimeUnit.SECONDS).until(() -> writes.get() > 0);
+            victoriaMetricsDataStorage.saveData(generateMockedMetricsData());
+
+            pendingWrite = producer.submit(
+                    () -> victoriaMetricsDataStorage.saveData(generateMockedMetricsData()));
+
+            pendingWrite.get(3, TimeUnit.SECONDS);
+            assertThat(victoriaMetricsDataStorage.getDroppedMetricCount()).isGreaterThanOrEqualTo(1);
+        } finally {
+            if (pendingWrite != null) {
+                pendingWrite.cancel(true);
+            }
+            producer.shutdownNow();
+        }
+    }
+
    @Test
    void customLabelsKeepJobButCannotOverrideStorageInstance() {
        when(victoriaMetricsProperties.insert()).thenReturn(new VictoriaMetricsProperties.InsertConfig(