Merge pull request #1969 from utmstack/hotfix/index_removal

Address issue #1968

Author: osmontero

backend/src/main/java/com/park/utmstack/service/elasticsearch/ElasticsearchService.java

@@ -10,6 +10,7 @@
 import com.park.utmstack.service.MailService;
 import com.park.utmstack.service.UtmSpaceNotificationControlService;
 import com.park.utmstack.service.application_events.ApplicationEventService;
+import com.park.utmstack.service.index_policy.IndexPolicyService;
 import com.park.utmstack.util.chart_builder.IndexPropertyType;
 import com.park.utmstack.util.exceptions.OpenSearchIndexNotFoundException;
 import com.park.utmstack.util.exceptions.UtmElasticsearchException;
@@ -43,6 +44,7 @@
 import java.time.temporal.ChronoUnit;
 import java.util.*;
 import java.util.stream.Collectors;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author Leonardo M. López
@@ -56,16 +58,19 @@ public class ElasticsearchService {
     private final UserRepository userRepository;
     private final MailService mailService;
     private final UtmSpaceNotificationControlService spaceNotificationControlService;
+    private final IndexPolicyService indexPolicyService;
     private final OpensearchClientBuilder client;
 
     public ElasticsearchService(ApplicationEventService eventService, UserRepository userRepository,
                                 MailService mailService,
                                 UtmSpaceNotificationControlService spaceNotificationControlService,
+                                IndexPolicyService indexPolicyService,
                                 OpensearchClientBuilder client) {
         this.eventService = eventService;
         this.userRepository = userRepository;
         this.mailService = mailService;
         this.spaceNotificationControlService = spaceNotificationControlService;
+        this.indexPolicyService = indexPolicyService;
         this.client = client;
     }
 
@@ -278,6 +283,16 @@ private void deleteOldestIndices() {
 
             // Indices are returned from oldest to newest ordered by creation.date asc
             for (IndicesRecord index : indices) {
+                Optional<ElasticCluster> opt = getClusterStatus();
+
+                if (opt.isEmpty() || opt.get().getResume().getDiskUsedPercent() < 70)
+                    break;
+
+                if (!indexPolicyService.isIndexRemovable(index.index())) {
+                    log.info("{} Skipping index {} because it is not in a removable state", ctx, index.index());
+                    continue;
+                }
+
                 try {
                     // Delete oldest indices
                     deleteIndex(Collections.singletonList(index.index()));
@@ -286,15 +301,12 @@ private void deleteOldestIndices() {
                                     "Docs Count: %3$s\n" +
                                     "Size: %4$s",
                             index.index(), index.creationDateString(), index.docsCount(), index.storeSize()), ApplicationEventType.INFO);
+                    TimeUnit.SECONDS.sleep(10);
                 } catch (Exception e) {
                     String msg = String.format("%1$s: Fail to delete index: %2$s with message: %3$s", ctx, index.index(), e.getMessage());
                     eventService.createEvent(msg, ApplicationEventType.WARNING);
                 }
 
-                Optional<ElasticCluster> opt = getClusterStatus();
-
-                if (opt.isEmpty() || opt.get().getResume().getDiskUsedPercent() < 70)
-                    break;
             }
         } catch (Exception e) {
             String msg = String.format("%1$s: %2$s", ctx, e.getMessage());
@@ -407,4 +419,4 @@ public <T> SearchSqlResponse<T> searchBySql(SqlQueryRequest request, Class<T> re
             throw new RuntimeException(ctx + ": " + e.getMessage());
         }
     }
-}
+}

backend/src/main/java/com/park/utmstack/service/index_policy/IndexPolicyService.java

@@ -1,6 +1,7 @@
 package com.park.utmstack.service.index_policy;
 
 import com.google.gson.Gson;
+import com.google.gson.JsonObject;
 import com.park.utmstack.config.Constants;
 import com.park.utmstack.domain.index_pattern.enums.SystemIndexPattern;
 import com.park.utmstack.domain.index_policy.*;
@@ -37,6 +38,59 @@ public IndexPolicyService(ApplicationContext applicationContext,
         this.client = client;
     }
 
+    /**
+     * Check if an index is in a removable state (STATE_DELETE or STATE_SAFE_DELETE)
+     *
+     * @param index: Index name
+     * @return True if removable, false otherwise
+     */
+    public boolean isIndexRemovable(String index) {
+        final String ctx = CLASSNAME + ".isIndexRemovable";
+        try {
+            final String url = String.format("_plugins/_ism/explain/%1$s", index);
+            try (Response rs = client.getClient().executeHttpRequest(url, null, null, HttpMethod.GET)) {
+                if (!rs.isSuccessful() || rs.body() == null)
+                    return false;
+
+                String body = rs.body().string();
+                JsonObject json = new Gson().fromJson(body, JsonObject.class);
+                if (json != null && json.has(index)) {
+                    JsonObject indexInfo = json.getAsJsonObject(index);
+                    String state = this.getCurrentState(indexInfo)
+                    if (state != null) {
+                        return Constants.STATE_DELETE.equals(state) || Constants.STATE_SAFE_DELETE.equals(state);
+                    }
+                }
+            }
+            return false;
+        } catch (Exception e) {
+            log.error("{}: {}", ctx, e.getMessage());
+            return false;
+        }
+    }
+
+
+    private String getCurrentState(JsonObject indexInfo) {
+        if (indexInfo.has("state") && indexInfo.get("state").isJsonObject()) {
+            JsonObject stateObj = indexInfo.getAsJsonObject("state");
+            if (stateObj.has("name")) {
+                return stateObj.get("name").getAsString();
+            }
+        }
+
+        String[] legacyPaths = {
+            "index.plugins.index_state_management.current_state",
+            "index.opendistro.index_state_management.current_state",
+            "opendistro.index_state_management.current_state"
+        };
+
+        return Arrays.stream(legacyPaths)
+            .filter(indexInfo::has)
+            .map(path -> indexInfo.get(path).getAsString())
+            .findFirst()
+            .orElse(null);
+    }
+
     @EventListener(IndexPatternsReadyEvent.class)
     public void init() throws Exception {
         final String ctx = CLASSNAME + ".init";