Merge pull request #1742 from emeroad/TraceV2/#1732_extract_serializer

#1732_extract_serializer
2026-08-24 20:26:16 +10:00 · 2016-05-04 13:35:28 +09:00
parent 83d91bbfdf 6cda79f26a
commit ec1360876a
10 changed files with 265 additions and 65 deletions
@@ -18,33 +18,27 @@ package com.navercorp.pinpoint.collector.dao.hbase;

 import com.navercorp.pinpoint.collector.dao.TracesDao;
 import com.navercorp.pinpoint.collector.dao.hbase.filter.SpanEventFilter;
+import com.navercorp.pinpoint.common.server.bo.serializer.AnnotationSerializer;
+import com.navercorp.pinpoint.common.server.bo.serializer.SpanEventSerializer;
 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;
 import com.navercorp.pinpoint.common.server.bo.SpanEventBo;
-import com.navercorp.pinpoint.common.buffer.AutomaticBuffer;
-import com.navercorp.pinpoint.common.buffer.Buffer;
 import static com.navercorp.pinpoint.common.hbase.HBaseTables.*;
 import com.navercorp.pinpoint.common.hbase.HbaseOperations2;
-import com.navercorp.pinpoint.common.util.BytesUtils;
 import com.navercorp.pinpoint.common.util.SpanUtils;
-import com.navercorp.pinpoint.thrift.dto.TAnnotation;
 import com.navercorp.pinpoint.thrift.dto.TSpan;
 import com.navercorp.pinpoint.thrift.dto.TSpanChunk;
 import com.navercorp.pinpoint.thrift.dto.TSpanEvent;
 import com.sematext.hbase.wd.AbstractRowKeyDistributor;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.hadoop.hbase.client.Put;
-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.Repository;

-import java.util.ArrayList;
 import java.util.List;

 /**
@@ -67,6 +61,12 @@ public class HbaseTraceDao implements TracesDao {
    @Autowired
    private SpanSerializer spanSerializer;

+    @Autowired
+    private SpanEventSerializer spanEventSerializer;
+
+    @Autowired
+    private AnnotationSerializer annotationSerializer;
+
    @Autowired
    @Qualifier("traceDistributor")
    private AbstractRowKeyDistributor rowKeyDistributor;
@@ -83,16 +83,8 @@ public class HbaseTraceDao implements TracesDao {
        final Put put = new Put(rowKey);

        this.spanSerializer.serialize(spanBo, put, null);
+        this.annotationSerializer.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.
-        final byte[] spanId = Bytes.toBytes(spanBo.getSpanId());
-
-        List<TAnnotation> annotations = span.getAnnotations();
-        if (CollectionUtils.isNotEmpty(annotations)) {
-            byte[] bytes = writeAnnotation(annotations);
-            put.addColumn(TRACES_CF_ANNOTATION, spanId, bytes);
-        }

        addNestedSpanEvent(put, span);

@@ -147,24 +139,8 @@ public class HbaseTraceDao implements TracesDao {
        if (!spanEventFilter.filter(spanEventBo)) {
            return;
        }
-
-        byte[] rowId = BytesUtils.add(spanEventBo.getSpanId(), spanEventBo.getSequence(), spanEventBo.getAsyncId(), spanEventBo.getAsyncSequence());
-        byte[] value = spanEventBo.writeValue();
-        final long acceptedTime = acceptedTimeService.getAcceptedTime();
-
-        put.addColumn(TRACES_CF_TERMINALSPAN, rowId, acceptedTime, value);
+        this.spanEventSerializer.serialize(spanEventBo, put, null);
    }

-    private byte[] writeAnnotation(List<TAnnotation> annotations) {
-        List<AnnotationBo> boList = new ArrayList<>(annotations.size());
-        for (TAnnotation ano : annotations) {
-            AnnotationBo annotationBo = new AnnotationBo(ano);
-            boList.add(annotationBo);
-        }

-        Buffer buffer = new AutomaticBuffer(64);
-        AnnotationBoList annotationBoList = new AnnotationBoList(boList);
-        annotationBoList.writeValue(buffer);
-        return buffer.getBuffer();
-    }
 }
@@ -63,6 +63,10 @@ public class AnnotationBo {
        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) " + version);
@@ -85,6 +89,10 @@ public class AnnotationBo {
        return valueType;
    }

+    public byte getRawValueType() {
+        return valueType;
+    }
+
    public void setValueType(byte valueType) {
        this.valueType = valueType;
    }
@@ -105,6 +113,7 @@ public class AnnotationBo {
        this.value = value;
    }

+    @Deprecated
    public void writeValue(Buffer buffer) {
        // long timestamp;    // required 8
        // long duration;     // optional 8
@@ -26,15 +26,16 @@ import java.util.List;
 * @author emeroad
 */
 public class AnnotationBoList {
+
    private List<AnnotationBo> annotationBoList;

    public AnnotationBoList() {
-        this.annotationBoList = new ArrayList<AnnotationBo>();
+        this.annotationBoList = new ArrayList<>();
    }


    public AnnotationBoList(int annotationBoListSize) {
-        this.annotationBoList = new ArrayList<AnnotationBo>(annotationBoListSize);
+        this.annotationBoList = new ArrayList<>(annotationBoListSize);
    }

    public AnnotationBoList(List<AnnotationBo> annotationBoList) {
@@ -53,7 +54,8 @@ public class AnnotationBoList {
        this.annotationBoList.add(annotationBo);
    }

-    public void writeValue(Buffer writer){
+    @Deprecated
+    public void writeValue(Buffer writer) {

        int size = this.annotationBoList.size();
        writer.putVar(size);
@@ -67,7 +69,7 @@ public class AnnotationBoList {
        if (size == 0) {
            return;
        }
-        this.annotationBoList = new ArrayList<AnnotationBo>(size);
+        this.annotationBoList = new ArrayList<>(size);
        for (int i = 0; i < size; i++) {
            AnnotationBo bo = new AnnotationBo();
            bo.readValue(reader);
@@ -17,6 +17,7 @@
 package com.navercorp.pinpoint.common.server.bo;

 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;

 import com.navercorp.pinpoint.common.buffer.AutomaticBuffer;
@@ -128,7 +129,7 @@ public class SpanBo implements Span {
            this.exceptionMessage = exceptionInfo.getStringValue();
        }

-        setAnnotationList(span.getAnnotations());
+        this.annotationBoList = buildAnnotationList(span.getAnnotations());
    }

    public SpanBo(String traceAgentId, long traceAgentStartTime, long traceTransactionSequence, long startTime, int elapsed, long spanId) {
@@ -290,15 +291,16 @@ public class SpanBo implements Span {
        return annotationBoList;
    }

-    public void setAnnotationList(List<TAnnotation> anoList) {
+    private List<AnnotationBo> buildAnnotationList(List<TAnnotation> anoList) {
        if (anoList == null) {
-            return;
+            return Collections.emptyList();
        }
-        List<AnnotationBo> boList = new ArrayList<AnnotationBo>(anoList.size());
+        List<AnnotationBo> boList = new ArrayList<>(anoList.size());
        for (TAnnotation ano : anoList) {
-            boList.add(new AnnotationBo(ano));
+            final AnnotationBo annotationBo = new AnnotationBo(ano);
+            boList.add(annotationBo);
        }
-        this.annotationBoList = boList;
+        return boList;
    }

    public void setAnnotationBoList(List<AnnotationBo> anoList) {
@@ -310,7 +312,7 @@ public class SpanBo implements Span {

    public void addSpanEvent(SpanEventBo spanEventBo) {
        if (spanEventBoList == null) {
-            spanEventBoList = new ArrayList<SpanEventBo>();
+            spanEventBoList = new ArrayList<>();
        }
        spanEventBoList.add(spanEventBo);
    }
@@ -128,15 +128,15 @@ public class SpanEventBo implements Span {
            this.exceptionMessage = exceptionInfo.getStringValue();
        }
        
-        if(tSpanEvent.isSetAsyncId()) {
+        if (tSpanEvent.isSetAsyncId()) {
            this.asyncId = tSpanEvent.getAsyncId();
        }
        
-        if(tSpanEvent.isSetNextAsyncId()) {
+        if (tSpanEvent.isSetNextAsyncId()) {
            this.nextAsyncId = tSpanEvent.getNextAsyncId();
        }
        
-        if(tSpanEvent.isSetAsyncSequence()) {
+        if (tSpanEvent.isSetAsyncSequence()) {
            this.asyncSequence = tSpanEvent.getAsyncSequence();
        }
    }
@@ -196,11 +196,11 @@ public class SpanEventBo implements Span {
            this.asyncId = spanEvent.getAsyncId();
        }
        
-        if(spanEvent.isSetNextAsyncId()) {
+        if (spanEvent.isSetNextAsyncId()) {
            this.nextAsyncId = spanEvent.getNextAsyncId();
        }
        
-        if(spanEvent.isSetAsyncSequence()) {
+        if (spanEvent.isSetAsyncSequence()) {
            this.asyncSequence = spanEvent.getAsyncSequence();
        }
    }
@@ -223,6 +223,14 @@ public class SpanEventBo implements Span {
        this.agentId = agentId;
    }

+    public String getApplicationId() {
+        return applicationId;
+    }
+
+    public void setApplicationId(String applicationId) {
+        this.applicationId = applicationId;
+    }
+
    public long getAgentStartTime() {
        return this.agentStartTime;
    }
@@ -414,14 +422,12 @@ public class SpanEventBo implements Span {
        this.asyncSequence = asyncSequence;
    }

+    @Deprecated
    public byte[] writeValue() {
        final Buffer buffer = new AutomaticBuffer(512);

        buffer.put(version);

-        // buffer.put(mostTraceID);
-        // buffer.put(leastTraceID);
-
        buffer.putPrefixedString(agentId);
        buffer.putPrefixedString(applicationId);
        buffer.putVar(agentStartTime);
@@ -496,8 +502,8 @@ public class SpanEventBo implements Span {
        }

        this.annotationBoList = readAnnotation(buffer);
-        if(buffer.getOffset() < endOffset) {
-            nextAsyncId = buffer.readSVarInt();            
+        if (buffer.getOffset() < endOffset) {
+            nextAsyncId = buffer.readSVarInt();
        }
        
        return buffer.getOffset();
@@ -0,0 +1,70 @@
+package com.navercorp.pinpoint.common.server.bo.serializer;
+
+import com.navercorp.pinpoint.common.buffer.AutomaticBuffer;
+import com.navercorp.pinpoint.common.buffer.Buffer;
+import com.navercorp.pinpoint.common.server.bo.AnnotationBo;
+import com.navercorp.pinpoint.common.server.bo.SpanBo;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.springframework.stereotype.Component;
+
+import java.util.Collections;
+import java.util.List;
+
+import static com.navercorp.pinpoint.common.hbase.HBaseTables.TRACES_CF_ANNOTATION;
+
+/**
+ * @author Woonduk Kang(emeroad)
+ */
+@Component
+public class AnnotationSerializer implements HbaseSerializer<SpanBo, Put> {
+
+
+    @Override
+    public void serialize(SpanBo spanBo, Put put, SerializationContext context) {
+
+        // TODO  if we can identify whether the columnName is duplicated or not,
+        // we can also know whether the span id is duplicated or not.
+        final byte[] spanId = Bytes.toBytes(spanBo.getSpanId());
+
+        final List<AnnotationBo> annotations = spanBo.getAnnotationBoList();
+        if (CollectionUtils.isNotEmpty(annotations)) {
+            byte[] bytes = writeAnnotationList(annotations);
+            put.addColumn(TRACES_CF_ANNOTATION, spanId, bytes);
+        }
+
+    }
+
+    private byte[] writeAnnotationList(List<AnnotationBo> annotationList) {
+        final Buffer buffer = new AutomaticBuffer(64);
+        return writeAnnotationList(annotationList, buffer);
+    }
+
+    // for test
+    public byte[] writeAnnotationList(List<AnnotationBo> annotationList, Buffer buffer) {
+
+        if (annotationList == null) {
+            annotationList = Collections.emptyList();
+        }
+        final int size = annotationList.size();
+
+        buffer.putVar(size);
+        for (AnnotationBo annotationBo : annotationList) {
+            writeAnnotation(annotationBo, buffer);
+        }
+
+        return buffer.getBuffer();
+    }
+
+    // for test
+    public void writeAnnotation(AnnotationBo annotationBo, Buffer puffer) {
+        // int key;           // required 4
+        // int valueTypeCode; // required 4
+        // ByteBuffer value;  // optional 4 + buf.length
+        puffer.put(annotationBo.getRawVersion());
+        puffer.putSVar(annotationBo.getKey());
+        puffer.put(annotationBo.getRawValueType());
+        puffer.putPrefixedBytes(annotationBo.getByteValue());
+    }
+}
@@ -0,0 +1,92 @@
+package com.navercorp.pinpoint.common.server.bo.serializer;
+
+import com.navercorp.pinpoint.common.buffer.AutomaticBuffer;
+import com.navercorp.pinpoint.common.buffer.Buffer;
+import com.navercorp.pinpoint.common.server.bo.AnnotationBo;
+import com.navercorp.pinpoint.common.server.bo.AnnotationBoList;
+import com.navercorp.pinpoint.common.server.bo.SpanEventBo;
+import com.navercorp.pinpoint.common.server.util.AcceptedTimeService;
+import com.navercorp.pinpoint.common.util.BytesUtils;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.protobuf.generated.CellProtos;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+import static com.navercorp.pinpoint.common.hbase.HBaseTables.TRACES_CF_TERMINALSPAN;
+
+/**
+ * @author Woonduk Kang(emeroad)
+ */
+@Component
+public class SpanEventSerializer implements HbaseSerializer<SpanEventBo, Put> {
+
+    private AnnotationSerializer annotationSerializer;
+
+    private AcceptedTimeService acceptedTimeService;
+
+    @Autowired
+    public void setAnnotationSerializer(AnnotationSerializer annotationSerializer) {
+        this.annotationSerializer = annotationSerializer;
+    }
+
+    @Autowired
+    public void setAcceptedTimeService(AcceptedTimeService acceptedTimeService) {
+        this.acceptedTimeService = acceptedTimeService;
+    }
+
+    @Override
+    public void serialize(SpanEventBo spanEventBo, Put put, SerializationContext context) {
+
+        byte[] rowId = BytesUtils.add(spanEventBo.getSpanId(), spanEventBo.getSequence(), spanEventBo.getAsyncId(), spanEventBo.getAsyncSequence());
+
+        final byte[] value = writeValue(spanEventBo);
+        final long acceptedTime = acceptedTimeService.getAcceptedTime();
+
+        put.addColumn(TRACES_CF_TERMINALSPAN, rowId, acceptedTime, value);
+
+    }
+
+    public byte[] writeValue(SpanEventBo spanEventBo) {
+        final Buffer buffer = new AutomaticBuffer(512);
+
+        buffer.put(spanEventBo.getVersion());
+
+        buffer.putPrefixedString(spanEventBo.getAgentId());
+        buffer.putPrefixedString(spanEventBo.getApplicationId());
+        buffer.putVar(spanEventBo.getAgentStartTime());
+
+        buffer.putVar(spanEventBo.getStartElapsed());
+        buffer.putVar(spanEventBo.getEndElapsed());
+
+        // don't need to put sequence because it is set at Qualifier
+        // buffer.put(sequence);
+
+        buffer.putPrefixedString(spanEventBo.getRpc());
+        buffer.put(spanEventBo.getServiceType());
+        buffer.putPrefixedString(spanEventBo.getEndPoint());
+        buffer.putPrefixedString(spanEventBo.getDestinationId());
+        buffer.putSVar(spanEventBo.getApiId());
+
+        buffer.putSVar(spanEventBo.getDepth());
+        buffer.put(spanEventBo.getNextSpanId());
+
+        if (spanEventBo.hasException()) {
+            buffer.put(true);
+            buffer.putSVar(spanEventBo.getExceptionId());
+            buffer.putPrefixedString(spanEventBo.getExceptionMessage());
+        } else {
+            buffer.put(false);
+        }
+        final List<AnnotationBo> annotationBoList = spanEventBo.getAnnotationBoList();
+        this.annotationSerializer.writeAnnotationList(annotationBoList, buffer);
+
+        buffer.putSVar(spanEventBo.getNextAsyncId());
+
+        return buffer.getBuffer();
+    }
+
+}
@@ -90,6 +90,7 @@ public class SpanSerializer implements HbaseSerializer<SpanBo, Put> {
        }

        buffer.put(span.getLoggingTransactionInfo());
+        buffer.putPrefixedString(span.getAcceptorHost());

        return buffer.getBuffer();
    }
@@ -16,18 +16,31 @@

 package com.navercorp.pinpoint.common.server.bo;

-import com.navercorp.pinpoint.common.server.bo.AnnotationBo;
 import com.navercorp.pinpoint.common.buffer.AutomaticBuffer;
 import com.navercorp.pinpoint.common.buffer.Buffer;
+import com.navercorp.pinpoint.common.server.bo.serializer.AnnotationSerializer;
 import com.navercorp.pinpoint.common.trace.AnnotationKey;

+import org.apache.commons.lang.RandomStringUtils;
+import org.apache.commons.lang.math.RandomUtils;
 import org.junit.Assert;
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.Charset;

 /**
 * @author emeroad
 */
 public class AnnotationBoTest {
+
+    private static final Charset UTF_8 = Charset.forName("UTF-8");
+    
+    private AnnotationSerializer serializer = new AnnotationSerializer();
+
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
    @Test
    public void testGetVersion() throws Exception {

@@ -40,20 +53,29 @@ public class AnnotationBoTest {

    @Test
    public void testWriteValue() throws Exception {
-        AnnotationBo bo = new AnnotationBo();
-        bo.setKey(AnnotationKey.API.getCode());
-        bo.setByteValue("value".getBytes("UTF-8"));
+        AnnotationBo annotation1 = new AnnotationBo();
+        annotation1.setKey(AnnotationKey.API.getCode());
+
+        final String value = RandomStringUtils.random(RandomUtils.nextInt(20));
+        annotation1.setByteValue(value.getBytes(UTF_8));
+        AnnotationBo annotation = annotation1;
 //        int bufferSize = bo.getBufferSize();

        Buffer buffer = new AutomaticBuffer(128);
-        bo.writeValue(buffer);
+        this.serializer.writeAnnotation(annotation, buffer);
+
+
+        Buffer deprecatedBuffer = new AutomaticBuffer(128);
+        annotation.writeValue(deprecatedBuffer);
+        Assert.assertArrayEquals(buffer.getBuffer(), deprecatedBuffer.getBuffer());

        AnnotationBo bo2 = new AnnotationBo();
        buffer.setOffset(0);
        bo2.readValue(buffer);
-        Assert.assertEquals(bo.getKey(), bo2.getKey());
-        Assert.assertEquals(bo.getValueType(), bo2.getValueType());
-        Assert.assertArrayEquals(bo.getByteValue(), bo2.getByteValue());
+        Assert.assertEquals(annotation.getKey(), bo2.getKey());
+        Assert.assertEquals(annotation.getValueType(), bo2.getValueType());
+        Assert.assertArrayEquals(annotation.getByteValue(), bo2.getByteValue());
    }

+
 }
@@ -16,17 +16,32 @@

 package com.navercorp.pinpoint.common.server.bo;

-import com.navercorp.pinpoint.common.server.bo.SpanEventBo;
+import com.navercorp.pinpoint.common.server.bo.serializer.AnnotationSerializer;
+import com.navercorp.pinpoint.common.server.bo.serializer.SpanEventSerializer;
 import com.navercorp.pinpoint.common.trace.ServiceType;

 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;

 /**
 * @author emeroad
 */
 public class SpanEventBoTest {

+
+    private SpanEventSerializer serializer = new SpanEventSerializer();
+
+    @Before
+    public void setUp() throws Exception {
+
+        this.serializer = new SpanEventSerializer();
+        final AnnotationSerializer annotationSerializer = new AnnotationSerializer();
+        this.serializer.setAnnotationSerializer(annotationSerializer);
+
+    }
+
    @Test
    public void testSerialize() throws Exception {
        SpanEventBo spanEventBo = new SpanEventBo();
@@ -43,8 +58,11 @@ public class SpanEventBoTest {
        spanEventBo.setServiceType(ServiceType.STAND_ALONE.getCode());
        spanEventBo.setSpanId(12);
        spanEventBo.setStartElapsed(100);
+        spanEventBo.setNextAsyncId(1000);

-        byte[] bytes = spanEventBo.writeValue();
+        byte[] deprecatedBytes = spanEventBo.writeValue();
+        byte[] bytes = serializer.writeValue(spanEventBo);
+        Assert.assertArrayEquals(bytes, deprecatedBytes);

        SpanEventBo newSpanEventBo = new SpanEventBo();
        int i = newSpanEventBo.readValue(bytes, 0, bytes.length);
@@ -64,6 +82,8 @@ public class SpanEventBoTest {
        Assert.assertEquals(spanEventBo.getServiceType(), newSpanEventBo.getServiceType());
        Assert.assertEquals(spanEventBo.getStartElapsed(), newSpanEventBo.getStartElapsed());

+        Assert.assertEquals(spanEventBo.getNextAsyncId(), newSpanEventBo.getNextAsyncId());
+

        // we get these from the row key
        spanEventBo.setSpanId(1);