Skip to content

Commit

Permalink
[backport to 2.x] Enforce DOCUMENT Replication for AD Indices
Browse files Browse the repository at this point in the history 
In this PR, we temporarily enforce DOCUMENT replication for AD indices. This change is necessary due to the current limitation of SegRep, which doesn't support Get/MultiGet by ID. This measure will be in place until SegRep adds support for these operations.

This adjustment aligns with the modification made in the referenced PR: opensearch-project/job-scheduler#417

Signed-off-by: Kaituo Li <[email protected]>
  • Loading branch information
@kaituo
kaituo committed Jul 11, 2023
1 parent 0b9be84 commit 5ef9fc5
Showing 1 changed file with 27 additions and 6 deletions.
33 changes: 27 additions & 6 deletions src/main/java/org/opensearch/ad/indices/AnomalyDetectionIndices.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
import static org.opensearch.ad.settings.AnomalyDetectorSettings.ANOMALY_RESULTS_INDEX_MAPPING_FILE;
import static org.opensearch.ad.settings.AnomalyDetectorSettings.CHECKPOINT_INDEX_MAPPING_FILE;
import static org.opensearch.ad.settings.AnomalyDetectorSettings.MAX_PRIMARY_SHARDS;
import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_REPLICATION_TYPE;
import static org.opensearch.indices.replication.common.ReplicationType.DOCUMENT;

import java.io.IOException;
import java.net.URL;
Expand Down Expand Up @@ -529,7 +531,10 @@ public void initAnomalyDetectorIndexIfAbsent(ActionListener<CreateIndexResponse>
* @throws IOException IOException from {@link AnomalyDetectionIndices#getAnomalyDetectorMappings}
*/
public void initAnomalyDetectorIndex(ActionListener<CreateIndexResponse> actionListener) throws IOException {
CreateIndexRequest request = new CreateIndexRequest(AnomalyDetector.ANOMALY_DETECTORS_INDEX)
// AD indices need RAW (e.g., we want users to be able to consume AD results as soon as possible and send out an alert if
// anomalies found).
Settings replicationSettings = Settings.builder().put(SETTING_REPLICATION_TYPE, DOCUMENT.name()).build();
CreateIndexRequest request = new CreateIndexRequest(AnomalyDetector.ANOMALY_DETECTORS_INDEX, replicationSettings)
.mapping(getAnomalyDetectorMappings(), XContentType.JSON)
.settings(settings);
adminClient.indices().create(request, markMappingUpToDate(ADIndex.CONFIG, actionListener));
Expand Down Expand Up @@ -594,7 +599,10 @@ public void initAnomalyResultIndexDirectly(
ActionListener<CreateIndexResponse> actionListener
) throws IOException {
String mapping = getAnomalyResultMappings();
CreateIndexRequest request = new CreateIndexRequest(resultIndex).mapping(mapping, XContentType.JSON);
// AD indices need RAW (e.g., we want users to be able to consume AD results as soon as possible and send out an alert if
// anomalies found).
Settings replicationSettings = Settings.builder().put(SETTING_REPLICATION_TYPE, DOCUMENT.name()).build();
CreateIndexRequest request = new CreateIndexRequest(resultIndex, replicationSettings).mapping(mapping, XContentType.JSON);
if (alias != null) {
request.alias(new Alias(CommonName.ANOMALY_RESULT_INDEX_ALIAS));
}
Expand All @@ -613,7 +621,10 @@ public void initAnomalyResultIndexDirectly(
*/
public void initAnomalyDetectorJobIndex(ActionListener<CreateIndexResponse> actionListener) {
try {
CreateIndexRequest request = new CreateIndexRequest(AnomalyDetectorJob.ANOMALY_DETECTOR_JOB_INDEX)
// AD indices need RAW (e.g., we want users to be able to consume AD results as soon as possible and send out an alert if
// anomalies found).
Settings replicationSettings = Settings.builder().put(SETTING_REPLICATION_TYPE, DOCUMENT.name()).build();
CreateIndexRequest request = new CreateIndexRequest(AnomalyDetectorJob.ANOMALY_DETECTOR_JOB_INDEX, replicationSettings)
.mapping(getAnomalyDetectorJobMappings(), XContentType.JSON);
request
.settings(
Expand Down Expand Up @@ -645,7 +656,10 @@ public void initAnomalyDetectorJobIndex(ActionListener<CreateIndexResponse> acti
*/
public void initDetectionStateIndex(ActionListener<CreateIndexResponse> actionListener) {
try {
CreateIndexRequest request = new CreateIndexRequest(CommonName.DETECTION_STATE_INDEX)
// AD indices need RAW (e.g., we want users to be able to consume AD results as soon as possible and send out an alert if
// anomalies found).
Settings replicationSettings = Settings.builder().put(SETTING_REPLICATION_TYPE, DOCUMENT.name()).build();
CreateIndexRequest request = new CreateIndexRequest(CommonName.DETECTION_STATE_INDEX, replicationSettings)
.mapping(getDetectionStateMappings(), XContentType.JSON)
.settings(settings);
adminClient.indices().create(request, markMappingUpToDate(ADIndex.STATE, actionListener));
Expand All @@ -668,7 +682,11 @@ public void initCheckpointIndex(ActionListener<CreateIndexResponse> actionListen
} catch (IOException e) {
throw new EndRunException("", "Cannot find checkpoint mapping file", true);
}
CreateIndexRequest request = new CreateIndexRequest(CommonName.CHECKPOINT_INDEX_NAME).mapping(mapping, XContentType.JSON);
// AD indices need RAW (e.g., we want users to be able to consume AD results as soon as possible and send out an alert if
// anomalies found).
Settings replicationSettings = Settings.builder().put(SETTING_REPLICATION_TYPE, DOCUMENT.name()).build();
CreateIndexRequest request = new CreateIndexRequest(CommonName.CHECKPOINT_INDEX_NAME, replicationSettings)
.mapping(mapping, XContentType.JSON);
choosePrimaryShards(request);
adminClient.indices().create(request, markMappingUpToDate(ADIndex.CHECKPOINT, actionListener));
}
Expand Down Expand Up @@ -725,7 +743,10 @@ void rolloverAndDeleteHistoryIndex() {
}
CreateIndexRequest createRequest = rollOverRequest.getCreateIndexRequest();

createRequest.index(AD_RESULT_HISTORY_INDEX_PATTERN).mapping(adResultMapping, XContentType.JSON);
// time series indices need RAW (e.g., we want users to be able to consume AD results as soon as possible
// and send out an alert if anomalies found).
Settings replicationSettings = Settings.builder().put(SETTING_REPLICATION_TYPE, DOCUMENT.name()).build();
createRequest.index(AD_RESULT_HISTORY_INDEX_PATTERN).settings(replicationSettings).mapping(adResultMapping, XContentType.JSON);

choosePrimaryShards(createRequest);

Expand Down

0 comments on commit 5ef9fc5

Please sign in to comment.