feat: adapt to the gateway api

13 files changed, 82 insertions, 258 deletions

diff --git a/platform-base/src/main/java/com/zdjizhi/base/rule/async/FixedRateFetcher.java b/platform-base/src/main/java/com/zdjizhi/base/rule/async/FixedRateFetcher.java
index 5e81e20..9902fc4 100644
--- a/platform-base/src/main/java/com/zdjizhi/base/rule/async/FixedRateFetcher.java
+++ b/platform-base/src/main/java/com/zdjizhi/base/rule/async/FixedRateFetcher.java
@@ -41,7 +41,7 @@ public class FixedRateFetcher implements Fetcher, ScheduledSupport {
             return;
         }
         scheduledFuture = scheduledExecutorService()
-                .scheduleAtFixedRate(() -> {
+                .scheduleWithFixedDelay(() -> {
                     try {
                         runnable.run();
                         // invoke callback
diff --git a/platform-etl/src/main/java/com/zdjizhi/etl/common/CommonConfig.java b/platform-etl/src/main/java/com/zdjizhi/etl/common/CommonConfig.java
index 08277a7..5545c76 100644
--- a/platform-etl/src/main/java/com/zdjizhi/etl/common/CommonConfig.java
+++ b/platform-etl/src/main/java/com/zdjizhi/etl/common/CommonConfig.java
@@ -25,14 +25,15 @@ public class CommonConfig {
             .stringType()
             .defaultValue("DHCP,DNS,FTP,GRE,GTP,HTTP,HTTPS,ICMP,IMAP,IMAPS,IPSEC,ISAKMP,XMPP,L2TP,LDAP,MMS,NETBIOS,NETFLOW,NTP,POP3,POP3S,RDP,PPTP,RADIUS,RTCP,RTP,RTSP,SIP,SMB,SMTP,SMTPS,SNMP,SSDP,SSH,SSL,STUN,TELNET,TFTP,OPENVPN,RTMP,TEREDO,FTPS,DTLS,SPDY,BJNP,QUIC,MDNS,Unknown TCP,Unknown UDP,Unknown Other,IKE,MAIL,SOCKS,DoH,SLP,SSL with ESNI,ISATAP,Stratum,SSL with ECH");
 
-    public static final ConfigOption<String> KNOWLEDGE_URL = ConfigOptions.key("knowledge.url")
+    public static final ConfigOption<String> GATEWAY_ADDRESS = ConfigOptions.key("gateway.address")
             .stringType()
             .noDefaultValue();
-
-    public static final ConfigOption<String> KNOWLEDGE_URL_TOKEN = ConfigOptions.key("knowledge.url.token")
+    public static final ConfigOption<String> GATEWAY_PORT = ConfigOptions.key("gateway.port")
             .stringType()
-            .defaultValue("1a653ea0-d39b-4246-94b0-1ba95db4b6a7");
-
+            .defaultValue("9999");
+    public static final ConfigOption<String> KNOWLEDGE_PATH = ConfigOptions.key("knowledge.path")
+            .stringType()
+            .defaultValue("/v1/knowledge_base");
     public static final ConfigOption<Integer> KNOWLEDGE_URL_INTERVAL_MS = ConfigOptions.key("knowledge.url.interval.ms")
             .intType()
             .defaultValue(30000);
diff --git a/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/CustomKnowledge.java b/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/CustomKnowledge.java
index cfd25c0..85a5792 100644
--- a/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/CustomKnowledge.java
+++ b/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/CustomKnowledge.java
@@ -5,6 +5,7 @@ import com.zdjizhi.base.utils.HttpClientUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -16,7 +17,7 @@ public class CustomKnowledge {
     private final Map<Long, Map<String, Object>> metadataMap = new HashMap<>();
 
     public void addMetadata(Map<String, Object> metadata) {
-        Long id = Long.parseLong((String) metadata.get("id"));
+        Long id = Long.parseLong((String) metadata.get("kb_id"));
         metadataMap.put(id, metadata);
     }
 
@@ -38,8 +39,8 @@ public class CustomKnowledge {
     public Map<Long, byte[]> downloadKnowledge(int maxRetry) {
         Map<Long, byte[]> contents = new HashMap<>(this.metadataMap.size());
         this.metadataMap.values().forEach(metadata -> contents.put(
-                Long.parseLong((String) metadata.get("id")),
-                downloadFile((String) metadata.get("path"), (String) metadata.get("sha256"), (Integer) metadata.get("isValid"), maxRetry)));
+                Long.parseLong((String) metadata.get("kb_id")),
+                downloadFile((String) metadata.get("path"), (String) metadata.get("sha256"), (Integer) metadata.get("is_valid"), maxRetry)));
         return contents;
     }
 
@@ -59,7 +60,8 @@ public class CustomKnowledge {
             }
         } while (!downloadFileSha256.equals(sha256) && downloadCount < maxRetry);
         if (downloadCount >= maxRetry) {
-            logger.warn("warning: file url: " + fileUrl + " download more than specified number of times");
+            int length = Math.min(1000, content.length);
+            logger.warn("warning: file url: " + fileUrl + " download more than specified number of times, the part of content is :" + new String(content, 0, length, StandardCharsets.UTF_8));
         }
         return content;
     }
diff --git a/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/KnowledgeManager.java b/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/KnowledgeManager.java
index d3f2ed1..38fae26 100644
--- a/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/KnowledgeManager.java
+++ b/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/KnowledgeManager.java
@@ -54,7 +54,7 @@ public class KnowledgeManager {
     private void checkMetaList(ArrayList<Map<String, Object>> metaList) {
         List<String> types = new ArrayList<>(metaList.size());
         for (Map<String, Object> metadata : metaList) {
-            String type = (String) metadata.get("type");
+            String type = (String) metadata.get("category");
             types.add(type);
         }
         check(types, configuration.get(CommonInternalConfig.IP_TYPE));
@@ -98,8 +98,8 @@ public class KnowledgeManager {
                 String format = (String) metadata.get("format");
                 String sha256 = (String) metadata.get("sha256");
                 String fileUrl = (String) metadata.get("path");
-                String type = (String) metadata.get("type");
-                Integer isValidInteger = (Integer) metadata.get("isValid");
+                String type = (String) metadata.get("category");
+                Integer isValidInteger = (Integer) metadata.get("is_valid");
                 int isValid = isValidInteger == null ? 1 : isValidInteger;
 
                 if (tagTypes.contains(type)) {
@@ -269,7 +269,7 @@ public class KnowledgeManager {
 
     private String buildExpr() {
         StringBuilder sb = new StringBuilder("");
-        sb.append("$.[?(@.version=='latest' && @.type in ['");
+        sb.append("$.[?(@.version=='latest' && @.category in ['");
         sb.append(configuration.get(CommonInternalConfig.IP_TYPE));
         sb.append("','");
         sb.append(configuration.get(CommonInternalConfig.ASN_TYPE));
@@ -301,7 +301,7 @@ public class KnowledgeManager {
         sb.append(configuration.get(CommonInternalConfig.IOC_DARKWEB_TYPE));
         sb.append("','");
         sb.append(configuration.get(CommonInternalConfig.IOC_MALWARE_TYPE));
-        sb.append("'])].['name','sha256','format','path','type','isValid','id']");
+        sb.append("'])].['name','sha256','format','path','category','is_valid','kb_id']");
         return sb.toString();
     }
 
diff --git a/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/MetadataListenerUtils.java b/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/MetadataListenerUtils.java
index b6761f6..2dde806 100644
--- a/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/MetadataListenerUtils.java
+++ b/platform-etl/src/main/java/com/zdjizhi/etl/knowledge/MetadataListenerUtils.java
@@ -15,6 +15,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -25,12 +26,22 @@ import java.util.concurrent.atomic.AtomicBoolean;
 public class MetadataListenerUtils {
 
     private final static AtomicBoolean ASYNC_LISTEN_RUNNING = new AtomicBoolean(false);
+    private final static CountDownLatch ASYNC_WAIT = new CountDownLatch(1);
 
-    public static void onListen(Configuration configuration) {
+    public static void onListen(Configuration configuration) throws InterruptedException {
         if (ASYNC_LISTEN_RUNNING.compareAndSet(false, true)) {
-            FixedRateFetcher fixedRateFetcher = new FixedRateFetcher(new FetchRunnable(configuration), configuration.get(CommonConfig.KNOWLEDGE_URL_INTERVAL_MS));
-            fixedRateFetcher.open();
+            try {
+                FetchRunnable fetchRunnable = new FetchRunnable(configuration);
+                fetchRunnable.run();
+                FixedRateFetcher fixedRateFetcher = new FixedRateFetcher(fetchRunnable, configuration.get(CommonConfig.KNOWLEDGE_URL_INTERVAL_MS));
+                fixedRateFetcher.open();
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            } finally {
+                ASYNC_WAIT.countDown();
+            }
         }
+        ASYNC_WAIT.await();
     }
 
     private static final class FetchRunnable implements Runnable {
@@ -48,14 +59,21 @@ public class MetadataListenerUtils {
         public FetchRunnable(Configuration configuration) {
             this.configuration = configuration;
             knowledgeManager = new KnowledgeManager(configuration);
-            url = configuration.get(CommonConfig.KNOWLEDGE_URL) + "/v1/knowledge_base";
+            url = getUrl();
+        }
+
+        private String getUrl() {
+            return "http://"
+                    + configuration.get(CommonConfig.GATEWAY_ADDRESS)
+                    + ":"
+                    + configuration.get(CommonConfig.GATEWAY_PORT)
+                    + configuration.get(CommonConfig.KNOWLEDGE_PATH);
         }
 
         @Override
         public void run() {
             final HttpGet httpGet = new HttpGet(url);
             httpGet.addHeader("Accept", "application/json");
-            httpGet.addHeader("Cn-Authorization", configuration.get(CommonConfig.KNOWLEDGE_URL_TOKEN));
             try {
                 CloseableHttpResponse response = HTTP_CLIENT.execute(httpGet);
                 HttpEntity entity = response.getEntity();
diff --git a/platform-etl/src/main/java/com/zdjizhi/etl/operator/EtlProcessFunc.java b/platform-etl/src/main/java/com/zdjizhi/etl/operator/EtlProcessFunc.java
index fad81a6..e5b2481 100644
--- a/platform-etl/src/main/java/com/zdjizhi/etl/operator/EtlProcessFunc.java
+++ b/platform-etl/src/main/java/com/zdjizhi/etl/operator/EtlProcessFunc.java
@@ -3,7 +3,7 @@ package com.zdjizhi.etl.operator;
 import com.alibaba.fastjson2.JSON;
 import com.zdjizhi.base.common.CnRecordLog;
 import com.zdjizhi.etl.common.CommonConfig;
-import com.zdjizhi.etl.knowledge.NacosManager;
+import com.zdjizhi.etl.knowledge.MetadataListenerUtils;
 import com.zdjizhi.etl.rule.RuleUpdateListener;
 import com.zdjizhi.etl.utils.*;
 import com.zdjizhi.etl.utils.fqdn.CategoryUtils;
@@ -31,8 +31,9 @@ public class EtlProcessFunc extends ProcessFunction<String, CnRecordLog> {
                 configuration.get(CommonConfig.IP_GEO_PATTERN_FIELD),
                 configuration.get(CommonConfig.IP_GEO_PATTERN_VALUE));
         CompletedUtils.init(configuration.get(CommonConfig.L7_PROTOCOL));
-        NacosManager nacosManager = new NacosManager(configuration);
-        nacosManager.startAsyncListen();
+        /*NacosManager nacosManager = new NacosManager(configuration);
+        nacosManager.startAsyncListen();*/
+        MetadataListenerUtils.onListen(configuration);
         RuleUpdateListener ruleUpdateListener = new RuleUpdateListener(configuration);
         ruleUpdateListener.registerListener();
     }
diff --git a/platform-schedule/src/main/resources/business.properties b/platform-schedule/src/main/resources/business.properties
index 846a4d9..ebdc14a 100644
--- a/platform-schedule/src/main/resources/business.properties
+++ b/platform-schedule/src/main/resources/business.properties
@@ -1,5 +1,10 @@
+# session-record-cn??
 cn.record.etl.class=com.zdjizhi.etl.CnRecordPersistence
+# ???
 cn.pre.metric.class=com.zdjizhi.pre.CnPreMetric
+# ?????????
 cn.pre.relation.metric.class=com.zdjizhi.pre.relation.CnRelationMetric
+# dns???
 cn.dns.pre.metric.class=com.zdjizhi.pre.dns.DnsPreMetric
+# detection
 cn.detection.indicator.class=com.zdjizhi.schedule.indicator.IndicatorSchedule
\ No newline at end of file
diff --git a/platform-schedule/src/main/resources/common.internal.properties b/platform-schedule/src/main/resources/common.internal.properties
deleted file mode 100644
index a6e2078..0000000
--- a/platform-schedule/src/main/resources/common.internal.properties
+++ /dev/null
@@ -1,43 +0,0 @@
-ipv4.type=cn_ipv4_location_built_in
-
-ipv6.type=cn_ipv6_location_built_in
-
-asnv4.type=cn_ipv4_asn_built_in
-
-asnv6.type=cn_ipv6_asn_built_in
-
-webskt.type=cn_fqdn_category_built_in
-
-dns.type=cn_dns_server_info_built_in
-
-icp.type=cn_fqdn_icp_built_in
-
-link.type=cn_link_direction_built_in
-
-idc.renter.type=cn_idc_renter_built_in
-
-internal.ip.type=cn_internal_ip_built_in
-
-appskt.type=cn_app_category_built_in
-
-whois.type=cn_fqdn_who_is_built_in
-
-ioc.type=cn_ioc_malware
-
-ip.tag.type=cn_ip_tag_user_defined
-
-domain.tag.type=cn_domain_tag_user_defined
-
-app.tag.type=cn_app_tag_user_defined
-
-psiphon.type=cn_psiphon3_ip
-
-ioc.darkweb.type=cn_ioc_darkweb
-
-ioc.malware.type=cn_ioc_malware
-
-etl.topic=SESSION-RECORD-TEMP
-
-init.check=1
-
-test.running=1
\ No newline at end of file
diff --git a/platform-schedule/src/main/resources/common.properties b/platform-schedule/src/main/resources/common.properties
index 0c6cda6..335c9f8 100644
--- a/platform-schedule/src/main/resources/common.properties
+++ b/platform-schedule/src/main/resources/common.properties
@@ -1,112 +1,37 @@
-#flink?????????????????????????????????????
-stream.execution.environment.parallelism=1
-
-#flink????????
+# ????
 stream.execution.job.name=ETL-METRIC
-
-#???SESSION-RECORD-COMPLETED
-session.record.completed.parallelism=1
-session.record.completed.topic=SESSION-RECORD-TEMP
-session.record.completed.group.id=test-0921
-
-#??kafka??
-kafka.input.bootstrap.servers=192.168.44.113:9092
-
-#???CN-RECORD
-cn.record.topic=test
-cn.record.parallelism=1
-
-#??kafka??
+# ?????
+stream.execution.environment.parallelism=2
+# kafka source???
+session.record.completed.parallelism=2
+# session-record-cn sink???
+cn.record.parallelism=2
+# ???sink???
+metric.output.parallelism=2
+# dns???sink???
+dns.metric.output.parallelism=2
+# ????sink???
+metric.entity.relation.output.parallelism=2
+# ????sink???
+metric.dynamic.attribute.output.parallelism=2
+etl.topic=SESSION-RECORD
+# kafka???
+kafka.input.bootstrap.servers=192.168.44.12:9092
+session.record.completed.topic=SESSION-RECORD
+session.record.completed.group.id=55-test
+# kafka???
 kafka.output.bootstrap.servers=192.168.44.55:9092
-
-#zookeerper??
-zookeeper.servers=192.168.44.55:2181
-
-data.center.id.num=15
-
-#ES ??,??: ip1:port,ip2:port...
-es.host=192.168.44.55:9200
-#?????es???
-es.sink.parallelism=1
-
-#kafka????????
-sasl.jaas.config.user=admin
-sasl.jaas.config.password=galaxy2019
-
-#??????kafka???????1???0??
+# SESSION-RECORD-CN??topic
+cn.record.topic=SESSION-RECORD-CN
+# kafka???sasl?? 0:? 1:?
 input.sasl.jaas.config.flag=0
-
-#??????kafka???????1???0??
+# kafka???sasl?? 0:? 1:?
 output.sasl.jaas.config.flag=0
-
-############################## Flink checkpoint ?? ######################################
-#????checkpoint,0:???;1:??
+# flink checkpoint?? 0:? 1:?
 flink.enable.checkpoint.flag=1
-#???? checkpoint ??,??:ms
-flink.enable.checkpoint.time=60
-#?? checkpoint ???????,??:ms
-flink.min.pause.between.checkpoints=30
-#?? checkpoint ????,??:ms
-flink.checkpoint.timeout=30
-#??????? checkpoint ??
-flink.tolerable.checkpoint.failure.number=2
-#????????? checkpoint ??
-flink.max.concurrent.checkpoint=1
-
-############################## Flink-Kafka ????? ######################################
-#producer???????
-kafka.producer.retries=0
-
-#?????????Batch????????????????Batch??????????????
-kafka.producer.linger.ms=10
-
-#???????????????????????????
-kafka.producer.request.timeout.ms=30000
-
-#producer????batch?????,???????:16384
-kafka.producer.batch.size=262144
-
-#Producer?????????????
-#128M
-kafka.producer.buffer.memory=134217728
-
-#????????????Kafka??????????,??1048576
-#10M
-kafka.producer.max.request.size=10485760
-
-#??kafka??????????
-kafka.producer.compression.type=snappy
-
-#common_app_label????????APP?????0???1??????0
-app.label.is.third=0
-
-http.pool.max.connection=400
-http.pool.max.per.route=80
-http.pool.request.timeout=60000
-http.pool.connect.timeout=60000
-http.pool.response.timeout=60000
-
-nacos.server.addr=192.168.44.113:8848
-nacos.namespace=
-nacos.username=nacos
-nacos.password=nacos
-nacos.data.id=knowledge_base.json
-nacos.group=DEFAULT_GROUP
-nacos.read.timeout=5000
-
-hos.token=c21f969b5f03d33d43e04f8f136e7682
-
-cluster.or.single=SINGLE
-
-############################## ?????????? ?? ######################################
-hdfs.path=/test/TEST/
-hdfs.uri.nn1=192.168.40.151:9000
-hdfs.uri.nn2=192.168.40.152:9000
-hdfs.user=cn
-
-############################## ???????????? ?? ######################################
-download.path=C://Users//admin//Desktop//tmp//data//
-
-knowledge.file.check.number=3
-
-watermark.seconds=1
\ No newline at end of file
+# nacos??
+nacos.server.addr=192.168.44.55:8848
+# api detection url
+rule.full.url=http://192.168.44.54:8090/v1/rule/detection
+rule.inc.url=http://192.168.44.54:8090/v1/rule/detection/increase
+gateway.address=192.168.44.55
\ No newline at end of file
diff --git a/platform-schedule/src/main/resources/dns_pre_metrics.properties b/platform-schedule/src/main/resources/dns_pre_metrics.properties
deleted file mode 100644
index 2d8c1ff..0000000
--- a/platform-schedule/src/main/resources/dns_pre_metrics.properties
+++ /dev/null
@@ -1,33 +0,0 @@
-#topN 统计窗口大小，默认：1min
-dns.pre.metrics.window.time=1
-
-#保留小数位数，默认4位
-dns.pre.metrics.round.scale=4
-
-#dns metric topic
-#dns.metric.server.ip.topic=METRIC-DNS-SERVER-IP
-#dns.metric.qname.topic=METRIC-DNS-QNAME
-#dns.metric.qtype.topic=METRIC-DNS-QTYPE
-#dns.metric.rcode.topic=METRIC-DNS-RCODE
-#dns.metric.rra.topic=METRIC-DNS-RR-A
-#dns.metric.rraaaa.topic=METRIC-DNS-RR-AAAA
-#dns.metric.rrcname.topic=METRIC-DNS-RR-CNAME
-
-#dns.metric.server.ip.topic=METRIC-DNS-SERVER-IP-TEST
-#dns.metric.qname.topic=METRIC-DNS-QNAME-TEST
-#dns.metric.qtype.topic=METRIC-DNS-QTYPE-TEST
-#dns.metric.rcode.topic=METRIC-DNS-RCODE-TEST
-#dns.metric.rra.topic=METRIC-DNS-RR-A-TEST
-#dns.metric.rraaaa.topic=METRIC-DNS-RR-AAAA-TEST
-#dns.metric.rrcname.topic=METRIC-DNS-RR-CNAME-TEST
-
-dns.metric.server.ip.topic=test
-dns.metric.qname.topic=test
-dns.metric.qtype.topic=test
-dns.metric.rcode.topic=test
-dns.metric.rra.topic=test
-dns.metric.rraaaa.topic=test
-dns.metric.rrcname.topic=test
-
-#metric输出并行度设置
-dns.metric.output.parallelism=1
\ No newline at end of file
diff --git a/platform-schedule/src/main/resources/etl.properties b/platform-schedule/src/main/resources/etl.properties
deleted file mode 100644
index f73eb4c..0000000
--- a/platform-schedule/src/main/resources/etl.properties
+++ /dev/null
@@ -1,10 +0,0 @@
-#??IP??????? 1??????????? 2?????????  3???common_direction????
-ip.internal.or.external.pattern=1
-
-#?ip.internal.or.external.pattern=2?,????????
-#?????????????????country/province/region
-ip.geo.pattern.field=province
-#??????????????????????(,)??????beijing,shanghai,tianjin
-ip.geo.pattern.value=beijing,shanghai,tianjin
-
-l7.protocol=DHCP,DNS,FTP,GRE,GTP,HTTP,HTTPS,ICMP,IMAP,IMAPS,IPSEC,ISAKMP,XMPP,L2TP,LDAP,MMS,NETBIOS,NETFLOW,NTP,POP3,POP3S,RDP,PPTP,RADIUS,RTCP,RTP,RTSP,SIP,SMB,SMTP,SMTPS,SNMP,SSDP,SSH,SSL,STUN,TELNET,TFTP,OPENVPN,RTMP,TEREDO,FTPS,DTLS,SPDY,BJNP,QUIC,MDNS,Unknown TCP,Unknown UDP,Unknown Other,IKE,MAIL,SOCKS,DoH,SLP,SSL with ESNI,ISATAP,Stratum,SSL with ECH
\ No newline at end of file
diff --git a/platform-schedule/src/main/resources/indicator-match.properties b/platform-schedule/src/main/resources/indicator-match.properties
deleted file mode 100644
index 24d904e..0000000
--- a/platform-schedule/src/main/resources/indicator-match.properties
+++ /dev/null
@@ -1,5 +0,0 @@
-sink.kafka.topic=SECURITY-EVENT-CN
-sink.kafka.props.bootstrap.servers=192.168.44.55:9092
-
-rule.full.url=http://192.168.44.54:8090/v1/rule/detection
-rule.inc.url=http://192.168.44.54:8090/v1/rule/detection/increase
\ No newline at end of file
diff --git a/platform-schedule/src/main/resources/pre-metrics.properties b/platform-schedule/src/main/resources/pre-metrics.properties
deleted file mode 100644
index 0236504..0000000
--- a/platform-schedule/src/main/resources/pre-metrics.properties
+++ /dev/null
@@ -1,37 +0,0 @@
-#topN ??????????1min
-pre.metrics.window.time=1
-
-#?????????4?
-pre.metrics.round.scale=10
-
-#metric topic?
-#metric.ip.topic=METRIC-IP-TEST
-#metric.region.topic=METRIC-REGION-TEST
-#metric.asn.topic=METRIC-ASN-TEST
-#metric.idc.renter.topic=METRIC-IDC-RENTER-TEST
-#metric.application.topic=METRIC-APPLICATION-TEST
-#metric.domain.topic=METRIC-DOMAIN-TEST
-#metric.http.host.topic=METRIC-HTTP-HOST-TEST
-#metric.ssl.sni.topic=METRIC-SSL-SNI-TEST
-#metric.protocol.topic=METRIC-PROTOCOL-TEST
-#metric.link.topic=METRIC-LINK-TEST
-#metric.relation.topic=METRIC-RELATION-TEST
-#metric.attribute.topic=METRIC-ATTRIBUTE-TEST
-
-
-metric.ip.topic=test
-metric.region.topic=test
-metric.asn.topic=test
-metric.idc.renter.topic=test
-metric.application.topic=test
-metric.domain.topic=test
-metric.http.host.topic=test
-metric.ssl.sni.topic=test
-metric.protocol.topic=test
-metric.link.topic=test
-metric.relation.topic=test
-metric.attribute.topic=test
-
-
-#metric???????
-metric.output.parallelism=1
\ No newline at end of file