You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by GitBox <gi...@apache.org> on 2022/03/17 07:05:05 UTC
[GitHub] [skywalking] mrproliu opened a new pull request #8696: Implement the eBPF profiling query and data collect protocol
mrproliu opened a new pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696
* Implement the eBPF profiling query and data collect protocol
* Migrate existing profile classes to the profiling package
* I will submit a new PR after rover side finished for E2E testing
- [ ] If this pull request closes/resolves/fixes an existing issue, replace the issue number. Closes #<issue number>.
- [x] Update the [`CHANGES` log](https://github.com/apache/skywalking/blob/master/CHANGES.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830569111
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * eBPF Profiling Analyzer working on data query and combine them for generate the Flame Graph.
+ */
+@Slf4j
+public class EBPFProfilingAnalyzer {
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_DATA_DURATION = TimeUnit.MINUTES.toMillis(2);
+ private static final ExecutorService FETCH_DATA_THREAD_POOL = Executors.newCachedThreadPool();
Review comment:
Yes, may we need two global thread pools for doing that. One for the profiling data analysis, the other one for the wait for the data query(works for the circle break the query when timeout). WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#issuecomment-1072955710
> How many threads for these op?
Do you mean Profiling Data Merging benchmark? I have declared the annotation `@Thread(4)` in the benchmark, so I think there using 4 threads.
> Should we set up a recommended max range for UI?
I think maybe two hours should be the max analysis duration.
1. Long-time range means that a lot of data need be obtained from the storage, which will take a longer time in query and cause a bad user experience
2. Long-time analysis is meaningless, usually a few minutes - one-hour duration is enough to explain the problem.
What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830555668
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
Review comment:
Add a new configuration item for the query timeout. default `30s`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng merged pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng merged pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830498906
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
Review comment:
Check is fine, you should add else.
Just a suggestion, not a block.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830461500
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
Review comment:
Make sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] kezhenxu94 commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
kezhenxu94 commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830482066
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * eBPF Profiling Analyzer working on data query and combine them for generate the Flame Graph.
+ */
+@Slf4j
+public class EBPFProfilingAnalyzer {
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+ private long maxAnalyzeTimeRangeInMillisecond;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager, int maxDurationOfAnalysisInMinute) {
+ this.moduleManager = moduleManager;
+ this.maxAnalyzeTimeRangeInMillisecond = TimeUnit.MINUTES.toMillis(maxDurationOfAnalysisInMinute);
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ String timeRangeValidate = validateIsOutOfTimeRangeLimit(ranges);
+ if (StringUtil.isNotEmpty(timeRangeValidate)) {
+ analyzation.setTip(timeRangeValidate);
+ return analyzation;
+ }
+
+ // query data
+ final Stream<EBPFProfilingStack> stackStream = buildTimeRanges(ranges).parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ log.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ log.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct();
+
+ // analyze tree
+ generateTrees(analyzation, stackStream);
+
+ return analyzation;
+ }
+
+ private String validateIsOutOfTimeRangeLimit(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ return "please provide time ranges";
+ }
+
+ long totalDuration = 0;
+ for (EBPFProfilingAnalyzeTimeRange timeRange : timeRanges) {
+ final long duration = timeRange.getEnd() - timeRange.getStart();
+ if (duration <= 0) {
+ return "please validate the time duration data";
+ }
+ totalDuration += duration;
+ }
+
+ if (totalDuration > maxAnalyzeTimeRangeInMillisecond) {
+ return "time range is out of " +
+ TimeUnit.MILLISECONDS.toMinutes(this.maxAnalyzeTimeRangeInMillisecond) + " minute";
+ }
+ return null;
+ }
+
+ public void generateTrees(EBPFProfilingAnalyzation analyzation, Stream<EBPFProfilingStack> stackStream) {
+ Collection<EBPFProfilingTree> stackTrees = stackStream
+ // stack list cannot be empty
+ .filter(s -> CollectionUtils.isNotEmpty(s.getSymbols()))
+ // analyze the symbol and combine as trees
+ .collect(Collectors.groupingBy(s -> s.getSymbols()
+ .get(0), ANALYZE_COLLECTOR)).values();
+
+ analyzation.getTrees().addAll(stackTrees);
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Split time ranges to insure the start time and end time is small then {@link #FETCH_FETCH_DURATION}
+ */
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ protected IEBPFProfilingDataDAO getDataDAO() {
+ if (dataDAO == null) {
+ dataDAO = moduleManager.find(StorageModule.NAME)
+ .provider()
Review comment:
Codes are not formatted
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * eBPF Profiling Analyzer working on data query and combine them for generate the Flame Graph.
+ */
+@Slf4j
+public class EBPFProfilingAnalyzer {
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+ private long maxAnalyzeTimeRangeInMillisecond;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager, int maxDurationOfAnalysisInMinute) {
+ this.moduleManager = moduleManager;
+ this.maxAnalyzeTimeRangeInMillisecond = TimeUnit.MINUTES.toMillis(maxDurationOfAnalysisInMinute);
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ String timeRangeValidate = validateIsOutOfTimeRangeLimit(ranges);
+ if (StringUtil.isNotEmpty(timeRangeValidate)) {
+ analyzation.setTip(timeRangeValidate);
+ return analyzation;
+ }
+
+ // query data
+ final Stream<EBPFProfilingStack> stackStream = buildTimeRanges(ranges).parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ log.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ log.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct();
+
+ // analyze tree
+ generateTrees(analyzation, stackStream);
+
+ return analyzation;
+ }
+
+ private String validateIsOutOfTimeRangeLimit(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ return "please provide time ranges";
+ }
+
+ long totalDuration = 0;
+ for (EBPFProfilingAnalyzeTimeRange timeRange : timeRanges) {
+ final long duration = timeRange.getEnd() - timeRange.getStart();
+ if (duration <= 0) {
+ return "please validate the time duration data";
+ }
+ totalDuration += duration;
+ }
+
+ if (totalDuration > maxAnalyzeTimeRangeInMillisecond) {
+ return "time range is out of " +
+ TimeUnit.MILLISECONDS.toMinutes(this.maxAnalyzeTimeRangeInMillisecond) + " minute";
+ }
+ return null;
+ }
+
+ public void generateTrees(EBPFProfilingAnalyzation analyzation, Stream<EBPFProfilingStack> stackStream) {
+ Collection<EBPFProfilingTree> stackTrees = stackStream
+ // stack list cannot be empty
+ .filter(s -> CollectionUtils.isNotEmpty(s.getSymbols()))
+ // analyze the symbol and combine as trees
+ .collect(Collectors.groupingBy(s -> s.getSymbols()
+ .get(0), ANALYZE_COLLECTOR)).values();
+
+ analyzation.getTrees().addAll(stackTrees);
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Split time ranges to insure the start time and end time is small then {@link #FETCH_FETCH_DURATION}
+ */
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ protected IEBPFProfilingDataDAO getDataDAO() {
+ if (dataDAO == null) {
+ dataDAO = moduleManager.find(StorageModule.NAME)
+ .provider()
+ .getService(IEBPFProfilingDataDAO.class);
+ }
+ return dataDAO;
+ }
+
+ private static class TimeRange {
+ private long minTime;
Review comment:
These fields can be `final` and use `@RequiredArgsConstructor`
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingStackNode.java
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import com.google.common.base.Objects;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingStackElement;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.function.Consumer;
+
+/**
+ * EBPF profiling data analyze intermediate state data
+ */
+public class EBPFProfilingStackNode {
+
+ private EBPFProfilingStack.Symbol codeSignature;
+ private List<EBPFProfilingStackNode> children;
+ private long dumpCount;
+
+ /**
+ * create new empty, un-init node
+ */
+ public static EBPFProfilingStackNode newNode() {
+ EBPFProfilingStackNode emptyNode = new EBPFProfilingStackNode();
+ emptyNode.children = new ArrayList<>();
+ return emptyNode;
+ }
+
+ /**
+ * accumulate {@link EBPFProfilingStack} to this tree, it will invoke on the tree root node
+ */
+ public void accumulateFrom(EBPFProfilingStack stack) {
+ List<EBPFProfilingStack.Symbol> stackList = stack.getSymbols();
+ if (codeSignature == null) {
+ codeSignature = stackList.get(0);
+ }
+ // add detected stack
+ this.detectedBy(stack);
+
+ // handle stack children
+ EBPFProfilingStackNode parent = this;
+ for (int depth = 1; depth < stackList.size(); depth++) {
+ EBPFProfilingStack.Symbol elementCodeSignature = stackList.get(depth);
+
+ // find same code signature children
+ EBPFProfilingStackNode childElement = null;
+ for (EBPFProfilingStackNode child : parent.children) {
+ if (Objects.equal(child.codeSignature, elementCodeSignature)) {
+ childElement = child;
+ break;
+ }
+ }
+
+ if (childElement != null) {
+ // add detected stack
+ childElement.detectedBy(stack);
+ parent = childElement;
+ } else {
+ // add children
+ EBPFProfilingStackNode childNode = newNode();
+ childNode.codeSignature = elementCodeSignature;
+ childNode.detectedBy(stack);
+
+ parent.children.add(childNode);
+ parent = childNode;
+ }
+ }
+ }
+
+ /**
+ * combine from other {@link EBPFProfilingStackNode}
+ */
+ public EBPFProfilingStackNode combine(EBPFProfilingStackNode node) {
+ // combine this node
+ this.combineDetectedStacks(node);
+
+ // merge tree using LDR to traversal tree node
+ // using stack to avoid recursion
+ // merge key.children <- value.children
+ LinkedList<Pair<EBPFProfilingStackNode, EBPFProfilingStackNode>> stack = new LinkedList<>();
+ stack.add(new Pair<>(this, node));
+ while (!stack.isEmpty()) {
+ Pair<EBPFProfilingStackNode, EBPFProfilingStackNode> needCombineNode = stack.pop();
+
+ // merge value children to key
+ // add to stack if need to keep traversal
+ combineChildrenNodes(needCombineNode.key, needCombineNode.value, stack::add);
+ }
+
+ return this;
+ }
+
+ /**
+ * merge all children nodes to appoint node
+ */
+ private void combineChildrenNodes(EBPFProfilingStackNode targetNode, EBPFProfilingStackNode beingMergedNode,
+ Consumer<Pair<EBPFProfilingStackNode, EBPFProfilingStackNode>> continueChildrenMerging) {
+ if (beingMergedNode.children.isEmpty()) {
+ return;
+ }
+
+ for (EBPFProfilingStackNode childrenNode : targetNode.children) {
+ // find node from being merged node children
+ for (ListIterator<EBPFProfilingStackNode> it = beingMergedNode.children.listIterator(); it.hasNext(); ) {
+ EBPFProfilingStackNode node = it.next();
+ if (node != null && node.matches(childrenNode)) {
+ childrenNode.combineDetectedStacks(node);
+ continueChildrenMerging.accept(new Pair<>(childrenNode, node));
+
+ it.set(null);
+ break;
+ }
+ }
+ }
+
+ for (EBPFProfilingStackNode node : beingMergedNode.children) {
+ if (node != null) {
+ targetNode.children.add(node);
+ }
+ }
+ }
+
+ /**
+ * build GraphQL result, calculate duration and count data using parallels
+ */
+ public EBPFProfilingTree buildAnalyzeResult() {
+ // all nodes add to single-level list (such as flat), work for parallel calculating
+ LinkedList<Pair<EBPFProfilingStackElement, EBPFProfilingStackNode>> nodeMapping = new LinkedList<>();
+ int idGenerator = 1;
+
+ EBPFProfilingStackElement root = buildElement(idGenerator++);
+ nodeMapping.add(new Pair<>(root, this));
+
+ // same with combine logic
+ LinkedList<Pair<EBPFProfilingStackElement, EBPFProfilingStackNode>> stack = new LinkedList<>();
+ stack.add(new Pair<>(root, this));
+ while (!stack.isEmpty()) {
+ Pair<EBPFProfilingStackElement, EBPFProfilingStackNode> mergingPair = stack.pop();
+ EBPFProfilingStackElement respElement = mergingPair.key;
+
+ // generate children node and add to stack and all node mapping
+ for (EBPFProfilingStackNode children : mergingPair.value.children) {
+ EBPFProfilingStackElement element = children.buildElement(idGenerator++);
+ element.setParentId(respElement.getId());
+
+ Pair<EBPFProfilingStackElement, EBPFProfilingStackNode> pair = new Pair<>(element, children);
+ stack.add(pair);
+ nodeMapping.add(pair);
+ }
+ }
+
+ EBPFProfilingTree tree = new EBPFProfilingTree();
+ nodeMapping.forEach(n -> tree.getElements().add(n.key));
+
+ return tree;
+ }
+
+ private void detectedBy(EBPFProfilingStack stack) {
+ this.dumpCount += stack.getDumpCount();
+ }
+
+ private void combineDetectedStacks(EBPFProfilingStackNode node) {
+ this.dumpCount += node.dumpCount;
+ }
+
+ private EBPFProfilingStackElement buildElement(int id) {
+ EBPFProfilingStackElement element = new EBPFProfilingStackElement();
+ element.setId(id);
+ element.setSymbol(this.codeSignature.getName());
+ element.setStackType(this.codeSignature.getStackType());
+ element.setDumpCount(this.dumpCount);
+ return element;
+ }
+
+ private boolean matches(EBPFProfilingStackNode node) {
+ return Objects.equal(this.codeSignature, node.codeSignature);
+ }
+
+ private static class Pair<K, V> {
Review comment:
Consider using existing one: `io.vavr.Tuple2`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829174312
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
+ if (CollectionUtils.isEmpty(stacks)) {
+ return null;
+ }
+
+ // using parallel stream
+ Map<EBPFProfilingStack.Symbol, EBPFProfilingTree> stackTrees = stacks.parallelStream()
+ // stack list cannot be empty
+ .filter(s -> CollectionUtils.isNotEmpty(s.getSymbols()))
+ .collect(Collectors.groupingBy(s -> s.getSymbols()
+ .get(0), ANALYZE_COLLECTOR));
+
+ return new ArrayList<>(stackTrees.values());
+ }
+
+ protected IEBPFProfilingDataDAO getDataDAO() {
+ if (dataDAO == null) {
+ dataDAO = moduleManager.find(StorageModule.NAME)
+ .provider()
+ .getService(IEBPFProfilingDataDAO.class);
+ }
+ return dataDAO;
+ }
+
+ private static class TimeRange {
Review comment:
What does TimeRange mean?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#issuecomment-1072974070
Then let's set a configuration to limit range, how about default in 10 mins(max)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830478752
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingScheduleEsDAO.java
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.RangeQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingScheduleRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingSchedule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingScheduleDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingScheduleEsDAO extends EsDAO implements IEBPFProfilingScheduleDAO {
+ private final int scheduleTaskSize;
+
+ public EBPFProfilingScheduleEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.scheduleTaskSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingSchedule> querySchedules(String taskId, long startTimeBucket, long endTimeBucket) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingScheduleRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (StringUtil.isNotEmpty(taskId)) {
+ query.must(Query.term(EBPFProfilingScheduleRecord.TASK_ID, taskId));
+ }
+ RangeQueryBuilder timeRangeQuery = null;
+ Supplier<RangeQueryBuilder> timeRangeQuerySupplier = () -> Query.range(EBPFProfilingScheduleRecord.START_TIME);
+ if (startTimeBucket > 0) {
+ timeRangeQuery = Optional.ofNullable(timeRangeQuery).orElseGet(timeRangeQuerySupplier)
+ .gte(TimeBucket.getTimestamp(startTimeBucket));
+ }
+ if (endTimeBucket > 0) {
Review comment:
Same as previous review, `startTimeBucket` and `endTimeBucket` should not be null in any case, please throw an illegalArgumentException if so. Recheck for all DAO implementations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829177816
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
Review comment:
Also, we need some benchmarks for this merging, as you provide a list of durations, so we could face a larger merging.
At the same time, we need a `circuit break` to avoid timeout. We should have the capability to provide a half-finished profiling tree with a message of `circuit break`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830569111
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * eBPF Profiling Analyzer working on data query and combine them for generate the Flame Graph.
+ */
+@Slf4j
+public class EBPFProfilingAnalyzer {
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_DATA_DURATION = TimeUnit.MINUTES.toMillis(2);
+ private static final ExecutorService FETCH_DATA_THREAD_POOL = Executors.newCachedThreadPool();
Review comment:
Yes, may we need two global thread pools for doing that. One for the profiling data analysis, the other one for the wait for the data query(works for the circle break the query when timeout). Also, these two pool sizes are the same. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830563562
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * eBPF Profiling Analyzer working on data query and combine them for generate the Flame Graph.
+ */
+@Slf4j
+public class EBPFProfilingAnalyzer {
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_DATA_DURATION = TimeUnit.MINUTES.toMillis(2);
+ private static final ExecutorService FETCH_DATA_THREAD_POOL = Executors.newCachedThreadPool();
Review comment:
The maximumPoolSize of the thread pool should be fixed and configurable. Default, it is Int.MAX, which could easily crash OAP server. We need to reject or timeout when face too many analysis requests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829940649
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.PROCESS_FIND_TYPE, finder.getFinderType().value()));
+ }
+ if (finder != null && finder.getServiceId() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.SERVICE_ID, finder.getServiceId()));
+ }
+ if (finder != null && finder.getInstanceId() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.INSTANCE_ID, finder.getInstanceId()));
+ }
+ if (finder != null && finder.getProcessIdList() != null) {
+ query.must(Query.terms(EBPFProfilingTaskRecord.PROCESS_ID, finder.getProcessIdList()));
+ }
+ if (targetType != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.TARGET_TYPE, targetType.value()));
+ }
+ if (taskStartTime > 0) {
+ query.must(Query.range(EBPFProfilingTaskRecord.START_TIME).gte(taskStartTime));
+ }
+ if (latestUpdateTime > 0) {
+ query.must(Query.range(EBPFProfilingTaskRecord.LAST_UPDATE_TIME).gt(latestUpdateTime));
+ }
+
+ final SearchBuilder search = Search.builder().query(query)
+ .sort(EBPFProfilingTaskRecord.CREATE_TIME, Sort.Order.DESC)
+ .size(taskMaxSize);
+
+ final SearchResponse response = getClient().search(index, search.build());
Review comment:
`EBPFProfilingTask` means the task which user created. The `EBPFProfilingData` is the profiling reported data, this one in the ES storage have using the scroll to query.
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
+ if (CollectionUtils.isEmpty(stacks)) {
+ return null;
+ }
+
+ // using parallel stream
+ Map<EBPFProfilingStack.Symbol, EBPFProfilingTree> stackTrees = stacks.parallelStream()
+ // stack list cannot be empty
+ .filter(s -> CollectionUtils.isNotEmpty(s.getSymbols()))
+ .collect(Collectors.groupingBy(s -> s.getSymbols()
+ .get(0), ANALYZE_COLLECTOR));
+
+ return new ArrayList<>(stackTrees.values());
+ }
+
+ protected IEBPFProfilingDataDAO getDataDAO() {
+ if (dataDAO == null) {
+ dataDAO = moduleManager.find(StorageModule.NAME)
+ .provider()
+ .getService(IEBPFProfilingDataDAO.class);
+ }
+ return dataDAO;
+ }
+
+ private static class TimeRange {
Review comment:
The `TimeRange` using to split the query, if the analysis time range is bigger than 2 minutes, then split them.
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/storage/EBPFProfilingTaskRecord.java
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.profiling.ebpf.storage;
+
+import com.google.common.base.Charsets;
+import com.google.common.hash.Hashing;
+import lombok.Data;
+import org.apache.skywalking.oap.server.core.analysis.Stream;
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
+import org.apache.skywalking.oap.server.core.analysis.worker.NoneStreamProcessor;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.source.ScopeDeclaration;
+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 static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.EBPF_PROFILING_TASK;
+
+@Data
+@ScopeDeclaration(id = EBPF_PROFILING_TASK, name = "EBPFProfilingTask")
+@Stream(name = EBPFProfilingTaskRecord.INDEX_NAME, scopeId = EBPF_PROFILING_TASK,
+ builder = EBPFProfilingTaskRecord.Builder.class, processor = NoneStreamProcessor.class)
+public class EBPFProfilingTaskRecord extends NoneStream {
Review comment:
Yes, now these three concepts are `EBPFProfilingTaskRecord`, `EBPFProfilingScheduleTraffic`, and `EBPFProfilingDataRecord`.
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
Review comment:
being added in the core level check before the storage query.
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
Review comment:
I have added a [benchmark](https://github.com/apache/skywalking/pull/8696/files#diff-15d820525f50abd439c83f34d6afd86a01134ed964142dc0fce907b7c1d4cc33) working on checking the analysis time without query data. I think profiling data analysis is not a big problem, we should add the `circuit break` when the query timeout. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830461537
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
+ if (CollectionUtils.isEmpty(stacks)) {
+ return null;
+ }
+
+ // using parallel stream
+ Map<EBPFProfilingStack.Symbol, EBPFProfilingTree> stackTrees = stacks.parallelStream()
+ // stack list cannot be empty
+ .filter(s -> CollectionUtils.isNotEmpty(s.getSymbols()))
+ .collect(Collectors.groupingBy(s -> s.getSymbols()
+ .get(0), ANALYZE_COLLECTOR));
+
+ return new ArrayList<>(stackTrees.values());
+ }
+
+ protected IEBPFProfilingDataDAO getDataDAO() {
+ if (dataDAO == null) {
+ dataDAO = moduleManager.find(StorageModule.NAME)
+ .provider()
+ .getService(IEBPFProfilingDataDAO.class);
+ }
+ return dataDAO;
+ }
+
+ private static class TimeRange {
Review comment:
Then please add comments. A lot of codes should be explained in comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#issuecomment-1072951779
How many threads for these op? Should we set up a recommended max range for UI?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830564393
##########
File path: docs/en/setup/backend/configuration-vocabulary.md
##########
@@ -46,6 +46,8 @@ core|default|role|Option values: `Mixed/Receiver/Aggregator`. **Receiver** mode
| - | - | maxSizeOfAnalyzeProfileSnapshot| The maximum number of snapshots analyzed by the OAP. | - | 12000 |
| - | - | prepareThreads| The number of threads used to prepare metrics data to the storage. | SW_CORE_PREPARE_THREADS | 2 |
| - | - | enableEndpointNameGroupingByOpenapi | Automatically groups endpoints by the given OpenAPI definitions. | SW_CORE_ENABLE_ENDPOINT_NAME_GROUPING_BY_OPAENAPI | true |
+| - | - | maxDurationOfAnalyzeEBPFProfiling| The maximum duration(in minute) of analyze the eBPF profiling data. | - | 10 |
+| - | - | maxDurationOfQueryEBPFProfilingData| The maximum timeout duration(in second) of query the eBPF profiling data when analysis. | - | 30 |
Review comment:
`maximum` and `timeout` is duplicated :) Choose one. There is not a minimal timeout, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829182955
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
Review comment:
Don't check this, we should throw NPE or IllegalArgumentException, rather than doing a full table scan.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829942294
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
+ if (CollectionUtils.isEmpty(stacks)) {
+ return null;
+ }
+
+ // using parallel stream
+ Map<EBPFProfilingStack.Symbol, EBPFProfilingTree> stackTrees = stacks.parallelStream()
+ // stack list cannot be empty
+ .filter(s -> CollectionUtils.isNotEmpty(s.getSymbols()))
+ .collect(Collectors.groupingBy(s -> s.getSymbols()
+ .get(0), ANALYZE_COLLECTOR));
+
+ return new ArrayList<>(stackTrees.values());
+ }
+
+ protected IEBPFProfilingDataDAO getDataDAO() {
+ if (dataDAO == null) {
+ dataDAO = moduleManager.find(StorageModule.NAME)
+ .provider()
+ .getService(IEBPFProfilingDataDAO.class);
+ }
+ return dataDAO;
+ }
+
+ private static class TimeRange {
Review comment:
The `TimeRange` using to split the query, if the analysis time range is bigger than 2 minutes, then split them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#issuecomment-1072978302
> Then let's set a configuration to limit range, how about default in 10 mins(max)?
Added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830568332
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * eBPF Profiling Analyzer working on data query and combine them for generate the Flame Graph.
+ */
+@Slf4j
+public class EBPFProfilingAnalyzer {
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_DATA_DURATION = TimeUnit.MINUTES.toMillis(2);
+ private static final ExecutorService FETCH_DATA_THREAD_POOL = Executors.newCachedThreadPool();
Review comment:
Just say, using `Runtime.getRuntime().availableProcessors()` in every time we need a thread pool, globally, the number of threads would also be crazy :) Isn't it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830569643
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * eBPF Profiling Analyzer working on data query and combine them for generate the Flame Graph.
+ */
+@Slf4j
+public class EBPFProfilingAnalyzer {
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_DATA_DURATION = TimeUnit.MINUTES.toMillis(2);
+ private static final ExecutorService FETCH_DATA_THREAD_POOL = Executors.newCachedThreadPool();
Review comment:
Let's consider that later. For now, let's just make sure `Runtime.getRuntime().availableProcessors()` as default and support configuration through application.yml
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830555771
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
Review comment:
Ok, I have updated all of the `EBPFProfilingSchedule` and `EBPFProfilignData` unnecessary checks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830567656
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * eBPF Profiling Analyzer working on data query and combine them for generate the Flame Graph.
+ */
+@Slf4j
+public class EBPFProfilingAnalyzer {
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_DATA_DURATION = TimeUnit.MINUTES.toMillis(2);
+ private static final ExecutorService FETCH_DATA_THREAD_POOL = Executors.newCachedThreadPool();
Review comment:
`Runtime.getRuntime().availableProcessors()` could be default, but it should be configurable for different use cases. Such as some users may want less CPU working for analysis, more for streaming aggregation process.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830478511
##########
File path: oap-server/server-receiver-plugin/skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/provider/handler/EBPFProfilingServiceHandler.java
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.receiver.ebpf.provider.handler;
+
+import com.google.common.base.Joiner;
+import io.grpc.stub.StreamObserver;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.apm.network.common.v3.Commands;
+import org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFOnCPUProfiling;
+import org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingServiceGrpc;
+import org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingStackMetadata;
+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.command.CommandService;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingStackType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.query.type.Process;
+import org.apache.skywalking.oap.server.core.source.EBPFProfilingData;
+import org.apache.skywalking.oap.server.core.source.EBPFProcessProfilingSchedule;
+import org.apache.skywalking.oap.server.core.source.SourceReceiver;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.core.storage.query.IMetadataQueryDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.server.grpc.GRPCHandler;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class EBPFProfilingServiceHandler extends EBPFProfilingServiceGrpc.EBPFProfilingServiceImplBase implements GRPCHandler {
+ public static final List<EBPFProfilingStackType> COMMON_STACK_TYPE_ORDER = Arrays.asList(
+ EBPFProfilingStackType.KERNEL_SPACE, EBPFProfilingStackType.USER_SPACE);
+
+ private IEBPFProfilingTaskDAO taskDAO;
+ private IMetadataQueryDAO metadataQueryDAO;
+ private final SourceReceiver sourceReceiver;
+ private final CommandService commandService;
+
+ public EBPFProfilingServiceHandler(ModuleManager moduleManager) {
+ this.metadataQueryDAO = moduleManager.find(StorageModule.NAME).provider().getService(IMetadataQueryDAO.class);
+ this.taskDAO = moduleManager.find(StorageModule.NAME).provider().getService(IEBPFProfilingTaskDAO.class);
+ this.sourceReceiver = moduleManager.find(CoreModule.NAME).provider().getService(SourceReceiver.class);
+ this.commandService = moduleManager.find(CoreModule.NAME).provider().getService(CommandService.class);
+ }
+
+ @Override
+ public void queryTasks(EBPFProfilingTaskQuery request, StreamObserver<Commands> responseObserver) {
+ String agentId = request.getRoverInstanceId();
+ final long latestUpdateTime = request.getLatestUpdateTime();
+ try {
+ // find exists process from agent
+ final List<Process> processes = metadataQueryDAO.listProcesses(null, null, agentId);
+ if (CollectionUtils.isEmpty(processes)) {
+ responseObserver.onNext(Commands.newBuilder().build());
+ responseObserver.onCompleted();
+ return;
+ }
+
+ // fetch tasks from process id list
+ final EBPFProfilingProcessFinder finder = EBPFProfilingProcessFinder.builder()
+ .processIdList(processes.stream().map(Process::getId).collect(Collectors.toList())).build();
+ final List<EBPFProfilingTask> tasks = taskDAO.queryTasks(finder, null, 0, latestUpdateTime);
+ final Commands.Builder builder = Commands.newBuilder();
+ tasks.stream().map(t -> commandService.newEBPFProcessProfilingTaskCommand(t).serialize()).forEach(builder::addCommands);
+ responseObserver.onNext(builder.build());
+ responseObserver.onCompleted();
+ return;
+ } catch (Exception e) {
+ log.warn("query ebpf process profiling task failure", e);
+ }
+ responseObserver.onNext(Commands.newBuilder().build());
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public StreamObserver<org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingData> collectProfilingData(StreamObserver<Commands> responseObserver) {
+ return new StreamObserver<org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingData>() {
+ private volatile boolean isFirst = true;
+ private EBPFProfilingTaskMetadata task;
+ private String scheduleId;
+
+ @Override
+ public void onNext(org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingData ebpfProfilingData) {
+ if (isFirst || ebpfProfilingData.hasTask()) {
+ task = ebpfProfilingData.getTask();
+
+ // update schedule metadata
+ final EBPFProcessProfilingSchedule schedule = new EBPFProcessProfilingSchedule();
+ schedule.setProcessId(task.getProcessId());
+ schedule.setTaskId(task.getTaskId());
+ schedule.setStartTime(task.getProfilingStartTime());
+ schedule.setCurrentTime(task.getCurrentTime());
+ sourceReceiver.receive(schedule);
+
+ scheduleId = schedule.getEntityId();
+ }
+ isFirst = false;
+
+ // add profiling data
+ final EBPFProfilingData data = new EBPFProfilingData();
+ data.setScheduleId(scheduleId);
+ data.setTaskId(task.getTaskId());
+ data.setUploadTime(task.getCurrentTime());
+ switch (ebpfProfilingData.getProfilingCase()) {
+ case ONCPU:
+ try {
+ processOnCPUProfiling(data, ebpfProfilingData.getOnCPU());
+ } catch (IOException e) {
+ log.warn("process ON_CPU profiling data failure", e);
+ }
+ break;
+ case PROFILING_NOT_SET:
Review comment:
Should be `default`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830564521
##########
File path: docs/en/setup/backend/configuration-vocabulary.md
##########
@@ -46,6 +46,8 @@ core|default|role|Option values: `Mixed/Receiver/Aggregator`. **Receiver** mode
| - | - | maxSizeOfAnalyzeProfileSnapshot| The maximum number of snapshots analyzed by the OAP. | - | 12000 |
| - | - | prepareThreads| The number of threads used to prepare metrics data to the storage. | SW_CORE_PREPARE_THREADS | 2 |
| - | - | enableEndpointNameGroupingByOpenapi | Automatically groups endpoints by the given OpenAPI definitions. | SW_CORE_ENABLE_ENDPOINT_NAME_GROUPING_BY_OPAENAPI | true |
+| - | - | maxDurationOfAnalyzeEBPFProfiling| The maximum duration(in minute) of analyze the eBPF profiling data. | - | 10 |
+| - | - | maxDurationOfQueryEBPFProfilingData| The maximum timeout duration(in second) of query the eBPF profiling data when analysis. | - | 30 |
Review comment:
And according to your codes, this is related to time of the database query, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829943408
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/storage/EBPFProfilingTaskRecord.java
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.profiling.ebpf.storage;
+
+import com.google.common.base.Charsets;
+import com.google.common.hash.Hashing;
+import lombok.Data;
+import org.apache.skywalking.oap.server.core.analysis.Stream;
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
+import org.apache.skywalking.oap.server.core.analysis.worker.NoneStreamProcessor;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.source.ScopeDeclaration;
+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 static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.EBPF_PROFILING_TASK;
+
+@Data
+@ScopeDeclaration(id = EBPF_PROFILING_TASK, name = "EBPFProfilingTask")
+@Stream(name = EBPFProfilingTaskRecord.INDEX_NAME, scopeId = EBPF_PROFILING_TASK,
+ builder = EBPFProfilingTaskRecord.Builder.class, processor = NoneStreamProcessor.class)
+public class EBPFProfilingTaskRecord extends NoneStream {
Review comment:
Yes, now these three concepts are `EBPFProfilingTaskRecord`, `EBPFProfilingScheduleTraffic`, and `EBPFProfilingDataRecord`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829182401
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.PROCESS_FIND_TYPE, finder.getFinderType().value()));
+ }
+ if (finder != null && finder.getServiceId() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.SERVICE_ID, finder.getServiceId()));
+ }
+ if (finder != null && finder.getInstanceId() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.INSTANCE_ID, finder.getInstanceId()));
+ }
+ if (finder != null && finder.getProcessIdList() != null) {
+ query.must(Query.terms(EBPFProfilingTaskRecord.PROCESS_ID, finder.getProcessIdList()));
+ }
+ if (targetType != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.TARGET_TYPE, targetType.value()));
+ }
+ if (taskStartTime > 0) {
+ query.must(Query.range(EBPFProfilingTaskRecord.START_TIME).gte(taskStartTime));
+ }
+ if (latestUpdateTime > 0) {
+ query.must(Query.range(EBPFProfilingTaskRecord.LAST_UPDATE_TIME).gt(latestUpdateTime));
+ }
+
+ final SearchBuilder search = Search.builder().query(query)
+ .sort(EBPFProfilingTaskRecord.CREATE_TIME, Sort.Order.DESC)
+ .size(taskMaxSize);
+
+ final SearchResponse response = getClient().search(index, search.build());
Review comment:
As ebpf profiling could be much longer, `long taskStartTime, long latestUpdateTime` could mean over 5k+ profiling reports, you need to use `scroll` query. Take `MetadataQueryEsDAO#listServices` as a reference.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829978441
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
Review comment:
I have added a [benchmark](https://github.com/apache/skywalking/pull/8696/files#diff-15d820525f50abd439c83f34d6afd86a01134ed964142dc0fce907b7c1d4cc33) working on checking the analysis time without query data. I think profiling data analysis is not a big problem, we should add the `circuit break` when the query timeout. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829971322
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
Review comment:
being added in the core level check before the storage query.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830565822
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * eBPF Profiling Analyzer working on data query and combine them for generate the Flame Graph.
+ */
+@Slf4j
+public class EBPFProfilingAnalyzer {
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_DATA_DURATION = TimeUnit.MINUTES.toMillis(2);
+ private static final ExecutorService FETCH_DATA_THREAD_POOL = Executors.newCachedThreadPool();
Review comment:
This thread pool size is the same as the java ForkJoin pool size, should I update to `Runtime.getRuntime().availableProcessors()`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830564992
##########
File path: docs/en/setup/backend/configuration-vocabulary.md
##########
@@ -46,6 +46,8 @@ core|default|role|Option values: `Mixed/Receiver/Aggregator`. **Receiver** mode
| - | - | maxSizeOfAnalyzeProfileSnapshot| The maximum number of snapshots analyzed by the OAP. | - | 12000 |
| - | - | prepareThreads| The number of threads used to prepare metrics data to the storage. | SW_CORE_PREPARE_THREADS | 2 |
| - | - | enableEndpointNameGroupingByOpenapi | Automatically groups endpoints by the given OpenAPI definitions. | SW_CORE_ENABLE_ENDPOINT_NAME_GROUPING_BY_OPAENAPI | true |
+| - | - | maxDurationOfAnalyzeEBPFProfiling| The maximum duration(in minute) of analyze the eBPF profiling data. | - | 10 |
+| - | - | maxDurationOfQueryEBPFProfilingData| The maximum timeout duration(in second) of query the eBPF profiling data when analysis. | - | 30 |
Review comment:
Yes, `maxDurationOfAnalyzeEBPFProfiling` means the user chooses the time range max duration. `maxDurationOfQueryEBPFProfilingData ` means the profiling data query to storage max duration. the second one uses to avoid users waiting for the query taking a long time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830555826
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingScheduleEsDAO.java
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.RangeQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingScheduleRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingSchedule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingScheduleDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingScheduleEsDAO extends EsDAO implements IEBPFProfilingScheduleDAO {
+ private final int scheduleTaskSize;
+
+ public EBPFProfilingScheduleEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.scheduleTaskSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingSchedule> querySchedules(String taskId, long startTimeBucket, long endTimeBucket) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingScheduleRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (StringUtil.isNotEmpty(taskId)) {
+ query.must(Query.term(EBPFProfilingScheduleRecord.TASK_ID, taskId));
+ }
+ RangeQueryBuilder timeRangeQuery = null;
+ Supplier<RangeQueryBuilder> timeRangeQuerySupplier = () -> Query.range(EBPFProfilingScheduleRecord.START_TIME);
+ if (startTimeBucket > 0) {
+ timeRangeQuery = Optional.ofNullable(timeRangeQuery).orElseGet(timeRangeQuerySupplier)
+ .gte(TimeBucket.getTimestamp(startTimeBucket));
+ }
+ if (endTimeBucket > 0) {
Review comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830497637
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
Review comment:
I just use one method to cover all the different ways to get the tasks, should I change to the multiple methods to avoid the null check?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830567538
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * eBPF Profiling Analyzer working on data query and combine them for generate the Flame Graph.
+ */
+@Slf4j
+public class EBPFProfilingAnalyzer {
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_DATA_DURATION = TimeUnit.MINUTES.toMillis(2);
+ private static final ExecutorService FETCH_DATA_THREAD_POOL = Executors.newCachedThreadPool();
Review comment:
```java
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
```
From my local source codes, the 2nd parameter is `maximumPoolSize`, which is `Integer.MAX_VALUE`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829183545
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/storage/EBPFProfilingTaskRecord.java
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.profiling.ebpf.storage;
+
+import com.google.common.base.Charsets;
+import com.google.common.hash.Hashing;
+import lombok.Data;
+import org.apache.skywalking.oap.server.core.analysis.Stream;
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
+import org.apache.skywalking.oap.server.core.analysis.worker.NoneStreamProcessor;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.source.ScopeDeclaration;
+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 static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.EBPF_PROFILING_TASK;
+
+@Data
+@ScopeDeclaration(id = EBPF_PROFILING_TASK, name = "EBPFProfilingTask")
+@Stream(name = EBPFProfilingTaskRecord.INDEX_NAME, scopeId = EBPF_PROFILING_TASK,
+ builder = EBPFProfilingTaskRecord.Builder.class, processor = NoneStreamProcessor.class)
+public class EBPFProfilingTaskRecord extends NoneStream {
Review comment:
```suggestion
public class EBPFProfilingRecord extends NoneStream {
```
Don't take `task` every place. You have `task`, `schedule`, `record` concepts, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829174312
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
+ if (CollectionUtils.isEmpty(stacks)) {
+ return null;
+ }
+
+ // using parallel stream
+ Map<EBPFProfilingStack.Symbol, EBPFProfilingTree> stackTrees = stacks.parallelStream()
+ // stack list cannot be empty
+ .filter(s -> CollectionUtils.isNotEmpty(s.getSymbols()))
+ .collect(Collectors.groupingBy(s -> s.getSymbols()
+ .get(0), ANALYZE_COLLECTOR));
+
+ return new ArrayList<>(stackTrees.values());
+ }
+
+ protected IEBPFProfilingDataDAO getDataDAO() {
+ if (dataDAO == null) {
+ dataDAO = moduleManager.find(StorageModule.NAME)
+ .provider()
+ .getService(IEBPFProfilingDataDAO.class);
+ }
+ return dataDAO;
+ }
+
+ private static class TimeRange {
Review comment:
What does TimeRange mean?
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
Review comment:
This method should have very strong UTs to cover merging logic. This is the key of the whole visualization result.
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
Review comment:
Also, we need some benchmarks for this merging, as you provide a list of durations, so we could face a larger merging.
At the same time, we need a `circuit break` to avoid timeout. We should have the capability to provide a half-finished profiling tree with a message of `circuit break`.
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.PROCESS_FIND_TYPE, finder.getFinderType().value()));
+ }
+ if (finder != null && finder.getServiceId() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.SERVICE_ID, finder.getServiceId()));
+ }
+ if (finder != null && finder.getInstanceId() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.INSTANCE_ID, finder.getInstanceId()));
+ }
+ if (finder != null && finder.getProcessIdList() != null) {
+ query.must(Query.terms(EBPFProfilingTaskRecord.PROCESS_ID, finder.getProcessIdList()));
+ }
+ if (targetType != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.TARGET_TYPE, targetType.value()));
+ }
+ if (taskStartTime > 0) {
+ query.must(Query.range(EBPFProfilingTaskRecord.START_TIME).gte(taskStartTime));
+ }
+ if (latestUpdateTime > 0) {
+ query.must(Query.range(EBPFProfilingTaskRecord.LAST_UPDATE_TIME).gt(latestUpdateTime));
+ }
+
+ final SearchBuilder search = Search.builder().query(query)
+ .sort(EBPFProfilingTaskRecord.CREATE_TIME, Sort.Order.DESC)
+ .size(taskMaxSize);
+
+ final SearchResponse response = getClient().search(index, search.build());
Review comment:
As ebpf profiling could be much longer, `long taskStartTime, long latestUpdateTime` could mean over 5k+ profiling reports, you need to use `scroll` query. Take `MetadataQueryEsDAO#listServices` as a reference.
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
Review comment:
Don't check this, we should throw NPE or IllegalArgumentException, rather than doing a full table scan.
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/storage/EBPFProfilingTaskRecord.java
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.profiling.ebpf.storage;
+
+import com.google.common.base.Charsets;
+import com.google.common.hash.Hashing;
+import lombok.Data;
+import org.apache.skywalking.oap.server.core.analysis.Stream;
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
+import org.apache.skywalking.oap.server.core.analysis.worker.NoneStreamProcessor;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.source.ScopeDeclaration;
+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 static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.EBPF_PROFILING_TASK;
+
+@Data
+@ScopeDeclaration(id = EBPF_PROFILING_TASK, name = "EBPFProfilingTask")
+@Stream(name = EBPFProfilingTaskRecord.INDEX_NAME, scopeId = EBPF_PROFILING_TASK,
+ builder = EBPFProfilingTaskRecord.Builder.class, processor = NoneStreamProcessor.class)
+public class EBPFProfilingTaskRecord extends NoneStream {
Review comment:
```suggestion
public class EBPFProfilingRecord extends NoneStream {
```
Don't take `task` every place. You have `task`, `schedule`, `record` concepts, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829940649
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.PROCESS_FIND_TYPE, finder.getFinderType().value()));
+ }
+ if (finder != null && finder.getServiceId() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.SERVICE_ID, finder.getServiceId()));
+ }
+ if (finder != null && finder.getInstanceId() != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.INSTANCE_ID, finder.getInstanceId()));
+ }
+ if (finder != null && finder.getProcessIdList() != null) {
+ query.must(Query.terms(EBPFProfilingTaskRecord.PROCESS_ID, finder.getProcessIdList()));
+ }
+ if (targetType != null) {
+ query.must(Query.term(EBPFProfilingTaskRecord.TARGET_TYPE, targetType.value()));
+ }
+ if (taskStartTime > 0) {
+ query.must(Query.range(EBPFProfilingTaskRecord.START_TIME).gte(taskStartTime));
+ }
+ if (latestUpdateTime > 0) {
+ query.must(Query.range(EBPFProfilingTaskRecord.LAST_UPDATE_TIME).gt(latestUpdateTime));
+ }
+
+ final SearchBuilder search = Search.builder().query(query)
+ .sort(EBPFProfilingTaskRecord.CREATE_TIME, Sort.Order.DESC)
+ .size(taskMaxSize);
+
+ final SearchResponse response = getClient().search(index, search.build());
Review comment:
`EBPFProfilingTask` means the task which user created. The `EBPFProfilingData` is the profiling reported data, this one in the ES storage have using the scroll to query.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: [WIP] Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r829175079
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
Review comment:
This method should have very strong UTs to cover merging logic. This is the key of the whole visualization result.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830461457
##########
File path: oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingTaskEsDAO.java
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.storage.plugin.elasticsearch.query;
+
+import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Query;
+import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchHit;
+import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTargetType;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTaskRecord;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingTriggerType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingProcessFinderType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingTaskEsDAO extends EsDAO implements IEBPFProfilingTaskDAO {
+ private final int taskMaxSize;
+
+ public EBPFProfilingTaskEsDAO(ElasticSearchClient client, StorageModuleElasticsearchConfig config) {
+ super(client);
+ this.taskMaxSize = config.getProfileTaskQueryMaxSize();
+ }
+
+ @Override
+ public List<EBPFProfilingTask> queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, long taskStartTime, long latestUpdateTime) throws IOException {
+ final String index =
+ IndexController.LogicIndicesRegister.getPhysicalTableName(EBPFProfilingTaskRecord.INDEX_NAME);
+ final BoolQueryBuilder query = Query.bool();
+
+ if (finder != null && finder.getFinderType() != null) {
Review comment:
I mean you should not check parameters not null in DAO, if they are required. We need to avoid in some case, really calling DAO with all NULL, and cause full table scan.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830555782
##########
File path: oap-server/server-receiver-plugin/skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/provider/handler/EBPFProfilingServiceHandler.java
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.receiver.ebpf.provider.handler;
+
+import com.google.common.base.Joiner;
+import io.grpc.stub.StreamObserver;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.apm.network.common.v3.Commands;
+import org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFOnCPUProfiling;
+import org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingServiceGrpc;
+import org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingStackMetadata;
+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.command.CommandService;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingStackType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTask;
+import org.apache.skywalking.oap.server.core.query.type.Process;
+import org.apache.skywalking.oap.server.core.source.EBPFProfilingData;
+import org.apache.skywalking.oap.server.core.source.EBPFProcessProfilingSchedule;
+import org.apache.skywalking.oap.server.core.source.SourceReceiver;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.EBPFProfilingProcessFinder;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
+import org.apache.skywalking.oap.server.core.storage.query.IMetadataQueryDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.server.grpc.GRPCHandler;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class EBPFProfilingServiceHandler extends EBPFProfilingServiceGrpc.EBPFProfilingServiceImplBase implements GRPCHandler {
+ public static final List<EBPFProfilingStackType> COMMON_STACK_TYPE_ORDER = Arrays.asList(
+ EBPFProfilingStackType.KERNEL_SPACE, EBPFProfilingStackType.USER_SPACE);
+
+ private IEBPFProfilingTaskDAO taskDAO;
+ private IMetadataQueryDAO metadataQueryDAO;
+ private final SourceReceiver sourceReceiver;
+ private final CommandService commandService;
+
+ public EBPFProfilingServiceHandler(ModuleManager moduleManager) {
+ this.metadataQueryDAO = moduleManager.find(StorageModule.NAME).provider().getService(IMetadataQueryDAO.class);
+ this.taskDAO = moduleManager.find(StorageModule.NAME).provider().getService(IEBPFProfilingTaskDAO.class);
+ this.sourceReceiver = moduleManager.find(CoreModule.NAME).provider().getService(SourceReceiver.class);
+ this.commandService = moduleManager.find(CoreModule.NAME).provider().getService(CommandService.class);
+ }
+
+ @Override
+ public void queryTasks(EBPFProfilingTaskQuery request, StreamObserver<Commands> responseObserver) {
+ String agentId = request.getRoverInstanceId();
+ final long latestUpdateTime = request.getLatestUpdateTime();
+ try {
+ // find exists process from agent
+ final List<Process> processes = metadataQueryDAO.listProcesses(null, null, agentId);
+ if (CollectionUtils.isEmpty(processes)) {
+ responseObserver.onNext(Commands.newBuilder().build());
+ responseObserver.onCompleted();
+ return;
+ }
+
+ // fetch tasks from process id list
+ final EBPFProfilingProcessFinder finder = EBPFProfilingProcessFinder.builder()
+ .processIdList(processes.stream().map(Process::getId).collect(Collectors.toList())).build();
+ final List<EBPFProfilingTask> tasks = taskDAO.queryTasks(finder, null, 0, latestUpdateTime);
+ final Commands.Builder builder = Commands.newBuilder();
+ tasks.stream().map(t -> commandService.newEBPFProcessProfilingTaskCommand(t).serialize()).forEach(builder::addCommands);
+ responseObserver.onNext(builder.build());
+ responseObserver.onCompleted();
+ return;
+ } catch (Exception e) {
+ log.warn("query ebpf process profiling task failure", e);
+ }
+ responseObserver.onNext(Commands.newBuilder().build());
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public StreamObserver<org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingData> collectProfilingData(StreamObserver<Commands> responseObserver) {
+ return new StreamObserver<org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingData>() {
+ private volatile boolean isFirst = true;
+ private EBPFProfilingTaskMetadata task;
+ private String scheduleId;
+
+ @Override
+ public void onNext(org.apache.skywalking.apm.network.ebpf.profiling.v3.EBPFProfilingData ebpfProfilingData) {
+ if (isFirst || ebpfProfilingData.hasTask()) {
+ task = ebpfProfilingData.getTask();
+
+ // update schedule metadata
+ final EBPFProcessProfilingSchedule schedule = new EBPFProcessProfilingSchedule();
+ schedule.setProcessId(task.getProcessId());
+ schedule.setTaskId(task.getTaskId());
+ schedule.setStartTime(task.getProfilingStartTime());
+ schedule.setCurrentTime(task.getCurrentTime());
+ sourceReceiver.receive(schedule);
+
+ scheduleId = schedule.getEntityId();
+ }
+ isFirst = false;
+
+ // add profiling data
+ final EBPFProfilingData data = new EBPFProfilingData();
+ data.setScheduleId(scheduleId);
+ data.setTaskId(task.getTaskId());
+ data.setUploadTime(task.getCurrentTime());
+ switch (ebpfProfilingData.getProfilingCase()) {
+ case ONCPU:
+ try {
+ processOnCPUProfiling(data, ebpfProfilingData.getOnCPU());
+ } catch (IOException e) {
+ log.warn("process ON_CPU profiling data failure", e);
+ }
+ break;
+ case PROFILING_NOT_SET:
Review comment:
Update to the `default`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830555853
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class EBPFProfilingAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EBPFProfilingAnalyzer.class);
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_FETCH_DURATION = TimeUnit.MINUTES.toMillis(2);
+
+ private final ModuleManager moduleManager;
+ protected IEBPFProfilingDataDAO dataDAO;
+
+ public EBPFProfilingAnalyzer(ModuleManager moduleManager) {
+ this.moduleManager = moduleManager;
+ }
+
+ /**
+ * search data and analyze
+ */
+ public EBPFProfilingAnalyzation analyze(String taskId, List<EBPFProfilingAnalyzeTimeRange> ranges) throws IOException {
+ EBPFProfilingAnalyzation analyzation = new EBPFProfilingAnalyzation();
+
+ List<TimeRange> timeRanges = buildTimeRanges(ranges);
+ if (CollectionUtils.isEmpty(timeRanges)) {
+ analyzation.setTip("data not found");
+ return analyzation;
+ }
+
+ // query data
+ List<EBPFProfilingStack> stacks = timeRanges.parallelStream().map(r -> {
+ try {
+ return getDataDAO().queryData(taskId, r.minTime, r.maxTime);
+ } catch (IOException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return Collections.<EBPFProfilingDataRecord>emptyList();
+ }
+ }).flatMap(Collection::stream).map(e -> {
+ try {
+ return EBPFProfilingStack.deserialize(e);
+ } catch (Exception ex) {
+ LOGGER.warn("could not deserialize the stack", ex);
+ return null;
+ }
+ }).filter(Objects::nonNull).distinct().collect(Collectors.toList());
+
+ // analyze
+ final List<EBPFProfilingTree> trees = analyze(stacks);
+ if (trees != null) {
+ analyzation.getTrees().addAll(trees);
+ }
+
+ return analyzation;
+ }
+
+ protected List<TimeRange> buildTimeRanges(List<EBPFProfilingAnalyzeTimeRange> timeRanges) {
+ return timeRanges.parallelStream()
+ .map(r -> buildTimeRanges(r.getStart(), r.getEnd()))
+ .filter(Objects::nonNull)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ protected List<TimeRange> buildTimeRanges(long start, long end) {
+ if (start >= end) {
+ return null;
+ }
+
+ // include latest millisecond
+ end += 1;
+
+ final List<TimeRange> timeRanges = new ArrayList<>();
+ do {
+ long batchEnd = Math.min(start + FETCH_FETCH_DURATION, end);
+ timeRanges.add(new TimeRange(start, batchEnd));
+ start = batchEnd;
+ }
+ while (start < end);
+
+ return timeRanges;
+ }
+
+ /**
+ * Analyze records
+ */
+ protected List<EBPFProfilingTree> analyze(List<EBPFProfilingStack> stacks) {
+ if (CollectionUtils.isEmpty(stacks)) {
+ return null;
+ }
+
+ // using parallel stream
+ Map<EBPFProfilingStack.Symbol, EBPFProfilingTree> stackTrees = stacks.parallelStream()
+ // stack list cannot be empty
+ .filter(s -> CollectionUtils.isNotEmpty(s.getSymbols()))
+ .collect(Collectors.groupingBy(s -> s.getSymbols()
+ .get(0), ANALYZE_COLLECTOR));
+
+ return new ArrayList<>(stackTrees.values());
+ }
+
+ protected IEBPFProfilingDataDAO getDataDAO() {
+ if (dataDAO == null) {
+ dataDAO = moduleManager.find(StorageModule.NAME)
+ .provider()
+ .getService(IEBPFProfilingDataDAO.class);
+ }
+ return dataDAO;
+ }
+
+ private static class TimeRange {
Review comment:
Added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830566747
##########
File path: docs/en/setup/backend/configuration-vocabulary.md
##########
@@ -46,6 +46,8 @@ core|default|role|Option values: `Mixed/Receiver/Aggregator`. **Receiver** mode
| - | - | maxSizeOfAnalyzeProfileSnapshot| The maximum number of snapshots analyzed by the OAP. | - | 12000 |
| - | - | prepareThreads| The number of threads used to prepare metrics data to the storage. | SW_CORE_PREPARE_THREADS | 2 |
| - | - | enableEndpointNameGroupingByOpenapi | Automatically groups endpoints by the given OpenAPI definitions. | SW_CORE_ENABLE_ENDPOINT_NAME_GROUPING_BY_OPAENAPI | true |
+| - | - | maxDurationOfAnalyzeEBPFProfiling| The maximum duration(in minute) of analyze the eBPF profiling data. | - | 10 |
+| - | - | maxDurationOfQueryEBPFProfilingData| The maximum timeout duration(in second) of query the eBPF profiling data when analysis. | - | 30 |
Review comment:
```suggestion
| - | - | maxDurationOfQueryEBPFProfilingData| The maximum duration(in second) of query the eBPF profiling data from database. | - | 30 |
```
I mean this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] mrproliu commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
mrproliu commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830571195
##########
File path: oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/profiling/ebpf/analyze/EBPFProfilingAnalyzer.java
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.profiling.ebpf.analyze;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingDataRecord;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzeTimeRange;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingTree;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * eBPF Profiling Analyzer working on data query and combine them for generate the Flame Graph.
+ */
+@Slf4j
+public class EBPFProfilingAnalyzer {
+
+ private static final EBPFProfilingAnalyzeCollector ANALYZE_COLLECTOR = new EBPFProfilingAnalyzeCollector();
+ private static final Long FETCH_DATA_DURATION = TimeUnit.MINUTES.toMillis(2);
+ private static final ExecutorService FETCH_DATA_THREAD_POOL = Executors.newCachedThreadPool();
Review comment:
Added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [skywalking] wu-sheng commented on a change in pull request #8696: Implement the eBPF profiling query and data collect protocol
Posted by GitBox <gi...@apache.org>.
wu-sheng commented on a change in pull request #8696:
URL: https://github.com/apache/skywalking/pull/8696#discussion_r830564872
##########
File path: apm-protocol/apm-network/src/main/java/org/apache/skywalking/oap/server/network/trace/component/command/EBPFProfilingTaskCommand.java
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.network.trace.component.command;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import org.apache.skywalking.apm.network.common.v3.Command;
+import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
+
+/**
+ * eBPF profiling task command use to send task to the ebpf agent side
Review comment:
```suggestion
* eBPF profiling task command, OAP uses this to send a task to the ebpf agent side
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org