diff --git a/README.md b/README.md index 97c32bb00..8e3bd8200 100644 --- a/README.md +++ b/README.md @@ -19,16 +19,6 @@ Sky Walking | [中文](https://github.com/wu-sheng/sky-walking/wiki/sky-walking- * High performance streaming analysis. * The UI released on [wu-sheng/sky-walking-ui](https://github.com/wu-sheng/sky-walking-ui) -___ - - -OneAPM Open Source Achievement Award - - -In October 2016, Sky Walking won `OneAPM Open Source Achievement Award`. The award appreciates sky walking for its "*contribution to popularization of APM technology*".


-Thanks all users of sky walking project. - -___ # Contributors _In chronological order_ diff --git a/apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkywalkingSpan.java b/apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkywalkingSpan.java index 430fa46e6..a9d08faac 100644 --- a/apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkywalkingSpan.java +++ b/apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkywalkingSpan.java @@ -68,16 +68,19 @@ public class SkywalkingSpan implements Span { return SkywalkingContext.INSTANCE; } + @NeedSnifferActivation( + "1. ContextManager#activeSpan()" + + "2. SkywalkingSpan#setTag(String, String)") @Override public Span setTag(String key, String value) { - return null; + return this; } @Override public Span setTag(String key, boolean value) { - return null; + return setTag(key, String.valueOf(value)); } @Override public Span setTag(String key, Number value) { - return null; + return setTag(key, String.valueOf(value)); } @Override diff --git a/apm-commons/apm-datacarrier/src/main/java/org/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java b/apm-commons/apm-datacarrier/src/main/java/org/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java index 75a2ad651..e6d4e7602 100644 --- a/apm-commons/apm-datacarrier/src/main/java/org/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java +++ b/apm-commons/apm-datacarrier/src/main/java/org/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java @@ -20,6 +20,7 @@ public class ConsumerPool { this(channels, num); for (int i = 0; i < num; i++) { consumerThreads[i] = new ConsumerThread("DataCarrier.Consumser." + i + ".Thread", getNewConsumerInstance(consumerClass)); + consumerThreads[i].setDaemon(true); } } diff --git a/apm-network/src/main/proto/DiscoveryService.proto b/apm-network/src/main/proto/DiscoveryService.proto index c1a14bcfb..cda0fa286 100644 --- a/apm-network/src/main/proto/DiscoveryService.proto +++ b/apm-network/src/main/proto/DiscoveryService.proto @@ -20,7 +20,8 @@ service InstanceDiscoveryService { message ApplicationInstance { int32 applicationId = 1; - int64 registerTime = 2; + string agentUUID = 2; + int64 registerTime = 3; } message ApplicationInstanceMapping { @@ -61,4 +62,4 @@ message ServiceNameMappingElement { message ServiceNameElement { string serviceName = 1; int32 applicationId = 2; -} \ No newline at end of file +} diff --git a/apm-network/src/main/proto/TraceSegmentService.proto b/apm-network/src/main/proto/TraceSegmentService.proto index 315165ebe..a1e0aa134 100644 --- a/apm-network/src/main/proto/TraceSegmentService.proto +++ b/apm-network/src/main/proto/TraceSegmentService.proto @@ -12,12 +12,16 @@ service TraceSegmentService { } message UpstreamSegment { - repeated string globalTraceIds = 1; + repeated UniqueId globalTraceIds = 1; bytes segment = 2; // the byte array of TraceSegmentObject } +message UniqueId { + repeated int64 idParts = 1; +} + message TraceSegmentObject { - string traceSegmentId = 1; + UniqueId traceSegmentId = 1; repeated TraceSegmentReference refs = 2; repeated SpanObject spans = 3; int32 applicationId = 4; @@ -26,13 +30,15 @@ message TraceSegmentObject { message TraceSegmentReference { RefType refType = 1; - string parentTraceSegmentId = 2; + UniqueId parentTraceSegmentId = 2; int32 parentSpanId = 3; int32 parentApplicationInstanceId = 4; string networkAddress = 5; int32 networkAddressId = 6; string entryServiceName = 7; int32 entryServiceId = 8; + string parentServiceName = 9; + int32 parentServiceId = 10; } message SpanObject { diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/BootService.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/BootService.java index d3538f915..be80b2c86 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/BootService.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/BootService.java @@ -13,4 +13,6 @@ public interface BootService { void boot() throws Throwable; void afterBoot() throws Throwable; + + void shutdown() throws Throwable; } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/ServiceManager.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/ServiceManager.java index 2d7338994..2c5337ef2 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/ServiceManager.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/ServiceManager.java @@ -27,6 +27,16 @@ public enum ServiceManager { afterBoot(); } + public void shutdown() { + for (BootService service : bootedServices.values()) { + try { + service.shutdown(); + } catch (Throwable e) { + logger.error(e, "ServiceManager try to shutdown [{}] fail.", service.getClass().getName()); + } + } + } + private Map loadAllServices() { HashMap bootedServices = new HashMap(); Iterator serviceIterator = load().iterator(); diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/conf/Config.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/conf/Config.java index 6ffa2cfb5..affced6f7 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/conf/Config.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/conf/Config.java @@ -60,12 +60,28 @@ public class Config { public static String DISCOVERY_SERVICE_NAME = "/grpc/addresses"; } + public static class Jvm { + /** + * The buffer size of collected JVM info. + */ + public static int BUFFER_SIZE = 60 * 10; + } + public static class Buffer { public static int CHANNEL_SIZE = 5; public static int BUFFER_SIZE = 300; } + public static class Dictionary { + /** + * The buffer size of application codes and peer + */ + public static int APPLICATION_CODE_BUFFER_SIZE = 10 * 10000; + + public static int OPERATION_NAME_BUFFER_SIZE = 1000 * 10000; + } + public static class Logging { /** * Log file name. diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/AbstractTracerContext.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/AbstractTracerContext.java index 3b9d569d1..35099a516 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/AbstractTracerContext.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/AbstractTracerContext.java @@ -55,7 +55,7 @@ public interface AbstractTracerContext { * @return the string represents the id. * @see {@link TracingContext} and {@link IgnoredTracerContext} */ - String getGlobalTraceId(); + String getReadableGlobalTraceId(); /** * Create an entry span diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextCarrier.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextCarrier.java index fad0ab726..1b94a3e25 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextCarrier.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextCarrier.java @@ -1,9 +1,9 @@ package org.skywalking.apm.agent.core.context; import java.io.Serializable; -import java.util.LinkedList; import java.util.List; import org.skywalking.apm.agent.core.context.ids.DistributedTraceId; +import org.skywalking.apm.agent.core.context.ids.ID; import org.skywalking.apm.agent.core.context.ids.PropagatedTraceId; import org.skywalking.apm.agent.core.context.trace.TraceSegment; import org.skywalking.apm.agent.core.dictionary.DictionaryUtil; @@ -19,7 +19,7 @@ public class ContextCarrier implements Serializable { /** * {@link TraceSegment#traceSegmentId} */ - private String traceSegmentId; + private ID traceSegmentId; private int spanId = -1; @@ -29,10 +29,12 @@ public class ContextCarrier implements Serializable { private String entryOperationName; + private String parentOperationName; + /** * {@link DistributedTraceId} */ - private List distributedTraceIds; + private DistributedTraceId primaryDistributedTraceId; /** * Serialize this {@link ContextCarrier} to a {@link String}, @@ -43,12 +45,13 @@ public class ContextCarrier implements Serializable { public String serialize() { if (this.isValid()) { return StringUtil.join('|', - this.getTraceSegmentId(), + this.getTraceSegmentId().toBase64(), this.getSpanId() + "", this.getApplicationInstanceId() + "", this.getPeerHost(), this.getEntryOperationName(), - this.serializeDistributedTraceIds()); + this.getParentOperationName(), + this.getPrimaryDistributedTraceId()); } else { return ""; } @@ -61,15 +64,16 @@ public class ContextCarrier implements Serializable { */ public ContextCarrier deserialize(String text) { if (text != null) { - String[] parts = text.split("\\|", 6); - if (parts.length == 6) { + String[] parts = text.split("\\|", 7); + if (parts.length == 7) { try { - this.traceSegmentId = parts[0]; + this.traceSegmentId = new ID(parts[0]); this.spanId = Integer.parseInt(parts[1]); this.applicationInstanceId = Integer.parseInt(parts[2]); this.peerHost = parts[3]; this.entryOperationName = parts[4]; - this.distributedTraceIds = deserializeDistributedTraceIds(parts[5]); + this.parentOperationName = parts[5]; + this.primaryDistributedTraceId = new PropagatedTraceId(parts[6]); } catch (NumberFormatException e) { } @@ -84,12 +88,13 @@ public class ContextCarrier implements Serializable { * @return true for unbroken {@link ContextCarrier} or no-initialized. Otherwise, false; */ public boolean isValid() { - return !StringUtil.isEmpty(traceSegmentId) + return traceSegmentId != null && getSpanId() > -1 && applicationInstanceId != DictionaryUtil.nullValue() && !StringUtil.isEmpty(peerHost) && !StringUtil.isEmpty(entryOperationName) - && distributedTraceIds != null; + && !StringUtil.isEmpty(parentOperationName) + && primaryDistributedTraceId != null; } public String getEntryOperationName() { @@ -104,7 +109,15 @@ public class ContextCarrier implements Serializable { this.entryOperationName = entryOperationId + ""; } - public String getTraceSegmentId() { + void setParentOperationName(String parentOperationName) { + this.parentOperationName = '#' + parentOperationName; + } + + void setParentOperationId(int parentOperationId) { + this.parentOperationName = parentOperationId + ""; + } + + public ID getTraceSegmentId() { return traceSegmentId; } @@ -112,7 +125,7 @@ public class ContextCarrier implements Serializable { return spanId; } - void setTraceSegmentId(String traceSegmentId) { + void setTraceSegmentId(ID traceSegmentId) { this.traceSegmentId = traceSegmentId; } @@ -140,51 +153,19 @@ public class ContextCarrier implements Serializable { this.peerHost = peerId + ""; } - public List getDistributedTraceIds() { - return distributedTraceIds; + public DistributedTraceId getDistributedTraceId() { + return primaryDistributedTraceId; } public void setDistributedTraceIds(List distributedTraceIds) { - this.distributedTraceIds = distributedTraceIds; + this.primaryDistributedTraceId = distributedTraceIds.get(0); } - /** - * Serialize {@link #distributedTraceIds} to a string, with ',' split. - * - * @return string, represents all {@link DistributedTraceId} - */ - private String serializeDistributedTraceIds() { - StringBuilder traceIdString = new StringBuilder(); - if (distributedTraceIds != null) { - boolean first = true; - for (DistributedTraceId distributedTraceId : distributedTraceIds) { - if (first) { - first = false; - } else { - traceIdString.append(","); - } - traceIdString.append(distributedTraceId.get()); - } - } - return traceIdString.toString(); + private String getPrimaryDistributedTraceId() { + return primaryDistributedTraceId.toBase64(); } - /** - * Deserialize {@link #distributedTraceIds} from a text, whith - * - * @param text - * @return - */ - private List deserializeDistributedTraceIds(String text) { - if (StringUtil.isEmpty(text)) { - return null; - } - String[] propagationTraceIdValues = text.split(","); - List traceIds = new LinkedList(); - for (String propagationTraceIdValue : propagationTraceIdValues) { - traceIds.add(new PropagatedTraceId(propagationTraceIdValue)); - } - return traceIds; + public String getParentOperationName() { + return parentOperationName; } - } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java index a34458464..01765f638 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java @@ -76,7 +76,7 @@ public class ContextManager implements TracingContextListener, BootService, Igno if (segment == null) { return "N/A"; } else { - return segment.getGlobalTraceId(); + return segment.getReadableGlobalTraceId(); } } @@ -162,6 +162,10 @@ public class ContextManager implements TracingContextListener, BootService, Igno } + @Override public void shutdown() throws Throwable { + + } + @Override public void afterFinished(TraceSegment traceSegment) { CONTEXT.remove(); diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextSnapshot.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextSnapshot.java index 755a2275d..7543fab3a 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextSnapshot.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextSnapshot.java @@ -2,6 +2,8 @@ package org.skywalking.apm.agent.core.context; import java.util.List; import org.skywalking.apm.agent.core.context.ids.DistributedTraceId; +import org.skywalking.apm.agent.core.context.ids.ID; +import org.skywalking.apm.util.StringUtil; /** * The ContextSnapshot is a snapshot for current context. The snapshot carries the info for building @@ -13,30 +15,52 @@ public class ContextSnapshot { /** * trace segment id of the parent trace segment. */ - private String traceSegmentId; + private ID traceSegmentId; /** * span id of the parent span, in parent trace segment. */ private int spanId = -1; + private String entryOperationName; + + private String parentOperationName; + /** * {@link DistributedTraceId} */ - private List distributedTraceIds; + private DistributedTraceId primaryDistributedTraceId; - ContextSnapshot(String traceSegmentId, int spanId, + ContextSnapshot(ID traceSegmentId, int spanId, List distributedTraceIds) { this.traceSegmentId = traceSegmentId; this.spanId = spanId; - this.distributedTraceIds = distributedTraceIds; + if (distributedTraceIds != null) { + this.primaryDistributedTraceId = distributedTraceIds.get(0); + } } - public List getDistributedTraceIds() { - return distributedTraceIds; + public void setEntryOperationName(String entryOperationName) { + this.entryOperationName = "#" + entryOperationName; } - public String getTraceSegmentId() { + public void setEntryOperationId(int entryOperationId) { + this.entryOperationName = entryOperationId + ""; + } + + public void setParentOperationName(String parentOperationName) { + this.parentOperationName = "#" + parentOperationName; + } + + public void setParentOperationId(int parentOperationId) { + this.parentOperationName = parentOperationId + ""; + } + + public DistributedTraceId getDistributedTraceId() { + return primaryDistributedTraceId; + } + + public ID getTraceSegmentId() { return traceSegmentId; } @@ -44,10 +68,19 @@ public class ContextSnapshot { return spanId; } + public String getParentOperationName() { + return parentOperationName; + } + public boolean isValid() { return traceSegmentId != null && spanId > -1 - && distributedTraceIds != null - && distributedTraceIds.size() > 0; + && primaryDistributedTraceId != null + && !StringUtil.isEmpty(entryOperationName) + && !StringUtil.isEmpty(parentOperationName); + } + + public String getEntryOperationName() { + return entryOperationName; } } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/IgnoredTracerContext.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/IgnoredTracerContext.java index 4ad37e361..891cdf6f4 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/IgnoredTracerContext.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/IgnoredTracerContext.java @@ -41,7 +41,7 @@ public class IgnoredTracerContext implements AbstractTracerContext { } @Override - public String getGlobalTraceId() { + public String getReadableGlobalTraceId() { return "[Ignored Trace]"; } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/TracingContext.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/TracingContext.java index 15e41c2a8..871cbbb75 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/TracingContext.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/TracingContext.java @@ -96,8 +96,8 @@ public class TracingContext implements AbstractTracerContext { String operationName; if (refs != null && refs.size() > 0) { TraceSegmentRef ref = refs.get(0); - operationId = ref.getOperationId(); - operationName = ref.getOperationName(); + operationId = ref.getEntryOperationId(); + operationName = ref.getEntryOperationName(); } else { AbstractTracingSpan firstSpan = first(); operationId = firstSpan.getOperationId(); @@ -109,6 +109,13 @@ public class TracingContext implements AbstractTracerContext { carrier.setEntryOperationId(operationId); } + int parentOperationId = first().getOperationId(); + if (parentOperationId == DictionaryUtil.nullValue()) { + carrier.setParentOperationName(first().getOperationName()); + } else { + carrier.setParentOperationId(parentOperationId); + } + carrier.setDistributedTraceIds(this.segment.getRelatedGlobalTraces()); } @@ -121,7 +128,7 @@ public class TracingContext implements AbstractTracerContext { @Override public void extract(ContextCarrier carrier) { this.segment.ref(new TraceSegmentRef(carrier)); - this.segment.relatedGlobalTraces(carrier.getDistributedTraceIds()); + this.segment.relatedGlobalTraces(carrier.getDistributedTraceId()); } /** @@ -132,10 +139,33 @@ public class TracingContext implements AbstractTracerContext { */ @Override public ContextSnapshot capture() { - return new ContextSnapshot(segment.getTraceSegmentId(), + List refs = this.segment.getRefs(); + ContextSnapshot snapshot = new ContextSnapshot(segment.getTraceSegmentId(), activeSpan().getSpanId(), - segment.getRelatedGlobalTraces() - ); + segment.getRelatedGlobalTraces()); + int entryOperationId; + String entryOperationName; + AbstractTracingSpan firstSpan = first(); + if (refs != null && refs.size() > 0) { + TraceSegmentRef ref = refs.get(0); + entryOperationId = ref.getEntryOperationId(); + entryOperationName = ref.getEntryOperationName(); + } else { + entryOperationId = firstSpan.getOperationId(); + entryOperationName = firstSpan.getOperationName(); + } + if (entryOperationId == DictionaryUtil.nullValue()) { + snapshot.setEntryOperationName(entryOperationName); + } else { + snapshot.setEntryOperationId(entryOperationId); + } + + if (firstSpan.getOperationId() == DictionaryUtil.nullValue()) { + snapshot.setParentOperationName(firstSpan.getOperationName()); + } else { + snapshot.setParentOperationId(firstSpan.getOperationId()); + } + return snapshot; } /** @@ -147,15 +177,15 @@ public class TracingContext implements AbstractTracerContext { @Override public void continued(ContextSnapshot snapshot) { this.segment.ref(new TraceSegmentRef(snapshot)); - this.segment.relatedGlobalTraces(snapshot.getDistributedTraceIds()); + this.segment.relatedGlobalTraces(snapshot.getDistributedTraceId()); } /** * @return the first global trace id. */ @Override - public String getGlobalTraceId() { - return segment.getRelatedGlobalTraces().get(0).get(); + public String getReadableGlobalTraceId() { + return segment.getRelatedGlobalTraces().get(0).toString(); } /** @@ -256,13 +286,13 @@ public class TracingContext implements AbstractTracerContext { .doInCondition( new PossibleFound.FoundAndObtain() { @Override - public Object doProcess(int peerId) { - return new ExitSpan(spanIdGenerator++, parentSpanId, applicationId, peerId); + public Object doProcess(int operationId) { + return new ExitSpan(spanIdGenerator++, parentSpanId, operationId, applicationId); } }, new PossibleFound.NotFoundAndObtain() { @Override public Object doProcess() { - return new ExitSpan(spanIdGenerator++, parentSpanId, applicationId, remotePeer); + return new ExitSpan(spanIdGenerator++, parentSpanId, operationName, remotePeer); } }); } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/DistributedTraceId.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/DistributedTraceId.java index 58b1f9d17..f6e5623ef 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/DistributedTraceId.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/DistributedTraceId.java @@ -1,5 +1,7 @@ package org.skywalking.apm.agent.core.context.ids; +import org.skywalking.apm.network.proto.UniqueId; + /** * The DistributedTraceId presents a distributed call chain. *

@@ -14,14 +16,27 @@ package org.skywalking.apm.agent.core.context.ids; * @author wusheng */ public abstract class DistributedTraceId { - private String id; + private ID id; - public DistributedTraceId(String id) { + public DistributedTraceId(ID id) { this.id = id; } - public String get() { - return id; + public DistributedTraceId(String id) { + this.id = new ID(id); + } + + public String toBase64() { + return id.toBase64(); + } + + @Override + public String toString() { + return id.toString(); + } + + public UniqueId toUniqueId() { + return id.transform(); } /** @@ -38,7 +53,7 @@ public abstract class DistributedTraceId { if (o == null || getClass() != o.getClass()) return false; - DistributedTraceId id1 = (DistributedTraceId) o; + DistributedTraceId id1 = (DistributedTraceId)o; return id != null ? id.equals(id1.id) : id1.id == null; } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/DistributedTraceIds.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/DistributedTraceIds.java index fab3ad63d..94ab46fa0 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/DistributedTraceIds.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/DistributedTraceIds.java @@ -1,10 +1,5 @@ package org.skywalking.apm.agent.core.context.ids; -import com.google.gson.TypeAdapter; -import com.google.gson.annotations.JsonAdapter; -import com.google.gson.stream.JsonReader; -import com.google.gson.stream.JsonWriter; -import java.io.IOException; import java.util.Collections; import java.util.LinkedList; import java.util.List; @@ -12,7 +7,6 @@ import java.util.List; /** * @author wusheng */ -@JsonAdapter(DistributedTraceIds.Serializer.class) public class DistributedTraceIds { private LinkedList relatedGlobalTraces; @@ -32,35 +26,4 @@ public class DistributedTraceIds { relatedGlobalTraces.add(distributedTraceId); } } - - public static class Serializer extends TypeAdapter { - - @Override - public void write(JsonWriter out, DistributedTraceIds value) throws IOException { - List globalTraces = value.getRelatedGlobalTraces(); - - if (globalTraces.size() > 0) { - out.beginArray(); - for (DistributedTraceId trace : globalTraces) { - out.value(trace.get()); - } - out.endArray(); - } - } - - @Override - public DistributedTraceIds read(JsonReader in) throws IOException { - DistributedTraceIds distributedTraceIds = new DistributedTraceIds(); - in.beginArray(); - try { - while (in.hasNext()) { - PropagatedTraceId traceId = new PropagatedTraceId(in.nextString()); - distributedTraceIds.append(traceId); - } - } finally { - in.endArray(); - } - return distributedTraceIds; - } - } } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/GlobalIdGenerator.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/GlobalIdGenerator.java index d9720fbcf..042e909c6 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/GlobalIdGenerator.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/GlobalIdGenerator.java @@ -1,34 +1,90 @@ package org.skywalking.apm.agent.core.context.ids; -import java.util.UUID; -import org.skywalking.apm.util.MachineInfo; -import org.skywalking.apm.util.StringUtil; +import java.util.Random; +import org.skywalking.apm.agent.core.conf.RemoteDownstreamConfig; +import org.skywalking.apm.agent.core.dictionary.DictionaryUtil; public final class GlobalIdGenerator { - private static final ThreadLocal THREAD_ID_SEQUENCE = new ThreadLocal() { + private static final ThreadLocal THREAD_ID_SEQUENCE = new ThreadLocal() { @Override - protected Integer initialValue() { - return 0; + protected IDContext initialValue() { + return new IDContext(System.currentTimeMillis(), (short)0); } }; - private static final int PROCESS_UUID; - - static { - String uuid = UUID.randomUUID().toString().replaceAll("-", ""); - PROCESS_UUID = uuid.substring(uuid.length() - 7).hashCode(); - } - private GlobalIdGenerator() { } - public static String generate(String type) { - Integer seq = THREAD_ID_SEQUENCE.get(); - seq++; - THREAD_ID_SEQUENCE.set(seq); + /** + * Generate a new id, combined by three long numbers. + * + * The first one represents application instance id. (most likely just an integer value, would be helpful in + * protobuf) + * + * The second one represents thread id. (most likely just an integer value, would be helpful in protobuf) + * + * The third one also has two parts,
+ * 1) a timestamp, measured in milliseconds
+ * 2) a seq, in current thread, between 0(included) and 9999(included) + * + * Notice, a long costs 8 bytes, three longs cost 24 bytes. And at the same time, a char costs 2 bytes. So + * sky-walking's old global and segment id like this: "S.1490097253214.-866187727.57515.1.1" which costs at least 72 + * bytes. + * + * @return an array contains three long numbers, which represents a unique id. + */ + public static ID generate() { + if (RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID == DictionaryUtil.nullValue()) { + throw new IllegalStateException(); + } + IDContext context = THREAD_ID_SEQUENCE.get(); - return StringUtil.join('.', - type + "", System.currentTimeMillis() + "", PROCESS_UUID + "", - MachineInfo.getProcessNo() + "", Thread.currentThread().getId() + "", seq + ""); + return new ID( + RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID, + Thread.currentThread().getId(), + context.nextSeq() + ); + } + + private static class IDContext { + private long lastTimestamp; + private short threadSeq; + + // Just for considering time-shift-back only. + private long runRandomTimestamp; + private int lastRandomValue; + private Random random; + + private IDContext(long lastTimestamp, short threadSeq) { + this.lastTimestamp = lastTimestamp; + this.threadSeq = threadSeq; + } + + private long nextSeq() { + return timestamp() * 10000 + nextThreadSeq(); + } + + private long timestamp() { + long currentTimeMillis = System.currentTimeMillis(); + + if (currentTimeMillis < lastTimestamp) { + // Just for considering time-shift-back by Ops or OS. @hanahmily 's suggestion. + if (random == null) { + random = new Random(); + } + if (runRandomTimestamp != currentTimeMillis) { + lastRandomValue = random.nextInt(); + runRandomTimestamp = currentTimeMillis; + } + return lastRandomValue; + } else { + lastTimestamp = currentTimeMillis; + return lastTimestamp; + } + } + + private short nextThreadSeq() { + return threadSeq++; + } } } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/ID.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/ID.java new file mode 100644 index 000000000..db936aee5 --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/ID.java @@ -0,0 +1,114 @@ +package org.skywalking.apm.agent.core.context.ids; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import org.skywalking.apm.agent.core.context.ids.base64.Base64; +import org.skywalking.apm.network.proto.UniqueId; + +/** + * @author wusheng + */ +public class ID { + private static final Base64.Encoder ENCODER = Base64.getEncoder(); + private static final Base64.Decoder DECODER = Base64.getDecoder(); + + private long part1; + private long part2; + private long part3; + + public ID(long part1, long part2, long part3) { + this.part1 = part1; + this.part2 = part2; + this.part3 = part3; + } + + public ID(String base64String) { + int index = 0; + for (int part = 0; part < 3; part++) { + String encodedString; + char potentialTypeChar = base64String.charAt(index); + long value; + if (potentialTypeChar == '#') { + encodedString = base64String.substring(index + 1, index + 5); + index += 5; + value = ByteBuffer.wrap(DECODER.decode(encodedString)).order(ByteOrder.LITTLE_ENDIAN).asShortBuffer().get(0); + } else if (potentialTypeChar == '$') { + encodedString = base64String.substring(index + 1, index + 9); + index += 9; + value = ByteBuffer.wrap(DECODER.decode(encodedString)).order(ByteOrder.LITTLE_ENDIAN).asIntBuffer().get(0); + } else { + encodedString = base64String.substring(index, index + 12); + index += 12; + value = ByteBuffer.wrap(DECODER.decode(encodedString)).order(ByteOrder.LITTLE_ENDIAN).asLongBuffer().get(0); + } + + if (part == 0) { + part1 = value; + } else if (part == 1) { + part2 = value; + } else { + part3 = value; + } + + } + } + + public String toBase64() { + return long2Base64(part1) + long2Base64(part2) + long2Base64(part3); + } + + private String long2Base64(long partN) { + if (partN < 0) { + throw new IllegalArgumentException("negative value."); + } + if (partN < 32768) { + // 0 - 32767 + // "#" as a prefix of a short value with base64 encoding. + byte[] data = new byte[2]; + ByteBuffer.wrap(data).order(ByteOrder.LITTLE_ENDIAN).asShortBuffer().put((short)partN); + return '#' + ENCODER.encodeToString(data); + } else if (partN <= 2147483647) { + // 32768 - 2147483647 + // "$" as a prefix of an integer value (greater than a short) with base64 encoding. + byte[] data = new byte[4]; + ByteBuffer.wrap(data).order(ByteOrder.LITTLE_ENDIAN).asIntBuffer().put((int)partN); + return '$' + ENCODER.encodeToString(data); + } else { + // > 2147483647 + // a long value (greater than an integer) + byte[] data = new byte[8]; + ByteBuffer.wrap(data).order(ByteOrder.LITTLE_ENDIAN).asLongBuffer().put(partN); + return ENCODER.encodeToString(data); + } + } + + @Override public String toString() { + return part1 + "." + part2 + '.' + part3; + } + + @Override public boolean equals(Object o) { + if (this == o) + return true; + if (o == null || getClass() != o.getClass()) + return false; + + ID id = (ID)o; + + if (part1 != id.part1) + return false; + if (part2 != id.part2) + return false; + return part3 == id.part3; + } + + @Override public int hashCode() { + int result = (int)(part1 ^ (part1 >>> 32)); + result = 31 * result + (int)(part2 ^ (part2 >>> 32)); + result = 31 * result + (int)(part3 ^ (part3 >>> 32)); + return result; + } + + public UniqueId transform() { + return UniqueId.newBuilder().addIdParts(part1).addIdParts(part2).addIdParts(part3).build(); + } +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/NewDistributedTraceId.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/NewDistributedTraceId.java index fc058e218..53c4fc726 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/NewDistributedTraceId.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/NewDistributedTraceId.java @@ -6,9 +6,7 @@ package org.skywalking.apm.agent.core.context.ids; * @author wusheng */ public class NewDistributedTraceId extends DistributedTraceId { - private static final String ID_TYPE = "T"; - public NewDistributedTraceId() { - super(GlobalIdGenerator.generate(ID_TYPE)); + super(GlobalIdGenerator.generate()); } } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/base64/Base64.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/base64/Base64.java new file mode 100644 index 000000000..48a08f95b --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/base64/Base64.java @@ -0,0 +1,984 @@ +package org.skywalking.apm.agent.core.context.ids.base64; + +import java.io.FilterOutputStream; +import java.io.InputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.util.Arrays; + +/** + * This class consists exclusively of static methods for obtaining + * encoders and decoders for the Base64 encoding scheme. The + * implementation of this class supports the following types of Base64 + * as specified in + * RFC 4648 and + * RFC 2045. + * + *

    + *
  • Basic + *

    Uses "The Base64 Alphabet" as specified in Table 1 of + * RFC 4648 and RFC 2045 for encoding and decoding operation. + * The encoder does not add any line feed (line separator) + * character. The decoder rejects data that contains characters + * outside the base64 alphabet.

  • + * + *
  • URL and Filename safe + *

    Uses the "URL and Filename safe Base64 Alphabet" as specified + * in Table 2 of RFC 4648 for encoding and decoding. The + * encoder does not add any line feed (line separator) character. + * The decoder rejects data that contains characters outside the + * base64 alphabet.

  • + * + *
  • MIME + *

    Uses the "The Base64 Alphabet" as specified in Table 1 of + * RFC 2045 for encoding and decoding operation. The encoded output + * must be represented in lines of no more than 76 characters each + * and uses a carriage return {@code '\r'} followed immediately by + * a linefeed {@code '\n'} as the line separator. No line separator + * is added to the end of the encoded output. All line separators + * or other characters not found in the base64 alphabet table are + * ignored in decoding operation.

  • + *
+ * + *

Unless otherwise noted, passing a {@code null} argument to a + * method of this class will cause a {@link java.lang.NullPointerException + * NullPointerException} to be thrown. + * + * @author Xueming Shen + * @since 1.8 + */ + +public class Base64 { + + private Base64() {} + + /** + * Returns a {@link Encoder} that encodes using the + * Basic type base64 encoding scheme. + * + * @return A Base64 encoder. + */ + public static Encoder getEncoder() { + return Encoder.RFC4648; + } + + /** + * Returns a {@link Encoder} that encodes using the + * URL and Filename safe type base64 + * encoding scheme. + * + * @return A Base64 encoder. + */ + public static Encoder getUrlEncoder() { + return Encoder.RFC4648_URLSAFE; + } + + /** + * Returns a {@link Encoder} that encodes using the + * MIME type base64 encoding scheme. + * + * @return A Base64 encoder. + */ + public static Encoder getMimeEncoder() { + return Encoder.RFC2045; + } + + /** + * Returns a {@link Encoder} that encodes using the + * MIME type base64 encoding scheme + * with specified line length and line separators. + * + * @param lineLength + * the length of each output line (rounded down to nearest multiple + * of 4). If {@code lineLength <= 0} the output will not be separated + * in lines + * @param lineSeparator + * the line separator for each output line + * + * @return A Base64 encoder. + * + * @throws IllegalArgumentException if {@code lineSeparator} includes any + * character of "The Base64 Alphabet" as specified in Table 1 of + * RFC 2045. + */ + public static Encoder getMimeEncoder(int lineLength, byte[] lineSeparator) { + Objects.requireNonNull(lineSeparator); + int[] base64 = Decoder.fromBase64; + for (byte b : lineSeparator) { + if (base64[b & 0xff] != -1) + throw new IllegalArgumentException( + "Illegal base64 line separator character 0x" + Integer.toString(b, 16)); + } + if (lineLength <= 0) { + return Encoder.RFC4648; + } + return new Encoder(false, lineSeparator, lineLength >> 2 << 2, true); + } + + /** + * Returns a {@link Decoder} that decodes using the + * Basic type base64 encoding scheme. + * + * @return A Base64 decoder. + */ + public static Decoder getDecoder() { + return Decoder.RFC4648; + } + + /** + * Returns a {@link Decoder} that decodes using the + * URL and Filename safe type base64 + * encoding scheme. + * + * @return A Base64 decoder. + */ + public static Decoder getUrlDecoder() { + return Decoder.RFC4648_URLSAFE; + } + + /** + * Returns a {@link Decoder} that decodes using the + * MIME type base64 decoding scheme. + * + * @return A Base64 decoder. + */ + public static Decoder getMimeDecoder() { + return Decoder.RFC2045; + } + + /** + * This class implements an encoder for encoding byte data using + * the Base64 encoding scheme as specified in RFC 4648 and RFC 2045. + * + *

Instances of {@link Encoder} class are safe for use by + * multiple concurrent threads. + * + *

Unless otherwise noted, passing a {@code null} argument to + * a method of this class will cause a + * {@link java.lang.NullPointerException NullPointerException} to + * be thrown. + * + * @see Decoder + * @since 1.8 + */ + public static class Encoder { + + private final byte[] newline; + private final int linemax; + private final boolean isURL; + private final boolean doPadding; + + private Encoder(boolean isURL, byte[] newline, int linemax, boolean doPadding) { + this.isURL = isURL; + this.newline = newline; + this.linemax = linemax; + this.doPadding = doPadding; + } + + /** + * This array is a lookup table that translates 6-bit positive integer + * index values into their "Base64 Alphabet" equivalents as specified + * in "Table 1: The Base64 Alphabet" of RFC 2045 (and RFC 4648). + */ + private static final char[] toBase64 = { + 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', + 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z', + 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', + 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z', + '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '+', '/' + }; + + /** + * It's the lookup table for "URL and Filename safe Base64" as specified + * in Table 2 of the RFC 4648, with the '+' and '/' changed to '-' and + * '_'. This table is used when BASE64_URL is specified. + */ + private static final char[] toBase64URL = { + 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', + 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z', + 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', + 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z', + '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '-', '_' + }; + + private static final int MIMELINEMAX = 76; + private static final byte[] CRLF = new byte[] {'\r', '\n'}; + + static final Encoder RFC4648 = new Encoder(false, null, -1, true); + static final Encoder RFC4648_URLSAFE = new Encoder(true, null, -1, true); + static final Encoder RFC2045 = new Encoder(false, CRLF, MIMELINEMAX, true); + + private final int outLength(int srclen) { + int len = 0; + if (doPadding) { + len = 4 * ((srclen + 2) / 3); + } else { + int n = srclen % 3; + len = 4 * (srclen / 3) + (n == 0 ? 0 : n + 1); + } + if (linemax > 0) // line separators + len += (len - 1) / linemax * newline.length; + return len; + } + + /** + * Encodes all bytes from the specified byte array into a newly-allocated + * byte array using the {@link Base64} encoding scheme. The returned byte + * array is of the length of the resulting bytes. + * + * @param src + * the byte array to encode + * @return A newly-allocated byte array containing the resulting + * encoded bytes. + */ + public byte[] encode(byte[] src) { + int len = outLength(src.length); // dst array size + byte[] dst = new byte[len]; + int ret = encode0(src, 0, src.length, dst); + if (ret != dst.length) + return Arrays.copyOf(dst, ret); + return dst; + } + + /** + * Encodes all bytes from the specified byte array using the + * {@link Base64} encoding scheme, writing the resulting bytes to the + * given output byte array, starting at offset 0. + * + *

It is the responsibility of the invoker of this method to make + * sure the output byte array {@code dst} has enough space for encoding + * all bytes from the input byte array. No bytes will be written to the + * output byte array if the output byte array is not big enough. + * + * @param src + * the byte array to encode + * @param dst + * the output byte array + * @return The number of bytes written to the output byte array + * + * @throws IllegalArgumentException if {@code dst} does not have enough + * space for encoding all input bytes. + */ + public int encode(byte[] src, byte[] dst) { + int len = outLength(src.length); // dst array size + if (dst.length < len) + throw new IllegalArgumentException( + "Output byte array is too small for encoding all input bytes"); + return encode0(src, 0, src.length, dst); + } + + /** + * Encodes the specified byte array into a String using the {@link Base64} + * encoding scheme. + * + *

This method first encodes all input bytes into a base64 encoded + * byte array and then constructs a new String by using the encoded byte + * array and the {@link StandardCharsets#ISO_8859_1 + * ISO-8859-1} charset. + * + *

In other words, an invocation of this method has exactly the same + * effect as invoking + * {@code new String(encode(src), StandardCharsets.ISO_8859_1)}. + * + * @param src + * the byte array to encode + * @return A String containing the resulting Base64 encoded characters + */ + @SuppressWarnings("deprecation") + public String encodeToString(byte[] src) { + byte[] encoded = encode(src); + return new String(encoded, 0, 0, encoded.length); + } + + /** + * Encodes all remaining bytes from the specified byte buffer into + * a newly-allocated ByteBuffer using the {@link Base64} encoding + * scheme. + * + * Upon return, the source buffer's position will be updated to + * its limit; its limit will not have been changed. The returned + * output buffer's position will be zero and its limit will be the + * number of resulting encoded bytes. + * + * @param buffer + * the source ByteBuffer to encode + * @return A newly-allocated byte buffer containing the encoded bytes. + */ + public ByteBuffer encode(ByteBuffer buffer) { + int len = outLength(buffer.remaining()); + byte[] dst = new byte[len]; + int ret = 0; + if (buffer.hasArray()) { + ret = encode0(buffer.array(), + buffer.arrayOffset() + buffer.position(), + buffer.arrayOffset() + buffer.limit(), + dst); + buffer.position(buffer.limit()); + } else { + byte[] src = new byte[buffer.remaining()]; + buffer.get(src); + ret = encode0(src, 0, src.length, dst); + } + if (ret != dst.length) + dst = Arrays.copyOf(dst, ret); + return ByteBuffer.wrap(dst); + } + + /** + * Wraps an output stream for encoding byte data using the {@link Base64} + * encoding scheme. + * + *

It is recommended to promptly close the returned output stream after + * use, during which it will flush all possible leftover bytes to the underlying + * output stream. Closing the returned output stream will close the underlying + * output stream. + * + * @param os + * the output stream. + * @return the output stream for encoding the byte data into the + * specified Base64 encoded format + */ + public OutputStream wrap(OutputStream os) { + Objects.requireNonNull(os); + return new EncOutputStream(os, isURL ? toBase64URL : toBase64, + newline, linemax, doPadding); + } + + /** + * Returns an encoder instance that encodes equivalently to this one, + * but without adding any padding character at the end of the encoded + * byte data. + * + *

The encoding scheme of this encoder instance is unaffected by + * this invocation. The returned encoder instance should be used for + * non-padding encoding operation. + * + * @return an equivalent encoder that encodes without adding any + * padding character at the end + */ + public Encoder withoutPadding() { + if (!doPadding) + return this; + return new Encoder(isURL, newline, linemax, false); + } + + private int encode0(byte[] src, int off, int end, byte[] dst) { + char[] base64 = isURL ? toBase64URL : toBase64; + int sp = off; + int slen = (end - off) / 3 * 3; + int sl = off + slen; + if (linemax > 0 && slen > linemax / 4 * 3) + slen = linemax / 4 * 3; + int dp = 0; + while (sp < sl) { + int sl0 = Math.min(sp + slen, sl); + for (int sp0 = sp, dp0 = dp ; sp0 < sl0; ) { + int bits = (src[sp0++] & 0xff) << 16 | + (src[sp0++] & 0xff) << 8 | + (src[sp0++] & 0xff); + dst[dp0++] = (byte)base64[(bits >>> 18) & 0x3f]; + dst[dp0++] = (byte)base64[(bits >>> 12) & 0x3f]; + dst[dp0++] = (byte)base64[(bits >>> 6) & 0x3f]; + dst[dp0++] = (byte)base64[bits & 0x3f]; + } + int dlen = (sl0 - sp) / 3 * 4; + dp += dlen; + sp = sl0; + if (dlen == linemax && sp < end) { + for (byte b : newline){ + dst[dp++] = b; + } + } + } + if (sp < end) { // 1 or 2 leftover bytes + int b0 = src[sp++] & 0xff; + dst[dp++] = (byte)base64[b0 >> 2]; + if (sp == end) { + dst[dp++] = (byte)base64[(b0 << 4) & 0x3f]; + if (doPadding) { + dst[dp++] = PADDING_CHAR; + dst[dp++] = PADDING_CHAR; + } + } else { + int b1 = src[sp++] & 0xff; + dst[dp++] = (byte)base64[(b0 << 4) & 0x3f | (b1 >> 4)]; + dst[dp++] = (byte)base64[(b1 << 2) & 0x3f]; + if (doPadding) { + dst[dp++] = PADDING_CHAR; + } + } + } + return dp; + } + } + + /** + * This class implements a decoder for decoding byte data using the + * Base64 encoding scheme as specified in RFC 4648 and RFC 2045. + * + *

The Base64 padding character {@code PADDING_CHAR} is accepted and + * interpreted as the end of the encoded byte data, but is not + * required. So if the final unit of the encoded byte data only has + * two or three Base64 characters (without the corresponding padding + * character(s) padded), they are decoded as if followed by padding + * character(s). If there is a padding character present in the + * final unit, the correct number of padding character(s) must be + * present, otherwise {@code IllegalArgumentException} ( + * {@code IOException} when reading from a Base64 stream) is thrown + * during decoding. + * + *

Instances of {@link Decoder} class are safe for use by + * multiple concurrent threads. + * + *

Unless otherwise noted, passing a {@code null} argument to + * a method of this class will cause a + * {@link java.lang.NullPointerException NullPointerException} to + * be thrown. + * + * @see Encoder + * @since 1.8 + */ + public static class Decoder { + + private final boolean isURL; + private final boolean isMIME; + + private Decoder(boolean isURL, boolean isMIME) { + this.isURL = isURL; + this.isMIME = isMIME; + } + + /** + * Lookup table for decoding unicode characters drawn from the + * "Base64 Alphabet" (as specified in Table 1 of RFC 2045) into + * their 6-bit positive integer equivalents. Characters that + * are not in the Base64 alphabet but fall within the bounds of + * the array are encoded to -1. + * + */ + private static final int[] fromBase64 = new int[256]; + static { + Arrays.fill(fromBase64, -1); + for (int i = 0; i < Encoder.toBase64.length; i++) + fromBase64[Encoder.toBase64[i]] = i; + fromBase64[PADDING_CHAR] = -2; + } + + /** + * Lookup table for decoding "URL and Filename safe Base64 Alphabet" + * as specified in Table2 of the RFC 4648. + */ + private static final int[] fromBase64URL = new int[256]; + + static { + Arrays.fill(fromBase64URL, -1); + for (int i = 0; i < Encoder.toBase64URL.length; i++) + fromBase64URL[Encoder.toBase64URL[i]] = i; + fromBase64URL[PADDING_CHAR] = -2; + } + + static final Decoder RFC4648 = new Decoder(false, false); + static final Decoder RFC4648_URLSAFE = new Decoder(true, false); + static final Decoder RFC2045 = new Decoder(false, true); + + /** + * Decodes all bytes from the input byte array using the {@link Base64} + * encoding scheme, writing the results into a newly-allocated output + * byte array. The returned byte array is of the length of the resulting + * bytes. + * + * @param src + * the byte array to decode + * + * @return A newly-allocated byte array containing the decoded bytes. + * + * @throws IllegalArgumentException + * if {@code src} is not in valid Base64 scheme + */ + public byte[] decode(byte[] src) { + byte[] dst = new byte[outLength(src, 0, src.length)]; + int ret = decode0(src, 0, src.length, dst); + if (ret != dst.length) { + dst = Arrays.copyOf(dst, ret); + } + return dst; + } + + /** + * Decodes a Base64 encoded String into a newly-allocated byte array + * using the {@link Base64} encoding scheme. + * + *

An invocation of this method has exactly the same effect as invoking + * {@code decode(src.getBytes(StandardCharsets.ISO_8859_1))} + * + * @param src + * the string to decode + * + * @return A newly-allocated byte array containing the decoded bytes. + * + * @throws IllegalArgumentException + * if {@code src} is not in valid Base64 scheme + */ + public byte[] decode(String src) { + return decode(src.getBytes(StandardCharsets.ISO_8859_1)); + } + + /** + * Decodes all bytes from the input byte array using the {@link Base64} + * encoding scheme, writing the results into the given output byte array, + * starting at offset 0. + * + *

It is the responsibility of the invoker of this method to make + * sure the output byte array {@code dst} has enough space for decoding + * all bytes from the input byte array. No bytes will be be written to + * the output byte array if the output byte array is not big enough. + * + *

If the input byte array is not in valid Base64 encoding scheme + * then some bytes may have been written to the output byte array before + * IllegalargumentException is thrown. + * + * @param src + * the byte array to decode + * @param dst + * the output byte array + * + * @return The number of bytes written to the output byte array + * + * @throws IllegalArgumentException + * if {@code src} is not in valid Base64 scheme, or {@code dst} + * does not have enough space for decoding all input bytes. + */ + public int decode(byte[] src, byte[] dst) { + int len = outLength(src, 0, src.length); + if (dst.length < len) + throw new IllegalArgumentException( + "Output byte array is too small for decoding all input bytes"); + return decode0(src, 0, src.length, dst); + } + + /** + * Decodes all bytes from the input byte buffer using the {@link Base64} + * encoding scheme, writing the results into a newly-allocated ByteBuffer. + * + *

Upon return, the source buffer's position will be updated to + * its limit; its limit will not have been changed. The returned + * output buffer's position will be zero and its limit will be the + * number of resulting decoded bytes + * + *

{@code IllegalArgumentException} is thrown if the input buffer + * is not in valid Base64 encoding scheme. The position of the input + * buffer will not be advanced in this case. + * + * @param buffer + * the ByteBuffer to decode + * + * @return A newly-allocated byte buffer containing the decoded bytes + * + * @throws IllegalArgumentException + * if {@code src} is not in valid Base64 scheme. + */ + public ByteBuffer decode(ByteBuffer buffer) { + int pos0 = buffer.position(); + try { + byte[] src; + int sp, sl; + if (buffer.hasArray()) { + src = buffer.array(); + sp = buffer.arrayOffset() + buffer.position(); + sl = buffer.arrayOffset() + buffer.limit(); + buffer.position(buffer.limit()); + } else { + src = new byte[buffer.remaining()]; + buffer.get(src); + sp = 0; + sl = src.length; + } + byte[] dst = new byte[outLength(src, sp, sl)]; + return ByteBuffer.wrap(dst, 0, decode0(src, sp, sl, dst)); + } catch (IllegalArgumentException iae) { + buffer.position(pos0); + throw iae; + } + } + + /** + * Returns an input stream for decoding {@link Base64} encoded byte stream. + * + *

The {@code read} methods of the returned {@code InputStream} will + * throw {@code IOException} when reading bytes that cannot be decoded. + * + *

Closing the returned input stream will close the underlying + * input stream. + * + * @param is + * the input stream + * + * @return the input stream for decoding the specified Base64 encoded + * byte stream + */ + public InputStream wrap(InputStream is) { + Objects.requireNonNull(is); + return new DecInputStream(is, isURL ? fromBase64URL : fromBase64, isMIME); + } + + private int outLength(byte[] src, int sp, int sl) { + int[] base64 = isURL ? fromBase64URL : fromBase64; + int paddings = 0; + int len = sl - sp; + if (len == 0) + return 0; + if (len < 2) { + if (isMIME && base64[0] == -1) + return 0; + throw new IllegalArgumentException( + "Input byte[] should at least have 2 bytes for base64 bytes"); + } + if (isMIME) { + // scan all bytes to fill out all non-alphabet. a performance + // trade-off of pre-scan or Arrays.copyOf + int n = 0; + while (sp < sl) { + int b = src[sp++] & 0xff; + if (b == PADDING_CHAR) { + len -= (sl - sp + 1); + break; + } + if ((b = base64[b]) == -1) + n++; + } + len -= n; + } else { + if (src[sl - 1] == PADDING_CHAR) { + paddings++; + if (src[sl - 2] == PADDING_CHAR) + paddings++; + } + } + if (paddings == 0 && (len & 0x3) != 0) + paddings = 4 - (len & 0x3); + return 3 * ((len + 3) / 4) - paddings; + } + + private int decode0(byte[] src, int sp, int sl, byte[] dst) { + int[] base64 = isURL ? fromBase64URL : fromBase64; + int dp = 0; + int bits = 0; + int shiftto = 18; // pos of first byte of 4-byte atom + while (sp < sl) { + int b = src[sp++] & 0xff; + if ((b = base64[b]) < 0) { + if (b == -2) { // padding byte PADDING_CHAR + // = shiftto==18 unnecessary padding + // x= shiftto==12 a dangling single x + // x to be handled together with non-padding case + // xx= shiftto==6&&sp==sl missing last = + // xx=y shiftto==6 last is not = + if (shiftto == 6 && (sp == sl || src[sp++] != PADDING_CHAR) || + shiftto == 18) { + throw new IllegalArgumentException( + "Input byte array has wrong 4-byte ending unit"); + } + break; + } + if (isMIME) // skip if for rfc2045 + continue; + else + throw new IllegalArgumentException( + "Illegal base64 character " + + Integer.toString(src[sp - 1], 16)); + } + bits |= (b << shiftto); + shiftto -= 6; + if (shiftto < 0) { + dst[dp++] = (byte)(bits >> 16); + dst[dp++] = (byte)(bits >> 8); + dst[dp++] = (byte)(bits); + shiftto = 18; + bits = 0; + } + } + // reached end of byte array or hit padding PADDING_CHAR characters. + if (shiftto == 6) { + dst[dp++] = (byte)(bits >> 16); + } else if (shiftto == 0) { + dst[dp++] = (byte)(bits >> 16); + dst[dp++] = (byte)(bits >> 8); + } else if (shiftto == 12) { + // dangling single "x", incorrectly encoded. + throw new IllegalArgumentException( + "Last unit does not have enough valid bits"); + } + // anything left is invalid, if is not MIME. + // if MIME, ignore all non-base64 character + while (sp < sl) { + if (isMIME && base64[src[sp++]] < 0) + continue; + throw new IllegalArgumentException( + "Input byte array has incorrect ending byte at " + sp); + } + return dp; + } + } + + /* + * An output stream for encoding bytes into the Base64. + */ + private static class EncOutputStream extends FilterOutputStream { + + private int leftover = 0; + private int b0, b1, b2; + private boolean closed = false; + + private final char[] base64; // byte->base64 mapping + private final byte[] newline; // line separator, if needed + private final int linemax; + private final boolean doPadding;// whether or not to pad + private int linepos = 0; + + EncOutputStream(OutputStream os, char[] base64, + byte[] newline, int linemax, boolean doPadding) { + super(os); + this.base64 = base64; + this.newline = newline; + this.linemax = linemax; + this.doPadding = doPadding; + } + + @Override + public void write(int b) throws IOException { + byte[] buf = new byte[1]; + buf[0] = (byte)(b & 0xff); + write(buf, 0, 1); + } + + private void checkNewline() throws IOException { + if (linepos == linemax) { + out.write(newline); + linepos = 0; + } + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + if (closed) + throw new IOException("Stream is closed"); + if (off < 0 || len < 0 || len > b.length - off) + throw new ArrayIndexOutOfBoundsException(); + if (len == 0) + return; + if (leftover != 0) { + if (leftover == 1) { + b1 = b[off++] & 0xff; + len--; + if (len == 0) { + leftover++; + return; + } + } + b2 = b[off++] & 0xff; + len--; + checkNewline(); + out.write(base64[b0 >> 2]); + out.write(base64[(b0 << 4) & 0x3f | (b1 >> 4)]); + out.write(base64[(b1 << 2) & 0x3f | (b2 >> 6)]); + out.write(base64[b2 & 0x3f]); + linepos += 4; + } + int nBits24 = len / 3; + leftover = len - (nBits24 * 3); + while (nBits24-- > 0) { + checkNewline(); + int bits = (b[off++] & 0xff) << 16 | + (b[off++] & 0xff) << 8 | + (b[off++] & 0xff); + out.write(base64[(bits >>> 18) & 0x3f]); + out.write(base64[(bits >>> 12) & 0x3f]); + out.write(base64[(bits >>> 6) & 0x3f]); + out.write(base64[bits & 0x3f]); + linepos += 4; + } + if (leftover == 1) { + b0 = b[off++] & 0xff; + } else if (leftover == 2) { + b0 = b[off++] & 0xff; + b1 = b[off++] & 0xff; + } + } + + @Override + public void close() throws IOException { + if (!closed) { + closed = true; + if (leftover == 1) { + checkNewline(); + out.write(base64[b0 >> 2]); + out.write(base64[(b0 << 4) & 0x3f]); + if (doPadding) { + out.write(PADDING_CHAR); + out.write(PADDING_CHAR); + } + } else if (leftover == 2) { + checkNewline(); + out.write(base64[b0 >> 2]); + out.write(base64[(b0 << 4) & 0x3f | (b1 >> 4)]); + out.write(base64[(b1 << 2) & 0x3f]); + if (doPadding) { + out.write(PADDING_CHAR); + } + } + leftover = 0; + out.close(); + } + } + } + + /* + * An input stream for decoding Base64 bytes + */ + private static class DecInputStream extends InputStream { + + private final InputStream is; + private final boolean isMIME; + private final int[] base64; // base64 -> byte mapping + private int bits = 0; // 24-bit buffer for decoding + private int nextin = 18; // next available "off" in "bits" for input; + // -> 18, 12, 6, 0 + private int nextout = -8; // next available "off" in "bits" for output; + // -> 8, 0, -8 (no byte for output) + private boolean eof = false; + private boolean closed = false; + + DecInputStream(InputStream is, int[] base64, boolean isMIME) { + this.is = is; + this.base64 = base64; + this.isMIME = isMIME; + } + + private byte[] sbBuf = new byte[1]; + + @Override + public int read() throws IOException { + return read(sbBuf, 0, 1) == -1 ? -1 : sbBuf[0] & 0xff; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + if (closed) + throw new IOException("Stream is closed"); + if (eof && nextout < 0) // eof and no leftover + return -1; + if (off < 0 || len < 0 || len > b.length - off) + throw new IndexOutOfBoundsException(); + int oldOff = off; + if (nextout >= 0) { // leftover output byte(s) in bits buf + do { + if (len == 0) + return off - oldOff; + b[off++] = (byte)(bits >> nextout); + len--; + nextout -= 8; + } while (nextout >= 0); + bits = 0; + } + while (len > 0) { + int v = is.read(); + if (v == -1) { + eof = true; + if (nextin != 18) { + if (nextin == 12) + throw new IOException("Base64 stream has one un-decoded dangling byte."); + // treat ending xx/xxx without padding character legal. + // same logic as v == PADDING_CHAR below + b[off++] = (byte)(bits >> (16)); + len--; + if (nextin == 0) { // only one padding byte + if (len == 0) { // no enough output space + bits >>= 8; // shift to lowest byte + nextout = 0; + } else { + b[off++] = (byte) (bits >> 8); + } + } + } + if (off == oldOff) + return -1; + else + return off - oldOff; + } + if (v == PADDING_CHAR) { // padding byte(s) + // = shiftto==18 unnecessary padding + // x= shiftto==12 dangling x, invalid unit + // xx= shiftto==6 && missing last PADDING_CHAR + // xx=y or last is not PADDING_CHAR + if (nextin == 18 || nextin == 12 || + nextin == 6 && is.read() != PADDING_CHAR) { + throw new IOException("Illegal base64 ending sequence:" + nextin); + } + b[off++] = (byte)(bits >> (16)); + len--; + if (nextin == 0) { // only one padding byte + if (len == 0) { // no enough output space + bits >>= 8; // shift to lowest byte + nextout = 0; + } else { + b[off++] = (byte) (bits >> 8); + } + } + eof = true; + break; + } + if ((v = base64[v]) == -1) { + if (isMIME) // skip if for rfc2045 + continue; + else + throw new IOException("Illegal base64 character " + + Integer.toString(v, 16)); + } + bits |= (v << nextin); + if (nextin == 0) { + nextin = 18; // clear for next + nextout = 16; + while (nextout >= 0) { + b[off++] = (byte)(bits >> nextout); + len--; + nextout -= 8; + if (len == 0 && nextout >= 0) { // don't clean "bits" + return off - oldOff; + } + } + bits = 0; + } else { + nextin -= 6; + } + } + return off - oldOff; + } + + @Override + public int available() throws IOException { + if (closed) + throw new IOException("Stream is closed"); + return is.available(); // TBD: + } + + @Override + public void close() throws IOException { + if (!closed) { + closed = true; + is.close(); + } + } + } + + /** + * This is a tiny adjustment for BASE64 encode and decode. + * I replace the padding char of base64 to '*' from '='. + * + * We found that the dubbox's head didn't support '=' as a part of value. :( + * + * @author wusheng + */ + private static byte PADDING_CHAR = '*'; + +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/base64/Objects.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/base64/Objects.java new file mode 100644 index 000000000..d65ea3032 --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/base64/Objects.java @@ -0,0 +1,157 @@ +package org.skywalking.apm.agent.core.context.ids.base64; + +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; + +/** + * This class consists of {@code static} utility methods for operating + * on objects. These utilities include {@code null}-safe or {@code + * null}-tolerant methods for computing the hash code of an object, + * returning a string for an object, and comparing two objects. + * + * @since 1.7 + */ +public final class Objects { + private Objects() { + throw new AssertionError("No java.util.Objects instances for you!"); + } + + /** + * Returns {@code true} if the arguments are equal to each other + * and {@code false} otherwise. + * Consequently, if both arguments are {@code null}, {@code true} + * is returned and if exactly one argument is {@code null}, {@code + * false} is returned. Otherwise, equality is determined by using + * the {@link Object#equals equals} method of the first + * argument. + * + * @param a an object + * @param b an object to be compared with {@code a} for equality + * @return {@code true} if the arguments are equal to each other + * and {@code false} otherwise + * @see Object#equals(Object) + */ + public static boolean equals(Object a, Object b) { + return (a == b) || (a != null && a.equals(b)); + } + + /** + * Returns the hash code of a non-{@code null} argument and 0 for + * a {@code null} argument. + * + * @param o an object + * @return the hash code of a non-{@code null} argument and 0 for + * a {@code null} argument + * @see Object#hashCode + */ + public static int hashCode(Object o) { + return o != null ? o.hashCode() : 0; + } + + /** + * Generates a hash code for a sequence of input values. The hash + * code is generated as if all the input values were placed into an + * array, and that array were hashed by calling {@link + * Arrays#hashCode(Object[])}. + * + *

This method is useful for implementing {@link + * Object#hashCode()} on objects containing multiple fields. For + * example, if an object that has three fields, {@code x}, {@code + * y}, and {@code z}, one could write: + * + *

+     * @Override public int hashCode() {
+     *     return Objects.hash(x, y, z);
+     * }
+     * 
+ * + * Warning: When a single object reference is supplied, the returned + * value does not equal the hash code of that object reference. This + * value can be computed by calling {@link #hashCode(Object)}. + * + * @param values the values to be hashed + * @return a hash value of the sequence of input values + * @see Arrays#hashCode(Object[]) + * @see List#hashCode + */ + public static int hash(Object... values) { + return Arrays.hashCode(values); + } + + /** + * Returns the result of calling {@code toString} for a non-{@code + * null} argument and {@code "null"} for a {@code null} argument. + * + * @param o an object + * @return the result of calling {@code toString} for a non-{@code + * null} argument and {@code "null"} for a {@code null} argument + * @see Object#toString + * @see String#valueOf(Object) + */ + public static String toString(Object o) { + return String.valueOf(o); + } + + /** + * Returns the result of calling {@code toString} on the first + * argument if the first argument is not {@code null} and returns + * the second argument otherwise. + * + * @param o an object + * @param nullDefault string to return if the first argument is + * {@code null} + * @return the result of calling {@code toString} on the first + * argument if it is not {@code null} and the second argument + * otherwise. + */ + public static String toString(Object o, String nullDefault) { + return (o != null) ? o.toString() : nullDefault; + } + + /** + * Returns 0 if the arguments are identical and {@code + * c.compare(a, b)} otherwise. + * Consequently, if both arguments are {@code null} 0 + * is returned. + * + *

Note that if one of the arguments is {@code null}, a {@code + * NullPointerException} may or may not be thrown depending on + * what ordering policy, if any, the {@link Comparator Comparator} + * chooses to have for {@code null} values. + * + * @param the type of the objects being compared + * @param a an object + * @param b an object to be compared with {@code a} + * @param c the {@code Comparator} to compare the first two arguments + * @return 0 if the arguments are identical and {@code + * c.compare(a, b)} otherwise. + * @see Comparable + * @see Comparator + */ + public static int compare(T a, T b, Comparator c) { + return (a == b) ? 0 : c.compare(a, b); + } + + /** + * Checks that the specified object reference is not {@code null}. This + * method is designed primarily for doing parameter validation in methods + * and constructors, as demonstrated below: + *

+     * public Foo(Bar bar) {
+     *     this.bar = Objects.requireNonNull(bar);
+     * }
+     * 
+ * + * @param obj the object reference to check for nullity + * @param the type of the reference + * @return {@code obj} if not {@code null} + * @throws NullPointerException if {@code obj} is {@code null} + */ + public static T requireNonNull(T obj) { + if (obj == null) + throw new NullPointerException(); + return obj; + } + +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/base64/StandardCharsets.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/base64/StandardCharsets.java new file mode 100644 index 000000000..cdc70f7df --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ids/base64/StandardCharsets.java @@ -0,0 +1,44 @@ +package org.skywalking.apm.agent.core.context.ids.base64; + +import java.nio.charset.Charset; + +/** + * Constant definitions for the standard {@link Charset Charsets}. These + * charsets are guaranteed to be available on every implementation of the Java + * platform. + * + * @see Standard Charsets + * @since 1.7 + */ +public final class StandardCharsets { + + private StandardCharsets() { + throw new AssertionError("No java.nio.charset.StandardCharsets instances for you!"); + } + /** + * Seven-bit ASCII, a.k.a. ISO646-US, a.k.a. the Basic Latin block of the + * Unicode character set + */ + public static final Charset US_ASCII = Charset.forName("US-ASCII"); + /** + * ISO Latin Alphabet No. 1, a.k.a. ISO-LATIN-1 + */ + public static final Charset ISO_8859_1 = Charset.forName("ISO-8859-1"); + /** + * Eight-bit UCS Transformation Format + */ + public static final Charset UTF_8 = Charset.forName("UTF-8"); + /** + * Sixteen-bit UCS Transformation Format, big-endian byte order + */ + public static final Charset UTF_16BE = Charset.forName("UTF-16BE"); + /** + * Sixteen-bit UCS Transformation Format, little-endian byte order + */ + public static final Charset UTF_16LE = Charset.forName("UTF-16LE"); + /** + * Sixteen-bit UCS Transformation Format, byte order identified by an + * optional byte-order mark + */ + public static final Charset UTF_16 = Charset.forName("UTF-16"); +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/TraceSegment.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/TraceSegment.java index 000e71956..6857c8b8f 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/TraceSegment.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/TraceSegment.java @@ -6,6 +6,7 @@ import org.skywalking.apm.agent.core.conf.RemoteDownstreamConfig; import org.skywalking.apm.agent.core.context.ids.DistributedTraceId; import org.skywalking.apm.agent.core.context.ids.DistributedTraceIds; import org.skywalking.apm.agent.core.context.ids.GlobalIdGenerator; +import org.skywalking.apm.agent.core.context.ids.ID; import org.skywalking.apm.agent.core.context.ids.NewDistributedTraceId; import org.skywalking.apm.logging.ILog; import org.skywalking.apm.logging.LogManager; @@ -31,7 +32,7 @@ public class TraceSegment { * The id of this trace segment. * Every segment has its unique-global-id. */ - private String traceSegmentId; + private ID traceSegmentId; /** * The refs of parent trace segments, except the primary one. @@ -71,7 +72,7 @@ public class TraceSegment { * and generate a new segment id. */ public TraceSegment() { - this.traceSegmentId = GlobalIdGenerator.generate(ID_TYPE); + this.traceSegmentId = GlobalIdGenerator.generate(); this.spans = new LinkedList(); this.relatedGlobalTraces = new DistributedTraceIds(); this.relatedGlobalTraces.append(new NewDistributedTraceId()); @@ -93,16 +94,9 @@ public class TraceSegment { /** * Establish the line between this segment and all relative global trace ids. - * - * @param distributedTraceIds multi global trace ids. @see {@link DistributedTraceId} */ - public void relatedGlobalTraces(List distributedTraceIds) { - if (distributedTraceIds == null || distributedTraceIds.size() == 0) { - return; - } - for (DistributedTraceId distributedTraceId : distributedTraceIds) { - relatedGlobalTraces.append(distributedTraceId); - } + public void relatedGlobalTraces(DistributedTraceId distributedTraceId) { + relatedGlobalTraces.append(distributedTraceId); } /** @@ -124,7 +118,7 @@ public class TraceSegment { return this; } - public String getTraceSegmentId() { + public ID getTraceSegmentId() { return traceSegmentId; } @@ -164,13 +158,13 @@ public class TraceSegment { public UpstreamSegment transform() { UpstreamSegment.Builder upstreamBuilder = UpstreamSegment.newBuilder(); for (DistributedTraceId distributedTraceId : getRelatedGlobalTraces()) { - upstreamBuilder = upstreamBuilder.addGlobalTraceIds(distributedTraceId.get()); + upstreamBuilder = upstreamBuilder.addGlobalTraceIds(distributedTraceId.toUniqueId()); } TraceSegmentObject.Builder traceSegmentBuilder = TraceSegmentObject.newBuilder(); /** * Trace Segment */ - traceSegmentBuilder.setTraceSegmentId(this.traceSegmentId); + traceSegmentBuilder.setTraceSegmentId(this.traceSegmentId.transform()); // TraceSegmentReference if (this.refs != null) { for (TraceSegmentRef ref : this.refs) { diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/TraceSegmentRef.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/TraceSegmentRef.java index 1e2da1217..74dc4bff9 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/TraceSegmentRef.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/TraceSegmentRef.java @@ -2,6 +2,7 @@ package org.skywalking.apm.agent.core.context.trace; import org.skywalking.apm.agent.core.context.ContextCarrier; import org.skywalking.apm.agent.core.context.ContextSnapshot; +import org.skywalking.apm.agent.core.context.ids.ID; import org.skywalking.apm.agent.core.dictionary.DictionaryUtil; import org.skywalking.apm.network.proto.RefType; import org.skywalking.apm.network.proto.TraceSegmentReference; @@ -15,7 +16,7 @@ import org.skywalking.apm.network.proto.TraceSegmentReference; public class TraceSegmentRef { private SegmentRefType type; - private String traceSegmentId; + private ID traceSegmentId; private int spanId = -1; @@ -25,9 +26,13 @@ public class TraceSegmentRef { private int peerId = DictionaryUtil.nullValue(); - private String operationName; + private String entryOperationName; - private int operationId = DictionaryUtil.nullValue(); + private int entryOperationId = DictionaryUtil.nullValue(); + + private String parentOperationName; + + private int parentOperationId = DictionaryUtil.nullValue(); /** * Transform a {@link ContextCarrier} to the TraceSegmentRef @@ -47,9 +52,15 @@ public class TraceSegmentRef { } String entryOperationName = carrier.getEntryOperationName(); if (entryOperationName.charAt(0) == '#') { - this.operationName = entryOperationName.substring(1); + this.entryOperationName = entryOperationName.substring(1); } else { - this.operationId = Integer.parseInt(entryOperationName); + this.entryOperationId = Integer.parseInt(entryOperationName); + } + String parentOperationName = carrier.getParentOperationName(); + if (parentOperationName.charAt(0) == '#') { + this.parentOperationName = parentOperationName.substring(1); + } else { + this.parentOperationId = Integer.parseInt(parentOperationName); } } @@ -57,14 +68,26 @@ public class TraceSegmentRef { this.type = SegmentRefType.CROSS_THREAD; this.traceSegmentId = snapshot.getTraceSegmentId(); this.spanId = snapshot.getSpanId(); + String entryOperationName = snapshot.getEntryOperationName(); + if (entryOperationName.charAt(0) == '#') { + this.entryOperationName = entryOperationName.substring(1); + } else { + this.entryOperationId = Integer.parseInt(entryOperationName); + } + String parentOperationName = snapshot.getParentOperationName(); + if (parentOperationName.charAt(0) == '#') { + this.parentOperationName = parentOperationName.substring(1); + } else { + this.parentOperationId = Integer.parseInt(parentOperationName); + } } - public String getOperationName() { - return operationName; + public String getEntryOperationName() { + return entryOperationName; } - public int getOperationId() { - return operationId; + public int getEntryOperationId() { + return entryOperationId; } public TraceSegmentReference transform() { @@ -77,17 +100,22 @@ public class TraceSegmentRef { } else { refBuilder.setNetworkAddressId(peerId); } - if (operationId == DictionaryUtil.nullValue()) { - refBuilder.setEntryServiceName(operationName); - } else { - refBuilder.setEntryServiceId(operationId); - } } else { refBuilder.setRefType(RefType.CrossThread); } - refBuilder.setParentTraceSegmentId(traceSegmentId); - refBuilder.setParentSpanId(spanId); + refBuilder.setParentTraceSegmentId(traceSegmentId.transform()); + refBuilder.setParentSpanId(spanId); + if (entryOperationId == DictionaryUtil.nullValue()) { + refBuilder.setEntryServiceName(entryOperationName); + } else { + refBuilder.setEntryServiceId(entryOperationId); + } + if (parentOperationId == DictionaryUtil.nullValue()) { + refBuilder.setParentServiceName(parentOperationName); + } else { + refBuilder.setParentServiceId(parentOperationId); + } return refBuilder.build(); } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/dictionary/ApplicationDictionary.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/dictionary/ApplicationDictionary.java index 295f4476d..016a607dd 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/dictionary/ApplicationDictionary.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/dictionary/ApplicationDictionary.java @@ -9,6 +9,8 @@ import org.skywalking.apm.network.proto.ApplicationMapping; import org.skywalking.apm.network.proto.ApplicationRegisterServiceGrpc; import org.skywalking.apm.network.proto.KeyWithIntegerValue; +import static org.skywalking.apm.agent.core.conf.Config.Dictionary.APPLICATION_CODE_BUFFER_SIZE; + /** * Map of application id to application code, which is from the collector side. * @@ -24,7 +26,9 @@ public enum ApplicationDictionary { if (applicationId != null) { return new Found(applicationId); } else { - unRegisterApplications.add(applicationCode); + if (applicationDictionary.size() + unRegisterApplications.size() < APPLICATION_CODE_BUFFER_SIZE) { + unRegisterApplications.add(applicationCode); + } return new NotFound(); } } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/dictionary/OperationNameDictionary.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/dictionary/OperationNameDictionary.java index 9f575b50a..2abd18cb5 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/dictionary/OperationNameDictionary.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/dictionary/OperationNameDictionary.java @@ -10,6 +10,8 @@ import org.skywalking.apm.network.proto.ServiceNameElement; import org.skywalking.apm.network.proto.ServiceNameMappingCollection; import org.skywalking.apm.network.proto.ServiceNameMappingElement; +import static org.skywalking.apm.agent.core.conf.Config.Dictionary.OPERATION_NAME_BUFFER_SIZE; + /** * @author wusheng */ @@ -19,12 +21,17 @@ public enum OperationNameDictionary { private Set unRegisterOperationNames = new ConcurrentSet(); public PossibleFound find(int applicationId, String operationName) { + if (operationName == null || operationName.length() == 0) { + return new NotFound(); + } OperationNameKey key = new OperationNameKey(applicationId, operationName); Integer operationId = operationNameDictionary.get(key); if (operationId != null) { return new Found(applicationId); } else { - unRegisterOperationNames.add(key); + if (operationNameDictionary.size() + unRegisterOperationNames.size() < OPERATION_NAME_BUFFER_SIZE) { + unRegisterOperationNames.add(key); + } return new NotFound(); } } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/jvm/JVMService.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/jvm/JVMService.java index bf3189c76..6ac9a0a2b 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/jvm/JVMService.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/jvm/JVMService.java @@ -1,15 +1,14 @@ package org.skywalking.apm.agent.core.jvm; import io.grpc.ManagedChannel; -import java.text.SimpleDateFormat; -import java.util.Date; import java.util.LinkedList; import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.ReentrantLock; import org.skywalking.apm.agent.core.boot.BootService; import org.skywalking.apm.agent.core.boot.ServiceManager; +import org.skywalking.apm.agent.core.conf.Config; import org.skywalking.apm.agent.core.conf.RemoteDownstreamConfig; import org.skywalking.apm.agent.core.dictionary.DictionaryUtil; import org.skywalking.apm.agent.core.jvm.cpu.CPUProvider; @@ -36,16 +35,14 @@ import static org.skywalking.apm.agent.core.remote.GRPCChannelStatus.CONNECTED; */ public class JVMService implements BootService, Runnable { private static final ILog logger = LogManager.getLogger(JVMService.class); - private ReentrantLock lock = new ReentrantLock(); - private volatile LinkedList buffer = new LinkedList(); - private SimpleDateFormat sdf = new SimpleDateFormat("ss"); + private LinkedBlockingQueue queue; private volatile ScheduledFuture collectMetricFuture; private volatile ScheduledFuture sendMetricFuture; - private volatile int lastBlockIdx = -1; private Sender sender; @Override public void beforeBoot() throws Throwable { + queue = new LinkedBlockingQueue(Config.Jvm.BUFFER_SIZE); sender = new Sender(); ServiceManager.INSTANCE.findService(GRPCChannelManager.class).addChannelListener(sender); } @@ -57,7 +54,7 @@ public class JVMService implements BootService, Runnable { .scheduleAtFixedRate(this, 0, 1, TimeUnit.SECONDS); sendMetricFuture = Executors .newSingleThreadScheduledExecutor() - .scheduleAtFixedRate(sender, 0, 15, TimeUnit.SECONDS); + .scheduleAtFixedRate(sender, 0, 1, TimeUnit.SECONDS); } @Override @@ -65,38 +62,33 @@ public class JVMService implements BootService, Runnable { } + @Override + public void shutdown() throws Throwable { + collectMetricFuture.cancel(true); + sendMetricFuture.cancel(true); + } + @Override public void run() { if (RemoteDownstreamConfig.Agent.APPLICATION_ID != DictionaryUtil.nullValue() && RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID != DictionaryUtil.nullValue() ) { long currentTimeMillis = System.currentTimeMillis(); - Date day = new Date(currentTimeMillis); - String second = sdf.format(day); - int blockIndex = Integer.parseInt(second) / 15; - if (blockIndex != lastBlockIdx) { - lastBlockIdx = blockIndex; - try { - JVMMetric.Builder jvmBuilder = JVMMetric.newBuilder(); - jvmBuilder.setTime(currentTimeMillis); - jvmBuilder.setCpu(CPUProvider.INSTANCE.getCpuMetric()); - jvmBuilder.addAllMemory(MemoryProvider.INSTANCE.getMemoryMetricList()); - jvmBuilder.addAllMemoryPool(MemoryPoolProvider.INSTANCE.getMemoryPoolMetricList()); - jvmBuilder.addAllGc(GCProvider.INSTANCE.getGCList()); + try { + JVMMetric.Builder jvmBuilder = JVMMetric.newBuilder(); + jvmBuilder.setTime(currentTimeMillis); + jvmBuilder.setCpu(CPUProvider.INSTANCE.getCpuMetric()); + jvmBuilder.addAllMemory(MemoryProvider.INSTANCE.getMemoryMetricList()); + jvmBuilder.addAllMemoryPool(MemoryPoolProvider.INSTANCE.getMemoryPoolMetricList()); + jvmBuilder.addAllGc(GCProvider.INSTANCE.getGCList()); - JVMMetric jvmMetric = jvmBuilder.build(); - lock.lock(); - try { - buffer.add(jvmMetric); - while (buffer.size() > 4) { - buffer.removeFirst(); - } - } finally { - lock.unlock(); - } - } catch (Exception e) { - logger.error(e, "Collect JVM info fail."); + JVMMetric jvmMetric = jvmBuilder.build(); + if (queue.offer(jvmMetric)) { + queue.poll(); + queue.offer(jvmMetric); } + } catch (Exception e) { + logger.error(e, "Collect JVM info fail."); } } } @@ -113,13 +105,9 @@ public class JVMService implements BootService, Runnable { if (status == GRPCChannelStatus.CONNECTED) { try { JVMMetrics.Builder builder = JVMMetrics.newBuilder(); - lock.lock(); - try { - builder.addAllMetrics(buffer); - buffer.clear(); - } finally { - lock.unlock(); - } + LinkedList buffer = new LinkedList(); + queue.drainTo(buffer); + builder.addAllMetrics(buffer); builder.setApplicationInstanceId(RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID); stub.collect(builder.build()); diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java index af24fc93d..2493616b4 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java @@ -1,6 +1,7 @@ package org.skywalking.apm.agent.core.remote; import io.grpc.ManagedChannel; +import java.util.UUID; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -33,6 +34,7 @@ import static org.skywalking.apm.agent.core.remote.GRPCChannelStatus.CONNECTED; */ public class AppAndServiceRegisterClient implements BootService, GRPCChannelListener, Runnable, TracingContextListener { private static final ILog logger = LogManager.getLogger(AppAndServiceRegisterClient.class); + private static final String PROCESS_UUID = UUID.randomUUID().toString().replaceAll("-", ""); private volatile GRPCChannelStatus status = GRPCChannelStatus.DISCONNECT; private volatile ApplicationRegisterServiceGrpc.ApplicationRegisterServiceBlockingStub applicationRegisterServiceBlockingStub; @@ -77,6 +79,11 @@ public class AppAndServiceRegisterClient implements BootService, GRPCChannelList TracingContext.ListenerManager.add(this); } + @Override + public void shutdown() throws Throwable { + applicationRegisterFuture.cancel(true); + } + @Override public void run() { if (CONNECTED.equals(status)) { @@ -95,6 +102,7 @@ public class AppAndServiceRegisterClient implements BootService, GRPCChannelList ApplicationInstanceMapping instanceMapping = instanceDiscoveryServiceBlockingStub.register(ApplicationInstance.newBuilder() .setApplicationId(RemoteDownstreamConfig.Agent.APPLICATION_ID) + .setAgentUUID(PROCESS_UUID) .setRegisterTime(System.currentTimeMillis()) .build()); if (instanceMapping.getApplicationInstanceId() != DictionaryUtil.nullValue()) { diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/CollectorDiscoveryService.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/CollectorDiscoveryService.java index 76461b985..0b73added 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/CollectorDiscoveryService.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/CollectorDiscoveryService.java @@ -1,6 +1,7 @@ package org.skywalking.apm.agent.core.remote; import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import org.skywalking.apm.agent.core.boot.BootService; import org.skywalking.apm.agent.core.conf.Config; @@ -11,6 +12,8 @@ import org.skywalking.apm.agent.core.conf.Config; * @author wusheng */ public class CollectorDiscoveryService implements BootService { + private ScheduledFuture future; + @Override public void beforeBoot() throws Throwable { @@ -18,7 +21,7 @@ public class CollectorDiscoveryService implements BootService { @Override public void boot() throws Throwable { - Executors.newSingleThreadScheduledExecutor() + future = Executors.newSingleThreadScheduledExecutor() .scheduleAtFixedRate(new DiscoveryRestServiceClient(), 0, Config.Collector.DISCOVERY_CHECK_INTERVAL, TimeUnit.SECONDS); } @@ -27,4 +30,9 @@ public class CollectorDiscoveryService implements BootService { public void afterBoot() throws Throwable { } + + @Override + public void shutdown() throws Throwable { + future.cancel(true); + } } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelManager.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelManager.java index 6042c55bf..a56525e3a 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelManager.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelManager.java @@ -49,6 +49,12 @@ public class GRPCChannelManager implements BootService, Runnable { } + @Override + public void shutdown() throws Throwable { + connectCheckFuture.cancel(true); + managedChannel.shutdownNow(); + } + @Override public void run() { if (reconnect) { diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java index d04a625f1..288fa7e4f 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java @@ -49,6 +49,11 @@ public class TraceSegmentServiceClient implements BootService, IConsumer spanList = SegmentHelper.getSpan(actualSegment); assertThat(spanList.size(), is(2)); @@ -205,7 +202,7 @@ public class ContextManagerTest { @Test public void testTransform() throws InvalidProtocolBufferException { - ContextCarrier contextCarrier = new ContextCarrier().deserialize("S.1499176688384.581928182.80935.69.1|3|1|#192.168.1.8 :18002|#/portal/|T.1499176688386.581928182.80935.69.2"); + ContextCarrier contextCarrier = new ContextCarrier().deserialize("#AQA*#AQA*4WcWe0tQNQA*|3|1|#127.0.0.1:8080|#/portal/|#/testEntrySpan|#AQA*#AQA*Et0We0tQNQA*"); assertTrue(contextCarrier.isValid()); AbstractSpan firstEntrySpan = ContextManager.createEntrySpan("/testFirstEntry", contextCarrier); @@ -238,7 +235,7 @@ public class ContextManagerTest { TraceSegmentReference reference = traceSegmentObject.getRefs(0); assertThat(reference.getEntryServiceName(), is("/portal/")); - assertThat(reference.getNetworkAddress(), is("192.168.1.8 :18002")); + assertThat(reference.getNetworkAddress(), is("127.0.0.1:8080")); assertThat(reference.getParentSpanId(), is(3)); assertThat(traceSegmentObject.getApplicationId(), is(1)); diff --git a/apm-sniffer/apm-agent/src/main/java/org/skywalking/apm/agent/SkyWalkingAgent.java b/apm-sniffer/apm-agent/src/main/java/org/skywalking/apm/agent/SkyWalkingAgent.java index ea411dd9a..1edf9f841 100644 --- a/apm-sniffer/apm-agent/src/main/java/org/skywalking/apm/agent/SkyWalkingAgent.java +++ b/apm-sniffer/apm-agent/src/main/java/org/skywalking/apm/agent/SkyWalkingAgent.java @@ -44,6 +44,12 @@ public class SkyWalkingAgent { ServiceManager.INSTANCE.boot(); + Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() { + @Override public void run() { + ServiceManager.INSTANCE.shutdown(); + } + }, "skywalking service shutdown thread")); + new AgentBuilder.Default().type(pluginFinder.buildMatch()).transform(new AgentBuilder.Transformer() { @Override public DynamicType.Builder transform(DynamicType.Builder builder, TypeDescription typeDescription, diff --git a/apm-sniffer/apm-sdk-plugin/dubbo-plugin/src/test/java/org/skywalking/apm/plugin/dubbo/DubboInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/dubbo-plugin/src/test/java/org/skywalking/apm/plugin/dubbo/DubboInterceptorTest.java index c8dd580ba..380cc56f2 100644 --- a/apm-sniffer/apm-sdk-plugin/dubbo-plugin/src/test/java/org/skywalking/apm/plugin/dubbo/DubboInterceptorTest.java +++ b/apm-sniffer/apm-sdk-plugin/dubbo-plugin/src/test/java/org/skywalking/apm/plugin/dubbo/DubboInterceptorTest.java @@ -148,7 +148,7 @@ public class DubboInterceptorTest { @Test public void testProviderWithAttachment() throws Throwable { when(rpcContext.isConsumerSide()).thenReturn(false); - when(rpcContext.getAttachment(Config.Plugin.Propagation.HEADER_NAME)).thenReturn("S.1499176688384.581928182.80935.69.1|3|1|#192.168.1.8 :18002|#/portal/|T.1499176688386.581928182.80935.69.2"); + when(rpcContext.getAttachment(Config.Plugin.Propagation.HEADER_NAME)).thenReturn("#AQA*#AQA*4WcWe0tQNQA*|3|1|#192.168.1.8 :18002|#/portal/|#/testEntrySpan|#AQA*#AQA*Et0We0tQNQA*"); dubboInterceptor.beforeMethod(enhancedInstance, "invoke", allArguments, argumentTypes, methodInterceptResult); dubboInterceptor.afterMethod(enhancedInstance, "invoke", allArguments, argumentTypes, result); @@ -160,7 +160,7 @@ public class DubboInterceptorTest { when(rpcContext.isConsumerSide()).thenReturn(false); FieldSetter.setStaticValue(BugFixActive.class, "ACTIVE", true); - testParam.setTraceContext("S.1499176688384.581928182.80935.69.1|3|1|#192.168.1.8 :18002|#/portal/|T.1499176688386.581928182.80935.69.2"); + testParam.setTraceContext("#AQA*#AQA*4WcWe0tQNQA*|3|1|#192.168.1.8 :18002|#/portal/|#/testEntrySpan|#AQA*#AQA*Et0We0tQNQA*"); dubboInterceptor.beforeMethod(enhancedInstance, "invoke", allArguments, argumentTypes, methodInterceptResult); dubboInterceptor.afterMethod(enhancedInstance, "invoke", allArguments, argumentTypes, result); @@ -194,7 +194,7 @@ public class DubboInterceptorTest { private void assertTraceSegmentRef(TraceSegmentRef actual) { assertThat(SegmentRefHelper.getSpanId(actual), is(3)); - assertThat(SegmentRefHelper.getTraceSegmentId(actual), is("S.1499176688384.581928182.80935.69.1")); + assertThat(SegmentRefHelper.getTraceSegmentId(actual).toString(), is("1.1.15006458883500001")); } private void assertProviderSpan(AbstractTracingSpan span) { diff --git a/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/AbstractHttpClientInstrumentation.java b/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/AbstractHttpClientInstrumentation.java index a1dcbc484..9da99c7ef 100644 --- a/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/AbstractHttpClientInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/AbstractHttpClientInstrumentation.java @@ -20,7 +20,7 @@ import static org.skywalking.apm.agent.core.plugin.match.NameMatch.byName; */ public class AbstractHttpClientInstrumentation extends HttpClientInstrumentation { - private static final String ENHANCE_CLASS = "org.apache.http.impl.discovery.AbstractHttpClient"; + private static final String ENHANCE_CLASS = "org.apache.http.impl.client.AbstractHttpClient"; @Override public ClassMatch enhanceClass() { diff --git a/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/InternalHttpClientInstrumentation.java b/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/InternalHttpClientInstrumentation.java index e873cfd52..deac327d4 100644 --- a/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/InternalHttpClientInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/InternalHttpClientInstrumentation.java @@ -17,7 +17,7 @@ import static org.skywalking.apm.agent.core.plugin.match.NameMatch.byName; */ public class InternalHttpClientInstrumentation extends HttpClientInstrumentation { - private static final String ENHANCE_CLASS = "org.apache.http.impl.discovery.InternalHttpClient"; + private static final String ENHANCE_CLASS = "org.apache.http.impl.client.InternalHttpClient"; @Override public ClassMatch enhanceClass() { diff --git a/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/MinimalHttpClientInstrumentation.java b/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/MinimalHttpClientInstrumentation.java index 714e4f2f7..3017b7b0e 100644 --- a/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/MinimalHttpClientInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/httpClient-4.x-plugin/src/main/java/org/skywalking/apm/plugin/httpClient/v4/define/MinimalHttpClientInstrumentation.java @@ -20,7 +20,7 @@ import static org.skywalking.apm.agent.core.plugin.match.NameMatch.byName; */ public class MinimalHttpClientInstrumentation extends HttpClientInstrumentation { - private static final String ENHANCE_CLASS = "org.apache.http.impl.discovery.MinimalHttpClient"; + private static final String ENHANCE_CLASS = "org.apache.http.impl.client.MinimalHttpClient"; @Override public ClassMatch enhanceClass() { diff --git a/apm-sniffer/apm-sdk-plugin/motan-plugin/src/test/java/org/skywalking/apm/plugin/motan/MotanProviderInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/motan-plugin/src/test/java/org/skywalking/apm/plugin/motan/MotanProviderInterceptorTest.java index dffaa628d..fe7541112 100644 --- a/apm-sniffer/apm-sdk-plugin/motan-plugin/src/test/java/org/skywalking/apm/plugin/motan/MotanProviderInterceptorTest.java +++ b/apm-sniffer/apm-sdk-plugin/motan-plugin/src/test/java/org/skywalking/apm/plugin/motan/MotanProviderInterceptorTest.java @@ -91,7 +91,7 @@ public class MotanProviderInterceptorTest { @Test public void testInvokerWithRefSegment() throws Throwable { HashMap attachments = new HashMap(); - attachments.put(Config.Plugin.Propagation.HEADER_NAME, "S.1499176688384.581928182.80935.69.1|3|1|#192.168.1.8:18002|#/portal/|T.1499176688386.581928182.80935.69.2"); + attachments.put(Config.Plugin.Propagation.HEADER_NAME, "#AQA*#AQA*4WcWe0tQNQA*|3|1|#192.168.1.8:18002|#/portal/|#/testEntrySpan|#AQA*#AQA*Et0We0tQNQA*"); when(request.getAttachments()).thenReturn(attachments); invokeInterceptor.beforeMethod(enhancedInstance, "execute", arguments, argumentType, null); @@ -135,7 +135,7 @@ public class MotanProviderInterceptorTest { } private void assertRefSegment(TraceSegmentRef primaryRef) { - assertThat(SegmentRefHelper.getTraceSegmentId(primaryRef), is("S.1499176688384.581928182.80935.69.1")); + assertThat(SegmentRefHelper.getTraceSegmentId(primaryRef).toString(), is("1.1.15006458883500001")); assertThat(SegmentRefHelper.getSpanId(primaryRef), is(3)); assertThat(SegmentRefHelper.getPeerHost(primaryRef), is("192.168.1.8:18002")); } diff --git a/apm-sniffer/apm-sdk-plugin/resin-3.x-plugin/src/test/java/org/skywalking/apm/plugin/resin/v3/ResinV3InterceptorTest.java b/apm-sniffer/apm-sdk-plugin/resin-3.x-plugin/src/test/java/org/skywalking/apm/plugin/resin/v3/ResinV3InterceptorTest.java index 2019dff0d..71b3e35d6 100644 --- a/apm-sniffer/apm-sdk-plugin/resin-3.x-plugin/src/test/java/org/skywalking/apm/plugin/resin/v3/ResinV3InterceptorTest.java +++ b/apm-sniffer/apm-sdk-plugin/resin-3.x-plugin/src/test/java/org/skywalking/apm/plugin/resin/v3/ResinV3InterceptorTest.java @@ -90,7 +90,7 @@ public class ResinV3InterceptorTest { @Test public void testWithSerializedContextData() throws Throwable { - when(request.getHeader(Config.Plugin.Propagation.HEADER_NAME)).thenReturn("S.1499176688384.581928182.80935.69.1|3|1|#192.168.1.8:18002|#/portal/|T.1499176688386.581928182.80935.69.2"); + when(request.getHeader(Config.Plugin.Propagation.HEADER_NAME)).thenReturn("#AQA*#AQA*4WcWe0tQNQA*|3|1|#192.168.1.8:18002|#/portal/|#/testEntrySpan|#AQA*#AQA*Et0We0tQNQA*"); interceptor.beforeMethod(enhancedInstance, "service", arguments, argumentType, methodInterceptResult); interceptor.afterMethod(enhancedInstance, "service", arguments, argumentType, null); @@ -121,7 +121,7 @@ public class ResinV3InterceptorTest { private void assertTraceSegmentRef(TraceSegmentRef ref) { assertThat(SegmentRefHelper.getSpanId(ref), is(3)); - assertThat(SegmentRefHelper.getTraceSegmentId(ref), is("S.1499176688384.581928182.80935.69.1")); + assertThat(SegmentRefHelper.getTraceSegmentId(ref).toString(), is("1.1.15006458883500001")); } private void assertHttpSpan(AbstractTracingSpan span) { diff --git a/apm-sniffer/apm-sdk-plugin/resin-4.x-plugin/src/test/java/org/skywalking/apm/plugin/resin/v4/ResinV4InterceptorTest.java b/apm-sniffer/apm-sdk-plugin/resin-4.x-plugin/src/test/java/org/skywalking/apm/plugin/resin/v4/ResinV4InterceptorTest.java index 1256dba6d..280211f6f 100644 --- a/apm-sniffer/apm-sdk-plugin/resin-4.x-plugin/src/test/java/org/skywalking/apm/plugin/resin/v4/ResinV4InterceptorTest.java +++ b/apm-sniffer/apm-sdk-plugin/resin-4.x-plugin/src/test/java/org/skywalking/apm/plugin/resin/v4/ResinV4InterceptorTest.java @@ -92,7 +92,7 @@ public class ResinV4InterceptorTest { @Test public void testWithSerializedContextData() throws Throwable { - when(request.getHeader(Config.Plugin.Propagation.HEADER_NAME)).thenReturn("S.1499176688384.581928182.80935.69.1|3|1|#192.168.1.8:18002|#/portal/|T.1499176688386.581928182.80935.69.2"); + when(request.getHeader(Config.Plugin.Propagation.HEADER_NAME)).thenReturn("#AQA*#AQA*4WcWe0tQNQA*|3|1|#192.168.1.8:18002|#/portal/|#/testEntrySpan|#AQA*#AQA*Et0We0tQNQA*"); interceptor.beforeMethod(enhancedInstance, "service", arguments, argumentType, methodInterceptResult); interceptor.afterMethod(enhancedInstance, "service", arguments, argumentType, null); @@ -123,7 +123,7 @@ public class ResinV4InterceptorTest { private void assertTraceSegmentRef(TraceSegmentRef ref) { assertThat(SegmentRefHelper.getSpanId(ref), is(3)); - assertThat(SegmentRefHelper.getTraceSegmentId(ref), is("S.1499176688384.581928182.80935.69.1")); + assertThat(SegmentRefHelper.getTraceSegmentId(ref).toString(), is("1.1.15006458883500001")); } private void assertHttpSpan(AbstractTracingSpan span) { diff --git a/apm-sniffer/apm-sdk-plugin/tomcat-7.x-8.x-plugin/src/test/java/org/skywalking/apm/plugin/tomcat78x/TomcatInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/tomcat-7.x-8.x-plugin/src/test/java/org/skywalking/apm/plugin/tomcat78x/TomcatInterceptorTest.java index f9bfbe738..19b94a5b2 100644 --- a/apm-sniffer/apm-sdk-plugin/tomcat-7.x-8.x-plugin/src/test/java/org/skywalking/apm/plugin/tomcat78x/TomcatInterceptorTest.java +++ b/apm-sniffer/apm-sdk-plugin/tomcat-7.x-8.x-plugin/src/test/java/org/skywalking/apm/plugin/tomcat78x/TomcatInterceptorTest.java @@ -82,7 +82,7 @@ public class TomcatInterceptorTest { @Test public void testWithSerializedContextData() throws Throwable { - when(request.getHeader(Config.Plugin.Propagation.HEADER_NAME)).thenReturn("S.1499176688384.581928182.80935.69.1|3|1|#192.168.1.8:18002|#/portal/|T.1499176688386.581928182.80935.69.2"); + when(request.getHeader(Config.Plugin.Propagation.HEADER_NAME)).thenReturn("#AQA*#AQA*4WcWe0tQNQA*|3|1|#192.168.1.8:18002|#/portal/|#/testEntrySpan|#AQA*#AQA*Et0We0tQNQA*"); tomcatInterceptor.beforeMethod(enhancedInstance, "invoke", arguments, argumentType, methodInterceptResult); tomcatInterceptor.afterMethod(enhancedInstance, "invoke", arguments, argumentType, null); @@ -113,7 +113,7 @@ public class TomcatInterceptorTest { private void assertTraceSegmentRef(TraceSegmentRef ref) { assertThat(SegmentRefHelper.getSpanId(ref), is(3)); - assertThat(SegmentRefHelper.getTraceSegmentId(ref), is("S.1499176688384.581928182.80935.69.1")); + assertThat(SegmentRefHelper.getTraceSegmentId(ref).toString(), is("1.1.15006458883500001")); } private void assertHttpSpan(AbstractTracingSpan span) { diff --git a/apm-sniffer/apm-test-tools/src/main/java/org/skywalking/apm/agent/test/helper/SegmentRefHelper.java b/apm-sniffer/apm-test-tools/src/main/java/org/skywalking/apm/agent/test/helper/SegmentRefHelper.java index 957e8af76..f0ce52860 100644 --- a/apm-sniffer/apm-test-tools/src/main/java/org/skywalking/apm/agent/test/helper/SegmentRefHelper.java +++ b/apm-sniffer/apm-test-tools/src/main/java/org/skywalking/apm/agent/test/helper/SegmentRefHelper.java @@ -1,5 +1,6 @@ package org.skywalking.apm.agent.test.helper; +import org.skywalking.apm.agent.core.context.ids.ID; import org.skywalking.apm.agent.core.context.trace.TraceSegmentRef; public class SegmentRefHelper { @@ -12,7 +13,7 @@ public class SegmentRefHelper { return null; } - public static String getTraceSegmentId(TraceSegmentRef ref) { + public static ID getTraceSegmentId(TraceSegmentRef ref) { try { return FieldGetter.getValue(ref, "traceSegmentId"); } catch (Exception e) { diff --git a/apm-sniffer/apm-test-tools/src/main/java/org/skywalking/apm/agent/test/tools/SegmentRefAssert.java b/apm-sniffer/apm-test-tools/src/main/java/org/skywalking/apm/agent/test/tools/SegmentRefAssert.java index ddedf5d5c..c5d88304a 100644 --- a/apm-sniffer/apm-test-tools/src/main/java/org/skywalking/apm/agent/test/tools/SegmentRefAssert.java +++ b/apm-sniffer/apm-test-tools/src/main/java/org/skywalking/apm/agent/test/tools/SegmentRefAssert.java @@ -8,7 +8,7 @@ import static org.hamcrest.MatcherAssert.assertThat; public class SegmentRefAssert { public static void assertSegmentId(TraceSegmentRef ref, String segmentId) { - assertThat(SegmentRefHelper.getTraceSegmentId(ref), is(segmentId)); + assertThat(SegmentRefHelper.getTraceSegmentId(ref).toString(), is(segmentId)); } public static void assertSpanId(TraceSegmentRef ref, int spanId) { diff --git a/apm-sniffer/apm-toolkit-activation/apm-toolkit-opentracing-activation/src/main/java/org/skywalking/apm/toolkit/activation/opentracing/span/SkywalkingSpanActivation.java b/apm-sniffer/apm-toolkit-activation/apm-toolkit-opentracing-activation/src/main/java/org/skywalking/apm/toolkit/activation/opentracing/span/SkywalkingSpanActivation.java index e68396cc8..eb22cff3e 100644 --- a/apm-sniffer/apm-toolkit-activation/apm-toolkit-opentracing-activation/src/main/java/org/skywalking/apm/toolkit/activation/opentracing/span/SkywalkingSpanActivation.java +++ b/apm-sniffer/apm-toolkit-activation/apm-toolkit-opentracing-activation/src/main/java/org/skywalking/apm/toolkit/activation/opentracing/span/SkywalkingSpanActivation.java @@ -48,6 +48,7 @@ public class SkywalkingSpanActivation extends ClassInstanceMethodsEnhancePluginD private static final String FINISH_METHOD_INTERCEPTOR = "org.skywalking.apm.toolkit.activation.opentracing.span.SpanFinishInterceptor"; private static final String LOG_INTERCEPTOR = "org.skywalking.apm.toolkit.activation.opentracing.span.SpanLogInterceptor"; private static final String SET_OPERATION_NAME_INTERCEPTOR = "org.skywalking.apm.toolkit.activation.opentracing.span.SpanSetOperationNameInterceptor"; + private static final String SET_TAG_INTERCEPTOR = "org.skywalking.apm.toolkit.activation.opentracing.span.SpanSetTagInterceptor"; @Override protected ClassMatch enhanceClass() { @@ -132,6 +133,19 @@ public class SkywalkingSpanActivation extends ClassInstanceMethodsEnhancePluginD public boolean isOverrideArgs() { return false; } + }, + new InstanceMethodsInterceptPoint() { + @Override public ElementMatcher getMethodsMatcher() { + return named("setTag").and(takesArgument(0, String.class)).and(takesArgument(1, String.class)); + } + + @Override public String getMethodsInterceptor() { + return SET_TAG_INTERCEPTOR; + } + + @Override public boolean isOverrideArgs() { + return false; + } } }; } diff --git a/apm-sniffer/apm-toolkit-activation/apm-toolkit-opentracing-activation/src/main/java/org/skywalking/apm/toolkit/activation/opentracing/span/SpanSetTagInterceptor.java b/apm-sniffer/apm-toolkit-activation/apm-toolkit-opentracing-activation/src/main/java/org/skywalking/apm/toolkit/activation/opentracing/span/SpanSetTagInterceptor.java new file mode 100644 index 000000000..3abc0ebde --- /dev/null +++ b/apm-sniffer/apm-toolkit-activation/apm-toolkit-opentracing-activation/src/main/java/org/skywalking/apm/toolkit/activation/opentracing/span/SpanSetTagInterceptor.java @@ -0,0 +1,37 @@ +package org.skywalking.apm.toolkit.activation.opentracing.span; + +import io.opentracing.tag.Tags; +import org.skywalking.apm.agent.core.context.ContextManager; +import org.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; +import org.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; + +public class SpanSetTagInterceptor implements InstanceMethodsAroundInterceptor { + @Override + public void beforeMethod(EnhancedInstance objInst, String methodName, Object[] allArguments, + Class[] argumentsTypes, MethodInterceptResult result) throws Throwable { + } + + @Override + public Object afterMethod(EnhancedInstance objInst, String methodName, Object[] allArguments, + Class[] argumentsTypes, Object ret) throws Throwable { + AbstractSpan activeSpan = ContextManager.activeSpan(); + String tagKey = String.valueOf(allArguments[0]); + String tagValue = String.valueOf(allArguments[1]); + if (Tags.COMPONENT.getKey().equals(tagKey)) { + activeSpan.setComponent(tagValue); + } else if (Tags.PEER_SERVICE.getKey().equals(tagKey)) { + activeSpan.setOperationName(tagValue); + } else { + activeSpan.tag(tagKey, tagValue); + } + return ret; + } + + @Override + public void handleMethodException(EnhancedInstance objInst, String methodName, Object[] allArguments, + Class[] argumentsTypes, Throwable t) { + + } +} diff --git a/apm-sniffer/apm-toolkit-activation/apm-toolkit-opentracing-activation/src/test/java/org/skywalking/apm/toolkit/activation/opentracing/SkywalkingSpanActivationTest.java b/apm-sniffer/apm-toolkit-activation/apm-toolkit-opentracing-activation/src/test/java/org/skywalking/apm/toolkit/activation/opentracing/SkywalkingSpanActivationTest.java index 279ffb7b7..474ce1777 100644 --- a/apm-sniffer/apm-toolkit-activation/apm-toolkit-opentracing-activation/src/test/java/org/skywalking/apm/toolkit/activation/opentracing/SkywalkingSpanActivationTest.java +++ b/apm-sniffer/apm-toolkit-activation/apm-toolkit-opentracing-activation/src/test/java/org/skywalking/apm/toolkit/activation/opentracing/SkywalkingSpanActivationTest.java @@ -181,15 +181,15 @@ public class SkywalkingSpanActivationTest { .withTag(Tags.PEER_HOST_IPV4.getKey(), "127.0.0.1").withTag(Tags.PEER_PORT.getKey(), 8080); startSpan(); extractInterceptor.afterMethod(enhancedInstance, "extract", - new Object[] {"S.1499746282749.1100157028.88023.1.1|0|1|#127.0.0.1:8080|#testOperationName|T.1499746282768.1100157028.88023.1.2"}, new Class[] {String.class}, null); + new Object[] {"#AQA*#AQA*4WcWe0tQNQA*|3|1|#127.0.0.1:8080|#/portal/|#/testEntrySpan|#AQA*#AQA*Et0We0tQNQA*"}, new Class[] {String.class}, null); stopSpan(); TraceSegment tracingSegment = assertTraceSemgnets(); List spans = SegmentHelper.getSpans(tracingSegment); assertThat(tracingSegment.getRefs().size(), is(1)); TraceSegmentRef ref = tracingSegment.getRefs().get(0); - assertSegmentId(ref, "S.1499746282749.1100157028.88023.1.1"); - assertSpanId(ref, 0); + assertSegmentId(ref, "1.1.15006458883500001"); + assertSpanId(ref, 3); assertPeerHost(ref, "127.0.0.1:8080"); assertThat(spans.size(), is(1)); assertSpanCommonsAttribute(spans.get(0)); @@ -201,7 +201,7 @@ public class SkywalkingSpanActivationTest { .withTag(Tags.PEER_HOST_IPV4.getKey(), "127.0.0.1").withTag(Tags.PEER_PORT.getKey(), 8080); startSpan(); extractInterceptor.afterMethod(enhancedInstance, "extract", - new Object[] {"S.1499746282749.1100157028.88023.1.1|0|1|#127.0.0.1:8080|#testOperationName"}, new Class[] {String.class}, null); + new Object[] {"#AQA*#AQA*4WcWe0tQNQA*|3|#192.168.1.8:18002|#/portal/|#/testEntrySpan|#AQA*#AQA*Et0We0tQNQA*"}, new Class[] {String.class}, null); stopSpan(); TraceSegment tracingSegment = assertTraceSemgnets(); diff --git a/pom.xml b/pom.xml index 55aa23768..45d1a7924 100644 --- a/pom.xml +++ b/pom.xml @@ -172,7 +172,9 @@ - org/skywalking/apm/network/trace/proto/*.class + org/skywalking/apm/network/**/*.class + org/skywalking/apm/collector/remote/grpc/**/*.class + org/skywalking/apm/agent/core/context/ids/base64/*.class @@ -198,7 +200,9 @@ true true org/skywalking/apm/network/**/*.java, - org/skywalking/apm/collector/remote/grpc/**/*.java + org/skywalking/apm/collector/remote/grpc/**/*.java, + org/skywalking/apm/agent/core/context/ids/base64/*.java + check