序
本文主要研究一下 elasticsearch 的 PeerFinder
PeersRequest
elasticsearch-7.0.0/server/src/main/java/org/elasticsearch/discovery/PeersRequest.java
public class PeersRequest extends TransportRequest {
private final DiscoveryNode sourceNode;
private final List<DiscoveryNode> knownPeers;
public PeersRequest(DiscoveryNode sourceNode, List<DiscoveryNode> knownPeers) {assert knownPeers.contains(sourceNode) == false : "local node is not a peer";
this.sourceNode = sourceNode;
this.knownPeers = knownPeers;
}
public PeersRequest(StreamInput in) throws IOException {super(in);
sourceNode = new DiscoveryNode(in);
knownPeers = in.readList(DiscoveryNode::new);
}
@Override
public void writeTo(StreamOutput out) throws IOException {super.writeTo(out);
sourceNode.writeTo(out);
out.writeList(knownPeers);
}
public List<DiscoveryNode> getKnownPeers() {return knownPeers;}
public DiscoveryNode getSourceNode() {return sourceNode;}
@Override
public String toString() {
return "PeersRequest{" +
"sourceNode=" + sourceNode +
", knownPeers=" + knownPeers +
'}';
}
@Override
public boolean equals(Object o) {if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
PeersRequest that = (PeersRequest) o;
return Objects.equals(sourceNode, that.sourceNode) &&
Objects.equals(knownPeers, that.knownPeers);
}
@Override
public int hashCode() {return Objects.hash(sourceNode, knownPeers);
}
}- PeersRequest 有两个属性，分别是 sourceNode 以及 knownPeers
PeersResponse
elasticsearch-7.0.0/server/src/main/java/org/elasticsearch/cluster/coordination/PeersResponse.java
public class PeersResponse extends TransportResponse {
private final Optional<DiscoveryNode> masterNode;
private final List<DiscoveryNode> knownPeers;
private final long term;
public PeersResponse(Optional<DiscoveryNode> masterNode, List<DiscoveryNode> knownPeers, long term) {assert masterNode.isPresent() == false || knownPeers.isEmpty();
this.masterNode = masterNode;
this.knownPeers = knownPeers;
this.term = term;
}
public PeersResponse(StreamInput in) throws IOException {masterNode = Optional.ofNullable(in.readOptionalWriteable(DiscoveryNode::new));
knownPeers = in.readList(DiscoveryNode::new);
term = in.readLong();
assert masterNode.isPresent() == false || knownPeers.isEmpty();
}
@Override
public void writeTo(StreamOutput out) throws IOException {super.writeTo(out);
out.writeOptionalWriteable(masterNode.orElse(null));
out.writeList(knownPeers);
out.writeLong(term);
}
/**
* @return the node that is currently leading, according to the responding node.
*/
public Optional<DiscoveryNode> getMasterNode() {return masterNode;}
/**
* @return the collection of known peers of the responding node, or an empty collection if the responding node believes there
* is currently a leader.
*/
public List<DiscoveryNode> getKnownPeers() {return knownPeers;}
/**
* @return the current term of the responding node. If the responding node is the leader then this is the term in which it is
* currently leading.
*/
public long getTerm() {return term;}
@Override
public String toString() {
return "PeersResponse{" +
"masterNode=" + masterNode +
", knownPeers=" + knownPeers +
", term=" + term +
'}';
}
@Override
public boolean equals(Object o) {if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
PeersResponse that = (PeersResponse) o;
return term == that.term &&
Objects.equals(masterNode, that.masterNode) &&
Objects.equals(knownPeers, that.knownPeers);
}
@Override
public int hashCode() {return Objects.hash(masterNode, knownPeers, term);
}
}- PeersResponse 有三个属性，分别是 masterNode、knownPeers、term
PeerFinder
elasticsearch-7.0.0/server/src/main/java/org/elasticsearch/discovery/PeerFinder.java
public abstract class PeerFinder {private static final Logger logger = LogManager.getLogger(PeerFinder.class);
public static final String REQUEST_PEERS_ACTION_NAME = "internal:discovery/request_peers";
// the time between attempts to find all peers
public static final Setting<TimeValue> DISCOVERY_FIND_PEERS_INTERVAL_SETTING =
Setting.timeSetting("discovery.find_peers_interval",
TimeValue.timeValueMillis(1000), TimeValue.timeValueMillis(1), Setting.Property.NodeScope);
public static final Setting<TimeValue> DISCOVERY_REQUEST_PEERS_TIMEOUT_SETTING =
Setting.timeSetting("discovery.request_peers_timeout",
TimeValue.timeValueMillis(3000), TimeValue.timeValueMillis(1), Setting.Property.NodeScope);
private final Settings settings;
private final TimeValue findPeersInterval;
private final TimeValue requestPeersTimeout;
private final Object mutex = new Object();
private final TransportService transportService;
private final TransportAddressConnector transportAddressConnector;
private final ConfiguredHostsResolver configuredHostsResolver;
private volatile long currentTerm;
private boolean active;
private DiscoveryNodes lastAcceptedNodes;
private final Map<TransportAddress, Peer> peersByAddress = new LinkedHashMap<>();
private Optional<DiscoveryNode> leader = Optional.empty();
private volatile List<TransportAddress> lastResolvedAddresses = emptyList();
public PeerFinder(Settings settings, TransportService transportService, TransportAddressConnector transportAddressConnector,
ConfiguredHostsResolver configuredHostsResolver) {
this.settings = settings;
findPeersInterval = DISCOVERY_FIND_PEERS_INTERVAL_SETTING.get(settings);
requestPeersTimeout = DISCOVERY_REQUEST_PEERS_TIMEOUT_SETTING.get(settings);
this.transportService = transportService;
this.transportAddressConnector = transportAddressConnector;
this.configuredHostsResolver = configuredHostsResolver;
transportService.registerRequestHandler(REQUEST_PEERS_ACTION_NAME, Names.GENERIC, false, false,
PeersRequest::new,
(request, channel, task) -> channel.sendResponse(handlePeersRequest(request)));
transportService.registerRequestHandler(UnicastZenPing.ACTION_NAME, Names.GENERIC, false, false,
UnicastZenPing.UnicastPingRequest::new, new Zen1UnicastPingRequestHandler());
}
public void activate(final DiscoveryNodes lastAcceptedNodes) {logger.trace("activating with {}", lastAcceptedNodes);
synchronized (mutex) {assert assertInactiveWithNoKnownPeers();
active = true;
this.lastAcceptedNodes = lastAcceptedNodes;
leader = Optional.empty();
handleWakeUp(); // return value discarded: there are no known peers, so none can be disconnected}
onFoundPeersUpdated(); // trigger a check for a quorum already}
public void deactivate(DiscoveryNode leader) {
final boolean peersRemoved;
synchronized (mutex) {logger.trace("deactivating and setting leader to {}", leader);
active = false;
peersRemoved = handleWakeUp();
this.leader = Optional.of(leader);
assert assertInactiveWithNoKnownPeers();}
if (peersRemoved) {onFoundPeersUpdated();
}
}
// exposed to subclasses for testing
protected final boolean holdsLock() {return Thread.holdsLock(mutex);
}
private boolean assertInactiveWithNoKnownPeers() {assert holdsLock() : "PeerFinder mutex not held";
assert active == false;
assert peersByAddress.isEmpty() : peersByAddress.keySet();
return true;
}
PeersResponse handlePeersRequest(PeersRequest peersRequest) {synchronized (mutex) {assert peersRequest.getSourceNode().equals(getLocalNode()) == false;
final List<DiscoveryNode> knownPeers;
if (active) {assert leader.isPresent() == false : leader;
if (peersRequest.getSourceNode().isMasterNode()) {startProbe(peersRequest.getSourceNode().getAddress());
}
peersRequest.getKnownPeers().stream().map(DiscoveryNode::getAddress).forEach(this::startProbe);
knownPeers = getFoundPeersUnderLock();} else {assert leader.isPresent() || lastAcceptedNodes == null;
knownPeers = emptyList();}
return new PeersResponse(leader, knownPeers, currentTerm);
}
}
// exposed for checking invariant in o.e.c.c.Coordinator (public since this is a different package)
public Optional<DiscoveryNode> getLeader() {synchronized (mutex) {return leader;}
}
// exposed for checking invariant in o.e.c.c.Coordinator (public since this is a different package)
public long getCurrentTerm() {return currentTerm;}
public void setCurrentTerm(long currentTerm) {this.currentTerm = currentTerm;}
private DiscoveryNode getLocalNode() {final DiscoveryNode localNode = transportService.getLocalNode();
assert localNode != null;
return localNode;
}
protected abstract void onActiveMasterFound(DiscoveryNode masterNode, long term);
protected abstract void onFoundPeersUpdated();
public List<TransportAddress> getLastResolvedAddresses() {return lastResolvedAddresses;}
public Iterable<DiscoveryNode> getFoundPeers() {synchronized (mutex) {return getFoundPeersUnderLock();
}
}
private List<DiscoveryNode> getFoundPeersUnderLock() {assert holdsLock() : "PeerFinder mutex not held";
return peersByAddress.values().stream()
.map(Peer::getDiscoveryNode).filter(Objects::nonNull).distinct().collect(Collectors.toList());
}
private Peer createConnectingPeer(TransportAddress transportAddress) {Peer peer = new Peer(transportAddress);
peer.establishConnection();
return peer;
}
/**
* @return whether any peers were removed due to disconnection
*/
private boolean handleWakeUp() {assert holdsLock() : "PeerFinder mutex not held";
final boolean peersRemoved = peersByAddress.values().removeIf(Peer::handleWakeUp);
if (active == false) {logger.trace("not active");
return peersRemoved;
}
logger.trace("probing master nodes from cluster state: {}", lastAcceptedNodes);
for (ObjectCursor<DiscoveryNode> discoveryNodeObjectCursor : lastAcceptedNodes.getMasterNodes().values()) {startProbe(discoveryNodeObjectCursor.value.getAddress());
}
configuredHostsResolver.resolveConfiguredHosts(providedAddresses -> {synchronized (mutex) {
lastResolvedAddresses = providedAddresses;
logger.trace("probing resolved transport addresses {}", providedAddresses);
providedAddresses.forEach(this::startProbe);
}
});
transportService.getThreadPool().scheduleUnlessShuttingDown(findPeersInterval, Names.GENERIC, new AbstractRunnable() {
@Override
public boolean isForceExecution() {return true;}
@Override
public void onFailure(Exception e) {
assert false : e;
logger.debug("unexpected exception in wakeup", e);
}
@Override
protected void doRun() {synchronized (mutex) {if (handleWakeUp() == false) {return;}
}
onFoundPeersUpdated();}
@Override
public String toString() {return "PeerFinder handling wakeup";}
});
return peersRemoved;
}
protected void startProbe(TransportAddress transportAddress) {assert holdsLock() : "PeerFinder mutex not held";
if (active == false) {logger.trace("startProbe({}) not running", transportAddress);
return;
}
if (transportAddress.equals(getLocalNode().getAddress())) {logger.trace("startProbe({}) not probing local node", transportAddress);
return;
}
peersByAddress.computeIfAbsent(transportAddress, this::createConnectingPeer);
}
private class Zen1UnicastPingRequestHandler implements TransportRequestHandler<UnicastZenPing.UnicastPingRequest> {
@Override
public void messageReceived(UnicastZenPing.UnicastPingRequest request, TransportChannel channel, Task task) throws Exception {final PeersRequest peersRequest = new PeersRequest(request.pingResponse.node(),
Optional.ofNullable(request.pingResponse.master()).map(Collections::singletonList).orElse(emptyList()));
final PeersResponse peersResponse = handlePeersRequest(peersRequest);
final List<ZenPing.PingResponse> pingResponses = new ArrayList<>();
final ClusterName clusterName = ClusterName.CLUSTER_NAME_SETTING.get(settings);
pingResponses.add(new ZenPing.PingResponse(createDiscoveryNodeWithImpossiblyHighId(transportService.getLocalNode()),
peersResponse.getMasterNode().orElse(null),
clusterName, ClusterState.UNKNOWN_VERSION));
peersResponse.getKnownPeers().forEach(dn -> pingResponses.add(
new ZenPing.PingResponse(ZenPing.PingResponse.FAKE_PING_ID,
isZen1Node(dn) ? dn : createDiscoveryNodeWithImpossiblyHighId(dn), null, clusterName, ClusterState.UNKNOWN_VERSION)));
channel.sendResponse(new UnicastZenPing.UnicastPingResponse(request.id, pingResponses.toArray(new ZenPing.PingResponse[0])));
}
}
//......
}- PeerFinder 的构造器注册了两个 handler，一个是针对 REQUEST_PEERS_ACTION_NAME，执行 handlePeersRequest；一个是针对 UnicastZenPing.ACTION_NAME，其 handler 为 Zen1UnicastPingRequestHandler，该 handler 里头也调用了 handlePeersRequest 方法
- handlePeersRequest 主要是针对 masterNode 及 peersRequest.getKnownPeers() 挨个执行 startProbe，然后通过 getFoundPeersUnderLock 设置 knownPeers，最后通过 leader、knownPeers 及 currentTerm 构造 PeersResponse 返回
- startProbe 方法主要是执行 createConnectingPeer，并将结果放入 key 为 transportAddress 的名为 peersByAddress 的 map 中；createConnectingPeer 方法主要是创建 Peer，然后调用 Peer 的 establishConnection 方法
Peer
elasticsearch-7.0.0/server/src/main/java/org/elasticsearch/discovery/PeerFinder.java
private class Peer {
private final TransportAddress transportAddress;
private SetOnce<DiscoveryNode> discoveryNode = new SetOnce<>();
private volatile boolean peersRequestInFlight;
Peer(TransportAddress transportAddress) {this.transportAddress = transportAddress;}
@Nullable
DiscoveryNode getDiscoveryNode() {return discoveryNode.get();
}
boolean handleWakeUp() {assert holdsLock() : "PeerFinder mutex not held";
if (active == false) {return true;}
final DiscoveryNode discoveryNode = getDiscoveryNode();
// may be null if connection not yet established
if (discoveryNode != null) {if (transportService.nodeConnected(discoveryNode)) {if (peersRequestInFlight == false) {requestPeers();
}
} else {logger.trace("{} no longer connected", this);
return true;
}
}
return false;
}
void establishConnection() {assert holdsLock() : "PeerFinder mutex not held";
assert getDiscoveryNode() == null : "unexpectedly connected to" + getDiscoveryNode();
assert active;
logger.trace("{} attempting connection", this);
transportAddressConnector.connectToRemoteMasterNode(transportAddress, new ActionListener<DiscoveryNode>() {
@Override
public void onResponse(DiscoveryNode remoteNode) {assert remoteNode.isMasterNode() : remoteNode + "is not master-eligible";
assert remoteNode.equals(getLocalNode()) == false : remoteNode + "is the local node";
synchronized (mutex) {if (active == false) {return;}
assert discoveryNode.get() == null : "discoveryNode unexpectedly already set to" + discoveryNode.get();
discoveryNode.set(remoteNode);
requestPeers();}
assert holdsLock() == false : "PeerFinder mutex is held in error";
onFoundPeersUpdated();}
@Override
public void onFailure(Exception e) {logger.debug(() -> new ParameterizedMessage("{} connection failed", Peer.this), e);
synchronized (mutex) {peersByAddress.remove(transportAddress);
}
}
});
}
private void requestPeers() {assert holdsLock() : "PeerFinder mutex not held";
assert peersRequestInFlight == false : "PeersRequest already in flight";
assert active;
final DiscoveryNode discoveryNode = getDiscoveryNode();
assert discoveryNode != null : "cannot request peers without first connecting";
if (discoveryNode.equals(getLocalNode())) {logger.trace("{} not requesting peers from local node", this);
return;
}
logger.trace("{} requesting peers", this);
peersRequestInFlight = true;
final List<DiscoveryNode> knownNodes = getFoundPeersUnderLock();
final TransportResponseHandler<PeersResponse> peersResponseHandler = new TransportResponseHandler<PeersResponse>() {
@Override
public PeersResponse read(StreamInput in) throws IOException {return new PeersResponse(in);
}
@Override
public void handleResponse(PeersResponse response) {logger.trace("{} received {}", Peer.this, response);
synchronized (mutex) {if (active == false) {return;}
peersRequestInFlight = false;
response.getMasterNode().map(DiscoveryNode::getAddress).ifPresent(PeerFinder.this::startProbe);
response.getKnownPeers().stream().map(DiscoveryNode::getAddress).forEach(PeerFinder.this::startProbe);
}
if (response.getMasterNode().equals(Optional.of(discoveryNode))) {
// Must not hold lock here to avoid deadlock
assert holdsLock() == false : "PeerFinder mutex is held in error";
onActiveMasterFound(discoveryNode, response.getTerm());
}
}
@Override
public void handleException(TransportException exp) {
peersRequestInFlight = false;
logger.debug(new ParameterizedMessage("{} peers request failed", Peer.this), exp);
}
@Override
public String executor() {return Names.GENERIC;}
};
final String actionName;
final TransportRequest transportRequest;
final TransportResponseHandler<?> transportResponseHandler;
if (isZen1Node(discoveryNode)) {
actionName = UnicastZenPing.ACTION_NAME;
transportRequest = new UnicastZenPing.UnicastPingRequest(1, ZenDiscovery.PING_TIMEOUT_SETTING.get(settings),
new ZenPing.PingResponse(createDiscoveryNodeWithImpossiblyHighId(getLocalNode()), null,
ClusterName.CLUSTER_NAME_SETTING.get(settings), ClusterState.UNKNOWN_VERSION));
transportResponseHandler = peersResponseHandler.wrap(ucResponse -> {Optional<DiscoveryNode> optionalMasterNode = Arrays.stream(ucResponse.pingResponses)
.filter(pr -> discoveryNode.equals(pr.node()) && discoveryNode.equals(pr.master()))
.map(ZenPing.PingResponse::node)
.findFirst();
List<DiscoveryNode> discoveredNodes = new ArrayList<>();
if (optionalMasterNode.isPresent() == false) {Arrays.stream(ucResponse.pingResponses).map(PingResponse::master).filter(Objects::nonNull)
.forEach(discoveredNodes::add);
Arrays.stream(ucResponse.pingResponses).map(PingResponse::node).forEach(discoveredNodes::add);
}
return new PeersResponse(optionalMasterNode, discoveredNodes, 0L);
}, UnicastZenPing.UnicastPingResponse::new);
} else {
actionName = REQUEST_PEERS_ACTION_NAME;
transportRequest = new PeersRequest(getLocalNode(), knownNodes);
transportResponseHandler = peersResponseHandler;
}
transportService.sendRequest(discoveryNode, actionName,
transportRequest,
TransportRequestOptions.builder().withTimeout(requestPeersTimeout).build(),
transportResponseHandler);
}
@Override
public String toString() {
return "Peer{" +
"transportAddress=" + transportAddress +
", discoveryNode=" + discoveryNode.get() +
", peersRequestInFlight=" + peersRequestInFlight +
'}';
}
}- Peer 的 establishConnection 方法主要是通过 transportAddressConnector.connectToRemoteMasterNode 请求 masterNode，在结果成功返回时执行 requestPeers 及 onFoundPeersUpdated 方法；如果出现异常则从 peersByAddress 移除该 transportAddress 对应的 Peer
- requestPeers 方法通过 getFoundPeersUnderLock 方法获取 knownNodes 构造 PeersRequest，然后通过 transportService.sendRequest 向 discoveryNode 发送请求，请求成功返回则执行 masterNode 的 startProbe 以及 knownPeers 的 startProbe 方法，如果当前 discoveryNode 是 masterNode，则触发 onActiveMasterFound 方法
- 7.0 版本的代码针对之前的版本做了不同的处理，requestPeers 方法发送的 request 为 UnicastZenPing.UnicastPingRequest
小结
- PeerFinder 的构造器注册了两个 handler，一个是针对 REQUEST_PEERS_ACTION_NAME，执行 handlePeersRequest；一个是针对 UnicastZenPing.ACTION_NAME，其 handler 为 Zen1UnicastPingRequestHandler，该 handler 里头也调用了 handlePeersRequest 方法
- handlePeersRequest 主要是针对 masterNode 及 peersRequest.getKnownPeers() 挨个执行 startProbe，然后通过 getFoundPeersUnderLock 设置 knownPeers，最后通过 leader、knownPeers 及 currentTerm 构造 PeersResponse 返回；startProbe 方法主要是执行 createConnectingPeer，并将结果放入 key 为 transportAddress 的名为 peersByAddress 的 map 中；createConnectingPeer 方法主要是创建 Peer，然后调用 Peer 的 establishConnection 方法
- Peer 的 establishConnection 方法主要是通过 transportAddressConnector.connectToRemoteMasterNode 请求 masterNode，在结果成功返回时执行 requestPeers 及 onFoundPeersUpdated 方法；如果出现异常则从 peersByAddress 移除该 transportAddress 对应的 Peer；requestPeers 方法通过 getFoundPeersUnderLock 方法获取 knownNodes 构造 PeersRequest，然后通过 transportService.sendRequest 向 discoveryNode 发送请求，请求成功返回则执行 masterNode 的 startProbe 以及 knownPeers 的 startProbe 方法，如果当前 discoveryNode 是 masterNode，则触发 onActiveMasterFound 方法
doc
- PeerFinder