基础功能全部跑通！

.gitignore

@@ -24,5 +24,5 @@ target/
 *.patch
 *.tmp
 .logs
-
+logs/
 

modules/hydra-collector-service/src/main/java/com/jd/bdp/hydra/collector/service/Bootstrap.java

@@ -9,22 +9,15 @@
  * Time: 上午11:01
  */
 public class Bootstrap {
-    private CollectorService collectorService;
-    private final String topic = "hydra_test";
 
     public static void main(String[] strings){
         try {
             ApplicationContext context = new ClassPathXmlApplicationContext("hydra-collector-service.xml");
-            Bootstrap bootstrap = new Bootstrap();
-            bootstrap.collectorService = (CollectorService)context.getBean("collectorService");
-            bootstrap.start();
+            CollectorSerService collectorSerService = (CollectorSerService)context.getBean("collectorService");
+            collectorSerService.subscribe();
         } catch (Exception e) {
             e.printStackTrace();
         }
     }
 
-    public void start()throws Exception{
-        collectorService.subscribe(topic);
-    }
-
 }

modules/hydra-collector-service/src/main/java/com/jd/bdp/hydra/collector/service/CollectorService.java → modules/hydra-collector-service/src/main/java/com/jd/bdp/hydra/collector/service/CollectorSerService.java

@@ -21,8 +21,9 @@
  * Date: 13-4-17
  * Time: 下午5:03
  */
-public class CollectorService {
-    private static final Logger log = LoggerFactory.getLogger(CollectorService.class);
+public class CollectorSerService {
+    private static final Logger log = LoggerFactory.getLogger(CollectorSerService.class);
+    private String topic;
     private MessageConsumer consumer;
     private HbaseService hbaseService;
     private ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
@@ -36,6 +37,10 @@ public void setHbaseService(HbaseService hbaseService) {
         this.hbaseService = hbaseService;
     }
 
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
     class HbaseConsumer implements MessageListener{
         @Override
         public void recieveMessages(Message message) {
@@ -49,15 +54,14 @@ public Executor getExecutor() {
     }
 
 
-    public void subscribe(String topic)throws Exception{
+    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());
-            log.info("成功接受message");
         }catch (Exception e){
            log.error(e.getMessage());
         }

modules/hydra-collector-service/src/main/resources/hydra-collector-service.xml

@@ -22,7 +22,10 @@
         default-autowire="byName">
 
     <bean id="hbaseService" class="com.jd.bdp.hydra.hbase.service.impl.HbaseServiceImpl"/>
-    <bean id="collectorService" class="com.jd.bdp.hydra.collector.service.CollectorService" />
+    <bean id="collectorSerService" class="com.jd.bdp.hydra.collector.service.CollectorSerService" >
+        <property name="topic" value="hydra_test"></property>
+    </bean>
+
 
     <bean id="sessionFactory" class="com.taobao.metamorphosis.client.MetaMessageSessionFactory">
         <constructor-arg>

modules/hydra-collector/src/main/resources/metaq.prop

@@ -1,3 +1,3 @@
-metaq.zk=192.168.227.83:2181,192.168.227.86:2181,192.168.228.85:2181
+metaq.zk=192.168.227.83:2181
 metaq.topic=hydra_test
 metaq.zk.root=/meta

modules/hydra-example/hydra-example-benchmark/src/main/java/com/jd/bdp/hydra/benchmark/startCollectorService/StartCollectorService.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright jd
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+package com.jd.bdp.hydra.benchmark.startCollectorService;
+
+import com.jd.bdp.hydra.agent.CollectorService;
+import com.jd.bdp.hydra.collector.service.CollectorSerService;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+/**
+ * User: xiangkui
+ * Date: 13-4-23
+ * Time: 下午6:19
+ */
+public class StartCollectorService {
+
+    public static void main(String[] args) throws Exception {
+        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{
+                "classpath*:hydra-collector-service.xml"
+        });
+        CollectorSerService collectorSerService = (CollectorSerService)context.getBean("collectorSerService");
+        collectorSerService.subscribe();
+        Thread.sleep(999999999);
+    }
+}

modules/hydra-example/hydra-example-benchmark/src/main/java/com/jd/bdp/hydra/benchmark/startTrigger/StartTrigger.java

@@ -17,7 +17,7 @@ public static void main(String[] args)  {
             context.start();
             Trigger trigger=(Trigger)context.getBean("trigger");
             // 每隔3s触发一次调用
-            trigger.startWorkWithSleep(5,500);
+            trigger.startWorkWithSleep(20,500);
             Thread.sleep(999999999);
         } catch (Exception e) {
             e.printStackTrace();

modules/hydra-example/pom.xml

@@ -34,6 +34,11 @@
             <groupId>com.jd.bdp</groupId>
             <artifactId>hydra-collector</artifactId>
         </dependency>
+        <!-- 测试hydra-collector-service -->
+        <dependency>
+            <groupId>com.jd.bdp</groupId>
+            <artifactId>hydra-collector-service</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>org.springframework</groupId>

modules/hydra-hbase/src/main/java/com/jd/bdp/hydra/hbase/service/impl/HbaseServiceImpl.java

@@ -144,34 +144,43 @@ public void annotationIndex(Span span) {
 
 
     public void durationIndex(Span span) {
-        List<Annotation> alist = span.getAnnotations();
-        Annotation cs = getCsAnnotation(alist);
-        Annotation cr = getCrAnnotation(alist);
-        if (cs != null) {
-            long duration = cs.getTimestamp() - cr.getTimestamp();
-            String rowkey = cs.getHost().getServiceName() + ":" + cs.getTimestamp();
-            Put put = new Put(rowkey.getBytes());
-            //rowkey:serviceId:csTime
-            //每列的timestamp为duration
-            //每列列名为traceId，值为1（用来区分1ms内的跟踪）
-            put.add(duration_index_family_column.getBytes(), long2ByteArray(span.getTraceId()), duration,  "1".getBytes());
-            HTableInterface htable = null;
-            try {
-                htable = POOL.getTable(duration_index);
-                htable.put(put);
-            } catch (IOException e) {
-                e.printStackTrace();
-            }finally {
-                if(htable != null){
-                    try {
-                        htable.close();
-                    } catch (IOException e) {
-                        e.printStackTrace();
+        if (isRootSpan(span)) {
+            List<Annotation> alist = span.getAnnotations();
+            Annotation cs = getCsAnnotation(alist);
+            Annotation cr = getCrAnnotation(alist);
+            if (cs != null) {
+                long duration = cr.getTimestamp() - cs.getTimestamp();
+                String rowkey = cs.getHost().getServiceName() + ":" + cs.getTimestamp();
+                Put put = new Put(rowkey.getBytes());
+                //rowkey:serviceId:csTime
+                //每列的timestamp为duration
+                //每列列名为traceId，值为1（用来区分1ms内的跟踪）
+                put.add(duration_index_family_column.getBytes(), long2ByteArray(span.getTraceId()), duration,  "1".getBytes());
+                HTableInterface htable = null;
+                try {
+                    htable = POOL.getTable(duration_index);
+                    System.out.println(1);
+                    htable.put(put);
+                } catch (IOException e) {
+                    e.printStackTrace();
+                }catch (Exception e) {
+                    e.printStackTrace();
+                }finally {
+                    if(htable != null){
+                        try {
+                            htable.close();
+                        } catch (IOException e) {
+                            e.printStackTrace();
+                        }
                     }
                 }
             }
         }
     }
 
+    private boolean isRootSpan(Span span) {
+        return span.getParentId() == null;
+    }
+
 
 }

modules/hydra-hbase/src/test/java/com/jd/bdp/hydra/hbase/service/HbaseServiceTest.java

@@ -42,7 +42,7 @@ public static void main(String[] args){
                 cs.setTimestamp(System.currentTimeMillis());
                 cs.setValue("cs");
                 Annotation cr = new Annotation();
-                cr.setTimestamp(System.currentTimeMillis()-(int)(Math.random() * 100));
+                cr.setTimestamp(System.currentTimeMillis()+(int)(Math.random() * 100));
                 cr.setValue("cr");
                 cr.setHost(endpoint);
                 span.setId(new Long(i));
@@ -68,10 +68,10 @@ public static void main(String[] args){
                 Annotation sr = new Annotation();
                 sr.setHost(endpoint);
                 sr.setTimestamp(System.currentTimeMillis());
-                sr.setValue("ss");
+                sr.setValue("sr");
                 Annotation ss = new Annotation();
-                ss.setTimestamp(System.currentTimeMillis()-(int)(Math.random() * 100));
-                ss.setValue("sr");
+                ss.setTimestamp(System.currentTimeMillis()+(int)(Math.random() * 100));
+                ss.setValue("ss");
                 ss.setHost(endpoint);
                 span.setParentId(new Long(i+1));
                 span.setId(new Long(i));

modules/hydra-web/src/main/java/com/jd/bdp/hydra/rest/TraceController.java

@@ -26,11 +26,6 @@ public class TraceController {
     @ResponseBody
     public JSONArray getTraces(@PathVariable String serviceId, @PathVariable long startTime, @PathVariable int durationMin, @PathVariable int durationMax, @PathVariable int sum) {
         try {
-            serviceId = "22001";
-            startTime = 1366614281166L;
-            durationMin = 20;
-            durationMax = 80;
-            sum = 500;
             return queryService.getTracesByDuration(serviceId, startTime, sum, durationMin, durationMax);
         }catch (Exception e){
             e.printStackTrace();

modules/hydra-web/src/main/webapp/statics/js/service/sequence-service.js

@@ -23,7 +23,6 @@ angular.module('hydra.services.sequence', [])
                 var spanIndex = {index: 0}
                 getMySpan(span, spanIndex);
 
-
                 function getMySpan(span, spanIndex) {
                     var anMap = {};
                     for (var i in span.annotations) {