聊聊skywalking的DatabaseSlowStatement
序
本文主要研究一下skywalking的DatabaseSlowStatement
DatabaseSlowStatement
skywalking-6.6.0/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DatabaseSlowStatement.java
@ScopeDeclaration(id = DATABASE_SLOW_STATEMENT, name = "DatabaseSlowStatement")
public class DatabaseSlowStatement extends Source {
@Getter @Setter private String id;
@Getter @Setter private int databaseServiceId;
@Getter @Setter private String statement;
@Getter @Setter private long latency;
@Getter @Setter private String traceId;
@Override public int scope() {
return DefaultScopeDefine.DATABASE_SLOW_STATEMENT;
}
@Override public String getEntityId() {
return Const.EMPTY_STRING;
}
}
- DatabaseSlowStatement继承了Source，它定义了id、databaseServiceId、statement、latency、traceId属性，其scope方法返回DefaultScopeDefine.DATABASE_SLOW_STATEMENT
MultiScopesSpanListener
skywalking-6.6.0/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/endpoint/MultiScopesSpanListener.java
public class MultiScopesSpanListener implements EntrySpanListener, ExitSpanListener, GlobalTraceIdsListener {
private static final Logger logger = LoggerFactory.getLogger(MultiScopesSpanListener.class);
private final SourceReceiver sourceReceiver;
private final ServiceInstanceInventoryCache instanceInventoryCache;
private final ServiceInventoryCache serviceInventoryCache;
private final EndpointInventoryCache endpointInventoryCache;
private final List<SourceBuilder> entrySourceBuilders;
private final List<SourceBuilder> exitSourceBuilders;
private final List<DatabaseSlowStatement> slowDatabaseAccesses;
private final TraceServiceModuleConfig config;
private final NetworkAddressInventoryCache networkAddressInventoryCache;
private SpanDecorator entrySpanDecorator;
private long minuteTimeBucket;
private String traceId;
private MultiScopesSpanListener(ModuleManager moduleManager, TraceServiceModuleConfig config) {
this.sourceReceiver = moduleManager.find(CoreModule.NAME).provider().getService(SourceReceiver.class);
this.entrySourceBuilders = new LinkedList<>();
this.exitSourceBuilders = new LinkedList<>();
this.slowDatabaseAccesses = new ArrayList<>(10);
this.instanceInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(ServiceInstanceInventoryCache.class);
this.serviceInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(ServiceInventoryCache.class);
this.endpointInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(EndpointInventoryCache.class);
this.networkAddressInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(NetworkAddressInventoryCache.class);
this.config = config;
this.traceId = null;
}
@Override public boolean containsPoint(Point point) {
return Point.Entry.equals(point) || Point.Exit.equals(point) || Point.TraceIds.equals(point);
}
@Override
public void parseEntry(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreInfo) {
this.minuteTimeBucket = segmentCoreInfo.getMinuteTimeBucket();
if (spanDecorator.getRefsCount() > 0) {
for (int i = 0; i < spanDecorator.getRefsCount(); i++) {
ReferenceDecorator reference = spanDecorator.getRefs(i);
SourceBuilder sourceBuilder = new SourceBuilder();
if (reference.getParentEndpointId() == Const.INEXISTENCE_ENDPOINT_ID) {
sourceBuilder.setSourceEndpointId(Const.USER_ENDPOINT_ID);
} else {
sourceBuilder.setSourceEndpointId(reference.getParentEndpointId());
}
final int networkAddressId = reference.getNetworkAddressId();
final int serviceIdByPeerId = serviceInventoryCache.getServiceId(networkAddressId);
final String address = networkAddressInventoryCache.get(networkAddressId).getName();
if (spanDecorator.getSpanLayer().equals(SpanLayer.MQ) || config.getUninstrumentedGatewaysConfig().isAddressConfiguredAsGateway(address)) {
int instanceIdByPeerId = instanceInventoryCache.getServiceInstanceId(serviceIdByPeerId, networkAddressId);
sourceBuilder.setSourceServiceInstanceId(instanceIdByPeerId);
sourceBuilder.setSourceServiceId(serviceIdByPeerId);
} else {
sourceBuilder.setSourceServiceInstanceId(reference.getParentServiceInstanceId());
sourceBuilder.setSourceServiceId(instanceInventoryCache.get(reference.getParentServiceInstanceId()).getServiceId());
}
sourceBuilder.setDestEndpointId(spanDecorator.getOperationNameId());
sourceBuilder.setDestServiceInstanceId(segmentCoreInfo.getServiceInstanceId());
sourceBuilder.setDestServiceId(segmentCoreInfo.getServiceId());
sourceBuilder.setDetectPoint(DetectPoint.SERVER);
sourceBuilder.setComponentId(spanDecorator.getComponentId());
setPublicAttrs(sourceBuilder, spanDecorator);
entrySourceBuilders.add(sourceBuilder);
}
} else {
SourceBuilder sourceBuilder = new SourceBuilder();
sourceBuilder.setSourceEndpointId(Const.USER_ENDPOINT_ID);
sourceBuilder.setSourceServiceInstanceId(Const.USER_INSTANCE_ID);
sourceBuilder.setSourceServiceId(Const.USER_SERVICE_ID);
sourceBuilder.setDestEndpointId(spanDecorator.getOperationNameId());
sourceBuilder.setDestServiceInstanceId(segmentCoreInfo.getServiceInstanceId());
sourceBuilder.setDestServiceId(segmentCoreInfo.getServiceId());
sourceBuilder.setDetectPoint(DetectPoint.SERVER);
sourceBuilder.setComponentId(spanDecorator.getComponentId());
setPublicAttrs(sourceBuilder, spanDecorator);
entrySourceBuilders.add(sourceBuilder);
}
this.entrySpanDecorator = spanDecorator;
}
@Override public void parseExit(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreInfo) {
if (this.minuteTimeBucket == 0) {
this.minuteTimeBucket = segmentCoreInfo.getMinuteTimeBucket();
}
SourceBuilder sourceBuilder = new SourceBuilder();
int peerId = spanDecorator.getPeerId();
if (peerId == 0) {
return;
}
int destServiceId = serviceInventoryCache.getServiceId(peerId);
int mappingServiceId = serviceInventoryCache.get(destServiceId).getMappingServiceId();
int destInstanceId = instanceInventoryCache.getServiceInstanceId(destServiceId, peerId);
int mappingServiceInstanceId = instanceInventoryCache.get(destInstanceId).getMappingServiceInstanceId();
sourceBuilder.setSourceServiceInstanceId(segmentCoreInfo.getServiceInstanceId());
sourceBuilder.setSourceServiceId(segmentCoreInfo.getServiceId());
if (Const.NONE == mappingServiceId) {
sourceBuilder.setDestServiceId(destServiceId);
} else {
sourceBuilder.setDestServiceId(mappingServiceId);
}
if (Const.NONE == mappingServiceInstanceId) {
sourceBuilder.setDestServiceInstanceId(destInstanceId);
} else {
sourceBuilder.setDestServiceInstanceId(mappingServiceInstanceId);
}
sourceBuilder.setDetectPoint(DetectPoint.CLIENT);
sourceBuilder.setComponentId(spanDecorator.getComponentId());
setPublicAttrs(sourceBuilder, spanDecorator);
exitSourceBuilders.add(sourceBuilder);
if (sourceBuilder.getType().equals(RequestType.DATABASE)) {
boolean isSlowDBAccess = false;
DatabaseSlowStatement statement = new DatabaseSlowStatement();
statement.setId(segmentCoreInfo.getSegmentId() + "-" + spanDecorator.getSpanId());
statement.setDatabaseServiceId(sourceBuilder.getDestServiceId());
statement.setLatency(sourceBuilder.getLatency());
statement.setTimeBucket(TimeBucket.getRecordTimeBucket(segmentCoreInfo.getStartTime()));
statement.setTraceId(traceId);
for (KeyStringValuePair tag : spanDecorator.getAllTags()) {
if (SpanTags.DB_STATEMENT.equals(tag.getKey())) {
String sqlStatement = tag.getValue();
if (StringUtil.isEmpty(sqlStatement)) {
statement.setStatement("[No statement]/" + sourceBuilder.getDestEndpointName());
} else if (sqlStatement.length() > config.getMaxSlowSQLLength()) {
statement.setStatement(sqlStatement.substring(0, config.getMaxSlowSQLLength()));
} else {
statement.setStatement(sqlStatement);
}
} else if (SpanTags.DB_TYPE.equals(tag.getKey())) {
String dbType = tag.getValue();
DBLatencyThresholdsAndWatcher thresholds = config.getDbLatencyThresholdsAndWatcher();
int threshold = thresholds.getThreshold(dbType);
if (sourceBuilder.getLatency() > threshold) {
isSlowDBAccess = true;
}
}
}
if (isSlowDBAccess) {
slowDatabaseAccesses.add(statement);
}
}
}
//......
@Override public void build() {
entrySourceBuilders.forEach(entrySourceBuilder -> {
entrySourceBuilder.setTimeBucket(minuteTimeBucket);
sourceReceiver.receive(entrySourceBuilder.toAll());
sourceReceiver.receive(entrySourceBuilder.toService());
sourceReceiver.receive(entrySourceBuilder.toServiceInstance());
sourceReceiver.receive(entrySourceBuilder.toEndpoint());
sourceReceiver.receive(entrySourceBuilder.toServiceRelation());
sourceReceiver.receive(entrySourceBuilder.toServiceInstanceRelation());
EndpointRelation endpointRelation = entrySourceBuilder.toEndpointRelation();
/**
* Parent endpoint could be none, because in SkyWalking Cross Process Propagation Headers Protocol v2,
* endpoint in ref could be empty, based on that, endpoint relation maybe can't be established.
* So, I am making this source as optional.
*
* Also, since 6.6.0, source endpoint could be none, if this trace begins by an internal task(local span or exit span), such as Timer,
* rather than, normally begin as an entry span, like a RPC server side.
*/
if (endpointRelation != null) {
sourceReceiver.receive(endpointRelation);
}
});
exitSourceBuilders.forEach(exitSourceBuilder -> {
if (nonNull(entrySpanDecorator)) {
exitSourceBuilder.setSourceEndpointId(entrySpanDecorator.getOperationNameId());
} else {
exitSourceBuilder.setSourceEndpointId(Const.USER_ENDPOINT_ID);
}
exitSourceBuilder.setSourceEndpointName(endpointInventoryCache.get(exitSourceBuilder.getSourceEndpointId()).getName());
exitSourceBuilder.setTimeBucket(minuteTimeBucket);
sourceReceiver.receive(exitSourceBuilder.toServiceRelation());
sourceReceiver.receive(exitSourceBuilder.toServiceInstanceRelation());
if (RequestType.DATABASE.equals(exitSourceBuilder.getType())) {
sourceReceiver.receive(exitSourceBuilder.toDatabaseAccess());
}
});
slowDatabaseAccesses.forEach(sourceReceiver::receive);
}
//......
}
-
MultiScopesSpanListener实现了EntrySpanListener、ExitSpanListener、GlobalTraceIdsListener接口，其parseExit方法在sourceBuilder.getType()为RequestType.DATABASE的时候会创建DatabaseSlowStatement，在tag.getKey()为SpanTags.DB_TYPE时，通过config.getDbLatencyThresholdsAndWatcher()获取DBLatencyThresholdsAndWatcher，然后在latency大于threshold时更新isSlowDBAccess为true，最后将DatabaseSlowStatement添加到slowDatabaseAccesses中；其build方法通过
slowDatabaseAccesses.forEach(sourceReceiver::receive)接收
SourceReceiverImpl
skywalking-6.6.0/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/SourceReceiverImpl.java
public class SourceReceiverImpl implements SourceReceiver {
@Getter
private final DispatcherManager dispatcherManager;
public SourceReceiverImpl() {
this.dispatcherManager = new DispatcherManager();
}
@Override public void receive(Source source) {
dispatcherManager.forward(source);
}
public void scan() throws IOException, InstantiationException, IllegalAccessException {
dispatcherManager.scan();
}
}
- SourceReceiverImpl实现了SourceReceiver接口，其receive方法执行dispatcherManager.forward(source)
小结
DatabaseSlowStatement继承了Source，它定义了id、databaseServiceId、statement、latency、traceId属性，其scope方法返回DefaultScopeDefine.DATABASE_SLOW_STATEMENT；MultiScopesSpanListener实现了EntrySpanListener、ExitSpanListener、GlobalTraceIdsListener接口，其parseExit方法在sourceBuilder.getType()为RequestType.DATABASE的时候会创建DatabaseSlowStatement，在tag.getKey()为SpanTags.DB_TYPE时，通过config.getDbLatencyThresholdsAndWatcher()获取DBLatencyThresholdsAndWatcher，然后在latency大于threshold时更新isSlowDBAccess为true，最后将DatabaseSlowStatement添加到slowDatabaseAccesses中；其build方法通过 slowDatabaseAccesses.forEach(sourceReceiver::receive)
接收
doc
- DatabaseSlowStatement
正文到此结束
- 本文标签： http id db UI final ask MQ tk tab map list core IO tag ssl sql value entity cache Watcher apache cat key src plugin Statement java find parse ORM equals tar App client Service ArrayList CTO build ACE LinkedList https IDE provider
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
Loading...
![[HBLOG]公众号](http://www.liuhaihua.cn/img/qrcode_gzh.jpg)
