pinpoint-apm#1734 Extract Seralization in XxxBo

 - refactoring SpanBo

collector/src/main/java/com/navercorp/pinpoint/collector/dao/hbase/HbaseApplicationTraceIndexDao.java

@@ -19,7 +19,7 @@
 import static com.navercorp.pinpoint.common.hbase.HBaseTables.*;
 
 import com.navercorp.pinpoint.collector.dao.ApplicationTraceIndexDao;
-import com.navercorp.pinpoint.collector.util.AcceptedTimeService;
+import com.navercorp.pinpoint.common.server.util.AcceptedTimeService;
 import com.navercorp.pinpoint.common.buffer.AutomaticBuffer;
 import com.navercorp.pinpoint.common.buffer.Buffer;
 import com.navercorp.pinpoint.common.hbase.HbaseOperations2;

collector/src/main/java/com/navercorp/pinpoint/collector/dao/hbase/HbaseHostApplicationMapDao.java

@@ -17,7 +17,7 @@
 package com.navercorp.pinpoint.collector.dao.hbase;
 
 import com.navercorp.pinpoint.collector.dao.HostApplicationMapDao;
-import com.navercorp.pinpoint.collector.util.AcceptedTimeService;
+import com.navercorp.pinpoint.common.server.util.AcceptedTimeService;
 import com.navercorp.pinpoint.collector.util.AtomicLongUpdateMap;
 import com.navercorp.pinpoint.common.buffer.AutomaticBuffer;
 import com.navercorp.pinpoint.common.buffer.Buffer;

collector/src/main/java/com/navercorp/pinpoint/collector/dao/hbase/HbaseMapResponseTimeDao.java

@@ -18,7 +18,7 @@
 
 import com.navercorp.pinpoint.collector.dao.MapResponseTimeDao;
 import com.navercorp.pinpoint.collector.dao.hbase.statistics.*;
-import com.navercorp.pinpoint.collector.util.AcceptedTimeService;
+import com.navercorp.pinpoint.common.server.util.AcceptedTimeService;
 import com.navercorp.pinpoint.collector.util.ConcurrentCounterMap;
 import com.navercorp.pinpoint.common.hbase.HbaseOperations2;
 import com.navercorp.pinpoint.common.trace.ServiceType;

collector/src/main/java/com/navercorp/pinpoint/collector/dao/hbase/HbaseMapStatisticsCalleeDao.java

@@ -20,7 +20,7 @@
 
 import com.navercorp.pinpoint.collector.dao.MapStatisticsCalleeDao;
 import com.navercorp.pinpoint.collector.dao.hbase.statistics.*;
-import com.navercorp.pinpoint.collector.util.AcceptedTimeService;
+import com.navercorp.pinpoint.common.server.util.AcceptedTimeService;
 import com.navercorp.pinpoint.collector.util.ConcurrentCounterMap;
 import com.navercorp.pinpoint.common.hbase.HbaseOperations2;
 import com.navercorp.pinpoint.common.trace.ServiceType;

collector/src/main/java/com/navercorp/pinpoint/collector/dao/hbase/HbaseMapStatisticsCallerDao.java

@@ -20,7 +20,7 @@
 
 import com.navercorp.pinpoint.collector.dao.MapStatisticsCallerDao;
 import com.navercorp.pinpoint.collector.dao.hbase.statistics.*;
-import com.navercorp.pinpoint.collector.util.AcceptedTimeService;
+import com.navercorp.pinpoint.common.server.util.AcceptedTimeService;
 import com.navercorp.pinpoint.collector.util.ConcurrentCounterMap;
 import com.navercorp.pinpoint.common.hbase.HbaseOperations2;
 import com.navercorp.pinpoint.common.trace.ServiceType;

collector/src/main/java/com/navercorp/pinpoint/collector/dao/hbase/HbaseTraceDao.java

@@ -18,7 +18,8 @@
 
 import com.navercorp.pinpoint.collector.dao.TracesDao;
 import com.navercorp.pinpoint.collector.dao.hbase.filter.SpanEventFilter;
-import com.navercorp.pinpoint.collector.util.AcceptedTimeService;
+import com.navercorp.pinpoint.common.server.bo.serializer.SpanSerializer;
+import com.navercorp.pinpoint.common.server.util.AcceptedTimeService;
 import com.navercorp.pinpoint.common.server.bo.AnnotationBo;
 import com.navercorp.pinpoint.common.server.bo.AnnotationBoList;
 import com.navercorp.pinpoint.common.server.bo.SpanBo;
@@ -63,6 +64,9 @@ public class HbaseTraceDao implements TracesDao {
     @Autowired
     private SpanEventFilter spanEventFilter;
 
+    @Autowired
+    private SpanSerializer spanSerializer;
+
     @Autowired
     @Qualifier("traceDistributor")
     private AbstractRowKeyDistributor rowKeyDistributor;
@@ -73,18 +77,16 @@ public void insert(final TSpan span) {
             throw new NullPointerException("span must not be null");
         }
 
-        SpanBo spanBo = new SpanBo(span);
+        final SpanBo spanBo = new SpanBo(span);
+
         final byte[] rowKey = getDistributeRowKey(SpanUtils.getTransactionId(span));
-        Put put = new Put(rowKey);
+        final Put put = new Put(rowKey);
 
-        byte[] spanValue = spanBo.writeValue();
+        this.spanSerializer.serialize(spanBo, put, null);
 
         // TODO  if we can identify whether the columnName is duplicated or not,
         // we can also know whether the span id is duplicated or not.
-        byte[] spanId = Bytes.toBytes(spanBo.getSpanId());
-
-        long acceptedTime = acceptedTimeService.getAcceptedTime();
-        put.addColumn(TRACES_CF_SPAN, spanId, acceptedTime, spanValue);
+        final byte[] spanId = Bytes.toBytes(spanBo.getSpanId());
 
         List<TAnnotation> annotations = span.getAnnotations();
         if (CollectionUtils.isNotEmpty(annotations)) {

collector/src/main/java/com/navercorp/pinpoint/collector/receiver/AbstractDispatchHandler.java

@@ -24,7 +24,7 @@
 import com.navercorp.pinpoint.collector.handler.Handler;
 import com.navercorp.pinpoint.collector.handler.RequestResponseHandler;
 import com.navercorp.pinpoint.collector.handler.SimpleHandler;
-import com.navercorp.pinpoint.collector.util.AcceptedTimeService;
+import com.navercorp.pinpoint.common.server.util.AcceptedTimeService;
 
 /**
  * @author emeroad

collector/src/main/resources/applicationContext-collector.xml

@@ -20,7 +20,9 @@
                         com.navercorp.pinpoint.collector.manage,
                         com.navercorp.pinpoint.collector.mapper,
                         com.navercorp.pinpoint.collector.util,
-                        com.navercorp.pinpoint.collector.config" />
+                        com.navercorp.pinpoint.collector.config,
+                        com.navercorp.pinpoint.common.server.util,
+                        com.navercorp.pinpoint.common.server.bo.serializer" />
 
     <bean id="propertyConfigurer"
           class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">

commons-server/pom.xml

@@ -11,6 +11,11 @@
     <name>pinpoint-commons-server</name>
     <packaging>jar</packaging>
 
+    <properties>
+        <jdk.version>1.7</jdk.version>
+        <jdk.home>${env.JAVA_7_HOME}</jdk.home>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>com.navercorp.pinpoint</groupId>
@@ -22,6 +27,30 @@
             <artifactId>pinpoint-commons</artifactId>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>com.navercorp.pinpoint</groupId>
+            <artifactId>pinpoint-commons-hbase</artifactId>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-core</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>commons-logging</groupId>
+                    <artifactId>commons-logging</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-context</artifactId>
+        </dependency>
 
         <!-- commons module may be used at agent so there will be problems. you should set optional -->
         <dependency>

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/SpanBo.java

@@ -70,8 +70,7 @@ public class SpanBo implements Span {
     private String exceptionMessage;
     private String exceptionClass;
 
-    private boolean hasApplicationServiceType = false;
-    private short applicationServiceType;
+    private Short applicationServiceType;
 
 
     private String remoteAddr; // optional
@@ -116,7 +115,6 @@ public SpanBo(TSpan span) {
         // FIXME (2015.03) Legacy - applicationServiceType added in v1.1.0
         // applicationServiceType is not saved for older versions where applicationServiceType does not exist.
         if (span.isSetApplicationServiceType()) {
-            this.hasApplicationServiceType = true;
             this.applicationServiceType = span.getApplicationServiceType();
         }
 
@@ -153,6 +151,11 @@ public int getVersion() {
         return version & 0xFF;
     }
 
+
+    public byte getRawVersion() {
+        return version;
+    }
+
     public void setVersion(int version) {
         if (version < 0 || version > 255) {
             throw new IllegalArgumentException("out of range (0~255)");
@@ -258,7 +261,7 @@ public void setParentSpanId(long parentSpanId) {
         this.parentSpanId = parentSpanId;
     }
 
-    public int getFlag() {
+    public short getFlag() {
         return flag;
     }
 
@@ -363,6 +366,13 @@ public String getExceptionMessage() {
         return exceptionMessage;
     }
 
+    public void setExceptionInfo(int exceptionId, String exceptionMessage) {
+        this.hasException = true;
+        this.exceptionId = exceptionId;
+        this.exceptionMessage = exceptionMessage;
+    }
+
+
     public String getExceptionClass() {
         return exceptionClass;
     }
@@ -371,13 +381,16 @@ public void setExceptionClass(String exceptionClass) {
         this.exceptionClass = exceptionClass;
     }
 
-    public void setApplicationServiceType(short applicationServiceType) {
-        this.hasApplicationServiceType = true;
+    public void setApplicationServiceType(Short applicationServiceType) {
         this.applicationServiceType  = applicationServiceType;
     }
-
+
+    public boolean hasApplicationServiceType() {
+        return applicationServiceType != null;
+    }
+
     public short getApplicationServiceType() {
-        if (this.hasApplicationServiceType) {
+        if (hasApplicationServiceType()) {
             return this.applicationServiceType;
         } else {
             return this.serviceType;
@@ -388,11 +401,9 @@ public byte getLoggingTransactionInfo() {
         return loggingTransactionInfo;
     }
 
-    public void setLoggingTransactionInfo(byte loggingTransactionInfo) {
-        this.loggingTransactionInfo = loggingTransactionInfo;
-    }
-
     // Variable encoding has been added in case of write io operation. The data size can be reduced by about 10%.
+    // for test
+    @Deprecated
     public byte[] writeValue() {
         /*
            It is difficult to calculate the size of buffer. It's not impossible.
@@ -403,9 +414,6 @@ public byte[] writeValue() {
 
         buffer.put(version);
 
-        // buffer.put(mostTraceID);
-        // buffer.put(leastTraceID);
-
         buffer.putPrefixedString(agentId);
 
         // Using var makes the sie of time smaller based on the present time. That consumes only 6 bytes.
@@ -438,19 +446,24 @@ public byte[] writeValue() {
         }
 
         buffer.put(flag);
-        
-        if (this.hasApplicationServiceType) {
+
+        if (hasApplicationServiceType()) {
             buffer.put(true);
             buffer.put(this.applicationServiceType);
         } else {
             buffer.put(false);
         }
-        
+
         buffer.put(loggingTransactionInfo);
 
         return buffer.getBuffer();
     }
 
+
+    public void setLoggingTransactionInfo(byte loggingTransactionInfo) {
+        this.loggingTransactionInfo = loggingTransactionInfo;
+    }
+
     public int readValue(byte[] bytes, int offset) {
         final Buffer buffer = new OffsetFixedBuffer(bytes, offset);
 
@@ -488,8 +501,8 @@ public int readValue(byte[] bytes, int offset) {
         // FIXME (2015.03) Legacy - applicationServiceType added in v1.1.0
         // Defaults to span's service type for older versions where applicationServiceType does not exist.
         if (buffer.limit() > 0) {
-            this.hasApplicationServiceType = buffer.readBoolean();
-            if (this.hasApplicationServiceType) {
+            final boolean hasApplicationServiceType = buffer.readBoolean();
+            if (hasApplicationServiceType) {
                 this.applicationServiceType = buffer.readShort();
             }
         }
@@ -501,6 +514,7 @@ public int readValue(byte[] bytes, int offset) {
         return buffer.getOffset();
     }
 
+
     @Override
     public String toString() {
         final StringBuilder sb = new StringBuilder(256);
@@ -529,10 +543,7 @@ public String toString() {
         sb.append(", exceptionId=").append(exceptionId);
         sb.append(", exceptionMessage='").append(exceptionMessage).append('\'');
         sb.append(", remoteAddr='").append(remoteAddr).append('\'');
-        sb.append(", hasApplicationServiceType=").append(hasApplicationServiceType);
-        if (hasApplicationServiceType) {
-            sb.append(", applicationServiceType=").append(applicationServiceType);
-        }
+        sb.append(", applicationServiceType=").append(applicationServiceType);
         sb.append('}');
         return sb.toString();
     }

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/serializer/HbaseSerializer.java

@@ -0,0 +1,12 @@
+package com.navercorp.pinpoint.common.server.bo.serializer;
+
+import org.apache.hadoop.hbase.client.Mutation;
+
+/**
+ * @author Woonduk Kang(emeroad)
+ */
+public interface HbaseSerializer<T, M extends Mutation> {
+
+    void serialize(T value, M mutation, SerializationContext context);
+
+}

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/serializer/SerializationContext.java

@@ -0,0 +1,8 @@
+package com.navercorp.pinpoint.common.server.bo.serializer;
+
+/**
+ * @author Woonduk Kang(emeroad)
+ */
+public interface SerializationContext {
+    // not implemented yet
+}