You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/10/15 06:33:26 UTC
(skywalking) 01/01: Build the dated Elasticsearch index names through one policy, and fix three index-list bugs
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch refactor/es-index-name-policy
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 8fc4a45301be83fa1b01afd0ffada5163de5ebe0
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Thu Oct 15 14:25:23 2026 +0800
Build the dated Elasticsearch index names through one policy, and fix three index-list bugs
The day step of a physical table is now decided once, from its models:
IndexRolling (NONE, DAY_STEP, SUPER_DATASET_DAY_STEP) is recorded per
physical table by IndexController.LogicIndicesRegister.registerRelation,
which rejects a table registered with two kinds; the catalog maps are
concurrent, as runtime rules register models after boot. IndexNames builds
every dated index name from an IndexFamily, so a caller no longer picks the
step. TimeSeriesUtils and TimeRangeIndexNameGenerator are removed, and
ElasticSearchClient.search takes an index array instead of a Supplier.
Apart from the fixes below, index names are unchanged.
Fixes:
- Zipkin getTraces listed one daily index per day since 1970 when the query
started at or before the epoch. It now searches the zipkin_span family in
that case, as its time filter already did.
- A model applied without a schema change skipped the isExists probe, the
only place its table was registered, so with logicSharding on a bundled
MAL rule restored after its runtime override kept the override's table.
StorageEsInstaller now registers in populateLocalCacheOnly.
- OTLP queryTraceById joined the day indices into one name, so the namespace
went on the first only and a missing day failed the scroll. The scroller
now takes the indices as an array, each namespaced, skipping missing days.
---
docs/en/changes/changes.md | 3 +
.../client/elasticsearch/ElasticSearchClient.java | 25 +-
.../elasticsearch/ElasticSearchScroller.java | 9 +-
.../elasticsearch/bulk/ElasticSearchIT.java | 35 ++
.../StorageModuleElasticsearchProvider.java | 10 +-
.../storage/plugin/elasticsearch/base/EsDAO.java | 7 +-
.../elasticsearch/base/HistoryDeleteEsDAO.java | 4 +-
.../plugin/elasticsearch/base/IndexController.java | 46 ++-
.../plugin/elasticsearch/base/IndexFamily.java | 61 +++
.../plugin/elasticsearch/base/IndexNames.java | 208 +++++++++++
.../plugin/elasticsearch/base/IndexRolling.java | 49 +++
.../plugin/elasticsearch/base/MetricsEsDAO.java | 6 +-
.../plugin/elasticsearch/base/NoneStreamEsDAO.java | 2 +-
.../plugin/elasticsearch/base/RecordEsDAO.java | 2 +-
.../elasticsearch/base/StorageEsInstaller.java | 12 +-
.../base/TimeRangeIndexNameGenerator.java | 44 ---
.../plugin/elasticsearch/base/TimeSeriesUtils.java | 189 ----------
.../query/AIAgentConversationQueryEsDAO.java | 28 +-
.../elasticsearch/query/AggregationQueryEsDAO.java | 9 +-
.../query/GenAIEvaluationRecordQueryEsDAO.java | 10 +-
.../plugin/elasticsearch/query/LogQueryEsDAO.java | 10 +-
.../elasticsearch/query/MetricsQueryEsDAO.java | 40 +-
.../query/TagAutoCompleteQueryDAO.java | 19 +-
.../elasticsearch/query/TraceQueryEsDAO.java | 9 +-
.../query/otlp/OTLPTraceQueryEsDAO.java | 32 +-
.../query/zipkin/ZipkinQueryEsDAO.java | 16 +-
.../plugin/elasticsearch/base/IndexNamesTest.java | 181 +++++++++
.../elasticsearch/base/StorageEsInstallerTest.java | 66 ++++
.../elasticsearch/base/TimeSeriesUtilsTest.java | 128 -------
.../base/TimeSeriesUtilsTestCase.java | 48 ---
.../query/AggregationQueryEsDAOTest.java | 18 +-
.../query/EsReadIndexSelectionTest.java | 416 +++++++++++++++++++++
32 files changed, 1177 insertions(+), 565 deletions(-)
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 965af5538d..e8e2b5a1a3 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -44,6 +44,9 @@
* Cut the slow statements a LAL `SlowSQL` output writes, as the default `mysql-slowsql`, `pgsql-slowsql` and `redis-slowsql` rules do, to the 2000 characters of the `top_n_database_statement` column, as the agent path already cuts them to `maxSlowSQLLength`. A longer statement was refused by a strict SQL store, and stored whole on Elasticsearch and BanyanDB; it is now cut there too.
* Fix a UI template update on BanyanDB leaving a disabled template disabled. `PUT /ui-management/templates` wrote only the configuration there, while on ElasticSearch and JDBC it replaces the whole record and enables it. An update is the only way to bring a disabled template back on those two stores, where adding refuses an existing id; the BanyanDB update now writes the whole record too, so the same call works on every storage.
* Fix the Elasticsearch storage reading metrics from the wrong daily indices when `dayStep` and `superDatasetDayStep` differ. The top N query, the labeled metrics read without an entity, the entity listing of a metric and the tag autocomplete query located their indices with `superDatasetDayStep`, though metrics are written to indices rolled by `dayStep`, and Elasticsearch ignores the missing indices, so these queries returned nothing or only part of the range. They now use `dayStep`.
+* Fix the Elasticsearch Zipkin trace search listing the daily indices of every day since 1970 when the query starts at or before the epoch, as a `lookback` equal to `endTs` makes it. With the default day step of 1 that is over 20,000 index names, which can exceed the request line Elasticsearch accepts. The search now uses the whole `zipkin_span` index family in that case, as its time filter already did. Queries with a usual `lookback` are not affected.
+* Fix Elasticsearch reads of a metric a peer OAP reinstalls without a schema change, such as a bundled MAL rule restored after the runtime rule overriding it is deleted. The peer kept the index it had registered for the runtime rule. This only matters with `logicSharding` enabled and when the two rules use different functions, since the function decides the index there; with the default merged indices every metric stays in `metrics-all`.
+* Build the dated Elasticsearch index names in one place, `IndexNames`, with the day step decided once per physical table from its models, replacing `TimeSeriesUtils` and `TimeRangeIndexNameGenerator`. Apart from the fixes above, index names are unchanged.
#### UI
* Add a Virtual GenAI evaluation-record page and evaluation-score chart in Horizon UI, so operators can inspect evaluation result, level, reason, judge model, timestamp, trace linkage, and the `gen_ai_model_evaluation_score_ppm` trend for evaluated records.
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 5a6a7933f5..952b5f2172 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
@@ -33,7 +33,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
-import java.util.function.Supplier;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.library.elasticsearch.requests.search.Query;
@@ -290,19 +289,27 @@ public class ElasticSearchClient implements Client, HealthCheckable {
return es.get().templates().delete(indexName);
}
- public SearchResponse search(Supplier<String[]> indices, Search search) {
- final String[] indexNames =
- Arrays.stream(indices.get())
+ public SearchResponse search(final String[] indexNames, final Search search) {
+ return search(indexNames, search, new SearchParams());
+ }
+
+ /**
+ * Searches several indices, each named without the namespace, which is added here to every name. Some of
+ * the indices may not exist, such as days without data in a time range, so the search skips them instead
+ * of failing. The given params keep their other settings, such as routing and scroll.
+ */
+ public SearchResponse search(final String[] indexNames, final Search search, final SearchParams params) {
+ final String[] namespacedNames =
+ Arrays.stream(indexNames)
.map(indexNameConverter)
.toArray(String[]::new);
- final SearchParams params = new SearchParams()
- .allowNoIndices(true)
- .ignoreUnavailable(true)
- .expandWildcards("open");
+ params.allowNoIndices(true)
+ .ignoreUnavailable(true)
+ .expandWildcards("open");
return es.get().search(
search,
params,
- indexNames);
+ namespacedNames);
}
public SearchResponse search(String indexName, Search search) {
diff --git a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchScroller.java b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchScroller.java
index 62504e83fc..7cefe125bb 100644
--- a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchScroller.java
+++ b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchScroller.java
@@ -38,6 +38,11 @@ public class ElasticSearchScroller<T> {
final ElasticSearchClient client;
final Search search;
final String index;
+ /**
+ * When set, the scroll searches these indices instead of {@link #index}, see
+ * {@link ElasticSearchClient#search(String[], Search, SearchParams)}.
+ */
+ final String[] indices;
@Builder.Default
final int queryMaxSize = 0;
@Builder.Default
@@ -50,7 +55,9 @@ public class ElasticSearchScroller<T> {
params.scroll(SCROLL_CONTEXT_RETENTION);
- var response = client.search(index, search, params);
+ var response = indices != null
+ ? client.search(indices, search, params)
+ : client.search(index, search, params);
try {
while (true) {
diff --git a/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/library/elasticsearch/bulk/ElasticSearchIT.java b/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/library/elasticsearch/bulk/ElasticSearchIT.java
index bbe471b178..e60e46c326 100644
--- a/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/library/elasticsearch/bulk/ElasticSearchIT.java
+++ b/oap-server/server-library/library-client/src/test/java/org/apache/skywalking/library/elasticsearch/bulk/ElasticSearchIT.java
@@ -51,6 +51,7 @@ import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
@@ -393,6 +394,40 @@ public class ElasticSearchIT {
server.stop();
}
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("versions")
+ public void scrollOverSeveralIndices(final ElasticsearchContainer server,
+ final String namespace) {
+ server.start();
+
+ final ElasticSearchClient client = new ElasticSearchClient(
+ moduleManager,
+ server.getHttpHostAddress(),
+ "http", "", "", "", "", "test", "test",
+ indexNameConverter(namespace), 500, 6000,
+ 0, 15
+ );
+ client.connect();
+
+ final String indexName = "test_scroll_over_several_indices";
+ client.forceInsert(indexName + "-20200807", "first", ImmutableMap.of("user", "sw", "day", "20200807"));
+ client.forceInsert(indexName + "-20200810", "second", ImmutableMap.of("user", "sw", "day", "20200810"));
+
+ // The middle day has no index. Each name gets the namespace, and the missing one is skipped.
+ final var scroller = ElasticSearchScroller
+ .<String>builder()
+ .client(client)
+ .search(Search.builder().size(10).query(Query.term("user", "sw")).build())
+ .indices(new String[] {indexName + "-20200807", indexName + "-20200809", indexName + "-20200810"})
+ .resultConverter(hit -> (String) hit.getSource().get("day"))
+ .build();
+ Assertions.assertEquals(
+ new HashSet<>(Arrays.asList("20200807", "20200810")), new HashSet<>(scroller.scroll()));
+
+ client.shutdown();
+ server.stop();
+ }
+
private static Function<String, String> indexNameConverter(String namespace) {
return indexName -> {
if (StringUtil.isNotEmpty(namespace)) {
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 88a955490b..df41f89a83 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
@@ -82,7 +82,7 @@ import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.Histor
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.StorageEsDAO;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.StorageEsInstaller;
-import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeSeriesUtils;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.cache.NetworkAddressAliasEsDAO;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.AggregationQueryEsDAO;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.AlarmQueryEsDAO;
@@ -166,13 +166,7 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider {
} else {
config.setNamespace(config.getNamespace().toLowerCase());
}
- if (config.getDayStep() > 1) {
- TimeSeriesUtils.setDAY_STEP(config.getDayStep());
- TimeSeriesUtils.setSUPER_DATASET_DAY_STEP(config.getDayStep());
- }
- if (config.getSuperDatasetDayStep() > 0) {
- TimeSeriesUtils.setSUPER_DATASET_DAY_STEP(config.getSuperDatasetDayStep());
- }
+ IndexNames.configure(config.getDayStep(), config.getSuperDatasetDayStep());
if (!StringUtil.isEmpty(config.getSecretsManagementFile())) {
MultipleFilesChangeMonitor monitor = new MultipleFilesChangeMonitor(
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/EsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/EsDAO.java
index e070eccb6b..0a7cc6e0b5 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/EsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/EsDAO.java
@@ -24,7 +24,6 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.function.Supplier;
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.Documents;
@@ -42,17 +41,17 @@ public abstract class EsDAO extends AbstractDAO<ElasticSearchClient> {
super(client);
}
- protected SearchResponse searchDebuggable(Supplier<String[]> indices, Search search) {
+ protected SearchResponse searchDebuggable(final String[] indexNames, final Search search) {
DebuggingTraceContext traceContext = DebuggingTraceContext.TRACE_CONTEXT.get();
DebuggingSpan span = null;
try {
StringBuilder builder = new StringBuilder();
if (traceContext != null) {
span = traceContext.createSpan("Query Elasticsearch: search");
- builder.append("Condition: ").append("indices: ").append(Arrays.toString(indices.get()));
+ builder.append("Condition: ").append("indices: ").append(Arrays.toString(indexNames));
span.setMsg(builder.toString());
}
- SearchResponse response = getClient().search(indices, search);
+ SearchResponse response = getClient().search(indexNames, search);
if (traceContext != null && traceContext.isDumpStorageRsp()) {
builder.append("\n").append(" Response: ").append(new Gson().toJson(response));
span.setMsg(builder.toString());
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/HistoryDeleteEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/HistoryDeleteEsDAO.java
index ebafa6451f..5147ed9d06 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/HistoryDeleteEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/HistoryDeleteEsDAO.java
@@ -64,7 +64,7 @@ public class HistoryDeleteEsDAO extends EsDAO implements IHistoryDeleteDAO {
return;
}
- String latestIndex = TimeSeriesUtils.latestWriteIndexName(model);
+ String latestIndex = IndexNames.latestIndex(IndexFamily.of(model), System.currentTimeMillis());
if (!client.isExistsIndex(latestIndex)) {
try {
client.createIndex(latestIndex);
@@ -91,7 +91,7 @@ public class HistoryDeleteEsDAO extends EsDAO implements IHistoryDeleteDAO {
List<String> prepareDeleteIndexes = new ArrayList<>();
for (String index : indices) {
- long timeSeries = TimeSeriesUtils.isolateTimeFromIndexName(index);
+ long timeSeries = IndexNames.indexStartDay(index);
if (deadline >= timeSeries) {
prepareDeleteIndexes.add(index);
}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexController.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexController.java
index 3dfd46a130..6e3cce0f96 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexController.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexController.java
@@ -32,9 +32,9 @@ import org.apache.skywalking.oap.server.library.util.CollectionUtils;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* The metrics data, that generated by OAL or MAL, would be partitioned to storage by the functions of the OAL or MAL.
@@ -136,13 +136,16 @@ public enum IndexController {
public static class LogicIndicesRegister {
/**
- * The relations of the logic table and the physical table.
+ * The relations of the logic table and the physical table. Runtime rules register models after boot, so
+ * the maps here are read while they are written.
*/
- private static final Map<String, String> LOGIC_INDICES_CATALOG = new HashMap<>();
+ private static final Map<String, String> LOGIC_INDICES_CATALOG = new ConcurrentHashMap<>();
- private static final Map<String/*physical index name*/, Map<String/*column name*/, ModelColumn>> PHYSICAL_INDICES_COLUMNS = new HashMap<>();
+ private static final Map<String/*physical index name*/, Map<String/*column name*/, ModelColumn>> PHYSICAL_INDICES_COLUMNS = new ConcurrentHashMap<>();
- private static final Map<String/*logic index name*/, Map<String/*column name*/, String/*alias*/>> LEGACY_COLUMNS_NAMES = new HashMap<>();
+ private static final Map<String/*logic index name*/, Map<String/*column name*/, String/*alias*/>> LEGACY_COLUMNS_NAMES = new ConcurrentHashMap<>();
+
+ private static final Map<String/*physical index name*/, IndexRolling> PHYSICAL_INDICES_ROLLING = new ConcurrentHashMap<>();
/**
* The metric table name in aggregation physical storage.
@@ -160,7 +163,8 @@ public enum IndexController {
public static final String MANAGEMENT_TABLE_NAME = "management_table";
public static String getPhysicalTableName(String logicName) {
- final String catalogEntry = LOGIC_INDICES_CATALOG.get(logicName);
+ // A null name resolves to null, as it did before the catalog became concurrent.
+ final String catalogEntry = logicName == null ? null : LOGIC_INDICES_CATALOG.get(logicName);
if (catalogEntry != null) {
return catalogEntry;
}
@@ -173,10 +177,24 @@ public enum IndexController {
return foreign != null ? METRICS_LOGIC_TABLE_NAME : logicName;
}
- public static void registerRelation(Model model, String physicalName) {
- LOGIC_INDICES_CATALOG.put(model.getName(), physicalName);
+ /**
+ * Registers which physical table a model lives in. A model registered again, or moved to another
+ * table by a runtime rule reshape, replaces its earlier relation. The physical table keeps its columns
+ * and rolling after its models are removed, as its indices stay.
+ *
+ * @throws IllegalArgumentException when the model's columns or rolling conflict with the physical
+ * table's
+ */
+ public static synchronized void registerRelation(final Model model, final String physicalName) {
+ final IndexRolling rolling = IndexRolling.of(model);
+ final IndexRolling registeredRolling = PHYSICAL_INDICES_ROLLING.putIfAbsent(physicalName, rolling);
+ if (registeredRolling != null && registeredRolling != rolling) {
+ throw new IllegalArgumentException(
+ model.getName() + " is rolled by " + rolling + ", but index " + physicalName
+ + " is rolled by " + registeredRolling);
+ }
Map<String, ModelColumn> columns = PHYSICAL_INDICES_COLUMNS.computeIfAbsent(
- physicalName, v -> new HashMap<>());
+ physicalName, v -> new ConcurrentHashMap<>());
if (!IndexController.INSTANCE.logicSharding || !model.isTimeSeries()) {
model.getColumns().forEach(modelColumn -> {
String columnName = modelColumn.getColumnName().getName();
@@ -192,7 +210,7 @@ public enum IndexController {
String legacyName = modelColumn.getElasticSearchExtension().getLegacyColumnName();
if (!Strings.isNullOrEmpty(legacyName)) {
Map<String, String> legacyNames = LEGACY_COLUMNS_NAMES.computeIfAbsent(
- model.getName(), v -> new HashMap<>());
+ model.getName(), v -> new ConcurrentHashMap<>());
legacyNames.put(modelColumn.getColumnName().getName(), legacyName);
columnName = legacyName;
}
@@ -203,7 +221,15 @@ public enum IndexController {
}
});
}
+ // Published last, so a reader never resolves the model to a table whose checks failed.
+ LOGIC_INDICES_CATALOG.put(model.getName(), physicalName);
+ }
+ /**
+ * @return how the physical table's indices are rolled, or null when no model registered the table
+ */
+ public static IndexRolling getRolling(final String physicalName) {
+ return physicalName == null ? null : PHYSICAL_INDICES_ROLLING.get(physicalName);
}
public static boolean isMergedTable(String logicName) {
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexFamily.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexFamily.java
new file mode 100644
index 0000000000..12504a0a09
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexFamily.java
@@ -0,0 +1,61 @@
+/*
+ * 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 lombok.AccessLevel;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+
+/**
+ * A physical table and how its dated indices are rolled. The physical name is also the alias every dated
+ * index of the table carries, see {@link StorageEsInstaller}.
+ */
+@Getter
+@EqualsAndHashCode
+@ToString
+@RequiredArgsConstructor(access = AccessLevel.PACKAGE)
+public final class IndexFamily {
+ /**
+ * The merged metrics index a foreign metric of the inspect API lives in. It is used where the read names
+ * the index itself instead of resolving the metric through the inspect overlay.
+ */
+ public static final IndexFamily FOREIGN_METRICS =
+ new IndexFamily(IndexController.METRICS_LOGIC_TABLE_NAME, IndexRolling.DAY_STEP);
+
+ private final String physicalName;
+ private final IndexRolling rolling;
+
+ public static IndexFamily of(final Model model) {
+ return new IndexFamily(IndexController.INSTANCE.getTableName(model), IndexRolling.of(model));
+ }
+
+ /**
+ * Resolves a logical name the way {@link IndexController.LogicIndicesRegister#getPhysicalTableName} does.
+ * A table no model registered is rolled by {@code dayStep}: only metric reads pass names that may be
+ * unregistered (a foreign metric, a metric added by a runtime rule), and metrics are rolled by
+ * {@code dayStep}.
+ */
+ public static IndexFamily of(final String logicName) {
+ final String physicalName = IndexController.LogicIndicesRegister.getPhysicalTableName(logicName);
+ final IndexRolling registered = IndexController.LogicIndicesRegister.getRolling(physicalName);
+ return new IndexFamily(physicalName, registered != null ? registered : IndexRolling.DAY_STEP);
+ }
+}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexNames.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexNames.java
new file mode 100644
index 0000000000..0ff1313475
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexNames.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.storage.plugin.elasticsearch.base;
+
+import java.util.ArrayList;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.skywalking.oap.server.core.Const;
+import org.apache.skywalking.oap.server.core.UnexpectedException;
+import org.apache.skywalking.oap.server.core.analysis.DownSampling;
+import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
+import org.apache.skywalking.oap.server.core.query.enumeration.Step;
+import org.apache.skywalking.oap.server.core.query.input.Duration;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+import org.joda.time.DateTime;
+import org.joda.time.Days;
+import org.joda.time.format.DateTimeFormat;
+import org.joda.time.format.DateTimeFormatter;
+
+/**
+ * Builds the names of Elasticsearch indices. A time series table is stored in dated indices,
+ * {@code <physical table>-<yyyyMMdd>}, each holding a group of days. The group size, the day step, comes from the
+ * table's {@link IndexRolling}, so a caller only says which table and which time.
+ */
+public final class IndexNames {
+ private static final DateTimeFormatter TIME_BUCKET_FORMATTER = DateTimeFormat.forPattern("yyyyMMdd");
+ /**
+ * Day groups are counted from this day, so a day lands in the same index whenever the OAP starts.
+ */
+ private static final DateTime DAY_ONE = TIME_BUCKET_FORMATTER.parseDateTime("20000101");
+ private static int DAY_STEP = 1;
+ private static int SUPER_DATASET_DAY_STEP = 1;
+
+ private IndexNames() {
+ }
+
+ /**
+ * @param dayStep days per index of metrics and records; a value below 2 means 1
+ * @param superDatasetDayStep days per index of super dataset records; a value below 1 means the day step
+ */
+ public static void configure(final int dayStep, final int superDatasetDayStep) {
+ DAY_STEP = dayStep > 1 ? dayStep : 1;
+ SUPER_DATASET_DAY_STEP = superDatasetDayStep > 0 ? superDatasetDayStep : DAY_STEP;
+ }
+
+ /**
+ * @return the index a document of the model with the given time bucket is written to
+ */
+ public static String writeIndex(final Model model, final long timeBucket) {
+ final IndexFamily family = IndexFamily.of(model);
+ if (family.getRolling() == IndexRolling.NONE) {
+ return family.getPhysicalName();
+ }
+ return datedIndex(family, toDay(timeBucket, model.getDownsampling()));
+ }
+
+ /**
+ * @return the index the point of a time series is read from
+ */
+ public static String pointIndex(final IndexFamily family, final long pointOfTB, final Step step) {
+ if (StringUtil.isBlank(family.getPhysicalName()) || pointOfTB <= 0) {
+ throw new IllegalArgumentException(
+ "Arguments [tableName]: " + family.getPhysicalName() + " can not be blank and [pointOfTB]: "
+ + pointOfTB + " can not <= 0");
+ }
+ return datedIndex(family, toDay(pointOfTB, step));
+ }
+
+ /**
+ * @return the indices holding the duration's days, in day order, or the table's alias when there is no
+ * duration
+ */
+ public static String[] rangeIndices(final IndexFamily family, @Nullable final Duration duration) {
+ if (duration == null) {
+ return new String[] {family.getPhysicalName()};
+ }
+ return rangeIndices(family, duration.getStartTimeBucketInSec(), duration.getEndTimeBucketInSec());
+ }
+
+ /**
+ * @return the indices holding the days between the two timestamps. The caller decides what a timestamp
+ * that is not positive means.
+ */
+ public static String[] rangeIndicesBetweenTimestamps(final IndexFamily family,
+ final long startMillis,
+ final long endMillis) {
+ return rangeIndices(
+ family, TimeBucket.getRecordTimeBucket(startMillis), TimeBucket.getRecordTimeBucket(endMillis));
+ }
+
+ /**
+ * @return today's index of the time series table
+ */
+ public static String latestIndex(final IndexFamily family, final long currentTimeMillis) {
+ return datedIndex(family, TimeBucket.getTimeBucket(currentTimeMillis, DownSampling.Day));
+ }
+
+ /**
+ * @param concreteIndexName a dated index name, with or without the namespace
+ * @return the first day the index holds
+ */
+ public static long indexStartDay(final String concreteIndexName) {
+ return Long.parseLong(concreteIndexName.substring(concreteIndexName.lastIndexOf(Const.LINE) + 1));
+ }
+
+ /**
+ * A zero bucket, and an end day before the start day, mean the whole table, as the alias.
+ */
+ private static String[] rangeIndices(final IndexFamily family, final long startSecondTB, final long endSecondTB) {
+ if (startSecondTB == 0 || endSecondTB == 0) {
+ return new String[] {family.getPhysicalName()};
+ }
+ final DateTime startDay = TIME_BUCKET_FORMATTER.parseDateTime(startSecondTB / 1000000 + "");
+ final DateTime endDay = TIME_BUCKET_FORMATTER.parseDateTime(endSecondTB / 1000000 + "");
+ final List<DateTime> days = new ArrayList<>(16);
+ for (int i = 0; i <= Days.daysBetween(startDay, endDay).getDays(); i++) {
+ days.add(startDay.plusDays(i));
+ }
+ if (days.isEmpty()) {
+ return new String[] {family.getPhysicalName()};
+ }
+ final int dayStep = dayStepOf(family);
+ return days.stream()
+ .map(day -> family.getPhysicalName() + Const.LINE + rollDay(day, dayStep))
+ .distinct()
+ .toArray(String[]::new);
+ }
+
+ private static String datedIndex(final IndexFamily family, final long day) {
+ return family.getPhysicalName() + Const.LINE + rollDay(day, dayStepOf(family));
+ }
+
+ private static int dayStepOf(final IndexFamily family) {
+ switch (family.getRolling()) {
+ case DAY_STEP:
+ return DAY_STEP;
+ case SUPER_DATASET_DAY_STEP:
+ return SUPER_DATASET_DAY_STEP;
+ default:
+ throw new IllegalArgumentException("Index " + family.getPhysicalName() + " has no dated indices");
+ }
+ }
+
+ /**
+ * @return the first day of the group the day falls in. With a day step of 11, 20000105 is in the group of
+ * 20000101, 20000115 in the group of 20000112, and 20000123 starts its own group.
+ */
+ private static long rollDay(final long day, final int dayStep) {
+ if (dayStep <= 1) {
+ return day;
+ }
+ return rollDay(TIME_BUCKET_FORMATTER.parseDateTime(String.valueOf(day)), dayStep);
+ }
+
+ private static long rollDay(final DateTime day, final int dayStep) {
+ if (dayStep <= 1) {
+ return Long.parseLong(day.toString(TIME_BUCKET_FORMATTER));
+ }
+ final int groupOffset = Days.daysBetween(DAY_ONE, day).getDays() % dayStep;
+ return Long.parseLong(day.minusDays(groupOffset).toString(TIME_BUCKET_FORMATTER));
+ }
+
+ private static long toDay(final long timeBucket, final DownSampling downSampling) {
+ switch (downSampling) {
+ case Second:
+ return timeBucket / 1000000;
+ case Minute:
+ return timeBucket / 10000;
+ case Hour:
+ return timeBucket / 100;
+ case Day:
+ return timeBucket;
+ default:
+ throw new UnexpectedException("Unexpected down sampling value, " + downSampling);
+ }
+ }
+
+ private static long toDay(final long pointOfTB, final Step step) {
+ switch (step) {
+ case SECOND:
+ return pointOfTB / 1000000;
+ case MINUTE:
+ return pointOfTB / 10000;
+ case HOUR:
+ return pointOfTB / 100;
+ case DAY:
+ return pointOfTB;
+ default:
+ throw new UnexpectedException("Unexpected step value, " + step);
+ }
+ }
+}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexRolling.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexRolling.java
new file mode 100644
index 0000000000..53fcf4dc73
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexRolling.java
@@ -0,0 +1,49 @@
+/*
+ * 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.core.storage.model.Model;
+
+/**
+ * How the dated indices of one physical table are rolled. Every model sharing a physical table has the same
+ * rolling, see {@link IndexController.LogicIndicesRegister#registerRelation}.
+ */
+public enum IndexRolling {
+ /**
+ * Management data: one concrete index, no date suffix.
+ */
+ NONE,
+ /**
+ * Metrics of every downsampling, and records that are not super datasets, rolled by {@code dayStep}.
+ */
+ DAY_STEP,
+ /**
+ * Super dataset records (traces, logs, spans), rolled by {@code superDatasetDayStep}.
+ */
+ SUPER_DATASET_DAY_STEP;
+
+ public static IndexRolling of(final Model model) {
+ if (!model.isTimeSeries()) {
+ return NONE;
+ }
+ if (model.isRecord() && model.isSuperDataset()) {
+ return SUPER_DATASET_DAY_STEP;
+ }
+ return DAY_STEP;
+ }
+}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/MetricsEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/MetricsEsDAO.java
index 75027a6fd5..bb99694153 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/MetricsEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/MetricsEsDAO.java
@@ -58,7 +58,7 @@ public class MetricsEsDAO extends EsDAO implements IMetricsDAO {
if (model.isTimeRelativeID()) {
metrics.forEach(metric -> {
// Try to use with timestamp index name(write index),
- String indexName = TimeSeriesUtils.writeIndexName(model, metric.getTimeBucket());
+ String indexName = IndexNames.writeIndex(model, metric.getTimeBucket());
groupIndices.computeIfAbsent(indexName, v -> new ArrayList<>()).add(metric);
});
@@ -102,7 +102,7 @@ public class MetricsEsDAO extends EsDAO implements IMetricsDAO {
final ElasticSearchConverter.ToStorage toStorage = new ElasticSearchConverter.ToStorage(model.getName());
storageBuilder.entity2Storage(metrics, toStorage);
Map<String, Object> builder = IndexController.INSTANCE.appendTableColumn(model, toStorage.obtain());
- String modelName = TimeSeriesUtils.writeIndexName(model, metrics.getTimeBucket());
+ String modelName = IndexNames.writeIndex(model, metrics.getTimeBucket());
String id = IndexController.INSTANCE.generateDocId(model, metrics.id().build());
return new MetricIndexRequestWrapper(getClient().prepareInsert(modelName, id, builder), callback);
}
@@ -113,7 +113,7 @@ public class MetricsEsDAO extends EsDAO implements IMetricsDAO {
storageBuilder.entity2Storage(metrics, toStorage);
Map<String, Object> builder =
IndexController.INSTANCE.appendTableColumn(model, toStorage.obtain());
- String modelName = TimeSeriesUtils.writeIndexName(model, metrics.getTimeBucket());
+ String modelName = IndexNames.writeIndex(model, metrics.getTimeBucket());
String id = IndexController.INSTANCE.generateDocId(model, metrics.id().build());
return new MetricIndexUpdateWrapper(getClient().prepareUpdate(modelName, id, builder), callback);
}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/NoneStreamEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/NoneStreamEsDAO.java
index 7ec21a0ccb..eeddf62b0a 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/NoneStreamEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/NoneStreamEsDAO.java
@@ -44,7 +44,7 @@ public class NoneStreamEsDAO extends EsDAO implements INoneStreamDAO {
storageBuilder.entity2Storage(noneStream, toStorage);
Map<String, Object> builder =
IndexController.INSTANCE.appendTableColumn(model, toStorage.obtain());
- String modelName = TimeSeriesUtils.writeIndexName(model, noneStream.getTimeBucket());
+ String modelName = IndexNames.writeIndex(model, noneStream.getTimeBucket());
String id = IndexController.INSTANCE.generateDocId(model, noneStream.id().build());
getClient().forceInsert(modelName, id, builder);
}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RecordEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RecordEsDAO.java
index 5d428efd74..bed5cd4514 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RecordEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RecordEsDAO.java
@@ -43,7 +43,7 @@ public class RecordEsDAO extends EsDAO implements IRecordDAO {
final ElasticSearchConverter.ToStorage toStorage = new ElasticSearchConverter.ToStorage(model.getName());
storageBuilder.entity2Storage(record, toStorage);
Map<String, Object> builder = IndexController.INSTANCE.appendTableColumn(model, toStorage.obtain());
- String modelName = TimeSeriesUtils.writeIndexName(model, record.getTimeBucket());
+ String modelName = IndexNames.writeIndex(model, record.getTimeBucket());
String id = IndexController.INSTANCE.generateDocId(model, record.id().build());
Optional<String> routingValue = RoutingUtils.getRoutingValue(model, toStorage);
return getClient().prepareInsert(modelName, id, routingValue, builder);
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java
index c0cc609d87..f0fa766f12 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java
@@ -147,6 +147,16 @@ public class StorageEsInstaller extends ModelInstaller {
return installInfo;
}
+ /**
+ * Registers the model's physical table, the local state reads translate a model name with. A node that
+ * applies a model without a schema change, such as a peer reinstalling a bundled rule, never calls
+ * {@link #isExists}, which registers it on every other path.
+ */
+ @Override
+ protected void populateLocalCacheOnly(final Model model, final StorageManipulationOpt opt) {
+ IndexController.LogicIndicesRegister.registerRelation(model, IndexController.INSTANCE.getTableName(model));
+ }
+
@Override
public void createTable(Model model) throws StorageException {
if (model.isTimeSeries()) {
@@ -198,7 +208,7 @@ public class StorageEsInstaller extends ModelInstaller {
String tableName = IndexController.INSTANCE.getTableName(model);
Map<String, Object> settings = createSetting(model);
Mappings mapping = createMapping(model);
- String indexName = TimeSeriesUtils.latestWriteIndexName(model);
+ String indexName = IndexNames.latestIndex(IndexFamily.of(model), System.currentTimeMillis());
try {
boolean shouldUpdateTemplate = !esClient.isExistsTemplate(tableName);
shouldUpdateTemplate = shouldUpdateTemplate
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeRangeIndexNameGenerator.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeRangeIndexNameGenerator.java
deleted file mode 100644
index 70ca4aeedb..0000000000
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeRangeIndexNameGenerator.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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 java.util.function.Supplier;
-
-/**
- * the time range index maker works for super size dataset
- */
-public class TimeRangeIndexNameGenerator implements Supplier<String[]> {
-
- private final long startSecondTB;
- private final long endSecondTB;
- private final String indexName;
-
- public TimeRangeIndexNameGenerator(final String indexName,
- final long startSecondTB,
- final long endSecondTB) {
- this.startSecondTB = startSecondTB;
- this.endSecondTB = endSecondTB;
- this.indexName = indexName;
- }
-
- @Override
- public String[] get() {
- 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
deleted file mode 100644
index fa29f67b67..0000000000
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtils.java
+++ /dev/null
@@ -1,189 +0,0 @@
-/*
- * 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 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;
-import org.apache.skywalking.oap.server.core.analysis.DownSampling;
-import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
-import org.apache.skywalking.oap.server.core.query.enumeration.Step;
-import org.apache.skywalking.oap.server.core.storage.model.Model;
-import org.apache.skywalking.oap.server.library.util.StringUtil;
-import org.joda.time.DateTime;
-import org.joda.time.Days;
-import org.joda.time.format.DateTimeFormat;
-import org.joda.time.format.DateTimeFormatter;
-
-/**
- * TimeSeriesUtils sets up and splits the time suffix of index name.
- */
-public class TimeSeriesUtils {
- private static DateTimeFormatter TIME_BUCKET_FORMATTER = DateTimeFormat.forPattern("yyyyMMdd");
- /**
- * We are far from the first day of 2000, so we set it as the day one to make sure the index based on {@link
- * #DAY_STEP} is consistently no matter whenever the OAP starts up.
- */
- private static final DateTime DAY_ONE = TIME_BUCKET_FORMATTER.parseDateTime("20000101");
- @Setter
- private static int DAY_STEP = 1;
- @Setter
- private static int SUPER_DATASET_DAY_STEP = 1;
-
- /**
- * @return formatted latest index name, based on current timestamp.
- */
- public static String latestWriteIndexName(Model model) {
- String tableName = IndexController.INSTANCE.getTableName(model);
- long dayTimeBucket = TimeBucket.getTimeBucket(System.currentTimeMillis(), DownSampling.Day);
- if (model.isRecord() && model.isSuperDataset()) {
- return tableName + Const.LINE + compressTimeBucket(dayTimeBucket, SUPER_DATASET_DAY_STEP);
- }
- return tableName + Const.LINE + compressTimeBucket(dayTimeBucket, DAY_STEP);
- }
-
- /**
- * @return Concrete index name for super dataset index
- */
- public static String[] superDatasetIndexNames(String indexName, long startSecondTB, long endSecondTB) {
- return timeRangeIndexNames(indexName, startSecondTB, endSecondTB, SUPER_DATASET_DAY_STEP);
- }
-
- /**
- * @return Concrete index names for metrics index
- */
- public static String[] metricsIndexNames(String indexName, long startSecondTB, long endSecondTB) {
- return timeRangeIndexNames(indexName, startSecondTB, endSecondTB, DAY_STEP);
- }
-
- private static String[] timeRangeIndexNames(String indexName, long startSecondTB, long endSecondTB, int dayStep) {
- 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<DateTime> 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, dayStep))
- .distinct()
- .toArray(String[]::new);
- }
- }
-
- public static String queryIndexName(String tableName,
- long pointOfTB,
- Step step,
- boolean isRecord,
- boolean isSuperDataSet) {
- if (StringUtil.isBlank(tableName) || pointOfTB <= 0) {
- throw new IllegalArgumentException(
- "Arguments [tableName]: " + tableName + " can not be blank and [pointOfTB]: " + pointOfTB + " can not <= 0");
- }
- if (isRecord && isSuperDataSet) {
- return tableName + Const.LINE + compressTimeBucket(pointOfTB / 1000000, SUPER_DATASET_DAY_STEP);
- }
- switch (step) {
- case DAY:
- return tableName + Const.LINE + compressTimeBucket(pointOfTB, DAY_STEP);
- case HOUR:
- return tableName + Const.LINE + compressTimeBucket(pointOfTB / 100, DAY_STEP);
- case MINUTE:
- return tableName + Const.LINE + compressTimeBucket(pointOfTB / 10000, DAY_STEP);
- case SECOND:
- return tableName + Const.LINE + compressTimeBucket(pointOfTB / 1000000, DAY_STEP);
- }
-
- throw new UnexpectedException("Failed to get the index name from tableName:" + tableName + ", pointOfTB:" + pointOfTB + ", step:" + step.name());
- }
-
- /**
- * @return index name based on model definition and given time bucket.
- */
- static String writeIndexName(Model model, long timeBucket) {
- String tableName = IndexController.INSTANCE.getTableName(model);
- if (model.isRecord() && model.isSuperDataset()) {
- return tableName + Const.LINE + compressTimeBucket(timeBucket / 1000000, SUPER_DATASET_DAY_STEP);
- } else {
- switch (model.getDownsampling()) {
- case None:
- return tableName;
- case Hour:
- return tableName + Const.LINE + compressTimeBucket(timeBucket / 100, DAY_STEP);
- case Minute:
- return tableName + Const.LINE + compressTimeBucket(timeBucket / 10000, DAY_STEP);
- case Day:
- return tableName + Const.LINE + compressTimeBucket(timeBucket, DAY_STEP);
- case Second:
- return tableName + Const.LINE + compressTimeBucket(timeBucket / 1000000, DAY_STEP);
- default:
- throw new UnexpectedException("Unexpected down sampling value, " + model.getDownsampling());
- }
- }
- }
-
- /**
- * @return the index represented time, which is included in the index name.
- */
- static long isolateTimeFromIndexName(String indexName) {
- return Long.parseLong(indexName.substring(indexName.lastIndexOf(Const.LINE) + 1));
- }
-
- /**
- * Follow the dayStep to re-format the time bucket literal long value.
- *
- * Such as, in dayStep == 11,
- *
- * 20000105 re-formatted time bucket is 20000101, 20000115 re-formatted time bucket is 20000112, 20000123
- * re-formatted time bucket is 20000123
- */
- static long compressTimeBucket(long timeBucket, int dayStep) {
- if (dayStep > 1) {
- DateTime time = TIME_BUCKET_FORMATTER.parseDateTime("" + timeBucket);
- 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 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/AIAgentConversationQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AIAgentConversationQueryEsDAO.java
index 2779244639..c8f07a042b 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AIAgentConversationQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AIAgentConversationQueryEsDAO.java
@@ -41,7 +41,8 @@ import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSear
import org.apache.skywalking.oap.server.library.util.StringUtil;
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.TimeRangeIndexNameGenerator;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexFamily;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
/**
* Both models are super datasets, so each has its own index family; the queries are term filters on the
@@ -113,10 +114,7 @@ public class AIAgentConversationQueryEsDAO extends EsDAO implements IAIAgentConv
}
}
final SearchResponse response = searchDebuggable(
- new TimeRangeIndexNameGenerator(
- IndexController.LogicIndicesRegister.getPhysicalTableName(AIAgentSessionFlowRecord.INDEX_NAME),
- startSecondTB, endSecondTB
- ),
+ IndexNames.rangeIndices(IndexFamily.of(AIAgentSessionFlowRecord.INDEX_NAME), duration),
search.build()
);
return rounds(response, includeBody);
@@ -141,8 +139,7 @@ public class AIAgentConversationQueryEsDAO extends EsDAO implements IAIAgentConv
.size(1);
search.source(AIAgentSessionFlowRecord.ROUND);
final SearchResponse response = searchDebuggable(
- new TimeRangeIndexNameGenerator(
- IndexController.LogicIndicesRegister.getPhysicalTableName(AIAgentSessionFlowRecord.INDEX_NAME), 0, 0),
+ new String[] {IndexController.LogicIndicesRegister.getPhysicalTableName(AIAgentSessionFlowRecord.INDEX_NAME)},
search.build()
);
long head = 0;
@@ -178,8 +175,7 @@ public class AIAgentConversationQueryEsDAO extends EsDAO implements IAIAgentConv
.size(RESULT_WINDOW);
// every index of the family: the rounds of a long conversation span days
final SearchResponse response = searchDebuggable(
- new TimeRangeIndexNameGenerator(
- IndexController.LogicIndicesRegister.getPhysicalTableName(AIAgentSessionFlowRecord.INDEX_NAME), 0, 0),
+ new String[] {IndexController.LogicIndicesRegister.getPhysicalTableName(AIAgentSessionFlowRecord.INDEX_NAME)},
search.build()
);
return rounds(response, true);
@@ -243,19 +239,15 @@ public class AIAgentConversationQueryEsDAO extends EsDAO implements IAIAgentConv
query.must(Query.range(AIAgentSessionDataRecord.TIMESTAMP).gte(fromTimestamp).lte(toTimestamp));
// a range with no start names every index of the family, as the round reads do; expanded day by day it
// would name every day since 1970, and the request line would be longer than Elasticsearch accepts
- final long startSecondTB = fromTimestamp <= 0 ? 0 : TimeBucket.getRecordTimeBucket(fromTimestamp);
- final long endSecondTB = fromTimestamp <= 0 ? 0 : TimeBucket.getRecordTimeBucket(toTimestamp);
+ final String[] indices = fromTimestamp <= 0
+ ? new String[] {IndexController.LogicIndicesRegister.getPhysicalTableName(AIAgentSessionDataRecord.INDEX_NAME)}
+ : IndexNames.rangeIndicesBetweenTimestamps(
+ IndexFamily.of(AIAgentSessionDataRecord.INDEX_NAME), fromTimestamp, toTimestamp);
final SearchBuilder search = Search.builder()
.query(query)
.sort(AIAgentSessionDataRecord.SEQ, Sort.Order.ASC)
.size(RESULT_WINDOW);
- final SearchResponse response = searchDebuggable(
- new TimeRangeIndexNameGenerator(
- IndexController.LogicIndicesRegister.getPhysicalTableName(AIAgentSessionDataRecord.INDEX_NAME),
- startSecondTB, endSecondTB
- ),
- search.build()
- );
+ final SearchResponse response = searchDebuggable(indices, search.build());
final List<AIAgentSessionDataRecord> files = new ArrayList<>();
for (final SearchHit hit : response.getHits().getHits()) {
final Map<String, Object> source = hit.getSource();
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAO.java
index f00cabac23..7ab113e2cd 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAO.java
@@ -40,7 +40,8 @@ import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSear
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.IndexController;
-import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeSeriesUtils;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexFamily;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
import java.util.ArrayList;
import java.util.List;
@@ -116,10 +117,8 @@ public class AggregationQueryEsDAO extends EsDAO implements IAggregationQueryDAO
.collectMode(TermsAggregationBuilder.CollectMode.BREADTH_FIRST)
.build());
- final SearchResponse response = searchDebuggable(() -> TimeSeriesUtils.metricsIndexNames(
- IndexController.LogicIndicesRegister.getPhysicalTableName(condition.getName()),
- duration.getStartTimeBucketInSec(),
- duration.getEndTimeBucketInSec()), search.build());
+ final SearchResponse response = searchDebuggable(
+ IndexNames.rangeIndices(IndexFamily.of(condition.getName()), duration), search.build());
final List<SelectedRecord> topNList = new ArrayList<>();
if (Objects.nonNull(response.getAggregations())) {
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/GenAIEvaluationRecordQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/GenAIEvaluationRecordQueryEsDAO.java
index 6e52324168..34b54b1b4a 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/GenAIEvaluationRecordQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/GenAIEvaluationRecordQueryEsDAO.java
@@ -39,7 +39,8 @@ import org.apache.skywalking.oap.server.core.storage.query.IGenAIEvaluationRecor
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
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.TimeRangeIndexNameGenerator;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexFamily;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
import static java.util.Objects.nonNull;
import static org.apache.skywalking.oap.server.library.util.StringUtil.isNotEmpty;
@@ -134,11 +135,8 @@ public class GenAIEvaluationRecordQueryEsDAO extends EsDAO implements IGenAIEval
.size(limit)
.from(from);
- final SearchResponse response = searchDebuggable(new TimeRangeIndexNameGenerator(
- IndexController.LogicIndicesRegister.getPhysicalTableName(GenAIEvaluationRecord.INDEX_NAME),
- startSecondTB,
- endSecondTB
- ), search.build());
+ final SearchResponse response = searchDebuggable(
+ IndexNames.rangeIndices(IndexFamily.of(GenAIEvaluationRecord.INDEX_NAME), duration), search.build());
final GenAIEvaluationRecords records = new GenAIEvaluationRecords();
for (SearchHit searchHit : response.getHits().getHits()) {
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java
index f3359d6eff..2fe233e235 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java
@@ -44,8 +44,9 @@ import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSear
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.IndexController;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexFamily;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.MatchCNameBuilder;
-import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeRangeIndexNameGenerator;
import static java.util.Objects.nonNull;
import static org.apache.skywalking.oap.server.library.util.StringUtil.isNotEmpty;
@@ -145,11 +146,8 @@ public class LogQueryEsDAO extends EsDAO implements ILogQueryDAO {
.size(limit)
.from(from);
- SearchResponse response = searchDebuggable(new TimeRangeIndexNameGenerator(
- IndexController.LogicIndicesRegister.getPhysicalTableName(LogRecord.INDEX_NAME),
- startSecondTB,
- endSecondTB
- ), search.build());
+ SearchResponse response = searchDebuggable(
+ IndexNames.rangeIndices(IndexFamily.of(LogRecord.INDEX_NAME), duration), search.build());
Logs logs = new Logs();
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetricsQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetricsQueryEsDAO.java
index be641be9ec..518d600a5d 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetricsQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetricsQueryEsDAO.java
@@ -53,7 +53,8 @@ import org.apache.skywalking.oap.server.core.storage.query.IMetricsQueryDAO;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
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.TimeSeriesUtils;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexFamily;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
@@ -77,7 +78,7 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
}
final String realValueColumn = IndexController.LogicIndicesRegister.getPhysicalColumnName(condition.getName(), valueColumnName);
final boolean mergedTable = IndexController.LogicIndicesRegister.isMergedTable(condition.getName());
- final String tableName = IndexController.LogicIndicesRegister.getPhysicalTableName(condition.getName());
+ final IndexFamily family = IndexFamily.of(condition.getName());
final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
Map<String, List<String>> indexIdsGroup = new HashMap<>();
@@ -86,8 +87,7 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
if (mergedTable) {
id = IndexController.INSTANCE.generateDocId(condition.getName(), id);
}
- String indexName = TimeSeriesUtils.queryIndexName(
- tableName, pointOfTime.getPoint(), duration.getStep(), false, false);
+ String indexName = IndexNames.pointIndex(family, pointOfTime.getPoint(), duration.getStep());
indexIdsGroup.computeIfAbsent(indexName, v -> new ArrayList<>()).add(id);
return id;
}).collect(Collectors.toList());
@@ -139,10 +139,10 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
}
final String realValueColumn = IndexController.LogicIndicesRegister.getPhysicalColumnName(condition.getName(), valueColumnName);
final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
- final String tableName = IndexController.LogicIndicesRegister.getPhysicalTableName(condition.getName());
+ final IndexFamily family = IndexFamily.of(condition.getName());
Map<String, List<String>> indexIdsGroup = new HashMap<>();
- boolean aggregationMode = !tableName.equals(condition.getName());
+ boolean aggregationMode = !family.getPhysicalName().equals(condition.getName());
List<String> ids = new ArrayList<>(pointOfTimes.size());
pointOfTimes.forEach(pointOfTime -> {
String id = pointOfTime.id(condition.getEntity().buildId());
@@ -150,8 +150,7 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
id = IndexController.INSTANCE.generateDocId(condition.getName(), id);
}
ids.add(id);
- String indexName = TimeSeriesUtils.queryIndexName(
- tableName, pointOfTime.getPoint(), duration.getStep(), false, false);
+ String indexName = IndexNames.pointIndex(family, pointOfTime.getPoint(), duration.getStep());
indexIdsGroup.computeIfAbsent(indexName, v -> new ArrayList<>()).add(id);
});
@@ -190,10 +189,8 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
));
search.query(query);
}
- final SearchResponse response = getClient().search(() -> TimeSeriesUtils.metricsIndexNames(
- IndexController.LogicIndicesRegister.getPhysicalTableName(metricName),
- duration.getStartTimeBucketInSec(),
- duration.getEndTimeBucketInSec()), search.build());
+ final SearchResponse response = getClient().search(
+ IndexNames.rangeIndices(IndexFamily.of(metricName), duration), search.build());
final String realValueColumn = IndexController.LogicIndicesRegister.getPhysicalColumnName(metricName, valueColumnName);
final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
Map<String, DataTable> idMap = new HashMap<>();
@@ -223,7 +220,7 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
// valueType != null signals a foreign metric (not defined on this OAP). The value column
// is unused by ES entity enumeration; only the physical index + discriminator differ.
final boolean foreign = valueType != null;
- final String physicalIndex;
+ final IndexFamily family;
final boolean filterByDiscriminator;
if (foreign) {
// Resolve from running config, not the local registry. In the default merged mode
@@ -236,10 +233,10 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
+ "physical index is derived from the metric's stream class, which this OAP "
+ "does not have for " + metricName);
}
- physicalIndex = IndexController.METRICS_LOGIC_TABLE_NAME;
+ family = IndexFamily.FOREIGN_METRICS;
filterByDiscriminator = true;
} else {
- physicalIndex = IndexController.LogicIndicesRegister.getPhysicalTableName(metricName);
+ family = IndexFamily.of(metricName);
filterByDiscriminator = IndexController.LogicIndicesRegister.isMergedTable(metricName);
}
final SearchBuilder search = Search.builder().size(limit);
@@ -257,10 +254,7 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
));
}
search.query(query);
- final SearchResponse response = getClient().search(() -> TimeSeriesUtils.metricsIndexNames(
- physicalIndex,
- duration.getStartTimeBucketInSec(),
- duration.getEndTimeBucketInSec()), search.build());
+ final SearchResponse response = getClient().search(IndexNames.rangeIndices(family, duration), search.build());
// Top-N hits across the time range, dedup client-side on entity_id. LinkedHashSet
// preserves the time_bucket-DESC ordering ES returned.
final LinkedHashSet<String> entityIds = new LinkedHashSet<>();
@@ -279,11 +273,10 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
final Duration duration) {
final String realValueColumn = IndexController.LogicIndicesRegister.getPhysicalColumnName(condition.getName(), valueColumnName);
final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
- String tableName =
- IndexController.LogicIndicesRegister.getPhysicalTableName(condition.getName());
+ final IndexFamily family = IndexFamily.of(condition.getName());
Map<String, List<String>> indexIdsGroup = new HashMap<>();
- boolean aggregationMode = !tableName.equals(condition.getName());
+ boolean aggregationMode = !family.getPhysicalName().equals(condition.getName());
List<String> ids = new ArrayList<>(pointOfTimes.size());
pointOfTimes.forEach(pointOfTime -> {
String id = pointOfTime.id(condition.getEntity().buildId());
@@ -291,8 +284,7 @@ public class MetricsQueryEsDAO extends EsDAO implements IMetricsQueryDAO {
id = IndexController.INSTANCE.generateDocId(condition.getName(), id);
}
ids.add(id);
- String indexName = TimeSeriesUtils.queryIndexName(
- tableName, pointOfTime.getPoint(), duration.getStep(), false, false);
+ String indexName = IndexNames.pointIndex(family, pointOfTime.getPoint(), duration.getStep());
indexIdsGroup.computeIfAbsent(indexName, v -> new ArrayList<>()).add(id);
});
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TagAutoCompleteQueryDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TagAutoCompleteQueryDAO.java
index 3446624663..f99e7a4f84 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TagAutoCompleteQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TagAutoCompleteQueryDAO.java
@@ -40,9 +40,8 @@ import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.ElasticSearchConverter;
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.TimeSeriesUtils;
-
-import static java.util.Objects.nonNull;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexFamily;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
public class TagAutoCompleteQueryDAO extends EsDAO implements ITagAutoCompleteQueryDAO {
public TagAutoCompleteQueryDAO(ElasticSearchClient client) {
@@ -55,8 +54,6 @@ public class TagAutoCompleteQueryDAO extends EsDAO implements ITagAutoCompleteQu
final Duration duration) throws IOException {
// Tags combine records by day and es rolling index by day, need search the whole day
// just use the time to locate the physical indexes and ignore time conditions.
- final long startSecondTB = nonNull(duration) ? duration.getStartTimeBucketInSec() : 0;
- final long endSecondTB = nonNull(duration) ? duration.getEndTimeBucketInSec() : 0;
BoolQueryBuilder query = Query.bool();
query.must(Query.term(TagAutocompleteData.TAG_TYPE, tagType.name()));
if (IndexController.LogicIndicesRegister.isMergedTable(TagAutocompleteData.INDEX_NAME)) {
@@ -68,10 +65,7 @@ public class TagAutoCompleteQueryDAO extends EsDAO implements ITagAutoCompleteQu
.size(limit));
final SearchResponse response = getClient().search(
- () -> TimeSeriesUtils.metricsIndexNames(
- IndexController.LogicIndicesRegister.getPhysicalTableName(TagAutocompleteData.INDEX_NAME),
- startSecondTB, endSecondTB
- ),
+ IndexNames.rangeIndices(IndexFamily.of(TagAutocompleteData.INDEX_NAME), duration),
search.build()
);
Set<String> tagKeys = new HashSet<>();
@@ -95,8 +89,6 @@ public class TagAutoCompleteQueryDAO extends EsDAO implements ITagAutoCompleteQu
public Set<String> queryTagAutocompleteValues(final TagType tagType, final String tagKey,
final int limit,
final Duration duration) throws IOException {
- final long startSecondTB = nonNull(duration) ? duration.getStartTimeBucketInSec() : 0;
- final long endSecondTB = nonNull(duration) ? duration.getEndTimeBucketInSec() : 0;
BoolQueryBuilder query = Query.bool().must(Query.term(TagAutocompleteData.TAG_KEY, tagKey));
query.must(Query.term(TagAutocompleteData.TAG_TYPE, tagType.name()));
if (IndexController.LogicIndicesRegister.isMergedTable(TagAutocompleteData.INDEX_NAME)) {
@@ -105,10 +97,7 @@ public class TagAutoCompleteQueryDAO extends EsDAO implements ITagAutoCompleteQu
final SearchBuilder search = Search.builder().query(query).size(limit);
final SearchResponse response = getClient().search(
- () -> TimeSeriesUtils.metricsIndexNames(
- IndexController.LogicIndicesRegister.getPhysicalTableName(TagAutocompleteData.INDEX_NAME),
- startSecondTB, endSecondTB
- ),
+ IndexNames.rangeIndices(IndexFamily.of(TagAutocompleteData.INDEX_NAME), duration),
search.build()
);
Set<String> tagValues = new HashSet<>();
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 8153c4b140..d0ddbde2eb 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
@@ -50,8 +50,9 @@ import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.ElasticSearchConverter;
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.IndexFamily;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.RoutingUtils;
-import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeRangeIndexNameGenerator;
import static java.util.Objects.nonNull;
@@ -149,11 +150,7 @@ public class TraceQueryEsDAO extends EsDAO implements ITraceQueryDAO {
search.size(limit).from(from);
final SearchResponse response = searchDebuggable(
- new TimeRangeIndexNameGenerator(
- IndexController.LogicIndicesRegister.getPhysicalTableName(SegmentRecord.INDEX_NAME),
- startSecondTB,
- endSecondTB
- ), search.build());
+ IndexNames.rangeIndices(IndexFamily.of(SegmentRecord.INDEX_NAME), duration), search.build());
final TraceBrief traceBrief = new TraceBrief();
for (SearchHit searchHit : response.getHits().getHits()) {
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/otlp/OTLPTraceQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/otlp/OTLPTraceQueryEsDAO.java
index b91cbe9ac6..5acbe1d8d2 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/otlp/OTLPTraceQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/otlp/OTLPTraceQueryEsDAO.java
@@ -37,7 +37,6 @@ import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.B
import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.TermsAggregationBuilder;
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.analysis.manual.searchtag.Tag;
import org.apache.skywalking.oap.server.core.otlp.OTLPServiceRelationTraffic;
import org.apache.skywalking.oap.server.core.otlp.OTLPServiceSpanTraffic;
@@ -55,8 +54,9 @@ import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.ElasticSearchConverter;
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.IndexFamily;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.RoutingUtils;
-import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeRangeIndexNameGenerator;
/**
* Reads {@code otlp_span} on Elasticsearch: a terms aggregation on {@code trace_id} selects the traces, then the
@@ -128,20 +128,17 @@ public class OTLPTraceQueryEsDAO extends EsDAO implements IOTLPTraceQueryDAO {
public List<SpanWrapper> queryTraceById(final String traceId, @Nullable final Duration duration) {
final String physicalIndex = IndexController.LogicIndicesRegister.getPhysicalTableName(OTLPSpanRecord.INDEX_NAME);
final BoolQueryBuilder query = Query.bool().must(Query.term(OTLPSpanRecord.TRACE_ID, traceId));
- final String index;
- if (duration != null && duration.getStartTimestamp() > 0 && duration.getEndTimestamp() > 0) {
+ final boolean inRange = duration != null && duration.getStartTimestamp() > 0 && duration.getEndTimestamp() > 0;
+ if (inRange) {
query.must(Query.range(OTLPSpanRecord.START_TIME)
.gte(duration.getStartTimestamp())
.lte(duration.getEndTimestamp()));
- // One comma-separated index expression: the scroller takes a single index name.
- index = String.join(",", new TimeRangeIndexNameGenerator(
- physicalIndex,
- TimeBucket.getRecordTimeBucket(duration.getStartTimestamp()),
- TimeBucket.getRecordTimeBucket(duration.getEndTimestamp())
- ).get());
- } else {
- index = physicalIndex;
}
+ // The day indices go to the client as an array, so the namespace is added to each of them, and the
+ // days without an index are skipped.
+ final String[] indices = inRange ? IndexNames.rangeIndicesBetweenTimestamps(
+ IndexFamily.of(OTLPSpanRecord.INDEX_NAME), duration.getStartTimestamp(), duration.getEndTimestamp()
+ ) : null;
final SearchBuilder search = Search.builder().query(query).size(SCROLLING_BATCH_SIZE);
final SearchParams params = new SearchParams();
RoutingUtils.addRoutingValueToSearchParam(params, traceId);
@@ -149,11 +146,12 @@ public class OTLPTraceQueryEsDAO extends EsDAO implements IOTLPTraceQueryDAO {
.<SpanWrapper>builder()
.client(getClient())
.search(search.build())
- .index(index)
+ .index(physicalIndex)
+ .indices(indices)
.params(params)
.resultConverter(searchHit -> toSpanWrapper(searchHit.getSource()))
.build();
- return scrollDebuggable(scroller, index, params);
+ return scrollDebuggable(scroller, indices != null ? String.join(",", indices) : physicalIndex, params);
}
@Override
@@ -221,10 +219,8 @@ public class OTLPTraceQueryEsDAO extends EsDAO implements IOTLPTraceQueryDAO {
final SearchBuilder search = Search.builder().query(query).aggregation(traceIdAggregation);
final SearchResponse traceIdResponse;
if (duration != null && duration.getStartTimestamp() > 0 && duration.getEndTimestamp() > 0) {
- traceIdResponse = searchDebuggable(new TimeRangeIndexNameGenerator(
- IndexController.LogicIndicesRegister.getPhysicalTableName(OTLPSpanRecord.INDEX_NAME),
- TimeBucket.getRecordTimeBucket(duration.getStartTimestamp()),
- TimeBucket.getRecordTimeBucket(duration.getEndTimestamp())
+ traceIdResponse = searchDebuggable(IndexNames.rangeIndicesBetweenTimestamps(
+ IndexFamily.of(OTLPSpanRecord.INDEX_NAME), duration.getStartTimestamp(), duration.getEndTimestamp()
), search.build());
} else {
traceIdResponse = searchDebuggable(
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/zipkin/ZipkinQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/zipkin/ZipkinQueryEsDAO.java
index c455ddbf53..255d5b10c6 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/zipkin/ZipkinQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/zipkin/ZipkinQueryEsDAO.java
@@ -30,7 +30,6 @@ import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.B
import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.TermsAggregationBuilder;
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.query.input.Duration;
import org.apache.skywalking.oap.server.core.storage.query.IZipkinQueryDAO;
import org.apache.skywalking.oap.server.core.zipkin.ZipkinServiceRelationTraffic;
@@ -44,8 +43,9 @@ import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.ElasticSearchConverter;
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.IndexFamily;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.RoutingUtils;
-import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeRangeIndexNameGenerator;
import zipkin2.Span;
import zipkin2.storage.QueryRequest;
@@ -201,11 +201,13 @@ public class ZipkinQueryEsDAO extends EsDAO implements IZipkinQueryDAO {
.order(BucketOrder.aggregation(ZipkinSpanRecord.TIMESTAMP_MILLIS, false));
SearchBuilder search = Search.builder().query(query).aggregation(traceIdAggregation);
- SearchResponse traceIdResponse = searchDebuggable(new TimeRangeIndexNameGenerator(
- IndexController.LogicIndicesRegister.getPhysicalTableName(ZipkinSpanRecord.INDEX_NAME),
- TimeBucket.getRecordTimeBucket(startTimeMillis),
- TimeBucket.getRecordTimeBucket(endTimeMillis)
- ), search.build());
+ // Day indices are listed only for the range the time filter above uses. A lookback that reaches back
+ // to 1970 would otherwise list one index per day since then.
+ final String[] indices = startTimeMillis > 0 && endTimeMillis > 0
+ ? IndexNames.rangeIndicesBetweenTimestamps(
+ IndexFamily.of(ZipkinSpanRecord.INDEX_NAME), startTimeMillis, endTimeMillis)
+ : new String[] {IndexController.LogicIndicesRegister.getPhysicalTableName(ZipkinSpanRecord.INDEX_NAME)};
+ SearchResponse traceIdResponse = searchDebuggable(indices, search.build());
Set<String> traceIds = new HashSet<>();
if (Objects.nonNull(traceIdResponse.getAggregations())) {
final Map<String, Object> idTerms =
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexNamesTest.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexNamesTest.java
new file mode 100644
index 0000000000..6be8cf7d0c
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndexNamesTest.java
@@ -0,0 +1,181 @@
+/*
+ * 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 java.util.Collections;
+import org.apache.skywalking.oap.server.core.analysis.DownSampling;
+import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
+import org.apache.skywalking.oap.server.core.analysis.record.Record;
+import org.apache.skywalking.oap.server.core.query.enumeration.Step;
+import org.apache.skywalking.oap.server.core.query.input.Duration;
+import org.apache.skywalking.oap.server.core.storage.model.BanyanDBModelExtension;
+import org.apache.skywalking.oap.server.core.storage.model.ElasticSearchModelExtension;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.core.storage.model.SQLDatabaseModelExtension;
+import org.joda.time.DateTime;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class IndexNamesTest {
+
+ private static final IndexFamily METRICS = new IndexFamily("metrics-apdex", IndexRolling.DAY_STEP);
+ private static final IndexFamily SEGMENT = new IndexFamily("segment", IndexRolling.SUPER_DATASET_DAY_STEP);
+ private static final IndexFamily MANAGEMENT = new IndexFamily("management", IndexRolling.NONE);
+
+ @AfterEach
+ void tearDown() {
+ IndexNames.configure(1, -1);
+ }
+
+ @Test
+ void configure_withoutAPositiveSuperDatasetDayStep_usesTheDayStepForSuperDatasets() {
+ IndexNames.configure(3, -1);
+
+ assertEquals("segment-20200807", IndexNames.pointIndex(SEGMENT, 20200809L, Step.DAY));
+ }
+
+ @Test
+ void pointIndex_rollsTheDayIntoItsGroupCountedFrom2000() {
+ IndexNames.configure(11, -1);
+
+ assertEquals("metrics-apdex-20000101", IndexNames.pointIndex(METRICS, 20000105L, Step.DAY));
+ assertEquals("metrics-apdex-20000101", IndexNames.pointIndex(METRICS, 20000111L, Step.DAY));
+ assertEquals("metrics-apdex-20000112", IndexNames.pointIndex(METRICS, 20000112L, Step.DAY));
+ assertEquals("metrics-apdex-20000112", IndexNames.pointIndex(METRICS, 20000122L, Step.DAY));
+ assertEquals("metrics-apdex-20000123", IndexNames.pointIndex(METRICS, 20000123L, Step.DAY));
+ assertEquals("metrics-apdex-20000123", IndexNames.pointIndex(METRICS, 20000125L, Step.DAY));
+ }
+
+ @Test
+ void pointIndex_takesTheDayOfThePointAtTheStepPrecision() {
+ assertEquals("metrics-apdex-20220710", IndexNames.pointIndex(METRICS, 20220710111111L, Step.SECOND));
+ assertEquals("metrics-apdex-20220710", IndexNames.pointIndex(METRICS, 202207101111L, Step.MINUTE));
+ assertEquals("metrics-apdex-20220710", IndexNames.pointIndex(METRICS, 2022071011L, Step.HOUR));
+ assertEquals("metrics-apdex-20220710", IndexNames.pointIndex(METRICS, 20220710L, Step.DAY));
+ assertEquals("segment-20220710", IndexNames.pointIndex(SEGMENT, 20220710111111L, Step.SECOND));
+ }
+
+ @Test
+ void pointIndex_rejectsABlankTableOrAPointThatIsNotPositive() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> IndexNames.pointIndex(new IndexFamily(" ", IndexRolling.DAY_STEP), 20220710L, Step.DAY)
+ );
+ assertThrows(IllegalArgumentException.class, () -> IndexNames.pointIndex(METRICS, 0, Step.DAY));
+ }
+
+ @Test
+ void writeIndex_rollsMetricsAndRecordsByDayStepAndSuperDatasetsBySuperDatasetDayStep() {
+ IndexNames.configure(3, 1);
+ final Model superDataset = model("superDatasetModel", DownSampling.Second, true, Record.class);
+ final Model record = model("normalRecordModel", DownSampling.Second, false, Record.class);
+ final Model metrics = model("normalMetricsModel", DownSampling.Minute, false, Metrics.class);
+
+ assertEquals("superDatasetModel-20200809", IndexNames.writeIndex(superDataset, 2020_0809_1010_59L));
+ assertEquals("records-all-20200807", IndexNames.writeIndex(record, 2020_0809_1010_59L));
+ assertEquals("metrics-all-20200807", IndexNames.writeIndex(metrics, 2020_0809_1010L));
+ assertEquals("superDatasetModel-20200810", IndexNames.writeIndex(superDataset, 2020_0810_1010_59L));
+ assertEquals("records-all-20200810", IndexNames.writeIndex(record, 2020_0810_1010_59L));
+ assertEquals("metrics-all-20200810", IndexNames.writeIndex(metrics, 2020_0810_1010L));
+ }
+
+ @Test
+ void writeIndex_ofManagementData_isTheTableWithoutADate() {
+ final Model management = model("ui_template", DownSampling.None, false, Record.class);
+
+ assertEquals("management", IndexNames.writeIndex(management, 20200809L));
+ }
+
+ @Test
+ void rangeIndices_listsEachRolledDayOnce() {
+ assertArrayEquals(
+ new String[] {"segment-20200601", "segment-20200602", "segment-20200603", "segment-20200604", "segment-20200605"},
+ IndexNames.rangeIndices(SEGMENT, duration("2020-06-01 1400", "2020-06-05 1400"))
+ );
+ IndexNames.configure(1, 2);
+ assertArrayEquals(
+ new String[] {"segment-20200531", "segment-20200602", "segment-20200604"},
+ IndexNames.rangeIndices(SEGMENT, duration("2020-06-01 1400", "2020-06-05 1400"))
+ );
+ }
+
+ @Test
+ void rangeIndices_withinOneDay_isThatDaysIndex() {
+ assertArrayEquals(
+ new String[] {"segment-20200605"},
+ IndexNames.rangeIndices(SEGMENT, duration("2020-06-05 1400", "2020-06-05 1200"))
+ );
+ }
+
+ @Test
+ void rangeIndices_endingOnAnEarlierDayOrWithoutDuration_isTheAlias() {
+ assertArrayEquals(
+ new String[] {"segment"}, IndexNames.rangeIndices(SEGMENT, duration("2020-06-05 1400", "2020-06-01 1400")));
+ assertArrayEquals(new String[] {"segment"}, IndexNames.rangeIndices(SEGMENT, null));
+ }
+
+ @Test
+ void rangeIndicesBetweenTimestamps_listsTheDaysOfTheLocalTimestamps() {
+ IndexNames.configure(1, 3);
+
+ assertArrayEquals(
+ new String[] {"segment-20200807", "segment-20200810"},
+ IndexNames.rangeIndicesBetweenTimestamps(
+ SEGMENT, new DateTime(2020, 8, 8, 12, 0).getMillis(), new DateTime(2020, 8, 10, 12, 0).getMillis())
+ );
+ }
+
+ @Test
+ void latestIndex_isTodaysIndex() {
+ IndexNames.configure(3, 1);
+ final long today = new DateTime(2020, 8, 9, 10, 10).getMillis();
+
+ assertEquals("metrics-apdex-20200807", IndexNames.latestIndex(METRICS, today));
+ assertEquals("segment-20200809", IndexNames.latestIndex(SEGMENT, today));
+ assertThrows(IllegalArgumentException.class, () -> IndexNames.latestIndex(MANAGEMENT, today));
+ }
+
+ @Test
+ void indexStartDay_isTheDateSuffix() {
+ assertEquals(20190602L, IndexNames.indexStartDay("Index_Test-20190602"));
+ assertEquals(20190602L, IndexNames.indexStartDay("namespace_metrics-all-20190602"));
+ }
+
+ private static Model model(final String name,
+ final DownSampling downSampling,
+ final boolean superDataset,
+ final Class<?> streamClass) {
+ return new Model(
+ name, Collections.emptyList(), 0, downSampling, superDataset, streamClass, true, false,
+ new SQLDatabaseModelExtension(), new BanyanDBModelExtension(), new ElasticSearchModelExtension()
+ );
+ }
+
+ private static Duration duration(final String start, final String end) {
+ final Duration duration = new Duration();
+ duration.setStep(Step.MINUTE);
+ duration.setStart(start);
+ duration.setEnd(end);
+ return duration;
+ }
+}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstallerTest.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstallerTest.java
new file mode 100644
index 0000000000..422f6e4ff4
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstallerTest.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.oap.server.storage.plugin.elasticsearch.base;
+
+import java.util.Collections;
+import org.apache.skywalking.oap.server.core.analysis.DownSampling;
+import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
+import org.apache.skywalking.oap.server.core.storage.model.BanyanDBModelExtension;
+import org.apache.skywalking.oap.server.core.storage.model.ElasticSearchModelExtension;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.core.storage.model.SQLDatabaseModelExtension;
+import org.apache.skywalking.oap.server.core.storage.model.StorageManipulationOpt;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchConfig;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+@ExtendWith(MockitoExtension.class)
+class StorageEsInstallerTest {
+
+ @Mock
+ private ElasticSearchClient client;
+
+ @Mock
+ private ModuleManager moduleManager;
+
+ @Test
+ void whenCreating_withoutSchemaChange_registersThePhysicalTableWithoutCallingElasticsearch() throws Exception {
+ final String metric = "installer_test_metric";
+ final Model model = new Model(
+ metric, Collections.emptyList(), 0, DownSampling.Minute, false, Metrics.class, true, false,
+ new SQLDatabaseModelExtension(), new BanyanDBModelExtension(), new ElasticSearchModelExtension()
+ );
+
+ new StorageEsInstaller(client, moduleManager, new StorageModuleElasticsearchConfig())
+ .whenCreating(model, StorageManipulationOpt.withoutSchemaChange());
+
+ assertEquals(
+ IndexController.METRICS_LOGIC_TABLE_NAME,
+ IndexController.LogicIndicesRegister.getPhysicalTableName(metric)
+ );
+ verifyNoInteractions(client);
+ }
+}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTest.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTest.java
deleted file mode 100644
index 0002de624b..0000000000
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTest.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
- * 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 com.google.common.collect.Lists;
-import org.apache.skywalking.oap.server.core.analysis.DownSampling;
-import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
-import org.apache.skywalking.oap.server.core.analysis.record.Record;
-import org.apache.skywalking.oap.server.core.query.enumeration.Step;
-import org.apache.skywalking.oap.server.core.storage.model.BanyanDBModelExtension;
-import org.apache.skywalking.oap.server.core.storage.model.ElasticSearchModelExtension;
-import org.apache.skywalking.oap.server.core.storage.model.Model;
-import org.apache.skywalking.oap.server.core.storage.model.SQLDatabaseModelExtension;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeSeriesUtils.compressTimeBucket;
-import static org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeSeriesUtils.writeIndexName;
-
-public class TimeSeriesUtilsTest {
-
- private Model superDatasetModel;
- private Model normalRecordModel;
- private Model normalMetricsModel;
-
- @BeforeEach
- public void prepare() {
- superDatasetModel = new Model("superDatasetModel", Lists.newArrayList(),
- 0, DownSampling.Second, true, Record.class, true, false,
- new SQLDatabaseModelExtension(), new BanyanDBModelExtension(), new ElasticSearchModelExtension()
- );
- normalRecordModel = new Model("normalRecordModel", Lists.newArrayList(),
- 0, DownSampling.Second, false, Record.class, true, false,
- new SQLDatabaseModelExtension(), new BanyanDBModelExtension(), new ElasticSearchModelExtension()
- );
- normalMetricsModel = new Model("normalMetricsModel", Lists.newArrayList(),
- 0, DownSampling.Minute, false, Metrics.class, true, false,
- new SQLDatabaseModelExtension(), new BanyanDBModelExtension(), new ElasticSearchModelExtension()
- );
- TimeSeriesUtils.setSUPER_DATASET_DAY_STEP(1);
- TimeSeriesUtils.setDAY_STEP(3);
- }
-
- @Test
- public void testCompressTimeBucket() {
- Assertions.assertEquals(20000101L, compressTimeBucket(20000105, 11));
- Assertions.assertEquals(20000101L, compressTimeBucket(20000111, 11));
- Assertions.assertEquals(20000112L, compressTimeBucket(20000112, 11));
- Assertions.assertEquals(20000112L, compressTimeBucket(20000122, 11));
- Assertions.assertEquals(20000123L, compressTimeBucket(20000123, 11));
- Assertions.assertEquals(20000123L, compressTimeBucket(20000125, 11));
- }
-
- @Test
- public void testIndexRolling() {
- long secondTimeBucket = 2020_0809_1010_59L;
- long minuteTimeBucket = 2020_0809_1010L;
-
- Assertions.assertEquals(
- "superDatasetModel-20200809",
- writeIndexName(superDatasetModel, secondTimeBucket)
- );
- Assertions.assertEquals(
- "records-all-20200807",
- writeIndexName(normalRecordModel, secondTimeBucket)
- );
- Assertions.assertEquals(
- "metrics-all-20200807",
- writeIndexName(normalMetricsModel, minuteTimeBucket)
- );
- secondTimeBucket += 1000000;
- minuteTimeBucket += 10000;
- Assertions.assertEquals(
- "superDatasetModel-20200810",
- writeIndexName(superDatasetModel, secondTimeBucket)
- );
- Assertions.assertEquals(
- "records-all-20200810",
- writeIndexName(normalRecordModel, secondTimeBucket)
- );
- Assertions.assertEquals(
- "metrics-all-20200810",
- writeIndexName(normalMetricsModel, minuteTimeBucket)
- );
- }
-
- @Test
- public void queryIndexNameTest() {
- Assertions.assertEquals(
- "metrics-apdex-20220710",
- TimeSeriesUtils.queryIndexName("metrics-apdex", 20220710111111L, Step.SECOND, false, false)
- );
- Assertions.assertEquals(
- "metrics-apdex-20220710",
- TimeSeriesUtils.queryIndexName("metrics-apdex", 202207101111L, Step.MINUTE, false, false)
- );
- Assertions.assertEquals(
- "metrics-apdex-20220710",
- TimeSeriesUtils.queryIndexName("metrics-apdex", 2022071011L, Step.HOUR, false, false)
- );
- Assertions.assertEquals(
- "metrics-apdex-20220710",
- TimeSeriesUtils.queryIndexName("metrics-apdex", 20220710L, Step.DAY, false, false)
- );
- Assertions.assertEquals(
- "metrics-apdex-20220710",
- TimeSeriesUtils.queryIndexName("metrics-apdex", 20220710111111L, Step.DAY, true, true)
- );
- }
-
-}
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
deleted file mode 100644
index 05ba998d8f..0000000000
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTestCase.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.core.analysis.manual.segment.SegmentRecord;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-public class TimeSeriesUtilsTestCase {
-
- @Test
- public void indexTimeSeries() {
- Assertions.assertEquals(20190602, TimeSeriesUtils.isolateTimeFromIndexName("Index_Test-20190602"));
- }
-
- @Test
- public void querySuperDatasetIndices() {
- String[] indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200601140000L, 20200605140000L);
- Assertions.assertEquals(indices.length, 5);
- indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200605140000L, 20200605140000L);
- Assertions.assertEquals(indices.length, 1);
- indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200605140000L, 20200601140000L);
- Assertions.assertEquals(indices.length, 1);
- TimeSeriesUtils.setSUPER_DATASET_DAY_STEP(2);
- indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200601140000L, 20200605140000L);
- Assertions.assertEquals(indices.length, 3);
- indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200605140000L, 20200605140000L);
- Assertions.assertEquals(indices.length, 1);
- indices = TimeSeriesUtils.superDatasetIndexNames(SegmentRecord.INDEX_NAME, 20200605140000L, 20200601140000L);
- Assertions.assertEquals(indices.length, 1);
-
- }
-}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAOTest.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAOTest.java
index ebf5b072d2..a9d244597d 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAOTest.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/AggregationQueryEsDAOTest.java
@@ -19,7 +19,6 @@
package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query;
import java.util.Collections;
-import java.util.function.Supplier;
import org.apache.skywalking.library.elasticsearch.requests.search.Search;
import org.apache.skywalking.library.elasticsearch.response.search.SearchResponse;
import org.apache.skywalking.oap.server.core.query.enumeration.Order;
@@ -27,7 +26,7 @@ import org.apache.skywalking.oap.server.core.query.enumeration.Step;
import org.apache.skywalking.oap.server.core.query.input.Duration;
import org.apache.skywalking.oap.server.core.query.input.TopNCondition;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
-import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeSeriesUtils;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -49,20 +48,17 @@ class AggregationQueryEsDAOTest {
@BeforeEach
void setUp() {
- TimeSeriesUtils.setDAY_STEP(3);
- TimeSeriesUtils.setSUPER_DATASET_DAY_STEP(1);
+ IndexNames.configure(3, 1);
}
@AfterEach
void tearDown() {
- TimeSeriesUtils.setDAY_STEP(1);
- TimeSeriesUtils.setSUPER_DATASET_DAY_STEP(1);
+ IndexNames.configure(1, -1);
}
@Test
- @SuppressWarnings("unchecked")
void sortMetrics_shouldSearchTheIndexRolledByDayStep() {
- when(client.search(any(Supplier.class), any(Search.class))).thenReturn(new SearchResponse());
+ when(client.search(any(String[].class), any(Search.class))).thenReturn(new SearchResponse());
final TopNCondition condition = new TopNCondition();
condition.setName("service_resp_time");
@@ -75,10 +71,10 @@ class AggregationQueryEsDAOTest {
new AggregationQueryEsDAO(client).sortMetrics(condition, "value", duration, Collections.emptyList());
- final ArgumentCaptor<Supplier<String[]>> indices = ArgumentCaptor.forClass(Supplier.class);
+ final ArgumentCaptor<String[]> indices = ArgumentCaptor.forClass(String[].class);
verify(client).search(indices.capture(), any(Search.class));
// With dayStep 3, the minutes of 2020-08-09 are written to the index of 2020-08-07,
- // see TimeSeriesUtilsTest#testIndexRolling.
- assertThat(indices.getValue().get()).containsExactly("service_resp_time-20200807");
+ // see IndexNamesTest#writeIndex_rollsMetricsAndRecordsByDayStepAndSuperDatasetsBySuperDatasetDayStep.
+ assertThat(indices.getValue()).containsExactly("service_resp_time-20200807");
}
}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EsReadIndexSelectionTest.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EsReadIndexSelectionTest.java
new file mode 100644
index 0000000000..9e4a8ff83f
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/EsReadIndexSelectionTest.java
@@ -0,0 +1,416 @@
+/*
+ * 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 java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+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;
+import org.apache.skywalking.oap.server.core.analysis.DownSampling;
+import org.apache.skywalking.oap.server.core.analysis.manual.searchtag.TagType;
+import org.apache.skywalking.oap.server.core.analysis.record.Record;
+import org.apache.skywalking.oap.server.core.query.enumeration.Order;
+import org.apache.skywalking.oap.server.core.query.enumeration.Scope;
+import org.apache.skywalking.oap.server.core.query.enumeration.Step;
+import org.apache.skywalking.oap.server.core.query.input.Duration;
+import org.apache.skywalking.oap.server.core.query.input.Entity;
+import org.apache.skywalking.oap.server.core.query.input.MetricsCondition;
+import org.apache.skywalking.oap.server.core.query.input.OTLPTraceQueryCondition;
+import org.apache.skywalking.oap.server.core.query.type.QueryOrder;
+import org.apache.skywalking.oap.server.core.query.type.TraceState;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
+import org.apache.skywalking.oap.server.core.storage.model.BanyanDBModelExtension;
+import org.apache.skywalking.oap.server.core.storage.model.ElasticSearchModelExtension;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.core.storage.model.SQLDatabaseModelExtension;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexNames;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.otlp.OTLPTraceQueryEsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.zipkin.ZipkinQueryEsDAO;
+import org.joda.time.DateTime;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import zipkin2.storage.QueryRequest;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Pins the indices each Elasticsearch read sends, so a change to how index names are built can be checked
+ * against what the reads searched before it.
+ *
+ * <p>Index groups count days from 2000-01-01. With a step of 3, 2020-08-07 and 2020-08-10 start a group, so
+ * {@link #RANGE} (2020-08-08 to 2020-08-10) names two indices, where a step of 1 names three. Each test rolls
+ * one step by 3 and the other by 1, so the names show which step the read used.
+ */
+@ExtendWith(MockitoExtension.class)
+class EsReadIndexSelectionTest {
+
+ private static final String METRIC = "index_selection_test_metric";
+
+ private static final Duration RANGE = duration(Step.MINUTE, "2020-08-08 1200", "2020-08-10 1200");
+
+ @Mock
+ private ElasticSearchClient client;
+
+ /**
+ * The OAP registers every static model's table at boot, before any read, and the super dataset reads rely on
+ * it to roll by the super dataset day step.
+ */
+ @BeforeAll
+ static void registerSuperDatasetTables() {
+ for (final String table : new String[] {
+ "segment", "log", "gen_ai_evaluation_record", "ai_agent_session_flow", "ai_agent_session_data",
+ "otlp_span", "zipkin_span"
+ }) {
+ IndexController.LogicIndicesRegister.registerRelation(
+ new Model(
+ table, Collections.emptyList(), 0, DownSampling.Second, true, Record.class, true, false,
+ new SQLDatabaseModelExtension(), new BanyanDBModelExtension(), new ElasticSearchModelExtension()
+ ),
+ table
+ );
+ }
+ }
+
+ @BeforeEach
+ void setUp() {
+ ValueColumnMetadata.INSTANCE.putIfAbsent(METRIC, "value", Column.ValueDataType.COMMON_VALUE, 0, 1);
+ }
+
+ @AfterEach
+ void tearDown() {
+ ValueColumnMetadata.INSTANCE.remove(METRIC);
+ IndexNames.configure(1, 1);
+ }
+
+ @Test
+ void readLabeledMetricsValuesWithoutEntity_searchesTheMetricRolledByDayStep() {
+ IndexNames.configure(3, 1);
+ answerRangeSearch();
+
+ // A minute step over RANGE is more points than a Duration allows, so this read takes it by the hour.
+ new MetricsQueryEsDAO(client).readLabeledMetricsValuesWithoutEntity(
+ METRIC, "value", Collections.emptyList(), duration(Step.HOUR, "2020-08-08 12", "2020-08-10 12"));
+
+ assertThat(rangeSearchIndices()).containsExactly(METRIC + "-20200807", METRIC + "-20200810");
+ }
+
+ @Test
+ void listEntityIdsInRange_searchesTheMetricRolledByDayStep() throws Exception {
+ IndexNames.configure(3, 1);
+ answerRangeSearch();
+
+ new MetricsQueryEsDAO(client).listEntityIdsInRange(METRIC, "value", null, RANGE, 10);
+
+ assertThat(rangeSearchIndices()).containsExactly(METRIC + "-20200807", METRIC + "-20200810");
+ }
+
+ @Test
+ void listEntityIdsInRange_ofAForeignMetric_searchesTheMergedMetricsIndexRolledByDayStep() throws Exception {
+ IndexNames.configure(3, 1);
+ answerRangeSearch();
+
+ new MetricsQueryEsDAO(client).listEntityIdsInRange("absent_metric", "value", "COMMON", RANGE, 10);
+
+ assertThat(rangeSearchIndices()).containsExactly("metrics-all-20200807", "metrics-all-20200810");
+ }
+
+ @Test
+ void readMetricsValues_readsEachPointFromTheIndexRolledByDayStep() throws Exception {
+ IndexNames.configure(3, 1);
+ answerIdsRead();
+
+ new MetricsQueryEsDAO(client).readMetricsValues(
+ serviceCondition(), "value", duration(Step.DAY, "2020-08-09", "2020-08-10"));
+
+ assertThat(idsReadIndices()).containsExactlyInAnyOrder(METRIC + "-20200807", METRIC + "-20200810");
+ }
+
+ @Test
+ void readLabeledMetricsValues_readsEachPointFromTheIndexRolledByDayStep() throws Exception {
+ IndexNames.configure(3, 1);
+ answerIdsRead();
+
+ new MetricsQueryEsDAO(client).readLabeledMetricsValues(
+ serviceCondition(), "value", Collections.emptyList(), duration(Step.HOUR, "2020-08-09 23", "2020-08-10 00"));
+
+ assertThat(idsReadIndices()).containsExactlyInAnyOrder(METRIC + "-20200807", METRIC + "-20200810");
+ }
+
+ @Test
+ void readHeatMap_readsEachPointFromTheIndexRolledByDayStep() {
+ IndexNames.configure(3, 1);
+ answerIdsRead();
+
+ new MetricsQueryEsDAO(client).readHeatMap(
+ serviceCondition(), "value", duration(Step.MINUTE, "2020-08-09 2359", "2020-08-10 0000"));
+
+ assertThat(idsReadIndices()).containsExactlyInAnyOrder(METRIC + "-20200807", METRIC + "-20200810");
+ }
+
+ @Test
+ void queryTagAutocompleteKeys_searchesTheTagIndexRolledByDayStep() throws Exception {
+ IndexNames.configure(3, 1);
+ answerRangeSearch();
+
+ new TagAutoCompleteQueryDAO(client).queryTagAutocompleteKeys(TagType.TRACE, 10, RANGE);
+
+ assertThat(rangeSearchIndices()).containsExactly("tag_autocomplete-20200807", "tag_autocomplete-20200810");
+ }
+
+ @Test
+ void queryTagAutocompleteValues_withoutDuration_searchesTheWholeTagIndexFamily() throws Exception {
+ IndexNames.configure(3, 1);
+ answerRangeSearch();
+
+ new TagAutoCompleteQueryDAO(client).queryTagAutocompleteValues(TagType.TRACE, "http.method", 10, null);
+
+ assertThat(rangeSearchIndices()).containsExactly("tag_autocomplete");
+ }
+
+ @Test
+ void queryBasicTraces_searchesSegmentsRolledBySuperDatasetDayStep() throws Exception {
+ IndexNames.configure(1, 3);
+ answerRangeSearch();
+
+ new TraceQueryEsDAO(client, 100).queryBasicTraces(
+ RANGE, 0, 0, null, null, null, null, 20, 0, TraceState.ALL, QueryOrder.BY_START_TIME,
+ Collections.emptyList());
+
+ assertThat(rangeSearchIndices()).containsExactly("segment-20200807", "segment-20200810");
+ }
+
+ @Test
+ void queryLogs_searchesLogsRolledBySuperDatasetDayStep() throws Exception {
+ IndexNames.configure(1, 3);
+ answerRangeSearch();
+
+ new LogQueryEsDAO(client).queryLogs(
+ null, null, null, null, Order.DES, 0, 20, RANGE, Collections.emptyList(), Collections.emptyList(),
+ Collections.emptyList());
+
+ assertThat(rangeSearchIndices()).containsExactly("log-20200807", "log-20200810");
+ }
+
+ @Test
+ void queryGenAIEvaluationRecord_searchesRecordsRolledBySuperDatasetDayStep() throws Exception {
+ IndexNames.configure(1, 3);
+ answerRangeSearch();
+
+ new GenAIEvaluationRecordQueryEsDAO(client).queryGenAIEvaluationRecord(
+ null, null, null, null, null, null, null, null, null, null, null, null, Order.DES, 0, 20, RANGE);
+
+ assertThat(rangeSearchIndices()).containsExactly(
+ "gen_ai_evaluation_record-20200807", "gen_ai_evaluation_record-20200810");
+ }
+
+ @Test
+ void queryRounds_searchesRoundsRolledBySuperDatasetDayStep() throws Exception {
+ IndexNames.configure(1, 3);
+ answerRangeSearch();
+
+ new AIAgentConversationQueryEsDAO(client).queryRounds("service", null, null, RANGE, 10, false);
+
+ assertThat(rangeSearchIndices()).containsExactly(
+ "ai_agent_session_flow-20200807", "ai_agent_session_flow-20200810");
+ }
+
+ @Test
+ void queryHeadRound_searchesTheWholeRoundIndexFamily() throws Exception {
+ IndexNames.configure(1, 3);
+ answerRangeSearch();
+
+ new AIAgentConversationQueryEsDAO(client).queryHeadRound("service", null, "conversation", false);
+
+ assertThat(rangeSearchIndices()).containsExactly("ai_agent_session_flow");
+ }
+
+ @Test
+ void queryRoundsByNumber_searchesTheWholeRoundIndexFamily() throws Exception {
+ IndexNames.configure(1, 3);
+ answerRangeSearch();
+
+ new AIAgentConversationQueryEsDAO(client).queryRoundsByNumber(
+ "service", null, "conversation", 1, 2, 1024, false);
+
+ assertThat(rangeSearchIndices()).containsExactly("ai_agent_session_flow");
+ }
+
+ @Test
+ void queryFiles_searchesFilesRolledBySuperDatasetDayStep() throws Exception {
+ IndexNames.configure(1, 3);
+ answerRangeSearch();
+
+ new AIAgentConversationQueryEsDAO(client).queryFiles(
+ "service", null, "session", millis(2020, 8, 8, 12), millis(2020, 8, 10, 12), 0, 10, 1024, false);
+
+ assertThat(rangeSearchIndices()).containsExactly(
+ "ai_agent_session_data-20200807", "ai_agent_session_data-20200810");
+ }
+
+ @Test
+ void queryFiles_withoutAStartTime_searchesTheWholeFileIndexFamily() throws Exception {
+ IndexNames.configure(1, 3);
+ answerRangeSearch();
+
+ new AIAgentConversationQueryEsDAO(client).queryFiles(
+ "service", null, "session", 0, millis(2020, 8, 10, 12), 0, 10, 1024, false);
+
+ assertThat(rangeSearchIndices()).containsExactly("ai_agent_session_data");
+ }
+
+ @Test
+ void queryTraces_searchesOTLPSpansRolledBySuperDatasetDayStep() {
+ IndexNames.configure(1, 3);
+ answerRangeSearch();
+
+ final OTLPTraceQueryCondition condition = new OTLPTraceQueryCondition();
+ condition.setQueryDuration(RANGE);
+ new OTLPTraceQueryEsDAO(client).queryTraces(condition);
+
+ assertThat(rangeSearchIndices()).containsExactly("otlp_span-20200807", "otlp_span-20200810");
+ }
+
+ @Test
+ void queryTraces_withoutDuration_searchesTheOTLPSpanAlias() {
+ IndexNames.configure(1, 3);
+ when(client.search(anyString(), any(Search.class))).thenReturn(new SearchResponse());
+
+ new OTLPTraceQueryEsDAO(client).queryTraces(new OTLPTraceQueryCondition());
+
+ verify(client).search(eq("otlp_span"), any(Search.class));
+ }
+
+ @Test
+ void queryTraceById_scrollsEachOTLPSpanDayIndexRolledBySuperDatasetDayStep() {
+ IndexNames.configure(1, 3);
+ when(client.search(any(String[].class), any(Search.class), any(SearchParams.class)))
+ .thenReturn(new SearchResponse());
+
+ new OTLPTraceQueryEsDAO(client).queryTraceById("trace", RANGE);
+
+ final ArgumentCaptor<String[]> indices = ArgumentCaptor.forClass(String[].class);
+ verify(client).search(indices.capture(), any(Search.class), any(SearchParams.class));
+ assertThat(indices.getValue()).containsExactly("otlp_span-20200807", "otlp_span-20200810");
+ }
+
+ @Test
+ void queryTraceById_withoutDuration_scrollsTheOTLPSpanAlias() {
+ IndexNames.configure(1, 3);
+ when(client.search(anyString(), any(Search.class), any(SearchParams.class))).thenReturn(new SearchResponse());
+
+ new OTLPTraceQueryEsDAO(client).queryTraceById("trace", null);
+
+ verify(client).search(eq("otlp_span"), any(Search.class), any(SearchParams.class));
+ }
+
+ @Test
+ void getTraces_searchesZipkinSpansRolledBySuperDatasetDayStep() {
+ IndexNames.configure(1, 3);
+ answerRangeSearch();
+ when(client.search(anyString(), any(Search.class), any(SearchParams.class))).thenReturn(new SearchResponse());
+
+ final QueryRequest request = QueryRequest.newBuilder()
+ .endTs(millis(2020, 8, 10, 12))
+ .lookback(2L * 24 * 3600 * 1000)
+ .limit(10)
+ .build();
+ new ZipkinQueryEsDAO(client).getTraces(request, duration(Step.SECOND, "2020-08-08 120000", "2020-08-10 120000"));
+
+ assertThat(rangeSearchIndices()).containsExactly("zipkin_span-20200807", "zipkin_span-20200810");
+ }
+
+ @Test
+ void getTraces_withAStartBefore1970_searchesTheWholeZipkinSpanIndexFamily() {
+ IndexNames.configure(1, 3);
+ answerRangeSearch();
+ when(client.search(anyString(), any(Search.class), any(SearchParams.class))).thenReturn(new SearchResponse());
+
+ final QueryRequest request = QueryRequest.newBuilder()
+ .endTs(millis(2020, 8, 10, 12))
+ .lookback(millis(2020, 8, 10, 12))
+ .limit(10)
+ .build();
+ // A lookback as long as endTs, the Zipkin API's documented default, puts the start at the epoch. A start
+ // on 1969-12-31 stays before the epoch in every time zone.
+ new ZipkinQueryEsDAO(client).getTraces(request, duration(Step.SECOND, "1969-12-31 000000", "2020-08-10 120000"));
+
+ assertThat(rangeSearchIndices()).containsExactly("zipkin_span");
+ }
+
+ private void answerRangeSearch() {
+ when(client.search(any(String[].class), any(Search.class))).thenReturn(new SearchResponse());
+ }
+
+ private String[] rangeSearchIndices() {
+ final ArgumentCaptor<String[]> indices = ArgumentCaptor.forClass(String[].class);
+ verify(client).search(indices.capture(), any(Search.class));
+ return indices.getValue();
+ }
+
+ private void answerIdsRead() {
+ when(client.ids(any())).thenReturn(Optional.empty());
+ }
+
+ @SuppressWarnings("unchecked")
+ private Iterable<String> idsReadIndices() {
+ final ArgumentCaptor<Map<String, List<String>>> indexIdsGroup = ArgumentCaptor.forClass(Map.class);
+ verify(client).ids(indexIdsGroup.capture());
+ return indexIdsGroup.getValue().keySet();
+ }
+
+ private static MetricsCondition serviceCondition() {
+ final Entity entity = new Entity();
+ entity.setScope(Scope.Service);
+ entity.setServiceName("service");
+ entity.setNormal(true);
+ final MetricsCondition condition = new MetricsCondition();
+ condition.setName(METRIC);
+ condition.setEntity(entity);
+ return condition;
+ }
+
+ private static Duration duration(final Step step, final String start, final String end) {
+ final Duration duration = new Duration();
+ duration.setStep(step);
+ duration.setStart(start);
+ duration.setEnd(end);
+ return duration;
+ }
+
+ private static long millis(final int year, final int month, final int day, final int hour) {
+ return new DateTime(year, month, day, hour, 0).getMillis();
+ }
+}