alluxio-master-2.x/core/client/fs/src/main/java/alluxio/client/file/RetryHandlingFileSystemMasterClient.java
/*
 * 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.client.file;

import alluxio.AbstractMasterClient;
import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.exception.status.AlluxioStatusException;
import alluxio.grpc.CancelSyncMetadataPRequest;
import alluxio.grpc.CancelSyncMetadataPResponse;
import alluxio.grpc.CheckAccessPOptions;
import alluxio.grpc.CheckAccessPRequest;
import alluxio.grpc.CheckConsistencyPOptions;
import alluxio.grpc.CheckConsistencyPRequest;
import alluxio.grpc.CompleteFilePOptions;
import alluxio.grpc.CompleteFilePRequest;
import alluxio.grpc.CreateDirectoryPOptions;
import alluxio.grpc.CreateDirectoryPRequest;
import alluxio.grpc.CreateFilePOptions;
import alluxio.grpc.CreateFilePRequest;
import alluxio.grpc.DeletePOptions;
import alluxio.grpc.DeletePRequest;
import alluxio.grpc.ExistsPOptions;
import alluxio.grpc.ExistsPRequest;
import alluxio.grpc.FileSystemMasterClientServiceGrpc;
import alluxio.grpc.FreePOptions;
import alluxio.grpc.FreePRequest;
import alluxio.grpc.GetFilePathPRequest;
import alluxio.grpc.GetJobProgressPRequest;
import alluxio.grpc.GetJobProgressPResponse;
import alluxio.grpc.GetMountTablePRequest;
import alluxio.grpc.GetNewBlockIdForFilePOptions;
import alluxio.grpc.GetNewBlockIdForFilePRequest;
import alluxio.grpc.GetStateLockHoldersPOptions;
import alluxio.grpc.GetStateLockHoldersPRequest;
import alluxio.grpc.GetStatusPOptions;
import alluxio.grpc.GetStatusPRequest;
import alluxio.grpc.GetSyncPathListPRequest;
import alluxio.grpc.GetSyncProgressPRequest;
import alluxio.grpc.GetSyncProgressPResponse;
import alluxio.grpc.GrpcUtils;
import alluxio.grpc.JobProgressPOptions;
import alluxio.grpc.JobProgressReportFormat;
import alluxio.grpc.ListStatusPOptions;
import alluxio.grpc.ListStatusPRequest;
import alluxio.grpc.ListStatusPartialPOptions;
import alluxio.grpc.ListStatusPartialPRequest;
import alluxio.grpc.MountPOptions;
import alluxio.grpc.MountPRequest;
import alluxio.grpc.NeedsSyncRequest;
import alluxio.grpc.RenamePOptions;
import alluxio.grpc.RenamePRequest;
import alluxio.grpc.ReverseResolvePRequest;
import alluxio.grpc.ScheduleAsyncPersistencePOptions;
import alluxio.grpc.ScheduleAsyncPersistencePRequest;
import alluxio.grpc.ServiceType;
import alluxio.grpc.SetAclAction;
import alluxio.grpc.SetAclPOptions;
import alluxio.grpc.SetAclPRequest;
import alluxio.grpc.SetAttributePOptions;
import alluxio.grpc.SetAttributePRequest;
import alluxio.grpc.StartSyncPRequest;
import alluxio.grpc.StopJobPRequest;
import alluxio.grpc.StopJobPResponse;
import alluxio.grpc.StopSyncPRequest;
import alluxio.grpc.SubmitJobPRequest;
import alluxio.grpc.SubmitJobPResponse;
import alluxio.grpc.SyncMetadataAsyncPResponse;
import alluxio.grpc.SyncMetadataPOptions;
import alluxio.grpc.SyncMetadataPRequest;
import alluxio.grpc.SyncMetadataPResponse;
import alluxio.grpc.UnmountPOptions;
import alluxio.grpc.UnmountPRequest;
import alluxio.grpc.UpdateMountPRequest;
import alluxio.grpc.UpdateUfsModePOptions;
import alluxio.grpc.UpdateUfsModePRequest;
import alluxio.job.JobDescription;
import alluxio.job.JobRequest;
import alluxio.master.MasterClientContext;
import alluxio.retry.CountingRetry;
import alluxio.security.authorization.AclEntry;
import alluxio.util.FileSystemOptionsUtils;
import alluxio.wire.SyncPointInfo;

import com.google.protobuf.ByteString;
import org.apache.commons.lang3.SerializationUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import javax.annotation.concurrent.ThreadSafe;

/**
 * A wrapper for the gRPC client to interact with the file system master, used by alluxio clients.
 *
 */
@ThreadSafe
public final class RetryHandlingFileSystemMasterClient extends AbstractMasterClient
    implements FileSystemMasterClient {
  private static final Logger RPC_LOG = LoggerFactory.getLogger(FileSystemMasterClient.class);

  private FileSystemMasterClientServiceGrpc.FileSystemMasterClientServiceBlockingStub mClient =
      null;

  /**
   * Creates a new {@link RetryHandlingFileSystemMasterClient} instance.
   *
   * @param conf master client configuration
   */
  public RetryHandlingFileSystemMasterClient(MasterClientContext conf) {
    super(conf);
  }

  @Override
  protected ServiceType getRemoteServiceType() {
    return ServiceType.FILE_SYSTEM_MASTER_CLIENT_SERVICE;
  }

  @Override
  protected String getServiceName() {
    return Constants.FILE_SYSTEM_MASTER_CLIENT_SERVICE_NAME;
  }

  @Override
  protected long getServiceVersion() {
    return Constants.FILE_SYSTEM_MASTER_CLIENT_SERVICE_VERSION;
  }

  @Override
  protected void afterConnect() {
    mClient = FileSystemMasterClientServiceGrpc.newBlockingStub(mChannel);
  }

  @Override
  public void checkAccess(AlluxioURI path, CheckAccessPOptions options)
      throws AlluxioStatusException {
    retryRPC(() -> mClient.checkAccess(
        CheckAccessPRequest.newBuilder().setPath(getTransportPath(path))
            .setOptions(options).build()),
        RPC_LOG, "CheckAccess", "path=%s,options=%s", path, options);
  }

  @Override
  public List<AlluxioURI> checkConsistency(final AlluxioURI path,
      final CheckConsistencyPOptions options) throws AlluxioStatusException {
    return retryRPC(() -> {
      List<String> inconsistentPaths = mClient.checkConsistency(CheckConsistencyPRequest
          .newBuilder().setPath(getTransportPath(path)).setOptions(options).build())
          .getInconsistentPathsList();
      List<AlluxioURI> inconsistentUris = new ArrayList<>(inconsistentPaths.size());
      for (String inconsistentPath : inconsistentPaths) {
        inconsistentUris.add(new AlluxioURI(inconsistentPath));
      }
      return inconsistentUris;
    }, RPC_LOG, "CheckConsistency", "path=%s,options=%s", path, options);
  }

  @Override
  public void createDirectory(final AlluxioURI path,
      final CreateDirectoryPOptions options) throws AlluxioStatusException {
    retryRPC(
        () -> mClient.createDirectory(CreateDirectoryPRequest.newBuilder()
            .setPath(getTransportPath(path)).setOptions(options).build()),
        RPC_LOG, "CreateDirectory", "path=%s,options=%s", path, options);
  }

  @Override
  public URIStatus createFile(final AlluxioURI path, final CreateFilePOptions options)
      throws AlluxioStatusException {
    return retryRPC(
        () -> new URIStatus(GrpcUtils.fromProto(mClient.createFile(CreateFilePRequest.newBuilder()
            .setPath(getTransportPath(path)).setOptions(options).build()).getFileInfo())),
        RPC_LOG, "CreateFile", "path=%s,options=%s", path, options);
  }

  @Override
  public void completeFile(final AlluxioURI path, final CompleteFilePOptions options)
      throws AlluxioStatusException {
    retryRPC(() -> mClient.completeFile(CompleteFilePRequest.newBuilder()
        .setPath(getTransportPath(path)).setOptions(options).build()), RPC_LOG, "CompleteFile",
        "path=%s,options=%s", path, options);
  }

  @Override
  public void delete(final AlluxioURI path, final DeletePOptions options)
      throws AlluxioStatusException {
    retryRPC(() -> mClient.remove(DeletePRequest.newBuilder().setPath(getTransportPath(path))
        .setOptions(options).build()), RPC_LOG, "Delete",
        "path=%s,options=%s", path, options);
  }

  @Override
  public boolean exists(final AlluxioURI path, final ExistsPOptions options)
      throws AlluxioStatusException {
    return retryRPC(() -> mClient.exists(ExistsPRequest.newBuilder()
        .setPath(getTransportPath(path))
        .setOptions(options).build()).getExists(), RPC_LOG, "Exists",
        "path=%s,options=%s", path, options);
  }

  @Override
  public void free(final AlluxioURI path, final FreePOptions options)
      throws AlluxioStatusException {
    retryRPC(() -> mClient.free(FreePRequest.newBuilder().setPath(getTransportPath(path))
        .setOptions(options).build()), RPC_LOG, "Free", "path=%s,options=%s", path, options);
  }

  @Override
  public String getFilePath(long fileId) throws AlluxioStatusException {
    return retryRPC(() -> mClient.getFilePath(GetFilePathPRequest
            .newBuilder().setFileId(fileId).build()).getPath(), RPC_LOG, "GetFilePath", "fileId=%d",
        fileId);
  }

  @Override
  public URIStatus getStatus(final AlluxioURI path, final GetStatusPOptions options)
      throws AlluxioStatusException {
    return retryRPC(() -> new URIStatus(GrpcUtils
        .fromProto(mClient.getStatus(GetStatusPRequest.newBuilder().setPath(getTransportPath(path))
            .setOptions(options).build()).getFileInfo())),
        RPC_LOG, "GetStatus", "path=%s,options=%s", path, options);
  }

  @Override
  public synchronized List<SyncPointInfo> getSyncPathList() throws AlluxioStatusException {
    return retryRPC(() -> mClient.getSyncPathList(GetSyncPathListPRequest.getDefaultInstance())
        .getSyncPathsList().stream().map(x -> alluxio.wire.SyncPointInfo.fromProto(x))
        .collect(Collectors.toList()), RPC_LOG, "GetSyncPathList", "");
  }

  @Override
  public long getNewBlockIdForFile(final AlluxioURI path)
      throws AlluxioStatusException {
    return retryRPC(
        () -> mClient.getNewBlockIdForFile(
            GetNewBlockIdForFilePRequest.newBuilder().setPath(getTransportPath(path))
                .setOptions(GetNewBlockIdForFilePOptions.newBuilder().build()).build())
            .getId(),
        RPC_LOG, "GetNewBlockIdForFile", "path=%s", path);
  }

  @Override
  public Map<String, alluxio.wire.MountPointInfo> getMountTable(boolean checkUfs)
      throws AlluxioStatusException {
    return retryRPC(() -> {
      Map<String, alluxio.wire.MountPointInfo> mountTableWire = new HashMap<>();
      for (Map.Entry<String, alluxio.grpc.MountPointInfo> entry : mClient
          .getMountTable(GetMountTablePRequest.newBuilder().setCheckUfs(checkUfs).build())
          .getMountPointsMap()
          .entrySet()) {
        mountTableWire.put(entry.getKey(), GrpcUtils.fromProto(entry.getValue()));
      }
      return mountTableWire;
    }, RPC_LOG, "GetMountTable", "");
  }

  @Override
  public void iterateStatus(final AlluxioURI path, final ListStatusPOptions options,
      Consumer<? super URIStatus> action)
      throws AlluxioStatusException {
    retryRPC(
         new CountingRetry(0),
        () ->  {
          StreamSupport.stream(
              Spliterators.spliteratorUnknownSize(
                  mClient.listStatus(ListStatusPRequest.newBuilder()
                      .setPath(getTransportPath(path)).setOptions(options).build()),
                  Spliterator.ORDERED),
              false)
              .flatMap(pListStatusResponse -> pListStatusResponse.getFileInfosList().stream()
                  .map(pFileInfo -> new URIStatus(GrpcUtils.fromProto(pFileInfo))))
              .forEach(action);
          return null;
        },
        RPC_LOG, "ListStatus", "path=%s,options=%s", path, options);
  }

  @Override
  public List<URIStatus> listStatus(final AlluxioURI path, final ListStatusPOptions options)
      throws AlluxioStatusException {
    return retryRPC(() -> {
      List<URIStatus> result = new ArrayList<>();
      mClient
          .listStatus(ListStatusPRequest.newBuilder().setPath(getTransportPath(path))
              .setOptions(options).build())
          .forEachRemaining(
              (pListStatusResponse) -> result.addAll(pListStatusResponse.getFileInfosList().stream()
                  .map((pFileInfo) -> new URIStatus(GrpcUtils.fromProto(pFileInfo)))
                  .collect(Collectors.toList())));
      return result;
    }, RPC_LOG, "ListStatus", "path=%s,options=%s", path, options);
  }

  @Override
  public ListStatusPartialResult listStatusPartial(
      final AlluxioURI path, final ListStatusPartialPOptions options)
      throws AlluxioStatusException {
    return retryRPC(() -> ListStatusPartialResult.fromProto(mClient
        .listStatusPartial(ListStatusPartialPRequest.newBuilder().setPath(getTransportPath(path))
            .setOptions(options).build())), RPC_LOG,
        "ListStatusPartial", "path=%s,options=%s", path, options);
  }

  @Override
  public void mount(final AlluxioURI alluxioPath, final AlluxioURI ufsPath,
      final MountPOptions options) throws AlluxioStatusException {
    retryRPC(
        () -> mClient.mount(MountPRequest.newBuilder().setAlluxioPath(alluxioPath.toString())
            .setUfsPath(ufsPath.toString()).setOptions(options).build()),
        RPC_LOG, "Mount", "alluxioPath=%s,ufsPath=%s,options=%s", alluxioPath, ufsPath, options);
  }

  @Override
  public void updateMount(final AlluxioURI alluxioPath, final MountPOptions options)
      throws AlluxioStatusException {
    retryRPC(
        () -> mClient.updateMount(UpdateMountPRequest.newBuilder()
            .setAlluxioPath(alluxioPath.toString())
            .setOptions(options).build()),
        RPC_LOG, "UpdateMount", "path=%s,options=%s", alluxioPath, options);
  }

  @Override
  public void rename(final AlluxioURI src, final AlluxioURI dst)
      throws AlluxioStatusException {
    rename(src, dst, FileSystemOptionsUtils.renameDefaults(mContext.getClusterConf()));
  }

  @Override
  public void rename(final AlluxioURI src, final AlluxioURI dst,
      final RenamePOptions options) throws AlluxioStatusException {
    retryRPC(() -> mClient.rename(RenamePRequest.newBuilder().setPath(getTransportPath(src))
        .setDstPath(getTransportPath(dst)).setOptions(options).build()), RPC_LOG, "Rename",
        "src=%s,dst=%s,options=%s", src, dst, options);
  }

  @Override
  public AlluxioURI reverseResolve(final AlluxioURI ufsUri) throws AlluxioStatusException {
    return retryRPC(() -> new AlluxioURI(mClient.reverseResolve(ReverseResolvePRequest.newBuilder()
        .setUfsUri(ufsUri.toString()).build()).getAlluxioPath()), RPC_LOG, "ReverseResolve",
        "ufsUri=%s", ufsUri);
  }

  @Override
  public void setAcl(AlluxioURI path, SetAclAction action, List<AclEntry> entries,
      SetAclPOptions options) throws AlluxioStatusException {
    retryRPC(() -> mClient.setAcl(
        SetAclPRequest.newBuilder().setPath(getTransportPath(path)).setAction(action)
            .addAllEntries(entries.stream().map(GrpcUtils::toProto).collect(Collectors.toList()))
            .setOptions(options).build()),
        RPC_LOG, "SetAcl", "path=%s,action=%s,entries=%s,options=%s",
        path, action, entries, options);
  }

  @Override
  public void setAttribute(final AlluxioURI path, final SetAttributePOptions options)
      throws AlluxioStatusException {
    retryRPC(() -> mClient.setAttribute(SetAttributePRequest.newBuilder()
        .setPath(getTransportPath(path)).setOptions(options).build()), RPC_LOG, "SetAttribute",
        "path=%s,options=%s", path, options);
  }

  @Override
  public void scheduleAsyncPersist(final AlluxioURI path, ScheduleAsyncPersistencePOptions options)
      throws AlluxioStatusException {
    retryRPC(() -> mClient.scheduleAsyncPersistence(ScheduleAsyncPersistencePRequest.newBuilder()
        .setPath(getTransportPath(path)).setOptions(options).build()), RPC_LOG,
        "ScheduleAsyncPersist", "path=%s,options=%s", path, options);
  }

  @Override
  public synchronized void startSync(final AlluxioURI path) throws AlluxioStatusException {
    retryRPC(
        () -> mClient
            .startSync(StartSyncPRequest.newBuilder().setPath(getTransportPath(path)).build()),
        RPC_LOG, "StartSync", "path=%s", path);
  }

  @Override
  public synchronized void stopSync(final AlluxioURI path) throws AlluxioStatusException {
    retryRPC(
        () -> mClient
            .stopSync(StopSyncPRequest.newBuilder().setPath(getTransportPath(path)).build()),
        RPC_LOG, "StopSync", "path=%s", path);
  }

  @Override
  public void unmount(final AlluxioURI alluxioPath) throws AlluxioStatusException {
    retryRPC(() -> mClient
        .unmount(UnmountPRequest.newBuilder().setAlluxioPath(getTransportPath(alluxioPath))
            .setOptions(UnmountPOptions.newBuilder().build()).build()),
        RPC_LOG, "Unmount", "path=%s", alluxioPath);
  }

  @Override
  public void updateUfsMode(final AlluxioURI ufsUri,
      final UpdateUfsModePOptions options) throws AlluxioStatusException {
    retryRPC(
        () -> mClient.updateUfsMode(UpdateUfsModePRequest.newBuilder()
            .setUfsPath(ufsUri.getRootPath()).setOptions(options).build()),
        RPC_LOG, "UpdateUfsMode", "ufsUri=%s,options=%s", ufsUri, options);
  }

  @Override
  public List<String> getStateLockHolders()
      throws AlluxioStatusException {
    return retryRPC(() -> {
      final ArrayList<String> result = new ArrayList<>();
      mClient.getStateLockHolders(GetStateLockHoldersPRequest.newBuilder()
          .setOptions(GetStateLockHoldersPOptions.newBuilder().build()).build()).getThreadsList()
          .forEach((thread) -> result.add(thread));
      return result;
    }, RPC_LOG, "GetStateLockHolders", "");
  }

  @Override
  public void needsSync(AlluxioURI path) throws AlluxioStatusException {
    retryRPC(
        () -> mClient.needsSync(
            NeedsSyncRequest.newBuilder().setPath(getTransportPath(path)).build()),
        RPC_LOG, "NeedsSync", "path=%s", path);
  }

  @Override
  public Optional<String> submitJob(JobRequest job) {
    connectWithRuntimeException();
    final ByteString requestBody = ByteString.copyFrom(SerializationUtils.serialize(job));
    SubmitJobPRequest request = SubmitJobPRequest
        .newBuilder()
        .setRequestBody(requestBody)
        .build();
    SubmitJobPResponse response = mClient.submitJob(request);
    return response.hasJobId() ? Optional.of(response.getJobId()) : Optional.empty();
  }

  @Override
  public boolean stopJob(JobDescription jobDescription) {
    connectWithRuntimeException();
    StopJobPResponse response = mClient.stopJob(StopJobPRequest
        .newBuilder()
        .setJobDescription(alluxio.grpc.JobDescription
            .newBuilder()
            .setType(jobDescription.getType())
            .setPath(jobDescription.getPath())
            .build())
        .build());
    return response.getJobStopped();
  }

  @Override
  public String getJobProgress(JobDescription jobDescription,
      JobProgressReportFormat format, boolean verbose) {
    JobProgressPOptions.Builder options = JobProgressPOptions.newBuilder()
            .setVerbose(verbose)
            .setFormat(format);
    connectWithRuntimeException();
    GetJobProgressPResponse response = mClient.getJobProgress(
        GetJobProgressPRequest.newBuilder()
            .setJobDescription(alluxio.grpc.JobDescription
            .newBuilder()
            .setType(jobDescription.getType())
            .setPath(jobDescription.getPath())
            .build())
            .setOptions(options.build())
            .build());
    return response.getProgressReport();
  }

  @Override
  public SyncMetadataPResponse syncMetadata(AlluxioURI path, SyncMetadataPOptions options)
      throws AlluxioStatusException {
    return retryRPC(() -> {
      SyncMetadataPRequest request = SyncMetadataPRequest.newBuilder()
          .setPath(path.getPath())
          .setOptions(options)
          .build();
      SyncMetadataPResponse response = mClient.syncMetadata(request);
      return response;
    }, RPC_LOG, "SyncMetadata", "path=%s,options=%s", path, options);
  }

  @Override
  public SyncMetadataAsyncPResponse syncMetadataAsync(AlluxioURI path, SyncMetadataPOptions options)
      throws AlluxioStatusException {
    return retryRPC(() -> {
      SyncMetadataPRequest request = SyncMetadataPRequest.newBuilder()
          .setPath(path.getPath())
          .setOptions(options)
          .build();
      SyncMetadataAsyncPResponse response = mClient.syncMetadataAsync(request);
      return response;
    }, RPC_LOG, "SyncMetadataAsync", "path=%s,options=%s", path, options);
  }

  @Override
  public GetSyncProgressPResponse getSyncProgress(long taskGroupId) throws AlluxioStatusException {
    return retryRPC(() -> {
      GetSyncProgressPRequest request = GetSyncProgressPRequest.newBuilder()
          .setTaskGroupId(taskGroupId)
          .build();
      return mClient.getSyncProgress(request);
    }, RPC_LOG, "GetSyncProgress", "taskGroupId=%s", taskGroupId);
  }

  @Override
  public CancelSyncMetadataPResponse cancelSyncMetadata(long taskGroupId)
      throws AlluxioStatusException {
    return retryRPC(() -> {
      CancelSyncMetadataPRequest request = CancelSyncMetadataPRequest.newBuilder()
          .setTaskGroupId(taskGroupId)
          .build();
      return mClient.cancelSyncMetadata(request);
    }, RPC_LOG, "CancelSyncMetadata", "taskGroupId=%s", taskGroupId);
  }

  /**
   * Gets the path that will be transported to master.
   *
   * @param uri uri
   * @return transport path
   */
  private static String getTransportPath(AlluxioURI uri) {
    if (uri.hasScheme() && !uri.getScheme().equals(Constants.SCHEME)) {
      // Return full URI for non-Alluxio path.
      return uri.toString();
    } else {
      // Scheme-less URIs are assumed to be Alluxio paths
      // and getPath() is used to avoid string conversion.
      return uri.getPath();
    }
  }
}
