diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java index ae93a1e7e2c..3b88bc88d81 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorage.java @@ -36,6 +36,8 @@ 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; @@ -100,16 +102,23 @@ 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 HashedWheelTimer metricsFlushTimer = null; private MetricsFlushTask metricsFlushtask = null; + private final AtomicBoolean immediateFlushPending = new AtomicBoolean(false); + private final AtomicBoolean closed = new AtomicBoolean(false); + private final AtomicLong droppedMetricCount = new AtomicLong(); + private List retryBatch = Collections.emptyList(); private boolean isBatchImportEnabled = false; @@ -136,7 +145,7 @@ private void initializeFlushTimer() { Thread thread = new Thread(r, "victoria-metrics-flush-timer"); thread.setDaemon(true); return thread; - }, 1, TimeUnit.SECONDS, 512); + }, 100, TimeUnit.MILLISECONDS, 512); metricsFlushtask = new MetricsFlushTask(); this.metricsFlushTimer.newTimeout(metricsFlushtask, 0, TimeUnit.SECONDS); } @@ -175,6 +184,10 @@ private boolean checkVictoriaMetricsDatasourceAvailable() { @Override public void saveData(CollectRep.MetricsData metricsData) { + if (closed.get()) { + log.warn("[Victoria Metrics] Rejecting metrics after storage shutdown"); + return; + } if (!isServerAvailable()) { serverAvailable = checkVictoriaMetricsDatasourceAvailable(); } @@ -278,9 +291,22 @@ public void saveData(CollectRep.MetricsData metricsData) { @Override public void destroy() { + synchronized (metricsFlushLock) { + if (!closed.compareAndSet(false, true)) { + return; + } + } if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) { metricsFlushTimer.stop(); } + immediateFlushPending.set(false); + while (hasPendingMetrics()) { + if (!flushBufferedMetrics()) { + log.error("[Victoria Metrics] Unable to flush {} buffered metrics during shutdown", + pendingMetricCount()); + break; + } + } } @Override @@ -579,7 +605,11 @@ public Map> getHistoryIntervalMetricData(String instance, St /** * Save metric data to victoria-metric via HTTP call */ - public void doSaveData(List contentList){ + public void doSaveData(List contentList) { + trySaveData(contentList); + } + + private boolean trySaveData(List contentList) { try { HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); @@ -599,12 +629,15 @@ public void doSaveData(List c httpEntity, String.class); if (responseEntity.getStatusCode().is2xxSuccessful()) { log.debug("insert metrics data to victoria-metrics success."); + return true; } else { - log.error("insert metrics data to victoria-metrics failed. {}", responseEntity.getBody()); + log.error("insert metrics data to victoria-metrics failed with status {}", + responseEntity.getStatusCode()); } } catch (Exception e){ log.error("flush metrics data to victoria-metrics error: {}.", e.getMessage(), e); } + return false; } /** @@ -612,49 +645,114 @@ public void doSaveData(List c * @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); + } + } + + 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() { + List batch; + synchronized (metricsFlushLock) { + if (retryBatch.isEmpty()) { + List nextBatch = + new ArrayList<>(vmInsertProps.bufferSize()); + metricsBufferQueue.drainTo(nextBatch, vmInsertProps.bufferSize()); + retryBatch = nextBatch; + } + batch = retryBatch; + } + if (batch.isEmpty()) { + return true; + } + if (!trySaveData(batch)) { + log.warn("[Victoria Metrics] Retaining {} metrics items for retry", batch.size()); + return false; + } + synchronized (metricsFlushLock) { + if (retryBatch == batch) { + retryBatch = Collections.emptyList(); + } + } + log.debug("[Victoria Metrics] Flushed {} metrics items", batch.size()); + return true; + } + + private boolean hasPendingMetrics() { + synchronized (metricsFlushLock) { + return !retryBatch.isEmpty() || !metricsBufferQueue.isEmpty(); + } + } + + private int pendingMetricCount() { + synchronized (metricsFlushLock) { + return retryBatch.size() + metricsBufferQueue.size(); + } + } + + private void schedulePeriodicFlush() { + if (closed.get() || metricsFlushTimer == null || metricsFlushTimer.isStop()) { + return; + } + try { + metricsFlushTimer.newTimeout(metricsFlushtask, vmInsertProps.flushInterval(), TimeUnit.SECONDS); + } catch (RuntimeException e) { + if (!closed.get()) { + log.warn("[Victoria Metrics] Unable to schedule periodic flush: {}", e.getMessage()); + } + } } /** @@ -663,18 +761,47 @@ private void triggerImmediateFlush() { private class MetricsFlushTask implements TimerTask { @Override public void run(Timeout timeout) { + boolean flushSucceeded = false; try { - List batch = new ArrayList<>(vmInsertProps.bufferSize()); - metricsBufferQueue.drainTo(batch, vmInsertProps.bufferSize()); - if (!batch.isEmpty()) { - doSaveData(batch); - log.debug("[Victoria Metrics] Flushed {} metrics items", batch.size()); - } - if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) { - metricsFlushTimer.newTimeout(this, vmInsertProps.flushInterval(), TimeUnit.SECONDS); - } + flushSucceeded = flushBufferedMetrics(); } catch (Exception e) { log.error("[VictoriaMetrics] flush task error: {}", e.getMessage(), e); + } finally { + if (!flushSucceeded && hasPendingMetrics() && !closed.get()) { + scheduleImmediateFlush(FAILED_FLUSH_RETRY_SECONDS, TimeUnit.SECONDS); + } + schedulePeriodicFlush(); + } + } + } + + /** + * Executes an immediate flush without creating another periodic chain. + */ + private class ImmediateMetricsFlushTask implements TimerTask { + @Override + public void run(Timeout timeout) { + boolean flushSucceeded = false; + try { + 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 5714e42186e..08c8142179f 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 @@ -36,6 +36,7 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.zip.GZIPOutputStream; import com.google.common.collect.Maps; @@ -99,15 +100,22 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage { 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 VictoriaMetricsProperties victoriaMetricsProp; private final RestTemplate restTemplate; private final BlockingQueue metricsBufferQueue; + private final Object metricsFlushLock = new Object(); private HashedWheelTimer metricsFlushTimer = null; + private MetricsFlushTask metricsFlushTask = null; private final VictoriaMetricsProperties.InsertConfig insertConfig; - private final AtomicBoolean draining = new AtomicBoolean(false); + private final AtomicBoolean immediateFlushPending = new AtomicBoolean(false); + private final AtomicBoolean closed = new AtomicBoolean(false); + private final AtomicLong droppedMetricCount = new AtomicLong(); + private List retryBatch = Collections.emptyList(); public VictoriaMetricsDataStorage(VictoriaMetricsProperties victoriaMetricsProperties, RestTemplate restTemplate) { if (victoriaMetricsProperties == null) { @@ -125,12 +133,12 @@ public VictoriaMetricsDataStorage(VictoriaMetricsProperties victoriaMetricsPrope private void initializeFlushTimer() { this.metricsFlushTimer = new HashedWheelTimer(r -> { - Thread thread = new Thread(r, "victoria-metrics-cluster-flush-timer"); + Thread thread = new Thread(r, "victoria-metrics-flush-timer"); thread.setDaemon(true); return thread; - }, 1, TimeUnit.SECONDS, 512); - // start flush interval timer - this.metricsFlushTimer.newTimeout(new MetricsFlushTask(null), insertConfig.flushInterval(), TimeUnit.SECONDS); + }, 100, TimeUnit.MILLISECONDS, 512); + metricsFlushTask = new MetricsFlushTask(); + this.metricsFlushTimer.newTimeout(metricsFlushTask, insertConfig.flushInterval(), TimeUnit.SECONDS); } private boolean checkVictoriaMetricsDatasourceAvailable() { @@ -159,6 +167,10 @@ private boolean checkVictoriaMetricsDatasourceAvailable() { @Override public void saveData(CollectRep.MetricsData metricsData) { + if (closed.get()) { + log.warn("[Victoria Metrics] Rejecting metrics after storage shutdown"); + return; + } if (!isServerAvailable()) { serverAvailable = checkVictoriaMetricsDatasourceAvailable(); } @@ -257,9 +269,22 @@ public void saveData(CollectRep.MetricsData metricsData) { @Override public void destroy() { + synchronized (metricsFlushLock) { + if (!closed.compareAndSet(false, true)) { + return; + } + } if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) { metricsFlushTimer.stop(); } + immediateFlushPending.set(false); + while (hasPendingMetrics()) { + if (!flushBufferedMetrics()) { + log.error("[Victoria Metrics] Unable to flush {} buffered metrics during shutdown", + pendingMetricCount()); + break; + } + } } @Override @@ -570,54 +595,112 @@ public static final class VictoriaMetricsContent { * @param contentList victoriaMetricsContent List */ private void sendVictoriaMetrics(List contentList) { - for (VictoriaMetricsDataStorage.VictoriaMetricsContent content : contentList) { - boolean offered = false; - int retryCount = 0; - while (!offered && retryCount < MAX_RETRIES) { + for (int index = 0; index < contentList.size(); index++) { + VictoriaMetricsDataStorage.VictoriaMetricsContent content = contentList.get(index); + boolean offered = metricsBufferQueue.offer(content); + for (int attempt = 1; attempt <= MAX_BUFFER_OFFER_ATTEMPTS && !offered; attempt++) { + if (closed.get()) { + return; + } + triggerImmediateFlush(); try { - // Attempt to add to the queue for a limited time offered = metricsBufferQueue.offer(content, MAX_WAIT_MS, TimeUnit.MILLISECONDS); - if (!offered) { - // If the queue is still full, trigger an immediate refresh to free up space - if (retryCount == 0) { - log.debug("victoria metrics buffer queue is full, triggering immediate flush"); - triggerImmediateFlush(); - } - retryCount++; - // The short sleep allows the queue to clear out - if (retryCount < MAX_RETRIES) { - Thread.sleep(100L * retryCount); - } - } } catch (InterruptedException e) { Thread.currentThread().interrupt(); - log.error("[Victoria Metrics] Interrupted while offering metrics to buffer queue", e); - break; + recordDroppedMetrics(contentList.size() - index, "producer interrupted"); + return; } } - // When the maximum number of retries is reached, if it still cannot be added to the queue, the data is saved directly if (!offered) { - log.warn("[Victoria Metrics] Failed to add metrics to buffer after {} retries, saving directly", MAX_RETRIES); - try { - doSaveData(contentList); - } catch (Exception e) { - log.error("[Victoria Metrics] Failed to save metrics directly: {}", e.getMessage(), e); - } + recordDroppedMetrics(contentList.size() - index, "buffer remained full"); + return; + } + if (metricsBufferQueue.size() >= insertConfig.bufferSize() * 0.8) { + triggerImmediateFlush(); } } - // Refresh in advance to avoid waiting - if (metricsBufferQueue.size() >= insertConfig.bufferSize() * 0.8 - && draining.compareAndSet(false, true)) { - triggerImmediateFlush(); + } + + private void recordDroppedMetrics(int count, String reason) { + long total = droppedMetricCount.addAndGet(count); + if (total == count || (total & (total - 1)) == 0 || total % 100 == 0) { + log.error("[Victoria Metrics] Dropped {} metrics because {}; cumulative dropped metrics: {}", + count, reason, total); } } + long getDroppedMetricCount() { + return droppedMetricCount.get(); + } + private void triggerImmediateFlush() { - List batch = new ArrayList<>(insertConfig.bufferSize()); - metricsBufferQueue.drainTo(batch, insertConfig.bufferSize()); - draining.set(false); - if (!batch.isEmpty()) { - metricsFlushTimer.newTimeout(new MetricsFlushTask(batch), 0, TimeUnit.MILLISECONDS); + scheduleImmediateFlush(0, TimeUnit.MILLISECONDS); + } + + private void scheduleImmediateFlush(long delay, TimeUnit unit) { + if (closed.get() || metricsFlushTimer == null || metricsFlushTimer.isStop()) { + return; + } + if (immediateFlushPending.compareAndSet(false, true)) { + try { + metricsFlushTimer.newTimeout(new ImmediateMetricsFlushTask(), delay, unit); + } catch (RuntimeException e) { + immediateFlushPending.set(false); + if (!closed.get()) { + log.warn("[Victoria Metrics] Unable to schedule immediate flush: {}", e.getMessage()); + } + } + } + } + + private boolean flushBufferedMetrics() { + List batch; + synchronized (metricsFlushLock) { + if (retryBatch.isEmpty()) { + List nextBatch = new ArrayList<>(insertConfig.bufferSize()); + metricsBufferQueue.drainTo(nextBatch, insertConfig.bufferSize()); + retryBatch = nextBatch; + } + batch = retryBatch; + } + if (batch.isEmpty()) { + return true; + } + if (!trySaveData(batch)) { + log.warn("[Victoria Metrics] Retaining {} metrics items for retry", batch.size()); + return false; + } + synchronized (metricsFlushLock) { + if (retryBatch == batch) { + retryBatch = Collections.emptyList(); + } + } + log.debug("[Victoria Metrics] Flushed {} metrics items", batch.size()); + return true; + } + + private boolean hasPendingMetrics() { + synchronized (metricsFlushLock) { + return !retryBatch.isEmpty() || !metricsBufferQueue.isEmpty(); + } + } + + private int pendingMetricCount() { + synchronized (metricsFlushLock) { + return retryBatch.size() + metricsBufferQueue.size(); + } + } + + private void schedulePeriodicFlush() { + if (closed.get() || metricsFlushTimer == null || metricsFlushTimer.isStop()) { + return; + } + try { + metricsFlushTimer.newTimeout(metricsFlushTask, insertConfig.flushInterval(), TimeUnit.SECONDS); + } catch (RuntimeException e) { + if (!closed.get()) { + log.warn("[Victoria Metrics] Unable to schedule periodic flush: {}", e.getMessage()); + } } } @@ -625,42 +708,46 @@ private void triggerImmediateFlush() { * Regularly refresh the buffer queue to the vm */ private class MetricsFlushTask implements TimerTask { - private final List batch; - - public MetricsFlushTask(List batch) { - this.batch = batch; - } - @Override public void run(Timeout timeout) { + boolean flushSucceeded = false; try { - if (batch == null) { - // If the batch is null, it means that the timer is triggered by flush interval timer - List batchT = new ArrayList<>(insertConfig.bufferSize()); - metricsBufferQueue.drainTo(batchT, insertConfig.bufferSize()); - triggerDoSaveData(batchT); - // Reschedule the next flush task - triggerIntervalFlushTimer(); - } else { - // If the batch is not null, it means that the timer is triggered by the immediate flush - triggerDoSaveData(batch); - } + flushSucceeded = flushBufferedMetrics(); } catch (Exception e) { log.error("[VictoriaMetrics] flush task error: {}", e.getMessage(), e); + } finally { + if (!flushSucceeded && hasPendingMetrics() && !closed.get()) { + scheduleImmediateFlush(FAILED_FLUSH_RETRY_SECONDS, TimeUnit.SECONDS); + } + schedulePeriodicFlush(); } } + } - private void triggerDoSaveData(List batch) { - if (!batch.isEmpty()) { - doSaveData(batch); - log.debug("[Victoria Metrics] Flushed {} metrics items", batch.size()); - } - } - - private void triggerIntervalFlushTimer() { - if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) { - metricsFlushTimer.newTimeout(new MetricsFlushTask(null), insertConfig.flushInterval(), TimeUnit.SECONDS); - log.debug("[Victoria Metrics] Rescheduled next flush task in {} seconds.", insertConfig.flushInterval()); + private class ImmediateMetricsFlushTask implements TimerTask { + @Override + public void run(Timeout timeout) { + boolean flushSucceeded = false; + try { + int flushedBatches = 0; + do { + flushSucceeded = flushBufferedMetrics(); + flushedBatches++; + } while (flushSucceeded + && hasPendingMetrics() + && !closed.get() + && flushedBatches < MAX_BATCHES_PER_FLUSH_TASK); + } catch (Exception e) { + log.error("[VictoriaMetrics] immediate flush task error: {}", e.getMessage(), e); + } finally { + immediateFlushPending.set(false); + if (hasPendingMetrics() && !closed.get()) { + if (flushSucceeded) { + scheduleImmediateFlush(0, TimeUnit.MILLISECONDS); + } else { + scheduleImmediateFlush(FAILED_FLUSH_RETRY_SECONDS, TimeUnit.SECONDS); + } + } } } } @@ -668,7 +755,7 @@ private void triggerIntervalFlushTimer() { /** * Save metric data to victoria-metric via HTTP call */ - private void doSaveData(List contentList) { + private boolean trySaveData(List contentList) { try { HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); @@ -705,12 +792,15 @@ private void doSaveData(List contentList) { httpEntity, String.class); if (responseEntity.getStatusCode().is2xxSuccessful()) { log.debug("insert metrics data to victoria-metrics success."); + return true; } else { - log.error("insert metrics data to victoria-metrics failed. {}", responseEntity.getBody()); + log.error("insert metrics data to victoria-metrics failed with status {}", + responseEntity.getStatusCode()); } } catch (Exception e){ log.error("flush metrics data to victoria-metrics error: {}.", e.getMessage(), e); } + return false; } } diff --git a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java new file mode 100644 index 00000000000..2cf9fb9a291 --- /dev/null +++ b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsClusterDataStorageTest.java @@ -0,0 +1,211 @@ +/* + * 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 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 f11fdb905dc..d6b5afe3580 100644 --- a/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorageTest.java +++ b/hertzbeat-warehouse/src/test/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorageTest.java @@ -51,6 +51,10 @@ import org.springframework.web.client.RestTemplate; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -184,6 +188,70 @@ void testMultiThreadSaveDataBySize() { .isGreaterThanOrEqualTo(threadCount * writeSize / bufferSize)); } + @Test + void failedSingleNodeFlushRetainsTheBatchAndRetriesQuickly() { + when(victoriaMetricsProperties.insert()).thenReturn(new VictoriaMetricsProperties.InsertConfig( + 10, 1, new VictoriaMetricsProperties.Compression(false))); + List attemptedBodies = new CopyOnWriteArrayList<>(); + AtomicInteger writes = new AtomicInteger(); + when(restTemplate.postForEntity( + startsWith(victoriaMetricsProperties.url()), + any(HttpEntity.class), + eq(String.class) + )).thenAnswer(invocation -> { + HttpEntity request = invocation.getArgument(1); + attemptedBodies.add(request.getBody()); + if (writes.getAndIncrement() == 0) { + return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).build(); + } + return ResponseEntity.noContent().build(); + }); + victoriaMetricsDataStorage = new VictoriaMetricsDataStorage(victoriaMetricsProperties, restTemplate); + + victoriaMetricsDataStorage.saveData(generateMockedMetricsData()); + + Awaitility.await() + .pollInterval(250, TimeUnit.MILLISECONDS) + .atMost(7, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(attemptedBodies).hasSizeGreaterThanOrEqualTo(2)); + assertThat(attemptedBodies.get(1)).isEqualTo(attemptedBodies.get(0)); + } + + @Test + void persistentSingleNodeFailureDoesNotBlockTheWarehouseProducerIndefinitely() throws Exception { + when(victoriaMetricsProperties.insert()).thenReturn(new VictoriaMetricsProperties.InsertConfig( + 1, 3600, new VictoriaMetricsProperties.Compression(false))); + AtomicInteger writes = new AtomicInteger(); + when(restTemplate.postForEntity( + startsWith(victoriaMetricsProperties.url()), + any(HttpEntity.class), + eq(String.class) + )).thenAnswer(invocation -> { + writes.incrementAndGet(); + return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).build(); + }); + victoriaMetricsDataStorage = new VictoriaMetricsDataStorage(victoriaMetricsProperties, restTemplate); + ExecutorService producer = Executors.newSingleThreadExecutor(); + Future pendingWrite = null; + + try { + victoriaMetricsDataStorage.saveData(generateMockedMetricsData()); + Awaitility.await().atMost(3, TimeUnit.SECONDS).until(() -> writes.get() > 0); + victoriaMetricsDataStorage.saveData(generateMockedMetricsData()); + + pendingWrite = producer.submit( + () -> victoriaMetricsDataStorage.saveData(generateMockedMetricsData())); + + pendingWrite.get(3, TimeUnit.SECONDS); + assertThat(victoriaMetricsDataStorage.getDroppedMetricCount()).isGreaterThanOrEqualTo(1); + } finally { + if (pendingWrite != null) { + pendingWrite.cancel(true); + } + producer.shutdownNow(); + } + } + @AfterEach void stop() { if (victoriaMetricsDataStorage != null) {