源码角度了解Skywalking之服务端OAP对Trace的处理
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了源码角度了解Skywalking之服务端OAP对Trace的处理相关的知识，希望对你有一定的参考价值。
源码角度了解Skywalking之服务端OAP对Trace的处理
从前几篇的文章我们知道Skywalking对Trace信息进行生成收集后，将TraceSegment对象转换为UpstreamSegment对象，通过GRPC发送给OAP服务端，服务端处理对应的模块是skywalking-trace-receiver-plugin模块
TraceModuleProvider向GRPCHandlerRegister中添加处理器TraceSegmentReportServiceHandler
接收Agent数据
TraceSegmentReportServiceHandler的collect()方法接收Agent的数据，调用SegmentParseV2.Producer的send()方法发送
SegmentParseV2.Producer的send()方法：
public void send(UpstreamSegment segment, SegmentSource source)
SegmentParseV2 segmentParse = new SegmentParseV2(moduleManager, listenerManager, config);
segmentParse.setStandardizationWorker(standardizationWorker);
segmentParse.parse(new BufferData<>(segment), source);
- 创建SegmentParseV2解析器
- 调用SegmentParseV2的parse()方法
SegmentParseV2的parse()方法：
- 创建SpanListener集合
- 获取UpstreamSegment对象
- 获取UpstreamSegment对象中关联的所有的TraceID
- 如果UpstreamSegment中的SegmentObject实例为空，就解析UpstreamSegment实例得到SegmentObject对象进行填充
- 重新检查段信息是否来自文件缓冲区，如果缓存中不存这个段信息对应的服务实例Id，然后返回true
- 把SegmentObject对象封装成SegmentDecorator对象，这里是装饰者模式的体现
- 调用preBuild()方法进行预构建操作，预构建不成功写入缓存文件中，构建成功会通知具体的监听器来进行构建
预构建
SegmentParseV2的preBuild()方法：
- 构建SegmentCoreInfo对象中的segmentId
- 调用notifyGlobalsListener()方法通知这个TraceSegment所关联的TraceId对应的监听器进行解析TraceId，需要采样的进行采样
- 将Segment信息填充到SegmentCoreInfo对象中
- 遍历TraceSegment的所有span，如果是TraceSegment的第一个span，调用notifyFirstListener()方法解析第一个span，将SegmentCoreInfo对象的属性添加到Segment对象中，记录firstEndpointId和firstEndpointName，其实就是对应的请求URL，根据Span类型通知不同的监听类
通知监听器者构建
这个方法遍历所有的span调用SegmentSpanListener的build()方法，设置Segment信息的端点id和端点名后调用SourceReceiverImpl的receive()方法，最终调用SegmentDispatcher的dispatch()方法
SegmentDispatcher的dispatch()方法：
@Override public void dispatch(Segment source)
SegmentRecord segment = new SegmentRecord();
segment.setSegmentId(source.getSegmentId());
segment.setTraceId(source.getTraceId());
segment.setServiceId(source.getServiceId());
segment.setServiceInstanceId(source.getServiceInstanceId());
segment.setEndpointName(source.getEndpointName());
segment.setEndpointId(source.getEndpointId());
segment.setStartTime(source.getStartTime());
segment.setEndTime(source.getEndTime());
segment.setLatency(source.getLatency());
segment.setIsError(source.getIsError());
segment.setDataBinary(source.getDataBinary());
segment.setTimeBucket(source.getTimeBucket());
segment.setVersion(source.getVersion());
RecordStreamProcessor.getInstance().in(segment);
组装SegmentRecord对象，通过RecordStreamProcessor创建实例，in()方法中调用RecordPersistentWorker来批量异步插入ES数据库中。
总结
这篇文章主要讲解了Skywalking的OAP接收到Agent发来的Trace信息的处理逻辑，入口是TraceSegmentReportServiceHandler的collect()方法，会对Agent封装的UpstreamSegment对象进行反序列化，构建Segment、Span等信息，最终由RecordStreamProcessor来批量异步把SegmentRecord写入ES数据库中
❤️ 感谢大家
如果你觉得这篇内容对你挺有有帮助的话：
- 欢迎关注我❤️，点赞
以上是关于源码角度了解Skywalking之服务端OAP对Trace的处理的主要内容，如果未能解决你的问题，请参考以下文章