home/pinpoint
2
0
Fork 0
mirror of https://github.com/wahyd4/pinpoint.git synced 2026-08-28 06:08:43 +10:00
Code Issues Packages Projects Releases Wiki Activity

[강운덕] [LUCYSUS-1744] root Span만 조회하여 trace 데이터를 구성하도록 수정.

Browse Source 
git-svn-id: http://svn.bds.nhncorp.com/pe/hippo-web/trunk@822 84d0f5b1-2673-498c-a247-62c4ff18d310
This commit is contained in:
Woonduk Kang
2012-10-25 09:46:00 +00:00
parent e722f4667b
commit 067abb052c
11 changed files with 170 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/nhn/hippo/web/controller/FlowChartController.java
+9 -1
View File
@@ -7,6 +7,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.util.StopWatch;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
@@ -47,9 +48,16 @@ public class FlowChartController {
@RequestMapping(value = "/flowserver", method = RequestMethod.GET)
public String flowserver(Model model, @RequestParam("host") String[] hosts, @RequestParam("from") long from, @RequestParam("to") long to) {
String[] agentIds = flow.selectAgentIds(hosts);
// TODO 제거 하거나, interceptor로 할것.
StopWatch watch = new StopWatch();
watch.start("scanTraceindex");
Set<TraceId> traceIds = flow.selectTraceIdsFromTraceIndex(agentIds, from, to);
watch.stop();
logger.info("time:{} {}", watch.getLastTaskTimeMillis(), traceIds.size());
watch.start("selectServerCallTree");
ServerCallTree callTree = flow.selectServerCallTree(traceIds);
watch.stop();
logger.info("time:{}", watch.getLastTaskTimeMillis());
model.addAttribute("nodes", callTree.getNodes());
model.addAttribute("links", callTree.getLinks());
src/main/java/com/nhn/hippo/web/dao/HbaseRootTraceIndexDao.java
+86
View File
@@ -0,0 +1,86 @@
package com.nhn.hippo.web.dao;
import com.profiler.common.hbase.HBaseTables;
import com.profiler.common.hbase.HbaseOperations2;
import com.profiler.common.util.BytesUtils;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.stereotype.Repository;
import java.util.ArrayList;
import java.util.List;
/**
*
*/
@Repository
public class HbaseRootTraceIndexDao implements RootTraceIndexDao {
private Logger logger = LoggerFactory.getLogger(this.getClass());
private final byte[] COLFAM_TRACE = Bytes.toBytes("Trace");
private final byte[] COLNAME_ID = Bytes.toBytes("ID");
@Autowired
private HbaseOperations2 hbaseOperations2;
@Autowired
@Qualifier("traceIndexMapper")
private RowMapper<byte[]> traceIndexMapper;
private int scanCacheSize = 40;
public void setScanCacheSize(int scanCacheSize) {
this.scanCacheSize = scanCacheSize;
}
@Override
public List<byte[]> scanTraceIndex(String agent, long start, long end) {
Scan scan = createScan(agent, start, end);
return hbaseOperations2.find(HBaseTables.ROOT_TRACE_INDEX, scan, traceIndexMapper);
}
@Override
public List<List<byte[]>> multiScanTraceIndex(String[] agents, long start, long end) {
final List<Scan> multiScan = new ArrayList<Scan>(agents.length);
for (String agent : agents) {
Scan scan = createScan(agent, start, end);
multiScan.add(scan);
}
return hbaseOperations2.find(HBaseTables.ROOT_TRACE_INDEX, multiScan, traceIndexMapper);
}
private Scan createScan(String agent, long start, long end) {
Scan scan = new Scan();
// cache size를 지정해야 되는거 같음.??
scan.setCaching(this.scanCacheSize);
byte[] bAgent = Bytes.toBytes(agent);
byte[] bStart = BytesUtils.add(bAgent, start);
scan.setStartRow(bStart);
// TODO 추가 filter를 구현하여 scan시 중복된 값을 제가 할수 있음. 단 server에도 Filter 클래스가 배포되어야 한다.
// scan.setFilter(new ValueFilter());
byte[] bEnd = BytesUtils.add(bAgent, end);
scan.setStopRow(bEnd);
scan.addColumn(COLFAM_TRACE, COLNAME_ID);
scan.setId("rootTraceIndexScan");
// json으로 변화해서 로그를 찍어서. 최초 변환 속도가 느림.
logger.debug("create scan:{}", scan);
return scan;
}
@Override
public List parallelScanTraceIndex(String[] agents, long start, long end) {
return null; //To change body of implemented methods use File | Settings | File Templates.
}
}
src/main/java/com/nhn/hippo/web/dao/HbaseTraceIndexDao.java
+2 -3
View File
@@ -34,7 +34,7 @@ public class HbaseTraceIndexDao implements TraceIndexDao {
private RowMapper<byte[]> traceIndexMapper;
private int scanCacheSize = 20;
private int scanCacheSize = 40;
public void setScanCacheSize(int scanCacheSize) {
this.scanCacheSize = scanCacheSize;
@@ -59,12 +59,12 @@ public class HbaseTraceIndexDao implements TraceIndexDao {
private Scan createScan(String agent, long start, long end) {
byte[] bAgent = Bytes.toBytes(agent);
Scan scan = new Scan();
// cache size를 지정해야 되는거 같음.??
scan.setCaching(this.scanCacheSize);
byte[] bAgent = Bytes.toBytes(agent);
byte[] bStart = BytesUtils.add(bAgent, start);
scan.setStartRow(bStart);
// TODO 추가 filter를 구현하여 scan시 중복된 값을 제가 할수 있음. 단 server에도 Filter 클래스가 배포되어야 한다.
@@ -72,7 +72,6 @@ public class HbaseTraceIndexDao implements TraceIndexDao {
byte[] bEnd = BytesUtils.add(bAgent, end);
scan.setStopRow(bEnd);
scan.addColumn(COLFAM_TRACE, COLNAME_ID);
scan.setId("traceIndexScan");
src/main/java/com/nhn/hippo/web/dao/RootTraceIndexDao.java
+15
View File
@@ -0,0 +1,15 @@
package com.nhn.hippo.web.dao;
import java.util.List;
/**
*
*/
public interface RootTraceIndexDao {
List<byte[]> scanTraceIndex(String agent, long start, long end);
List<List<byte[]>> multiScanTraceIndex(String[] agents, long start, long end);
List parallelScanTraceIndex(String[] agents, long start, long end);
}
src/main/java/com/nhn/hippo/web/mapper/AnnotationMapper.java
+5 -6
View File
@@ -31,27 +31,26 @@ public class AnnotationMapper implements RowMapper<Map<Long, List<AnnotationBo>>
int offset = kv.getValueOffset();
if (kv.getFamilyLength() == HBaseTables.TRACES_CF_ANNOTATION.length) {
// byte[] value = kv.getValue();
// if(value == null) {
// continue;
// }
int valueLength = kv.getValueLength();
if (valueLength == 0) {
continue;
}
int size = BytesUtils.bytesToInt(buffer, offset);
offset += 4;
if (size == 0) {
continue;
}
offset += 4;
List<AnnotationBo> bos = new ArrayList<AnnotationBo>(size);
for (int i = 0; i < size; i++) {
AnnotationBo annotationBo = new AnnotationBo();
annotationBo.setSpanId(spanId);
offset = annotationBo.readValue(buffer, offset);
bos.add(annotationBo);
logger.trace("read annotation:{}", annotationBo);
if (logger.isTraceEnabled()) {
logger.trace("read annotation:{}", annotationBo);
}
}
annotationList.put(spanId, bos);
}
src/main/java/com/nhn/hippo/web/mapper/BinaryAnnotationDecoder.java
-11
View File
@@ -1,11 +0,0 @@
package com.nhn.hippo.web.mapper;
import com.profiler.common.dto.thrift.BinaryAnnotation;
import com.profiler.common.dto.thrift.Span;
/**
*
*/
public interface BinaryAnnotationDecoder {
void decode(Span span);
}
src/main/java/com/nhn/hippo/web/mapper/JavaObjectDecoder.java
-46
View File
@@ -1,46 +0,0 @@
package com.nhn.hippo.web.mapper;
import com.profiler.common.dto.thrift.Annotation;
import com.profiler.common.dto.thrift.BinaryAnnotation;
import com.profiler.common.dto.thrift.Span;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
import java.util.ArrayList;
import java.util.List;
/**
* 임시 객체
*/
public class JavaObjectDecoder implements BinaryAnnotationDecoder {
private Logger logger = LoggerFactory.getLogger(this.getClass());
@Override
public void decode(Span span) {
// List<Annotation> annotations = span.getAnnotations();
//
// List<BinaryAnnotation> binaryAnnotations = span.getBinaryAnnotations();
// for (BinaryAnnotation binaryAnnotation : binaryAnnotations) {
// Object decode = decode(binaryAnnotation);
// Annotation annotation = new Annotation(binaryAnnotation.getTimestamp(), binaryAnnotation.getKey() + ":" + decode.toString());
// annotations.add(annotation);
// }
}
// private Object decode(BinaryAnnotation binaryAnnotation) {
// ByteArrayInputStream ins = new ByteArrayInputStream(binaryAnnotation.getValue());
// try {
// ObjectInputStream in = new ObjectInputStream(ins);
// Object readValue = in.readValue();
// return readValue;
// } catch (IOException e) {
// logger.warn("binaryAnnotation decode fail Cause:{}", e.getMessage(), e);
// return "binaryAnnotation decode fail Cause:" + e.getMessage();
// } catch (ClassNotFoundException e) {
// logger.warn("binaryAnnotation decode fail Cause:{}", e.getMessage(), e);
// return "binaryAnnotation decode fail Cause:" + e.getMessage();
// }
// }
}
src/main/java/com/nhn/hippo/web/mapper/SpanMapper.java
-32
View File
@@ -68,38 +68,6 @@ public class SpanMapper implements RowMapper<List<SpanBo>> {
return spanList;
// NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(COLFAM_SPAN);
// if (familyMap == null) {
// return Collections.emptyList();
// }
// List<SpanBo> spanList = new ArrayList<SpanBo>(familyMap.size());
// Put put = new Put(SpanUtils.getTracesRowkey(span), span.getTimestamp());
// // TODO columName이 중복일 경우를 확인가능하면 span id 중복 발급을 알수 있음.
// put.add(COLFAM_SPAN, Bytes.toBytes(span.getSpanID()), value);
// byte[] rowKey = result.getRow();
// long most = BytesUtils.bytesToFirstLong(rowKey);
// long least = BytesUtils.bytesToSecondLong(rowKey);
//
// for (NavigableMap.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
// SpanBo spanBo = new SpanBo();
//
// spanBo.setMostTraceID(most);
// spanBo.setLeastTraceID(least);
// spanBo.setSpanID(Bytes.toLong(entry.getKey()));
// //
// //byte[] spanId = entry.getKey();
//// if (binaryAnnotationDecoder != null) {
//// binaryAnnotationDecoder.decode(span);
//// }
//
// if (logger.isDebugEnabled()) {
// logger.debug("read span :{}", spanBo);
// }
// spanList.add(spanBo);
// }
// return spanList;
}
private void addAnnotation(List<SpanBo> spanList, Map<Long, List<AnnotationBo>> annotationMap) {
src/main/java/com/nhn/hippo/web/service/FlowChartServiceImpl.java
+9 -2
View File
@@ -2,6 +2,7 @@ package com.nhn.hippo.web.service;
import com.nhn.hippo.web.calltree.rpc.RPCCallTree;
import com.nhn.hippo.web.calltree.server.ServerCallTree;
import com.nhn.hippo.web.dao.RootTraceIndexDao;
import com.nhn.hippo.web.dao.TraceDao;
import com.nhn.hippo.web.dao.TraceIndexDao;
import com.nhn.hippo.web.service.TracesProcessor.SpanHandler;
@@ -19,6 +20,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;
import java.util.*;
@@ -37,6 +39,9 @@ public class FlowChartServiceImpl implements FlowChartService {
@Autowired
private TraceDao traceDao;
@Autowired
private RootTraceIndexDao rootTraceIndexDao;
@Autowired
private TraceIndexDao traceIndexDao;
@@ -71,7 +76,8 @@ public class FlowChartServiceImpl implements FlowChartService {
if (logger.isTraceEnabled()) {
logger.trace("scan {}, {}, {}", new Object[]{agentIds[0], from, to});
}
List<byte[]> bytes = this.traceIndexDao.scanTraceIndex(agentIds[0], from, to);
// List<byte[]> bytes = this.traceIndexDao.scanTraceIndex(agentIds[0], from, to);
List<byte[]> bytes = this.rootTraceIndexDao.scanTraceIndex(agentIds[0], from, to);
// 이런 필터로직을 scan filter에서 할수 없나?
Set<TraceId> result = new HashSet<TraceId>();
for (byte[] traceId : bytes) {
@@ -82,7 +88,8 @@ public class FlowChartServiceImpl implements FlowChartService {
return result;
} else {
// multi scan 가능한 동일 htable 에서 액세스함.
List<List<byte[]>> multiScan = this.traceIndexDao.multiScanTraceIndex(agentIds, from, to);
// List<List<byte[]>> multiScan = this.traceIndexDao.multiScanTraceIndex(agentIds, from, to);
List<List<byte[]>> multiScan = this.rootTraceIndexDao.multiScanTraceIndex(agentIds, from, to);
Set<TraceId> result = new HashSet<TraceId>();
for (List<byte[]> scan : multiScan) {
for (byte[] traceId : scan) {