Skip to content

Commit

Permalink
Address PR comments
Browse files Browse the repository at this point in the history 
Signed-off-by: Shivansh Arora <[email protected]>
  • Loading branch information
@shiv0408
shiv0408 committed Apr 30, 2024
1 parent 6c637c0 commit 494aacc
Show file tree
Hide file tree
Showing 2 changed files with 76 additions and 68 deletions.
97 changes: 53 additions & 44 deletions server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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];
}
Expand Down Expand Up @@ -126,58 +176,17 @@ private static Map<String, UploadedMetadataAttribute> customMetadata(Object[] fi

private static final ConstructingObjectParser<ClusterMetadataManifest, Void> 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<ClusterMetadataManifest, Void> 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<ClusterMetadataManifest, Void> 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<ClusterMetadataManifest, Void> CURRENT_PARSER = PARSER_V2;
Expand Down
47 changes: 23 additions & 24 deletions server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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
Expand Down

0 comments on commit 494aacc

Please sign in to comment.