/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hdfs.server.namenode.ha;

import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.anyBoolean;
import static org.mockito.Matchers.anyInt;
import static org.mockito.Matchers.anyLong;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

import java.io.IOException;
import java.util.Collection;
import java.util.LinkedList;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.ha.ServiceFailedException;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.HAUtil;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.MiniDFSNNTopology;
import org.apache.hadoop.hdfs.server.namenode.EditLogInputException;
import org.apache.hadoop.hdfs.server.namenode.EditLogInputStream;
import org.apache.hadoop.hdfs.server.namenode.FSEditLog;
import org.apache.hadoop.hdfs.server.namenode.FSEditLogOp;
import org.apache.hadoop.hdfs.server.namenode.NameNode;
import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;

import com.google.common.collect.ImmutableList;

public class TestFailureToReadEdits {
  
  private static final Log LOG = LogFactory.getLog(TestFailureToReadEdits.class);
  
  private static final String TEST_DIR1 = "/test1";
  private static final String TEST_DIR2 = "/test2";
  private static final String TEST_DIR3 = "/test3";
  
  private Configuration conf;
  private Runtime mockRuntime = mock(Runtime.class);
  private MiniDFSCluster cluster;
  private NameNode nn0;
  private NameNode nn1;
  private FileSystem fs;
  
  @Before
  public void setUpCluster() throws Exception {
    conf = new Configuration();
    conf.setInt(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_CHECK_PERIOD_KEY, 1);
    conf.setInt(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_TXNS_KEY, 1);
    conf.setInt(DFSConfigKeys.DFS_NAMENODE_NUM_CHECKPOINTS_RETAINED_KEY, 10);
    conf.setInt(DFSConfigKeys.DFS_HA_TAILEDITS_PERIOD_KEY, 1);
    HAUtil.setAllowStandbyReads(conf, true);
    
    MiniDFSNNTopology topology = new MiniDFSNNTopology()
      .addNameservice(new MiniDFSNNTopology.NSConf("ns1")
        .addNN(new MiniDFSNNTopology.NNConf("nn1").setHttpPort(10001))
        .addNN(new MiniDFSNNTopology.NNConf("nn2").setHttpPort(10002)));
    cluster = new MiniDFSCluster.Builder(conf)
      .nnTopology(topology)
      .numDataNodes(0)
      .build();
    
    cluster.waitActive();
    
    nn0 = cluster.getNameNode(0);
    nn1 = cluster.getNameNode(1);
    nn1.getNamesystem().getEditLogTailer().setRuntime(mockRuntime);
    
    cluster.transitionToActive(0);
    fs = HATestUtil.configureFailoverFs(cluster, conf);
  }
  
  @After
  public void tearDownCluster() throws Exception {
    if (fs != null) {
      fs.close();
    }
    
    if (cluster != null) {
      cluster.shutdown();
    }
  }

  /**
   * Test that the standby NN won't double-replay earlier edits if it encounters
   * a failure to read a later edit.
   */
  @Test
  public void testFailuretoReadEdits() throws Exception {
    assertTrue(fs.mkdirs(new Path(TEST_DIR1)));
    HATestUtil.waitForStandbyToCatchUp(nn0, nn1);
    
    // If these two ops are applied twice, the first op will throw an
    // exception the second time its replayed.
    fs.setOwner(new Path(TEST_DIR1), "foo", "bar");
    assertTrue(fs.delete(new Path(TEST_DIR1), true));
    
    // This op should get applied just fine.
    assertTrue(fs.mkdirs(new Path(TEST_DIR2)));
    
    // This is the op the mocking will cause to fail to be read.
    assertTrue(fs.mkdirs(new Path(TEST_DIR3)));
    
    LimitedEditLogAnswer answer = causeFailureOnEditLogRead();
    
    try {
      HATestUtil.waitForStandbyToCatchUp(nn0, nn1);
      fail("Standby fully caught up, but should not have been able to");
    } catch (HATestUtil.CouldNotCatchUpException e) {
      verify(mockRuntime, times(0)).exit(anyInt());
    }
    
    // Null because it was deleted.
    assertNull(NameNodeAdapter.getFileInfo(nn1,
        TEST_DIR1, false));
    // Should have been successfully created.
    assertTrue(NameNodeAdapter.getFileInfo(nn1,
        TEST_DIR2, false).isDir());
    // Null because it hasn't been created yet.
    assertNull(NameNodeAdapter.getFileInfo(nn1,
        TEST_DIR3, false));
    
    // Now let the standby read ALL the edits.
    answer.setThrowExceptionOnRead(false);
    HATestUtil.waitForStandbyToCatchUp(nn0, nn1);
    
    // Null because it was deleted.
    assertNull(NameNodeAdapter.getFileInfo(nn1,
        TEST_DIR1, false));
    // Should have been successfully created.
    assertTrue(NameNodeAdapter.getFileInfo(nn1,
        TEST_DIR2, false).isDir());
    // Should now have been successfully created.
    assertTrue(NameNodeAdapter.getFileInfo(nn1,
        TEST_DIR3, false).isDir());
  }
  
  /**
   * Test the following case:
   * 1. SBN is reading a finalized edits file when NFS disappears halfway
   *    through (or some intermittent error happens)
   * 2. SBN performs a checkpoint and uploads it to the NN
   * 3. NN receives a checkpoint that doesn't correspond to the end of any log
   *    segment
   * 4. Both NN and SBN should be able to restart at this point.
   * 
   * This is a regression test for HDFS-2766.
   */
  @Test
  public void testCheckpointStartingMidEditsFile() throws Exception {
    assertTrue(fs.mkdirs(new Path(TEST_DIR1)));
    
    HATestUtil.waitForStandbyToCatchUp(nn0, nn1);
    
    // Once the standby catches up, it should notice that it needs to
    // do a checkpoint and save one to its local directories.
    HATestUtil.waitForCheckpoint(cluster, 1, ImmutableList.of(0, 3));
    
    // It should also upload it back to the active.
    HATestUtil.waitForCheckpoint(cluster, 0, ImmutableList.of(0, 3));
    
    causeFailureOnEditLogRead();
    
    assertTrue(fs.mkdirs(new Path(TEST_DIR2)));
    assertTrue(fs.mkdirs(new Path(TEST_DIR3)));
    
    try {
      HATestUtil.waitForStandbyToCatchUp(nn0, nn1);
      fail("Standby fully caught up, but should not have been able to");
    } catch (HATestUtil.CouldNotCatchUpException e) {
      verify(mockRuntime, times(0)).exit(anyInt());
    }
    
    // 5 because we should get OP_START_LOG_SEGMENT and one successful OP_MKDIR
    HATestUtil.waitForCheckpoint(cluster, 1, ImmutableList.of(0, 3, 5));
    
    // It should also upload it back to the active.
    HATestUtil.waitForCheckpoint(cluster, 0, ImmutableList.of(0, 3, 5));

    // Restart the active NN
    cluster.restartNameNode(0);
    
    HATestUtil.waitForCheckpoint(cluster, 0, ImmutableList.of(0, 3, 5));
    
    FileSystem fs0 = null;
    try {
      // Make sure that when the active restarts, it loads all the edits.
      fs0 = FileSystem.get(NameNode.getUri(nn0.getNameNodeAddress()),
          conf);
      
      assertTrue(fs0.exists(new Path(TEST_DIR1)));
      assertTrue(fs0.exists(new Path(TEST_DIR2)));
      assertTrue(fs0.exists(new Path(TEST_DIR3)));
    } finally {
      if (fs0 != null)
        fs0.close();
    }
  }

  /**
   * Ensure that the standby fails to become active if it cannot read all
   * available edits in the shared edits dir when it is transitioning to active
   * state.
   */
  @Test
  public void testFailureToReadEditsOnTransitionToActive() throws Exception {
    assertTrue(fs.mkdirs(new Path(TEST_DIR1)));
    
    HATestUtil.waitForStandbyToCatchUp(nn0, nn1);
    
    // It should also upload it back to the active.
    HATestUtil.waitForCheckpoint(cluster, 0, ImmutableList.of(0, 3));
    
    causeFailureOnEditLogRead();
    
    assertTrue(fs.mkdirs(new Path(TEST_DIR2)));
    assertTrue(fs.mkdirs(new Path(TEST_DIR3)));
    
    try {
      HATestUtil.waitForStandbyToCatchUp(nn0, nn1);
      fail("Standby fully caught up, but should not have been able to");
    } catch (HATestUtil.CouldNotCatchUpException e) {
      verify(mockRuntime, times(0)).exit(anyInt());
    }
    
    // Shutdown the active NN.
    cluster.shutdownNameNode(0);
    
    try {
      // Transition the standby to active.
      cluster.transitionToActive(1);
      fail("Standby transitioned to active, but should not have been able to");
    } catch (ServiceFailedException sfe) {
      LOG.info("got expected exception: " + sfe.toString(), sfe);
      assertTrue("Standby failed to catch up for some reason other than "
          + "failure to read logs", sfe.toString().contains(
              EditLogInputException.class.getName()));
    }
  }
  
  private LimitedEditLogAnswer causeFailureOnEditLogRead() throws IOException {
    FSEditLog spyEditLog = spy(nn1.getNamesystem().getEditLogTailer()
        .getEditLog());
    LimitedEditLogAnswer answer = new LimitedEditLogAnswer(); 
    doAnswer(answer).when(spyEditLog).selectInputStreams(
        anyLong(), anyLong(), anyBoolean());
    nn1.getNamesystem().getEditLogTailer().setEditLog(spyEditLog);
    
    return answer;
  }
  
  private static class LimitedEditLogAnswer
      implements Answer<Collection<EditLogInputStream>> {
    
    private boolean throwExceptionOnRead = true;

    @SuppressWarnings("unchecked")
    @Override
    public Collection<EditLogInputStream> answer(InvocationOnMock invocation)
        throws Throwable {
      Collection<EditLogInputStream> streams = (Collection<EditLogInputStream>)
          invocation.callRealMethod();
  
      if (!throwExceptionOnRead) {
        return streams;
      } else {
        Collection<EditLogInputStream> ret = new LinkedList<EditLogInputStream>();
        for (EditLogInputStream stream : streams) {
          EditLogInputStream spyStream = spy(stream);
          doAnswer(new Answer<FSEditLogOp>() {

            @Override
            public FSEditLogOp answer(InvocationOnMock invocation)
                throws Throwable {
              FSEditLogOp op = (FSEditLogOp) invocation.callRealMethod();
              if (throwExceptionOnRead &&
                  TEST_DIR3.equals(NameNodeAdapter.getMkdirOpPath(op))) {
                throw new IOException("failed to read op creating " + TEST_DIR3);
              } else {
                return op;
              }
            }
            
          }).when(spyStream).readOp();
          ret.add(spyStream);
        }
        return ret;
      }
    }
    
    public void setThrowExceptionOnRead(boolean throwExceptionOnRead) {
      this.throwExceptionOnRead = throwExceptionOnRead;
    }
  }
  
}
