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

Reduce the count of process index and adding time range when query process index (#12681)

This commit is contained in:
mrproliu 2024-10-08 21:10:37 +07:00 committed by GitHub
parent 180ca71375
commit 44e3c34c9c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 30 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -4,6 +4,7 @@
* Skip processing OTLP metrics data points with flag `FLAG_NO_RECORDED_VALUE`, which causes exceptional result.
* Add self observability metrics for GraphQL query, `graphql_query_latency`.
* Reduce the count of process index and adding time range when query process index.
#### UI

6
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/process/ProcessTraffic.java
View File

@ -139,6 +139,12 @@ public class ProcessTraffic extends Metrics {
if (StringUtil.isNotEmpty(processTraffic.getLabelsJson())) {
this.labelsJson = processTraffic.getLabelsJson();
}
/**
* Keep the time bucket as the same time inserted.
*/
if (this.getTimeBucket() > metrics.getTimeBucket()) {
this.setTimeBucket(metrics.getTimeBucket());
}
return true;
}

2
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IMetadataQueryDAO.java
View File

@ -78,7 +78,7 @@ public interface IMetadataQueryDAO extends DAO {
/**
* @param agentId the agent id of the process.
*/
List<Process> listProcesses(final String agentId) throws IOException;
List<Process> listProcesses(final String agentId, long startPingTimeBucket, long endPingTimeBucket) throws IOException;
/**
* @param serviceId the service id of the process

13
oap-server/server-receiver-plugin/skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/provider/handler/EBPFProfilingServiceHandler.java
View File

@ -33,6 +33,8 @@ import org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingStackMet
import org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingTaskMetadata;
import org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingTaskQuery;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.analysis.DownSampling;
import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
import org.apache.skywalking.oap.server.core.command.CommandService;
import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingStackType;
import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
@ -55,6 +57,7 @@ import org.apache.skywalking.oap.server.network.trace.component.command.EBPFProf
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@ -70,6 +73,10 @@ public class EBPFProfilingServiceHandler extends EBPFProfilingServiceGrpc.EBPFPr
private static final Gson GSON = new Gson();
public static final List<EBPFProfilingStackType> COMMON_STACK_TYPE_ORDER = Arrays.asList(
EBPFProfilingStackType.KERNEL_SPACE, EBPFProfilingStackType.USER_SPACE);
/**
* When querying profiling tasks, processes from the last few minutes would be queried.
*/
public static final int QUERY_TASK_PROCESSES_RANGE_MINUTES = 5;
private IEBPFProfilingTaskDAO taskDAO;
private IMetadataQueryDAO metadataQueryDAO;
@ -88,8 +95,12 @@ public class EBPFProfilingServiceHandler extends EBPFProfilingServiceGrpc.EBPFPr
String agentId = request.getRoverInstanceId();
final long latestUpdateTime = request.getLatestUpdateTime();
try {
final Calendar now = Calendar.getInstance();
long endTimeBucket = TimeBucket.getTimeBucket(now.getTimeInMillis(), DownSampling.Minute);
now.add(Calendar.MINUTE, -QUERY_TASK_PROCESSES_RANGE_MINUTES);
long startTimeBucket = TimeBucket.getTimeBucket(now.getTimeInMillis(), DownSampling.Minute);
// find exists process from agent
final List<Process> processes = metadataQueryDAO.listProcesses(agentId);
final List<Process> processes = metadataQueryDAO.listProcesses(agentId, startTimeBucket, endTimeBucket);
if (CollectionUtils.isEmpty(processes)) {
responseObserver.onNext(Commands.newBuilder().build());
responseObserver.onCompleted();

4
oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java
View File

@ -262,7 +262,7 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
}
@Override
public List<Process> listProcesses(String agentId) throws IOException {
public List<Process> listProcesses(String agentId, long startPingTimeBucket, long endPingTimeBucket) throws IOException {
MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(ProcessTraffic.INDEX_NAME, DownSampling.Minute);
MeasureQueryResponse resp = query(schema,
PROCESS_TRAFFIC_TAGS,
@ -271,6 +271,8 @@ public class BanyanDBMetadataQueryDAO extends AbstractBanyanDBDAO implements IMe
@Override
protected void apply(MeasureQuery query) {
query.and(eq(ProcessTraffic.AGENT_ID, agentId));
query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, startPingTimeBucket));
query.and(lte(ProcessTraffic.LAST_PING_TIME_BUCKET, endPingTimeBucket));
query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
}
});

5
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetadataQueryEsDAO.java
View File

@ -311,7 +311,7 @@ public class MetadataQueryEsDAO extends EsDAO implements IMetadataQueryDAO {
}
@Override
public List<Process> listProcesses(String agentId) {
public List<Process> listProcesses(String agentId, long startPingTimeBucket, long endPingTimeBucket) {
final String index =
IndexController.LogicIndicesRegister.getPhysicalTableName(ProcessTraffic.INDEX_NAME);
@ -320,7 +320,8 @@ public class MetadataQueryEsDAO extends EsDAO implements IMetadataQueryDAO {
query.must(Query.term(IndexController.LogicIndicesRegister.METRIC_TABLE_NAME, ProcessTraffic.INDEX_NAME));
}
final SearchBuilder search = Search.builder().query(query).size(queryMaxSize);
appendProcessWhereQuery(query, null, null, agentId, null, 0, 0, false);
appendProcessWhereQuery(query, null, null, agentId, null,
startPingTimeBucket, endPingTimeBucket, false);
final var scroller = ElasticSearchScroller
.<Process>builder()

25
oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCMetadataQueryDAO.java
View File

@ -248,16 +248,7 @@ public class JDBCMetadataQueryDAO implements IMetadataQueryDAO {
@Override
@SneakyThrows
public List<Process> listProcesses(String serviceId, ProfilingSupportStatus supportStatus, long lastPingStartTimeBucket, long lastPingEndTimeBucket) {
List<String> tables;
if (lastPingStartTimeBucket > 0 && lastPingEndTimeBucket > 0) {
tables = tableHelper.getTablesForRead(
ProcessTraffic.INDEX_NAME,
lastPingStartTimeBucket,
lastPingEndTimeBucket
);
} else {
tables = tableHelper.getTablesWithinTTL(ProcessTraffic.INDEX_NAME);
}
List<String> tables = tableHelper.getTablesWithinTTL(ProcessTraffic.INDEX_NAME);
final var results = new ArrayList<Process>();
for (String table : tables) {
@ -297,11 +288,7 @@ public class JDBCMetadataQueryDAO implements IMetadataQueryDAO {
@Override
@SneakyThrows
public List<Process> listProcesses(String serviceInstanceId, Duration duration, boolean includeVirtual) {
final var tables = tableHelper.getTablesForRead(
ProcessTraffic.INDEX_NAME,
duration.getStartTimeBucket(),
duration.getEndTimeBucket()
);
final List<String> tables = tableHelper.getTablesWithinTTL(ProcessTraffic.INDEX_NAME);
final var results = new ArrayList<Process>();
for (String table : tables) {
@ -337,7 +324,7 @@ public class JDBCMetadataQueryDAO implements IMetadataQueryDAO {
@Override
@SneakyThrows
public List<Process> listProcesses(String agentId) {
public List<Process> listProcesses(String agentId, long startPingTimeBucket, long endPingTimeBucket) {
final var tables = tableHelper.getTablesWithinTTL(ProcessTraffic.INDEX_NAME);
final var results = new ArrayList<Process>();
@ -364,11 +351,7 @@ public class JDBCMetadataQueryDAO implements IMetadataQueryDAO {
@Override
@SneakyThrows
public long getProcessCount(String serviceId, ProfilingSupportStatus profilingSupportStatus, long lastPingStartTimeBucket, long lastPingEndTimeBucket) {
final var tables = tableHelper.getTablesForRead(
ProcessTraffic.INDEX_NAME,
lastPingStartTimeBucket,
lastPingEndTimeBucket
);
final var tables = tableHelper.getTablesWithinTTL(ProcessTraffic.INDEX_NAME);
long total = 0;
for (String table : tables) {