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

import com.google.common.util.concurrent.AtomicDouble;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleEvent;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
import org.apache.hadoop.hdds.scm.container.SCMContainerManager;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerInfo;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB;
import org.apache.hadoop.hdds.scm.server.SCMChillModeManager;
import org.apache.hadoop.hdds.scm.server.SCMClientProtocolServer;
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
import org.apache.hadoop.ozone.HddsDatanodeService;
import org.apache.hadoop.ozone.MiniOzoneCluster;
import org.apache.hadoop.ozone.MiniOzoneClusterImpl;
import org.apache.hadoop.ozone.OzoneConfigKeys;
import org.apache.hadoop.ozone.TestStorageContainerManagerHelper;
import org.apache.hadoop.ozone.om.exceptions.OMException;
import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.test.LambdaTestUtils;
import org.apache.hadoop.util.Time;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

/**
 * Test Ozone Manager operation in distributed handler scenario.
 */
public class TestScmChillMode {

  private final static Logger LOG = LoggerFactory
      .getLogger(TestScmChillMode.class);
  private static MiniOzoneCluster cluster = null;
  private static MiniOzoneCluster.Builder builder = null;
  private static OzoneConfiguration conf;
  private static OzoneManager om;
  private static StorageContainerLocationProtocolClientSideTranslatorPB
      storageContainerLocationClient;


  @Rule
  public Timeout timeout = new Timeout(1000 * 200);

  /**
   * Create a MiniDFSCluster for testing.
   * <p>
   * Ozone is made active by setting OZONE_ENABLED = true and
   * OZONE_HANDLER_TYPE_KEY = "distributed"
   *
   * @throws IOException
   */
  @Before
  public void init() throws Exception {
    conf = new OzoneConfiguration();
    builder = MiniOzoneCluster.newBuilder(conf)
        .setHbInterval(1000)
        .setHbProcessorInterval(500)
        .setStartDataNodes(false);
    cluster = builder.build();
    cluster.startHddsDatanodes();
    cluster.waitForClusterToBeReady();
    om = cluster.getOzoneManager();
    storageContainerLocationClient = cluster
        .getStorageContainerLocationClient();
  }

  /**
   * Shutdown MiniDFSCluster.
   */
  @After
  public void shutdown() {
    if (cluster != null) {
      try {
        cluster.shutdown();
      } catch (Exception e) {
        // do nothing.
      }
    }
  }

  @Test
  public void testChillModeOperations() throws Exception {
    final AtomicReference<MiniOzoneCluster> miniCluster =
        new AtomicReference<>();
    // Create {numKeys} random names keys.
    TestStorageContainerManagerHelper helper =
        new TestStorageContainerManagerHelper(cluster, conf);
    Map<String, OmKeyInfo> keyLocations = helper.createKeys(100, 4096);
    final List<ContainerInfo> containers = cluster
        .getStorageContainerManager()
        .getContainerManager().getStateManager().getAllContainers();
    GenericTestUtils.waitFor(() -> {
      return containers.size() > 10;
    }, 100, 1000);

    String volumeName = "volume" + RandomStringUtils.randomNumeric(5);
    String bucketName = "bucket" + RandomStringUtils.randomNumeric(5);
    String keyName = "key" + RandomStringUtils.randomNumeric(5);
    String userName = "user" + RandomStringUtils.randomNumeric(5);
    String adminName = "admin" + RandomStringUtils.randomNumeric(5);
    OmKeyArgs keyArgs = new OmKeyArgs.Builder()
        .setVolumeName(volumeName)
        .setBucketName(bucketName)
        .setKeyName(keyName)
        .setDataSize(1000)
        .build();
    OmVolumeArgs volArgs = new OmVolumeArgs.Builder()
        .setAdminName(adminName)
        .setCreationTime(Time.monotonicNow())
        .setQuotaInBytes(10000)
        .setVolume(volumeName)
        .setOwnerName(userName)
        .build();
    OmBucketInfo bucketInfo = new OmBucketInfo.Builder()
        .setBucketName(bucketName)
        .setIsVersionEnabled(false)
        .setVolumeName(volumeName)
        .build();
    om.createVolume(volArgs);
    om.createBucket(bucketInfo);
    om.openKey(keyArgs);
    //om.commitKey(keyArgs, 1);

    cluster.stop();

    new Thread(() -> {
      try {
        miniCluster.set(builder.build());
      } catch (IOException e) {
        fail("failed");
      }
    }).start();

    StorageContainerManager scm;
    GenericTestUtils.waitFor(() -> {
      return miniCluster.get() != null;
    }, 100, 1000 * 3);
    cluster = miniCluster.get();

    scm = cluster.getStorageContainerManager();
    Assert.assertTrue(scm.isInChillMode());

    om = miniCluster.get().getOzoneManager();

    LambdaTestUtils.intercept(OMException.class,
        "ChillModePrecheck failed for allocateBlock",
        () -> om.openKey(keyArgs));
  }

  /**
   * Tests inChillMode & forceExitChillMode api calls.
   */
  @Test
  public void testIsScmInChillModeAndForceExit() throws Exception {
    final AtomicReference<MiniOzoneCluster> miniCluster =
        new AtomicReference<>();
    // Test 1: SCM should be out of chill mode.
    Assert.assertFalse(storageContainerLocationClient.inChillMode());
    cluster.stop();
    // Restart the cluster with same metadata dir.
    new Thread(() -> {
      try {
        miniCluster.set(builder.build());
      } catch (IOException e) {
        Assert.fail("Cluster startup failed.");
      }
    }).start();
    GenericTestUtils.waitFor(() -> {
      return miniCluster.get() != null;
    }, 10, 1000 * 3);
    cluster = miniCluster.get();

    // Test 2: Scm should be in chill mode as datanodes are not started yet.
    storageContainerLocationClient = cluster
        .getStorageContainerLocationClient();
    Assert.assertTrue(storageContainerLocationClient.inChillMode());
    // Force scm out of chill mode.
    cluster.getStorageContainerManager().getClientProtocolServer()
        .forceExitChillMode();
    // Test 3: SCM should be out of chill mode.
    GenericTestUtils.waitFor(() -> {
      try {
        return !cluster.getStorageContainerManager().getClientProtocolServer()
            .inChillMode();
      } catch (IOException e) {
        Assert.fail("Cluster");
        return false;
      }
    }, 10, 1000 * 5);

  }

  @Test
  public void testSCMChillMode() throws Exception {
    MiniOzoneCluster.Builder clusterBuilder = MiniOzoneCluster.newBuilder(conf)
        .setHbInterval(1000)
        .setNumDatanodes(3)
        .setStartDataNodes(false)
        .setHbProcessorInterval(500);
    MiniOzoneClusterImpl miniCluster = (MiniOzoneClusterImpl) clusterBuilder
        .build();
    // Test1: Test chill mode  when there are no containers in system.
    assertTrue(miniCluster.getStorageContainerManager().isInChillMode());
    miniCluster.startHddsDatanodes();
    miniCluster.waitForClusterToBeReady();
    assertFalse(miniCluster.getStorageContainerManager().isInChillMode());

    // Test2: Test chill mode  when containers are there in system.
    // Create {numKeys} random names keys.
    TestStorageContainerManagerHelper helper =
        new TestStorageContainerManagerHelper(miniCluster, conf);
    Map<String, OmKeyInfo> keyLocations = helper.createKeys(100 * 2, 4096);
    final List<ContainerInfo> containers = miniCluster
        .getStorageContainerManager().getContainerManager()
        .getStateManager().getAllContainers();
    GenericTestUtils.waitFor(() -> {
      return containers.size() > 10;
    }, 100, 1000 * 2);

    // Removing some container to keep them open.
    containers.remove(0);
    containers.remove(1);
    containers.remove(2);
    containers.remove(3);

    // Close remaining containers
    SCMContainerManager mapping = (SCMContainerManager) miniCluster
        .getStorageContainerManager().getContainerManager();
    containers.forEach(c -> {
      try {
        mapping.updateContainerState(c.getContainerID(),
            HddsProtos.LifeCycleEvent.FINALIZE);
        mapping.updateContainerState(c.getContainerID(),
            LifeCycleEvent.CLOSE);
      } catch (IOException e) {
        LOG.info("Failed to change state of open containers.", e);
      }
    });
    miniCluster.stop();

    GenericTestUtils.LogCapturer logCapturer = GenericTestUtils.LogCapturer
        .captureLogs(SCMChillModeManager.getLogger());
    logCapturer.clearOutput();
    AtomicReference<MiniOzoneCluster> miniClusterOzone
        = new AtomicReference<>();
    new Thread(() -> {
      try {
        miniClusterOzone.set(clusterBuilder.setStartDataNodes(false).build());
      } catch (IOException e) {
        fail("failed");
      }
    }).start();

    StorageContainerManager scm;
    GenericTestUtils.waitFor(() -> {
      return miniClusterOzone.get() != null;
    }, 100, 1000 * 3);

    miniCluster = (MiniOzoneClusterImpl) miniClusterOzone.get();

    scm = miniCluster.getStorageContainerManager();
    assertTrue(scm.isInChillMode());
    assertFalse(logCapturer.getOutput().contains("SCM exiting chill mode."));
    assertTrue(scm.getCurrentContainerThreshold() == 0);
    AtomicDouble curThreshold = new AtomicDouble();
    AtomicDouble lastReportedThreshold = new AtomicDouble();
    for (HddsDatanodeService dn : miniCluster.getHddsDatanodes()) {
      dn.start(null);
      GenericTestUtils.waitFor(() -> {
        curThreshold.set(scm.getCurrentContainerThreshold());
        return curThreshold.get() > lastReportedThreshold.get();
      }, 100, 1000 * 5);
      lastReportedThreshold.set(curThreshold.get());
    }
    cluster = miniCluster;
    double chillModeCutoff = conf
        .getDouble(HddsConfigKeys.HDDS_SCM_CHILLMODE_THRESHOLD_PCT,
            HddsConfigKeys.HDDS_SCM_CHILLMODE_THRESHOLD_PCT_DEFAULT);
    assertTrue(scm.getCurrentContainerThreshold() >= chillModeCutoff);
    assertTrue(logCapturer.getOutput().contains("SCM exiting chill mode."));
    assertFalse(scm.isInChillMode());
  }

  @Test
  public void testSCMChillModeRestrictedOp() throws Exception {
    conf.set(OzoneConfigKeys.OZONE_METADATA_STORE_IMPL,
        OzoneConfigKeys.OZONE_METADATA_STORE_IMPL_LEVELDB);
    cluster.stop();
    cluster = builder.build();
    StorageContainerManager scm = cluster.getStorageContainerManager();
    assertTrue(scm.isInChillMode());

    LambdaTestUtils.intercept(SCMException.class,
        "ChillModePrecheck failed for allocateContainer", () -> {
          scm.getClientProtocolServer()
              .allocateContainer(ReplicationType.STAND_ALONE,
                  ReplicationFactor.ONE, "");
        });

    cluster.startHddsDatanodes();
    cluster.waitForClusterToBeReady();
    assertFalse(scm.isInChillMode());

    TestStorageContainerManagerHelper helper =
        new TestStorageContainerManagerHelper(cluster, conf);
    helper.createKeys(10, 4096);
    SCMClientProtocolServer clientProtocolServer = cluster
        .getStorageContainerManager().getClientProtocolServer();
    assertFalse((scm.getClientProtocolServer()).getChillModeStatus());
    final List<ContainerInfo> containers = scm.getContainerManager()
        .getStateManager().getAllContainers();
    scm.getEventQueue().fireEvent(SCMEvents.CHILL_MODE_STATUS, true);
    GenericTestUtils.waitFor(() -> {
      return clientProtocolServer.getChillModeStatus();
    }, 50, 1000 * 5);
    assertTrue(clientProtocolServer.getChillModeStatus());

    LambdaTestUtils.intercept(SCMException.class,
        "Open container " + containers.get(0).getContainerID() + " "
            + "doesn't have enough replicas to service this operation in Chill"
            + " mode.", () -> clientProtocolServer
            .getContainerWithPipeline(containers.get(0).getContainerID()));
  }

}