skywalking后端处理业务逻辑的梳理

 接下来会进入下面的代理处理TraceSegment

接下来会调用到TraceAnalyzer的doAnalysis方法，代码如下

public void doAnalysisSegmentObject segmentObject) {
        if segmentObject.getSpansList).size) == 0) {
            return;
        }

        createSpanListeners);

        try {
            notifySegmentListenersegmentObject);

            segmentObject.getSpansList).forEachspanObject -> {
                if spanObject.getSpanId) == 0) {
                    notifyFirstListenerspanObject, segmentObject);
                }

                if SpanType.Exit.equalsspanObject.getSpanType))) {
                    notifyExitListenerspanObject, segmentObject);
                } else if SpanType.Entry.equalsspanObject.getSpanType))) {
                    notifyEntryListenerspanObject, segmentObject);
                } else if SpanType.Local.equalsspanObject.getSpanType))) {
                    notifyLocalListenerspanObject, segmentObject);
                } else {
                    log.error"span type value was unexpected, span type name: {}", spanObject.getSpanType)
                                                                                              .name));
                }
            });

            notifyListenerToBuild);
        } catch Throwable e) {
            log.errore.getMessage), e);
        }
    }

我们首先来看方法的第一个业务逻辑createSpanListeners)，改方法主要产生处于TraceSegment。以及各类span的listener，默认情况下产生的listener情况如下

 默认情况下产生上面三个listener

1、SegmentAnalysisListener主要是处理TraceSegment，将整个TraceSegment转行成segement对象存储到数据库中

2、MultiScopesAnalysisListener主要用来处理TraceSegment中的span，依据span的不同类型进行单独处理，然后依据不同的span类型产生不同的指标信息

这是收到的skywalking的出

skywalking后端oap集群收到TraceSegment的处理信息之后，会进行处理，这里处理会对TraceSegment中保存的span进行分类，span分为下面的几种类型

enum Point {
        Entry, Exit, Local, First, Segment
    }

每一种类型会对应一个listen处理器与之对应，经过处理器处理之后，在延伸到后续的流程和操作

五个listen的类型如下面所示

代码中创建了三个listener情况如下

 接下来我们分析完成了createSpanListeners);接下来是执行

notifySegmentListenersegmentObject);我们来看下改方法是如何进行处理的

 private void notifySegmentListenerSegmentObject segmentObject) {
        analysisListeners.forEachlistener -> {
            if listener.containsPointAnalysisListener.Point.Segment)) {
                SegmentListener) listener).parseSegmentsegmentObject);
            }能够
        });
    }

在上面创建的三个listener中只有SegmentAnalysisListener能够处理TraceSegment，我们来看下后面的代码

 @Override
    public void parseSegmentSegmentObject segmentObject) {
        segment.setTraceIdsegmentObject.getTraceId));
        segmentObject.getSpansList).forEachspan -> {
            if startTimestamp == 0 || startTimestamp > span.getStartTime)) {
                startTimestamp = span.getStartTime);
            }
            if span.getEndTime) > endTimestamp) {
                endTimestamp = span.getEndTime);
            }
            isError = isError || segmentStatusAnalyzer.isErrorspan);
            appendSearchableTagsspan);
        });
        final long accurateDuration = endTimestamp - startTimestamp;
        duration = accurateDuration > Integer.MAX_VALUE ? Integer.MAX_VALUE : int) accurateDuration;

        if sampleStatus.equalsSAMPLE_STATUS.UNKNOWN) || sampleStatus.equalsSAMPLE_STATUS.IGNORE)) {
            if sampler.shouldSamplesegmentObject.getTraceId))) {
                sampleStatus = SAMPLE_STATUS.SAMPLED;
            } else if isError && forceSampleErrorSegment) {
                sampleStatus = SAMPLE_STATUS.SAMPLED;
            } else if traceLatencyThresholdsAndWatcher.shouldSampleduration)) {
                sampleStatus = SAMPLE_STATUS.SAMPLED;
            } else {
                sampleStatus = SAMPLE_STATUS.IGNORE;
            }
        }
    }

我们来重点看下SegmentAnalysisListener下的parseSegment的方法的流程，方法传入的参数segmentObject的结构体如下

traceId: "f5411eeefb3b403f8cafa89d0b82a82f.48.16092665850380005"
traceSegmentId: "ec5de79a4a8848e88146ebf696d157dc.73.16092665874590000"
spans {
  parentSpanId: -1
  startTime: 1609266587459
  endTime: 1609266589471
  refs {
    traceId: "f5411eeefb3b403f8cafa89d0b82a82f.48.16092665850380005"
    parentTraceSegmentId: "f5411eeefb3b403f8cafa89d0b82a82f.48.16092665850380004"
    parentSpanId: 1
    parentService: "DEMO-webapp"
    parentServiceInstance: "e171fdf7a1b4436b8289159e6d3f90b1@192.168.43.80"
    parentEndpoint: "{GET}/hello/{words}"
    networkAddressUsedAtPeer: "169.254.122.166:20880"
  }
  operationName: "com.xxx.HelloService.sayString)"
  spanLayer: RPCFramework
  componentId: 3
  tags {
    key: "url"
    value: "dubbo://169.254.122.166:20880/com.xxx.HelloService.sayString)"
  }
}
service: "DEMO-provider"
serviceInstance: "c5ec906d3dd24d0b8907fa8549c8306b@192.168.43.80"

这里首先处理的是dubbo-provider上传的TraceSegment，产生的是一个EntrySpan，改EntrySpan中管理了dubbo-webapp的TraceSegment

 

 

 这里在处理的时候如果当前的TraceSegment中存在5个span，那么当前TraceSegment的开始时间等于5个span中最小的starttime时间，最大时间为5个span中最大的endTime的时间，如果当前的TraceSegment中只要存在一个span的熟悉中为iserror，则改TraceSegment就为false

 这个代码是判断当前的TraceSegment是否被采用处理了，如果被采样了就可以存储到数据库中，如果没有采用改TraceSegment就丢弃不用做任何处理

 notifySegmentListenersegmentObject);本质上就是将后端上传的TraceSegment转化成一个segment对象

接下来继续执行

 取出当前TraceSegment中的span列表，然后对没有span进行判断，判断当前的span是EntrySpan、localspan、exitspan，依据不同的span传递给不同的listener进行处理

这里当前的TraceSegment中只有一个span

span的对象为

parentSpanId: -1
startTime: 1609266587459
endTime: 1609266589471
refs {
  traceId: "f5411eeefb3b403f8cafa89d0b82a82f.48.16092665850380005"
  parentTraceSegmentId: "f5411eeefb3b403f8cafa89d0b82a82f.48.16092665850380004"
  parentSpanId: 1
  parentService: "DEMO-webapp"
  parentServiceInstance: "e171fdf7a1b4436b8289159e6d3f90b1@192.168.43.80"
  parentEndpoint: "{GET}/hello/{words}"
  networkAddressUsedAtPeer: "169.254.122.166:20880"
}
operationName: "com.xxx.HelloService.sayString)"
spanLayer: RPCFramework
componentId: 3
tags {
  key: "url"
  value: "dubbo://169.254.122.166:20880/com.xxx.HelloService.sayString)"
}

因为当前的span为EntrySpan，对应的spanID值就是0，会被下面的listener进行处理

  notifyFirstListenerspanObject, segmentObject);底层会调用SegmentAnalysisListener的parseFirst方法，我们来看下改方法的业务逻辑

@Override
    public void parseFirstSpanObject span, SegmentObject segmentObject) {
        if sampleStatus.equalsSAMPLE_STATUS.IGNORE)) {
            return;
        }

        if StringUtil.isEmptyserviceId)) {
            serviceName = namingControl.formatServiceNamesegmentObject.getService));
            serviceId = IDManager.ServiceID.buildId
                serviceName,
                NodeType.Normal
            );
        }

        long timeBucket = TimeBucket.getRecordTimeBucketstartTimestamp);

        segment.setSegmentIdsegmentObject.getTraceSegmentId));
        segment.setServiceIdserviceId);
        segment.setServiceInstanceIdIDManager.ServiceInstanceID.buildId
            serviceId,
            namingControl.formatInstanceNamesegmentObject.getServiceInstance))
        ));
        segment.setLatencyduration);
        segment.setStartTimestartTimestamp);
        segment.setTimeBuckettimeBucket);
        segment.setEndTimeendTimestamp);
        segment.setIsErrorBooleanUtils.booleanToValueisError));
        segment.setDataBinarysegmentObject.toByteArray));
        segment.setVersion3);

        endpointName = namingControl.formatEndpointNameserviceName, span.getOperationName));
        endpointId = IDManager.EndpointID.buildId
            serviceId,
            endpointName
        );
    }

1、当前service: “DEMO-provider”我们需要将”DEMO-provider”转行成对应应用id，存储到segment对象中

2、同理应用实例serviceInstance: “c5ec906d3dd24d0b8907fa8549c8306b@192.168.43.80″我们也要转化成对应的应用实例id存储到segement对象中

 

segement中的timeBucket字段就是当前TraceSegment中span集合中，比较各个span的开始时间，取出最小的时间，最小的时间作为segement的timeBucket

 接下来设置segemnt的开始时间，开始时间就是比较各个span的开始时间，取出最小的开始时间作为segemnt的开始时间

segemnt的结束时间结束始时间就是比较各个span的结束时间，取出最大的结束时间作为segemnt的结束时间

整个segemnt的持续时间就是开始时间减去结束时间，此外还将整个segemeOBj对象转化成二进制文件存储在字段setDataBinary中

经过整个parseFirst方法之后，整个segemnt对象的结构为

 接下来我们继续执行下面的方法

 当前的span为EntrySpan，接下来会执行 notifyEntryListenerspanObject, segmentObject);方法

 

 这里MultiScopesAnalysisListener和SegmentAnalysisListener都会对整个EntrySpan进行处理，我们来分析下下面的代码

首先来看下MultiScopesAnalysisListener对parseEntry的方法对EntrySpan的处理

 改方法的主要作用是获得当前EntrySpan关联的上游的TraceSegmentRef对象，然后将TraceSegmentRef对象封装到SourceBuilder对象中

我们先来看下当前EntrySpan中获得的TraceSegmentRef对象的信息如下

traceId: "f5411eeefb3b403f8cafa89d0b82a82f.48.16092665850380005"
parentTraceSegmentId: "f5411eeefb3b403f8cafa89d0b82a82f.48.16092665850380004"
parentSpanId: 1
parentService: "DEMO-webapp"
parentServiceInstance: "e171fdf7a1b4436b8289159e6d3f90b1@192.168.43.80"
parentEndpoint: "{GET}/hello/{words}"
networkAddressUsedAtPeer: "169.254.122.166:20880"

 在 SourceBuilder 中记录了上下游两个系统的基础信息，核心字段如下所示：

// 上游系统的 service 信息、serviceInstance 信息以及 Endpoint 信息

@Getter @Setter private int sourceServiceId;

@Getter @Setter private String sourceServiceName;

@Getter @Setter private int sourceServiceInstanceId;

@Getter @Setter private String sourceServiceInstanceName;

@Getter @Setter private int sourceEndpointId;

@Getter @Setter private String sourceEndpointName;

// 下游系统的 service 信息、serviceInstance 信息以及 Endpoint 信息

@Getter @Setter private int destServiceId;

@Getter @Setter private String destServiceName;

@Getter @Setter private int destServiceInstanceId;

@Getter @Setter private String destServiceInstanceName;

@Getter @Setter private int destEndpointId;

@Getter @Setter private String destEndpointName;

// 当前系统的组件类型

@Getter @Setter private int componentId;

// 在当前系统中的耗时

@Getter @Setter private int latency;

// 当前系统是否发生Error

@Getter @Setter private boolean status;

@Getter @Setter private int responseCode; // 默认为0

@Getter @Setter private RequestType type; // 请求类型

// 调用关系中的角色，是调用方Client)、被调用方Server)还是代理Proxy)

@Getter @Setter private DetectPoint detectPoint; 

// TraceSegment 起始时间所在分钟级时间窗口

@Getter @Setter private long timeBucket;

这里有一个关键点

 就是如果当前的EntrySpan中没有TraceSegmentRef没有父亲入口，比如dubbo-webapp入口的EntrySpan，在http请求头信息是没有任何信息的，获得的TraceSegmentRef就是为null，这里代码中设置当前的span的sourceBuilder中的  sourceBuilder.setSourceEndpointNameConst.USER_ENDPOINT_NAME);，public static final String USER_ENDPOINT_NAME = “User”;的函数表示就是浏览器或者postman等入口调用了dubbo-webapp，作为dubbo-webapp调用的入口

这里dubbo-provider的执行的是else条件，入口的方法就是dubbo-webapp的方法

 

 这里设置sourceBuilder的上游节点内容的时候，需要判断上面有MQ类型还是一般应用，这里是一般应用执行下面的else的方法

 上游节点的信息设置完成之后，接下来是设置下游的节点信息

 

 通过这个sourcebuild对象就将当前的EntrySpan和dubb-webapp中产线的exitspan关联起来了，

接下来将创建好的sourceBuilder添加到集合entrySourceBuilders中，后续会进行处理

这里TraceSegment中的span集合中如果存在5个EntrySpan，就会产生5个sourceBuilder对象

接下来继续执行 parseLogicEndpointsspan, segmentObject);

 我们来看看 parseLogicEndpointsspan, segmentObject);具体的业务代码，改方法的具体业务逻辑后面再进行分析

MultiScopesAnalysisListener的parseEntry执行完成之后，接下来执行NetworkAddressAliasMappingListener类的parseEntry方法

我们来看下改方法的具体业务逻辑，就是将当前EntrySpan所在的网络、应用名称、应用实例封装成一个networkAddressAliasSetup对象，然后将这个对象

传递给sourceReceiver进行处理

 整个networkAddressAliasSetup的结构如下

 SegmentAnalysisListener的parseEntry方法就是将应用名称应用实例名称转化成对于的id

 

 经过上面的转化之后执行notifyListenerToBuild)方法

private void notifyListenerToBuild) {
    analysisListeners.forEachAnalysisListener::build);
}执行各个listener的build方法，主要是执行SegmentAnalysisListener的build方法和MultiScopesAnalysisListener的build方法

我们先介绍SegmentAnalysisListener的build方法，然后在介绍MultiScopesAnalysisListener的build方法

 收到判断当前的segment是否是被采样忽略的，如果是被采样忽略的

skywalking后端采样率要设置为一样的原因
一个全局的请求，在dubbo-webapp产生了一个TraceSegmentA，在dubbo-dubbo产生了一个TraceSegmentB，需要有这两个TraceSegment才能还原这一次Trace
TraceSegmentA传递给后端的oap集群的A节点处理，TraceSegmentB传递给后端的oap集群的B节点处理，
如果A节点和B节点在后端的采样率设置不一致，就可能导致TraceSegmentA被处理了存储在es中，TraceSegmentB被忽略直接丢弃，导致无法还原当前请求的trace情况
所以oap集群后端每个节点的采样率必须设置为一致

  收到判断当前的segment是否是被采样忽略的，如果是被采样忽略的就直接忽略，如果不是继续执行后续的代码

设置segemnt中endpointId的值，这里endpointID是这样计算出来的

        endpointId = IDManager.EndpointID.buildId
            serviceId,
            endpointName
        )

当前的应用为dubbo-provider对于的service_id的值为REVNTy13ZWJhcHA=.1

endpointName为当前span对应的  operationName: “com.xxx.HelloService.sayString)”

 然后通过上面的这两个参数获得对应的值

         */
        public static String buildIdString serviceId, String endpointName) {
            return serviceId
                + Const.ID_CONNECTOR
                + encodeendpointName);
        }

算出来了之后 

经过处理之后整个TraceSegment的结构如下

 2、预构建 ：替换url peer端口、servicenam等信息

1、接下来将TraceSegment中各个span的类型分配给不同的spanlistener进行处理

 接下来对每一个span进行解析

parentSpanId: -1
startTime: 1609266200857
endTime: 1609266202857
refs {
traceId: “f5411eeefb3b403f8cafa89d0b82a82f.57.16092661992120001”
parentTraceSegmentId: “f5411eeefb3b403f8cafa89d0b82a82f.57.16092661992120000”
parentSpanId: 1
parentService: “DEMO-webapp”
parentServiceInstance: “e171fdf7a1b4436b8289159e6d3f90b1@192.168.43.80”
parentEndpoint: “{GET}/hello/{words}”
networkAddressUsedAtPeer: “169.254.122.166:20880”
}
operationName: “com.xxx.HelloService.sayString)”
spanLayer: RPCFramework
componentId: 3
tags {
key: “url”
value: “dubbo://169.254.122.166:20880/com.xxx.HelloService.sayString)”
}

依据不同的span类型进行解析，传递给不同的listen对象，然后产生不同的数据指标