ref: collector部分根据文档要求修改调整数据处理逻辑

ref: 移除kafka的protobuf部分（目前尚不考虑使用）
ref: collector部分配置项增加kafka的linger批量提交配置
todo: 数据收集部分改写为异步接口模式；优化大批量数据入库db时的性能

rition-center/api/pom.xml

@@ -43,5 +43,10 @@
             <artifactId>notify</artifactId>
             <version>${revision}</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+        </dependency>
     </dependencies>
 </project>

rition-center/api/src/main/java/rition/backend/api/interceptor/ResponseIdInterceptor.java

@@ -41,7 +41,7 @@ public class ResponseIdInterceptor implements HandlerInterceptor {
             if (StringUtils.isEmpty(existsRequestId)) {
                 var requestId = UUID.randomUUID().toString();
                 response.setHeader("X-Request-ID", requestId);
-                request.setAttribute("X-Request-ID", requestId);
+                request.setAttribute("requestId", requestId);
             }
         }
 

rition-center/api/src/main/java/rition/backend/api/resolver/RequestIdArgumentResolver.java

@@ -37,7 +37,7 @@ public class RequestIdArgumentResolver implements HandlerMethodArgumentResolver
         // 如果Attribute上下文和ResponseHeader都有RequestId，则现在生成一个新的
         // 相应的方法已经使用@WithRequestIdResponse注解在拦截器中设置，
         // 则其实可以断言Attribute上下文和ResponseHeader都有RequestId
-        String requestId = (String) request.getAttribute("X-Request-ID");
+        String requestId = (String) request.getAttribute("requestId");
         if (requestId == null) {
             requestId = response.getHeader("X-Request-ID");
             if (requestId == null) {
@@ -45,7 +45,7 @@ public class RequestIdArgumentResolver implements HandlerMethodArgumentResolver
                 response.setHeader("X-Request-ID", requestId);
             }
 
-            request.setAttribute("X-Request-ID", requestId);
+            request.setAttribute("requestId", requestId);
             return requestId;
         } else {
             if (response.getHeader("X-Request-ID") == null) {

rition-center/api/src/main/java/rition/backend/api/v1/DataCollectingController.java

@@ -11,7 +11,9 @@ import rition.backend.service.MetricDataCollectingService;
 import rition.common.data.dto.MetricDataDto;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 @RestController
 @RequestMapping("/metric")
@@ -26,20 +28,25 @@ public class DataCollectingController {
     @PostMapping("/put")
     @WithRequestIdResponse
     public Response<Object> putData(@RequestBody List<MetricDataUploadRequest> uploadDataList) {
-
+        // dto转换，按时间组装数据
-        // dto转换
+        Map<Long, MetricDataDto> metricDataTimestampMap = new HashMap<>(uploadDataList.size() / 20);
-        MetricDataDto metricDataDto = new MetricDataDto();
+        for (MetricDataUploadRequest uploadData : uploadDataList) {
-        List<MetricDataDto.MetricData> dataList = uploadDataList.stream()
+            MetricDataDto dataList = metricDataTimestampMap.get(uploadData.getTimestamp());
-                .map(data -> MetricDataDto.MetricData.builder()
+            if (dataList == null) {
-                        .metric(data.getMetric())
+                dataList = MetricDataDto.builder().dataList(new ArrayList<>(20)).build();
-                        .value(data.getValue())
+                metricDataTimestampMap.put(uploadData.getTimestamp(), dataList);
-                        .timestamp(data.getTimestamp())
+            }
-                        .tags(data.getTags())
+
-                        .build())
+            dataList.getDataList().add(MetricDataDto.MetricData.builder()
-                .toList();
+                    .metric(uploadData.getMetric())
-
+                    .value(uploadData.getValue())
-        metricDataDto.setDataList(dataList);
+                    .timestamp(uploadData.getTimestamp())
-        metricDataCollectingService.receiveData(metricDataDto);
+                    .tags(uploadData.getTags())
+                    .build()
+            );
+        }
+
+        metricDataCollectingService.receiveData(metricDataTimestampMap);
 
         return Response.success();
     }

rition-center/api/src/main/java/rition/backend/service/MetricDataCollectingService.java

@@ -2,9 +2,10 @@ package rition.backend.service;
 
 import org.springframework.stereotype.Service;
 import rition.common.data.dto.MetricDataDto;
-import rition.common.data.dto.MetricDataProto;
 import rition.service.collector.MetricCollectingService;
 
+import java.util.Map;
+
 /**
  * MetricCollectorService 指标数据收集类，负责收集来自云主机探针上报的数据，并将其入库等，
  * 这里只是个代理类，实际的实现在collector模块，
@@ -24,7 +25,7 @@ public class MetricDataCollectingService {
      * 接收处理好的监控指标数据
      * @param collectedMetricData 监控指标数据，protobuf对象
      */
-    public void receiveData(MetricDataDto collectedMetricData) {
+    public void receiveData(Map<Long, MetricDataDto> collectedMetricData) {
         metricCollectingService.receiveData(collectedMetricData);
     }
 }

rition-center/api/src/main/resources/application.yml

@@ -1,6 +1,7 @@
 server:
   port: 22019
-
+  servlet:
+    context-path: /api
 spring:
   profiles:
     include: collector

rition-center/common/pom.xml

@@ -20,39 +20,15 @@
     </properties>
 
     <dependencies>
-        <dependency>
+
-            <groupId>com.google.protobuf</groupId>
-            <artifactId>protobuf-java</artifactId>
-            <version>${protobuf.version}</version>
-        </dependency>
     </dependencies>
 
     <build>
         <extensions>
-            <extension>
+
-                <groupId>kr.motd.maven</groupId>
-                <artifactId>os-maven-plugin</artifactId>
-                <version>1.7.1</version>
-            </extension>
         </extensions>
         <plugins>
-            <plugin>
+
-                <groupId>org.xolstice.maven.plugins</groupId>
-                <artifactId>protobuf-maven-plugin</artifactId>
-                <version>0.6.1</version>
-                <extensions>true</extensions>
-                <configuration>
-                    <protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}</protocArtifact>
-                </configuration>
-                <executions>
-                    <execution>
-                        <goals>
-                            <goal>compile</goal>
-                            <goal>test-compile</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
         </plugins>
     </build>
 </project>

rition-center/pom.xml

@@ -53,6 +53,10 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-aop</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>

rition-center/service/collector/src/main/java/rition/service/collector/MetricCollectingService.java

@@ -11,7 +11,7 @@ import rition.service.collector.dao.entity.MetricRecordEntity;
 import rition.service.collector.dao.mapper.MetricRecordMapper;
 
 import java.time.Instant;
-import java.util.ArrayList;
+import java.util.Map;
 
 /**
  * DataCollectorService 指标数据收集类，负责收集来自云主机探针上报的数据，并将其入库等
@@ -21,7 +21,6 @@ import java.util.ArrayList;
 public class MetricCollectingService {
 
     private final KafkaTemplate<String, MetricDataDto> kafkaTemplate;
-    private final KafkaConfigure kafkaConfigure;
 
     private final String collectedDataTopic;
     private final MetricRecordMapper metricRecordMapper;
@@ -29,55 +28,27 @@ public class MetricCollectingService {
     public MetricCollectingService(KafkaTemplate<String, MetricDataDto> kafkaTemplate,
                                    KafkaConfigure kafkaConfigure, MetricRecordMapper metricRecordMapper) {
         this.kafkaTemplate = kafkaTemplate;
-        this.kafkaConfigure = kafkaConfigure;
         this.collectedDataTopic = kafkaConfigure.getDataCollecting().getTopic();
         this.metricRecordMapper = metricRecordMapper;
     }
 
-    private static final int MAX_BATCH_INSERT_SIZE = 512;
-
     /**
-     * 接收处理好的监控指标数据
+     * 接收处理好的监控指标数据，以时间戳为分组入库和发送mq
      *
-     * @param collectedMetricData 监控指标数据，protobuf对象
+     * @param collectedMetricData 监控指标数据，以时间戳为组
      */
-    public void receiveData(MetricDataDto collectedMetricData) {
+    public void receiveData(Map<Long, MetricDataDto> collectedMetricData) {
-        // 对大于MAX_BATCH_INSERT_SIZE大小的数据进行分组的批量插入
+        for (Long timestamp : collectedMetricData.keySet()) {
-        var metricDataList = collectedMetricData.getDataList();
+            var metricDataDto = collectedMetricData.get(timestamp);
-        if (metricDataList.size() <= MAX_BATCH_INSERT_SIZE) {
+            var metricDataList = metricDataDto.getDataList();
-            var metricRecordEntityList = new ArrayList<MetricRecordEntity>(metricDataList.size());
+            var metricRecordEntityList = metricDataList.stream().map(this::convert).toList();
-            for (MetricDataDto.MetricData metricData : metricDataList) {
-                MetricRecordEntity entity = this.convert(metricData);
-
-                metricRecordEntityList.add(entity);
-            }
 
-            kafkaTemplate.send(this.collectedDataTopic, collectedMetricData);
+            kafkaTemplate.send(this.collectedDataTopic, metricDataDto);
             metricRecordMapper.insertBatchSomeColumn(metricRecordEntityList);
-        } else {
-            var batch = metricDataList.size() / MAX_BATCH_INSERT_SIZE;
-            if (metricDataList.size() % MAX_BATCH_INSERT_SIZE != 0) {
-                batch++;
-            }
-
-            var metricRecordEntityList = new ArrayList<MetricRecordEntity>(MAX_BATCH_INSERT_SIZE);
-            for (int i = 0; i < batch; i++) {
-                metricRecordEntityList.clear();
-                var from = i * MAX_BATCH_INSERT_SIZE;
-                var to = (i + 1) * MAX_BATCH_INSERT_SIZE;
-                var subList = metricDataList.subList(from, Math.min(metricDataList.size(), to));
-                for (MetricDataDto.MetricData metricData : subList) {
-                    MetricRecordEntity entity = this.convert(metricData);
-                    metricRecordEntityList.add(entity);
-                }
-
-                kafkaTemplate.send(this.collectedDataTopic, MetricDataDto.builder().dataList(subList).build());
-                metricRecordMapper.insertBatchSomeColumn(metricRecordEntityList);
-            }
         }
     }
 
-    public MetricRecordEntity convert(MetricDataDto.MetricData metricData) {
+    private MetricRecordEntity convert(MetricDataDto.MetricData metricData) {
         MetricRecordEntity entity = new MetricRecordEntity();
         entity.setId(YitIdHelper.nextId());
         entity.setInstanceId(metricData.getTags().get("instanceId"));
@@ -85,7 +56,7 @@ public class MetricCollectingService {
         entity.setValue(metricData.getValue());
         entity.setStatus(CommonEntityStatus.STATUS_NORMAL.getValue());
 
-        var time = Instant.ofEpochMilli(metricData.getTimestamp());
+        var time = Instant.ofEpochSecond(metricData.getTimestamp());
         entity.setTime(time);
         entity.setUpdateTime(time);
 

rition-center/service/collector/src/main/java/rition/service/collector/mq/CollectedMetricDataDeserializer.java

@@ -1,18 +0,0 @@
-package rition.service.collector.mq;
-
-import com.google.protobuf.InvalidProtocolBufferException;
-import org.apache.kafka.common.serialization.Deserializer;
-import rition.common.data.dto.MetricDataProto;
-
-@Deprecated
-public class CollectedMetricDataDeserializer implements Deserializer<MetricDataProto.CollectedMetricData> {
-
-    @Override
-    public MetricDataProto.CollectedMetricData deserialize(String topic, byte[] data) {
-        try {
-            return MetricDataProto.CollectedMetricData.parseFrom(data);
-        } catch (InvalidProtocolBufferException e) {
-            throw new RuntimeException("exception happen while parsing protobuf data: ", e);
-        }
-    }
-}

rition-center/service/collector/src/main/java/rition/service/collector/mq/ProtobufMessageSerializer.java

@@ -1,13 +0,0 @@
-package rition.service.collector.mq;
-
-import com.google.protobuf.GeneratedMessageV3;
-import org.apache.kafka.common.serialization.Serializer;
-
-@Deprecated
-public class ProtobufMessageSerializer implements Serializer<GeneratedMessageV3> {
-
-    @Override
-    public byte[] serialize(String topic, GeneratedMessageV3 data) {
-        return data.toByteArray();
-    }
-}

rition-center/service/collector/src/main/resources/application-collector.yml

@@ -1,7 +1,7 @@
 spring:
   datasource:
     driver-class-name: com.mysql.cj.jdbc.Driver
-    url: jdbc:mysql://127.0.0.1:3306/rition?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai&rewriteBatchedStatements=true
+    url: jdbc:mysql://127.0.0.1:3306/rition?useUnicode=true&characterEncoding=utf-8&useSSL=true&serverTimezone=Asia/Shanghai&rewriteBatchedStatements=true
     username: root
     password: Test2333!
   kafka:
@@ -12,6 +12,10 @@ spring:
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
       acks: 1
+      properties:
+        linger:
+          ms: 200
+
 rition:
   kafka:
     data-collecting:

rition-probe/main.go

@@ -32,6 +32,6 @@ func main() {
 	}
 
 	srv := service.NewService(config)
-	//srv.Run()
+	srv.Run()
-	srv.Report()
+	//srv.Report()
 }