ism index patterns based on indexAlias/index #5117

Signed-off-by: kranthikirang <[email protected]>
opensearch-project · Oct 26, 2024 · d06fcb8 · d06fcb8
1 parent 9065acb
commit d06fcb8
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 6 deletions.
diff --git a/...n/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/AbstractIndexManager.java b/...n/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/AbstractIndexManager.java
@@ -228,7 +228,7 @@ public void setupIndex() throws IOException {
     private void checkAndCreateIndexTemplate() throws IOException {
         final boolean isISMEnabled = checkISMEnabled();
         final Optional<String> policyIdOptional = isISMEnabled ?
-                ismPolicyManagementStrategy.checkAndCreatePolicy() :
+                ismPolicyManagementStrategy.checkAndCreatePolicy(configuredIndexAlias) :
                 Optional.empty();
         if (!openSearchSinkConfiguration.getIndexConfiguration().getIndexTemplate().isEmpty()) {
             checkAndCreateIndexTemplate(isISMEnabled, policyIdOptional.orElse(null));
@@ -258,8 +258,8 @@ final void checkAndCreateIndexTemplate(final boolean isISMEnabled, final String
         templateStrategy.createTemplate(indexTemplate);
     }
 
-    final Optional<String> checkAndCreatePolicy() throws IOException {
-        return ismPolicyManagementStrategy.checkAndCreatePolicy();
+    final Optional<String> checkAndCreatePolicy(final String indexAlias) throws IOException {
+        return ismPolicyManagementStrategy.checkAndCreatePolicy(indexAlias);
     }
 
     public void checkAndCreateIndex() throws IOException {
@@ -322,6 +322,7 @@ private void attachPolicy(
             indexTemplate.putCustomSetting(IndexConstants.ISM_POLICY_ID_SETTING, ismPolicyId);
         }
         indexTemplate.putCustomSetting(IndexConstants.ISM_ROLLOVER_ALIAS_SETTING, rolloverAlias);
+	indexTemplate.putCustomSetting(IndexConstants.PLUGINS_ROLLOVER_ALIAS_SETTING, rolloverAlias);
     }
 
 }
diff --git a/...rc/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/IndexConstants.java b/...rc/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/IndexConstants.java
@@ -21,6 +21,7 @@ public class IndexConstants {
   public static final String ISM_ENABLED_SETTING = "opendistro.index_state_management.enabled";
   public static final String ISM_POLICY_ID_SETTING = "opendistro.index_state_management.policy_id";
   public static final String ISM_ROLLOVER_ALIAS_SETTING = "opendistro.index_state_management.rollover_alias";
+  public static final String PLUGINS_ROLLOVER_ALIAS_SETTING = "plugins.index_state_management.rollover_alias";
   // TODO: extract out version number into version enum
   public static final String SERVICE_MAP_DEFAULT_TEMPLATE_FILE = "otel-v1-apm-service-map-index-template.json";
 

diff --git a/...in/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/IsmPolicyManagement.java b/...in/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/IsmPolicyManagement.java
@@ -10,6 +10,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
 import io.micrometer.core.instrument.util.StringUtils;
 import org.opensearch.client.Request;
 import org.opensearch.client.ResponseException;
@@ -93,10 +94,19 @@ public IsmPolicyManagement(final OpenSearchClient openSearchClient,
     }
 
     @Override
-    public Optional<String> checkAndCreatePolicy() throws IOException {
+    public Optional<String> checkAndCreatePolicy(final String indexAlias) throws IOException {
         final String policyManagementEndpoint = POLICY_MANAGEMENT_ENDPOINT + policyName;
 
         String policyJsonString = retrievePolicyJsonString(policyFile);
+	if(!indexAlias.isEmpty()) {
+                final ObjectMapper mapper = new ObjectMapper();
+                final JsonNode jsonNode = mapper.readTree(policyJsonString);
+                final ArrayNode iparray = mapper.createArrayNode();
+                iparray.add(indexAlias + "*");
+                ((ObjectNode) jsonNode.get("policy").get("ism_template")).put("index_patterns", iparray);
+                policyJsonString = jsonNode.toString();
+        }
+        LOG.debug("Got the policystring as {} and indexAlias as {}", policyJsonString, indexAlias);
         Request request = createPolicyRequestFromFile(policyManagementEndpoint, policyJsonString);
 
         try {

diff --git a/...org/opensearch/dataprepper/plugins/sink/opensearch/index/IsmPolicyManagementStrategy.java b/...org/opensearch/dataprepper/plugins/sink/opensearch/index/IsmPolicyManagementStrategy.java
@@ -13,7 +13,7 @@
 
 interface IsmPolicyManagementStrategy {
 
-    Optional<String> checkAndCreatePolicy() throws IOException;
+    Optional<String> checkAndCreatePolicy(final String indexAlias) throws IOException;
 
     List<String> getIndexPatterns(final String indexAlias);
 

diff --git a/.../java/org/opensearch/dataprepper/plugins/sink/opensearch/index/NoIsmPolicyManagement.java b/.../java/org/opensearch/dataprepper/plugins/sink/opensearch/index/NoIsmPolicyManagement.java
@@ -34,7 +34,7 @@ public NoIsmPolicyManagement(final OpenSearchClient openSearchClient,
     }
 
     @Override
-    public Optional<String> checkAndCreatePolicy() throws IOException {
+    public Optional<String> checkAndCreatePolicy(final String indexAlias) throws IOException {
         return Optional.empty();
     }