/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.fs.http.server;

import com.google.common.base.Charsets;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.XAttrCodec;
import org.apache.hadoop.fs.XAttrSetFlag;
import org.apache.hadoop.fs.http.client.HttpFSFileSystem;
import org.apache.hadoop.fs.http.client.HttpFSUtils;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.AccessTimeParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.AclPermissionParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.BlockSizeParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.DataParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.DestinationParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.FilterParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.GroupParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.LenParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.ModifiedTimeParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.NewLengthParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.OffsetParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.OldSnapshotNameParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.OperationParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.OverwriteParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.OwnerParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.PermissionParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.PolicyNameParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.RecursiveParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.ReplicationParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.SourcesParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.SnapshotNameParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.XAttrEncodingParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.XAttrNameParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.XAttrSetFlagParam;
import org.apache.hadoop.fs.http.server.HttpFSParametersProvider.XAttrValueParam;
import org.apache.hadoop.hdfs.web.JsonUtil;
import org.apache.hadoop.lib.service.FileSystemAccess;
import org.apache.hadoop.lib.service.FileSystemAccessException;
import org.apache.hadoop.lib.service.Groups;
import org.apache.hadoop.lib.service.Instrumentation;
import org.apache.hadoop.lib.servlet.FileSystemReleaseFilter;
import org.apache.hadoop.lib.wsrs.InputStreamEntity;
import org.apache.hadoop.lib.wsrs.Parameters;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.delegation.web.HttpUserGroupInformation;
import org.json.simple.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import javax.servlet.http.HttpServletRequest;
import javax.ws.rs.Consumes;
import javax.ws.rs.DELETE;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriBuilder;
import javax.ws.rs.core.UriInfo;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.security.AccessControlException;
import java.security.PrivilegedExceptionAction;
import java.text.MessageFormat;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;

/**
 * Main class of HttpFSServer server.
 * <p>
 * The <code>HttpFSServer</code> class uses Jersey JAX-RS to binds HTTP requests to the
 * different operations.
 */
@Path(HttpFSFileSystem.SERVICE_VERSION)
@InterfaceAudience.Private
public class HttpFSServer {
  private static Logger AUDIT_LOG = LoggerFactory.getLogger("httpfsaudit");
  private static final Logger LOG = LoggerFactory.getLogger(HttpFSServer.class);

  /**
   * Executes a {@link FileSystemAccess.FileSystemExecutor} using a filesystem for the effective
   * user.
   *
   * @param ugi user making the request.
   * @param executor FileSystemExecutor to execute.
   *
   * @return FileSystemExecutor response
   *
   * @throws IOException thrown if an IO error occurs.
   * @throws FileSystemAccessException thrown if a FileSystemAccess releated error occurred. Thrown
   * exceptions are handled by {@link HttpFSExceptionProvider}.
   */
  private <T> T fsExecute(UserGroupInformation ugi, FileSystemAccess.FileSystemExecutor<T> executor)
    throws IOException, FileSystemAccessException {
    FileSystemAccess fsAccess = HttpFSServerWebApp.get().get(FileSystemAccess.class);
    Configuration conf = HttpFSServerWebApp.get().get(FileSystemAccess.class).getFileSystemConfiguration();
    return fsAccess.execute(ugi.getShortUserName(), conf, executor);
  }

  /**
   * Returns a filesystem instance. The fileystem instance is wired for release at the completion of
   * the current Servlet request via the {@link FileSystemReleaseFilter}.
   * <p>
   * If a do-as user is specified, the current user must be a valid proxyuser, otherwise an
   * <code>AccessControlException</code> will be thrown.
   *
   * @param ugi principal for whom the filesystem instance is.
   *
   * @return a filesystem for the specified user or do-as user.
   *
   * @throws IOException thrown if an IO error occurred. Thrown exceptions are
   * handled by {@link HttpFSExceptionProvider}.
   * @throws FileSystemAccessException thrown if a FileSystemAccess releated error occurred. Thrown
   * exceptions are handled by {@link HttpFSExceptionProvider}.
   */
  private FileSystem createFileSystem(UserGroupInformation ugi)
      throws IOException, FileSystemAccessException {
    String hadoopUser = ugi.getShortUserName();
    FileSystemAccess fsAccess = HttpFSServerWebApp.get().get(FileSystemAccess.class);
    Configuration conf = HttpFSServerWebApp.get().get(FileSystemAccess.class).getFileSystemConfiguration();
    FileSystem fs = fsAccess.createFileSystem(hadoopUser, conf);
    FileSystemReleaseFilter.setFileSystem(fs);
    return fs;
  }

  private void enforceRootPath(HttpFSFileSystem.Operation op, String path) {
    if (!path.equals("/")) {
      throw new UnsupportedOperationException(
        MessageFormat.format("Operation [{0}], invalid path [{1}], must be '/'",
                             op, path));
    }
  }

  /**
   * Special binding for '/' as it is not handled by the wildcard binding.
   *
   * @param op the HttpFS operation of the request.
   * @param params the HttpFS parameters of the request.
   *
   * @return the request response.
   *
   * @throws IOException thrown if an IO error occurred. Thrown exceptions are
   * handled by {@link HttpFSExceptionProvider}.
   * @throws FileSystemAccessException thrown if a FileSystemAccess releated
   * error occurred. Thrown exceptions are handled by
   * {@link HttpFSExceptionProvider}.
   */
  @GET
  @Produces(MediaType.APPLICATION_JSON)
  public Response getRoot(@QueryParam(OperationParam.NAME) OperationParam op,
                          @Context Parameters params,
                          @Context HttpServletRequest request)
    throws IOException, FileSystemAccessException {
    return get("", op, params, request);
  }

  private String makeAbsolute(String path) {
    return "/" + ((path != null) ? path : "");
  }

  /**
   * Binding to handle GET requests, supported operations are
   *
   * @param path the path for operation.
   * @param op the HttpFS operation of the request.
   * @param params the HttpFS parameters of the request.
   *
   * @return the request response.
   *
   * @throws IOException thrown if an IO error occurred. Thrown exceptions are
   * handled by {@link HttpFSExceptionProvider}.
   * @throws FileSystemAccessException thrown if a FileSystemAccess releated
   * error occurred. Thrown exceptions are handled by
   * {@link HttpFSExceptionProvider}.
   */
  @GET
  @Path("{path:.*}")
  @Produces({MediaType.APPLICATION_OCTET_STREAM, MediaType.APPLICATION_JSON})
  public Response get(@PathParam("path") String path,
                      @QueryParam(OperationParam.NAME) OperationParam op,
                      @Context Parameters params,
                      @Context HttpServletRequest request)
    throws IOException, FileSystemAccessException {
    UserGroupInformation user = HttpUserGroupInformation.get();
    Response response;
    path = makeAbsolute(path);
    MDC.put(HttpFSFileSystem.OP_PARAM, op.value().name());
    MDC.put("hostname", request.getRemoteAddr());
    switch (op.value()) {
    case OPEN: {
      //Invoking the command directly using an unmanaged FileSystem that is
      // released by the FileSystemReleaseFilter
      final FSOperations.FSOpen command = new FSOperations.FSOpen(path);
      final FileSystem fs = createFileSystem(user);
      InputStream is = null;
      UserGroupInformation ugi = UserGroupInformation
          .createProxyUser(user.getShortUserName(),
              UserGroupInformation.getLoginUser());
      try {
        is = ugi.doAs(new PrivilegedExceptionAction<InputStream>() {
          @Override
          public InputStream run() throws Exception {
            return command.execute(fs);
          }
        });
      } catch (InterruptedException ie) {
        LOG.info("Open interrupted.", ie);
        Thread.currentThread().interrupt();
      }
      Long offset = params.get(OffsetParam.NAME, OffsetParam.class);
      Long len = params.get(LenParam.NAME, LenParam.class);
      AUDIT_LOG.info("[{}] offset [{}] len [{}]",
          new Object[] { path, offset, len });
      InputStreamEntity entity = new InputStreamEntity(is, offset, len);
      response =
          Response.ok(entity).type(MediaType.APPLICATION_OCTET_STREAM).build();
      break;
    }
    case GETFILESTATUS: {
      FSOperations.FSFileStatus command = new FSOperations.FSFileStatus(path);
      Map json = fsExecute(user, command);
      AUDIT_LOG.info("[{}]", path);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case LISTSTATUS: {
      String filter = params.get(FilterParam.NAME, FilterParam.class);
      FSOperations.FSListStatus command =
          new FSOperations.FSListStatus(path, filter);
      Map json = fsExecute(user, command);
      AUDIT_LOG.info("[{}] filter [{}]", path, (filter != null) ? filter : "-");
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case GETHOMEDIRECTORY: {
      enforceRootPath(op.value(), path);
      FSOperations.FSHomeDir command = new FSOperations.FSHomeDir();
      JSONObject json = fsExecute(user, command);
      AUDIT_LOG.info("");
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case INSTRUMENTATION: {
      enforceRootPath(op.value(), path);
      Groups groups = HttpFSServerWebApp.get().get(Groups.class);
      List<String> userGroups = groups.getGroups(user.getShortUserName());
      if (!userGroups.contains(HttpFSServerWebApp.get().getAdminGroup())) {
        throw new AccessControlException(
            "User not in HttpFSServer admin group");
      }
      Instrumentation instrumentation =
          HttpFSServerWebApp.get().get(Instrumentation.class);
      Map snapshot = instrumentation.getSnapshot();
      response = Response.ok(snapshot).build();
      break;
    }
    case GETCONTENTSUMMARY: {
      FSOperations.FSContentSummary command =
          new FSOperations.FSContentSummary(path);
      Map json = fsExecute(user, command);
      AUDIT_LOG.info("[{}]", path);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case GETQUOTAUSAGE: {
      FSOperations.FSQuotaUsage command =
          new FSOperations.FSQuotaUsage(path);
      Map json = fsExecute(user, command);
      AUDIT_LOG.info("[{}]", path);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case GETFILECHECKSUM: {
      FSOperations.FSFileChecksum command =
          new FSOperations.FSFileChecksum(path);
      Map json = fsExecute(user, command);
      AUDIT_LOG.info("[{}]", path);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case GETFILEBLOCKLOCATIONS: {
      long offset = 0;
      // In case length is not given, reset to max long
      // in order to retrieve all file block locations
      long len = Long.MAX_VALUE;
      Long offsetParam = params.get(OffsetParam.NAME, OffsetParam.class);
      Long lenParam = params.get(LenParam.NAME, LenParam.class);
      AUDIT_LOG.info("[{}] offset [{}] len [{}]",
          new Object[] {path, offsetParam, lenParam});
      if (offsetParam != null && offsetParam.longValue() > 0) {
        offset = offsetParam.longValue();
      }
      if (lenParam != null && lenParam.longValue() > 0) {
        len = lenParam.longValue();
      }
      FSOperations.FSFileBlockLocations command =
          new FSOperations.FSFileBlockLocations(path, offset, len);
      @SuppressWarnings("rawtypes") Map locations = fsExecute(user, command);
      final String json = JsonUtil.toJsonString("BlockLocations", locations);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case GETACLSTATUS: {
      FSOperations.FSAclStatus command = new FSOperations.FSAclStatus(path);
      Map json = fsExecute(user, command);
      AUDIT_LOG.info("ACL status for [{}]", path);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case GETXATTRS: {
      List<String> xattrNames =
          params.getValues(XAttrNameParam.NAME, XAttrNameParam.class);
      XAttrCodec encoding =
          params.get(XAttrEncodingParam.NAME, XAttrEncodingParam.class);
      FSOperations.FSGetXAttrs command =
          new FSOperations.FSGetXAttrs(path, xattrNames, encoding);
      @SuppressWarnings("rawtypes") Map json = fsExecute(user, command);
      AUDIT_LOG.info("XAttrs for [{}]", path);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case LISTXATTRS: {
      FSOperations.FSListXAttrs command = new FSOperations.FSListXAttrs(path);
      @SuppressWarnings("rawtypes") Map json = fsExecute(user, command);
      AUDIT_LOG.info("XAttr names for [{}]", path);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case LISTSTATUS_BATCH: {
      String startAfter = params.get(
          HttpFSParametersProvider.StartAfterParam.NAME,
          HttpFSParametersProvider.StartAfterParam.class);
      byte[] token = HttpFSUtils.EMPTY_BYTES;
      if (startAfter != null) {
        token = startAfter.getBytes(Charsets.UTF_8);
      }
      FSOperations.FSListStatusBatch command = new FSOperations
          .FSListStatusBatch(path, token);
      @SuppressWarnings("rawtypes") Map json = fsExecute(user, command);
      AUDIT_LOG.info("[{}] token [{}]", path, token);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case GETTRASHROOT: {
      FSOperations.FSTrashRoot command = new FSOperations.FSTrashRoot(path);
      JSONObject json = fsExecute(user, command);
      AUDIT_LOG.info("[{}]", path);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case GETALLSTORAGEPOLICY: {
      FSOperations.FSGetAllStoragePolicies command =
          new FSOperations.FSGetAllStoragePolicies();
      JSONObject json = fsExecute(user, command);
      AUDIT_LOG.info("[{}]", path);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    case GETSTORAGEPOLICY: {
      FSOperations.FSGetStoragePolicy command =
          new FSOperations.FSGetStoragePolicy(path);
      JSONObject json = fsExecute(user, command);
      AUDIT_LOG.info("[{}]", path);
      response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
      break;
    }
    default: {
      throw new IOException(
          MessageFormat.format("Invalid HTTP GET operation [{0}]", op.value()));
    }
    }
    return response;
  }


  /**
   * Binding to handle DELETE requests.
   *
   * @param path the path for operation.
   * @param op the HttpFS operation of the request.
   * @param params the HttpFS parameters of the request.
   *
   * @return the request response.
   *
   * @throws IOException thrown if an IO error occurred. Thrown exceptions are
   * handled by {@link HttpFSExceptionProvider}.
   * @throws FileSystemAccessException thrown if a FileSystemAccess releated
   * error occurred. Thrown exceptions are handled by
   * {@link HttpFSExceptionProvider}.
   */
  @DELETE
  @Path("{path:.*}")
  @Produces(MediaType.APPLICATION_JSON)
  public Response delete(@PathParam("path") String path,
                         @QueryParam(OperationParam.NAME) OperationParam op,
                         @Context Parameters params,
                         @Context HttpServletRequest request)
    throws IOException, FileSystemAccessException {
    UserGroupInformation user = HttpUserGroupInformation.get();
    Response response;
    path = makeAbsolute(path);
    MDC.put(HttpFSFileSystem.OP_PARAM, op.value().name());
    MDC.put("hostname", request.getRemoteAddr());
    switch (op.value()) {
      case DELETE: {
        Boolean recursive =
          params.get(RecursiveParam.NAME,  RecursiveParam.class);
        AUDIT_LOG.info("[{}] recursive [{}]", path, recursive);
        FSOperations.FSDelete command =
          new FSOperations.FSDelete(path, recursive);
        JSONObject json = fsExecute(user, command);
        response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
        break;
      }
      case DELETESNAPSHOT: {
        String snapshotName = params.get(SnapshotNameParam.NAME,
            SnapshotNameParam.class);
        FSOperations.FSDeleteSnapshot command =
                new FSOperations.FSDeleteSnapshot(path, snapshotName);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] deleted snapshot [{}]", path, snapshotName);
        response = Response.ok().build();
        break;
      }
      default: {
        throw new IOException(
          MessageFormat.format("Invalid HTTP DELETE operation [{0}]",
                               op.value()));
      }
    }
    return response;
  }

  /**
   * Binding to handle POST requests.
   *
   * @param is the inputstream for the request payload.
   * @param uriInfo the of the request.
   * @param path the path for operation.
   * @param op the HttpFS operation of the request.
   * @param params the HttpFS parameters of the request.
   *
   * @return the request response.
   *
   * @throws IOException thrown if an IO error occurred. Thrown exceptions are
   * handled by {@link HttpFSExceptionProvider}.
   * @throws FileSystemAccessException thrown if a FileSystemAccess releated
   * error occurred. Thrown exceptions are handled by
   * {@link HttpFSExceptionProvider}.
   */
  @POST
  @Path("{path:.*}")
  @Consumes({"*/*"})
  @Produces({MediaType.APPLICATION_JSON})
  public Response post(InputStream is,
                       @Context UriInfo uriInfo,
                       @PathParam("path") String path,
                       @QueryParam(OperationParam.NAME) OperationParam op,
                       @Context Parameters params,
                       @Context HttpServletRequest request)
    throws IOException, FileSystemAccessException {
    UserGroupInformation user = HttpUserGroupInformation.get();
    Response response;
    path = makeAbsolute(path);
    MDC.put(HttpFSFileSystem.OP_PARAM, op.value().name());
    MDC.put("hostname", request.getRemoteAddr());
    switch (op.value()) {
      case APPEND: {
        Boolean hasData = params.get(DataParam.NAME, DataParam.class);
        if (!hasData) {
          response = Response.temporaryRedirect(
            createUploadRedirectionURL(uriInfo,
              HttpFSFileSystem.Operation.APPEND)).build();
        } else {
          FSOperations.FSAppend command =
            new FSOperations.FSAppend(is, path);
          fsExecute(user, command);
          AUDIT_LOG.info("[{}]", path);
          response = Response.ok().type(MediaType.APPLICATION_JSON).build();
        }
        break;
      }
      case CONCAT: {
        System.out.println("HTTPFS SERVER CONCAT");
        String sources = params.get(SourcesParam.NAME, SourcesParam.class);

        FSOperations.FSConcat command =
            new FSOperations.FSConcat(path, sources.split(","));
        fsExecute(user, command);
        AUDIT_LOG.info("[{}]", path);
        System.out.println("SENT RESPONSE");
        response = Response.ok().build();
        break;
      }
      case TRUNCATE: {
        Long newLength = params.get(NewLengthParam.NAME, NewLengthParam.class);
        FSOperations.FSTruncate command = 
            new FSOperations.FSTruncate(path, newLength);
        JSONObject json = fsExecute(user, command);
        AUDIT_LOG.info("Truncate [{}] to length [{}]", path, newLength);
        response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
        break;
      }
      case UNSETSTORAGEPOLICY: {
        FSOperations.FSUnsetStoragePolicy command =
             new FSOperations.FSUnsetStoragePolicy(path);
         fsExecute(user, command);
         AUDIT_LOG.info("Unset storage policy [{}]", path);
         response = Response.ok().build();
         break;
      }
      default: {
        throw new IOException(
          MessageFormat.format("Invalid HTTP POST operation [{0}]",
                               op.value()));
      }
    }
    return response;
  }

  /**
   * Creates the URL for an upload operation (create or append).
   *
   * @param uriInfo uri info of the request.
   * @param uploadOperation operation for the upload URL.
   *
   * @return the URI for uploading data.
   */
  protected URI createUploadRedirectionURL(UriInfo uriInfo, Enum<?> uploadOperation) {
    UriBuilder uriBuilder = uriInfo.getRequestUriBuilder();
    uriBuilder = uriBuilder.replaceQueryParam(OperationParam.NAME, uploadOperation).
      queryParam(DataParam.NAME, Boolean.TRUE);
    return uriBuilder.build(null);
  }


  /**
   * Binding to handle PUT requests.
   *
   * @param is the inputstream for the request payload.
   * @param uriInfo the of the request.
   * @param path the path for operation.
   * @param op the HttpFS operation of the request.
   * @param params the HttpFS parameters of the request.
   *
   * @return the request response.
   *
   * @throws IOException thrown if an IO error occurred. Thrown exceptions are
   * handled by {@link HttpFSExceptionProvider}.
   * @throws FileSystemAccessException thrown if a FileSystemAccess releated
   * error occurred. Thrown exceptions are handled by
   * {@link HttpFSExceptionProvider}.
   */
  @PUT
  @Path("{path:.*}")
  @Consumes({"*/*"})
  @Produces({MediaType.APPLICATION_JSON})
  public Response put(InputStream is,
                       @Context UriInfo uriInfo,
                       @PathParam("path") String path,
                       @QueryParam(OperationParam.NAME) OperationParam op,
                       @Context Parameters params,
                       @Context HttpServletRequest request)
    throws IOException, FileSystemAccessException {
    UserGroupInformation user = HttpUserGroupInformation.get();
    Response response;
    path = makeAbsolute(path);
    MDC.put(HttpFSFileSystem.OP_PARAM, op.value().name());
    MDC.put("hostname", request.getRemoteAddr());
    switch (op.value()) {
      case CREATE: {
        Boolean hasData = params.get(DataParam.NAME, DataParam.class);
        if (!hasData) {
          response = Response.temporaryRedirect(
            createUploadRedirectionURL(uriInfo,
              HttpFSFileSystem.Operation.CREATE)).build();
        } else {
          Short permission = params.get(PermissionParam.NAME,
                                         PermissionParam.class);
          Boolean override = params.get(OverwriteParam.NAME,
                                        OverwriteParam.class);
          Short replication = params.get(ReplicationParam.NAME,
                                         ReplicationParam.class);
          Long blockSize = params.get(BlockSizeParam.NAME,
                                      BlockSizeParam.class);
          FSOperations.FSCreate command =
            new FSOperations.FSCreate(is, path, permission, override,
                                      replication, blockSize);
          fsExecute(user, command);
          AUDIT_LOG.info(
            "[{}] permission [{}] override [{}] replication [{}] blockSize [{}]",
            new Object[]{path, permission, override, replication, blockSize});
          response = Response.status(Response.Status.CREATED).build();
        }
        break;
      }
      case CREATESNAPSHOT: {
        String snapshotName = params.get(SnapshotNameParam.NAME,
            SnapshotNameParam.class);
        FSOperations.FSCreateSnapshot command =
            new FSOperations.FSCreateSnapshot(path, snapshotName);
        String json = fsExecute(user, command);
        AUDIT_LOG.info("[{}] snapshot created as [{}]", path, snapshotName);
        response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
        break;
      }
      case SETXATTR: {
        String xattrName = params.get(XAttrNameParam.NAME, 
            XAttrNameParam.class);
        String xattrValue = params.get(XAttrValueParam.NAME, 
            XAttrValueParam.class);
        EnumSet<XAttrSetFlag> flag = params.get(XAttrSetFlagParam.NAME, 
            XAttrSetFlagParam.class);

        FSOperations.FSSetXAttr command = new FSOperations.FSSetXAttr(
            path, xattrName, xattrValue, flag);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] to xAttr [{}]", path, xattrName);
        response = Response.ok().build();
        break;
      }
      case RENAMESNAPSHOT: {
        String oldSnapshotName = params.get(OldSnapshotNameParam.NAME,
            OldSnapshotNameParam.class);
        String snapshotName = params.get(SnapshotNameParam.NAME,
            SnapshotNameParam.class);
        FSOperations.FSRenameSnapshot command =
                new FSOperations.FSRenameSnapshot(path, oldSnapshotName,
                    snapshotName);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] renamed snapshot [{}] to [{}]", path,
            oldSnapshotName, snapshotName);
        response = Response.ok().build();
        break;
      }
      case REMOVEXATTR: {
        String xattrName = params.get(XAttrNameParam.NAME, XAttrNameParam.class);
        FSOperations.FSRemoveXAttr command = new FSOperations.FSRemoveXAttr(
            path, xattrName);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] removed xAttr [{}]", path, xattrName);
        response = Response.ok().build();
        break;
      }
      case MKDIRS: {
        Short permission = params.get(PermissionParam.NAME,
                                       PermissionParam.class);
        FSOperations.FSMkdirs command =
          new FSOperations.FSMkdirs(path, permission);
        JSONObject json = fsExecute(user, command);
        AUDIT_LOG.info("[{}] permission [{}]", path, permission);
        response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
        break;
      }
      case RENAME: {
        String toPath = params.get(DestinationParam.NAME, DestinationParam.class);
        FSOperations.FSRename command =
          new FSOperations.FSRename(path, toPath);
        JSONObject json = fsExecute(user, command);
        AUDIT_LOG.info("[{}] to [{}]", path, toPath);
        response = Response.ok(json).type(MediaType.APPLICATION_JSON).build();
        break;
      }
      case SETOWNER: {
        String owner = params.get(OwnerParam.NAME, OwnerParam.class);
        String group = params.get(GroupParam.NAME, GroupParam.class);
        FSOperations.FSSetOwner command =
          new FSOperations.FSSetOwner(path, owner, group);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] to (O/G)[{}]", path, owner + ":" + group);
        response = Response.ok().build();
        break;
      }
      case SETPERMISSION: {
        Short permission = params.get(PermissionParam.NAME,
                                      PermissionParam.class);
        FSOperations.FSSetPermission command =
          new FSOperations.FSSetPermission(path, permission);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] to [{}]", path, permission);
        response = Response.ok().build();
        break;
      }
      case SETREPLICATION: {
        Short replication = params.get(ReplicationParam.NAME,
                                       ReplicationParam.class);
        FSOperations.FSSetReplication command =
          new FSOperations.FSSetReplication(path, replication);
        JSONObject json = fsExecute(user, command);
        AUDIT_LOG.info("[{}] to [{}]", path, replication);
        response = Response.ok(json).build();
        break;
      }
      case SETTIMES: {
        Long modifiedTime = params.get(ModifiedTimeParam.NAME,
                                       ModifiedTimeParam.class);
        Long accessTime = params.get(AccessTimeParam.NAME,
                                     AccessTimeParam.class);
        FSOperations.FSSetTimes command =
          new FSOperations.FSSetTimes(path, modifiedTime, accessTime);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] to (M/A)[{}]", path,
                       modifiedTime + ":" + accessTime);
        response = Response.ok().build();
        break;
      }
      case SETACL: {
        String aclSpec = params.get(AclPermissionParam.NAME,
                AclPermissionParam.class);
        FSOperations.FSSetAcl command =
                new FSOperations.FSSetAcl(path, aclSpec);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] to acl [{}]", path, aclSpec);
        response = Response.ok().build();
        break;
      }
      case REMOVEACL: {
        FSOperations.FSRemoveAcl command =
                new FSOperations.FSRemoveAcl(path);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] removed acl", path);
        response = Response.ok().build();
        break;
      }
      case MODIFYACLENTRIES: {
        String aclSpec = params.get(AclPermissionParam.NAME,
                AclPermissionParam.class);
        FSOperations.FSModifyAclEntries command =
                new FSOperations.FSModifyAclEntries(path, aclSpec);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] modify acl entry with [{}]", path, aclSpec);
        response = Response.ok().build();
        break;
      }
      case REMOVEACLENTRIES: {
        String aclSpec = params.get(AclPermissionParam.NAME,
                AclPermissionParam.class);
        FSOperations.FSRemoveAclEntries command =
                new FSOperations.FSRemoveAclEntries(path, aclSpec);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] remove acl entry [{}]", path, aclSpec);
        response = Response.ok().build();
        break;
      }
      case REMOVEDEFAULTACL: {
        FSOperations.FSRemoveDefaultAcl command =
                new FSOperations.FSRemoveDefaultAcl(path);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] remove default acl", path);
        response = Response.ok().build();
        break;
      }
      case SETSTORAGEPOLICY: {
        String policyName = params.get(PolicyNameParam.NAME,
            PolicyNameParam.class);
        FSOperations.FSSetStoragePolicy command =
            new FSOperations.FSSetStoragePolicy(path, policyName);
        fsExecute(user, command);
        AUDIT_LOG.info("[{}] to policy [{}]", path, policyName);
        response = Response.ok().build();
        break;
      }
      default: {
        throw new IOException(
          MessageFormat.format("Invalid HTTP PUT operation [{0}]",
                               op.value()));
      }
    }
    return response;
  }

}