  /**
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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 org.apache.commons.lang3.RandomStringUtils;
import org.apache.hadoop.hdds.protocol.StorageType;
import org.apache.hadoop.hdfs.LogVerificationAppender;
import org.apache.hadoop.ozone.OzoneTestUtils;
import org.apache.hadoop.ozone.client.ObjectStore;
import org.apache.hadoop.ozone.client.OzoneBucket;
import org.apache.hadoop.ozone.client.BucketArgs;
import org.apache.hadoop.ozone.client.OzoneVolume;
import org.apache.hadoop.ozone.client.OzoneClientFactory;
import org.apache.hadoop.ozone.client.OzoneClient;
import org.apache.hadoop.ozone.client.VolumeArgs;
import org.apache.hadoop.ozone.om.exceptions.OMException;
import org.apache.hadoop.ozone.om.ha.OMFailoverProxyProvider;
import org.apache.hadoop.ozone.om.ha.OMProxyInfo;
import org.apache.hadoop.ozone.om.helpers.OMRatisHelper;
import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer;
import org.apache.hadoop.ozone.om.request.volume.OMVolumeCreateRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CreateVolumeRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.VolumeInfo;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.log4j.Logger;
import org.apache.ratis.protocol.ClientId;
import org.apache.ratis.protocol.Message;
import org.apache.ratis.protocol.RaftClientReply;
import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.server.RaftServer;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;

import javax.management.MBeanInfo;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.lang.management.ManagementFactory;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.time.Instant;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.Iterator;
import java.util.UUID;

import static org.apache.hadoop.ozone.MiniOzoneHAClusterImpl.NODE_FAILURE_TIMEOUT;
import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT;

import static org.apache.ratis.metrics.RatisMetrics.RATIS_APPLICATION_NAME_METRICS;
import static org.junit.Assert.fail;

/**
 * Test Ozone Manager Metadata operation in distributed handler scenario.
 */
public class TestOzoneManagerHAMetadataOnly extends TestOzoneManagerHA {

  private OzoneVolume createAndCheckVolume(String volumeName)
      throws Exception {
    String userName = "user" + RandomStringUtils.randomNumeric(5);
    String adminName = "admin" + RandomStringUtils.randomNumeric(5);
    VolumeArgs createVolumeArgs = VolumeArgs.newBuilder()
        .setOwner(userName)
        .setAdmin(adminName)
        .build();

    ObjectStore objectStore = getObjectStore();
    objectStore.createVolume(volumeName, createVolumeArgs);

    OzoneVolume retVolume = objectStore.getVolume(volumeName);

    Assert.assertTrue(retVolume.getName().equals(volumeName));
    Assert.assertTrue(retVolume.getOwner().equals(userName));
    Assert.assertTrue(retVolume.getAdmin().equals(adminName));

    return retVolume;
  }
  @Test
  public void testAllVolumeOperations() throws Exception {

    String volumeName = "volume" + RandomStringUtils.randomNumeric(5);

    createAndCheckVolume(volumeName);

    ObjectStore objectStore = getObjectStore();
    objectStore.deleteVolume(volumeName);

    OzoneTestUtils.expectOmException(OMException.ResultCodes.VOLUME_NOT_FOUND,
        () -> objectStore.getVolume(volumeName));

    OzoneTestUtils.expectOmException(OMException.ResultCodes.VOLUME_NOT_FOUND,
        () -> objectStore.deleteVolume(volumeName));
  }


  @Test
  public void testAllBucketOperations() throws Exception {

    String volumeName = "volume" + RandomStringUtils.randomNumeric(5);
    String bucketName = "volume" + RandomStringUtils.randomNumeric(5);

    OzoneVolume retVolume = createAndCheckVolume(volumeName);

    BucketArgs bucketArgs =
        BucketArgs.newBuilder().setStorageType(StorageType.DISK)
            .setVersioning(true).build();


    retVolume.createBucket(bucketName, bucketArgs);


    OzoneBucket ozoneBucket = retVolume.getBucket(bucketName);

    Assert.assertEquals(volumeName, ozoneBucket.getVolumeName());
    Assert.assertEquals(bucketName, ozoneBucket.getName());
    Assert.assertTrue(ozoneBucket.getVersioning());
    Assert.assertEquals(StorageType.DISK, ozoneBucket.getStorageType());
    Assert.assertFalse(ozoneBucket.getCreationTime().isAfter(Instant.now()));


    // Change versioning to false
    ozoneBucket.setVersioning(false);

    ozoneBucket = retVolume.getBucket(bucketName);
    Assert.assertFalse(ozoneBucket.getVersioning());

    retVolume.deleteBucket(bucketName);

    OzoneTestUtils.expectOmException(OMException.ResultCodes.BUCKET_NOT_FOUND,
        () -> retVolume.deleteBucket(bucketName));
  }

  /**
   * Test that OMFailoverProxyProvider creates an OM proxy for each OM in the
   * cluster.
   */
  @Test
  public void testOMProxyProviderInitialization() throws Exception {
    OzoneClient rpcClient = getCluster().getRpcClient();

    OMFailoverProxyProvider omFailoverProxyProvider =
        OmFailoverProxyUtil.getFailoverProxyProvider(
            rpcClient.getObjectStore().getClientProxy());

    List<OMProxyInfo> omProxies =
        omFailoverProxyProvider.getOMProxyInfos();

    Assert.assertEquals(getNumOfOMs(), omProxies.size());

    for (int i = 0; i < getNumOfOMs(); i++) {
      InetSocketAddress omRpcServerAddr =
          getCluster().getOzoneManager(i).getOmRpcServerAddr();
      boolean omClientProxyExists = false;
      for (OMProxyInfo omProxyInfo : omProxies) {
        if (omProxyInfo.getAddress().equals(omRpcServerAddr)) {
          omClientProxyExists = true;
          break;
        }
      }
      Assert.assertTrue("There is no OM Client Proxy corresponding to OM " +
              "node" + getCluster().getOzoneManager(i).getOMNodeId(),
          omClientProxyExists);
    }
  }

  /**
   * Test OMFailoverProxyProvider failover on connection exception to OM client.
   */
  @Ignore("This test randomly failing. Let's enable once its fixed.")
  @Test
  public void testOMProxyProviderFailoverOnConnectionFailure()
      throws Exception {
    ObjectStore objectStore = getObjectStore();
    OMFailoverProxyProvider omFailoverProxyProvider =
        OmFailoverProxyUtil
            .getFailoverProxyProvider(objectStore.getClientProxy());
    String firstProxyNodeId = omFailoverProxyProvider.getCurrentProxyOMNodeId();

    createVolumeTest(true);

    // On stopping the current OM Proxy, the next connection attempt should
    // failover to a another OM proxy.
    getCluster().stopOzoneManager(firstProxyNodeId);
    Thread.sleep(OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT * 4);

    // Next request to the proxy provider should result in a failover
    createVolumeTest(true);
    Thread.sleep(OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT);

    // Get the new OM Proxy NodeId
    String newProxyNodeId = omFailoverProxyProvider.getCurrentProxyOMNodeId();

    // Verify that a failover occured. the new proxy nodeId should be
    // different from the old proxy nodeId.
    Assert.assertNotEquals("Failover did not occur as expected",
        firstProxyNodeId, newProxyNodeId);
  }

  /**
   * Test OMFailoverProxyProvider failover when current OM proxy is not
   * the current OM Leader.
   */
  @Test
  public void testOMProxyProviderFailoverToCurrentLeader() throws Exception {
    ObjectStore objectStore = getObjectStore();
    OMFailoverProxyProvider omFailoverProxyProvider = OmFailoverProxyUtil
        .getFailoverProxyProvider(objectStore.getClientProxy());

    // Run couple of createVolume tests to discover the current Leader OM
    createVolumeTest(true);
    createVolumeTest(true);

    // The OMFailoverProxyProvider will point to the current leader OM node.
    String leaderOMNodeId = omFailoverProxyProvider.getCurrentProxyOMNodeId();

    // Perform a manual failover of the proxy provider to move the
    // currentProxyIndex to a node other than the leader OM.
    omFailoverProxyProvider.performFailoverToNextProxy();

    String newProxyNodeId = omFailoverProxyProvider.getCurrentProxyOMNodeId();
    Assert.assertNotEquals(leaderOMNodeId, newProxyNodeId);

    // Once another request is sent to this new proxy node, the leader
    // information must be returned via the response and a failover must
    // happen to the leader proxy node.
    createVolumeTest(true);
    Thread.sleep(2000);

    String newLeaderOMNodeId =
        omFailoverProxyProvider.getCurrentProxyOMNodeId();

    // The old and new Leader OM NodeId must match since there was no new
    // election in the Ratis ring.
    Assert.assertEquals(leaderOMNodeId, newLeaderOMNodeId);
  }

  @Test
  public void testOMRetryProxy() throws Exception {
    int maxFailoverAttempts = getOzoneClientFailoverMaxAttempts();
    // Stop all the OMs.
    for (int i = 0; i < getNumOfOMs(); i++) {
      getCluster().stopOzoneManager(i);
    }

    final LogVerificationAppender appender = new LogVerificationAppender();
    final Logger logger = Logger.getRootLogger();
    logger.addAppender(appender);

    try {
      createVolumeTest(true);
      // After making N (set maxRetries value) connection attempts to OMs,
      // the RpcClient should give up.
      fail("TestOMRetryProxy should fail when there are no OMs running");
    } catch (ConnectException e) {
      Assert.assertEquals(1,
          appender.countLinesWithMessage("Failed to connect to OMs:"));
      Assert.assertEquals(maxFailoverAttempts,
          appender.countLinesWithMessage("Trying to failover"));
      Assert.assertEquals(1,
          appender.countLinesWithMessage("Attempted " +
              maxFailoverAttempts + " failovers."));
    }
  }

  @Test
  public void testReadRequest() throws Exception {
    String volumeName = "volume" + RandomStringUtils.randomNumeric(5);
    ObjectStore objectStore = getObjectStore();
    objectStore.createVolume(volumeName);

    OMFailoverProxyProvider omFailoverProxyProvider = OmFailoverProxyUtil
        .getFailoverProxyProvider(objectStore.getClientProxy());

    String currentLeaderNodeId = omFailoverProxyProvider
        .getCurrentProxyOMNodeId();

    // A read request from any proxy should failover to the current leader OM
    for (int i = 0; i < getNumOfOMs(); i++) {
      // Failover OMFailoverProxyProvider to OM at index i
      OzoneManager ozoneManager = getCluster().getOzoneManager(i);

      // Get the ObjectStore and FailoverProxyProvider for OM at index i
      final ObjectStore store = OzoneClientFactory.getRpcClient(
          getOmServiceId(), getConf()).getObjectStore();
      final OMFailoverProxyProvider proxyProvider =
          OmFailoverProxyUtil.getFailoverProxyProvider(store.getClientProxy());

      // Failover to the OM node that the objectStore points to
      omFailoverProxyProvider.performFailoverIfRequired(
          ozoneManager.getOMNodeId());

      // A read request should result in the proxyProvider failing over to
      // leader node.
      OzoneVolume volume = store.getVolume(volumeName);
      Assert.assertEquals(volumeName, volume.getName());

      Assert.assertEquals(currentLeaderNodeId,
          proxyProvider.getCurrentProxyOMNodeId());
    }
  }

  @Ignore("This test randomly failing. Let's enable once its fixed.")
  @Test
  public void testListVolumes() throws Exception {
    String userName = UserGroupInformation.getCurrentUser().getUserName();
    String adminName = userName;
    ObjectStore objectStore = getObjectStore();

    Set<String> expectedVolumes = new TreeSet<>();
    for (int i=0; i < 100; i++) {
      String volumeName = "vol" + i;
      expectedVolumes.add(volumeName);
      VolumeArgs createVolumeArgs = VolumeArgs.newBuilder()
          .setOwner(userName)
          .setAdmin(adminName)
          .build();
      objectStore.createVolume(volumeName, createVolumeArgs);
    }

    validateVolumesList(userName, expectedVolumes);

    // Stop leader OM, and then validate list volumes for user.
    stopLeaderOM();
    Thread.sleep(NODE_FAILURE_TIMEOUT * 2);

    validateVolumesList(userName, expectedVolumes);

  }

  @Test
  public void testJMXMetrics() throws Exception {
    // Verify any one ratis metric is exposed by JMX MBeanServer
    OzoneManagerRatisServer ratisServer =
        getCluster().getOzoneManager(0).getOmRatisServer();
    ObjectName oname = new ObjectName(RATIS_APPLICATION_NAME_METRICS, "name",
        RATIS_APPLICATION_NAME_METRICS + ".log_worker." +
            ratisServer.getRaftPeerId().toString() +
            "@" + ratisServer.getRaftGroup().getGroupId() + ".flushCount");
    MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
    MBeanInfo mBeanInfo = mBeanServer.getMBeanInfo(oname);
    Assert.assertNotNull(mBeanInfo);
    Object flushCount = mBeanServer.getAttribute(oname, "Count");
    Assert.assertTrue((long) flushCount >= 0);
  }

  @Test
  public void testOMRetryCache() throws Exception {
    ObjectStore objectStore = getObjectStore();
    objectStore.createVolume(UUID.randomUUID().toString());


    OMFailoverProxyProvider omFailoverProxyProvider = OmFailoverProxyUtil
        .getFailoverProxyProvider(objectStore.getClientProxy());

    String currentLeaderNodeId = omFailoverProxyProvider
        .getCurrentProxyOMNodeId();

    OzoneManagerRatisServer ozoneManagerRatisServer =
        getCluster().getOzoneManager(currentLeaderNodeId).getOmRatisServer();

    RaftServer raftServer = ozoneManagerRatisServer.getServer();

    ClientId clientId = ClientId.randomId();
    long callId = 2000L;
    String userName = UserGroupInformation.getCurrentUser().getUserName();
    String volumeName = UUID.randomUUID().toString();


    GenericTestUtils.LogCapturer logCapturer = GenericTestUtils.LogCapturer
        .captureLogs(OMVolumeCreateRequest.getLogger());
    OMRequest omRequest =
        OMRequest.newBuilder().setCreateVolumeRequest(
            CreateVolumeRequest.newBuilder().setVolumeInfo(
                VolumeInfo.newBuilder().setOwnerName(userName)
                    .setAdminName(userName).setVolume(volumeName).build())
                .build()).setClientId(UUID.randomUUID().toString())
            .setCmdType(OzoneManagerProtocolProtos.Type.CreateVolume).build();

    RaftClientReply raftClientReply =
        raftServer.submitClientRequest(RaftClientRequest.newBuilder()
            .setClientId(clientId)
            .setServerId(raftServer.getId())
            .setGroupId(ozoneManagerRatisServer.getRaftGroup().getGroupId())
            .setCallId(callId)
            .setMessage(
                Message.valueOf(
                    OMRatisHelper.convertRequestToByteString(omRequest)))
            .setType(RaftClientRequest.writeRequestType())
            .build());

    Assert.assertTrue(raftClientReply.isSuccess());

    Assert.assertTrue(logCapturer.getOutput().contains("created volume:"
        + volumeName));

    logCapturer.clearOutput();

    raftClientReply =
        raftServer.submitClientRequest(RaftClientRequest.newBuilder()
            .setClientId(clientId)
            .setServerId(raftServer.getId())
            .setGroupId(ozoneManagerRatisServer.getRaftGroup().getGroupId())
            .setCallId(callId)
            .setMessage(
                Message.valueOf(
                    OMRatisHelper.convertRequestToByteString(omRequest)))
            .setType(RaftClientRequest.writeRequestType())
            .build());

    Assert.assertTrue(raftClientReply.isSuccess());

    // As second time with same client id and call id, this request should
    // not be executed ratis server should return from cache.
    // If 2nd time executed, it will fail with Volume creation failed. check
    // for that.
    Assert.assertFalse(logCapturer.getOutput().contains(
        "Volume creation failed"));

    //Sleep for little above retry cache duration to get cache clear.
    Thread.sleep(getRetryCacheDuration().toMillis() + 5000);

    raftClientReply =
        raftServer.submitClientRequest(RaftClientRequest.newBuilder()
            .setClientId(clientId)
            .setServerId(raftServer.getId())
            .setGroupId(ozoneManagerRatisServer.getRaftGroup().getGroupId())
            .setCallId(callId)
            .setMessage(
                Message.valueOf(
                    OMRatisHelper.convertRequestToByteString(omRequest)))
            .setType(RaftClientRequest.writeRequestType())
            .build());

    Assert.assertTrue(raftClientReply.isSuccess());

    // As second time with same client id and call id, this request should
    // be executed by ratis server as we are sending this request after cache
    // expiry duration.
    Assert.assertTrue(logCapturer.getOutput().contains(
        "Volume creation failed"));

  }

  private void validateVolumesList(String userName,
      Set<String> expectedVolumes) throws Exception {
    ObjectStore objectStore = getObjectStore();

    int expectedCount = 0;
    Iterator<? extends OzoneVolume> volumeIterator =
        objectStore.listVolumesByUser(userName, "", "");

    while (volumeIterator.hasNext()) {
      OzoneVolume next = volumeIterator.next();
      Assert.assertTrue(expectedVolumes.contains(next.getName()));
      expectedCount++;
    }

    Assert.assertEquals(expectedVolumes.size(),  expectedCount);
  }
}