Adding Index Settings validation before starting replication (#461)

* Adding Index Settings validation before starting replication

Signed-off-by: Gaurav Bafna <gbbafna@amazon.com>

* Retrieving default index settings before starting replication

Signed-off-by: Gaurav Bafna <gbbafna@amazon.com>

Author: gbbafna

src/main/kotlin/org/opensearch/replication/action/index/TransportReplicateIndexAction.kt

@@ -33,7 +33,7 @@ import org.opensearch.action.support.HandledTransportAction
 import org.opensearch.action.support.IndicesOptions
 import org.opensearch.client.Client
 import org.opensearch.cluster.ClusterState
-import org.opensearch.cluster.metadata.IndexMetadata
+import org.opensearch.cluster.metadata.MetadataCreateIndexService
 import org.opensearch.common.inject.Inject
 import org.opensearch.common.settings.Settings
 import org.opensearch.env.Environment
@@ -48,7 +48,8 @@ class TransportReplicateIndexAction @Inject constructor(transportService: Transp
                                                         val threadPool: ThreadPool,
                                                         actionFilters: ActionFilters,
                                                         private val client : Client,
-                                                        private val environment: Environment) :
+                                                        private val environment: Environment,
+                                                        private val metadataCreateIndexService: MetadataCreateIndexService) :
         HandledTransportAction<ReplicateIndexRequest, ReplicateIndexResponse>(ReplicateIndexAction.NAME,
                 transportService, actionFilters, ::ReplicateIndexRequest),
     CoroutineScope by GlobalScope {
@@ -102,7 +103,13 @@ class TransportReplicateIndexAction @Inject constructor(transportService: Transp
                     throw IllegalArgumentException("Cannot replicate k-NN index - ${request.leaderIndex}")
                 }
 
-                ValidationUtil.validateAnalyzerSettings(environment, leaderSettings, request.settings)
+                ValidationUtil.validateIndexSettings(
+                    environment,
+                    request.followerIndex,
+                    leaderSettings,
+                    request.settings,
+                    metadataCreateIndexService
+                )
 
                 // Setup checks are successful and trigger replication for the index
                 // permissions evaluation to trigger replication is based on the current security context set
@@ -128,7 +135,7 @@ class TransportReplicateIndexAction @Inject constructor(transportService: Transp
 
     private suspend fun getLeaderIndexSettings(leaderAlias: String, leaderIndex: String): Settings {
         val remoteClient = client.getRemoteClusterClient(leaderAlias)
-        val getSettingsRequest = GetSettingsRequest().includeDefaults(false).indices(leaderIndex)
+        val getSettingsRequest = GetSettingsRequest().includeDefaults(true).indices(leaderIndex)
         val settingsResponse = remoteClient.suspending(remoteClient.admin().indices()::getSettings,
                 injectSecurityContext = true)(getSettingsRequest)
         return settingsResponse.indexToSettings.get(leaderIndex) ?: throw IndexNotFoundException("${leaderAlias}:${leaderIndex}")

src/main/kotlin/org/opensearch/replication/util/ValidationUtil.kt

@@ -31,6 +31,27 @@ object ValidationUtil {
 
     private val log = LogManager.getLogger(ValidationUtil::class.java)
 
+    fun validateIndexSettings(
+        environment: Environment,
+        followerIndex: String,
+        leaderSettings: Settings,
+        overriddenSettings: Settings,
+        metadataCreateIndexService: MetadataCreateIndexService
+    ) {
+        val settingsList = arrayOf(leaderSettings, overriddenSettings)
+        val desiredSettingsBuilder = Settings.builder()
+        // Desired settings are taking leader Settings and then overriding them with desired settings
+        for (settings in settingsList) {
+            for (key in settings.keySet()) {
+                desiredSettingsBuilder.copy(key, settings);
+            }
+        }
+        val desiredSettings = desiredSettingsBuilder.build()
+
+        metadataCreateIndexService.validateIndexSettings(followerIndex,desiredSettings, false)
+        validateAnalyzerSettings(environment, leaderSettings, overriddenSettings)
+    }
+
     fun validateAnalyzerSettings(environment: Environment, leaderSettings: Settings, overriddenSettings: Settings) {
         val analyserSettings = leaderSettings.filter { k: String? -> k!!.matches(Regex("index.analysis.*path")) }
         for (analyserSetting in analyserSettings.keySet()) {

src/test/kotlin/org/opensearch/replication/integ/rest/StartReplicationIT.kt

@@ -1169,6 +1169,26 @@ class StartReplicationIT: MultiClusterRestTestCase() {
         }
     }
 
+    fun `test start replication invalid settings`() {
+        val followerClient = getClientForCluster(FOLLOWER)
+        val leaderClient = getClientForCluster(LEADER)
+
+        createConnectionBetweenClusters(FOLLOWER, LEADER)
+
+        val createIndexResponse = leaderClient.indices().create(CreateIndexRequest(leaderIndexName), RequestOptions.DEFAULT)
+        assertThat(createIndexResponse.isAcknowledged).isTrue()
+        val settings = Settings.builder()
+            .put("index.data_path", "/random-path/invalid-setting")
+            .build()
+
+        try {
+            followerClient.startReplication(StartReplicationRequest("source", leaderIndexName, followerIndexName, settings = settings))
+        } catch (e: ResponseException) {
+            Assert.assertEquals(400, e.response.statusLine.statusCode)
+            Assert.assertTrue(e.message!!.contains("Validation Failed: 1: custom path [/random-path/invalid-setting] is not a sub-path of path.shared_data"))
+        }
+    }
+
     fun `test that replication is not started when all primary shards are not in active state`() {
         val followerClient = getClientForCluster(FOLLOWER)
         val leaderClient = getClientForCluster(LEADER)