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

import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.CoreAdminRequest;
import org.apache.solr.client.solrj.response.CoreAdminResponse;
import org.apache.solr.client.solrj.response.RequestStatusState;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.util.IdUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MoveReplicaTest extends SolrCloudTestCase {
  private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

  // used by MoveReplicaHDFSTest
  protected boolean inPlaceMove = true;

  protected String getConfigSet() {
    return "cloud-dynamic";
  }

  @Before
  public void beforeTest() throws Exception {
    inPlaceMove = true;

    configureCluster(4)
        .addConfig("conf1", configset(getConfigSet()))
        .addConfig("conf2", configset(getConfigSet()))
        .withSolrXml(TEST_PATH().resolve("solr.xml"))
        .configure();

    NamedList<Object> overSeerStatus = cluster.getSolrClient().request(CollectionAdminRequest.getOverseerStatus());
    JettySolrRunner overseerJetty = null;
    String overseerLeader = (String) overSeerStatus.get("leader");
    for (JettySolrRunner jetty : cluster.getJettySolrRunners()) {
      if (jetty.getNodeName().equals(overseerLeader)) {
        overseerJetty = jetty;
        break;
      }
    }
    if (overseerJetty == null) {
      fail("no overseer leader!");
    }
  }

  @After
  public void afterTest() throws Exception {
    try {
      shutdownCluster();
    } finally {
      super.tearDown();
    }
  }

  @Test
  // commented out on: 17-Feb-2019   @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // annotated on: 24-Dec-2018
  public void test() throws Exception {
    String coll = getTestClass().getSimpleName() + "_coll_" + inPlaceMove;
    if (log.isInfoEnabled()) {
      log.info("total_jettys: {}", cluster.getJettySolrRunners().size());
    }
    int REPLICATION = 2;

    CloudSolrClient cloudClient = cluster.getSolrClient();

    // random create tlog or pull type replicas with nrt
    boolean isTlog = random().nextBoolean();
    CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(coll, "conf1", 2, 1, isTlog ? 1 : 0, !isTlog ? 1 : 0);
    create.setMaxShardsPerNode(2);
    create.setAutoAddReplicas(false);
    cloudClient.request(create);

    addDocs(coll, 100);

    Replica replica = getRandomReplica(coll, cloudClient);
    Set<String> liveNodes = cloudClient.getZkStateReader().getClusterState().getLiveNodes();
    ArrayList<String> l = new ArrayList<>(liveNodes);
    Collections.shuffle(l, random());
    String targetNode = null;
    for (String node : liveNodes) {
      if (!replica.getNodeName().equals(node)) {
        targetNode = node;
        break;
      }
    }
    assertNotNull(targetNode);
    String shardId = null;
    for (Slice slice : cloudClient.getZkStateReader().getClusterState().getCollection(coll).getSlices()) {
      if (slice.getReplicas().contains(replica)) {
        shardId = slice.getName();
      }
    }

    int sourceNumCores = getNumOfCores(cloudClient, replica.getNodeName(), coll, replica.getType().name());
    int targetNumCores = getNumOfCores(cloudClient, targetNode, coll, replica.getType().name());

    CollectionAdminRequest.MoveReplica moveReplica = createMoveReplicaRequest(coll, replica, targetNode);
    moveReplica.setInPlaceMove(inPlaceMove);
    String asyncId = IdUtils.randomId();
    moveReplica.processAsync(asyncId, cloudClient);
    CollectionAdminRequest.RequestStatus requestStatus = CollectionAdminRequest.requestStatus(asyncId);
    // wait for async request success
    boolean success = false;
    for (int i = 0; i < 200; i++) {
      CollectionAdminRequest.RequestStatusResponse rsp = requestStatus.process(cloudClient);
      if (rsp.getRequestStatus() == RequestStatusState.COMPLETED) {
        success = true;
        break;
      }
      assertNotSame(rsp.getRequestStatus(), RequestStatusState.FAILED);
      Thread.sleep(500);
    }
    assertTrue(success);
    assertEquals("should be one less core on the source node!", sourceNumCores - 1, getNumOfCores(cloudClient, replica.getNodeName(), coll, replica.getType().name()));
    assertEquals("should be one more core on target node!", targetNumCores + 1, getNumOfCores(cloudClient, targetNode, coll, replica.getType().name()));
    // wait for recovery
    boolean recovered = false;
    for (int i = 0; i < 300; i++) {
      DocCollection collState = getCollectionState(coll);
      log.debug("###### {}", collState);
      Collection<Replica> replicas = collState.getSlice(shardId).getReplicas();
      boolean allActive = true;
      boolean hasLeaders = true;
      if (replicas != null && !replicas.isEmpty()) {
        for (Replica r : replicas) {
          if (!r.getNodeName().equals(targetNode)) {
            continue;
          }
          if (!r.isActive(Collections.singleton(targetNode))) {
            log.info("Not active: {}", r);
            allActive = false;
          }
        }
      } else {
        allActive = false;
      }
      for (Slice slice : collState.getSlices()) {
        if (slice.getLeader() == null) {
          hasLeaders = false;
        }
      }
      if (allActive && hasLeaders) {
        // check the number of active replicas
        assertEquals("total number of replicas", REPLICATION, replicas.size());
        recovered = true;
        break;
      } else {
        log.info("--- waiting, allActive={}, hasLeaders={}", allActive, hasLeaders);
        Thread.sleep(1000);
      }
    }
    assertTrue("replica never fully recovered", recovered);

    assertEquals(100, cluster.getSolrClient().query(coll, new SolrQuery("*:*")).getResults().getNumFound());

    moveReplica = createMoveReplicaRequest(coll, replica, targetNode, shardId);
    moveReplica.setInPlaceMove(inPlaceMove);
    moveReplica.process(cloudClient);
    checkNumOfCores(cloudClient, replica.getNodeName(), coll, sourceNumCores);
    // wait for recovery
    recovered = false;
    for (int i = 0; i < 300; i++) {
      DocCollection collState = getCollectionState(coll);
      log.debug("###### {}", collState);
      Collection<Replica> replicas = collState.getSlice(shardId).getReplicas();
      boolean allActive = true;
      boolean hasLeaders = true;
      if (replicas != null && !replicas.isEmpty()) {
        for (Replica r : replicas) {
          if (!r.getNodeName().equals(replica.getNodeName())) {
            continue;
          }
          if (!r.isActive(Collections.singleton(replica.getNodeName()))) {
            log.info("Not active yet: {}", r);
            allActive = false;
          }
        }
      } else {
        allActive = false;
      }
      for (Slice slice : collState.getSlices()) {
        if (slice.getLeader() == null) {
          hasLeaders = false;
        }
      }
      if (allActive && hasLeaders) {
        assertEquals("total number of replicas", REPLICATION, replicas.size());
        recovered = true;
        break;
      } else {
        Thread.sleep(1000);
      }
    }
    assertTrue("replica never fully recovered", recovered);

    assertEquals(100, cluster.getSolrClient().query(coll, new SolrQuery("*:*")).getResults().getNumFound());
  }

  //Commented out 5-Dec-2017
  // @AwaitsFix(bugUrl = "https://issues.apache.org/jira/browse/SOLR-11458")
  @Test
  // 12-Jun-2018 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // 17-Mar-2018 This JIRA is fixed, but this test still fails
  //17-Aug-2018 commented  @LuceneTestCase.BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // 2-Aug-2018
  // commented out on: 17-Feb-2019   @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // annotated on: 24-Dec-2018
  public void testFailedMove() throws Exception {
    String coll = getTestClass().getSimpleName() + "_failed_coll_" + inPlaceMove;
    int REPLICATION = 2;

    CloudSolrClient cloudClient = cluster.getSolrClient();

    // random create tlog or pull type replicas with nrt
    boolean isTlog = random().nextBoolean();
    CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(coll, "conf1", 2, 1, isTlog ? 1 : 0, !isTlog ? 1 : 0);
    create.setAutoAddReplicas(false);
    cloudClient.request(create);

    addDocs(coll, 100);

    NamedList<Object> overSeerStatus = cluster.getSolrClient().request(CollectionAdminRequest.getOverseerStatus());
    String overseerLeader = (String) overSeerStatus.get("leader");

    // don't kill overseer in this test
    Replica replica;
    int count = 10;
    do {
      replica = getRandomReplica(coll, cloudClient);
    } while (!replica.getNodeName().equals(overseerLeader) && count-- > 0);
    assertNotNull("could not find non-overseer replica???", replica);
    Set<String> liveNodes = cloudClient.getZkStateReader().getClusterState().getLiveNodes();
    ArrayList<String> l = new ArrayList<>(liveNodes);
    Collections.shuffle(l, random());
    String targetNode = null;
    for (String node : liveNodes) {
      if (!replica.getNodeName().equals(node) && !overseerLeader.equals(node)) {
        targetNode = node;
        break;
      }
    }
    assertNotNull(targetNode);
    CollectionAdminRequest.MoveReplica moveReplica = createMoveReplicaRequest(coll, replica, targetNode);
    moveReplica.setInPlaceMove(inPlaceMove);
    // start moving
    String asyncId = IdUtils.randomId();
    moveReplica.processAsync(asyncId, cloudClient);
    // shut down target node
    for (int i = 0; i < cluster.getJettySolrRunners().size(); i++) {
      if (cluster.getJettySolrRunner(i).getNodeName().equals(targetNode)) {
        JettySolrRunner j = cluster.stopJettySolrRunner(i);
        cluster.waitForJettyToStop(j);
        break;
      }
    }
    CollectionAdminRequest.RequestStatus requestStatus = CollectionAdminRequest.requestStatus(asyncId);
    // wait for async request success
    boolean success = true;
    for (int i = 0; i < 200; i++) {
      CollectionAdminRequest.RequestStatusResponse rsp = requestStatus.process(cloudClient);
      assertNotSame(rsp.getRequestStatus().toString(), rsp.getRequestStatus(), RequestStatusState.COMPLETED);
      if (rsp.getRequestStatus() == RequestStatusState.FAILED) {
        success = false;
        break;
      }
      Thread.sleep(500);
    }
    assertFalse(success);

    if (log.isInfoEnabled()) {
      log.info("--- current collection state: {}", cloudClient.getZkStateReader().getClusterState().getCollection(coll));
    }
    assertEquals(100, cluster.getSolrClient().query(coll, new SolrQuery("*:*")).getResults().getNumFound());
  }

  private CollectionAdminRequest.MoveReplica createMoveReplicaRequest(String coll, Replica replica, String targetNode, String shardId) {
    return new CollectionAdminRequest.MoveReplica(coll, shardId, targetNode, replica.getNodeName());
  }

  private CollectionAdminRequest.MoveReplica createMoveReplicaRequest(String coll, Replica replica, String targetNode) {
    return new CollectionAdminRequest.MoveReplica(coll, replica.getName(), targetNode);
  }

  private Replica getRandomReplica(String coll, CloudSolrClient cloudClient) {
    List<Replica> replicas = cloudClient.getZkStateReader().getClusterState().getCollection(coll).getReplicas();
    Collections.shuffle(replicas, random());
    return replicas.get(0);
  }

  private void checkNumOfCores(CloudSolrClient cloudClient, String nodeName, String collectionName, int expectedCores) throws IOException, SolrServerException {
    assertEquals(nodeName + " does not have expected number of cores", expectedCores, getNumOfCores(cloudClient, nodeName, collectionName));
  }

  private int getNumOfCores(CloudSolrClient cloudClient, String nodeName, String collectionName) throws IOException, SolrServerException {
    return getNumOfCores(cloudClient, nodeName, collectionName, null);
  }

  private int getNumOfCores(CloudSolrClient cloudClient, String nodeName, String collectionName, String replicaType) throws IOException, SolrServerException {
    try (HttpSolrClient coreclient = getHttpSolrClient(cloudClient.getZkStateReader().getBaseUrlForNodeName(nodeName))) {
      CoreAdminResponse status = CoreAdminRequest.getStatus(null, coreclient);
      if (status.getCoreStatus().size() == 0) {
        return 0;
      }
      if (collectionName == null && replicaType == null) {
        return status.getCoreStatus().size();
      }
      // filter size by collection name
      int size = 0;
      for (Map.Entry<String, NamedList<Object>> stringNamedListEntry : status.getCoreStatus()) {
        if (collectionName != null) {
          String coll = (String) stringNamedListEntry.getValue().findRecursive("cloud", "collection");
          if (!collectionName.equals(coll)) {
            continue;
          }
        }
        if (replicaType != null) {
          String type = (String) stringNamedListEntry.getValue().findRecursive("cloud", "replicaType");
          if (!replicaType.equals(type)) {
            continue;
          }
        }
        size++;
      }
      return size;
    }
  }

  protected void addDocs(String collection, int numDocs) throws Exception {
    SolrClient solrClient = cluster.getSolrClient();
    for (int docId = 1; docId <= numDocs; docId++) {
      SolrInputDocument doc = new SolrInputDocument();
      doc.addField("id", docId);
      solrClient.add(collection, doc);
    }
    solrClient.commit(collection);
    Thread.sleep(5000);
  }
}