OCT

REM:
1. kafka发送消息

config/application-dispose.properties

@@ -53,6 +53,11 @@ kafka.producer.linger=1
 kafka.producer.buffer.memory=33554432
 kafka.producer.servers=172.21.44.189:9092,172.21.44.9:9092,172.21.44.244:9092,172.21.44.236:9092,172.21.44.80:9092
 kafka.dispose.topic=ddos-vip-customer-ck
+kafka.consumer.group-id = testGroup
+kafka.consumer.auto-offset-reset = earliest
+kafka.consumer.enable-auto-commit = true
+kafka.consumer.auto-commit-interval = 100
+kafka.consumer.max-poll-records =5
 
 #信任主机配置
 # 白名单开关

pom.xml

@@ -89,7 +89,6 @@
             <artifactId>spring-kafka</artifactId>
             <version>2.7.2</version>
         </dependency>
-
         <!--mysql -->
         <dependency>
             <groupId>mysql</groupId>

src/main/java/com/dispose/PhoenixBootApplication.java

@@ -5,6 +5,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
 import org.springframework.boot.web.servlet.ServletComponentScan;
 import org.springframework.context.annotation.EnableAspectJAutoProxy;
 import org.springframework.scheduling.annotation.EnableAsync;
@@ -16,7 +17,7 @@ import org.springframework.transaction.annotation.EnableTransactionManagement;
  *
  * @author <huangxin@cmhi.chinamoblie.com>
  */
-@SpringBootApplication
+@SpringBootApplication(exclude = {KafkaAutoConfiguration.class})
 @EnableAsync
 @EnableScheduling
 @EnableAspectJAutoProxy

src/main/java/com/dispose/controller/kafkaController.java

@@ -0,0 +1,108 @@
+package com.dispose.controller;
+
+import com.dispose.common.ErrorCode;
+import com.dispose.config.KafkaConfiguration;
+import com.dispose.pojo.dto.protocol.base.BaseRespStatus;
+import com.dispose.pojo.dto.protocol.base.ProtocolReqDTO;
+import com.dispose.pojo.dto.protocol.base.ProtocolRespDTO;
+import com.dispose.pojo.dto.protocol.kafka.AlarmInfo;
+import com.dispose.pojo.dto.protocol.kafka.AlarmInfoReq;
+import com.dispose.security.annotation.Decryption;
+import com.dispose.security.annotation.Encryption;
+import com.dispose.validation.group.ValidGroups;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Controller;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.validation.annotation.Validated;
+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.ResponseBody;
+
+import javax.annotation.Resource;
+import java.util.Objects;
+
+/**
+ * The type Auth controller.
+ *
+ * @author <huangxin@cmhi.chinamoblie.com>
+ */
+@Controller
+@RequestMapping(value = "/kafka")
+@Slf4j
+@Api(value = "处置平台发送消息接口", tags = "处置平台发送消息接口")
+@Component
+@Validated
+@Encryption
+@Decryption
+public class kafkaController {
+    /**
+     * The Object mapper.
+     */
+    @Resource
+    private ObjectMapper objectMapper;
+
+    /**
+     * The Kafka configuration.
+     */
+//    final private KafkaConfiguration kafkaConfiguration =
+//            SpringBootBeanUtil.getBean(com.dispose.config.KafkaConfiguration.class);
+    @Resource
+    private KafkaConfiguration kafkaConfiguration;
+
+    /**
+     * Dispatch command sent to kafka.
+     *
+     * @param mr the mr
+     * @return the protocol resp dto
+     */
+    @PostMapping("/dispatchCommand")
+    @ResponseBody
+    @ApiOperation("发送消息")
+    public ProtocolRespDTO<BaseRespStatus> dispatchCommand(
+            @Validated(ValidGroups.ProtocolCommonValid.class)
+            @RequestBody ProtocolReqDTO<AlarmInfoReq> mr) throws JsonProcessingException {
+        //获取入参信息，进行所需数据格式拼接
+        log.info("alarm message information :{}", mr.getMsgContent().getAlarmInfo());
+        AlarmInfo alarmInfo = objectMapper.readValue(mr.getMsgContent().getAlarmInfo(), new TypeReference<AlarmInfo>() {
+        });
+
+        //推动数据格式到kafka
+        log.info("send alarm message :{}", alarmInfo.toString());
+        String sendMessage = mr.getMsgContent().getAlarmInfo();
+        ListenableFuture<SendResult<String, String>> sendResult = kafkaConfiguration
+                .kafkaTemplate()
+                .sendDefault(0, System.currentTimeMillis(), "dispose", sendMessage);
+
+        sendResult.addCallback(v -> log.info("Kafka send {} to {} at {}", sendMessage,
+                        Objects.requireNonNull(v)
+                                .getRecordMetadata()
+                                .topic(), v.getRecordMetadata().partition()),
+                ex -> log.error("Kafka send error: {}", ex.getMessage()));
+
+        BaseRespStatus rspInfo = new BaseRespStatus();
+        rspInfo.setStatus(ErrorCode.ERR_OK.getCode());
+        rspInfo.setMessage(new String[]{ErrorCode.ERR_OK.getMsg()});
+        return ProtocolRespDTO.result(ErrorCode.ERR_OK, rspInfo);
+    }
+
+
+//    @KafkaListener(topics = {"ddos-vip-customer-ck"})
+//    public void kafkaListen(ConsumerRecord<String, String> consumerRecord) {
+//        //判断消息是否为null
+//        Optional<String> kafkaMessage = Optional.ofNullable(consumerRecord.value());
+//        log.info(">>>>>>>>>>> record = " + kafkaMessage);
+//        if (kafkaMessage.isPresent()) {
+//            String consumerMsg = kafkaMessage.get();
+//            log.info("消费消息：" + consumerMsg);
+//        }
+//    }
+
+}

src/main/java/com/dispose/pojo/dto/protocol/kafka/AlarmInfo.java

@@ -0,0 +1,84 @@
+package com.dispose.pojo.dto.protocol.kafka;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.util.List;
+
+@EqualsAndHashCode()
+@Data
+@Builder
+@AllArgsConstructor
+@JsonPropertyOrder({"alarmId", "dstIp", "attackType", "bpspps", "dstProvince", "dstCity", "srcIpLs", "startTime",
+        "endTime", "disposeType", "disposeTime", "maxBps", "maxPps"})
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class AlarmInfo {
+    /**
+     * 告警id.
+     */
+    private String alarmId;
+
+    /**
+     * 被攻击ip.
+     */
+    private String dstIp;
+
+    /**
+     * 攻击类型(类似HTTP Flood字符串形式).
+     */
+    private String attackType;
+
+    /**
+     * 被攻击ip.
+     */
+    private String bpspps;
+
+    /**
+     * 目的ip省份.
+     */
+    private String dstProvince;
+
+    /**
+     * 目的ip市.
+     */
+    private String dstCity;
+
+    /**
+     * 告警源ip列表,没有就给empty的list.
+     */
+    private List<String> srcIpLs;
+
+    /**
+     * 告警开始时间.
+     */
+    private String startTime;
+
+    /**
+     * 告警结束时间.
+     */
+    private String endTime;
+
+    /**
+     * 处置类型(1:清洗,2:黑洞,3:高防),默认清洗.
+     */
+    private Integer disposeType;
+
+    /**
+     * 处置时长.
+     */
+    private Integer disposeTime;
+
+    /**
+     * 流量峰值(单位:bps).
+     */
+    private String maxBps;
+
+    /**
+     * 包数峰值(单位:pps).
+     */
+    private String maxPps;
+}

src/main/java/com/dispose/pojo/dto/protocol/kafka/AlarmInfoReq.java

@@ -0,0 +1,29 @@
+package com.dispose.pojo.dto.protocol.kafka;
+
+import com.dispose.validation.group.ValidGroups;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import javax.validation.constraints.NotBlank;
+
+
+/**
+ * The type Login req.
+ *
+ * @author <huangxin@cmhi.chinamoblie.com>
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class AlarmInfoReq {
+    /**
+     * 告警信息.
+     */
+    @NotBlank(message = "alarmInfo 告警信息不能为空", groups = ValidGroups.ProtocolCommonValid.class)
+    private String alarmInfo;
+}

src/test/java/com/dispose/test/dev/controller/KafkaControllerTest.java

@@ -0,0 +1,126 @@
+package com.dispose.test.dev.controller;
+
+import cn.hutool.json.JSONObject;
+import com.dispose.common.ConstValue;
+import com.dispose.common.ErrorCode;
+import com.dispose.common.ProtoCryptoType;
+import com.dispose.pojo.dto.protocol.base.BaseRespStatus;
+import com.dispose.pojo.dto.protocol.base.ProtocolReqDTO;
+import com.dispose.pojo.dto.protocol.base.ProtocolRespDTO;
+import com.dispose.pojo.dto.protocol.kafka.AlarmInfoReq;
+import com.dispose.test.dev.Global.InitTestEnvironment;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jodd.net.HttpStatus;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.Assert;
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.MethodSorters;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.http.MediaType;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.annotation.Rollback;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
+import org.springframework.transaction.annotation.Transactional;
+
+import javax.annotation.Resource;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+/**
+ * The type Auth controller test.
+ *
+ * @author <huangxin@cmhi.chinamoblie.com>
+ */
+@AutoConfigureMockMvc
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+@Transactional
+@Rollback
+@Slf4j
+public class KafkaControllerTest extends InitTestEnvironment {
+    /**
+     * The Mock mvc.
+     */
+    @Resource
+    private MockMvc mockMvc;
+    /**
+     * The Object mapper.
+     */
+    @Resource
+    private ObjectMapper objectMapper;
+
+    /**
+     * A 1 login.
+     *
+     * @throws Exception the exception
+     */
+    @Test
+    public void a1_dispatchCommand() throws Exception {
+        JSONObject disposeParam = new JSONObject();
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+        disposeParam.set("alarmId", "1");
+        disposeParam.set("dstIp", "192.168.1.1");
+        disposeParam.set("attackType", "1");
+        disposeParam.set("bpspps", "bps");
+        disposeParam.set("dstProvince", "ZHEJIANG");
+        disposeParam.set("dstCity", "HANGZHOU");
+        disposeParam.set("startTime", sdf.format(new Date()));
+        // 1清洗，2流控，3黑洞
+        disposeParam.set("disposeType", 1);
+        disposeParam.set("disposeTime", 30);
+        disposeParam.set("endTime", new Date());
+        List<String> srcIp = new ArrayList<>();
+        srcIp.add("192.168.10.1");
+        srcIp.add("192.168.10.2");
+        disposeParam.set("srcIpLs", srcIp);
+        disposeParam.set("maxBps", "20bps");
+        disposeParam.set("maxPps", "20pps");
+        String jsonAlarmInfo = disposeParam.toString();
+        log.info("============ jsonAlarmInfo :{}", jsonAlarmInfo);
+
+        AlarmInfoReq alarmInfoReq = AlarmInfoReq.builder()
+                .alarmInfo(jsonAlarmInfo).build();
+
+        ProtocolReqDTO<AlarmInfoReq> reqInfo = new ProtocolReqDTO<>();
+
+        reqInfo.setVer(ConstValue.Protocol.VERSION);
+        reqInfo.setCryptoType(ProtoCryptoType.CRYPTO_NONE.getCode());
+        reqInfo.setTimeStamp(System.currentTimeMillis());
+        reqInfo.setMsgContent(alarmInfoReq);
+
+        String ret = mockMvc.perform(MockMvcRequestBuilders
+                        .post("/kafka/dispatchCommand")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .header("Authorization", ConstValue.STRING_HTTP_AUTH_HEAD + getLoginToken())
+                        .content(objectMapper.writeValueAsString(reqInfo)))
+                .andDo(print()).andExpect(status().isOk())
+                .andExpect(jsonPath("$.code").value(HttpStatus.ok().status()))
+                .andReturn()
+                .getResponse()
+                .getContentAsString();
+
+        ProtocolRespDTO<BaseRespStatus> rspInfo = objectMapper.readValue(ret,
+                new TypeReference<ProtocolRespDTO<BaseRespStatus>>() {
+                });
+
+        verifyRespProtocol(rspInfo);
+
+        log.debug(objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(rspInfo));
+        Assert.assertEquals(ErrorCode.ERR_OK.getCode(), (long) rspInfo.getMsgContent().getStatus());
+    }
+}

src/test/java/com/dispose/test/dev/debug/demo.java

@@ -1,5 +1,6 @@
 package com.dispose.test.dev.debug;
 
+import cn.hutool.json.JSONObject;
 import com.dispose.common.ConstValue;
 import com.dispose.common.DDoSAttackType;
 import com.dispose.common.DisposeConfigValue;
@@ -26,8 +27,10 @@ import java.lang.reflect.Modifier;
 import java.nio.charset.StandardCharsets;
 import java.security.InvalidKeyException;
 import java.security.NoSuchAlgorithmException;
+import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Date;
 import java.util.Hashtable;
 import java.util.List;
 import java.util.Map;
@@ -684,4 +687,30 @@ public class demo {
             }
         }
     }
+
+    @Test
+    public void alarmInfoTest() {
+        JSONObject disposeParam = new JSONObject();
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+        disposeParam.set("alarmId", "1");
+        disposeParam.set("dstIp", "192.168.1.1");
+        disposeParam.set("attackType", "1");
+        disposeParam.set("bpspps", "bps");
+        disposeParam.set("dstProvince", "ZHEJIANG");
+        disposeParam.set("dstCity", "HANGZHOU");
+        disposeParam.set("startTime", sdf.format(new Date()));
+        // 1清洗，2流控，3黑洞
+        disposeParam.set("disposeType", 1);
+        disposeParam.set("disposeTime", 30);
+        disposeParam.set("endTime", "");
+        List<String> srcIp = new ArrayList<>();
+        srcIp.add("192.168.10.1");
+        srcIp.add("192.168.10.2");
+        disposeParam.set("srcIpLs", srcIp);
+        disposeParam.set("maxBps", "20bps");
+        disposeParam.set("maxPps", "20pps");
+        String jsonAlarmInfo = disposeParam.toString();
+        log.info("============ jsonAlarmInfo :{}", jsonAlarmInfo);
+    }
 }