diff --git a/pom.xml b/pom.xml index 50510ee6e..ed0b176ac 100644 --- a/pom.xml +++ b/pom.xml @@ -404,6 +404,37 @@ 1.1.0 + + + org.codehaus.jackson + jackson-core-asl + 1.9.9 + + + + + org.codehaus.jackson + jackson-mapper-asl + 1.9.9 + + + + + org.codehaus.jackson + jackson-xc + 1.9.9 + + + + + org.codehaus.jackson + jackson-jaxrs + 1.9.9 + + + + + org.slf4j diff --git a/src/main/java/com/nhn/hippo/web/dao/HbaseTraceDao.java b/src/main/java/com/nhn/hippo/web/dao/HbaseTraceDao.java index 9f33c94a5..ce58ddeab 100644 --- a/src/main/java/com/nhn/hippo/web/dao/HbaseTraceDao.java +++ b/src/main/java/com/nhn/hippo/web/dao/HbaseTraceDao.java @@ -1,10 +1,9 @@ package com.nhn.hippo.web.dao; -import com.nhn.hippo.web.mapper.SpanMapper; import com.nhn.hippo.web.vo.TraceId; import com.profiler.common.dto.thrift.Span; import com.profiler.common.hbase.HBaseTables; -import com.profiler.common.hbase.HbaseTemplate2; +import com.profiler.common.hbase.HbaseOperations2; import com.profiler.common.util.BytesUtils; import org.apache.hadoop.hbase.client.Get; import org.apache.hadoop.hbase.util.Bytes; @@ -30,7 +29,7 @@ public class HbaseTraceDao implements TraceDao { private Logger logger = LoggerFactory.getLogger(this.getClass()); @Autowired - private HbaseTemplate2 template2; + private HbaseOperations2 template2; @Autowired @Qualifier("spanMapper") diff --git a/src/main/java/com/nhn/hippo/web/dao/HbaseTraceIndexDao.java b/src/main/java/com/nhn/hippo/web/dao/HbaseTraceIndexDao.java new file mode 100644 index 000000000..38e272634 --- /dev/null +++ b/src/main/java/com/nhn/hippo/web/dao/HbaseTraceIndexDao.java @@ -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 HbaseTraceIndexDao implements TraceIndexDao { + + private final 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 traceIndexMapper; + + + private int scanCacheSize = 20; + + public void setScanCacheSize(int scanCacheSize) { + this.scanCacheSize = scanCacheSize; + } + + + @Override + public List scanTraceIndex(String agent, long start, long end) { + Scan scan = createScan(agent, start, end); + return hbaseOperations2.find(HBaseTables.TRACE_INDEX, scan, traceIndexMapper); + } + + @Override + public List> multiScanTraceIndex(String[] agents, long start, long end) { + final List multiScan = new ArrayList(agents.length); + for (String agent : agents) { + Scan scan = createScan(agent, start, end); + multiScan.add(scan); + } + return hbaseOperations2.find(HBaseTables.TRACE_INDEX, multiScan, traceIndexMapper); + } + + + 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[] bStart = BytesUtils.add(bAgent, start); + scan.setStartRow(bStart); + + byte[] bEnd = BytesUtils.add(bAgent, end); + scan.setStopRow(bEnd); + + scan.addColumn(COLFAM_TRACE, COLNAME_ID); + + // json으로 변화해서 로그를 찍어서. 최초 변환 속도가 느림. + logger.debug("create scan:{}", scan); + return scan; + } + + + @Override + public List parallelScanTraceIndex(String[] agents, long start, long end) { + throw new UnsupportedOperationException(); + } +} diff --git a/src/main/java/com/nhn/hippo/web/dao/TraceIndexDao.java b/src/main/java/com/nhn/hippo/web/dao/TraceIndexDao.java index 32fe9ebf4..77dd48fc3 100644 --- a/src/main/java/com/nhn/hippo/web/dao/TraceIndexDao.java +++ b/src/main/java/com/nhn/hippo/web/dao/TraceIndexDao.java @@ -1,7 +1,14 @@ package com.nhn.hippo.web.dao; +import java.util.List; + /** * */ -public class TraceIndexDao { +public interface TraceIndexDao { + List scanTraceIndex(String agent, long start, long end); + + List> multiScanTraceIndex(String[] agents, long start, long end); + + List parallelScanTraceIndex(String[] agents, long start, long end); } diff --git a/src/main/java/com/nhn/hippo/web/mapper/SpanMapper.java b/src/main/java/com/nhn/hippo/web/mapper/SpanMapper.java index b6258e142..1d66f5aa0 100644 --- a/src/main/java/com/nhn/hippo/web/mapper/SpanMapper.java +++ b/src/main/java/com/nhn/hippo/web/mapper/SpanMapper.java @@ -18,7 +18,7 @@ import java.util.NavigableMap; * */ @Component -public class SpanMapper> implements RowMapper> { +public class SpanMapper implements RowMapper> { private final byte[] COLFAM_SPAN = Bytes.toBytes("Span"); private final Logger logger = LoggerFactory.getLogger(this.getClass()); @@ -31,11 +31,14 @@ public class SpanMapper> implements RowMapper> { if (familyMap == null) { return Collections.emptyList(); } + List spanList = new ArrayList(familyMap.size()); // TODO thrift 포멧이 아니고 따로 풀어서 넣어야 될거 같음. TDeserializer de = new TDeserializer(); for (NavigableMap.Entry entry : familyMap.entrySet()) { Span span = new Span(); + // spainid가 이미 value에 들어 있어서 일단 필요가 없음. + //byte[] spanId = entry.getKey(); de.deserialize(span, entry.getValue()); if (binaryAnnotationDecoder != null) { binaryAnnotationDecoder.decode(span); diff --git a/src/main/java/com/nhn/hippo/web/mapper/TraceIndexMapper.java b/src/main/java/com/nhn/hippo/web/mapper/TraceIndexMapper.java new file mode 100644 index 000000000..3e317d77e --- /dev/null +++ b/src/main/java/com/nhn/hippo/web/mapper/TraceIndexMapper.java @@ -0,0 +1,21 @@ +package com.nhn.hippo.web.mapper; + +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.util.Bytes; +import org.springframework.data.hadoop.hbase.RowMapper; +import org.springframework.stereotype.Component; + +/** + * + */ +@Component +public class TraceIndexMapper implements RowMapper { + + private final byte[] COLFAM_TRACE = Bytes.toBytes("Trace"); + private final byte[] COLNAME_ID = Bytes.toBytes("ID"); + + @Override + public byte[] mapRow(Result result, int rowNum) throws Exception { + return result.getValue(COLFAM_TRACE, COLNAME_ID); + } +} diff --git a/src/main/java/com/nhn/hippo/web/service/FlowChartServiceImpl.java b/src/main/java/com/nhn/hippo/web/service/FlowChartServiceImpl.java index ce06f40f4..8c81fc201 100755 --- a/src/main/java/com/nhn/hippo/web/service/FlowChartServiceImpl.java +++ b/src/main/java/com/nhn/hippo/web/service/FlowChartServiceImpl.java @@ -1,26 +1,9 @@ package com.nhn.hippo.web.service; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Iterator; -import java.util.List; -import java.util.Map; -import java.util.Set; - -import com.nhn.hippo.web.dao.TraceDao; -import org.apache.commons.lang.ArrayUtils; -import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.Result; -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.stereotype.Service; - import com.nhn.hippo.web.calltree.rpc.RPCCallTree; import com.nhn.hippo.web.calltree.server.ServerCallTree; +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.dto.thrift.Span; @@ -28,6 +11,15 @@ 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 java.util.*; /** * @author netspider @@ -44,6 +36,9 @@ public class FlowChartServiceImpl implements FlowChartService { @Autowired private TraceDao traceDao; + @Autowired + private TraceIndexDao traceIndexDao; + @Override public String[] selectAgentIds(String[] hosts) { List column = new ArrayList(); @@ -52,37 +47,44 @@ public class FlowChartServiceImpl implements FlowChartService { HBaseQuery query = new HBaseQuery(HBaseTables.SERVERS, null, null, column); Iterator> iterator = client.getHBaseData(query); - while (iterator.hasNext()) { - System.out.println("selectedAgentId=" + iterator.next()); + if (logger.isDebugEnabled()) { + while (iterator.hasNext()) { + logger.debug("selectedAgentId={}", iterator.next()); + } + logger.debug("!!!==============WARNING==============!!!"); + logger.debug("!!! selectAgentIds IS NOT IMPLEMENTED !!!"); + logger.debug("!!!===================================!!!"); } - System.out.println("!!!==============WARNING==============!!!"); - System.out.println("!!! selectAgentIds IS NOT IMPLEMENTED !!!"); - System.out.println("!!!===================================!!!"); - return hosts; } @Override public Set selectTraceIdsFromTraceIndex(String[] agentIds, long from, long to) { - List column = new ArrayList(); - column.add(new HbaseColumn("Trace", "ID")); - - Set set = new HashSet(); - - for (String agentId : agentIds) { - byte[] s = ArrayUtils.addAll(Bytes.toBytes(agentId), Bytes.toBytes(from)); - byte[] e = ArrayUtils.addAll(Bytes.toBytes(agentId), Bytes.toBytes(to)); - - HBaseQuery query = new HBaseQuery(HBaseTables.TRACE_INDEX, s, e, column); - Iterator> result = client.getHBaseData(query); - - while (result.hasNext()) { - set.add(new TraceId(result.next().get("ID"))); - } + if (agentIds == null) { + throw new NullPointerException("agentIds"); } - return set; + if (agentIds.length == 1) { + // single scan + List bytes = this.traceIndexDao.scanTraceIndex(agentIds[0], from, to); + // 이런 필터로직을 scan filter에서 할수 없나? + Set result = new HashSet(); + for (byte[] traceId : bytes) { + result.add(new TraceId(traceId)); + } + return result; + } else { + // multi scan 가능한 동일 htable 에서 액세스함. + List> multiScan = this.traceIndexDao.multiScanTraceIndex(agentIds, from, to); + Set result = new HashSet(); + for (List scan : multiScan) { + for (byte[] traceId : scan) { + result.add(new TraceId(traceId)); + } + } + return result; + } } @Override diff --git a/src/main/java/com/nhn/hippo/web/vo/TraceId.java b/src/main/java/com/nhn/hippo/web/vo/TraceId.java index 6a919e5c7..e35863c72 100644 --- a/src/main/java/com/nhn/hippo/web/vo/TraceId.java +++ b/src/main/java/com/nhn/hippo/web/vo/TraceId.java @@ -2,7 +2,6 @@ package com.nhn.hippo.web.vo; import com.profiler.common.util.BytesUtils; -import java.util.Arrays; import java.util.UUID; public class TraceId { @@ -15,12 +14,12 @@ public class TraceId { if (traceId == null) { throw new NullPointerException("traceId"); } - if (traceId.length < 8) { + if (traceId.length < 16) { throw new IllegalArgumentException("invalid traceId"); } this.id = traceId; - this.most = BytesUtils.bytesToFirstLong(traceId); - this.least = BytesUtils.bytesToSecondLong(traceId); + this.most = BytesUtils.bytesToFirstLong(id); + this.least = BytesUtils.bytesToSecondLong(id); } public byte[] getBytes() { @@ -50,6 +49,6 @@ public class TraceId { @Override public String toString() { UUID uuid = new UUID(most, least); - return "TraceId [id=" + uuid + "]"; + return "TraceId [" + uuid + "]"; } } diff --git a/src/main/resources/log4j.xml b/src/main/resources/log4j.xml index 3c36e3868..e5c91d05c 100644 --- a/src/main/resources/log4j.xml +++ b/src/main/resources/log4j.xml @@ -18,11 +18,11 @@ - + - +