Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Updating multi-field mapping at follower #671

Merged
merged 4 commits into from
Jan 10, 2023
Merged

Updating multi-field mapping at follower #671

Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ae5a4ac
Updating multi-field mapping at follower
sricharanvuppu Dec 29, 2022
a8f3d9a
Integ tests were written for multi-field mapping update at follower
sricharanvuppu Jan 3, 2023
541b604
Modified Integ test cases for Multi-field mapping at follower
sricharanvuppu Jan 3, 2023
c7334b2
UpdateFollowerMapping method added instead of SyncRemoteMapping
sricharanvuppu Jan 10, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 41 additions & 0 deletions src/main/kotlin/org/opensearch/replication/task/index/IndexReplicationTask.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,9 @@ import org.opensearch.action.admin.indices.alias.get.GetAliasesRequest
import org.opensearch.action.admin.indices.delete.DeleteIndexRequest
import org.opensearch.action.admin.indices.settings.get.GetSettingsRequest
import org.opensearch.action.admin.indices.settings.put.UpdateSettingsRequest
import org.opensearch.action.admin.indices.mapping.get.GetMappingsRequest
import org.opensearch.action.admin.indices.mapping.put.PutMappingRequest
import org.opensearch.action.support.IndicesOptions
import org.opensearch.client.Client
import org.opensearch.client.Requests
import org.opensearch.cluster.ClusterChangedEvent
Expand All @@ -75,6 +78,7 @@ import org.opensearch.common.unit.ByteSizeValue
import org.opensearch.common.xcontent.ToXContent
import org.opensearch.common.xcontent.ToXContentObject
import org.opensearch.common.xcontent.XContentBuilder
import org.opensearch.common.xcontent.XContentType
import org.opensearch.index.Index
import org.opensearch.index.IndexService
import org.opensearch.index.IndexSettings
Expand All @@ -88,6 +92,7 @@ import org.opensearch.persistent.PersistentTasksCustomMetadata.PersistentTask
import org.opensearch.persistent.PersistentTasksNodeService
import org.opensearch.persistent.PersistentTasksService
import org.opensearch.replication.ReplicationException
import org.opensearch.replication.MappingNotAvailableException
import org.opensearch.replication.ReplicationPlugin.Companion.REPLICATION_INDEX_TRANSLOG_PRUNING_ENABLED_SETTING
import org.opensearch.rest.RestStatus
import org.opensearch.tasks.TaskId
Expand All @@ -100,6 +105,7 @@ import kotlin.coroutines.resume
import kotlin.coroutines.resumeWithException
import kotlin.coroutines.suspendCoroutine
import kotlin.streams.toList
import org.opensearch.cluster.DiffableUtils

open class IndexReplicationTask(id: Long, type: String, action: String, description: String,
parentTask: TaskId,
Expand Down Expand Up @@ -407,6 +413,25 @@ open class IndexReplicationTask(id: Long, type: String, action: String, descript
}
}

private suspend fun syncRemoteMapping(leaderAlias: String, leaderIndex: String,
followerIndex: String) {
log.debug("Syncing mappings from ${leaderAlias}:${leaderIndex} -> $followerIndex...")
val remoteClient = client.getRemoteClusterClient(leaderAlias)
val options = IndicesOptions.strictSingleIndexNoExpandForbidClosed()
val getMappingsRequest = GetMappingsRequest().indices(leaderIndex).indicesOptions(options)
val getMappingsResponse = remoteClient.suspending(remoteClient.admin().indices()::getMappings, injectSecurityContext = true)(getMappingsRequest)
val mappingSource = getMappingsResponse?.mappings()?.get(leaderIndex)?.source()?.string()
if (null == mappingSource) {
log.error("Mapping response: $getMappingsResponse")
throw MappingNotAvailableException("Mapping for the index $leaderIndex is not available")
}
val putMappingRequest = PutMappingRequest().indices(followerIndex).indicesOptions(options)
.source(mappingSource, XContentType.JSON)
val updateMappingRequest = UpdateMetadataRequest(followerIndex, UpdateMetadataRequest.Type.MAPPING, putMappingRequest)
client.suspendExecute(UpdateMetadataAction.INSTANCE, updateMappingRequest, injectSecurityContext = true)
log.debug("Mappings synced for $followerIndex")
}

private suspend fun pollForMetadata(scope: CoroutineScope) {
while (scope.isActive) {
try {
Expand Down Expand Up @@ -547,6 +572,22 @@ open class IndexReplicationTask(id: Long, type: String, action: String, descript
} else {
metadataUpdate = null
}
val options = IndicesOptions.strictSingleIndexNoExpandForbidClosed()
var gmr = GetMappingsRequest().indices(this.leaderIndex.name).indicesOptions(options)
var mappingResponse = remoteClient.suspending(remoteClient.admin().indices()::getMappings, injectSecurityContext = true)(gmr)
val leaderMappings = mappingResponse.mappings().get(this.leaderIndex.name).sourceAsMap().toMap()
val leadProperties = leaderMappings.get("properties") as Map<String,Any>
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's combine the two calls rather than invoking multiple ones.

gmr = GetMappingsRequest().indices(this.followerIndexName).indicesOptions(options)
mappingResponse = client.suspending(client.admin().indices()::getMappings, injectSecurityContext = true)(gmr)
val followerMappings = mappingResponse.mappings().get(this.followerIndexName).sourceAsMap().toMap()
val followerProperties = followerMappings.get("properties") as Map<String,Any>
for(iter in followerProperties) {
if(leadProperties.containsKey(iter.key) && leadProperties.getValue(iter.key).toString()!=(iter.value).toString()){
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we add integ test case for this?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added Integ tests

log.info("Updating Multi-field Mapping at Follower")
syncRemoteMapping(this.leaderAlias,this.leaderIndex.name, this.followerIndexName)
break;
}
}

} catch (e: Exception) {
log.error("Error in getting the required metadata ${e.stackTraceToString()}")
Expand Down