kazusa
/
skywalking
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Support analyzing cache related spans to provide metrics and slow commands for cache services from client side. (#9622) 

* Support analyzing cache related spans to provide metrics and slow commands for cache services from client side
* Optimize virtual database, fix dynamic config watcher NPE when default value is null 
* [UI] Add virtual cache dashboard
This commit is contained in:
pg.yang 2022-09-19 22:10:32 +08:00 committed by GitHub
parent c3e0b12648
commit bbe0a7c4c2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
48 changed files with 2320 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
docs/en/changes/changes.md
View File

@ -12,6 +12,8 @@
* [**Breaking Change**] Change the LAL script format(Add layer property).
* Adapt ElasticSearch 8.1+, migrate from removed APIs to recommended APIs.
* Support monitoring MySQL slow SQLs.
* Support analyzing cache related spans to provide metrics and slow commands for cache services from client side
* Optimize virtual database, fix dynamic config watcher NPE when default value is null
* Remove physical index existing check and keep template existing check only to avoid meaningless `retry wait`
in `no-init` mode.
* Make sure instance list ordered in TTL processor to avoid TTL timer never runs.
@ -32,6 +34,7 @@
* Enhance the process topology graph to support dragging nodes.
* UI-template: Fix metrics calculation in `general-service/mesh-service/faas-function` top-list dashboard.
* Update MySQL dashboard to visualize collected slow SQLs.
* Add virtual cache dashboard
#### Documentation

51
docs/en/concepts-and-designs/scope-definitions.md
View File

@ -286,3 +286,54 @@ This calculates the metrics data from [events](event.md).
| type | The type of the event, `Normal` or `Error`. | | string|
| message | The message of the event. | | string |
| parameters | The parameters in the `message`, see [parameters](event.md#parameters). | | string |
### SCOPE `DatabaseAccess`
This calculates the metrics data from each request of cache system.
| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The service name of virtual database service. | | string |
| databaseTypeId | The ID of the component used in this call. | | int |
| latency | The time taken by each request. | | int(in ms)|
| status | Indicates the success or failure of the request.| | boolean |
### SCOPE `DatabaseSlowStatement`
This calculates the metrics data from slow request of cache system , which is used for `write` or `read` operation.
| Name | Remarks | Group Key | Type |
|---|---|---|---|
| databaseServiceId | The service id of virtual cache service. | | string |
| statement | The sql statement . | | string |
| latency | The time taken by each request. | | int(in ms)|
| traceId | The traceId of this slow statement| | string|
### SCOPE `CacheAccess`
This calculates the metrics data from each request of cache system.
| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The service name of virtual cache service. | | string |
| cacheTypeId | The ID of the component used in this call. | | int |
| latency | The time taken by each request. | | int(in ms)|
| status | Indicates the success or failure of the request.| | boolean |
| op | Indicates this access is used for `write` or `read` | | string |
### SCOPE `CacheSlowAccess`
This calculates the metrics data from slow request of cache system , which is used for `write` or `read` operation.
| Name | Remarks | Group Key | Type |
|---|---|---|---|
| cacheServiceId | The service id of virtual cache service. | | string |
| command | The cache command . | | string |
| key | The cache command key. | | string|
| latency | The time taken by each request. | | int(in ms)|
| traceId | The traceId of this slow access| | string|
| status | Indicates the success or failure of the request.| | boolean |
| op | Indicates this access is used for `write` or `read` | | string |

2
docs/en/setup/backend/configuration-vocabulary.md
View File

@ -154,6 +154,8 @@ The Configuration Vocabulary lists all available configurations provided by `app
| - | - | segmentStatusAnalysisStrategy | Determines the final segment status from span status. Available values are `FROM_SPAN_STATUS` , `FROM_ENTRY_SPAN`, and `FROM_FIRST_SPAN`. `FROM_SPAN_STATUS` indicates that the segment status would be error if any span has an error status. `FROM_ENTRY_SPAN` means that the segment status would only be determined by the status of entry spans. `FROM_FIRST_SPAN` means that the segment status would only be determined by the status of the first span. | SW_SEGMENT_STATUS_ANALYSIS_STRATEGY | FROM_SPAN_STATUS |
| - | - | noUpstreamRealAddressAgents | Exit spans with the component in the list would not generate client-side instance relation metrics, since some tracing plugins (e.g. Nginx-LUA and Envoy) can't collect the real peer IP address. | SW_NO_UPSTREAM_REAL_ADDRESS | 6000,9000 |
| - | - | meterAnalyzerActiveFiles | Indicates which files could be instrumented and analyzed. Multiple files are split by ",". | SW_METER_ANALYZER_ACTIVE_FILES || |
| - | - | slowCacheWriteThreshold | The threshold of slow command which is used for writing operation (in milliseconds). | SW_SLOW_CACHE_WRITE_THRESHOLD | `default:20,redis:10` |
| - | - | slowCacheReadThreshold | The threshold of slow command which is used for reading (getting) operation (in milliseconds). | SW_SLOW_CACHE_READ_THRESHOLD | `default:20,redis:10` |
| receiver-sharing-server | default | Sharing server provides new gRPC and restful servers for data collection. Ana designates that servers in the core module are to be used for internal communication only. | - | - | |
| - | - | restHost | Binding IP of RESTful services. Services include GraphQL query and HTTP data report. | SW_RECEIVER_SHARING_REST_HOST | - |
| - | - | restPort | Binding port of RESTful services. | SW_RECEIVER_SHARING_REST_PORT | - |

13
docs/en/setup/backend/slow-cache-command.md Normal file
View File

@ -0,0 +1,13 @@
# Slow Cache Command
Slow Cache command are sensitive for you to identify bottlenecks of a system which relies on cache system.
Slow Cache command are based on sampling. Right now, the core samples are the top 50 slowest every 10 minutes.
Note that the duration of these command must be slower than the threshold.
Here's the format of the settings (in milliseconds):
> cache-type:thresholdValue,cache-type2:thresholdValue2
The default settings are `default:20,redis:10`. `Reserved Cache type` is **default**, which is the default threshold for all
cache types, unless set explicitly.
**Note**: The threshold should not be set too small, like `1ms`. Although it works in theory, OAP performance issues may arise if your system statement access time is usually more than 1ms.

16
docs/en/setup/service-agent/virtual-cache.md Normal file
View File

@ -0,0 +1,16 @@
# Virtual Cache
Virtual cache represent the cache nodes detected by [server agents' plugins](server-agents.md). The performance
metrics of the cache are also from the Cache client-side perspective.
For example, Redis plugins in the Java agent could detect the latency of command
As a result, SkyWalking would show traffic, latency, success rate, and sampled slow operations(write/read) powered by backend analysis capabilities in this dashboard.
The cache operation span should have
- It is an **Exit** or **Local** span
- **Span's layer == CACHE**
- Tag key = `cache.type`, value = The type of cache system , e.g. redis
- Tag key = `cache.op`, value = the operation of command , indicates the command is used for `write` or `read` operation
- Tag key = `cache.command`, value = the cache command , e.g. get,set,del
- Tag key = `cache.key`, value = the cache key
- If the cache system is in-memory (e.g. Guava-cache), agents' plugin would create a local span usually, and the span's peer would be null ,otherwise the peer is the network address(IP or domain) of Cache server.

4
docs/menu.yml
View File

@ -109,6 +109,8 @@ catalog:
path: "/en/setup/backend/trace-sampling"
- name: "Detect Slow Database Statement"
path: "/en/setup/backend/slow-db-statement"
- name: "Detect Slow Cache Command"
path: "/en/setup/backend/slow-cache-command"
- name: "Message Queue Performance"
path: "/en/setup/backend/mq"
- name: "Uninstrumented Gateways"
@ -159,6 +161,8 @@ catalog:
path: "/en/setup/service-agent/agent-compatibility"
- name: "Virtual Database"
path: "/en/setup/service-agent/virtual-database"
- name: "Virtual Cache"
path: "/en/setup/service-agent/virtual-cache"
- name: "Service Mesh"
catalog:
- name: "Observe Service Mesh"

19
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/AnalyzerModuleConfig.java
View File

@ -23,6 +23,8 @@ import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.skywalking.oap.server.analyzer.provider.trace.CacheReadLatencyThresholdsAndWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.CacheWriteLatencyThresholdsAndWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.DBLatencyThresholdsAndWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.TraceSamplingPolicyWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.UninstrumentedGatewaysConfig;
@ -60,6 +62,23 @@ public class AnalyzerModuleConfig extends ModuleConfig {
@Setter
@Getter
private DBLatencyThresholdsAndWatcher dbLatencyThresholdsAndWatcher;
@Setter
@Getter
private String slowCacheWriteThreshold = "default:20,redis:10";
@Setter
@Getter
private CacheWriteLatencyThresholdsAndWatcher cacheWriteLatencyThresholdsAndWatcher;
@Setter
@Getter
private String slowCacheReadThreshold = "default:20,redis:10";
@Setter
@Getter
private CacheReadLatencyThresholdsAndWatcher cacheReadLatencyThresholdsAndWatcher;
@Setter
@Getter
private UninstrumentedGatewaysConfig uninstrumentedGatewaysConfig;

28
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/AnalyzerModuleProvider.java
View File

@ -18,14 +18,14 @@
package org.apache.skywalking.oap.server.analyzer.provider;
import java.util.List;
import lombok.Getter;
import org.apache.skywalking.oap.server.analyzer.module.AnalyzerModule;
import org.apache.skywalking.oap.server.analyzer.provider.meter.config.MeterConfig;
import org.apache.skywalking.oap.server.analyzer.provider.meter.config.MeterConfigs;
import org.apache.skywalking.oap.server.analyzer.provider.meter.process.IMeterProcessService;
import org.apache.skywalking.oap.server.analyzer.provider.meter.process.MeterProcessService;
import org.apache.skywalking.oap.server.analyzer.provider.trace.CacheReadLatencyThresholdsAndWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.CacheWriteLatencyThresholdsAndWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.DBLatencyThresholdsAndWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.TraceSamplingPolicyWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.UninstrumentedGatewaysConfig;
@ -33,9 +33,10 @@ import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.ISegmentP
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.SegmentParserListenerManager;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.SegmentParserServiceImpl;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.EndpointDepFromCrossThreadAnalysisListener;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.RPCAnalysisListener;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.NetworkAddressAliasMappingListener;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.RPCAnalysisListener;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.SegmentAnalysisListener;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.VirtualServiceAnalysisListener;
import org.apache.skywalking.oap.server.configuration.api.ConfigurationModule;
import org.apache.skywalking.oap.server.configuration.api.DynamicConfigurationService;
import org.apache.skywalking.oap.server.core.CoreModule;
@ -48,11 +49,15 @@ import org.apache.skywalking.oap.server.library.module.ModuleStartException;
import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException;
import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
import java.util.List;
public class AnalyzerModuleProvider extends ModuleProvider {
@Getter
private final AnalyzerModuleConfig moduleConfig;
@Getter
private DBLatencyThresholdsAndWatcher thresholds;
private DBLatencyThresholdsAndWatcher dbLatencyThresholdsAndWatcher;
private CacheReadLatencyThresholdsAndWatcher cacheReadLatencyThresholdsAndWatcher;
private CacheWriteLatencyThresholdsAndWatcher cacheWriteLatencyThresholdsAndWatcher;
@Getter
private UninstrumentedGatewaysConfig uninstrumentedGatewaysConfig;
@Getter
@ -85,15 +90,17 @@ public class AnalyzerModuleProvider extends ModuleProvider {
@Override
public void prepare() throws ServiceNotProvidedException, ModuleStartException {
thresholds = new DBLatencyThresholdsAndWatcher(moduleConfig.getSlowDBAccessThreshold(), this);
dbLatencyThresholdsAndWatcher = new DBLatencyThresholdsAndWatcher(moduleConfig.getSlowDBAccessThreshold(), this);
uninstrumentedGatewaysConfig = new UninstrumentedGatewaysConfig(this);
traceSamplingPolicyWatcher = new TraceSamplingPolicyWatcher(moduleConfig, this);
cacheReadLatencyThresholdsAndWatcher = new CacheReadLatencyThresholdsAndWatcher(moduleConfig.getSlowCacheReadThreshold(), this);
cacheWriteLatencyThresholdsAndWatcher = new CacheWriteLatencyThresholdsAndWatcher(moduleConfig.getSlowCacheWriteThreshold(), this);
moduleConfig.setDbLatencyThresholdsAndWatcher(thresholds);
moduleConfig.setDbLatencyThresholdsAndWatcher(dbLatencyThresholdsAndWatcher);
moduleConfig.setUninstrumentedGatewaysConfig(uninstrumentedGatewaysConfig);
moduleConfig.setTraceSamplingPolicyWatcher(traceSamplingPolicyWatcher);
moduleConfig.setCacheReadLatencyThresholdsAndWatcher(cacheReadLatencyThresholdsAndWatcher);
moduleConfig.setCacheWriteLatencyThresholdsAndWatcher(cacheWriteLatencyThresholdsAndWatcher);
segmentParserService = new SegmentParserServiceImpl(getManager(), moduleConfig);
this.registerServiceImplementation(ISegmentParserService.class, segmentParserService);
@ -116,9 +123,11 @@ public class AnalyzerModuleProvider extends ModuleProvider {
.provider()
.getService(
DynamicConfigurationService.class);
dynamicConfigurationService.registerConfigChangeWatcher(thresholds);
dynamicConfigurationService.registerConfigChangeWatcher(dbLatencyThresholdsAndWatcher);
dynamicConfigurationService.registerConfigChangeWatcher(uninstrumentedGatewaysConfig);
dynamicConfigurationService.registerConfigChangeWatcher(traceSamplingPolicyWatcher);
dynamicConfigurationService.registerConfigChangeWatcher(cacheReadLatencyThresholdsAndWatcher);
dynamicConfigurationService.registerConfigChangeWatcher(cacheWriteLatencyThresholdsAndWatcher);
segmentParserService.setListenerManager(listenerManager());
@ -147,6 +156,7 @@ public class AnalyzerModuleProvider extends ModuleProvider {
listenerManager.add(new NetworkAddressAliasMappingListener.Factory(getManager()));
}
listenerManager.add(new SegmentAnalysisListener.Factory(getManager(), moduleConfig));
listenerManager.add(new VirtualServiceAnalysisListener.Factory(getManager()));
return listenerManager;
}

80
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/CacheReadLatencyThresholdsAndWatcher.java Normal file
View File

@ -0,0 +1,80 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.analyzer.provider.trace;
import com.google.common.base.Splitter;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.skywalking.oap.server.analyzer.module.AnalyzerModule;
import org.apache.skywalking.oap.server.configuration.api.ConfigChangeWatcher;
import org.apache.skywalking.oap.server.library.module.ModuleProvider;
public class CacheReadLatencyThresholdsAndWatcher extends ConfigChangeWatcher {
private AtomicReference<Map<String, Integer>> thresholds;
private final String initialSettingsString;
private volatile String dynamicSettingsString;
public CacheReadLatencyThresholdsAndWatcher(String config, ModuleProvider provider) {
super(AnalyzerModule.NAME, provider, "slowCacheReadThreshold");
thresholds = new AtomicReference<>(new HashMap<>());
initialSettingsString = config;
activeSetting(config);
}
private void activeSetting(String config) {
Map<String, Integer> newThresholds = new HashMap<>();
List<String> settings = Splitter.on(',').splitToList(config);
for (String setting : settings) {
List<String> typeValue = Splitter.on(":").splitToList(setting);
if (typeValue.size() == 2) {
newThresholds.put(typeValue.get(0).trim().toLowerCase(), Integer.parseInt(typeValue.get(1).trim()));
}
}
thresholds.set(newThresholds);
}
public int getThreshold(String type) {
type = type.toLowerCase();
if (thresholds.get().containsKey(type)) {
return thresholds.get().get(type);
} else {
return Optional.ofNullable(thresholds.get().get("default")).orElse(Integer.MAX_VALUE);
}
}
@Override
public void notify(ConfigChangeEvent value) {
if (EventType.DELETE.equals(value.getEventType())) {
dynamicSettingsString = null;
activeSetting(initialSettingsString);
} else {
dynamicSettingsString = value.getNewValue();
activeSetting(value.getNewValue());
}
}
@Override
public String value() {
return dynamicSettingsString;
}
}

80
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/CacheWriteLatencyThresholdsAndWatcher.java Normal file
View File

@ -0,0 +1,80 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.analyzer.provider.trace;
import com.google.common.base.Splitter;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.skywalking.oap.server.analyzer.module.AnalyzerModule;
import org.apache.skywalking.oap.server.configuration.api.ConfigChangeWatcher;
import org.apache.skywalking.oap.server.library.module.ModuleProvider;
public class CacheWriteLatencyThresholdsAndWatcher extends ConfigChangeWatcher {
private AtomicReference<Map<String, Integer>> thresholds;
private final String initialSettingsString;
private volatile String dynamicSettingsString;
public CacheWriteLatencyThresholdsAndWatcher(String config, ModuleProvider provider) {
super(AnalyzerModule.NAME, provider, "slowCacheWriteThreshold");
thresholds = new AtomicReference<>(new HashMap<>());
initialSettingsString = config;
activeSetting(config);
}
private void activeSetting(String config) {
Map<String, Integer> newThresholds = new HashMap<>();
List<String> settings = Splitter.on(',').splitToList(config);
for (String setting : settings) {
List<String> typeValue = Splitter.on(":").splitToList(setting);
if (typeValue.size() == 2) {
newThresholds.put(typeValue.get(0).trim().toLowerCase(), Integer.parseInt(typeValue.get(1).trim()));
}
}
thresholds.set(newThresholds);
}
public int getThreshold(String type) {
type = type.toLowerCase();
if (thresholds.get().containsKey(type)) {
return thresholds.get().get(type);
} else {
return Optional.ofNullable(thresholds.get().get("default")).orElse(Integer.MAX_VALUE);
}
}
@Override
public void notify(ConfigChangeEvent value) {
if (EventType.DELETE.equals(value.getEventType())) {
dynamicSettingsString = null;
activeSetting(initialSettingsString);
} else {
dynamicSettingsString = value.getNewValue();
activeSetting(value.getNewValue());
}
}
@Override
public String value() {
return dynamicSettingsString;
}
}

3
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/DBLatencyThresholdsAndWatcher.java
View File

@ -20,6 +20,7 @@ package org.apache.skywalking.oap.server.analyzer.provider.trace;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.skywalking.oap.server.analyzer.module.AnalyzerModule;
import org.apache.skywalking.oap.server.configuration.api.ConfigChangeWatcher;
@ -56,7 +57,7 @@ public class DBLatencyThresholdsAndWatcher extends ConfigChangeWatcher {
if (thresholds.get().containsKey(type)) {
return thresholds.get().get(type);
} else {
return thresholds.get().get("default");
return Optional.ofNullable(thresholds.get().get("default")).orElse(Integer.MAX_VALUE);
}
}

8
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/SpanTags.java
View File

@ -37,6 +37,14 @@ public class SpanTags {
public static final String DB_TYPE = "db.type";
public static final String CACHE_TYPE = "cache.type";
public static final String CACHE_OP = "cache.op";
public static final String CACHE_CMD = "cache.cmd";
public static final String CACHE_KEY = "cache.key";
/**
* Tag, x-le(extension logic endpoint) series tag. Value is JSON format.
* <pre>

57
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/RPCAnalysisListener.java
View File

@ -20,18 +20,14 @@ package org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import java.util.ArrayList;
import java.util.List;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
import org.apache.skywalking.apm.network.language.agent.v3.SegmentReference;
import org.apache.skywalking.apm.network.language.agent.v3.SpanLayer;
import org.apache.skywalking.apm.network.language.agent.v3.SpanObject;
import org.apache.skywalking.apm.network.language.agent.v3.SpanType;
import org.apache.skywalking.oap.server.analyzer.provider.AnalyzerModuleConfig;
import org.apache.skywalking.oap.server.analyzer.provider.trace.DBLatencyThresholdsAndWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.SpanTags;
import org.apache.skywalking.oap.server.core.Const;
import org.apache.skywalking.oap.server.core.CoreModule;
@ -50,6 +46,9 @@ import org.apache.skywalking.oap.server.core.source.SourceReceiver;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import java.util.ArrayList;
import java.util.List;
import static org.apache.skywalking.oap.server.analyzer.provider.trace.parser.SpanTags.LOGIC_ENDPOINT;
/**
@ -60,7 +59,6 @@ import static org.apache.skywalking.oap.server.analyzer.provider.trace.parser.Sp
public class RPCAnalysisListener extends CommonAnalysisListener implements EntryAnalysisListener, ExitAnalysisListener, LocalAnalysisListener {
private final List<RPCTrafficSourceBuilder> callingInTraffic = new ArrayList<>(10);
private final List<RPCTrafficSourceBuilder> callingOutTraffic = new ArrayList<>(10);
private final List<DatabaseSlowStatementBuilder> dbSlowStatementBuilders = new ArrayList<>(10);
private final List<EndpointSourceBuilder> logicEndpointBuilders = new ArrayList<>(10);
private final Gson gson = new Gson();
private final SourceReceiver sourceReceiver;
@ -197,45 +195,6 @@ public class RPCAnalysisListener extends CommonAnalysisListener implements Entry
sourceBuilder.setComponentId(span.getComponentId());
setPublicAttrs(sourceBuilder, span);
callingOutTraffic.add(sourceBuilder);
if (RequestType.DATABASE.equals(sourceBuilder.getType())) {
boolean isSlowDBAccess = false;
DatabaseSlowStatementBuilder slowStatementBuilder = new DatabaseSlowStatementBuilder(namingControl);
slowStatementBuilder.setServiceName(networkAddress);
slowStatementBuilder.setId(segmentObject.getTraceSegmentId() + "-" + span.getSpanId());
slowStatementBuilder.setLatency(sourceBuilder.getLatency());
slowStatementBuilder.setTimeBucket(TimeBucket.getRecordTimeBucket(span.getStartTime()));
slowStatementBuilder.setTraceId(segmentObject.getTraceId());
for (KeyStringValuePair tag : span.getTagsList()) {
if (SpanTags.DB_STATEMENT.equals(tag.getKey())) {
String sqlStatement = tag.getValue();
if (StringUtil.isNotEmpty(sqlStatement)) {
if (sqlStatement.length() > config.getMaxSlowSQLLength()) {
slowStatementBuilder.setStatement(sqlStatement.substring(0, config.getMaxSlowSQLLength()));
} else {
slowStatementBuilder.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 (StringUtil.isEmpty(slowStatementBuilder.getStatement())) {
String statement = StringUtil.isEmpty(
span.getOperationName()) ? "[No statement]" : "[No statement]/" + span.getOperationName();
slowStatementBuilder.setStatement(statement);
}
if (isSlowDBAccess) {
dbSlowStatementBuilders.add(slowStatementBuilder);
}
}
}
private void setPublicAttrs(RPCTrafficSourceBuilder sourceBuilder, SpanObject span) {
@ -318,17 +277,7 @@ public class RPCAnalysisListener extends CommonAnalysisListener implements Entry
if (serviceInstanceRelation != null) {
sourceReceiver.receive(serviceInstanceRelation);
}
if (RequestType.DATABASE.equals(callingOut.getType())) {
sourceReceiver.receive(callingOut.toServiceMeta());
sourceReceiver.receive(callingOut.toDatabaseAccess());
}
});
dbSlowStatementBuilders.forEach(dbSlowStatBuilder -> {
dbSlowStatBuilder.prepare();
sourceReceiver.receive(dbSlowStatBuilder.toDatabaseSlowStatement());
});
logicEndpointBuilders.forEach(logicEndpointBuilder -> {
logicEndpointBuilder.prepare();
sourceReceiver.receive(logicEndpointBuilder.toEndpoint());

32
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/RPCTrafficSourceBuilder.java
View File

@ -22,13 +22,10 @@ import lombok.Getter;
import lombok.Setter;
import org.apache.skywalking.oap.server.core.analysis.Layer;
import org.apache.skywalking.oap.server.core.config.NamingControl;
import org.apache.skywalking.oap.server.core.source.DatabaseAccess;
import org.apache.skywalking.oap.server.core.source.EndpointRelation;
import org.apache.skywalking.oap.server.core.source.RequestType;
import org.apache.skywalking.oap.server.core.source.Service;
import org.apache.skywalking.oap.server.core.source.ServiceInstance;
import org.apache.skywalking.oap.server.core.source.ServiceInstanceRelation;
import org.apache.skywalking.oap.server.core.source.ServiceMeta;
import org.apache.skywalking.oap.server.core.source.ServiceRelation;
import org.apache.skywalking.oap.server.library.util.StringUtil;
@ -204,33 +201,4 @@ class RPCTrafficSourceBuilder extends EndpointSourceBuilder {
endpointRelation.setTimeBucket(timeBucket);
return endpointRelation;
}
/**
* Service meta is only for building the service list, but wouldn't be same as {@link #toService()}, which could
* generate traffic and metrics both.
*/
ServiceMeta toServiceMeta() {
ServiceMeta service = new ServiceMeta();
service.setName(destServiceName);
service.setLayer(destLayer);
service.setLayer(destLayer);
service.setTimeBucket(timeBucket);
return service;
}
/**
* Database traffic metrics source. The metrics base on the OAL scripts.
*/
DatabaseAccess toDatabaseAccess() {
if (!RequestType.DATABASE.equals(type)) {
return null;
}
DatabaseAccess databaseAccess = new DatabaseAccess();
databaseAccess.setDatabaseTypeId(componentId);
databaseAccess.setLatency(latency);
databaseAccess.setName(destServiceName);
databaseAccess.setStatus(status);
databaseAccess.setTimeBucket(timeBucket);
return databaseAccess;
}
}

89
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/VirtualServiceAnalysisListener.java Normal file
View File

@ -0,0 +1,89 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener;
import lombok.RequiredArgsConstructor;
import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
import org.apache.skywalking.apm.network.language.agent.v3.SpanObject;
import org.apache.skywalking.oap.server.analyzer.provider.AnalyzerModuleConfig;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.vservice.VirtualCacheProcessor;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.vservice.VirtualDatabaseProcessor;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.vservice.VirtualServiceProcessor;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.config.NamingControl;
import org.apache.skywalking.oap.server.core.source.SourceReceiver;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import java.util.Arrays;
import java.util.List;
/**
* Virtual Service represent remote service
*/
@RequiredArgsConstructor
public class VirtualServiceAnalysisListener implements ExitAnalysisListener, LocalAnalysisListener {
private final SourceReceiver sourceReceiver;
private final List<VirtualServiceProcessor> virtualServiceProcessors;
@Override
public void build() {
virtualServiceProcessors.forEach(p -> p.emitTo(sourceReceiver::receive));
}
@Override
public boolean containsPoint(Point point) {
return point == Point.Local || point == Point.Exit;
}
@Override
public void parseExit(SpanObject span, SegmentObject segmentObject) {
virtualServiceProcessors.forEach(p -> p.prepareVSIfNecessary(span, segmentObject));
}
@Override
public void parseLocal(SpanObject span, SegmentObject segmentObject) {
virtualServiceProcessors.forEach(p -> p.prepareVSIfNecessary(span, segmentObject));
}
public static class Factory implements AnalysisListenerFactory {
private final SourceReceiver sourceReceiver;
private final NamingControl namingControl;
public Factory(ModuleManager moduleManager) {
this.sourceReceiver = moduleManager.find(CoreModule.NAME).provider().getService(SourceReceiver.class);
this.namingControl = moduleManager.find(CoreModule.NAME)
.provider()
.getService(NamingControl.class);
}
@Override
public AnalysisListener create(ModuleManager moduleManager, AnalyzerModuleConfig config) {
return new VirtualServiceAnalysisListener(sourceReceiver,
Arrays.asList(
new VirtualCacheProcessor(namingControl, config),
new VirtualDatabaseProcessor(namingControl, config)
)
);
}
}
}

128
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/vservice/VirtualCacheProcessor.java Normal file
View File

@ -0,0 +1,128 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.vservice;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
import org.apache.skywalking.apm.network.language.agent.v3.SpanLayer;
import org.apache.skywalking.apm.network.language.agent.v3.SpanObject;
import org.apache.skywalking.oap.server.analyzer.provider.AnalyzerModuleConfig;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.SpanTags;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
import org.apache.skywalking.oap.server.core.analysis.Layer;
import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
import org.apache.skywalking.oap.server.core.config.NamingControl;
import org.apache.skywalking.oap.server.core.source.ServiceMeta;
import org.apache.skywalking.oap.server.core.source.Source;
import org.apache.skywalking.oap.server.core.source.VirtualCacheAccess;
import org.apache.skywalking.oap.server.core.source.VirtualCacheOperation;
import org.apache.skywalking.oap.server.core.source.VirtualCacheSlowAccess;
import org.apache.skywalking.oap.server.library.util.StringUtil;
@Slf4j
@RequiredArgsConstructor
public class VirtualCacheProcessor implements VirtualServiceProcessor {
private final NamingControl namingControl;
private final AnalyzerModuleConfig config;
private final List<Source> sourceList = new ArrayList<>();
@Override
public void prepareVSIfNecessary(SpanObject span, SegmentObject segmentObject) {
if (span.getSpanLayer() != SpanLayer.Cache) {
return;
}
Map<String, String> tags = span.getTagsList().stream()
.collect(
Collectors.toMap(KeyStringValuePair::getKey, KeyStringValuePair::getValue));
String cacheType = tags.get(SpanTags.CACHE_TYPE);
if (StringUtil.isBlank(cacheType)) {
return;
}
cacheType = cacheType.toLowerCase();
String peer = span.getPeer();
// peer is blank if it's a local span.
if (StringUtil.isBlank(peer)) {
peer = tags.get(SpanTags.CACHE_TYPE) + "-local";
}
long timeBucket = TimeBucket.getMinuteTimeBucket(span.getStartTime());
String serviceName = namingControl.formatServiceName(peer);
int latency = (int) (span.getEndTime() - span.getStartTime());
sourceList.add(parseServiceMeta(serviceName, timeBucket));
VirtualCacheOperation op = parseOperation(tags.get(SpanTags.CACHE_OP));
if ((op == VirtualCacheOperation.Write && latency > config.getCacheWriteLatencyThresholdsAndWatcher()
.getThreshold(cacheType))
|| (op == VirtualCacheOperation.Read && latency > config.getCacheReadLatencyThresholdsAndWatcher()
.getThreshold(cacheType))) {
VirtualCacheSlowAccess slowAccess = new VirtualCacheSlowAccess();
slowAccess.setCacheServiceId(IDManager.ServiceID.buildId(serviceName, false));
slowAccess.setLatency(latency);
slowAccess.setId(segmentObject.getTraceSegmentId() + "-" + span.getSpanId());
slowAccess.setStatus(!span.getIsError());
slowAccess.setTraceId(segmentObject.getTraceId());
slowAccess.setCommand(tags.get(SpanTags.CACHE_CMD));
slowAccess.setKey(tags.get(SpanTags.CACHE_KEY));
slowAccess.setTimeBucket(TimeBucket.getRecordTimeBucket(span.getStartTime()));
slowAccess.setOperation(op);
sourceList.add(slowAccess);
}
VirtualCacheAccess access = new VirtualCacheAccess();
access.setCacheTypeId(span.getComponentId());
access.setLatency(latency);
access.setName(serviceName);
access.setStatus(!span.getIsError());
access.setTimeBucket(timeBucket);
access.setOperation(op);
sourceList.add(access);
}
private ServiceMeta parseServiceMeta(String serviceName, long timeBucket) {
ServiceMeta service = new ServiceMeta();
service.setName(serviceName);
service.setLayer(Layer.VIRTUAL_CACHE);
service.setTimeBucket(timeBucket);
return service;
}
private VirtualCacheOperation parseOperation(String op) {
if ("write".equals(op)) {
return VirtualCacheOperation.Write;
}
if ("read".equals(op)) {
return VirtualCacheOperation.Read;
}
return VirtualCacheOperation.Others;
}
@Override
public void emitTo(Consumer<Source> consumer) {
sourceList.forEach(consumer);
}
}

122
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/vservice/VirtualDatabaseProcessor.java Normal file
View File

@ -0,0 +1,122 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.vservice;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
import org.apache.skywalking.apm.network.language.agent.v3.SpanLayer;
import org.apache.skywalking.apm.network.language.agent.v3.SpanObject;
import org.apache.skywalking.oap.server.analyzer.provider.AnalyzerModuleConfig;
import org.apache.skywalking.oap.server.analyzer.provider.trace.DBLatencyThresholdsAndWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.SpanTags;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
import org.apache.skywalking.oap.server.core.analysis.Layer;
import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
import org.apache.skywalking.oap.server.core.config.NamingControl;
import org.apache.skywalking.oap.server.core.source.DatabaseAccess;
import org.apache.skywalking.oap.server.core.source.DatabaseSlowStatement;
import org.apache.skywalking.oap.server.core.source.ServiceMeta;
import org.apache.skywalking.oap.server.core.source.Source;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.function.Consumer;
@Slf4j
@RequiredArgsConstructor
public class VirtualDatabaseProcessor implements VirtualServiceProcessor {
private final NamingControl namingControl;
private final AnalyzerModuleConfig config;
private List<Source> recordList = new ArrayList<>();
@Override
public void prepareVSIfNecessary(SpanObject span, SegmentObject segmentObject) {
if (span.getSpanLayer() != SpanLayer.Database) {
return;
}
String peer = span.getPeer();
long timeBucket = TimeBucket.getMinuteTimeBucket(span.getStartTime());
String serviceName = namingControl.formatServiceName(peer);
int latency = (int) (span.getEndTime() - span.getStartTime());
recordList.add(toServiceMeta(serviceName, timeBucket));
recordList.add(toDatabaseAccess(span, serviceName, timeBucket, latency));
readStatementIfSlow(span.getTagsList(), latency).ifPresent(statement -> {
DatabaseSlowStatement dbSlowStat = new DatabaseSlowStatement();
dbSlowStat.setId(segmentObject.getTraceSegmentId() + "-" + span.getSpanId());
dbSlowStat.setTraceId(segmentObject.getTraceId());
dbSlowStat.setDatabaseServiceId(IDManager.ServiceID.buildId(serviceName, false));
dbSlowStat.setStatement(statement);
dbSlowStat.setLatency(latency);
dbSlowStat.setTimeBucket(TimeBucket.getRecordTimeBucket(span.getStartTime()));
recordList.add(dbSlowStat);
});
}
private Optional<String> readStatementIfSlow(List<KeyStringValuePair> tags, int latency) {
String statement = null;
boolean isSlowDBAccess = false;
for (KeyStringValuePair tag : tags) {
if (SpanTags.DB_STATEMENT.equals(tag.getKey())) {
statement = StringUtil.cut(tag.getValue(), config.getMaxSlowSQLLength());
} else if (SpanTags.DB_TYPE.equals(tag.getKey())) {
String dbType = tag.getValue();
DBLatencyThresholdsAndWatcher thresholds = config.getDbLatencyThresholdsAndWatcher();
int threshold = thresholds.getThreshold(dbType);
if (latency > threshold) {
isSlowDBAccess = true;
}
}
}
if (isSlowDBAccess) {
return Optional.ofNullable(statement).filter(StringUtil::isNotBlank);
}
return Optional.empty();
}
private ServiceMeta toServiceMeta(String serviceName, Long timeBucket) {
ServiceMeta service = new ServiceMeta();
service.setName(serviceName);
service.setLayer(Layer.VIRTUAL_DATABASE);
service.setTimeBucket(timeBucket);
return service;
}
private DatabaseAccess toDatabaseAccess(SpanObject span, String serviceName, long timeBucket, int latency) {
DatabaseAccess databaseAccess = new DatabaseAccess();
databaseAccess.setDatabaseTypeId(span.getComponentId());
databaseAccess.setLatency(latency);
databaseAccess.setName(serviceName);
databaseAccess.setStatus(!span.getIsError());
databaseAccess.setTimeBucket(timeBucket);
return databaseAccess;
}
@Override
public void emitTo(Consumer<Source> consumer) {
recordList.forEach(consumer);
}
}

42
oap-server/analyzer/agent-analyzer/src/main/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/vservice/VirtualServiceProcessor.java Normal file
View File

@ -0,0 +1,42 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.vservice;
import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
import org.apache.skywalking.apm.network.language.agent.v3.SpanObject;
import org.apache.skywalking.oap.server.core.source.Source;
import java.util.function.Consumer;
/**
* Virtual Service represent remote service
*/
public interface VirtualServiceProcessor {
/**
* Parse virtual service metadata and metrics data if the span it's appropriate
*/
void prepareVSIfNecessary(final SpanObject span, final SegmentObject segmentObject);
/**
* Emit collected metadata , metrics data to consumer
*/
void emitTo(Consumer<Source> consumer);
}

190
oap-server/analyzer/agent-analyzer/src/test/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/vservice/VirtualCacheProcessorTest.java Normal file
View File

@ -0,0 +1,190 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.vservice;
import com.google.protobuf.ByteString;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
import org.apache.skywalking.apm.network.language.agent.v3.SpanLayer;
import org.apache.skywalking.apm.network.language.agent.v3.SpanObject;
import org.apache.skywalking.apm.network.language.agent.v3.SpanType;
import org.apache.skywalking.oap.server.analyzer.provider.AnalyzerModuleConfig;
import org.apache.skywalking.oap.server.analyzer.provider.trace.CacheReadLatencyThresholdsAndWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.SpanTags;
import org.apache.skywalking.oap.server.core.analysis.Layer;
import org.apache.skywalking.oap.server.core.config.NamingControl;
import org.apache.skywalking.oap.server.core.config.group.EndpointNameGrouping;
import org.apache.skywalking.oap.server.core.source.ServiceMeta;
import org.apache.skywalking.oap.server.core.source.Source;
import org.apache.skywalking.oap.server.core.source.VirtualCacheAccess;
import org.apache.skywalking.oap.server.core.source.VirtualCacheOperation;
import org.apache.skywalking.oap.server.core.source.VirtualCacheSlowAccess;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.junit.Assert;
import org.junit.Test;
public class VirtualCacheProcessorTest {
@Test
public void testEmptySpan() {
SpanObject spanObject = SpanObject.newBuilder().setSpanLayer(SpanLayer.Cache).build();
SegmentObject segmentObject = SegmentObject.newBuilder().build();
VirtualCacheProcessor cacheVirtualServiceProcessor = buildCacheVirtualServiceProcessor();
cacheVirtualServiceProcessor.prepareVSIfNecessary(spanObject, segmentObject);
ArrayList<Source> sources = new ArrayList<>();
cacheVirtualServiceProcessor.emitTo(sources::add);
Assert.assertTrue(sources.isEmpty());
}
@Test
public void testExitSpan() {
SpanObject spanObject = SpanObject.newBuilder()
.setSpanLayer(SpanLayer.Cache)
.setSpanId(0)
.addAllTags(buildTags())
.setSpanType(SpanType.Exit)
.setPeerBytes(
ByteString.copyFrom("127.0.0.1:6379".getBytes(StandardCharsets.UTF_8)))
.setStartTime(getTimeInMillis("2022-09-12 14:13:12.790"))
.setEndTime(getTimeInMillis("2022-09-12 14:13:13.790"))
.build();
SegmentObject segmentObject = SegmentObject.newBuilder().setTraceId("trace-id").build();
VirtualCacheProcessor cacheVirtualServiceProcessor = buildCacheVirtualServiceProcessor();
cacheVirtualServiceProcessor.prepareVSIfNecessary(spanObject, segmentObject);
ArrayList<Source> sources = new ArrayList<>();
cacheVirtualServiceProcessor.emitTo(sources::add);
Assert.assertEquals(sources.size(), 3);
ServiceMeta serviceMeta = (ServiceMeta) sources.get(0);
Assert.assertEquals("127.0.0.1:6379", serviceMeta.getName());
Assert.assertEquals(202209121413L, serviceMeta.getTimeBucket());
Assert.assertEquals(Layer.VIRTUAL_CACHE, serviceMeta.getLayer());
VirtualCacheSlowAccess slowAccess = (VirtualCacheSlowAccess) sources.get(1);
Assert.assertEquals("MTI3LjAuMC4xOjYzNzk=.0", slowAccess.getCacheServiceId());
Assert.assertEquals(1000, slowAccess.getLatency());
Assert.assertEquals(20220912141312L, slowAccess.getTimeBucket());
Assert.assertEquals(VirtualCacheOperation.Read, slowAccess.getOperation());
Assert.assertNotNull(slowAccess.getTraceId());
Assert.assertNotNull(slowAccess.getCommand());
Assert.assertNotNull(slowAccess.getKey());
VirtualCacheAccess cacheAccess = (VirtualCacheAccess) sources.get(2);
Assert.assertEquals("127.0.0.1:6379", cacheAccess.getName());
Assert.assertEquals(1000, cacheAccess.getLatency());
Assert.assertEquals(202209121413L, cacheAccess.getTimeBucket());
Assert.assertNotNull(cacheAccess.getOperation());
}
@Test
public void testExitSpanLessThreshold() {
SpanObject spanObject = SpanObject.newBuilder()
.setSpanLayer(SpanLayer.Cache)
.setSpanId(0)
.addAllTags(buildTags())
.setSpanType(SpanType.Exit)
.setPeerBytes(
ByteString.copyFrom("127.0.0.1:6379".getBytes(StandardCharsets.UTF_8)))
.setStartTime(getTimeInMillis("2022-09-12 14:13:12.790"))
.setEndTime(getTimeInMillis("2022-09-12 14:13:12.793"))
.build();
SegmentObject segmentObject = SegmentObject.newBuilder().build();
VirtualCacheProcessor cacheVirtualServiceProcessor = buildCacheVirtualServiceProcessor();
cacheVirtualServiceProcessor.prepareVSIfNecessary(spanObject, segmentObject);
ArrayList<Source> sources = new ArrayList<>();
cacheVirtualServiceProcessor.emitTo(sources::add);
Assert.assertEquals(sources.size(), 2);
ServiceMeta serviceMeta = (ServiceMeta) sources.get(0);
Assert.assertEquals("127.0.0.1:6379", serviceMeta.getName());
Assert.assertEquals(202209121413L, serviceMeta.getTimeBucket());
Assert.assertEquals(Layer.VIRTUAL_CACHE, serviceMeta.getLayer());
VirtualCacheAccess cacheAccess = (VirtualCacheAccess) sources.get(1);
Assert.assertEquals("127.0.0.1:6379", cacheAccess.getName());
Assert.assertEquals(3, cacheAccess.getLatency());
Assert.assertEquals(202209121413L, cacheAccess.getTimeBucket());
Assert.assertNotNull(cacheAccess.getOperation());
}
@Test
public void testLocalSpan() {
SpanObject spanObject = SpanObject.newBuilder()
.setSpanLayer(SpanLayer.Cache)
.setSpanId(0)
.addAllTags(buildTags())
.setSpanType(SpanType.Local)
.setStartTime(getTimeInMillis("2022-09-12 14:13:12.790"))
.setEndTime(getTimeInMillis("2022-09-12 14:13:13.790"))
.build();
SegmentObject segmentObject = SegmentObject.newBuilder().build();
VirtualCacheProcessor cacheVirtualServiceProcessor = buildCacheVirtualServiceProcessor();
cacheVirtualServiceProcessor.prepareVSIfNecessary(spanObject, segmentObject);
ArrayList<Source> sources = new ArrayList<>();
cacheVirtualServiceProcessor.emitTo(sources::add);
Assert.assertEquals(sources.size(), 3);
ServiceMeta serviceMeta = (ServiceMeta) sources.get(0);
Assert.assertEquals("redis-local", serviceMeta.getName());
Assert.assertEquals(202209121413L, serviceMeta.getTimeBucket());
Assert.assertEquals(Layer.VIRTUAL_CACHE, serviceMeta.getLayer());
VirtualCacheSlowAccess slowAccess = (VirtualCacheSlowAccess) sources.get(1);
Assert.assertEquals("cmVkaXMtbG9jYWw=.0", slowAccess.getCacheServiceId());
Assert.assertEquals(1000, slowAccess.getLatency());
Assert.assertEquals(20220912141312L, slowAccess.getTimeBucket());
Assert.assertEquals(VirtualCacheOperation.Read, slowAccess.getOperation());
Assert.assertNotNull(slowAccess.getTraceId());
Assert.assertNotNull(slowAccess.getCommand());
Assert.assertNotNull(slowAccess.getKey());
VirtualCacheAccess cacheAccess = (VirtualCacheAccess) sources.get(2);
Assert.assertEquals("redis-local", cacheAccess.getName());
Assert.assertEquals(1000, cacheAccess.getLatency());
Assert.assertEquals(202209121413L, cacheAccess.getTimeBucket());
Assert.assertNotNull(cacheAccess.getOperation());
}
private long getTimeInMillis(String s) {
return DateTime.parse(s, DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSS")).getMillis();
}
private List<KeyStringValuePair> buildTags() {
return Arrays.asList(
KeyStringValuePair.newBuilder().setKey(SpanTags.CACHE_KEY).setValue("test_key").build(),
KeyStringValuePair.newBuilder().setKey(SpanTags.CACHE_TYPE).setValue("redis").build(),
KeyStringValuePair.newBuilder().setKey(SpanTags.CACHE_CMD).setValue("get").build(),
KeyStringValuePair.newBuilder().setKey(SpanTags.CACHE_OP).setValue("read").build()
);
}
private VirtualCacheProcessor buildCacheVirtualServiceProcessor() {
NamingControl namingControl = new NamingControl(512, 512, 512, new EndpointNameGrouping());
AnalyzerModuleConfig config = new AnalyzerModuleConfig();
config.setCacheReadLatencyThresholdsAndWatcher(new CacheReadLatencyThresholdsAndWatcher("default:10", null));
return new VirtualCacheProcessor(namingControl, config);
}
}

145
oap-server/analyzer/agent-analyzer/src/test/java/org/apache/skywalking/oap/server/analyzer/provider/trace/parser/listener/vservice/VirtualDatabaseProcessorTest.java Normal file
View File

@ -0,0 +1,145 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.vservice;
import com.google.protobuf.ByteString;
import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
import org.apache.skywalking.apm.network.language.agent.v3.SpanLayer;
import org.apache.skywalking.apm.network.language.agent.v3.SpanObject;
import org.apache.skywalking.apm.network.language.agent.v3.SpanType;
import org.apache.skywalking.oap.server.analyzer.provider.AnalyzerModuleConfig;
import org.apache.skywalking.oap.server.analyzer.provider.trace.DBLatencyThresholdsAndWatcher;
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.SpanTags;
import org.apache.skywalking.oap.server.core.analysis.Layer;
import org.apache.skywalking.oap.server.core.config.NamingControl;
import org.apache.skywalking.oap.server.core.config.group.EndpointNameGrouping;
import org.apache.skywalking.oap.server.core.source.DatabaseAccess;
import org.apache.skywalking.oap.server.core.source.DatabaseSlowStatement;
import org.apache.skywalking.oap.server.core.source.ServiceMeta;
import org.apache.skywalking.oap.server.core.source.Source;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.junit.Assert;
import org.junit.Test;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
public class VirtualDatabaseProcessorTest {
@Test
public void testEmptySpan() {
SpanObject spanObject = SpanObject.newBuilder().setSpanLayer(SpanLayer.Cache).build();
SegmentObject segmentObject = SegmentObject.newBuilder().build();
VirtualDatabaseProcessor processor = buildVirtualServiceProcessor();
processor.prepareVSIfNecessary(spanObject, segmentObject);
ArrayList<Source> sources = new ArrayList<>();
processor.emitTo(sources::add);
Assert.assertTrue(sources.isEmpty());
}
@Test
public void testExitSpan() {
SpanObject spanObject = SpanObject.newBuilder()
.setSpanLayer(SpanLayer.Database)
.setSpanId(0)
.addAllTags(buildTags())
.setSpanType(SpanType.Exit)
.setPeerBytes(ByteString.copyFrom("127.0.0.1:3306".getBytes(StandardCharsets.UTF_8)))
.setStartTime(getTimeInMillis("2022-09-12 14:13:12.790"))
.setEndTime(getTimeInMillis("2022-09-12 14:13:13.790"))
.build();
SegmentObject segmentObject = SegmentObject.newBuilder()
.setTraceId("trace-id-1")
.build();
VirtualDatabaseProcessor processor = buildVirtualServiceProcessor();
processor.prepareVSIfNecessary(spanObject, segmentObject);
ArrayList<Source> sources = new ArrayList<>();
processor.emitTo(sources::add);
Assert.assertEquals(sources.size(), 3);
ServiceMeta serviceMeta = (ServiceMeta) sources.get(0);
Assert.assertEquals("127.0.0.1:3306", serviceMeta.getName());
Assert.assertEquals(202209121413L, serviceMeta.getTimeBucket());
Assert.assertEquals(Layer.VIRTUAL_DATABASE, serviceMeta.getLayer());
DatabaseAccess databaseAccess = (DatabaseAccess) sources.get(1);
Assert.assertEquals("127.0.0.1:3306", databaseAccess.getName());
Assert.assertEquals(1000, databaseAccess.getLatency());
Assert.assertEquals(202209121413L, databaseAccess.getTimeBucket());
DatabaseSlowStatement slowStatement = (DatabaseSlowStatement) sources.get(2);
Assert.assertEquals("MTI3LjAuMC4xOjMzMDY=.0", slowStatement.getDatabaseServiceId());
Assert.assertEquals(1000, slowStatement.getLatency());
Assert.assertEquals(20220912141312L, slowStatement.getTimeBucket());
Assert.assertEquals("trace-id-1", slowStatement.getTraceId());
}
@Test
public void testExitSpanLessThreshold() {
SpanObject spanObject = SpanObject.newBuilder()
.setSpanLayer(SpanLayer.Database)
.setSpanId(0)
.addAllTags(buildTags())
.setSpanType(SpanType.Exit)
.setPeerBytes(ByteString.copyFrom("127.0.0.1:3306".getBytes(StandardCharsets.UTF_8)))
.setStartTime(getTimeInMillis("2022-09-12 14:13:12.790"))
.setEndTime(getTimeInMillis("2022-09-12 14:13:12.793"))
.build();
SegmentObject segmentObject = SegmentObject.newBuilder().build();
VirtualDatabaseProcessor processor = buildVirtualServiceProcessor();
processor.prepareVSIfNecessary(spanObject, segmentObject);
ArrayList<Source> sources = new ArrayList<>();
processor.emitTo(sources::add);
Assert.assertEquals(sources.size(), 2);
ServiceMeta serviceMeta = (ServiceMeta) sources.get(0);
Assert.assertEquals("127.0.0.1:3306", serviceMeta.getName());
Assert.assertEquals(202209121413L, serviceMeta.getTimeBucket());
Assert.assertEquals(Layer.VIRTUAL_DATABASE, serviceMeta.getLayer());
DatabaseAccess databaseAccess = (DatabaseAccess) sources.get(1);
Assert.assertEquals("127.0.0.1:3306", databaseAccess.getName());
Assert.assertEquals(3, databaseAccess.getLatency());
Assert.assertEquals(202209121413L, databaseAccess.getTimeBucket());
}
private long getTimeInMillis(String s) {
return DateTime.parse(s, DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSS")).getMillis();
}
private List<KeyStringValuePair> buildTags() {
return Arrays.asList(
KeyStringValuePair.newBuilder().setKey(SpanTags.DB_STATEMENT).setValue("select * from dual").build(),
KeyStringValuePair.newBuilder().setKey(SpanTags.DB_TYPE).setValue("Mysql").build()
);
}
private VirtualDatabaseProcessor buildVirtualServiceProcessor() {
NamingControl namingControl = new NamingControl(512, 512, 512, new EndpointNameGrouping());
AnalyzerModuleConfig config = new AnalyzerModuleConfig();
config.setDbLatencyThresholdsAndWatcher(new DBLatencyThresholdsAndWatcher("default:10", null));
return new VirtualDatabaseProcessor(namingControl, config);
}
}

2
oap-server/oal-grammar/src/main/antlr4/org/apache/skywalking/oal/rt/grammar/OALLexer.g4
View File

@ -45,6 +45,8 @@ SRC_SERVICE_INSTANCE_CLR_GC: 'ServiceInstanceCLRGC';
SRC_SERVICE_INSTANCE_CLR_THREAD: 'ServiceInstanceCLRThread';
SRC_ENVOY_INSTANCE_METRIC: 'EnvoyInstanceMetric';
SRC_EVENT: 'Event';
SRC_CACHE_ACCESS: 'VirtualCacheAccess';
// Browser keywords
SRC_BROWSER_APP_PERF: 'BrowserAppPerf';

2
oap-server/oal-grammar/src/main/antlr4/org/apache/skywalking/oal/rt/grammar/OALParser.g4
View File

@ -50,7 +50,7 @@ filterExpression
;
source
: SRC_SERVICE | SRC_DATABASE_ACCESS | SRC_SERVICE_INSTANCE | SRC_ENDPOINT |
: SRC_SERVICE | SRC_DATABASE_ACCESS | SRC_SERVICE_INSTANCE | SRC_ENDPOINT | SRC_CACHE_ACCESS |
SRC_SERVICE_RELATION | SRC_SERVICE_INSTANCE_RELATION | SRC_ENDPOINT_RELATION |
SRC_SERVICE_INSTANCE_CLR_CPU | SRC_SERVICE_INSTANCE_CLR_GC | SRC_SERVICE_INSTANCE_CLR_THREAD |
SRC_SERVICE_INSTANCE_JVM_CPU | SRC_SERVICE_INSTANCE_JVM_MEMORY | SRC_SERVICE_INSTANCE_JVM_MEMORY_POOL | SRC_SERVICE_INSTANCE_JVM_GC | SRC_SERVICE_INSTANCE_JVM_THREAD | SRC_SERVICE_INSTANCE_JVM_CLASS |// JVM source of service instance

51
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/cache/CacheSlowAccessDispatcher.java vendored Normal file
View File

@ -0,0 +1,51 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.core.analysis.manual.cache;
import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher;
import org.apache.skywalking.oap.server.core.analysis.worker.TopNStreamProcessor;
import org.apache.skywalking.oap.server.core.source.VirtualCacheOperation;
import org.apache.skywalking.oap.server.core.source.VirtualCacheSlowAccess;
public class CacheSlowAccessDispatcher implements SourceDispatcher<VirtualCacheSlowAccess> {
@Override
public void dispatch(VirtualCacheSlowAccess source) {
// There are only two kinds of Operation : write or read .Refer VirtualCacheProcessor#prepareVSIfNecessary
if (source.getOperation() == VirtualCacheOperation.Read) {
TopNCacheReadCommand readCommand = new TopNCacheReadCommand();
readCommand.setId(source.getId());
readCommand.setCommand(source.getCommand() + " " + source.getKey());
readCommand.setLatency(source.getLatency());
readCommand.setTraceId(source.getTraceId());
readCommand.setServiceId(source.getCacheServiceId());
readCommand.setTimeBucket(source.getTimeBucket());
TopNStreamProcessor.getInstance().in(readCommand);
} else if (source.getOperation() == VirtualCacheOperation.Write) {
TopNCacheWriteCommand writeCommand = new TopNCacheWriteCommand();
writeCommand.setId(source.getId());
writeCommand.setCommand(source.getCommand() + " " + source.getKey());
writeCommand.setLatency(source.getLatency());
writeCommand.setTraceId(source.getTraceId());
writeCommand.setServiceId(source.getCacheServiceId());
writeCommand.setTimeBucket(source.getTimeBucket());
TopNStreamProcessor.getInstance().in(writeCommand);
}
}
}

88
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/cache/TopNCacheReadCommand.java vendored Normal file
View File

@ -0,0 +1,88 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.core.analysis.manual.cache;
import java.util.Objects;
import lombok.Getter;
import lombok.Setter;
import org.apache.skywalking.oap.server.core.analysis.Stream;
import org.apache.skywalking.oap.server.core.analysis.topn.TopN;
import org.apache.skywalking.oap.server.core.analysis.worker.TopNStreamProcessor;
import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
import org.apache.skywalking.oap.server.core.storage.type.Convert2Entity;
import org.apache.skywalking.oap.server.core.storage.type.Convert2Storage;
import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
/**
* Database TopN statement, including Database SQL statement, mongoDB and Redis commands.
*/
@Stream(name = TopNCacheReadCommand.INDEX_NAME, scopeId = DefaultScopeDefine.CACHE_SLOW_ACCESS, builder = TopNCacheReadCommand.Builder.class, processor = TopNStreamProcessor.class)
public class TopNCacheReadCommand extends TopN {
public static final String INDEX_NAME = "top_n_cache_read_command";
@Setter
private String id;
@Getter
@Setter
@Column(columnName = STATEMENT, length = 2000, lengthEnvVariable = "SW_SLOW_DB_THRESHOLD", storageOnly = true)
private String command;
@Override
public String id() {
return id;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
TopNCacheReadCommand statement = (TopNCacheReadCommand) o;
return Objects.equals(getServiceId(), statement.getServiceId());
}
@Override
public int hashCode() {
return Objects.hash(getServiceId());
}
public static class Builder implements StorageBuilder<TopNCacheReadCommand> {
@Override
public TopNCacheReadCommand storage2Entity(final Convert2Entity converter) {
TopNCacheReadCommand statement = new TopNCacheReadCommand();
statement.setCommand((String) converter.get(STATEMENT));
statement.setTraceId((String) converter.get(TRACE_ID));
statement.setLatency(((Number) converter.get(LATENCY)).longValue());
statement.setServiceId((String) converter.get(SERVICE_ID));
statement.setTimeBucket(((Number) converter.get(TIME_BUCKET)).longValue());
return statement;
}
@Override
public void entity2Storage(final TopNCacheReadCommand storageData, final Convert2Storage converter) {
converter.accept(STATEMENT, storageData.getCommand());
converter.accept(TRACE_ID, storageData.getTraceId());
converter.accept(LATENCY, storageData.getLatency());
converter.accept(SERVICE_ID, storageData.getServiceId());
converter.accept(TIME_BUCKET, storageData.getTimeBucket());
}
}
}

89
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/cache/TopNCacheWriteCommand.java vendored Normal file
View File

@ -0,0 +1,89 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.core.analysis.manual.cache;
import lombok.Getter;
import lombok.Setter;
import org.apache.skywalking.oap.server.core.analysis.Stream;
import org.apache.skywalking.oap.server.core.analysis.topn.TopN;
import org.apache.skywalking.oap.server.core.analysis.worker.TopNStreamProcessor;
import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
import org.apache.skywalking.oap.server.core.storage.type.Convert2Entity;
import org.apache.skywalking.oap.server.core.storage.type.Convert2Storage;
import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
import java.util.Objects;
/**
* Database TopN statement, including Database SQL statement, mongoDB and Redis commands.
*/
@Stream(name = TopNCacheWriteCommand.INDEX_NAME, scopeId = DefaultScopeDefine.CACHE_SLOW_ACCESS, builder = TopNCacheWriteCommand.Builder.class, processor = TopNStreamProcessor.class)
public class TopNCacheWriteCommand extends TopN {
public static final String INDEX_NAME = "top_n_cache_write_command";
@Setter
private String id;
@Getter
@Setter
@Column(columnName = STATEMENT, length = 2000, lengthEnvVariable = "SW_SLOW_DB_THRESHOLD", storageOnly = true)
private String command;
@Override
public String id() {
return id;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
TopNCacheWriteCommand statement = (TopNCacheWriteCommand) o;
return Objects.equals(getServiceId(), statement.getServiceId());
}
@Override
public int hashCode() {
return Objects.hash(getServiceId());
}
public static class Builder implements StorageBuilder<TopNCacheWriteCommand> {
@Override
public TopNCacheWriteCommand storage2Entity(final Convert2Entity converter) {
TopNCacheWriteCommand statement = new TopNCacheWriteCommand();
statement.setCommand((String) converter.get(STATEMENT));
statement.setTraceId((String) converter.get(TRACE_ID));
statement.setLatency(((Number) converter.get(LATENCY)).longValue());
statement.setServiceId((String) converter.get(SERVICE_ID));
statement.setTimeBucket(((Number) converter.get(TIME_BUCKET)).longValue());
return statement;
}
@Override
public void entity2Storage(final TopNCacheWriteCommand storageData, final Convert2Storage converter) {
converter.accept(STATEMENT, storageData.getCommand());
converter.accept(TRACE_ID, storageData.getTraceId());
converter.accept(LATENCY, storageData.getLatency());
converter.accept(SERVICE_ID, storageData.getServiceId());
converter.accept(TIME_BUCKET, storageData.getTimeBucket());
}
}
}

2
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/database/TopNDatabaseStatement.java
View File

@ -56,7 +56,7 @@ public class TopNDatabaseStatement extends TopN {
if (o == null || getClass() != o.getClass())
return false;
TopNDatabaseStatement statement = (TopNDatabaseStatement) o;
return getServiceId() == statement.getServiceId();
return Objects.equals(getServiceId(), statement.getServiceId());
}
@Override

1
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/management/ui/template/UITemplateInitializer.java
View File

@ -54,6 +54,7 @@ public class UITemplateInitializer {
Layer.BROWSER.name(),
Layer.SO11Y_OAP.name(),
Layer.VIRTUAL_DATABASE.name(),
Layer.VIRTUAL_CACHE.name(),
Layer.K8S_SERVICE.name(),
Layer.SO11Y_SATELLITE.name(),
Layer.FAAS.name(),

2
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java
View File

@ -106,6 +106,8 @@ public class DefaultScopeDefine {
public static final int ZIPKIN_SERVICE_SPAN = 52;
public static final int ZIPKIN_SERVICE_RELATION = 53;
public static final int PROCESS_RELATION = 54;
public static final int CACHE_ACCESS = 55;
public static final int CACHE_SLOW_ACCESS = 56;
/**
* Catalog of scope, the metrics processor could use this to group all generated metrics by oal rt.

58
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/VirtualCacheAccess.java Normal file
View File

@ -0,0 +1,58 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.core.source;
import lombok.Getter;
import lombok.Setter;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.CACHE_ACCESS;
import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE_CATALOG_NAME;
@ScopeDeclaration(id = CACHE_ACCESS, name = "VirtualCacheAccess", catalog = SERVICE_CATALOG_NAME)
@ScopeDefaultColumn.VirtualColumnDefinition(fieldName = "entityId", columnName = "entity_id", isID = true, type = String.class)
public class VirtualCacheAccess extends Source {
@Override
public int scope() {
return CACHE_ACCESS;
}
@Override
public String getEntityId() {
return IDManager.ServiceID.buildId(name, false);
}
@Getter
@Setter
@ScopeDefaultColumn.DefinedByField(columnName = "name", requireDynamicActive = true)
private String name;
@Getter
@Setter
private int cacheTypeId;
@Getter
@Setter
private int latency;
@Getter
@Setter
private boolean status;
@Getter
@Setter
private VirtualCacheOperation operation;
}

28
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/VirtualCacheOperation.java Normal file
View File

@ -0,0 +1,28 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.core.source;
/**
* Represents an operation of cache access
*/
public enum VirtualCacheOperation {
Write,
Read,
Others,
}

67
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/VirtualCacheSlowAccess.java Normal file
View File

@ -0,0 +1,67 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.core.source;
import lombok.Getter;
import lombok.Setter;
import org.apache.skywalking.oap.server.core.Const;
import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.CACHE_SLOW_ACCESS;
import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE_CATALOG_NAME;
@ScopeDeclaration(id = CACHE_SLOW_ACCESS, name = "VirtualCacheSlowAccess", catalog = SERVICE_CATALOG_NAME)
public class VirtualCacheSlowAccess extends Source {
@Getter
@Setter
private String id;
@Getter
@Setter
private String cacheServiceId;
@Getter
@Setter
private String command;
@Getter
@Setter
private String key;
@Getter
@Setter
private long latency;
@Getter
@Setter
private String traceId;
@Getter
@Setter
private boolean status;
@Getter
@Setter
private VirtualCacheOperation operation;
@Override
public int scope() {
return DefaultScopeDefine.CACHE_SLOW_ACCESS;
}
@Override
public String getEntityId() {
return Const.EMPTY_STRING;
}
}

13
oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/RPCAnalysisListenerTest.java
View File

@ -19,7 +19,6 @@
package org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener;
import com.google.gson.JsonObject;
import java.util.List;
import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
import org.apache.skywalking.apm.network.language.agent.v3.RefType;
import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
@ -34,19 +33,16 @@ import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.
import org.apache.skywalking.oap.server.analyzer.provider.trace.parser.listener.RPCAnalysisListener;
import org.apache.skywalking.oap.server.core.Const;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
import org.apache.skywalking.oap.server.core.analysis.Layer;
import org.apache.skywalking.oap.server.core.analysis.manual.networkalias.NetworkAddressAlias;
import org.apache.skywalking.oap.server.core.cache.NetworkAddressAliasCache;
import org.apache.skywalking.oap.server.core.config.NamingControl;
import org.apache.skywalking.oap.server.core.config.group.EndpointNameGrouping;
import org.apache.skywalking.oap.server.core.source.DatabaseAccess;
import org.apache.skywalking.oap.server.core.source.Endpoint;
import org.apache.skywalking.oap.server.core.source.EndpointRelation;
import org.apache.skywalking.oap.server.core.source.ISource;
import org.apache.skywalking.oap.server.core.source.Service;
import org.apache.skywalking.oap.server.core.source.ServiceInstance;
import org.apache.skywalking.oap.server.core.source.ServiceInstanceRelation;
import org.apache.skywalking.oap.server.core.source.ServiceMeta;
import org.apache.skywalking.oap.server.core.source.ServiceRelation;
import org.junit.Assert;
import org.junit.Before;
@ -55,6 +51,8 @@ import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
import java.util.List;
import static org.apache.skywalking.oap.server.analyzer.provider.trace.parser.SpanTags.LOGIC_ENDPOINT;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.when;
@ -418,18 +416,13 @@ public class RPCAnalysisListenerTest {
listener.build();
final List<ISource> receivedSources = mockReceiver.getReceivedSources();
Assert.assertEquals(4, receivedSources.size());
Assert.assertEquals(2, receivedSources.size());
final ServiceRelation serviceRelation = (ServiceRelation) receivedSources.get(0);
final ServiceInstanceRelation serviceInstanceRelation = (ServiceInstanceRelation) receivedSources.get(1);
final ServiceMeta serviceMeta = (ServiceMeta) receivedSources.get(2);
final DatabaseAccess databaseAccess = (DatabaseAccess) receivedSources.get(3);
Assert.assertEquals("mock-service", serviceRelation.getSourceServiceName());
Assert.assertEquals("127.0.0.1:8080", serviceRelation.getDestServiceName());
Assert.assertEquals("mock-instance", serviceInstanceRelation.getSourceServiceInstanceName());
Assert.assertEquals("127.0.0.1:8080", serviceInstanceRelation.getDestServiceInstanceName());
Assert.assertEquals("127.0.0.1:8080", serviceMeta.getName());
Assert.assertEquals(Layer.VIRTUAL_DATABASE, serviceMeta.getLayer());
Assert.assertEquals("127.0.0.1:8080", databaseAccess.getName());
}
/**

2
oap-server/server-starter/src/main/resources/application.yml
View File

@ -235,6 +235,8 @@ agent-analyzer:
# Exit spans with the component in the list would not generate the client-side instance relation metrics.
noUpstreamRealAddressAgents: ${SW_NO_UPSTREAM_REAL_ADDRESS:6000,9000}
meterAnalyzerActiveFiles: ${SW_METER_ANALYZER_ACTIVE_FILES:datasource,threadpool,satellite} # Which files could be meter analyzed, files split by ","
slowCacheReadThreshold: ${SW_SLOW_CACHE_SLOW_READ_THRESHOLD:default:20,redis:10} # The slow cache write operation thresholds. Unit ms.
slowCacheWriteThreshold: ${SW_SLOW_CACHE_SLOW_WRITE_THRESHOLD:default:20,redis:10} # The slow cache write operation thresholds. Unit ms.
log-analyzer:
selector: ${SW_LOG_ANALYZER:default}

15
oap-server/server-starter/src/main/resources/oal/core.oal
View File

@ -71,3 +71,18 @@ database_access_resp_time = from(DatabaseAccess.latency).longAvg();
database_access_sla = from(DatabaseAccess.*).percent(status == true);
database_access_cpm = from(DatabaseAccess.*).cpm();
database_access_percentile = from(DatabaseAccess.latency).percentile(10);
cache_read_resp_time = from(VirtualCacheAccess.latency).filter(operation == VirtualCacheOperation.Read).longAvg();
cache_read_sla = from(VirtualCacheAccess.*).filter(operation == VirtualCacheOperation.Read).percent(status == true);
cache_read_cpm = from(VirtualCacheAccess.*).filter(operation == VirtualCacheOperation.Read).cpm();
cache_read_percentile = from(VirtualCacheAccess.latency).filter(operation == VirtualCacheOperation.Read).percentile(10);
cache_write_resp_time = from(VirtualCacheAccess.latency).filter(operation == VirtualCacheOperation.Write).longAvg();
cache_write_sla = from(VirtualCacheAccess.*).filter(operation == VirtualCacheOperation.Write).percent(status == true);
cache_write_cpm = from(VirtualCacheAccess.*).filter(operation == VirtualCacheOperation.Write).cpm();
cache_write_percentile = from(VirtualCacheAccess.latency).filter(operation == VirtualCacheOperation.Write).percentile(10);
cache_access_resp_time = from(VirtualCacheAccess.latency).longAvg();
cache_access_sla = from(VirtualCacheAccess.*).percent(status == true);
cache_access_cpm = from(VirtualCacheAccess.*).cpm();
cache_access_percentile = from(VirtualCacheAccess.latency).percentile(10);

101
oap-server/server-starter/src/main/resources/ui-initialized-templates/virtual_cache/virtual-cache-root.json Normal file
View File

@ -0,0 +1,101 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
[
{
"id": "Virtual-Cache-Root",
"configuration": {
"children": [
{
"x": 0,
"y": 2,
"w": 24,
"h": 52,
"i": "0",
"type": "Widget",
"widget": {
"title": "Virtual Cache"
},
"graph": {
"type": "ServiceList",
"dashboardName": "Virtual-Cache-Service",
"fontSize": 12,
"showXAxis": false,
"showYAxis": false,
"showGroup": false
},
"metrics": [
"cache_access_resp_time",
"cache_access_sla",
"cache_access_cpm"
],
"metricTypes": [
"readMetricsValues",
"readMetricsValues",
"readMetricsValues"
],
"moved": false,
"metricConfig": [
{
"unit": "ms",
"label": "Access Latency",
"calculation": "average"
},
{
"label": "Successful Access Rate",
"unit": "%",
"calculation": "percentageAvg"
},
{
"label": "Access Traffic",
"unit": "calls / min",
"calculation": "average"
}
]
},
{
"x": 0,
"y": 0,
"w": 24,
"h": 2,
"i": "100",
"type": "Text",
"metricTypes": [
""
],
"metrics": [
""
],
"graph": {
"fontColor": "blue",
"backgroundColor": "white",
"content": "Observe the Virtual Cache which is conjectured by language agent through various plugins.",
"fontSize": 14,
"textAlign": "left",
"url": "https://skywalking.apache.org/docs/main/latest/en/setup/service-agent/virtual-cache/"
},
"moved": false
}
],
"id": "Virtual-Cache-Root",
"layer": "VIRTUAL_CACHE",
"entity": "All",
"name": "Virtual-Cache-Root",
"isRoot": true
}
}
]

622
oap-server/server-starter/src/main/resources/ui-initialized-templates/virtual_cache/virtual-cache-service.json Normal file
View File

@ -0,0 +1,622 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
[
{
"id": "Virtual-Cache-Service",
"configuration": {
"children": [
{
"x": 0,
"y": 0,
"w": 6,
"h": 13,
"i": "1",
"type": "Widget",
"widget": {
"title": "Access Avg Response Time (ms)",
"name": "Access_Avg_Response"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_access_resp_time"
],
"metricTypes": [
"readMetricsValues"
],
"value": "6",
"label": "Access_Avg_Response",
"filters": {
"dataIndex": 17,
"sourceId": "4"
},
"associate": [
{
"widgetId": "2"
},
{
"widgetId": "3"
},
{
"widgetId": "4"
}
]
},
{
"x": 6,
"y": 0,
"w": 6,
"h": 13,
"i": "2",
"type": "Widget",
"widget": {
"title": "Access Successful Rate (%)",
"name": "Access_Successful_Rate"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_access_sla"
],
"metricTypes": [
"readMetricsValues"
],
"metricConfig": [
{
"calculation": "percentage"
}
],
"value": "3",
"label": "Successful_Rate",
"filters": {
"dataIndex": 17,
"sourceId": "4"
},
"associate": [
{
"widgetId": "1"
},
{
"widgetId": "3"
},
{
"widgetId": "4"
}
]
},
{
"x": 12,
"y": 0,
"w": 6,
"h": 13,
"i": "3",
"type": "Widget",
"widget": {
"title": "Access Traffic (calls / min)",
"name": "Access_Traffic"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_access_cpm"
],
"metricTypes": [
"readMetricsValues"
],
"value": "4",
"label": "Cache_Traffic",
"associate": [
{
"widgetId": "1"
},
{
"widgetId": "2"
},
{
"widgetId": "4"
}
]
},
{
"x": 18,
"y": 0,
"w": 6,
"h": 13,
"i": "4",
"type": "Widget",
"widget": {
"title": "Access Latency Percentile (ms)",
"name": "Access_Latency_Percentile"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_access_percentile"
],
"metricTypes": [
"readLabeledMetricsValues"
],
"metricConfig": [
{
"label": "P50, P75, P90, P95, P99",
"labelsIndex": "0,1,2,3,4"
}
],
"value": "5",
"label": "5",
"associate": [
{
"widgetId": "1"
},
{
"widgetId": "2"
},
{
"widgetId": "3"
}
],
"filters": {
"dataIndex": 17,
"sourceId": "4"
}
},
{
"x": 0,
"y": 13,
"w": 6,
"h": 13,
"i": "5",
"type": "Widget",
"widget": {
"title": "Read Avg Response Time (ms)",
"name": "Read_Avg_Response"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_read_resp_time"
],
"metricTypes": [
"readMetricsValues"
],
"value": "6",
"label": "Cache_Read_Avg_Response",
"filters": {
"dataIndex": 17,
"sourceId": "4"
},
"associate": [
{
"widgetId": "6"
},
{
"widgetId": "7"
},
{
"widgetId": "8"
}
]
},
{
"x": 6,
"y": 13,
"w": 6,
"h": 13,
"i": "6",
"type": "Widget",
"widget": {
"title": "Read Successful Rate (%)",
"name": "Read_Successful_Rate"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_read_sla"
],
"metricTypes": [
"readMetricsValues"
],
"metricConfig": [
{
"calculation": "percentage"
}
],
"value": "3",
"label": "Successful_Rate",
"associate": [
{
"widgetId": "5"
},
{
"widgetId": "7"
},
{
"widgetId": "8"
}
],
"filters": {
"dataIndex": 17,
"sourceId": "4"
}
},
{
"x": 12,
"y": 13,
"w": 6,
"h": 13,
"i": "9",
"type": "Widget",
"widget": {
"title": "Write Avg Response Time (ms)",
"name": "Write_Avg_Response"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_write_resp_time"
],
"metricTypes": [
"readMetricsValues"
],
"value": "6",
"label": "Read_Avg_Response",
"filters": {
"dataIndex": 17,
"sourceId": "4"
},
"associate": [
{
"widgetId": "10"
},
{
"widgetId": "11"
},
{
"widgetId": "12"
}
]
},
{
"x": 18,
"y": 13,
"w": 6,
"h": 13,
"i": "10",
"type": "Widget",
"widget": {
"title": "Write Successful Rate (%)",
"name": "Write_Successful_Rate"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_write_sla"
],
"metricTypes": [
"readMetricsValues"
],
"metricConfig": [
{
"calculation": "percentage"
}
],
"value": "3",
"label": "Successful_Rate",
"filters": {
"dataIndex": 17,
"sourceId": "4"
},
"associate": [
{
"widgetId": "9"
},
{
"widgetId": "11"
},
{
"widgetId": "12"
}
]
},
{
"x": 0,
"y": 26,
"w": 6,
"h": 13,
"i": "7",
"type": "Widget",
"widget": {
"title": "Read Traffic (calls / min)",
"name": "Read_Traffic"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_read_cpm"
],
"metricTypes": [
"readMetricsValues"
],
"value": "4",
"label": "Cache_Traffic",
"associate": [
{
"widgetId": "5"
},
{
"widgetId": "6"
},
{
"widgetId": "8"
}
]
},
{
"x": 6,
"y": 26,
"w": 6,
"h": 13,
"i": "8",
"type": "Widget",
"widget": {
"title": "Read Latency Percentile (ms)",
"name": "Read_Latency_Percentile"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_read_percentile"
],
"metricTypes": [
"readLabeledMetricsValues"
],
"metricConfig": [
{
"label": "P50, P75, P90, P95, P99",
"labelsIndex": "0,1,2,3,4"
}
],
"value": "5",
"label": "5",
"associate": [
{
"widgetId": "5"
},
{
"widgetId": "6"
},
{
"widgetId": "7"
}
],
"filters": {
"dataIndex": 17,
"sourceId": "4"
}
},
{
"x": 12,
"y": 26,
"w": 6,
"h": 13,
"i": "11",
"type": "Widget",
"widget": {
"title": "Write Traffic (calls / min)",
"name": "Write_Traffic"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_write_cpm"
],
"metricTypes": [
"readMetricsValues"
],
"value": "4",
"label": "Database_Traffic",
"associate": [
{
"widgetId": "9"
},
{
"widgetId": "10"
},
{
"widgetId": "12"
}
]
},
{
"x": 18,
"y": 26,
"w": 6,
"h": 13,
"i": "12",
"type": "Widget",
"widget": {
"title": "Write Latency Percentile (ms)",
"name": "Write_Latency_Percentile"
},
"graph": {
"type": "Line",
"step": false,
"smooth": false,
"showSymbol": false,
"showXAxis": true,
"showYAxis": true
},
"metrics": [
"cache_write_percentile"
],
"metricTypes": [
"readLabeledMetricsValues"
],
"metricConfig": [
{
"label": "P50, P75, P90, P95, P99",
"labelsIndex": "0,1,2,3,4"
}
],
"value": "5",
"label": "5",
"associate": [
{
"widgetId": "9"
},
{
"widgetId": "10"
},
{
"widgetId": "11"
}
],
"filters": {
"dataIndex": 17,
"sourceId": "4"
}
},
{
"x": 0,
"y": 39,
"w": 12,
"h": 24,
"i": "13",
"type": "Widget",
"widget": {
"title": "Slow Read Command (ms)"
},
"graph": {
"type": "TopList",
"color": "purple"
},
"metrics": [
"top_n_cache_read_command"
],
"metricTypes": [
"readSampledRecords"
],
"value": "2",
"label": "2"
},
{
"x": 12,
"y": 39,
"w": 12,
"h": 24,
"i": "14",
"type": "Widget",
"widget": {
"title": "Slow Write Command (ms)"
},
"graph": {
"type": "TopList",
"color": "purple"
},
"metrics": [
"top_n_cache_write_command"
],
"metricTypes": [
"readSampledRecords"
],
"value": "2",
"label": "2"
}
],
"layer": "VIRTUAL_CACHE",
"entity": "Service",
"name": "Virtual-Cache-Service",
"id": "Virtual-Cache-Service",
"isRoot": false
}
}
]

2
skywalking-ui

@ -1 +1 @@
Subproject commit 26817e9f927dd07fb439f415227bf371f45a9645
Subproject commit 214b34ddfd259dffcd343a6775c7c05c62ae788b

7
test/e2e-v2/cases/meter/docker-compose.yml
View File

@ -19,6 +19,10 @@ services:
oap:
environment:
SW_METER_ANALYZER_ACTIVE_FILES: spring-sleuth,batch-meter
#virtual cache test case
SW_SLOW_CACHE_SLOW_WRITE_THRESHOLD: default:-1
SW_SLOW_CACHE_SLOW_READ_THRESHOLD: default:-1
SW_CORE_TOPN_REPORT_PERIOD: 1 # TopN period
extends:
file: ../../script/docker-compose/base-compose.yml
service: oap
@ -33,12 +37,13 @@ services:
service: provider
environment:
SW_METER_REPORT_INTERVAL: 5
# Activate guava-cache plugin for virtual cache test case
command: ["bash" , "-c" , "cp /skywalking/agent/optional-plugins/apm-guava-cache-plugin*.jar /skywalking/agent/plugins/ && java -jar /services_provider.jar"]
depends_on:
oap:
condition: service_healthy
ports:
- 9090
sender:
image: "eclipse-temurin:8-jre"
volumes:

3
test/e2e-v2/cases/meter/e2e.yaml
View File

@ -40,7 +40,8 @@ trigger:
verify:
retry:
count: 20
# The total time must max than 1min
count: 30
interval: 3s
cases:
- includes:

47
test/e2e-v2/cases/meter/expected/metrics-has-value-percentile.yml Normal file
View File

@ -0,0 +1,47 @@
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
{{- contains . }}
- key: 0
value:
{{- contains .value }}
- key: {{ notEmpty .key }}
value: {{ ge .value 0 }}
{{- end }}
- key: 1
value:
{{- contains .value }}
- key: {{ notEmpty .key }}
value: {{ ge .value 0 }}
{{- end }}
- key: 2
value:
{{- contains .value }}
- key: {{ notEmpty .key }}
value: {{ ge .value 0 }}
{{- end }}
- key: 3
value:
{{- contains .value }}
- key: {{ notEmpty .key }}
value: {{ ge .value 0 }}
{{- end }}
- key: 4
value:
{{- contains .value }}
- key: {{ notEmpty .key }}
value: {{ ge .value 0 }}
{{- end }}
{{- end }}

19
test/e2e-v2/cases/meter/expected/metrics-has-value0.yml Normal file
View File

@ -0,0 +1,19 @@
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
{{- contains . }}
- key: {{ notEmpty .key }}
value: {{ ge .value 0 }}
{{- end }}

29
test/e2e-v2/cases/meter/expected/record-has-value.yml Normal file
View File

@ -0,0 +1,29 @@
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
{{- contains . }}
- key: 0
value:
{{- contains .value }}
- key: name
value: {{ notEmpty .value}}
- key: id
value: ""
- key: value
value: {{ notEmpty .value}}
- key: refid
value:
{{- end }}
{{- end }}

16
test/e2e-v2/cases/meter/expected/service.yml
View File

@ -21,4 +21,20 @@
normal: true
layers:
- GENERAL
- id: {{ b64enc "localhost:-1" }}.0
name: localhost:-1
group: ""
shortname: localhost:-1
normal: false
layers:
- VIRTUAL_DATABASE
- id: {{ b64enc "GuavaCache-local" }}.0
name: GuavaCache-local
group: ""
shortname: GuavaCache-local
normal: false
layers:
- VIRTUAL_CACHE
{{- end }}

19
test/e2e-v2/cases/meter/meter-cases.yaml
View File

@ -31,4 +31,21 @@
curl -s -XPOST http://${sender_host}:${sender_9093}/sendBatchMetrics > /dev/null;
sleep 10;
swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name=batch_test --instance-name=test-instance --service-name=test-service |yq e 'to_entries' -
expected: expected/metrics-has-value.yml
expected: expected/metrics-has-value.yml
# virtual cache
- query: |
swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name=cache_read_resp_time --service-id=R3VhdmFDYWNoZS1sb2NhbA==.0 | yq e 'to_entries' -
expected: expected/metrics-has-value0.yml
- query: |
swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name=cache_write_sla --service-id=R3VhdmFDYWNoZS1sb2NhbA==.0 | yq e 'to_entries' -
expected: expected/metrics-has-value.yml
- query: |
swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name=cache_access_cpm --service-id=R3VhdmFDYWNoZS1sb2NhbA==.0 | yq e 'to_entries' -
expected: expected/metrics-has-value.yml
- query: |
swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics sampled-record --name=top_n_cache_write_command --service-id=R3VhdmFDYWNoZS1sb2NhbA==.0 | yq e 'to_entries | with(.[] ; .value=(.value | to_entries))' -
expected: expected/record-has-value.yml
# virtual database
- query: |
swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql metrics linear --name=database_access_resp_time --service-id=bG9jYWxob3N0Oi0x.0 | yq e 'to_entries' -
expected: expected/metrics-has-value0.yml

2
test/e2e-v2/cases/storage/storage-cases.yaml
View File

@ -47,7 +47,7 @@ cases:
- query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls
expected: expected/traces-list.yml
# negative tags search: relationship should be logical AND instead of logical OR
- query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls --tags http.method=POST,http.status_code=200
- query: swctl --display yaml --base-url=http://${oap_host}:${oap_12800}/graphql trace ls --tags http.method=POST,http.status_code=201
expected: expected/empty-traces-list.yml
# trace detail
- query: |

6
test/e2e-v2/java-test-service/e2e-service-provider/pom.xml
View File

@ -105,6 +105,12 @@
<artifactId>apm-toolkit-trace</artifactId>
<version>${sw.version}</version>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>23.0</version>
</dependency>
</dependencies>
<build>

27
test/e2e-v2/java-test-service/e2e-service-provider/src/main/java/org/apache/skywalking/e2e/controller/UserController.java
View File

@ -18,10 +18,13 @@
package org.apache.skywalking.e2e.controller;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import lombok.RequiredArgsConstructor;
import org.apache.skywalking.apm.toolkit.trace.TraceContext;
import org.apache.skywalking.e2e.User;
import org.apache.skywalking.e2e.UserRepo;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
@ -29,6 +32,9 @@ import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import org.slf4j.LoggerFactory;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
@RestController
@ -37,6 +43,10 @@ import java.util.Random;
public class UserController {
private static final org.slf4j.Logger LOGBACK_LOGGER = LoggerFactory.getLogger(UserController.class);
private final Cache<String, String> guavaCache = CacheBuilder.newBuilder()
.concurrencyLevel(Runtime.getRuntime().availableProcessors())
.build();
private final UserRepo userRepo;
private final int sleepMin = 500;
private final int sleepMax = 1000;
@ -51,6 +61,8 @@ public class UserController {
@PostMapping("/users")
public User createAuthor(@RequestBody final User user) throws InterruptedException {
Thread.sleep(randomSleepLong(sleepMin, sleepMax));
//virtual cache test case
testCacheService();
return userRepo.save(user);
}
@ -68,4 +80,19 @@ public class UserController {
int randomNumber = rand.nextInt((max - min) + 1) + min;
return randomNumber;
}
private void testCacheService() {
String userInfo = guavaCache.getIfPresent("user_1");
if (!StringUtils.hasLength(userInfo)) {
guavaCache.put("user_1", "name:John,address:earth");
}
Map<String, String> users = new HashMap<>();
users.put("user_2", "name:Tom,address:earth");
users.put("user_3", "name:Jack,address:earth");
guavaCache.putAll(users);
guavaCache.getAllPresent(Arrays.asList("user_2", "user_3"));
guavaCache.invalidate("user_1");
guavaCache.invalidate("user_2");
guavaCache.invalidateAll();
}
}

2
test/e2e-v2/script/env
View File

@ -13,7 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
SW_AGENT_JAVA_COMMIT=b83865f6cd9477f12c8ae8bc3379c31d7d105072
SW_AGENT_JAVA_COMMIT=3f88d735ba2bfd1196aff946502447d4b14450c8
SW_AGENT_SATELLITE_COMMIT=ea27a3f4e126a24775fe12e2aa2695bcb23d99c3
SW_AGENT_NGINX_LUA_COMMIT=c3cee4841798a147d83b96a10914d4ac0e11d0aa
SW_AGENT_NODEJS_COMMIT=2e7560518aff846befd4d6bc815fe5e38c704a11