/**
 * 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.tools.dynamometer.workloadgenerator.audit;

import com.google.common.base.Splitter;
import org.apache.hadoop.tools.dynamometer.workloadgenerator.WorkloadDriver;
import java.io.IOException;
import java.net.URI;
import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.TimeUnit;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.counters.GenericCounter;
import org.apache.hadoop.security.UserGroupInformation;

import org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditReplayMapper.REPLAYCOUNTERS;
import org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditReplayMapper.ReplayCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditReplayMapper.INDIVIDUAL_COMMANDS_COUNTER_GROUP;
import static org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditReplayMapper.INDIVIDUAL_COMMANDS_COUNT_SUFFIX;
import static org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditReplayMapper.INDIVIDUAL_COMMANDS_INVALID_SUFFIX;
import static org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditReplayMapper.INDIVIDUAL_COMMANDS_LATENCY_SUFFIX;

/**
 * This class replays each audit log entry at a specified timestamp in the
 * future. Each of these threads maintains a {@link DelayQueue} into which items
 * are inserted by the {@link AuditReplayMapper}. Once an item is ready, this
 * thread will fetch the command from the queue and attempt to replay it.
 */
public class AuditReplayThread extends Thread {

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

  private DelayQueue<AuditReplayCommand> commandQueue;
  private ConcurrentMap<String, FileSystem> fsCache;
  private URI namenodeUri;
  private UserGroupInformation loginUser;
  private Configuration mapperConf;
  // If any exception is encountered it will be stored here
  private Exception exception;
  private long startTimestampMs;
  private boolean createBlocks;

  // Counters are not thread-safe so we store a local mapping in our thread
  // and merge them all together at the end.
  private Map<REPLAYCOUNTERS, Counter> replayCountersMap = new HashMap<>();
  private Map<String, Counter> individualCommandsMap = new HashMap<>();

  AuditReplayThread(Mapper.Context mapperContext,
      DelayQueue<AuditReplayCommand> queue,
      ConcurrentMap<String, FileSystem> fsCache) throws IOException {
    commandQueue = queue;
    this.fsCache = fsCache;
    loginUser = UserGroupInformation.getLoginUser();
    mapperConf = mapperContext.getConfiguration();
    namenodeUri = URI.create(mapperConf.get(WorkloadDriver.NN_URI));
    startTimestampMs = mapperConf.getLong(WorkloadDriver.START_TIMESTAMP_MS,
        -1);
    createBlocks = mapperConf.getBoolean(AuditReplayMapper.CREATE_BLOCKS_KEY,
        AuditReplayMapper.CREATE_BLOCKS_DEFAULT);
    LOG.info("Start timestamp: " + startTimestampMs);
    for (REPLAYCOUNTERS rc : REPLAYCOUNTERS.values()) {
      replayCountersMap.put(rc, new GenericCounter());
    }
    for (ReplayCommand replayCommand : ReplayCommand.values()) {
      individualCommandsMap.put(
          replayCommand + INDIVIDUAL_COMMANDS_COUNT_SUFFIX,
          new GenericCounter());
      individualCommandsMap.put(
          replayCommand + INDIVIDUAL_COMMANDS_LATENCY_SUFFIX,
          new GenericCounter());
      individualCommandsMap.put(
          replayCommand + INDIVIDUAL_COMMANDS_INVALID_SUFFIX,
          new GenericCounter());
    }
  }

  /**
   * Merge all of this thread's counter values into the counters contained
   * within the passed context.
   *
   * @param context The context holding the counters to increment.
   */
  void drainCounters(Mapper.Context context) {
    for (Map.Entry<REPLAYCOUNTERS, Counter> ent : replayCountersMap
        .entrySet()) {
      context.getCounter(ent.getKey()).increment(ent.getValue().getValue());
    }
    for (Map.Entry<String, Counter> ent : individualCommandsMap.entrySet()) {
      context.getCounter(INDIVIDUAL_COMMANDS_COUNTER_GROUP, ent.getKey())
          .increment(ent.getValue().getValue());
    }
  }

  /**
   * Add a command to this thread's processing queue.
   *
   * @param cmd Command to add.
   */
  void addToQueue(AuditReplayCommand cmd) {
    commandQueue.put(cmd);
  }

  /**
   * Get the Exception that caused this thread to stop running, if any, else
   * null. Should not be called until this thread has already completed (i.e.,
   * after {@link #join()} has been called).
   *
   * @return The exception which was thrown, if any.
   */
  Exception getException() {
    return exception;
  }

  @Override
  public void run() {
    long currentEpoch = System.currentTimeMillis();
    long delay = startTimestampMs - currentEpoch;
    try {
      if (delay > 0) {
        LOG.info("Sleeping for " + delay + " ms");
        Thread.sleep(delay);
      } else {
        LOG.warn("Starting late by " + (-1 * delay) + " ms");
      }

      AuditReplayCommand cmd = commandQueue.take();
      while (!cmd.isPoison()) {
        replayCountersMap.get(REPLAYCOUNTERS.TOTALCOMMANDS).increment(1);
        delay = cmd.getDelay(TimeUnit.MILLISECONDS);
        if (delay < -5) { // allow some tolerance here
          replayCountersMap.get(REPLAYCOUNTERS.LATECOMMANDS).increment(1);
          replayCountersMap.get(REPLAYCOUNTERS.LATECOMMANDSTOTALTIME)
              .increment(-1 * delay);
        }
        if (!replayLog(cmd)) {
          replayCountersMap.get(REPLAYCOUNTERS.TOTALINVALIDCOMMANDS)
              .increment(1);
        }
        cmd = commandQueue.take();
      }
    } catch (InterruptedException e) {
      LOG.error("Interrupted; exiting from thread.", e);
    } catch (Exception e) {
      exception = e;
      LOG.error("ReplayThread encountered exception; exiting.", e);
    }
  }

  /**
   * Attempt to replay the provided command. Updates counters accordingly.
   *
   * @param command The command to replay
   * @return True iff the command was successfully replayed (i.e., no exceptions
   *         were thrown).
   */
  private boolean replayLog(final AuditReplayCommand command) {
    final String src = command.getSrc();
    final String dst = command.getDest();
    FileSystem proxyFs = fsCache.get(command.getSimpleUgi());
    if (proxyFs == null) {
      UserGroupInformation ugi = UserGroupInformation
          .createProxyUser(command.getSimpleUgi(), loginUser);
      proxyFs = ugi.doAs((PrivilegedAction<FileSystem>) () -> {
        try {
          FileSystem fs = new DistributedFileSystem();
          fs.initialize(namenodeUri, mapperConf);
          return fs;
        } catch (IOException ioe) {
          throw new RuntimeException(ioe);
        }
      });
      fsCache.put(command.getSimpleUgi(), proxyFs);
    }
    final FileSystem fs = proxyFs;
    ReplayCommand replayCommand;
    try {
      replayCommand = ReplayCommand
          .valueOf(command.getCommand().split(" ")[0].toUpperCase());
    } catch (IllegalArgumentException iae) {
      LOG.warn("Unsupported/invalid command: " + command);
      replayCountersMap.get(REPLAYCOUNTERS.TOTALUNSUPPORTEDCOMMANDS)
          .increment(1);
      return false;
    }
    try {
      long startTime = System.currentTimeMillis();
      switch (replayCommand) {
      case CREATE:
        FSDataOutputStream fsDos = fs.create(new Path(src));
        if (createBlocks) {
          fsDos.writeByte(0);
        }
        fsDos.close();
        break;
      case GETFILEINFO:
        fs.getFileStatus(new Path(src));
        break;
      case CONTENTSUMMARY:
        fs.getContentSummary(new Path(src));
        break;
      case MKDIRS:
        fs.mkdirs(new Path(src));
        break;
      case RENAME:
        fs.rename(new Path(src), new Path(dst));
        break;
      case LISTSTATUS:
        ((DistributedFileSystem) fs).getClient().listPaths(src,
            HdfsFileStatus.EMPTY_NAME);
        break;
      case APPEND:
        fs.append(new Path(src));
        return true;
      case DELETE:
        fs.delete(new Path(src), true);
        break;
      case OPEN:
        fs.open(new Path(src)).close();
        break;
      case SETPERMISSION:
        fs.setPermission(new Path(src), FsPermission.getDefault());
        break;
      case SETOWNER:
        fs.setOwner(new Path(src),
            UserGroupInformation.getCurrentUser().getShortUserName(),
            UserGroupInformation.getCurrentUser().getPrimaryGroupName());
        break;
      case SETTIMES:
        fs.setTimes(new Path(src), System.currentTimeMillis(),
            System.currentTimeMillis());
        break;
      case SETREPLICATION:
        fs.setReplication(new Path(src), (short) 1);
        break;
      case CONCAT:
        // dst is like [path1, path2] - strip brackets and split on comma
        String bareDist = dst.length() < 2 ? ""
            : dst.substring(1, dst.length() - 1).trim();
        List<Path> dsts = new ArrayList<>();
        for (String s : Splitter.on(",").omitEmptyStrings().trimResults()
            .split(bareDist)) {
          dsts.add(new Path(s));
        }
        fs.concat(new Path(src), dsts.toArray(new Path[] {}));
        break;
      default:
        throw new RuntimeException("Unexpected command: " + replayCommand);
      }
      long latency = System.currentTimeMillis() - startTime;
      switch (replayCommand.getType()) {
      case WRITE:
        replayCountersMap.get(REPLAYCOUNTERS.TOTALWRITECOMMANDLATENCY)
            .increment(latency);
        replayCountersMap.get(REPLAYCOUNTERS.TOTALWRITECOMMANDS).increment(1);
        break;
      case READ:
        replayCountersMap.get(REPLAYCOUNTERS.TOTALREADCOMMANDLATENCY)
            .increment(latency);
        replayCountersMap.get(REPLAYCOUNTERS.TOTALREADCOMMANDS).increment(1);
        break;
      default:
        throw new RuntimeException("Unexpected command type: "
            + replayCommand.getType());
      }
      individualCommandsMap
          .get(replayCommand + INDIVIDUAL_COMMANDS_LATENCY_SUFFIX)
          .increment(latency);
      individualCommandsMap
          .get(replayCommand + INDIVIDUAL_COMMANDS_COUNT_SUFFIX).increment(1);
      return true;
    } catch (IOException e) {
      LOG.debug("IOException: " + e.getLocalizedMessage());
      individualCommandsMap
          .get(replayCommand + INDIVIDUAL_COMMANDS_INVALID_SUFFIX).increment(1);
      return false;
    }
  }
}
