test/dataflow
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

成功解析第一个包

Browse Source
This commit is contained in:
eson 2022-06-07 18:23:16 +08:00
parent 6e7a35b76d
commit bbf4a47ed3
26 changed files with 1156 additions and 1245 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
src/main/java/com/yuandian/dataflow/.proto/decode/AppFlow.java
View File

@ -1,101 +0,0 @@
package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
public class AppFlow extends PacketBase {
public static int SIZE = 104;
public long srcIp;
public int srcPort;
public long dstIp;
public int dstPort;
public long startTvSec;
public long startTvUsec;
public long lastTvSec;
public long lastTvUsec;
public long endTvSec;
public long endTvUsec;
public int inputPackets;
public int outputPackets;
public int inputBytes;
public int outputBytes;
public String protocaol;
public int appId;
public int appGroupId;
public int probeIf;
public int appStyle;
public long timeFlag;
public int vlanId;
public int mplsLable;
public int tos;
@Override
public PacketBase Parse(PacketHeader header,ByteBuffer data)
throws Exception {
AppFlow nlf = new AppFlow();
nlf.srcIp = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.srcPort = ByteUtil.getInteger(data, offset, 2);
offset += 2;
nlf.dstIp = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.dstPort = ByteUtil.getInteger(data, offset, 2);
offset += 2;
nlf.startTvSec = ByteUtil.getLong(data, offset, 4);
offset += 4;
nlf.startTvUsec = ByteUtil.getLong(data, offset, 4);
offset += 4;
nlf.lastTvSec = ByteUtil.getLong(data, offset, 4);
offset += 4;
nlf.lastTvUsec = ByteUtil.getLong(data, offset, 4);
offset += 4;
nlf.endTvSec = ByteUtil.getLong(data, offset, 4);
offset += 4;
nlf.endTvUsec = ByteUtil.getLong(data, offset, 4);
offset += 4;
nlf.inputPackets = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.outputPackets = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.inputBytes = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.outputBytes = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.protocaol = ByteUtil.getString(data, offset, 20);
offset += 20;
nlf.appId = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.appGroupId = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.probeIf = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.appStyle = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.timeFlag = ByteUtil.getLong(data, offset, 4);
offset += 4;
nlf.vlanId = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.mplsLable = ByteUtil.getInteger(data, offset, 4);
offset += 4;
nlf.tos = ByteUtil.getInteger(data, offset, 4);
offset += 4;
return nlf;
}
@Override
public int getUnitPacketLength() {
return 0;
}
@Override
public String[] getData() {
return null;
}
@Override
public int getInterfaceNumber() {
return 0;
}
}

98
src/main/java/com/yuandian/dataflow/.proto/decode/BacktrackingFlow.java
View File

@ -1,98 +0,0 @@
package com.yuandian.dataflow.proto.decode;
/**
* @author User
*
*/
public class BacktrackingFlow extends PacketBase{
public static int SIZE = 108;
//tuple StatisTuple10 56 10元组信息
private long macSrc; //源MAC 8
private long macDst; //目的MAC 8
private long ipSrc; //源IP 8
private long ipDst; //目的IP 8
private long portSrc; //源端口，如果没有，为-1
private long portDst; //目标端口，如果没有，为-1
private int l3Proto; //第三层协议ID，如果没有，为-1
private int l4Proto; //第四层协议ID，如果没有，为-1
private int tos; //Tos，如果没有，为-1
private int vlanId; //vlan ID，如果没有，为-1
private long bytes; // 8 字节总数
private long packets; // 8 数据包总数
private long tcpSp; // 8 tcp同步包数
private long tcpScpn;// 8 tcp同步确认包数
private long tcpSrp; // 8 tcp同步重置包数
private long appId; // 4 appID
private long appGroupId;// 4 app组ID
private long mplsLabel;// 4
@Override
public PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception {
BacktrackingFlow backFlow = new BacktrackingFlow();
backFlow.macSrc = ByteUtil.getLong(data, offset, 8); //源MAC 8
offset+=8;
backFlow.macDst = ByteUtil.getLong(data, offset, 8); //目的MAC 8
offset+=8;
backFlow.ipSrc = ByteUtil.getLong(data, offset, 8); //源IP 8
offset+=8;
backFlow.ipDst = ByteUtil.getLong(data, offset, 8); //目的IP 8
offset+=8;
backFlow.portSrc = ByteUtil.getInteger(data, offset, 4); //源端口，如果没有，为-1
offset+=4;
backFlow.portDst = ByteUtil.getInteger(data, offset, 4); //目标端口，如果没有，为-1
offset+=4;
backFlow.l3Proto = ByteUtil.getInteger(data, offset, 4); //第三层协议ID，如果没有，为-1
offset+=4;
backFlow.l4Proto = ByteUtil.getInteger(data, offset, 4); //第四层协议ID，如果没有，为-1
offset+=4;
backFlow.tos = ByteUtil.getInteger(data, offset, 4); //Tos，如果没有，为-1
offset+=4;
backFlow.vlanId = ByteUtil.getInteger(data, offset, 4); //vlan ID，如果没有，为-1
offset+=4;
backFlow.bytes = ByteUtil.getLong(data, offset, 8); // 8 字节总数
offset+=8;
backFlow.packets = ByteUtil.getLong(data, offset, 8); // 8 数据包总数
offset+=8;
backFlow.tcpSp = ByteUtil.getLong(data, offset, 8); // 8 tcp同步包数
offset+=8;
backFlow.tcpScpn = ByteUtil.getLong(data, offset, 8);// 8 tcp同步确认包数
offset+=8;
backFlow.tcpSrp = ByteUtil.getLong(data, offset, 8); // 8 tcp同步重置包数
offset+=8;
backFlow.appId = ByteUtil.getInteger(data, offset, 4); // 4 appID
offset+=4;
backFlow.appGroupId = ByteUtil.getInteger(data, offset, 4);// 4 app组ID
offset+=4;
backFlow.mplsLabel = ByteUtil.getInteger(data, offset, 4);// 4
offset+=4;
return backFlow;
}
@Override
public int getUnitPacketLength() {
// TODO Auto-generated method stub
return 0;
}
@Override
public String[] getData() {
// TODO Auto-generated method stub
return null;
}
@Override
public int getInterfaceNumber() {
// TODO Auto-generated method stub
return 0;
}
}

82
src/main/java/com/yuandian/dataflow/.proto/decode/BasicTrafficFlow.java
View File

@ -1,82 +0,0 @@
package com.yuandian.dataflow.proto.decode;
public class BasicTrafficFlow extends PacketBase {
public static int SIZE = 56;
public long capPort;
public int requestPort;
public int responsePort;
public long requestIp;
public long responseIp;
public long startTime;
public long totalBytes;
public long totalPackets;
public long spackets64;
public long spackets128;
public long spackets256;
public long spackets512;
public long spackets1024;
public long spackets;
public long sendTime;
@Override
public PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception {
BasicTrafficFlow btf = new BasicTrafficFlow();
btf.setPacketHeader(header);
btf.capPort = ByteUtil.getLong(data, offset, 4);
offset += 4;
// btf.requestPort = ByteUtil.getInteger(data, offset, 2);
offset += 2;
// btf.responsePort = ByteUtil.getInteger(data, offset, 2);
offset += 2;
// btf.requestIp = ByteUtil.getLong(data, offset, 4);
offset += 4;
// btf.responseIp = ByteUtil.getLong(data, offset, 4);
offset += 4;
btf.startTime = ByteUtil.getLong(data, offset, 4);
offset += 4;
btf.totalBytes = ByteUtil.getLong(data, offset, 4);
offset += 4;
btf.totalPackets = ByteUtil.getLong(data, offset, 4);
offset += 4;
btf.spackets64 = ByteUtil.getLong(data, offset, 4);
offset += 4;
btf.spackets128 = ByteUtil.getLong(data, offset, 4);
offset += 4;
btf.spackets256 = ByteUtil.getLong(data, offset, 4);
offset += 4;
btf.spackets512 = ByteUtil.getLong(data, offset, 4);
offset += 4;
btf.spackets1024 = ByteUtil.getLong(data, offset, 4);
offset += 4;
btf.spackets = ByteUtil.getLong(data, offset, 4);
offset += 4;
btf.sendTime = ByteUtil.getLong(data, offset, 4);
offset += 4;
return btf;
}
@Override
public String[] getData() {
// TODO Auto-generated method stub
return null;
}
@Override
public int getInterfaceNumber() {
// TODO Auto-generated method stub
return 0;
}
@Override
public int getUnitPacketLength() {
// TODO Auto-generated method stub
return 0;
}
}

111
src/main/java/com/yuandian/dataflow/.proto/decode/BussFlowOrl.java
View File

@ -1,111 +0,0 @@
package com.yuandian.dataflow.proto.decode;
public class BussFlowOrl extends PacketBase {
public int msg_type;
public int msg_version;
public int msg_seq;
public int msg_len;
public long request_mac;
public long response_mac;
public long request_ip;
public int request_port;
public long response_ip;
public int response_port;
public int probeif;
public int protocol;
public long start_tv_sec;//Web开始时间秒
public long start_tv_usec;//开始时间毫秒
public long end_tv_sec;//结束时间秒
public long end_tv_usec;//结束时间微妙
public int req_len;
public int res_len;
public int busi_msg_len;
public int key_msg_len;
public int detail_msg_len;
public int remain_len;
public String business_code;
public String sessionid;
public String req_data;
public String res_data;
public String busi_msg;
public String busi_key_msg;
public String busi_detail_msg;
public String remain_data;
@Override
public PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception {
BussFlowOrl orl = new BussFlowOrl();
orl.msg_type = header.getNow_type();
orl.request_mac = ByteUtil.getLong(data, offset, 8);
offset += 8;
orl.response_mac = ByteUtil.getLong(data, offset, 8);
offset += 8;
orl.request_ip = ByteUtil.getLong(data, offset, 4);
offset += 4;
orl.request_port = ByteUtil.getInteger(data, offset, 4);
offset += 4;
orl.response_ip = ByteUtil.getLong(data, offset, 4);
offset += 4;
orl.response_port = ByteUtil.getInteger(data, offset, 4);
offset += 4;
orl.probeif = ByteUtil.getInteger(data, offset, 4);
offset += 4;
orl.protocol = ByteUtil.getInteger(data, offset, 4);
offset += 4;
orl.start_tv_sec = ByteUtil.getLong(data, offset, 4);
offset += 4;
orl.start_tv_usec = ByteUtil.getLong(data, offset, 4);
offset += 4;
orl.end_tv_sec = ByteUtil.getLong(data, offset, 4);
offset += 4;
orl.end_tv_usec = ByteUtil.getLong(data, offset, 4);
offset += 4;
int req_len = ByteUtil.getInteger(data, offset, 4);
offset += 4;
int res_len = ByteUtil.getInteger(data, offset, 4);
offset += 4;
int busi_msg_len = ByteUtil.getInteger(data, offset, 4);
offset += 4;
int key_msg_len = ByteUtil.getInteger(data, offset, 4);
offset += 4;
int detail_msg_len = ByteUtil.getInteger(data, offset, 4);
offset += 4;
int remain_len = ByteUtil.getInteger(data, offset, 4);
offset += 4;
orl.business_code = ByteUtil.getString(data, offset, 32);
offset += 32;
orl.sessionid = ByteUtil.getString(data, offset, 80);
offset += 80;
orl.req_data = ByteUtil.getString(data, offset, req_len);
offset += req_len;
orl.res_data = ByteUtil.getString(data, offset, res_len);
offset += res_len;
orl.busi_msg = ByteUtil.getString(data, offset, busi_msg_len);
offset += busi_msg_len;
orl.busi_key_msg = ByteUtil.getString(data, offset, key_msg_len);
offset += key_msg_len;
orl.busi_detail_msg = ByteUtil.getString(data, offset, detail_msg_len);
offset += detail_msg_len;
orl.remain_data = ByteUtil.getString(data, offset, remain_len);
offset += remain_len;
return orl;
}
@Override
public int getUnitPacketLength() {
return 0;
}
@Override
public String[] getData() {
return new String[0];
}
@Override
public int getInterfaceNumber() {
return 0;
}
}

160
src/main/java/com/yuandian/dataflow/.proto/decode/DataFlow.java
View File

@ -1,160 +0,0 @@
package com.yuandian.dataflow.proto.decode;
public class DataFlow extends PacketBase {
private final int UNIT_DATA_LENGTH = 157;//贵阳
public String id;
public Integer msg_type; //消息类型
public Integer msg_version; //消息版本
public Integer msg_seq; //序列号
public Integer msg_len; //消息长度
public Integer probe_if; //接口号
public Long timestamp; //时间戳
public Long mac_src; //源物理地址
public Long mac_dst; //目的物理地址
public Integer vlan_id; //vlan_id
public Long l3_proto; //l3层协议
public Long l4_proto; //l4层协议
public Integer tos; //tos
public Integer retran_count; //重传次数
public Integer reset_count; //重置次数
public Integer zerowin_count; //零窗口次数
public Integer protocol; //协议名
public Long seq;
public Long ack;
public Integer recog_status; //识别类型标识
public Long bytes; //总字节
public Long packets; //总包数
public Integer start_tv_sec;//Web开始时间秒
public Long start_tv_usec;//开始时间毫秒
public Integer end_tv_sec;//结束时间秒
public Long end_tv_usec;//结束时间微妙
public Integer server_start_tv_sec;//服务器响应开始时间秒
public Long server_start_tv_usec;//服务器响应开始时间毫秒
public Integer server_end_tv_sec;//服务器响应结束时间秒
public Long server_end_tv_usec;//服务器响应结束时间微妙
public Long server_response_time;//Web服务器响应时间
public Long client_translate_time;//Web客户端传输耗时
public Long server_translate_time;//Web服务器传输耗时
public Long bytes_in;
public Long bytes_out;
public Long packets_in;
public Long packets_out;
public Long ip_src; //源IP
public Long ip_dst; //目的IP
public Long port_src; //源端口
public Long port_dst; //目的端口
public Long probeIP; //探针IP
public Long intodb_time;
public Long count = 1L;
@Override
public PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception {
if (offset + UNIT_DATA_LENGTH > data.length)
throw new Exception("data length error!");
DataFlow dflow = new DataFlow();
// dflow.m_Header = header;
dflow.msg_type = header.getTableID();
// offset += 4;
// dflow.msg_version = ByteUtil.getInteger(data, offset, 1);
// offset += 1;
// dflow.msg_seq = ByteUtil.getInteger(data, offset, 4);
// offset += 4;
// dflow.msg_len = ByteUtil.getInteger(data, offset, 4);
// offset += 4;
dflow.ip_src = ByteUtil.getLong(data, offset, 4);
offset += 4;
dflow.ip_dst = ByteUtil.getLong(data, offset, 4);
offset += 4;
// dflow.port_src = ByteUtil.getLong(data, offset, 2);
offset += 2;
// dflow.port_dst = ByteUtil.getLong(data, offset, 2);
offset += 2;
dflow.probe_if = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// dflow.timestamp = ByteUtil.getLong(data, offset, 4);
offset += 4;
// dflow.mac_src = ByteUtil.getLong(data, offset, 8);
offset += 8;
// dflow.mac_dst = ByteUtil.getLong(data, offset, 8);
offset += 8;
// dflow.vlan_id = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// dflow.l3_proto = ByteUtil.getLong(data, offset, 4);
offset += 4;
// dflow.l4_proto = ByteUtil.getLong(data, offset, 4);
offset += 4;
// dflow.tos = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// dflow.retran_count = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// dflow.reset_count = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// dflow.zerowin_count = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// dflow.protocol=ByteUtil.getInteger(data, offset, 4);
offset += 4;
// dflow.seq = ByteUtil.getLong(data, offset, 4);
offset += 4;
// dflow.ack = ByteUtil.getLong(data, offset, 4);
offset += 4;
// dflow.recog_status = ByteUtil.getInteger(data, offset, 4);
offset += 4;
dflow.bytes_in = ByteUtil.getLong(data, offset, 4);
offset += 4;
dflow.bytes_out = ByteUtil.getLong(data, offset, 4);
offset += 4;
dflow.packets_in = ByteUtil.getLong(data, offset, 4);
offset += 4;
dflow.packets_out = ByteUtil.getLong(data, offset, 4);
offset += 4;
dflow.bytes = ByteUtil.getLong(data, offset, 4);
offset += 4;
dflow.packets = ByteUtil.getLong(data, offset, 4);
offset += 4;
// dflow.start_tv_sec = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// dflow.start_tv_usec = ByteUtil.getLong(data, offset, 4);
offset += 4;
// dflow.end_tv_sec = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// dflow.end_tv_usec = ByteUtil.getLong(data, offset, 4);
offset += 4;
// dflow.server_start_tv_sec = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// dflow.server_start_tv_usec = ByteUtil.getLong(data, offset, 4);
offset += 4;
// dflow.server_end_tv_sec = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// dflow.server_end_tv_usec = ByteUtil.getLong(data, offset, 4);
offset += 4;
// dflow.server_translate_time = ByteUtil.getLong(data, offset, 8);
offset += 8;
// dflow.client_translate_time = ByteUtil.getLong(data, offset, 8);
offset += 8;
// dflow.server_response_time = ByteUtil.getLong(data, offset, 8);
offset += 8;
return dflow;
}
@Override
public int getUnitPacketLength() {
return UNIT_DATA_LENGTH;
}
@Override
public String[] getData() {
String data[] = new String[45];
return data;
}
@Override
public int getInterfaceNumber() {
return Integer.parseInt(probe_if.toString());
}
}

27
src/main/java/com/yuandian/dataflow/.proto/decode/PacketBase.java
View File

@ -1,27 +0,0 @@
package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
public abstract class PacketBase {
protected PacketHeader m_Header = null;
public PacketHeader getPacketHead(){
return m_Header;
}
public void setPacketHeader(PacketHeader header)
{
this.m_Header = header;
}
//抽象方法，每个数据操作类实现不一样
public abstract PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception;
public abstract int getUnitPacketLength();
public abstract String[] getData();
public abstract int getInterfaceNumber();
}

55
src/main/java/com/yuandian/dataflow/.proto/decode/PacketHeader.java
View File

@ -1,55 +0,0 @@
package com.yuandian.dataflow.proto.decode;
public class PacketHeader {
private int m_TableID = -1; //数据类型
private int m_RecCount = 0; //数据总条数
private int msg_len; //数据报文总长度，60010端口为压缩后长度
private long timestamp; //60010端口发送数据时间
private int probe_if; //60010端口抓包口
private int umcompr_len;//60010端口数据报文压缩前长度
private int now_type; //记录25类型数据数据当前数据类型 22 23 24
public int getTableID() {
return m_TableID;
}
public int getMsg_len() {
return msg_len;
}
public int getRecCount() {
return m_RecCount;
}
public long getTimestamp() {
return timestamp ;
}
public int getProbe_id(){
return probe_if;
}
public int getUmcompr_len(){
return umcompr_len;
}
public PacketHeader(byte[] data) throws Exception {
if (data.length != 12 && data.length != 20) throw new Exception("Packet header length error!");
this.m_TableID = ByteUtil.getInteger(data, 0, 4);//22
this.m_RecCount = ByteUtil.getInteger(data, 4, 4);//4000
this.msg_len = ByteUtil.getInteger(data, 8, 4);
}
public void parseNextHeader_60010(byte[] data) throws Exception {
if (data.length != 12 && data.length != 20) throw new Exception("Packet header length error!");
this.timestamp = ByteUtil.getLong(data, 0, 4);
this.probe_if = ByteUtil.getInteger(data, 4, 4);
this.umcompr_len = ByteUtil.getInteger(data, 8, 4);
}
public void setNow_type(int now_type) {
this.now_type = now_type;
}
public int getNow_type(){
return now_type;
}
}

1
src/main/java/com/yuandian/dataflow/Server.java
View File

@ -76,7 +76,6 @@ public class Server {
String groupId = "jraft"; String groupId = "jraft";
Configuration conf = JRaftUtils.getConfiguration("localhost:4440,localhost:4441,localhost:4442"); Configuration conf = JRaftUtils.getConfiguration("localhost:4440,localhost:4441,localhost:4442");
PeerId serverId = JRaftUtils.getPeerId(peeridstr); PeerId serverId = JRaftUtils.getPeerId(peeridstr);

27
src/main/java/com/yuandian/dataflow/master/Header.java
View File

@ -13,7 +13,8 @@ import java.nio.ByteBuffer;
import java.nio.ByteOrder; import java.nio.ByteOrder;
import java.util.ArrayList; import java.util.ArrayList;
import com.yuandian.dataflow.utils.PacketHeader; import com.yuandian.dataflow.proto.decode.PacketBase;
import com.yuandian.dataflow.proto.decode.PacketHeader;
import lombok.Cleanup; import lombok.Cleanup;
import lombok.Getter; import lombok.Getter;
@ -28,12 +29,7 @@ import lombok.extern.slf4j.Slf4j;
public class Header { public class Header {
public static void main(String[] args) throws Exception { public static void main(String[] args) throws Exception {
ArrayList<Thread> threads = new ArrayList<Thread>() ;
for(int i = 0 ; i < 100 ; i++) {
Thread thread = new Thread(() -> {
try {
var addr = new InetSocketAddress("192.168.1.248", 60001); var addr = new InetSocketAddress("192.168.1.248", 60001);
@Cleanup @Cleanup
var sock = new Socket(); var sock = new Socket();
@ -45,28 +41,11 @@ public class Header {
var out = new DataOutputStream(sock.getOutputStream()); var out = new DataOutputStream(sock.getOutputStream());
// 发送验证字符串 // 发送验证字符串
out.write("public".getBytes()); out.write("public".getBytes());
// var buf = ByteBuffer.wrap( in.readNBytes(4)).order(ByteOrder.LITTLE_ENDIAN);
log.error("{}", PacketHeader.PacketCode(in)); log.error("{}", PacketHeader.PacketCode(in));
var pheader = new PacketHeader(in); var pheader = new PacketHeader(in);
// buf = ByteBuffer.wrap(in.readNBytes(12)).order(ByteOrder.LITTLE_ENDIAN);
log.error("{}", pheader); log.error("{}", pheader);
} catch (Exception e) { log.error("{}",PacketBase.createPacketBase(pheader));
e.printStackTrace();
}
}) ;
threads.add(thread);
thread.start();
}
threads.forEach((t)->{
try {
t.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
} }
} }

6
src/main/java/com/yuandian/dataflow/.proto/decode/ApmBaseDataFlow.java → src/main/java/com/yuandian/dataflow/proto/decode/ApmBaseDataFlow.java
View File

@ -1,7 +1,9 @@
package com.yuandian.dataflow.proto.decode; package com.yuandian.dataflow.proto.decode;
import java.io.BufferedReader;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -178,6 +180,7 @@ public class ApmBaseDataFlow extends PacketBase {
@Override @Override
public PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception { public PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception {
ApmBaseDataFlow oneData = new ApmBaseDataFlow(); ApmBaseDataFlow oneData = new ApmBaseDataFlow();
oneData.setPacketHeader(header); oneData.setPacketHeader(header);
oneData.probeIf = data.getLong(4); oneData.probeIf = data.getLong(4);
@ -263,8 +266,7 @@ public class ApmBaseDataFlow extends PacketBase {
oneData.scAlert = data.getLong(4); oneData.scAlert = data.getLong(4);
oneData.protocal = data.slice().toString() ;
oneData.protocal = new String(data.array());
return oneData; return oneData;
} }

100
src/main/java/com/yuandian/dataflow/proto/decode/AppFlow.java Normal file
View File

@ -0,0 +1,100 @@
package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
public class AppFlow extends PacketBase {
public static int SIZE = 104;
public long srcIp;
public int srcPort;
public long dstIp;
public int dstPort;
public long startTvSec;
public long startTvUsec;
public long lastTvSec;
public long lastTvUsec;
public long endTvSec;
public long endTvUsec;
public int inputPackets;
public int outputPackets;
public int inputBytes;
public int outputBytes;
public String protocaol;
public int appId;
public int appGroupId;
public int probeIf;
public int appStyle;
public long timeFlag;
public int vlanId;
public int mplsLable;
public int tos;
@Override
public PacketBase Parse(PacketHeader header,ByteBuffer data)
throws Exception {
AppFlow nlf = new AppFlow();
nlf.srcIp = data.getInt(4);
nlf.srcPort = data.getInt(2);
nlf.dstIp = data.getInt(4);
nlf.dstPort = data.getInt(2);
nlf.startTvSec = data.getLong(4);
nlf.startTvUsec = data.getLong(4);
nlf.lastTvSec = data.getLong(4);
nlf.lastTvUsec = data.getLong(4);
nlf.endTvSec = data.getLong(4);
nlf.endTvUsec = data.getLong(4);
nlf.inputPackets = data.getInt(4);
nlf.outputPackets = data.getInt(4);
nlf.inputBytes = data.getInt(4);
nlf.outputBytes = data.getInt(4);
nlf.protocaol = data.alignedSlice(20).toString();
nlf.appId = data.getInt(4);
nlf.appGroupId = data.getInt(4);
nlf.probeIf = data.getInt(4);
nlf.appStyle = data.getInt(4);
nlf.timeFlag = data.getLong(4);
nlf.vlanId = data.getInt(4);
nlf.mplsLable = data.getInt(4);
nlf.tos = data.getInt(4);
return nlf;
}
@Override
public int getUnitPacketLength() {
return 0;
}
@Override
public String[] getData() {
return null;
}
@Override
public int getInterfaceNumber() {
return 0;
}
}

98
src/main/java/com/yuandian/dataflow/proto/decode/BacktrackingFlow.java Normal file
View File

@ -0,0 +1,98 @@
package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
/**
* @author User
*
*/
public class BacktrackingFlow extends PacketBase{
public static int SIZE = 108;
//tuple StatisTuple10 56 10元组信息
private long macSrc; //源MAC 8
private long macDst; //目的MAC 8
private long ipSrc; //源IP 8
private long ipDst; //目的IP 8
private long portSrc; //源端口，如果没有，为-1
private long portDst; //目标端口，如果没有，为-1
private int l3Proto; //第三层协议ID，如果没有，为-1
private int l4Proto; //第四层协议ID，如果没有，为-1
private int tos; //Tos，如果没有，为-1
private int vlanId; //vlan ID，如果没有，为-1
private long bytes; // 8 字节总数
private long packets; // 8 数据包总数
private long tcpSp; // 8 tcp同步包数
private long tcpScpn;// 8 tcp同步确认包数
private long tcpSrp; // 8 tcp同步重置包数
private long appId; // 4 appID
private long appGroupId;// 4 app组ID
private long mplsLabel;// 4
@Override
public PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception {
BacktrackingFlow backFlow = new BacktrackingFlow();
backFlow.macSrc = data.getLong(8); //源MAC 8
backFlow.macDst = data.getLong(8); //目的MAC 8
backFlow.ipSrc = data.getLong(8); //源IP 8
backFlow.ipDst = data.getLong(8); //目的IP 8
backFlow.portSrc = data.getInt(4); //源端口，如果没有，为-1
backFlow.portDst = data.getInt(4); //目标端口，如果没有，为-1
backFlow.l3Proto = data.getInt(4); //第三层协议ID，如果没有，为-1
backFlow.l4Proto = data.getInt(4); //第四层协议ID，如果没有，为-1
backFlow.tos = data.getInt(4); //Tos，如果没有，为-1
backFlow.vlanId = data.getInt(4); //vlan ID，如果没有，为-1
backFlow.bytes = data.getLong(8); // 8 字节总数
backFlow.packets = data.getLong(8); // 8 数据包总数
backFlow.tcpSp = data.getLong(8); // 8 tcp同步包数
backFlow.tcpScpn = data.getLong(8);// 8 tcp同步确认包数
backFlow.tcpSrp = data.getLong(8); // 8 tcp同步重置包数
backFlow.appId = data.getInt(4); // 4 appID
backFlow.appGroupId = data.getInt(4);// 4 app组ID
backFlow.mplsLabel = data.getInt(4);// 4
return backFlow;
}
@Override
public int getUnitPacketLength() {
// TODO Auto-generated method stub
return 0;
}
@Override
public String[] getData() {
// TODO Auto-generated method stub
return null;
}
@Override
public int getInterfaceNumber() {
// TODO Auto-generated method stub
return 0;
}
}

73
src/main/java/com/yuandian/dataflow/proto/decode/BasicTrafficFlow.java Normal file
View File

@ -0,0 +1,73 @@
package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
public class BasicTrafficFlow extends PacketBase {
public static int SIZE = 56;
public long capPort;
public int requestPort;
public int responsePort;
public long requestIp;
public long responseIp;
public long startTime;
public long totalBytes;
public long totalPackets;
public long spackets64;
public long spackets128;
public long spackets256;
public long spackets512;
public long spackets1024;
public long spackets;
public long sendTime;
@Override
public PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception {
BasicTrafficFlow btf = new BasicTrafficFlow();
btf.setPacketHeader(header);
btf.capPort = data.getLong(4);
data.position(data.position() + 12);
btf.startTime = data.getLong(4);
btf.totalBytes = data.getLong(4);
btf.totalPackets = data.getLong(4);
btf.spackets64 = data.getLong(4);
btf.spackets128 = data.getLong(4);
btf.spackets256 = data.getLong(4);
btf.spackets512 = data.getLong(4);
btf.spackets1024 = data.getLong(4);
btf.spackets = data.getLong(4);
btf.sendTime = data.getLong(4);
return btf;
}
@Override
public String[] getData() {
// TODO Auto-generated method stub
return null;
}
@Override
public int getInterfaceNumber() {
// TODO Auto-generated method stub
return 0;
}
@Override
public int getUnitPacketLength() {
// TODO Auto-generated method stub
return 0;
}
}

0
src/main/java/com/yuandian/dataflow/.proto/decode/BusinessBodyData.java → src/main/java/com/yuandian/dataflow/proto/decode/BusinessBodyData.java
View File

203
src/main/java/com/yuandian/dataflow/.proto/decode/BussFlowDb.java → src/main/java/com/yuandian/dataflow/proto/decode/BussFlowDb.java
View File

@ -1,6 +1,8 @@
package com.yuandian.dataflow.proto.decode; package com.yuandian.dataflow.proto.decode;
import com.ud.module.common.SystemConfigManager; import java.nio.ByteBuffer;
// import com.ud.module.common.SystemConfigManager;
@ -82,128 +84,107 @@ public class BussFlowDb extends PacketBase {
// buss.m_Header = header; // buss.m_Header = header;
buss.msg_type = header.getTableID(); buss.msg_type = header.getTableID();
// offset += 4; //
// buss.msg_version = ByteUtil.getInteger(data, offset, 1); // buss.msg_version = data.getInt(1);
// offset += 1; // offset += 1;
// buss.msg_seq = ByteUtil.getInteger(data, offset, 4); // buss.msg_seq = data.getInt(4);
// offset += 4; //
// buss.msg_len = header.getMsg_len(); // buss.msg_len = header.getMsg_len();
// offset += 4; //
// UNIT_DATA_LENGTH=msg_len; // UNIT_DATA_LENGTH=msg_len;
buss.src_mac = ByteUtil.getLong(data, offset, 8); buss.src_mac = data.getLong(8);
offset += 8;
buss.dst_mac = ByteUtil.getLong(data, offset, 8); buss.dst_mac = data.getLong(8);
if(SystemConfigManager.macFilterFlag && buss.src_mac!=0 && buss.dst_mac==0){
boolean macFilterFlag = false;
if(macFilterFlag && buss.src_mac!=0 && buss.dst_mac==0){
return null; return null;
} }
offset += 8;
// buss.vlan_id = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// buss.tos = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// buss.retran_count = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// buss.reset_count = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// buss.zerowin_count = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.protocol=ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.bytes_out = ByteUtil.getLong(data, offset, 4);
offset += 4;
buss.bytes_in = ByteUtil.getLong(data, offset, 4);
offset += 4;
buss.package_out = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.package_in = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.session_serial_number = ByteUtil.getString(data, offset, 24);
offset += 24;
// buss.recog_status = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.probe_if=ByteUtil.getInteger(data, offset,4);
offset += 4;
// buss.channel=ByteUtil.getString(data, offset, 24);
offset+=24;
buss.name = ByteUtil.getString(data, offset, 64);
offset+=64;
buss.request_ip=ByteUtil.getLong(data, offset,4); data.position(data.position() + 28);
offset += 4;
buss.request_port=ByteUtil.getInteger(data, offset,4); buss.protocol= data.getInt(4);
offset += 4;
buss.response_ip=ByteUtil.getLong(data, offset,4); buss.bytes_out = data.getLong(4);
offset += 4;
buss.response_port=ByteUtil.getInteger(data, offset,4); buss.bytes_in = data.getLong(4);
offset += 4;
// buss.deal_state=ByteUtil.getInteger(data, offset,8); buss.package_out = data.getInt(4);
offset += 8;
buss.server_translate_time=ByteUtil.getLong(data, offset,8); buss.package_in = data.getInt(4);
offset += 8;
buss.client_translate_time=ByteUtil.getLong(data, offset,8);
offset += 8; buss.session_serial_number = data.alignedSlice(24).toString();
buss.server_response_time=ByteUtil.getLong(data, offset,8);
offset += 8;
buss.server_start_tv_sec=ByteUtil.getInteger(data, offset,4);
offset +=4; buss.probe_if= data.getInt();
buss.server_start_tv_usec=ByteUtil.getLong(data, offset,4);
offset +=4;
data.position(data.position() + 24);
buss.name = data.alignedSlice(64).toString();
buss.request_ip=data.getLong();
buss.request_port= data.getInt();
buss.response_ip=data.getLong();
buss.response_port= data.getInt();
// buss.deal_state= data.getInt();
buss.server_translate_time=data.getLong();
buss.client_translate_time=data.getLong();
buss.server_response_time=data.getLong();
buss.server_start_tv_sec= data.getInt();
buss.server_start_tv_usec=data.getLong();
//结束时间 //结束时间
buss.server_end_tv_sec=ByteUtil.getInteger(data, offset,4); buss.server_end_tv_sec= data.getInt();
offset +=4;
buss.server_end_tv_usec=ByteUtil.getLong(data, offset,4); buss.server_end_tv_usec=data.getLong();
offset +=4;
buss.start_tv_sec=ByteUtil.getLong(data, offset,4); buss.start_tv_sec=data.getLong();
offset +=4;
buss.start_tv_usec=ByteUtil.getLong(data, offset,4); buss.start_tv_usec=data.getLong();
offset +=4;
//结束时间 //结束时间
buss.end_tv_sec=ByteUtil.getLong(data, offset,4); buss.end_tv_sec=data.getLong();
offset +=4;
buss.end_tv_usec=ByteUtil.getLong(data, offset,4);
offset +=4;
buss.isUncomplete = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.time_flag=ByteUtil.getInteger(data, offset, 4);
offset += 4;
int sqlLen = ByteUtil.getInteger(data, offset, 4); buss.end_tv_usec=data.getLong();
offset += 4;
int detailMsgLen = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.request_msg_length = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.response_msg_length = ByteUtil.getInteger(data, offset, 4);
offset += 4;
int reservedLen = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.ori_sql=ByteUtil.getString(data, offset,sqlLen); buss.isUncomplete = data.getInt(4);
offset += sqlLen;
buss.business_detail_mesg=ByteUtil.getString(data, offset, detailMsgLen);
offset += detailMsgLen;
buss.request_msg_detail = ByteUtil.getStringIgnoreZero(data, offset, buss.request_msg_length);
offset += buss.request_msg_length;
buss.response_msg_detail = ByteUtil.getStringIgnoreZero(data, offset, buss.response_msg_length);
offset += buss.response_msg_length;
buss.reserved = ByteUtil.getString(data, offset, reservedLen);
offset += reservedLen;
/*if(SystemConfigManager.projectName.contains("GDYDBPM")){ buss.time_flag= data.getInt(4);
boolean exist=false; int sqlLen = data.getInt(4);
if(SystemConfigManager.reduceFlag){
redisTemplate = (RedisTemplate<String, String>) ApplicationContextUtil.getBean("redisTemplate"); int detailMsgLen = data.getInt(4);
exist = reduceByKey(buss);
}
//不存在就加入缓存 buss.request_msg_length = data.getInt(4);
if(!exist)return buss;
//存在就返回空 buss.response_msg_length = data.getInt(4);
return null;
}*/ int reservedLen = data.getInt(4);
buss.ori_sql = data.alignedSlice(sqlLen).toString();
buss.business_detail_mesg= data.alignedSlice( detailMsgLen).toString();
;
buss.request_msg_detail = utils.ByteBufferUTF8String(data, buss.request_msg_length);
buss.response_msg_detail = utils.ByteBufferUTF8String(data, buss.response_msg_length);
buss.reserved = data.alignedSlice( reservedLen).toString();
return buss; return buss;
} }
@ -224,6 +205,12 @@ public class BussFlowDb extends PacketBase {
return 0; return 0;
} }
// @Override
// public PacketBase Parse(PacketHeader header, ByteBuffer data) throws Exception {
// // TODO Auto-generated method stub
// return null;
// }
/** /**
* 如果已经存在这个session_id（即key），则说明是重复数据，不再进行后续操作 * 如果已经存在这个session_id（即key），则说明是重复数据，不再进行后续操作
* @param buss * @param buss

117
src/main/java/com/yuandian/dataflow/.proto/decode/BussFlowExternal.java → src/main/java/com/yuandian/dataflow/proto/decode/BussFlowExternal.java
View File

@ -1,5 +1,6 @@
package com.yuandian.dataflow.proto.decode; package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
import java.util.Date; import java.util.Date;
@ -15,69 +16,73 @@ public class BussFlowExternal extends PacketBase {
public BussFlowExternal Parse(PacketHeader header,ByteBuffer data) public BussFlowExternal Parse(PacketHeader header,ByteBuffer data)
throws Exception { throws Exception {
if (offset + UNIT_DATA_LENGTH > data.length)
if (data.position() + UNIT_DATA_LENGTH > data.array().length)
throw new Exception("data length error!"); throw new Exception("data length error!");
BussFlowExternal buss = new BussFlowExternal(); BussFlowExternal buss = new BussFlowExternal();
// buss.m_Header = header; // buss.m_Header = header;
buss.probe_if=ByteUtil.getLong(data, offset,4); buss.probe_if= data.getLong(4);
offset += 4;
buss.channel=ByteUtil.getString(data, offset,24); buss.channel= data.alignedSlice(24).toString();
offset += 24;
buss.interfaceid=ByteUtil.getString(data, offset,100); buss.interfaceid= data.alignedSlice(100).toString();
offset += 100;
buss.systemName=ByteUtil.getString(data, offset,32); buss.systemName= data.alignedSlice(32).toString();
offset += 32;
buss.net_type=ByteUtil.getString(data, offset,6); buss.net_type= data.alignedSlice(6).toString();
offset += 6;
buss.net_segment = ByteUtil.getString(data, offset, 5); buss.net_segment = data.alignedSlice( 5).toString();
offset += 5;
buss.session_id=ByteUtil.getString(data, offset,80); buss.session_id= data.alignedSlice(80).toString();
offset += 80;
buss.phoneid=ByteUtil.getString(data, offset,12); buss.phoneid= data.alignedSlice(12).toString();
offset += 12;
buss.request_ip= data.getLong(4);
buss.request_port= data.getLong(4);
buss.response_ip= data.getLong(4);
buss.response_port= data.getLong(4);
buss.request_ip=ByteUtil.getLong(data, offset,4);
offset += 4;
buss.request_port=ByteUtil.getLong(data, offset,4);
offset += 4;
buss.response_ip=ByteUtil.getLong(data, offset,4);
offset += 4;
buss.response_port=ByteUtil.getLong(data, offset,4);
offset += 4;
//开始时间 //开始时间
buss.start_tv_sec=ByteUtil.getLong(data, offset,4); buss.start_tv_sec= data.getLong(4);
offset +=4;
buss.start_tv_usec=ByteUtil.getLong(data, offset,4); buss.start_tv_usec= data.getLong(4);
offset +=4;
//结束时间 //结束时间
buss.end_tv_sec=ByteUtil.getLong(data, offset,4); buss.end_tv_sec= data.getLong(4);
offset +=4;
buss.end_tv_usec=ByteUtil.getLong(data, offset,4); buss.end_tv_usec= data.getLong(4);
offset +=4;
buss.deal_state= data.getInt(4);
buss.server_translate_time= data.getLong(8);
buss.server_response_time= data.getLong(8);
buss.begin_url= data.alignedSlice(100).toString();
buss.operating_sytem= data.alignedSlice(20).toString();
buss.server_res_code= data.getInt(2);
buss.browser= data.alignedSlice(30).toString();
buss.business_detail_mesg= data.alignedSlice(200).toString();
buss.business_involve_msg = data.alignedSlice(200).toString();
isUncomplete = data.getInt( 4);
buss.time_flag= data.getLong( 4);
buss.deal_state=ByteUtil.getInteger(data, offset,4);
offset += 4;
buss.server_translate_time=ByteUtil.getLong(data, offset,8);
offset += 8;
buss.server_response_time=ByteUtil.getLong(data, offset,8);
offset += 8;
buss.begin_url=ByteUtil.getString(data, offset,100);
offset += 100;
buss.operating_sytem=ByteUtil.getString(data, offset,20);
offset += 20;
buss.server_res_code=ByteUtil.getInteger(data, offset,2);
offset += 2;
buss.browser=ByteUtil.getString(data, offset,30);
offset += 30;
buss.business_detail_mesg=ByteUtil.getString(data, offset,200);
offset += 200;
buss.business_involve_msg = ByteUtil.getString(data, offset,200);
offset += 200;
isUncomplete = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.time_flag=ByteUtil.getLong(data, offset, 4);
offset += 4;
return buss; return buss;
} }

209
src/main/java/com/yuandian/dataflow/.proto/decode/BussFlowMidd.java → src/main/java/com/yuandian/dataflow/proto/decode/BussFlowMidd.java
View File

@ -1,12 +1,11 @@
package com.yuandian.dataflow.proto.decode; package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import com.ud.module.common.ApplicationContextUtil;
import com.ud.module.common.SystemConfigManager;
import com.ud.module.util.code.CodingGeneration;
public class BussFlowMidd extends PacketBase { public class BussFlowMidd extends PacketBase {
private int UNIT_DATA_LENGTH =3569; private int UNIT_DATA_LENGTH =3569;
@ -85,131 +84,115 @@ public class BussFlowMidd extends PacketBase {
// buss.m_Header = header; // buss.m_Header = header;
buss.msg_type = header.getTableID(); buss.msg_type = header.getTableID();
// offset += 4;
// buss.msg_version = ByteUtil.getInteger(data, offset, 1);
// offset += 1;
// buss.msg_seq = ByteUtil.getInteger(data, offset, 4);
// offset += 4;
// buss.msg_len = ByteUtil.getInteger(data, offset, 4);
// offset += 4;
// UNIT_DATA_LENGTH=msg_len; buss.src_mac = data.getLong(8);
buss.src_mac = ByteUtil.getLong(data, offset, 8);
offset += 8; buss.dst_mac = data.getLong(8);
buss.dst_mac = ByteUtil.getLong(data, offset, 8); var macFilterFlag = false ;
if(SystemConfigManager.macFilterFlag && buss.src_mac!=0 && buss.dst_mac==0){ if(macFilterFlag && buss.src_mac!=0 && buss.dst_mac==0){
return null; return null;
} }
offset += 8;
// buss.vlan_id = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// buss.tos = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// buss.retran_count = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// buss.reset_count = ByteUtil.getInteger(data, offset, 4);
offset += 4;
// buss.zerowin_count = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.protocol=ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.bytes_out = ByteUtil.getLong(data, offset, 4);
offset += 4;
buss.bytes_in = ByteUtil.getLong(data, offset, 4);
offset += 4;
buss.package_out = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.package_in = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.session_serial_number = ByteUtil.getString(data, offset, 24);
offset += 24;
// buss.recog_status = ByteUtil.getInteger(data, offset, 4); data.position( data.position() + 28);
offset += 4;
// buss.req_method = ByteUtil.getLong(data, offset, 4); buss.protocol= data.getInt( 4);
// offset += 4;
// buss.content_type = ByteUtil.getString(data, offset, 40); buss.bytes_out = data.getLong(4);
buss.bytes_in = data.getLong(4);
buss.package_out = data.getInt( 4);
buss.package_in = data.getInt( 4);
buss.session_serial_number = data.alignedSlice( 24).toString();
data.position( data.position() + 4);
// buss.req_method = data.getLong(4);
//
// buss.content_type = data.alignedSlice( 40).toString();
// offset += 40; // offset += 40;
// buss.accept = ByteUtil.getString(data, offset, 40); // buss.accept = data.alignedSlice( 40).toString();
// offset += 40; // offset += 40;
buss.probe_if=ByteUtil.getInteger(data, offset,4); buss.probe_if= data.getInt(4);
offset += 4;
// buss.buss_type =ByteUtil.getString(data, offset,10); // buss.buss_type =data.alignedSlice(10).toString();
// offset += 10; // offset += 10;
buss.channel=ByteUtil.getString(data, offset,24); buss.channel=data.alignedSlice(24).toString();
offset += 24;
// buss.session_id=ByteUtil.getString(data, offset,36); // buss.session_id=data.alignedSlice(36).toString();
// offset += 36; // offset += 36;
// buss.net_segment=ByteUtil.getString(data, offset,6); // buss.net_segment=data.alignedSlice(6).toString();
// offset += 6; // offset += 6;
buss.request_ip=ByteUtil.getLong(data, offset,4); buss.request_ip=data.getLong();
offset += 4;
buss.request_port=ByteUtil.getInteger(data, offset,4); buss.request_port= data.getInt(4);
offset += 4;
buss.response_ip=ByteUtil.getLong(data, offset,4); buss.response_ip=data.getLong();
offset += 4;
buss.response_port=ByteUtil.getInteger(data, offset,4); buss.response_port= data.getInt(4);
offset += 4;
//开始时间 //开始时间
buss.start_tv_sec=ByteUtil.getLong(data, offset,4); buss.start_tv_sec=data.getLong();
offset +=4;
buss.start_tv_usec=ByteUtil.getLong(data, offset,4); buss.start_tv_usec=data.getLong();
offset +=4;
//结束时间 //结束时间
buss.end_tv_sec=ByteUtil.getLong(data, offset,4); buss.end_tv_sec=data.getLong();
offset +=4;
buss.end_tv_usec=ByteUtil.getLong(data, offset,4); buss.end_tv_usec=data.getLong();
offset +=4;
// buss.deal_state=ByteUtil.getInteger(data, offset,4); // buss.deal_state= data.getInt(4);
offset +=4;
buss.server_res_code = ByteUtil.getInteger(data, offset, 2); buss.server_res_code = data.getInt( 2);
offset +=2;
buss.server_translate_time=ByteUtil.getLong(data, offset,8); buss.server_translate_time=data.getLong();
offset += 8;
buss.server_start_tv_sec= data.getInt(4);
buss.server_start_tv_usec=data.getLong();
buss.server_start_tv_sec=ByteUtil.getInteger(data, offset,4);
offset +=4;
buss.server_start_tv_usec=ByteUtil.getLong(data, offset,4);
offset +=4;
//结束时间 //结束时间
buss.server_end_tv_sec=ByteUtil.getInteger(data, offset,4); buss.server_end_tv_sec= data.getInt(4);
offset +=4;
buss.server_end_tv_usec=ByteUtil.getLong(data, offset,4);
offset +=4;
buss.server_response_time=ByteUtil.getLong(data, offset,8); buss.server_end_tv_usec=data.getLong();
offset += 8;
buss.client_translate_time=ByteUtil.getLong(data, offset,8);
offset += 8;
buss.isUncomplete = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.time_flag=ByteUtil.getInteger(data, offset, 4);
offset +=4;
int detailMsgLen = ByteUtil.getInteger(data, offset, 4);
offset += 4;
int keyMsgLen = ByteUtil.getInteger(data, offset, 4);
offset += 4;
int apiLen = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.request_msg_length = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.response_msg_length = ByteUtil.getInteger(data, offset, 4);
offset += 4;
int remainLen = ByteUtil.getInteger(data, offset, 4);
offset += 4;
buss.business_detail_mesg=ByteUtil.getString(data, offset,detailMsgLen);
offset += detailMsgLen; buss.server_response_time=data.getLong();
buss.bussiness_key_mesg = ByteUtil.getString(data, offset, keyMsgLen);
offset += keyMsgLen; buss.client_translate_time=data.getLong();
buss.ori_api=ByteUtil.getString(data, offset, apiLen);
offset += apiLen; buss.isUncomplete = data.getInt( 4);
buss.request_msg_detail = ByteUtil.getStringIgnoreZero(data, offset, buss.request_msg_length);
offset += buss.request_msg_length; buss.time_flag= data.getInt( 4);
buss.response_msg_detail = ByteUtil.getStringIgnoreZero(data, offset, buss.response_msg_length);
offset += buss.response_msg_length; int detailMsgLen = data.getInt( 4);
// buss.remain_data = ByteUtil.getString(data, offset, remainLen);
offset += remainLen; int keyMsgLen = data.getInt( 4);
int apiLen = data.getInt( 4);
buss.request_msg_length = data.getInt( 4);
buss.response_msg_length = data.getInt( 4);
int remainLen = data.getInt( 4);
buss.business_detail_mesg=data.alignedSlice(detailMsgLen).toString();
buss.bussiness_key_mesg = data.alignedSlice( keyMsgLen).toString();
buss.ori_api=data.alignedSlice( apiLen).toString();
buss.request_msg_detail = utils.ByteBufferUTF8String(data, buss.request_msg_length);
buss.response_msg_detail = utils.ByteBufferUTF8String(data, buss.response_msg_length);
data.position(data.position()+remainLen);

111
src/main/java/com/yuandian/dataflow/proto/decode/BussFlowOrl.java Normal file
View File

@ -0,0 +1,111 @@
package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
public class BussFlowOrl extends PacketBase {
public int msg_type;
public int msg_version;
public int msg_seq;
public int msg_len;
public long request_mac;
public long response_mac;
public long request_ip;
public int request_port;
public long response_ip;
public int response_port;
public int probeif;
public int protocol;
public long start_tv_sec;//Web开始时间秒
public long start_tv_usec;//开始时间毫秒
public long end_tv_sec;//结束时间秒
public long end_tv_usec;//结束时间微妙
public int req_len;
public int res_len;
public int busi_msg_len;
public int key_msg_len;
public int detail_msg_len;
public int remain_len;
public String business_code;
public String sessionid;
public String req_data;
public String res_data;
public String busi_msg;
public String busi_key_msg;
public String busi_detail_msg;
public String remain_data;
@Override
public PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception {
BussFlowOrl orl = new BussFlowOrl();
orl.msg_type = header.getNowType();
orl.request_mac =data.getLong(8);
orl.response_mac =data.getLong(8);
orl.request_ip =data.getLong(4);
orl.request_port = data.getInt(4);
orl.response_ip =data.getLong(4);
orl.response_port = data.getInt(4);
orl.probeif = data.getInt(4);
orl.protocol = data.getInt(4);
orl.start_tv_sec =data.getLong(4);
orl.start_tv_usec =data.getLong(4);
orl.end_tv_sec =data.getLong(4);
orl.end_tv_usec =data.getLong(4);
int req_len = data.getInt(4);
int res_len = data.getInt(4);
int busi_msg_len = data.getInt(4);
int key_msg_len = data.getInt(4);
int detail_msg_len = data.getInt(4);
int remain_len = data.getInt(4);
orl.business_code = data.alignedSlice( 32).toString();
orl.sessionid = data.alignedSlice( 80).toString();
orl.req_data = data.alignedSlice( req_len).toString();
orl.res_data = data.alignedSlice( res_len).toString();
orl.busi_msg = data.alignedSlice( busi_msg_len).toString();
orl.busi_key_msg = data.alignedSlice( key_msg_len).toString();
orl.busi_detail_msg = data.alignedSlice( detail_msg_len).toString();
orl.remain_data = data.alignedSlice( remain_len).toString();
return orl;
}
@Override
public int getUnitPacketLength() {
return 0;
}
@Override
public String[] getData() {
return new String[0];
}
@Override
public int getInterfaceNumber() {
return 0;
}
}

263
src/main/java/com/yuandian/dataflow/.proto/decode/BussFlowWeb.java → src/main/java/com/yuandian/dataflow/proto/decode/BussFlowWeb.java
View File

@ -1,16 +1,15 @@
package com.yuandian.dataflow.proto.decode; package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
import java.util.regex.Matcher; import java.util.regex.Matcher;
import java.util.regex.Pattern; import java.util.regex.Pattern;
import com.ud.module.common.Constants;
import com.ud.module.common.SystemConfigManager;
import com.yuandian.dataflow.proto.decode.PacketBase; import com.yuandian.dataflow.proto.decode.PacketBase;
import com.yuandian.dataflow.proto.decode.PacketHeader; import com.yuandian.dataflow.proto.decode.PacketHeader;
import com.ud.module.util.flow.JXYDBPMDetailMesgUtils;
import com.ud.module.util.flow.URLUtil;
public class BussFlowWeb extends PacketBase { public class BussFlowWeb extends PacketBase {
@ -31,160 +30,152 @@ public class BussFlowWeb extends PacketBase {
buss.msg_type = header.getTableID(); buss.msg_type = header.getTableID();
// offset += 4; // offset += 4;
// buss.msg_version = ByteUtil.getInteger(data, offset, 1); // buss.msg_version = data.getInt(1);
// offset += 1; // offset += 1;
// buss.msg_seq = ByteUtil.getInteger(data, offset, 4); // buss.msg_seq = data.getInt(4);
// offset += 4; // offset += 4;
// buss.msg_len = ByteUtil.getInteger(data, offset, 4); // buss.msg_len = data.getInt(4);
// offset += 4; // offset += 4;
buss.src_mac = ByteUtil.getLong(data, offset, 8); buss.src_mac = data.getLong(8);
offset += 8;
buss.dst_mac = ByteUtil.getLong(data, offset, 8); buss.dst_mac = data.getLong(8);
offset += 8;
// buss.vlan_id = ByteUtil.getInteger(data, offset, 4);
offset += 4; data.position(data.position()+20);
// buss.tos = ByteUtil.getInteger(data, offset, 4);
offset += 4; buss.protocol=data.getInt(4);
// buss.retran_count = ByteUtil.getInteger(data, offset, 4);
offset += 4; buss.bytes_out = data.getLong(4);
// buss.reset_count = ByteUtil.getInteger(data, offset, 4);
offset += 4; buss.bytes_in = data.getLong(4);
// buss.zerowin_count = ByteUtil.getInteger(data, offset, 4);
offset += 4; buss.package_out = data.getInt(4);
buss.protocol=ByteUtil.getInteger(data, offset, 4);
offset += 4; buss.package_in = data.getInt(4);
buss.bytes_out = ByteUtil.getLong(data, offset, 4);
offset += 4; buss.session_serial_number = data.alignedSlice(24).toString(); //5268109200
buss.bytes_in = ByteUtil.getLong(data, offset, 4);
offset += 4; // buss.recog_status = data.getInt(4);
buss.package_out = ByteUtil.getInteger(data, offset, 4);
offset += 4; data.position(data.position()+4);
buss.package_in = ByteUtil.getInteger(data, offset, 4); buss.req_method = data.getLong(4); //1
offset += 4;
buss.session_serial_number = ByteUtil.getString(data, offset, 24); //5268109200 buss.content_type = data.alignedSlice(40).toString();
offset += 24;
// buss.recog_status = ByteUtil.getInteger(data, offset, 4); buss.accept = data.alignedSlice(40).toString(); //image/png, image/svg+xml, image/*;q=0.8
offset += 4;
buss.req_method = ByteUtil.getLong(data, offset, 4); //1 buss.probe_if=data.getInt();
offset += 4;
buss.content_type = ByteUtil.getString(data, offset, 40); buss.channel=data.alignedSlice(4).toString();
offset += 40;
buss.accept = ByteUtil.getString(data, offset, 40); //image/png, image/svg+xml, image/*;q=0.8 buss.session_id=data.alignedSlice(0).toString(); //7Ip6bYet8n0wfxeJMJtAm7v2FW76ZTImJVyNjbYoRbUdUf71QaxQ!-202316728
offset += 40;
buss.probe_if=ByteUtil.getInteger(data, offset,4); buss.request_ip=data.getLong(); //3232235850
offset += 4;
buss.channel=ByteUtil.getString(data, offset,24); buss.request_port=data.getInt(); //3309
offset += 24;
buss.session_id=ByteUtil.getString(data, offset,80); //7Ip6bYet8n0wfxeJMJtAm7v2FW76ZTImJVyNjbYoRbUdUf71QaxQ!-202316728 buss.response_ip=data.getLong(); //1971882754
offset += 80;
buss.request_ip=ByteUtil.getLong(data, offset,4); //3232235850 buss.response_port=data.getInt(); //80
offset += 4;
buss.request_port=ByteUtil.getInteger(data, offset,4); //3309
offset += 4;
buss.response_ip=ByteUtil.getLong(data, offset,4); //1971882754
offset += 4;
buss.response_port=ByteUtil.getInteger(data, offset,4); //80
offset += 4;
//开始时间 //开始时间
buss.start_tv_sec=ByteUtil.getLong(data, offset,4); //1492473697 buss.start_tv_sec=data.getLong(); //1492473697
offset +=4;
buss.start_tv_usec=ByteUtil.getLong(data, offset,4); //459461 buss.start_tv_usec=data.getLong(); //459461
offset +=4;
//结束时间 //结束时间
buss.end_tv_sec=ByteUtil.getLong(data, offset,4); //1492473697 buss.end_tv_sec=data.getLong(); //1492473697
offset +=4;
buss.end_tv_usec=ByteUtil.getLong(data, offset,4); //459490
offset +=4;
buss.server_translate_time=ByteUtil.getLong(data, offset,8); buss.end_tv_usec=data.getLong(); //459490
offset += 8;
buss.server_translate_time=data.getLong();
buss.server_start_tv_sec=data.getInt();
buss.server_start_tv_usec=data.getLong();
buss.server_start_tv_sec=ByteUtil.getInteger(data, offset,4);
offset +=4;
buss.server_start_tv_usec=ByteUtil.getLong(data, offset,4);
offset +=4;
//结束时间 //结束时间
buss.server_end_tv_sec=ByteUtil.getInteger(data, offset,4); buss.server_end_tv_sec=data.getInt();
offset +=4;
buss.server_end_tv_usec=ByteUtil.getLong(data, offset,4);
offset +=4;
buss.server_response_time=ByteUtil.getLong(data, offset,8); buss.server_end_tv_usec=data.getLong();
offset += 8;
buss.client_translate_time=ByteUtil.getLong(data, offset,8);
offset += 8; buss.server_response_time=data.getLong();
// buss.locate_server_translate_time=ByteUtil.getLong(data, offset,8);
offset += 8; buss.client_translate_time=data.getLong();
// buss.locate_server_response_time=ByteUtil.getLong(data, offset,8);
offset += 8;
// buss.locate_client_translate_time=ByteUtil.getLong(data, offset,8); data.position(data.position()+24);
offset += 8;
buss.x_requested_with = ByteUtil.getString(data, offset, 20); buss.x_requested_with = data.alignedSlice(20).toString();
offset += 20;
buss.operating_system=ByteUtil.getString(data, offset,20); //Windows NT 6.1 buss.operating_system=data.alignedSlice(0).toString(); //Windows NT 6.1
offset += 20;
buss.server_res_code=ByteUtil.getInteger(data, offset,2); buss.server_res_code=data.getInt();
offset += 2;
buss.browser=ByteUtil.getString(data, offset,30); buss.browser=data.alignedSlice(0).toString();
offset += 30;
buss.isUncomplete = ByteUtil.getInteger(data, offset, 4); buss.isUncomplete = data.getInt(4);
offset += 4;
buss.time_flag=ByteUtil.getInteger(data, offset, 4); buss.time_flag=data.getInt(4);
offset += 4;
int detailMsgLen = ByteUtil.getInteger(data, offset, 4); int detailMsgLen = data.getInt(4);
offset += 4;
int keyMsgLen = ByteUtil.getInteger(data, offset, 4); int keyMsgLen = data.getInt(4);
offset += 4;
int reqMsgLeng= ByteUtil.getInteger(data, offset, 4); int reqMsgLeng= data.getInt(4);
buss.request_msg_length =reqMsgLeng; buss.request_msg_length =reqMsgLeng;
offset += 4;
int resMsgLength=ByteUtil.getInteger(data, offset, 4); int resMsgLength=data.getInt(4);
buss.response_msg_length = resMsgLength; buss.response_msg_length = resMsgLength;
offset += 4;
int cookieLength=ByteUtil.getInteger(data, offset, 4); int cookieLength=data.getInt(4);
buss.req_cookie_leng=cookieLength; buss.req_cookie_leng=cookieLength;
offset +=4;
int url_len=ByteUtil.getInteger(data, offset, 4); int url_len=data.getInt(4);
offset +=4;
int refer_len=ByteUtil.getInteger(data, offset, 4); int refer_len=data.getInt(4);
offset += 4;
int remainLen = ByteUtil.getInteger(data, offset, 4); int remainLen = data.getInt(4);
offset += 4;
// if (SystemConfigManager.systemCode.equals("JXYDBPM")) { // if (SystemConfigManager.systemCode.equals("JXYDBPM")) {
// buss.business_detail_mesg=JXYDBPMDetailMesgUtils.decodeValue(ByteUtil.getString(data, offset,detailMsgLen)); // buss.business_detail_mesg=JXYDBPMDetailMesgUtils.decodeValue(data.alignedSlice(etailMsgLen).toString());
// } else { // } else {
// buss.business_detail_mesg=ByteUtil.getString(data, offset,detailMsgLen); // buss.business_detail_mesg=data.alignedSlice(etailMsgLen).toString();
// } // }
buss.business_detail_mesg=ByteUtil.getString(data, offset,detailMsgLen); buss.business_detail_mesg=data.alignedSlice(detailMsgLen).toString();
offset += detailMsgLen;
buss.bussiness_key_mesg = ByteUtil.getString(data, offset,keyMsgLen); buss.bussiness_key_mesg = data.alignedSlice(keyMsgLen).toString();
offset += keyMsgLen;
buss.request_msg_detail = ByteUtil.getStringIgnoreZero(data, offset, reqMsgLeng); buss.request_msg_detail = utils.ByteBufferUTF8String(data, reqMsgLeng);
offset += reqMsgLeng;
buss.response_msg_detail = ByteUtil.getStringIgnoreZero(data, offset,resMsgLength); buss.response_msg_detail = utils.ByteBufferUTF8String(data, resMsgLength);
offset +=resMsgLength;
buss.req_cookie_detail = ByteUtil.getString(data, offset,cookieLength); buss.req_cookie_detail = data.alignedSlice(cookieLength).toString();
offset += cookieLength;
buss.ori_url=ByteUtil.getString(data, offset,url_len); // /images/index/220130.gif buss.ori_url=data.alignedSlice(url_len).toString(); // /images/index/220130.gif
if(buss.ori_url!=null && !"".equals(buss.ori_url.trim())){ if(buss.ori_url!=null && !"".equals(buss.ori_url.trim())){
String result = buss.ori_url.replaceAll("\\?", "").replaceAll("\\/", "").replaceAll("\\*", "").replaceAll("\\=", "").replaceAll("\\&", "").replaceAll("\\+", ""); String result = buss.ori_url.replaceAll("\\?", "").replaceAll("\\/", "").replaceAll("\\*", "").replaceAll("\\=", "").replaceAll("\\&", "").replaceAll("\\+", "");
if(isMessyCode(result)){//如果中文乱码 if(isMessyCode(result)){//如果中文乱码
String withoutValue = URLUtil.getInsertURL(buss.ori_url, Constants.SPLIT_NULL).trim();//去参数值 // String withoutValue = URLUtil.getInsertURL(buss.ori_url, Constants.SPLIT_NULL).trim();//去参数值
String result2 = withoutValue.replaceAll("\\?", "").replaceAll("\\/", "").replaceAll("\\*", "").replaceAll("\\=", "").replaceAll("\\&", "").replaceAll("\\+", "");
if(isMessyCode(result2)){//判断是否还乱码 // String result2 = withoutValue.replaceAll("\\?", "").replaceAll("\\/", "").replaceAll("\\*", "").replaceAll("\\=", "").replaceAll("\\&", "").replaceAll("\\+", "");
buss.ori_url=""; // if(isMessyCode(result2)){//判断是否还乱码
}else{ // buss.ori_url="";
buss.ori_url=withoutValue; // }else{
} // buss.ori_url=withoutValue;
// }
}; };
} }
offset += url_len;
buss.reter_url = ByteUtil.getString(data, offset, refer_len); // http://www.10086.cn/gd/index_200_200.html buss.reter_url = data.alignedSlice(refer_len).toString(); // http://www.10086.cn/gd/index_200_200.html
offset += refer_len;
buss.remain_data = ByteUtil.getString(data, offset, remainLen); buss.remain_data = data.alignedSlice(remainLen).toString();
offset += remainLen;
/*if(SystemConfigManager.projectName.contains("GDYDBPM")){ /*if(SystemConfigManager.projectName.contains("GDYDBPM")){
boolean exist=false; boolean exist=false;
if(SystemConfigManager.reduceFlag){ if(SystemConfigManager.reduceFlag){

78
src/main/java/com/yuandian/dataflow/.proto/decode/DataBaseModel.java → src/main/java/com/yuandian/dataflow/proto/decode/DataBaseModel.java
View File

@ -3,8 +3,11 @@ package com.yuandian.dataflow.proto.decode;
import java.io.Serializable; import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.Date; import java.util.Date;
import io.netty.buffer.ByteBuf;
/** /**
@ -73,53 +76,52 @@ public class DataBaseModel extends PacketBase implements Serializable{
return strb.toString(); return strb.toString();
} }
public DataBaseModel Parse(byte[] data, int offset)throws Exception { public DataBaseModel Parse(ByteBuffer data)throws Exception {
if (offset + 108 > data.length) if (data.position() + 108 > data.array().length)
throw new Exception("data length error!"); throw new Exception("data length error!");
DataBaseModel db = new DataBaseModel(); DataBaseModel db = new DataBaseModel();
db.mac_src = ByteUtil.getLong(data, offset, 8); db.mac_src = data.getLong( 8);
offset+=8;
db.mac_dst = ByteUtil.getLong(data, offset, 8);
offset+=8;
db.ip_src = ByteUtil.getLong(data, offset, 8);
offset+=8;
db.ip_dst = ByteUtil.getLong(data, offset, 8);
offset+=8;
db.port_src = ByteUtil.getInteger(data, offset, 4); db.mac_dst = data.getLong( 8);
offset+=4;
db.port_dst = ByteUtil.getInteger(data, offset, 4); db.ip_src = data.getLong( 8);
offset+=4;
db.l3_proto = ByteUtil.getInteger(data, offset, 4); db.ip_dst = data.getLong( 8);
offset+=4;
db.l4_proto = ByteUtil.getInteger(data, offset, 4); db.port_src = data.getInt(4);
offset+=4;
db.tos = ByteUtil.getInteger(data, offset, 4); db.port_dst = data.getInt(4);
offset+=4;
db.vlan_id = ByteUtil.getInteger(data, offset, 4); db.l3_proto = data.getInt(4);
offset+=4;
db.bytes = ByteUtil.getLong(data, offset, 8); db.l4_proto = data.getInt(4);
offset+=8;
db.packets = ByteUtil.getLong(data, offset, 8); db.tos = data.getInt(4);
offset+=8;
db.packets_syn = ByteUtil.getLong(data, offset, 8); db.vlan_id = data.getInt(4);
db.bytes = data.getLong( 8);
db.packets = data.getLong( 8);
db.packets_syn = data.getLong( 8);
db.packets_syn_ack = data.getLong( 8);
db.packets_syn_rst = data.getLong( 8);
offset+=8;
db.packets_syn_ack = ByteUtil.getLong(data, offset, 8);
offset+=8;
db.packets_syn_rst = ByteUtil.getLong(data, offset, 8);
offset+=8;
/** /**
* 2015-12-10txy 新增3个解析 * 2015-12-10txy 新增3个解析
*/ */
db.appid=ByteUtil.getLong(data, offset, 4); db.appid=data.getLong( 4);
offset+=4;
db.app_group_id=ByteUtil.getLong(data, offset, 4); db.app_group_id=data.getLong( 4);
offset+=4;
db.mpls_label=ByteUtil.getLong(data, offset, 4); db.mpls_label=data.getLong( 4);
offset+=4;
return db; return db;
} }

112
src/main/java/com/yuandian/dataflow/proto/decode/DataFlow.java Normal file
View File

@ -0,0 +1,112 @@
package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
public class DataFlow extends PacketBase {
private final int UNIT_DATA_LENGTH = 157;//贵阳
public String id;
public Integer msg_type; //消息类型
public Integer msg_version; //消息版本
public Integer msg_seq; //序列号
public Integer msg_len; //消息长度
public Integer probe_if; //接口号
public Long timestamp; //时间戳
public Long mac_src; //源物理地址
public Long mac_dst; //目的物理地址
public Integer vlan_id; //vlan_id
public Long l3_proto; //l3层协议
public Long l4_proto; //l4层协议
public Integer tos; //tos
public Integer retran_count; //重传次数
public Integer reset_count; //重置次数
public Integer zerowin_count; //零窗口次数
public Integer protocol; //协议名
public Long seq;
public Long ack;
public Integer recog_status; //识别类型标识
public Long bytes; //总字节
public Long packets; //总包数
public Integer start_tv_sec;//Web开始时间秒
public Long start_tv_usec;//开始时间毫秒
public Integer end_tv_sec;//结束时间秒
public Long end_tv_usec;//结束时间微妙
public Integer server_start_tv_sec;//服务器响应开始时间秒
public Long server_start_tv_usec;//服务器响应开始时间毫秒
public Integer server_end_tv_sec;//服务器响应结束时间秒
public Long server_end_tv_usec;//服务器响应结束时间微妙
public Long server_response_time;//Web服务器响应时间
public Long client_translate_time;//Web客户端传输耗时
public Long server_translate_time;//Web服务器传输耗时
public Long bytes_in;
public Long bytes_out;
public Long packets_in;
public Long packets_out;
public Long ip_src; //源IP
public Long ip_dst; //目的IP
public Long port_src; //源端口
public Long port_dst; //目的端口
public Long probeIP; //探针IP
public Long intodb_time;
public Long count = 1L;
@Override
public PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception {
if (data.position() + UNIT_DATA_LENGTH > data.array().length)
throw new Exception("data length error!");
DataFlow dflow = new DataFlow();
// dflow.m_Header = header;
dflow.msg_type = header.getTableID();
dflow.ip_src = data.getLong(4);
dflow.ip_dst = data.getLong(4);
data.position(data.position()+4);
dflow.probe_if = data.getInt(4);
data.position(data.position() + 32 + 12 + 12 + 8 );
dflow.bytes_in = data.getLong(4);
dflow.bytes_out = data.getLong(4);
dflow.packets_in = data.getLong(4);
dflow.packets_out = data.getLong(4);
dflow.bytes = data.getLong(4);
dflow.packets = data.getLong(4);
data.position(data.position() + 12 + 12 + 16 + 16 );
return dflow;
}
@Override
public int getUnitPacketLength() {
return UNIT_DATA_LENGTH;
}
@Override
public String[] getData() {
String data[] = new String[45];
return data;
}
@Override
public int getInterfaceNumber() {
return Integer.parseInt(probe_if.toString());
}
}

80
src/main/java/com/yuandian/dataflow/proto/decode/PacketBase.java Normal file
View File

@ -0,0 +1,80 @@
package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
import lombok.ToString;
@ToString
public abstract class PacketBase {
protected PacketHeader m_Header = null;
public PacketHeader getPacketHead(){
return m_Header;
}
public void setPacketHeader(PacketHeader header)
{
this.m_Header = header;
}
//抽象方法，每个数据操作类实现不一样
public abstract PacketBase Parse(PacketHeader header,ByteBuffer data) throws Exception;
public abstract int getUnitPacketLength();
public abstract String[] getData();
public abstract int getInterfaceNumber();
public static PacketBase createPacketBase(PacketHeader packetHeader) {
// 获取报文头
// 响应表ID
int table_id = packetHeader.getTableID();
switch (table_id) {
case 17:
AppFlow appFlow = new AppFlow();
appFlow.setPacketHeader(packetHeader);
return appFlow;
case 18:
QoeFlow qoeFlow = new QoeFlow();
qoeFlow.setPacketHeader(packetHeader);
return qoeFlow;
case 20:
SstFlow sstFlow = new SstFlow();
sstFlow.setPacketHeader(packetHeader);
return sstFlow;
case 22:
BussFlowWeb tBusiness = new BussFlowWeb();
tBusiness.setPacketHeader(packetHeader);
return tBusiness;
case 23:
BussFlowMidd secondBusiness = new BussFlowMidd();
secondBusiness.setPacketHeader(packetHeader);
return secondBusiness;
case 24:
BussFlowDb thirdBusiness = new BussFlowDb();
thirdBusiness.setPacketHeader(packetHeader);
return thirdBusiness;
case 25:
BussFlowOrl tExternalBusiness = new BussFlowOrl();
tExternalBusiness.setPacketHeader(packetHeader);
return tExternalBusiness;
case 27:
DataFlow dataFlow = new DataFlow();
dataFlow.setPacketHeader(packetHeader);
return dataFlow;
case 28:
ApmBaseDataFlow apmBaseDataFlow = new ApmBaseDataFlow();
apmBaseDataFlow.setPacketHeader(packetHeader);
return apmBaseDataFlow;
case 29:
BasicTrafficFlow BasicTrafficFlow = new BasicTrafficFlow();
BasicTrafficFlow.setPacketHeader(packetHeader);
return BasicTrafficFlow;
default:
return null;
}
}
}

2
src/main/java/com/yuandian/dataflow/utils/PacketHeader.java → src/main/java/com/yuandian/dataflow/proto/decode/PacketHeader.java
View File

@ -4,7 +4,7 @@
* @author eson * @author eson
*2022年6月07日-11:09:21 *2022年6月07日-11:09:21
*/ */
package com.yuandian.dataflow.utils; package com.yuandian.dataflow.proto.decode;
import java.io.DataInputStream; import java.io.DataInputStream;
import java.io.IOException; import java.io.IOException;

152
src/main/java/com/yuandian/dataflow/.proto/decode/QoeFlow.java → src/main/java/com/yuandian/dataflow/proto/decode/QoeFlow.java
View File

@ -59,82 +59,82 @@ public int vlanId; //uint32_t 4
@Override @Override
public QoeFlow Parse(PacketHeader header,ByteBuffer data) throws Exception { public QoeFlow Parse(PacketHeader header,ByteBuffer data) throws Exception {
QoeFlow qoeFlow = new QoeFlow(); QoeFlow qoeFlow = new QoeFlow();
qoeFlow.srcIp=ByteUtil.getInteger(data, offset, 4); qoeFlow.srcIp=data.getInt( 4);
offset += 4;
qoeFlow.dstIp=ByteUtil.getInteger(data, offset, 4); qoeFlow.dstIp=data.getInt( 4);
offset += 4;
qoeFlow.stvSec=ByteUtil.getInteger(data, offset, 4); qoeFlow.stvSec=data.getInt( 4);
offset += 4;
qoeFlow.stvUsec=ByteUtil.getInteger(data, offset, 4); qoeFlow.stvUsec=data.getInt( 4);
offset += 4;
qoeFlow.ltvSec=ByteUtil.getInteger(data, offset, 4); qoeFlow.ltvSec=data.getInt( 4);
offset += 4;
qoeFlow.ltvUsec=ByteUtil.getInteger(data, offset, 4); qoeFlow.ltvUsec=data.getInt( 4);
offset += 4;
qoeFlow.dst2ResponNum=ByteUtil.getInteger(data, offset, 4); qoeFlow.dst2ResponNum=data.getInt( 4);
offset += 4;
qoeFlow.dst2Fast=ByteUtil.getInteger(data, offset, 4); qoeFlow.dst2Fast=data.getInt( 4);
offset += 4;
qoeFlow.dst2FastExpected=ByteUtil.getInteger(data, offset, 4); qoeFlow.dst2FastExpected=data.getInt( 4);
offset += 4;
qoeFlow.dst2ExpectedDegrated=ByteUtil.getInteger(data, offset, 4); qoeFlow.dst2ExpectedDegrated=data.getInt( 4);
offset += 4;
qoeFlow.dst2DegratedService=ByteUtil.getInteger(data, offset, 4); qoeFlow.dst2DegratedService=data.getInt( 4);
offset += 4;
qoeFlow.dst2ServiceAvailability=ByteUtil.getInteger(data, offset, 4); qoeFlow.dst2ServiceAvailability=data.getInt( 4);
offset += 4;
qoeFlow.dst2ResponTimeout=ByteUtil.getInteger(data, offset, 4); qoeFlow.dst2ResponTimeout=data.getInt( 4);
offset += 4;
qoeFlow.dst2ResponSuccess=ByteUtil.getInteger(data, offset, 4); qoeFlow.dst2ResponSuccess=data.getInt( 4);
offset += 4;
qoeFlow.dst2ResponFail=ByteUtil.getInteger(data, offset, 4); qoeFlow.dst2ResponFail=data.getInt( 4);
offset += 4;
qoeFlow.dst2ResponPeek=ByteUtil.getInteger(data, offset, 4); qoeFlow.dst2ResponPeek=data.getInt( 4);
offset += 4;
qoeFlow.dst2ResponAverage=ByteUtil.getInteger(data, offset, 4); qoeFlow.dst2ResponAverage=data.getInt( 4);
offset += 4;
qoeFlow.csWindow=ByteUtil.getInteger(data, offset, 4); qoeFlow.csWindow=data.getInt( 4);
offset += 4;
qoeFlow.scWindow=ByteUtil.getInteger(data, offset, 4); qoeFlow.scWindow=data.getInt( 4);
offset += 4;
qoeFlow.csReset=ByteUtil.getInteger(data, offset, 4); qoeFlow.csReset=data.getInt( 4);
offset += 4;
qoeFlow.scReset=ByteUtil.getInteger(data, offset, 4); qoeFlow.scReset=data.getInt( 4);
offset += 4;
qoeFlow.csRetran=ByteUtil.getInteger(data, offset, 4); qoeFlow.csRetran=data.getInt( 4);
offset += 4;
qoeFlow.scRetran=ByteUtil.getInteger(data, offset, 4); qoeFlow.scRetran=data.getInt( 4);
offset += 4;
qoeFlow.appId=ByteUtil.getInteger(data, offset, 4); qoeFlow.appId=data.getInt( 4);
offset += 4;
qoeFlow.appGroupId=ByteUtil.getInteger(data, offset, 4); qoeFlow.appGroupId=data.getInt( 4);
offset += 4;
qoeFlow.probeIf=ByteUtil.getInteger(data, offset, 4); qoeFlow.probeIf=data.getInt( 4);
offset += 4;
qoeFlow.appStyle=ByteUtil.getInteger(data, offset, 4); qoeFlow.appStyle=data.getInt( 4);
offset += 4;
qoeFlow.timeFlag=ByteUtil.getInteger(data, offset, 4); qoeFlow.timeFlag=data.getInt( 4);
offset += 4;
qoeFlow.connSetupTm=ByteUtil.getInteger(data, offset, 4); qoeFlow.connSetupTm=data.getInt( 4);
offset += 4;
qoeFlow.dataTransferTm=ByteUtil.getInteger(data, offset, 4); qoeFlow.dataTransferTm=data.getInt( 4);
offset += 4;
qoeFlow.retransDelayTm=ByteUtil.getInteger(data, offset, 4); qoeFlow.retransDelayTm=data.getInt( 4);
offset += 4;
qoeFlow.networkInbound=ByteUtil.getInteger(data, offset, 4); qoeFlow.networkInbound=data.getInt( 4);
offset += 4;
qoeFlow.networkOutbound=ByteUtil.getInteger(data, offset, 4); qoeFlow.networkOutbound=data.getInt( 4);
offset += 4;
qoeFlow.newSession=ByteUtil.getInteger(data, offset, 4); qoeFlow.newSession=data.getInt( 4);
offset += 4;
qoeFlow.userEvents=ByteUtil.getInteger(data, offset, 4); qoeFlow.userEvents=data.getInt( 4);
offset += 4;
qoeFlow.serverEvents=ByteUtil.getInteger(data, offset, 4); qoeFlow.serverEvents=data.getInt( 4);
offset += 4;
qoeFlow.connSetupPeek=ByteUtil.getInteger(data, offset, 4); qoeFlow.connSetupPeek=data.getInt( 4);
offset += 4;
qoeFlow.vlanId=ByteUtil.getInteger(data, offset, 4); qoeFlow.vlanId=data.getInt( 4);
offset += 4;
return qoeFlow; return qoeFlow;
} }
@Override @Override

75
src/main/java/com/yuandian/dataflow/.proto/decode/SstFlow.java → src/main/java/com/yuandian/dataflow/proto/decode/SstFlow.java
View File

@ -1,5 +1,6 @@
package com.yuandian.dataflow.proto.decode; package com.yuandian.dataflow.proto.decode;
import java.nio.ByteBuffer;
import java.util.Date; import java.util.Date;
@ -51,45 +52,45 @@ public class SstFlow extends PacketBase{
public PacketBase Parse(PacketHeader header,ByteBuffer data) public PacketBase Parse(PacketHeader header,ByteBuffer data)
throws Exception { throws Exception {
SstFlow sstFlow = new SstFlow(); SstFlow sstFlow = new SstFlow();
sstFlow.mac_src = ByteUtil.getLong(data, offset, 8); sstFlow.mac_src = data.getLong(8);
offset += 8;
sstFlow.mac_dst = ByteUtil.getLong(data, offset, 8); sstFlow.mac_dst = data.getLong(8);
offset += 8;
sstFlow.ip_src = ByteUtil.getLong(data, offset, 8); sstFlow.ip_src = data.getLong(8);
offset += 8;
sstFlow.ip_dst = ByteUtil.getLong(data, offset, 8); sstFlow.ip_dst = data.getLong(8);
offset += 8;
sstFlow.port_src = ByteUtil.getInteger(data, offset, 4); sstFlow.port_src = data.getInt(4);
offset += 4;
sstFlow.port_dst = ByteUtil.getInteger(data, offset, 4); sstFlow.port_dst = data.getInt(4);
offset += 4;
sstFlow.l3_proto = ByteUtil.getInteger(data, offset, 4); sstFlow.l3_proto = data.getInt(4);
offset += 4;
sstFlow.l4_proto = ByteUtil.getInteger(data, offset, 4); sstFlow.l4_proto = data.getInt(4);
offset += 4;
sstFlow.tos = ByteUtil.getInteger(data, offset, 4); sstFlow.tos = data.getInt(4);
offset += 4;
sstFlow.vlan_id = ByteUtil.getInteger(data, offset, 4); sstFlow.vlan_id = data.getInt(4);
offset += 4;
sstFlow.bytes = data.getLong(8);
sstFlow.packets = data.getLong(8);
sstFlow.packets_syn = data.getLong(8);
sstFlow.packets_syn_ack = data.getLong(8);
sstFlow.packets_syn_rst = data.getLong(8);
sstFlow.appid = data.getInt(4);
sstFlow.app_group_id = data.getInt(4);
sstFlow.mpls_label = data.getInt(4);
sstFlow.bytes = ByteUtil.getLong(data, offset, 8);
offset += 8;
sstFlow.packets = ByteUtil.getLong(data, offset, 8);
offset += 8;
sstFlow.packets_syn = ByteUtil.getLong(data, offset, 8);
offset += 8;
sstFlow.packets_syn_ack = ByteUtil.getLong(data, offset, 8);
offset += 8;
sstFlow.packets_syn_rst = ByteUtil.getLong(data, offset, 8);
offset += 8;
sstFlow.appid = ByteUtil.getInteger(data, offset, 4);
offset += 4;
sstFlow.app_group_id = ByteUtil.getInteger(data, offset, 4);
offset += 4;
sstFlow.mpls_label = ByteUtil.getInteger(data, offset, 4);
offset += 4;
sstFlow.timestamp = header.getTimestamp(); sstFlow.timestamp = header.getTimestamp();
sstFlow.probe_if = header.getProbe_id(); sstFlow.probe_if = header.getProbeIf();
return sstFlow; return sstFlow;
} }

22
src/main/java/com/yuandian/dataflow/proto/decode/utils.java Normal file
View File

@ -0,0 +1,22 @@
/**
* description
*
* @author eson
*2022年6月07日-16:18:34
*/
package com.yuandian.dataflow.proto.decode;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
/**
* description
*
* @author eson
*2022年6月07日-16:18:34
*/
public class utils {
public static String ByteBufferUTF8String(ByteBuffer data, int length) throws UnsupportedEncodingException {
return new String( data.alignedSlice(length).array(), "utf-8");
}
}