Skip to content

Commit c6c5f9f

Browse files
ywangdywangd
committed
Move existing client management code across
1 parent cca38e4 commit c6c5f9f

File tree

3 files changed

+124
-54
lines changed

3 files changed

+124
-54
lines changed

‎modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageClientsManager.java‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,15 +11,126 @@
1111

1212
import org.elasticsearch.cluster.ClusterChangedEvent;
1313
import org.elasticsearch.cluster.ClusterStateApplier;
14+
import org.elasticsearch.common.CheckedBiFunction;
15+
import org.elasticsearch.common.Strings;
16+
import org.elasticsearch.common.util.Maps;
1417
import org.elasticsearch.logging.LogManager;
1518
import org.elasticsearch.logging.Logger;
1619

20+
import java.io.IOException;
21+
import java.util.Map;
22+
import java.util.stream.Collectors;
23+
24+
import static java.util.Collections.emptyMap;
25+
import static org.elasticsearch.core.Strings.format;
26+
1727
public class GoogleCloudStorageClientsManager implements ClusterStateApplier {
1828

1929
private static final Logger logger = LogManager.getLogger(GoogleCloudStorageClientsManager.class);
2030

31+
private final CheckedBiFunction<
32+
GoogleCloudStorageClientSettings,
33+
GcsRepositoryStatsCollector,
34+
MeteredStorage,
35+
IOException> clientBuilder;
36+
private final ClientsHolder clusterClientsHolder;
37+
38+
public GoogleCloudStorageClientsManager(
39+
CheckedBiFunction<GoogleCloudStorageClientSettings, GcsRepositoryStatsCollector, MeteredStorage, IOException> clientBuilder
40+
) {
41+
this.clientBuilder = clientBuilder;
42+
this.clusterClientsHolder = new ClientsHolder();
43+
}
44+
2145
@Override
2246
public void applyClusterState(ClusterChangedEvent event) {
2347

2448
}
49+
50+
void refreshAndClearCache(Map<String, GoogleCloudStorageClientSettings> clientsSettings) {
51+
clusterClientsHolder.refreshAndClearCache(clientsSettings);
52+
}
53+
54+
MeteredStorage client(final String clientName, final String repositoryName, final GcsRepositoryStatsCollector statsCollector)
55+
throws IOException {
56+
return clusterClientsHolder.client(clientName, repositoryName, statsCollector);
57+
}
58+
59+
void closeRepositoryClients(String repositoryName) {
60+
clusterClientsHolder.closeRepositoryClients(repositoryName);
61+
}
62+
63+
class ClientsHolder {
64+
65+
private volatile Map<String, GoogleCloudStorageClientSettings> clientSettings = emptyMap();
66+
/**
67+
* Dictionary of client instances. Client instances are built lazily from the
68+
* latest settings. Clients are cached by a composite repositoryName key.
69+
*/
70+
private volatile Map<String, MeteredStorage> clientCache = emptyMap();
71+
72+
/**
73+
* Attempts to retrieve a client from the cache. If the client does not exist it
74+
* will be created from the latest settings and will populate the cache. The
75+
* returned instance should not be cached by the calling code. Instead, for each
76+
* use, the (possibly updated) instance should be requested by calling this
77+
* method.
78+
*
79+
* @param clientName name of the client settings used to create the client
80+
* @param repositoryName name of the repository that would use the client
81+
* @return a cached client storage instance that can be used to manage objects
82+
* (blobs)
83+
*/
84+
MeteredStorage client(final String clientName, final String repositoryName, final GcsRepositoryStatsCollector statsCollector)
85+
throws IOException {
86+
{
87+
final MeteredStorage storage = clientCache.get(repositoryName);
88+
if (storage != null) {
89+
return storage;
90+
}
91+
}
92+
synchronized (this) {
93+
final MeteredStorage existing = clientCache.get(repositoryName);
94+
95+
if (existing != null) {
96+
return existing;
97+
}
98+
99+
final GoogleCloudStorageClientSettings settings = clientSettings.get(clientName);
100+
101+
if (settings == null) {
102+
throw new IllegalArgumentException(
103+
"Unknown client name ["
104+
+ clientName
105+
+ "]. Existing client configs: "
106+
+ Strings.collectionToDelimitedString(clientSettings.keySet(), ",")
107+
);
108+
}
109+
110+
logger.debug(() -> format("creating GCS client with client_name [%s], endpoint [%s]", clientName, settings.getHost()));
111+
final MeteredStorage storage = clientBuilder.apply(settings, statsCollector);
112+
clientCache = Maps.copyMapWithAddedEntry(clientCache, repositoryName, storage);
113+
return storage;
114+
}
115+
}
116+
117+
/**
118+
* Refreshes the client settings and clears the client cache. Subsequent calls to
119+
* {@code GoogleCloudStorageService#client} will return new clients constructed
120+
* using the parameter settings.
121+
*
122+
* @param clientsSettings the new settings used for building clients for subsequent requests
123+
*/
124+
synchronized void refreshAndClearCache(Map<String, GoogleCloudStorageClientSettings> clientsSettings) {
125+
this.clientCache = emptyMap();
126+
this.clientSettings = Maps.ofEntries(clientsSettings.entrySet());
127+
}
128+
129+
synchronized void closeRepositoryClients(String repositoryName) {
130+
clientCache = clientCache.entrySet()
131+
.stream()
132+
.filter(entry -> entry.getKey().equals(repositoryName) == false)
133+
.collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
134+
}
135+
}
25136
}

‎modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageService.java‎

Lines changed: 9 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@
2929
import org.elasticsearch.cluster.project.ProjectResolver;
3030
import org.elasticsearch.cluster.service.ClusterService;
3131
import org.elasticsearch.common.Strings;
32-
import org.elasticsearch.common.util.Maps;
3332
import org.elasticsearch.core.Nullable;
3433
import org.elasticsearch.core.SuppressForbidden;
3534
import org.elasticsearch.core.TimeValue;
@@ -46,49 +45,38 @@
4645
import java.net.UnknownHostException;
4746
import java.security.KeyStore;
4847
import java.util.Map;
49-
import java.util.stream.Collectors;
5048

5149
import static java.nio.charset.StandardCharsets.UTF_8;
52-
import static java.util.Collections.emptyMap;
53-
import static org.elasticsearch.core.Strings.format;
5450

5551
public class GoogleCloudStorageService {
5652

5753
private static final Logger logger = LogManager.getLogger(GoogleCloudStorageService.class);
58-
private final GoogleCloudStorageClientsManager googleCloudStorageClientsManager;
59-
60-
private volatile Map<String, GoogleCloudStorageClientSettings> clientSettings = emptyMap();
54+
private final GoogleCloudStorageClientsManager clientsManager;
6155

6256
private final boolean isServerless;
6357

58+
@SuppressWarnings("this-escape")
6459
public GoogleCloudStorageService(ClusterService clusterService, ProjectResolver projectResolver) {
6560
this.isServerless = DiscoveryNode.isStateless(clusterService.getSettings());
66-
this.googleCloudStorageClientsManager = new GoogleCloudStorageClientsManager();
61+
this.clientsManager = new GoogleCloudStorageClientsManager(this::createClient);
6762
if (projectResolver.supportsMultipleProjects()) {
68-
clusterService.addHighPriorityApplier(this.googleCloudStorageClientsManager);
63+
clusterService.addHighPriorityApplier(this.clientsManager);
6964
}
7065
}
7166

7267
public boolean isServerless() {
7368
return isServerless;
7469
}
7570

76-
/**
77-
* Dictionary of client instances. Client instances are built lazily from the
78-
* latest settings. Clients are cached by a composite repositoryName key.
79-
*/
80-
private volatile Map<String, MeteredStorage> clientCache = emptyMap();
81-
8271
/**
8372
* Refreshes the client settings and clears the client cache. Subsequent calls to
8473
* {@code GoogleCloudStorageService#client} will return new clients constructed
8574
* using the parameter settings.
8675
*
8776
* @param clientsSettings the new settings used for building clients for subsequent requests
8877
*/
89-
public synchronized void refreshAndClearCache(Map<String, GoogleCloudStorageClientSettings> clientsSettings) {
90-
this.clientCache = emptyMap();
91-
this.clientSettings = Maps.ofEntries(clientsSettings.entrySet());
78+
public void refreshAndClearCache(Map<String, GoogleCloudStorageClientSettings> clientsSettings) {
79+
clientsManager.refreshAndClearCache(clientsSettings);
9280
}
9381

9482
/**
@@ -105,42 +93,11 @@ public synchronized void refreshAndClearCache(Map<String, GoogleCloudStorageClie
10593
*/
10694
public MeteredStorage client(final String clientName, final String repositoryName, final GcsRepositoryStatsCollector statsCollector)
10795
throws IOException {
108-
{
109-
final MeteredStorage storage = clientCache.get(repositoryName);
110-
if (storage != null) {
111-
return storage;
112-
}
113-
}
114-
synchronized (this) {
115-
final MeteredStorage existing = clientCache.get(repositoryName);
116-
117-
if (existing != null) {
118-
return existing;
119-
}
120-
121-
final GoogleCloudStorageClientSettings settings = clientSettings.get(clientName);
122-
123-
if (settings == null) {
124-
throw new IllegalArgumentException(
125-
"Unknown client name ["
126-
+ clientName
127-
+ "]. Existing client configs: "
128-
+ Strings.collectionToDelimitedString(clientSettings.keySet(), ",")
129-
);
130-
}
131-
132-
logger.debug(() -> format("creating GCS client with client_name [%s], endpoint [%s]", clientName, settings.getHost()));
133-
final MeteredStorage storage = createClient(settings, statsCollector);
134-
clientCache = Maps.copyMapWithAddedEntry(clientCache, repositoryName, storage);
135-
return storage;
136-
}
96+
return clientsManager.client(clientName, repositoryName, statsCollector);
13797
}
13898

139-
synchronized void closeRepositoryClients(String repositoryName) {
140-
clientCache = clientCache.entrySet()
141-
.stream()
142-
.filter(entry -> entry.getKey().equals(repositoryName) == false)
143-
.collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
99+
void closeRepositoryClients(String repositoryName) {
100+
clientsManager.closeRepositoryClients(repositoryName);
144101
}
145102

146103
/**

‎modules/repository-gcs/src/test/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobContainerStatsTests.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -106,8 +106,10 @@ public void createStorageService() throws Exception {
106106
threadPool = new TestThreadPool(getTestClass().getName());
107107
httpServer = MockHttpServer.createHttp(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
108108
httpServer.start();
109-
googleCloudStorageService = new GoogleCloudStorageService(ClusterServiceUtils.createClusterService(threadPool),
110-
TestProjectResolvers.DEFAULT_PROJECT_ONLY);
109+
googleCloudStorageService = new GoogleCloudStorageService(
110+
ClusterServiceUtils.createClusterService(threadPool),
111+
TestProjectResolvers.DEFAULT_PROJECT_ONLY
112+
);
111113
googleCloudStorageHttpHandler = new GoogleCloudStorageHttpHandler(BUCKET);
112114
httpServer.createContext("/", googleCloudStorageHttpHandler);
113115
httpServer.createContext("/token", new FakeOAuth2HttpHandler());

0 commit comments

Comments
 (0)