Choose the rollup settings to use based on cluster configuration (opensearch-project#91)

Author: thalurur

src/main/kotlin/org/opensearch/indexmanagement/IndexManagementPlugin.kt

@@ -299,6 +299,13 @@ class IndexManagementPlugin : JobSchedulerExtension, NetworkPlugin, ActionPlugin
     ): Collection<Any> {
         val settings = environment.settings()
         this.clusterService = clusterService
+        rollupInterceptor = RollupInterceptor(clusterService, settings, indexNameExpressionResolver)
+        val jvmService = JvmService(environment.settings())
+        val transformRunner = TransformRunner.initialize(client, clusterService, xContentRegistry, settings, indexNameExpressionResolver, jvmService)
+        fieldCapsFilter = FieldCapsFilter(clusterService, settings, indexNameExpressionResolver)
+        this.indexNameExpressionResolver = indexNameExpressionResolver
+
+        val skipFlag = SkipExecution(client, clusterService)
         val rollupRunner = RollupRunner
             .registerClient(client)
             .registerClusterService(clusterService)
@@ -311,13 +318,7 @@ class IndexManagementPlugin : JobSchedulerExtension, NetworkPlugin, ActionPlugin
             .registerSearcher(RollupSearchService(settings, clusterService, client))
             .registerMetadataServices(RollupMetadataService(client, xContentRegistry))
             .registerConsumers()
-        rollupInterceptor = RollupInterceptor(clusterService, settings, indexNameExpressionResolver)
-        val jvmService = JvmService(environment.settings())
-        val transformRunner = TransformRunner.initialize(client, clusterService, xContentRegistry, settings, indexNameExpressionResolver, jvmService)
-        fieldCapsFilter = FieldCapsFilter(clusterService, settings, indexNameExpressionResolver)
-        this.indexNameExpressionResolver = indexNameExpressionResolver
-
-        val skipFlag = SkipExecution(client, clusterService)
+            .registerClusterConfigurationProvider(skipFlag)
         indexManagementIndices = IndexManagementIndices(settings, client.admin().indices(), clusterService)
         val indexStateManagementHistory =
             IndexStateManagementHistory(

src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/SkipExecution.kt

@@ -49,6 +49,9 @@ class SkipExecution(
 
     @Volatile final var flag: Boolean = false
         private set
+    // To track if there are any legacy IM plugin nodes part of the cluster
+    @Volatile final var hasLegacyPlugin: Boolean = false
+        private set
 
     init {
         clusterService.addListener(this)
@@ -68,6 +71,7 @@ class SkipExecution(
             object : ActionListener<NodesInfoResponse> {
                 override fun onResponse(response: NodesInfoResponse) {
                     val versionSet = mutableSetOf<String>()
+                    val legacyVersionSet = mutableSetOf<String>()
 
                     response.nodes.map { it.getInfo(PluginsAndModules::class.java).pluginInfos }
                         .forEach {
@@ -77,13 +81,24 @@ class SkipExecution(
                                 ) {
                                     versionSet.add(nodePlugin.version)
                                 }
+
+                                if (nodePlugin.name == "opendistro-index-management" ||
+                                    nodePlugin.name == "opendistro_index_management"
+                                ) {
+                                    legacyVersionSet.add(nodePlugin.version)
+                                }
                             }
                         }
 
-                    if (versionSet.size > 1) {
+                    if ((versionSet.size + legacyVersionSet.size) > 1) {
                         flag = true
-                        logger.info("There are multiple versions of Index Management plugins in the cluster: $versionSet")
+                        logger.info("There are multiple versions of Index Management plugins in the cluster: [$versionSet, $legacyVersionSet]")
                     } else flag = false
+
+                    if (versionSet.isNotEmpty() && legacyVersionSet.isNotEmpty()) {
+                        hasLegacyPlugin = true
+                        logger.info("Found legacy plugin versions [$legacyVersionSet] and opensearch plugins versions [$versionSet] in the cluster")
+                    } else hasLegacyPlugin = false
                 }
 
                 override fun onFailure(e: Exception) {

src/main/kotlin/org/opensearch/indexmanagement/rollup/RollupMapperService.kt

@@ -51,6 +51,7 @@ import org.opensearch.indexmanagement.rollup.action.mapping.UpdateRollupMappingA
 import org.opensearch.indexmanagement.rollup.action.mapping.UpdateRollupMappingRequest
 import org.opensearch.indexmanagement.rollup.model.Rollup
 import org.opensearch.indexmanagement.rollup.model.RollupJobValidationResult
+import org.opensearch.indexmanagement.rollup.settings.LegacyOpenDistroRollupSettings
 import org.opensearch.indexmanagement.rollup.settings.RollupSettings
 import org.opensearch.indexmanagement.rollup.util.isRollupIndex
 import org.opensearch.indexmanagement.util.IndexUtils.Companion._META
@@ -89,13 +90,13 @@ class RollupMapperService(
     // If the target index mappings doesn't contain rollup job attempts to update the mappings.
     // TODO: error handling
     @Suppress("ReturnCount")
-    suspend fun attemptCreateRollupTargetIndex(job: Rollup): RollupJobValidationResult {
+    suspend fun attemptCreateRollupTargetIndex(job: Rollup, hasLegacyPlugin: Boolean): RollupJobValidationResult {
         if (indexExists(job.targetIndex)) {
             return validateAndAttemptToUpdateTargetIndex(job)
         } else {
             val errorMessage = "Failed to create target index [${job.targetIndex}]"
             return try {
-                val response = createTargetIndex(job)
+                val response = createTargetIndex(job, hasLegacyPlugin)
                 if (response.isAcknowledged) {
                     updateRollupIndexMappings(job)
                 } else {
@@ -112,9 +113,14 @@ class RollupMapperService(
         }
     }
 
-    private suspend fun createTargetIndex(job: Rollup): CreateIndexResponse {
+    private suspend fun createTargetIndex(job: Rollup, hasLegacyPlugin: Boolean): CreateIndexResponse {
+        val settings = if (hasLegacyPlugin) {
+            Settings.builder().put(LegacyOpenDistroRollupSettings.ROLLUP_INDEX.key, true).build()
+        } else {
+            Settings.builder().put(RollupSettings.ROLLUP_INDEX.key, true).build()
+        }
         val request = CreateIndexRequest(job.targetIndex)
-            .settings(Settings.builder().put(RollupSettings.ROLLUP_INDEX.key, true).build())
+            .settings(settings)
             .mapping(_DOC, IndexManagementIndices.rollupTargetMappings, XContentType.JSON)
         // TODO: Perhaps we can do better than this for mappings... as it'll be dynamic for rest
         //  Can we read in the actual mappings from the source index and use that?

src/main/kotlin/org/opensearch/indexmanagement/rollup/RollupRunner.kt

@@ -40,6 +40,7 @@ import org.opensearch.cluster.service.ClusterService
 import org.opensearch.common.settings.Settings
 import org.opensearch.common.unit.TimeValue
 import org.opensearch.common.xcontent.NamedXContentRegistry
+import org.opensearch.indexmanagement.indexstatemanagement.SkipExecution
 import org.opensearch.indexmanagement.opensearchapi.suspendUntil
 import org.opensearch.indexmanagement.rollup.action.get.GetRollupAction
 import org.opensearch.indexmanagement.rollup.action.get.GetRollupRequest
@@ -86,6 +87,7 @@ object RollupRunner :
     private lateinit var rollupIndexer: RollupIndexer
     private lateinit var rollupSearchService: RollupSearchService
     private lateinit var rollupMetadataService: RollupMetadataService
+    private lateinit var clusterConfigurationProvider: SkipExecution
 
     fun registerClusterService(clusterService: ClusterService): RollupRunner {
         this.clusterService = clusterService
@@ -139,6 +141,11 @@ object RollupRunner :
         return this
     }
 
+    fun registerClusterConfigurationProvider(clusterConfigurationProvider: SkipExecution): RollupRunner {
+        this.clusterConfigurationProvider = clusterConfigurationProvider
+        return this
+    }
+
     fun registerConsumers(): RollupRunner {
         return this
     }
@@ -250,7 +257,7 @@ object RollupRunner :
                 }
             }
 
-            when (val result = rollupMapperService.attemptCreateRollupTargetIndex(updatableJob)) {
+            when (val result = rollupMapperService.attemptCreateRollupTargetIndex(updatableJob, clusterConfigurationProvider.hasLegacyPlugin)) {
                 is RollupJobValidationResult.Failure -> {
                     setFailedMetadataAndDisableJob(updatableJob, result.message, metadata)
                     return
@@ -390,7 +397,7 @@ object RollupRunner :
         // we validate target index only if there is metadata document in the rollup
         if (metadata != null) {
             logger.debug("Attempting to create/validate target index [${job.targetIndex}] for rollup job [${job.id}]")
-            return rollupMapperService.attemptCreateRollupTargetIndex(job)
+            return rollupMapperService.attemptCreateRollupTargetIndex(job, clusterConfigurationProvider.hasLegacyPlugin)
         }
 
         return RollupJobValidationResult.Valid