From 419e0e8658164d00888b2af95a854095962680a2 Mon Sep 17 00:00:00 2001 From: Logic Date: Thu, 30 Jul 2026 17:34:11 +0800 Subject: [PATCH 1/2] maintenance: make VictoriaMetrics flush recovery durable --- .../vm/VictoriaMetricsClusterDataStorage.java | 190 ++++++++++++++---- ...VictoriaMetricsClusterDataStorageTest.java | 176 ++++++++++++++++ 2 files changed, 328 insertions(+), 38 deletions(-) create mode 100644 hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java index ae93a1e7e2c..4e246d80b24 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java @@ -36,6 +36,7 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -100,16 +101,20 @@ 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 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 metricsBufferQueue; + private final Object metricsFlushLock = new Object(); private HashedWheelTimer metricsFlushTimer = null; private MetricsFlushTask metricsFlushtask = null; + private final AtomicBoolean immediateFlushPending = new AtomicBoolean(false); + private final AtomicBoolean closed = new AtomicBoolean(false); + private List retryBatch = Collections.emptyList(); private boolean isBatchImportEnabled = false; @@ -175,6 +180,10 @@ private boolean checkVictoriaMetricsDatasourceAvailable() { @Override public void saveData(CollectRep.MetricsData metricsData) { + if (closed.get()) { + log.warn("[Victoria Metrics] Rejecting metrics after storage shutdown"); + return; + } if (!isServerAvailable()) { serverAvailable = checkVictoriaMetricsDatasourceAvailable(); } @@ -278,9 +287,22 @@ public void saveData(CollectRep.MetricsData metricsData) { @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", + pendingMetricCount()); + break; + } + } } @Override @@ -579,7 +601,11 @@ public Map> getHistoryIntervalMetricData(String instance, St /** * Save metric data to victoria-metric via HTTP call */ - public void doSaveData(List contentList){ + public void doSaveData(List contentList) { + trySaveData(contentList); + } + + private boolean trySaveData(List contentList) { try { HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); @@ -599,12 +625,15 @@ public void doSaveData(List c 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; } /** @@ -614,39 +643,34 @@ public void doSaveData(List c private void sendVictoriaMetrics(List 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"); - triggerImmediateFlush(); - } - retryCount++; - // The short sleep allows the queue to clear out - if (retryCount < MAX_RETRIES) { - Thread.sleep(100L * retryCount); - } + boolean backpressureLogged = false; + while (!offered) { + synchronized (metricsFlushLock) { + if (closed.get()) { + break; } + offered = metricsBufferQueue.offer(content); + } + if (offered) { + break; + } + if (!backpressureLogged) { + log.warn("[Victoria Metrics] Buffer is full; applying backpressure until a flush succeeds"); + backpressureLogged = true; + } + triggerImmediateFlush(); + try { + Thread.sleep(MAX_WAIT_MS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - log.error("[Victoria Metrics] Interrupted while offering metrics to buffer queue", e); + log.error("[Victoria Metrics] Interrupted while buffering metrics"); break; } } - // 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); - } + log.warn("[Victoria Metrics] Metric was not buffered because storage is shutting down"); + return; } - // Refresh in advance to avoid waiting if (metricsBufferQueue.size() >= vmInsertProps.bufferSize() * 0.8) { triggerImmediateFlush(); } @@ -654,7 +678,75 @@ private void sendVictoriaMetrics(List batch; + synchronized (metricsFlushLock) { + if (retryBatch.isEmpty()) { + List 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; + } + + private boolean hasPendingMetrics() { + synchronized (metricsFlushLock) { + return !retryBatch.isEmpty() || !metricsBufferQueue.isEmpty(); + } + } + + private int pendingMetricCount() { + 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()); + } + } } /** @@ -663,18 +755,40 @@ private void triggerImmediateFlush() { private class MetricsFlushTask implements TimerTask { @Override public void run(Timeout timeout) { + boolean flushSucceeded = false; try { - List 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 { + flushSucceeded = flushBufferedMetrics(); + } 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); + } + } } } } diff --git a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java new file mode 100644 index 00000000000..6ea69dd7239 --- /dev/null +++ b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java @@ -0,0 +1,176 @@ +/* + * 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.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 successfulBodies = new CopyOnWriteArrayList<>(); + AtomicInteger writes = new AtomicInteger(); + when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class))) + .thenAnswer(invocation -> { + HttpEntity 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 retriesPeriodicFlushFailuresQuicklyWhenTheConfiguredIntervalIsLong() throws Exception { + mockHealthCheck(); + List attemptedBodies = new CopyOnWriteArrayList<>(); + AtomicInteger writes = new AtomicInteger(); + when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class))) + .thenAnswer(invocation -> { + HttpEntity 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 successfulBodies = new CopyOnWriteArrayList<>(); + when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class))) + .thenAnswer(invocation -> { + HttpEntity 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); + } + + 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(); + } +} From ab4fc80bd056f1ce6dc2ec12c800240568ee019f Mon Sep 17 00:00:00 2001 From: Logic Date: Tue, 4 Aug 2026 20:48:45 +0800 Subject: [PATCH 2/2] Bound VictoriaMetrics producer backpressure --- .../vm/VictoriaMetricsClusterDataStorage.java | 57 +++-- .../tsdb/vm/VictoriaMetricsDataStorage.java | 232 ++++++++++++------ ...VictoriaMetricsClusterDataStorageTest.java | 35 +++ .../vm/VictoriaMetricsDataStorageTest.java | 68 +++++ 4 files changed, 299 insertions(+), 93 deletions(-) diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java index 4e246d80b24..3b88bc88d81 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java @@ -37,6 +37,7 @@ 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,6 +102,8 @@ 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_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; @@ -114,6 +117,7 @@ public class VictoriaMetricsClusterDataStorage extends AbstractHistoryDataStorag private MetricsFlushTask metricsFlushtask = null; private final AtomicBoolean immediateFlushPending = new AtomicBoolean(false); private final AtomicBoolean closed = new AtomicBoolean(false); + private final AtomicLong droppedMetricCount = new AtomicLong(); private List retryBatch = Collections.emptyList(); private boolean isBatchImportEnabled = false; @@ -141,7 +145,7 @@ private void initializeFlushTimer() { 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); } @@ -641,34 +645,24 @@ private boolean trySaveData(List contentList) { - for (VictoriaMetricsDataStorage.VictoriaMetricsContent content : contentList) { - boolean offered = false; - boolean backpressureLogged = false; - while (!offered) { - synchronized (metricsFlushLock) { - if (closed.get()) { - break; - } - offered = metricsBufferQueue.offer(content); - } - if (offered) { - break; - } - if (!backpressureLogged) { - log.warn("[Victoria Metrics] Buffer is full; applying backpressure until a flush succeeds"); - backpressureLogged = true; + 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(); try { - Thread.sleep(MAX_WAIT_MS); + offered = metricsBufferQueue.offer(content, MAX_WAIT_MS, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - log.error("[Victoria Metrics] Interrupted while buffering metrics"); - break; + recordDroppedMetrics(contentList.size() - index, "producer interrupted"); + return; } } if (!offered) { - log.warn("[Victoria Metrics] Metric was not buffered because storage is shutting down"); + recordDroppedMetrics(contentList.size() - index, "buffer remained full"); return; } if (metricsBufferQueue.size() >= vmInsertProps.bufferSize() * 0.8) { @@ -677,6 +671,18 @@ private void sendVictoriaMetrics(List metricsBufferQueue; + private final Object metricsFlushLock = new Object(); 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 closed = new AtomicBoolean(false); + private final AtomicLong droppedMetricCount = new AtomicLong(); + private List retryBatch = Collections.emptyList(); public VictoriaMetricsDataStorage(VictoriaMetricsProperties victoriaMetricsProperties, RestTemplate restTemplate) { if (victoriaMetricsProperties == null) { @@ -125,12 +133,12 @@ public VictoriaMetricsDataStorage(VictoriaMetricsProperties victoriaMetricsPrope 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() { @@ -159,6 +167,10 @@ private boolean checkVictoriaMetricsDatasourceAvailable() { @Override public void saveData(CollectRep.MetricsData metricsData) { + if (closed.get()) { + log.warn("[Victoria Metrics] Rejecting metrics after storage shutdown"); + return; + } if (!isServerAvailable()) { serverAvailable = checkVictoriaMetricsDatasourceAvailable(); } @@ -257,9 +269,22 @@ public void saveData(CollectRep.MetricsData metricsData) { @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", + pendingMetricCount()); + break; + } + } } @Override @@ -570,54 +595,112 @@ public static final class VictoriaMetricsContent { * @param contentList victoriaMetricsContent List */ private void sendVictoriaMetrics(List contentList) { - for (VictoriaMetricsDataStorage.VictoriaMetricsContent content : contentList) { - boolean offered = false; - int retryCount = 0; - while (!offered && retryCount < MAX_RETRIES) { + 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(); 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"); - triggerImmediateFlush(); - } - retryCount++; - // The short sleep allows the queue to clear out - if (retryCount < MAX_RETRIES) { - Thread.sleep(100L * retryCount); - } - } } 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; + } + if (metricsBufferQueue.size() >= insertConfig.bufferSize() * 0.8) { + triggerImmediateFlush(); } } - // Refresh in advance to avoid waiting - if (metricsBufferQueue.size() >= insertConfig.bufferSize() * 0.8 - && draining.compareAndSet(false, true)) { - 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); } } + long getDroppedMetricCount() { + return droppedMetricCount.get(); + } + private void triggerImmediateFlush() { - List 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() { + List batch; + synchronized (metricsFlushLock) { + if (retryBatch.isEmpty()) { + List 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; + } + + private boolean hasPendingMetrics() { + synchronized (metricsFlushLock) { + return !retryBatch.isEmpty() || !metricsBufferQueue.isEmpty(); + } + } + + private int pendingMetricCount() { + 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()); + } } } @@ -625,42 +708,46 @@ private void triggerImmediateFlush() { * Regularly refresh the buffer queue to the vm */ private class MetricsFlushTask implements TimerTask { - private final List batch; - - public MetricsFlushTask(List 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 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 batch) { - if (!batch.isEmpty()) { - doSaveData(batch); - log.debug("[Victoria Metrics] Flushed {} metrics items", batch.size()); - } - } - - 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()); + 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); + } + } } } } @@ -668,7 +755,7 @@ private void triggerIntervalFlushTimer() { /** * Save metric data to victoria-metric via HTTP call */ - private void doSaveData(List contentList) { + private boolean trySaveData(List contentList) { try { HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); @@ -705,12 +792,15 @@ private void doSaveData(List contentList) { 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; } } diff --git a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java index 6ea69dd7239..2cf9fb9a291 100644 --- a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java +++ b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java @@ -27,6 +27,9 @@ 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; @@ -147,6 +150,38 @@ void destroyFlushesBufferedMetricsAndRejectsLaterWrites() { 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(), diff --git a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorageTest.java b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorageTest.java index f11fdb905dc..d6b5afe3580 100644 --- a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorageTest.java +++ b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorageTest.java @@ -51,6 +51,10 @@ import org.springframework.web.client.RestTemplate; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +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; @@ -184,6 +188,70 @@ void testMultiThreadSaveDataBySize() { .isGreaterThanOrEqualTo(threadCount * writeSize / bufferSize)); } + @Test + void failedSingleNodeFlushRetainsTheBatchAndRetriesQuickly() { + when(victoriaMetricsProperties.insert()).thenReturn(new VictoriaMetricsProperties.InsertConfig( + 10, 1, new VictoriaMetricsProperties.Compression(false))); + List attemptedBodies = new CopyOnWriteArrayList<>(); + AtomicInteger writes = new AtomicInteger(); + when(restTemplate.postForEntity( + startsWith(victoriaMetricsProperties.url()), + any(HttpEntity.class), + eq(String.class) + )).thenAnswer(invocation -> { + HttpEntity 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 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(); + } + } + @AfterEach void stop() { if (victoriaMetricsDataStorage != null) {