/*
 * 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.solr.handler.admin;

import static org.apache.solr.common.cloud.DocCollection.DOC_ROUTER;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.commons.lang.StringUtils;
import org.apache.lucene.index.DirectoryReader;
import org.apache.lucene.search.MatchAllDocsQuery;
import org.apache.lucene.store.Directory;
import org.apache.lucene.util.IOUtils;
import org.apache.solr.cloud.CloudDescriptor;
import org.apache.solr.cloud.Overseer;
import org.apache.solr.cloud.SyncStrategy;
import org.apache.solr.cloud.ZkController;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.DocRouter;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.CoreAdminParams;
import org.apache.solr.common.params.CoreAdminParams.CoreAdminAction;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.params.UpdateParams;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.SimpleOrderedMap;
import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.CoreDescriptor;
import org.apache.solr.core.DirectoryFactory;
import org.apache.solr.core.DirectoryFactory.DirContext;
import org.apache.solr.core.SolrCore;
import org.apache.solr.core.SolrXMLCoresLocator;
import org.apache.solr.handler.RequestHandlerBase;
import org.apache.solr.request.LocalSolrQueryRequest;
import org.apache.solr.request.SolrQueryRequest;
import org.apache.solr.response.SolrQueryResponse;
import org.apache.solr.search.SolrIndexSearcher;
import org.apache.solr.update.CommitUpdateCommand;
import org.apache.solr.update.MergeIndexesCommand;
import org.apache.solr.update.SplitIndexCommand;
import org.apache.solr.update.UpdateLog;
import org.apache.solr.update.processor.UpdateRequestProcessor;
import org.apache.solr.update.processor.UpdateRequestProcessorChain;
import org.apache.solr.util.DefaultSolrThreadFactory;
import org.apache.solr.util.NumberUtils;
import org.apache.solr.util.RefCounted;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;

/**
 *
 * @since solr 1.3
 */
public class CoreAdminHandler extends RequestHandlerBase {
  protected static Logger log = LoggerFactory.getLogger(CoreAdminHandler.class);
  protected final CoreContainer coreContainer;
  protected static HashMap<String, Map<String, TaskObject>> requestStatusMap =
      new HashMap<String,Map<String, TaskObject>>();

  protected final ExecutorService parallelExecutor = Executors.newFixedThreadPool(50,
      new DefaultSolrThreadFactory("parallelCoreAdminExecutor"));

  protected static int MAX_TRACKED_REQUESTS = 100;
  public static String RUNNING = "running";
  public static String COMPLETED = "completed";
  public static String FAILED = "failed";
  public static String RESPONSE = "Response";
  public static String RESPONSE_STATUS = "STATUS";
  public static String RESPONSE_MESSAGE = "msg";

  static {
    requestStatusMap.put(RUNNING, Collections.synchronizedMap(new LinkedHashMap<String, TaskObject>()));
    requestStatusMap.put(COMPLETED, Collections.synchronizedMap(new LinkedHashMap<String, TaskObject>()));
    requestStatusMap.put(FAILED, Collections.synchronizedMap(new LinkedHashMap<String, TaskObject>()));
  }

  public CoreAdminHandler() {
    super();
    // Unlike most request handlers, CoreContainer initialization 
    // should happen in the constructor...  
    this.coreContainer = null;
  }


  /**
   * Overloaded ctor to inject CoreContainer into the handler.
   *
   * @param coreContainer Core Container of the solr webapp installed.
   */
  public CoreAdminHandler(final CoreContainer coreContainer) {
    this.coreContainer = coreContainer;
  }


  @Override
  final public void init(NamedList args) {
    throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
            "CoreAdminHandler should not be configured in solrconf.xml\n" +
                    "it is a special Handler configured directly by the RequestDispatcher");
  }

  /**
   * The instance of CoreContainer this handler handles. This should be the CoreContainer instance that created this
   * handler.
   *
   * @return a CoreContainer instance
   */
  public CoreContainer getCoreContainer() {
    return this.coreContainer;
  }

  @Override
  public void handleRequestBody(SolrQueryRequest req, SolrQueryResponse rsp) throws Exception {
    // Make sure the cores is enabled
    CoreContainer cores = getCoreContainer();
    if (cores == null) {
      throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
              "Core container instance missing");
    }
    //boolean doPersist = false;
    String taskId = req.getParams().get("async");
    TaskObject taskObject = new TaskObject(taskId);

    if(taskId != null) {
      // Put the tasks into the maps for tracking
      if (getMap(RUNNING).containsKey(taskId) || getMap(COMPLETED).containsKey(taskId) || getMap(FAILED).containsKey(taskId)) {
        throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
            "Duplicate request with the same requestid found.");
      }

      addTask(RUNNING, taskObject);
    }

    // Pick the action
    SolrParams params = req.getParams();
    CoreAdminAction action = CoreAdminAction.STATUS;
    String a = params.get(CoreAdminParams.ACTION);
    if (a != null) {
      action = CoreAdminAction.get(a);
      if (action == null) {
        this.handleCustomAction(req, rsp);
      }
    }

    if (taskId == null) {
      handleRequestInternal(req, rsp, action);
    } else {
      ParallelCoreAdminHandlerThread parallelHandlerThread = new ParallelCoreAdminHandlerThread(req, rsp, action, taskObject);
      parallelExecutor.execute(parallelHandlerThread);
    }
  }

  protected void handleRequestInternal(SolrQueryRequest req, SolrQueryResponse rsp, CoreAdminAction action) throws Exception {
    if (action != null) {
      switch (action) {
        case CREATE: {
          this.handleCreateAction(req, rsp);
          break;
        }

        case RENAME: {
          this.handleRenameAction(req, rsp);
          break;
        }

        case UNLOAD: {
          this.handleUnloadAction(req, rsp);
          break;
        }

        case STATUS: {
          this.handleStatusAction(req, rsp);
          break;

        }

        case PERSIST: {
          this.handlePersistAction(req, rsp);
          break;
        }

        case RELOAD: {
          this.handleReloadAction(req, rsp);
          break;
        }

        case SWAP: {
          this.handleSwapAction(req, rsp);
          break;
        }

        case MERGEINDEXES: {
          this.handleMergeAction(req, rsp);
          break;
        }

        case SPLIT: {
          this.handleSplitAction(req, rsp);
          break;
        }

        case PREPRECOVERY: {
          this.handleWaitForStateAction(req, rsp);
          break;
        }

        case REQUESTRECOVERY: {
          this.handleRequestRecoveryAction(req, rsp);
          break;
        }

        case REQUESTSYNCSHARD: {
          this.handleRequestSyncAction(req, rsp);
          break;
        }

        // todo : Can this be done by the regular RecoveryStrategy route?
        case REQUESTAPPLYUPDATES: {
          this.handleRequestApplyUpdatesAction(req, rsp);
          break;
        }
        case REQUESTBUFFERUPDATES:  {
          this.handleRequestBufferUpdatesAction(req, rsp);
          break;
        }
        case REQUESTSTATUS: {
          this.handleRequestActionStatus(req, rsp);
          break;
        }
        case OVERSEEROP:{
          ZkController zkController = coreContainer.getZkController();
          if(zkController != null){
           String op = req.getParams().get("op");
           String electionNode = req.getParams().get("electionNode");
           if(electionNode != null) {
             zkController.rejoinOverseerElection(electionNode, "rejoinAtHead".equals(op));
           } else {
             log.info("electionNode is required param");
           }
          }
          break;
        }
        default: {
          this.handleCustomAction(req, rsp);
          break;
        }
        case LOAD:
          break;
      }
    }
    rsp.setHttpCaching(false);
  }


  /**
   * Handle the core admin SPLIT action.
   */
  protected void handleSplitAction(SolrQueryRequest adminReq, SolrQueryResponse rsp) throws IOException {
    SolrParams params = adminReq.getParams();
    List<DocRouter.Range> ranges = null;

    String[] pathsArr = params.getParams("path");
    String rangesStr = params.get(CoreAdminParams.RANGES);    // ranges=a-b,c-d,e-f
    if (rangesStr != null)  {
      String[] rangesArr = rangesStr.split(",");
      if (rangesArr.length == 0) {
        throw new SolrException(ErrorCode.BAD_REQUEST, "There must be at least one range specified to split an index");
      } else  {
        ranges = new ArrayList<>(rangesArr.length);
        for (String r : rangesArr) {
          try {
            ranges.add(DocRouter.DEFAULT.fromString(r));
          } catch (Exception e) {
            throw new SolrException(ErrorCode.BAD_REQUEST, "Exception parsing hexadecimal hash range: " + r, e);
          }
        }
      }
    }
    String splitKey = params.get("split.key");
    String[] newCoreNames = params.getParams("targetCore");
    String cname = params.get(CoreAdminParams.CORE, "");

    if ((pathsArr == null || pathsArr.length == 0) && (newCoreNames == null || newCoreNames.length == 0)) {
      throw new SolrException(ErrorCode.BAD_REQUEST, "Either path or targetCore param must be specified");
    }

    log.info("Invoked split action for core: " + cname);
    SolrCore core = coreContainer.getCore(cname);
    SolrQueryRequest req = new LocalSolrQueryRequest(core, params);
    List<SolrCore> newCores = null;

    try {
      // TODO: allow use of rangesStr in the future
      List<String> paths = null;
      int partitions = pathsArr != null ? pathsArr.length : newCoreNames.length;

      DocRouter router = null;
      String routeFieldName = null;
      if (coreContainer.isZooKeeperAware()) {
        ClusterState clusterState = coreContainer.getZkController().getClusterState();
        String collectionName = req.getCore().getCoreDescriptor().getCloudDescriptor().getCollectionName();
        DocCollection collection = clusterState.getCollection(collectionName);
        String sliceName = req.getCore().getCoreDescriptor().getCloudDescriptor().getShardId();
        Slice slice = clusterState.getSlice(collectionName, sliceName);
        router = collection.getRouter() != null ? collection.getRouter() : DocRouter.DEFAULT;
        if (ranges == null) {
          DocRouter.Range currentRange = slice.getRange();
          ranges = currentRange != null ? router.partitionRange(partitions, currentRange) : null;
        }
        Object routerObj = collection.get(DOC_ROUTER); // for back-compat with Solr 4.4
        if (routerObj != null && routerObj instanceof Map) {
          Map routerProps = (Map) routerObj;
          routeFieldName = (String) routerProps.get("field");
        }
      }

      if (pathsArr == null) {
        newCores = new ArrayList<>(partitions);
        for (String newCoreName : newCoreNames) {
          SolrCore newcore = coreContainer.getCore(newCoreName);
          if (newcore != null) {
            newCores.add(newcore);
          } else {
            throw new SolrException(ErrorCode.BAD_REQUEST, "Core with core name " + newCoreName + " expected but doesn't exist.");
          }
        }
      } else {
        paths = Arrays.asList(pathsArr);
      }


      SplitIndexCommand cmd = new SplitIndexCommand(req, paths, newCores, ranges, router, routeFieldName, splitKey);
      core.getUpdateHandler().split(cmd);

      // After the split has completed, someone (here?) should start the process of replaying the buffered updates.

    } catch (Exception e) {
      log.error("ERROR executing split:", e);
      throw new RuntimeException(e);

    } finally {
      if (req != null) req.close();
      if (core != null) core.close();
      if (newCores != null) {
        for (SolrCore newCore : newCores) {
          newCore.close();
        }
      }
    }

  }


  protected void handleMergeAction(SolrQueryRequest req, SolrQueryResponse rsp) throws Exception {
    SolrParams params = req.getParams();
    String cname = params.required().get(CoreAdminParams.CORE);
    SolrCore core = coreContainer.getCore(cname);
    SolrQueryRequest wrappedReq = null;

    List<SolrCore> sourceCores = Lists.newArrayList();
    List<RefCounted<SolrIndexSearcher>> searchers = Lists.newArrayList();
    // stores readers created from indexDir param values
    List<DirectoryReader> readersToBeClosed = Lists.newArrayList();
    List<Directory> dirsToBeReleased = Lists.newArrayList();
    if (core != null) {
      try {
        String[] dirNames = params.getParams(CoreAdminParams.INDEX_DIR);
        if (dirNames == null || dirNames.length == 0) {
          String[] sources = params.getParams("srcCore");
          if (sources == null || sources.length == 0)
            throw new SolrException( SolrException.ErrorCode.BAD_REQUEST,
                "At least one indexDir or srcCore must be specified");

          for (int i = 0; i < sources.length; i++) {
            String source = sources[i];
            SolrCore srcCore = coreContainer.getCore(source);
            if (srcCore == null)
              throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
                  "Core: " + source + " does not exist");
            sourceCores.add(srcCore);
          }
        } else  {
          DirectoryFactory dirFactory = core.getDirectoryFactory();
          for (int i = 0; i < dirNames.length; i++) {
            Directory dir = dirFactory.get(dirNames[i], DirContext.DEFAULT, core.getSolrConfig().indexConfig.lockType);
            dirsToBeReleased.add(dir);
            // TODO: why doesn't this use the IR factory? what is going on here?
            readersToBeClosed.add(DirectoryReader.open(dir));
          }
        }

        List<DirectoryReader> readers = null;
        if (readersToBeClosed.size() > 0)  {
          readers = readersToBeClosed;
        } else {
          readers = Lists.newArrayList();
          for (SolrCore solrCore: sourceCores) {
            // record the searchers so that we can decref
            RefCounted<SolrIndexSearcher> searcher = solrCore.getSearcher();
            searchers.add(searcher);
            readers.add(searcher.get().getIndexReader());
          }
        }

        UpdateRequestProcessorChain processorChain =
                core.getUpdateProcessingChain(params.get(UpdateParams.UPDATE_CHAIN));
        wrappedReq = new LocalSolrQueryRequest(core, req.getParams());
        UpdateRequestProcessor processor =
                processorChain.createProcessor(wrappedReq, rsp);
        processor.processMergeIndexes(new MergeIndexesCommand(readers, req));
      } catch (Exception e) {
        // log and rethrow so that if the finally fails we don't lose the original problem
        log.error("ERROR executing merge:", e);
        throw e;
      } finally {
        for (RefCounted<SolrIndexSearcher> searcher : searchers) {
          if (searcher != null) searcher.decref();
        }
        for (SolrCore solrCore : sourceCores) {
          if (solrCore != null) solrCore.close();
        }
        IOUtils.closeWhileHandlingException(readersToBeClosed);
        for (Directory dir : dirsToBeReleased) {
          DirectoryFactory dirFactory = core.getDirectoryFactory();
          dirFactory.release(dir);
        }
        if (wrappedReq != null) wrappedReq.close();
        core.close();
      }
    }
  }

  /**
   * Handle Custom Action.
   * <p/>
   * This method could be overridden by derived classes to handle custom actions. <br> By default - this method throws a
   * solr exception. Derived classes are free to write their derivation if necessary.
   */
  protected void handleCustomAction(SolrQueryRequest req, SolrQueryResponse rsp) {
    throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Unsupported operation: " +
            req.getParams().get(CoreAdminParams.ACTION));
  }

  public static ImmutableMap<String, String> paramToProp = ImmutableMap.<String, String>builder()
      .put(CoreAdminParams.CONFIG, CoreDescriptor.CORE_CONFIG)
      .put(CoreAdminParams.SCHEMA, CoreDescriptor.CORE_SCHEMA)
      .put(CoreAdminParams.DATA_DIR, CoreDescriptor.CORE_DATADIR)
      .put(CoreAdminParams.ULOG_DIR, CoreDescriptor.CORE_ULOGDIR)
      .put(CoreAdminParams.CONFIGSET, CoreDescriptor.CORE_CONFIGSET)
      .put(CoreAdminParams.LOAD_ON_STARTUP, CoreDescriptor.CORE_LOADONSTARTUP)
      .put(CoreAdminParams.TRANSIENT, CoreDescriptor.CORE_TRANSIENT)
      .put(CoreAdminParams.SHARD, CoreDescriptor.CORE_SHARD)
      .put(CoreAdminParams.COLLECTION, CoreDescriptor.CORE_COLLECTION)
      .put(CoreAdminParams.ROLES, CoreDescriptor.CORE_ROLES)
      .put(CoreAdminParams.CORE_NODE_NAME, CoreDescriptor.CORE_NODE_NAME)
      .put(ZkStateReader.NUM_SHARDS_PROP, CloudDescriptor.NUM_SHARDS)
      .build();

  public static ImmutableMap<String, String> cloudParamToProp;

  protected static CoreDescriptor buildCoreDescriptor(SolrParams params, CoreContainer container) {

    String name = checkNotEmpty(params.get(CoreAdminParams.NAME),
        "Missing parameter [" + CoreAdminParams.NAME + "]");

    Properties coreProps = new Properties();
    for (String param : paramToProp.keySet()) {
      String value = params.get(param, null);
      if (StringUtils.isNotEmpty(value)) {
        coreProps.setProperty(paramToProp.get(param), value);
      }
    }
    Iterator<String> paramsIt = params.getParameterNamesIterator();
    while (paramsIt.hasNext()) {
      String param = paramsIt.next();
      if (!param.startsWith(CoreAdminParams.PROPERTY_PREFIX))
        continue;
      String propName = param.substring(CoreAdminParams.PROPERTY_PREFIX.length());
      String propValue = params.get(param);
      coreProps.setProperty(propName, propValue);
    }

    String instancedir = params.get(CoreAdminParams.INSTANCE_DIR);
    if (StringUtils.isEmpty(instancedir) && coreProps.getProperty(CoreAdminParams.INSTANCE_DIR) != null) {
      instancedir = coreProps.getProperty(CoreAdminParams.INSTANCE_DIR);
    } else if (StringUtils.isEmpty(instancedir)){
      instancedir = name; // will be resolved later against solr.home
      //instancedir = container.getSolrHome() + "/" + name;
    }

    return new CoreDescriptor(container, name, instancedir, coreProps, params);
  }

  private static String checkNotEmpty(String value, String message) {
    if (StringUtils.isEmpty(value))
      throw new SolrException(ErrorCode.BAD_REQUEST, message);
    return value;
  }

  /**
   * Handle 'CREATE' action.
   *
   * @throws SolrException in case of a configuration error.
   */
  protected void handleCreateAction(SolrQueryRequest req, SolrQueryResponse rsp) throws SolrException {

    SolrParams params = req.getParams();
    log.info("core create command {}", params);
    CoreDescriptor dcore = buildCoreDescriptor(params, coreContainer);

    if (coreContainer.getAllCoreNames().contains(dcore.getName())) {
      log.warn("Creating a core with existing name is not allowed");
      throw new SolrException(ErrorCode.SERVER_ERROR,
          "Core with name '" + dcore.getName() + "' already exists.");
    }

    // TODO this should be moved into CoreContainer, really...
    boolean preExisitingZkEntry = false;
    try {
      if (coreContainer.getZkController() != null) {
        if (!Overseer.isLegacy(coreContainer.getZkController().getZkStateReader().getClusterProps())) {
          if (dcore.getCloudDescriptor().getCoreNodeName() == null) {
            throw new SolrException(ErrorCode.SERVER_ERROR,
                "non legacy mode coreNodeName missing " + params);
            
          }
        }
        
        preExisitingZkEntry = checkIfCoreNodeNameAlreadyExists(dcore);

      }
      
      SolrCore core = coreContainer.create(dcore);
      
      // only write out the descriptor if the core is successfully created
      coreContainer.getCoresLocator().create(coreContainer, dcore);
      
      if (coreContainer.getCoresLocator() instanceof SolrXMLCoresLocator) {
        // hack - in this case we persist once more because a core create race might
        // have dropped entries.
        coreContainer.getCoresLocator().create(coreContainer);
      }
      rsp.add("core", core.getName());
    }
    catch (Exception ex) {
      if (coreContainer.isZooKeeperAware() && dcore != null && !preExisitingZkEntry) {
        try {
          coreContainer.getZkController().unregister(dcore.getName(), dcore);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          SolrException.log(log, null, e);
        } catch (KeeperException e) {
          SolrException.log(log, null, e);
        }
      }
      
      Throwable tc = ex;
      Throwable c = null;
      do {
        tc = tc.getCause();
        if (tc != null) {
          c = tc;
        }
      } while (tc != null);
      
      String rootMsg = "";
      if (c != null) {
        rootMsg = " Caused by: " + c.getMessage();
      }
      
      throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
          "Error CREATEing SolrCore '" + dcore.getName() + "': " +
          ex.getMessage() + rootMsg, ex);
    }
  }


  private boolean checkIfCoreNodeNameAlreadyExists(CoreDescriptor dcore) {
    ZkStateReader zkStateReader = coreContainer.getZkController()
        .getZkStateReader();
    DocCollection collection = zkStateReader.getClusterState().getCollectionOrNull(dcore.getCollectionName());
    if (collection != null) {
      Collection<Slice> slices = collection.getSlices();
      
      for (Slice slice : slices) {
        Collection<Replica> replicas = slice.getReplicas();
        for (Replica replica : replicas) {
          if (replica.getName().equals(
              dcore.getCloudDescriptor().getCoreNodeName())) {
            return true;
          }
        }
      }
    }
    return false;
  }

  /**
   * Handle "RENAME" Action
   */
  protected void handleRenameAction(SolrQueryRequest req, SolrQueryResponse rsp) throws SolrException {
    SolrParams params = req.getParams();

    String name = params.get(CoreAdminParams.OTHER);
    String cname = params.get(CoreAdminParams.CORE);

    if (cname.equals(name)) return;

    coreContainer.rename(cname, name);

  }

  /**
   * Handle "ALIAS" action
   */
  @Deprecated
  protected void handleAliasAction(SolrQueryRequest req, SolrQueryResponse rsp) {
    throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "The ALIAS action is no longer supported");
  }


  /**
   * Handle "UNLOAD" Action
   */
  protected void handleUnloadAction(SolrQueryRequest req, SolrQueryResponse rsp) throws SolrException {

    SolrParams params = req.getParams();
    String cname = params.get(CoreAdminParams.CORE);
    boolean deleteIndexDir = params.getBool(CoreAdminParams.DELETE_INDEX, false);
    boolean deleteDataDir = params.getBool(CoreAdminParams.DELETE_DATA_DIR, false);
    boolean deleteInstanceDir = params.getBool(CoreAdminParams.DELETE_INSTANCE_DIR, false);

    coreContainer.unload(cname, deleteIndexDir, deleteDataDir, deleteInstanceDir);

  }

  /**
   * Handle "STATUS" action
   */
  protected void handleStatusAction(SolrQueryRequest req, SolrQueryResponse rsp)
          throws SolrException {
    SolrParams params = req.getParams();

    String cname = params.get(CoreAdminParams.CORE);
    String indexInfo = params.get(CoreAdminParams.INDEX_INFO);
    boolean isIndexInfoNeeded = Boolean.parseBoolean(null == indexInfo ? "true" : indexInfo);
    NamedList<Object> status = new SimpleOrderedMap<>();
    Map<String, Exception> failures = new HashMap<>();
    for (Map.Entry<String, CoreContainer.CoreLoadFailure> failure : coreContainer.getCoreInitFailures().entrySet()) {
      failures.put(failure.getKey(), failure.getValue().exception);
    }
    try {
      if (cname == null) {
        rsp.add("defaultCoreName", coreContainer.getDefaultCoreName());
        for (String name : coreContainer.getAllCoreNames()) {
          status.add(name, getCoreStatus(coreContainer, name, isIndexInfoNeeded));
        }
        rsp.add("initFailures", failures);
      } else {
        failures = failures.containsKey(cname)
          ? Collections.singletonMap(cname, failures.get(cname))
          : Collections.<String, Exception>emptyMap();
        rsp.add("initFailures", failures);
        status.add(cname, getCoreStatus(coreContainer, cname, isIndexInfoNeeded));
      }
      rsp.add("status", status);
    } catch (Exception ex) {
      throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
              "Error handling 'status' action ", ex);
    }
  }

  /**
   * Handler "PERSIST" action
   */
  protected void handlePersistAction(SolrQueryRequest req, SolrQueryResponse rsp)
          throws SolrException {
    rsp.add("message", "The PERSIST action has been deprecated");
  }

  /**
   * Handler "RELOAD" action
   */
  protected void handleReloadAction(SolrQueryRequest req, SolrQueryResponse rsp) {
    SolrParams params = req.getParams();
    String cname = params.get(CoreAdminParams.CORE);

    if(!coreContainer.getCoreNames().contains(cname)) {
      throw new SolrException(ErrorCode.BAD_REQUEST, "Core with core name [" + cname + "] does not exist.");
    }

    try {
      coreContainer.reload(cname);
    } catch (Exception ex) {
      throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Error handling 'reload' action", ex);
    }
  }

  /**
   * Handle "REQUESTSTATUS" action
   */
  protected void handleRequestActionStatus(SolrQueryRequest req, SolrQueryResponse rsp) {
    SolrParams params = req.getParams();
    String requestId = params.get(CoreAdminParams.REQUESTID);
    log.info("Checking request status for : " + requestId);

    if (mapContainsTask(RUNNING, requestId)) {
      rsp.add(RESPONSE_STATUS, RUNNING);
    } else if(mapContainsTask(COMPLETED, requestId)) {
      rsp.add(RESPONSE_STATUS, COMPLETED);
      rsp.add(RESPONSE, getMap(COMPLETED).get(requestId).getRspObject());
    } else if(mapContainsTask(FAILED, requestId)) {
      rsp.add(RESPONSE_STATUS, FAILED);
      rsp.add(RESPONSE, getMap(FAILED).get(requestId).getRspObject());
    } else {
      rsp.add(RESPONSE_STATUS, "notfound");
      rsp.add(RESPONSE_MESSAGE, "No task found in running, completed or failed tasks");
    }
  }

  /**
   * Handle "SWAP" action
   */
  protected void handleSwapAction(SolrQueryRequest req, SolrQueryResponse rsp) {
    final SolrParams params = req.getParams();
    final SolrParams required = params.required();

    final String cname = params.get(CoreAdminParams.CORE);
    String other = required.get(CoreAdminParams.OTHER);
    coreContainer.swap(cname, other);

  }
  
  protected void handleRequestRecoveryAction(SolrQueryRequest req,
      SolrQueryResponse rsp) throws IOException {
    final SolrParams params = req.getParams();
    log.info("It has been requested that we recover: core="+params.get(CoreAdminParams.CORE));
    Thread thread = new Thread() {
      @Override
      public void run() {
        String cname = params.get(CoreAdminParams.CORE);
        if (cname == null) {
          cname = "";
        }
        try (SolrCore core = coreContainer.getCore(cname)) {

          if (core != null) {
            // try to publish as recovering right away
            try {
              coreContainer.getZkController().publish(core.getCoreDescriptor(), ZkStateReader.RECOVERING);
            }  catch (InterruptedException e) {
              Thread.currentThread().interrupt();
              SolrException.log(log, "", e);
            } catch (Throwable e) {
              SolrException.log(log, "", e);
              if (e instanceof Error) {
                throw (Error) e;
              }
            }
            
            core.getUpdateHandler().getSolrCoreState().doRecovery(coreContainer, core.getCoreDescriptor());
          } else {
            SolrException.log(log, "Could not find core to call recovery:" + cname);
          }
        }
      }
    };
    
    thread.start();
  }
  
  protected void handleRequestSyncAction(SolrQueryRequest req,
      SolrQueryResponse rsp) throws IOException {
    final SolrParams params = req.getParams();

    log.info("I have been requested to sync up my shard");
    ZkController zkController = coreContainer.getZkController();
    if (zkController == null) {
      throw new SolrException(ErrorCode.BAD_REQUEST, "Only valid for SolrCloud");
    }
    
    String cname = params.get(CoreAdminParams.CORE);
    if (cname == null) {
      throw new IllegalArgumentException(CoreAdminParams.CORE + " is required");
    }

    SyncStrategy syncStrategy = null;
    try (SolrCore core = coreContainer.getCore(cname)) {

      if (core != null) {
        syncStrategy = new SyncStrategy(core.getCoreDescriptor().getCoreContainer());
        
        Map<String,Object> props = new HashMap<>();
        props.put(ZkStateReader.BASE_URL_PROP, zkController.getBaseUrl());
        props.put(ZkStateReader.CORE_NAME_PROP, cname);
        props.put(ZkStateReader.NODE_NAME_PROP, zkController.getNodeName());
        
        boolean success = syncStrategy.sync(zkController, core, new ZkNodeProps(props), true);
        // solrcloud_debug
        if (log.isDebugEnabled()) {
          try {
            RefCounted<SolrIndexSearcher> searchHolder = core
                .getNewestSearcher(false);
            SolrIndexSearcher searcher = searchHolder.get();
            try {
              log.debug(core.getCoreDescriptor().getCoreContainer()
                  .getZkController().getNodeName()
                  + " synched "
                  + searcher.search(new MatchAllDocsQuery(), 1).totalHits);
            } finally {
              searchHolder.decref();
            }
          } catch (Exception e) {
            throw new SolrException(ErrorCode.SERVER_ERROR, null, e);
          }
        }
        if (!success) {
          throw new SolrException(ErrorCode.SERVER_ERROR, "Sync Failed");
        }
      } else {
        SolrException.log(log, "Cound not find core to call sync:" + cname);
      }
    } finally {
      // no recoveryStrat close for now
      if (syncStrategy != null) {
        syncStrategy.close();
      }
    }
    

  }
  
  protected void handleWaitForStateAction(SolrQueryRequest req,
      SolrQueryResponse rsp) throws IOException, InterruptedException, KeeperException {
    final SolrParams params = req.getParams();
    
    String cname = params.get(CoreAdminParams.CORE);
    if (cname == null) {
      cname = "";
    }
    
    String nodeName = params.get("nodeName");
    String coreNodeName = params.get("coreNodeName");
    String waitForState = params.get("state");
    Boolean checkLive = params.getBool("checkLive");
    Boolean onlyIfLeader = params.getBool("onlyIfLeader");
    Boolean onlyIfLeaderActive = params.getBool("onlyIfLeaderActive");

    log.info("Going to wait for coreNodeName: " + coreNodeName + ", state: " + waitForState
        + ", checkLive: " + checkLive + ", onlyIfLeader: " + onlyIfLeader
        + ", onlyIfLeaderActive: "+onlyIfLeaderActive);

    int maxTries = 0; 
    String state = null;
    boolean live = false;
    int retry = 0;
    while (true) {
      try (SolrCore core = coreContainer.getCore(cname)) {
        if (core == null && retry == 30) {
          throw new SolrException(ErrorCode.BAD_REQUEST, "core not found:"
              + cname);
        }
        if (core != null) {
          if (onlyIfLeader != null && onlyIfLeader) {
           if (!core.getCoreDescriptor().getCloudDescriptor().isLeader()) {
             throw new SolrException(ErrorCode.BAD_REQUEST, "We are not the leader");
           }
          }
          
          // wait until we are sure the recovering node is ready
          // to accept updates
          CloudDescriptor cloudDescriptor = core.getCoreDescriptor()
              .getCloudDescriptor();
          
          if (retry % 15 == 0) {
            if (retry > 0 && log.isInfoEnabled())
              log.info("After " + retry + " seconds, core " + cname + " (" +
                  cloudDescriptor.getShardId() + " of " +
                  cloudDescriptor.getCollectionName() + ") still does not have state: " +
                  waitForState + "; forcing ClusterState update from ZooKeeper");
            
            // force a cluster state update
            coreContainer.getZkController().getZkStateReader().updateClusterState(true);
          }

          if (maxTries == 0) {
            // wait long enough for the leader conflict to work itself out plus a little extra
            int conflictWaitMs = coreContainer.getZkController().getLeaderConflictResolveWait();
            maxTries = (int) Math.round(conflictWaitMs / 1000) + 3;
            log.info("Will wait a max of " + maxTries + " seconds to see " + cname + " (" +
                cloudDescriptor.getShardId() + " of " +
                cloudDescriptor.getCollectionName() + ") have state: " + waitForState);
          }
          
          ClusterState clusterState = coreContainer.getZkController()
              .getClusterState();
          String collection = cloudDescriptor.getCollectionName();
          Slice slice = clusterState.getSlice(collection,
              cloudDescriptor.getShardId());
          if (slice != null) {
            ZkNodeProps nodeProps = slice.getReplicasMap().get(coreNodeName);
            if (nodeProps != null) {
              state = nodeProps.getStr(ZkStateReader.STATE_PROP);
              live = clusterState.liveNodesContain(nodeName);
              
              String localState = cloudDescriptor.getLastPublished();

              // TODO: This is funky but I've seen this in testing where the replica asks the
              // leader to be in recovery? Need to track down how that happens ... in the meantime,
              // this is a safeguard 
              boolean leaderDoesNotNeedRecovery = (onlyIfLeader != null && 
                  onlyIfLeader && 
                  core.getName().equals(nodeProps.getStr("core")) &&
                  ZkStateReader.RECOVERING.equals(waitForState) && 
                  ZkStateReader.ACTIVE.equals(localState) && 
                  ZkStateReader.ACTIVE.equals(state));
              
              if (leaderDoesNotNeedRecovery) {
                log.warn("Leader "+core.getName()+" ignoring request to be in the recovering state because it is live and active.");
              }              
              
              boolean onlyIfActiveCheckResult = onlyIfLeaderActive != null && onlyIfLeaderActive && (localState == null || !localState.equals(ZkStateReader.ACTIVE));
              log.info("In WaitForState("+waitForState+"): collection="+collection+", shard="+slice.getName()+
                  ", thisCore="+core.getName()+", leaderDoesNotNeedRecovery="+leaderDoesNotNeedRecovery+
                  ", isLeader? "+core.getCoreDescriptor().getCloudDescriptor().isLeader()+
                  ", live="+live+", checkLive="+checkLive+", currentState="+state+", localState="+localState+", nodeName="+nodeName+
                  ", coreNodeName="+coreNodeName+", onlyIfActiveCheckResult="+onlyIfActiveCheckResult+", nodeProps: "+nodeProps);

              if (!onlyIfActiveCheckResult && nodeProps != null && (state.equals(waitForState) || leaderDoesNotNeedRecovery)) {
                if (checkLive == null) {
                  break;
                } else if (checkLive && live) {
                  break;
                } else if (!checkLive && !live) {
                  break;
                }
              }
            }
          }
        }

        if (retry++ == maxTries) {
          String collection = null;
          String leaderInfo = null;
          String shardId = null;
          try {
            CloudDescriptor cloudDescriptor =
                core.getCoreDescriptor().getCloudDescriptor();
            collection = cloudDescriptor.getCollectionName();
            shardId = cloudDescriptor.getShardId();
            leaderInfo = coreContainer.getZkController().
                getZkStateReader().getLeaderUrl(collection, shardId, 5000);
          } catch (Exception exc) {
            leaderInfo = "Not available due to: " + exc;
          }

          throw new SolrException(ErrorCode.BAD_REQUEST,
              "I was asked to wait on state " + waitForState + " for "
                  + shardId + " in " + collection + " on " + nodeName
                  + " but I still do not see the requested state. I see state: "
                  + state + " live:" + live + " leader from ZK: " + leaderInfo
          );
        }
        
        if (coreContainer.isShutDown()) {
          throw new SolrException(ErrorCode.BAD_REQUEST,
              "Solr is shutting down");
        }
        
        // solrcloud_debug
        if (log.isDebugEnabled()) {
          try {
            LocalSolrQueryRequest r = new LocalSolrQueryRequest(core,
                new ModifiableSolrParams());
            CommitUpdateCommand commitCmd = new CommitUpdateCommand(r, false);
            commitCmd.softCommit = true;
            core.getUpdateHandler().commit(commitCmd);
            RefCounted<SolrIndexSearcher> searchHolder = core
                .getNewestSearcher(false);
            SolrIndexSearcher searcher = searchHolder.get();
            try {
              log.debug(core.getCoreDescriptor().getCoreContainer()
                  .getZkController().getNodeName()
                  + " to replicate "
                  + searcher.search(new MatchAllDocsQuery(), 1).totalHits
                  + " gen:"
                  + core.getDeletionPolicy().getLatestCommit().getGeneration()
                  + " data:" + core.getDataDir());
            } finally {
              searchHolder.decref();
            }
          } catch (Exception e) {
            throw new SolrException(ErrorCode.SERVER_ERROR, null, e);
          }
        }
      }
      Thread.sleep(1000);
    }

    log.info("Waited coreNodeName: " + coreNodeName + ", state: " + waitForState
        + ", checkLive: " + checkLive + ", onlyIfLeader: " + onlyIfLeader + " for: " + retry + " seconds.");
  }

  private void handleRequestApplyUpdatesAction(SolrQueryRequest req, SolrQueryResponse rsp) {
    SolrParams params = req.getParams();
    String cname = params.get(CoreAdminParams.NAME, "");
    log.info("Applying buffered updates on core: " + cname);
    try (SolrCore core = coreContainer.getCore(cname)) {
      if (core == null)
        throw new SolrException(ErrorCode.BAD_REQUEST, "Core [" + cname + "] not found");
      UpdateLog updateLog = core.getUpdateHandler().getUpdateLog();
      if (updateLog.getState() != UpdateLog.State.BUFFERING)  {
        throw new SolrException(ErrorCode.SERVER_ERROR, "Core " + cname + " not in buffering state");
      }
      Future<UpdateLog.RecoveryInfo> future = updateLog.applyBufferedUpdates();
      if (future == null) {
        log.info("No buffered updates available. core=" + cname);
        rsp.add("core", cname);
        rsp.add("status", "EMPTY_BUFFER");
        return;
      }
      UpdateLog.RecoveryInfo report = future.get();
      if (report.failed) {
        SolrException.log(log, "Replay failed");
        throw new SolrException(ErrorCode.SERVER_ERROR, "Replay failed");
      }
      coreContainer.getZkController().publish(core.getCoreDescriptor(), ZkStateReader.ACTIVE);
      rsp.add("core", cname);
      rsp.add("status", "BUFFER_APPLIED");
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      log.warn("Recovery was interrupted", e);
    } catch (Exception e) {
      if (e instanceof SolrException)
        throw (SolrException)e;
      else
        throw new SolrException(ErrorCode.SERVER_ERROR, "Could not apply buffered updates", e);
    } finally {
      if (req != null) req.close();
    }
    
  }

  private void handleRequestBufferUpdatesAction(SolrQueryRequest req, SolrQueryResponse rsp) {
    SolrParams params = req.getParams();
    String cname = params.get(CoreAdminParams.NAME, "");
    log.info("Starting to buffer updates on core:" + cname);

    try (SolrCore core = coreContainer.getCore(cname)) {
      if (core == null)
        throw new SolrException(ErrorCode.BAD_REQUEST, "Core [" + cname + "] does not exist");
      UpdateLog updateLog = core.getUpdateHandler().getUpdateLog();
      if (updateLog.getState() != UpdateLog.State.ACTIVE)  {
        throw new SolrException(ErrorCode.SERVER_ERROR, "Core " + cname + " not in active state");
      }
      updateLog.bufferUpdates();
      rsp.add("core", cname);
      rsp.add("status", "BUFFERING");
    } catch (Throwable e) {
      if (e instanceof SolrException)
        throw (SolrException)e;
      else
        throw new SolrException(ErrorCode.SERVER_ERROR, "Could not start buffering updates", e);
    } finally {
      if (req != null) req.close();
    }
  }

  /**
   * Returns the core status for a particular core.
   * @param cores - the enclosing core container
   * @param cname - the core to return
   * @param isIndexInfoNeeded - add what may be expensive index information. NOT returned if the core is not loaded
   * @return - a named list of key/value pairs from the core.
   * @throws IOException - LukeRequestHandler can throw an I/O exception
   */
  protected NamedList<Object> getCoreStatus(CoreContainer cores, String cname, boolean isIndexInfoNeeded)  throws IOException {
    NamedList<Object> info = new SimpleOrderedMap<>();

    if (!cores.isLoaded(cname)) { // Lazily-loaded core, fill in what we can.
      // It would be a real mistake to load the cores just to get the status
      CoreDescriptor desc = cores.getUnloadedCoreDescriptor(cname);
      if (desc != null) {
        info.add("name", desc.getName());
        info.add("isDefaultCore", desc.getName().equals(cores.getDefaultCoreName()));
        info.add("instanceDir", desc.getInstanceDir());
        // None of the following are guaranteed to be present in a not-yet-loaded core.
        String tmp = desc.getDataDir();
        if (StringUtils.isNotBlank(tmp)) info.add("dataDir", tmp);
        tmp = desc.getConfigName();
        if (StringUtils.isNotBlank(tmp)) info.add("config", tmp);
        tmp = desc.getSchemaName();
        if (StringUtils.isNotBlank(tmp)) info.add("schema", tmp);
        info.add("isLoaded", "false");
      }
    } else {
      try (SolrCore core = cores.getCore(cname)) {
        if (core != null) {
          info.add("name", core.getName());
          info.add("isDefaultCore", core.getName().equals(cores.getDefaultCoreName()));
          info.add("instanceDir", normalizePath(core.getResourceLoader().getInstanceDir()));
          info.add("dataDir", normalizePath(core.getDataDir()));
          info.add("config", core.getConfigResource());
          info.add("schema", core.getSchemaResource());
          info.add("startTime", new Date(core.getStartTime()));
          info.add("uptime", System.currentTimeMillis() - core.getStartTime());
          if (isIndexInfoNeeded) {
            RefCounted<SolrIndexSearcher> searcher = core.getSearcher();
            try {
              SimpleOrderedMap<Object> indexInfo = LukeRequestHandler.getIndexInfo(searcher.get().getIndexReader());
              long size = getIndexSize(core);
              indexInfo.add("sizeInBytes", size);
              indexInfo.add("size", NumberUtils.readableSize(size));
              info.add("index", indexInfo);
            } finally {
              searcher.decref();
            }
          }
        }
      }
    }
    return info;
  }

  private long getIndexSize(SolrCore core) {
    Directory dir;
    long size = 0;
    try {

      dir = core.getDirectoryFactory().get(core.getIndexDir(), DirContext.DEFAULT, core.getSolrConfig().indexConfig.lockType); 

      try {
        size = DirectoryFactory.sizeOfDirectory(dir);
      } finally {
        core.getDirectoryFactory().release(dir);
      }
    } catch (IOException e) {
      SolrException.log(log, "IO error while trying to get the size of the Directory", e);
    }
    return size;
  }

  protected static String normalizePath(String path) {
    if (path == null)
      return null;
    path = path.replace('/', File.separatorChar);
    path = path.replace('\\', File.separatorChar);
    return path;
  }

  public static ModifiableSolrParams params(String... params) {
    ModifiableSolrParams msp = new ModifiableSolrParams();
    for (int i=0; i<params.length; i+=2) {
      msp.add(params[i], params[i+1]);
    }
    return msp;
  }

  //////////////////////// SolrInfoMBeans methods //////////////////////

  @Override
  public String getDescription() {
    return "Manage Multiple Solr Cores";
  }

  @Override
  public String getSource() {
    return null;
  }

  /**
   * Class to implement multi-threaded CoreAdminHandler behaviour.
   * This accepts all of the context from handleRequestBody.
   */
  protected class ParallelCoreAdminHandlerThread implements Runnable {
    SolrQueryRequest req;
    SolrQueryResponse rsp;
    CoreAdminAction action;
    TaskObject taskObject;

    public ParallelCoreAdminHandlerThread (SolrQueryRequest req, SolrQueryResponse rsp,
                                           CoreAdminAction action, TaskObject taskObject){
      this.req = req;
      this.rsp = rsp;
      this.action = action;
      this.taskObject = taskObject;
    }

    public void run() {
      boolean exceptionCaught = false;
      try {
        handleRequestInternal(req, rsp, action);
        taskObject.setRspObject(rsp);
      } catch (Exception e) {
        exceptionCaught = true;
        taskObject.setRspObjectFromException(e);
      } finally {
        removeTask("running", taskObject.taskId);
        if(exceptionCaught) {
          addTask("failed", taskObject, true);
        } else
          addTask("completed", taskObject, true);
      }

    }

  }

  /**
   * Helper class to manage the tasks to be tracked.
   * This contains the taskId, request and the response (if available).
   */
  private class TaskObject {
    String taskId;
    String rspInfo;

    public TaskObject(String taskId) {
      this.taskId = taskId;
    }

    public String getRspObject() {
      return rspInfo;
    }

    public void setRspObject(SolrQueryResponse rspObject) {
      this.rspInfo = rspObject.getToLogAsString("TaskId: " + this.taskId + " ");
    }

    public void setRspObjectFromException(Exception e) {
      this.rspInfo = e.getMessage();
    }
  }

  /**
   * Helper method to add a task to a tracking map.
   */
  protected void addTask(String map, TaskObject o, boolean limit) {
    synchronized (getMap(map)) {
      if(limit && getMap(map).size() == MAX_TRACKED_REQUESTS) {
        String key = getMap(map).entrySet().iterator().next().getKey();
        getMap(map).remove(key);
      }
      addTask(map, o);
    }
  }


  protected void addTask(String map, TaskObject o) {
    synchronized (getMap(map)) {
      getMap(map).put(o.taskId, o);
    }
  }

  /**
   * Helper method to remove a task from a tracking map.
   */
  protected void removeTask(String map, String taskId) {
    synchronized (getMap(map)) {
      getMap(map).remove(taskId);
    }
  }

  /**
   * Helper method to check if a map contains a taskObject with the given taskId.
   */
  protected boolean mapContainsTask(String map, String taskId) {
    return getMap(map).containsKey(taskId);
  }

  /**
   * Helper method to get a TaskObject given a map and a taskId.
   */
  protected TaskObject getTask(String map, String taskId) {
    return getMap(map).get(taskId);
  }

  /**
   * Helper method to get a request status map given the name.
   */
  private Map<String, TaskObject> getMap(String map) {
    return requestStatusMap.get(map);
  }

  /**
   * Method to ensure shutting down of the ThreadPool Executor.
   */
  public void shutdown() {
    if (parallelExecutor != null && !parallelExecutor.isShutdown())
      ExecutorUtil.shutdownAndAwaitTermination(parallelExecutor);
  }
}