Automatic purge of alarm log indices

services/alarm-logger/doc/index.rst

@@ -32,3 +32,19 @@ Examples:
 * **Commands** 
 
  e.g. a user actions to *Acknowledge* an alarm
+
+****************************************
+Automatic purge of Elasticsearch indices
+****************************************
+
+To avoid issues related to a high number of Elasticsearch indices, automatic purge can be enabled in order to delete
+indices considered obsolete. This is done by setting the preferences ``date_span_units`` and ``retain_indices_count`` such
+that they evaluate to a number larger or equal to 100. The default ``retain_indices_count`` is 0, i.e. automatic purge is disabled by default.
+
+The automatic purge is run using a cron expression defined in preference ``purge_cron_expr``, default is
+``0 0 0 * * SUN``, i.e. midnight each Sunday. See the SpringDocumentation_ on how to define the cron expression.
+
+An Elasticsearch index is considered eligible for deletion if the last inserted message date is before current time
+minus the number of days computed from ``date_span_units`` and ``retain_indices_count``.
+
+.. _SpringDocumentation: https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/scheduling/support/CronExpression.html

services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmLoggingConfiguration.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright (C) 2023 European Spallation Source ERIC.
+ *
+ *  This program is free software; you can redistribute it and/or
+ *  modify it under the terms of the GNU General Public License
+ *  as published by the Free Software Foundation; either version 2
+ *  of the License, or (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+ *
+ */
+
+package org.phoebus.alarm.logging;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class AlarmLoggingConfiguration {
+
+    @Value("${days:150}")
+    public int days;
+
+    @Bean
+    public int getDays(){
+        return days;
+    }
+}

services/alarm-logger/src/main/java/org/phoebus/alarm/logging/AlarmLoggingService.java

@@ -20,8 +20,11 @@
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 @SpringBootApplication
+@EnableScheduling
 public class AlarmLoggingService {
 
     /** Alarm system logger */

services/alarm-logger/src/main/java/org/phoebus/alarm/logging/purge/ElasticIndexPurger.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright (C) 2023 European Spallation Source ERIC.
+ *
+ *  This program is free software; you can redistribute it and/or
+ *  modify it under the terms of the GNU General Public License
+ *  as published by the Free Software Foundation; either version 2
+ *  of the License, or (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+ *
+ */
+
+package org.phoebus.alarm.logging.purge;
+
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
+import co.elastic.clients.elasticsearch._types.FieldSort;
+import co.elastic.clients.elasticsearch._types.SortOptions;
+import co.elastic.clients.elasticsearch._types.SortOrder;
+import co.elastic.clients.elasticsearch._types.query_dsl.MatchAllQuery;
+import co.elastic.clients.elasticsearch.cat.IndicesResponse;
+import co.elastic.clients.elasticsearch.cat.indices.IndicesRecord;
+import co.elastic.clients.elasticsearch.core.SearchRequest;
+import co.elastic.clients.elasticsearch.core.SearchResponse;
+import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest;
+import co.elastic.clients.elasticsearch.indices.DeleteIndexResponse;
+import org.phoebus.alarm.logging.ElasticClientHelper;
+import org.phoebus.alarm.logging.rest.AlarmLogMessage;
+import org.phoebus.alarm.logging.rest.AlarmLogSearchUtil;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.io.IOException;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * Utility class purging Elasticsearch from indices considered obsolete based on the date_span_units and retain_indices_count
+ * application properties. If these result in a value below 100 (days), this {@link Component} will not be instantiated.
+ * To determine last updated date of an index, each Elasticsearch index considered related to alarms is queried for last
+ * inserted document. The message_time field of that document is compared to the retention period to determine
+ * if the index should be deleted.
+ * A cron expression application property is used to define when to run the purging process.
+ */
+@Component
+// Enable only of retention period is >= 100 days
+@ConditionalOnExpression("#{T(org.phoebus.alarm.logging.purge.ElasticIndexPurger.EnableCondition).getRetentionDays('${date_span_units}', '${retain_indices_count}') >= 100}")
+public class ElasticIndexPurger {
+
+    private static final Logger logger = Logger.getLogger(ElasticIndexPurger.class.getName());
+
+    private ElasticsearchClient elasticsearchClient;
+
+    @SuppressWarnings("unused")
+    @Value("${retention_period_days:0}")
+    private int retentionPeriod;
+
+    @SuppressWarnings("unused")
+    @PostConstruct
+    public void init() {
+        elasticsearchClient = ElasticClientHelper.getInstance().getClient();
+    }
+
+    /**
+     * Deletes Elasticsearch indices based on the {@link AlarmLogMessage#getMessage_time()} for each index found
+     * by the client. The message time {@link Instant} is compared to current time minus the number of days specified as
+     * application property.
+     */
+    @SuppressWarnings("unused")
+    @Scheduled(cron = "${purge_cron_expr}")
+    public void purgeElasticIndices() {
+        try {
+            IndicesResponse indicesResponse = elasticsearchClient.cat().indices();
+            List<IndicesRecord> indicesRecords = indicesResponse.valueBody();
+            Instant toInstant = Instant.now().minus(retentionPeriod, ChronoUnit.DAYS);
+            for (IndicesRecord indicesRecord : indicesRecords) {
+                // Elasticsearch may contain indices other than alarm indices...
+                String indexName = indicesRecord.index();
+                if (indexName != null && !indexName.startsWith("_alarms") && (indexName.contains("_alarms_state") ||
+                        indexName.contains("_alarms_cmd") ||
+                        indexName.contains("_alarms_config"))) {
+                    // Find most recent document - based on message_time - in the alarm index.
+                    SearchRequest searchRequest = SearchRequest.of(s ->
+                            s.index(indexName)
+                                    .query(new MatchAllQuery.Builder().build()._toQuery())
+                                    .size(1)
+                                    .sort(SortOptions.of(so -> so.field(FieldSort.of(f -> f.field("message_time").order(SortOrder.Desc))))));
+                    SearchResponse<AlarmLogMessage> searchResponse = elasticsearchClient.search(searchRequest, AlarmLogMessage.class);
+                    if (!searchResponse.hits().hits().isEmpty()) {
+                        AlarmLogMessage alarmLogMessage = searchResponse.hits().hits().get(0).source();
+                        if (alarmLogMessage != null && alarmLogMessage.getMessage_time().isBefore(toInstant)) {
+                            DeleteIndexRequest deleteIndexRequest = DeleteIndexRequest.of(d -> d.index(indexName));
+                            DeleteIndexResponse deleteIndexResponse = elasticsearchClient.indices().delete(deleteIndexRequest);
+                            logger.log(Level.INFO, "Delete index " + indexName + " acknowledged: " + deleteIndexResponse.acknowledged());
+                        }
+                    } else {
+                        logger.log(Level.WARNING, "Index " + indexName + " cannot be evaluated for removal as document count is zero.");
+                    }
+                }
+            }
+        } catch (IOException e) {
+            logger.log(Level.WARNING, "Elastic query failed", e);
+        }
+    }
+
+    /**
+     * Helper class used to determine whether this service should be enabled or not
+     */
+    public static class EnableCondition {
+
+        /**
+         *
+         * @param dateSpanUnits Any of the values Y, M, W, D
+         * @param retainIndicesCountString String value of the retain_indices_count preference
+         * @return A number computed from input. In case input arguments are invalid (e.g. non-numerical value
+         * for retain_indices_coun), then 0 is returned to indicate that this {@link Component} should not be enabled.
+         */
+        @SuppressWarnings("unused")
+        public static int getRetentionDays(String dateSpanUnits, String retainIndicesCountString) {
+            int days = AlarmLogSearchUtil.getDateSpanInDays(dateSpanUnits);
+            if (days == -1) {
+                return 0;
+            }
+            try {
+                int retainIndicesCount = Integer.parseInt(retainIndicesCountString);
+                return days * retainIndicesCount;
+            } catch (NumberFormatException e) {
+                return 0;
+            }
+        }
+    }
+}

services/alarm-logger/src/main/java/org/phoebus/alarm/logging/rest/AlarmLogSearchUtil.java

@@ -360,21 +360,7 @@ public static List<String> findIndexNames(String baseIndexName, Instant fromInst
             if (fromIndex.equalsIgnoreCase(toIndex)) {
                 indexList.add(fromIndex);
             } else {
-                int indexDateSpanDayValue = -1;
-                switch (indexDateSpanUnits) {
-                    case "Y":
-                        indexDateSpanDayValue = 365;
-                        break;
-                    case "M":
-                        indexDateSpanDayValue = 30;
-                        break;
-                    case "W":
-                        indexDateSpanDayValue = 7;
-                        break;
-                    case "D":
-                        indexDateSpanDayValue = 1;
-                        break;
-                }
+                int indexDateSpanDayValue = getDateSpanInDays(indexDateSpanUnits);
                 indexList.add(fromIndex);
                 while (!fromIndex.equalsIgnoreCase(toIndex)) {
                     fromInstant = fromInstant.plus(indexDateSpanDayValue, ChronoUnit.DAYS);
@@ -386,4 +372,25 @@ public static List<String> findIndexNames(String baseIndexName, Instant fromInst
 
         return indexList;
     }
+
+    /**
+     *
+     * @param indexDateSpanUnits A single char string from [Y, M, W, D]
+     * @return Number of days corresponding to the unit, or -1 if the input does not match
+     * supported chars.
+     */
+    public static int getDateSpanInDays(String indexDateSpanUnits){
+        switch (indexDateSpanUnits) {
+            case "Y":
+                return 365;
+            case "M":
+                return 30;
+            case "W":
+                return 7;
+            case "D":
+                return 1;
+            default:
+                return -1;
+        }
+    }
 }

services/alarm-logger/src/main/resources/application.properties

@@ -43,4 +43,17 @@ thread_pool_size=4
 
 ############################## REST Logging ###############################
 # DEBUG level will log all requests and responses to and from the REST end points
-logging.level.org.springframework.web.filter.CommonsRequestLoggingFilter=INFO
+logging.level.org.springframework.web.filter.CommonsRequestLoggingFilter=INFO
+
+############################## Index purge settings #######################
+# How many indices to retain (e.g. if you have selected date_span_units as M and set the retain_indices_count to 6, then indices
+# older than 6 months will be deleted.
+# Number of days computed form this setting and date_span_units must be greater or equal to 100
+# for automatic purge to be enabled.
+retain_indices_count=0
+
+# Cron expression used by Spring scheduler running automatic purge, default every Sunday at midnight.
+# See https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/scheduling/support/CronExpression.html
+# Incorrect syntax will fail service startup if retention_period_days >= 100.
+purge_cron_expr=0 0 0 * * SUN
+##############################################################################