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 5059caf63..5e84b0ba5 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 @@ -32,6 +32,7 @@ import java.security.KeyStoreException; import java.security.NoSuchAlgorithmException; import java.security.cert.CertificateException; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.LinkedList; @@ -80,6 +81,7 @@ import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.support.ActiveShardCount; +import org.elasticsearch.action.support.IndicesOptions; import org.elasticsearch.action.support.WriteRequest; import org.elasticsearch.client.Response; import org.elasticsearch.client.RestClient; @@ -325,15 +327,24 @@ public class ElasticSearchClient implements Client, HealthCheckable { public boolean deleteTemplate(String indexName) throws IOException { indexName = formatIndexName(indexName); - - Response response = client.getLowLevelClient() - .performRequest(HttpDelete.METHOD_NAME, "/_template/" + indexName); + Response response = client.getLowLevelClient().performRequest(HttpDelete.METHOD_NAME, "/_template/" + indexName); return response.getStatusLine().getStatusCode() == HttpStatus.SC_OK; } + public SearchResponse search(IndexNameMaker indexNameMaker, SearchSourceBuilder searchSourceBuilder) throws IOException { + String[] indexNames = Arrays.stream(indexNameMaker.make()).map(this::formatIndexName).toArray(String[]::new); + return doSearch(searchSourceBuilder, indexNames); + } + public SearchResponse search(String indexName, SearchSourceBuilder searchSourceBuilder) throws IOException { indexName = formatIndexName(indexName); - SearchRequest searchRequest = new SearchRequest(indexName); + return doSearch(searchSourceBuilder, indexName); + } + + protected SearchResponse doSearch(SearchSourceBuilder searchSourceBuilder, + String... indexNames) throws IOException { + SearchRequest searchRequest = new SearchRequest(indexNames); + searchRequest.indicesOptions(IndicesOptions.fromOptions(true, true, true, false)); searchRequest.types(TYPE); searchRequest.source(searchSourceBuilder); try { @@ -501,7 +512,8 @@ public class ElasticSearchClient implements Client, HealthCheckable { return indexName; } - @Override public void registerChecker(HealthChecker healthChecker) { + @Override + public void registerChecker(HealthChecker healthChecker) { this.healthChecker.register(healthChecker); } } diff --git a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/IndexNameMaker.java b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/IndexNameMaker.java new file mode 100644 index 000000000..237e3dd6e --- /dev/null +++ b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/IndexNameMaker.java @@ -0,0 +1,27 @@ +/* + * 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.library.client.elasticsearch; + +/** + * Implementation supports to get concrete index name + */ +@FunctionalInterface +public interface IndexNameMaker { + String[] make(); +} diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeRangeIndexNameMaker.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeRangeIndexNameMaker.java new file mode 100644 index 000000000..1b73b548c --- /dev/null +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeRangeIndexNameMaker.java @@ -0,0 +1,42 @@ +/* + * 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.base; + +import org.apache.skywalking.oap.server.library.client.elasticsearch.IndexNameMaker; + +/** + * the time range index maker works for super size dataset + */ +public class TimeRangeIndexNameMaker implements IndexNameMaker { + + private final long startSecondTB; + private final long endSecondTB; + private final String indexName; + + public TimeRangeIndexNameMaker(final String indexName, final long startSecondTB, final long endSecondTB) { + this.startSecondTB = startSecondTB; + this.endSecondTB = endSecondTB; + this.indexName = indexName; + } + + @Override + public String[] make() { + return TimeSeriesUtils.superDatasetIndexNames(indexName, startSecondTB, endSecondTB); + } +} diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtils.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtils.java index 6764def8f..5b596757b 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtils.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtils.java @@ -17,6 +17,8 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base; +import java.util.ArrayList; +import java.util.List; import lombok.Setter; import org.apache.skywalking.oap.server.core.Const; import org.apache.skywalking.oap.server.core.UnexpectedException; @@ -60,6 +62,29 @@ public class TimeSeriesUtils { } } + /** + * @return Concrete index name for super dataset index + */ + public static String[] superDatasetIndexNames(String indexName, long startSecondTB, long endSecondTB) { + if (startSecondTB == 0 || endSecondTB == 0) { + return new String[] {indexName}; + } + DateTime startDateTime = TIME_BUCKET_FORMATTER.parseDateTime(startSecondTB / 1000000 + ""); + DateTime endDateTime = TIME_BUCKET_FORMATTER.parseDateTime(endSecondTB / 1000000 + ""); + List timeRanges = new ArrayList<>(16); + for (int i = 0; i <= Days.daysBetween(startDateTime, endDateTime).getDays(); i++) { + timeRanges.add(startDateTime.plusDays(i)); + } + if (timeRanges.isEmpty()) { + return new String[] {indexName}; + } else { + return timeRanges.stream() + .map(item -> indexName + Const.LINE + compressDateTime(item, SUPER_DATASET_DAY_STEP)) + .distinct() + .toArray(String[]::new); + } + } + /** * @return index name based on model definition and given time bucket. */ @@ -116,4 +141,18 @@ public class TimeSeriesUtils { return timeBucket; } } + + static long compressDateTime(DateTime time, int dayStep) { + if (dayStep > 1) { + int days = Days.daysBetween(DAY_ONE, time).getDays(); + int groupBucketOffset = days % dayStep; + return Long.parseLong(time.minusDays(groupBucketOffset).toString(TIME_BUCKET_FORMATTER)); + } else { + /** + * No calculation required. dayStep is for lower traffic. For normally configuration, there is pointless to calculate. + */ + return Long.parseLong(time.toString(TIME_BUCKET_FORMATTER)); + } + } + } diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TraceQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TraceQueryEsDAO.java index 047be7a55..827cef4af 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TraceQueryEsDAO.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TraceQueryEsDAO.java @@ -38,6 +38,7 @@ import org.apache.skywalking.oap.server.library.util.BooleanUtils; import org.apache.skywalking.oap.server.library.util.CollectionUtils; import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO; import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.MatchCNameBuilder; +import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeRangeIndexNameMaker; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.QueryBuilder; @@ -132,9 +133,7 @@ public class TraceQueryEsDAO extends EsDAO implements ITraceQueryDAO { } sourceBuilder.size(limit); sourceBuilder.from(from); - - SearchResponse response = getClient().search(SegmentRecord.INDEX_NAME, sourceBuilder); - + SearchResponse response = getClient().search(new TimeRangeIndexNameMaker(SegmentRecord.INDEX_NAME, startSecondTB, endSecondTB), sourceBuilder); TraceBrief traceBrief = new TraceBrief(); traceBrief.setTotal((int) response.getHits().totalHits); diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTestCase.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTestCase.java index 8e434e876..07e4fbead 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTestCase.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTestCase.java @@ -17,6 +17,7 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base; +import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord; import org.junit.Assert; import org.junit.Test; @@ -26,4 +27,22 @@ public class TimeSeriesUtilsTestCase { public void indexTimeSeries() { Assert.assertEquals(20190602, TimeSeriesUtils.isolateTimeFromIndexName("Index_Test-20190602")); } + + @Test + public void querySuperDatasetIndices() { + String[] indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200601140000L, 20200605140000L); + Assert.assertEquals(indices.length, 5); + indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200605140000L, 20200605140000L); + Assert.assertEquals(indices.length, 1); + indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200605140000L, 20200601140000L); + Assert.assertEquals(indices.length, 1); + TimeSeriesUtils.setSUPER_DATASET_DAY_STEP(2); + indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200601140000L, 20200605140000L); + Assert.assertEquals(indices.length, 3); + indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200605140000L, 20200605140000L); + Assert.assertEquals(indices.length, 1); + indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200605140000L, 20200601140000L); + Assert.assertEquals(indices.length, 1); + + } } diff --git a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java index b1534b0a3..2f2d518ee 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java @@ -192,9 +192,9 @@ public class ElasticSearch7Client extends ElasticSearchClient { return acknowledgedResponse.isAcknowledged(); } - public SearchResponse search(String indexName, SearchSourceBuilder searchSourceBuilder) throws IOException { - indexName = formatIndexName(indexName); - SearchRequest searchRequest = new SearchRequest(indexName); + @Override + public SearchResponse doSearch(SearchSourceBuilder searchSourceBuilder, String... indexNames) throws IOException { + SearchRequest searchRequest = new SearchRequest(indexNames); searchRequest.source(searchSourceBuilder); try { SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT); diff --git a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/query/TraceQueryEs7DAO.java b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/query/TraceQueryEs7DAO.java index 2077fa96f..7b255ee26 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/query/TraceQueryEs7DAO.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/query/TraceQueryEs7DAO.java @@ -32,6 +32,7 @@ import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSear import org.apache.skywalking.oap.server.library.util.BooleanUtils; import org.apache.skywalking.oap.server.library.util.CollectionUtils; import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.MatchCNameBuilder; +import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeRangeIndexNameMaker; import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.TraceQueryEsDAO; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.index.query.BoolQueryBuilder; @@ -124,8 +125,7 @@ public class TraceQueryEs7DAO extends TraceQueryEsDAO { } sourceBuilder.size(limit); sourceBuilder.from(from); - - SearchResponse response = getClient().search(SegmentRecord.INDEX_NAME, sourceBuilder); + SearchResponse response = getClient().search(new TimeRangeIndexNameMaker(SegmentRecord.INDEX_NAME, startSecondTB, endSecondTB), sourceBuilder); TraceBrief traceBrief = new TraceBrief(); traceBrief.setTotal((int) response.getHits().getTotalHits().value);