| /** |
| * 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 |
| * <p> |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * <p> |
| * 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.util.concurrent.ThreadFactoryBuilder; |
| import org.apache.commons.lang3.tuple.Pair; |
| import org.apache.hadoop.hdds.client.ECReplicationConfig; |
| import org.apache.hadoop.hdds.conf.OzoneConfiguration; |
| import org.apache.hadoop.hdds.protocol.DatanodeDetails; |
| import org.apache.hadoop.hdds.protocol.proto.HddsProtos; |
| import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState; |
| import org.apache.hadoop.hdds.scm.DatanodeAdminError; |
| import org.apache.hadoop.hdds.scm.ScmConfigKeys; |
| import org.apache.hadoop.hdds.scm.container.ContainerID; |
| import org.apache.hadoop.hdds.scm.container.ContainerInfo; |
| import org.apache.hadoop.hdds.scm.container.ContainerManager; |
| import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException; |
| import org.apache.hadoop.hdds.scm.container.replication.ReplicationManager; |
| import org.apache.hadoop.hdds.scm.ha.SCMContext; |
| import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException; |
| import org.apache.hadoop.hdds.server.events.EventPublisher; |
| import org.apache.hadoop.hdfs.DFSConfigKeys; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.net.InetAddress; |
| import java.net.URI; |
| import java.net.URISyntaxException; |
| import java.net.UnknownHostException; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.Comparator; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.TimeUnit; |
| import java.util.stream.Collectors; |
| |
| /** |
| * Class used to manage datanodes scheduled for maintenance or decommission. |
| */ |
| public class NodeDecommissionManager { |
| |
| private final ScheduledExecutorService executor; |
| private final DatanodeAdminMonitor monitor; |
| |
| private final NodeManager nodeManager; |
| private ContainerManager containerManager; |
| private final SCMContext scmContext; |
| private final boolean useHostnames; |
| private Integer maintenanceReplicaMinimum; |
| private Integer maintenanceRemainingRedundancy; |
| |
| // Decommissioning and Maintenance mode progress related metrics. |
| private final NodeDecommissionMetrics metrics; |
| |
| private static final Logger LOG = |
| LoggerFactory.getLogger(NodeDecommissionManager.class); |
| |
| static final class HostDefinition { |
| private String rawHostname; |
| private String hostname; |
| private int port; |
| |
| HostDefinition(String hostname) throws InvalidHostStringException { |
| this.rawHostname = hostname; |
| parseHostname(); |
| } |
| |
| public String getRawHostname() { |
| return rawHostname; |
| } |
| |
| public String getHostname() { |
| return hostname; |
| } |
| |
| public int getPort() { |
| return port; |
| } |
| |
| private void parseHostname() throws InvalidHostStringException { |
| try { |
| // A URI *must* have a scheme, so just create a fake one |
| URI uri = new URI("empty://" + rawHostname.trim()); |
| this.hostname = uri.getHost(); |
| this.port = uri.getPort(); |
| |
| if (this.hostname == null) { |
| throw new InvalidHostStringException("The string " + rawHostname + |
| " does not contain a value hostname or hostname:port definition"); |
| } |
| } catch (URISyntaxException e) { |
| throw new InvalidHostStringException( |
| "Unable to parse the hoststring " + rawHostname, e); |
| } |
| } |
| } |
| |
| private List<DatanodeDetails> mapHostnamesToDatanodes(List<String> hosts, |
| List<DatanodeAdminError> errors) { |
| List<DatanodeDetails> results = new LinkedList<>(); |
| HostDefinition host; |
| InetAddress addr; |
| String msg; |
| for (String hostString : hosts) { |
| try { |
| host = new HostDefinition(hostString); |
| addr = InetAddress.getByName(host.getHostname()); |
| } catch (InvalidHostStringException | UnknownHostException e) { |
| LOG.warn("Unable to resolve host {} ", hostString, e); |
| errors.add(new DatanodeAdminError(hostString, |
| e.getMessage())); |
| continue; |
| } |
| String dnsName; |
| if (useHostnames) { |
| dnsName = addr.getHostName(); |
| } else { |
| dnsName = addr.getHostAddress(); |
| } |
| List<DatanodeDetails> found = nodeManager.getNodesByAddress(dnsName); |
| if (found.isEmpty()) { |
| msg = "Host " + host.getRawHostname() |
| + " (" + dnsName + ") is not running any datanodes registered" |
| + " with SCM. Please check the host name."; |
| LOG.warn(msg); |
| errors.add(new DatanodeAdminError(host.getRawHostname(), msg)); |
| } else if (found.size() == 1) { |
| if (host.getPort() != -1 && |
| !validateDNPortMatch(host.getPort(), found.get(0))) { |
| msg = "Host " + host.getRawHostname() |
| + " is running a datanode registered with SCM," |
| + " but the port number doesn't match." |
| + " Please check the port number."; |
| LOG.warn(msg); |
| errors.add(new DatanodeAdminError(host.getRawHostname(), msg)); |
| continue; |
| } |
| results.add(found.get(0)); |
| } else { |
| // Here we either have multiple DNs on the same host / IP, and they |
| // should have different ports. Or, we have a case where a DN was |
| // registered from a host, then stopped and formatted, changing its |
| // UUID and registered again. In that case, the ports of all hosts |
| // should be the same, and we should just use the one with the most |
| // recent heartbeat. |
| if (host.getPort() != -1) { |
| HostDefinition finalHost = host; |
| found.removeIf(dn -> !validateDNPortMatch(finalHost.getPort(), dn)); |
| } |
| if (found.isEmpty()) { |
| msg = "Host " + host.getRawHostname() |
| + " is running multiple datanodes registered with SCM," |
| + " but no port numbers match." |
| + " Please check the port number."; |
| LOG.warn(msg); |
| errors.add(new DatanodeAdminError(host.getRawHostname(), msg)); |
| continue; |
| } else if (found.size() == 1) { |
| results.add(found.get(0)); |
| continue; |
| } |
| // Here we have at least 2 DNs matching the passed in port, or no port |
| // was passed so we may have all the same ports in SCM or a mix of |
| // ports. |
| if (allPortsMatch(found)) { |
| // All ports match, so just use the most recent heartbeat as it is |
| // not possible for a host to have 2 DNs coming from the same port. |
| DatanodeDetails mostRecent = findDnWithMostRecentHeartbeat(found); |
| if (mostRecent == null) { |
| msg = "Host " + host.getRawHostname() |
| + " has multiple datanodes registered with SCM." |
| + " All have identical ports, but none have a newest" |
| + " heartbeat."; |
| LOG.warn(msg); |
| errors.add(new DatanodeAdminError(host.getRawHostname(), msg)); |
| continue; |
| } |
| results.add(mostRecent); |
| } else { |
| // We have no passed in port or the ports in SCM do not all match, so |
| // we cannot decide which DN to use. |
| msg = "Host " + host.getRawHostname() |
| + " is running multiple datanodes registered with SCM," |
| + " but no port numbers match." |
| + " Please check the port number."; |
| LOG.warn(msg); |
| errors.add(new DatanodeAdminError(host.getRawHostname(), msg)); |
| } |
| } |
| } |
| return results; |
| } |
| |
| private boolean allPortsMatch(List<DatanodeDetails> dns) { |
| if (dns.size() < 2) { |
| return true; |
| } |
| int port = dns.get(0).getPort(DatanodeDetails.Port.Name.RATIS).getValue(); |
| for (int i = 1; i < dns.size(); i++) { |
| if (dns.get(i).getPort(DatanodeDetails.Port.Name.RATIS).getValue() |
| != port) { |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| private DatanodeDetails findDnWithMostRecentHeartbeat( |
| List<DatanodeDetails> dns) { |
| if (dns.size() < 2) { |
| return dns.isEmpty() ? null : dns.get(0); |
| } |
| List<Pair<DatanodeDetails, Long>> dnsWithHeartbeat = dns.stream() |
| .map(dn -> Pair.of(dn, nodeManager.getLastHeartbeat(dn))) |
| .sorted(Comparator.comparingLong(Pair::getRight)) |
| .collect(Collectors.toList()); |
| // The last element should have the largest (newest) heartbeat. But also |
| // check it is not identical to the last but 1 element, as then we cannot |
| // determine which node to decommission. |
| Pair<DatanodeDetails, Long> last = dnsWithHeartbeat.get( |
| dnsWithHeartbeat.size() - 1); |
| if (last.getRight() > dnsWithHeartbeat.get( |
| dnsWithHeartbeat.size() - 2).getRight()) { |
| return last.getLeft(); |
| } |
| return null; |
| } |
| |
| /** |
| * Check if the passed port is used by the given DatanodeDetails object. If |
| * it is, return true, otherwise return false. |
| * @param port Port number to check if it is used by the datanode |
| * @param dn Datanode to check if it is using the given port |
| * @return True if port is used by the datanode. False otherwise. |
| */ |
| private boolean validateDNPortMatch(int port, DatanodeDetails dn) { |
| for (DatanodeDetails.Port p : dn.getPorts()) { |
| if (p.getValue() == port) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| public NodeDecommissionManager(OzoneConfiguration config, NodeManager nm, ContainerManager cm, |
| SCMContext scmContext, |
| EventPublisher eventQueue, ReplicationManager rm) { |
| this.nodeManager = nm; |
| this.containerManager = cm; |
| this.scmContext = scmContext; |
| |
| executor = Executors.newScheduledThreadPool(1, |
| new ThreadFactoryBuilder() |
| .setNameFormat( |
| scmContext.threadNamePrefix() + "DatanodeAdminManager-%d") |
| .setDaemon(true) |
| .build() |
| ); |
| |
| useHostnames = config.getBoolean( |
| DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME, |
| DFSConfigKeys.DFS_DATANODE_USE_DN_HOSTNAME_DEFAULT); |
| |
| long monitorInterval = config.getTimeDuration( |
| ScmConfigKeys.OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL, |
| ScmConfigKeys.OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL_DEFAULT, |
| TimeUnit.SECONDS); |
| if (monitorInterval <= 0) { |
| LOG.warn("{} must be greater than zero, defaulting to {}", |
| ScmConfigKeys.OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL, |
| ScmConfigKeys.OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL_DEFAULT); |
| config.set(ScmConfigKeys.OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL, |
| ScmConfigKeys.OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL_DEFAULT); |
| monitorInterval = config.getTimeDuration( |
| ScmConfigKeys.OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL, |
| ScmConfigKeys.OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL_DEFAULT, |
| TimeUnit.SECONDS); |
| } |
| setMaintenanceConfigs(config.getInt("hdds.scm.replication.maintenance.replica.minimum", 2), |
| config.getInt("hdds.scm.replication.maintenance.remaining.redundancy", 1)); |
| |
| monitor = new DatanodeAdminMonitorImpl(config, eventQueue, nodeManager, |
| rm); |
| this.metrics = NodeDecommissionMetrics.create(); |
| monitor.setMetrics(this.metrics); |
| executor.scheduleAtFixedRate(monitor, monitorInterval, monitorInterval, |
| TimeUnit.SECONDS); |
| } |
| |
| public Map<String, List<ContainerID>> getContainersPendingReplication(DatanodeDetails dn) |
| throws NodeNotFoundException { |
| return getMonitor().getContainersPendingReplication(dn); |
| } |
| |
| @VisibleForTesting |
| public DatanodeAdminMonitor getMonitor() { |
| return monitor; |
| } |
| |
| public synchronized List<DatanodeAdminError> decommissionNodes( |
| List<String> nodes, boolean force) { |
| List<DatanodeAdminError> errors = new ArrayList<>(); |
| List<DatanodeDetails> dns = mapHostnamesToDatanodes(nodes, errors); |
| // add check for fail-early if force flag is not set |
| if (!force) { |
| LOG.info("Force flag = {}. Checking if decommission is possible for dns: {}", force, dns); |
| boolean decommissionPossible = checkIfDecommissionPossible(dns, errors); |
| if (!decommissionPossible) { |
| LOG.error("Cannot decommission nodes as sufficient node are not available."); |
| errors.add(new DatanodeAdminError("AllHosts", "Sufficient nodes are not available.")); |
| return errors; |
| } |
| } else { |
| LOG.info("Force flag = {}. Skip checking if decommission is possible for dns: {}", force, dns); |
| } |
| for (DatanodeDetails dn : dns) { |
| try { |
| startDecommission(dn); |
| } catch (NodeNotFoundException e) { |
| // We already validated the host strings and retrieved the DnDetails |
| // object from the node manager. Therefore we should never get a |
| // NodeNotFoundException here expect if the node is remove in the |
| // very short window between validation and starting decom. Therefore |
| // log a warning and ignore the exception |
| LOG.warn("The host {} was not found in SCM. Ignoring the request to " + |
| "decommission it", dn.getHostName()); |
| errors.add(new DatanodeAdminError(dn.getHostName(), |
| "The host was not found in SCM")); |
| } catch (InvalidNodeStateException e) { |
| errors.add(new DatanodeAdminError(dn.getHostName(), e.getMessage())); |
| } |
| } |
| return errors; |
| } |
| |
| /** |
| * If a SCM is restarted, then upon re-registration the datanode will already |
| * be in DECOMMISSIONING or ENTERING_MAINTENANCE state. In that case, it |
| * needs to be added back into the monitor to track its progress. |
| * @param dn Datanode to add back to tracking. |
| */ |
| public synchronized void continueAdminForNode(DatanodeDetails dn) |
| throws NodeNotFoundException { |
| if (!scmContext.isLeader()) { |
| LOG.info("follower SCM ignored continue admin for datanode {}", dn); |
| return; |
| } |
| NodeOperationalState opState = getNodeStatus(dn).getOperationalState(); |
| if (opState == NodeOperationalState.DECOMMISSIONING |
| || opState == NodeOperationalState.ENTERING_MAINTENANCE |
| || opState == NodeOperationalState.IN_MAINTENANCE) { |
| LOG.info("Continue admin for datanode {}", dn); |
| monitor.startMonitoring(dn); |
| } |
| } |
| |
| public synchronized void startDecommission(DatanodeDetails dn) |
| throws NodeNotFoundException, InvalidNodeStateException { |
| NodeStatus nodeStatus = getNodeStatus(dn); |
| NodeOperationalState opState = nodeStatus.getOperationalState(); |
| if (opState == NodeOperationalState.IN_SERVICE) { |
| LOG.info("Starting Decommission for node {}", dn); |
| nodeManager.setNodeOperationalState( |
| dn, NodeOperationalState.DECOMMISSIONING); |
| monitor.startMonitoring(dn); |
| } else if (nodeStatus.isDecommission()) { |
| LOG.info("Start Decommission called on node {} in state {}. Nothing to " + |
| "do.", dn, opState); |
| } else { |
| LOG.error("Cannot decommission node {} in state {}", dn, opState); |
| throw new InvalidNodeStateException("Cannot decommission node " + |
| dn + " in state " + opState); |
| } |
| } |
| |
| private synchronized boolean checkIfDecommissionPossible(List<DatanodeDetails> dns, List<DatanodeAdminError> errors) { |
| int numDecom = dns.size(); |
| List<DatanodeDetails> validDns = new ArrayList<>(dns); |
| int inServiceTotal = nodeManager.getNodeCount(NodeStatus.inServiceHealthy()); |
| for (DatanodeDetails dn : dns) { |
| try { |
| NodeStatus nodeStatus = getNodeStatus(dn); |
| NodeOperationalState opState = nodeStatus.getOperationalState(); |
| if (opState != NodeOperationalState.IN_SERVICE) { |
| numDecom--; |
| validDns.remove(dn); |
| } |
| } catch (NodeNotFoundException ex) { |
| numDecom--; |
| validDns.remove(dn); |
| } |
| } |
| |
| for (DatanodeDetails dn : validDns) { |
| Set<ContainerID> containers; |
| try { |
| containers = nodeManager.getContainers(dn); |
| } catch (NodeNotFoundException ex) { |
| LOG.warn("The host {} was not found in SCM. Ignoring the request to " + |
| "decommission it", dn.getHostName()); |
| continue; // ignore the DN and continue to next one |
| } |
| |
| for (ContainerID cid : containers) { |
| ContainerInfo cif; |
| try { |
| cif = containerManager.getContainer(cid); |
| } catch (ContainerNotFoundException ex) { |
| LOG.warn("Could not find container info for container {}.", cid); |
| continue; // ignore the container and continue to next one |
| } |
| synchronized (cif) { |
| if (cif.getState().equals(HddsProtos.LifeCycleState.DELETED) || |
| cif.getState().equals(HddsProtos.LifeCycleState.DELETING)) { |
| continue; |
| } |
| int reqNodes = cif.getReplicationConfig().getRequiredNodes(); |
| if ((inServiceTotal - numDecom) < reqNodes) { |
| LOG.info("Cannot decommission nodes. Tried to decommission {} nodes of which valid nodes = {}. " + |
| "Cluster state: In-service nodes = {}, nodes required for replication = {}. " + |
| "Failing due to datanode : {}, container : {}", |
| dns.size(), numDecom, inServiceTotal, reqNodes, dn, cid); |
| return false; |
| } |
| } |
| } |
| } |
| return true; |
| } |
| |
| public synchronized List<DatanodeAdminError> recommissionNodes( |
| List<String> nodes) { |
| List<DatanodeAdminError> errors = new ArrayList<>(); |
| List<DatanodeDetails> dns = mapHostnamesToDatanodes(nodes, errors); |
| for (DatanodeDetails dn : dns) { |
| try { |
| recommission(dn); |
| } catch (NodeNotFoundException e) { |
| // We already validated the host strings and retrieved the DnDetails |
| // object from the node manager. Therefore we should never get a |
| // NodeNotFoundException here expect if the node is remove in the |
| // very short window between validation and starting decom. Therefore |
| // log a warning and ignore the exception |
| LOG.warn("Host {} was not found in SCM. Ignoring the request to " + |
| "recommission it.", dn.getHostName()); |
| errors.add(new DatanodeAdminError(dn.getHostName(), |
| "The host was not found in SCM")); |
| } |
| } |
| return errors; |
| } |
| |
| public synchronized void recommission(DatanodeDetails dn) |
| throws NodeNotFoundException { |
| NodeStatus nodeStatus = getNodeStatus(dn); |
| NodeOperationalState opState = nodeStatus.getOperationalState(); |
| if (opState != NodeOperationalState.IN_SERVICE) { |
| // The node will be set back to IN_SERVICE when it is processed by the |
| // monitor |
| monitor.stopMonitoring(dn); |
| LOG.info("Queued node {} for recommission", dn); |
| } else { |
| LOG.info("Recommission called on node {} with state {}. " + |
| "Nothing to do.", dn, opState); |
| } |
| } |
| |
| public synchronized List<DatanodeAdminError> startMaintenanceNodes( |
| List<String> nodes, int endInHours, boolean force) { |
| List<DatanodeAdminError> errors = new ArrayList<>(); |
| List<DatanodeDetails> dns = mapHostnamesToDatanodes(nodes, errors); |
| // add check for fail-early if force flag is not set |
| if (!force) { |
| LOG.info("Force flag = {}. Checking if maintenance is possible for dns: {}", force, dns); |
| boolean maintenancePossible = checkIfMaintenancePossible(dns, errors); |
| if (!maintenancePossible) { |
| LOG.error("Cannot put nodes to maintenance as sufficient node are not available."); |
| errors.add(new DatanodeAdminError("AllHosts", "Sufficient nodes are not available.")); |
| return errors; |
| } |
| } else { |
| LOG.info("Force flag = {}. Skip checking if maintenance is possible for dns: {}", force, dns); |
| } |
| for (DatanodeDetails dn : dns) { |
| try { |
| startMaintenance(dn, endInHours); |
| } catch (NodeNotFoundException e) { |
| // We already validated the host strings and retrieved the DnDetails |
| // object from the node manager. Therefore we should never get a |
| // NodeNotFoundException here expect if the node is remove in the |
| // very short window between validation and starting decom. Therefore |
| // log a warning and ignore the exception |
| LOG.warn("The host {} was not found in SCM. Ignoring the request to " + |
| "start maintenance on it", dn.getHostName()); |
| } catch (InvalidNodeStateException e) { |
| errors.add(new DatanodeAdminError(dn.getHostName(), e.getMessage())); |
| } |
| } |
| return errors; |
| } |
| |
| // TODO - If startMaintenance is called on a host already in maintenance, |
| // then we should update the end time? |
| public synchronized void startMaintenance(DatanodeDetails dn, int endInHours) |
| throws NodeNotFoundException, InvalidNodeStateException { |
| NodeStatus nodeStatus = getNodeStatus(dn); |
| NodeOperationalState opState = nodeStatus.getOperationalState(); |
| |
| long maintenanceEnd = 0; |
| if (endInHours != 0) { |
| maintenanceEnd = |
| (System.currentTimeMillis() / 1000L) + (endInHours * 60L * 60L); |
| } |
| if (opState == NodeOperationalState.IN_SERVICE) { |
| nodeManager.setNodeOperationalState( |
| dn, NodeOperationalState.ENTERING_MAINTENANCE, maintenanceEnd); |
| monitor.startMonitoring(dn); |
| LOG.info("Starting Maintenance for node {}", dn); |
| } else if (nodeStatus.isMaintenance()) { |
| LOG.info("Starting Maintenance called on node {} with state {}. " + |
| "Nothing to do.", dn, opState); |
| } else { |
| LOG.error("Cannot start maintenance on node {} in state {}", dn, opState); |
| throw new InvalidNodeStateException("Cannot start maintenance on node " + |
| dn + " in state " + opState); |
| } |
| } |
| |
| private synchronized boolean checkIfMaintenancePossible(List<DatanodeDetails> dns, List<DatanodeAdminError> errors) { |
| int numMaintenance = dns.size(); |
| List<DatanodeDetails> validDns = dns.stream().collect(Collectors.toList()); |
| Collections.copy(validDns, dns); |
| int inServiceTotal = nodeManager.getNodeCount(NodeStatus.inServiceHealthy()); |
| for (DatanodeDetails dn : dns) { |
| try { |
| NodeStatus nodeStatus = getNodeStatus(dn); |
| NodeOperationalState opState = nodeStatus.getOperationalState(); |
| if (opState != NodeOperationalState.IN_SERVICE) { |
| numMaintenance--; |
| validDns.remove(dn); |
| } |
| } catch (NodeNotFoundException ex) { |
| numMaintenance--; |
| validDns.remove(dn); |
| } |
| } |
| |
| for (DatanodeDetails dn : validDns) { |
| Set<ContainerID> containers; |
| try { |
| containers = nodeManager.getContainers(dn); |
| } catch (NodeNotFoundException ex) { |
| LOG.warn("The host {} was not found in SCM. Ignoring the request to " + |
| "enter maintenance", dn.getHostName()); |
| errors.add(new DatanodeAdminError(dn.getHostName(), |
| "The host was not found in SCM")); |
| continue; // ignore the DN and continue to next one |
| } |
| |
| for (ContainerID cid : containers) { |
| ContainerInfo cif; |
| try { |
| cif = containerManager.getContainer(cid); |
| } catch (ContainerNotFoundException ex) { |
| continue; // ignore the container and continue to next one |
| } |
| synchronized (cif) { |
| if (cif.getState().equals(HddsProtos.LifeCycleState.DELETED) || |
| cif.getState().equals(HddsProtos.LifeCycleState.DELETING)) { |
| continue; |
| } |
| |
| int minInService; |
| HddsProtos.ReplicationType replicationType = cif.getReplicationType(); |
| if (replicationType.equals(HddsProtos.ReplicationType.EC)) { |
| int reqNodes = cif.getReplicationConfig().getRequiredNodes(); |
| int data = ((ECReplicationConfig)cif.getReplicationConfig()).getData(); |
| minInService = Math.min((data + maintenanceRemainingRedundancy), reqNodes); |
| } else { |
| minInService = maintenanceReplicaMinimum; |
| } |
| if ((inServiceTotal - numMaintenance) < minInService) { |
| LOG.info("Cannot enter nodes into maintenance. Tried to start maintenance for {} nodes " + |
| "of which valid nodes = {}. " + |
| "Cluster state: In-service nodes = {}, nodes required for replication = {}. " + |
| "Failing due to datanode : {}, container : {}", |
| dns.size(), numMaintenance, inServiceTotal, minInService, dn, cid); |
| return false; |
| } |
| } |
| } |
| } |
| return true; |
| } |
| |
| /** |
| * Stops the decommission monitor from running when SCM is shutdown. |
| */ |
| public void stop() { |
| metrics.unRegister(); |
| if (executor != null) { |
| executor.shutdown(); |
| } |
| } |
| |
| private NodeStatus getNodeStatus(DatanodeDetails dn) |
| throws NodeNotFoundException { |
| return nodeManager.getNodeStatus(dn); |
| } |
| |
| /** |
| * Called in SCMStateMachine#notifyLeaderChanged when current SCM becomes |
| * leader. |
| */ |
| public void onBecomeLeader() { |
| nodeManager.getAllNodes().forEach(datanodeDetails -> { |
| try { |
| continueAdminForNode(datanodeDetails); |
| } catch (NodeNotFoundException e) { |
| // Should not happen, as the node has just registered to call this event |
| // handler. |
| LOG.warn("NodeNotFound when adding the node to the decommissionManager", |
| e); |
| } |
| }); |
| } |
| |
| @VisibleForTesting |
| public void setMaintenanceConfigs(int replicaMinimum, int remainingRedundancy) { |
| synchronized (this) { |
| maintenanceRemainingRedundancy = remainingRedundancy; |
| maintenanceReplicaMinimum = replicaMinimum; |
| } |
| } |
| } |
