content: /*
   * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
   * (the "License"). You may not use this work except in compliance with the License, which is
   * available at www.apache.org/licenses/LICENSE-2.0
   *
   * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
   * either express or implied, as more fully set forth in the License.
   *
   * See the NOTICE file distributed with this work for information regarding copyright ownership.
   */
  
  package alluxio.master.file.replication;
  
  import alluxio.AlluxioURI;
  import alluxio.Constants;
  import alluxio.client.job.JobMasterClientPool;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.BlockInfoException;
  import alluxio.exception.FileDoesNotExistException;
  import alluxio.exception.JobDoesNotExistException;
  import alluxio.exception.status.ResourceExhaustedException;
  import alluxio.exception.status.UnavailableException;
  import alluxio.heartbeat.HeartbeatExecutor;
  import alluxio.job.plan.replicate.DefaultReplicationHandler;
  import alluxio.job.plan.replicate.ReplicationHandler;
  import alluxio.job.wire.Status;
  import alluxio.master.SafeModeManager;
  import alluxio.master.block.BlockMaster;
  import alluxio.master.file.meta.InodeFile;
  import alluxio.master.file.meta.InodeTree;
  import alluxio.master.file.meta.InodeTree.LockPattern;
  import alluxio.master.file.meta.LockedInodePath;
  import alluxio.master.file.meta.PersistenceState;
  import alluxio.master.journal.NoopJournalContext;
  import alluxio.metrics.MetricKey;
  import alluxio.metrics.MetricsSystem;
  import alluxio.util.logging.SamplingLogger;
  import alluxio.wire.BlockInfo;
  import alluxio.wire.BlockLocation;
  
  import com.google.common.collect.HashBiMap;
  import com.google.common.collect.ImmutableSet;
  import org.apache.commons.lang3.tuple.ImmutableTriple;
  import org.apache.commons.lang3.tuple.Triple;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import javax.annotation.concurrent.ThreadSafe;
  
  /**
   * The executor to check block replication level periodically and handle over-replicated and
   * under-replicated blocks correspondingly.
   */
  @ThreadSafe
  public final class ReplicationChecker implements HeartbeatExecutor {
    private static final Logger LOG = LoggerFactory.getLogger(ReplicationChecker.class);
    private static final Logger SAMPLING_LOG = new SamplingLogger(LOG, 10L * Constants.MINUTE_MS);
  
    /** Maximum number of active jobs to be submitted to the job service. **/
    private final int mMaxActiveJobs;
  
    /** Handler to the inode tree. */
    private final InodeTree mInodeTree;
    /** Handler to the block master. */
    private final BlockMaster mBlockMaster;
    /** Handler for adjusting block replication level. */
    private final ReplicationHandler mReplicationHandler;
    /** Manager of master safe mode state. */
    private final SafeModeManager mSafeModeManager;
  
    private final HashBiMap<Long, Long> mActiveJobToInodeID;
  
    private enum Mode {
      EVICT,
      REPLICATE
    }
  
    /**
     * Constructs a new {@link ReplicationChecker} using default (job service) handler to replicate
     * and evict blocks.
     *
     * @param inodeTree inode tree of the filesystem master
     * @param blockMaster block master
     * @param safeModeManager manager of master safe mode state
     * @param jobMasterClientPool job master client pool
     */
    public ReplicationChecker(InodeTree inodeTree, BlockMaster blockMaster,
        SafeModeManager safeModeManager, JobMasterClientPool jobMasterClientPool) {
      this(inodeTree, blockMaster, safeModeManager,
          new DefaultReplicationHandler(jobMasterClientPool));
    }
  
    /**
     * Constructs a new {@link ReplicationChecker} with specified replicate and evict handlers (for
     * unit testing).
     *
     * @param inodeTree inode tree of the filesystem master
     * @param blockMaster block master
     * @param safeModeManager manager of master safe mode state
     * @param replicationHandler handler to replicate blocks
     */
    public ReplicationChecker(InodeTree inodeTree, BlockMaster blockMaster,
        SafeModeManager safeModeManager, ReplicationHandler replicationHandler) {
      mInodeTree = inodeTree;
      mBlockMaster = blockMaster;
      mSafeModeManager = safeModeManager;
      mReplicationHandler = replicationHandler;
  
      // Do not use more than 10% of the job service
      mMaxActiveJobs = Math.max(1,
          (int) (Configuration.getLong(PropertyKey.JOB_MASTER_JOB_CAPACITY) * 0.1));
      mActiveJobToInodeID = HashBiMap.create();
      MetricsSystem.registerCachedGaugeIfAbsent(
          MetricsSystem.getMetricName(MetricKey.MASTER_REPLICA_MGMT_ACTIVE_JOB_SIZE.getName()),
          mActiveJobToInodeID::size);
    }
  
    private boolean shouldRun() {
      // In unit tests there may not be workers, but we still want the ReplicationChecker to execute
      if (Configuration.getBoolean(PropertyKey.TEST_MODE)) {
        return true;
      }
      if (mSafeModeManager.isInSafeMode() || mBlockMaster.getWorkerCount() == 0) {
        LOG.debug("Skip the ReplicationChecker in safe mode and when there are no workers");
        return false;
      }
      return true;
    }
  
    /**
     * {@inheritDoc}
     *
     * During this heartbeat, this class will check:
     * <p>
     * (1) Is there any block from the pinned files becomes under replicated (i.e., the number of
     * existing copies is smaller than the target replication min for this file, possibly due to node
     * failures), and schedule replicate jobs to increase the replication level when found;
     *
     * (2) Is there any blocks over replicated, schedule evict jobs to reduce the replication level.
     */
    @Override
    public void heartbeat(long timeLimitMs) throws InterruptedException {
      if (!shouldRun()) {
        return;
      }
      final Set<Long> activeJobIds = new HashSet<>();
      try {
        if (!mActiveJobToInodeID.isEmpty()) {
          final List<Long> activeEvictJobIds =
              mReplicationHandler.findJobs("Evict",
                  ImmutableSet.of(Status.RUNNING, Status.CREATED));
          final List<Long> activeMoveJobIds =
              mReplicationHandler.findJobs("Move",
                  ImmutableSet.of(Status.RUNNING, Status.CREATED));
          final List<Long> activeReplicateJobIds =
              mReplicationHandler.findJobs("Replicate",
                  ImmutableSet.of(Status.RUNNING, Status.CREATED));
  
          activeJobIds.addAll(activeEvictJobIds);
          activeJobIds.addAll(activeMoveJobIds);
          activeJobIds.addAll(activeReplicateJobIds);
          mActiveJobToInodeID.keySet().removeIf(jobId -> !activeJobIds.contains(jobId));
        }
      } catch (IOException e) {
        // It is possible the job master process is not answering rpcs,
        // log but do not throw the exception
        // which will kill the replication checker thread.
        LOG.debug("Failed to contact job master to get updated list of replication jobs", e);
      }
  
      Set<Long> inodes;
  
      // Check the set of files that could possibly be under-replicated
      inodes = mInodeTree.getPinIdSet();
      check(inodes, mReplicationHandler, Mode.REPLICATE);
  
      // Check the set of files that could possibly be over-replicated
      inodes = mInodeTree.getReplicationLimitedFileIds();
      check(inodes, mReplicationHandler, Mode.EVICT);
  
      // Check the set of files that could possibly be mis-replicated
      inodes = mInodeTree.getPinIdSet();
      checkMisreplicated(inodes, mReplicationHandler);
    }
  
    @Override
    public void close() {
      // Nothing to clean up
    }
  
    /**
     * Find a set of from and to locations for our moveBlock operation.
     * After the move, the file will be in the desired pinned medium at least minReplication times.
     *
     * @param file the file to scan for misplaced blocks
     * @param blockInfo blockInfo containing information about the block locations
     * @return a map that maps from workerHost to desired medium
     */
    private Map<String, String> findMisplacedBlock(
        InodeFile file, BlockInfo blockInfo) {
      Set<String> pinnedMediumTypes = file.getMediumTypes();
      if (pinnedMediumTypes.isEmpty()) {
        // nothing needs to be moved
        return Collections.emptyMap();
      }
      Map<String, String> movement = new HashMap<>();
      // at least pinned to one medium type
      String firstPinnedMedium = pinnedMediumTypes.iterator().next();
      int minReplication = file.getReplicationMin();
      int correctReplication = 0;
      List<String> candidates = new ArrayList<>();
      for (BlockLocation loc: blockInfo.getLocations()) {
        if (pinnedMediumTypes.contains(loc.getMediumType())) {
          correctReplication++;
        } else {
          candidates.add(loc.getWorkerAddress().getHost());
        }
      }
      if (correctReplication >= minReplication) {
        // there are more than minReplication in the right location
        return Collections.emptyMap();
      } else {
        int toMove = minReplication - correctReplication;
        for (String candidate : candidates) {
          // We are only addressing pinning to a single medium case, in the future we might
          // address pinning to multiple medium types.
          movement.put(candidate, firstPinnedMedium);
          toMove--;
          if (toMove == 0) {
            return movement;
          }
        }
      }
      return movement;
    }
  
    private void checkMisreplicated(Set<Long> inodes, ReplicationHandler handler)
        throws InterruptedException {
      for (long inodeId : inodes) {
        if (mActiveJobToInodeID.size() >= mMaxActiveJobs) {
          return;
        }
        if (mActiveJobToInodeID.containsValue(inodeId)) {
          continue;
        }
        // Throw if interrupted.
        if (Thread.interrupted()) {
          throw new InterruptedException("ReplicationChecker interrupted.");
        }
        try (LockedInodePath inodePath =
                 mInodeTree.lockFullInodePath(inodeId, LockPattern.READ, NoopJournalContext.INSTANCE)
        ) {
          InodeFile file = inodePath.getInodeFile();
          for (long blockId : file.getBlockIds()) {
            BlockInfo blockInfo = null;
            try {
              blockInfo = mBlockMaster.getBlockInfo(blockId);
            } catch (BlockInfoException e) {
              // Cannot find this block in Alluxio from BlockMaster, possibly persisted in UFS
            } catch (UnavailableException e) {
              // The block master is not available, wait for the next heartbeat
              LOG.warn("The block master is not available: {}", e.toString());
              return;
            }
            if (blockInfo == null) {
              // no block info available, we simply log and return;
              LOG.warn("Block info is null");
              return;
            }
  
            for (Map.Entry<String, String> entry
                : findMisplacedBlock(file, blockInfo).entrySet()) {
              try {
                final long jobId =
                    handler.migrate(inodePath.getUri(), blockId, entry.getKey(), entry.getValue());
                mActiveJobToInodeID.put(jobId, inodeId);
              } catch (Exception e) {
                LOG.warn(
                    "Unexpected exception encountered when starting a migration job (uri={},"
                        + " block ID={}, workerHost= {}) : {}",
                    inodePath.getUri(), blockId, entry.getKey(), e.toString());
                LOG.debug("Exception: ", e);
              }
            }
          }
        } catch (FileDoesNotExistException e) {
          LOG.warn("Failed to check replication level for inode id {} : {}", inodeId, e.toString());
        }
      }
    }
  
    private Set<Long> check(Set<Long> inodes, ReplicationHandler handler, Mode mode)
        throws InterruptedException {
      Set<Long> processedFileIds = new HashSet<>();
      for (long inodeId : inodes) {
        if (mActiveJobToInodeID.size() >= mMaxActiveJobs) {
          return processedFileIds;
        }
        if (mActiveJobToInodeID.containsValue(inodeId)) {
          continue;
        }
        Set<Triple<AlluxioURI, Long, Integer>> requests = new HashSet<>();
        // Throw if interrupted.
        if (Thread.interrupted()) {
          throw new InterruptedException("ReplicationChecker interrupted.");
        }
        // TODO(binfan): calling lockFullInodePath locks the entire path from root to the target
        // file and may increase lock contention in this tree. Investigate if we could avoid
        // locking the entire path but just the inode file since this access is read-only.
        try (LockedInodePath inodePath = mInodeTree.lockFullInodePath(
            inodeId, LockPattern.READ, NoopJournalContext.INSTANCE)
        ) {
          InodeFile file = inodePath.getInodeFile();
          for (long blockId : file.getBlockIds()) {
            BlockInfo blockInfo = null;
            try {
              blockInfo = mBlockMaster.getBlockInfo(blockId);
            } catch (BlockInfoException e) {
              // Cannot find this block in Alluxio from BlockMaster, possibly persisted in UFS
            } catch (UnavailableException e) {
              // The block master is not available, wait for the next heartbeat
              LOG.warn("The block master is not available: {}", e.toString());
              return processedFileIds;
            }
            int currentReplicas = (blockInfo == null) ? 0 : blockInfo.getLocations().size();
            switch (mode) {
              case EVICT:
                int maxReplicas = file.getReplicationMax();
                if (file.getPersistenceState() == PersistenceState.TO_BE_PERSISTED
                    && file.getReplicationDurable() > maxReplicas) {
                  maxReplicas = file.getReplicationDurable();
                }
                if (currentReplicas > maxReplicas) {
                  requests.add(new ImmutableTriple<>(inodePath.getUri(), blockId,
                      maxReplicas));
                }
                break;
              case REPLICATE:
                int minReplicas = file.getReplicationMin();
                if (file.getPersistenceState() == PersistenceState.TO_BE_PERSISTED
                    && file.getReplicationDurable() > minReplicas) {
                  minReplicas = file.getReplicationDurable();
                }
                if (currentReplicas < minReplicas) {
                  // if this file is not persisted and block master thinks it is lost, no effort made
                  if (!file.isPersisted() && mBlockMaster.isBlockLost(blockId)) {
                    continue;
                  }
                  requests.add(new ImmutableTriple<>(inodePath.getUri(), blockId,
                      minReplicas));
                }
                break;
              default:
                LOG.warn("Unexpected replication mode {}.", mode);
            }
          }
        } catch (FileDoesNotExistException e) {
          LOG.warn("Failed to check replication level for inode id {} : {}", inodeId, e.toString());
        }
  
        for (Triple<AlluxioURI, Long, Integer> entry : requests) {
          AlluxioURI uri = entry.getLeft();
          long blockId = entry.getMiddle();
          int numReplicas = entry.getRight();
          try {
            long jobId;
            switch (mode) {
              case EVICT :
              case REPLICATE:
                jobId = handler.setReplica(uri, blockId, numReplicas);
                break;
              default:
                throw new RuntimeException(String.format("Unexpected replication mode %s.", mode));
            }
            processedFileIds.add(inodeId);
            mActiveJobToInodeID.put(jobId, inodeId);
          } catch (JobDoesNotExistException | ResourceExhaustedException e) {
            LOG.warn("The job service is busy, will retry later. {}", e.toString());
            return processedFileIds;
          } catch (UnavailableException e) {
            LOG.warn("Unable to complete the replication check: {}, will retry later.", e.toString());
            return processedFileIds;
          } catch (Exception e) {
            SAMPLING_LOG.warn(
                "Unexpected exception encountered when starting a {} job (uri={},"
                    + " block ID={}, num replicas={}) : {}",
                mode, uri, blockId, numReplicas, e.toString());
            LOG.debug("Job service unexpected exception: ", e);
          }
        }
      }
      return processedFileIds;
    }
  }
  