Skip to content

Commit 3e3dfc8

Browse files
soosinhasoosinha
committed
Refactor discoverynodes and clusterblocks
1 parent 043a170 commit 3e3dfc8

8 files changed

+253
-121
lines changed

‎server/src/main/java/org/opensearch/gateway/remote/RemoteClusterBlocks.java

+109
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.gateway.remote;
10+
11+
import static org.opensearch.gateway.remote.RemoteClusterStateAttributesManager.CLUSTER_STATE_ATTRIBUTES_CURRENT_CODEC_VERSION;
12+
import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER;
13+
import static org.opensearch.gateway.remote.RemoteClusterStateUtils.METADATA_NAME_FORMAT;
14+
15+
import java.io.IOException;
16+
import java.io.InputStream;
17+
import java.util.List;
18+
import org.opensearch.cluster.block.ClusterBlocks;
19+
import org.opensearch.common.io.Streams;
20+
import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedMetadata;
21+
import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedMetadataAttribute;
22+
import org.opensearch.index.remote.RemoteStoreUtils;
23+
import org.opensearch.index.translog.transfer.BlobStoreTransferService;
24+
import org.opensearch.repositories.blobstore.BlobStoreRepository;
25+
import org.opensearch.repositories.blobstore.ChecksumBlobStoreFormat;
26+
import org.opensearch.threadpool.ThreadPool;
27+
28+
public class RemoteClusterBlocks extends AbstractRemoteBlobStoreObject<ClusterBlocks> {
29+
30+
public static final String CLUSTER_BLOCKS = "blocks";
31+
public static final ChecksumBlobStoreFormat<ClusterBlocks> CLUSTER_BLOCKS_FORMAT = new ChecksumBlobStoreFormat<>(
32+
"blocks",
33+
METADATA_NAME_FORMAT,
34+
ClusterBlocks::fromXContent
35+
);
36+
37+
private ClusterBlocks clusterBlocks;
38+
private long stateVersion;
39+
private String blobName;
40+
private final String clusterUUID;
41+
42+
public RemoteClusterBlocks(ClusterBlocks clusterBlocks, long stateVersion, String clusterUUID, BlobStoreTransferService blobStoreTransferService,
43+
BlobStoreRepository blobStoreRepository, String clusterName,
44+
ThreadPool threadPool) {
45+
super(blobStoreTransferService, blobStoreRepository, clusterName, threadPool);
46+
this.clusterBlocks = clusterBlocks;
47+
this.stateVersion = stateVersion;
48+
this.clusterUUID = clusterUUID;
49+
}
50+
51+
public RemoteClusterBlocks(String blobName, String clusterUUID, BlobStoreTransferService blobStoreTransferService, BlobStoreRepository blobStoreRepository,
52+
String clusterName,
53+
ThreadPool threadPool) {
54+
super(blobStoreTransferService, blobStoreRepository, clusterName, threadPool);
55+
this.blobName = blobName;
56+
this.clusterUUID = clusterUUID;
57+
}
58+
59+
@Override
60+
public BlobPathParameters getBlobPathParameters() {
61+
return new BlobPathParameters(List.of("transient"), CLUSTER_BLOCKS);
62+
}
63+
64+
@Override
65+
public String getFullBlobName() {
66+
return blobName;
67+
}
68+
69+
@Override
70+
public String generateBlobFileName() {
71+
// 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/transient/<componentPrefix>__<inverted_state_version>__<inverted__timestamp>__<codec_version>
72+
String blobFileName = String.join(
73+
DELIMITER,
74+
getBlobPathParameters().getFilePrefix(),
75+
RemoteStoreUtils.invertLong(stateVersion),
76+
RemoteStoreUtils.invertLong(System.currentTimeMillis()),
77+
String.valueOf(CLUSTER_STATE_ATTRIBUTES_CURRENT_CODEC_VERSION)
78+
);
79+
// setting the full blob path with name for future access
80+
this.blobName = getBlobPathForUpload().buildAsString() + blobFileName;
81+
return blobFileName;
82+
}
83+
84+
@Override
85+
public UploadedMetadata getUploadedMetadata() {
86+
assert blobName != null;
87+
return new UploadedMetadataAttribute(CLUSTER_BLOCKS, blobName);
88+
}
89+
90+
@Override
91+
public ClusterBlocks get() {
92+
return clusterBlocks;
93+
}
94+
95+
@Override
96+
public String clusterUUID() {
97+
return clusterUUID;
98+
}
99+
100+
@Override
101+
public InputStream serialize() throws IOException {
102+
return CLUSTER_BLOCKS_FORMAT.serialize(clusterBlocks, generateBlobFileName(), getCompressor(), RemoteClusterStateUtils.FORMAT_PARAMS).streamInput();
103+
}
104+
105+
@Override
106+
public ClusterBlocks deserialize(InputStream inputStream) throws IOException {
107+
return CLUSTER_BLOCKS_FORMAT.deserialize(blobName, getBlobStoreRepository().getNamedXContentRegistry(), Streams.readFully(inputStream));
108+
}
109+
}

‎server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateAttributesManager.java

+30-68
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,9 @@
1818
import org.opensearch.common.blobstore.BlobContainer;
1919
import org.opensearch.core.action.ActionListener;
2020
import org.opensearch.core.xcontent.ToXContent;
21+
import org.opensearch.gateway.remote.RemoteClusterStateUtils.RemoteStateTransferException;
2122
import org.opensearch.index.remote.RemoteStoreUtils;
23+
import org.opensearch.index.translog.transfer.BlobStoreTransferService;
2224
import org.opensearch.repositories.blobstore.BlobStoreRepository;
2325
import org.opensearch.repositories.blobstore.ChecksumBlobStoreFormat;
2426
import org.opensearch.threadpool.ThreadPool;
@@ -35,23 +37,17 @@ public class RemoteClusterStateAttributesManager {
3537
public static final String DISCOVERY_NODES = "nodes";
3638
public static final String CLUSTER_BLOCKS = "blocks";
3739
public static final String CUSTOM_PREFIX = "custom";
38-
public static final ChecksumBlobStoreFormat<DiscoveryNodes> DISCOVERY_NODES_FORMAT = new ChecksumBlobStoreFormat<>(
39-
"nodes",
40-
METADATA_NAME_FORMAT,
41-
DiscoveryNodes::fromXContent
42-
);
43-
public static final ChecksumBlobStoreFormat<ClusterBlocks> CLUSTER_BLOCKS_FORMAT = new ChecksumBlobStoreFormat<>(
44-
"blocks",
45-
METADATA_NAME_FORMAT,
46-
ClusterBlocks::fromXContent
47-
);
4840
public static final int CLUSTER_STATE_ATTRIBUTES_CURRENT_CODEC_VERSION = 1;
41+
private final BlobStoreTransferService blobStoreTransferService;
4942
private final BlobStoreRepository blobStoreRepository;
5043
private final ThreadPool threadPool;
44+
private final String clusterName;
5145

52-
RemoteClusterStateAttributesManager(BlobStoreRepository repository, ThreadPool threadPool) {
46+
RemoteClusterStateAttributesManager(BlobStoreTransferService blobStoreTransferService, BlobStoreRepository repository, ThreadPool threadPool, String clusterName) {
47+
this.blobStoreTransferService = blobStoreTransferService;
5348
this.blobStoreRepository = repository;
5449
this.threadPool = threadPool;
50+
this.clusterName = clusterName;
5551
}
5652

5753
/**
@@ -60,82 +56,48 @@ public class RemoteClusterStateAttributesManager {
6056
CheckedRunnable<IOException> getAsyncMetadataWriteAction(
6157
ClusterState clusterState,
6258
String component,
63-
ChecksumBlobStoreFormat componentMetadataBlobStore,
6459
ToXContent componentData,
6560
LatchedActionListener<ClusterMetadataManifest.UploadedMetadata> latchedActionListener
6661
) {
67-
final BlobContainer remoteStateAttributeContainer = clusterStateAttributeContainer(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID());
68-
final String componentMetadataFilename = clusterStateAttributeFileName(component, clusterState.metadata().version());
62+
AbstractRemoteBlobStoreObject remoteObject = getRemoteObject(componentData, clusterState.version(), clusterState.metadata().clusterUUID());
6963
ActionListener<Void> completionListener = ActionListener.wrap(
7064
resp -> latchedActionListener.onResponse(
71-
new ClusterMetadataManifest.UploadedMetadataAttribute(component, componentMetadataFilename)
65+
remoteObject.getUploadedMetadata()
7266
),
7367
ex -> latchedActionListener.onFailure(new RemoteClusterStateUtils.RemoteStateTransferException(component, ex))
7468
);
75-
return () -> componentMetadataBlobStore.writeAsyncWithUrgentPriority(
76-
componentData,
77-
remoteStateAttributeContainer,
78-
componentMetadataFilename,
79-
blobStoreRepository.getCompressor(),
80-
completionListener,
81-
FORMAT_PARAMS
82-
);
69+
return remoteObject.writeAsync(completionListener);
70+
}
71+
72+
private AbstractRemoteBlobStoreObject getRemoteObject(ToXContent componentData, long stateVersion, String clusterUUID) {
73+
if (componentData instanceof DiscoveryNodes) {
74+
return new RemoteDiscoveryNodes((DiscoveryNodes)componentData, stateVersion, clusterUUID, blobStoreTransferService, blobStoreRepository, clusterName, threadPool);
75+
} else if (componentData instanceof ClusterBlocks) {
76+
return new RemoteClusterBlocks((ClusterBlocks) componentData, stateVersion, clusterUUID, blobStoreTransferService, blobStoreRepository, clusterName, threadPool);
77+
} else {
78+
throw new RemoteStateTransferException("Remote object not found for "+ componentData.getClass());
79+
}
8380
}
8481

8582
public CheckedRunnable<IOException> getAsyncMetadataReadAction(
86-
String clusterName,
8783
String clusterUUID,
8884
String component,
8985
String uploadedFilename,
90-
ChecksumBlobStoreFormat componentBlobStore,
9186
LatchedActionListener<RemoteClusterStateUtils.RemoteReadResult> listener
9287
) {
93-
String[] splitFilename = uploadedFilename.split("/");
94-
return () -> componentBlobStore.readAsync(
95-
clusterStateAttributeContainer(clusterName, clusterUUID),
96-
splitFilename[splitFilename.length -1],
97-
blobStoreRepository.getNamedXContentRegistry(),
98-
threadPool.executor(ThreadPool.Names.GENERIC),
99-
ActionListener.wrap(response -> listener.onResponse(new RemoteClusterStateUtils.RemoteReadResult((ToXContent) response, CLUSTER_STATE_ATTRIBUTE, component)), listener::onFailure)
100-
);
88+
AbstractRemoteBlobStoreObject remoteObject = getRemoteObject(component, uploadedFilename, clusterUUID);
89+
ActionListener actionListener = ActionListener.wrap(response -> listener.onResponse(new RemoteClusterStateUtils.RemoteReadResult((ToXContent) response, CLUSTER_STATE_ATTRIBUTE, component)), listener::onFailure);
90+
return () -> remoteObject.readAsync(actionListener);
10191
}
10292

103-
public ToXContent readMetadata(ChecksumBlobStoreFormat componentMetadataBlobStore, String clusterName, String clusterUUID, String fileName) {
104-
final BlobContainer remoteStateAttributeContainer = clusterStateAttributeContainer(clusterName, clusterUUID);
105-
try {
106-
// Fetch custom metadata
107-
if (fileName != null) {
108-
String[] splitPath = fileName.split("/");
109-
return componentMetadataBlobStore.read(
110-
remoteStateAttributeContainer,
111-
splitPath[splitPath.length - 1],
112-
blobStoreRepository.getNamedXContentRegistry()
113-
);
114-
} else {
115-
return TemplatesMetadata.EMPTY_METADATA;
116-
}
117-
} catch (IOException e) {
118-
throw new IllegalStateException(
119-
String.format(Locale.ROOT, "Error while downloading Templates Metadata - %s", fileName),
120-
e
121-
);
93+
private AbstractRemoteBlobStoreObject getRemoteObject(String component, String blobName, String clusterUUID) {
94+
if (component.equals(RemoteDiscoveryNodes.DISCOVERY_NODES)) {
95+
return new RemoteDiscoveryNodes(blobName, clusterUUID, blobStoreTransferService, blobStoreRepository, clusterName, threadPool);
96+
} else if (component.equals(RemoteClusterBlocks.CLUSTER_BLOCKS)) {
97+
return new RemoteClusterBlocks(blobName, clusterUUID, blobStoreTransferService, blobStoreRepository, clusterName, threadPool);
98+
} else {
99+
throw new RemoteStateTransferException("Remote object not found for "+ component);
122100
}
123101
}
124102

125-
private BlobContainer clusterStateAttributeContainer(String clusterName, String clusterUUID) {
126-
// 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/
127-
return blobStoreRepository.blobStore()
128-
.blobContainer(getCusterMetadataBasePath(blobStoreRepository, clusterName, clusterUUID));
129-
}
130-
131-
private static String clusterStateAttributeFileName(String componentPrefix, Long metadataVersion) {
132-
// 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/<componentPrefix>__<inverted_metadata_version>__<inverted__timestamp>__<codec_version>
133-
return String.join(
134-
DELIMITER,
135-
componentPrefix,
136-
RemoteStoreUtils.invertLong(metadataVersion),
137-
RemoteStoreUtils.invertLong(System.currentTimeMillis()),
138-
String.valueOf(CLUSTER_STATE_ATTRIBUTES_CURRENT_CODEC_VERSION)
139-
);
140-
}
141103
}

‎server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManager.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@
3737
import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.GLOBAL_METADATA_FORMAT;
3838
import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.GLOBAL_METADATA_PATH_TOKEN;
3939
import static org.opensearch.gateway.remote.RemoteIndexMetadata.INDEX_METADATA_FORMAT;
40-
import static org.opensearch.gateway.remote.RemoteIndexMetadataManager.INDEX_PATH_TOKEN;
40+
import static org.opensearch.gateway.remote.RemoteIndexMetadata.INDEX_PATH_TOKEN;
4141
import static org.opensearch.gateway.remote.RemoteManifestManager.MANIFEST_FILE_PREFIX;
4242
import static org.opensearch.gateway.remote.RemoteManifestManager.MANIFEST_PATH_TOKEN;
4343

0 commit comments

Comments
 (0)