From 16ea3f2a5a012233d34c05b2efdb2496cd5f6d49 Mon Sep 17 00:00:00 2001 From: Woonduk Kang Date: Mon, 29 Oct 2012 02:50:44 +0000 Subject: [PATCH] =?UTF-8?q?[=EA=B0=95=EC=9A=B4=EB=8D=95]=20[LUCYSUS-1744]?= =?UTF-8?q?=20index=EC=9D=98=20rowkey=20=EC=9D=98=20agentid=EB=A5=BC=20fix?= =?UTF-8?q?ed=ED=95=98=EA=B2=8C=20=EC=A0=80=EC=9E=A5=ED=95=98=EB=8F=84?= =?UTF-8?q?=EB=A1=9D=20=EC=88=98=EC=A0=95.=20root=20span=EB=A7=8C=20?= =?UTF-8?q?=EC=8A=A4=EC=BA=94=ED=95=98=EB=8D=98=EA=B2=83=EC=9D=84=20?= =?UTF-8?q?=EC=A0=84=EC=B2=B4=20=EC=8A=A4=EC=BA=94=ED=95=98=EB=8F=84?= =?UTF-8?q?=EB=A1=9D=20=EB=8B=A4=EC=8B=9C=20=EB=B3=80=EA=B2=BD.=20root=20s?= =?UTF-8?q?pan=EB=A7=8C=20=EC=8A=A4=EC=BA=94=ED=95=98=EB=8B=88=20=EC=A1=B0?= =?UTF-8?q?=ED=9A=8C=EC=8B=9C=20=ED=95=9C=EA=B3=84=EA=B0=80=20=EC=9E=88?= =?UTF-8?q?=EC=9D=8C.=20hbase=20client=EC=9D=98=20thread=20limit=20?= =?UTF-8?q?=EC=A1=B0=EC=A0=95.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit git-svn-id: http://svn.bds.nhncorp.com/pe/hippo-commons/trunk@832 84d0f5b1-2673-498c-a247-62c4ff18d310 --- .../common/hbase/HBaseAdminTemplate.java | 78 +++ .../profiler/common/hbase/HBaseClient.java | 500 ++++++++---------- .../java/com/profiler/common/util/Buffer.java | 8 + .../common/util/HeaderTBaseDeserializer.java | 169 +++--- .../{HeaderUtil.java => HeaderUtils.java} | 4 +- .../com/profiler/common/util/RowKeyUtils.java | 24 + .../com/profiler/common/util/SpanUtils.java | 18 +- .../common/hbase/HBaseClientTest.java | 95 ++-- .../com/profiler/common/util/BufferTest.java | 76 +++ .../profiler/common/util/SpanUtilsTest.java | 58 ++ src/test/resources/test-hbase.properties | 4 +- 11 files changed, 627 insertions(+), 407 deletions(-) create mode 100644 src/main/java/com/profiler/common/hbase/HBaseAdminTemplate.java rename src/main/java/com/profiler/common/util/{HeaderUtil.java => HeaderUtils.java} (56%) create mode 100644 src/main/java/com/profiler/common/util/RowKeyUtils.java create mode 100644 src/test/java/com/profiler/common/util/BufferTest.java create mode 100644 src/test/java/com/profiler/common/util/SpanUtilsTest.java diff --git a/src/main/java/com/profiler/common/hbase/HBaseAdminTemplate.java b/src/main/java/com/profiler/common/hbase/HBaseAdminTemplate.java new file mode 100644 index 000000000..a7ca54959 --- /dev/null +++ b/src/main/java/com/profiler/common/hbase/HBaseAdminTemplate.java @@ -0,0 +1,78 @@ +package com.profiler.common.hbase; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.MasterNotRunningException; +import org.apache.hadoop.hbase.ZooKeeperConnectionException; +import org.apache.hadoop.hbase.client.HBaseAdmin; +import org.springframework.data.hadoop.hbase.HbaseSystemException; + +import java.io.IOException; + +/** + * + */ +public class HBaseAdminTemplate { + + private final HBaseAdmin hBaseAdmin; + + public HBaseAdminTemplate(Configuration configuration) { + try { + this.hBaseAdmin = new HBaseAdmin(configuration); + } catch (MasterNotRunningException e) { + throw new HbaseSystemException(e); + } catch (ZooKeeperConnectionException e) { + throw new HbaseSystemException(e); + } + } + + public boolean createTableIfNotExist(HTableDescriptor htd) { + try { + if (!hBaseAdmin.tableExists(htd.getName())) { + this.hBaseAdmin.createTable(htd); + return true; + } + return false; + } catch (IOException e) { + throw new HbaseSystemException(e); + } + } + + public boolean tableExists(String tableName) { + try { + return hBaseAdmin.tableExists(tableName); + } catch (IOException e) { + throw new HbaseSystemException(e); + } + } + + public boolean dropTableIfExist(String tableName) { + try { + if (hBaseAdmin.tableExists(tableName)) { + this.hBaseAdmin.disableTable(tableName); + this.hBaseAdmin.deleteTable(tableName); + return true; + } + return false; + } catch (IOException e) { + throw new HbaseSystemException(e); + } + } + + public void dropTable(String tableName) { + try { + this.hBaseAdmin.disableTable(tableName); + this.hBaseAdmin.deleteTable(tableName); + } catch (IOException e) { + throw new HbaseSystemException(e); + } + } + + public void close() { + try { + this.hBaseAdmin.close(); + } catch (IOException e) { + throw new HbaseSystemException(e); + } + } +} diff --git a/src/main/java/com/profiler/common/hbase/HBaseClient.java b/src/main/java/com/profiler/common/hbase/HBaseClient.java index b66f77a9c..e7322cabd 100644 --- a/src/main/java/com/profiler/common/hbase/HBaseClient.java +++ b/src/main/java/com/profiler/common/hbase/HBaseClient.java @@ -11,12 +11,8 @@ import java.util.Properties; import org.apache.commons.lang.math.NumberUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.HTableDescriptor; -import org.apache.hadoop.hbase.MasterNotRunningException; -import org.apache.hadoop.hbase.ZooKeeperConnectionException; import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HBaseAdmin; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.HTablePool; @@ -31,315 +27,281 @@ import com.profiler.common.hbase.HBaseQuery.HbaseColumn; public class HBaseClient { - private static final Logger LOG = LoggerFactory.getLogger(HBaseClient.class); + private static final Logger LOG = LoggerFactory.getLogger(HBaseClient.class); - private HTablePool tablePool; - private HBaseAdmin admin; + private Configuration configuration; + private HTablePool tablePool; - private final Map htableList = new HashMap(); + private final Map htableList = new HashMap(); - public HBaseClient(Properties properties) { - String host = properties.getProperty("hbase.client.host", "localhost"); - String port = properties.getProperty("hbase.client.port", "2181"); - Integer poolSize = NumberUtils.toInt(properties.getProperty("hbase.client.poolSize"), 16); - init(host, port, poolSize, null); - } + public HBaseClient(Properties properties) { + String host = properties.getProperty("hbase.client.host", "localhost"); + String port = properties.getProperty("hbase.client.port", "2181"); + this.configuration = createConfiguration(host, port); + Integer poolSize = NumberUtils.toInt(properties.getProperty("hbase.client.poolSize"), 16); + init(configuration, poolSize); + } - public HBaseClient(String zk, String port, int poolSize) { - init(zk, port, poolSize, null); - } + public HBaseClient(String zk, String port, int poolSize) { + Configuration cfg = createConfiguration(zk, port); + init(cfg, poolSize); + } - public HBaseClient(String zk, String port, int poolSize, Configuration configuration) { - init(zk, port, poolSize, configuration); - } + private Configuration createConfiguration(String zk, String port) { + Configuration cfg = HBaseConfiguration.create(); + cfg.set("hbase.zookeeper.quorum", zk); + cfg.set("hbase.zookeeper.property.clientPort", port); + return cfg; + } - public HBaseClient(Configuration configuration, int poolSize) { - init(null, null, poolSize, configuration); - } + public HBaseClient(Configuration configuration, int poolSize) { + init(configuration, poolSize); + } - private void init(String zk, String port, int poolSize, Configuration configuration) { - if (configuration == null) { - Configuration cfg = HBaseConfiguration.create(); - if (zk != null) { - cfg.set("hbase.zookeeper.quorum", zk); - } - if (port != null) { - cfg.set("hbase.zookeeper.property.clientPort", port); - } - configuration = cfg; - } + private void init(Configuration configuration, int poolSize) { + tablePool = new HTablePool(configuration, poolSize); + } - tablePool = new HTablePool(configuration, poolSize); - try { - admin = new HBaseAdmin(configuration); - } catch (MasterNotRunningException e) { - e.printStackTrace(); - } catch (ZooKeeperConnectionException e) { - e.printStackTrace(); - } - } + public Configuration getConfiguration() { + return configuration; + } - public void close() { - // htableList는 안지워워도 되지. - try { - tablePool.close(); - } catch (IOException e) { - // TODO - e.printStackTrace(); // To change body of catch statement use File | - // Settings | File Templates. - } - } + public void close() { + // htableList는 안지워워도 되지. + try { + tablePool.close(); + } catch (IOException e) { + // TODO + e.printStackTrace(); // To change body of catch statement use File | + // Settings | File Templates. + } + } - HTablePool getTablePool() { - return tablePool; - } + HTablePool getTablePool() { + return tablePool; + } - public Iterator> getHBaseData(HBaseQuery query) { - ResultSetIterator r = new ResultSetIterator(query); - return r.getIterator(); - } + public Iterator> getHBaseData(HBaseQuery query) { + ResultSetIterator r = new ResultSetIterator(query); + return r.getIterator(); + } - private HTableInterface getHTable(String tableName) { - HTableInterface htable = null; - synchronized (htableList) { - htable = htableList.get(tableName); - if (htable == null) { - htable = tablePool.getTable(tableName); - htableList.put(tableName, htable); - } - } - return htable; - } + private HTableInterface getHTable(String tableName) { + HTableInterface htable = null; + synchronized (htableList) { + htable = htableList.get(tableName); + if (htable == null) { + htable = tablePool.getTable(tableName); + htableList.put(tableName, htable); + } + } + return htable; + } - public boolean isTableExists(final String tableName) { - try { - return admin.tableExists(tableName); - } catch (IOException e) { - e.printStackTrace(); - return false; - } - } - public void dropTable(final String tableName) { - try { - admin.disableTable(tableName); - admin.deleteTable(tableName); - } catch (IOException e) { - e.printStackTrace(); - } - } + public void insert(String tableName, Put put) { + HTable htable = (HTable) tablePool.getTable(tableName); + try { + htable.put(put); + } catch (IOException e) { + e.printStackTrace(); + } finally { + closeHTable(htable); + } + } - public void createTable(HTableDescriptor td) { - try { - admin.createTable(td); - } catch (IOException e) { - e.printStackTrace(); - } - } + public void insert(String tableName, List put) { + HTable htable = (HTable) tablePool.getTable(tableName); + try { + htable.put(put); + } catch (IOException e) { + e.printStackTrace(); + } finally { + closeHTable(htable); + } + } - public void insert(String tableName, Put put) { - HTable htable = (HTable) tablePool.getTable(tableName); - try { - htable.put(put); - } catch (IOException e) { - e.printStackTrace(); - } finally { - closeHTable(htable); - } - } + public void delete(String tableName, Delete delete) { + HTable htable = (HTable) tablePool.getTable(tableName); + try { + htable.delete(delete); + } catch (IOException e) { + e.printStackTrace(); + } finally { + closeHTable(htable); + } + } - public void insert(String tableName, List put) { - HTable htable = (HTable) tablePool.getTable(tableName); - try { - htable.put(put); - } catch (IOException e) { - e.printStackTrace(); - } finally { - closeHTable(htable); - } - } + public void execute(String tableName, HTableCallBack callBack) { + HTable htable = (HTable) tablePool.getTable(tableName); + try { + callBack.doExecute(htable); + } catch (IOException e) { + e.printStackTrace(); + // TODO ex 처리 + } finally { + closeHTable(htable); + } + } - public void delete(String tableName, Delete delete) { - HTable htable = (HTable) tablePool.getTable(tableName); - try { - htable.delete(delete); - } catch (IOException e) { - e.printStackTrace(); - } finally { - closeHTable(htable); - } - } + public Result[] get(String tablename, List get) { + HTable htable = (HTable) tablePool.getTable(tablename); - public void execute(String tableName, HTableCallBack callBack) { - HTable htable = (HTable) tablePool.getTable(tableName); - try { - callBack.doExecute(htable); - } catch (IOException e) { - e.printStackTrace(); - // TODO ex 처리 - } finally { - closeHTable(htable); - } - } + try { + return htable.get(get); + } catch (IOException e) { + e.printStackTrace(); + // TODO ex 처리 + } finally { + closeHTable(htable); + } + return new Result[0]; + } - public Result[] get(String tablename, List get) { - HTable htable = (HTable) tablePool.getTable(tablename); + private void closeHTable(HTable htable) { + if (htable != null) { + try { + htable.close(); + } catch (IOException e) { + LOG.warn(e.getMessage(), e); + } + } + } - try { - return htable.get(get); - } catch (IOException e) { - e.printStackTrace(); - // TODO ex 처리 - } finally { - closeHTable(htable); - } - return new Result[0]; - } + private class ResultSetIterator { + ResultScanner resultScanner = null; + Iterator resultIterator; + List columns; + Iterator> rSetIterator; - private void closeHTable(HTable htable) { - if (htable != null) { - try { - htable.close(); - } catch (IOException e) { - LOG.warn(e.getMessage(), e); - } - } - } + public ResultSetIterator(HBaseQuery query) { + try { + byte[] startRow = query.getStartRow(); + byte[] stopRow = query.getStopRow(); + String tableName = query.getTableName(); - private class ResultSetIterator { - ResultScanner resultScanner = null; - Iterator resultIterator; - List columns; - Iterator> rSetIterator; + columns = query.getColumns(); + HTableInterface htable = getHTable(tableName); - public ResultSetIterator(HBaseQuery query) { - try { - byte[] startRow = query.getStartRow(); - byte[] stopRow = query.getStopRow(); - String tableName = query.getTableName(); + if (query.isSingleRow()) { + Get get = new Get(startRow); - columns = query.getColumns(); - HTableInterface htable = getHTable(tableName); + if (columns != null) { + for (HbaseColumn column : columns) { + get.addColumn(column.getFamily().getBytes(), column.getColumnName().getBytes()); + } + } - if (query.isSingleRow()) { - Get get = new Get(startRow); + Result result = htable.get(get); + List resultList = new ArrayList(1); + resultList.add(result); + resultIterator = resultList.iterator(); + } else { + Scan scan = new Scan(); - if (columns != null) { - for (HbaseColumn column : columns) { - get.addColumn(column.getFamily().getBytes(), column.getColumnName().getBytes()); - } - } + if (startRow != null) { + scan.setStartRow(startRow); + } - Result result = htable.get(get); - List resultList = new ArrayList(1); - resultList.add(result); - resultIterator = resultList.iterator(); - } else { - Scan scan = new Scan(); + if (stopRow != null) { + scan.setStopRow(stopRow); + } - if (startRow != null) { - scan.setStartRow(startRow); - } + if (columns != null) { + for (HbaseColumn column : columns) { + scan.addColumn(column.getFamily().getBytes(), column.getColumnName().getBytes()); + } + } - if (stopRow != null) { - scan.setStopRow(stopRow); - } + LOG.debug("Executing scanner: " + query); - if (columns != null) { - for (HbaseColumn column : columns) { - scan.addColumn(column.getFamily().getBytes(), column.getColumnName().getBytes()); - } - } + long start = System.currentTimeMillis(); - LOG.debug("Executing scanner: " + query); + resultScanner = htable.getScanner(scan); - long start = System.currentTimeMillis(); + LOG.trace("Time taken for scanner: " + (System.currentTimeMillis() - start)); - resultScanner = htable.getScanner(scan); + resultIterator = resultScanner.iterator(); + } + } catch (Exception e) { + throw new RuntimeException("UNable to execute SCANNER : " + query, e); + } - LOG.trace("Time taken for scanner: " + (System.currentTimeMillis() - start)); + if (!resultIterator.hasNext()) { + rSetIterator = new ArrayList>().iterator(); + return; + } - resultIterator = resultScanner.iterator(); - } - } catch (Exception e) { - throw new RuntimeException("UNable to execute SCANNER : " + query, e); - } + rSetIterator = new Iterator>() { + public boolean hasNext() { + return hasnext(); + } - if (!resultIterator.hasNext()) { - rSetIterator = new ArrayList>().iterator(); - return; - } + public Map next() { + return getARow(); + } - rSetIterator = new Iterator>() { - public boolean hasNext() { - return hasnext(); - } + public void remove() { + } + }; + } - public Map next() { - return getARow(); - } + private Iterator> getIterator() { + return rSetIterator; + } - public void remove() { - } - }; - } + private Map getARow() { + if (resultIterator == null) + return null; + Result res = resultIterator.next(); - private Iterator> getIterator() { - return rSetIterator; - } + Map result = new HashMap(); + if (!res.isEmpty()) { + byte[] value; + if (columns != null) { + for (HbaseColumn column : columns) { + String colName = column.getColumnName(); + value = res.getValue(column.getFamily().getBytes(), column.getColumnName().getBytes()); - private Map getARow() { - if (resultIterator == null) - return null; - Result res = resultIterator.next(); + if (value == null) { + continue; + } - Map result = new HashMap(); - if (!res.isEmpty()) { - byte[] value; - if (columns != null) { - for (HbaseColumn column : columns) { - String colName = column.getColumnName(); - value = res.getValue(column.getFamily().getBytes(), column.getColumnName().getBytes()); + result.put(colName, value); + } + } + value = res.getRow(); + } - if (value == null) { - continue; - } + return result; + } - result.put(colName, value); - } - } - value = res.getRow(); - } + private boolean hasnext() { + if (resultIterator == null) + return false; + try { + if (resultIterator.hasNext()) { + return true; + } else { + close(); + return false; + } + } catch (Exception e) { + close(); + e.printStackTrace(); + return false; + } + } - return result; - } - - private boolean hasnext() { - if (resultIterator == null) - return false; - try { - if (resultIterator.hasNext()) { - return true; - } else { - close(); - return false; - } - } catch (Exception e) { - close(); - e.printStackTrace(); - return false; - } - } - - private void close() { - try { - if (resultScanner != null) { - resultScanner.close(); - } - } catch (Exception e) { - e.printStackTrace(); - } finally { - } - } - } + private void close() { + try { + if (resultScanner != null) { + resultScanner.close(); + } + } catch (Exception e) { + e.printStackTrace(); + } finally { + } + } + } } diff --git a/src/main/java/com/profiler/common/util/Buffer.java b/src/main/java/com/profiler/common/util/Buffer.java index 041f9f5b0..641d9a362 100644 --- a/src/main/java/com/profiler/common/util/Buffer.java +++ b/src/main/java/com/profiler/common/util/Buffer.java @@ -78,6 +78,10 @@ public class Buffer { if (size == 0) { return EMPTY; } + return readBytes(size); + } + + private byte[] readBytes(int size) { byte[] b = new byte[size]; System.arraycopy(buffer, offset, b, 0, size); this.offset = offset + size; @@ -89,6 +93,10 @@ public class Buffer { if (size == 0) { return ""; } + return readString(size); + } + + private String readString(int size) { String s = new String(buffer, offset, size, UTF8); this.offset = offset + size; return s; diff --git a/src/main/java/com/profiler/common/util/HeaderTBaseDeserializer.java b/src/main/java/com/profiler/common/util/HeaderTBaseDeserializer.java index a973b40a2..349f3ba68 100644 --- a/src/main/java/com/profiler/common/util/HeaderTBaseDeserializer.java +++ b/src/main/java/com/profiler/common/util/HeaderTBaseDeserializer.java @@ -8,97 +8,94 @@ import org.apache.thrift.protocol.*; import org.apache.thrift.transport.TMemoryInputTransport; public class HeaderTBaseDeserializer { - - private final TProtocol protocol_; - private final TMemoryInputTransport trans_; - /** - * Create a new TDeserializer that uses the TBinaryProtocol by default. - */ - public HeaderTBaseDeserializer() { - this(new TBinaryProtocol.Factory()); - } + private final TProtocol protocol_; + private final TMemoryInputTransport trans_; - /** - * Create a new TDeserializer. It will use the TProtocol specified by the - * factory that is passed in. - * - * @param protocolFactory - * Factory to create a protocol - */ - public HeaderTBaseDeserializer(TProtocolFactory protocolFactory) { - trans_ = new TMemoryInputTransport(); - protocol_ = protocolFactory.getProtocol(trans_); - } + /** + * Create a new TDeserializer that uses the TBinaryProtocol by default. + */ + public HeaderTBaseDeserializer() { + this(new TBinaryProtocol.Factory()); + } - /** - * Deserialize the Thrift object from a byte array. - * - * @param locator - * The object to read into - * @param bytes - * The array to read from - */ - public TBase deserialize(TBaseLocator locator, byte[] bytes) throws TException { - try { - trans_.reset(bytes); - Header header = readHeader(); - validate(header); - TBase base = locator.tBaseLookup(header.getType()); - base.read(protocol_); - return base; - } finally { - trans_.clear(); - protocol_.reset(); - } - } + /** + * Create a new TDeserializer. It will use the TProtocol specified by the + * factory that is passed in. + * + * @param protocolFactory Factory to create a protocol + */ + public HeaderTBaseDeserializer(TProtocolFactory protocolFactory) { + trans_ = new TMemoryInputTransport(); + protocol_ = protocolFactory.getProtocol(trans_); + } - private void validate(Header header) { - boolean accepted = HeaderUtil.validateSignature(header.getSignature()); - if (!accepted) { - throw new IllegalArgumentException("Invalid Signature:" + header); - } - } + /** + * Deserialize the Thrift object from a byte array. + * + * @param locator The object to read into + * @param bytes The array to read from + */ + public TBase deserialize(TBaseLocator locator, byte[] bytes) throws TException { + try { + trans_.reset(bytes); + Header header = readHeader(); + validate(header); + TBase base = locator.tBaseLookup(header.getType()); + base.read(protocol_); + return base; + } finally { + trans_.clear(); + protocol_.reset(); + } + } - private Header readHeader() throws TException { - byte signature = protocol_.readByte(); - byte version = protocol_.readByte(); - short type = protocol_.readI16(); - return new Header(signature, version, type); - } + private void validate(Header header) { + boolean accepted = HeaderUtils.validateSignature(header.getSignature()); + if (!accepted) { + throw new IllegalArgumentException("Invalid Signature:" + header); + } + } - /** - * Deserialize the Thrift object from a Java string, using a specified - * character set for decoding. - * - * @param base - * The object to read into - * @param data - * The string to read from - * @param charset - * Valid JVM charset - */ - // public void deserialize(TBase base, String data, String charset) throws - // TException { - // try { - // deserialize(base, data.getBytes(charset)); - // } catch (UnsupportedEncodingException uex) { - // throw new TException("JVM DOES NOT SUPPORT ENCODING: " + charset); - // } finally { - // protocol_.reset(); - // } - // } + private Header readHeader() throws TException { + byte signature = protocol_.readByte(); + byte version = protocol_.readByte(); + short type = protocol_.readI16(); + return new Header(signature, version, type); + } - /** - * Deserialize the Thrift object from a Java string, using the default JVM - * charset encoding. - * - * @param base - * The object to read into - * @param data - * The string to read from - */ - // public void fromString(TBase base, String data) throws TException { - // deserialize(base, data.getBytes()); - // } + /** + * Deserialize the Thrift object from a Java string, using a specified + * character set for decoding. + * + * @param base + * The object to read into + * @param data + * The string to read from + * @param charset + * Valid JVM charset + */ + // public void deserialize(TBase base, String data, String charset) throws + // TException { + // try { + // deserialize(base, data.getBytes(charset)); + // } catch (UnsupportedEncodingException uex) { + // throw new TException("JVM DOES NOT SUPPORT ENCODING: " + charset); + // } finally { + // protocol_.reset(); + // } + // } + + /** + * Deserialize the Thrift object from a Java string, using the default JVM + * charset encoding. + * + * @param base + * The object to read into + * @param data + * The string to read from + */ + // public void fromString(TBase base, String data) throws TException { + // deserialize(base, data.getBytes()); + // } } diff --git a/src/main/java/com/profiler/common/util/HeaderUtil.java b/src/main/java/com/profiler/common/util/HeaderUtils.java similarity index 56% rename from src/main/java/com/profiler/common/util/HeaderUtil.java rename to src/main/java/com/profiler/common/util/HeaderUtils.java index b96831cfc..f4e519698 100644 --- a/src/main/java/com/profiler/common/util/HeaderUtil.java +++ b/src/main/java/com/profiler/common/util/HeaderUtils.java @@ -3,8 +3,8 @@ package com.profiler.common.util; import com.profiler.common.dto.Header; -public class HeaderUtil { - public static boolean validateSignature(byte signature){ +public class HeaderUtils { + public static boolean validateSignature(byte signature) { return Header.SIGNATURE == signature; } } diff --git a/src/main/java/com/profiler/common/util/RowKeyUtils.java b/src/main/java/com/profiler/common/util/RowKeyUtils.java new file mode 100644 index 000000000..b0eb074b4 --- /dev/null +++ b/src/main/java/com/profiler/common/util/RowKeyUtils.java @@ -0,0 +1,24 @@ +package com.profiler.common.util; + +import org.apache.hadoop.hbase.util.Bytes; + +/** + * + */ +public class RowKeyUtils { + + public static int LONG_BYTE_LENGTH = 8; + + public static byte[] concatFixedByteAndLong(byte[] fixedBytes, int maxFixedLength, long l) { + if (fixedBytes == null) { + throw new IllegalArgumentException("fixedBytes must not null"); + } + if (fixedBytes.length > maxFixedLength) { + throw new IllegalArgumentException("fixedBytes.length too big. length:" + fixedBytes.length); + } + byte[] rowKey = new byte[maxFixedLength + LONG_BYTE_LENGTH]; + Bytes.putBytes(rowKey, 0, fixedBytes, 0, fixedBytes.length); + BytesUtils.writeLong(l, rowKey, maxFixedLength); + return rowKey; + } +} diff --git a/src/main/java/com/profiler/common/util/SpanUtils.java b/src/main/java/com/profiler/common/util/SpanUtils.java index daf0624e7..eaeb8a4b3 100644 --- a/src/main/java/com/profiler/common/util/SpanUtils.java +++ b/src/main/java/com/profiler/common/util/SpanUtils.java @@ -1,11 +1,27 @@ package com.profiler.common.util; import com.profiler.common.dto.thrift.Span; +import org.apache.hadoop.hbase.util.Bytes; public class SpanUtils { + public static final int AGENT_NAME_LIMIT = 24; + + public static byte[] getTraceIndexRowKey(Span span) { - return BytesUtils.add(span.getAgentId(), span.getTimestamp()); + return getTraceIndexRowKey(span.getAgentId(), span.getTimestamp()); + } + + public static byte[] getTraceIndexRowKey(byte[] agentId, long time) { + return RowKeyUtils.concatFixedByteAndLong(agentId, AGENT_NAME_LIMIT, time); + } + + public static byte[] getTraceIndexRowKey(String agentId, long time) { + if (agentId == null) { + throw new IllegalArgumentException("agentId must not null"); + } + byte[] bAgentId = BytesUtils.getBytes(agentId); + return getTraceIndexRowKey(bAgentId, time); } public static byte[] getTraceId(Span span) { diff --git a/src/test/java/com/profiler/common/hbase/HBaseClientTest.java b/src/test/java/com/profiler/common/hbase/HBaseClientTest.java index fbd198ac9..5b197e6e7 100644 --- a/src/test/java/com/profiler/common/hbase/HBaseClientTest.java +++ b/src/test/java/com/profiler/common/hbase/HBaseClientTest.java @@ -17,65 +17,64 @@ import java.util.*; public class HBaseClientTest { - private static final String TABLE_NAME = "TEST_TABLE"; - private static final String COLUMN_FAMILY = "COLUMN_FAMILY"; - private static HBaseClient client; + private static final String TABLE_NAME = "TEST_TABLE"; + private static final String COLUMN_FAMILY = "COLUMN_FAMILY"; + private static HBaseClient client; - @BeforeClass - public static void init() throws IOException { + private static HBaseAdminTemplate adminTemplate; + + @BeforeClass + public static void init() throws IOException { URL resource = HBaseClientTest.class.getClassLoader().getResource("test-hbase.properties"); Properties properties = PropertyUtils.readProperties(resource.getPath()); client = new HBaseClient(properties); - if (client.isTableExists(TABLE_NAME)) { - client.dropTable(TABLE_NAME); - } - Assert.assertNotNull(client); - } + adminTemplate = new HBaseAdminTemplate(client.getConfiguration()); + adminTemplate.dropTableIfExist(TABLE_NAME); - @AfterClass - public static void destroy() { - if (client.isTableExists(TABLE_NAME)) { - client.dropTable(TABLE_NAME); - } - client.close(); - } + Assert.assertNotNull(client); + } - @Test - public void manageTable() { - client.createTable(new HTableDescriptor(TABLE_NAME)); + @AfterClass + public static void destroy() { + adminTemplate.dropTableIfExist(TABLE_NAME); + client.close(); + } - Assert.assertTrue("Table is not exists", client.isTableExists(TABLE_NAME)); + @Test + public void manageTable() { + adminTemplate.createTableIfNotExist(new HTableDescriptor(TABLE_NAME)); - client.dropTable(TABLE_NAME); - client.isTableExists(TABLE_NAME); + Assert.assertTrue("Table is not exists", adminTemplate.tableExists(TABLE_NAME)); + adminTemplate.dropTableIfExist(TABLE_NAME); + adminTemplate.tableExists(TABLE_NAME); - Assert.assertFalse("Table is not dropped", client.isTableExists(TABLE_NAME)); - } + Assert.assertFalse("Table is not dropped", adminTemplate.tableExists(TABLE_NAME)); + } - @Test - public void insertRow() { - HTableDescriptor desc = new HTableDescriptor(TABLE_NAME); - desc.addFamily(new HColumnDescriptor(COLUMN_FAMILY)); - client.createTable(desc); - Assert.assertTrue("Table is not exists", client.isTableExists(TABLE_NAME)); + @Test + public void insertRow() { + HTableDescriptor desc = new HTableDescriptor(TABLE_NAME); + desc.addFamily(new HColumnDescriptor(COLUMN_FAMILY)); + adminTemplate.createTableIfNotExist(desc); + Assert.assertTrue("Table is not exists", adminTemplate.tableExists(TABLE_NAME)); - List putList = new ArrayList(); - for (int i = 0; i < 10; i++) { - Put put = new Put(Bytes.toBytes(i + "row")); - put.add(Bytes.toBytes(COLUMN_FAMILY), Bytes.toBytes("qual1"), Bytes.toBytes("val1")); - put.add(Bytes.toBytes(COLUMN_FAMILY), Bytes.toBytes("qual2"), Bytes.toBytes("val2")); - putList.add(put); - } - client.insert(TABLE_NAME, putList); + List putList = new ArrayList(); + for (int i = 0; i < 10; i++) { + Put put = new Put(Bytes.toBytes(i + "row")); + put.add(Bytes.toBytes(COLUMN_FAMILY), Bytes.toBytes("qual1"), Bytes.toBytes("val1")); + put.add(Bytes.toBytes(COLUMN_FAMILY), Bytes.toBytes("qual2"), Bytes.toBytes("val2")); + putList.add(put); + } + client.insert(TABLE_NAME, putList); - List list = new ArrayList(); - list.add(new HbaseColumn(COLUMN_FAMILY, "qual1")); - list.add(new HbaseColumn(COLUMN_FAMILY, "qual2")); - Iterator> result = client.getHBaseData(new HBaseQuery(TABLE_NAME, Bytes.toBytes("0"), Bytes.toBytes("9"), list)); + List list = new ArrayList(); + list.add(new HbaseColumn(COLUMN_FAMILY, "qual1")); + list.add(new HbaseColumn(COLUMN_FAMILY, "qual2")); + Iterator> result = client.getHBaseData(new HBaseQuery(TABLE_NAME, Bytes.toBytes("0"), Bytes.toBytes("9"), list)); - while (result.hasNext()) { - Map next = result.next(); - System.out.println(next); - } - } + while (result.hasNext()) { + Map next = result.next(); + System.out.println(next); + } + } } diff --git a/src/test/java/com/profiler/common/util/BufferTest.java b/src/test/java/com/profiler/common/util/BufferTest.java new file mode 100644 index 000000000..4d3eea510 --- /dev/null +++ b/src/test/java/com/profiler/common/util/BufferTest.java @@ -0,0 +1,76 @@ +package com.profiler.common.util; + +import junit.framework.Assert; +import org.junit.Test; + +/** + * + */ +public class BufferTest { + @Test + public void testPutPrefixedBytes() throws Exception { + String test = "test"; + int expected = 3333; + + Buffer buffer = new Buffer(1024); + buffer.putPrefixedBytes(test.getBytes("UTF-8")); + + buffer.put(expected); + byte[] buffer1 = buffer.getBuffer(); + + Buffer actual = new Buffer(buffer1); + String s = actual.readPrefixedString(); + Assert.assertEquals(test, s); + + int i = actual.readInt(); + Assert.assertEquals(expected, i); + + + } + + @Test + public void testReadByte() throws Exception { + + } + + @Test + public void testReadBoolean() throws Exception { + + } + + @Test + public void testReadInt() throws Exception { + + } + + @Test + public void testReadLong() throws Exception { + + } + + @Test + public void testReadPrefixedBytes() throws Exception { + + } + + @Test + public void testReadPrefixedString() throws Exception { + + } + + @Test + public void testPut() throws Exception { + + } + + + @Test + public void testGetBuffer() throws Exception { + + } + + @Test + public void testGetOffset() throws Exception { + + } +} diff --git a/src/test/java/com/profiler/common/util/SpanUtilsTest.java b/src/test/java/com/profiler/common/util/SpanUtilsTest.java new file mode 100644 index 000000000..a5955a501 --- /dev/null +++ b/src/test/java/com/profiler/common/util/SpanUtilsTest.java @@ -0,0 +1,58 @@ +package com.profiler.common.util; + +import com.profiler.common.dto.thrift.Span; +import junit.framework.Assert; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.Test; + +/** + * + */ +public class SpanUtilsTest { + @Test + public void testGetTraceIndexRowKey1() throws Exception { + String agentId = "test"; + long time = System.currentTimeMillis(); + check(agentId, time); + } + + @Test + public void testGetTraceIndexRowKey2() throws Exception { + String agentId = ""; + for (int i = 0; i < SpanUtils.AGENT_NAME_LIMIT; i++) { + agentId += "1"; + } + + long time = System.currentTimeMillis(); + check(agentId, time); + } + + @Test + public void testGetTraceIndexRowKey3() throws Exception { + String agentId = ""; + for (int i = 0; i < SpanUtils.AGENT_NAME_LIMIT + 1; i++) { + agentId += "1"; + } + + long time = System.currentTimeMillis(); + try { + check(agentId, time); + Assert.fail(); + } catch (Exception e) { + } + } + + private void check(String agentId0, long l1) { + Span span = new Span(); + span.setAgentId(agentId0); + span.setTimestamp(l1); + + byte[] traceIndexRowKey = SpanUtils.getTraceIndexRowKey(span); + + String agentId = Bytes.toString(traceIndexRowKey, 0, agentId0.length()); + Assert.assertEquals(agentId0, agentId); + + long time = Bytes.toLong(traceIndexRowKey, SpanUtils.AGENT_NAME_LIMIT); + Assert.assertEquals(time, l1); + } +} diff --git a/src/test/resources/test-hbase.properties b/src/test/resources/test-hbase.properties index 293ae9edd..e19d6bb6b 100644 --- a/src/test/resources/test-hbase.properties +++ b/src/test/resources/test-hbase.properties @@ -2,4 +2,6 @@ #hbase.client.host=10.64.84.188 hbase.client.host=10.25.131.38 -hbase.client.port=2181 \ No newline at end of file +hbase.client.port=2181 + +hbase.htable.threads.max=32 \ No newline at end of file