2727import org .elasticsearch .cluster .ClusterName ;
2828import org .elasticsearch .cluster .node .DiscoveryNode ;
2929import org .elasticsearch .cluster .node .DiscoveryNodeRole ;
30+ import org .elasticsearch .common .Strings ;
3031import org .elasticsearch .common .io .stream .StreamInput ;
3132import org .elasticsearch .common .io .stream .StreamOutput ;
3233import org .elasticsearch .common .io .stream .Writeable ;
3738import org .elasticsearch .common .xcontent .XContentBuilder ;
3839
3940import java .io .IOException ;
40- import java .util .Arrays ;
4141import java .util .Collections ;
42- import java .util .HashSet ;
4342import java .util .List ;
4443import java .util .Map ;
4544import java .util .Objects ;
46- import java .util .Set ;
4745import java .util .concurrent .atomic .AtomicInteger ;
4846import java .util .concurrent .atomic .AtomicLong ;
4947import java .util .concurrent .atomic .AtomicReference ;
5048import java .util .function .Supplier ;
51- import java .util .stream .Collectors ;
5249import java .util .stream .Stream ;
5350
5451import static org .elasticsearch .common .settings .Setting .boolSetting ;
5754public class ProxyConnectionStrategy extends RemoteConnectionStrategy {
5855
5956 /**
60- * A list of addresses for remote cluster connections. The connections will be opened to the configured addresses in a round-robin
61- * fashion.
57+ * The remote address for the proxy. The connections will be opened to the configured address.
6258 */
63- public static final Setting .AffixSetting <List < String > > REMOTE_CLUSTER_ADDRESSES = Setting .affixKeySetting (
59+ public static final Setting .AffixSetting <String > REMOTE_CLUSTER_ADDRESSES = Setting .affixKeySetting (
6460 "cluster.remote." ,
65- "proxy_addresses" ,
66- (ns , key ) -> Setting .listSetting (key , Collections .emptyList (), s -> {
67- // validate address
68- parsePort (s );
69- return s ;
70- }, new StrategyValidator <>(ns , key , ConnectionStrategy .PROXY ),
71- Setting .Property .Dynamic , Setting .Property .NodeScope ));
61+ "proxy_address" ,
62+ (ns , key ) -> Setting .simpleString (key , new StrategyValidator <>(ns , key , ConnectionStrategy .PROXY , s -> {
63+ if (Strings .hasLength (s )) {
64+ parsePort (s );
65+ }
66+ }), Setting .Property .Dynamic , Setting .Property .NodeScope ));
7267
7368 /**
7469 * The maximum number of socket connections that will be established to a remote cluster. The default is 18.
@@ -95,9 +90,9 @@ public class ProxyConnectionStrategy extends RemoteConnectionStrategy {
9590
9691 private final int maxNumConnections ;
9792 private final AtomicLong counter = new AtomicLong (0 );
98- private final List < String > configuredAddresses ;
93+ private final String configuredAddress ;
9994 private final boolean includeServerName ;
100- private final List < Supplier <TransportAddress >> addresses ;
95+ private final Supplier <TransportAddress > address ;
10196 private final AtomicReference <ClusterName > remoteClusterName = new AtomicReference <>();
10297 private final ConnectionProfile profile ;
10398 private final ConnectionManager .ConnectionValidator clusterNameValidator ;
@@ -114,28 +109,26 @@ public class ProxyConnectionStrategy extends RemoteConnectionStrategy {
114109 }
115110
116111 ProxyConnectionStrategy (String clusterAlias , TransportService transportService , RemoteConnectionManager connectionManager ,
117- int maxNumConnections , List <String > configuredAddresses ) {
118- this (clusterAlias , transportService , connectionManager , maxNumConnections , configuredAddresses ,
119- configuredAddresses .stream ().map (address ->
120- (Supplier <TransportAddress >) () -> resolveAddress (address )).collect (Collectors .toList ()), false );
112+ int maxNumConnections , String configuredAddress ) {
113+ this (clusterAlias , transportService , connectionManager , maxNumConnections , configuredAddress ,
114+ () -> resolveAddress (configuredAddress ), false );
121115 }
122116
123117 ProxyConnectionStrategy (String clusterAlias , TransportService transportService , RemoteConnectionManager connectionManager ,
124- int maxNumConnections , List <String > configuredAddresses , boolean includeServerName ) {
125- this (clusterAlias , transportService , connectionManager , maxNumConnections , configuredAddresses ,
126- configuredAddresses .stream ().map (address ->
127- (Supplier <TransportAddress >) () -> resolveAddress (address )).collect (Collectors .toList ()), includeServerName );
118+ int maxNumConnections , String configuredAddress , boolean includeServerName ) {
119+ this (clusterAlias , transportService , connectionManager , maxNumConnections , configuredAddress ,
120+ () -> resolveAddress (configuredAddress ), includeServerName );
128121 }
129122
130123 ProxyConnectionStrategy (String clusterAlias , TransportService transportService , RemoteConnectionManager connectionManager ,
131- int maxNumConnections , List < String > configuredAddresses , List < Supplier <TransportAddress >> addresses ,
124+ int maxNumConnections , String configuredAddress , Supplier <TransportAddress > address ,
132125 boolean includeServerName ) {
133126 super (clusterAlias , transportService , connectionManager );
134127 this .maxNumConnections = maxNumConnections ;
135- this .configuredAddresses = configuredAddresses ;
128+ this .configuredAddress = configuredAddress ;
136129 this .includeServerName = includeServerName ;
137- assert addresses .isEmpty () == false : "Cannot use proxy connection strategy with no configured addresses" ;
138- this .addresses = addresses ;
130+ assert Strings .isEmpty (configuredAddress ) == false : "Cannot use proxy connection strategy with no configured addresses" ;
131+ this .address = address ;
139132 // TODO: Move into the ConnectionManager
140133 this .profile = new ConnectionProfile .Builder ()
141134 .addConnections (1 , TransportRequestOptions .Type .REG , TransportRequestOptions .Type .PING )
@@ -172,9 +165,9 @@ protected boolean shouldOpenMoreConnections() {
172165
173166 @ Override
174167 protected boolean strategyMustBeRebuilt (Settings newSettings ) {
175- List < String > addresses = REMOTE_CLUSTER_ADDRESSES .getConcreteSettingForNamespace (clusterAlias ).get (newSettings );
168+ String address = REMOTE_CLUSTER_ADDRESSES .getConcreteSettingForNamespace (clusterAlias ).get (newSettings );
176169 int numOfSockets = REMOTE_SOCKET_CONNECTIONS .getConcreteSettingForNamespace (clusterAlias ).get (newSettings );
177- return numOfSockets != maxNumConnections || addressesChanged ( configuredAddresses , addresses ) ;
170+ return numOfSockets != maxNumConnections || configuredAddress . equals ( address ) == false ;
178171 }
179172
180173 @ Override
@@ -189,7 +182,7 @@ protected void connectImpl(ActionListener<Void> listener) {
189182
190183 @ Override
191184 public RemoteConnectionInfo .ModeInfo getModeInfo () {
192- return new ProxyModeInfo (configuredAddresses , maxNumConnections , connectionManager .size ());
185+ return new ProxyModeInfo (configuredAddress , maxNumConnections , connectionManager .size ());
193186 }
194187
195188 private void performProxyConnectionProcess (ActionListener <Void > listener ) {
@@ -198,7 +191,7 @@ private void performProxyConnectionProcess(ActionListener<Void> listener) {
198191
199192 private void openConnections (ActionListener <Void > finished , int attemptNumber ) {
200193 if (attemptNumber <= MAX_CONNECT_ATTEMPTS_PER_RUN ) {
201- List < TransportAddress > resolved = addresses . stream (). map ( Supplier :: get ). collect ( Collectors . toList () );
194+ TransportAddress resolved = address . get ( );
202195
203196 int remaining = maxNumConnections - connectionManager .size ();
204197 ActionListener <Void > compositeListener = new ActionListener <>() {
@@ -228,15 +221,14 @@ public void onFailure(Exception e) {
228221
229222
230223 for (int i = 0 ; i < remaining ; ++i ) {
231- TransportAddress address = nextAddress (resolved );
232- String id = clusterAlias + "#" + address ;
224+ String id = clusterAlias + "#" + resolved ;
233225 Map <String , String > attributes ;
234226 if (includeServerName ) {
235- attributes = Collections .singletonMap ("server_name" , address .address ().getHostString ());
227+ attributes = Collections .singletonMap ("server_name" , resolved .address ().getHostString ());
236228 } else {
237229 attributes = Collections .emptyMap ();
238230 }
239- DiscoveryNode node = new DiscoveryNode (id , address , attributes , DiscoveryNodeRole .BUILT_IN_ROLES ,
231+ DiscoveryNode node = new DiscoveryNode (id , resolved , attributes , DiscoveryNodeRole .BUILT_IN_ROLES ,
240232 Version .CURRENT .minimumCompatibilityVersion ());
241233
242234 connectionManager .connectToNode (node , profile , clusterNameValidator , new ActionListener <>() {
@@ -248,7 +240,7 @@ public void onResponse(Void v) {
248240 @ Override
249241 public void onFailure (Exception e ) {
250242 logger .debug (new ParameterizedMessage ("failed to open remote connection [remote cluster: {}, address: {}]" ,
251- clusterAlias , address ), e );
243+ clusterAlias , resolved ), e );
252244 compositeListener .onFailure (e );
253245 }
254246 });
@@ -276,48 +268,35 @@ private static TransportAddress resolveAddress(String address) {
276268 return new TransportAddress (parseConfiguredAddress (address ));
277269 }
278270
279- private boolean addressesChanged (final List <String > oldAddresses , final List <String > newAddresses ) {
280- if (oldAddresses .size () != newAddresses .size ()) {
281- return true ;
282- }
283- Set <String > oldSeeds = new HashSet <>(oldAddresses );
284- Set <String > newSeeds = new HashSet <>(newAddresses );
285- return oldSeeds .equals (newSeeds ) == false ;
286- }
287-
288271 static class ProxyModeInfo implements RemoteConnectionInfo .ModeInfo {
289272
290- private final List < String > addresses ;
273+ private final String address ;
291274 private final int maxSocketConnections ;
292275 private final int numSocketsConnected ;
293276
294- ProxyModeInfo (List < String > addresses , int maxSocketConnections , int numSocketsConnected ) {
295- this .addresses = addresses ;
277+ ProxyModeInfo (String address , int maxSocketConnections , int numSocketsConnected ) {
278+ this .address = address ;
296279 this .maxSocketConnections = maxSocketConnections ;
297280 this .numSocketsConnected = numSocketsConnected ;
298281 }
299282
300283 private ProxyModeInfo (StreamInput input ) throws IOException {
301- addresses = Arrays . asList ( input .readStringArray () );
284+ address = input .readString ( );
302285 maxSocketConnections = input .readVInt ();
303286 numSocketsConnected = input .readVInt ();
304287 }
305288
306289 @ Override
307290 public XContentBuilder toXContent (XContentBuilder builder , Params params ) throws IOException {
308- builder .startArray ("addresses" );
309- for (String address : addresses ) {
310- builder .value (address );
311- }
312- builder .endArray ();
291+ builder .field ("address" , address );
313292 builder .field ("num_sockets_connected" , numSocketsConnected );
314293 builder .field ("max_socket_connections" , maxSocketConnections );
315294 return builder ;
316295 }
317296
318297 @ Override
319298 public void writeTo (StreamOutput out ) throws IOException {
320- out .writeStringArray ( addresses . toArray ( new String [ 0 ]) );
299+ out .writeString ( address );
321300 out .writeVInt (maxSocketConnections );
322301 out .writeVInt (numSocketsConnected );
323302 }
@@ -344,12 +323,12 @@ public boolean equals(Object o) {
344323 ProxyModeInfo otherProxy = (ProxyModeInfo ) o ;
345324 return maxSocketConnections == otherProxy .maxSocketConnections &&
346325 numSocketsConnected == otherProxy .numSocketsConnected &&
347- Objects .equals (addresses , otherProxy .addresses );
326+ Objects .equals (address , otherProxy .address );
348327 }
349328
350329 @ Override
351330 public int hashCode () {
352- return Objects .hash (addresses , maxSocketConnections , numSocketsConnected );
331+ return Objects .hash (address , maxSocketConnections , numSocketsConnected );
353332 }
354333 }
355334}
0 commit comments