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

[강운덕] [LUCYSUS-1744] index의 rowkey 의 agentid를 fixed하게 저장하도록 수정. root span만 스캔하던것을 전체 스캔하도록 다시 변경. root span만 스캔하니 조회시 한계가 있음. hbase client의 thread limit 조정.

Browse Source 
git-svn-id: http://svn.bds.nhncorp.com/pe/hippo-web/trunk@832 84d0f5b1-2673-498c-a247-62c4ff18d310
This commit is contained in:
Woonduk Kang
2012-10-29 02:50:44 +00:00
parent fc9da41124
commit 1fa55eb654
9 changed files with 16 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/nhn/hippo/web/dao/HbaseRootTraceIndexDao.java
+5 -7
View File
@@ -2,7 +2,7 @@ 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 com.profiler.common.util.SpanUtils;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
@@ -63,13 +63,11 @@ public class HbaseRootTraceIndexDao implements RootTraceIndexDao {
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[] traceIndexStartKey = SpanUtils.getTraceIndexRowKey(bAgent, start);
scan.setStartRow(traceIndexStartKey);
byte[] bEnd = BytesUtils.add(bAgent, end);
scan.setStopRow(bEnd);
byte[] traceIndexEndKey = SpanUtils.getTraceIndexRowKey(bAgent, end);
scan.setStopRow(traceIndexEndKey);
scan.addColumn(COLFAM_TRACE, COLNAME_ID);
scan.setId("rootTraceIndexScan");
src/main/java/com/nhn/hippo/web/dao/HbaseTraceIndexDao.java
+5 -11
View File
@@ -2,7 +2,7 @@ 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 com.profiler.common.util.SpanUtils;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
@@ -65,13 +65,13 @@ public class HbaseTraceIndexDao implements TraceIndexDao {
scan.setCaching(this.scanCacheSize);
byte[] bAgent = Bytes.toBytes(agent);
byte[] bStart = BytesUtils.add(bAgent, start);
scan.setStartRow(bStart);
byte[] traceIndexStartKey = SpanUtils.getTraceIndexRowKey(bAgent, start);
scan.setStartRow(traceIndexStartKey);
// TODO 추가 filter를 구현하여 scan시 중복된 값을 제가 할수 있음. 단 server에도 Filter 클래스가 배포되어야 한다.
// scan.setFilter(new ValueFilter());
byte[] bEnd = BytesUtils.add(bAgent, end);
scan.setStopRow(bEnd);
byte[] traceIndexEndKey = SpanUtils.getTraceIndexRowKey(bAgent, end);
scan.setStopRow(traceIndexEndKey);
scan.addColumn(COLFAM_TRACE, COLNAME_ID);
scan.setId("traceIndexScan");
@@ -80,10 +80,4 @@ public class HbaseTraceIndexDao implements TraceIndexDao {
return scan;
}
// private ExecutorService executor = Executors.newFixedThreadPool(100);
@Override
public List parallelScanTraceIndex(String[] agents, long start, long end) {
// executor.invokeAll();
throw new UnsupportedOperationException();
}
}
src/main/java/com/nhn/hippo/web/dao/TraceIndexDao.java
-2
View File
@@ -9,6 +9,4 @@ public interface TraceIndexDao {
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/service/FlowChartService.java
-8
View File
@@ -32,14 +32,6 @@ public interface FlowChartService {
*/
public Set<TraceId> selectTraceIdsFromTraceIndex(String[] agentIds, long from, long to);
/**
* select Traces from Trace table
*
* @param traceIds
* @return
*/
public Map<byte[], List<Span>> selectTraces(List<byte[]> traceIds);
/**
* select call tree
*
src/main/java/com/nhn/hippo/web/service/FlowChartServiceImpl.java
+3 -38
View File
@@ -5,22 +5,17 @@ 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;
import com.nhn.hippo.web.vo.TraceId;
import com.profiler.common.bo.SpanBo;
import com.profiler.common.dto.thrift.Span;
import com.profiler.common.hbase.HBaseClient;
import com.profiler.common.hbase.HBaseQuery;
import com.profiler.common.hbase.HBaseQuery.HbaseColumn;
import com.profiler.common.hbase.HBaseTables;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
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.stereotype.Service;
import org.springframework.util.StopWatch;
import java.util.*;
@@ -76,9 +71,7 @@ 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.rootTraceIndexDao.scanTraceIndex(agentIds[0], from, to);
// 이런 필터로직을 scan filter에서 할수 없나?
List<byte[]> bytes = this.traceIndexDao.scanTraceIndex(agentIds[0], from, to);
Set<TraceId> result = new HashSet<TraceId>();
for (byte[] traceId : bytes) {
TraceId tid = new TraceId(traceId);
@@ -87,9 +80,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.rootTraceIndexDao.multiScanTraceIndex(agentIds, from, to);
// multi scan 가능한 동일 open htable 에서 액세스함.
List<List<byte[]>> multiScan = this.traceIndexDao.multiScanTraceIndex(agentIds, from, to);
Set<TraceId> result = new HashSet<TraceId>();
for (List<byte[]> scan : multiScan) {
for (byte[] traceId : scan) {
@@ -100,33 +92,6 @@ public class FlowChartServiceImpl implements FlowChartService {
}
}
@Override
public Map<byte[], List<Span>> selectTraces(List<byte[]> traceIds) {
List<Get> gets = new ArrayList<Get>(traceIds.size());
for (byte[] traceId : traceIds) {
gets.add(new Get(traceId));
}
Result[] results = client.get(HBaseTables.TRACES, gets);
// traceId, SpanList
final Map<byte[], List<Span>> result = new HashMap<byte[], List<Span>>();
TracesProcessor.process(results, new SpanHandler() {
@Override
public void handleSpan(byte[] row, byte[] family, byte[] column, Span span) {
if (result.containsKey(row)) {
result.get(row).add(span);
} else {
List<Span> list = new ArrayList<Span>();
list.add(span);
result.put(row, list);
}
}
});
return result;
}
@Override
public RPCCallTree selectRPCCallTree(Set<TraceId> traceIds) {
src/main/java/com/nhn/hippo/web/service/TracesProcessor.java
-61
View File
@@ -1,61 +0,0 @@
package com.nhn.hippo.web.service;
import java.util.Map.Entry;
import java.util.NavigableMap;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
import com.profiler.common.dto.thrift.Span;
/**
*
* @author netspider
*
*/
public class TracesProcessor {
public static interface SpanHandler {
void handleSpan(byte[] row, byte[] family, byte[] column, Span span);
}
public static void process(Result[] results, SpanHandler handler) {
TDeserializer deserializer = new TDeserializer();
for (Result res : results) {
/**
* res.getMap() represent
* "Map<FAMILY, Map<COLUMN_NAME, Map<Timestamp, VALUE>>>"
*/
NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> map = res.getMap();
for (Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> entry : map.entrySet()) {
byte[] family = entry.getKey();
NavigableMap<byte[], NavigableMap<Long, byte[]>> values = entry.getValue();
/**
* For each column (SpanID)
*/
for (Entry<byte[], NavigableMap<Long, byte[]>> value : values.entrySet()) {
byte[] colname = value.getKey();
NavigableMap<Long, byte[]> valueSeries = value.getValue();
/**
* Decode span object
*/
for (Entry<Long, byte[]> v : valueSeries.entrySet()) {
Span span = new Span();
try {
deserializer.deserialize(span, v.getValue());
handler.handleSpan(res.getRow(), family, colname, span);
} catch (TException e) {
e.printStackTrace();
}
}
}
}
}
}
}
src/main/java/com/nhn/hippo/web/timeline/Timeline.java
-11
View File
@@ -1,11 +0,0 @@
package com.nhn.hippo.web.timeline;
/**
* Trace Timeline
*
* @author netspider
*
*/
public class Timeline {
}