diff --git a/src/main/java/com/profiler/server/dao/TerminalStatistics.java b/src/main/java/com/profiler/server/dao/TerminalStatistics.java new file mode 100644 index 000000000..f76459074 --- /dev/null +++ b/src/main/java/com/profiler/server/dao/TerminalStatistics.java @@ -0,0 +1,10 @@ +package com.profiler.server.dao; + +/** + * + * @author netspider + * + */ +public interface TerminalStatistics { + void update(String sourceApplicationName, String destApplicationName, short destServiceType); +} diff --git a/src/main/java/com/profiler/server/dao/Traces.java b/src/main/java/com/profiler/server/dao/Traces.java index 63e904ae9..aef494658 100644 --- a/src/main/java/com/profiler/server/dao/Traces.java +++ b/src/main/java/com/profiler/server/dao/Traces.java @@ -4,4 +4,5 @@ import com.profiler.common.dto.thrift.Span; public interface Traces { void insert(String applicationName, Span span); + void insertTerminalSpan(String applicationName, Span span); } diff --git a/src/main/java/com/profiler/server/dao/hbase/HbaseTerminalStatisticsDao.java b/src/main/java/com/profiler/server/dao/hbase/HbaseTerminalStatisticsDao.java new file mode 100644 index 000000000..550b78da7 --- /dev/null +++ b/src/main/java/com/profiler/server/dao/hbase/HbaseTerminalStatisticsDao.java @@ -0,0 +1,40 @@ +package com.profiler.server.dao.hbase; + +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.util.Bytes; +import org.springframework.beans.factory.annotation.Autowired; + +import com.profiler.common.ServiceType; +import com.profiler.common.hbase.HBaseTables; +import com.profiler.common.hbase.HbaseOperations2; +import com.profiler.common.util.TerminalSpanUtils; +import com.profiler.common.util.TimeSlot; +import com.profiler.server.dao.TerminalStatistics; + +/** + * + * @author netspider + * + */ +public class HbaseTerminalStatisticsDao implements TerminalStatistics { + + @Autowired + private HbaseOperations2 hbaseTemplate; + + @Override + public void update(String sourceApplicationName, String destApplicationName, short destServiceType) { + System.out.println("[WritingTerminalStatistics] " + sourceApplicationName + " -> " + destApplicationName + " (" + ServiceType.parse(destServiceType) + ")"); + + byte[] columnName = TerminalSpanUtils.makeColumnName(destServiceType, destApplicationName); + // TODO 시간을 재는 위치 변경 + long timeSlot = TimeSlot.getSlot(System.currentTimeMillis()); + final byte[] rowKey = TerminalSpanUtils.makeRowKey(sourceApplicationName, timeSlot); + + final Put put = new Put(rowKey); + put.add(HBaseTables.TERMINAL_STATISTICS_CF_COUNTER, columnName, Bytes.toBytes(1)); + + hbaseTemplate.put(HBaseTables.TERMINAL_STATISTICS, put); + + // TODO respose time histogram, agentId list가 추가되어야 함. + } +} diff --git a/src/main/java/com/profiler/server/dao/hbase/HbaseTraceDao.java b/src/main/java/com/profiler/server/dao/hbase/HbaseTraceDao.java index e4675acbb..feadf64a8 100644 --- a/src/main/java/com/profiler/server/dao/hbase/HbaseTraceDao.java +++ b/src/main/java/com/profiler/server/dao/hbase/HbaseTraceDao.java @@ -21,6 +21,7 @@ public class HbaseTraceDao implements Traces { final byte[] COLFAM_SPAN = HBaseTables.TRACES_CF_SPAN; final byte[] COLFAM_ANNOTATION = HBaseTables.TRACES_CF_ANNOTATION; + final byte[] COLFAM_TERMINAL_SPAN = HBaseTables.TRACES_CF_TERMINALSPAN; @Autowired private HbaseOperations2 hbaseTemplate; @@ -43,6 +44,25 @@ public class HbaseTraceDao implements Traces { hbaseTemplate.put(HBaseTables.TRACES, put); } + + @Override + public void insertTerminalSpan(final String applicationName, final Span span) { + SpanBo spanBo = new SpanBo(span); + byte[] value = spanBo.writeValue(); + // TODO 서버 시간으로 변경해댜 될듯 함. + Put put = new Put(SpanUtils.getTraceId(span), spanBo.getStartTime()); + // TODO columName이 중복일 경우를 확인가능하면 span id 중복 발급을 알수 있음. + byte[] spanId = Bytes.toBytes(spanBo.getSpanId()); + put.add(COLFAM_TERMINAL_SPAN, spanId, value); + + List annotations = span.getAnnotations(); + if (annotations.size() != 0) { + byte[] bytes = wrietBuffer(annotations); + put.add(COLFAM_ANNOTATION, spanId, bytes); + } + + hbaseTemplate.put(HBaseTables.TRACES, put); + } private byte[] wrietBuffer(List annotations) { int size = 0; diff --git a/src/main/java/com/profiler/server/handler/SpanHandler.java b/src/main/java/com/profiler/server/handler/SpanHandler.java index afdb504d8..639fe6e05 100644 --- a/src/main/java/com/profiler/server/handler/SpanHandler.java +++ b/src/main/java/com/profiler/server/handler/SpanHandler.java @@ -12,64 +12,81 @@ import com.profiler.common.dto.thrift.Span; import com.profiler.server.dao.AgentIdApplicationIndex; import com.profiler.server.dao.ApplicationTraceIndex; import com.profiler.server.dao.RootTraceIndexDao; +import com.profiler.server.dao.TerminalStatistics; import com.profiler.server.dao.TraceIndex; import com.profiler.server.dao.Traces; public class SpanHandler implements Handler { - private final Logger logger = LoggerFactory.getLogger(SpanHandler.class.getName()); + private final Logger logger = LoggerFactory.getLogger(SpanHandler.class.getName()); - @Autowired - private TraceIndex traceIndexDao; + @Autowired + private TraceIndex traceIndexDao; - @Autowired - private Traces traceDao; + @Autowired + private Traces traceDao; - @Autowired - private RootTraceIndexDao rootTraceIndexDao; + @Autowired + private RootTraceIndexDao rootTraceIndexDao; - @Autowired - private ApplicationTraceIndex applicationTraceIndexDao; + @Autowired + private ApplicationTraceIndex applicationTraceIndexDao; - @Autowired - private AgentIdApplicationIndex agentIdApplicationIndexDao; + @Autowired + private AgentIdApplicationIndex agentIdApplicationIndexDao; + + @Autowired + private TerminalStatistics terminalStatistics; - public void handler(TBase tbase, DatagramPacket datagramPacket) { - assert (tbase instanceof Span); + public void handler(TBase tbase, DatagramPacket datagramPacket) { + assert (tbase instanceof Span); - try { - Span span = (Span) tbase; + try { + Span span = (Span) tbase; - if (logger.isInfoEnabled()) { - logger.info("Received SPAN=" + span); - } + if (logger.isInfoEnabled()) { + logger.info("Received SPAN=" + span); + } - String applicationName = agentIdApplicationIndexDao.selectApplicationName(span.getAgentId()); + String applicationName = agentIdApplicationIndexDao.selectApplicationName(span.getAgentId()); - if (applicationName == null) { - logger.warn("Applicationname '{}' not found. Drop the log.", applicationName); - return; - } else { - logger.info("Applicationname '{}' found. Write the log.", applicationName); - } + if (applicationName == null) { + logger.warn("Applicationname '{}' not found. Drop the log.", applicationName); + return; + } else { + logger.info("Applicationname '{}' found. Write the log.", applicationName); + } - if (logger.isDebugEnabled()) { - logger.debug("Found Applicationname={}", applicationName); - } + if (logger.isDebugEnabled()) { + logger.debug("Found Applicationname={}", applicationName); + } - traceDao.insert(applicationName, span); - if (span.getParentSpanId() == -1) { - rootTraceIndexDao.insert(span); - } - - if (ServiceType.parse(span.getServiceType()).isIndexable()) { + ServiceType serviceType = ServiceType.parse(span.getServiceType()); + + // insert span + if (serviceType.isTerminal()) { + traceDao.insertTerminalSpan(applicationName, span); + + // if terminal update statistics + terminalStatistics.update(applicationName, span.getServiceName(), serviceType.getCode()); + } else { + traceDao.insert(applicationName, span); + } + + // indexing root span + if (span.getParentSpanId() == -1) { + rootTraceIndexDao.insert(span); + } + + // indexing non-terminal span + if (serviceType.isIndexable()) { traceIndexDao.insert(span); applicationTraceIndexDao.insert(applicationName, span); } else { logger.debug("Skip writing index. '{}'", span); } - } catch (Exception e) { - logger.warn("Span handle error " + e.getMessage(), e); - } - } + } catch (Exception e) { + logger.warn("Span handle error " + e.getMessage(), e); + } + } } diff --git a/src/main/resources/applicationContext.xml b/src/main/resources/applicationContext.xml index 8664a31c3..99a3ddad0 100644 --- a/src/main/resources/applicationContext.xml +++ b/src/main/resources/applicationContext.xml @@ -42,6 +42,7 @@ + \ No newline at end of file