You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ke...@apache.org on 2022/05/06 01:35:34 UTC
[skywalking] branch master updated: Clean up scroll context after going through the results (#8967)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/master by this push:
new 369949b29a Clean up scroll context after going through the results (#8967)
369949b29a is described below
commit 369949b29a716b9d8d857b0f75a742c49971b71d
Author: kezhenxu94 <ke...@apache.org>
AuthorDate: Fri May 6 09:35:27 2022 +0800
Clean up scroll context after going through the results (#8967)
---
.github/workflows/skywalking.yaml | 22 +++++++--------
docs/en/changes/changes.md | 1 +
.../client/elasticsearch/ElasticSearchClient.java | 14 +++++++---
.../library/elasticsearch/ElasticSearch.java | 4 +++
.../library/elasticsearch/client/SearchClient.java | 23 ++++++++++++++++
.../requests/factory/SearchFactory.java | 5 ++++
.../factory/common/CommonSearchFactory.java | 23 ++++++++++++++--
.../cache/NetworkAddressAliasEsDAO.java | 31 +++++++++++++--------
.../query/EBPFProfilingDataEsDAO.java | 24 ++++++++++------
.../elasticsearch/query/MetadataQueryEsDAO.java | 32 +++++++++++++---------
10 files changed, 129 insertions(+), 50 deletions(-)
diff --git a/.github/workflows/skywalking.yaml b/.github/workflows/skywalking.yaml
index 562c82532e..707dc0cf8a 100644
--- a/.github/workflows/skywalking.yaml
+++ b/.github/workflows/skywalking.yaml
@@ -85,7 +85,7 @@ jobs:
token: ${{ github.token }}
# The following is a single composite pattern that allows next CI steps,
# the pattern is in form of [not (foo or bar)] to be safe.
- # When new files come in, the CI will NOT ignore them unless listed,
+ # When new files come in, the CI will NOT ignore them unless listed,
# so remember to extend here if they do not serve functional purposes.
# NOTE: careful with using ** in expression, keep !**/{old, new things}.
filters: |
@@ -103,10 +103,10 @@ jobs:
component-libraries.yml\
}"
list-files: json # logs matched files
-
+
dist-tar:
if: |
- always() &&
+ always() &&
((github.event_name == 'schedule' && github.repository == 'apache/skywalking') || needs.changes.outputs.oap == 'true')
name: Build dist tar
needs: [ sanity-check, changes ]
@@ -135,7 +135,7 @@ jobs:
docker:
if: |
- always() &&
+ always() &&
((github.event_name == 'schedule' && github.repository == 'apache/skywalking') || needs.changes.outputs.oap == 'true')
name: Docker images
needs: [ dist-tar, changes ]
@@ -173,7 +173,7 @@ jobs:
dependency-license:
if: |
- always() &&
+ always() &&
((github.event_name == 'schedule' && github.repository == 'apache/skywalking') || needs.changes.outputs.oap == 'true')
name: Dependency licenses
needs: [ dist-tar, changes ]
@@ -197,7 +197,7 @@ jobs:
unit-test:
if: |
- always() &&
+ always() &&
((github.event_name == 'schedule' && github.repository == 'apache/skywalking') || needs.changes.outputs.oap == 'true')
name: Unit test
needs: [ sanity-check, changes ]
@@ -229,7 +229,7 @@ jobs:
integration-test:
if: |
- always() &&
+ always() &&
((github.event_name == 'schedule' && github.repository == 'apache/skywalking') || needs.changes.outputs.oap == 'true')
name: Integration test
needs: [ sanity-check, changes ]
@@ -259,7 +259,7 @@ jobs:
e2e-test:
if: |
- always() &&
+ always() &&
((github.event_name == 'schedule' && github.repository == 'apache/skywalking') || needs.changes.outputs.oap == 'true')
name: E2E test
needs: [ docker ]
@@ -518,7 +518,7 @@ jobs:
e2e-test-istio:
if: |
- always() &&
+ always() &&
((github.event_name == 'schedule' && github.repository == 'apache/skywalking') || needs.changes.outputs.oap == 'true')
name: E2E test
needs: [ docker ]
@@ -554,7 +554,7 @@ jobs:
e2e-test-java-versions:
if: |
- always() &&
+ always() &&
((github.event_name == 'schedule' && github.repository == 'apache/skywalking') || needs.changes.outputs.oap == 'true')
name: E2E test
needs: [ docker ]
@@ -607,7 +607,7 @@ jobs:
timeout-minutes: 10
steps:
- name: Merge Requirement
- # check changes, sanity, dependency license, unit, integration, e2e, e2e-istio and e2e-java-versions,
+ # check changes, sanity, dependency license, unit, integration, e2e, e2e-istio and e2e-java-versions,
# if all of them are working as expected then naturally exits else return error code
run: |
execute=${{ needs.changes.outputs.oap }}
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 06a82323ae..f963806c81 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -41,6 +41,7 @@
* Enhance Converter mechanism in kernel level to make BanyanDB native feature more effective.
* Add TermsAggregation properties collect_mode and execution_hint.
* Add "execution_hint": "map", "collect_mode": "breadth_first" for aggregation and topology query to improve 5-10x performance.
+* Clean up scroll contexts after used.
#### UI
diff --git a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java
index cdbd8750ee..482ed58745 100644
--- a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java
+++ b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java
@@ -255,10 +255,7 @@ public class ElasticSearchClient implements Client, HealthCheckable {
.toArray(String[]::new);
return es.get().search(
search,
- new SearchParams()
- .ignoreUnavailable(true)
- .allowNoIndices(true)
- .expandWildcards("open"),
+ null,
indexNames);
}
@@ -278,6 +275,15 @@ public class ElasticSearchClient implements Client, HealthCheckable {
return es.get().scroll(contextRetention, scrollId);
}
+ public boolean deleteScrollContextQuietly(String scrollId) {
+ try {
+ return es.get().deleteScrollContext(scrollId);
+ } catch (Exception e) {
+ log.warn("Failed to delete scroll context: {}", scrollId, e);
+ return false;
+ }
+ }
+
public Optional<Document> get(String indexName, String id) {
indexName = indexNameConverter.apply(indexName);
diff --git a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearch.java b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearch.java
index acb7fdaea8..de6b7414f7 100644
--- a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearch.java
+++ b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/ElasticSearch.java
@@ -187,6 +187,10 @@ public final class ElasticSearch implements Closeable {
.build());
}
+ public boolean deleteScrollContext(String scrollId) {
+ return searchClient.deleteScrollContext(scrollId);
+ }
+
@Override
public void close() {
endpointGroup.removeListener(healthyEndpointListener);
diff --git a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/client/SearchClient.java b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/client/SearchClient.java
index b442605850..17aff2b3ab 100644
--- a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/client/SearchClient.java
+++ b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/client/SearchClient.java
@@ -102,4 +102,27 @@ public final class SearchClient {
});
return future.get();
}
+
+ @SneakyThrows
+ public boolean deleteScrollContext(String scrollId) {
+ final CompletableFuture<Boolean> future =
+ version.thenCompose(
+ v -> client.execute(v.requestFactory().search().deleteScrollContext(scrollId))
+ .aggregate().thenApply(response -> {
+ if (response.status() == HttpStatus.OK) {
+ return true;
+ }
+ throw new RuntimeException(response.contentUtf8());
+ }));
+ future.whenComplete((result, exception) -> {
+ if (exception != null) {
+ log.error("Failed to delete scroll context, request {}, {}", scrollId, exception);
+ return;
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Succeeded to delete scroll context, {}", result);
+ }
+ });
+ return future.get();
+ }
}
diff --git a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/SearchFactory.java b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/SearchFactory.java
index 51c8f5e989..aa6bf36379 100644
--- a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/SearchFactory.java
+++ b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/SearchFactory.java
@@ -33,6 +33,11 @@ public interface SearchFactory {
*/
HttpRequest scroll(Scroll scroll);
+ /**
+ * Returns a request to delete the scroll context.
+ */
+ HttpRequest deleteScrollContext(String scrollId);
+
/**
* Returns a request to search documents.
*/
diff --git a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/common/CommonSearchFactory.java b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/common/CommonSearchFactory.java
index 4ca76893a4..a54de130aa 100644
--- a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/common/CommonSearchFactory.java
+++ b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/factory/common/CommonSearchFactory.java
@@ -17,17 +17,19 @@
package org.apache.skywalking.library.elasticsearch.requests.factory.common;
+import java.util.HashMap;
+import java.util.Map;
import com.linecorp.armeria.common.HttpRequest;
import com.linecorp.armeria.common.HttpRequestBuilder;
import com.linecorp.armeria.common.MediaType;
-import lombok.RequiredArgsConstructor;
-import lombok.SneakyThrows;
-import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.library.elasticsearch.ElasticSearchVersion;
import org.apache.skywalking.library.elasticsearch.requests.factory.SearchFactory;
import org.apache.skywalking.library.elasticsearch.requests.search.Scroll;
import org.apache.skywalking.library.elasticsearch.requests.search.Search;
import org.apache.skywalking.library.elasticsearch.requests.search.SearchParams;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
@Slf4j
@RequiredArgsConstructor
@@ -75,4 +77,19 @@ public final class CommonSearchFactory implements SearchFactory {
return builder.content(MediaType.JSON, content).build();
}
+
+ @SneakyThrows
+ @Override
+ public HttpRequest deleteScrollContext(String scrollId) {
+ final HttpRequestBuilder builder = HttpRequest.builder().delete("/_search/scroll");
+ final Map<String, String> params = new HashMap<>();
+ params.put("scroll_id", scrollId);
+ final byte[] content = version.codec().encode(params);
+
+ if (log.isDebugEnabled()) {
+ log.debug("Delete scroll context request: {}", new String(content));
+ }
+
+ return builder.content(MediaType.JSON, content).build();
+ }
}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/cache/NetworkAddressAliasEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/cache/NetworkAddressAliasEsDAO.java
index 80b207ab42..402ce4c720 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/cache/NetworkAddressAliasEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/cache/NetworkAddressAliasEsDAO.java
@@ -62,18 +62,27 @@ public class NetworkAddressAliasEsDAO extends EsDAO implements INetworkAddressAl
SearchResponse results =
getClient().search(NetworkAddressAlias.INDEX_NAME, search, params);
- while (results.getHits().getTotal() > 0) {
- for (SearchHit searchHit : results.getHits()) {
- networkAddressAliases.add(
- builder.storage2Entity(new HashMapConverter.ToEntity(searchHit.getSource())));
+ while (true) {
+ final String scrollId = results.getScrollId();
+ try {
+ if (results.getHits().getTotal() == 0) {
+ break;
+ }
+ for (SearchHit searchHit : results.getHits()) {
+ networkAddressAliases.add(
+ builder.storage2Entity(
+ new HashMapConverter.ToEntity(searchHit.getSource())));
+ }
+ if (results.getHits().getTotal() < batchSize) {
+ break;
+ }
+ if (networkAddressAliases.size() >= resultWindowMaxSize) {
+ break;
+ }
+ results = getClient().scroll(SCROLL_CONTEXT_RETENTION, scrollId);
+ } finally {
+ getClient().deleteScrollContextQuietly(scrollId);
}
- if (results.getHits().getTotal() < batchSize) {
- break;
- }
- if (networkAddressAliases.size() >= resultWindowMaxSize) {
- break;
- }
- results = getClient().scroll(SCROLL_CONTEXT_RETENTION, results.getScrollId());
}
} catch (Throwable t) {
log.error(t.getMessage(), t);
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingDataEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingDataEsDAO.java
index 47dd3c2b1d..b10575da74 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingDataEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EBPFProfilingDataEsDAO.java
@@ -59,14 +59,22 @@ public class EBPFProfilingDataEsDAO extends EsDAO implements IEBPFProfilingDataD
final List<EBPFProfilingDataRecord> records = new ArrayList<>();
SearchResponse results = getClient().search(index, search.build(), params);
- while (results.getHits().getTotal() > 0) {
- final List<EBPFProfilingDataRecord> batch = buildDataList(results);
- records.addAll(batch);
- // The last iterate, there is no more data
- if (batch.size() < scrollingBatchSize) {
- break;
+ while (true) {
+ final String scrollId = results.getScrollId();
+ try {
+ if (results.getHits().getTotal() == 0) {
+ break;
+ }
+ final List<EBPFProfilingDataRecord> batch = buildDataList(results);
+ records.addAll(batch);
+ // The last iterate, there is no more data
+ if (batch.size() < scrollingBatchSize) {
+ break;
+ }
+ results = getClient().scroll(SCROLL_CONTEXT_RETENTION, scrollId);
+ } finally {
+ getClient().deleteScrollContextQuietly(scrollId);
}
- results = getClient().scroll(SCROLL_CONTEXT_RETENTION, results.getScrollId());
}
return records;
}
@@ -80,4 +88,4 @@ public class EBPFProfilingDataEsDAO extends EsDAO implements IEBPFProfilingDataD
}
return records;
}
-}
\ No newline at end of file
+}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetadataQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetadataQueryEsDAO.java
index 354b5e8827..59f78369a1 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetadataQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetadataQueryEsDAO.java
@@ -26,7 +26,6 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-
import org.apache.commons.lang3.StringUtils;
import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
import org.apache.skywalking.library.elasticsearch.requests.search.Query;
@@ -59,7 +58,6 @@ import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModu
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.MatchCNameBuilder;
-
import static org.apache.skywalking.oap.server.core.analysis.manual.instance.InstanceTraffic.PropertyUtil.LANGUAGE;
public class MetadataQueryEsDAO extends EsDAO implements IMetadataQueryDAO {
@@ -95,18 +93,26 @@ public class MetadataQueryEsDAO extends EsDAO implements IMetadataQueryDAO {
final List<Service> services = new ArrayList<>();
SearchResponse results = getClient().search(index, search.build(), params);
- while (results.getHits().getTotal() > 0) {
- final List<Service> batch = buildServices(results);
- services.addAll(batch);
- // The last iterate, there is no more data
- if (batch.size() < batchSize) {
- break;
- }
- // We've got enough data
- if (services.size() >= queryMaxSize) {
- break;
+ while (true) {
+ final String scrollId = results.getScrollId();
+ try {
+ if (results.getHits().getTotal() == 0) {
+ break;
+ }
+ final List<Service> batch = buildServices(results);
+ services.addAll(batch);
+ // The last iterate, there is no more data
+ if (batch.size() < batchSize) {
+ break;
+ }
+ // We've got enough data
+ if (services.size() >= queryMaxSize) {
+ break;
+ }
+ results = getClient().scroll(SCROLL_CONTEXT_RETENTION, scrollId);
+ } finally {
+ getClient().deleteScrollContextQuietly(scrollId);
}
- results = getClient().scroll(SCROLL_CONTEXT_RETENTION, results.getScrollId());
}
return services;
}