diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/CellToSpanHandler.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/CellToSpanHandler.java new file mode 100644 index 000000000..793b8fd1d --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/CellToSpanHandler.java @@ -0,0 +1,65 @@ +package com.ai.cloud.skywalking.analysis.chainbuild.util; + +import com.ai.cloud.skywalking.protocol.AckSpan; +import com.ai.cloud.skywalking.protocol.FullSpan; +import com.ai.cloud.skywalking.protocol.RequestSpan; +import com.ai.cloud.skywalking.protocol.exception.ConvertFailedException; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.util.Bytes; + +import java.util.HashMap; +import java.util.Map; + +public class CellToSpanHandler { + private Map levelIdToRequestSpans; + private Map levelIdToAckSpans; + + public CellToSpanHandler() { + levelIdToRequestSpans = new HashMap(); + levelIdToAckSpans = new HashMap(); + } + + public void addSpan(Cell cell) throws ConvertFailedException { + if (cell != null && cell.getValueArray().length > 0) { + String colId = + Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()); + + if (colId.endsWith("-ACK")) { + levelIdToAckSpans.put(colId.substring(0, colId.length() - 4), convertACKSpan(CellUtil.cloneValue(cell))); + } else { + levelIdToRequestSpans.put(colId, convertRequestSpan(CellUtil.cloneValue(cell))); + } + } + } + + private RequestSpan convertRequestSpan(byte[] originData) throws ConvertFailedException { + return RequestSpan.convert(originData); + } + + private AckSpan convertACKSpan(byte[] originData) throws ConvertFailedException { + return AckSpan.convert(originData); + } + + public Map handle() { + Map fullSpans = new HashMap(); + for (Map.Entry entry : levelIdToRequestSpans.entrySet()){ + FullSpan traceNodeInfo = new FullSpan(entry.getValue(), levelIdToAckSpans.get(entry.getKey())); + fullSpans.put(entry.getKey(),traceNodeInfo); + } + + return fullSpans; + } + + + public Map getRequestSpans() { + return levelIdToRequestSpans; + } + + public Map getAckSpans() { + return levelIdToAckSpans; + } + + + +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/HBaseUtil.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/HBaseUtil.java index 2dd224932..ae62d520d 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/HBaseUtil.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/HBaseUtil.java @@ -6,13 +6,10 @@ import com.ai.cloud.skywalking.analysis.chainbuild.entity.ChainNodeSpecificMinSu import com.ai.cloud.skywalking.analysis.chainbuild.entity.ChainNodeSpecificMonthSummary; import com.ai.cloud.skywalking.analysis.config.Config; import com.ai.cloud.skywalking.analysis.config.HBaseTableMetaData; -import com.ai.cloud.skywalking.protocol.AckSpan; import com.ai.cloud.skywalking.protocol.FullSpan; -import com.ai.cloud.skywalking.protocol.RequestSpan; import com.ai.cloud.skywalking.protocol.exception.ConvertFailedException; import com.google.gson.Gson; import com.google.gson.reflect.TypeToken; -import com.google.protobuf.InvalidProtocolBufferException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.*; import org.apache.hadoop.hbase.client.*; @@ -22,9 +19,7 @@ import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.ArrayList; -import java.util.HashMap; import java.util.List; -import java.util.Map; public class HBaseUtil { private static Logger logger = LoggerFactory.getLogger(HBaseUtil.class.getName()); @@ -265,24 +260,12 @@ public class HBaseUtil { } public static List fetchTraceSpansFromHBase(Result value) throws ConvertFailedException { - List spanList = new ArrayList(); - Map ackSpans = new HashMap(); + + CellToSpanHandler cellToSpanHandler = new CellToSpanHandler(); for (Cell cell : value.rawCells()) { - String traceLevelId = - Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()); - String[] traceLevelIdArray = traceLevelId.split("-"); - if (traceLevelIdArray.length == 2 && "ACK".equals(traceLevelIdArray[1])) { - ackSpans.put(traceLevelIdArray[0], new AckSpan(cell.getValueArray())); - } else { - spanList.add(new FullSpan(new RequestSpan(cell.getValueArray()))); - } - + cellToSpanHandler.addSpan(cell); } - for (FullSpan fullSpan : spanList) { - fullSpan.addAckSpan(ackSpans.get(fullSpan.getTraceLevelId())); - } - - return spanList; + return new ArrayList<>(cellToSpanHandler.handle().values()); } } diff --git a/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java b/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java index e02466b64..561057e77 100644 --- a/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java +++ b/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java @@ -55,6 +55,8 @@ public class AckSpan extends AbstractDataSerializable { private String applicationId; + private static final AckSpan INSTANCE = new AckSpan(); + public AckSpan(Span spanData) { this.traceId = spanData.getTraceId(); this.parentLevel = spanData.getParentLevel(); @@ -68,26 +70,10 @@ public class AckSpan extends AbstractDataSerializable { this.viewPointId = spanData.getViewPointId(); } - public AckSpan() { + private AckSpan() { } - public AckSpan(byte[] originData) throws ConvertFailedException { - TraceProtocol.AckSpan ackSpanProtocol = null; - try { - ackSpanProtocol = TraceProtocol.AckSpan.parseFrom(originData); - } catch (InvalidProtocolBufferException e) { - throw new ConvertFailedException(e.getMessage(), e); - } - this.setTraceId(ackSpanProtocol.getTraceId()); - this.setParentLevel(ackSpanProtocol.getParentLevel()); - this.setLevelId(ackSpanProtocol.getLevelId()); - this.setCost(ackSpanProtocol.getCost()); - this.viewPointId = ackSpanProtocol.getViewpointId(); - this.setExceptionStack(ackSpanProtocol.getExceptionStack()); - this.setStatusCode((byte) ackSpanProtocol.getStatusCode()); - } - public String getTraceId() { return traceId; } @@ -175,6 +161,10 @@ public class AckSpan extends AbstractDataSerializable { return ackSpan; } + public static AckSpan convert(byte[] data) throws ConvertFailedException { + return (AckSpan) INSTANCE.convertData(data); + } + public boolean isNull() { return false; } diff --git a/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSpan.java b/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSpan.java index cb846d579..87987ee7e 100644 --- a/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSpan.java +++ b/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSpan.java @@ -96,6 +96,9 @@ public class RequestSpan extends AbstractDataSerializable { */ protected String address = ""; + + private static final RequestSpan INSTANCE = new RequestSpan(); + public RequestSpan(Span spanData) { this.traceId = spanData.getTraceId(); this.parentLevel = spanData.getParentLevel(); @@ -105,33 +108,9 @@ public class RequestSpan extends AbstractDataSerializable { this.userId = spanData.getUserId(); } - public RequestSpan() { + private RequestSpan() { } - - public RequestSpan(byte[] originData) throws ConvertFailedException { - TraceProtocol.RequestSpan requestSpanByte = null; - try { - requestSpanByte = TraceProtocol.RequestSpan.parseFrom(originData); - } catch (InvalidProtocolBufferException e) { - throw new ConvertFailedException(e.getMessage(), e); - } - this.setTraceId(requestSpanByte.getTraceId()); - this.setParentLevel(requestSpanByte.getParentLevel()); - this.setLevelId(requestSpanByte.getLevelId()); - this.setApplicationId(requestSpanByte.getApplicationId()); - this.setCallType(requestSpanByte.getCallType()); - this.setSpanType(SpanType.convert(requestSpanByte.getSpanType())); - this.setSpanTypeDesc(requestSpanByte.getSpanTypeDesc()); - this.setStartDate(requestSpanByte.getStartDate()); - this.setUserId(requestSpanByte.getUserId()); - this.setViewPointId(requestSpanByte.getViewPointId()); - this.setBusinessKey(requestSpanByte.getBussinessKey()); - this.setAgentId(requestSpanByte.getAgentId()); - this.setProcessNo(requestSpanByte.getProcessNo()); - this.setAddress(requestSpanByte.getAddress()); - } - private boolean isEntrySpan() { return "0".equals(this.getParentLevel() + this.getLevelId()); } @@ -274,6 +253,10 @@ public class RequestSpan extends AbstractDataSerializable { return requestSpan; } + public static RequestSpan convert(byte[] data) throws ConvertFailedException { + return (RequestSpan)INSTANCE.convertData(data); + } + public void setBusinessKey(String businessKey) { this.businessKey = businessKey; } diff --git a/skywalking-protocol/target/maven-archiver/pom.properties b/skywalking-protocol/target/maven-archiver/pom.properties deleted file mode 100644 index 809c55dbd..000000000 --- a/skywalking-protocol/target/maven-archiver/pom.properties +++ /dev/null @@ -1,5 +0,0 @@ -#Generated by Maven -#Wed Aug 10 18:20:13 CST 2016 -version=1.0-Final -groupId=com.ai.cloud -artifactId=skywalking-protocol diff --git a/skywalking-webui/src/main/java/com/ai/cloud/skywalking/web/dao/impl/SpanDataHandler.java b/skywalking-webui/src/main/java/com/ai/cloud/skywalking/web/dao/impl/CellToSpanHandler.java similarity index 82% rename from skywalking-webui/src/main/java/com/ai/cloud/skywalking/web/dao/impl/SpanDataHandler.java rename to skywalking-webui/src/main/java/com/ai/cloud/skywalking/web/dao/impl/CellToSpanHandler.java index a3f0db0c9..cc6013547 100644 --- a/skywalking-webui/src/main/java/com/ai/cloud/skywalking/web/dao/impl/SpanDataHandler.java +++ b/skywalking-webui/src/main/java/com/ai/cloud/skywalking/web/dao/impl/CellToSpanHandler.java @@ -10,11 +10,11 @@ import org.apache.hadoop.hbase.util.Bytes; import java.util.*; -public class SpanDataHandler { +public class CellToSpanHandler { private Map levelIdToRequestSpans; private Map levelIdToAckSpans; - public SpanDataHandler() { + public CellToSpanHandler() { levelIdToRequestSpans = new HashMap(); levelIdToAckSpans = new HashMap(); } @@ -33,14 +33,14 @@ public class SpanDataHandler { } private RequestSpan convertRequestSpan(byte[] originData) throws ConvertFailedException { - return new RequestSpan(originData); + return RequestSpan.convert(originData); } private AckSpan convertACKSpan(byte[] originData) throws ConvertFailedException { - return new AckSpan(originData); + return AckSpan.convert(originData); } - public Map merge() { + public Map handle() { Map traceNodeInfos = new HashMap(); for (Map.Entry entry : levelIdToRequestSpans.entrySet()){ TraceNodeInfo traceNodeInfo = new TraceNodeInfo(entry.getValue(), levelIdToAckSpans.get(entry.getKey())); @@ -50,4 +50,15 @@ public class SpanDataHandler { return traceNodeInfos; } + + public Map getRequestSpans() { + return levelIdToRequestSpans; + } + + public Map getAckSpans() { + return levelIdToAckSpans; + } + + + } diff --git a/skywalking-webui/src/main/java/com/ai/cloud/skywalking/web/dao/impl/TraceNodeDao.java b/skywalking-webui/src/main/java/com/ai/cloud/skywalking/web/dao/impl/TraceNodeDao.java index dc07f2db6..1d2eeedd2 100644 --- a/skywalking-webui/src/main/java/com/ai/cloud/skywalking/web/dao/impl/TraceNodeDao.java +++ b/skywalking-webui/src/main/java/com/ai/cloud/skywalking/web/dao/impl/TraceNodeDao.java @@ -1,5 +1,6 @@ package com.ai.cloud.skywalking.web.dao.impl; +import com.ai.cloud.skywalking.protocol.RequestSpan; import com.ai.cloud.skywalking.protocol.exception.ConvertFailedException; import com.ai.cloud.skywalking.web.dto.TraceNodeInfo; import com.ai.cloud.skywalking.web.dto.TraceNodesResult; @@ -48,12 +49,12 @@ public class TraceNodeDao implements ITraceNodeDao { Map rpcMap = new HashMap(); TraceNodesResult result = new TraceNodesResult(); if (r.rawCells().length < Constants.MAX_SEARCH_SPAN_SIZE) { - SpanDataHandler spanDataHandler = new SpanDataHandler(); + CellToSpanHandler cellToSpanHandler = new CellToSpanHandler(); for (Cell cell : r.rawCells()) { - spanDataHandler.addSpan(cell); + cellToSpanHandler.addSpan(cell); } - for (Map.Entry entry : spanDataHandler.merge().entrySet()){ + for (Map.Entry entry : cellToSpanHandler.handle().entrySet()){ SortUtil.addCurNodeTreeMapKey(traceLogMap, entry.getKey(), entry.getValue()); } computeRPCInfo(rpcMap, traceLogMap); @@ -65,6 +66,8 @@ public class TraceNodeDao implements ITraceNodeDao { return result; } + + private static final String[] NODES = new String[] {"0","0-ACK","0.0","0.0-ACK"}; @Override @@ -81,13 +84,13 @@ public class TraceNodeDao implements ITraceNodeDao { Map traceLogMap = new HashMap(); Map rpcMap = new HashMap(); - SpanDataHandler spanDataHandler = new SpanDataHandler(); + CellToSpanHandler cellToSpanHandler = new CellToSpanHandler(); for (String node : NODES) { Cell cell = r.getColumnLatestCell("call-chain".getBytes(), node.getBytes()); - spanDataHandler.addSpan(cell); + cellToSpanHandler.addSpan(cell); } - for (Map.Entry entry : spanDataHandler.merge().entrySet()){ + for (Map.Entry entry : cellToSpanHandler.handle().entrySet()){ SortUtil.addCurNodeTreeMapKey(traceLogMap, entry.getKey(), entry.getValue()); } diff --git a/skywalking-webui/src/test/java/com/ai/cloud/skywalking/web/dao/impl/SpanDataHandlerTest.java b/skywalking-webui/src/test/java/com/ai/cloud/skywalking/web/dao/impl/SpanDataHandlerTest.java deleted file mode 100644 index 288a64c29..000000000 --- a/skywalking-webui/src/test/java/com/ai/cloud/skywalking/web/dao/impl/SpanDataHandlerTest.java +++ /dev/null @@ -1,77 +0,0 @@ -package com.ai.cloud.skywalking.web.dao.impl; - -import com.ai.cloud.skywalking.protocol.RequestSpan; -import com.ai.cloud.skywalking.protocol.proto.TraceProtocol; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.*; -import org.apache.hadoop.hbase.client.*; -import org.apache.hadoop.hbase.util.Bytes; -import org.junit.Test; - -import static org.apache.hadoop.hbase.util.Bytes.toBytes; -import static org.junit.Assert.*; - -/** - * Created by xin on 16/8/5. - */ -public class SpanDataHandlerTest { - - private Configuration configuration; - private Connection connection; - - @Test - public void addSpan0() throws Exception { - if (configuration == null) { - configuration = HBaseConfiguration.create(); - configuration.set("hbase.zookeeper.quorum", "swhbaseenv"); - configuration.set("hbase.zookeeper.property.clientPort", "2181"); - } - connection = ConnectionFactory.createConnection(configuration); - - Admin admin = connection.getAdmin(); - if (!admin.tableExists(TableName.valueOf("trace-data"))){ - HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf("trace-data")); - HColumnDescriptor family = new HColumnDescriptor(toBytes("call-chain")); - descriptor.addFamily(family); - admin.createTable(descriptor); - } - - TraceProtocol.RequestSpan requestSpan = TraceProtocol.RequestSpan.newBuilder().setUserId("1") - .setStartDate(System.currentTimeMillis() - 10 * 1000).setViewPointId("test").setAgentId("1") - .setApplicationId("test").setCallType("w").setLevelId(0).setParentLevel("test").setSpanType(1) - .setSpanTypeDesc("test").setTraceId("test").build(); - - - - - Put put = new Put(Bytes.toBytes(requestSpan.getTraceId())); - put.addColumn(Bytes.toBytes("call-chain"), Bytes.toBytes("0"), requestSpan.toByteArray()); - for (byte b : requestSpan.toByteArray()){ - System.out.print(b + " "); - } - Table table = connection.getTable(TableName.valueOf("trace-data")); - table.put(put); - - Get get = new Get(requestSpan.getTraceId().getBytes()); - Result result = table.get(get); - for (Cell cell : result.rawCells()){ - byte[] bytes = CellUtil.cloneValue(cell); - for (byte b : bytes){ - System.out.print(b + " "); - } - TraceProtocol.RequestSpan requestSpan1 = TraceProtocol.RequestSpan.parseFrom(cell.getValueArray()); - } - - - } - - @Test - public void addSpan() throws Exception { - TraceProtocol.RequestSpan requestSpan = - TraceProtocol.RequestSpan.newBuilder().setLevelId(0).setTraceId("test").setCallType("1") - .setParentLevel("").setApplicationId("1").setSpanType(1).setSpanTypeDesc("web").setAgentId("1") - .setViewPointId("tst").setStartDate(System.currentTimeMillis() - 10000).setUserId("1").build(); - new RequestSpan(requestSpan.toByteArray()); - } - -}