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.worker.block;
  
  import alluxio.Constants;
  import alluxio.ProcessUtils;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.ConnectionFailedException;
  import alluxio.exception.FailedToAcquireRegisterLeaseException;
  import alluxio.grpc.Command;
  import alluxio.heartbeat.HeartbeatExecutor;
  import alluxio.util.logging.SamplingLogger;
  import alluxio.wire.WorkerNetAddress;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import java.util.concurrent.atomic.AtomicReference;
  import javax.annotation.concurrent.NotThreadSafe;
  
  /**
   * Task that carries out the necessary block worker to master communications, including register and
   * heartbeat. This class manages its own {@link BlockMasterClient}.
   *
   * When running, this task first requests a block report from the
   * {@link alluxio.worker.block.BlockWorker}, then sends it to the master. The master may
   * respond to the heartbeat with a command which will be executed. After which, the task will wait
   * for the elapsed time since its last heartbeat has reached the heartbeat interval. Then the cycle
   * will continue.
   *
   * If the task fails to heartbeat to the master, it will destroy its old master client and recreate
   * it before retrying.
   */
  @NotThreadSafe
  public final class BlockMasterSync implements HeartbeatExecutor {
    private static final Logger LOG = LoggerFactory.getLogger(BlockMasterSync.class);
    private static final Logger SAMPLING_LOG = new SamplingLogger(LOG, 30L * Constants.SECOND);
    private static final long ACQUIRE_LEASE_WAIT_MAX_DURATION =
        Configuration.getMs(PropertyKey.WORKER_REGISTER_LEASE_RETRY_MAX_DURATION);
    private static final int HEARTBEAT_TIMEOUT_MS =
        (int) Configuration.getMs(PropertyKey.WORKER_BLOCK_HEARTBEAT_TIMEOUT_MS);
  
    /** The block worker responsible for interacting with Alluxio and UFS storage. */
    private final BlockWorker mBlockWorker;
  
    /** The worker ID for the worker. This may change if the master asks the worker to re-register. */
    private final AtomicReference<Long> mWorkerId;
  
    /** The net address of the worker. */
    private final WorkerNetAddress mWorkerAddress;
  
    /** Client-pool for all master communication. */
    private final BlockMasterClientPool mMasterClientPool;
    /** Client for all master communication. */
    private final BlockMasterClient mMasterClient;
  
    /** An async service to remove block. */
    private final AsyncBlockRemover mAsyncBlockRemover;
  
    /** Last System.currentTimeMillis() timestamp when a heartbeat successfully completed. */
    private long mLastSuccessfulHeartbeatMs;
  
    /** The helper instance for sync related methods. */
    private final BlockMasterSyncHelper mBlockMasterSyncHelper;
  
    /**
     * Creates a new instance of {@link BlockMasterSync}.
     *
     * @param blockWorker the {@link BlockWorker} this syncer is updating to
     * @param workerId the worker id of the worker, assigned by the block master
     * @param workerAddress the net address of the worker
     * @param masterClientPool the Alluxio master client pool
     */
    public BlockMasterSync(BlockWorker blockWorker, AtomicReference<Long> workerId,
        WorkerNetAddress workerAddress, BlockMasterClientPool masterClientPool) throws IOException {
      mBlockWorker = blockWorker;
      mWorkerId = workerId;
      mWorkerAddress = workerAddress;
      mMasterClientPool = masterClientPool;
      mMasterClient = mMasterClientPool.acquire();
      mAsyncBlockRemover = new AsyncBlockRemover(mBlockWorker);
      mBlockMasterSyncHelper = new BlockMasterSyncHelper(mMasterClient);
  
      registerWithMaster();
      mLastSuccessfulHeartbeatMs = System.currentTimeMillis();
    }
  
    /**
     * Registers with the Alluxio master. This should be called before the
     * continuous heartbeat thread begins.
     */
    private void registerWithMaster() throws IOException {
      BlockStoreMeta storeMeta = mBlockWorker.getStoreMetaFull();
      try {
        mBlockMasterSyncHelper.tryAcquireLease(mWorkerId.get(), storeMeta);
      } catch (FailedToAcquireRegisterLeaseException e) {
        mMasterClient.disconnect();
        if (Configuration.getBoolean(PropertyKey.TEST_MODE)) {
          throw new RuntimeException(String.format("Master register lease timeout exceeded: %dms",
              ACQUIRE_LEASE_WAIT_MAX_DURATION));
        }
        ProcessUtils.fatalError(LOG, "Master register lease timeout exceeded: %dms",
            ACQUIRE_LEASE_WAIT_MAX_DURATION);
      }
      mBlockMasterSyncHelper.registerToMaster(mWorkerId.get(), storeMeta);
    }
  
    /**
     * Heartbeats to the master node about the change in the worker's managed space.
     */
    @Override
    public void heartbeat(long timeLimitMs) {
      boolean success = mBlockMasterSyncHelper.heartbeat(
          mWorkerId.get(), mBlockWorker.getReport(),
          mBlockWorker.getStoreMeta(), this::handleMasterCommand);
      if (success) {
        mLastSuccessfulHeartbeatMs = System.currentTimeMillis();
      } else {
        if (HEARTBEAT_TIMEOUT_MS > 0) {
          if (System.currentTimeMillis() - mLastSuccessfulHeartbeatMs >= HEARTBEAT_TIMEOUT_MS) {
            if (Configuration.getBoolean(PropertyKey.TEST_MODE)) {
              throw new RuntimeException(
                  String.format("Master heartbeat timeout exceeded: %s", HEARTBEAT_TIMEOUT_MS));
            }
            // TODO(andrew): Propagate the exception to the main thread and exit there.
            ProcessUtils.fatalError(LOG, "Master heartbeat timeout exceeded: %d",
                HEARTBEAT_TIMEOUT_MS);
          }
        }
      }
    }
  
    @Override
    public void close() {
      mAsyncBlockRemover.shutDown();
      mMasterClientPool.release(mMasterClient);
    }
  
    /**
     * Handles a master command. The command is one of Unknown, Nothing, Register, Free, or Delete.
     * This call will block until the command is complete.
     *
     * @param cmd the command to execute
     * @throws IOException if I/O errors occur
     * @throws ConnectionFailedException if connection fails
     */
    // TODO(calvin): Evaluate the necessity of each command.
    private void handleMasterCommand(Command cmd) throws IOException, ConnectionFailedException {
      if (cmd == null) {
        return;
      }
      switch (cmd.getCommandType()) {
        // Currently unused
        case Delete:
          break;
        // Master requests blocks to be removed from Alluxio managed space.
        case Free:
          mAsyncBlockRemover.addBlocksToDelete(cmd.getDataList());
          break;
        // No action required
        case Nothing:
          break;
        // Master requests re-registration
        case Register:
          mWorkerId.set(mMasterClient.getId(mWorkerAddress));
          registerWithMaster();
          break;
        // Unknown request
        case Unknown:
          LOG.error("Master heartbeat sends unknown command {}", cmd);
          break;
        case Decommissioned:
          SAMPLING_LOG.info("This worker has been decommissioned");
          break;
        default:
          throw new RuntimeException("Un-recognized command from master " + cmd);
      }
    }
  }
  