This is an automated email from the ASF dual-hosted git repository.
Duansg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
The following commit(s) were added to refs/heads/master by this push:
new c884f11995 maintenance: make VictoriaMetrics flush recovery durable
(#4289)
c884f11995 is described below
commit c884f11995d46239a424e6d3251af113feb3a694
Author: Logic <[email protected]>
AuthorDate: Wed Aug 26 10:58:09 2026 +0800
maintenance: make VictoriaMetrics flush recovery durable (#4289)
Co-authored-by: Duansg <[email protected]>
---
.../controller/MetricsDataController.java | 15 +-
.../warehouse/service/MetricsDataService.java | 7 +
.../warehouse/service/WarehouseStorageStatus.java | 30 +++
.../service/impl/MetricsDataServiceImpl.java | 13 +-
.../store/history/tsdb/HistoryDataReader.java | 14 ++
.../tsdb/vm/VictoriaMetricsClusterDataStorage.java | 218 ++++++++++++++----
.../tsdb/vm/VictoriaMetricsDataStorage.java | 241 ++++++++++++++------
.../controller/MetricsDataControllerTest.java | 15 +-
.../warehouse/service/MetricsDataServiceTest.java | 4 +
.../vm/VictoriaMetricsClusterDataStorageTest.java | 249 +++++++++++++++++++++
.../tsdb/vm/VictoriaMetricsDataStorageTest.java | 111 +++++++++
11 files changed, 793 insertions(+), 124 deletions(-)
diff --git
a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/controller/MetricsDataController.java
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/controller/MetricsDataController.java
index 832059cd2d..b9844a8539 100644
---
a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/controller/MetricsDataController.java
+++
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/controller/MetricsDataController.java
@@ -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}")
diff --git
a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/MetricsDataService.java
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/MetricsDataService.java
index 6c39b78d58..ea2222cc31 100644
---
a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/MetricsDataService.java
+++
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/MetricsDataService.java
@@ -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
diff --git
a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/WarehouseStorageStatus.java
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/WarehouseStorageStatus.java
new file mode 100644
index 0000000000..439db822fd
--- /dev/null
+++
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/WarehouseStorageStatus.java
@@ -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) {
+}
diff --git
a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/impl/MetricsDataServiceImpl.java
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/impl/MetricsDataServiceImpl.java
index df336e5e3a..a1c81ebc8e 100644
---
a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/impl/MetricsDataServiceImpl.java
+++
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/impl/MetricsDataServiceImpl.java
@@ -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
diff --git
a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/HistoryDataReader.java
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/HistoryDataReader.java
index 08a42bb919..fd821a7bfd 100644
---
a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/HistoryDataReader.java
+++
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/HistoryDataReader.java
@@ -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
*/
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 3777bb1830..e808f62202 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.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
@@ -599,7 +625,11 @@ public class VictoriaMetricsClusterDataStorage extends
AbstractHistoryDataStorag
/**
* Save metric data to victoria-metric via HTTP call
*/
- public void
doSaveData(List<VictoriaMetricsDataStorage.VictoriaMetricsContent> contentList){
+ 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) {
+ 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;
}
- // 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);
+ }
+ }
}
}
}
diff --git
a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorage.java
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorage.java
index 5a80c9824c..7f199ff4f7 100644
---
a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorage.java
+++
b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorage.java
@@ -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) {
+ 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);
}
}
+ @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 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);
+ }
+ }
}
}
}
@@ -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;
}
}
diff --git
a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/controller/MetricsDataControllerTest.java
b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/controller/MetricsDataControllerTest.java
index 674f2b5d33..ce26d541f7 100644
---
a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/controller/MetricsDataControllerTest.java
+++
b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/controller/MetricsDataControllerTest.java
@@ -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();
}
diff --git
a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/service/MetricsDataServiceTest.java
b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/service/MetricsDataServiceTest.java
index b73646a9e9..bf8f88a834 100644
---
a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/service/MetricsDataServiceTest.java
+++
b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/service/MetricsDataServiceTest.java
@@ -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
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 0000000000..960b28ea89
--- /dev/null
+++
b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java
@@ -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();
+ }
+}
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 dd0ac7d9f3..31528e84cc 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
@@ -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(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
