| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF 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.apache.solr.cloud; |
| |
| import org.apache.http.NoHttpResponseException; |
| import org.apache.http.conn.ConnectTimeoutException; |
| import org.apache.solr.client.solrj.impl.HttpSolrClient; |
| import org.apache.solr.client.solrj.request.CoreAdminRequest.RequestRecovery; |
| import org.apache.solr.common.SolrException; |
| import org.apache.solr.common.SolrException.ErrorCode; |
| import org.apache.solr.common.cloud.Replica; |
| import org.apache.solr.common.cloud.ZkCoreNodeProps; |
| import org.apache.solr.common.cloud.ZkNodeProps; |
| import org.apache.solr.common.cloud.ZkStateReader; |
| import org.apache.solr.common.params.CoreAdminParams.CoreAdminAction; |
| import org.apache.solr.common.util.Utils; |
| import org.apache.solr.core.CoreContainer; |
| import org.apache.solr.core.CoreDescriptor; |
| import org.apache.zookeeper.KeeperException; |
| import org.apache.solr.util.RTimer; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.lang.invoke.MethodHandles; |
| import java.net.ConnectException; |
| import java.net.SocketException; |
| import java.net.UnknownHostException; |
| import java.util.List; |
| |
| /** |
| * Background daemon thread that tries to send the REQUESTRECOVERY to a downed |
| * replica; used by a shard leader to nag a replica into recovering after the |
| * leader experiences an error trying to send an update request to the replica. |
| */ |
| public class LeaderInitiatedRecoveryThread extends Thread { |
| |
| private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); |
| |
| protected ZkController zkController; |
| protected CoreContainer coreContainer; |
| protected String collection; |
| protected String shardId; |
| protected ZkCoreNodeProps nodeProps; |
| protected int maxTries; |
| private CoreDescriptor leaderCd; |
| |
| public LeaderInitiatedRecoveryThread(ZkController zkController, |
| CoreContainer cc, |
| String collection, |
| String shardId, |
| ZkCoreNodeProps nodeProps, |
| int maxTries, |
| CoreDescriptor leaderCd) |
| { |
| super("LeaderInitiatedRecoveryThread-"+nodeProps.getCoreName()); |
| this.zkController = zkController; |
| this.coreContainer = cc; |
| this.collection = collection; |
| this.shardId = shardId; |
| this.nodeProps = nodeProps; |
| this.maxTries = maxTries; |
| this.leaderCd = leaderCd; |
| setDaemon(true); |
| } |
| |
| public void run() { |
| RTimer timer = new RTimer(); |
| |
| String replicaCoreName = nodeProps.getCoreName(); |
| String replicaCoreNodeName = ((Replica) nodeProps.getNodeProps()).getName(); |
| String replicaNodeName = nodeProps.getNodeName(); |
| final String replicaUrl = nodeProps.getCoreUrl(); |
| |
| if (!zkController.isReplicaInRecoveryHandling(replicaUrl)) { |
| throw new SolrException(ErrorCode.INVALID_STATE, "Replica: " + replicaUrl + " should have been marked under leader initiated recovery in ZkController but wasn't."); |
| } |
| |
| boolean sendRecoveryCommand = publishDownState(replicaCoreName, replicaCoreNodeName, replicaNodeName, replicaUrl, false); |
| |
| if (sendRecoveryCommand) { |
| try { |
| sendRecoveryCommandWithRetry(); |
| } catch (Exception exc) { |
| log.error(getName()+" failed due to: "+exc, exc); |
| if (exc instanceof SolrException) { |
| throw (SolrException)exc; |
| } else { |
| throw new SolrException(ErrorCode.SERVER_ERROR, exc); |
| } |
| } finally { |
| zkController.removeReplicaFromLeaderInitiatedRecoveryHandling(replicaUrl); |
| } |
| } else { |
| // replica is no longer in recovery on this node (may be handled on another node) |
| zkController.removeReplicaFromLeaderInitiatedRecoveryHandling(replicaUrl); |
| } |
| log.info("{} completed successfully after running for {}ms", getName(), timer.getTime()); |
| } |
| |
| public boolean publishDownState(String replicaCoreName, String replicaCoreNodeName, String replicaNodeName, String replicaUrl, boolean forcePublishState) { |
| boolean sendRecoveryCommand = true; |
| boolean publishDownState = false; |
| |
| if (zkController.getZkStateReader().getClusterState().liveNodesContain(replicaNodeName)) { |
| try { |
| // create a znode that requires the replica needs to "ack" to verify it knows it was out-of-sync |
| updateLIRState(replicaCoreNodeName); |
| |
| log.info("Put replica core={} coreNodeName={} on " + |
| replicaNodeName + " into leader-initiated recovery.", replicaCoreName, replicaCoreNodeName); |
| publishDownState = true; |
| } catch (Exception e) { |
| Throwable setLirZnodeFailedCause = SolrException.getRootCause(e); |
| log.error("Leader failed to set replica " + |
| nodeProps.getCoreUrl() + " state to DOWN due to: " + setLirZnodeFailedCause, setLirZnodeFailedCause); |
| if (setLirZnodeFailedCause instanceof KeeperException.SessionExpiredException |
| || setLirZnodeFailedCause instanceof KeeperException.ConnectionLossException |
| || setLirZnodeFailedCause instanceof ZkController.NotLeaderException) { |
| // our session is expired, which means our state is suspect, so don't go |
| // putting other replicas in recovery (see SOLR-6511) |
| sendRecoveryCommand = false; |
| forcePublishState = false; // no need to force publish any state in this case |
| } // else will go ahead and try to send the recovery command once after this error |
| } |
| } else { |
| log.info("Node " + replicaNodeName + |
| " is not live, so skipping leader-initiated recovery for replica: core={} coreNodeName={}", |
| replicaCoreName, replicaCoreNodeName); |
| // publishDownState will be false to avoid publishing the "down" state too many times |
| // as many errors can occur together and will each call into this method (SOLR-6189) |
| forcePublishState = false; // no need to force publish the state because replica is not live |
| sendRecoveryCommand = false; // no need to send recovery messages as well |
| } |
| |
| try { |
| if (publishDownState || forcePublishState) { |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, "state", |
| ZkStateReader.STATE_PROP, Replica.State.DOWN.toString(), |
| ZkStateReader.BASE_URL_PROP, nodeProps.getBaseUrl(), |
| ZkStateReader.CORE_NAME_PROP, nodeProps.getCoreName(), |
| ZkStateReader.NODE_NAME_PROP, nodeProps.getNodeName(), |
| ZkStateReader.SHARD_ID_PROP, shardId, |
| ZkStateReader.COLLECTION_PROP, collection); |
| log.warn("Leader is publishing core={} coreNodeName ={} state={} on behalf of un-reachable replica {}", |
| replicaCoreName, replicaCoreNodeName, Replica.State.DOWN.toString(), replicaUrl); |
| zkController.getOverseerJobQueue().offer(Utils.toJSON(m)); |
| } |
| } catch (Exception e) { |
| log.error("Could not publish 'down' state for replicaUrl: {}", replicaUrl, e); |
| } |
| |
| return sendRecoveryCommand; |
| } |
| |
| /* |
| protected scope for testing purposes |
| */ |
| protected void updateLIRState(String replicaCoreNodeName) { |
| zkController.updateLeaderInitiatedRecoveryState(collection, |
| shardId, |
| replicaCoreNodeName, Replica.State.DOWN, leaderCd, true); |
| } |
| |
| protected void sendRecoveryCommandWithRetry() throws Exception { |
| int tries = 0; |
| long waitBetweenTriesMs = 5000L; |
| boolean continueTrying = true; |
| |
| String replicaCoreName = nodeProps.getCoreName(); |
| String recoveryUrl = nodeProps.getBaseUrl(); |
| String replicaNodeName = nodeProps.getNodeName(); |
| String coreNeedingRecovery = nodeProps.getCoreName(); |
| String replicaCoreNodeName = ((Replica) nodeProps.getNodeProps()).getName(); |
| String replicaUrl = nodeProps.getCoreUrl(); |
| |
| log.info(getName()+" started running to send REQUESTRECOVERY command to "+replicaUrl+ |
| "; will try for a max of "+(maxTries * (waitBetweenTriesMs/1000))+" secs"); |
| |
| RequestRecovery recoverRequestCmd = new RequestRecovery(); |
| recoverRequestCmd.setAction(CoreAdminAction.REQUESTRECOVERY); |
| recoverRequestCmd.setCoreName(coreNeedingRecovery); |
| |
| while (continueTrying && ++tries <= maxTries) { |
| if (tries > 1) { |
| log.warn("Asking core={} coreNodeName={} on " + recoveryUrl + |
| " to recover; unsuccessful after "+tries+" of "+maxTries+" attempts so far ...", coreNeedingRecovery, replicaCoreNodeName); |
| } else { |
| log.info("Asking core={} coreNodeName={} on " + recoveryUrl + " to recover", coreNeedingRecovery, replicaCoreNodeName); |
| } |
| |
| try (HttpSolrClient client = new HttpSolrClient.Builder(recoveryUrl).build()) { |
| client.setSoTimeout(60000); |
| client.setConnectionTimeout(15000); |
| try { |
| client.request(recoverRequestCmd); |
| |
| log.info("Successfully sent " + CoreAdminAction.REQUESTRECOVERY + |
| " command to core={} coreNodeName={} on " + recoveryUrl, coreNeedingRecovery, replicaCoreNodeName); |
| |
| continueTrying = false; // succeeded, so stop looping |
| } catch (Exception t) { |
| Throwable rootCause = SolrException.getRootCause(t); |
| boolean wasCommError = |
| (rootCause instanceof ConnectException || |
| rootCause instanceof ConnectTimeoutException || |
| rootCause instanceof NoHttpResponseException || |
| rootCause instanceof SocketException || |
| rootCause instanceof UnknownHostException); |
| |
| SolrException.log(log, recoveryUrl + ": Could not tell a replica to recover", t); |
| |
| if (!wasCommError) { |
| continueTrying = false; |
| } |
| } |
| } |
| |
| // wait a few seconds |
| if (continueTrying) { |
| try { |
| Thread.sleep(waitBetweenTriesMs); |
| } catch (InterruptedException ignoreMe) { |
| Thread.currentThread().interrupt(); |
| } |
| |
| if (coreContainer.isShutDown()) { |
| log.warn("Stop trying to send recovery command to downed replica core={} coreNodeName={} on " |
| + replicaNodeName + " because my core container is closed.", coreNeedingRecovery, replicaCoreNodeName); |
| continueTrying = false; |
| break; |
| } |
| |
| // see if the replica's node is still live, if not, no need to keep doing this loop |
| ZkStateReader zkStateReader = zkController.getZkStateReader(); |
| if (!zkStateReader.getClusterState().liveNodesContain(replicaNodeName)) { |
| log.warn("Node "+replicaNodeName+" hosting core "+coreNeedingRecovery+ |
| " is no longer live. No need to keep trying to tell it to recover!"); |
| continueTrying = false; |
| break; |
| } |
| |
| String leaderCoreNodeName = leaderCd.getCloudDescriptor().getCoreNodeName(); |
| // stop trying if I'm no longer the leader |
| if (leaderCoreNodeName != null && collection != null) { |
| String leaderCoreNodeNameFromZk = null; |
| try { |
| leaderCoreNodeNameFromZk = zkController.getZkStateReader().getLeaderRetry(collection, shardId, 1000).getName(); |
| } catch (Exception exc) { |
| log.error("Failed to determine if " + leaderCoreNodeName + " is still the leader for " + collection + |
| " " + shardId + " before starting leader-initiated recovery thread for " + replicaUrl + " due to: " + exc); |
| } |
| if (!leaderCoreNodeName.equals(leaderCoreNodeNameFromZk)) { |
| log.warn("Stop trying to send recovery command to downed replica core=" + coreNeedingRecovery + |
| ",coreNodeName=" + replicaCoreNodeName + " on " + replicaNodeName + " because " + |
| leaderCoreNodeName + " is no longer the leader! New leader is " + leaderCoreNodeNameFromZk); |
| continueTrying = false; |
| break; |
| } |
| if (!leaderCd.getCloudDescriptor().isLeader()) { |
| log.warn("Stop trying to send recovery command to downed replica core=" + coreNeedingRecovery + |
| ",coreNodeName=" + replicaCoreNodeName + " on " + replicaNodeName + " because " + |
| leaderCoreNodeName + " is no longer the leader!"); |
| continueTrying = false; |
| break; |
| } |
| } |
| |
| // additional safeguard against the replica trying to be in the active state |
| // before acknowledging the leader initiated recovery command |
| if (collection != null && shardId != null) { |
| try { |
| // call out to ZooKeeper to get the leader-initiated recovery state |
| final Replica.State lirState = zkController.getLeaderInitiatedRecoveryState(collection, shardId, replicaCoreNodeName); |
| |
| if (lirState == null) { |
| log.warn("Stop trying to send recovery command to downed replica core="+coreNeedingRecovery+ |
| ",coreNodeName=" + replicaCoreNodeName + " on "+replicaNodeName+" because the znode no longer exists."); |
| continueTrying = false; |
| break; |
| } |
| |
| if (lirState == Replica.State.RECOVERING) { |
| // replica has ack'd leader initiated recovery and entered the recovering state |
| // so we don't need to keep looping to send the command |
| continueTrying = false; |
| log.info("Replica "+coreNeedingRecovery+ |
| " on node "+replicaNodeName+" ack'd the leader initiated recovery state, " |
| + "no need to keep trying to send recovery command"); |
| } else { |
| String lcnn = zkStateReader.getLeaderRetry(collection, shardId, 5000).getName(); |
| List<ZkCoreNodeProps> replicaProps = |
| zkStateReader.getReplicaProps(collection, shardId, lcnn); |
| if (replicaProps != null && replicaProps.size() > 0) { |
| for (ZkCoreNodeProps prop : replicaProps) { |
| final Replica replica = (Replica) prop.getNodeProps(); |
| if (replicaCoreNodeName.equals(replica.getName())) { |
| if (replica.getState() == Replica.State.ACTIVE) { |
| // replica published its state as "active", |
| // which is bad if lirState is still "down" |
| if (lirState == Replica.State.DOWN) { |
| // OK, so the replica thinks it is active, but it never ack'd the leader initiated recovery |
| // so its state cannot be trusted and it needs to be told to recover again ... and we keep looping here |
| log.warn("Replica core={} coreNodeName={} set to active but the leader thinks it should be in recovery;" |
| + " forcing it back to down state to re-run the leader-initiated recovery process; props: " + replicaProps.get(0), coreNeedingRecovery, replicaCoreNodeName); |
| publishDownState(replicaCoreName, replicaCoreNodeName, replicaNodeName, replicaUrl, true); |
| } |
| } |
| break; |
| } |
| } |
| } |
| } |
| } catch (Exception ignoreMe) { |
| log.warn("Failed to determine state of core={} coreNodeName={} due to: "+ignoreMe, coreNeedingRecovery, replicaCoreNodeName); |
| // eventually this loop will exhaust max tries and stop so we can just log this for now |
| } |
| } |
| } |
| } |
| |
| // replica is no longer in recovery on this node (may be handled on another node) |
| zkController.removeReplicaFromLeaderInitiatedRecoveryHandling(replicaUrl); |
| |
| if (continueTrying) { |
| // ugh! this means the loop timed out before the recovery command could be delivered |
| // how exotic do we want to get here? |
| log.error("Timed out after waiting for "+(tries * (waitBetweenTriesMs/1000))+ |
| " secs to send the recovery request to: "+replicaUrl+"; not much more we can do here?"); |
| |
| // TODO: need to raise a JMX event to allow monitoring tools to take over from here |
| |
| } |
| } |
| } |
