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

ElasticSearch: scroll id should be updated when scrolling as it may change (#9199)

This commit is contained in:
kezhenxu94 2022-06-10 18:13:59 +08:00 committed by GitHub
parent 96dd6bf8d9
commit 9ccb04e6cf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 28 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -4,6 +4,8 @@
#### OAP Server
* ElasticSearch: scroll id should be updated when scrolling as it may change
#### Documentation
* Fix invalid links in release docs

8
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/cache/NetworkAddressAliasEsDAO.java vendored
View File

@ -19,7 +19,9 @@
package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.cache;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.library.elasticsearch.requests.search.Query;
import org.apache.skywalking.library.elasticsearch.requests.search.Search;
@ -62,9 +64,11 @@ public class NetworkAddressAliasEsDAO extends EsDAO implements INetworkAddressAl
SearchResponse results =
getClient().search(NetworkAddressAlias.INDEX_NAME, search, params);
final String scrollId = results.getScrollId();
final Set<String> scrollIds = new HashSet<>();
try {
while (true) {
final String scrollId = results.getScrollId();
scrollIds.add(scrollId);
if (results.getHits().getTotal() == 0) {
break;
}
@ -82,7 +86,7 @@ public class NetworkAddressAliasEsDAO extends EsDAO implements INetworkAddressAl
results = getClient().scroll(SCROLL_CONTEXT_RETENTION, scrollId);
}
} finally {
getClient().deleteScrollContextQuietly(scrollId);
scrollIds.forEach(getClient()::deleteScrollContextQuietly);
}
} catch (Throwable t) {
log.error(t.getMessage(), t);

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

@ -35,8 +35,10 @@ import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexC
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
public class EBPFProfilingDataEsDAO extends EsDAO implements IEBPFProfilingDataDAO {
private final int scrollingBatchSize;
@ -59,9 +61,11 @@ public class EBPFProfilingDataEsDAO extends EsDAO implements IEBPFProfilingDataD
final List<EBPFProfilingDataRecord> records = new ArrayList<>();
SearchResponse results = getClient().search(index, search.build(), params);
final String scrollId = results.getScrollId();
final Set<String> scrollIds = new HashSet<>();
try {
while (true) {
final String scrollId = results.getScrollId();
scrollIds.add(scrollId);
if (results.getHits().getTotal() == 0) {
break;
}
@ -74,7 +78,7 @@ public class EBPFProfilingDataEsDAO extends EsDAO implements IEBPFProfilingDataD
results = getClient().scroll(SCROLL_CONTEXT_RETENTION, scrollId);
}
} finally {
getClient().deleteScrollContextQuietly(scrollId);
scrollIds.forEach(getClient()::deleteScrollContextQuietly);
}
return records;
}

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

@ -24,8 +24,10 @@ import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import org.apache.skywalking.library.elasticsearch.requests.search.BoolQueryBuilder;
import org.apache.skywalking.library.elasticsearch.requests.search.Query;
@ -93,9 +95,11 @@ public class MetadataQueryEsDAO extends EsDAO implements IMetadataQueryDAO {
final List<Service> services = new ArrayList<>();
SearchResponse results = getClient().search(index, search.build(), params);
String scrollId = results.getScrollId();
Set<String> scrollIds = new HashSet<>();
try {
while (true) {
String scrollId = results.getScrollId();
scrollIds.add(scrollId);
if (results.getHits().getTotal() == 0) {
break;
}
@ -112,7 +116,7 @@ public class MetadataQueryEsDAO extends EsDAO implements IMetadataQueryDAO {
results = getClient().scroll(SCROLL_CONTEXT_RETENTION, scrollId);
}
} finally {
getClient().deleteScrollContextQuietly(scrollId);
scrollIds.forEach(getClient()::deleteScrollContextQuietly);
}
return services;
}

11
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/zipkin/ZipkinQueryEsDAO.java
View File

@ -151,16 +151,21 @@ public class ZipkinQueryEsDAO extends EsDAO implements IZipkinQueryDAO {
SearchBuilder search = Search.builder().query(query).size(5000); //max span size for 1 scroll
final SearchParams params = new SearchParams().scroll(SCROLL_CONTEXT_RETENTION);
SearchResponse response = getClient().search(index, search.build(), params);
String scrollId = response.getScrollId();
Map<String, List<Span>> groupedByTraceId = new LinkedHashMap<String, List<Span>>();
final Set<String> scrollIds = new HashSet<>();
try {
while (response.getHits().getHits().size() != 0) {
while (true) {
String scrollId = response.getScrollId();
scrollIds.add(scrollId);
if (response.getHits().getHits().size() == 0) {
break;
}
buildTraces(response, groupedByTraceId);
response = getClient().scroll(SCROLL_CONTEXT_RETENTION, scrollId);
}
} finally {
getClient().deleteScrollContextQuietly(scrollId);
scrollIds.forEach(getClient()::deleteScrollContextQuietly);
}
return new ArrayList<>(groupedByTraceId.values());
}