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

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

import org.assertj.core.api.Assertions;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.hadoop.fs.s3a.commit.files.PersistentCommitData;
import org.apache.hadoop.fs.s3a.commit.files.SinglePendingCommit;
import org.apache.hadoop.fs.s3a.commit.impl.CommitOperations;
import org.apache.hadoop.fs.s3a.performance.AbstractS3ACostTest;
import org.apache.hadoop.fs.statistics.IOStatisticsLogging;

import static org.apache.hadoop.fs.s3a.Statistic.ACTION_HTTP_GET_REQUEST;
import static org.apache.hadoop.fs.s3a.Statistic.COMMITTER_MAGIC_FILES_CREATED;
import static org.apache.hadoop.fs.s3a.Statistic.COMMITTER_MAGIC_MARKER_PUT;
import static org.apache.hadoop.fs.s3a.Statistic.DIRECTORIES_CREATED;
import static org.apache.hadoop.fs.s3a.Statistic.FAKE_DIRECTORIES_DELETED;
import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_BULK_DELETE_REQUEST;
import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_DELETE_REQUEST;
import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_LIST_REQUEST;
import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_METADATA_REQUESTS;
import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_MULTIPART_UPLOAD_INITIATED;
import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_PUT_REQUESTS;
import static org.apache.hadoop.fs.s3a.commit.CommitConstants.MAGIC_PATH_PREFIX;
import static org.apache.hadoop.fs.s3a.commit.CommitterTestHelper.JOB_ID;
import static org.apache.hadoop.fs.s3a.commit.CommitterTestHelper.assertIsMagicStream;
import static org.apache.hadoop.fs.s3a.commit.CommitterTestHelper.makeMagic;
import static org.apache.hadoop.fs.s3a.performance.OperationCost.LIST_FILES_LIST_OP;
import static org.apache.hadoop.fs.s3a.performance.OperationCost.LIST_OPERATION;
import static org.apache.hadoop.fs.s3a.performance.OperationCost.NO_HEAD_OR_LIST;
import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.ioStatisticsToPrettyString;
import static org.apache.hadoop.util.functional.RemoteIterators.toList;

/**
 * Assert cost of commit operations;
 * <ol>
 *   <li>Even on marker deleting filesystems,
 *       operations under magic dirs do not trigger marker deletion.</li>
 *   <li>Loading pending files from FileStatus entries skips HEAD checks.</li>
 *   <li>Mkdir under magic dirs doesn't check ancestor or dest type</li>
 * </ol>
 */
public class ITestCommitOperationCost extends AbstractS3ACostTest {

  private static final Logger LOG =
      LoggerFactory.getLogger(ITestCommitOperationCost.class);

  /**
   * Helper for the tests.
   */
  private CommitterTestHelper testHelper;

  /**
   * Create with markers kept, always.
   */
  public ITestCommitOperationCost() {
    super(false);
  }

  @Override
  public void setup() throws Exception {
    super.setup();
    testHelper = new CommitterTestHelper(getFileSystem());
  }

  @Override
  public void teardown() throws Exception {
    try {
      if (testHelper != null) {
        testHelper.abortMultipartUploadsUnderPath(methodPath());
      }
    } finally {
      super.teardown();
    }
  }

  /**
   * Get a method-relative path.
   * @param filename filename
   * @return new path
   * @throws IOException failure to create/parse the path.
   */
  private Path methodSubPath(String filename) throws IOException {
    return new Path(methodPath(), filename);
  }

  /**
   * Return the FS IOStats, prettified.
   * @return string for assertions.
   */
  protected String fileSystemIOStats() {
    return ioStatisticsToPrettyString(getFileSystem().getIOStatistics());
  }

  @Test
  public void testMagicMkdir() throws Throwable {
    describe("Mkdirs 'MAGIC PATH' always skips dir marker deletion");
    S3AFileSystem fs = getFileSystem();
    Path baseDir = methodPath();
    // create dest dir marker, always
    fs.mkdirs(baseDir);
    Path magicDir = new Path(baseDir, MAGIC_PATH_PREFIX + JOB_ID);
    verifyMetrics(() -> {
      fs.mkdirs(magicDir);
      return fileSystemIOStats();
    },
        with(OBJECT_BULK_DELETE_REQUEST, 0),
        with(OBJECT_DELETE_REQUEST, 0),
        with(DIRECTORIES_CREATED, 1));
    verifyMetrics(() -> {
      fs.delete(magicDir, true);
      return fileSystemIOStats();
    },
        with(OBJECT_BULK_DELETE_REQUEST, 0),
        with(OBJECT_DELETE_REQUEST, 1),
        with(DIRECTORIES_CREATED, 0));
    assertPathExists("parent", baseDir);
  }

  /**
   * When a magic subdir is deleted, parent dirs are not recreated.
   */
  @Test
  public void testMagicMkdirs() throws Throwable {
    describe("Mkdirs __magic_job-<jobId>/subdir always skips dir marker deletion");
    S3AFileSystem fs = getFileSystem();
    Path baseDir = methodPath();
    Path magicDir = new Path(baseDir, MAGIC_PATH_PREFIX + JOB_ID);
    fs.delete(baseDir, true);

    Path magicSubdir = new Path(magicDir, "subdir");
    verifyMetrics(() -> {
      fs.mkdirs(magicSubdir, FsPermission.getDirDefault());
      return "after mkdirs " + fileSystemIOStats();
    },
        always(LIST_OPERATION),
        with(OBJECT_BULK_DELETE_REQUEST, 0),
        with(OBJECT_DELETE_REQUEST, 0),
        with(DIRECTORIES_CREATED, 1));
    assertPathExists("magicSubdir", magicSubdir);

    verifyMetrics(() -> {
      fs.delete(magicSubdir, true);
      return "after delete " + fileSystemIOStats();
    },
        with(OBJECT_BULK_DELETE_REQUEST, 0),
        with(OBJECT_DELETE_REQUEST, 1),
        with(OBJECT_LIST_REQUEST, 1),
        with(OBJECT_METADATA_REQUESTS, 1),
        with(DIRECTORIES_CREATED, 0));
    // no marker dir creation
    assertPathDoesNotExist("magicDir", magicDir);
    assertPathDoesNotExist("baseDir", baseDir);
  }

  /**
   * Active stream; a field is used so closures can write to
   * it.
   */
  private FSDataOutputStream stream;

  /**
   * Abort any active stream.
   * @throws IOException failure
   */
  private void abortActiveStream() throws IOException {
    if (stream != null) {
      stream.abort();
      stream.close();
    }
  }

  @Test
  public void testCostOfCreatingMagicFile() throws Throwable {
    describe("Files created under magic paths skip existence checks and marker deletes");
    S3AFileSystem fs = getFileSystem();
    Path destFile = methodSubPath("file.txt");
    fs.delete(destFile.getParent(), true);
    Path magicDest = makeMagic(destFile);

    // when the file is created, there is no check for overwrites
    // or the dest being a directory, even if overwrite=false
    try {
      verifyMetrics(() -> {
        stream = fs.create(magicDest, false);
        return stream.toString();
      },
          always(NO_HEAD_OR_LIST),
          with(COMMITTER_MAGIC_FILES_CREATED, 1),
          with(COMMITTER_MAGIC_MARKER_PUT, 0),
          with(OBJECT_MULTIPART_UPLOAD_INITIATED, 1));
      assertIsMagicStream(stream);

      stream.write("hello".getBytes(StandardCharsets.UTF_8));

      // when closing, there will be no directories deleted
      // we do expect two PUT requests, because the marker and manifests
      // are both written
      LOG.info("closing magic stream to {}", magicDest);
      verifyMetrics(() -> {
        stream.close();
        return stream.toString();
      },
          always(NO_HEAD_OR_LIST),
          with(OBJECT_PUT_REQUESTS, 2),
          with(COMMITTER_MAGIC_MARKER_PUT, 2),
          with(OBJECT_BULK_DELETE_REQUEST, 0),
          with(OBJECT_DELETE_REQUEST, 0));

    } catch (Exception e) {
      abortActiveStream();
      throw e;
    }
    // list the manifests
    final CommitOperations commitOperations = new CommitOperations(fs);
    List<LocatedFileStatus> pending = verifyMetrics(() ->
            toList(commitOperations.
                locateAllSinglePendingCommits(magicDest.getParent(), false)),
        always(LIST_FILES_LIST_OP));
    Assertions.assertThat(pending)
        .describedAs("pending commits")
        .hasSize(1);

    // load the only pending commit
    SinglePendingCommit singleCommit = verifyMetrics(() ->
            PersistentCommitData.load(fs,
                pending.get(0),
                SinglePendingCommit.serializer()),
        always(NO_HEAD_OR_LIST),
        with(ACTION_HTTP_GET_REQUEST, 1));

    // commit it through the commit operations.
    verifyMetrics(() -> {
      commitOperations.commitOrFail(singleCommit);
      return ioStatisticsToPrettyString(
          commitOperations.getIOStatistics());
    },
        always(NO_HEAD_OR_LIST),  // no probes for the dest path
        with(FAKE_DIRECTORIES_DELETED, 0),  // no fake dirs
        with(OBJECT_DELETE_REQUEST, 0)); // no deletes

    LOG.info("Final Statistics {}",
        IOStatisticsLogging.ioStatisticsToPrettyString(stream.getIOStatistics()));
  }

  /**
   * saving pending files MUST NOT trigger HEAD/LIST calls
   * when created under a magic path; when opening
   * with an S3AFileStatus the HEAD will be skipped too.
   */
  @Test
  public void testCostOfSavingLoadingPendingFile() throws Throwable {
    describe("Verify costs of saving .pending file under a magic path");

    S3AFileSystem fs = getFileSystem();
    Path partDir = methodSubPath("file.pending");
    Path destFile = new Path(partDir, "file.pending");
    Path magicDest = makeMagic(destFile);
    // create a pending file with minimal values needed
    // for validation to work
    final SinglePendingCommit commit = new SinglePendingCommit();
    commit.touch(System.currentTimeMillis());
    commit.setUri(destFile.toUri().toString());
    commit.setBucket(fs.getBucket());
    commit.setLength(0);
    commit.setDestinationKey(fs.pathToKey(destFile));
    commit.setUploadId("uploadId");
    commit.setEtags(new ArrayList<>());
    // fail fast if the commit data is incomplete
    commit.validate();

    // save the file: no checks will be made
    verifyMetrics(() -> {
          commit.save(fs, magicDest,
              SinglePendingCommit.serializer());
          return commit.toString();
        },
        with(COMMITTER_MAGIC_FILES_CREATED, 0),
        always(NO_HEAD_OR_LIST),
        with(OBJECT_BULK_DELETE_REQUEST, 0),
        with(OBJECT_DELETE_REQUEST, 0)
    );

    LOG.info("File written; Validating");
    testHelper.assertFileLacksMarkerHeader(magicDest);
    FileStatus status = fs.getFileStatus(magicDest);

    LOG.info("Reading file {}", status);
    // opening a file with a status passed in will skip the HEAD
    verifyMetrics(() ->
            PersistentCommitData.load(fs, status, SinglePendingCommit.serializer()),
        always(NO_HEAD_OR_LIST),
        with(ACTION_HTTP_GET_REQUEST, 1));
  }

}
