fix: resolve rollover issue with index and ism policy improvements

This commit addresses the issue where rollover was not resolving as intended, leading to a single
index accumulating all documents without purging.

BREAKING CHANGE: The index name has been changed from "actions" to "events", requiring manual
deletion of the old "actions" index.

Key changes:

1. The index configuration now uses 3 shards and 1 replica to distribute the load evenly across
   the cluster's 3 data nodes. The index name has been changed to `events` to enforce this setting.
2. An index template with a `rollover_alias` has been created to ensure rollover operations work
   correctly and don't hang.
3. Updated ISM policy:
  - Rollover now occurs daily, allowing for more granular purging after 14 days.
  - Introduced a warm state for indexes older than 7 days an action reduces the number of replicas
    to 0.

Author: jboix

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/model/EventRequest.kt

@@ -27,7 +27,7 @@ import org.springframework.data.elasticsearch.annotations.FieldType
  * @property data Additional data associated with the event.
  * @property session Session data associated with the event, potentially updated later.
  */
-@Document(indexName = "actions", createIndex = false)
+@Document(indexName = "events", createIndex = false)
 data class EventRequest(
   @Id
   @JsonIgnore

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/setup/AliasSetupTask.kt

@@ -9,7 +9,6 @@ import org.springframework.core.io.ResourceLoader
 import org.springframework.http.HttpHeaders
 import org.springframework.http.HttpStatusCode
 import org.springframework.http.MediaType
-import org.springframework.http.ResponseEntity
 import org.springframework.stereotype.Component
 import org.springframework.web.reactive.function.client.WebClient
 import reactor.core.publisher.Mono
@@ -24,7 +23,7 @@ import reactor.core.publisher.Mono
  * @property resourceLoader Resource loader used to access the alias configuration JSON file.
  */
 @Component
-@Order(3)
+@Order(4)
 class AliasSetupTask(
   @Qualifier("openSearchWebClient")
   private val webClient: WebClient,
@@ -39,7 +38,7 @@ class AliasSetupTask(
     /**
      * Name of the filtered alias.
      */
-    private const val ALIAS_NAME = "filtered_actions"
+    private const val ALIAS_NAME = "user_events"
 
     /**
      * Path to check for the existence of the alias in OpenSearch.
@@ -61,7 +60,7 @@ class AliasSetupTask(
    */
   override fun run(): Mono<*> = checkAndCreateAlias()
 
-  private fun checkAndCreateAlias(): Mono<ResponseEntity<Void>> =
+  private fun checkAndCreateAlias(): Mono<*> =
     webClient
       .get()
       .uri(ALIAS_CHECK_PATH)
@@ -74,7 +73,7 @@ class AliasSetupTask(
         Mono.empty()
       }.toBodilessEntity()
 
-  private fun createAlias(): Mono<ResponseEntity<Void>> {
+  private fun createAlias(): Mono<*> {
     val indexTemplateJson = resourceLoader.loadResourceContent("classpath:opensearch/alias.json")
     return webClient
       .put()

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/setup/ISMPolicySetupTask.kt

@@ -9,7 +9,6 @@ import org.springframework.core.io.ResourceLoader
 import org.springframework.http.HttpHeaders
 import org.springframework.http.HttpStatusCode
 import org.springframework.http.MediaType
-import org.springframework.http.ResponseEntity
 import org.springframework.stereotype.Component
 import org.springframework.web.reactive.function.client.WebClient
 import reactor.core.publisher.Mono
@@ -39,7 +38,7 @@ class ISMPolicySetupTask(
     /**
      * Path for the Index State Management (ISM) policy in OpenSearch.
      */
-    private const val ISM_POLICY_PATH = "/_plugins/_ism/policies/actions_policy"
+    private const val ISM_POLICY_PATH = "/_plugins/_ism/policies/events_policy"
   }
 
   /**
@@ -51,7 +50,7 @@ class ISMPolicySetupTask(
    */
   override fun run(): Mono<*> = checkAndApplyISMPolicy()
 
-  private fun checkAndApplyISMPolicy(): Mono<ResponseEntity<Void>> =
+  private fun checkAndApplyISMPolicy(): Mono<*> =
     webClient
       .get()
       .uri(ISM_POLICY_PATH)
@@ -64,7 +63,7 @@ class ISMPolicySetupTask(
         Mono.empty()
       }.toBodilessEntity()
 
-  private fun applyISMPolicy(): Mono<ResponseEntity<Void>> {
+  private fun applyISMPolicy(): Mono<*> {
     val ismPolicyJson = resourceLoader.loadResourceContent("classpath:opensearch/ism_policy.json")
 
     return webClient

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/setup/IndexSetupTask.kt

@@ -9,7 +9,6 @@ import org.springframework.core.io.ResourceLoader
 import org.springframework.http.HttpHeaders
 import org.springframework.http.HttpStatusCode
 import org.springframework.http.MediaType
-import org.springframework.http.ResponseEntity
 import org.springframework.stereotype.Component
 import org.springframework.web.reactive.function.client.WebClient
 import reactor.core.publisher.Mono
@@ -25,7 +24,7 @@ import reactor.core.publisher.Mono
  * @property resourceLoader Resource loader used to access the index template JSON file.
  */
 @Component
-@Order(2)
+@Order(3)
 class IndexSetupTask(
   @Qualifier("openSearchWebClient")
   private val webClient: WebClient,
@@ -40,7 +39,12 @@ class IndexSetupTask(
     /**
      * Path for creating the OpenSearch index.
      */
-    private const val INDEX_CREATION_PATH = "/actions-000001"
+    private const val INDEX_CREATION_PATH = "/events-000001"
+
+    /**
+     * Path to check if the index exists.
+     */
+    private const val INDEX_CHECK_PATH = "/events"
   }
 
   /**
@@ -52,10 +56,10 @@ class IndexSetupTask(
    */
   override fun run(): Mono<*> = checkAndCreateIndex()
 
-  private fun checkAndCreateIndex(): Mono<ResponseEntity<Void>> =
+  private fun checkAndCreateIndex(): Mono<*> =
     webClient
       .head()
-      .uri(INDEX_CREATION_PATH)
+      .uri(INDEX_CHECK_PATH)
       .retrieve()
       .onStatus(HttpStatusCode::is4xxClientError) {
         logger.info("Index does not exist, creating index...")
@@ -65,13 +69,13 @@ class IndexSetupTask(
         Mono.empty()
       }.toBodilessEntity()
 
-  private fun createIndex(): Mono<ResponseEntity<Void>> {
-    val indexTemplateJson = resourceLoader.loadResourceContent("classpath:opensearch/index_template.json")
+  private fun createIndex(): Mono<*> {
+    val indexJson = resourceLoader.loadResourceContent("classpath:opensearch/index.json")
     return webClient
       .put()
       .uri(INDEX_CREATION_PATH)
       .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
-      .bodyValue(indexTemplateJson)
+      .bodyValue(indexJson)
       .retrieve()
       .toBodilessEntity()
       .doOnSuccess { logger.info("Index created successfully") }

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/setup/IndexTemplateSetupTask.kt

@@ -0,0 +1,77 @@
+package ch.srgssr.pillarbox.monitoring.event.setup
+
+import ch.srgssr.pillarbox.monitoring.io.loadResourceContent
+import ch.srgssr.pillarbox.monitoring.log.error
+import ch.srgssr.pillarbox.monitoring.log.logger
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.core.annotation.Order
+import org.springframework.core.io.ResourceLoader
+import org.springframework.http.HttpHeaders
+import org.springframework.http.HttpStatusCode
+import org.springframework.http.MediaType
+import org.springframework.stereotype.Component
+import org.springframework.web.reactive.function.client.WebClient
+import reactor.core.publisher.Mono
+
+/**
+ * Task responsible for setting up the OpenSearch index template.
+ *
+ * This task creates the index template stored in `resources/opensearch/index_template.json`.
+ *
+ * @property webClient WebClient instance used to interact with the OpenSearch API.
+ * @property resourceLoader Resource loader used to access the index template JSON file.
+ */
+@Component
+@Order(2)
+class IndexTemplateSetupTask(
+  @Qualifier("openSearchWebClient")
+  private val webClient: WebClient,
+  private val resourceLoader: ResourceLoader,
+) : OpenSearchSetupTask {
+  private companion object {
+    /**
+     * Logger instance for logging within this task.
+     */
+    private val logger = logger()
+
+    /**
+     * Path for creating the index template.
+     */
+    private const val INDEX_TEMPLATE_CREATION_PATH = "/_index_template/events_template"
+  }
+
+  /**
+   * Runs the index template setup task.
+   *
+   * Checks if the index template exists: if not, creates the index template.
+   *
+   * @return Mono indicating the completion of the task.
+   */
+  override fun run(): Mono<*> = checkAndCreateTemplate()
+
+  private fun checkAndCreateTemplate(): Mono<*> =
+    webClient
+      .get()
+      .uri(INDEX_TEMPLATE_CREATION_PATH)
+      .retrieve()
+      .onStatus(HttpStatusCode::is4xxClientError) {
+        logger.info("Index template does not exist, creating it...")
+        createTemplate().then(Mono.empty())
+      }.onStatus(HttpStatusCode::is2xxSuccessful) {
+        logger.info("Index template already exists, skipping creation.")
+        Mono.empty()
+      }.toBodilessEntity()
+
+  private fun createTemplate(): Mono<*> {
+    val indexTemplateJson = resourceLoader.loadResourceContent("classpath:opensearch/index_template.json")
+    return webClient
+      .put()
+      .uri(INDEX_TEMPLATE_CREATION_PATH)
+      .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
+      .bodyValue(indexTemplateJson)
+      .retrieve()
+      .toBodilessEntity()
+      .doOnSuccess { logger.info("Index template created successfully") }
+      .doOnError { e -> logger.error { "Failed to create index template: ${e.message}" } }
+  }
+}

src/main/kotlin/ch/srgssr/pillarbox/monitoring/event/setup/OpenSearchSetupService.kt

@@ -3,7 +3,6 @@ package ch.srgssr.pillarbox.monitoring.event.setup
 import ch.srgssr.pillarbox.monitoring.event.repository.OpenSearchConfigurationProperties
 import ch.srgssr.pillarbox.monitoring.log.logger
 import org.springframework.beans.factory.annotation.Qualifier
-import org.springframework.http.ResponseEntity
 import org.springframework.stereotype.Service
 import org.springframework.web.reactive.function.client.WebClient
 import reactor.core.publisher.Flux
@@ -52,7 +51,7 @@ class OpenSearchSetupService(
       .then(runSetupTasks())
       .doOnSuccess { logger.info("All setup tasks are completed, starting SSE client...") }
 
-  private fun checkOpenSearchHealth(): Mono<ResponseEntity<Void>> =
+  private fun checkOpenSearchHealth(): Mono<*> =
     webClient
       .get()
       .uri("/")

src/main/resources/opensearch/alias.json

@@ -2,8 +2,8 @@
   "actions": [
     {
       "add": {
-        "index": "actions-*",
-        "alias": "filtered_actions",
+        "index": "events*",
+        "alias": "user_events",
         "filter": {
           "bool": {
             "must_not": [

src/main/resources/opensearch/index.json

@@ -0,0 +1,7 @@
+{
+  "aliases": {
+    "events": {
+      "is_write_index": true
+    }
+  }
+}

src/main/resources/opensearch/index_template.json

@@ -1,18 +1,19 @@
 {
-  "settings": {
-    "number_of_shards": 1
-  },
-  "mappings": {
-    "properties": {
-      "@timestamp": {
-        "type": "date",
-        "format": "epoch_millis"
+  "index_patterns": ["events*"],
+  "priority": 100,
+  "template": {
+    "settings": {
+      "number_of_shards": 3,
+      "number_of_replicas": 1,
+      "plugins.index_state_management.rollover_alias": "events"
+    },
+    "mappings": {
+      "properties": {
+        "@timestamp": {
+          "type": "date",
+          "format": "epoch_millis"
+        }
       }
     }
-  },
-  "aliases": {
-    "actions": {
-      "is_write_index": true
-    }
   }
 }

src/main/resources/opensearch/ism_policy.json

@@ -1,14 +1,34 @@
 {
   "policy": {
-    "description": "Policy for managing actions indices",
+    "description": "Hot warm delete policy for events",
     "default_state": "hot",
+    "schema_version": 1,
     "states": [
       {
         "name": "hot",
         "actions": [
           {
             "rollover": {
-              "min_index_age": "14d"
+              "min_index_age": "1d",
+              "copy_alias": true
+            }
+          }
+        ],
+        "transitions": [
+          {
+            "state_name": "warm",
+            "conditions": {
+              "min_index_age": "7d"
+            }
+          }
+        ]
+      },
+      {
+        "name": "warm",
+        "actions": [
+          {
+            "replica_count": {
+              "number_of_replicas": 0
             }
           }
         ],
@@ -27,14 +47,12 @@
           {
             "delete": {}
           }
-        ],
-        "transitions": []
+        ]
       }
+
     ],
     "ism_template": {
-      "index_patterns": [
-        "actions*"
-      ],
+      "index_patterns": ["events*"],
       "priority": 100
     }
   }