/**
 * 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.base.Preconditions;
import com.google.common.base.Strings;
import org.apache.hadoop.hdds.DFSConfigKeysLegacy;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.LayoutVersionProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMRegisteredResponseProto.ErrorCode;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMVersionRequestProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.StorageReportProto;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.VersionInfo;
import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeMetric;
import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeStat;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.ha.SCMContext;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
import org.apache.hadoop.hdds.scm.node.states.NodeAlreadyExistsException;
import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
import org.apache.hadoop.hdds.scm.pipeline.PipelineNotFoundException;
import org.apache.hadoop.hdds.scm.server.SCMStorageConfig;
import org.apache.hadoop.hdds.scm.server.upgrade.FinalizationManager;
import org.apache.hadoop.hdds.server.events.EventPublisher;
import org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager;
import org.apache.hadoop.ipc.Server;
import org.apache.hadoop.metrics2.util.MBeans;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.protocol.VersionResponse;
import org.apache.hadoop.ozone.protocol.commands.CommandForDatanode;
import org.apache.hadoop.ozone.protocol.commands.FinalizeNewLayoutVersionCommand;
import org.apache.hadoop.ozone.protocol.commands.RefreshVolumeUsageCommand;
import org.apache.hadoop.ozone.protocol.commands.RegisteredCommand;
import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
import org.apache.hadoop.ozone.protocol.commands.SetNodeOperationalStateCommand;
import org.apache.hadoop.util.Time;
import org.apache.ratis.protocol.exceptions.NotLeaderException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.management.ObjectName;
import java.io.IOException;
import java.math.RoundingMode;
import java.net.InetAddress;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.Collectors;

import static org.apache.hadoop.hdds.protocol.DatanodeDetails.Port.Name.HTTP;
import static org.apache.hadoop.hdds.protocol.DatanodeDetails.Port.Name.HTTPS;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.IN_SERVICE;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.HEALTHY;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.HEALTHY_READONLY;

/**
 * Maintains information about the Datanodes on SCM side.
 * <p>
 * Heartbeats under SCM is very simple compared to HDFS heartbeatManager.
 * <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 SCMNodeManager implements NodeManager {

  public static final Logger LOG =
      LoggerFactory.getLogger(SCMNodeManager.class);

  private final NodeStateManager nodeStateManager;
  private final VersionInfo version;
  private final CommandQueue commandQueue;
  private final SCMNodeMetrics metrics;
  // Node manager MXBean
  private ObjectName nmInfoBean;
  private final SCMStorageConfig scmStorageConfig;
  private final NetworkTopology clusterMap;
  private final Function<String, String> nodeResolver;
  private final boolean useHostname;
  private final Map<String, Set<UUID>> dnsToUuidMap = new ConcurrentHashMap<>();
  private final int numPipelinesPerMetadataVolume;
  private final int heavyNodeCriteria;
  private final HDDSLayoutVersionManager scmLayoutVersionManager;
  private final EventPublisher scmNodeEventPublisher;
  private final SCMContext scmContext;
  private final Map<SCMCommandProto.Type,
      BiConsumer<DatanodeDetails, SCMCommand<?>>> sendCommandNotifyMap;

  /**
   * Lock used to synchronize some operation in Node manager to ensure a
   * consistent view of the node state.
   */
  private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  private static final String OPESTATE = "OPSTATE";
  private static final String COMSTATE = "COMSTATE";
  private static final String LASTHEARTBEAT = "LASTHEARTBEAT";
  private static final String USEDSPACEPERCENT = "USEDSPACEPERCENT";
  private static final String TOTALCAPACITY = "CAPACITY";
  /**
   * Constructs SCM machine Manager.
   */
  public SCMNodeManager(
      OzoneConfiguration conf,
      SCMStorageConfig scmStorageConfig,
      EventPublisher eventPublisher,
      NetworkTopology networkTopology,
      SCMContext scmContext,
      HDDSLayoutVersionManager layoutVersionManager) {
    this(conf, scmStorageConfig, eventPublisher, networkTopology, scmContext,
        layoutVersionManager, hostname -> null);
  }

  public SCMNodeManager(
      OzoneConfiguration conf,
      SCMStorageConfig scmStorageConfig,
      EventPublisher eventPublisher,
      NetworkTopology networkTopology,
      SCMContext scmContext,
      HDDSLayoutVersionManager layoutVersionManager,
      Function<String, String> nodeResolver) {
    this.scmNodeEventPublisher = eventPublisher;
    this.nodeStateManager = new NodeStateManager(conf, eventPublisher,
        layoutVersionManager, scmContext);
    this.version = VersionInfo.getLatestVersion();
    this.commandQueue = new CommandQueue();
    this.scmStorageConfig = scmStorageConfig;
    this.scmLayoutVersionManager = layoutVersionManager;
    LOG.info("Entering startup safe mode.");
    registerMXBean();
    this.metrics = SCMNodeMetrics.create(this);
    this.clusterMap = networkTopology;
    this.nodeResolver = nodeResolver;
    this.useHostname = conf.getBoolean(
        DFSConfigKeysLegacy.DFS_DATANODE_USE_DN_HOSTNAME,
        DFSConfigKeysLegacy.DFS_DATANODE_USE_DN_HOSTNAME_DEFAULT);
    this.numPipelinesPerMetadataVolume =
        conf.getInt(ScmConfigKeys.OZONE_SCM_PIPELINE_PER_METADATA_VOLUME,
            ScmConfigKeys.OZONE_SCM_PIPELINE_PER_METADATA_VOLUME_DEFAULT);
    String dnLimit = conf.get(ScmConfigKeys.OZONE_DATANODE_PIPELINE_LIMIT);
    this.heavyNodeCriteria = dnLimit == null ? 0 : Integer.parseInt(dnLimit);
    this.scmContext = scmContext;
    this.sendCommandNotifyMap = new HashMap<>();
  }

  @Override
  public void registerSendCommandNotify(SCMCommandProto.Type type,
      BiConsumer<DatanodeDetails, SCMCommand<?>> scmCommand) {
    this.sendCommandNotifyMap.put(type, scmCommand);
  }

  private void registerMXBean() {
    this.nmInfoBean = MBeans.register("SCMNodeManager",
        "SCMNodeManagerInfo", this);
  }

  private void unregisterMXBean() {
    if (this.nmInfoBean != null) {
      MBeans.unregister(this.nmInfoBean);
      this.nmInfoBean = null;
    }
  }

  protected NodeStateManager getNodeStateManager() {
    return nodeStateManager;
  }

  /**
   * Returns all datanode that are in the given state. This function works by
   * taking a snapshot of the current collection and then returning the list
   * from that collection. This means that real map might have changed by the
   * time we return this list.
   *
   * @return List of Datanodes that are known to SCM in the requested state.
   */
  @Override
  public List<DatanodeDetails> getNodes(NodeStatus nodeStatus) {
    return nodeStateManager.getNodes(nodeStatus)
        .stream()
        .map(node -> (DatanodeDetails)node).collect(Collectors.toList());
  }

  /**
   * Returns all datanode that are in the given states. Passing null for one of
   * of the states acts like a wildcard for that state. This function works by
   * taking a snapshot of the current collection and then returning the list
   * from that collection. This means that real map might have changed by the
   * time we return this list.
   *
   * @param opState The operational state of the node
   * @param health The health of the node
   * @return List of Datanodes that are known to SCM in the requested states.
   */
  @Override
  public List<DatanodeDetails> getNodes(
      NodeOperationalState opState, NodeState health) {
    return nodeStateManager.getNodes(opState, health)
        .stream()
        .map(node -> (DatanodeDetails)node).collect(Collectors.toList());
  }

  /**
   * Returns all datanodes that are known to SCM.
   *
   * @return List of DatanodeDetails
   */
  @Override
  public List<DatanodeDetails> getAllNodes() {
    return nodeStateManager.getAllNodes().stream()
        .map(node -> (DatanodeDetails) node).collect(Collectors.toList());
  }

  /**
   * Returns the Number of Datanodes by State they are in.
   *
   * @return count
   */
  @Override
  public int getNodeCount(NodeStatus nodeStatus) {
    return nodeStateManager.getNodeCount(nodeStatus);
  }

  /**
   * Returns the Number of Datanodes by State they are in. Passing null for
   * either of the states acts like a wildcard for that state.
   *
   * @param nodeOpState - The Operational State of the node
   * @param health - The health of the node
   * @return count
   */
  @Override
  public int getNodeCount(NodeOperationalState nodeOpState, NodeState health) {
    return nodeStateManager.getNodeCount(nodeOpState, health);
  }

  /**
   * Returns the node status of a specific node.
   *
   * @param datanodeDetails Datanode Details
   * @return NodeStatus for the node
   */
  @Override
  public NodeStatus getNodeStatus(DatanodeDetails datanodeDetails)
      throws NodeNotFoundException {
    return nodeStateManager.getNodeStatus(datanodeDetails);
  }

  /**
   * Set the operation state of a node.
   * @param datanodeDetails The datanode to set the new state for
   * @param newState The new operational state for the node
   */
  @Override
  public void setNodeOperationalState(DatanodeDetails datanodeDetails,
      NodeOperationalState newState) throws NodeNotFoundException {
    setNodeOperationalState(datanodeDetails, newState, 0);
  }

  /**
   * Set the operation state of a node.
   * @param datanodeDetails The datanode to set the new state for
   * @param newState The new operational state for the node
   * @param opStateExpiryEpocSec Seconds from the epoch when the operational
   *                             state should end. Zero indicates the state
   *                             never end.
   */
  @Override
  public void setNodeOperationalState(DatanodeDetails datanodeDetails,
      NodeOperationalState newState, long opStateExpiryEpocSec)
      throws NodeNotFoundException {
    nodeStateManager.setNodeOperationalState(
        datanodeDetails, newState, opStateExpiryEpocSec);
  }

  /**
   * Closes this stream and releases any system resources associated with it. If
   * the stream is already closed then invoking this method has no effect.
   *
   * @throws IOException if an I/O error occurs
   */
  @Override
  public void close() throws IOException {
    unregisterMXBean();
    metrics.unRegister();
    nodeStateManager.close();
  }

  /**
   * Gets the version info from SCM.
   *
   * @param versionRequest - version Request.
   * @return - returns SCM version info and other required information needed by
   * datanode.
   */
  @Override
  public VersionResponse getVersion(SCMVersionRequestProto versionRequest) {
    return VersionResponse.newBuilder()
        .setVersion(this.version.getVersion())
        .addValue(OzoneConsts.SCM_ID,
            this.scmStorageConfig.getScmId())
        .addValue(OzoneConsts.CLUSTER_ID, this.scmStorageConfig.getClusterID())
        .build();
  }

  @Override
  public RegisteredCommand register(
      DatanodeDetails datanodeDetails, NodeReportProto nodeReport,
      PipelineReportsProto pipelineReportsProto) {
    return register(datanodeDetails, nodeReport, pipelineReportsProto,
        LayoutVersionProto.newBuilder()
            .setMetadataLayoutVersion(
                scmLayoutVersionManager.getMetadataLayoutVersion())
            .setSoftwareLayoutVersion(
                scmLayoutVersionManager.getSoftwareLayoutVersion())
            .build());
  }

  /**
   * Register the node if the node finds that it is not registered with any
   * SCM.
   *
   * @param datanodeDetails - Send datanodeDetails with Node info.
   *                        This function generates and assigns new datanode ID
   *                        for the datanode. This allows SCM to be run
   *                        independent
   *                        of Namenode if required.
   * @param nodeReport      NodeReport.
   * @return SCMRegisteredResponseProto
   */
  @Override
  public RegisteredCommand register(
      DatanodeDetails datanodeDetails, NodeReportProto nodeReport,
      PipelineReportsProto pipelineReportsProto,
      LayoutVersionProto layoutInfo) {
    if (layoutInfo.getSoftwareLayoutVersion() !=
        scmLayoutVersionManager.getSoftwareLayoutVersion()) {
      return RegisteredCommand.newBuilder()
          .setErrorCode(ErrorCode.errorNodeNotPermitted)
          .setDatanode(datanodeDetails)
          .setClusterID(this.scmStorageConfig.getClusterID())
          .build();
    }

    InetAddress dnAddress = Server.getRemoteIp();
    if (dnAddress != null) {
      // Mostly called inside an RPC, update ip
      if (!useHostname) {
        datanodeDetails.setHostName(dnAddress.getHostName());
      }
      datanodeDetails.setIpAddress(dnAddress.getHostAddress());
    }

    final String ipAddress = datanodeDetails.getIpAddress();
    final String hostName = datanodeDetails.getHostName();
    datanodeDetails.setNetworkName(datanodeDetails.getUuidString());
    String networkLocation = nodeResolver.apply(
        useHostname ? hostName : ipAddress);
    if (networkLocation != null) {
      datanodeDetails.setNetworkLocation(networkLocation);
    }

    final UUID uuid = datanodeDetails.getUuid();
    if (!isNodeRegistered(datanodeDetails)) {
      try {
        clusterMap.add(datanodeDetails);
        nodeStateManager.addNode(datanodeDetails, layoutInfo);
        // Check that datanode in nodeStateManager has topology parent set
        DatanodeDetails dn = nodeStateManager.getNode(datanodeDetails);
        Preconditions.checkState(dn.getParent() != null);
        addToDnsToUuidMap(uuid, ipAddress, hostName);
        // Updating Node Report, as registration is successful
        processNodeReport(datanodeDetails, nodeReport);
        LOG.info("Registered datanode: {}", datanodeDetails.toDebugString());
        scmNodeEventPublisher.fireEvent(SCMEvents.NEW_NODE, datanodeDetails);
      } catch (NodeAlreadyExistsException e) {
        if (LOG.isTraceEnabled()) {
          LOG.trace("Datanode is already registered: {}",
              datanodeDetails);
        }
      } catch (NodeNotFoundException e) {
        LOG.error("Cannot find datanode {} from nodeStateManager",
            datanodeDetails);
      }
    } else {
      // Update datanode if it is registered but the ip or hostname changes
      try {
        final DatanodeInfo oldNode = nodeStateManager.getNode(datanodeDetails);
        if (updateDnsToUuidMap(oldNode.getHostName(), oldNode.getIpAddress(),
            hostName, ipAddress, uuid)) {
          LOG.info("Updating datanode {} from {} to {}",
                  datanodeDetails.getUuidString(),
                  oldNode,
                  datanodeDetails);
          clusterMap.update(oldNode, datanodeDetails);
          nodeStateManager.updateNode(datanodeDetails, layoutInfo);
          DatanodeDetails dn = nodeStateManager.getNode(datanodeDetails);
          Preconditions.checkState(dn.getParent() != null);
          processNodeReport(datanodeDetails, nodeReport);
          LOG.info("Updated datanode to: {}", dn);
          scmNodeEventPublisher.fireEvent(SCMEvents.NODE_ADDRESS_UPDATE, dn);
        }
      } catch (NodeNotFoundException e) {
        LOG.error("Cannot find datanode {} from nodeStateManager",
                datanodeDetails);
      }
    }

    return RegisteredCommand.newBuilder().setErrorCode(ErrorCode.success)
        .setDatanode(datanodeDetails)
        .setClusterID(this.scmStorageConfig.getClusterID())
        .build();
  }

  /**
   * Add an entry to the dnsToUuidMap, which maps hostname / IP to the DNs
   * running on that host. As each address can have many DNs running on it,
   * and each host can have multiple addresses,
   * this is a many to many mapping.
   *
   * @param uuid the UUID of the registered node.
   * @param addresses hostname and/or IP of the node
   */
  private synchronized void addToDnsToUuidMap(UUID uuid, String... addresses) {
    for (String addr : addresses) {
      if (!Strings.isNullOrEmpty(addr)) {
        dnsToUuidMap.computeIfAbsent(addr, k -> ConcurrentHashMap.newKeySet())
            .add(uuid);
      }
    }
  }

  private synchronized void removeFromDnsToUuidMap(UUID uuid, String address) {
    if (address != null) {
      Set<UUID> dnSet = dnsToUuidMap.get(address);
      if (dnSet != null && dnSet.remove(uuid) && dnSet.isEmpty()) {
        dnsToUuidMap.remove(address);
      }
    }
  }

  private boolean updateDnsToUuidMap(
      String oldHostName, String oldIpAddress,
      String newHostName, String newIpAddress,
      UUID uuid) {
    final boolean ipChanged = !Objects.equals(oldIpAddress, newIpAddress);
    final boolean hostNameChanged = !Objects.equals(oldHostName, newHostName);
    if (ipChanged || hostNameChanged) {
      synchronized (this) {
        if (ipChanged) {
          removeFromDnsToUuidMap(uuid, oldIpAddress);
          addToDnsToUuidMap(uuid, newIpAddress);
        }
        if (hostNameChanged) {
          removeFromDnsToUuidMap(uuid, oldHostName);
          addToDnsToUuidMap(uuid, newHostName);
        }
      }
    }
    return ipChanged || hostNameChanged;
  }

  /**
   * Send heartbeat to indicate the datanode is alive and doing well.
   *
   * @param datanodeDetails - DatanodeDetailsProto.
   * @return SCMheartbeat response.
   */
  @Override
  public List<SCMCommand> processHeartbeat(DatanodeDetails datanodeDetails,
                                           CommandQueueReportProto queueReport) {
    Preconditions.checkNotNull(datanodeDetails, "Heartbeat is missing " +
        "DatanodeDetails.");
    try {
      nodeStateManager.updateLastHeartbeatTime(datanodeDetails);
      metrics.incNumHBProcessed();
      updateDatanodeOpState(datanodeDetails);
    } catch (NodeNotFoundException e) {
      metrics.incNumHBProcessingFailed();
      LOG.error("SCM trying to process heartbeat from an " +
          "unregistered node {}. Ignoring the heartbeat.", datanodeDetails);
    }
    writeLock().lock();
    try {
      Map<SCMCommandProto.Type, Integer> summary =
          commandQueue.getDatanodeCommandSummary(datanodeDetails.getUuid());
      List<SCMCommand> commands =
          commandQueue.getCommand(datanodeDetails.getUuid());

      // Update the SCMCommand of deleteBlocksCommand Status
      for (SCMCommand<?> command : commands) {
        if (sendCommandNotifyMap.get(command.getType()) != null) {
          sendCommandNotifyMap.get(command.getType())
              .accept(datanodeDetails, command);
        }
      }

      if (queueReport != null) {
        processNodeCommandQueueReport(datanodeDetails, queueReport, summary);
      }
      return commands;
    } finally {
      writeLock().unlock();
    }
  }

  boolean opStateDiffers(DatanodeDetails dnDetails, NodeStatus nodeStatus) {
    return nodeStatus.getOperationalState() != dnDetails.getPersistedOpState()
        || nodeStatus.getOpStateExpiryEpochSeconds()
        != dnDetails.getPersistedOpStateExpiryEpochSec();
  }

  /**
   * This method should only be called when processing the heartbeat.
   *
   * On leader SCM, for a registered node, the information stored in SCM is
   * the source of truth. If the operational state or expiry reported in the
   * datanode heartbeat do not match those store in SCM, queue a command to
   * update the state persisted on the datanode. Additionally, ensure the
   * datanodeDetails stored in SCM match those reported in the heartbeat.
   *
   * On follower SCM, datanode notifies follower SCM its latest operational
   * state or expiry via heartbeat. If the operational state or expiry
   * reported in the datanode heartbeat do not match those stored in SCM,
   * just update the state in follower SCM accordingly.
   *
   * @param reportedDn The DatanodeDetails taken from the node heartbeat.
   * @throws NodeNotFoundException
   */
  protected void updateDatanodeOpState(DatanodeDetails reportedDn)
      throws NodeNotFoundException {
    NodeStatus scmStatus = getNodeStatus(reportedDn);
    if (opStateDiffers(reportedDn, scmStatus)) {
      if (scmContext.isLeader()) {
        LOG.info("Scheduling a command to update the operationalState " +
                "persisted on {} as the reported value ({}, {}) does not " +
                "match the value stored in SCM ({}, {})",
            reportedDn,
            reportedDn.getPersistedOpState(),
            reportedDn.getPersistedOpStateExpiryEpochSec(),
            scmStatus.getOperationalState(),
            scmStatus.getOpStateExpiryEpochSeconds());

        try {
          SCMCommand<?> command = new SetNodeOperationalStateCommand(
              Time.monotonicNow(),
              scmStatus.getOperationalState(),
              scmStatus.getOpStateExpiryEpochSeconds());
          command.setTerm(scmContext.getTermOfLeader());
          addDatanodeCommand(reportedDn.getUuid(), command);
        } catch (NotLeaderException nle) {
          LOG.warn("Skip sending SetNodeOperationalStateCommand,"
              + " since current SCM is not leader.", nle);
          return;
        }
      } else {
        LOG.info("Update the operationalState saved in follower SCM " +
                "for {} as the reported value ({}, {}) does not " +
                "match the value stored in SCM ({}, {})",
            reportedDn,
            reportedDn.getPersistedOpState(),
            reportedDn.getPersistedOpStateExpiryEpochSec(),
            scmStatus.getOperationalState(),
            scmStatus.getOpStateExpiryEpochSeconds());

        setNodeOperationalState(reportedDn, reportedDn.getPersistedOpState(),
            reportedDn.getPersistedOpStateExpiryEpochSec());
      }
    }
    DatanodeDetails scmDnd = nodeStateManager.getNode(reportedDn);
    scmDnd.setPersistedOpStateExpiryEpochSec(
        reportedDn.getPersistedOpStateExpiryEpochSec());
    scmDnd.setPersistedOpState(reportedDn.getPersistedOpState());
  }

  @Override
  public Boolean isNodeRegistered(DatanodeDetails datanodeDetails) {
    try {
      nodeStateManager.getNode(datanodeDetails);
      return true;
    } catch (NodeNotFoundException e) {
      return false;
    }
  }

  /**
   * Process node report.
   *
   * @param datanodeDetails
   * @param nodeReport
   */
  @Override
  public void processNodeReport(DatanodeDetails datanodeDetails,
      NodeReportProto nodeReport) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Processing node report from [datanode={}]",
          datanodeDetails.getHostName());
    }
    if (LOG.isTraceEnabled() && nodeReport != null) {
      LOG.trace("HB is received from [datanode={}]: <json>{}</json>",
          datanodeDetails.getHostName(),
          nodeReport.toString().replaceAll("\n", "\\\\n"));
    }
    try {
      DatanodeInfo datanodeInfo = nodeStateManager.getNode(datanodeDetails);
      if (nodeReport != null) {
        datanodeInfo.updateStorageReports(nodeReport.getStorageReportList());
        datanodeInfo.updateMetaDataStorageReports(nodeReport.
            getMetadataStorageReportList());
        metrics.incNumNodeReportProcessed();
      }
    } catch (NodeNotFoundException e) {
      metrics.incNumNodeReportProcessingFailed();
      LOG.warn("Got node report from unregistered datanode {}",
          datanodeDetails);
    }
  }

  /**
   * Process Layout Version report.
   *
   * @param datanodeDetails
   * @param layoutVersionReport
   */
  @Override
  public void processLayoutVersionReport(DatanodeDetails datanodeDetails,
                                LayoutVersionProto layoutVersionReport) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Processing Layout Version report from [datanode={}]",
          datanodeDetails.getHostName());
    }
    if (LOG.isTraceEnabled()) {
      LOG.trace("HB is received from [datanode={}]: <json>{}</json>",
          datanodeDetails.getHostName(),
          layoutVersionReport.toString().replaceAll("\n", "\\\\n"));
    }

    try {
      nodeStateManager.updateLastKnownLayoutVersion(datanodeDetails,
          layoutVersionReport);
    } catch (NodeNotFoundException e) {
      LOG.error("SCM trying to process Layout Version from an " +
          "unregistered node {}.", datanodeDetails);
      return;
    }

    // Software layout version is hardcoded to the SCM.
    int scmSlv = scmLayoutVersionManager.getSoftwareLayoutVersion();
    int dnSlv = layoutVersionReport.getSoftwareLayoutVersion();
    int dnMlv = layoutVersionReport.getMetadataLayoutVersion();

    // A datanode with a larger software layout version is from a future
    // version of ozone. It should not have been added to the cluster.
    if (dnSlv > scmSlv) {
      LOG.error("Invalid data node in the cluster : {}. " +
              "DataNode SoftwareLayoutVersion = {}, SCM " +
              "SoftwareLayoutVersion = {}",
          datanodeDetails.getHostName(), dnSlv, scmSlv);
    }

    if (FinalizationManager.shouldTellDatanodesToFinalize(
        scmContext.getFinalizationCheckpoint())) {
      // Because we have crossed the MLV_EQUALS_SLV checkpoint, SCM metadata
      // layout version will not change. We can now compare it to the
      // datanodes' metadata layout versions to tell them to finalize.
      int scmMlv = scmLayoutVersionManager.getMetadataLayoutVersion();

      // If the datanode mlv < scm mlv, it can not be allowed to be part of
      // any pipeline. However it can be allowed to join the cluster
      if (dnMlv < scmMlv) {
        LOG.warn("Data node {} can not be used in any pipeline in the " +
                "cluster. " + "DataNode MetadataLayoutVersion = {}, SCM " +
                "MetadataLayoutVersion = {}",
            datanodeDetails.getHostName(), dnMlv, scmMlv);

        FinalizeNewLayoutVersionCommand finalizeCmd =
            new FinalizeNewLayoutVersionCommand(true,
                LayoutVersionProto.newBuilder()
                    .setSoftwareLayoutVersion(dnSlv)
                    .setMetadataLayoutVersion(dnSlv).build());
        if (scmContext.isLeader()) {
          try {
            finalizeCmd.setTerm(scmContext.getTermOfLeader());

            // Send Finalize command to the data node. Its OK to
            // send Finalize command multiple times.
            scmNodeEventPublisher.fireEvent(SCMEvents.DATANODE_COMMAND,
                new CommandForDatanode<>(datanodeDetails.getUuid(),
                    finalizeCmd));
          } catch (NotLeaderException ex) {
            LOG.warn("Skip sending finalize upgrade command since current SCM" +
                " is not leader.", ex);
          }
        }
      }
    }
  }

  /**
   * Process Command Queue Reports from the Datanode Heartbeat.
   *
   * @param datanodeDetails
   * @param commandQueueReportProto
   * @param commandsToBeSent
   */
  private void processNodeCommandQueueReport(DatanodeDetails datanodeDetails,
      CommandQueueReportProto commandQueueReportProto,
      Map<SCMCommandProto.Type, Integer> commandsToBeSent) {
    LOG.debug("Processing Command Queue Report from [datanode={}]",
        datanodeDetails.getHostName());
    if (commandQueueReportProto == null) {
      LOG.debug("The Command Queue Report from [datanode={}] is null",
          datanodeDetails.getHostName());
      return;
    }
    if (LOG.isTraceEnabled()) {
      LOG.trace("Command Queue Report is received from [datanode={}]: " +
          "<json>{}</json>", datanodeDetails.getHostName(),
          commandQueueReportProto.toString().replaceAll("\n", "\\\\n"));
    }
    try {
      DatanodeInfo datanodeInfo = nodeStateManager.getNode(datanodeDetails);
      datanodeInfo.setCommandCounts(commandQueueReportProto,
          commandsToBeSent);
      metrics.incNumNodeCommandQueueReportProcessed();
      scmNodeEventPublisher.fireEvent(
          SCMEvents.DATANODE_COMMAND_COUNT_UPDATED, datanodeDetails);
    } catch (NodeNotFoundException e) {
      metrics.incNumNodeCommandQueueReportProcessingFailed();
      LOG.warn("Got Command Queue Report from unregistered datanode {}",
          datanodeDetails);
    }
  }

  /**
   * Get the number of commands of the given type queued on the datanode at the
   * last heartbeat. If the Datanode has not reported information for the given
   * command type, -1 will be returned.
   * @param cmdType
   * @return The queued count or -1 if no data has been received from the DN.
   */
  @Override
  public int getNodeQueuedCommandCount(DatanodeDetails datanodeDetails,
      SCMCommandProto.Type cmdType) throws NodeNotFoundException {
    readLock().lock();
    try {
      DatanodeInfo datanodeInfo = nodeStateManager.getNode(datanodeDetails);
      return datanodeInfo.getCommandCount(cmdType);
    } finally {
      readLock().unlock();
    }
  }

  /**
   * Get the number of commands of the given type queued in the SCM CommandQueue
   * for the given datanode.
   * @param dnID The UUID of the datanode.
   * @param cmdType The Type of command to query the current count for.
   * @return The count of commands queued, or zero if none.
   */
  @Override
  public int getCommandQueueCount(UUID dnID, SCMCommandProto.Type cmdType) {
    readLock().lock();
    try {
      return commandQueue.getDatanodeCommandCount(dnID, cmdType);
    } finally {
      readLock().unlock();
    }
  }

  /**
   * Get the total number of pending commands of the given type on the given
   * datanode. This includes both the number of commands queued in SCM which
   * will be sent to the datanode on the next heartbeat, and the number of
   * commands reported by the datanode in the last heartbeat.
   * If the datanode has not reported any information for the given command,
   * zero is assumed.
   * @param datanodeDetails The datanode to query.
   * @param cmdType The command Type To query.
   * @return The number of commands of the given type pending on the datanode.
   * @throws NodeNotFoundException
   */
  @Override
  public int getTotalDatanodeCommandCount(DatanodeDetails datanodeDetails,
      SCMCommandProto.Type cmdType) throws NodeNotFoundException {
    readLock().lock();
    try {
      int dnCount = getNodeQueuedCommandCount(datanodeDetails, cmdType);
      if (dnCount == -1) {
        LOG.warn("No command count information for datanode {} and command {}" +
            ". Assuming zero", datanodeDetails, cmdType);
        dnCount = 0;
      }
      return getCommandQueueCount(datanodeDetails.getUuid(), cmdType) + dnCount;
    } finally {
      readLock().unlock();
    }
  }

  /**
   * Get the total number of pending commands of the given types on the given
   * datanode. For each command, this includes both the number of commands
   * queued in SCM which will be sent to the datanode on the next heartbeat,
   * and the number of commands reported by the datanode in the last heartbeat.
   * If the datanode has not reported any information for the given command,
   * zero is assumed.
   * All commands are retrieved under a single read lock, so the counts are
   * consistent.
   * @param datanodeDetails The datanode to query.
   * @param cmdType The list of command Types To query.
   * @return A Map of commandType to Integer with an entry for each command type
   *         passed.
   * @throws NodeNotFoundException
   */
  @Override
  public Map<SCMCommandProto.Type, Integer> getTotalDatanodeCommandCounts(
      DatanodeDetails datanodeDetails, SCMCommandProto.Type... cmdType)
      throws NodeNotFoundException {
    Map<SCMCommandProto.Type, Integer> counts = new HashMap<>();
    readLock().lock();
    try {
      for (SCMCommandProto.Type type : cmdType) {
        counts.put(type, getTotalDatanodeCommandCount(datanodeDetails, type));
      }
      return counts;
    } finally {
      readLock().unlock();
    }
  }

  /**
   * Returns the aggregated node stats.
   *
   * @return the aggregated node stats.
   */
  @Override
  public SCMNodeStat getStats() {
    long capacity = 0L;
    long used = 0L;
    long remaining = 0L;
    long committed = 0L;
    long freeSpaceToSpare = 0L;

    for (SCMNodeStat stat : getNodeStats().values()) {
      capacity += stat.getCapacity().get();
      used += stat.getScmUsed().get();
      remaining += stat.getRemaining().get();
      committed += stat.getCommitted().get();
      freeSpaceToSpare += stat.getFreeSpaceToSpare().get();
    }
    return new SCMNodeStat(capacity, used, remaining, committed,
        freeSpaceToSpare);
  }

  /**
   * Return a map of node stats.
   *
   * @return a map of individual node stats (live/stale but not dead).
   */
  @Override
  public Map<DatanodeDetails, SCMNodeStat> getNodeStats() {

    final Map<DatanodeDetails, SCMNodeStat> nodeStats = new HashMap<>();

    final List<DatanodeInfo> healthyNodes = nodeStateManager
        .getNodes(null, HEALTHY);
    final List<DatanodeInfo> healthyReadOnlyNodes = nodeStateManager
        .getNodes(null, HEALTHY_READONLY);
    final List<DatanodeInfo> staleNodes = nodeStateManager
        .getStaleNodes();
    final List<DatanodeInfo> datanodes = new ArrayList<>(healthyNodes);
    datanodes.addAll(healthyReadOnlyNodes);
    datanodes.addAll(staleNodes);

    for (DatanodeInfo dnInfo : datanodes) {
      SCMNodeStat nodeStat = getNodeStatInternal(dnInfo);
      if (nodeStat != null) {
        nodeStats.put(dnInfo, nodeStat);
      }
    }
    return nodeStats;
  }

  /**
   * Gets a sorted list of most or least used DatanodeUsageInfo containing
   * healthy, in-service nodes. If the specified mostUsed is true, the returned
   * list is in descending order of usage. Otherwise, the returned list is in
   * ascending order of usage.
   *
   * @param mostUsed true if most used, false if least used
   * @return List of DatanodeUsageInfo
   */
  @Override
  public List<DatanodeUsageInfo> getMostOrLeastUsedDatanodes(
      boolean mostUsed) {
    List<DatanodeDetails> healthyNodes =
        getNodes(IN_SERVICE, NodeState.HEALTHY);

    List<DatanodeUsageInfo> datanodeUsageInfoList =
        new ArrayList<>(healthyNodes.size());

    // create a DatanodeUsageInfo from each DatanodeDetails and add it to the
    // list
    for (DatanodeDetails node : healthyNodes) {
      DatanodeUsageInfo datanodeUsageInfo = getUsageInfo(node);
      datanodeUsageInfoList.add(datanodeUsageInfo);
    }

    // sort the list according to appropriate comparator
    if (mostUsed) {
      datanodeUsageInfoList.sort(
          DatanodeUsageInfo.getMostUtilized().reversed());
    } else {
      datanodeUsageInfoList.sort(
          DatanodeUsageInfo.getMostUtilized());
    }
    return datanodeUsageInfoList;
  }

  /**
   * Get the usage info of a specified datanode.
   *
   * @param dn the usage of which we want to get
   * @return DatanodeUsageInfo of the specified datanode
   */
  @Override
  public DatanodeUsageInfo getUsageInfo(DatanodeDetails dn) {
    SCMNodeStat stat = getNodeStatInternal(dn);
    DatanodeUsageInfo usageInfo = new DatanodeUsageInfo(dn, stat);
    try {
      int containerCount = getContainers(dn).size();
      usageInfo.setContainerCount(containerCount);
    } catch (NodeNotFoundException ex) {
      LOG.error("Unknown datanode {}.", dn, ex);
    }
    return usageInfo;
  }

  /**
   * Return the node stat of the specified datanode.
   *
   * @param datanodeDetails - datanode ID.
   * @return node stat if it is live/stale, null if it is decommissioned or
   * doesn't exist.
   */
  @Override
  public SCMNodeMetric getNodeStat(DatanodeDetails datanodeDetails) {
    final SCMNodeStat nodeStat = getNodeStatInternal(datanodeDetails);
    return nodeStat != null ? new SCMNodeMetric(nodeStat) : null;
  }

  private SCMNodeStat getNodeStatInternal(DatanodeDetails datanodeDetails) {
    try {
      long capacity = 0L;
      long used = 0L;
      long remaining = 0L;
      long committed = 0L;
      long freeSpaceToSpare = 0L;

      final DatanodeInfo datanodeInfo = nodeStateManager
          .getNode(datanodeDetails);
      final List<StorageReportProto> storageReportProtos = datanodeInfo
          .getStorageReports();
      for (StorageReportProto reportProto : storageReportProtos) {
        capacity += reportProto.getCapacity();
        used += reportProto.getScmUsed();
        remaining += reportProto.getRemaining();
        committed += reportProto.getCommitted();
        freeSpaceToSpare += reportProto.getFreeSpaceToSpare();
      }
      return new SCMNodeStat(capacity, used, remaining, committed,
          freeSpaceToSpare);
    } catch (NodeNotFoundException e) {
      LOG.warn("Cannot generate NodeStat, datanode {} not found.",
          datanodeDetails.getUuidString());
      return null;
    }
  }

  @Override // NodeManagerMXBean
  public Map<String, Map<String, Integer>> getNodeCount() {
    Map<String, Map<String, Integer>> nodes = new HashMap<>();
    for (NodeOperationalState opState : NodeOperationalState.values()) {
      Map<String, Integer> states = new HashMap<>();
      for (NodeState health : NodeState.values()) {
        states.put(health.name(), 0);
      }
      nodes.put(opState.name(), states);
    }
    for (DatanodeInfo dni : nodeStateManager.getAllNodes()) {
      NodeStatus status = dni.getNodeStatus();
      nodes.get(status.getOperationalState().name())
          .compute(status.getHealth().name(), (k, v) -> v + 1);
    }
    return nodes;
  }

  // We should introduce DISK, SSD, etc., notion in
  // SCMNodeStat and try to use it.
  @Override // NodeManagerMXBean
  public Map<String, Long> getNodeInfo() {
    Map<String, Long> nodeInfo = new HashMap<>();
    // Compute all the possible stats from the enums, and default to zero:
    for (UsageStates s : UsageStates.values()) {
      for (UsageMetrics stat : UsageMetrics.values()) {
        nodeInfo.put(s.label + stat.name(), 0L);
      }
    }
    nodeInfo.put("TotalCapacity", 0L);
    nodeInfo.put("TotalUsed", 0L);

    for (DatanodeInfo node : nodeStateManager.getAllNodes()) {
      String keyPrefix = "";
      NodeStatus status = node.getNodeStatus();
      if (status.isMaintenance()) {
        keyPrefix = UsageStates.MAINT.getLabel();
      } else if (status.isDecommission()) {
        keyPrefix = UsageStates.DECOM.getLabel();
      } else if (status.isAlive()) {
        // Inservice but not dead
        keyPrefix = UsageStates.ONLINE.getLabel();
      } else {
        // dead inservice node, skip it
        continue;
      }
      List<StorageReportProto> storageReportProtos = node.getStorageReports();
      for (StorageReportProto reportProto : storageReportProtos) {
        if (reportProto.getStorageType() ==
            StorageContainerDatanodeProtocolProtos.StorageTypeProto.DISK) {
          nodeInfo.compute(keyPrefix + UsageMetrics.DiskCapacity.name(),
              (k, v) -> v + reportProto.getCapacity());
          nodeInfo.compute(keyPrefix + UsageMetrics.DiskRemaining.name(),
              (k, v) -> v + reportProto.getRemaining());
          nodeInfo.compute(keyPrefix + UsageMetrics.DiskUsed.name(),
              (k, v) -> v + reportProto.getScmUsed());
        } else if (reportProto.getStorageType() ==
            StorageContainerDatanodeProtocolProtos.StorageTypeProto.SSD) {
          nodeInfo.compute(keyPrefix + UsageMetrics.SSDCapacity.name(),
              (k, v) -> v + reportProto.getCapacity());
          nodeInfo.compute(keyPrefix + UsageMetrics.SSDRemaining.name(),
              (k, v) -> v + reportProto.getRemaining());
          nodeInfo.compute(keyPrefix + UsageMetrics.SSDUsed.name(),
              (k, v) -> v + reportProto.getScmUsed());
        }
        nodeInfo.compute("TotalCapacity", (k, v) -> v + reportProto.getCapacity());
        nodeInfo.compute("TotalUsed", (k, v) -> v + reportProto.getScmUsed());
      }
    }
    return nodeInfo;
  }

  @Override
  public Map<String, Map<String, String>> getNodeStatusInfo() {
    Map<String, Map<String, String>> nodes = new HashMap<>();
    for (DatanodeInfo dni : nodeStateManager.getAllNodes()) {
      String hostName = dni.getHostName();
      DatanodeDetails.Port httpPort = dni.getPort(HTTP);
      DatanodeDetails.Port httpsPort = dni.getPort(HTTPS);
      String opstate = "";
      String healthState = "";
      String heartbeatTimeDiff = "";
      if (dni.getNodeStatus() != null) {
        opstate = dni.getNodeStatus().getOperationalState().toString();
        healthState = dni.getNodeStatus().getHealth().toString();
        heartbeatTimeDiff = getLastHeartbeatTimeDiff(dni.getLastHeartbeatTime());
      }
      Map<String, String> map = new HashMap<>();
      map.put(OPESTATE, opstate);
      map.put(COMSTATE, healthState);
      map.put(LASTHEARTBEAT, heartbeatTimeDiff);
      if (httpPort != null) {
        map.put(httpPort.getName().toString(), httpPort.getValue().toString());
      }
      if (httpsPort != null) {
        map.put(httpsPort.getName().toString(),
                  httpsPort.getValue().toString());
      }
      String capacity = calculateStorageCapacity(dni.getStorageReports());
      map.put(TOTALCAPACITY, capacity);
      String[] storagePercentage = calculateStoragePercentage(
          dni.getStorageReports());
      String scmUsedPerc = storagePercentage[0];
      String nonScmUsedPerc = storagePercentage[1];
      map.put(USEDSPACEPERCENT,
          "Ozone: " + scmUsedPerc + "%, other: " + nonScmUsedPerc + "%");
      nodes.put(hostName, map);
    }
    return nodes;
  }

  /**
   * Calculate the storage capacity of the DataNode node.
   * @param storageReports Calculate the storage capacity corresponding
   *                       to the storage collection.
   * @return
   */
  public static String calculateStorageCapacity(
      List<StorageReportProto> storageReports) {
    long capacityByte = 0;
    if (storageReports != null && !storageReports.isEmpty()) {
      for (StorageReportProto storageReport : storageReports) {
        capacityByte += storageReport.getCapacity();
      }
    }

    double ua = capacityByte;
    StringBuilder unit = new StringBuilder("B");
    if (ua > 1024) {
      ua = ua / 1024;
      unit.replace(0, 1, "KB");
    }
    if (ua > 1024) {
      ua = ua / 1024;
      unit.replace(0, 2, "MB");
    }
    if (ua > 1024) {
      ua = ua / 1024;
      unit.replace(0, 2, "GB");
    }
    if (ua > 1024) {
      ua = ua / 1024;
      unit.replace(0, 2, "TB");
    }

    DecimalFormat decimalFormat = new DecimalFormat("#0.0");
    decimalFormat.setRoundingMode(RoundingMode.HALF_UP);
    String capacity = decimalFormat.format(ua);
    return capacity + unit.toString();
  }

  /**
   * Calculate the storage usage percentage of a DataNode node.
   * @param storageReports Calculate the storage percentage corresponding
   *                       to the storage collection.
   * @return
   */
  public static String[] calculateStoragePercentage(
      List<StorageReportProto> storageReports) {
    String[] storagePercentage = new String[2];
    String usedPercentage = "N/A";
    String nonUsedPercentage = "N/A";
    if (storageReports != null && !storageReports.isEmpty()) {
      long capacity = 0;
      long scmUsed = 0;
      long remaining = 0;
      for (StorageReportProto storageReport : storageReports) {
        capacity += storageReport.getCapacity();
        scmUsed += storageReport.getScmUsed();
        remaining += storageReport.getRemaining();
      }
      long scmNonUsed = capacity - scmUsed - remaining;

      DecimalFormat decimalFormat = new DecimalFormat("#0.00");
      decimalFormat.setRoundingMode(RoundingMode.HALF_UP);

      double usedPerc = ((double) scmUsed / capacity) * 100;
      usedPerc = usedPerc > 100.0 ? 100.0 : usedPerc;
      double nonUsedPerc = ((double) scmNonUsed / capacity) * 100;
      nonUsedPerc = nonUsedPerc > 100.0 ? 100.0 : nonUsedPerc;
      usedPercentage = decimalFormat.format(usedPerc);
      nonUsedPercentage = decimalFormat.format(nonUsedPerc);
    }

    storagePercentage[0] = usedPercentage;
    storagePercentage[1] = nonUsedPercentage;
    return storagePercentage;
  }

  /**
   * Based on the current time and the last heartbeat, calculate the time difference
   * and get a string of the relative value. E.g. "2s ago", "1m 2s ago", etc.
   *
   * @return string with the relative value of the time diff.
   */
  public String getLastHeartbeatTimeDiff(long lastHeartbeatTime) {
    long currentTime = Time.monotonicNow();
    long timeDiff = currentTime - lastHeartbeatTime;

    // Time is in ms. Calculate total time in seconds.
    long seconds = TimeUnit.MILLISECONDS.toSeconds(timeDiff);
    // Calculate days, convert the number back to seconds and subtract it from seconds.
    long days = TimeUnit.SECONDS.toDays(seconds);
    seconds -= TimeUnit.DAYS.toSeconds(days);
    // Calculate hours, convert the number back to seconds and subtract it from seconds.
    long hours = TimeUnit.SECONDS.toHours(seconds);
    seconds -= TimeUnit.HOURS.toSeconds(hours);
    // Calculate minutes, convert the number back to seconds and subtract it from seconds.
    long minutes = TimeUnit.SECONDS.toMinutes(seconds);
    seconds -= TimeUnit.MINUTES.toSeconds(minutes);

    StringBuilder stringBuilder = new StringBuilder();
    if (days > 0) {
      stringBuilder.append(days).append("d ");
    }
    if (hours > 0) {
      stringBuilder.append(hours).append("h ");
    }
    if (minutes > 0) {
      stringBuilder.append(minutes).append("m ");
    }
    if (seconds > 0) {
      stringBuilder.append(seconds).append("s ");
    }
    String str = stringBuilder.length() == 0 ? "Just now" : "ago";

    stringBuilder.append(str);

    return stringBuilder.toString();
  }

  private enum UsageMetrics {
    DiskCapacity,
    DiskUsed,
    DiskRemaining,
    SSDCapacity,
    SSDUsed,
    SSDRemaining
  }

  private enum UsageStates {
    ONLINE(""),
    MAINT("Maintenance"),
    DECOM("Decommissioned");

    private final String label;

    public String getLabel() {
      return label;
    }

    UsageStates(String label) {
      this.label = label;
    }
  }

  /**
   * Returns the min of no healthy volumes reported out of the set
   * of datanodes constituting the pipeline.
   */
  @Override
  public int minHealthyVolumeNum(List<DatanodeDetails> dnList) {
    List<Integer> volumeCountList = new ArrayList<>(dnList.size());
    for (DatanodeDetails dn : dnList) {
      try {
        volumeCountList.add(nodeStateManager.getNode(dn).
                getHealthyVolumeCount());
      } catch (NodeNotFoundException e) {
        LOG.warn("Cannot generate NodeStat, datanode {} not found.",
                dn.getUuid());
      }
    }
    Preconditions.checkArgument(!volumeCountList.isEmpty());
    return Collections.min(volumeCountList);
  }

  @Override
  public int totalHealthyVolumeCount() {
    int sum = 0;
    for (DatanodeInfo dn : nodeStateManager.getNodes(IN_SERVICE, HEALTHY)) {
      sum += dn.getHealthyVolumeCount();
    }
    return sum;
  }

  /**
   * Returns the pipeline limit for the datanode.
   * if the datanode pipeline limit is set, consider that as the max
   * pipeline limit.
   * In case, the pipeline limit is not set, the max pipeline limit
   * will be based on the no of raft log volume reported and provided
   * that it has atleast one healthy data volume.
   */
  @Override
  public int pipelineLimit(DatanodeDetails dn) {
    try {
      if (heavyNodeCriteria > 0) {
        return heavyNodeCriteria;
      } else if (nodeStateManager.getNode(dn).getHealthyVolumeCount() > 0) {
        return numPipelinesPerMetadataVolume *
            nodeStateManager.getNode(dn).getMetaDataVolumeCount();
      }
    } catch (NodeNotFoundException e) {
      LOG.warn("Cannot generate NodeStat, datanode {} not found.",
          dn.getUuid());
    }
    return 0;
  }

  /**
   * Returns the pipeline limit for set of datanodes.
   */
  @Override
  public int minPipelineLimit(List<DatanodeDetails> dnList) {
    List<Integer> pipelineCountList = new ArrayList<>(dnList.size());
    for (DatanodeDetails dn : dnList) {
      pipelineCountList.add(pipelineLimit(dn));
    }
    Preconditions.checkArgument(!pipelineCountList.isEmpty());
    return Collections.min(pipelineCountList);
  }

  @Override
  public Collection<DatanodeDetails> getPeerList(DatanodeDetails dn) {
    HashSet<DatanodeDetails> dns = new HashSet<>();
    Preconditions.checkNotNull(dn);
    Set<PipelineID> pipelines =
        nodeStateManager.getPipelineByDnID(dn.getUuid());
    PipelineManager pipelineManager = scmContext.getScm().getPipelineManager();
    if (!pipelines.isEmpty()) {
      pipelines.forEach(id -> {
        try {
          Pipeline pipeline = pipelineManager.getPipeline(id);
          List<DatanodeDetails> peers = pipeline.getNodes();
          dns.addAll(peers);
        } catch (PipelineNotFoundException pnfe) {
          //ignore the pipeline not found exception here
        }
      });
    }
    // renove self node from the set
    dns.remove(dn);
    return dns;
  }

  /**
   * Get set of pipelines a datanode is part of.
   *
   * @param datanodeDetails - datanodeID
   * @return Set of PipelineID
   */
  @Override
  public Set<PipelineID> getPipelines(DatanodeDetails datanodeDetails) {
    return nodeStateManager.getPipelineByDnID(datanodeDetails.getUuid());
  }

  /**
   * Get the count of pipelines a datanodes is associated with.
   * @param datanodeDetails DatanodeDetails
   * @return The number of pipelines
   */
  @Override
  public int getPipelinesCount(DatanodeDetails datanodeDetails) {
    return nodeStateManager.getPipelinesCount(datanodeDetails);
  }

  /**
   * Add pipeline information in the NodeManager.
   *
   * @param pipeline - Pipeline to be added
   */
  @Override
  public void addPipeline(Pipeline pipeline) {
    nodeStateManager.addPipeline(pipeline);
  }

  /**
   * Remove a pipeline information from the NodeManager.
   *
   * @param pipeline - Pipeline to be removed
   */
  @Override
  public void removePipeline(Pipeline pipeline) {
    nodeStateManager.removePipeline(pipeline);
  }

  @Override
  public void addContainer(final DatanodeDetails datanodeDetails,
      final ContainerID containerId)
      throws NodeNotFoundException {
    nodeStateManager.addContainer(datanodeDetails.getUuid(), containerId);
  }

  @Override
  public void removeContainer(final DatanodeDetails datanodeDetails,
                           final ContainerID containerId)
      throws NodeNotFoundException {
    nodeStateManager.removeContainer(datanodeDetails.getUuid(), containerId);
  }

  /**
   * Update set of containers available on a datanode.
   *
   * @param datanodeDetails - DatanodeID
   * @param containerIds    - Set of containerIDs
   * @throws NodeNotFoundException - if datanode is not known. For new datanode
   *                               use addDatanodeInContainerMap call.
   */
  @Override
  public void setContainers(DatanodeDetails datanodeDetails,
      Set<ContainerID> containerIds) throws NodeNotFoundException {
    nodeStateManager.setContainers(datanodeDetails.getUuid(),
        containerIds);
  }

  /**
   * Return set of containerIDs available on a datanode. This is a copy of the
   * set which resides inside NodeManager and hence can be modified without
   * synchronization or side effects.
   *
   * @param datanodeDetails - DatanodeID
   * @return - set of containerIDs
   */
  @Override
  public Set<ContainerID> getContainers(DatanodeDetails datanodeDetails)
      throws NodeNotFoundException {
    return nodeStateManager.getContainers(datanodeDetails.getUuid());
  }

  @Override
  public void addDatanodeCommand(UUID dnId, SCMCommand command) {
    writeLock().lock();
    try {
      this.commandQueue.addCommand(dnId, command);
    } finally {
      writeLock().unlock();
    }
  }

  /**
   * send refresh command to all the healthy datanodes to refresh
   * volume usage info immediately.
   */
  @Override
  public void refreshAllHealthyDnUsageInfo() {
    RefreshVolumeUsageCommand refreshVolumeUsageCommand =
        new RefreshVolumeUsageCommand();
    try {
      refreshVolumeUsageCommand.setTerm(scmContext.getTermOfLeader());
    } catch (NotLeaderException nle) {
      LOG.warn("Skip sending refreshVolumeUsage command,"
          + " since current SCM is not leader.", nle);
      return;
    }
    getNodes(IN_SERVICE, HEALTHY).forEach(datanode ->
        addDatanodeCommand(datanode.getUuid(), refreshVolumeUsageCommand));
  }

  /**
   * This method is called by EventQueue whenever someone adds a new
   * DATANODE_COMMAND to the Queue.
   *
   * @param commandForDatanode DatanodeCommand
   * @param ignored            publisher
   */
  @Override
  public void onMessage(CommandForDatanode commandForDatanode,
                        EventPublisher ignored) {
    addDatanodeCommand(commandForDatanode.getDatanodeId(),
        commandForDatanode.getCommand());
  }

  @Override
  public List<SCMCommand> getCommandQueue(UUID dnID) {
    // Getting the queue actually clears it and returns the commands, so this
    // is a write operation and not a read as the method name suggests.
    writeLock().lock();
    try {
      return commandQueue.getCommand(dnID);
    } finally {
      writeLock().unlock();
    }
  }

  /**
   * Given datanode uuid, returns the DatanodeDetails for the node.
   *
   * @param uuid node host address
   * @return the given datanode, or null if not found
   */
  @Override
  public DatanodeDetails getNodeByUuid(String uuid) {
    return uuid != null && !uuid.isEmpty()
        ? getNodeByUuid(UUID.fromString(uuid))
        : null;
  }

  @Override
  public DatanodeDetails getNodeByUuid(UUID uuid) {
    if (uuid == null) {
      return null;
    }

    try {
      return nodeStateManager.getNode(uuid);
    } catch (NodeNotFoundException e) {
      LOG.warn("Cannot find node for uuid {}", uuid);
      return null;
    }
  }

  /**
   * Given datanode address(Ipaddress or hostname), return a list of
   * DatanodeDetails for the datanodes registered on that address.
   *
   * @param address datanode address
   * @return the given datanode, or empty list if none found
   */
  @Override
  public List<DatanodeDetails> getNodesByAddress(String address) {
    List<DatanodeDetails> results = new LinkedList<>();
    if (Strings.isNullOrEmpty(address)) {
      LOG.warn("address is null");
      return results;
    }
    Set<UUID> uuids = dnsToUuidMap.get(address);
    if (uuids == null) {
      LOG.warn("Cannot find node for address {}", address);
      return results;
    }

    for (UUID uuid : uuids) {
      try {
        results.add(nodeStateManager.getNode(uuid));
      } catch (NodeNotFoundException e) {
        LOG.warn("Cannot find node for uuid {}", uuid);
      }
    }
    return results;
  }

  /**
   * Get cluster map as in network topology for this node manager.
   * @return cluster map
   */
  @Override
  public NetworkTopology getClusterNetworkTopologyMap() {
    return clusterMap;
  }

  /**
   * For the given node, retried the last heartbeat time.
   * @param datanodeDetails DatanodeDetails of the node.
   * @return The last heartbeat time in milliseconds or -1 if the node does not
   *         exist.
   */
  @Override
  public long getLastHeartbeat(DatanodeDetails datanodeDetails) {
    try {
      DatanodeInfo node = nodeStateManager.getNode(datanodeDetails);
      return node.getLastHeartbeatTime();
    } catch (NodeNotFoundException e) {
      return -1;
    }
  }

  /**
   * Test utility to stop heartbeat check process.
   *
   * @return ScheduledFuture of next scheduled check that got cancelled.
   */
  @VisibleForTesting
  ScheduledFuture pauseHealthCheck() {
    return nodeStateManager.pause();
  }

  /**
   * Test utility to resume the paused heartbeat check process.
   *
   * @return ScheduledFuture of the next scheduled check
   */
  @VisibleForTesting
  ScheduledFuture unpauseHealthCheck() {
    return nodeStateManager.unpause();
  }

  /**
   * Test utility to get the count of skipped heartbeat check iterations.
   *
   * @return count of skipped heartbeat check iterations
   */
  @VisibleForTesting
  long getSkippedHealthChecks() {
    return nodeStateManager.getSkippedHealthChecks();
  }

  /**
   * @return  HDDSLayoutVersionManager
   */
  @VisibleForTesting
  @Override
  public HDDSLayoutVersionManager getLayoutVersionManager() {
    return scmLayoutVersionManager;
  }

  @VisibleForTesting
  @Override
  public void forceNodesToHealthyReadOnly() {
    nodeStateManager.forceNodesToHealthyReadOnly();
  }

  private ReentrantReadWriteLock.WriteLock writeLock() {
    return lock.writeLock();
  }

  private ReentrantReadWriteLock.ReadLock readLock() {
    return lock.readLock();
  }

  /**
   * This API allows removal of only DECOMMISSIONED and DEAD nodes from NodeManager data structures and cleanup memory.
   * This API call is having a pre-condition before removal of node like following resources to be removed:
   *   --- all pipelines for datanode should be closed.
   *   --- all containers for datanode should be closed.
   *   --- remove all containers replicas maintained by datanode.
   *   --- clears all SCM DeletedBlockLog transaction records associated with datanode.
   *
   * @param datanodeDetails
   * @throws NodeNotFoundException
   */
  @Override
  public void removeNode(DatanodeDetails datanodeDetails) throws NodeNotFoundException, IOException {
    writeLock().lock();
    try {
      NodeStatus nodeStatus = this.getNodeStatus(datanodeDetails);
      if (datanodeDetails.isDecommissioned() || nodeStatus.isDead()) {
        if (clusterMap.contains(datanodeDetails)) {
          clusterMap.remove(datanodeDetails);
        }
        nodeStateManager.removeNode(datanodeDetails);
        removeFromDnsToUuidMap(datanodeDetails.getUuid(), datanodeDetails.getIpAddress());
        final List<SCMCommand> cmdList = getCommandQueue(datanodeDetails.getUuid());
        LOG.info("Clearing command queue of size {} for DN {}", cmdList.size(), datanodeDetails);
      } else {
        LOG.warn("Node not decommissioned or dead, cannot remove: {}", datanodeDetails);
      }
    } finally {
      writeLock().unlock();
    }
  }
}