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.grpc;
  
  import alluxio.conf.Configuration;
  import alluxio.grpc.WriteRequest;
  import alluxio.grpc.WriteResponse;
  import alluxio.metrics.MetricInfo;
  import alluxio.metrics.MetricKey;
  import alluxio.metrics.MetricsSystem;
  import alluxio.network.protocol.databuffer.DataBuffer;
  import alluxio.proto.dataserver.Protocol;
  import alluxio.resource.CloseableResource;
  import alluxio.security.authentication.AuthenticatedUserInfo;
  import alluxio.security.authorization.Mode;
  import alluxio.underfs.ContentHashable;
  import alluxio.underfs.UfsManager;
  import alluxio.underfs.UnderFileSystem;
  import alluxio.underfs.options.CreateOptions;
  import alluxio.util.proto.ProtoUtils;
  
  import com.google.common.base.Preconditions;
  import io.grpc.stub.StreamObserver;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import javax.annotation.concurrent.NotThreadSafe;
  
  /**
   * This handler handles writes to a file in the under file system. Due to the semantics enforced
   * on under file systems, the client must write all the data of a file through the same stream to
   * the under file system. This prevents us from handling the writes at a block level.
   *
   * For more information about the implementation of read/write buffering, see
   * {@link AbstractWriteHandler}.
   *
   * For more information about the implementation of the client side writer, see
   * UnderFileSystemFileOutStream.
   */
  @alluxio.annotation.SuppressFBWarnings(
      value = "BC_UNCONFIRMED_CAST_OF_RETURN_VALUE",
      justification = "false positive with superclass generics, "
          + "see more description in https://sourceforge.net/p/findbugs/bugs/1242/")
  @NotThreadSafe
  public final class UfsFileWriteHandler extends AbstractWriteHandler<UfsFileWriteRequestContext> {
    private static final Logger LOG = LoggerFactory.getLogger(UfsFileWriteHandler.class);
    private final UfsManager mUfsManager;
  
    /**
     * Creates an instance of {@link UfsFileWriteHandler}.
     *
     * @param ufsManager the file data manager
     * @param userInfo the authenticated user info
     */
    UfsFileWriteHandler(UfsManager ufsManager, StreamObserver<WriteResponse> responseObserver,
        AuthenticatedUserInfo userInfo) {
      super(responseObserver, userInfo);
      mUfsManager = ufsManager;
    }
  
    @Override
    protected UfsFileWriteRequestContext createRequestContext(WriteRequest msg) {
      return new UfsFileWriteRequestContext(msg);
    }
  
    @Override
    protected void completeRequest(UfsFileWriteRequestContext context)
        throws Exception {
      if (context == null) {
        return;
      }
      if (context.getOutputStream() == null) {
        createUfsFile(context);
      }
      Preconditions.checkState(context.getOutputStream() != null);
      context.getOutputStream().close();
      if (context.getOutputStream() instanceof ContentHashable) {
        try {
          ((ContentHashable) context.getOutputStream()).getContentHash()
              .ifPresent(context::setContentHash);
        } catch (IOException e) {
          LOG.warn("Error getting content hash after completing file", e);
        }
      }
      CreateOptions createOptions = context.getCreateOptions();
      if (createOptions != null) {
        try {
          // Set the owner/group of the file to the correct owner.
          context.getUfsResource().get()
              .setOwner(context.getRequest().getUfsPath(), createOptions.getOwner(),
                  createOptions.getGroup());
        } catch (IOException e) {
          LOG.warn("Failed to update ownership for ufs path: {} owner: {} group: {} error: {}",
              context.getRequest().getUfsPath(), createOptions.getOwner(), createOptions.getGroup(),
              e.toString());
        }
      }
      context.setOutputStream(null);
      context.setCreateOptions(null);
      context.getUfsResource().close();
    }
  
    @Override
    protected void cancelRequest(UfsFileWriteRequestContext context) throws Exception {
      if (context == null) {
        return;
      }
      UfsFileWriteRequest request = context.getRequest();
      // TODO(calvin): Consider adding cancel to the ufs stream api.
      if (context.getOutputStream() != null && context.getUfsResource() != null) {
        context.getOutputStream().close();
        context.getUfsResource().get().deleteExistingFile(request.getUfsPath());
        context.setOutputStream(null);
        context.setCreateOptions(null);
        context.getUfsResource().close();
      }
    }
  
    @Override
    protected void cleanupRequest(UfsFileWriteRequestContext context) throws Exception {
      cancelRequest(context);
    }
  
    @Override
    protected void flushRequest(UfsFileWriteRequestContext context)
        throws Exception {
      Preconditions.checkState(context != null);
      if (context.getOutputStream() != null) {
        context.getOutputStream().flush();
      }
    }
  
    @Override
    protected void writeBuf(UfsFileWriteRequestContext context,
        StreamObserver<WriteResponse> observer, DataBuffer buf, long pos) throws Exception {
      Preconditions.checkState(context != null);
      if (context.getOutputStream() == null) {
        createUfsFile(context);
      }
      buf.readBytes(context.getOutputStream(), buf.readableBytes());
    }
  
    @Override
    protected String getLocationInternal(UfsFileWriteRequestContext context) {
      return context.getRequest().getCreateUfsFileOptions().getUfsPath();
    }
  
    private void createUfsFile(UfsFileWriteRequestContext context)
        throws IOException {
      UfsFileWriteRequest request = context.getRequest();
      Preconditions.checkState(request != null);
      Protocol.CreateUfsFileOptions createUfsFileOptions = request.getCreateUfsFileOptions();
      UfsManager.UfsClient ufsClient = mUfsManager.get(createUfsFileOptions.getMountId());
      CloseableResource<UnderFileSystem> ufsResource = ufsClient.acquireUfsResource();
      context.setUfsResource(ufsResource);
      UnderFileSystem ufs = ufsResource.get();
      CreateOptions createOptions = CreateOptions.defaults(Configuration.global())
          .setCreateParent(true)
          .setOwner(createUfsFileOptions.getOwner()).setGroup(createUfsFileOptions.getGroup())
          .setMode(new Mode((short) createUfsFileOptions.getMode()));
      if (createUfsFileOptions.hasAcl()) {
        // This acl information will be ignored by all but HDFS implementations
        createOptions.setAcl(ProtoUtils.fromProto(createUfsFileOptions.getAcl()));
      }
      context.setOutputStream(ufs.createNonexistingFile(request.getUfsPath(), createOptions));
      context.setCreateOptions(createOptions);
      String ufsString = MetricsSystem.escape(ufsClient.getUfsMountPointUri());
  
      MetricKey counterKey = MetricKey.WORKER_BYTES_WRITTEN_UFS;
      MetricKey meterKey = MetricKey.WORKER_BYTES_WRITTEN_UFS_THROUGHPUT;
      context.setCounter(MetricsSystem.counterWithTags(counterKey.getName(),
          counterKey.isClusterAggregated(), MetricInfo.TAG_UFS, ufsString));
      context.setMeter(MetricsSystem.meterWithTags(meterKey.getName(),
          meterKey.isClusterAggregated(), MetricInfo.TAG_UFS, ufsString));
    }
  }
  