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> getWarehouseStorageServerStatus() { - Boolean status = metricsDataService.getWarehouseStorageServerStatus(); - if (Boolean.TRUE.equals(status)) { - return ResponseEntity.ok(Message.success()); + public ResponseEntity> getWarehouseStorageServerStatus() { + WarehouseStorageStatus status = metricsDataService.getWarehouseStorageStatus(); + Message 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 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 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 contentList){ + public void doSaveData(List contentList) { + trySaveData(contentList); + } + + private boolean trySaveData(List 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 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 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; + } 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 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 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 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 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 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 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; + } 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 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); + } + } } } } @@ -728,7 +824,7 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage { /** * 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); @@ -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 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 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 first = flushers.submit(() -> + ReflectionTestUtils.invokeMethod(storage, "flushBufferedMetrics")); + assertThat(firstWriteStarted.await(5, TimeUnit.SECONDS)).isTrue(); + Future 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 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); + } + + @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 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 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 first = flushers.submit(() -> + ReflectionTestUtils.invokeMethod(victoriaMetricsDataStorage, "flushBufferedMetrics")); + assertThat(firstWriteStarted.await(5, TimeUnit.SECONDS)).isTrue(); + Future 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(