impl: 基本完成数据收集模块，以及服务器状态探针

rition-center/api/pom.xml

@@ -28,5 +28,20 @@
             <artifactId>common</artifactId>
             <version>${revision}</version>
         </dependency>
+        <dependency>
+            <groupId>net.rition</groupId>
+            <artifactId>collector</artifactId>
+            <version>${revision}</version>
+        </dependency>
+        <dependency>
+            <groupId>net.rition</groupId>
+            <artifactId>monitor</artifactId>
+            <version>${revision}</version>
+        </dependency>
+        <dependency>
+            <groupId>net.rition</groupId>
+            <artifactId>notify</artifactId>
+            <version>${revision}</version>
+        </dependency>
     </dependencies>
 </project>

rition-center/api/src/main/java/rition/backend/RitionBackendMain.java

@@ -1,15 +1,26 @@
 package rition.backend;
 
+import com.github.yitter.contract.IdGeneratorOptions;
+import com.github.yitter.idgen.YitIdHelper;
+import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
 
 @EnableAsync
 @EnableScheduling
-@SpringBootApplication
+@SpringBootApplication(scanBasePackages = {
+        "rition",
+})
+@MapperScan("rition")
+@EnableTransactionManagement
 public class RitionBackendMain {
     public static void main(String[] args) {
+        IdGeneratorOptions options = new IdGeneratorOptions((short) 0);
+        YitIdHelper.setIdGenerator(options);
+
         SpringApplication.run(RitionBackendMain.class);
     }
 }

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

@@ -4,21 +4,43 @@ import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
-import rition.backend.annotation.paramter.RequestId;
 import rition.backend.annotation.WithRequestIdResponse;
-import rition.backend.api.v1.dto.request.MonitorDataUploadRequest;
+import rition.backend.api.v1.dto.request.MetricDataUploadRequest;
 import rition.backend.api.v1.dto.response.Response;
+import rition.backend.service.MetricDataCollectingService;
+import rition.common.data.dto.MetricDataDto;
 
+import java.util.ArrayList;
 import java.util.List;
 
 @RestController
 @RequestMapping("/metric")
 public class DataCollectingController {
 
+    private final MetricDataCollectingService metricDataCollectingService;
+
+    public DataCollectingController(MetricDataCollectingService metricDataCollectingService) {
+        this.metricDataCollectingService = metricDataCollectingService;
+    }
+
     @PostMapping("/put")
     @WithRequestIdResponse
-    public Response<Object> putData(@RequestBody List<MonitorDataUploadRequest> request,
+    public Response<Object> putData(@RequestBody List<MetricDataUploadRequest> uploadDataList) {
-                                    @RequestId String requestId) {
+
+        // dto转换
+        MetricDataDto metricDataDto = new MetricDataDto();
+        List<MetricDataDto.MetricData> dataList = uploadDataList.stream()
+                .map(data -> MetricDataDto.MetricData.builder()
+                        .metric(data.getMetric())
+                        .value(data.getValue())
+                        .timestamp(data.getTimestamp())
+                        .tags(data.getTags())
+                        .build())
+                .toList();
+
+        metricDataDto.setDataList(dataList);
+        metricDataCollectingService.receiveData(metricDataDto);
+
         return Response.success();
     }
 }

rition-center/api/src/main/java/rition/backend/api/v1/dto/request/MetricDataUploadRequest.java

@@ -6,7 +6,7 @@ import lombok.Data;
 import java.util.Map;
 
 @Data
-public class MonitorDataUploadRequest {
+public class MetricDataUploadRequest {
     /**
      * 指标名
      */
@@ -25,5 +25,5 @@ public class MonitorDataUploadRequest {
     /**
      * 值
      */
-    private Float value;
+    private String value;
 }

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

@@ -0,0 +1,30 @@
+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;
+
+/**
+ * MetricCollectorService 指标数据收集类，负责收集来自云主机探针上报的数据，并将其入库等，
+ * 这里只是个代理类，实际的实现在collector模块，
+ * 这样做只是为了方便日后将collector等模块独立拆分出来，
+ * 这种情况下实际上数据收集的入口应当直接放在collector里，而不是api部分
+ */
+@Service
+public class MetricDataCollectingService {
+
+    private final MetricCollectingService metricCollectingService;
+
+    public MetricDataCollectingService(MetricCollectingService metricCollectingService) {
+        this.metricCollectingService = metricCollectingService;
+    }
+
+    /**
+     * 接收处理好的监控指标数据
+     * @param collectedMetricData 监控指标数据，protobuf对象
+     */
+    public void receiveData(MetricDataDto collectedMetricData) {
+        metricCollectingService.receiveData(collectedMetricData);
+    }
+}

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

@@ -1,2 +1,6 @@
 server:
   port: 22019
+
+spring:
+  profiles:
+    include: collector

rition-center/common/pom.xml

@@ -15,6 +15,44 @@
         <maven.compiler.source>17</maven.compiler.source>
         <maven.compiler.target>17</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+
+        <protobuf.version>3.25.3</protobuf.version>
     </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/common/src/main/java/rition/common/data/dto/MetricDataDto.java

@@ -0,0 +1,48 @@
+
+package rition.common.data.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+import java.util.Map;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class MetricDataDto {
+
+    private List<MetricData> dataList;
+
+    /**
+     * 指标数据
+     */
+    @Data
+    @Builder
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class MetricData {
+        /**
+         * 指标名
+         */
+        private String metric;
+
+        /**
+         * 数据标签
+         */
+        private Map<String, String> tags;
+
+        /**
+         * 时间戳
+         */
+        private Long timestamp;
+
+        /**
+         * 值
+         */
+        private String value;
+    }
+}

rition-center/common/src/main/java/rition/common/data/enums/CommonEntityStatus.java

@@ -0,0 +1,17 @@
+package rition.common.data.enums;
+
+import lombok.Getter;
+
+@Getter
+public enum CommonEntityStatus {
+    STATUS_NORMAL(0),
+    STATUS_DELETED(1),
+    STATUS_HIDDEN(2),
+    ;
+
+    private final int value;
+
+    CommonEntityStatus(int value) {
+        this.value = value;
+    }
+}

rition-center/common/src/main/java/rition/common/data/enums/Constants.java

@@ -0,0 +1,7 @@
+package rition.common.data.enums;
+
+import java.time.ZoneId;
+
+public class Constants {
+    public static final ZoneId DefaultTimeZone = ZoneId.of("Asia/Shanghai");
+}

rition-center/common/src/main/proto/metric_data.proto

@@ -0,0 +1,17 @@
+syntax = "proto3";
+
+package rition.proto.collecting.v1;
+
+option java_package = "rition.common.data.dto";
+option java_outer_classname = "MetricDataProto";
+
+message MetricData {
+  string metric = 1;
+  string value = 2;
+  uint64 timestamp = 3;
+  map<string, string> tags = 4;
+}
+
+message CollectedMetricData {
+  repeated MetricData data = 4;
+}

rition-center/pom.xml

@@ -27,6 +27,8 @@
         <maven.compiler.source>17</maven.compiler.source>
         <maven.compiler.target>17</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+
+        <idgen.version>1.0.6</idgen.version>
     </properties>
 
     <dependencies>
@@ -56,6 +58,13 @@
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-annotations</artifactId>
         </dependency>
+
+        <!-- id生成 -->
+        <dependency>
+            <groupId>com.github.yitter</groupId>
+            <artifactId>yitter-idgenerator</artifactId>
+            <version>${idgen.version}</version>
+        </dependency>
     </dependencies>
 
     <build>

rition-center/service/collector/pom.xml

@@ -15,6 +15,43 @@
         <maven.compiler.source>17</maven.compiler.source>
         <maven.compiler.target>17</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+
+        <mybatis-sb-starter.version>3.0.3</mybatis-sb-starter.version>
+        <mybatis.version>3.5.16</mybatis.version>
+        <mybatis-plus.version>3.5.6</mybatis-plus.version>
     </properties>
 
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mybatis.spring.boot</groupId>
+            <artifactId>mybatis-spring-boot-starter</artifactId>
+            <version>${mybatis-sb-starter.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.mybatis.spring.boot</groupId>
+            <artifactId>mybatis-spring-boot-starter-test</artifactId>
+            <version>${mybatis-sb-starter.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mybatis</groupId>
+            <artifactId>mybatis</artifactId>
+            <version>${mybatis.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.baomidou</groupId>
+            <artifactId>mybatis-plus-spring-boot3-starter</artifactId>
+            <version>${mybatis-plus.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.mysql</groupId>
+            <artifactId>mysql-connector-j</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+    </dependencies>
 </project>

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

@@ -1,4 +0,0 @@
-package rition.service.collector;
-
-public class DataCollectorService {
-}

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

@@ -0,0 +1,94 @@
+package rition.service.collector;
+
+import com.github.yitter.idgen.YitIdHelper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+import rition.common.data.dto.MetricDataDto;
+import rition.common.data.enums.CommonEntityStatus;
+import rition.service.collector.configure.KafkaConfigure;
+import rition.service.collector.dao.entity.MetricRecordEntity;
+import rition.service.collector.dao.mapper.MetricRecordMapper;
+
+import java.time.Instant;
+import java.util.ArrayList;
+
+/**
+ * DataCollectorService 指标数据收集类，负责收集来自云主机探针上报的数据，并将其入库等
+ */
+@Slf4j
+@Service
+public class MetricCollectingService {
+
+    private final KafkaTemplate<String, MetricDataDto> kafkaTemplate;
+    private final KafkaConfigure kafkaConfigure;
+
+    private final String collectedDataTopic;
+    private final MetricRecordMapper metricRecordMapper;
+
+    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;
+
+    /**
+     * 接收处理好的监控指标数据
+     *
+     * @param collectedMetricData 监控指标数据，protobuf对象
+     */
+    public void receiveData(MetricDataDto collectedMetricData) {
+        // 对大于MAX_BATCH_INSERT_SIZE大小的数据进行分组的批量插入
+        var metricDataList = collectedMetricData.getDataList();
+        if (metricDataList.size() <= MAX_BATCH_INSERT_SIZE) {
+            var metricRecordEntityList = new ArrayList<MetricRecordEntity>(metricDataList.size());
+            for (MetricDataDto.MetricData metricData : metricDataList) {
+                MetricRecordEntity entity = this.convert(metricData);
+
+                metricRecordEntityList.add(entity);
+            }
+
+            kafkaTemplate.send(this.collectedDataTopic, collectedMetricData);
+            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) {
+        MetricRecordEntity entity = new MetricRecordEntity();
+        entity.setId(YitIdHelper.nextId());
+        entity.setInstanceId(metricData.getTags().get("instanceId"));
+        entity.setItem(metricData.getMetric());
+        entity.setValue(metricData.getValue());
+        entity.setStatus(CommonEntityStatus.STATUS_NORMAL.getValue());
+
+        var time = Instant.ofEpochMilli(metricData.getTimestamp());
+        entity.setTime(time);
+        entity.setUpdateTime(time);
+
+        return entity;
+    }
+}

rition-center/service/collector/src/main/java/rition/service/collector/configure/KafkaConfigure.java

@@ -0,0 +1,20 @@
+package rition.service.collector.configure;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@Data
+@Configuration
+@ConfigurationProperties("rition.kafka")
+public class KafkaConfigure {
+    private DataCollectingKafkaConfig dataCollecting;
+
+    @Data
+    public static class DataCollectingKafkaConfig {
+        private String topic;
+        private String group;
+    }
+
+
+}

rition-center/service/collector/src/main/java/rition/service/collector/configure/MybatisPlusConfigure.java

@@ -0,0 +1,8 @@
+package rition.service.collector.configure;
+
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class MybatisPlusConfigure {
+
+}

rition-center/service/collector/src/main/java/rition/service/collector/dao/BatchSQLInjector.java

@@ -0,0 +1,21 @@
+package rition.service.collector.dao;
+
+import com.baomidou.mybatisplus.core.injector.AbstractMethod;
+import com.baomidou.mybatisplus.core.injector.DefaultSqlInjector;
+import com.baomidou.mybatisplus.core.metadata.TableInfo;
+import com.baomidou.mybatisplus.extension.injector.methods.InsertBatchSomeColumn;
+import org.apache.ibatis.session.Configuration;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+@Component
+public class BatchSQLInjector extends DefaultSqlInjector {
+    @Override
+    public List<AbstractMethod> getMethodList(Configuration configuration, Class<?> mapperClass, TableInfo tableInfo) {
+        List<AbstractMethod> methodList = super.getMethodList(configuration, mapperClass, tableInfo);
+        methodList.add(new InsertBatchSomeColumn());
+
+        return methodList;
+    }
+}

rition-center/service/collector/src/main/java/rition/service/collector/dao/entity/MetricRecordEntity.java

@@ -0,0 +1,59 @@
+package rition.service.collector.dao.entity;
+
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.Instant;
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+@TableName("rition_record")
+public class MetricRecordEntity {
+
+    /**
+     * 数据记录id
+     */
+    private Long id;
+
+    /**
+     * 云主机实例id
+     */
+    private String instanceId;
+
+    /**
+     * 数据项
+     */
+    private String item;
+
+    /**
+     * 数据值，统一使用字符串存储
+     */
+    private String value;
+
+    /**
+     * create_time
+     */
+    private Instant time;
+
+    /**
+     * update_time
+     */
+    private Instant updateTime;
+
+    /**
+     * status
+     */
+    private Integer status;
+
+    public static class RecordValueTypes {
+        public static final int STRING = 0;
+        public static final int INTEGER = 1;
+        public static final int FLOAT = 2;
+        public static final int BOOLEAN = 3;
+    }
+}

rition-center/service/collector/src/main/java/rition/service/collector/dao/mapper/MetricRecordMapper.java

@@ -0,0 +1,12 @@
+package rition.service.collector.dao.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import org.apache.ibatis.annotations.Mapper;
+import rition.service.collector.dao.entity.MetricRecordEntity;
+
+import java.util.List;
+
+@Mapper
+public interface MetricRecordMapper extends BaseMapper<MetricRecordEntity> {
+    int insertBatchSomeColumn(List<MetricRecordEntity> entityList);
+}

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

@@ -0,0 +1,18 @@
+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

@@ -0,0 +1,13 @@
+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

@@ -0,0 +1,19 @@
+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
+    username: root
+    password: Test2333!
+  kafka:
+    bootstrap-servers: '127.0.0.1:9092'
+    producer:
+      retries: 4
+      compression-type: zstd
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
+      acks: 1
+rition:
+  kafka:
+    data-collecting:
+      topic: 'ecs-metric-data-topic'
+      group: 'ecs-metric-data-group'

rition-center/service/collector/src/main/resources/mapper/MetricRecordMapper.xml

@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="rition.service.collector.dao.mapper.MetricRecordMapper">
+
+</mapper>

rition-center/service/pom.xml

@@ -23,4 +23,11 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
 
+    <dependencies>
+        <dependency>
+            <groupId>net.rition</groupId>
+            <artifactId>common</artifactId>
+            <version>${revision}</version>
+        </dependency>
+    </dependencies>
 </project>

rition-probe/client/client.go

@@ -1,12 +1,9 @@
 package client
 
-import "time"
-
 type Config struct {
 	CenterServer   string        `json:"centerServer,omitempty"`
 	HostName       string        `json:"hostName,omitempty"`
 	InstanceId     string        `json:"instanceId,omitempty"`
-	ReportInterval time.Duration `json:"reportInterval,omitempty"`
 }
 
 type Client struct {

rition-probe/client/http.go

@@ -1,16 +1,79 @@
 package client
 
 import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
 	"time"
 )
 
 type ReportDataItem struct {
 	Metric    string            `json:"metric,omitempty"`
+	Value     any               `json:"value,omitempty"`
 	Tags      map[string]string `json:"tags,omitempty"`
-	Timestamp time.Time         `json:"timestamp"`
+	Timestamp int64             `json:"timestamp"`
+}
+
+type DataItem struct {
+	Metric    string `json:"metric,omitempty"`
 	Value     any    `json:"value,omitempty"`
+	Timestamp int64  `json:"timestamp"`
+}
+
+type ReportResponse struct {
+	RequestId string `json:"request_id,omitempty"`
+	Message   string `json:"message,omitempty"`
+}
+
+func (c *Client) Report(dataItem []DataItem) (*ReportResponse, error) {
+	tags := map[string]string{
+		"instanceId": c.config.InstanceId,
+		"hostname":   c.config.HostName,
+	}
+
+	// convert data
+	reportDataItems := make([]ReportDataItem, 0, len(dataItem))
+	for _, item := range dataItem {
+		reportDataItems = append(reportDataItems, ReportDataItem{
+			Metric:    item.Metric,
+			Value:     item.Value,
+			Tags:      tags,
+			Timestamp: time.Now().UnixMilli(),
+		})
 	}
 
-func (c *Client) Report() {
+	reportJsonBytes, err := json.Marshal(reportDataItems)
+	if err != nil {
+		return nil, err
+	}
+
+	fmt.Println(string(reportJsonBytes))
+
+	// send report data
+	resp, err := http.Post(c.config.CenterServer, "application/json", bytes.NewBuffer(reportJsonBytes))
+	if err != nil {
+		return nil, err
+	}
+
+	// get response
+	defer func(Body io.ReadCloser) {
+		_ = Body.Close()
+	}(resp.Body)
+
+	respBytes, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	fmt.Println(string(respBytes))
+
+	response := ReportResponse{}
+	err = json.Unmarshal(respBytes, &response)
+	if err != nil {
+		return nil, err
+	}
 
+	return &response, nil
 }

rition-probe/main.go

@@ -1,5 +1,37 @@
 package main
 
+import (
+	"flag"
+	"os"
+	"rition-probe/client"
+	"rition-probe/service"
+	"time"
+)
+
 func main() {
+	// --report '127.0.0.1:22019' --disk_dev nvme0n1 --mount / --net_dev wlp0s20f3 --interval 30
+	centerServer := flag.String("report", "", "上报中心")
+	diskDevice := flag.String("disk_dev", "", "监控的硬盘设备")
+	mount := flag.String("mount", "/", "监控的分区挂载位置")
+	netInterface := flag.String("net_dev", "", "监控的网卡设备")
+	interval := flag.Int("interval", 30, "上报间隔，单位为秒")
+	flag.Parse()
+
+	hostname, _ := os.Hostname()
+
+	config := service.Config{
+		ClientConfig: client.Config{
+			CenterServer: *centerServer,
+			HostName:     hostname,
+			InstanceId:   "5c4928f8-f27a-4ac9-85b4-c2dd03e086ad",
+		},
+		MonitorDiskDevice:       *diskDevice,
+		MonitorDiskMountPoint:   *mount,
+		MonitorNetworkInterface: *netInterface,
+		ReportInterval:          time.Duration(*interval) * time.Second,
+	}
 
+	srv := service.NewService(config)
+	//srv.Run()
+	srv.Report()
 }

rition-probe/probe/cpu.go

@@ -1,6 +1,8 @@
 package probe
 
-import "github.com/shirou/gopsutil/v3/cpu"
+import (
+	"github.com/shirou/gopsutil/v3/cpu"
+)
 
 func (p *Probe) CurrentCPUTotalPercent() float64 {
 	cpuPercents, _ := cpu.Percent(0, false)

rition-probe/probe/cpu_test.go

@@ -0,0 +1,10 @@
+package probe
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestProbe_CurrentCPUTotalPercent(t *testing.T) {
+	fmt.Println(probeTest.CurrentCPUTotalPercent())
+}

rition-probe/probe/host.go

@@ -4,7 +4,7 @@ import (
 	"github.com/shirou/gopsutil/v3/host"
 )
 
-func (p *Probe) HostUpTime() uint64 {
+func (p *Probe) HostUptime() uint64 {
 	upTime, _ := host.Uptime()
 	return upTime
 }

rition-probe/probe/network.go

@@ -9,13 +9,14 @@ func (p *Probe) GetNetworkCounterAll() []net.IOCountersStat {
 	return stats
 }
 
-func (p *Probe) GetNetworkCounterOne(name string) net.IOCountersStat {
+func (p *Probe) GetNetworkCounterOne(name string) (net.IOCountersStat, []net.ConnectionStat) {
-	stats, _ := net.IOCounters(true)
+	ioCounters, _ := net.IOCounters(true)
-	for _, stat := range stats {
+	tcpConnStat, _ := net.Connections("tcp")
+	for _, stat := range ioCounters {
 		if stat.Name == name {
-			return stat
+			return stat, tcpConnStat
 		}
 	}
 
-	return net.IOCountersStat{}
+	return net.IOCountersStat{}, tcpConnStat
 }

rition-probe/probe/network_test.go

@@ -10,4 +10,10 @@ func TestProbe_GetNetworkCounter(t *testing.T) {
 	for _, counter := range counters {
 		fmt.Println(counter)
 	}
+
+	networkIoCount, connStat := probeTest.GetNetworkCounterOne("wlp0s20f3")
+	fmt.Printf("%+v", networkIoCount)
+	fmt.Printf("--------------------------")
+	fmt.Printf("%+v", connStat)
+	println()
 }

rition-probe/service/probe.go

@@ -0,0 +1,106 @@
+package service
+
+import (
+	"fmt"
+	"rition-probe/client"
+	"time"
+)
+
+func (s *Service) Run() {
+	ticker := time.NewTicker(s.config.ReportInterval)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		s.Report()
+	}
+}
+
+func (s *Service) Report() {
+	dataItems := make([]client.DataItem, 0)
+	nowTime := time.Now()
+	now := nowTime.UnixMilli()
+
+	// cpu
+	_5minCPULoad := s.prevCpuLoadData
+	if nowTime.Sub(s.prevCpuLoadDataCollectTime).Minutes() >= 5 {
+		_5minCPULoad = s.probe.CurrentCPUTotalPercent()
+	}
+	dataItems = append(dataItems, client.DataItem{
+		Metric:    "node_load5",
+		Value:     _5minCPULoad,
+		Timestamp: now,
+	})
+
+	sysUptime := s.probe.HostUptime()
+	dataItems = append(dataItems, client.DataItem{
+		Metric:    "node_cpu_seconds_total",
+		Value:     sysUptime,
+		Timestamp: now,
+	})
+
+	// mem
+	memTotal, memUsed, memBuffers, memCached := s.probe.RamUsage()
+	dataItems = append(dataItems, client.DataItem{
+		Metric: "node_memory_MemTotal_bytes", Value: memTotal, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_memory_MemFree_bytes", Value: memTotal - memUsed, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_memory_Buffers_bytes", Value: memBuffers, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_memory_Cached_bytes", Value: memCached, Timestamp: now,
+	})
+
+	// disk
+	usageStat, diskIoCount := s.probe.DiskStatus(s.config.MonitorDiskMountPoint, s.config.MonitorDiskDevice)
+	dataItems = append(dataItems, client.DataItem{
+		Metric: "node_filesystem_avail_bytes", Value: usageStat.Free, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_filesystem_size_bytes", Value: usageStat.Total, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_filesystem_free_bytes", Value: usageStat.Free, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_disk_read_bytes_total", Value: diskIoCount.ReadBytes, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_disk_written_bytes_total", Value: diskIoCount.WriteBytes, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_disk_reads_completed_total", Value: diskIoCount.MergedWriteCount, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_disk_writes_completed_total", Value: diskIoCount.MergedWriteCount, Timestamp: now,
+	})
+
+	// network
+	networkIoCount, conns := s.probe.GetNetworkCounterOne(s.config.MonitorNetworkInterface)
+	dataItems = append(dataItems, client.DataItem{
+		Metric: "node_network_receive_bytes_total", Value: networkIoCount.BytesRecv, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_network_transmit_bytes_total", Value: networkIoCount.BytesSent, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_network_receive_packets_total", Value: networkIoCount.PacketsRecv, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_network_transmit_packets_total", Value: networkIoCount.PacketsSent, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_network_receive_drop_total", Value: networkIoCount.Dropin, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_network_transmit_drop_total", Value: networkIoCount.Dropout, Timestamp: now,
+	})
+
+	estabCnt, twCnt := 0, 0
+	for _, conn := range conns {
+		switch conn.Status {
+		case "ESTABLISHED":
+			estabCnt++
+		case "TIME_WAIT":
+			twCnt++
+		}
+	}
+	dataItems = append(dataItems, client.DataItem{
+		Metric: "node_netstat_Tcp_CurrEstab", Value: estabCnt, Timestamp: now,
+	}, client.DataItem{
+		Metric: "node_netstat_Tcp_tw", Value: twCnt, Timestamp: now,
+	})
+
+	_, err := s.client.Report(dataItems)
+	if err != nil {
+		fmt.Println("err: ", err.Error())
+	}
+}

rition-probe/service/service.go

@@ -1,22 +1,38 @@
 package service
 
-import "time"
+import (
+	"rition-probe/client"
+	"rition-probe/probe"
+	"time"
+)
 
 type Config struct {
-	CenterServer   string        `json:"centerServer,omitempty"`
+	ClientConfig            client.Config
-	HostName       string        `json:"hostName,omitempty"`
+	MonitorDiskDevice       string
-	InstanceId     string        `json:"instanceId,omitempty"`
+	MonitorDiskMountPoint   string
+	MonitorNetworkInterface string
+
 	ReportInterval time.Duration `json:"reportInterval,omitempty"`
 }
 
 type Service struct {
 	config Config
+	probe  *probe.Probe
+	client *client.Client
+
+	prevCpuLoadDataCollectTime time.Time
+	prevCpuLoadData            float64
 }
 
 func NewService(config Config) *Service {
 	service := Service{
 		config: config,
+		probe:  probe.NewProbe(),
+		client: client.NetClient(config.ClientConfig),
 	}
 
+	service.prevCpuLoadData = service.probe.CurrentCPUTotalPercent()
+	service.prevCpuLoadDataCollectTime = time.Now()
+
 	return &service
 }