/**
 * 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.smartdata.server.engine;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.smartdata.AbstractService;
import org.smartdata.action.ActionException;
import org.smartdata.action.ActionRegistry;
import org.smartdata.action.SmartAction;
import org.smartdata.conf.SmartConf;
import org.smartdata.conf.SmartConfKeys;
import org.smartdata.exception.QueueFullException;
import org.smartdata.hdfs.action.move.AbstractMoveFileAction;
import org.smartdata.hdfs.scheduler.ActionSchedulerService;
import org.smartdata.metastore.MetaStore;
import org.smartdata.metastore.MetaStoreException;
import org.smartdata.model.ActionInfo;
import org.smartdata.model.CmdletDescriptor;
import org.smartdata.model.CmdletInfo;
import org.smartdata.model.CmdletState;
import org.smartdata.model.DetailedFileAction;
import org.smartdata.model.LaunchAction;
import org.smartdata.model.UserInfo;
import org.smartdata.model.WhitelistHelper;
import org.smartdata.model.action.ActionScheduler;
import org.smartdata.model.action.ScheduleResult;
import org.smartdata.protocol.message.ActionStatus;
import org.smartdata.protocol.message.ActionStatusFactory;
import org.smartdata.protocol.message.CmdletStatus;
import org.smartdata.protocol.message.CmdletStatusUpdate;
import org.smartdata.protocol.message.LaunchCmdlet;
import org.smartdata.protocol.message.StatusMessage;
import org.smartdata.protocol.message.StatusReport;
import org.smartdata.server.cluster.ActiveServerNodeCmdletMetrics;
import org.smartdata.server.cluster.NodeCmdletMetrics;
import org.smartdata.server.engine.cmdlet.CmdletDispatcher;
import org.smartdata.server.engine.cmdlet.CmdletExecutorService;
import org.smartdata.server.engine.cmdlet.TaskTracker;
import org.smartdata.utils.StringUtil;

import java.io.IOException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * When a Cmdlet is submitted, it's string descriptor will be stored into set submittedCmdlets
 * to avoid duplicated Cmdlet, then enqueue into pendingCmdlet. When the Cmdlet is scheduled it
 * will be remove out of the queue and marked in the runningCmdlets.
 *
 * <p>The map idToCmdlets stores all the recent CmdletInfos, including pending and running Cmdlets.
 * After the Cmdlet is finished or cancelled or failed, it's status will be flush to DB.
 */
public class CmdletManager extends AbstractService {
  private static final Logger LOG = LoggerFactory.getLogger(CmdletManager.class);
  public static final int TIMEOUT_MULTIPLIER = 100;
  public static final int TIMEOUT_MIN_MILLISECOND = 30000;

  private ScheduledExecutorService executorService;
  private CmdletDispatcher dispatcher;
  private MetaStore metaStore;
  private AtomicLong maxActionId;
  private AtomicLong maxCmdletId;
  // cache sync threshold
  private int cacheCmdTh;

  private int maxNumPendingCmdlets;
  private List<Long> pendingCmdlet;
  private List<Long> schedulingCmdlet;
  private Queue<Long> scheduledCmdlet;
  private Map<Long, LaunchCmdlet> idToLaunchCmdlet;
  private List<Long> runningCmdlets;
  private Map<Long, CmdletInfo> idToCmdlets;
  // Track a CmdletDescriptor from the submission to
  // the finish.
  private TaskTracker tracker;
  private Map<Long, ActionInfo> idToActions;
  private Map<Long, CmdletInfo> cacheCmd;
  private List<Long> tobeDeletedCmd;
  private ListMultimap<String, ActionScheduler> schedulers = ArrayListMultimap.create();
  private List<ActionSchedulerService> schedulerServices = new ArrayList<>();

  private AtomicLong numCmdletsGen = new AtomicLong(0);
  private AtomicLong numCmdletsFinished = new AtomicLong(0);

  private long totalScheduled = 0;

  private ActionGroup tmpActions = new ActionGroup();

  private long timeout;

  private ActionGroup cache;

  public CmdletManager(ServerContext context) throws IOException {
    super(context);

    this.metaStore = context.getMetaStore();
    this.executorService = Executors.newScheduledThreadPool(4);
    this.runningCmdlets = new ArrayList<>();
    this.pendingCmdlet = new LinkedList<>();
    this.schedulingCmdlet = new LinkedList<>();
    this.scheduledCmdlet = new LinkedBlockingQueue<>();
    this.idToLaunchCmdlet = new ConcurrentHashMap<>();
    this.idToCmdlets = new ConcurrentHashMap<>();
    this.tracker = new TaskTracker();
    this.idToActions = new ConcurrentHashMap<>();
    this.cacheCmd = new ConcurrentHashMap<>();
    this.tobeDeletedCmd = new LinkedList<>();
    this.dispatcher = new CmdletDispatcher(context, this, scheduledCmdlet,
      idToLaunchCmdlet, runningCmdlets, schedulers);
    maxNumPendingCmdlets = context.getConf()
      .getInt(SmartConfKeys.SMART_CMDLET_MAX_NUM_PENDING_KEY,
        SmartConfKeys.SMART_CMDLET_MAX_NUM_PENDING_DEFAULT);
    cacheCmdTh = context.getConf()
      .getInt(SmartConfKeys.SMART_CMDLET_CACHE_BATCH,
        SmartConfKeys.SMART_CMDLET_CACHE_BATCH_DEFAULT);

    int reportPeriod = context.getConf().getInt(SmartConfKeys.SMART_STATUS_REPORT_PERIOD_KEY,
        SmartConfKeys.SMART_STATUS_REPORT_PERIOD_DEFAULT);
    // Max interval of status report, by default 500ms.
    int maxInterval = reportPeriod * context.getConf().getInt(
        SmartConfKeys.SMART_STATUS_REPORT_PERIOD_MULTIPLIER_KEY,
        SmartConfKeys.SMART_STATUS_REPORT_PERIOD_MULTIPLIER_DEFAULT);
    // TIMEOUT_MULTIPLIER * maxInterval, 50s by default, is a potential timeout
    // value. And the least timeout value is 30s according to the below code.
    this.timeout = TIMEOUT_MULTIPLIER * maxInterval < TIMEOUT_MIN_MILLISECOND
        ? TIMEOUT_MIN_MILLISECOND : TIMEOUT_MULTIPLIER * maxInterval;
  }

  @VisibleForTesting
  public void setTimeout(long timeout) {
    this.timeout = timeout;
  }

  @VisibleForTesting
  void setDispatcher(CmdletDispatcher dispatcher) {
    this.dispatcher = dispatcher;
  }

  @Override
  public void init() throws IOException {
    LOG.info("Initializing ...");
    try {
      maxActionId = new AtomicLong(metaStore.getMaxActionId());
      maxCmdletId = new AtomicLong(metaStore.getMaxCmdletId());
      numCmdletsFinished.addAndGet(metaStore.getNumCmdletsInTerminiatedStates());

      schedulerServices = AbstractServiceFactory.createActionSchedulerServices(
        getContext().getConf(), (ServerContext) getContext(), metaStore, false);

      for (ActionSchedulerService s : schedulerServices) {
        s.init();
        List<String> actions = s.getSupportedActions();
        for (String a : actions) {
          schedulers.put(a, s);
        }
      }
      recover();
      LOG.info("Initialized.");
    } catch (MetaStoreException e) {
      LOG.error("DB Connection error! Failed to get Max CmdletId/ActionId!", e);
      throw new IOException(e);
    } catch (IOException e) {
      throw e;
    } catch (Throwable t) {
      throw new IOException(t);
    }
  }

  private void recover() throws IOException {
    reloadCmdletsInDB();
  }

  @VisibleForTesting
  public List<ActionScheduler> getSchedulers(String actionName) {
    return schedulers.get(actionName);
  }

  private void reloadCmdletsInDB() throws IOException{
    LOG.info("reloading the dispatched and pending cmdlets in DB.");
    List<CmdletInfo> cmdletInfos;
    try {
      cmdletInfos = metaStore.getCmdlets(CmdletState.DISPATCHED);
      if (cmdletInfos != null && cmdletInfos.size() != 0) {
        for (CmdletInfo cmdletInfo : cmdletInfos) {
          //track reload cmdlets
          CmdletDescriptor cmdletDescriptor =
                  CmdletDescriptor.fromCmdletString(cmdletInfo.getParameters());
          cmdletDescriptor.setRuleId(cmdletInfo.getRid());
          tracker.track(cmdletInfo.getCid(), cmdletDescriptor);
          List<ActionInfo> actionInfos = getActions(cmdletInfo.getAids());
          for (ActionInfo actionInfo: actionInfos) {
            actionInfo.setCreateTime(cmdletInfo.getGenerateTime());
            actionInfo.setFinishTime(System.currentTimeMillis());
            // Recover scheduler status according to dispatched action.
            recoverSchedulerStatus(actionInfo);
          }
          syncCmdAction(cmdletInfo, actionInfos);
        }
      }

      cmdletInfos = metaStore.getCmdlets(CmdletState.PENDING);
      if (cmdletInfos != null && cmdletInfos.size() != 0) {
        for (CmdletInfo cmdletInfo : cmdletInfos) {
          CmdletDescriptor cmdletDescriptor =
                  CmdletDescriptor.fromCmdletString(cmdletInfo.getParameters());
          cmdletDescriptor.setRuleId(cmdletInfo.getRid());
          // Pending task also needs to be tracked.
          tracker.track(cmdletInfo.getCid(), cmdletDescriptor);
          LOG.debug(String.format("Reload pending cmdlet: {}", cmdletInfo));
          List<ActionInfo> actionInfos = getActions(cmdletInfo.getAids());
          syncCmdAction(cmdletInfo, actionInfos);
        }
      }
    } catch (MetaStoreException e) {
      LOG.error("DB connection error occurs when ssm is reloading cmdlets!");
      return;
    } catch (ParseException pe) {
      LOG.error("Failed to parse cmdlet string for tracking task", pe);
    }
  }

  /**
   * Only recover scheduler status according to dispatched task.
   */
  public void recoverSchedulerStatus(ActionInfo actionInfo) {
    try {
      CmdletInfo cmdletInfo = getCmdletInfo(actionInfo.getCmdletId());
      if (cmdletInfo.getState() != CmdletState.DISPATCHED) {
        return;
      }
    } catch (IOException e) {
      return;
    }
    for (ActionScheduler p : schedulers.get(actionInfo.getActionName())) {
      p.recover(actionInfo);
    }
  }

  /**
   * Check if action names in cmdletDescriptor are correct.
   * @param cmdletDescriptor
   * @throws IOException
   */
  private void checkActionNames(
    CmdletDescriptor cmdletDescriptor) throws IOException {
    for (int index = 0; index < cmdletDescriptor.getActionSize(); index++) {
      if (!ActionRegistry
        .registeredAction(cmdletDescriptor.getActionName(index))) {
        throw new IOException(
          String.format(
            "Submit Cmdlet %s error! Action names are not correct!",
            cmdletDescriptor));
      }
    }
  }

  /**
   * Let Scheduler check actioninfo onsubmit and add them to cmdletinfo.
   * TODO: remove useless actionIndex.
   * @param cmdletInfo
   * @param actionInfos
   * @throws IOException
   */
  private void checkActionsOnSubmit(CmdletInfo cmdletInfo,
      List<ActionInfo> actionInfos) throws IOException {
    for (ActionInfo actionInfo : actionInfos) {
      cmdletInfo.addAction(actionInfo.getActionId());
    }
    int actionIndex = 0;
    for (ActionInfo actionInfo : actionInfos) {
      for (ActionScheduler p : schedulers.get(actionInfo.getActionName())) {
        if (!p.onSubmit(cmdletInfo, actionInfo, actionIndex)) {
          throw new IOException(
            String.format("Action rejected by scheduler", actionInfo));
        }
      }
      actionIndex++;
    }
  }

  @Override
  public void start() throws IOException {
    LOG.info("Starting ...");
    executorService.scheduleAtFixedRate(new CmdletPurgeTask(getContext().getConf()),
        10, 5000, TimeUnit.MILLISECONDS);
    executorService.scheduleAtFixedRate(new ScheduleTask(), 100, 50, TimeUnit.MILLISECONDS);
    executorService.scheduleAtFixedRate(new FlushCachedCmdletsTask(), 200, 50,
        TimeUnit.MILLISECONDS);
    executorService.scheduleAtFixedRate(new DetectFailedActionTask(), 1000, 5000,
        TimeUnit.MILLISECONDS);

    for (ActionSchedulerService s : schedulerServices) {
      s.start();
    }
    dispatcher.start();
    LOG.info("Started.");
  }

  @Override
  public void stop() throws IOException {
    LOG.info("Stopping ...");
    dispatcher.stop();
    for (int i = schedulerServices.size() - 1; i >= 0; i--) {
      schedulerServices.get(i).stop();
    }
    executorService.shutdown();
    cacheCmdTh = Integer.MAX_VALUE;
    try {
      batchSyncCmdAction();
    } catch (Exception e) {
      throw new IOException(e);
    }
    dispatcher.shutDownExcutorServices();
    LOG.info("Stopped.");
  }

  /**
   * Register agentExecutorService & hazelcastExecutorService.
   */
  public void registerExecutorService(CmdletExecutorService executorService) {
    dispatcher.registerExecutorService(executorService);
  }

  public void addNewUser(UserInfo userInfo) throws MetaStoreException {
    metaStore.insertUserInfo(userInfo);
  }

  public void newPassword(
      UserInfo userInfo) throws MetaStoreException {
    try {
      metaStore.newPassword(userInfo);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  /**
   * Compare userInfo(userName, password) with the one recorded in metastore.
   * @param userInfo  its password should be encrypted by SHA512.
   * @return  true if the given user info equals the one recorded in metastore.
   * @throws MetaStoreException
   */
  public boolean authentic(UserInfo userInfo) throws MetaStoreException {
    try {
      UserInfo origin = metaStore.getUserInfoByUserName(userInfo.getUserName());
      if (origin == null) {
        LOG.warn("The given user is not registered: " + userInfo.getUserName());
        return false;
      }
      return origin.equals(userInfo);
    } catch (Exception e) {
      throw new MetaStoreException(e);
    }
  }

  public long submitCmdlet(String cmdlet) throws IOException {
    LOG.debug(String.format("Received Cmdlet -> [ %s ]", cmdlet));
    try {
      if (StringUtils.isBlank(cmdlet)) {
        throw new IOException("Cannot submit an empty action!");
      }
      CmdletDescriptor cmdletDescriptor = CmdletDescriptor.fromCmdletString(cmdlet);
      return submitCmdlet(cmdletDescriptor);
    } catch (ParseException e) {
      LOG.debug("Cmdlet -> [ {} ], format is not correct", cmdlet, e);
      throw new IOException(e);
    }
  }

  public long submitCmdlet(CmdletDescriptor cmdletDescriptor) throws IOException {
    // To avoid repeatedly submitting task. If tracker contains one CmdletDescriptor
    // with the same rule id and cmdlet string, return -1.
    if (tracker.contains(cmdletDescriptor)) {
      LOG.debug("Refuse to repeatedly submit Cmdlet for {}", cmdletDescriptor);
      return -1;
    }
    if (LOG.isDebugEnabled()) {
      LOG.debug(String.format("Received Cmdlet -> [ %s ]", cmdletDescriptor.getCmdletString()));
    }
    if (maxNumPendingCmdlets <= pendingCmdlet.size() + schedulingCmdlet.size()) {
      throw new QueueFullException("Pending cmdlets exceeds value specified by key '"
          + SmartConfKeys.SMART_CMDLET_MAX_NUM_PENDING_KEY + "' = " + maxNumPendingCmdlets);
    }
    long submitTime = System.currentTimeMillis();
    CmdletInfo cmdletInfo =
      new CmdletInfo(
        maxCmdletId.getAndIncrement(),
        cmdletDescriptor.getRuleId(),
        CmdletState.PENDING,
        cmdletDescriptor.getCmdletString(),
        submitTime,
        submitTime,
        submitTime + cmdletDescriptor.getDeferIntervalMs());
    List<ActionInfo> actionInfos =
        createActionInfos(cmdletDescriptor, cmdletInfo.getCid());
    // Check action names
    checkActionNames(cmdletDescriptor);
    // Check if action path is in whitelist
    if (WhitelistHelper.isEnabled(getContext().getConf())) {
      if (!WhitelistHelper.isCmdletInWhitelist(cmdletDescriptor)) {
        throw new IOException("This path is not in the whitelist.");
      }
    }
    // Let Scheduler check actioninfo onsubmit and add them to cmdletinfo
    checkActionsOnSubmit(cmdletInfo, actionInfos);
    // Insert cmdletinfo and actionInfos to metastore and cache.
    syncCmdAction(cmdletInfo, actionInfos);
    // Track in the submission portal. For cmdlets recovered from DB
    // (see #recover), they will be not be tracked.
    tracker.track(cmdletInfo.getCid(), cmdletDescriptor);
    return cmdletInfo.getCid();
  }

  /**
   * Insert cmdletinfo and actions to metastore and cache.
   *
   * @param cmdletInfo
   * @param actionInfos
   * @throws IOException
   */
  private void syncCmdAction(CmdletInfo cmdletInfo,
      List<ActionInfo> actionInfos) throws IOException {
    LOG.debug("Cache cmd {}", cmdletInfo);
    for (ActionInfo actionInfo : actionInfos) {
      idToActions.put(actionInfo.getActionId(), actionInfo);
    }
    idToCmdlets.put(cmdletInfo.getCid(), cmdletInfo);

    if (cmdletInfo.getState() == CmdletState.PENDING) {
      numCmdletsGen.incrementAndGet();
      cacheCmd.put(cmdletInfo.getCid(), cmdletInfo);
      synchronized (pendingCmdlet) {
        pendingCmdlet.add(cmdletInfo.getCid());
      }
    } else if (cmdletInfo.getState() == CmdletState.DISPATCHED) {
      runningCmdlets.add(cmdletInfo.getCid());
      LaunchCmdlet launchCmdlet = createLaunchCmdlet(cmdletInfo);
      idToLaunchCmdlet.put(cmdletInfo.getCid(), launchCmdlet);
    }
  }

  private void batchSyncCmdAction() throws Exception {
    if (cacheCmd.size() == 0 && tobeDeletedCmd.size() == 0) {
      return;
    }
    List<CmdletInfo> cmdletInfos = new ArrayList<>();
    List<ActionInfo> actionInfos = new ArrayList<>();
    List<CmdletInfo> cmdletFinished = new ArrayList<>();
    LOG.debug("Number of cached cmds {}", cacheCmd.size());
    int todelSize;
    synchronized (cacheCmd) {
      synchronized (tobeDeletedCmd) {
        todelSize = tobeDeletedCmd.size();
        for (Long cid : tobeDeletedCmd) {
          cacheCmd.remove(cid);
        }
      }
      for (Long cid : cacheCmd.keySet()) {
        CmdletInfo cmdletInfo = cacheCmd.remove(cid);
        if (cmdletInfo.getState() != CmdletState.DISABLED) {
          cmdletInfos.add(cmdletInfo);
          for (Long aid : cmdletInfo.getAids()) {
            ActionInfo actionInfo = idToActions.get(aid);
            if (actionInfo != null) {
              actionInfos.add(actionInfo);
            }
          }
        }
        if (CmdletState.isTerminalState(cmdletInfo.getState())) {
          cmdletFinished.add(cmdletInfo);
        }
        if (cmdletInfos.size() >= cacheCmdTh) {
          break;
        }
      }

      for (CmdletInfo cmdletInfo : cmdletFinished) {
        idToCmdlets.remove(cmdletInfo.getCid());
        try {
          tracker.untrack(cmdletInfo.getCid());
        } catch (Exception e) {
          LOG.warn("Failed to untrack task!", e);
        }

        for (Long aid : cmdletInfo.getAids()) {
          idToActions.remove(aid);
        }
      }
    }

    if (cmdletInfos.size() > 0) {
      LOG.debug("Number of cmds {} to submit", cmdletInfos.size());
      try {
        metaStore.insertActions(
            actionInfos.toArray(new ActionInfo[actionInfos.size()]));
        metaStore.insertCmdlets(
            cmdletInfos.toArray(new CmdletInfo[cmdletInfos.size()]));
      } catch (MetaStoreException e) {
        LOG.error("CmdletIds -> [ {} ], submit to DB error", cmdletInfos, e);
      }
    }

    if (todelSize > 0) {
      List<Long> del = new LinkedList<>();
      synchronized (tobeDeletedCmd) {
        del.addAll(tobeDeletedCmd.subList(0, todelSize > cacheCmdTh ? cacheCmdTh : todelSize));
        tobeDeletedCmd.removeAll(del);
      }

      if (del.size() > 0) {
        LOG.debug("Number of cmds {} to delete", del.size());
        try {
          metaStore.batchDeleteCmdlet(del);
          metaStore.batchDeleteCmdletActions(del);
        } catch (MetaStoreException e) {
          LOG.error("CmdletIds -> [ {} ], delete from DB error", del, e);
        }
      }
    }
  }

  private boolean shouldStopSchedule() {
    int left = dispatcher.getTotalSlotsLeft();
    int total = dispatcher.getTotalSlots();
    if (scheduledCmdlet.size() >= left + total * 0.2) {
      return true;
    }
    return false;
  }

  private int getNumPendingScheduleCmdlets() {
    return pendingCmdlet.size() + schedulingCmdlet.size();
  }

  public void updateNodeCmdletMetrics(ActiveServerNodeCmdletMetrics metrics) {
    metrics.setMaxPendingSchedule(maxNumPendingCmdlets);
    metrics.setNumPendingSchedule(getNumPendingScheduleCmdlets());
  }

  public Collection<NodeCmdletMetrics> getAllNodeCmdletMetrics() {
    return dispatcher.getNodeCmdletMetrics();
  }

  private int scheduleCmdlet() throws IOException {
    int nScheduled = 0;

    synchronized (pendingCmdlet) {
      if (pendingCmdlet.size() > 0) {
        schedulingCmdlet.addAll(pendingCmdlet);
        pendingCmdlet.clear();
      }
    }

    long curr = System.currentTimeMillis();
    Iterator<Long> it = schedulingCmdlet.iterator();
    while (it.hasNext() && !shouldStopSchedule()) {
      long id = it.next();
      if (nScheduled % 20 == 0) {
        curr = System.currentTimeMillis();
      }
      CmdletInfo cmdlet = idToCmdlets.get(id);
      if (cmdlet == null) {
        it.remove();
        continue;
      }

      synchronized (cmdlet) {
        switch (cmdlet.getState()) {
          case CANCELLED:
          case DISABLED:
            it.remove();
            break;

          case PENDING:
            if (cmdlet.getDeferedToTime() > curr) {
              break;
            }

            LaunchCmdlet launchCmdlet = createLaunchCmdlet(cmdlet);
            ScheduleResult result;
            try {
              result = scheduleCmdletActions(cmdlet, launchCmdlet);
            } catch (Throwable t) {
              LOG.error("Schedule " + cmdlet + " failed.", t);
              result = ScheduleResult.FAIL;
            }
            if (result != ScheduleResult.RETRY) {
              it.remove();
            } else {
              continue;
            }
            try {
              if (result == ScheduleResult.SUCCESS) {
                idToLaunchCmdlet.put(cmdlet.getCid(), launchCmdlet);
                cmdlet.setState(CmdletState.SCHEDULED);
                cmdlet.setStateChangedTime(System.currentTimeMillis());
                scheduledCmdlet.add(id);
                nScheduled++;
              } else if (result == ScheduleResult.FAIL) {
                cmdlet.updateState(CmdletState.CANCELLED);
                CmdletStatus cmdletStatus = new CmdletStatus(
                    cmdlet.getCid(), cmdlet.getStateChangedTime(), cmdlet.getState());
                // Mark all actions as finished
                cmdletFinishedInternal(cmdlet, false);
                onCmdletStatusUpdate(cmdletStatus);
              } else if (result == ScheduleResult.SUCCESS_NO_EXECUTION) {
                cmdlet.updateState(CmdletState.DONE);
                cmdletFinishedInternal(cmdlet, true);
                CmdletStatus cmdletStatus = new CmdletStatus(
                    cmdlet.getCid(), cmdlet.getStateChangedTime(), cmdlet.getState());
                onCmdletStatusUpdate(cmdletStatus);
              }
            } catch (Throwable t) {
              LOG.error("Post schedule cmdlet " + cmdlet + " error.", t);
            }
            break;
        }
      }
    }
    return nScheduled;
  }

  private ScheduleResult scheduleCmdletActions(CmdletInfo info,
      LaunchCmdlet launchCmdlet) {
    List<Long> actIds = info.getAids();
    int idx = 0;
    int schIdx = 0;
    ActionInfo actionInfo;
    LaunchAction launchAction;
    List<ActionScheduler> actSchedulers;
    boolean skipped = false;
    ScheduleResult scheduleResult = ScheduleResult.SUCCESS_NO_EXECUTION;
    ScheduleResult resultTmp;
    for (idx = 0; idx < actIds.size(); idx++) {
      actionInfo = idToActions.get(actIds.get(idx));
      launchAction = launchCmdlet.getLaunchActions().get(idx);
      actSchedulers = schedulers.get(actionInfo.getActionName());
      if (actSchedulers == null || actSchedulers.size() == 0) {
        skipped = true;
        continue;
      }

      for (schIdx = 0; schIdx < actSchedulers.size(); schIdx++) {
        ActionScheduler s = actSchedulers.get(schIdx);
        try {
          resultTmp = s.onSchedule(info, actionInfo, launchCmdlet, launchAction, idx);
        } catch (Throwable t) {
          actionInfo.appendLogLine("\nOnSchedule exception: " + t);
          resultTmp = ScheduleResult.FAIL;
        }

        if (resultTmp != ScheduleResult.SUCCESS
            && resultTmp != ScheduleResult.SUCCESS_NO_EXECUTION) {
          scheduleResult = resultTmp;
        } else {
          if (scheduleResult == ScheduleResult.SUCCESS_NO_EXECUTION) {
            scheduleResult = resultTmp;
          }
        }

        if (scheduleResult != ScheduleResult.SUCCESS
            && scheduleResult != ScheduleResult.SUCCESS_NO_EXECUTION) {
          break;
        }
      }

      if (scheduleResult != ScheduleResult.SUCCESS
          && scheduleResult != ScheduleResult.SUCCESS_NO_EXECUTION) {
        break;
      }
    }

    if (scheduleResult == ScheduleResult.SUCCESS
        || scheduleResult == ScheduleResult.SUCCESS_NO_EXECUTION) {
      idx--;
      schIdx--;
      if (skipped) {
        scheduleResult = ScheduleResult.SUCCESS;
      }
    }
    postscheduleCmdletActions(info, actIds, scheduleResult, idx, schIdx);
    return scheduleResult;
  }

  private void postscheduleCmdletActions(CmdletInfo cmdletInfo,
      List<Long> actions, ScheduleResult result,
      int lastAction, int lastScheduler) {
    List<ActionScheduler> actSchedulers;
    for (int aidx = lastAction; aidx >= 0; aidx--) {
      ActionInfo info = idToActions.get(actions.get(aidx));
      actSchedulers = schedulers.get(info.getActionName());
      if (actSchedulers == null || actSchedulers.size() == 0) {
        continue;
      }
      if (lastScheduler < 0) {
        lastScheduler = actSchedulers.size() - 1;
      }

      for (int sidx = lastScheduler; sidx >= 0; sidx--) {
        try {
          actSchedulers.get(sidx).postSchedule(cmdletInfo, info, sidx, result);
        } catch (Throwable t) {
          info.setLog((info.getLog() == null ? "" : info.getLog())
              + "\nPostSchedule exception: " + t);
        }
      }

      lastScheduler = -1;
    }
  }

  private LaunchCmdlet createLaunchCmdlet(CmdletInfo cmdletInfo) {
    if (cmdletInfo == null) {
      return null;
    }
    Map<String, String> args;
    List<LaunchAction> launchActions = new ArrayList<>();
    for (Long aid : cmdletInfo.getAids()) {
      if (idToActions.containsKey(aid)) {
        ActionInfo toLaunch = idToActions.get(aid);
        args =  new HashMap<>();
        args.putAll(toLaunch.getArgs());
        launchActions.add(
          new LaunchAction(toLaunch.getActionId(), toLaunch.getActionName(), args));
      }
    }
    return new LaunchCmdlet(cmdletInfo.getCid(), launchActions);
  }

  public CmdletInfo getCmdletInfo(long cid) throws IOException {
    if (idToCmdlets.containsKey(cid)) {
      return idToCmdlets.get(cid);
    }
    try {
      return metaStore.getCmdletById(cid);
    } catch (MetaStoreException e) {
      LOG.error("CmdletId -> [ {} ], get CmdletInfo from DB error", cid, e);
      throw new IOException(e);
    }
  }

  public CmdletGroup listCmdletsInfo(long rid, long pageIndex, long numPerPage,
      List<String> orderBy,
      List<Boolean> isDesc) throws IOException, MetaStoreException {
    List<CmdletInfo> cmdlets = metaStore.listPageCmdlets(rid,
      (pageIndex - 1) * numPerPage, numPerPage, orderBy, isDesc);
    return new CmdletGroup(cmdlets, metaStore.getNumCmdletsByRid(rid));
  }

  public List<CmdletInfo> listCmdletsInfo(long rid, CmdletState cmdletState) throws IOException {
    List<CmdletInfo> result = new ArrayList<>();
    try {
      if (rid == -1) {
        result.addAll(metaStore.getCmdlets(null, null, cmdletState));
      } else {
        result.addAll(metaStore.getCmdlets(null, String.format("= %d", rid), cmdletState));
      }
    } catch (MetaStoreException e) {
      LOG.error("RuleId -> [ {} ], List CmdletInfo from DB error", rid, e);
      throw new IOException(e);
    }
    for (CmdletInfo info : idToCmdlets.values()) {
      if (info.getRid() == rid && info.getState().equals(cmdletState)) {
        result.add(info);
      }
    }
    return result;
  }

  public List<CmdletInfo> listCmdletsInfo(long rid) throws IOException {
    Map<Long, CmdletInfo> result = new HashMap<>();
    try {
      String ridCondition = rid == -1 ? null : String.format("= %d", rid);
      for (CmdletInfo info : metaStore.getCmdlets(null, ridCondition, null)) {
        result.put(info.getCid(), info);
      }
    } catch (MetaStoreException e) {
      LOG.error("RuleId -> [ {} ], List CmdletInfo from DB error", rid, e);
      throw new IOException(e);
    }
    for (CmdletInfo info : idToCmdlets.values()) {
      if (info.getRid() == rid) {
        result.put(info.getCid(), info);
      }
    }
    return Lists.newArrayList(result.values());
  }

  public void activateCmdlet(long cid) throws IOException {
    // Currently the default cmdlet status is pending, do nothing here
  }

  public void disableCmdlet(long cid) throws IOException {
    if (idToCmdlets.containsKey(cid)) {
      CmdletInfo info = idToCmdlets.get(cid);
      onCmdletStatusUpdate(
        new CmdletStatus(info.getCid(), System.currentTimeMillis(), CmdletState.DISABLED));

      synchronized (pendingCmdlet) {
        if (pendingCmdlet.contains(cid)) {
          pendingCmdlet.remove(cid);
        }
      }

      if (schedulingCmdlet.contains(cid)) {
        schedulingCmdlet.remove(cid);
      }

      if (scheduledCmdlet.contains(cid)) {
        scheduledCmdlet.remove(cid);
      }

      // Wait status update from status reporter, so need to update to MetaStore
      if (runningCmdlets.contains(cid)) {
        dispatcher.stopCmdlet(cid);
      }
    }
  }

  /**
   * Drop all unfinished cmdlets.
   *
   * @param ruleId
   * @throws IOException
   */
  public void dropRuleCmdlets(long ruleId) throws IOException {
    List<Long> cids = new ArrayList<>();
    for (CmdletInfo info : idToCmdlets.values()) {
      if (info.getRid() == ruleId && !CmdletState.isTerminalState(info.getState())) {
        cids.add(info.getCid());
      }
    }
    batchDeleteCmdlet(cids);
  }

  //Todo: optimize this function.
  private void cmdletFinished(long cmdletId) throws IOException {
    numCmdletsFinished.incrementAndGet();
    CmdletInfo cmdletInfo = idToCmdlets.get(cmdletId);
    if (cmdletInfo == null) {
      LOG.debug("CmdletInfo [id={}] does not exist in idToCmdlets.", cmdletId);
      return;
    }

    dispatcher.onCmdletFinished(cmdletInfo.getCid());
    runningCmdlets.remove(cmdletId);
    idToLaunchCmdlet.remove(cmdletId);

    flushCmdletInfo(cmdletInfo);
  }

  private void cmdletFinishedInternal(CmdletInfo cmdletInfo, boolean success) throws IOException {
    numCmdletsFinished.incrementAndGet();
    ActionInfo actionInfo;
    for (Long aid : cmdletInfo.getAids()) {
      actionInfo = idToActions.get(aid);
      synchronized (actionInfo) {
        // Set all action as finished
        actionInfo.setProgress(1.0F);
        actionInfo.setFinished(true);
        actionInfo.setCreateTime(cmdletInfo.getStateChangedTime());
        actionInfo.setFinishTime(cmdletInfo.getStateChangedTime());
        actionInfo.setExecHost(ActiveServerInfo.getInstance().getId());
        actionInfo.setSuccessful(success);
      }
    }
  }

  public void deleteCmdlet(long cid) throws IOException {
    this.disableCmdlet(cid);
    try {
      metaStore.deleteCmdlet(cid);
      metaStore.deleteCmdletActions(cid);
    } catch (MetaStoreException e) {
      LOG.error("CmdletId -> [ {} ], delete from DB error", cid, e);
      throw new IOException(e);
    }
