Merge branch 'master' of https://github.com/jd-bdp/hydra

hhhcommon · May 27, 2013 · d9c811c · d9c811c
2 parents d774bab + 43be865
commit d9c811c
Show file tree

Hide file tree

Showing 58 changed files with 227 additions and 549 deletions.
diff --git a/docs/table-hbase/initTable b/docs/table-hbase/initTable
@@ -0,0 +1,3 @@
+create 'trace',{NAME=>'span',VERSIONS=>1},{SPLITS=>['01','02','03','04','05','06','07','08','09','10','11','12','13','14','15','16','17','18','19','20','21','22','23','24','25','26']}
+create 'duration_index',{NAME=>'trace',VERSIONS=>1},{SPLITS=>['01','02','03','04','05','06','07','08','09','10','11','12','13','14','15','16','17','18','19','20','21','22','23','24','25','26']}
+create 'annotation_index',{NAME=>'trace',VERSIONS=>1},{SPLITS=>['01','02','03','04','05','06','07','08','09','10','11','12','13','14','15','16','17','18','19','20','21','22','23','24','25','26']}
diff --git a/...llector-service/src/main/java/com/jd/bdp/hydra/collector/service/CollectorSerService.java b/...llector-service/src/main/java/com/jd/bdp/hydra/collector/service/CollectorSerService.java
@@ -39,7 +39,7 @@ public void setTopic(String topic) {
         this.topic = topic;
     }
 
-    class HbaseConsumer implements MessageListener{
+    class HbaseConsumer implements MessageListener {
         @Override
         public void recieveMessages(Message message) {
             persistent(message);
@@ -52,24 +52,27 @@ public Executor getExecutor() {
     }
 
 
-    public void subscribe()throws Exception{
-        consumer.subscribe(topic,1024*1024,new HbaseConsumer()).completeSubscribe();
+    public void subscribe() throws Exception {
+        consumer.subscribe(topic, 1024 * 1024, new HbaseConsumer()).completeSubscribe();
     }
 
-    public void persistent(Message message){
-        List<Span> spanList = new ArrayList<Span>();
-        try{
-            spanList = (List)PB.parsePBBytes(message.getData());
-        }catch (Exception e){
-           log.error(e.getMessage());
+    public void persistent(Message message) {
+        List<Span> spanList;
+        try {
+            spanList = (List) PB.parsePBBytes(message.getData());
+        } catch (Exception e) {
+            log.error(e.getMessage());
+            return;
         }
-        try{
-          for(Span s : spanList){
-              insertService.addSpan(s);
-              insertService.addAnnotation(s);
-              insertService.addTrace(s);
-          }
-        }catch (Exception e){
+        try {
+            if (spanList != null) {
+                for (Span s : spanList) {
+                    insertService.addSpan(s);
+                    insertService.addAnnotation(s);
+                    insertService.addTrace(s);
+                }
+            }
+        } catch (Exception e) {
             e.printStackTrace();
             log.error(e.getMessage());
         }

diff --git a/...a-hbase/src/main/resources/hbase-site.xml → ...service/src/main/resources/hbase-site.xml b/...a-hbase/src/main/resources/hbase-site.xml → ...service/src/main/resources/hbase-site.xml
diff --git a/modules/hydra-collector-service/src/main/resources/hydra-collector-service.xml b/modules/hydra-collector-service/src/main/resources/hydra-collector-service.xml
@@ -29,7 +29,7 @@
 
     <context:property-placeholder location="classpath:conf.properties"/>
 
-    <bean id="collectorSerService" class="com.jd.bdp.hydra.collector.service.CollectorSerService" >
+    <bean id="collectorService" class="com.jd.bdp.hydra.collector.service.CollectorSerService" >
         <property name="topic" value="${metaq.topic}"></property>
     </bean>
 

diff --git a/...a-manager-db/src/main/java/com/jd/bdp/hydra/mysql/persistent/dao/impl/SeedMapperImpl.java b/...a-manager-db/src/main/java/com/jd/bdp/hydra/mysql/persistent/dao/impl/SeedMapperImpl.java
@@ -29,6 +29,17 @@
 public class SeedMapperImpl implements SeedMapper {
     private SqlSessionTemplate sqlSession;
 
+    /**
+     * 验证该种子是否已经派发过
+     * 内部逻辑：
+     * 1：找最大索引Id行，获取种子value值 为V（有记录行）
+     * 2：若没有记录行，则返回false （无记录行）
+     * 2：若value>V,则返回false，否则返回true
+     * @param value 种子值
+     * @return 是否派发过
+     * @throws Exception 持久层异常
+     * @since 2.0
+     */
     @Override
     public boolean hasSeed(Integer value) throws Exception {
         Integer result_count = null;
@@ -41,6 +52,26 @@ public boolean hasSeed(Integer value) throws Exception {
         }
         return hasSeed;
     }
+    /**
+     * 找到种子标示值
+     * 内部逻辑：
+     * 1：找最大索引Id行，获取种子value值 为V
+     * 1.1返回记录行（有记录）
+     * 1.2返回null （无记录）
+     * @return 种子标识
+     * @throws Exception 持久层异常
+     * @since 2.0
+     */
+    @Override
+    public SeedData findTheSeed() throws Exception {
+        SeedData seedData = null;
+        try {
+            seedData = (SeedData) sqlSession.selectOne("findTheSeed");
+        } catch (Exception e) {
+            throw e;
+        }
+        return seedData;
+    }
 
     @Override
     public Integer getMaxSeedValue() throws Exception {
@@ -52,8 +83,6 @@ public Integer getMaxSeedValue() throws Exception {
         }
         return max_value;
     }
-
-    //TODO
     @Override
     public void addSeed(SeedData servicePara) throws Exception {
         Integer id=null;
@@ -62,12 +91,12 @@ public void addSeed(SeedData servicePara) throws Exception {
             flag=sqlSession.insert("addSeed",servicePara)>0?true:false;
             id=servicePara.getId();
         } catch (Exception e) {
-            //e.printStackTrace();
             flag=false;
             throw e;
         }
     }
 
+    @Deprecated
     @Override
     public void deleteSeed(SeedData servicePara) throws Exception {
         boolean flag = false;
@@ -90,6 +119,7 @@ public void updateSeed(SeedData servicePara) throws Exception {
         }
     }
 
+    @Deprecated
     @Override
     public SeedData getOneSeed(Integer id) throws Exception {
         SeedData seedData = null;
@@ -101,16 +131,7 @@ public SeedData getOneSeed(Integer id) throws Exception {
         return seedData;
     }
 
-    @Override
-    public SeedData findTheSeed() throws Exception {
-        SeedData seedData = null;
-        try {
-            seedData = (SeedData) sqlSession.selectOne("findTheSeed");
-        } catch (Exception e) {
-            throw e;
-        }
-        return seedData;
-    }
+
 
     //getter and setter
     public SqlSessionTemplate getSqlSession() {

diff --git a/modules/hydra-manager/src/main/resources/mysql.properties b/modules/hydra-manager/src/main/resources/mysql.properties
@@ -1,4 +1,3 @@
-jdbc.driverClassName=com.mysql.jdbc.Driver
-jdbc.url=jdbc:mysql://192.168.200.113:3306/hydra-config-test-pressure
+jdbc.url=jdbc:mysql://192.168.228.81:3306/hydra-config-test-pressure
 jdbc.username=root
 jdbc.password=root
diff --git a/...store/hydra-hbase/src/main/java/com/jd/bdp/hydra/hbase/service/impl/QueryServiceImpl.java b/...store/hydra-hbase/src/main/java/com/jd/bdp/hydra/hbase/service/impl/QueryServiceImpl.java
@@ -171,7 +171,6 @@ private JSONObject assembleTrace(List<KeyValue> list) {
                         spanAleadyExist.put("durationServer", getDurationServer(content));
                     }
                     ((JSONArray) spanAleadyExist.get("annotations")).addAll((JSONArray) content.get("annotations"));
-                    handleTheMinAndMaxTimestamp(trace, spanAleadyExist);
                 } else {
                     spanAleadyExist = content;
                     if (isClientSpan(kv)) {
@@ -208,28 +207,7 @@ private JSONObject assembleTrace(List<KeyValue> list) {
         return trace;
     }
 
-    private void handleTheMinAndMaxTimestamp(JSONObject trace, JSONObject span) {
-        for (Object obj : span.getJSONArray("annotations")) {
-            long timestamp = Long.parseLong(((JSONObject) obj).get("timestamp").toString());
-            if (trace.containsKey("minTimestamp")) {
-                long min = Long.parseLong(trace.get("minTimestamp").toString());
-                if (min > timestamp) {
-                    trace.put("minTimestamp", timestamp);
-                }
-            } else {
-                trace.put("minTimestamp", timestamp);
-            }
-            if (trace.containsKey("maxTimestamp")) {
-                long max = Long.parseLong(trace.get("maxTimestamp").toString());
-                if (max < timestamp) {
-                    trace.put("maxTimestamp", timestamp);
-                }
-            } else {
-                trace.put("maxTimestamp", timestamp);
-            }
-        }
-    }
-
+    //这里判断如果某个span没有收集全4个annotation，则判定为不可用，页面不展示图
     private boolean isSpanAvailable(JSONObject span) {
         return span.getJSONArray("annotations").size() == 4;
     }
@@ -282,24 +260,24 @@ private Long getDurationClient(JSONObject content) {
     }
 
 
-    public void setOneItem(String tableName, String familyColumnName, String rowkey, String columnName, byte[] valueParm) {
-        HTableInterface table = POOL.getTable(tableName);
-        table.setAutoFlush(true);//自动提交
-        try {
-            Put put = new Put(Bytes.toBytes(rowkey));
-            put.add(Bytes.toBytes(familyColumnName), Bytes.toBytes(columnName), valueParm);
-            table.put(put);
-//            table.flushCommits();//手动提交，最好每次close之前手动提交...
-        } catch (IOException e) {
-            e.printStackTrace();
-        } finally {
-            try {
-                table.close();
-            } catch (IOException e) {
-                e.printStackTrace();
-            }
-        }
-    }
+//    public void setOneItem(String tableName, String familyColumnName, String rowkey, String columnName, byte[] valueParm) {
+//        HTableInterface table = POOL.getTable(tableName);
+//        table.setAutoFlush(true);//自动提交
+//        try {
+//            Put put = new Put(Bytes.toBytes(rowkey));
+//            put.add(Bytes.toBytes(familyColumnName), Bytes.toBytes(columnName), valueParm);
+//            table.put(put);
+////            table.flushCommits();//手动提交，最好每次close之前手动提交...
+//        } catch (IOException e) {
+//            e.printStackTrace();
+//        } finally {
+//            try {
+//                table.close();
+//            } catch (IOException e) {
+//                e.printStackTrace();
+//            }
+//        }
+//    }
 
 //    /**
 //     * 删除指定表名的rowKey下某时间戳的数据。

diff --git a/modules/hydra-store/hydra-hbase/src/test/resources/hbase-site.xml b/modules/hydra-store/hydra-hbase/src/test/resources/hbase-site.xml
@@ -0,0 +1,10 @@
+<configuration>
+    <property>
+        <name>hbase.client.retries.number</name>
+        <value>3</value>
+    </property>
+    <property>
+        <name>hbase.zookeeper.quorum</name>
+        <value>boss,emp1,emp2</value>
+    </property>
+</configuration>
diff --git a/...store/hydra-mysql/src/main/java/com/jd/bdp/hydra/mysql/service/impl/QueryServiceImpl.java b/...store/hydra-mysql/src/main/java/com/jd/bdp/hydra/mysql/service/impl/QueryServiceImpl.java
@@ -77,35 +77,12 @@ private JSONObject assembleTrace(List<Span> spans, List<Absannotation> annotatio
                 }
             }
             setSpanDuration(mySpan);
-            handleTheMinAndMaxTimestamp(trace, mySpan);
             isAvailable = isAvailable && isSpanAvailable(mySpan);
         }
         trace.put("available", isAvailable);
         return trace;
     }
 
-    private void handleTheMinAndMaxTimestamp(JSONObject trace, JSONObject span) {
-        for (Object obj : span.getJSONArray("annotations")) {
-            long timestamp = Long.parseLong(((JSONObject) obj).get("timestamp").toString());
-            if (trace.containsKey("minTimestamp")) {
-                long min = Long.parseLong(trace.get("minTimestamp").toString());
-                if (min > timestamp) {
-                    trace.put("minTimestamp", timestamp);
-                }
-            } else {
-                trace.put("minTimestamp", timestamp);
-            }
-            if (trace.containsKey("maxTimestamp")) {
-                long max = Long.parseLong(trace.get("maxTimestamp").toString());
-                if (max < timestamp) {
-                    trace.put("maxTimestamp", timestamp);
-                }
-            } else {
-                trace.put("maxTimestamp", timestamp);
-            }
-        }
-    }
-
     public void setSpanDuration(JSONObject spanDuration) {
         JSONArray serverAnns = spanDuration.getJSONArray("annotations");
         Long sr = null;
@@ -134,6 +111,7 @@ public void setSpanDuration(JSONObject spanDuration) {
         }
     }
 
+    //这里判断如果某个span没有收集全4个annotation，则判定为不可用，页面不展示图
     private boolean isSpanAvailable(JSONObject span) {
         return span.getJSONArray("annotations").size() == 4;
     }

diff --git a/...integration/src/test/java/com/jd/bdp/hydra/tempTest/startCollect/StartCollectorMysql.java b/...integration/src/test/java/com/jd/bdp/hydra/tempTest/startCollect/StartCollectorMysql.java
diff --git a/...a-test-integration/src/test/java/com/jd/bdp/hydra/tempTest/startManager/StartManager.java b/...a-test-integration/src/test/java/com/jd/bdp/hydra/tempTest/startManager/StartManager.java
diff --git a/...a-test-integration/src/test/resources/hydra-collector/hydra-collector-integrationTest.xml b/...a-test-integration/src/test/resources/hydra-collector/hydra-collector-integrationTest.xml