保存代码

src/main/java/com/yuandian/dataflow/grpc/CollectPackets.java

@@ -6,6 +6,10 @@
  */
 package com.yuandian.dataflow.grpc;
 
+import java.time.Duration;
+import java.time.Instant;
+
+import com.google.protobuf.Any;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.yuandian.dataflow.proto.CollectPacketsServerGrpc;
 import com.yuandian.dataflow.proto.ApmBaseDataFlowOuterClass.ApmBaseDataFlow;
@@ -17,6 +21,7 @@ import com.yuandian.dataflow.proto.CollectPacketsServerGrpc.CollectPacketsServer
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.netty.shaded.io.netty.handler.codec.compression.CompressionOptions;
 import io.grpc.stub.StreamObserver;
+import lombok.extern.slf4j.Slf4j;
 
 /**
  * description
@@ -24,6 +29,7 @@ import io.grpc.stub.StreamObserver;
  * @author eson
  *2022年6月09日-16:29:17
  */
+@Slf4j
 public class CollectPackets extends CollectPacketsServerImplBase {
 
    
@@ -44,18 +50,34 @@ public class CollectPackets extends CollectPacketsServerImplBase {
     //     MyThing.Response response = blockingStub.sayHi(request);
     //     System.out.println(response.getName());
     // }
-    var managedChannelBuilder = ManagedChannelBuilder.forAddress("localhost", 4444);
+    var managedChannelBuilder = ManagedChannelBuilder.forAddress("localhost", 50051);
     var channel = managedChannelBuilder.usePlaintext().build();
     var stub = CollectPacketsServerGrpc.newBlockingStub(channel);
-    stub.withCompression("snappy");
+    // stub.withCompression("snappy");
       
-    var response =  stub.getPackets(Request.newBuilder().setVersion("v1.1.1").build());
+    var response =  stub.getPackets(Request.newBuilder().setVersion("0.1.0").build());
+     
     while(true){
+
+
+
         var iter = response.next();
+        
+        Instant now = Instant.now();
+
+        var i = 0;
         for(var p : iter.getPacketsList()) {
-            var flow = p.unpack( ApmBaseDataFlow.class);
             
+            if(p.is(ApmBaseDataFlow.class)) {
+                var flow = p.unpack( ApmBaseDataFlow.class);
+                
+                i ++;
+            }
         }
+
+      
+
+        log.info("条数:{}, {}:ms",i,Duration.between(now, Instant.now()).toMillis());
     }
     
 

src/main/java/com/yuandian/dataflow/proto/CollectPacketsServerGrpc.java

@@ -12,7 +12,7 @@ public final class CollectPacketsServerGrpc {
 
   private CollectPacketsServerGrpc() {}
 
-  public static final String SERVICE_NAME = "com.yuandian.dataflow.proto.CollectPacketsServer";
+  public static final String SERVICE_NAME = "dataflow.CollectPacketsServer";
 
   // Static method descriptors that strictly reflect the proto.
   private static volatile io.grpc.MethodDescriptor<com.yuandian.dataflow.proto.Base.Request,

src/main/proto/ApmBaseDataFlow.proto

@@ -1,96 +1,101 @@
 syntax = "proto3";
 
-package com.yuandian.dataflow.proto;
+option java_package = "com.yuandian.dataflow.proto";
+option go_package = "../grpc-gen;grpcgen";
+
+
+package dataflow;
+ 
 
 message ApmBaseDataFlow {
     
-    int32 tableID = 1;
+    int32 table_id = 1;
 
-    int32 probeIf = 2;
+    int32 probe_if = 2;
 	//四元组
 	/**
 	 * 请求端口
 	 */
-	int32 requestPort = 3;
+	int32 request_port = 3;
 	/**
 	 * 响应端口
 	 */
-	int32 responsePort = 4;
+	int32 response_port = 4;
 	/**
 	 * 请求IP
 	 */
-     int32 requestIp = 5;
+     int32 request_ip = 5;
 	/**
-	 * 响应Ip
+	 * 响应ip
 	 */
-     int32 responseIp = 6;
+     int32 response_ip = 6;
 
 	/**
 	 * 源mac
 	 */
-	int64 srcMac = 7;
+	int64 src_mac = 7;
 
 	/**
 	 * 目的mac
 	 */
-	int64 dstMac = 8;
+	int64 dst_mac = 8;
 
 	/**
 	 * 链路编号
 	 */
-     int32 vlanId = 9;
+     int32 vlan_id = 9;
 	
-     int32 tvSec = 10;
+     int32 tv_sec = 10;
 
-     int32 tvUsec = 11;
+     int32 tv_usec = 11;
 	
 	/**
 	 * 开始时间
 	 */
-     int32 startTm = 12;
+     int32 start_tm = 12;
 
 	/**
 	 * 总字节数
 	 */
-     int32 totalBytes = 13;
+     int32 total_bytes = 13;
 
 	/**
 	 * 总包数
 	 */
-     int32 totalPackets = 14;
+     int32 total_packets = 14;
 	/**
 	 * 总丢包数
 	 */
-     int32 totalDropPackets = 15;
+     int32 total_drop_packets = 15;
 	/**
 	 * 重传延时
 	 */
-     int32 retranTimeDelay = 16;
+     int32 retran_time_delay = 16;
 	/**
 	 * 客户端rtt
 	 */
-     int32 clientRtt = 17;
+     int32 client_rtt = 17;
 	
 	/**
 	 * 服务端Rtt
 	 */
-     int32 serverRtt = 18;
+     int32 server_rtt = 18;
 	
 	
 	/**
 	 * 用户响应时间
 	 */
-     int32 userResponseTime = 19;
+     int32 user_response_time = 19;
 	/**
 	 * 服务响应时间
 	 */
-     int32 serverResponseTime = 20;
+     int32 server_response_time = 20;
 
 
 	/**
 	 * tcp回话连接失败数
 	 */
-     int32 conFail = 21;
+     int32 con_fail = 21;
 
 	/**
 	 * 会话重置数
@@ -100,66 +105,66 @@ message ApmBaseDataFlow {
 	/**
 	 * 服务端总字节数
 	 */
-     int32 bytesIn = 23;
+     int32 bytes_in = 23;
 	/**
 	 * 客户端总字节数
 	 */
-     int32 bytesOut = 24;
+     int32 bytes_out = 24;
 	/**
 	 * 探针推送时间
 	 */
-     int32 timeFlag = 25;
+     int32 time_flag = 25;
 	
 	/**
 	 * 结束时间
 	 */
-     int32 endTm = 26;
+     int32 end_tm = 26;
 	
 	/**
 	 * 结束时间微秒
 	 */
-     int32 endTmUsec = 27;
+     int32 end_tm_usec = 27;
 	
 	/**
 	 * 总响应数
 	 */
-     int32	responNum = 28;
+     int32	respon_num = 28;
 	/**
 	 * 客户端零窗口数
 	 */
-     int32	csWindow = 29;
+     int32	cs_window = 29;
 	/**
 	 * 服务端零窗口数
 	 */
-     int32	scWindow = 30;
+     int32	sc_window = 30;
 	/**
 	 * 客户端重置数
 	 */
-     int32	csReset = 31;
+     int32	cs_reset = 31;
 	/**
 	 * 服务端重置数
 	 */
-     int32	scReset = 32;
+     int32	sc_reset = 32;
 	/**
 	 * 客户端重传数
 	 */
-     int32	csRetran = 33;
+     int32	cs_retran = 33;
 	/**
 	 * 服务端重传数
 	 */
-     int32	scRetran = 34;
+     int32	sc_retran = 34;
 	/**
 	 * 会话建立时间
 	 */
-     int32	connSetupTm = 35;
+     int32	conn_setup_tm = 35;
 	/**
 	 * 新建会话数
 	 */
-     int32	newSession = 36;
+     int32	new_session = 36;
 	
-     int32 csAlert = 37;
+     int32 cs_alert = 37;
 	
-     int32 scAlert = 38;
+     int32 sc_alert = 38;
 
 	string protocal = 39;
 }

src/main/proto/AppFlow.proto

@@ -0,0 +1,34 @@
+syntax = "proto3";
+
+option java_package = "com.yuandian.dataflow.proto";
+option go_package = "../grpc-gen;grpcgen";
+
+
+package dataflow;
+
+message AppFlow { 
+    int32 table_id = 1;
+    int32 src_ip = 2;
+    int32 src_port = 3;
+    int32 dst_ip = 4;
+    int32 dst_port = 5;
+    int32 start_tv_sec = 6;
+    int32 start_tv_usec = 7;
+    int32 last_tv_sec = 8;
+    int32 last_tv_usec = 9;
+    int32 end_tv_sec = 10;
+    int32 end_tv_usec = 11;
+    int32 input_packets = 12;
+    int32 output_packets = 13;
+    int32 input_bytes = 14;
+    int32 output_bytes = 15;
+    string protocaol = 16;
+    int32 app_id = 17;
+    int32 app_group_id = 18;
+    int32 probe_if = 19;
+    int32 app_style = 20;
+    int32 time_flag = 21;
+    int32 vlan_id = 22;
+    int32 mpls_lable = 23;
+    int32 tos = 24;
+}

src/main/proto/BacktrackingFlow.proto

@@ -0,0 +1,31 @@
+syntax = "proto3";
+
+option java_package = "com.yuandian.dataflow.proto";
+option go_package = "../grpc-gen;grpcgen";
+
+
+package dataflow;
+
+message BacktrackingFlow {
+
+    int32 table_id = 1;
+    int64 mac_src = 2;          //源MAC  8 
+	int64 mac_dst = 3;          //目的MAC 8
+	int64 ip_src = 4;              //源IP 8
+	int64 ip_dst = 5;              //目的IP 8
+	int32 port_src = 6;           //源端口，如果没有，为-1
+	int32 port_dst = 7;           //目标端口，如果没有，为-1
+	int32 l3_proto = 8;           //第三层协议ID，如果没有，为-1
+	int32 l4_proto = 9;          //第四层协议ID，如果没有，为-1
+	int32 tos = 10;                 //Tos，如果没有，为-1
+	int32 vlan_id = 11;            //vlan ID，如果没有，为-1
+	
+	int64 bytes = 12;	//	8	字节总数
+	int64 packets = 13;	//	8	数据包总数
+	int64 tcp_sp = 14;	//	8	tcp同步包数
+	int64 tcp_scpn = 15;//	8	tcp同步确认包数
+	int64 tcp_srp = 16;	//	8	tcp同步重置包数
+	int32 app_id = 17;	//	4	appID
+	int32 app_group_id = 18;//	4	app组ID
+	int32 mpls_label = 19;//	4	
+}

src/main/proto/Base.proto

@@ -1,12 +1,14 @@
 syntax = "proto3";
 
 import "google/protobuf/any.proto";
-import "ApmBaseDataFlow.proto";
+ 
 
 option java_multiple_files = false; //不要拆分成多个文件
 option java_package = "com.yuandian.dataflow.proto";
 
-package com.yuandian.dataflow.proto;
+option go_package = "../grpc-gen;grpcgen";
+ 
+package dataflow;
 
 message Response {
     int32 code = 1;
@@ -16,8 +18,8 @@ message Response {
 
 message Request {
     int32 type = 1; // 请求类型.  默认 为 1.  暂无意义
-    repeated google.protobuf.Any param = 2;
-    string version = 3; // 请求版本.  区分版本. 非必要时不使用 
+    string version = 2; // 请求版本.  区分版本. 非必要时不使用 
+    repeated google.protobuf.Any param = 3;
 }
 
 service CollectPacketsServer {

src/main/proto/BasicTrafficFlow.proto

@@ -0,0 +1,26 @@
+syntax = "proto3";
+
+option java_package = "com.yuandian.dataflow.proto";
+option go_package = "../grpc-gen;grpcgen";
+
+
+package dataflow;
+
+message BasicTrafficFlow {
+    int32 table_id = 1;
+    int32 cap_port = 2;
+	int32 request_port = 3;
+	int32 response_port = 4;
+	int32 request_ip = 5;
+	int32 response_ip = 6;
+	int32 start_time = 7;
+	int32 total_bytes = 8;
+	int32 total_packets = 9;
+	int32 spackets64 = 10;
+	int32 spackets128 = 11;
+	int32 spackets256 = 12;
+	int32 spackets512 = 13;
+	int32 spackets1024 = 14;
+	int32 spackets = 15;
+	int32 send_time = 16;
+}

src/main/proto/BusinessBodyData.proto

@@ -0,0 +1,33 @@
+syntax = "proto3";
+
+option java_package = "com.yuandian.dataflow.proto";
+option go_package = "../grpc-gen;grpcgen";
+
+
+package dataflow;
+
+message BusinessBodyData {
+    int32 table_id = 1;
+
+    string relvance_data_id = 2;
+	int64 request_ip = 3;
+	int32 request_port = 4;
+	int64 response_ip = 5;
+	int32 response_port = 6;
+	
+	int64 start_tv_sec = 7;//开始时间秒
+	int64 start_tv_usec = 8;//开始时间毫秒
+	int64 end_tv_sec = 9;//结束时间秒
+	int64 end_tv_usec = 10;//结束时间微妙
+	
+	string resource_code = 11;
+	string no_parameter_recognition = 12;
+	string original_recognition = 13;
+	string request_cookie = 14;
+	
+	string request_body_context = 15;
+	string response_body_context = 16;
+	
+	int32 filter_id = 17;
+	string business_detail_mesg = 18;
+}

src/main/proto/BussFlowDb.proto

@@ -0,0 +1,76 @@
+syntax = "proto3";
+
+option java_package = "com.yuandian.dataflow.proto";
+option go_package = "../grpc-gen;grpcgen";
+
+
+package dataflow;
+
+
+message BussFlowDb {
+
+    int32    table_id = 1;
+  
+	// redis资源归并,处理服务资源发现是需要设置识别串(正则表达式)
+	 string redisRegex = 2;
+
+	 string id = 3; //id
+	 int64 msg_len = 4;                 //消息长度
+	 int32 msg_type = 5;                //消息类型
+	 int64 src_mac = 6;
+	 int64 dst_mac = 7;
+	 int32 protocol = 8;                //协议名
+	 string session_serial_number = 9;//会话序列号			
+	 string  buss_type = 10;//业务服务资源编码（C_01）
+	 int64 request_ip = 11;//Web客户端IP
+	 int32 request_port = 12;//Web客户端端口
+	 int64  response_ip = 13;//Web服务器IP
+	 int32 response_port = 14;//Web服务器端口
+	 int64 start_tv_sec = 15;//Web开始时间秒
+	 int64 start_tv_usec = 16;//开始时间毫秒
+	 int64 end_tv_sec = 17;//结束时间秒
+	 int64 end_tv_usec = 18;//结束时间微妙
+	 string disc_resource_ident = 19;    //web：url   mid:apiBody   db：sql
+	 string name = 20;               //web:操作系统   midd：interfaceName  db:db_name
+	//web midd
+	 string session_id = 21;//sessionid
+	 int32 request_msg_length = 22;  //请求报文长度
+	 string request_msg_detail = 23;   //请求报文详情
+	 int32 response_msg_length = 24; //响应报文长度
+	 string response_msg_detail = 25;  //响应报文详情
+	//web段
+	 string reter_url = 26;           
+	 string x_requested_with = 27;
+	 int64 req_method = 28;         //请求方式
+	 string content_type = 29;     //请求类型
+	 string accept = 30;          //jieshou
+	 int32 req_cookie_leng = 31;//请求cookie报文长度
+	 string req_cookie_detail = 32;//请求cookie报文详情
+	 int64 t_intodb_time = 33;
+	 int32 load_or_step = 34;        //0: 页面   1：加载项   2：非web段数据 
+	 string business_detail_mesg = 35;
+	 string bussiness_key_mesg = 36;  //关键字  格式：key=val|key=val....
+	 int32 isUncomplete = 37;    //组包是否完全    0：组包完整  1：不完整
+	 int32 deal_state = 38; // 默认值为1
+	 int32 server_res_code = 39;
+	 int64 server_response_time = 40;
+	 int64 client_translate_time = 41;
+	 string browser = 42;
+	 int64 server_start_tv_sec = 43;
+	 int64 server_start_tv_usec = 44;
+	 int64 server_end_tv_sec = 45;
+	 int64 server_end_tv_usec = 46;
+	 string probe_ip = 47;
+	 int64 probe_if = 48;
+	 int64 server_translate_time = 49;
+	 int64 time_flag = 50;	
+	 string base_code = 51;
+	 string ori_sql = 52;
+	 string reserved = 53;
+	 int64 bytes_in = 54;
+	 int64 bytes_out = 55;
+	 int32 package_in = 56;
+	 int32 package_out = 57;
+	 string  data_id = 58;
+	 int32 filter_id = 59;
+}

src/main/proto/BussFlowExternal.proto

@@ -0,0 +1,42 @@
+syntax = "proto3";
+
+option java_package = "com.yuandian.dataflow.proto";
+option go_package = "../grpc-gen;grpcgen";
+
+
+package dataflow;
+
+message BussFlowExternal {
+    int32  table_id = 1;
+    string id = 2;
+	int64  probe_if = 3;//接口号
+	string channel = 4;//营业厅渠道：前台或者分析服务器给出,渠道标识
+	string system_name = 5;  //外部系统名称
+	string interface_id = 6;// 业务接口编码
+	string session_id = 7;//sessionid
+	string phone_id = 8;//受理手机号码
+	string net_type = 9;
+	string  net_segment = 10;//网段标识(客户-web)
+	int64 request_ip = 11;//Web客户端IP
+	int64 request_port = 12;//Web客户端端口
+	int64  response_ip = 13;//Web服务器IP
+	int64 response_port = 14;//Web服务器端口
+	
+	 int64 start_tv_sec = 15;//Web开始时间秒
+	int64 start_tv_usec = 16;//开始时间微秒
+	int64 end_tv_sec = 17;//结束时间秒
+	 int64 end_tv_usec = 18;//结束时间微妙
+	
+	int32 deal_state = 19;//Web操作成功/失败标识1成功0失败
+	int64 server_translate_time = 20;//Web服务器传输耗时
+	int64 server_response_time = 21;//Web服务器响应时间
+	string begin_url = 22;//url
+	string operating_sytem = 23;//操作系统
+	int32 server_res_code = 24; //Web系统返回码
+	string browser = 25;//浏览器
+	string business_detail_mesg = 26;//要获取的指标
+	int64 insert_time = 27;//插入时间? 没有被使用
+	string business_involve_msg = 28; //要关联的指标
+	int32 is_uncomplete = 29;
+	int64 time_flag = 30;
+}

src/main/proto/BussFlowMidd.proto

@@ -0,0 +1,72 @@
+syntax = "proto3";
+
+option java_package = "com.yuandian.dataflow.proto";
+option go_package = "../grpc-gen;grpcgen";
+
+
+package dataflow;
+
+message BussFlowMidd {
+    int32 table_id = 1;
+    string id = 2; //id
+	int64 src_mac = 3;
+	int64 dst_mac = 4;
+	int32 msg_len = 5;                 //消息长度
+	int32 msg_type = 6;                //消息类型
+	int32 protocol = 7;                //协议名
+	string session_serial_number = 8;//会话序列号			
+	string  buss_type = 9;//业务服务资源编码（C_01）
+	string  net_segment = 10;//网段标识(客户-web)
+	int64 request_ip = 11;//Web客户端IP
+	int32 request_port = 12;//Web客户端端口
+	int64  response_ip = 13;//Web服务器IP
+	int32 response_port = 14;//Web服务器端口
+	int64 start_tv_sec = 15;//Web开始时间秒
+	int64 start_tv_usec = 16;//开始时间毫秒
+	int64 end_tv_sec = 17;//结束时间秒
+	int64 end_tv_usec = 18;//结束时间微妙
+	string disc_resource_ident = 19;    //web：url   mid:apiBody   db：sql
+	string name = 20;               //web:操作系统   midd：interfaceName  db:db_name 
+	//web midd
+	string session_id = 21;//sessionid
+	int32 request_msg_length = 22;  //请求报文长度
+	string request_msg_detail = 23;   //请求报文详情
+	int32 response_msg_length = 24; //响应报文长度
+	string response_msg_detail = 25;  //响应报文详情
+	//web段
+	string reter_url = 26;           
+	string x_requested_with = 27;
+	int64 req_method = 28;         //请求方式
+	string content_type = 29;     //请求类型
+	string accept = 30;           //jieshou
+	int32 req_cookie_leng = 31;//请求cookie报文长度
+	string req_cookie_detail = 32;//请求cookie报文详情
+	int64 t_intodb_time = 33;
+	int32 load_or_step = 34;        //0: 页面   1：加载项   2：非web段数据 
+	string business_detail_mesg = 35;
+	string bussiness_key_mesg = 36;  //关键字  格式：key=val|key=val....
+	int32 is_uncomplete = 37;    //组包是否完全    0：组包完整  1：不完整
+	int32 deal_state = 38; // 默认值1
+	int32 server_res_code = 39;
+	int64 server_response_time = 40;
+	int64 client_translate_time = 41;
+	string browser = 42;
+	int32 server_start_tv_sec = 43;
+	int64 server_start_tv_usec = 44;
+	int32 server_end_tv_sec = 45;
+	int64 server_end_tv_usec = 46;
+	string probe_ip = 47;
+	int32 probe_if = 48;
+	int64 server_translate_time = 49;
+	int64 time_flag = 50;	
+	string channel = 51;
+	string base_code = 52;
+	string ori_api = 53;
+	string remain_data = 54;
+	int64 bytes_in = 55;
+	int64 bytes_out = 56;
+	int32 package_in = 57;
+	int32 package_out = 58;
+	string  data_id = 59;
+	int32 filter_id = 60;
+}

src/main/proto/BussFlowOrl.proto

@@ -0,0 +1,41 @@
+syntax = "proto3";
+
+option java_package = "com.yuandian.dataflow.proto";
+option go_package = "../grpc-gen;grpcgen";
+
+
+package dataflow;
+
+message BussFlowOrl {
+    int32 table_id = 1;
+    int32 msg_type = 2;
+    int32 msg_version = 3;
+    int32 msg_seq = 4;
+    int32 msg_len = 5;
+    int64 request_mac = 6;
+    int64 response_mac = 7;
+    int64 request_ip = 8;
+    int32 request_port = 9;
+    int64 response_ip = 10;
+    int32 response_port = 11;
+    int32 probe_if = 12;
+    int32 protocol = 13;
+    int64 start_tv_sec = 14;//Web开始时间秒
+    int64 start_tv_usec = 15;//开始时间毫秒
+    int64 end_tv_sec = 16;//结束时间秒
+    int64 end_tv_usec = 17;//结束时间微妙
+    int32 req_len = 18;
+    int32 res_len = 19;
+    int32 busi_msg_len = 20;
+    int32 key_msg_len = 21;
+    int32 detail_msg_len = 22;
+    int32 remain_len = 23;
+    string business_code = 24;
+    string session_id = 25;
+    string req_data = 26;
+    string res_data = 27;
+    string busi_msg = 28;
+    string busi_key_msg = 29;
+    string busi_detail_msg = 30;
+    string remain_data = 31;
+}

src/test/java/com/yuandian/dataflow/AppTest.java

@@ -190,4 +190,6 @@ public class AppTest {
 
     }
 
+
+ 
 }