Skip to content

Commit d7abd5b

Browse files
github-actions[bot]github-actions[bot]
committed
Refactor Exporters and Readers (#210)
* Move exporter config to query insights level Signed-off-by: Chenyang Ji <cyji@amazon.com> * use index mapping and define default index metadata for top n queries Signed-off-by: Chenyang Ji <cyji@amazon.com> * improve safegard for deleting top queries indices with index mapping Signed-off-by: Chenyang Ji <cyji@amazon.com> * fix unit tests Signed-off-by: Chenyang Ji <cyji@amazon.com> * fix bug on local index reader with generateLocalIndexDateHash Signed-off-by: Chenyang Ji <cyji@amazon.com> * fix exception in query grouper Signed-off-by: Chenyang Ji <cyji@amazon.com> * update PR based on comments and re-enable unit tests Signed-off-by: Chenyang Ji <cyji@amazon.com> --------- Signed-off-by: Chenyang Ji <cyji@amazon.com> (cherry picked from commit 2d14764) Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
1 parent 3e8701e commit d7abd5b

26 files changed

+1069
-489
lines changed

‎README.md

+1-2
Original file line numberDiff line numberDiff line change
@@ -52,8 +52,7 @@ A local index exporter allows you to export the top N queries to local OpenSearc
5252
PUT _cluster/settings
5353
{
5454
"persistent" : {
55-
"search.insights.top_queries.latency.exporter.type" : "local_index",
56-
"search.insights.top_queries.latency.exporter.config.index" : "YYYY.MM.dd"
55+
"search.insights.top_queries.exporter.type" : "local_index"
5756
}
5857
}
5958
```

‎src/main/java/org/opensearch/plugin/insights/QueryInsightsPlugin.java

+1-3
Original file line numberDiff line numberDiff line change
@@ -131,21 +131,19 @@ public List<Setting<?>> getSettings() {
131131
QueryInsightsSettings.TOP_N_LATENCY_QUERIES_ENABLED,
132132
QueryInsightsSettings.TOP_N_LATENCY_QUERIES_SIZE,
133133
QueryInsightsSettings.TOP_N_LATENCY_QUERIES_WINDOW_SIZE,
134-
QueryInsightsSettings.TOP_N_LATENCY_EXPORTER_SETTINGS,
135134
QueryInsightsSettings.TOP_N_CPU_QUERIES_ENABLED,
136135
QueryInsightsSettings.TOP_N_CPU_QUERIES_SIZE,
137136
QueryInsightsSettings.TOP_N_CPU_QUERIES_WINDOW_SIZE,
138-
QueryInsightsSettings.TOP_N_CPU_EXPORTER_SETTINGS,
139137
QueryInsightsSettings.TOP_N_MEMORY_QUERIES_ENABLED,
140138
QueryInsightsSettings.TOP_N_MEMORY_QUERIES_SIZE,
141139
QueryInsightsSettings.TOP_N_MEMORY_QUERIES_WINDOW_SIZE,
142-
QueryInsightsSettings.TOP_N_MEMORY_EXPORTER_SETTINGS,
143140
QueryInsightsSettings.TOP_N_QUERIES_GROUP_BY,
144141
QueryInsightsSettings.TOP_N_QUERIES_MAX_GROUPS_EXCLUDING_N,
145142
QueryInsightsSettings.TOP_N_QUERIES_GROUPING_FIELD_NAME,
146143
QueryInsightsSettings.TOP_N_QUERIES_GROUPING_FIELD_TYPE,
147144
QueryCategorizationSettings.SEARCH_QUERY_METRICS_ENABLED_SETTING,
148145
QueryInsightsSettings.TOP_N_EXPORTER_DELETE_AFTER,
146+
QueryInsightsSettings.TOP_N_EXPORTER_TYPE,
149147
QueryCategorizationSettings.SEARCH_QUERY_FIELD_TYPE_CACHE_SIZE_KEY
150148
);
151149
}

‎src/main/java/org/opensearch/plugin/insights/core/exporter/DebugExporter.java

+7-1
Original file line numberDiff line numberDiff line change
@@ -16,17 +16,23 @@
1616
/**
1717
* Debug exporter for development purpose
1818
*/
19-
public final class DebugExporter implements QueryInsightsExporter {
19+
public class DebugExporter implements QueryInsightsExporter {
2020
/**
2121
* Logger of the debug exporter
2222
*/
2323
private final Logger logger = LogManager.getLogger();
24+
private static final String EXPORTER_ID = "debug_exporter";
2425

2526
/**
2627
* Constructor of DebugExporter
2728
*/
2829
private DebugExporter() {}
2930

31+
@Override
32+
public String getId() {
33+
return EXPORTER_ID;
34+
}
35+
3036
private static class InstanceHolder {
3137
private static final DebugExporter INSTANCE = new DebugExporter();
3238
}

‎src/main/java/org/opensearch/plugin/insights/core/exporter/LocalIndexExporter.java

+153-45
Original file line numberDiff line numberDiff line change
@@ -8,55 +8,89 @@
88

99
package org.opensearch.plugin.insights.core.exporter;
1010

11-
import static org.opensearch.plugin.insights.core.service.TopQueriesService.isTopQueriesIndex;
11+
import static org.opensearch.plugin.insights.core.utils.ExporterReaderUtils.generateLocalIndexDateHash;
1212
import static org.opensearch.plugin.insights.settings.QueryInsightsSettings.DEFAULT_DELETE_AFTER_VALUE;
1313

14-
import java.time.Instant;
14+
import java.io.IOException;
15+
import java.nio.charset.Charset;
1516
import java.time.ZoneOffset;
1617
import java.time.ZonedDateTime;
1718
import java.time.format.DateTimeFormatter;
1819
import java.util.List;
19-
import java.util.Locale;
20-
import java.util.Map;
21-
import java.util.concurrent.TimeUnit;
20+
import java.util.Objects;
2221
import org.apache.logging.log4j.LogManager;
2322
import org.apache.logging.log4j.Logger;
23+
import org.opensearch.ExceptionsHelper;
24+
import org.opensearch.ResourceAlreadyExistsException;
25+
import org.opensearch.action.admin.indices.create.CreateIndexRequest;
26+
import org.opensearch.action.admin.indices.create.CreateIndexResponse;
27+
import org.opensearch.action.admin.indices.delete.DeleteIndexRequest;
2428
import org.opensearch.action.bulk.BulkRequestBuilder;
2529
import org.opensearch.action.bulk.BulkResponse;
2630
import org.opensearch.action.index.IndexRequest;
2731
import org.opensearch.client.Client;
28-
import org.opensearch.cluster.metadata.IndexMetadata;
32+
import org.opensearch.cluster.ClusterState;
33+
import org.opensearch.cluster.service.ClusterService;
34+
import org.opensearch.common.settings.Settings;
2935
import org.opensearch.common.unit.TimeValue;
3036
import org.opensearch.common.xcontent.XContentFactory;
3137
import org.opensearch.core.action.ActionListener;
3238
import org.opensearch.core.xcontent.ToXContent;
39+
import org.opensearch.index.IndexNotFoundException;
3340
import org.opensearch.plugin.insights.core.metrics.OperationalMetric;
3441
import org.opensearch.plugin.insights.core.metrics.OperationalMetricsCounter;
35-
import org.opensearch.plugin.insights.core.service.TopQueriesService;
3642
import org.opensearch.plugin.insights.rules.model.SearchQueryRecord;
3743

3844
/**
3945
* Local index exporter for exporting query insights data to local OpenSearch indices.
4046
*/
41-
public final class LocalIndexExporter implements QueryInsightsExporter {
47+
public class LocalIndexExporter implements QueryInsightsExporter {
4248
/**
4349
* Logger of the local index exporter
4450
*/
4551
private final Logger logger = LogManager.getLogger();
4652
private final Client client;
53+
private final ClusterService clusterService;
54+
private final String indexMapping;
4755
private DateTimeFormatter indexPattern;
4856
private int deleteAfter;
57+
private final String id;
58+
private static final int DEFAULT_NUMBER_OF_REPLICA = 1;
59+
private static final int DEFAULT_NUMBER_OF_SHARDS = 1;
60+
private static final List<String> DEFAULT_SORTED_FIELDS = List.of(
61+
"measurements.latency.number",
62+
"measurements.cpu.number",
63+
"measurements.memory.number"
64+
);
65+
private static final List<String> DEFAULT_SORTED_ORDERS = List.of("desc", "desc", "desc");
4966

5067
/**
5168
* Constructor of LocalIndexExporter
5269
*
5370
* @param client OS client
71+
* @param clusterService cluster service
5472
* @param indexPattern the pattern of index to export to
73+
* @param indexMapping the index mapping file
74+
* @param id id of the exporter
5575
*/
56-
public LocalIndexExporter(final Client client, final DateTimeFormatter indexPattern) {
76+
public LocalIndexExporter(
77+
final Client client,
78+
final ClusterService clusterService,
79+
final DateTimeFormatter indexPattern,
80+
final String indexMapping,
81+
final String id
82+
) {
5783
this.indexPattern = indexPattern;
5884
this.client = client;
85+
this.clusterService = clusterService;
86+
this.indexMapping = indexMapping;
5987
this.deleteAfter = DEFAULT_DELETE_AFTER_VALUE;
88+
this.id = id;
89+
}
90+
91+
@Override
92+
public String getId() {
93+
return id;
6094
}
6195

6296
/**
@@ -73,7 +107,7 @@ public DateTimeFormatter getIndexPattern() {
73107
*
74108
* @param indexPattern index pattern
75109
*/
76-
void setIndexPattern(DateTimeFormatter indexPattern) {
110+
public void setIndexPattern(DateTimeFormatter indexPattern) {
77111
this.indexPattern = indexPattern;
78112
}
79113

@@ -89,28 +123,76 @@ public void export(final List<SearchQueryRecord> records) {
89123
}
90124
try {
91125
final String indexName = buildLocalIndexName();
92-
final BulkRequestBuilder bulkRequestBuilder = client.prepareBulk().setTimeout(TimeValue.timeValueMinutes(1));
93-
for (SearchQueryRecord record : records) {
94-
bulkRequestBuilder.add(
95-
new IndexRequest(indexName).source(record.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS))
126+
if (!checkIndexExists(indexName)) {
127+
CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
128+
129+
createIndexRequest.settings(
130+
Settings.builder()
131+
.putList("index.sort.field", DEFAULT_SORTED_FIELDS)
132+
.putList("index.sort.order", DEFAULT_SORTED_ORDERS)
133+
.put("index.number_of_shards", DEFAULT_NUMBER_OF_SHARDS)
134+
.put("index.number_of_replicas", DEFAULT_NUMBER_OF_REPLICA)
96135
);
136+
createIndexRequest.mapping(readIndexMappings());
137+
138+
client.admin().indices().create(createIndexRequest, new ActionListener<>() {
139+
@Override
140+
public void onResponse(CreateIndexResponse createIndexResponse) {
141+
if (createIndexResponse.isAcknowledged()) {
142+
try {
143+
bulk(indexName, records);
144+
} catch (IOException e) {
145+
OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_EXCEPTIONS);
146+
logger.error("Unable to index query insights data: ", e);
147+
}
148+
}
149+
}
150+
151+
@Override
152+
public void onFailure(Exception e) {
153+
Throwable cause = ExceptionsHelper.unwrapCause(e);
154+
if (cause instanceof ResourceAlreadyExistsException) {
155+
try {
156+
bulk(indexName, records);
157+
} catch (IOException ex) {
158+
OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_EXCEPTIONS);
159+
logger.error("Unable to index query insights data: ", e);
160+
}
161+
} else {
162+
OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_EXCEPTIONS);
163+
logger.error("Unable to create query insights index: ", e);
164+
}
165+
}
166+
});
167+
} else {
168+
bulk(indexName, records);
97169
}
98-
bulkRequestBuilder.execute(new ActionListener<BulkResponse>() {
99-
@Override
100-
public void onResponse(BulkResponse bulkItemResponses) {}
101-
102-
@Override
103-
public void onFailure(Exception e) {
104-
OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_BULK_FAILURES);
105-
logger.error("Failed to execute bulk operation for query insights data: ", e);
106-
}
107-
});
108170
} catch (final Exception e) {
109171
OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_EXCEPTIONS);
110172
logger.error("Unable to index query insights data: ", e);
111173
}
112174
}
113175

176+
private void bulk(final String indexName, final List<SearchQueryRecord> records) throws IOException {
177+
final BulkRequestBuilder bulkRequestBuilder = client.prepareBulk().setTimeout(TimeValue.timeValueMinutes(1));
178+
for (SearchQueryRecord record : records) {
179+
bulkRequestBuilder.add(
180+
new IndexRequest(indexName).id(record.getId())
181+
.source(record.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS))
182+
);
183+
}
184+
bulkRequestBuilder.execute(new ActionListener<BulkResponse>() {
185+
@Override
186+
public void onResponse(BulkResponse bulkItemResponses) {}
187+
188+
@Override
189+
public void onFailure(Exception e) {
190+
OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_BULK_FAILURES);
191+
logger.error("Failed to execute bulk operation for query insights data: ", e);
192+
}
193+
});
194+
}
195+
114196
/**
115197
* Close the exporter sink
116198
*/
@@ -125,7 +207,8 @@ public void close() {
125207
* @return A string representing the index name in the format "top_queries-YYYY.MM.dd-01234".
126208
*/
127209
String buildLocalIndexName() {
128-
return indexPattern.format(ZonedDateTime.now(ZoneOffset.UTC)) + "-" + generateLocalIndexDateHash();
210+
ZonedDateTime currentTime = ZonedDateTime.now(ZoneOffset.UTC);
211+
return indexPattern.format(currentTime) + "-" + generateLocalIndexDateHash(currentTime.toLocalDate());
129212
}
130213

131214
/**
@@ -138,33 +221,58 @@ public void setDeleteAfter(final int deleteAfter) {
138221
}
139222

140223
/**
141-
* Delete Top N local indices older than the configured data retention period
224+
* Get local index exporter data retention period
142225
*
143-
* @param indexMetadataMap Map of index name {@link String} to {@link IndexMetadata}
226+
* @return the number of days after which Top N local indices should be deleted
144227
*/
145-
public void deleteExpiredTopNIndices(final Map<String, IndexMetadata> indexMetadataMap) {
146-
long expirationMillisLong = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(deleteAfter);
147-
for (Map.Entry<String, IndexMetadata> entry : indexMetadataMap.entrySet()) {
148-
String indexName = entry.getKey();
149-
if (isTopQueriesIndex(indexName) && entry.getValue().getCreationDate() <= expirationMillisLong) {
150-
// delete this index
151-
TopQueriesService.deleteSingleIndex(indexName, client);
152-
}
153-
}
228+
public int getDeleteAfter() {
229+
return deleteAfter;
154230
}
155231

156232
/**
157-
* Generates a consistent 5-digit numeric hash based on the current UTC date.
158-
* The generated hash is deterministic, meaning it will return the same result for the same date.
233+
* Deletes the specified index and logs any failure that occurs during the operation.
159234
*
160-
* @return A 5-digit numeric string representation of the current date's hash.
235+
* @param indexName The name of the index to delete.
236+
* @param client The OpenSearch client used to perform the deletion.
237+
*/
238+
public void deleteSingleIndex(String indexName, Client client) {
239+
Logger logger = LogManager.getLogger();
240+
client.admin().indices().delete(new DeleteIndexRequest(indexName), new ActionListener<>() {
241+
@Override
242+
// CS-SUPPRESS-SINGLE: RegexpSingleline It is not possible to use phrase "cluster manager" instead of master here
243+
public void onResponse(org.opensearch.action.support.master.AcknowledgedResponse acknowledgedResponse) {}
244+
245+
@Override
246+
public void onFailure(Exception e) {
247+
Throwable cause = ExceptionsHelper.unwrapCause(e);
248+
if (cause instanceof IndexNotFoundException) {
249+
return;
250+
}
251+
OperationalMetricsCounter.getInstance().incrementCounter(OperationalMetric.LOCAL_INDEX_EXPORTER_DELETE_FAILURES);
252+
logger.error("Failed to delete index '{}': ", indexName, e);
253+
}
254+
});
255+
}
256+
257+
/**
258+
* check if index exists
259+
* @return boolean
161260
*/
162-
public static String generateLocalIndexDateHash() {
163-
// Get the current date in UTC (yyyy-MM-dd format)
164-
String currentDate = DateTimeFormatter.ofPattern("yyyy-MM-dd", Locale.ROOT)
165-
.format(Instant.now().atOffset(ZoneOffset.UTC).toLocalDate());
261+
private boolean checkIndexExists(String indexName) {
262+
ClusterState clusterState = clusterService.state();
263+
return clusterState.getRoutingTable().hasIndex(indexName);
264+
}
166265

167-
// Generate a 5-digit numeric hash from the date's hashCode
168-
return String.format(Locale.ROOT, "%05d", (currentDate.hashCode() % 100000 + 100000) % 100000);
266+
/**
267+
* get correlation rule index mappings
268+
* @return mappings of correlation rule index
269+
* @throws IOException IOException
270+
*/
271+
private String readIndexMappings() throws IOException {
272+
return new String(
273+
Objects.requireNonNull(LocalIndexExporter.class.getClassLoader().getResourceAsStream(indexMapping)).readAllBytes(),
274+
Charset.defaultCharset()
275+
);
169276
}
277+
170278
}

‎src/main/java/org/opensearch/plugin/insights/core/exporter/QueryInsightsExporter.java

+2
Original file line numberDiff line numberDiff line change
@@ -22,4 +22,6 @@ public interface QueryInsightsExporter extends Closeable {
2222
* @param records list of {@link SearchQueryRecord}
2323
*/
2424
void export(final List<SearchQueryRecord> records);
25+
26+
String getId();
2527
}

0 commit comments

Comments
 (0)