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/03/07 15:06:36 UTC
[skywalking] branch kezhenxu94/issue8618 updated (104f444 -> 80e88bc)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a change to branch kezhenxu94/issue8618
in repository https://gitbox.apache.org/repos/asf/skywalking.git.
discard 104f444 Support large service list query and replace deprecated API of Armeria
add 057fdf5 Use the plugin version defined by the Apache maven parent (#8613)
add ff1103e Unify module name `receiver_zipkin` to `receiver-zipkin`, remove `receiver_jaeger` from `application.yaml`. (#8616)
add f06c220 Introduce the entity of Process Type (#8617)
add ddde0c9 Set the length of event#parameters to 2000 (#8615)
add ff024ce Limit the length of Event#parameters (#8622)
add 4b0504f Update codeStyle.xml to fix incompatibility (#8623)
new 80e88bc Support large service list query and replace deprecated API of Armeria
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (104f444)
\
N -- N -- N refs/heads/kezhenxu94/issue8618 (80e88bc)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGES.md | 15 ++
apm-protocol/apm-network/src/main/proto | 2 +-
codeStyle.xml | 8 +-
docs/en/concepts-and-designs/overview.md | 4 +-
docs/en/setup/backend/configuration-vocabulary.md | 2 +-
docs/en/setup/backend/zipkin-trace.md | 2 +-
.../server/exporter/provider/MetricFormatter.java | 4 +
oap-server/pom.xml | 3 +-
.../server/core/alarm/provider/NotifyHandler.java | 15 +-
.../core/alarm/provider/NotifyHandlerTest.java | 35 +++++
...iceMetaInAlarm.java => ProcessMetaInAlarm.java} | 8 +-
.../oap/server/core/analysis/IDManager.java | 18 +++
.../analysis/manual/process/ProcessDetectType.java | 71 +++++++++
.../ProcessDispatcher.java} | 25 ++--
.../ProcessTraffic.java} | 138 ++++++++++-------
.../analysis/metrics/MetricsEntityMetaInfo.java | 80 ++++++++++
.../core/analysis/metrics/MetricsMetaInfo.java | 13 +-
.../oap/server/core/source/DefaultScopeDefine.java | 32 +++-
.../skywalking/oap/server/core/source/Event.java | 16 +-
.../{ServiceInstanceUpdate.java => Process.java} | 44 ++++--
.../elasticsearch/requests/search/Scroll.java | 19 +++
.../requests/search/ScrollBuilder.java | 19 +++
.../requests/search/SearchParams.java | 19 +++
oap-server/server-receiver-plugin/pom.xml | 1 +
.../server-receiver-plugin/receiver-proto/pom.xml | 1 -
.../pom.xml | 2 +-
.../receiver/ebpf/module/EBPFReceiverModule.java} | 8 +-
.../ebpf/provider/EBPFReceiverProvider.java} | 33 ++--
.../handler/EBPFProcessServiceHandler.java | 166 +++++++++++++++++++++
...ywalking.oap.server.library.module.ModuleDefine | 2 +-
...alking.oap.server.library.module.ModuleProvider | 2 +-
.../receiver/zipkin/ZipkinReceiverModule.java | 2 +-
oap-server/server-starter/pom.xml | 5 +
.../src/main/resources/application.yml | 12 +-
.../config/ApplicationConfigLoaderTestCase.java | 6 +-
pom.xml | 21 ---
36 files changed, 698 insertions(+), 155 deletions(-)
copy oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/{ServiceMetaInAlarm.java => ProcessMetaInAlarm.java} (90%)
create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/process/ProcessDetectType.java
copy oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/{instance/InstanceUpdateDispatcher.java => process/ProcessDispatcher.java} (69%)
copy oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/{instance/InstanceTraffic.java => process/ProcessTraffic.java} (56%)
create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/MetricsEntityMetaInfo.java
copy oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/{ServiceInstanceUpdate.java => Process.java} (62%)
copy oap-server/server-receiver-plugin/{skywalking-profile-receiver-plugin => skywalking-ebpf-receiver-plugin}/pom.xml (95%)
copy oap-server/server-receiver-plugin/{skywalking-meter-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/meter/module/MeterReceiverModule.java => skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/module/EBPFReceiverModule.java} (82%)
copy oap-server/{exporter/src/main/java/org/apache/skywalking/oap/server/exporter/provider/grpc/GRPCExporterProvider.java => server-receiver-plugin/skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/provider/EBPFReceiverProvider.java} (66%)
create mode 100644 oap-server/server-receiver-plugin/skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/provider/handler/EBPFProcessServiceHandler.java
copy oap-server/{analyzer/log-analyzer => server-receiver-plugin/skywalking-ebpf-receiver-plugin}/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine (91%)
copy oap-server/{analyzer/agent-analyzer => server-receiver-plugin/skywalking-ebpf-receiver-plugin}/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider (91%)
[skywalking] 01/01: Support large service list query and replace deprecated API of Armeria
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch kezhenxu94/issue8618
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 80e88bc715f1f50da7d465c18ea2a17399e055c9
Author: kezhenxu94 <ke...@apache.org>
AuthorDate: Mon Mar 7 22:58:29 2022 +0800
Support large service list query and replace deprecated API of Armeria
---
CHANGES.md | 3 +
docs/en/setup/backend/configuration-vocabulary.md | 3 +-
.../client/elasticsearch/ElasticSearchClient.java | 23 +++++---
.../library/elasticsearch/ElasticSearch.java | 17 ++++--
.../library/elasticsearch/client/SearchClient.java | 34 ++++++++++-
.../requests/factory/SearchFactory.java | 10 +++-
.../factory/common/CommonSearchFactory.java | 23 ++++++--
.../elasticsearch/requests/search/Scroll.java | 37 ++++++++++++
.../requests/search/ScrollBuilder.java | 51 +++++++++++++++++
.../requests/search/SearchParams.java | 66 ++++++++++++++++++++++
.../response/search/SearchResponse.java | 3 +
.../src/main/resources/application.yml | 3 +-
.../StorageModuleElasticsearchConfig.java | 6 ++
.../StorageModuleElasticsearchProvider.java | 3 +-
.../elasticsearch/query/MetadataQueryEsDAO.java | 37 ++++++++++--
15 files changed, 290 insertions(+), 29 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 0e1edb5..c57d636 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -82,6 +82,9 @@ Release Notes.
* Introduce the entity of Process type.
* Set the length of event#parameters to 2000.
* Limit the length of Event#parameters.
+* Support large service list query by using ElasticSearch scrolling API, add `metadataQueryBatchSize` to configure scrolling page size.
+* Change default value of `metadataQueryMaxSize` from `5000` to `10000`
+* Replace deprecated Armeria API `BasicToken.of` with `AuthToken.ofBasic`.
#### UI
diff --git a/docs/en/setup/backend/configuration-vocabulary.md b/docs/en/setup/backend/configuration-vocabulary.md
index 6b39677..d95a779 100644
--- a/docs/en/setup/backend/configuration-vocabulary.md
+++ b/docs/en/setup/backend/configuration-vocabulary.md
@@ -106,7 +106,8 @@ core|default|role|Option values: `Mixed/Receiver/Aggregator`. **Receiver** mode
| - | - | flushInterval| Period of flush (in seconds). Does not matter whether `bulkActions` is reached or not. INT(flushInterval * 2/3) is used for index refresh period. | SW_STORAGE_ES_FLUSH_INTERVAL | 15 (index refresh period = 10)|
| - | - | concurrentRequests| The number of concurrent requests allowed to be executed. | SW_STORAGE_ES_CONCURRENT_REQUESTS| 2 |
| - | - | resultWindowMaxSize | The maximum size of dataset when the OAP loads cache, such as network aliases. | SW_STORAGE_ES_QUERY_MAX_WINDOW_SIZE | 10000|
-| - | - | metadataQueryMaxSize | The maximum size of metadata per query. | SW_STORAGE_ES_QUERY_MAX_SIZE | 5000 |
+| - | - | metadataQueryMaxSize | The maximum size of metadata per query. | SW_STORAGE_ES_QUERY_MAX_SIZE | 10000 |
+| - | - | metadataQueryBatchSize | The batch size of metadata per iteration when `metadataQueryMaxSize` is too large to be retrieved in one query. | SW_STORAGE_ES_QUERY_BATCH_SIZE | 5000 |
| - | - | segmentQueryMaxSize | The maximum size of trace segments per query. | SW_STORAGE_ES_QUERY_SEGMENT_SIZE | 200|
| - | - | profileTaskQueryMaxSize | The maximum size of profile task per query. | SW_STORAGE_ES_QUERY_PROFILE_TASK_SIZE | 200|
| - | - | advanced | All settings of ElasticSearch index creation. The value should be in JSON format. | SW_STORAGE_ES_ADVANCED | - |
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 05b29da..e4150e9 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
@@ -40,6 +40,7 @@ import org.apache.skywalking.library.elasticsearch.ElasticSearchVersion;
import org.apache.skywalking.library.elasticsearch.bulk.BulkProcessor;
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.SearchParams;
import org.apache.skywalking.library.elasticsearch.response.Document;
import org.apache.skywalking.library.elasticsearch.response.Index;
import org.apache.skywalking.library.elasticsearch.response.IndexTemplate;
@@ -254,13 +255,11 @@ public class ElasticSearchClient implements Client, HealthCheckable {
.toArray(String[]::new);
return es.get().search(
search,
- ImmutableMap.of(
- "ignore_unavailable", true,
- "allow_no_indices", true,
- "expand_wildcards", "open"
- ),
- indexNames
- );
+ new SearchParams()
+ .ignoreUnavailable(true)
+ .allowNoIndices(true)
+ .expandWildcards(true),
+ indexNames);
}
public SearchResponse search(String indexName, Search search) {
@@ -269,6 +268,16 @@ public class ElasticSearchClient implements Client, HealthCheckable {
return es.get().search(search, indexName);
}
+ public SearchResponse search(String indexName, Search search, SearchParams params) {
+ indexName = indexNameConverter.apply(indexName);
+
+ return es.get().search(search, params, indexName);
+ }
+
+ public SearchResponse scroll(Duration contextRetention, String scrollId) {
+ return es.get().scroll(contextRetention, scrollId);
+ }
+
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 e39e53a..acb7fda 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
@@ -31,7 +31,7 @@ import com.linecorp.armeria.client.retry.RetryingClient;
import com.linecorp.armeria.common.HttpData;
import com.linecorp.armeria.common.HttpStatus;
import com.linecorp.armeria.common.SessionProtocol;
-import com.linecorp.armeria.common.auth.BasicToken;
+import com.linecorp.armeria.common.auth.AuthToken;
import com.linecorp.armeria.common.util.Exceptions;
import java.io.Closeable;
import java.io.IOException;
@@ -39,7 +39,6 @@ import java.io.InputStream;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import lombok.Getter;
@@ -51,7 +50,9 @@ import org.apache.skywalking.library.elasticsearch.client.DocumentClient;
import org.apache.skywalking.library.elasticsearch.client.IndexClient;
import org.apache.skywalking.library.elasticsearch.client.SearchClient;
import org.apache.skywalking.library.elasticsearch.client.TemplateClient;
+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 org.apache.skywalking.library.elasticsearch.response.NodeInfo;
import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
import org.apache.skywalking.oap.server.library.util.StringUtil;
@@ -104,7 +105,7 @@ public final class ElasticSearch implements Closeable {
.maxTotalAttempts(3)
.newDecorator());
if (StringUtil.isNotBlank(username) && StringUtil.isNotBlank(password)) {
- builder.auth(BasicToken.of(username, password));
+ builder.auth(AuthToken.ofBasic(username, password));
}
client = builder.build();
version = new CompletableFuture<>();
@@ -170,7 +171,7 @@ public final class ElasticSearch implements Closeable {
return aliasClient;
}
- public SearchResponse search(Search search, Map<String, ?> params, String... index) {
+ public SearchResponse search(Search search, SearchParams params, String... index) {
return searchClient.search(search, params, index);
}
@@ -178,6 +179,14 @@ public final class ElasticSearch implements Closeable {
return search(search, null, index);
}
+ public SearchResponse scroll(Duration contextRetention, String scrollId) {
+ return searchClient.scroll(
+ Scroll.builder()
+ .contextRetention(contextRetention)
+ .scrollId(scrollId)
+ .build());
+ }
+
@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 d585a2f..b442605 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
@@ -22,13 +22,14 @@ import com.linecorp.armeria.common.HttpData;
import com.linecorp.armeria.common.HttpStatus;
import com.linecorp.armeria.common.util.Exceptions;
import java.io.InputStream;
-import java.util.Map;
import java.util.concurrent.CompletableFuture;
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.search.Scroll;
import org.apache.skywalking.library.elasticsearch.requests.search.Search;
+import org.apache.skywalking.library.elasticsearch.requests.search.SearchParams;
import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
@Slf4j
@@ -40,7 +41,7 @@ public final class SearchClient {
@SneakyThrows
public SearchResponse search(Search criteria,
- Map<String, ?> params,
+ SearchParams params,
String... index) {
final CompletableFuture<SearchResponse> future =
version.thenCompose(
@@ -72,4 +73,33 @@ public final class SearchClient {
});
return future.get();
}
+
+ @SneakyThrows
+ public SearchResponse scroll(Scroll scroll) {
+ final CompletableFuture<SearchResponse> future =
+ version.thenCompose(
+ v -> client.execute(v.requestFactory().search().scroll(scroll))
+ .aggregate().thenApply(response -> {
+ if (response.status() != HttpStatus.OK) {
+ throw new RuntimeException(response.contentUtf8());
+ }
+
+ try (final HttpData content = response.content();
+ final InputStream is = content.toInputStream()) {
+ return v.codec().decode(is, SearchResponse.class);
+ } catch (Exception e) {
+ return Exceptions.throwUnsafely(e);
+ }
+ }));
+ future.whenComplete((result, exception) -> {
+ if (exception != null) {
+ log.error("Failed to scroll, request {}, {}", scroll, exception);
+ return;
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("Succeeded to scroll, {}", 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 3824a4b..51c8f5e 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
@@ -18,14 +18,20 @@
package org.apache.skywalking.library.elasticsearch.requests.factory;
import com.linecorp.armeria.common.HttpRequest;
-import java.util.Map;
+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;
public interface SearchFactory {
/**
* Returns a request to search documents.
*/
- HttpRequest search(Search search, Map<String, ?> queryParams, String... index);
+ HttpRequest search(Search search, SearchParams params, String... index);
+
+ /**
+ * Returns a request to retrieve the next batch of results for a scrolling search.
+ */
+ HttpRequest scroll(Scroll scroll);
/**
* 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 2b4ff04..4ca7689 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
@@ -20,13 +20,14 @@ package org.apache.skywalking.library.elasticsearch.requests.factory.common;
import com.linecorp.armeria.common.HttpRequest;
import com.linecorp.armeria.common.HttpRequestBuilder;
import com.linecorp.armeria.common.MediaType;
-import java.util.Map;
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;
@Slf4j
@RequiredArgsConstructor
@@ -36,7 +37,7 @@ public final class CommonSearchFactory implements SearchFactory {
@SneakyThrows
@Override
public HttpRequest search(Search search,
- Map<String, ?> queryParams,
+ SearchParams params,
String... indices) {
final HttpRequestBuilder builder = HttpRequest.builder();
@@ -47,8 +48,8 @@ public final class CommonSearchFactory implements SearchFactory {
.pathParam("indices", String.join(",", indices));
}
- if (queryParams != null && !queryParams.isEmpty()) {
- queryParams.forEach(builder::queryParam);
+ if (params != null) {
+ params.forEach(e -> builder.queryParam(e.getKey(), e.getValue()));
}
final byte[] content = version.codec().encode(search);
@@ -60,4 +61,18 @@ public final class CommonSearchFactory implements SearchFactory {
return builder.content(MediaType.JSON, content)
.build();
}
+
+ @SneakyThrows
+ @Override
+ public HttpRequest scroll(Scroll scroll) {
+ final HttpRequestBuilder builder = HttpRequest.builder().get("/_search/scroll");
+
+ final byte[] content = version.codec().encode(scroll);
+
+ if (log.isDebugEnabled()) {
+ log.debug("Scroll request: {}", new String(content));
+ }
+
+ return builder.content(MediaType.JSON, content).build();
+ }
}
diff --git a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/Scroll.java b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/Scroll.java
new file mode 100644
index 0000000..6cd7411
--- /dev/null
+++ b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/Scroll.java
@@ -0,0 +1,37 @@
+/*
+ * 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.library.elasticsearch.requests.search;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+@Getter
+@RequiredArgsConstructor(access = AccessLevel.PACKAGE)
+public final class Scroll {
+ private final String scroll;
+ @JsonProperty("scroll_id")
+ private final String scrollId;
+
+ public static ScrollBuilder builder() {
+ return new ScrollBuilder();
+ }
+}
diff --git a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/ScrollBuilder.java b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/ScrollBuilder.java
new file mode 100644
index 0000000..1ac1e1a
--- /dev/null
+++ b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/ScrollBuilder.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.skywalking.library.elasticsearch.requests.search;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import java.time.Duration;
+import com.google.common.base.Strings;
+
+public final class ScrollBuilder {
+ private Duration contextRetention;
+ private String scrollId;
+
+ public ScrollBuilder contextRetention(Duration contextRetention) {
+ checkArgument(
+ contextRetention != null && !contextRetention.isNegative()
+ && !contextRetention.isZero(),
+ "contextRetention must be positive, but was %s", contextRetention);
+ this.contextRetention = contextRetention;
+ return this;
+ }
+
+ public ScrollBuilder scrollId(String scrollId) {
+ checkArgument(!Strings.isNullOrEmpty(scrollId), "scrollId cannot be blank");
+ this.scrollId = scrollId;
+ return this;
+ }
+
+ public Scroll build() {
+ checkState(contextRetention != null, "contextRetention is not set");
+ checkState(scrollId != null, "scrollId is not set");
+ return new Scroll(contextRetention.getSeconds() + "s", scrollId);
+ }
+}
diff --git a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/SearchParams.java b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/SearchParams.java
new file mode 100644
index 0000000..326f423
--- /dev/null
+++ b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/requests/search/SearchParams.java
@@ -0,0 +1,66 @@
+/*
+ * 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.library.elasticsearch.requests.search;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;;
+
+public final class SearchParams implements Iterable<Map.Entry<String, Object>> {
+ private static final String IGNORE_UNAVAILABLE = "ignore_unavailable";
+ private static final String ALLOW_NO_INDICES = "allow_no_indices";
+ private static final String EXPAND_WILDCARDS = "expand_wildcards";
+ private static final String SCROLL = "scroll";
+
+ private final Map<String, Object> params = new HashMap<>();
+
+ public SearchParams ignoreUnavailable(boolean ignoreUnavailable) {
+ params.put(IGNORE_UNAVAILABLE, ignoreUnavailable);
+ return this;
+ }
+
+ public SearchParams allowNoIndices(boolean allowNoIndices) {
+ params.put(ALLOW_NO_INDICES, allowNoIndices);
+ return this;
+ }
+
+ public SearchParams expandWildcards(boolean expandWildcards) {
+ params.put(EXPAND_WILDCARDS, expandWildcards);
+ return this;
+ }
+
+ public SearchParams scroll(Duration contextRetention) {
+ checkArgument(
+ contextRetention != null && !contextRetention.isNegative()
+ && !contextRetention.isZero(),
+ "contextRetention must be positive, but was %s",
+ contextRetention);
+ params.put(SCROLL, contextRetention.getSeconds() + "s");
+ return this;
+ }
+
+ @Override
+ public Iterator<Entry<String, Object>> iterator() {
+ return params.entrySet().iterator();
+ }
+}
diff --git a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/response/search/SearchResponse.java b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/response/search/SearchResponse.java
index 270ac22..746c0b1 100644
--- a/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/response/search/SearchResponse.java
+++ b/oap-server/server-library/library-elasticsearch-client/src/main/java/org/apache/skywalking/library/elasticsearch/response/search/SearchResponse.java
@@ -18,6 +18,7 @@
package org.apache.skywalking.library.elasticsearch.response.search;
import java.util.Map;
+import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Getter;
import lombok.Setter;
@@ -26,4 +27,6 @@ import lombok.Setter;
public final class SearchResponse {
private SearchHits hits = new SearchHits();
private Map<String, Object> aggregations;
+ @JsonProperty("_scroll_id")
+ private String scrollId;
}
diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml
index 8cd9909..e49a97e 100755
--- a/oap-server/server-starter/src/main/resources/application.yml
+++ b/oap-server/server-starter/src/main/resources/application.yml
@@ -156,7 +156,8 @@ storage:
flushInterval: ${SW_STORAGE_ES_FLUSH_INTERVAL:15}
concurrentRequests: ${SW_STORAGE_ES_CONCURRENT_REQUESTS:2} # the number of concurrent requests
resultWindowMaxSize: ${SW_STORAGE_ES_QUERY_MAX_WINDOW_SIZE:10000}
- metadataQueryMaxSize: ${SW_STORAGE_ES_QUERY_MAX_SIZE:5000}
+ metadataQueryMaxSize: ${SW_STORAGE_ES_QUERY_MAX_SIZE:10000}
+ metadataQueryBatchSize: ${SW_STORAGE_ES_QUERY_BATCH_SIZE:5000}
segmentQueryMaxSize: ${SW_STORAGE_ES_QUERY_SEGMENT_SIZE:200}
profileTaskQueryMaxSize: ${SW_STORAGE_ES_QUERY_PROFILE_TASK_SIZE:200}
oapAnalyzer: ${SW_STORAGE_ES_OAP_ANALYZER:"{\"analyzer\":{\"oap_analyzer\":{\"type\":\"stop\"}}}"} # the oap analyzer.
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchConfig.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchConfig.java
index 47a03d5..b6ede9b 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchConfig.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchConfig.java
@@ -108,6 +108,12 @@ public class StorageModuleElasticsearchConfig extends ModuleConfig {
private String trustStorePass;
private int resultWindowMaxSize = 10000;
private int metadataQueryMaxSize = 5000;
+ /**
+ * @since 9.0.0 The batch size that is used to scroll on the large results,
+ * if {@link #metadataQueryMaxSize} is larger than the maximum result window in
+ * ElasticSearch server, this can be used to retrieve all results.
+ */
+ private int metadataQueryBatchSize = 5000;
private int segmentQueryMaxSize = 200;
private int profileTaskQueryMaxSize = 200;
/**
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java
index b58d240..5e70359 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java
@@ -175,8 +175,7 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider {
this.registerServiceImplementation(
ITraceQueryDAO.class, new TraceQueryEsDAO(elasticSearchClient, config.getSegmentQueryMaxSize()));
this.registerServiceImplementation(IBrowserLogQueryDAO.class, new BrowserLogQueryEsDAO(elasticSearchClient));
- this.registerServiceImplementation(
- IMetadataQueryDAO.class, new MetadataQueryEsDAO(elasticSearchClient, config.getMetadataQueryMaxSize()));
+ this.registerServiceImplementation(IMetadataQueryDAO.class, new MetadataQueryEsDAO(elasticSearchClient, config));
this.registerServiceImplementation(IAggregationQueryDAO.class, new AggregationQueryEsDAO(elasticSearchClient));
this.registerServiceImplementation(IAlarmQueryDAO.class, new AlarmQueryEsDAO(elasticSearchClient));
this.registerServiceImplementation(ITopNRecordsQueryDAO.class, new TopNRecordsQueryEsDAO(elasticSearchClient));
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 39ccf47..99c0fe2 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
@@ -22,6 +22,7 @@ import com.google.common.base.Strings;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import java.io.IOException;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -29,6 +30,7 @@ import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuil
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.SearchParams;
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.Layer;
@@ -44,6 +46,7 @@ import org.apache.skywalking.oap.server.core.query.type.ServiceInstance;
import org.apache.skywalking.oap.server.core.storage.query.IMetadataQueryDAO;
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 org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.MatchCNameBuilder;
@@ -51,11 +54,17 @@ import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.MatchC
import static org.apache.skywalking.oap.server.core.analysis.manual.instance.InstanceTraffic.PropertyUtil.LANGUAGE;
public class MetadataQueryEsDAO extends EsDAO implements IMetadataQueryDAO {
+ private static final Duration SCROLL_CONTEXT_RETENTION = Duration.ofSeconds(60);
+
private final int queryMaxSize;
+ private final int queryBatchSize;
- public MetadataQueryEsDAO(ElasticSearchClient client, int queryMaxSize) {
+ public MetadataQueryEsDAO(
+ ElasticSearchClient client,
+ StorageModuleElasticsearchConfig config) {
super(client);
- this.queryMaxSize = queryMaxSize;
+ this.queryMaxSize = config.getMetadataQueryMaxSize();
+ this.queryBatchSize = config.getMetadataQueryBatchSize();
}
@Override
@@ -63,18 +72,34 @@ public class MetadataQueryEsDAO extends EsDAO implements IMetadataQueryDAO {
final String index =
IndexController.LogicIndicesRegister.getPhysicalTableName(ServiceTraffic.INDEX_NAME);
+ final int batchSize = Math.min(queryMaxSize, queryBatchSize);
final BoolQueryBuilder query =
Query.bool();
- final SearchBuilder search = Search.builder().query(query).size(queryMaxSize);
+ final SearchBuilder search = Search.builder().query(query).size(batchSize);
if (StringUtil.isNotEmpty(layer)) {
query.must(Query.term(ServiceTraffic.LAYER, Layer.valueOf(layer).value()));
}
if (StringUtil.isNotEmpty(group)) {
query.must(Query.term(ServiceTraffic.GROUP, group));
}
- final SearchResponse results = getClient().search(index, search.build());
-
- return buildServices(results);
+ final SearchParams params = new SearchParams().scroll(SCROLL_CONTEXT_RETENTION);
+ 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;
+ }
+ results = getClient().scroll(SCROLL_CONTEXT_RETENTION, results.getScrollId());
+ }
+ return services;
}
@Override