-
Notifications
You must be signed in to change notification settings - Fork 24.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Extract remote "sniffing" to connection strategy (#47253)
Currently the connection strategy used by the remote cluster service is implemented as a multi-step sniffing process in the RemoteClusterConnection. We intend to introduce a new connection strategy that will operate in a different manner. This commit extracts the sniffing logic to a dedicated strategy class. Additionally, it implements dedicated tests for this class. Additionally, in previous commits we moved away from a world where the remote cluster connection was mutable. Instead, when setting updates are made, the connection is torn down and rebuilt. We still had methods and tests hanging around for the mutable behavior. This commit removes those.
- Loading branch information
1 parent
bc96370
commit 50e0c96
Showing
8 changed files
with
1,007 additions
and
1,018 deletions.
There are no files selected for viewing
378 changes: 19 additions & 359 deletions
378
server/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
164 changes: 164 additions & 0 deletions
164
server/src/main/java/org/elasticsearch/transport/RemoteConnectionStrategy.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,164 @@ | ||
| /* | ||
| * Licensed to Elasticsearch under one or more contributor | ||
| * license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright | ||
| * ownership. Elasticsearch licenses this file to you under | ||
| * the Apache License, Version 2.0 (the "License"); you may | ||
| * not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.transport; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.apache.logging.log4j.message.ParameterizedMessage; | ||
| import org.apache.lucene.store.AlreadyClosedException; | ||
| import org.elasticsearch.action.ActionListener; | ||
| import org.elasticsearch.action.support.ContextPreservingActionListener; | ||
| import org.elasticsearch.cluster.node.DiscoveryNode; | ||
| import org.elasticsearch.common.util.concurrent.AbstractRunnable; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
|
|
||
| import java.io.Closeable; | ||
| import java.util.ArrayList; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.concurrent.RejectedExecutionException; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
|
|
||
| public abstract class RemoteConnectionStrategy implements TransportConnectionListener, Closeable { | ||
|
|
||
| protected static final Logger logger = LogManager.getLogger(RemoteConnectionStrategy.class); | ||
|
|
||
| private static final int MAX_LISTENERS = 100; | ||
| private final AtomicBoolean closed = new AtomicBoolean(false); | ||
| private final Object mutex = new Object(); | ||
| private final ThreadPool threadPool; | ||
| protected final RemoteConnectionManager connectionManager; | ||
| private List<ActionListener<Void>> listeners = new ArrayList<>(); | ||
|
|
||
| RemoteConnectionStrategy(ThreadPool threadPool, RemoteConnectionManager connectionManager) { | ||
| this.threadPool = threadPool; | ||
| this.connectionManager = connectionManager; | ||
| connectionManager.getConnectionManager().addListener(this); | ||
| } | ||
|
|
||
| /** | ||
| * Triggers a connect round unless there is one running already. If there is a connect round running, the listener will either | ||
| * be queued or rejected and failed. | ||
| */ | ||
| void connect(ActionListener<Void> connectListener) { | ||
| boolean runConnect = false; | ||
| final ActionListener<Void> listener = | ||
| ContextPreservingActionListener.wrapPreservingContext(connectListener, threadPool.getThreadContext()); | ||
| boolean closed; | ||
| synchronized (mutex) { | ||
| closed = this.closed.get(); | ||
| if (closed) { | ||
| assert listeners.isEmpty(); | ||
| } else { | ||
| if (listeners.size() >= MAX_LISTENERS) { | ||
| assert listeners.size() == MAX_LISTENERS; | ||
| listener.onFailure(new RejectedExecutionException("connect listener queue is full")); | ||
| return; | ||
| } else { | ||
| listeners.add(listener); | ||
| } | ||
| runConnect = listeners.size() == 1; | ||
| } | ||
| } | ||
| if (closed) { | ||
| connectListener.onFailure(new AlreadyClosedException("connect handler is already closed")); | ||
| return; | ||
| } | ||
| if (runConnect) { | ||
| ExecutorService executor = threadPool.executor(ThreadPool.Names.MANAGEMENT); | ||
| executor.submit(new AbstractRunnable() { | ||
| @Override | ||
| public void onFailure(Exception e) { | ||
| ActionListener.onFailure(getAndClearListeners(), e); | ||
| } | ||
|
|
||
| @Override | ||
| protected void doRun() { | ||
| connectImpl(new ActionListener<>() { | ||
| @Override | ||
| public void onResponse(Void aVoid) { | ||
| ActionListener.onResponse(getAndClearListeners(), aVoid); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(Exception e) { | ||
| ActionListener.onFailure(getAndClearListeners(), e); | ||
| } | ||
| }); | ||
| } | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) { | ||
| if (shouldOpenMoreConnections()) { | ||
| // try to reconnect and fill up the slot of the disconnected node | ||
| connect(ActionListener.wrap( | ||
| ignore -> logger.trace("successfully connected after disconnect of {}", node), | ||
| e -> logger.trace(() -> new ParameterizedMessage("failed to connect after disconnect of {}", node), e))); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| final List<ActionListener<Void>> toNotify; | ||
| synchronized (mutex) { | ||
| if (closed.compareAndSet(false, true)) { | ||
| connectionManager.getConnectionManager().removeListener(this); | ||
| toNotify = listeners; | ||
| listeners = Collections.emptyList(); | ||
| } else { | ||
| toNotify = Collections.emptyList(); | ||
| } | ||
| } | ||
| ActionListener.onFailure(toNotify, new AlreadyClosedException("connect handler is already closed")); | ||
| } | ||
|
|
||
| public boolean isClosed() { | ||
| return closed.get(); | ||
| } | ||
|
|
||
| // for testing only | ||
| boolean assertNoRunningConnections() { | ||
| synchronized (mutex) { | ||
| assert listeners.isEmpty(); | ||
| } | ||
| return true; | ||
| } | ||
|
|
||
| protected abstract boolean shouldOpenMoreConnections(); | ||
|
|
||
| protected abstract void connectImpl(ActionListener<Void> listener); | ||
|
|
||
| private List<ActionListener<Void>> getAndClearListeners() { | ||
| final List<ActionListener<Void>> result; | ||
| synchronized (mutex) { | ||
| if (listeners.isEmpty()) { | ||
| result = Collections.emptyList(); | ||
| } else { | ||
| result = listeners; | ||
| listeners = new ArrayList<>(); | ||
| } | ||
| } | ||
| return result; | ||
| } | ||
| } |
Oops, something went wrong.