Merge branch 'master' into doc112

apache · Jul 22, 2024 · 70d2141 · 70d2141
2 parents d78d518 + a482d24
commit 70d2141
Show file tree

Hide file tree

Showing 38 changed files with 1,402 additions and 173 deletions.
diff --git a/.github/workflows/backend-build-test.yml b/.github/workflows/backend-build-test.yml
@@ -61,11 +61,11 @@ jobs:
       run: |
         docker buildx create --use --name myBuilder --driver docker-container
         docker buildx use myBuilder
-
+        
         ./script/docker/server/build.sh
     - name: Run E2E
       run: |
         sudo curl -L https://github.com/docker/compose/releases/download/v2.23.0/docker-compose-linux-x86_64 -o /usr/local/bin/docker-compose
         sudo chmod u+x /usr/local/bin/docker-compose
-
-        cd e2e && ./start.sh
+        
+        cd e2e && chmod +x ./script/*.sh && ./script/start.sh
diff --git a/...ctor/src/test/java/org/apache/hertzbeat/collector/dispatch/MetricsCollectorQueueTest.java b/...ctor/src/test/java/org/apache/hertzbeat/collector/dispatch/MetricsCollectorQueueTest.java
@@ -17,28 +17,39 @@
 
 package org.apache.hertzbeat.collector.dispatch;
 
-import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.mock;
 
 /**
  * Test case for {@link MetricsCollectorQueue}
  */
 class MetricsCollectorQueueTest {
 
+    private MetricsCollectorQueue metricsCollectorQueue;
+
+    private MetricsCollect mockJob;
+
     @BeforeEach
     void setUp() {
-    }
 
-    @AfterEach
-    void tearDown() {
+        metricsCollectorQueue = new MetricsCollectorQueue();
+        mockJob = mock(MetricsCollect.class);
     }
 
     @Test
-    void addJob() {
+    void testAddJob() throws InterruptedException {
+
+        metricsCollectorQueue.addJob(mockJob);
+        assertEquals(mockJob, metricsCollectorQueue.getJob());
     }
 
     @Test
-    void getJob() {
+    void testGetJobTimeout() throws InterruptedException {
+
+        assertNull(metricsCollectorQueue.getJob());
     }
+
 }
diff --git a/collector/src/test/java/org/apache/hertzbeat/collector/dispatch/WorkerPoolTest.java b/collector/src/test/java/org/apache/hertzbeat/collector/dispatch/WorkerPoolTest.java
@@ -17,28 +17,48 @@
 
 package org.apache.hertzbeat.collector.dispatch;
 
-import org.junit.jupiter.api.AfterEach;
+import java.util.concurrent.RejectedExecutionException;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
 
 /**
  * Test case for {@link WorkerPool}
  */
 class WorkerPoolTest {
 
+    private WorkerPool workerPool;
+
+    private Runnable mockTask;
+
     @BeforeEach
     void setUp() {
+
+        workerPool = new WorkerPool();
+        mockTask = mock(Runnable.class);
     }
 
-    @AfterEach
-    void tearDown() {
+    @Test
+    void testExecuteJob() {
+
+        assertDoesNotThrow(() -> workerPool.executeJob(mockTask));
     }
 
     @Test
-    void executeJob() {
+    void testExecuteJobThrowsException() {
+
+        workerPool = mock(WorkerPool.class);
+        doThrow(new RejectedExecutionException()).when(workerPool).executeJob(mockTask);
+
+        assertThrows(RejectedExecutionException.class, () -> workerPool.executeJob(mockTask));
     }
 
     @Test
-    void destroy() {
+    void testDestroy() {
+        assertDoesNotThrow(() -> workerPool.destroy());
     }
+
 }
diff --git a/common/pom.xml b/common/pom.xml
@@ -106,6 +106,11 @@
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
         </dependency>
+        <!-- redis -->
+        <dependency>
+            <groupId>io.lettuce</groupId>
+            <artifactId>lettuce-core</artifactId>
+        </dependency>
         <!-- expr calculate -->
         <dependency>
             <groupId>org.apache.commons</groupId>

diff --git a/common/src/main/java/org/apache/hertzbeat/common/config/CommonProperties.java b/common/src/main/java/org/apache/hertzbeat/common/config/CommonProperties.java
@@ -73,6 +73,8 @@ public static class DataQueueProperties {
 
         private KafkaProperties kafka;
 
+        private RedisProperties redis;
+
         public QueueType getType() {
             return type;
         }
@@ -88,6 +90,16 @@ public KafkaProperties getKafka() {
         public void setKafka(KafkaProperties kafka) {
             this.kafka = kafka;
         }
+
+        public RedisProperties getRedis() {
+
+            return redis;
+        }
+
+        public void setRedis(RedisProperties redis) {
+
+            this.redis = redis;
+        }
     }
 
     /**
@@ -101,7 +113,106 @@ public enum QueueType {
         /** with netty connect **/
         Netty,
         /** rabbit mq **/
-        Rabbit_Mq
+        Rabbit_Mq,
+        /** redis **/
+        Redis
+    }
+
+    /**
+     * redis data queue properties
+     */
+    public static class RedisProperties {
+
+        /**
+         * redis server host.
+         */
+        private String redisHost;
+
+        /**
+         * redis server port.
+         */
+        private int redisPort;
+
+        /**
+         * Queue name for metrics data to alerter
+         */
+        private String metricsDataQueueNameToAlerter;
+
+        /**
+         * Queue name for metrics data to persistent storage
+         */
+        private String metricsDataQueueNameToPersistentStorage;
+
+        /**
+         * Queue name for metrics data to real-time storage
+         */
+        private String metricsDataQueueNameToRealTimeStorage;
+
+        /**
+         * Queue name for alerts data
+         */
+        private String alertsDataQueueName;
+
+        public int getRedisPort() {
+
+            return redisPort;
+        }
+
+        public void setRedisPort(int redisPort) {
+
+            this.redisPort = redisPort;
+        }
+
+        public String getRedisHost() {
+
+            return redisHost;
+        }
+
+        public void setRedisHost(String redisHost) {
+
+            this.redisHost = redisHost;
+        }
+
+        public String getMetricsDataQueueNameToAlerter() {
+
+            return metricsDataQueueNameToAlerter;
+        }
+
+        public void setMetricsDataQueueNameToAlerter(String metricsDataQueueNameToAlerter) {
+
+            this.metricsDataQueueNameToAlerter = metricsDataQueueNameToAlerter;
+        }
+
+        public String getMetricsDataQueueNameToPersistentStorage() {
+
+            return metricsDataQueueNameToPersistentStorage;
+        }
+
+        public void setMetricsDataQueueNameToPersistentStorage(String metricsDataQueueNameToPersistentStorage) {
+
+            this.metricsDataQueueNameToPersistentStorage = metricsDataQueueNameToPersistentStorage;
+        }
+
+        public String getMetricsDataQueueNameToRealTimeStorage() {
+
+            return metricsDataQueueNameToRealTimeStorage;
+        }
+
+        public void setMetricsDataQueueNameToRealTimeStorage(String metricsDataQueueNameToRealTimeStorage) {
+
+            this.metricsDataQueueNameToRealTimeStorage = metricsDataQueueNameToRealTimeStorage;
+        }
+
+        public String getAlertsDataQueueName() {
+
+            return alertsDataQueueName;
+        }
+
+        public void setAlertsDataQueueName(String alertsDataQueueName) {
+
+            this.alertsDataQueueName = alertsDataQueueName;
+        }
+
     }
 
     /**

diff --git a/common/src/main/java/org/apache/hertzbeat/common/constants/SignConstants.java b/common/src/main/java/org/apache/hertzbeat/common/constants/SignConstants.java
@@ -31,4 +31,6 @@ public interface SignConstants {
     String CARRIAGE_RETURN = "\r";
 
     String RIGHT_DASH = "/";
+
+    String COMMA = ",";
 }