[강운덕] [LUCYSUS-1744] flowchart의 traceIndex 조회 로직 최적화 120ms에서 17ms정도로 성능향상이 있는것으로 보임. 로직 최적화가 아니라 cache size때문에 그럴수도 있음.

hbasetemplate2에 multi scan관련 api를 추가함.

git-svn-id: http://svn.bds.nhncorp.com/pe/hippo-web/trunk@783 84d0f5b1-2673-498c-a247-62c4ff18d310

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")

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<byte[]> traceIndexMapper;
+
+
+    private int scanCacheSize = 20;
+
+    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.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.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();
+    }
+}

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<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/SpanMapper.java

@@ -18,7 +18,7 @@ import java.util.NavigableMap;
  *
  */
 @Component
-public class SpanMapper<T extends List<Span>> implements RowMapper<List<Span>> {
+public class SpanMapper implements RowMapper<List<Span>> {
 
     private final byte[] COLFAM_SPAN = Bytes.toBytes("Span");
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
@@ -31,11 +31,14 @@ public class SpanMapper<T extends List<Span>> implements RowMapper<List<Span>> {
         if (familyMap == null) {
             return Collections.emptyList();
         }
+
         List<Span> spanList = new ArrayList<Span>(familyMap.size());
         // TODO thrift 포멧이 아니고 따로 풀어서 넣어야 될거 같음.
         TDeserializer de = new TDeserializer();
         for (NavigableMap.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
             Span span = new Span();
+            // spainid가 이미 value에 들어 있어서 일단 필요가 없음.
+            //byte[] spanId = entry.getKey();
             de.deserialize(span, entry.getValue());
             if (binaryAnnotationDecoder != null) {
                 binaryAnnotationDecoder.decode(span);

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<byte[]> {
+
+    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);
+    }
+}

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<HbaseColumn> column = new ArrayList<HBaseQuery.HbaseColumn>();
@@ -52,37 +47,44 @@ public class FlowChartServiceImpl implements FlowChartService {
         HBaseQuery query = new HBaseQuery(HBaseTables.SERVERS, null, null, column);
         Iterator<Map<String, byte[]>> 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<TraceId> selectTraceIdsFromTraceIndex(String[] agentIds, long from, long to) {
-        List<HbaseColumn> column = new ArrayList<HBaseQuery.HbaseColumn>();
-        column.add(new HbaseColumn("Trace", "ID"));
-
-        Set<TraceId> set = new HashSet<TraceId>();
-
-        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<Map<String, byte[]>> 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<byte[]> bytes = this.traceIndexDao.scanTraceIndex(agentIds[0], from, to);
+            // 이런 필터로직을 scan filter에서 할수 없나?
+            Set<TraceId> result = new HashSet<TraceId>();
+            for (byte[] traceId : bytes) {
+                result.add(new TraceId(traceId));
+            }
+            return result;
+        } else {
+            // multi scan 가능한 동일 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) {
+                    result.add(new TraceId(traceId));
+                }
+            }
+            return result;
+        }
     }
 
     @Override

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 + "]";
     }
 }