| /** |
| * 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.hadoop.hdds.scm.node; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Preconditions; |
| import com.google.common.util.concurrent.ThreadFactoryBuilder; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.hdds.protocol.DatanodeDetails; |
| import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState; |
| import org.apache.hadoop.hdds.scm.HddsServerUtil; |
| import org.apache.hadoop.hdds.scm.container.ContainerID; |
| import org.apache.hadoop.hdds.scm.pipeline.Pipeline; |
| import org.apache.hadoop.hdds.scm.pipeline.PipelineID; |
| import org.apache.hadoop.hdds.scm.events.SCMEvents; |
| import org.apache.hadoop.hdds.scm.node.states.*; |
| import org.apache.hadoop.hdds.scm.node.states.Node2PipelineMap; |
| import org.apache.hadoop.hdds.server.events.Event; |
| import org.apache.hadoop.hdds.server.events.EventPublisher; |
| import org.apache.hadoop.ozone.common.statemachine |
| .InvalidStateTransitionException; |
| import org.apache.hadoop.ozone.common.statemachine.StateMachine; |
| import org.apache.hadoop.util.Time; |
| import org.apache.hadoop.util.concurrent.HadoopExecutors; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.Closeable; |
| import java.util.*; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.ScheduledFuture; |
| import java.util.concurrent.TimeUnit; |
| import java.util.function.Predicate; |
| |
| import static org.apache.hadoop.hdds.scm.ScmConfigKeys |
| .OZONE_SCM_DEADNODE_INTERVAL; |
| import static org.apache.hadoop.hdds.scm.ScmConfigKeys |
| .OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL; |
| import static org.apache.hadoop.hdds.scm.ScmConfigKeys |
| .OZONE_SCM_STALENODE_INTERVAL; |
| |
| /** |
| * NodeStateManager maintains the state of all the datanodes in the cluster. All |
| * the node state change should happen only via NodeStateManager. It also |
| * runs a heartbeat thread which periodically updates the node state. |
| * <p> |
| * The getNode(byState) functions make copy of node maps and then creates a list |
| * based on that. It should be assumed that these get functions always report |
| * *stale* information. For example, getting the deadNodeCount followed by |
| * getNodes(DEAD) could very well produce totally different count. Also |
| * getNodeCount(HEALTHY) + getNodeCount(DEAD) + getNodeCode(STALE), is not |
| * guaranteed to add up to the total nodes that we know off. Please treat all |
| * get functions in this file as a snap-shot of information that is inconsistent |
| * as soon as you read it. |
| */ |
| public class NodeStateManager implements Runnable, Closeable { |
| |
| /** |
| * Node's life cycle events. |
| */ |
| private enum NodeLifeCycleEvent { |
| TIMEOUT, RESTORE, RESURRECT, DECOMMISSION, DECOMMISSIONED |
| } |
| |
| private static final Logger LOG = LoggerFactory |
| .getLogger(NodeStateManager.class); |
| |
| /** |
| * StateMachine for node lifecycle. |
| */ |
| private final StateMachine<NodeState, NodeLifeCycleEvent> stateMachine; |
| /** |
| * This is the map which maintains the current state of all datanodes. |
| */ |
| private final NodeStateMap nodeStateMap; |
| /** |
| * Maintains the mapping from node to pipelines a node is part of. |
| */ |
| private final Node2PipelineMap node2PipelineMap; |
| /** |
| * Used for publishing node state change events. |
| */ |
| private final EventPublisher eventPublisher; |
| /** |
| * Maps the event to be triggered when a node state us updated. |
| */ |
| private final Map<NodeState, Event<DatanodeDetails>> state2EventMap; |
| /** |
| * ExecutorService used for scheduling heartbeat processing thread. |
| */ |
| private final ScheduledExecutorService executorService; |
| /** |
| * The frequency in which we have run the heartbeat processing thread. |
| */ |
| private final long heartbeatCheckerIntervalMs; |
| /** |
| * The timeout value which will be used for marking a datanode as stale. |
| */ |
| private final long staleNodeIntervalMs; |
| /** |
| * The timeout value which will be used for marking a datanode as dead. |
| */ |
| private final long deadNodeIntervalMs; |
| |
| /** |
| * The future is used to pause/unpause the scheduled checks. |
| */ |
| private ScheduledFuture<?> healthCheckFuture; |
| |
| /** |
| * Test utility - tracks if health check has been paused (unit tests). |
| */ |
| private boolean checkPaused; |
| |
| /** |
| * timestamp of the latest heartbeat check process. |
| */ |
| private long lastHealthCheck; |
| |
| /** |
| * number of times the heart beat check was skipped. |
| */ |
| private long skippedHealthChecks; |
| |
| /** |
| * Constructs a NodeStateManager instance with the given configuration. |
| * |
| * @param conf Configuration |
| */ |
| public NodeStateManager(Configuration conf, EventPublisher eventPublisher) { |
| this.nodeStateMap = new NodeStateMap(); |
| this.node2PipelineMap = new Node2PipelineMap(); |
| this.eventPublisher = eventPublisher; |
| this.state2EventMap = new HashMap<>(); |
| initialiseState2EventMap(); |
| Set<NodeState> finalStates = new HashSet<>(); |
| finalStates.add(NodeState.DECOMMISSIONED); |
| this.stateMachine = new StateMachine<>(NodeState.HEALTHY, finalStates); |
| initializeStateMachine(); |
| heartbeatCheckerIntervalMs = HddsServerUtil |
| .getScmheartbeatCheckerInterval(conf); |
| staleNodeIntervalMs = HddsServerUtil.getStaleNodeInterval(conf); |
| deadNodeIntervalMs = HddsServerUtil.getDeadNodeInterval(conf); |
| Preconditions.checkState(heartbeatCheckerIntervalMs > 0, |
| OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL + " should be greater than 0."); |
| Preconditions.checkState(staleNodeIntervalMs < deadNodeIntervalMs, |
| OZONE_SCM_STALENODE_INTERVAL + " should be less than" + |
| OZONE_SCM_DEADNODE_INTERVAL); |
| executorService = HadoopExecutors.newScheduledThreadPool(1, |
| new ThreadFactoryBuilder().setDaemon(true) |
| .setNameFormat("SCM Heartbeat Processing Thread - %d").build()); |
| |
| skippedHealthChecks = 0; |
| checkPaused = false; // accessed only from test functions |
| |
| scheduleNextHealthCheck(); |
| } |
| |
| /** |
| * Populates state2event map. |
| */ |
| private void initialiseState2EventMap() { |
| state2EventMap.put(NodeState.STALE, SCMEvents.STALE_NODE); |
| state2EventMap.put(NodeState.DEAD, SCMEvents.DEAD_NODE); |
| state2EventMap |
| .put(NodeState.HEALTHY, SCMEvents.NON_HEALTHY_TO_HEALTHY_NODE); |
| } |
| |
| /* |
| * |
| * Node and State Transition Mapping: |
| * |
| * State: HEALTHY -------------------> STALE |
| * Event: TIMEOUT |
| * |
| * State: STALE -------------------> DEAD |
| * Event: TIMEOUT |
| * |
| * State: STALE -------------------> HEALTHY |
| * Event: RESTORE |
| * |
| * State: DEAD -------------------> HEALTHY |
| * Event: RESURRECT |
| * |
| * State: HEALTHY -------------------> DECOMMISSIONING |
| * Event: DECOMMISSION |
| * |
| * State: STALE -------------------> DECOMMISSIONING |
| * Event: DECOMMISSION |
| * |
| * State: DEAD -------------------> DECOMMISSIONING |
| * Event: DECOMMISSION |
| * |
| * State: DECOMMISSIONING -------------------> DECOMMISSIONED |
| * Event: DECOMMISSIONED |
| * |
| * Node State Flow |
| * |
| * +--------------------------------------------------------+ |
| * | (RESURRECT) | |
| * | +--------------------------+ | |
| * | | (RESTORE) | | |
| * | | | | |
| * V V | | |
| * [HEALTHY]------------------->[STALE]------------------->[DEAD] |
| * | (TIMEOUT) | (TIMEOUT) | |
| * | | | |
| * | | | |
| * | | | |
| * | | | |
| * | (DECOMMISSION) | (DECOMMISSION) | (DECOMMISSION) |
| * | V | |
| * +------------------->[DECOMMISSIONING]<----------------+ |
| * | |
| * | (DECOMMISSIONED) |
| * | |
| * V |
| * [DECOMMISSIONED] |
| * |
| */ |
| |
| /** |
| * Initializes the lifecycle of node state machine. |
| */ |
| private void initializeStateMachine() { |
| stateMachine.addTransition( |
| NodeState.HEALTHY, NodeState.STALE, NodeLifeCycleEvent.TIMEOUT); |
| stateMachine.addTransition( |
| NodeState.STALE, NodeState.DEAD, NodeLifeCycleEvent.TIMEOUT); |
| stateMachine.addTransition( |
| NodeState.STALE, NodeState.HEALTHY, NodeLifeCycleEvent.RESTORE); |
| stateMachine.addTransition( |
| NodeState.DEAD, NodeState.HEALTHY, NodeLifeCycleEvent.RESURRECT); |
| stateMachine.addTransition( |
| NodeState.HEALTHY, NodeState.DECOMMISSIONING, |
| NodeLifeCycleEvent.DECOMMISSION); |
| stateMachine.addTransition( |
| NodeState.STALE, NodeState.DECOMMISSIONING, |
| NodeLifeCycleEvent.DECOMMISSION); |
| stateMachine.addTransition( |
| NodeState.DEAD, NodeState.DECOMMISSIONING, |
| NodeLifeCycleEvent.DECOMMISSION); |
| stateMachine.addTransition( |
| NodeState.DECOMMISSIONING, NodeState.DECOMMISSIONED, |
| NodeLifeCycleEvent.DECOMMISSIONED); |
| |
| } |
| |
| /** |
| * Adds a new node to the state manager. |
| * |
| * @param datanodeDetails DatanodeDetails |
| * |
| * @throws NodeAlreadyExistsException if the node is already present |
| */ |
| public void addNode(DatanodeDetails datanodeDetails) |
| throws NodeAlreadyExistsException { |
| nodeStateMap.addNode(datanodeDetails, stateMachine.getInitialState()); |
| eventPublisher.fireEvent(SCMEvents.NEW_NODE, datanodeDetails); |
| } |
| |
| /** |
| * Adds a pipeline in the node2PipelineMap. |
| * @param pipeline - Pipeline to be added |
| */ |
| public void addPipeline(Pipeline pipeline) { |
| node2PipelineMap.addPipeline(pipeline); |
| } |
| |
| /** |
| * Get information about the node. |
| * |
| * @param datanodeDetails DatanodeDetails |
| * |
| * @return DatanodeInfo |
| * |
| * @throws NodeNotFoundException if the node is not present |
| */ |
| public DatanodeInfo getNode(DatanodeDetails datanodeDetails) |
| throws NodeNotFoundException { |
| return nodeStateMap.getNodeInfo(datanodeDetails.getUuid()); |
| } |
| |
| /** |
| * Updates the last heartbeat time of the node. |
| * |
| * @throws NodeNotFoundException if the node is not present |
| */ |
| public void updateLastHeartbeatTime(DatanodeDetails datanodeDetails) |
| throws NodeNotFoundException { |
| nodeStateMap.getNodeInfo(datanodeDetails.getUuid()) |
| .updateLastHeartbeatTime(); |
| } |
| |
| /** |
| * Returns the current state of the node. |
| * |
| * @param datanodeDetails DatanodeDetails |
| * |
| * @return NodeState |
| * |
| * @throws NodeNotFoundException if the node is not present |
| */ |
| public NodeState getNodeState(DatanodeDetails datanodeDetails) |
| throws NodeNotFoundException { |
| return nodeStateMap.getNodeState(datanodeDetails.getUuid()); |
| } |
| |
| /** |
| * Returns all the node which are in healthy state. |
| * |
| * @return list of healthy nodes |
| */ |
| public List<DatanodeInfo> getHealthyNodes() { |
| return getNodes(NodeState.HEALTHY); |
| } |
| |
| /** |
| * Returns all the node which are in stale state. |
| * |
| * @return list of stale nodes |
| */ |
| public List<DatanodeInfo> getStaleNodes() { |
| return getNodes(NodeState.STALE); |
| } |
| |
| /** |
| * Returns all the node which are in dead state. |
| * |
| * @return list of dead nodes |
| */ |
| public List<DatanodeInfo> getDeadNodes() { |
| return getNodes(NodeState.DEAD); |
| } |
| |
| /** |
| * Returns all the node which are in the specified state. |
| * |
| * @param state NodeState |
| * |
| * @return list of nodes |
| */ |
| public List<DatanodeInfo> getNodes(NodeState state) { |
| List<DatanodeInfo> nodes = new ArrayList<>(); |
| nodeStateMap.getNodes(state).forEach( |
| uuid -> { |
| try { |
| nodes.add(nodeStateMap.getNodeInfo(uuid)); |
| } catch (NodeNotFoundException e) { |
| // This should not happen unless someone else other than |
| // NodeStateManager is directly modifying NodeStateMap and removed |
| // the node entry after we got the list of UUIDs. |
| LOG.error("Inconsistent NodeStateMap! " + nodeStateMap); |
| } |
| }); |
| return nodes; |
| } |
| |
| /** |
| * Returns all the nodes which have registered to NodeStateManager. |
| * |
| * @return all the managed nodes |
| */ |
| public List<DatanodeInfo> getAllNodes() { |
| List<DatanodeInfo> nodes = new ArrayList<>(); |
| nodeStateMap.getAllNodes().forEach( |
| uuid -> { |
| try { |
| nodes.add(nodeStateMap.getNodeInfo(uuid)); |
| } catch (NodeNotFoundException e) { |
| // This should not happen unless someone else other than |
| // NodeStateManager is directly modifying NodeStateMap and removed |
| // the node entry after we got the list of UUIDs. |
| LOG.error("Inconsistent NodeStateMap! " + nodeStateMap); |
| } |
| }); |
| return nodes; |
| } |
| |
| /** |
| * Gets set of pipelineID a datanode belongs to. |
| * @param dnId - Datanode ID |
| * @return Set of PipelineID |
| */ |
| public Set<PipelineID> getPipelineByDnID(UUID dnId) { |
| return node2PipelineMap.getPipelines(dnId); |
| } |
| |
| /** |
| * Returns the count of healthy nodes. |
| * |
| * @return healthy node count |
| */ |
| public int getHealthyNodeCount() { |
| return getNodeCount(NodeState.HEALTHY); |
| } |
| |
| /** |
| * Returns the count of stale nodes. |
| * |
| * @return stale node count |
| */ |
| public int getStaleNodeCount() { |
| return getNodeCount(NodeState.STALE); |
| } |
| |
| /** |
| * Returns the count of dead nodes. |
| * |
| * @return dead node count |
| */ |
| public int getDeadNodeCount() { |
| return getNodeCount(NodeState.DEAD); |
| } |
| |
| /** |
| * Returns the count of nodes in specified state. |
| * |
| * @param state NodeState |
| * |
| * @return node count |
| */ |
| public int getNodeCount(NodeState state) { |
| return nodeStateMap.getNodeCount(state); |
| } |
| |
| /** |
| * Returns the count of all nodes managed by NodeStateManager. |
| * |
| * @return node count |
| */ |
| public int getTotalNodeCount() { |
| return nodeStateMap.getTotalNodeCount(); |
| } |
| |
| /** |
| * Removes a pipeline from the node2PipelineMap. |
| * @param pipeline - Pipeline to be removed |
| */ |
| public void removePipeline(Pipeline pipeline) { |
| node2PipelineMap.removePipeline(pipeline); |
| } |
| |
| /** |
| * Adds the given container to the specified datanode. |
| * |
| * @param uuid - datanode uuid |
| * @param containerId - containerID |
| * @throws NodeNotFoundException - if datanode is not known. For new datanode |
| * use addDatanodeInContainerMap call. |
| */ |
| public void addContainer(final UUID uuid, |
| final ContainerID containerId) |
| throws NodeNotFoundException { |
| nodeStateMap.addContainer(uuid, containerId); |
| } |
| |
| /** |
| * Update set of containers available on a datanode. |
| * @param uuid - DatanodeID |
| * @param containerIds - Set of containerIDs |
| * @throws NodeNotFoundException - if datanode is not known. |
| */ |
| public void setContainers(UUID uuid, Set<ContainerID> containerIds) |
| throws NodeNotFoundException { |
| nodeStateMap.setContainers(uuid, containerIds); |
| } |
| |
| /** |
| * Return set of containerIDs available on a datanode. |
| * @param uuid - DatanodeID |
| * @return - set of containerIDs |
| */ |
| public Set<ContainerID> getContainers(UUID uuid) |
| throws NodeNotFoundException { |
| return nodeStateMap.getContainers(uuid); |
| } |
| |
| /** |
| * Move Stale or Dead node to healthy if we got a heartbeat from them. |
| * Move healthy nodes to stale nodes if it is needed. |
| * Move Stales node to dead if needed. |
| * |
| * @see Thread#run() |
| */ |
| @Override |
| public void run() { |
| |
| if (shouldSkipCheck()) { |
| skippedHealthChecks++; |
| LOG.info("Detected long delay in scheduling HB processing thread. " |
| + "Skipping heartbeat checks for one iteration."); |
| } else { |
| checkNodesHealth(); |
| } |
| |
| // we purposefully make this non-deterministic. Instead of using a |
| // scheduleAtFixedFrequency we will just go to sleep |
| // and wake up at the next rendezvous point, which is currentTime + |
| // heartbeatCheckerIntervalMs. This leads to the issue that we are now |
| // heart beating not at a fixed cadence, but clock tick + time taken to |
| // work. |
| // |
| // This time taken to work can skew the heartbeat processor thread. |
| // The reason why we don't care is because of the following reasons. |
| // |
| // 1. checkerInterval is general many magnitudes faster than datanode HB |
| // frequency. |
| // |
| // 2. if we have too much nodes, the SCM would be doing only HB |
| // processing, this could lead to SCM's CPU starvation. With this |
| // approach we always guarantee that HB thread sleeps for a little while. |
| // |
| // 3. It is possible that we will never finish processing the HB's in the |
| // thread. But that means we have a mis-configured system. We will warn |
| // the users by logging that information. |
| // |
| // 4. And the most important reason, heartbeats are not blocked even if |
| // this thread does not run, they will go into the processing queue. |
| scheduleNextHealthCheck(); |
| } |
| |
| private void checkNodesHealth() { |
| |
| /* |
| * |
| * staleNodeDeadline healthyNodeDeadline |
| * | | |
| * Dead | Stale | Healthy |
| * Node | Node | Node |
| * Window | Window | Window |
| * ----------------+----------------------------------+-------------------> |
| * >>-->> time-line >>-->> |
| * |
| * Here is the logic of computing the health of a node. |
|      * |
|      * 1. We get the current time and look back that the time |
|      *  when we got a heartbeat from a node. |
|      *  |
|      * 2. If the last heartbeat was within the window of healthy node we mark |
|      *  it as healthy. |
|      *  |
|      * 3. If the last HB Time stamp is longer and falls within the window of |
|      *  Stale Node time, we will mark it as Stale. |
|      *  |
|      * 4. If the last HB time is older than the Stale Window, then the node is |
|      * marked as dead. |
| * |
| * The Processing starts from current time and looks backwards in time. |
| */ |
| long processingStartTime = Time.monotonicNow(); |
| // After this time node is considered to be stale. |
| long healthyNodeDeadline = processingStartTime - staleNodeIntervalMs; |
| // After this time node is considered to be dead. |
| long staleNodeDeadline = processingStartTime - deadNodeIntervalMs; |
| |
| Predicate<Long> healthyNodeCondition = |
| (lastHbTime) -> lastHbTime >= healthyNodeDeadline; |
| // staleNodeCondition is superset of stale and dead node |
| Predicate<Long> staleNodeCondition = |
| (lastHbTime) -> lastHbTime < healthyNodeDeadline; |
| Predicate<Long> deadNodeCondition = |
| (lastHbTime) -> lastHbTime < staleNodeDeadline; |
| try { |
| for (NodeState state : NodeState.values()) { |
| List<UUID> nodes = nodeStateMap.getNodes(state); |
| for (UUID id : nodes) { |
| DatanodeInfo node = nodeStateMap.getNodeInfo(id); |
| switch (state) { |
| case HEALTHY: |
| // Move the node to STALE if the last heartbeat time is less than |
| // configured stale-node interval. |
| updateNodeState(node, staleNodeCondition, state, |
| NodeLifeCycleEvent.TIMEOUT); |
| break; |
| case STALE: |
| // Move the node to DEAD if the last heartbeat time is less than |
| // configured dead-node interval. |
| updateNodeState(node, deadNodeCondition, state, |
| NodeLifeCycleEvent.TIMEOUT); |
| // Restore the node if we have received heartbeat before configured |
| // stale-node interval. |
| updateNodeState(node, healthyNodeCondition, state, |
| NodeLifeCycleEvent.RESTORE); |
| break; |
| case DEAD: |
| // Resurrect the node if we have received heartbeat before |
| // configured stale-node interval. |
| updateNodeState(node, healthyNodeCondition, state, |
| NodeLifeCycleEvent.RESURRECT); |
| break; |
| // We don't do anything for DECOMMISSIONING and DECOMMISSIONED in |
| // heartbeat processing. |
| case DECOMMISSIONING: |
| case DECOMMISSIONED: |
| default: |
| } |
| } |
| } |
| } catch (NodeNotFoundException e) { |
| // This should not happen unless someone else other than |
| // NodeStateManager is directly modifying NodeStateMap and removed |
| // the node entry after we got the list of UUIDs. |
| LOG.error("Inconsistent NodeStateMap! " + nodeStateMap); |
| } |
| long processingEndTime = Time.monotonicNow(); |
| //If we have taken too much time for HB processing, log that information. |
| if ((processingEndTime - processingStartTime) > |
| heartbeatCheckerIntervalMs) { |
| LOG.error("Total time spend processing datanode HB's is greater than " + |
| "configured values for datanode heartbeats. Please adjust the" + |
| " heartbeat configs. Time Spend on HB processing: {} seconds " + |
| "Datanode heartbeat Interval: {} seconds.", |
| TimeUnit.MILLISECONDS |
| .toSeconds(processingEndTime - processingStartTime), |
| heartbeatCheckerIntervalMs); |
| } |
| |
| } |
| |
| private void scheduleNextHealthCheck() { |
| |
| if (!Thread.currentThread().isInterrupted() && |
| !executorService.isShutdown()) { |
| //BUGBUG: The return future needs to checked here to make sure the |
| // exceptions are handled correctly. |
| healthCheckFuture = executorService.schedule(this, |
| heartbeatCheckerIntervalMs, TimeUnit.MILLISECONDS); |
| } else { |
| LOG.warn("Current Thread is interrupted, shutting down HB processing " + |
| "thread for Node Manager."); |
| } |
| |
| lastHealthCheck = Time.monotonicNow(); |
| } |
| |
| /** |
| * if the time since last check exceeds the stale|dead node interval, skip. |
| * such long delays might be caused by a JVM pause. SCM cannot make reliable |
| * conclusions about datanode health in such situations. |
| * @return : true indicates skip HB checks |
| */ |
| private boolean shouldSkipCheck() { |
| |
| long currentTime = Time.monotonicNow(); |
| long minInterval = Math.min(staleNodeIntervalMs, deadNodeIntervalMs); |
| |
| return ((currentTime - lastHealthCheck) >= minInterval); |
| } |
| |
| /** |
| * Updates the node state if the condition satisfies. |
| * |
| * @param node DatanodeInfo |
| * @param condition condition to check |
| * @param state current state of node |
| * @param lifeCycleEvent NodeLifeCycleEvent to be applied if condition |
| * matches |
| * |
| * @throws NodeNotFoundException if the node is not present |
| */ |
| private void updateNodeState(DatanodeInfo node, Predicate<Long> condition, |
| NodeState state, NodeLifeCycleEvent lifeCycleEvent) |
| throws NodeNotFoundException { |
| try { |
| if (condition.test(node.getLastHeartbeatTime())) { |
| NodeState newState = stateMachine.getNextState(state, lifeCycleEvent); |
| nodeStateMap.updateNodeState(node.getUuid(), state, newState); |
| if (state2EventMap.containsKey(newState)) { |
| eventPublisher.fireEvent(state2EventMap.get(newState), node); |
| } |
| } |
| } catch (InvalidStateTransitionException e) { |
| LOG.warn("Invalid state transition of node {}." + |
| " Current state: {}, life cycle event: {}", |
| node, state, lifeCycleEvent); |
| } |
| } |
| |
| @Override |
| public void close() { |
| executorService.shutdown(); |
| try { |
| if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) { |
| executorService.shutdownNow(); |
| } |
| |
| if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) { |
| LOG.error("Unable to shutdown NodeStateManager properly."); |
| } |
| } catch (InterruptedException e) { |
| executorService.shutdownNow(); |
| Thread.currentThread().interrupt(); |
| } |
| } |
| |
| /** |
| * Test Utility : return number of times heartbeat check was skipped. |
| * @return : count of times HB process was skipped |
| */ |
| @VisibleForTesting |
| long getSkippedHealthChecks() { |
| return skippedHealthChecks; |
| } |
| |
| /** |
| * Test Utility : Pause the periodic node hb check. |
| * @return ScheduledFuture for the scheduled check that got cancelled. |
| */ |
| @VisibleForTesting |
| ScheduledFuture pause() { |
| |
| if (executorService.isShutdown() || checkPaused) { |
| return null; |
| } |
| |
| checkPaused = healthCheckFuture.cancel(false); |
| |
| return healthCheckFuture; |
| } |
| |
| /** |
| * Test utility : unpause the periodic node hb check. |
| * @return ScheduledFuture for the next scheduled check |
| */ |
| @VisibleForTesting |
| ScheduledFuture unpause() { |
| |
| if (executorService.isShutdown()) { |
| return null; |
| } |
| |
| if (checkPaused) { |
| Preconditions.checkState(((healthCheckFuture == null) |
| || healthCheckFuture.isCancelled() |
| || healthCheckFuture.isDone())); |
| |
| checkPaused = false; |
| /** |
| * We do not call scheduleNextHealthCheck because we are |
| * not updating the lastHealthCheck timestamp. |
| */ |
| healthCheckFuture = executorService.schedule(this, |
| heartbeatCheckerIntervalMs, TimeUnit.MILLISECONDS); |
| } |
| |
| return healthCheckFuture; |
| } |
| } |
