郑州模板建站平台,域名服务器ip查询网站,网页设计与网站建设试题及答案,犀牛云做网站推广怎么样文章目录 一、简介1 为什么写这篇文章2 跨进程传播协议-简介 二、协议1 Standard Header项2 Extension Header项3 Correlation Header项 三、跨进程传播协议的源码分析1 OpenTracing规范2 通过dubbo插件分析跨进程数据传播3 分析跨进程传播协议的核心源码 四、小结参考 一、简介… 文章目录 一、简介1 为什么写这篇文章2 跨进程传播协议-简介 二、协议1 Standard Header项2 Extension Header项3 Correlation Header项 三、跨进程传播协议的源码分析1 OpenTracing规范2 通过dubbo插件分析跨进程数据传播3 分析跨进程传播协议的核心源码 四、小结参考 一、简介
1 为什么写这篇文章
写这篇文章是为了让自己和大家梳理这些内容
SkyWalking的链路串联依赖跨进程数据传播他的跨进程传播协议是怎样的如果我想借助SkyWalking的跨进程传播协议实现传递全链路业务数据(如全局userId等)该如何实现
2 跨进程传播协议-简介
SkyWalking 跨进程传播协议是用于上下文的传播之前经历过sw3协议、sw6协议本文介绍是当前(2023年)最新的sw8协议。 该协议适用于不同语言、系统的探针之间传递上下文。
二、协议
Header项分为三类
Standard Header项Header名称sw8Extension Header项Header名称sw8-xCorrelation Header项Header名称sw8-correlation
协议的整体设计
下面详细讲解协议的Header项
1 Standard Header项
该Header项是上下文传播必须包含的。
Header名称sw8.Header值由-分隔的8个字段组成。Header值的长度应该小于2KB。
Header值中具体包含以下8个字段
采样Sample0 或 10 表示上下文存在但是可以也很可能被忽略而不做采样1 表示这个trace需要采样并发送到后端。追踪IDTrace Id是 Base64 编码的字符串其内容是由 . 分割的三个 long 类型值, 表示此trace的唯一标识。父追踪片段IDParent trace segment Id是 Base64 编码的字符串其内容是字符串且全局唯一。父跨度IDParent span Id是一个从 0 开始的整数这个跨度ID指向父追踪片段segment中的父跨度span。父服务名称Parent service是 Base64 编码的字符串其内容是一个长度小于或等于50个UTF-8编码的字符串。父服务实例标识Parent service instance是 Base64 编码的字符串其内容是一个长度小于或等于50个UTF-8编码的字符串。父服务的端点Parent endpoint是 Base64 编码的字符串其内容是父追踪片段segment中第一个入口跨度span的操作名由长度小于或等于50个UTF-8编码的字符组成。本请求的目标地址Peer是 Base64 编码的字符串其内容是客户端用于访问目标服务的网络地址不一定是 IP 端口。
示例值 1-TRACEID-SEGMENTID-3-PARENT_SERVICE-PARENT_INSTANCE-PARENT_ENDPOINT-IPPORT
2 Extension Header项
该Header项是可选的。扩展Header项是为高级特性设计的它提供了部署在上游和下游服务中的探针之间的交互功能。
Header名称sw8-x
Header值由-分割字段可扩展。
扩展Header值 当前值包括的字段
追踪模式Tracing Mode空、0或1默认为空或0。表示在这个上下文中生成的所有跨度span应该跳过分析。在默认情况下这个应该在上下文中传播到服务端除非它在跟踪过程中被更改。 客户端发送的时间戳用于异步RPC如MQ。一旦设置消费端将计算发送和接收之间的延迟并使用key transmission.latency自动在span中标记延迟。
示例值1-1621344125000
3 Correlation Header项
该Header项是是可选的。并非所有语言的探针都支持已知的是Java的探针是支持该协议。 该Header项用于跨进程传递用户自定义数据例如userId、orgId。 这个协议跟OpenTracing 的 Baggage很类似但是Correlation Header项相比在默认设置下会更有更严格的限制例如只能存放3个字段且有字段长度限制这个是为了安全、性能等考虑。 数据格式
Header名称sw8-correlation
Header值由,分割一对对key、value每对key、value逗号分割key、value的由Base64编码。
示例值a2V5MQ:dmFsdWUx,a2V5LTI:dmFsdWUy
三、跨进程传播协议的源码分析
1 OpenTracing规范
SkyWalking是基于OpenTracing标准的追踪系统参考吴晟老师翻译的OpenTracing规范的文章opentracing之Inject和ExtractOpenTracing定义了跨进程传播的几个要素
SpanContextSpanContext代表跨越进程边界传递到下级span的状态。在SkyWalking中的实现类是org.apache.skywalking.apm.agent.core.context.TracingContext Carrier传递跨进程数据的搬运工负责将追踪状态从一个进程carries携带传递到另一个进程 Inject 和 ExtractSpanContexts可以通过Inject(注入)操作向Carrier增加或者通过Extract(提取)从Carrier中获取跨进程通讯数据例如HTTP头。通过这种方式SpanContexts可以跨越进程边界并提供足够的信息来建立跨进程的span间关系因此可以实现跨进程连续追踪
2 通过dubbo插件分析跨进程数据传播
我们以SkyWalking java agent的dubbo-2.7.x-plugin插件为例其中跨进程传播数据的核心代码在org.apache.skywalking.apm.plugin.asf.dubbo.DubboInterceptor下面是该类跨进程传播的核心代码
public class DubboInterceptor implements InstanceMethodsAroundInterceptor {/*** Consumer: The serialized trace context data will* inject to the {link RpcContext#attachments} for transport to provider side.* p* Provider: The serialized trace context data will extract from* {link RpcContext#attachments}. current trace segment will ref if the serialization context data is not null.*/Overridepublic void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class?[] argumentsTypes,MethodInterceptResult result) throws Throwable {......if (isConsumer) { // 1、consumer端// ContextCarrierfinal ContextCarrier contextCarrier new ContextCarrier();// 1.1 createExitSpan()内部会调用TracerContext.inject(carrier)将TracerContext中的context数据inject(注入)到ContextCarrier的context中span ContextManager.createExitSpan(generateOperationName(requestURL, invocation), contextCarrier, host : port);CarrierItem next contextCarrier.items();// 1.2 遍历ContextCarrier从ContextCarrier的context获取数据注入到dubbo的attachment从consumer端传递到provider端while (next.hasNext()) {next next.next();rpcContext.setAttachment(next.getHeadKey(), next.getHeadValue());if (invocation.getAttachments().containsKey(next.getHeadKey())) {invocation.getAttachments().remove(next.getHeadKey());}}} else { // 2 provider端// 2.1 从consumer端传递到provider端的attachment中获取跨进程协议数据然后设置到contextContextCarrier contextCarrier new ContextCarrier();CarrierItem next contextCarrier.items();while (next.hasNext()) {next next.next();next.setHeadValue(rpcContext.getAttachment(next.getHeadKey()));}// 2.2 createEntrySpan()内部会调用TracerContext.extract(carrier)将ContextCarrier的context数据extract(提取)到将TracerContext中的context中span ContextManager.createEntrySpan(generateOperationName(requestURL, invocation), contextCarrier);span.setPeer(rpcContext.getRemoteAddressString());}}
}从上面的源码可以看出在服务调用方和被调用方都会用到ContextCarrier他是临时搬运工负责两个进程的TracerContext数据的传递。 下面分析ContextCarrier等类的核心源码。
3 分析跨进程传播协议的核心源码
TracingContext org.apache.skywalking.apm.agent.core.context.TracingContext是OpenTracing的SpanContext的一种实现里面包含了span的上下文包含在segment、correlationContext、extensionContext而inject()、extract()负责跨进程上下文透传。
public class TracingContext implements AbstractTracerContext {/*** The final {link TraceSegment}, which includes all finished spans.*/private TraceSegment segment;Getter(AccessLevel.PACKAGE)private final CorrelationContext correlationContext;Getter(AccessLevel.PACKAGE)private final ExtensionContext extensionContext;/*** Prepare for the cross-process propagation. How to initialize the carrier, depends on the implementation.** param carrier to carry the context for crossing process.*/void inject(ContextCarrier carrier);/*** Build the reference between this segment and a cross-process segment. How to build, depends on the* implementation.** param carrier carried the context from a cross-process segment.*/void extract(ContextCarrier carrier);
}ContextCarrier ContextCarrier作为传递跨进程数据的搬运工负责将追踪状态从一个进程carries携带传递到另一个进程其中包含了sw8协议里的Standard Header项、Extension Header项、Correlation Header项相关的上下文数据具体参考下面的代码
public class ContextCarrier implements Serializable {/*** extensionContext包含了在某些特定场景中用于增强分析的可选上下文对应sw8的Extension Header项*/private ExtensionContext extensionContext new ExtensionContext();/*** 用户的自定义上下文容器。此上下文与主追踪上下文一同传播。对应sw8的Correlation Header项*/private CorrelationContext correlationContext new CorrelationContext();/*** return 存在于当前tracing上下文中的item清单*/public CarrierItem items() {SW8ExtensionCarrierItem sw8ExtensionCarrierItem new SW8ExtensionCarrierItem(extensionContext, null);SW8CorrelationCarrierItem sw8CorrelationCarrierItem new SW8CorrelationCarrierItem(correlationContext, sw8ExtensionCarrierItem);SW8CarrierItem sw8CarrierItem new SW8CarrierItem(this, sw8CorrelationCarrierItem);return new CarrierItemHead(sw8CarrierItem);}/*** Extract the extension context to tracing context*/void extractExtensionTo(TracingContext tracingContext) {tracingContext.getExtensionContext().extract(this);// The extension context could have field not to propagate further, so, must use the this.* to process.this.extensionContext.handle(tracingContext.activeSpan());}/*** Extract the correlation context to tracing context*/void extractCorrelationTo(TracingContext tracingContext) {tracingContext.getCorrelationContext().extract(this);// The correlation context could have field not to propagate further, so, must use the this.* to process.this.correlationContext.handle(tracingContext.activeSpan());}/*** 序列化sw8的Standard Header项使用 - 分割各个字段* Serialize this {link ContextCarrier} to a {link String}, with | split.* return the serialization string.*/String serialize(HeaderVersion version) {if (this.isValid(version)) {return StringUtil.join(-,1,Base64.encode(this.getTraceId()),Base64.encode(this.getTraceSegmentId()),this.getSpanId() ,Base64.encode(this.getParentService()),Base64.encode(this.getParentServiceInstance()),Base64.encode(this.getParentEndpoint()),Base64.encode(this.getAddressUsedAtClient()));}return ;}/*** 反序列化sw8的Standard Header项* Initialize fields with the given text.* param text carries {link #traceSegmentId} and {link #spanId}, with | split.*/ContextCarrier deserialize(String text, HeaderVersion version) {if (text null) {return this;}if (HeaderVersion.v3.equals(version)) {String[] parts text.split(-, 8);if (parts.length 8) {try {// parts[0] is sample flag, always trace if header exists.this.traceId Base64.decode2UTFString(parts[1]);this.traceSegmentId Base64.decode2UTFString(parts[2]);this.spanId Integer.parseInt(parts[3]);this.parentService Base64.decode2UTFString(parts[4]);this.parentServiceInstance Base64.decode2UTFString(parts[5]);this.parentEndpoint Base64.decode2UTFString(parts[6]);this.addressUsedAtClient Base64.decode2UTFString(parts[7]);} catch (IllegalArgumentException ignored) {}}}return this;}
}CorrelationContext ContextCarrier里包含里sw8的Correlation Header项存放于CorrelationContext这个类非常有用适合我们去在全链路跨进程传递自定义的数据。 sw8协议里的Standard Header项、Extension Header项是比较固定的协议格式我们可以扩展这些协议例如Standard Header项当前固定是8位的对应8个字段我们可以扩展为9位第九位可以定义为userId。但是如果要这样改造就得修改ContextCarrier类序列化、反序列的逻辑要重新发布agent并考虑好新旧版本兼容性问题、以及不同语言的agent是否兼容。 而sw8的Correlation Header项使用起来就非常方便。先看下对应实现了CorrelationContext的源码
/*** Correlation context, use to propagation user custom data.* Correlation上下文用于传播用户自定义数据*/
public class CorrelationContext {private final MapString, String data;/*** Add or override the context. 添加或覆盖上下文数据** param key to add or locate the existing context* param value as new value* return old one if exist.*/public OptionalString put(String key, String value) {// 可以存放于span的tag中if (AUTO_TAG_KEYS.contains(key) ContextManager.isActive()) {ContextManager.activeSpan().tag(new StringTag(key), value);}// settingdata.put(key, value);return Optional.empty();}/*** param key to find the context 获取上下文数据* return value if exist.*/public OptionalString get(String key) {return Optional.ofNullable(data.get(key));}/*** Serialize this {link CorrelationContext} to a {link String} 序列化** return the serialization string.*/String serialize() {if (data.isEmpty()) {return ;}return data.entrySet().stream().map(entry - Base64.encode(entry.getKey()) : Base64.encode(entry.getValue())).collect(Collectors.joining(,));}/*** Deserialize data from {link String} 反序列化*/void deserialize(String value) {if (StringUtil.isEmpty(value)) {return;}for (String perData : value.split(,)) {// Only data with limited count of elements can be addedif (data.size() Config.Correlation.ELEMENT_MAX_NUMBER) {break;}final String[] parts perData.split(:);if (parts.length ! 2) {continue;}data.put(Base64.decode2UTFString(parts[0]), Base64.decode2UTFString(parts[1]));}}/*** Prepare for the cross-process propagation. Inject the {link #data} into {link* ContextCarrier#getCorrelationContext()}*/void inject(ContextCarrier carrier) {carrier.getCorrelationContext().data.putAll(this.data);}/*** Extra the {link ContextCarrier#getCorrelationContext()} into this context.*/void extract(ContextCarrier carrier) {......}/*** Clone the context data, work for capture to cross-thread. 克隆数据用于跨线程传递*/Overridepublic CorrelationContext clone() {final CorrelationContext context new CorrelationContext();context.data.putAll(this.data);return context;}/*** Continue the correlation context in another thread.传递到另外的线程** param snapshot holds the context.*/void continued(ContextSnapshot snapshot) {this.data.putAll(snapshot.getCorrelationContext().data);}
}通过源码可知CorrelationContext通过MapString, String来存放数据CorrelationContext数据支持跨线程、跨进程透传。
四、小结
分析Dubbo插件的跨进程核心代码了解了跨进程传播协议的核心实现逻辑。
其实在其他分布式追踪系统(如Zipkin、Jager)、全链路灰度系统等涉及到跨进程数据传播的系统中也是使用了类似于上面SkyWalking协议的思路。
参考
SkyWalking Cross Process Propagation Headers Protocol SkyWalking Cross Process Correlation Headers Protocol 详解 Apache SkyWalking 的跨进程传播协议
