| /** |
| * 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.container.replication; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import org.apache.hadoop.hdds.HddsConfigKeys; |
| import org.apache.hadoop.hdds.conf.Config; |
| import org.apache.hadoop.hdds.conf.ConfigGroup; |
| import org.apache.hadoop.hdds.conf.ConfigType; |
| import org.apache.hadoop.hdds.conf.ConfigurationSource; |
| import org.apache.hadoop.hdds.protocol.DatanodeDetails; |
| import org.apache.hadoop.hdds.protocol.proto.HddsProtos; |
| import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto; |
| import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State; |
| import org.apache.hadoop.hdds.scm.PlacementPolicy; |
| 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.ContainerReplica; |
| |
| import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport; |
| import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport.HealthState; |
| import org.apache.hadoop.hdds.scm.events.SCMEvents; |
| import org.apache.hadoop.hdds.scm.ha.SCMContext; |
| import org.apache.hadoop.hdds.scm.ha.SCMService; |
| import org.apache.hadoop.hdds.scm.node.NodeManager; |
| import org.apache.hadoop.hdds.scm.node.NodeStatus; |
| import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException; |
| import org.apache.hadoop.hdds.scm.server.StorageContainerManager; |
| import org.apache.hadoop.hdds.server.events.EventPublisher; |
| import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand; |
| import org.apache.hadoop.ozone.protocol.commands.CommandForDatanode; |
| import org.apache.hadoop.ozone.protocol.commands.SCMCommand; |
| import org.apache.hadoop.util.ExitUtil; |
| import org.apache.ratis.protocol.exceptions.NotLeaderException; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.time.Clock; |
| import java.time.Duration; |
| import java.util.Comparator; |
| import java.util.Iterator; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.PriorityQueue; |
| import java.util.Queue; |
| import java.util.Set; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| import java.util.concurrent.locks.Lock; |
| import java.util.concurrent.locks.ReentrantLock; |
| import java.util.stream.Collectors; |
| |
| import static org.apache.hadoop.hdds.conf.ConfigTag.OZONE; |
| import static org.apache.hadoop.hdds.conf.ConfigTag.SCM; |
| import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType.EC; |
| |
| /** |
| * Replication Manager (RM) is the one which is responsible for making sure |
| * that the containers are properly replicated. Replication Manager deals only |
| * with Quasi Closed / Closed container. |
| */ |
| public class ReplicationManager implements SCMService { |
| |
| public static final Logger LOG = |
| LoggerFactory.getLogger(ReplicationManager.class); |
| |
| /** |
| * Reference to the ContainerManager. |
| */ |
| private final ContainerManager containerManager; |
| |
| |
| /** |
| * SCMContext from StorageContainerManager. |
| */ |
| private final SCMContext scmContext; |
| |
| |
| /** |
| * ReplicationManager specific configuration. |
| */ |
| private final ReplicationManagerConfiguration rmConf; |
| private final NodeManager nodeManager; |
| |
| /** |
| * ReplicationMonitor thread is the one which wakes up at configured |
| * interval and processes all the containers. |
| */ |
| private Thread replicationMonitor; |
| |
| /** |
| * Flag used for checking if the ReplicationMonitor thread is running or |
| * not. |
| */ |
| private volatile boolean running; |
| |
| /** |
| * Report object that is refreshed each time replication Manager runs. |
| */ |
| private ReplicationManagerReport containerReport; |
| |
| /** |
| * Replication progress related metrics. |
| */ |
| private ReplicationManagerMetrics metrics; |
| |
| |
| /** |
| * Legacy RM will hopefully be removed after completing refactor |
| * for now, it is used to process non-EC container. |
| */ |
| private LegacyReplicationManager legacyReplicationManager; |
| |
| /** |
| * SCMService related variables. |
| * After leaving safe mode, replicationMonitor needs to wait for a while |
| * before really take effect. |
| */ |
| private final Lock serviceLock = new ReentrantLock(); |
| private ServiceStatus serviceStatus = ServiceStatus.PAUSING; |
| private final long waitTimeInMillis; |
| private long lastTimeToBeReadyInMillis = 0; |
| private final Clock clock; |
| private final ContainerReplicaPendingOps containerReplicaPendingOps; |
| private final ContainerHealthCheck ecContainerHealthCheck; |
| private final EventPublisher eventPublisher; |
| private final ReentrantLock lock = new ReentrantLock(); |
| private Queue<ContainerHealthResult.UnderReplicatedHealthResult> |
| underRepQueue; |
| private Queue<ContainerHealthResult.OverReplicatedHealthResult> |
| overRepQueue; |
| private final ECUnderReplicationHandler ecUnderReplicationHandler; |
| private final ECOverReplicationHandler ecOverReplicationHandler; |
| |
| /** |
| * Constructs ReplicationManager instance with the given configuration. |
| * |
| * @param conf OzoneConfiguration |
| * @param containerManager ContainerManager |
| * @param containerPlacement PlacementPolicy |
| * @param eventPublisher EventPublisher |
| */ |
| @SuppressWarnings("parameternumber") |
| public ReplicationManager(final ConfigurationSource conf, |
| final ContainerManager containerManager, |
| final PlacementPolicy containerPlacement, |
| final EventPublisher eventPublisher, |
| final SCMContext scmContext, |
| final NodeManager nodeManager, |
| final Clock clock, |
| final LegacyReplicationManager legacyReplicationManager, |
| final ContainerReplicaPendingOps replicaPendingOps) |
| throws IOException { |
| this.containerManager = containerManager; |
| this.scmContext = scmContext; |
| this.rmConf = conf.getObject(ReplicationManagerConfiguration.class); |
| this.running = false; |
| this.clock = clock; |
| this.containerReport = new ReplicationManagerReport(); |
| this.metrics = null; |
| this.eventPublisher = eventPublisher; |
| this.waitTimeInMillis = conf.getTimeDuration( |
| HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT, |
| HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT_DEFAULT, |
| TimeUnit.MILLISECONDS); |
| this.containerReplicaPendingOps = replicaPendingOps; |
| this.legacyReplicationManager = legacyReplicationManager; |
| this.ecContainerHealthCheck = new ECContainerHealthCheck(); |
| this.nodeManager = nodeManager; |
| this.underRepQueue = createUnderReplicatedQueue(); |
| this.overRepQueue = new LinkedList<>(); |
| ecUnderReplicationHandler = new ECUnderReplicationHandler( |
| containerPlacement, conf, nodeManager); |
| ecOverReplicationHandler = |
| new ECOverReplicationHandler(containerPlacement, nodeManager); |
| start(); |
| } |
| |
| /** |
| * Starts Replication Monitor thread. |
| */ |
| @Override |
| public synchronized void start() { |
| if (!isRunning()) { |
| LOG.info("Starting Replication Monitor Thread."); |
| running = true; |
| metrics = ReplicationManagerMetrics.create(this); |
| legacyReplicationManager.setMetrics(metrics); |
| replicationMonitor = new Thread(this::run); |
| replicationMonitor.setName("ReplicationMonitor"); |
| replicationMonitor.setDaemon(true); |
| replicationMonitor.start(); |
| } else { |
| LOG.info("Replication Monitor Thread is already running."); |
| } |
| } |
| |
| /** |
| * Returns true if the Replication Monitor Thread is running. |
| * |
| * @return true if running, false otherwise |
| */ |
| public boolean isRunning() { |
| if (!running) { |
| synchronized (this) { |
| return replicationMonitor != null |
| && replicationMonitor.isAlive(); |
| } |
| } |
| return true; |
| } |
| |
| /** |
| * Stops Replication Monitor thread. |
| */ |
| public synchronized void stop() { |
| if (running) { |
| LOG.info("Stopping Replication Monitor Thread."); |
| running = false; |
| legacyReplicationManager.clearInflightActions(); |
| metrics.unRegister(); |
| replicationMonitor.interrupt(); |
| } else { |
| LOG.info("Replication Monitor Thread is not running."); |
| } |
| } |
| |
| /** |
| * Process all the containers now, and wait for the processing to complete. |
| * This in intended to be used in tests. |
| */ |
| public synchronized void processAll() { |
| if (!shouldRun()) { |
| LOG.info("Replication Manager is not ready to run until {}ms after " + |
| "safemode exit", waitTimeInMillis); |
| return; |
| } |
| final long start = clock.millis(); |
| final List<ContainerInfo> containers = |
| containerManager.getContainers(); |
| ReplicationManagerReport report = new ReplicationManagerReport(); |
| Queue<ContainerHealthResult.UnderReplicatedHealthResult> |
| underReplicated = createUnderReplicatedQueue(); |
| Queue<ContainerHealthResult.OverReplicatedHealthResult> overReplicated = |
| new LinkedList<>(); |
| |
| for (ContainerInfo c : containers) { |
| if (!shouldRun()) { |
| break; |
| } |
| report.increment(c.getState()); |
| if (c.getReplicationType() != EC) { |
| legacyReplicationManager.processContainer(c, report); |
| continue; |
| } |
| try { |
| processContainer(c, underReplicated, overReplicated, report); |
| // TODO - send any commands contained in the health result |
| } catch (ContainerNotFoundException e) { |
| LOG.error("Container {} not found", c.getContainerID(), e); |
| } |
| } |
| report.setComplete(); |
| lock.lock(); |
| try { |
| underRepQueue = underReplicated; |
| overRepQueue = overReplicated; |
| } finally { |
| lock.unlock(); |
| } |
| this.containerReport = report; |
| LOG.info("Replication Monitor Thread took {} milliseconds for" + |
| " processing {} containers.", clock.millis() - start, |
| containers.size()); |
| } |
| |
| /** |
| * Retrieve the new highest priority container to be replicated from the |
| * under replicated queue. |
| * @return The new underReplicated container to be processed, or null if the |
| * queue is empty. |
| */ |
| public ContainerHealthResult.UnderReplicatedHealthResult |
| dequeueUnderReplicatedContainer() { |
| lock.lock(); |
| try { |
| return underRepQueue.poll(); |
| } finally { |
| lock.unlock(); |
| } |
| } |
| |
| /** |
| * Retrieve the new highest priority container to be replicated from the |
| * under replicated queue. |
| * @return The next over-replicated container to be processed, or null if the |
| * queue is empty. |
| */ |
| public ContainerHealthResult.OverReplicatedHealthResult |
| dequeueOverReplicatedContainer() { |
| lock.lock(); |
| try { |
| return overRepQueue.poll(); |
| } finally { |
| lock.unlock(); |
| } |
| } |
| |
| /** |
| * Add an under replicated container back to the queue if it was unable to |
| * be processed. Its retry count will be incremented before it is re-queued, |
| * reducing its priority. |
| * Note that the queue could have been rebuilt and replaced after this |
| * message was removed but before it is added back. This will result in a |
| * duplicate entry on the queue. However, when it is processed again, the |
| * result of the processing will end up with pending replicas scheduled. If |
| * instance 1 is processed and creates the pending replicas, when instance 2 |
| * is processed, it will find the pending containers and know it has no work |
| * to do, and be discarded. Additionally, the queue will be refreshed |
| * periodically removing any duplicates. |
| * @param underReplicatedHealthResult |
| */ |
| public void requeueUnderReplicatedContainer(ContainerHealthResult |
| .UnderReplicatedHealthResult underReplicatedHealthResult) { |
| underReplicatedHealthResult.incrementRequeueCount(); |
| lock.lock(); |
| try { |
| underRepQueue.add(underReplicatedHealthResult); |
| } finally { |
| lock.unlock(); |
| } |
| } |
| |
| public void requeueOverReplicatedContainer(ContainerHealthResult |
| .OverReplicatedHealthResult overReplicatedHealthResult) { |
| lock.lock(); |
| try { |
| overRepQueue.add(overReplicatedHealthResult); |
| } finally { |
| lock.unlock(); |
| } |
| } |
| |
| public Map<DatanodeDetails, SCMCommand<?>> processUnderReplicatedContainer( |
| final ContainerHealthResult result) throws IOException { |
| ContainerID containerID = result.getContainerInfo().containerID(); |
| Set<ContainerReplica> replicas = containerManager.getContainerReplicas( |
| containerID); |
| List<ContainerReplicaOp> pendingOps = |
| containerReplicaPendingOps.getPendingOps(containerID); |
| return ecUnderReplicationHandler.processAndCreateCommands(replicas, |
| pendingOps, result, 0); |
| } |
| |
| public Map<DatanodeDetails, SCMCommand<?>> processOverReplicatedContainer( |
| final ContainerHealthResult result) throws IOException { |
| ContainerID containerID = result.getContainerInfo().containerID(); |
| Set<ContainerReplica> replicas = containerManager.getContainerReplicas( |
| containerID); |
| List<ContainerReplicaOp> pendingOps = |
| containerReplicaPendingOps.getPendingOps(containerID); |
| return ecOverReplicationHandler.processAndCreateCommands(replicas, |
| pendingOps, result, 0); |
| } |
| |
| public long getScmTerm() throws NotLeaderException { |
| return scmContext.getTermOfLeader(); |
| } |
| |
| protected ContainerHealthResult processContainer(ContainerInfo containerInfo, |
| Queue<ContainerHealthResult.UnderReplicatedHealthResult> underRep, |
| Queue<ContainerHealthResult.OverReplicatedHealthResult> overRep, |
| ReplicationManagerReport report) throws ContainerNotFoundException { |
| |
| ContainerID containerID = containerInfo.containerID(); |
| Set<ContainerReplica> replicas = containerManager.getContainerReplicas( |
| containerID); |
| |
| if (containerInfo.getState() == HddsProtos.LifeCycleState.OPEN) { |
| if (!isOpenContainerHealthy(containerInfo, replicas)) { |
| report.incrementAndSample( |
| HealthState.OPEN_UNHEALTHY, containerID); |
| eventPublisher.fireEvent(SCMEvents.CLOSE_CONTAINER, containerID); |
| return new ContainerHealthResult.UnHealthyResult(containerInfo); |
| } |
| return new ContainerHealthResult.HealthyResult(containerInfo); |
| } |
| |
| if (containerInfo.getState() == HddsProtos.LifeCycleState.CLOSED) { |
| List<ContainerReplica> unhealthyReplicas = replicas.stream() |
| .filter(r -> !compareState(containerInfo.getState(), r.getState())) |
| .collect(Collectors.toList()); |
| |
| if (unhealthyReplicas.size() > 0) { |
| handleUnhealthyReplicas(containerInfo, unhealthyReplicas); |
| } |
| } |
| |
| List<ContainerReplicaOp> pendingOps = |
| containerReplicaPendingOps.getPendingOps(containerID); |
| ContainerHealthResult health = ecContainerHealthCheck |
| .checkHealth(containerInfo, replicas, pendingOps, 0); |
| // TODO - should the report have a HEALTHY state, rather than just bad |
| // states? It would need to be added to legacy RM too. |
| if (health.getHealthState() |
| == ContainerHealthResult.HealthState.UNDER_REPLICATED) { |
| report.incrementAndSample(HealthState.UNDER_REPLICATED, containerID); |
| ContainerHealthResult.UnderReplicatedHealthResult underHealth |
| = ((ContainerHealthResult.UnderReplicatedHealthResult) health); |
| if (underHealth.isUnrecoverable()) { |
| // TODO - do we need a new health state for unrecoverable EC? |
| report.incrementAndSample(HealthState.MISSING, containerID); |
| } |
| if (!underHealth.isSufficientlyReplicatedAfterPending() && |
| !underHealth.isUnrecoverable()) { |
| underRep.add(underHealth); |
| } |
| } else if (health.getHealthState() |
| == ContainerHealthResult.HealthState.OVER_REPLICATED) { |
| report.incrementAndSample(HealthState.OVER_REPLICATED, containerID); |
| ContainerHealthResult.OverReplicatedHealthResult overHealth |
| = ((ContainerHealthResult.OverReplicatedHealthResult) health); |
| if (!overHealth.isSufficientlyReplicatedAfterPending()) { |
| overRep.add(overHealth); |
| } |
| } |
| return health; |
| } |
| |
| /** |
| * Handles unhealthy container. |
| * A container is inconsistent if any of the replica state doesn't |
| * match the container state. We have to take appropriate action |
| * based on state of the replica. |
| * |
| * @param container ContainerInfo |
| * @param unhealthyReplicas List of ContainerReplica |
| */ |
| private void handleUnhealthyReplicas(final ContainerInfo container, |
| List<ContainerReplica> unhealthyReplicas) { |
| Iterator<ContainerReplica> iterator = unhealthyReplicas.iterator(); |
| while (iterator.hasNext()) { |
| final ContainerReplica replica = iterator.next(); |
| final ContainerReplicaProto.State state = replica.getState(); |
| if (state == State.OPEN || state == State.CLOSING) { |
| sendCloseCommand(container, replica.getDatanodeDetails(), true); |
| iterator.remove(); |
| } |
| } |
| } |
| |
| /** |
| * Sends close container command for the given container to the given |
| * datanode. |
| * |
| * @param container Container to be closed |
| * @param datanode The datanode on which the container |
| * has to be closed |
| * @param force Should be set to true if we want to force close. |
| */ |
| private void sendCloseCommand(final ContainerInfo container, |
| final DatanodeDetails datanode, final boolean force) { |
| |
| ContainerID containerID = container.containerID(); |
| LOG.info("Sending close container command for container {}" + |
| " to datanode {}.", containerID, datanode); |
| CloseContainerCommand closeContainerCommand = |
| new CloseContainerCommand(container.getContainerID(), |
| container.getPipelineID(), force); |
| try { |
| closeContainerCommand.setTerm(scmContext.getTermOfLeader()); |
| } catch (NotLeaderException nle) { |
| LOG.warn("Skip sending close container command," |
| + " since current SCM is not leader.", nle); |
| return; |
| } |
| closeContainerCommand.setEncodedToken(getContainerToken(containerID)); |
| eventPublisher.fireEvent(SCMEvents.DATANODE_COMMAND, |
| new CommandForDatanode<>(datanode.getUuid(), closeContainerCommand)); |
| } |
| |
| private String getContainerToken(ContainerID containerID) { |
| if (scmContext.getScm() instanceof StorageContainerManager) { |
| StorageContainerManager scm = |
| (StorageContainerManager) scmContext.getScm(); |
| return scm.getContainerTokenGenerator().generateEncodedToken(containerID); |
| } |
| return ""; // unit test |
| } |
| |
| /** |
| * Creates a priority queue of UnderReplicatedHealthResult, where the elements |
| * are ordered by the weighted redundancy of the container. This means that |
| * containers with the least remaining redundancy are at the front of the |
| * queue, and will be processed first. |
| * @return An empty instance of a PriorityQueue. |
| */ |
| protected PriorityQueue<ContainerHealthResult.UnderReplicatedHealthResult> |
| createUnderReplicatedQueue() { |
| return new PriorityQueue<>(Comparator.comparing(ContainerHealthResult |
| .UnderReplicatedHealthResult::getWeightedRedundancy) |
| .thenComparing(ContainerHealthResult |
| .UnderReplicatedHealthResult::getRequeueCount)); |
| } |
| |
| public ReplicationManagerReport getContainerReport() { |
| return containerReport; |
| } |
| |
| /** |
| * ReplicationMonitor thread runnable. This wakes up at configured |
| * interval and processes all the containers in the system. |
| */ |
| private synchronized void run() { |
| try { |
| while (running) { |
| processAll(); |
| wait(rmConf.getInterval()); |
| } |
| } catch (Throwable t) { |
| if (t instanceof InterruptedException) { |
| LOG.info("Replication Monitor Thread is stopped"); |
| Thread.currentThread().interrupt(); |
| } else { |
| // When we get runtime exception, we should terminate SCM. |
| LOG.error("Exception in Replication Monitor Thread.", t); |
| ExitUtil.terminate(1, t); |
| } |
| } |
| } |
| |
| /** |
| * Given a ContainerID, lookup the ContainerInfo and then return a |
| * ContainerReplicaCount object for the container. |
| * @param containerID The ID of the container |
| * @return ContainerReplicaCount for the given container |
| * @throws ContainerNotFoundException |
| */ |
| public ContainerReplicaCount getContainerReplicaCount(ContainerID containerID) |
| throws ContainerNotFoundException { |
| ContainerInfo container = containerManager.getContainer(containerID); |
| if (container.getReplicationType() == EC) { |
| return getECContainerReplicaCount(container); |
| } |
| return legacyReplicationManager.getContainerReplicaCount(container); |
| } |
| |
| /** |
| * An open container is healthy if all its replicas are in the same state as |
| * the container. |
| * @param container The container to check |
| * @param replicas The replicas belonging to the container |
| * @return True if the container is healthy, false otherwise |
| */ |
| private boolean isOpenContainerHealthy( |
| ContainerInfo container, Set<ContainerReplica> replicas) { |
| HddsProtos.LifeCycleState state = container.getState(); |
| return replicas.stream() |
| .allMatch(r -> compareState(state, r.getState())); |
| } |
| |
| /** |
| * Compares the container state with the replica state. |
| * |
| * @param containerState ContainerState |
| * @param replicaState ReplicaState |
| * @return true if the state matches, false otherwise |
| */ |
| public static boolean compareState( |
| final HddsProtos.LifeCycleState containerState, |
| final ContainerReplicaProto.State replicaState) { |
| switch (containerState) { |
| case OPEN: |
| return replicaState == ContainerReplicaProto.State.OPEN; |
| case CLOSING: |
| return replicaState == ContainerReplicaProto.State.CLOSING; |
| case QUASI_CLOSED: |
| return replicaState == ContainerReplicaProto.State.QUASI_CLOSED; |
| case CLOSED: |
| return replicaState == ContainerReplicaProto.State.CLOSED; |
| default: |
| return false; |
| } |
| } |
| |
| /** |
| * Configuration used by the Replication Manager. |
| */ |
| @ConfigGroup(prefix = "hdds.scm.replication") |
| public static class ReplicationManagerConfiguration { |
| /** |
| * The frequency in which ReplicationMonitor thread should run. |
| */ |
| @Config(key = "thread.interval", |
| type = ConfigType.TIME, |
| defaultValue = "300s", |
| tags = {SCM, OZONE}, |
| description = "There is a replication monitor thread running inside " + |
| "SCM which takes care of replicating the containers in the " + |
| "cluster. This property is used to configure the interval in " + |
| "which that thread runs." |
| ) |
| private long interval = Duration.ofSeconds(300).toMillis(); |
| |
| /** |
| * The frequency in which the Under Replicated queue is processed. |
| */ |
| @Config(key = "under.replicated.interval", |
| type = ConfigType.TIME, |
| defaultValue = "30s", |
| tags = {SCM, OZONE}, |
| description = "How frequently to check if there are work to process " + |
| " on the under replicated queue" |
| ) |
| private long underReplicatedInterval = Duration.ofSeconds(30).toMillis(); |
| |
| /** |
| * The frequency in which the Over Replicated queue is processed. |
| */ |
| @Config(key = "over.replicated.interval", |
| type = ConfigType.TIME, |
| defaultValue = "30s", |
| tags = {SCM, OZONE}, |
| description = "How frequently to check if there are work to process " + |
| " on the over replicated queue" |
| ) |
| private long overReplicatedInterval = Duration.ofSeconds(30).toMillis(); |
| |
| /** |
| * Timeout for container replication & deletion command issued by |
| * ReplicationManager. |
| */ |
| @Config(key = "event.timeout", |
| type = ConfigType.TIME, |
| defaultValue = "30m", |
| tags = {SCM, OZONE}, |
| description = "Timeout for the container replication/deletion commands " |
| + "sent to datanodes. After this timeout the command will be " |
| + "retried.") |
| private long eventTimeout = Duration.ofMinutes(30).toMillis(); |
| public void setInterval(Duration interval) { |
| this.interval = interval.toMillis(); |
| } |
| |
| public void setEventTimeout(Duration timeout) { |
| this.eventTimeout = timeout.toMillis(); |
| } |
| |
| /** |
| * The number of container replica which must be available for a node to |
| * enter maintenance. |
| */ |
| @Config(key = "maintenance.replica.minimum", |
| type = ConfigType.INT, |
| defaultValue = "2", |
| tags = {SCM, OZONE}, |
| description = "The minimum number of container replicas which must " + |
| " be available for a node to enter maintenance. If putting a " + |
| " node into maintenance reduces the available replicas for any " + |
| " container below this level, the node will remain in the " + |
| " entering maintenance state until a new replica is created.") |
| private int maintenanceReplicaMinimum = 2; |
| |
| public void setMaintenanceReplicaMinimum(int replicaCount) { |
| this.maintenanceReplicaMinimum = replicaCount; |
| } |
| |
| public long getInterval() { |
| return interval; |
| } |
| |
| public long getUnderReplicatedInterval() { |
| return underReplicatedInterval; |
| } |
| |
| public long getOverReplicatedInterval() { |
| return overReplicatedInterval; |
| } |
| |
| public long getEventTimeout() { |
| return eventTimeout; |
| } |
| |
| public int getMaintenanceReplicaMinimum() { |
| return maintenanceReplicaMinimum; |
| } |
| } |
| |
| @Override |
| public void notifyStatusChanged() { |
| serviceLock.lock(); |
| try { |
| // 1) SCMContext#isLeaderReady returns true. |
| // 2) not in safe mode. |
| if (scmContext.isLeaderReady() && !scmContext.isInSafeMode()) { |
| // transition from PAUSING to RUNNING |
| if (serviceStatus != ServiceStatus.RUNNING) { |
| LOG.info("Service {} transitions to RUNNING.", getServiceName()); |
| lastTimeToBeReadyInMillis = clock.millis(); |
| serviceStatus = ServiceStatus.RUNNING; |
| } |
| //now, as the current scm is leader and it`s state is up-to-date, |
| //we need to take some action about replicated inflight move options. |
| legacyReplicationManager.notifyStatusChanged(); |
| } else { |
| serviceStatus = ServiceStatus.PAUSING; |
| } |
| } finally { |
| serviceLock.unlock(); |
| } |
| } |
| |
| @Override |
| public boolean shouldRun() { |
| serviceLock.lock(); |
| try { |
| // If safe mode is off, then this SCMService starts to run with a delay. |
| return serviceStatus == ServiceStatus.RUNNING && |
| clock.millis() - lastTimeToBeReadyInMillis >= waitTimeInMillis; |
| } finally { |
| serviceLock.unlock(); |
| } |
| } |
| |
| @Override |
| public String getServiceName() { |
| return ReplicationManager.class.getSimpleName(); |
| } |
| |
| public ReplicationManagerMetrics getMetrics() { |
| return metrics; |
| } |
| |
| |
| /** |
| * following functions will be refactored in a separate jira. |
| */ |
| public CompletableFuture<LegacyReplicationManager.MoveResult> move( |
| ContainerID cid, DatanodeDetails src, DatanodeDetails tgt) |
| throws NodeNotFoundException, ContainerNotFoundException, |
| TimeoutException { |
| CompletableFuture<LegacyReplicationManager.MoveResult> ret = |
| new CompletableFuture<>(); |
| if (!isRunning()) { |
| ret.complete(LegacyReplicationManager.MoveResult.FAIL_NOT_RUNNING); |
| return ret; |
| } |
| |
| return legacyReplicationManager.move(cid, src, tgt); |
| } |
| |
| public Map<ContainerID, |
| CompletableFuture<LegacyReplicationManager.MoveResult>> |
| getInflightMove() { |
| return legacyReplicationManager.getInflightMove(); |
| } |
| |
| public LegacyReplicationManager.MoveScheduler getMoveScheduler() { |
| return legacyReplicationManager.getMoveScheduler(); |
| } |
| |
| @VisibleForTesting |
| public LegacyReplicationManager getLegacyReplicationManager() { |
| return legacyReplicationManager; |
| } |
| |
| public boolean isContainerReplicatingOrDeleting(ContainerID containerID) { |
| return legacyReplicationManager |
| .isContainerReplicatingOrDeleting(containerID); |
| } |
| |
| private ECContainerReplicaCount getECContainerReplicaCount( |
| ContainerInfo containerInfo) throws ContainerNotFoundException { |
| Set<ContainerReplica> replicas = containerManager.getContainerReplicas( |
| containerInfo.containerID()); |
| List<ContainerReplicaOp> pendingOps = |
| containerReplicaPendingOps.getPendingOps(containerInfo.containerID()); |
| // TODO: define maintenance redundancy for EC (HDDS-6975) |
| return new ECContainerReplicaCount(containerInfo, replicas, pendingOps, 0); |
| } |
| |
| /** |
| * Wrap the call to nodeManager.getNodeStatus, catching any |
| * NodeNotFoundException and instead throwing an IllegalStateException. |
| * @param dn The datanodeDetails to obtain the NodeStatus for |
| * @return NodeStatus corresponding to the given Datanode. |
| */ |
| static NodeStatus getNodeStatus(DatanodeDetails dn, NodeManager nm) { |
| try { |
| return nm.getNodeStatus(dn); |
| } catch (NodeNotFoundException e) { |
| throw new IllegalStateException("Unable to find NodeStatus for " + dn, e); |
| } |
| } |
| } |
| |
