|
11 | 11 | import org.apache.logging.log4j.LogManager; |
12 | 12 | import org.apache.logging.log4j.Logger; |
13 | 13 | import org.opensearch.Version; |
| 14 | +import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest; |
14 | 15 | import org.opensearch.cluster.ClusterState; |
15 | 16 | import org.opensearch.cluster.metadata.IndexMetadata; |
| 17 | +import org.opensearch.cluster.metadata.Metadata; |
16 | 18 | import org.opensearch.cluster.node.DiscoveryNode; |
17 | 19 | import org.opensearch.cluster.node.DiscoveryNodes; |
| 20 | +import org.opensearch.cluster.routing.RoutingTable; |
18 | 21 | import org.opensearch.common.collect.Tuple; |
19 | 22 | import org.opensearch.common.settings.Settings; |
20 | 23 | import org.opensearch.node.remotestore.RemoteStoreNodeAttribute; |
| 24 | +import org.opensearch.node.remotestore.RemoteStoreNodeService; |
21 | 25 |
|
22 | 26 | import java.nio.ByteBuffer; |
23 | 27 | import java.util.Arrays; |
24 | 28 | import java.util.Base64; |
| 29 | +import java.util.Collection; |
| 30 | +import java.util.Collections; |
25 | 31 | import java.util.HashMap; |
26 | 32 | import java.util.List; |
27 | 33 | import java.util.Locale; |
28 | 34 | import java.util.Map; |
29 | 35 | import java.util.Objects; |
30 | 36 | import java.util.Optional; |
31 | 37 | import java.util.function.Function; |
| 38 | +import java.util.stream.Collectors; |
32 | 39 |
|
| 40 | +import static org.opensearch.index.remote.RemoteMigrationIndexMetadataUpdater.indexHasRemoteStoreSettings; |
33 | 41 | import static org.opensearch.indices.RemoteStoreSettings.CLUSTER_REMOTE_STORE_PATH_HASH_ALGORITHM_SETTING; |
34 | 42 | import static org.opensearch.indices.RemoteStoreSettings.CLUSTER_REMOTE_STORE_PATH_TYPE_SETTING; |
35 | 43 | import static org.opensearch.indices.RemoteStoreSettings.CLUSTER_REMOTE_STORE_TRANSLOG_METADATA; |
@@ -250,4 +258,119 @@ public static Map<String, String> getRemoteStoreRepoName(DiscoveryNodes discover |
250 | 258 | .findFirst(); |
251 | 259 | return remoteNode.map(RemoteStoreNodeAttribute::getDataRepoNames).orElseGet(HashMap::new); |
252 | 260 | } |
| 261 | + |
| 262 | + /** |
| 263 | + * Invoked after a cluster settings update. |
| 264 | + * Checks if the applied cluster settings has switched the cluster to STRICT mode. |
| 265 | + * If so, checks and applies appropriate index settings depending on the current set |
| 266 | + * of node types in the cluster |
| 267 | + * This has been intentionally done after the cluster settings update |
| 268 | + * flow. That way we are not interfering with the usual settings update |
| 269 | + * and the cluster state mutation that comes along with it |
| 270 | + * |
| 271 | + * @param isCompatibilityModeChanging flag passed from cluster settings update call to denote if a compatibility mode change has been done |
| 272 | + * @param request request payload passed from cluster settings update |
| 273 | + * @param currentState cluster state generated after changing cluster settings were applied |
| 274 | + * @param logger Logger reference |
| 275 | + * @return Mutated cluster state with remote store index settings applied, no-op if the cluster is not switching to `STRICT` compatibility mode |
| 276 | + */ |
| 277 | + public static ClusterState checkAndFinalizeRemoteStoreMigration( |
| 278 | + boolean isCompatibilityModeChanging, |
| 279 | + ClusterUpdateSettingsRequest request, |
| 280 | + ClusterState currentState, |
| 281 | + Logger logger |
| 282 | + ) { |
| 283 | + if (isCompatibilityModeChanging && isSwitchToStrictCompatibilityMode(request)) { |
| 284 | + return finalizeMigration(currentState, logger); |
| 285 | + } |
| 286 | + return currentState; |
| 287 | + } |
| 288 | + |
| 289 | + /** |
| 290 | + * Finalizes the docrep to remote-store migration process by applying remote store based index settings |
| 291 | + * on indices that are missing them. No-Op if all indices already have the settings applied through |
| 292 | + * IndexMetadataUpdater |
| 293 | + * |
| 294 | + * @param incomingState mutated cluster state after cluster settings were applied |
| 295 | + * @return new cluster state with index settings updated |
| 296 | + */ |
| 297 | + public static ClusterState finalizeMigration(ClusterState incomingState, Logger logger) { |
| 298 | + Map<String, DiscoveryNode> discoveryNodeMap = incomingState.nodes().getNodes(); |
| 299 | + if (discoveryNodeMap.isEmpty() == false) { |
| 300 | + // At this point, we have already validated that all nodes in the cluster are of uniform type. |
| 301 | + // Either all of them are remote store enabled, or all of them are docrep enabled |
| 302 | + boolean remoteStoreEnabledNodePresent = discoveryNodeMap.values().stream().findFirst().get().isRemoteStoreNode(); |
| 303 | + if (remoteStoreEnabledNodePresent == true) { |
| 304 | + List<IndexMetadata> indicesWithoutRemoteStoreSettings = getIndicesWithoutRemoteStoreSettings(incomingState, logger); |
| 305 | + if (indicesWithoutRemoteStoreSettings.isEmpty() == true) { |
| 306 | + logger.info("All indices in the cluster has remote store based index settings"); |
| 307 | + } else { |
| 308 | + Metadata mutatedMetadata = applyRemoteStoreSettings(incomingState, indicesWithoutRemoteStoreSettings, logger); |
| 309 | + return ClusterState.builder(incomingState).metadata(mutatedMetadata).build(); |
| 310 | + } |
| 311 | + } else { |
| 312 | + logger.debug("All nodes in the cluster are not remote nodes. Skipping."); |
| 313 | + } |
| 314 | + } |
| 315 | + return incomingState; |
| 316 | + } |
| 317 | + |
| 318 | + /** |
| 319 | + * Filters out indices which does not have remote store based |
| 320 | + * index settings applied even after all shard copies have |
| 321 | + * migrated to remote store enabled nodes |
| 322 | + */ |
| 323 | + private static List<IndexMetadata> getIndicesWithoutRemoteStoreSettings(ClusterState clusterState, Logger logger) { |
| 324 | + Collection<IndexMetadata> allIndicesMetadata = clusterState.metadata().indices().values(); |
| 325 | + if (allIndicesMetadata.isEmpty() == false) { |
| 326 | + List<IndexMetadata> indicesWithoutRemoteSettings = allIndicesMetadata.stream() |
| 327 | + .filter(idxMd -> indexHasRemoteStoreSettings(idxMd.getSettings()) == false) |
| 328 | + .collect(Collectors.toList()); |
| 329 | + logger.debug( |
| 330 | + "Attempting to switch to strict mode. Count of indices without remote store settings {}", |
| 331 | + indicesWithoutRemoteSettings.size() |
| 332 | + ); |
| 333 | + return indicesWithoutRemoteSettings; |
| 334 | + } |
| 335 | + return Collections.emptyList(); |
| 336 | + } |
| 337 | + |
| 338 | + /** |
| 339 | + * Applies remote store index settings through {@link RemoteMigrationIndexMetadataUpdater} |
| 340 | + */ |
| 341 | + private static Metadata applyRemoteStoreSettings( |
| 342 | + ClusterState clusterState, |
| 343 | + List<IndexMetadata> indicesWithoutRemoteStoreSettings, |
| 344 | + Logger logger |
| 345 | + ) { |
| 346 | + Metadata.Builder metadataBuilder = Metadata.builder(clusterState.getMetadata()); |
| 347 | + RoutingTable currentRoutingTable = clusterState.getRoutingTable(); |
| 348 | + DiscoveryNodes currentDiscoveryNodes = clusterState.getNodes(); |
| 349 | + Settings currentClusterSettings = clusterState.metadata().settings(); |
| 350 | + for (IndexMetadata indexMetadata : indicesWithoutRemoteStoreSettings) { |
| 351 | + IndexMetadata.Builder indexMetadataBuilder = IndexMetadata.builder(indexMetadata); |
| 352 | + RemoteMigrationIndexMetadataUpdater indexMetadataUpdater = new RemoteMigrationIndexMetadataUpdater( |
| 353 | + currentDiscoveryNodes, |
| 354 | + currentRoutingTable, |
| 355 | + indexMetadata, |
| 356 | + currentClusterSettings, |
| 357 | + logger |
| 358 | + ); |
| 359 | + indexMetadataUpdater.maybeAddRemoteIndexSettings(indexMetadataBuilder, indexMetadata.getIndex().getName()); |
| 360 | + metadataBuilder.put(indexMetadataBuilder); |
| 361 | + } |
| 362 | + return metadataBuilder.build(); |
| 363 | + } |
| 364 | + |
| 365 | + /** |
| 366 | + * Checks if the incoming cluster settings payload is attempting to switch |
| 367 | + * the cluster to `STRICT` compatibility mode |
| 368 | + * Visible only for tests |
| 369 | + */ |
| 370 | + public static boolean isSwitchToStrictCompatibilityMode(ClusterUpdateSettingsRequest request) { |
| 371 | + Settings incomingSettings = Settings.builder().put(request.persistentSettings()).put(request.transientSettings()).build(); |
| 372 | + return RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING.get( |
| 373 | + incomingSettings |
| 374 | + ) == RemoteStoreNodeService.CompatibilityMode.STRICT; |
| 375 | + } |
253 | 376 | } |
0 commit comments