/**
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.hdds.scm.block;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.primitives.Longs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto
    .StorageContainerDatanodeProtocolProtos.ContainerBlocksDeletionACKProto;
import org.apache.hadoop.hdds.protocol.proto
    .StorageContainerDatanodeProtocolProtos.ContainerBlocksDeletionACKProto
    .DeleteBlockTransactionResult;
import org.apache.hadoop.hdds.scm.command
    .CommandStatusReportHandler.DeleteBlockStatus;
import org.apache.hadoop.hdds.scm.container.ContainerManager;
import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
import org.apache.hadoop.hdds.server.events.EventHandler;
import org.apache.hadoop.hdds.server.events.EventPublisher;
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdds.protocol.proto
    .StorageContainerDatanodeProtocolProtos.DeletedBlocksTransaction;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.utils.BatchOperation;
import org.apache.hadoop.utils.MetadataStore;
import org.apache.hadoop.utils.MetadataStoreBuilder;
import org.eclipse.jetty.util.ConcurrentHashSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

import static java.lang.Math.min;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys
    .OZONE_SCM_BLOCK_DELETION_MAX_RETRY;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys
    .OZONE_SCM_BLOCK_DELETION_MAX_RETRY_DEFAULT;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys
    .OZONE_SCM_DB_CACHE_SIZE_DEFAULT;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys
    .OZONE_SCM_DB_CACHE_SIZE_MB;
import static org.apache.hadoop.hdds.server.ServerUtils.getOzoneMetaDirPath;
import static org.apache.hadoop.ozone.OzoneConsts.DELETED_BLOCK_DB;

/**
 * A implement class of {@link DeletedBlockLog}, and it uses
 * K/V db to maintain block deletion transactions between scm and datanode.
 * This is a very basic implementation, it simply scans the log and
 * memorize the position that scanned by last time, and uses this to
 * determine where the next scan starts. It has no notion about weight
 * of each transaction so as long as transaction is still valid, they get
 * equally same chance to be retrieved which only depends on the nature
 * order of the transaction ID.
 */
public class DeletedBlockLogImpl
    implements DeletedBlockLog, EventHandler<DeleteBlockStatus> {

  public static final Logger LOG =
      LoggerFactory.getLogger(DeletedBlockLogImpl.class);

  private static final byte[] LATEST_TXID =
      DFSUtil.string2Bytes("#LATEST_TXID#");

  private final int maxRetry;
  private final MetadataStore deletedStore;
  private final ContainerManager containerManager;
  private final Lock lock;
  // The latest id of deleted blocks in the db.
  private long lastTxID;
  // Maps txId to set of DNs which are successful in committing the transaction
  private Map<Long, Set<UUID>> transactionToDNsCommitMap;

  public DeletedBlockLogImpl(Configuration conf,
     ContainerManager containerManager) throws IOException {
    maxRetry = conf.getInt(OZONE_SCM_BLOCK_DELETION_MAX_RETRY,
        OZONE_SCM_BLOCK_DELETION_MAX_RETRY_DEFAULT);

    File metaDir = getOzoneMetaDirPath(conf);
    String scmMetaDataDir = metaDir.getPath();
    File deletedLogDbPath = new File(scmMetaDataDir, DELETED_BLOCK_DB);
    int cacheSize = conf.getInt(OZONE_SCM_DB_CACHE_SIZE_MB,
        OZONE_SCM_DB_CACHE_SIZE_DEFAULT);
    // Load store of all transactions.
    deletedStore = MetadataStoreBuilder.newBuilder()
        .setCreateIfMissing(true)
        .setConf(conf)
        .setDbFile(deletedLogDbPath)
        .setCacheSize(cacheSize * OzoneConsts.MB)
        .build();
    this.containerManager = containerManager;

    this.lock = new ReentrantLock();
    // start from the head of deleted store.
    lastTxID = findLatestTxIDInStore();

    // transactionToDNsCommitMap is updated only when
    // transaction is added to the log and when it is removed.

    // maps transaction to dns which have committed it.
    transactionToDNsCommitMap = new ConcurrentHashMap<>();
  }

  @VisibleForTesting
  public MetadataStore getDeletedStore() {
    return deletedStore;
  }

  /**
   * There is no need to lock before reading because
   * it's only used in construct method.
   *
   * @return latest txid.
   * @throws IOException
   */
  private long findLatestTxIDInStore() throws IOException {
    long txid = 0;
    byte[] value = deletedStore.get(LATEST_TXID);
    if (value != null) {
      txid = Longs.fromByteArray(value);
    }
    return txid;
  }

  @Override
  public List<DeletedBlocksTransaction> getFailedTransactions()
      throws IOException {
    lock.lock();
    try {
      final List<DeletedBlocksTransaction> failedTXs = Lists.newArrayList();
      deletedStore.iterate(null, (key, value) -> {
        if (!Arrays.equals(LATEST_TXID, key)) {
          DeletedBlocksTransaction delTX =
              DeletedBlocksTransaction.parseFrom(value);
          if (delTX.getCount() == -1) {
            failedTXs.add(delTX);
          }
        }
        return true;
      });
      return failedTXs;
    } finally {
      lock.unlock();
    }
  }

  /**
   * {@inheritDoc}
   *
   * @param txIDs - transaction ID.
   * @throws IOException
   */
  @Override
  public void incrementCount(List<Long> txIDs) throws IOException {
    BatchOperation batch = new BatchOperation();
    lock.lock();
    try {
      for(Long txID : txIDs) {
        try {
          byte[] deleteBlockBytes =
              deletedStore.get(Longs.toByteArray(txID));
          if (deleteBlockBytes == null) {
            LOG.warn("Delete txID {} not found", txID);
            continue;
          }
          DeletedBlocksTransaction block = DeletedBlocksTransaction
              .parseFrom(deleteBlockBytes);
          DeletedBlocksTransaction.Builder builder = block.toBuilder();
          int currentCount = block.getCount();
          if (currentCount > -1) {
            builder.setCount(++currentCount);
          }
          // if the retry time exceeds the maxRetry value
          // then set the retry value to -1, stop retrying, admins can
          // analyze those blocks and purge them manually by SCMCli.
          if (currentCount > maxRetry) {
            builder.setCount(-1);
          }
          deletedStore.put(Longs.toByteArray(txID),
              builder.build().toByteArray());
        } catch (IOException ex) {
          LOG.warn("Cannot increase count for txID " + txID, ex);
        }
      }
      deletedStore.writeBatch(batch);
    } finally {
      lock.unlock();
    }
  }

  private DeletedBlocksTransaction constructNewTransaction(long txID,
      long containerID, List<Long> blocks) {
    return DeletedBlocksTransaction.newBuilder()
        .setTxID(txID)
        .setContainerID(containerID)
        .addAllLocalID(blocks)
        .setCount(0)
        .build();
  }

  /**
   * {@inheritDoc}
   *
   * @param transactionResults - transaction IDs.
   * @param dnID - Id of Datanode which has acknowledged a delete block command.
   * @throws IOException
   */
  @Override
  public void commitTransactions(
      List<DeleteBlockTransactionResult> transactionResults, UUID dnID) {
    lock.lock();
    try {
      Set<UUID> dnsWithCommittedTxn;
      for (DeleteBlockTransactionResult transactionResult :
          transactionResults) {
        if (isTransactionFailed(transactionResult)) {
          continue;
        }
        try {
          long txID = transactionResult.getTxID();
          // set of dns which have successfully committed transaction txId.
          dnsWithCommittedTxn = transactionToDNsCommitMap.get(txID);
          Long containerId = transactionResult.getContainerID();
          if (dnsWithCommittedTxn == null) {
            LOG.warn("Transaction txId={} commit by dnId={} for containerID={} "
                    + "failed. Corresponding entry not found.", txID, dnID,
                containerId);
            return;
          }

          dnsWithCommittedTxn.add(dnID);
          Pipeline pipeline =
              containerManager.getContainerWithPipeline(containerId)
                  .getPipeline();
          Collection<DatanodeDetails> containerDnsDetails =
              pipeline.getDatanodes().values();
          // The delete entry can be safely removed from the log if all the
          // corresponding nodes commit the txn. It is required to check that
          // the nodes returned in the pipeline match the replication factor.
          if (min(containerDnsDetails.size(), dnsWithCommittedTxn.size())
              >= pipeline.getFactor().getNumber()) {
            List<UUID> containerDns = containerDnsDetails.stream()
                .map(DatanodeDetails::getUuid)
                .collect(Collectors.toList());
            if (dnsWithCommittedTxn.containsAll(containerDns)) {
              transactionToDNsCommitMap.remove(txID);
              LOG.debug("Purging txId={} from block deletion log", txID);
              deletedStore.delete(Longs.toByteArray(txID));
            }
          }
          LOG.debug("Datanode txId={} containerId={} committed by dnId={}",
              txID, containerId, dnID);
        } catch (IOException e) {
          LOG.warn("Could not commit delete block transaction: " +
              transactionResult.getTxID(), e);
        }
      }
    } finally {
      lock.unlock();
    }
  }

  private boolean isTransactionFailed(DeleteBlockTransactionResult result) {
    if (LOG.isDebugEnabled()) {
      LOG.debug(
          "Got block deletion ACK from datanode, TXIDs={}, " + "success={}",
          result.getTxID(), result.getSuccess());
    }
    if (!result.getSuccess()) {
      LOG.warn("Got failed ACK for TXID={}, prepare to resend the "
          + "TX in next interval", result.getTxID());
      return true;
    }
    return false;
  }

  /**
   * {@inheritDoc}
   *
   * @param containerID - container ID.
   * @param blocks - blocks that belong to the same container.
   * @throws IOException
   */
  @Override
  public void addTransaction(long containerID, List<Long> blocks)
      throws IOException {
    BatchOperation batch = new BatchOperation();
    lock.lock();
    try {
      DeletedBlocksTransaction tx = constructNewTransaction(lastTxID + 1,
          containerID, blocks);
      byte[] key = Longs.toByteArray(lastTxID + 1);

      batch.put(key, tx.toByteArray());
      batch.put(LATEST_TXID, Longs.toByteArray(lastTxID + 1));

      deletedStore.writeBatch(batch);
      lastTxID += 1;
    } finally {
      lock.unlock();
    }
  }

  @Override
  public int getNumOfValidTransactions() throws IOException {
    lock.lock();
    try {
      final AtomicInteger num = new AtomicInteger(0);
      deletedStore.iterate(null, (key, value) -> {
        // Exclude latest txid record
        if (!Arrays.equals(LATEST_TXID, key)) {
          DeletedBlocksTransaction delTX =
              DeletedBlocksTransaction.parseFrom(value);
          if (delTX.getCount() > -1) {
            num.incrementAndGet();
          }
        }
        return true;
      });
      return num.get();
    } finally {
      lock.unlock();
    }
  }

  /**
   * {@inheritDoc}
   *
   * @param containerBlocksMap a map of containerBlocks.
   * @throws IOException
   */
  @Override
  public void addTransactions(
      Map<Long, List<Long>> containerBlocksMap)
      throws IOException {
    BatchOperation batch = new BatchOperation();
    lock.lock();
    try {
      long currentLatestID = lastTxID;
      for (Map.Entry<Long, List<Long>> entry :
          containerBlocksMap.entrySet()) {
        currentLatestID += 1;
        byte[] key = Longs.toByteArray(currentLatestID);
        DeletedBlocksTransaction tx = constructNewTransaction(currentLatestID,
            entry.getKey(), entry.getValue());
        batch.put(key, tx.toByteArray());
      }
      lastTxID = currentLatestID;
      batch.put(LATEST_TXID, Longs.toByteArray(lastTxID));
      deletedStore.writeBatch(batch);
    } finally {
      lock.unlock();
    }
  }

  @Override
  public void close() throws IOException {
    if (deletedStore != null) {
      deletedStore.close();
    }
  }

  @Override
  public Map<Long, Long> getTransactions(
      DatanodeDeletedBlockTransactions transactions) throws IOException {
    lock.lock();
    try {
      Map<Long, Long> deleteTransactionMap = new HashMap<>();
      deletedStore.iterate(null, (key, value) -> {
        if (!Arrays.equals(LATEST_TXID, key)) {
          DeletedBlocksTransaction block = DeletedBlocksTransaction
              .parseFrom(value);

          if (block.getCount() > -1 && block.getCount() <= maxRetry) {
            if (transactions.addTransaction(block,
                transactionToDNsCommitMap.get(block.getTxID()))) {
              deleteTransactionMap.put(block.getContainerID(), block.getTxID());
              transactionToDNsCommitMap
                  .putIfAbsent(block.getTxID(), new ConcurrentHashSet<>());
            }
          }
          return !transactions.isFull();
        }
        return true;
      });
      return deleteTransactionMap;
    } finally {
      lock.unlock();
    }
  }

  @Override
  public void onMessage(DeleteBlockStatus deleteBlockStatus,
      EventPublisher publisher) {
    ContainerBlocksDeletionACKProto ackProto =
        deleteBlockStatus.getCmdStatus().getBlockDeletionAck();
    commitTransactions(ackProto.getResultsList(),
        UUID.fromString(ackProto.getDnId()));
  }
}