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

import alluxio.metrics.MetricsSystem;
import alluxio.network.protocol.RPCProtoMessage;
import alluxio.proto.dataserver.Protocol;
import alluxio.security.authorization.Mode;
import alluxio.underfs.UfsManager;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.options.CreateOptions;

import com.codahale.metrics.Counter;
import io.netty.buffer.ByteBuf;

import java.io.IOException;
import java.io.OutputStream;
import java.util.concurrent.ExecutorService;

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 DataServerWriteHandler}.
 *
 * For more information about the implementation of the client side writer, see
 * UnderFileSystemFileOutStream.
 */
@NotThreadSafe
final class DataServerUfsFileWriteHandler extends DataServerWriteHandler {
  private final UfsManager mUfsManager;

  private class FileWriteRequestInternal extends WriteRequestInternal {
    private final String mUfsPath;
    private final UnderFileSystem mUnderFileSystem;
    private final OutputStream mOutputStream;

    FileWriteRequestInternal(Protocol.WriteRequest request) throws Exception {
      super(request.getId());
      Protocol.CreateUfsFileOptions createUfsFileOptions = request.getCreateUfsFileOptions();
      mUfsPath = createUfsFileOptions.getUfsPath();
      mUnderFileSystem = mUfsManager.get(createUfsFileOptions.getMountId()).getUfs();
      mOutputStream = mUnderFileSystem.create(mUfsPath,
          CreateOptions.defaults().setOwner(createUfsFileOptions.getOwner())
              .setGroup(createUfsFileOptions.getGroup())
              .setMode(new Mode((short) createUfsFileOptions.getMode())));
    }

    @Override
    public void close() throws IOException {
      mOutputStream.close();
    }

    @Override
    void cancel() throws IOException {
      // TODO(calvin): Consider adding cancel to the ufs stream api.
      mOutputStream.close();
      mUnderFileSystem.deleteFile(mUfsPath);
    }

    @Override
    void cleanup() throws IOException {
      cancel();
    }
  }

  /**
   * Creates an instance of {@link DataServerUfsFileWriteHandler}.
   *
   * @param executorService the executor service to run {@link PacketWriter}s
   * @param ufsManager the file data manager
   */
  DataServerUfsFileWriteHandler(ExecutorService executorService, UfsManager ufsManager) {
    super(executorService);
    mUfsManager = ufsManager;
  }

  @Override
  protected boolean acceptMessage(Object object) {
    if (!super.acceptMessage(object)) {
      return false;
    }
    Protocol.WriteRequest request = ((RPCProtoMessage) object).getMessage().asWriteRequest();
    return request.getType() == Protocol.RequestType.UFS_FILE;
  }

  /**
   * Initializes the handler if necessary.
   *
   * @param msg the block write request
   */
  @Override
  protected void initializeRequest(RPCProtoMessage msg) throws Exception {
    super.initializeRequest(msg);
    if (mRequest == null) {
      mRequest = new FileWriteRequestInternal(msg.getMessage().asWriteRequest());
    }
  }

  @Override
  protected void writeBuf(ByteBuf buf, long pos) throws Exception {
    buf.readBytes(((FileWriteRequestInternal) mRequest).mOutputStream, buf.readableBytes());
  }

  @Override
  protected void incrementMetrics(long bytesWritten) {
    Metrics.BYTES_WRITTEN_UFS.inc(bytesWritten);
  }

  /**
   * Class that contains metrics for BlockDataServerHandler.
   */
  private static final class Metrics {
    private static final Counter BYTES_WRITTEN_UFS = MetricsSystem.workerCounter("BytesWrittenUFS");

    private Metrics() {
    } // prevent instantiation
  }
}

@@ -742,24 +742,63 @@
 UfsManager;%0A
+import alluxio.underfs.UfsManager.Ufs;%0A
 import allux
@@ -817,32 +817,32 @@
 nderFileSystem;%0A
-
 import alluxio.u
@@ -1952,16 +1952,52 @@
 tStream;
+%0A    private final Counter mCounter;
 %0A%0A    Fi
@@ -2251,32 +2251,23 @@
 ;%0A      
-mUnderFileSystem
+Ufs ufs
  = mUfsM
@@ -2311,16 +2311,46 @@
 untId())
+;%0A      mUnderFileSystem = ufs
 .getUfs(
@@ -2613,24 +2613,227 @@
 tMode())));%0A
+      String ufsName = ufs.getUfsUri().toString().replace(%22/%22, %22_%22);%0A      String metricName = String.format(%22BytesWrittenUfs-Ufs:%25s%22, ufsName);%0A      mCounter = MetricsSystem.workerCounter(metricName);%0A
     %7D%0A%0A    @
@@ -4452,253 +4452,60 @@
     
-Metrics.BYTES_WRITTEN_UFS.inc(bytesWritten);%0A  %7D%0A%0A  /**%0A   * Class that contains metrics for BlockDataServerHandler.%0A   */%0A  private static final class Metrics %7B%0A    private static final Counter BYTES_WRITTEN_UFS = MetricsSystem.worker
+((FileWriteRequestInternal) mRequest).m
 Counter
-(%22B
+.inc(b
 ytes
@@ -4515,70 +4515,10 @@
 tten
-UFS%22);%0A%0A    private Metrics() %7B%0A    %7D // prevent instantiation
+);
 %0A  %7D

Include ufs name in ufs write metric