From 494aacc7c73671d76e298284fcbcee1a3072636f Mon Sep 17 00:00:00 2001 From: Shivansh Arora Date: Tue, 30 Apr 2024 20:30:25 +0530 Subject: [PATCH] Address PR comments Signed-off-by: Shivansh Arora --- .../remote/ClusterMetadataManifest.java | 97 ++++++++++--------- .../remote/RemoteClusterStateService.java | 47 +++++---- 2 files changed, 76 insertions(+), 68 deletions(-) diff --git a/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java b/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java index 023eb06b8eb86..de2e6727e9507 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java +++ b/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java @@ -59,6 +59,56 @@ public class ClusterMetadataManifest implements Writeable, ToXContentFragment { private static final ParseField UPLOADED_TEMPLATES_METADATA = new ParseField("uploaded_templates_metadata"); private static final ParseField UPLOADED_CUSTOM_METADATA = new ParseField("uploaded_custom_metadata"); + private static ClusterMetadataManifest.Builder manifestV0Builder(Object[] fields) { + return ClusterMetadataManifest.builder() + .clusterTerm(term(fields)) + .stateVersion(version(fields)) + .clusterUUID(clusterUUID(fields)) + .stateUUID(stateUUID(fields)) + .opensearchVersion(opensearchVersion(fields)) + .nodeId(nodeId(fields)) + .committed(committed(fields)) + .codecVersion(CODEC_V0) + .indices(indices(fields)) + .previousClusterUUID(previousClusterUUID(fields)) + .clusterUUIDCommitted(clusterUUIDCommitted(fields)); + } + + private static ClusterMetadataManifest.Builder manifestV1Builder(Object[] fields) { + return ClusterMetadataManifest.builder() + .clusterTerm(term(fields)) + .stateVersion(version(fields)) + .clusterUUID(clusterUUID(fields)) + .stateUUID(stateUUID(fields)) + .opensearchVersion(opensearchVersion(fields)) + .nodeId(nodeId(fields)) + .committed(committed(fields)) + .codecVersion(codecVersion(fields)) + .globalMetadataFileName(globalMetadataFileName(fields)) + .indices(indices(fields)) + .previousClusterUUID(previousClusterUUID(fields)) + .clusterUUIDCommitted(clusterUUIDCommitted(fields)); + } + + private static ClusterMetadataManifest.Builder manifestV2Builder(Object[] fields) { + return ClusterMetadataManifest.builder() + .clusterTerm(term(fields)) + .stateVersion(version(fields)) + .clusterUUID(clusterUUID(fields)) + .stateUUID(stateUUID(fields)) + .opensearchVersion(opensearchVersion(fields)) + .nodeId(nodeId(fields)) + .committed(committed(fields)) + .codecVersion(codecVersion(fields)) + .indices(indices(fields)) + .previousClusterUUID(previousClusterUUID(fields)) + .clusterUUIDCommitted(clusterUUIDCommitted(fields)) + .coordinationMetadata(coordinationMetadata(fields)) + .settingMetadata(settingsMetadata(fields)) + .templatesMetadata(templatesMetadata(fields)) + .customMetadataMap(customMetadata(fields)); + } + private static long term(Object[] fields) { return (long) fields[0]; } @@ -126,58 +176,17 @@ private static Map customMetadata(Object[] fi private static final ConstructingObjectParser PARSER_V0 = new ConstructingObjectParser<>( "cluster_metadata_manifest", - fields -> ClusterMetadataManifest.builder() - .clusterTerm(term(fields)) - .stateVersion(version(fields)) - .clusterUUID(clusterUUID(fields)) - .stateUUID(stateUUID(fields)) - .opensearchVersion(opensearchVersion(fields)) - .nodeId(nodeId(fields)) - .committed(committed(fields)) - .codecVersion(CODEC_V0) - .indices(indices(fields)) - .previousClusterUUID(previousClusterUUID(fields)) - .clusterUUIDCommitted(clusterUUIDCommitted(fields)) - .build() + fields -> manifestV0Builder(fields).build() ); private static final ConstructingObjectParser PARSER_V1 = new ConstructingObjectParser<>( "cluster_metadata_manifest", - fields -> ClusterMetadataManifest.builder() - .clusterTerm(term(fields)) - .stateVersion(version(fields)) - .clusterUUID(clusterUUID(fields)) - .stateUUID(stateUUID(fields)) - .opensearchVersion(opensearchVersion(fields)) - .nodeId(nodeId(fields)) - .committed(committed(fields)) - .codecVersion(codecVersion(fields)) - .globalMetadataFileName(globalMetadataFileName(fields)) - .indices(indices(fields)) - .previousClusterUUID(previousClusterUUID(fields)) - .clusterUUIDCommitted(clusterUUIDCommitted(fields)) - .build() + fields -> manifestV1Builder(fields).build() ); private static final ConstructingObjectParser PARSER_V2 = new ConstructingObjectParser<>( "cluster_metadata_manifest", - fields -> ClusterMetadataManifest.builder() - .clusterTerm(term(fields)) - .stateVersion(version(fields)) - .clusterUUID(clusterUUID(fields)) - .stateUUID(stateUUID(fields)) - .opensearchVersion(opensearchVersion(fields)) - .nodeId(nodeId(fields)) - .committed(committed(fields)) - .codecVersion(codecVersion(fields)) - .indices(indices(fields)) - .previousClusterUUID(previousClusterUUID(fields)) - .clusterUUIDCommitted(clusterUUIDCommitted(fields)) - .coordinationMetadata(coordinationMetadata(fields)) - .settingMetadata(settingsMetadata(fields)) - .templatesMetadata(templatesMetadata(fields)) - .customMetadataMap(customMetadata(fields)) - .build() + fields -> manifestV2Builder(fields).build() ); private static final ConstructingObjectParser CURRENT_PARSER = PARSER_V2; diff --git a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java index cb66323be8994..fa3d5f3773685 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java @@ -210,6 +210,10 @@ public class RemoteClusterStateService implements Closeable { private final AtomicBoolean deleteStaleMetadataRunning = new AtomicBoolean(false); private final RemotePersistenceStats remoteStateStats; + private final String CLUSTER_STATE_UPLOAD_TIME_LOG_STRING = "writing cluster state for version [{}] took [{}ms]"; + private final String METADATA_UPDATE_LOG_STRING = "wrote metadata for [{}] indices and skipped [{}] unchanged " + + "indices, coordination metadata updated : [{}], settings metadata updated : [{}], templates metadata " + + "updated : [{}], custom metadata updated : [{}]"; public static final int INDEX_METADATA_CURRENT_CODEC_VERSION = 1; public static final int MANIFEST_CURRENT_CODEC_VERSION = ClusterMetadataManifest.CODEC_V2; public static final int GLOBAL_METADATA_CURRENT_CODEC_VERSION = 1; @@ -442,34 +446,29 @@ public ClusterMetadataManifest writeIncrementalMetadata( final long durationMillis = TimeValue.nsecToMSec(relativeTimeNanosSupplier.getAsLong() - startTimeNanos); remoteStateStats.stateSucceeded(); remoteStateStats.stateTook(durationMillis); + ParameterizedMessage clusterStateUploadTimeMessage = new ParameterizedMessage( + CLUSTER_STATE_UPLOAD_TIME_LOG_STRING, + manifest.getStateVersion(), + durationMillis + ); + ParameterizedMessage metadataUpdateMessage = new ParameterizedMessage( + METADATA_UPDATE_LOG_STRING, + numIndicesUpdated, + numIndicesUnchanged, + updateCoordinationMetadata, + updateSettingsMetadata, + updateTemplatesMetadata, + customsToUpload.size() + ); if (durationMillis >= slowWriteLoggingThreshold.getMillis()) { logger.warn( - "writing cluster state took [{}ms] which is above the warn threshold of [{}]; " - + "wrote metadata for [{}] indices and skipped [{}] unchanged indices, coordination metadata updated : [{}], " - + "settings metadata updated : [{}], templates metadata updated : [{}], custom metadata updated : [{}]", - durationMillis, + "{} which is above the warn threshold of [{}]; {}", + clusterStateUploadTimeMessage, slowWriteLoggingThreshold, - numIndicesUpdated, - numIndicesUnchanged, - updateCoordinationMetadata, - updateSettingsMetadata, - updateTemplatesMetadata, - customsToUpload.size() + metadataUpdateMessage ); } else { - logger.info( - "writing cluster state for version [{}] took [{}ms]; " - + "wrote metadata for [{}] indices and skipped [{}] unchanged indices, coordination metadata updated : [{}], " - + "settings metadata updated : [{}], templates metadata updated : [{}], custom metadata updated : [{}]", - manifest.getStateVersion(), - durationMillis, - numIndicesUpdated, - numIndicesUnchanged, - updateCoordinationMetadata, - updateSettingsMetadata, - updateTemplatesMetadata, - customsToUpload.size() - ); + logger.info("{}; {}", clusterStateUploadTimeMessage, metadataUpdateMessage); } return manifest; } @@ -549,7 +548,7 @@ private UploadedMetadataResults writeMetadataInParallel( ); }); indexToUpload.forEach(indexMetadata -> { - uploadTasks.put(indexMetadata.getIndexName(), getIndexMetadataAsyncAction(clusterState, indexMetadata, listener)); + uploadTasks.put(indexMetadata.getIndex().getName(), getIndexMetadataAsyncAction(clusterState, indexMetadata, listener)); }); // start async upload of all required metadata files