/**
 * 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.server;

import javax.management.ObjectName;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.RemovalListener;
import com.google.protobuf.BlockingService;

import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.annotation.InterfaceAudience;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
import org.apache.hadoop.hdds.scm.PipelineChoosePolicy;
import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.container.ContainerManagerImpl;
import org.apache.hadoop.hdds.scm.container.ContainerManagerV2;
import org.apache.hadoop.hdds.scm.ha.SCMContext;
import org.apache.hadoop.hdds.scm.ha.SCMHAManager;
import org.apache.hadoop.hdds.scm.ha.SCMHAManagerImpl;
import org.apache.hadoop.hdds.scm.ha.SCMHANodeDetails;
import org.apache.hadoop.hdds.scm.ha.SCMServiceManager;
import org.apache.hadoop.hdds.scm.ha.SCMNodeDetails;
import org.apache.hadoop.hdds.scm.ha.SCMRatisServerImpl;
import org.apache.hadoop.hdds.scm.ha.SCMHAUtils;
import org.apache.hadoop.hdds.scm.ha.SequenceIdGenerator;
import org.apache.hadoop.hdds.scm.ScmInfo;
import org.apache.hadoop.hdds.utils.HAUtils;
import org.apache.hadoop.hdds.utils.HddsServerUtil;
import org.apache.hadoop.hdds.scm.ScmConfig;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.block.BlockManager;
import org.apache.hadoop.hdds.scm.block.BlockManagerImpl;
import org.apache.hadoop.hdds.scm.block.DeletedBlockLogImplV2;
import org.apache.hadoop.hdds.scm.block.PendingDeleteHandler;
import org.apache.hadoop.hdds.scm.command.CommandStatusReportHandler;
import org.apache.hadoop.hdds.scm.container.CloseContainerEventHandler;
import org.apache.hadoop.hdds.scm.container.ContainerActionsHandler;
import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import org.apache.hadoop.hdds.scm.container.ContainerReportHandler;
import org.apache.hadoop.hdds.scm.container.IncrementalContainerReportHandler;
import org.apache.hadoop.hdds.scm.container.ReplicationManager;
import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementPolicyFactory;
import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementMetrics;
import org.apache.hadoop.hdds.scm.container.placement.metrics.ContainerStat;
import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMMetrics;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes;
import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStore;
import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStoreImpl;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
import org.apache.hadoop.hdds.scm.net.NetworkTopologyImpl;
import org.apache.hadoop.hdds.scm.node.DeadNodeHandler;
import org.apache.hadoop.hdds.scm.node.NewNodeHandler;
import org.apache.hadoop.hdds.scm.node.StartDatanodeAdminHandler;
import org.apache.hadoop.hdds.scm.node.NonHealthyToHealthyNodeHandler;
import org.apache.hadoop.hdds.scm.node.NodeManager;
import org.apache.hadoop.hdds.scm.node.NodeReportHandler;
import org.apache.hadoop.hdds.scm.node.SCMNodeManager;
import org.apache.hadoop.hdds.scm.node.StaleNodeHandler;
import org.apache.hadoop.hdds.scm.node.NodeDecommissionManager;
import org.apache.hadoop.hdds.scm.pipeline.PipelineActionHandler;
import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
import org.apache.hadoop.hdds.scm.pipeline.PipelineReportHandler;
import org.apache.hadoop.hdds.scm.pipeline.PipelineManagerV2Impl;
import org.apache.hadoop.hdds.scm.pipeline.choose.algorithms.PipelineChoosePolicyFactory;
import org.apache.hadoop.hdds.scm.safemode.SCMSafeModeManager;
import org.apache.hadoop.hdds.security.exception.SCMSecurityException;
import org.apache.hadoop.hdds.security.x509.SecurityConfig;
import org.apache.hadoop.hdds.security.x509.certificate.authority.CertificateServer;
import org.apache.hadoop.hdds.security.x509.certificate.authority.DefaultCAServer;
import org.apache.hadoop.hdds.security.x509.certificate.utils.CertificateCodec;
import org.apache.hadoop.hdds.server.ServiceRuntimeInfoImpl;
import org.apache.hadoop.hdds.server.events.EventPublisher;
import org.apache.hadoop.hdds.server.events.EventQueue;
import org.apache.hadoop.hdds.utils.HddsVersionInfo;
import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.metrics2.MetricsSystem;
import org.apache.hadoop.metrics2.util.MBeans;
import org.apache.hadoop.ozone.OzoneConfigKeys;
import org.apache.hadoop.ozone.OzoneSecurityUtil;
import org.apache.hadoop.ozone.common.Storage.StorageState;
import org.apache.hadoop.ozone.lease.LeaseManager;
import org.apache.hadoop.ozone.lock.LockManager;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
import org.apache.hadoop.security.authentication.client.AuthenticationException;
import org.apache.hadoop.util.JvmPauseMonitor;
import org.apache.ratis.grpc.GrpcTlsConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_SCM_WATCHER_TIMEOUT_DEFAULT;
import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ADMINISTRATORS_WILDCARD;
import static org.apache.hadoop.ozone.OzoneConsts.CRL_SEQUENCE_ID_KEY;

/**
 * StorageContainerManager is the main entry point for the service that
 * provides information about
 * which SCM nodes host containers.
 *
 * <p>DataNodes report to StorageContainerManager using heartbeat messages.
 * SCM allocates containers
 * and returns a pipeline.
 *
 * <p>A client once it gets a pipeline (a list of datanodes) will connect to
 * the datanodes and create a container, which then can be used to store data.
 */
@InterfaceAudience.LimitedPrivate({"HDFS", "CBLOCK", "OZONE", "HBASE"})
public final class StorageContainerManager extends ServiceRuntimeInfoImpl
    implements SCMMXBean, OzoneStorageContainerManager {

  private static final Logger LOG = LoggerFactory
      .getLogger(StorageContainerManager.class);

  /**
   * SCM metrics.
   */
  private static SCMMetrics metrics;

  /*
   * RPC Endpoints exposed by SCM.
   */
  private final SCMDatanodeProtocolServer datanodeProtocolServer;
  private final SCMBlockProtocolServer blockProtocolServer;
  private final SCMClientProtocolServer clientProtocolServer;
  private SCMSecurityProtocolServer securityProtocolServer;

  /*
   * State Managers of SCM.
   */
  private NodeManager scmNodeManager;
  private PipelineManager pipelineManager;
  private ContainerManagerV2 containerManager;
  private BlockManager scmBlockManager;
  private final SCMStorageConfig scmStorageConfig;
  private NodeDecommissionManager scmDecommissionManager;

  private SCMMetadataStore scmMetadataStore;
  private SCMHAManager scmHAManager;
  private SCMContext scmContext;
  private SequenceIdGenerator sequenceIdGen;

  private final EventQueue eventQueue;
  private final SCMServiceManager serviceManager;

  /*
   * HTTP endpoint for JMX access.
   */
  private StorageContainerManagerHttpServer httpServer;
  /**
   * SCM super user.
   */
  private final Collection<String> scmAdminUsernames;
  /**
   * SCM mxbean.
   */
  private ObjectName scmInfoBeanName;
  /**
   * Key = DatanodeUuid, value = ContainerStat.
   */
  private final Cache<String, ContainerStat> containerReportCache;

  private ReplicationManager replicationManager;

  private final LeaseManager<Long> commandWatcherLeaseManager;

  private SCMSafeModeManager scmSafeModeManager;
  private CertificateServer certificateServer;
  private GrpcTlsConfig grpcTlsConfig;

  private JvmPauseMonitor jvmPauseMonitor;
  private final OzoneConfiguration configuration;
  private SCMContainerMetrics scmContainerMetrics;
  private SCMContainerPlacementMetrics placementMetrics;
  private MetricsSystem ms;

  /**
   *  Network topology Map.
   */
  private NetworkTopology clusterMap;
  private PipelineChoosePolicy pipelineChoosePolicy;

  private final SCMHANodeDetails scmHANodeDetails;

  /**
   * Creates a new StorageContainerManager. Configuration will be
   * updated with information on the actual listening addresses used
   * for RPC servers.
   *
   * @param conf configuration
   */
  private StorageContainerManager(OzoneConfiguration conf)
      throws IOException, AuthenticationException {
    // default empty configurator means default managers will be used.
    this(conf, new SCMConfigurator());
  }

  /**
   * This constructor offers finer control over how SCM comes up.
   * To use this, user needs to create a SCMConfigurator and set various
   * managers that user wants SCM to use, if a value is missing then SCM will
   * use the default value for that manager.
   *
   * @param conf - Configuration
   * @param configurator - configurator
   */
  private StorageContainerManager(OzoneConfiguration conf,
                                  SCMConfigurator configurator)
      throws IOException, AuthenticationException  {
    super(HddsVersionInfo.HDDS_VERSION_INFO);

    Objects.requireNonNull(configurator, "configurator cannot not be null");
    Objects.requireNonNull(conf, "configuration cannot not be null");

    scmHANodeDetails = SCMHANodeDetails.loadSCMHAConfig(conf);

    configuration = conf;
    initMetrics();
    containerReportCache = buildContainerReportCache();

    /**
     * It is assumed the scm --init command creates the SCM Storage Config.
     */
    scmStorageConfig = new SCMStorageConfig(conf);
    if (scmStorageConfig.getState() != StorageState.INITIALIZED) {
      String errMsg = "Please make sure you have run \'ozone scm --init\' " +
          "command to generate all the required metadata to " +
          scmStorageConfig.getStorageDir();
      if (SCMHAUtils.isSCMHAEnabled(conf)) {
        errMsg += " or make sure you have run \'ozone scm --bootstrap\' cmd to "
            + "add the SCM to existing SCM HA group";
      }
      LOG.error(errMsg + ".");
      throw new SCMException("SCM not initialized due to storage config " +
          "failure.", ResultCodes.SCM_NOT_INITIALIZED);
    }

    /**
     * Important : This initialization sequence is assumed by some of our tests.
     * The testSecureOzoneCluster assumes that security checks have to be
     * passed before any artifacts like SCM DB is created. So please don't
     * add any other initialization above the Security checks please.
     */
    if (OzoneSecurityUtil.isSecurityEnabled(conf)) {
      loginAsSCMUser(conf);
    }

    // Creates the SCM DBs or opens them if it exists.
    // A valid pointer to the store is required by all the other services below.
    initalizeMetadataStore(conf, configurator);
    // Authenticate SCM if security is enabled, this initialization can only
    // be done after the metadata store is initialized.
    if (OzoneSecurityUtil.isSecurityEnabled(conf)) {
      initializeCAnSecurityProtocol(conf, configurator);
    } else {
      // if no Security, we do not create a Certificate Server at all.
      // This allows user to boot SCM without security temporarily
      // and then come back and enable it without any impact.
      certificateServer = null;
      securityProtocolServer = null;
    }

    eventQueue = new EventQueue();
    serviceManager = new SCMServiceManager();

    long watcherTimeout =
        conf.getTimeDuration(ScmConfigKeys.HDDS_SCM_WATCHER_TIMEOUT,
            HDDS_SCM_WATCHER_TIMEOUT_DEFAULT, TimeUnit.MILLISECONDS);
    commandWatcherLeaseManager = new LeaseManager<>("CommandWatcher",
        watcherTimeout);
    initializeSystemManagers(conf, configurator);

    CloseContainerEventHandler closeContainerHandler =
        new CloseContainerEventHandler(
            pipelineManager, containerManager, scmContext);
    NodeReportHandler nodeReportHandler =
        new NodeReportHandler(scmNodeManager);
    PipelineReportHandler pipelineReportHandler =
        new PipelineReportHandler(
            scmSafeModeManager, pipelineManager, scmContext, conf);
    CommandStatusReportHandler cmdStatusReportHandler =
        new CommandStatusReportHandler();

    NewNodeHandler newNodeHandler = new NewNodeHandler(pipelineManager,
        scmDecommissionManager, conf, serviceManager);
    StaleNodeHandler staleNodeHandler =
        new StaleNodeHandler(scmNodeManager, pipelineManager, conf);
    DeadNodeHandler deadNodeHandler = new DeadNodeHandler(scmNodeManager,
        pipelineManager, containerManager);
    StartDatanodeAdminHandler datanodeStartAdminHandler =
        new StartDatanodeAdminHandler(scmNodeManager, pipelineManager);
    NonHealthyToHealthyNodeHandler nonHealthyToHealthyNodeHandler =
        new NonHealthyToHealthyNodeHandler(conf, serviceManager);
    ContainerActionsHandler actionsHandler = new ContainerActionsHandler();
    PendingDeleteHandler pendingDeleteHandler =
        new PendingDeleteHandler(scmBlockManager.getSCMBlockDeletingService());

    ContainerReportHandler containerReportHandler =
        new ContainerReportHandler(
            scmNodeManager, containerManager, scmContext, conf);

    IncrementalContainerReportHandler incrementalContainerReportHandler =
        new IncrementalContainerReportHandler(
            scmNodeManager, containerManager, scmContext);

    PipelineActionHandler pipelineActionHandler =
        new PipelineActionHandler(pipelineManager, scmContext, conf);

    scmAdminUsernames = conf.getTrimmedStringCollection(OzoneConfigKeys
        .OZONE_ADMINISTRATORS);
    String scmUsername = UserGroupInformation.getCurrentUser().getUserName();
    if (!scmAdminUsernames.contains(scmUsername)) {
      scmAdminUsernames.add(scmUsername);
    }

    datanodeProtocolServer = new SCMDatanodeProtocolServer(conf, this,
        eventQueue);
    blockProtocolServer = new SCMBlockProtocolServer(conf, this);
    clientProtocolServer = new SCMClientProtocolServer(conf, this);
    eventQueue.addHandler(SCMEvents.DATANODE_COMMAND, scmNodeManager);
    eventQueue.addHandler(SCMEvents.RETRIABLE_DATANODE_COMMAND, scmNodeManager);
    eventQueue.addHandler(SCMEvents.NODE_REPORT, nodeReportHandler);
    eventQueue.addHandler(SCMEvents.CONTAINER_REPORT, containerReportHandler);
    eventQueue.addHandler(SCMEvents.INCREMENTAL_CONTAINER_REPORT,
        incrementalContainerReportHandler);
    eventQueue.addHandler(SCMEvents.CONTAINER_ACTIONS, actionsHandler);
    eventQueue.addHandler(SCMEvents.CLOSE_CONTAINER, closeContainerHandler);
    eventQueue.addHandler(SCMEvents.NEW_NODE, newNodeHandler);
    eventQueue.addHandler(SCMEvents.STALE_NODE, staleNodeHandler);
    eventQueue.addHandler(SCMEvents.NON_HEALTHY_TO_HEALTHY_NODE,
        nonHealthyToHealthyNodeHandler);
    eventQueue.addHandler(SCMEvents.DEAD_NODE, deadNodeHandler);
    eventQueue.addHandler(SCMEvents.START_ADMIN_ON_NODE,
        datanodeStartAdminHandler);
    eventQueue.addHandler(SCMEvents.CMD_STATUS_REPORT, cmdStatusReportHandler);
    eventQueue
        .addHandler(SCMEvents.PENDING_DELETE_STATUS, pendingDeleteHandler);
    eventQueue.addHandler(SCMEvents.DELETE_BLOCK_STATUS,
        (DeletedBlockLogImplV2) scmBlockManager.getDeletedBlockLog());
    eventQueue.addHandler(SCMEvents.PIPELINE_ACTIONS, pipelineActionHandler);
    eventQueue.addHandler(SCMEvents.PIPELINE_REPORT, pipelineReportHandler);

    // Emit initial safe mode status, as now handlers are registered.
    scmSafeModeManager.emitSafeModeStatus();

    registerMXBean();
    registerMetricsSource(this);
  }

  public OzoneConfiguration getConfiguration() {
    return configuration;
  }

  /**
   * Create an SCM instance based on the supplied configuration.
   *
   * @param conf        HDDS configuration
   * @param configurator SCM configurator
   * @return SCM instance
   * @throws IOException, AuthenticationException
   */
  public static StorageContainerManager createSCM(
      OzoneConfiguration conf, SCMConfigurator configurator)
      throws IOException, AuthenticationException {
    return new StorageContainerManager(conf, configurator);
  }

  /**
   * Create an SCM instance based on the supplied configuration.
   *
   * @param conf        HDDS configuration
   * @return SCM instance
   * @throws IOException, AuthenticationException
   */
  public static StorageContainerManager createSCM(OzoneConfiguration conf)
      throws IOException, AuthenticationException {
    return createSCM(conf, new SCMConfigurator());
  }

  /**
   * This function initializes the following managers. If the configurator
   * specifies a value, we will use it, else we will use the default value.
   *
   *  Node Manager
   *  Pipeline Manager
   *  Container Manager
   *  Block Manager
   *  Replication Manager
   *  Safe Mode Manager
   *
   * @param conf - Ozone Configuration.
   * @param configurator - A customizer which allows different managers to be
   *                    used if needed.
   * @throws IOException - on Failure.
   */
  private void initializeSystemManagers(OzoneConfiguration conf,
                                       SCMConfigurator configurator)
      throws IOException {
    if (configurator.getNetworkTopology() != null) {
      clusterMap = configurator.getNetworkTopology();
    } else {
      clusterMap = new NetworkTopologyImpl(conf);
    }

    if (configurator.getSCMHAManager() != null) {
      scmHAManager = configurator.getSCMHAManager();
    } else {
      scmHAManager = new SCMHAManagerImpl(conf, this);
    }

    // inline upgrade for SequenceIdGenerator
    SequenceIdGenerator.upgradeToSequenceId(scmMetadataStore);
    // Distributed sequence id generator
    sequenceIdGen = new SequenceIdGenerator(
        conf, scmHAManager, scmMetadataStore.getSequenceIdTable());

    if (configurator.getScmContext() != null) {
      scmContext = configurator.getScmContext();
    } else {
      // When term equals SCMContext.INVALID_TERM, the isLeader() check
      // and getTermOfLeader() will always pass.
      long term = SCMHAUtils.isSCMHAEnabled(conf) ? 0 : SCMContext.INVALID_TERM;
      // non-leader of term 0, in safe mode, preCheck not completed.
      scmContext = new SCMContext.Builder()
          .setLeader(false)
          .setTerm(term)
          .setIsInSafeMode(true)
          .setIsPreCheckComplete(false)
          .setSCM(this)
          .build();
    }

    if(configurator.getScmNodeManager() != null) {
      scmNodeManager = configurator.getScmNodeManager();
    } else {
      scmNodeManager = new SCMNodeManager(
          conf, scmStorageConfig, eventQueue, clusterMap, scmContext);
    }

    placementMetrics = SCMContainerPlacementMetrics.create();
    PlacementPolicy containerPlacementPolicy =
        ContainerPlacementPolicyFactory.getPolicy(conf, scmNodeManager,
            clusterMap, true, placementMetrics);

    if (configurator.getPipelineManager() != null) {
      pipelineManager = configurator.getPipelineManager();
    } else {
      pipelineManager =
          PipelineManagerV2Impl.newPipelineManager(
              conf,
              scmHAManager,
              scmNodeManager,
              scmMetadataStore.getPipelineTable(),
              eventQueue,
              scmContext,
              serviceManager);
    }

    if (configurator.getContainerManager() != null) {
      containerManager = configurator.getContainerManager();
    } else {
      containerManager = new ContainerManagerImpl(conf, scmHAManager,
          sequenceIdGen, pipelineManager, scmMetadataStore.getContainerTable());
    }

    pipelineChoosePolicy = PipelineChoosePolicyFactory.getPolicy(conf);
    if (configurator.getScmBlockManager() != null) {
      scmBlockManager = configurator.getScmBlockManager();
    } else {
      scmBlockManager = new BlockManagerImpl(conf, this);
    }
    if (configurator.getReplicationManager() != null) {
      replicationManager = configurator.getReplicationManager();
    }  else {
      replicationManager = new ReplicationManager(
          conf,
          containerManager,
          containerPlacementPolicy,
          eventQueue,
          scmContext,
          serviceManager,
          new LockManager<>(conf),
          scmNodeManager);
    }
    if(configurator.getScmSafeModeManager() != null) {
      scmSafeModeManager = configurator.getScmSafeModeManager();
    } else {
      scmSafeModeManager = new SCMSafeModeManager(conf,
          containerManager.getContainers(),
          pipelineManager, eventQueue, serviceManager, scmContext);
    }
    scmDecommissionManager = new NodeDecommissionManager(conf, scmNodeManager,
        containerManager, eventQueue, replicationManager);
  }

  /**
   * If security is enabled we need to have the Security Protocol and a
   * default CA. This function initializes those values based on the
   * configurator.
   *
   * @param conf - Config
   * @param configurator - configurator
   * @throws IOException - on Failure
   * @throws AuthenticationException - on Failure
   */
  private void initializeCAnSecurityProtocol(OzoneConfiguration conf,
      SCMConfigurator configurator) throws IOException {
    if(configurator.getCertificateServer() != null) {
      this.certificateServer = configurator.getCertificateServer();
    } else {
      // This assumes that SCM init has run, and DB metadata stores are created.
      certificateServer = initializeCertificateServer(
          getScmStorageConfig().getClusterID(),
          getScmStorageConfig().getScmId());
    }
    // TODO: Support Intermediary CAs in future.
    certificateServer.init(new SecurityConfig(conf),
        CertificateServer.CAType.SELF_SIGNED_CA);
    securityProtocolServer = new SCMSecurityProtocolServer(conf,
        certificateServer, this);

    grpcTlsConfig = createTlsClientConfigForSCM(new SecurityConfig(conf),
            certificateServer);
  }

  public CertificateServer getCertificateServer() {
    return certificateServer;
  }

  // For Internal gRPC client from SCM to DN with gRPC TLS
  static GrpcTlsConfig createTlsClientConfigForSCM(SecurityConfig conf,
      CertificateServer certificateServer) throws IOException {
    if (conf.isSecurityEnabled() && conf.isGrpcTlsEnabled()) {
      try {
        X509Certificate caCert =
            CertificateCodec.getX509Certificate(
                certificateServer.getCACertificate());
        return new GrpcTlsConfig(null, null,
            caCert, false);
      } catch (CertificateException ex) {
        throw new SCMSecurityException("Fail to find SCM CA certificate.", ex);
      }
    }
    return null;
  }
  /**
   * Init the metadata store based on the configurator.
   * @param conf - Config
   * @param configurator - configurator
   * @throws IOException - on Failure
   */
  private void initalizeMetadataStore(OzoneConfiguration conf,
                                      SCMConfigurator configurator)
      throws IOException {
    if(configurator.getMetadataStore() != null) {
      scmMetadataStore = configurator.getMetadataStore();
    } else {
      scmMetadataStore = new SCMMetadataStoreImpl(conf);
    }
  }

  /**
   * Login as the configured user for SCM.
   *
   * @param conf
   */
  private void loginAsSCMUser(ConfigurationSource conf)
      throws IOException, AuthenticationException {
    if (LOG.isDebugEnabled()) {
      ScmConfig scmConfig = configuration.getObject(ScmConfig.class);
      LOG.debug("Ozone security is enabled. Attempting login for SCM user. "
              + "Principal: {}, keytab: {}",
          scmConfig.getKerberosPrincipal(),
          scmConfig.getKerberosKeytab());
    }

    Configuration hadoopConf =
        LegacyHadoopConfigurationSource.asHadoopConfiguration(conf);
    if (SecurityUtil.getAuthenticationMethod(hadoopConf).equals(
        AuthenticationMethod.KERBEROS)) {
      UserGroupInformation.setConfiguration(hadoopConf);
      InetSocketAddress socAddr = HddsServerUtil
          .getScmBlockClientBindAddress(conf);
      SecurityUtil.login(hadoopConf,
            ScmConfig.ConfigStrings.HDDS_SCM_KERBEROS_KEYTAB_FILE_KEY,
            ScmConfig.ConfigStrings.HDDS_SCM_KERBEROS_PRINCIPAL_KEY,
            socAddr.getHostName());
    } else {
      throw new AuthenticationException(SecurityUtil.getAuthenticationMethod(
          hadoopConf) + " authentication method not support. "
          + "SCM user login failed.");
    }
    LOG.info("SCM login successful.");
  }

  /**
   * This function creates/initializes a certificate server as needed.
   * This function is idempotent, so calling this again and again after the
   * server is initialized is not a problem.
   *
   * @param clusterID - Cluster ID
   * @param scmID     - SCM ID
   */
  private CertificateServer initializeCertificateServer(String clusterID,
      String scmID) throws IOException {
    // TODO: Support Certificate Server loading via Class Name loader.
    // So it is easy to use different Certificate Servers if needed.
    String subject = "scm@" + InetAddress.getLocalHost().getHostName();
    if(this.scmMetadataStore == null) {
      LOG.error("Cannot initialize Certificate Server without a valid meta " +
          "data layer.");
      throw new SCMException("Cannot initialize CA without a valid metadata " +
          "store", ResultCodes.SCM_NOT_INITIALIZED);
    }
    SCMCertStore certStore = new SCMCertStore(this.scmMetadataStore,
        getLastSequenceIdForCRL());
    return new DefaultCAServer(subject, clusterID, scmID, certStore);
  }

  long getLastSequenceIdForCRL() throws IOException {
    Long sequenceId =
        scmMetadataStore.getCRLSequenceIdTable().get(CRL_SEQUENCE_ID_KEY);
    // If the CRL_SEQUENCE_ID_KEY does not exist in DB return 0 so that new
    // CRL requests can have sequence id starting from 1.
    if (sequenceId == null) {
      return 0L;
    }
    // If there exists a last sequence id in the DB, the new incoming
    // CRL requests must have sequence ids greater than the one stored in the DB
    return sequenceId;
  }

  /**
   * Builds a message for logging startup information about an RPC server.
   *
   * @param description RPC server description
   * @param addr        RPC server listening address
   * @return server startup message
   */
  public static String buildRpcServerStartMessage(String description,
                                                  InetSocketAddress addr) {
    return addr != null
        ? String.format("%s is listening at %s", description, addr.toString())
        : String.format("%s not started", description);
  }

  /**
   * Starts an RPC server, if configured.
   *
   * @param conf configuration
   * @param addr configured address of RPC server
   * @param protocol RPC protocol provided by RPC server
   * @param instance RPC protocol implementation instance
   * @param handlerCount RPC server handler count
   * @return RPC server
   * @throws IOException if there is an I/O error while creating RPC server
   */
  public static RPC.Server startRpcServer(
      OzoneConfiguration conf,
      InetSocketAddress addr,
      Class<?> protocol,
      BlockingService instance,
      int handlerCount)
      throws IOException {
    RPC.Server rpcServer =
        new RPC.Builder(conf)
            .setProtocol(protocol)
            .setInstance(instance)
            .setBindAddress(addr.getHostString())
            .setPort(addr.getPort())
            .setNumHandlers(handlerCount)
            .setVerbose(false)
            .setSecretManager(null)
            .build();

    HddsServerUtil.addPBProtocol(conf, protocol, instance, rpcServer);
    return rpcServer;
  }

  /**
   * Routine to bootstrap the StorageContainerManager. This will connect to a
   * running SCM instance which has valid cluster id and fetch the cluster id
   * from there.
   *
   * TODO: once SCM HA security is enabled, CSR cerificates will be fetched from
   * running scm leader instance as well.
   *
   * @param conf OzoneConfiguration
   * @return true if SCM bootstrap is successful, false otherwise.
   * @throws IOException if init fails due to I/O error
   */
  public static boolean scmBootstrap(OzoneConfiguration conf)
      throws IOException {
    if (!SCMHAUtils.isSCMHAEnabled(conf)) {
      LOG.error("Bootstrap is not supported without SCM HA.");
      return false;
    }
    // The node here will try to fetch the cluster id from any of existing
    // running SCM instances.
    SCMHANodeDetails scmhaNodeDetails = SCMHANodeDetails.loadSCMHAConfig(conf);
    String primordialSCM = SCMHAUtils.getPrimordialSCM(conf);
    String selfNodeId = scmhaNodeDetails.getLocalNodeDetails().getNodeId();
    if (primordialSCM != null && SCMHAUtils.isPrimordialSCM(conf, selfNodeId)) {
      LOG.info(
          "SCM bootstrap command can only be executed in non-Primordial SCM "
              + "{}, self id {} "
              + "Ignoring it.", primordialSCM, selfNodeId);
      return true;
    }
    OzoneConfiguration config =
        SCMHAUtils.removeSelfId(conf,
            scmhaNodeDetails.getLocalNodeDetails().getNodeId());
    final ScmInfo scmInfo = HAUtils.getScmInfo(config);
    SCMStorageConfig scmStorageConfig = new SCMStorageConfig(conf);
    final String persistedClusterId = scmStorageConfig.getClusterID();
    final String fetchedId = scmInfo.getClusterId();
    Preconditions.checkNotNull(fetchedId);
    StorageState state = scmStorageConfig.getState();
    if (state == StorageState.INITIALIZED) {
      Preconditions.checkNotNull(scmStorageConfig.getScmId());
      if (!fetchedId.equals(persistedClusterId)) {
        LOG.error(
            "Could not bootstrap as SCM is already initialized with cluster "
                + "id {} but cluster id for existing leader SCM instance "
                + "is {}", persistedClusterId, fetchedId);
        return false;
      }
    } else {
      try {
        scmStorageConfig.setClusterId(fetchedId);
        // It will write down the cluster Id fetched from already
        // running SCM as well as the local SCM Id.

        // SCM Node info containing hostname to scm Id mappings
        // will be persisted into the version file once this node gets added
        // to existing SCM ring post node regular start up.
        scmStorageConfig.initialize();
      } catch (IOException ioe) {
        LOG.error("Could not initialize SCM version file", ioe);
        return false;
      }
    }
    return true;
  }

  /**
   * Routine to set up the Version info for StorageContainerManager.
   *
   * @param conf OzoneConfiguration
   * @return true if SCM initialization is successful, false otherwise.
   * @throws IOException if init fails due to I/O error
   */
  public static boolean scmInit(OzoneConfiguration conf,
      String clusterId) throws IOException {
    SCMStorageConfig scmStorageConfig = new SCMStorageConfig(conf);
    StorageState state = scmStorageConfig.getState();
    final SCMHANodeDetails haDetails = SCMHANodeDetails.loadSCMHAConfig(conf);
    String primordialSCM = SCMHAUtils.getPrimordialSCM(conf);
    String selfNodeId = haDetails.getLocalNodeDetails().getNodeId();
    if (primordialSCM != null && !SCMHAUtils
        .isPrimordialSCM(conf, selfNodeId)) {
      LOG.info(
          "SCM init command can only be executed in Primordial SCM {}, "
              + "self id {} "
              + "Ignoring it.", primordialSCM, selfNodeId);
      return true;
    }
    if (state != StorageState.INITIALIZED) {
      try {
        if (clusterId != null && !clusterId.isEmpty()) {
          // clusterId must be an UUID
          Preconditions.checkNotNull(UUID.fromString(clusterId));
          scmStorageConfig.setClusterId(clusterId);
        }
        scmStorageConfig.initialize();
        if (SCMHAUtils.isSCMHAEnabled(conf)) {
          SCMRatisServerImpl.initialize(scmStorageConfig.getClusterID(),
              scmStorageConfig.getScmId(), haDetails.getLocalNodeDetails(),
              conf);
        }
        LOG.info("SCM initialization succeeded. Current cluster id for sd={}"
                + "; cid={}; layoutVersion={}; scmId={}",
            scmStorageConfig.getStorageDir(), scmStorageConfig.getClusterID(),
            scmStorageConfig.getLayoutVersion(), scmStorageConfig.getScmId());
        return true;
      } catch (IOException ioe) {
        LOG.error("Could not initialize SCM version file", ioe);
        return false;
      }
    } else {
      clusterId = scmStorageConfig.getClusterID();
      LOG.info("SCM already initialized. Reusing existing cluster id for sd={}"
              + ";cid={};layoutVersion={}", scmStorageConfig.getStorageDir(),
          clusterId, scmStorageConfig.getLayoutVersion());
      if (SCMHAUtils.isSCMHAEnabled(conf)) {
        SCMRatisServerImpl.reinitialize(clusterId, scmStorageConfig.getScmId(),
            haDetails.getLocalNodeDetails(), conf);
      }
      return true;
    }
  }

  /**
   * Initialize SCM metrics.
   */
  public static void initMetrics() {
    metrics = SCMMetrics.create();
  }

  /**
   * Return SCM metrics instance.
   */
  public static SCMMetrics getMetrics() {
    return metrics == null ? SCMMetrics.create() : metrics;
  }

  public SCMStorageConfig getScmStorageConfig() {
    return scmStorageConfig;
  }

  public SCMDatanodeProtocolServer getDatanodeProtocolServer() {
    return datanodeProtocolServer;
  }

  public SCMBlockProtocolServer getBlockProtocolServer() {
    return blockProtocolServer;
  }

  public SCMClientProtocolServer getClientProtocolServer() {
    return clientProtocolServer;
  }

  public SCMSecurityProtocolServer getSecurityProtocolServer() {
    return securityProtocolServer;
  }

  /**
   * Initialize container reports cache that sent from datanodes.
   */
  @SuppressWarnings("UnstableApiUsage")
  private Cache<String, ContainerStat> buildContainerReportCache() {
    return
        CacheBuilder.newBuilder()
            .expireAfterAccess(Long.MAX_VALUE, TimeUnit.MILLISECONDS)
            .maximumSize(Integer.MAX_VALUE)
            .removalListener((
                RemovalListener<String, ContainerStat>) removalNotification -> {
                  synchronized (containerReportCache) {
                    ContainerStat stat = removalNotification.getValue();
                    if (stat != null) {
                      // TODO: Are we doing the right thing here?
                      // remove invalid container report
                      metrics.decrContainerStat(stat);
                    }
                    if (LOG.isDebugEnabled()) {
                      LOG.debug("Remove expired container stat entry for " +
                          "datanode: {}.", removalNotification.getKey());
                    }
                  }
                })
            .build();
  }

  private void registerMXBean() {
    final Map<String, String> jmxProperties = new HashMap<>();
    jmxProperties.put("component", "ServerRuntime");
    this.scmInfoBeanName = HddsUtils.registerWithJmxProperties(
        "StorageContainerManager", "StorageContainerManagerInfo",
        jmxProperties, this);
  }

  private void registerMetricsSource(SCMMXBean scmMBean) {
    scmContainerMetrics = SCMContainerMetrics.create(scmMBean);
  }

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

  @VisibleForTesting
  public ContainerInfo getContainerInfo(long containerID) throws
      IOException {
    return containerManager.getContainer(ContainerID.valueOf(containerID));
  }

  /**
   * Returns listening address of StorageLocation Protocol RPC server.
   *
   * @return listen address of StorageLocation RPC server
   */
  @VisibleForTesting
  public InetSocketAddress getClientRpcAddress() {
    return getClientProtocolServer().getClientRpcAddress();
  }

  @Override
  public String getClientRpcPort() {
    InetSocketAddress addr = getClientRpcAddress();
    return addr == null ? "0" : Integer.toString(addr.getPort());
  }

  /**
   * Returns listening address of StorageDatanode Protocol RPC server.
   *
   * @return Address where datanode are communicating.
   */
  @Override
  public InetSocketAddress getDatanodeRpcAddress() {
    return getDatanodeProtocolServer().getDatanodeRpcAddress();
  }

  @Override
  public SCMNodeDetails getScmNodeDetails() {
    return scmHANodeDetails.getLocalNodeDetails();
  }

  public SCMHANodeDetails getSCMHANodeDetails() {
    return scmHANodeDetails;
  }

  @Override
  public String getDatanodeRpcPort() {
    InetSocketAddress addr = getDatanodeRpcAddress();
    return addr == null ? "0" : Integer.toString(addr.getPort());
  }

  /**
   * Start service.
   */
  @Override
  public void start() throws IOException {
    if (LOG.isInfoEnabled()) {
      LOG.info(buildRpcServerStartMessage(
          "StorageContainerLocationProtocol RPC server",
          getClientRpcAddress()));
    }

    scmHAManager.start();

    ms = HddsServerUtil
        .initializeMetrics(configuration, "StorageContainerManager");

    commandWatcherLeaseManager.start();
    getClientProtocolServer().start();

    if (LOG.isInfoEnabled()) {
      LOG.info(buildRpcServerStartMessage("ScmBlockLocationProtocol RPC " +
          "server", getBlockProtocolServer().getBlockRpcAddress()));
    }
    getBlockProtocolServer().start();

    if (LOG.isInfoEnabled()) {
      LOG.info(buildRpcServerStartMessage("ScmDatanodeProtocl RPC " +
          "server", getDatanodeProtocolServer().getDatanodeRpcAddress()));
    }
    getDatanodeProtocolServer().start();
    if (getSecurityProtocolServer() != null) {
      getSecurityProtocolServer().start();
    }

    scmBlockManager.start();

    // Start jvm monitor
    jvmPauseMonitor = new JvmPauseMonitor();
    jvmPauseMonitor.init(configuration);
    jvmPauseMonitor.start();

    try {
      httpServer = new StorageContainerManagerHttpServer(configuration, this);
      httpServer.start();
    } catch (Exception ex) {
      // SCM HttpServer start-up failure should be non-fatal
      LOG.error("SCM HttpServer failed to start.", ex);
    }

    setStartTime();
  }

  /**
   * Stop service.
   */
  @Override
  public void stop() {
    try {
      LOG.info("Stopping Replication Manager Service.");
      replicationManager.stop();
    } catch (Exception ex) {
      LOG.error("Replication manager service stop failed.", ex);
    }

    try {
      LOG.info("Stopping the Datanode Admin Monitor.");
      scmDecommissionManager.stop();
    } catch (Exception ex) {
      LOG.error("The Datanode Admin Monitor failed to stop", ex);
    }

    try {
      LOG.info("Stopping Lease Manager of the command watchers");
      commandWatcherLeaseManager.shutdown();
    } catch (Exception ex) {
      LOG.error("Lease Manager of the command watchers stop failed");
    }

    try {
      LOG.info("Stopping datanode service RPC server");
      getDatanodeProtocolServer().stop();

    } catch (Exception ex) {
      LOG.error("Storage Container Manager datanode RPC stop failed.", ex);
    }

    try {
      LOG.info("Stopping block service RPC server");
      getBlockProtocolServer().stop();
    } catch (Exception ex) {
      LOG.error("Storage Container Manager blockRpcServer stop failed.", ex);
    }

    try {
      LOG.info("Stopping the StorageContainerLocationProtocol RPC server");
      getClientProtocolServer().stop();
    } catch (Exception ex) {
      LOG.error("Storage Container Manager clientRpcServer stop failed.", ex);
    }

    try {
      LOG.info("Stopping Storage Container Manager HTTP server.");
      httpServer.stop();
    } catch (Exception ex) {
      LOG.error("Storage Container Manager HTTP server stop failed.", ex);
    }

    if (getSecurityProtocolServer() != null) {
      getSecurityProtocolServer().stop();
    }

    try {
      LOG.info("Stopping Block Manager Service.");
      scmBlockManager.stop();
    } catch (Exception ex) {
      LOG.error("SCM block manager service stop failed.", ex);
    }

    if (containerReportCache != null) {
      containerReportCache.invalidateAll();
      containerReportCache.cleanUp();
    }

    if (metrics != null) {
      metrics.unRegister();
    }

    unregisterMXBean();
    if (scmContainerMetrics != null) {
      scmContainerMetrics.unRegister();
    }
    if (placementMetrics != null) {
      placementMetrics.unRegister();
    }

    // Event queue must be stopped before the DB store is closed at the end.
    try {
      LOG.info("Stopping SCM Event Queue.");
      eventQueue.close();
    } catch (Exception ex) {
      LOG.error("SCM Event Queue stop failed", ex);
    }

    if (jvmPauseMonitor != null) {
      jvmPauseMonitor.stop();
    }

    try {
      scmHAManager.shutdown();
    } catch (Exception ex) {
      LOG.error("SCM HA Manager stop failed", ex);
    }

    IOUtils.cleanupWithLogger(LOG, containerManager);
    IOUtils.cleanupWithLogger(LOG, pipelineManager);

    try {
      scmMetadataStore.stop();
    } catch (Exception ex) {
      LOG.error("SCM Metadata store stop failed", ex);
    }

    if (ms != null) {
      ms.stop();
    }

    scmSafeModeManager.stop();
  }
  
  /**
   * Wait until service has completed shutdown.
   */
  @Override
  public void join() {
    try {
      getBlockProtocolServer().join();
      getClientProtocolServer().join();
      getDatanodeProtocolServer().join();
      if (getSecurityProtocolServer() != null) {
        getSecurityProtocolServer().join();
      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      LOG.info("Interrupted during StorageContainerManager join.");
    }
  }

  /**
   * Returns the Number of Datanodes that are communicating with SCM.
   *
   * @param nodestate Healthy, Dead etc.
   * @return int -- count
   */
  public int getNodeCount(NodeState nodestate) {
    // TODO - decomm - this probably needs to accept opState and health
    return scmNodeManager.getNodeCount(null, nodestate);
  }

  /**
   * Returns the node decommission manager.
   *
   * @return NodeDecommissionManager The decommission manger for the used by
   *         scm
   */
  public NodeDecommissionManager getScmDecommissionManager() {
    return scmDecommissionManager;
  }

  /**
   * Returns SCMHAManager.
   */
  public SCMHAManager getScmHAManager() {
    return scmHAManager;
  }

  /**
   * Returns SCM container manager.
   */
  @VisibleForTesting
  @Override
  public ContainerManagerV2 getContainerManager() {
    return containerManager;
  }

  /**
   * Returns node manager.
   *
   * @return - Node Manager
   */
  @VisibleForTesting
  @Override
  public NodeManager getScmNodeManager() {
    return scmNodeManager;
  }

  /**
   * Returns pipeline manager.
   *
   * @return - Pipeline Manager
   */
  @VisibleForTesting
  @Override
  public PipelineManager getPipelineManager() {
    return pipelineManager;
  }

  @VisibleForTesting
  @Override
  public BlockManager getScmBlockManager() {
    return scmBlockManager;
  }

  @VisibleForTesting
  public SCMSafeModeManager getScmSafeModeManager() {
    return scmSafeModeManager;
  }

  @VisibleForTesting
  @Override
  public ReplicationManager getReplicationManager() {
    return replicationManager;
  }

  /**
   * Check if the current scm is the leader and ready for accepting requests.
   * @return - if the current scm is the leader and is ready.
   */
  public boolean checkLeader() {
    // For NON-HA setup, the node will always be the leader
    if (!SCMHAUtils.isSCMHAEnabled(configuration)) {
      Preconditions.checkArgument(scmContext.isLeader());
      return true;
    } else {
      // FOR HA setup, the node has to be the leader and ready to serve
      // requests.
      return scmContext.isLeader() && getScmHAManager().getRatisServer()
          .getDivision().getInfo().isLeaderReady();
    }
  }

  public void checkAdminAccess(String remoteUser) throws IOException {
    if (remoteUser != null && !scmAdminUsernames.contains(remoteUser) &&
        !scmAdminUsernames.contains(OZONE_ADMINISTRATORS_WILDCARD)) {
      throw new IOException(
          "Access denied for user " + remoteUser + ". Superuser privilege " +
              "is required.");
    }
  }

  /**
   * Invalidate container stat entry for given datanode.
   *
   * @param datanodeUuid
   */
  public void removeContainerReport(String datanodeUuid) {
    synchronized (containerReportCache) {
      containerReportCache.invalidate(datanodeUuid);
    }
  }

  /**
   * Get container stat of specified datanode.
   *
   * @param datanodeUuid
   * @return
   */
  public ContainerStat getContainerReport(String datanodeUuid) {
    ContainerStat stat = null;
    synchronized (containerReportCache) {
      stat = containerReportCache.getIfPresent(datanodeUuid);
    }

    return stat;
  }

  /**
   * Returns a view of the container stat entries. Modifications made to the
   * map will directly
   * affect the cache.
   *
   * @return
   */
  public ConcurrentMap<String, ContainerStat> getContainerReportCache() {
    return containerReportCache.asMap();
  }

  @Override
  public Map<String, String> getContainerReport() {
    Map<String, String> id2StatMap = new HashMap<>();
    synchronized (containerReportCache) {
      ConcurrentMap<String, ContainerStat> map = containerReportCache.asMap();
      for (Map.Entry<String, ContainerStat> entry : map.entrySet()) {
        id2StatMap.put(entry.getKey(), entry.getValue().toJsonString());
      }
    }

    return id2StatMap;
  }

  /**
   * Returns live safe mode container threshold.
   *
   * @return String
   */
  @Override
  public double getSafeModeCurrentContainerThreshold() {
    return getCurrentContainerThreshold();
  }

  /**
   * Returns safe mode status.
   * @return boolean
   */
  @Override
  public boolean isInSafeMode() {
    return scmSafeModeManager.getInSafeMode();
  }

  /**
   * Returns EventPublisher.
   */
  public EventPublisher getEventQueue() {
    return eventQueue;
  }

  /**
   * Returns SCMContext.
   */
  public SCMContext getScmContext() {
    return scmContext;
  }

  /**
   * Returns SequenceIdGen.
   */
  public SequenceIdGenerator getSequenceIdGen() {
    return sequenceIdGen;
  }

  /**
   * Returns SCMServiceManager.
   */
  public SCMServiceManager getSCMServiceManager() {
    return serviceManager;
  }

  /**
   * Force SCM out of safe mode.
   */
  public boolean exitSafeMode() {
    scmSafeModeManager.exitSafeMode(eventQueue);
    return true;
  }

  @VisibleForTesting
  public double getCurrentContainerThreshold() {
    return scmSafeModeManager.getCurrentContainerThreshold();
  }

  @Override
  public Map<String, Integer> getContainerStateCount() {
    Map<String, Integer> nodeStateCount = new HashMap<>();
    for (HddsProtos.LifeCycleState state : HddsProtos.LifeCycleState.values()) {
      nodeStateCount.put(state.toString(),
          containerManager.getContainers(state).size());
    }
    return nodeStateCount;
  }

  /**
   * Returns the SCM metadata Store.
   * @return SCMMetadataStore
   */
  public SCMMetadataStore getScmMetadataStore() {
    return scmMetadataStore;
  }

  /**
   * Returns the SCM network topology cluster.
   * @return NetworkTopology
   */
  public NetworkTopology getClusterMap() {
    return this.clusterMap;
  }

  /**
   * Get the safe mode status of all rules.
   *
   * @return map of rule statuses.
   */
  public Map<String, Pair<Boolean, String>> getRuleStatus() {
    return scmSafeModeManager.getRuleStatus();
  }

  @Override
  public Map<String, String[]> getSafeModeRuleStatus() {
    Map<String, String[]> map = new HashMap<>();
    for (Map.Entry<String, Pair<Boolean, String>> entry :
        scmSafeModeManager.getRuleStatus().entrySet()) {
      String[] status =
          {entry.getValue().getRight(), entry.getValue().getLeft().toString()};
      map.put(entry.getKey(), status);
    }
    return map;
  }

  public PipelineChoosePolicy getPipelineChoosePolicy() {
    return this.pipelineChoosePolicy;
  }

  @Override
  public String getScmId() {
    return getScmStorageConfig().getScmId();
  }

  @Override
  public String getClusterId() {
    return getScmStorageConfig().getClusterID();
  }

  /**
   * Return the node Id of this SCM.
   * @return node Id.
   */
  public String getSCMNodeId() {
    return scmHANodeDetails.getLocalNodeDetails().getNodeId();
  }
}