Merge branch 'master' into fix-npe-victoria-metrics-data-storage

Signed-off-by: tomsun28 <tomsun28@outlook.com>

hertzbeat-manager/src/main/resources/application.yml

@@ -138,7 +138,7 @@ warehouse:
       username: root
       password: root
       insert:
-        buffer-size: 1000
+        buffer-size: 100
         flush-interval: 3
       cluster:
         enabled: false

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorage.java

@@ -100,10 +100,10 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {
     private final VictoriaMetricsProperties victoriaMetricsProp;
     private final RestTemplate restTemplate;
     private final BlockingQueue<VictoriaMetricsDataStorage.VictoriaMetricsContent> metricsBufferQueue;
-
-    private boolean isBatchImportEnabled = false;
+    
     private HashedWheelTimer metricsFlushTimer = null;
     private MetricsFlushTask metricsFlushtask = null;
+    private final VictoriaMetricsProperties.InsertConfig insertConfig;
 
     public VictoriaMetricsDataStorage(VictoriaMetricsProperties victoriaMetricsProperties, RestTemplate restTemplate) {
         if (victoriaMetricsProperties == null) {
@@ -114,11 +114,9 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {
         victoriaMetricsProp = victoriaMetricsProperties;
         serverAvailable = checkVictoriaMetricsDatasourceAvailable();
         serverAvailable = checkVictoriaMetricsDatasourceAvailable();
-        metricsBufferQueue = new LinkedBlockingQueue<>(victoriaMetricsProperties.insert().bufferSize());
-        isBatchImportEnabled = victoriaMetricsProperties.insert().flushInterval() != 0 && victoriaMetricsProperties.insert().bufferSize() != 0;
-        if (isBatchImportEnabled){
-            initializeFlushTimer();
-        }
+        insertConfig = victoriaMetricsProperties.insert() == null ? new VictoriaMetricsProperties.InsertConfig(100, 3) : victoriaMetricsProperties.insert();
+        metricsBufferQueue = new LinkedBlockingQueue<>(insertConfig.bufferSize());
+        initializeFlushTimer();
     }
 
     private void initializeFlushTimer() {
@@ -244,10 +242,6 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {
             log.info("[warehouse victoria-metrics] flush metrics data {} is empty, ignore.", metricsData.getId());
             return;
         }
-        if (!isBatchImportEnabled){
-            doSaveData(contentList);
-            return;
-        }
         sendVictoriaMetrics(contentList);
     }
 
@@ -585,10 +579,10 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {
                     log.error("[Victoria Metrics] Failed to save metrics directly: {}", e.getMessage(), e);
                 }
             }
-            // Refresh in advance to avoid waiting
-            if (metricsBufferQueue.size() >= victoriaMetricsProp.insert().bufferSize() * 0.8) {
-                triggerImmediateFlush();
-            }
+        }
+        // Refresh in advance to avoid waiting
+        if (metricsBufferQueue.size() >= insertConfig.bufferSize() * 0.8) {
+            triggerImmediateFlush();
         }
     }
 
@@ -603,14 +597,14 @@ public class VictoriaMetricsDataStorage extends AbstractHistoryDataStorage {
         @Override
         public void run(Timeout timeout) {
             try {
-                List<VictoriaMetricsDataStorage.VictoriaMetricsContent> batch = new ArrayList<>(victoriaMetricsProp.insert().bufferSize());
-                metricsBufferQueue.drainTo(batch, victoriaMetricsProp.insert().bufferSize());
+                List<VictoriaMetricsDataStorage.VictoriaMetricsContent> batch = new ArrayList<>(insertConfig.bufferSize());
+                metricsBufferQueue.drainTo(batch, insertConfig.bufferSize());
                 if (!batch.isEmpty()) {
                     doSaveData(batch);
                     log.debug("[Victoria Metrics] Flushed {} metrics items", batch.size());
                 }
                 if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) {
-                    metricsFlushTimer.newTimeout(this, victoriaMetricsProp.insert().flushInterval(), TimeUnit.SECONDS);
+                    metricsFlushTimer.newTimeout(this, insertConfig.flushInterval(), TimeUnit.SECONDS);
                 }
             } catch (Exception e) {
                 log.error("[VictoriaMetrics] flush task error: {}", e.getMessage(), e);

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsProperties.java

@@ -34,7 +34,7 @@ public record VictoriaMetricsProperties(@DefaultValue("false") boolean enabled,
                                         String password,
                                         InsertConfig insert) {
 
-    record InsertConfig(@DefaultValue("1000") int bufferSize,
+    record InsertConfig(@DefaultValue("100") int bufferSize,
                         @DefaultValue("3") int flushInterval) {
 
     }

script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/application.yml

@@ -125,6 +125,9 @@ warehouse:
       url: http://victoria-metrics:8428      
       username: root
       password: root
+      insert:
+        buffer-size: 100
+        flush-interval: 3
   # store real-time metrics data, enable only one below
   real-time:
     memory:

script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/application.yml

@@ -124,6 +124,9 @@ warehouse:
       url: http://victoria-metrics:8428
       username: root
       password: root
+      insert:
+        buffer-size: 100
+        flush-interval: 3
   # store real-time metrics data, enable only one below
   real-time:
     memory: