/*
 * 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.yarn.server.resourcemanager.scheduler;

import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableMap;
import org.apache.hadoop.thirdparty.com.google.common.collect.Maps;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.metrics2.MetricsSource;
import org.apache.hadoop.metrics2.MetricsSystem;
import org.apache.hadoop.metrics2.impl.MetricsSystemImpl;
import org.apache.hadoop.metrics2.lib.MutableCounterLong;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceInformation;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.metrics.CustomResourceMetricValue;
import org.apache.hadoop.yarn.resourcetypes.ResourceTypesTestHelper;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;


import org.apache.hadoop.yarn.util.resource.ResourceUtils;
import org.junit.Before;
import org.junit.Test;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.function.Function;

import static org.apache.hadoop.yarn.conf.YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB;
import static org.apache.hadoop.yarn.conf.YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES;
import static org.apache.hadoop.yarn.resourcetypes.ResourceTypesTestHelper
    .extractCustomResourcesAsStrings;
import static org.apache.hadoop.yarn.resourcetypes.ResourceTypesTestHelper.newResource;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
    .ResourceMetricsChecker.ResourceMetricsKey.AGGREGATE_CONTAINERS_ALLOCATED;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
    .ResourceMetricsChecker.ResourceMetricsKey.AGGREGATE_CONTAINERS_RELEASED;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
    .ResourceMetricsChecker.ResourceMetricsKey.AGGREGATE_MEMORY_MB_SECONDS_PREEMPTED;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
    .ResourceMetricsChecker.ResourceMetricsKey.AGGREGATE_VCORE_SECONDS_PREEMPTED;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
    .ResourceMetricsChecker.ResourceMetricsKey.ALLOCATED_CONTAINERS;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
    .ResourceMetricsChecker.ResourceMetricsKey.ALLOCATED_MB;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
    .ResourceMetricsChecker.ResourceMetricsKey.ALLOCATED_V_CORES;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.AVAILABLE_MB;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.AVAILABLE_V_CORES;

import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
    .ResourceMetricsChecker.ResourceMetricsKey.PENDING_CONTAINERS;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
    .ResourceMetricsChecker.ResourceMetricsKey.PENDING_MB;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
    .ResourceMetricsChecker.ResourceMetricsKey.PENDING_V_CORES;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.RESERVED_CONTAINERS;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.RESERVED_MB;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.RESERVED_V_CORES;

import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.ALLOCATED_CUSTOM_RES1;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.ALLOCATED_CUSTOM_RES2;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.AVAILABLE_CUSTOM_RES1;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.AVAILABLE_CUSTOM_RES2;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.PENDING_CUSTOM_RES1;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.PENDING_CUSTOM_RES2;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.RESERVED_CUSTOM_RES1;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.RESERVED_CUSTOM_RES2;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.AGGREGATE_PREEMPTED_SECONDS_CUSTOM_RES1;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceMetricsChecker.ResourceMetricsKey.AGGREGATE_PREEMPTED_SECONDS_CUSTOM_RES2;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.TestQueueMetrics.queueSource;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;

public class TestQueueMetricsForCustomResources {
  public enum MetricsForCustomResource {
    ALLOCATED, AVAILABLE, PENDING, RESERVED, AGGREGATE_PREEMPTED_SECONDS
  }

  public static final long GB = 1024; // MB
  private static final Configuration CONF = new Configuration();
  public static final String CUSTOM_RES_1 = "custom_res_1";
  public static final String CUSTOM_RES_2 = "custom_res_2";
  public static final String USER = "alice";
  private Resource defaultResource;
  private MetricsSystem ms;

  @Before
  public void setUp() {
    ms = new MetricsSystemImpl();
    QueueMetrics.clearQueueMetrics();
    initializeResourceTypes();
    createDefaultResource();
  }

  private void createDefaultResource() {
    defaultResource = newResource(4 * GB, 4,
        ImmutableMap.<String, String> builder()
            .put(CUSTOM_RES_1, String.valueOf(15 * GB))
            .put(CUSTOM_RES_2, String.valueOf(20 * GB))
            .build());
  }

  private void initializeResourceTypes() {
    Map<String, ResourceInformation> riMap = new HashMap<>();

    ResourceInformation memory = ResourceInformation.newInstance(
        ResourceInformation.MEMORY_MB.getName(),
        ResourceInformation.MEMORY_MB.getUnits(),
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB,
        DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB);
    ResourceInformation vcores = ResourceInformation.newInstance(
        ResourceInformation.VCORES.getName(),
        ResourceInformation.VCORES.getUnits(),
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES,
        DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES);
    ResourceInformation res1 = ResourceInformation.newInstance(CUSTOM_RES_1,
        ResourceInformation.VCORES.getUnits(), 0, 2000);
    ResourceInformation res2 = ResourceInformation.newInstance(CUSTOM_RES_2,
        ResourceInformation.VCORES.getUnits(), 0, 2000);

    riMap.put(ResourceInformation.MEMORY_URI, memory);
    riMap.put(ResourceInformation.VCORES_URI, vcores);
    riMap.put(CUSTOM_RES_1, res1);
    riMap.put(CUSTOM_RES_2, res2);
    ResourceUtils.initializeResourcesFromResourceInformationMap(riMap);
  }

  private static void assertCustomResourceValue(QueueMetrics metrics,
      MetricsForCustomResource metricsType,
      Function<QueueMetrics, Resource> func,
      String resourceName,
      long expectedValue) {
    Resource res = func.apply(metrics);
    Long value = res.getResourceValue(resourceName);
    assertCustomResourceValueInternal(metricsType, resourceName,
        expectedValue, value);
  }

  private static void assertCustomResourceValueInternal(
      MetricsForCustomResource metricsType, String resourceName, long
      expectedValue, Long value) {
    assertNotNull(
            "QueueMetrics should have custom resource metrics value " +
                "for resource: " + resourceName, value);
    assertEquals(String.format(
            "QueueMetrics should have custom resource metrics value %d " +
                "for resource: %s for metrics type %s",
            expectedValue, resourceName, metricsType), expectedValue,
            (long) value);
  }

  private static Map<String, String> getCustomResourcesWithValue(long value) {
    return ImmutableMap.<String, String>builder()
        .put(CUSTOM_RES_1, String.valueOf(value))
        .put(CUSTOM_RES_2, String.valueOf(value))
        .build();
  }

  private QueueInfo createFourLevelQueueHierarchy() {
    QueueInfo root = new QueueInfo(null, "root", ms, CONF, USER);
    QueueInfo sub = new QueueInfo(root, "root.subQ", ms, CONF, USER);
    QueueInfo sub2 = new QueueInfo(sub, "root.subQ2", ms, CONF, USER);
    return new QueueInfo(sub2, "root.subQ2.leafQ", ms, CONF, USER);
  }

  private QueueInfo createBasicQueueHierarchy() {
    QueueInfo root = new QueueInfo(null, "root", ms, CONF, USER);
    return new QueueInfo(root, "root.leaf", ms, CONF, USER);
  }

  private QueueMetricsTestData.Builder
      createQueueMetricsTestDataWithContainers(int containers) {
    return createDefaultQueueMetricsTestData()
        .withContainers(containers);
  }

  private QueueMetricsTestData.Builder createDefaultQueueMetricsTestData() {
    return QueueMetricsTestData.Builder.create()
        .withUser(USER)
        .withPartition(RMNodeLabelsManager.NO_LABEL);
  }

  private void testIncreasePendingResources(QueueMetricsTestData testData) {
    testIncreasePendingResourcesInternal(testData.containers, testData);
  }

  private void testIncreasePendingResourcesWithoutContainer(
      QueueMetricsTestData testData) {
    testIncreasePendingResourcesInternal(1, testData);
  }

  private void testIncreasePendingResourcesInternal(int containers,
      QueueMetricsTestData testData) {
    testData.leafQueue.queueMetrics.incrPendingResources(testData.partition,
        testData.user, containers, testData.resource);

    ResourceMetricsChecker checker = ResourceMetricsChecker
        .create()
        .gaugeInt(PENDING_CONTAINERS, containers)
        .gaugeLong(PENDING_MB, containers *
            testData.resource.getMemorySize())
        .gaugeInt(PENDING_V_CORES, containers *
              testData.resource.getVirtualCores())
        .gaugeLong(PENDING_CUSTOM_RES1,
            containers * testData.customResourceValues.get(CUSTOM_RES_1))
        .gaugeLong(PENDING_CUSTOM_RES2,
            containers * testData.customResourceValues.get(CUSTOM_RES_2));
    assertAllMetrics(testData.leafQueue, checker,
        QueueMetrics::getPendingResources,
        MetricsForCustomResource.PENDING, computeExpectedCustomResourceValues(
            testData.customResourceValues, (k, v) -> v * containers));
  }

  private void testAllocateResources(boolean decreasePending,
      QueueMetricsTestData testData) {
    testData.leafQueue.queueMetrics.allocateResources(testData.partition,
        testData.user, testData.containers, testData.resource, decreasePending);

    ResourceMetricsChecker checker = ResourceMetricsChecker
        .create()
        .gaugeInt(ALLOCATED_CONTAINERS, testData.containers)
        .counter(AGGREGATE_CONTAINERS_ALLOCATED, testData.containers)
        .gaugeLong(ALLOCATED_MB, testData.containers *
            testData.resource.getMemorySize())
        .gaugeInt(ALLOCATED_V_CORES, testData.containers *
            testData.resource.getVirtualCores())
        .gaugeInt(PENDING_CONTAINERS, 0)
        .gaugeLong(PENDING_MB, 0)
        .gaugeInt(PENDING_V_CORES, 0)
        .gaugeLong(ALLOCATED_CUSTOM_RES1,
            testData.containers
                * testData.customResourceValues.get(CUSTOM_RES_1))
        .gaugeLong(ALLOCATED_CUSTOM_RES2,
            testData.containers
                * testData.customResourceValues.get(CUSTOM_RES_2))
        .checkAgainst(testData.leafQueue.queueSource);
    if (decreasePending) {
      assertAllMetrics(testData.leafQueue, checker,
          QueueMetrics::getPendingResources,
          MetricsForCustomResource.PENDING,
          computeExpectedCustomResourceValues(testData.customResourceValues,
              (k, v) -> 0L));
    }
    if (!testData.customResourceValues.isEmpty()) {
      assertAllMetrics(testData.leafQueue, checker,
          QueueMetrics::getAllocatedResources,
          MetricsForCustomResource.ALLOCATED,
          computeExpectedCustomResourceValues(testData.customResourceValues,
              (k, v) -> v * testData.containers));
    }
  }

  private void testUpdatePreemptedSeconds(QueueMetricsTestData testData,
      int seconds) {
    testData.leafQueue.queueMetrics.updatePreemptedMemoryMBSeconds(
        testData.resource.getMemorySize() * seconds);
    testData.leafQueue.queueMetrics.updatePreemptedVcoreSeconds(
        testData.resource.getVirtualCores() * seconds);
    testData.leafQueue.queueMetrics.updatePreemptedSecondsForCustomResources(
        testData.resource, seconds);

    ResourceMetricsChecker checker = ResourceMetricsChecker
        .create()
        .counter(AGGREGATE_MEMORY_MB_SECONDS_PREEMPTED,
            testData.resource.getMemorySize() * seconds)
        .counter(AGGREGATE_VCORE_SECONDS_PREEMPTED,
            testData.resource.getVirtualCores() * seconds)
        .gaugeLong(AGGREGATE_PREEMPTED_SECONDS_CUSTOM_RES1,
            testData.customResourceValues.get(CUSTOM_RES_1) * seconds)
        .gaugeLong(AGGREGATE_PREEMPTED_SECONDS_CUSTOM_RES2,
            testData.customResourceValues.get(CUSTOM_RES_2) * seconds);

    assertQueueMetricsOnly(testData.leafQueue, checker,
        this::convertPreemptedSecondsToResource,
        MetricsForCustomResource.AGGREGATE_PREEMPTED_SECONDS,
        computeExpectedCustomResourceValues(testData.customResourceValues,
            (k, v) -> v * seconds));
  }

  private Resource convertPreemptedSecondsToResource(QueueMetrics qm) {
    CustomResourceMetricValue customValues = qm
        .getAggregatedPreemptedSecondsResources();
    MutableCounterLong vcoreSeconds = qm
        .getAggregateVcoreSecondsPreempted();
    MutableCounterLong memorySeconds = qm
        .getAggregateMemoryMBSecondsPreempted();
    return Resource.newInstance(
        memorySeconds.value(), (int) vcoreSeconds.value(),
        customValues.getValues());
  }

  private void testReserveResources(QueueMetricsTestData testData) {
    testData.leafQueue.queueMetrics.reserveResource(testData.partition,
        testData.user, testData.resource);

    ResourceMetricsChecker checker = ResourceMetricsChecker
        .create()
        .gaugeInt(RESERVED_CONTAINERS, 1)
        .gaugeLong(RESERVED_MB, testData.resource.getMemorySize())
        .gaugeInt(RESERVED_V_CORES, testData.resource.getVirtualCores())
        .gaugeLong(RESERVED_CUSTOM_RES1,
            testData.customResourceValues.get(CUSTOM_RES_1))
        .gaugeLong(RESERVED_CUSTOM_RES2,
            testData.customResourceValues.get(CUSTOM_RES_2))
        .checkAgainst(testData.leafQueue.queueSource);
    assertAllMetrics(testData.leafQueue, checker,
        QueueMetrics::getReservedResources,
        MetricsForCustomResource.RESERVED,
        computeExpectedCustomResourceValues(
            testData.customResourceValues, (k, v) -> v));
  }

  private void testGetAllocatedResources(QueueMetricsTestData testData) {
    testAllocateResources(false, testData);

    Resource res = testData.leafQueue.queueMetrics.getAllocatedResources();
    if (testData.customResourceValues.size() > 0) {
      assertCustomResourceValueInternal(MetricsForCustomResource.ALLOCATED,
          CUSTOM_RES_1,
          testData.customResourceValues.get(CUSTOM_RES_1) * testData.containers,
          res.getResourceValue(CUSTOM_RES_1));
      assertCustomResourceValueInternal(MetricsForCustomResource.ALLOCATED,
          CUSTOM_RES_2,
          testData.customResourceValues.get(CUSTOM_RES_2) * testData.containers,
          res.getResourceValue(CUSTOM_RES_2));
    }
  }

  private void assertAllMetrics(QueueInfo queueInfo,
      ResourceMetricsChecker checker,
      Function<QueueMetrics, Resource> func,
      MetricsForCustomResource metricsType,
      Map<String, Long> expectedCustomResourceValues) {
    assertAllQueueMetrics(queueInfo, checker, func, metricsType,
        expectedCustomResourceValues);

    //assert leaf and root userSources
    checker = ResourceMetricsChecker.createFromChecker(checker)
        .checkAgainst(queueInfo.userSource);
    ResourceMetricsChecker.createFromChecker(checker)
        .checkAgainst(queueInfo.getRoot().userSource);
  }

  private void assertQueueMetricsOnly(QueueInfo queueInfo,
      ResourceMetricsChecker checker,
      Function<QueueMetrics, Resource> func,
      MetricsForCustomResource metricsType,
      Map<String, Long> expectedCustomResourceValues) {
    assertAllQueueMetrics(queueInfo, checker, func, metricsType,
        expectedCustomResourceValues);
  }

  private void assertAllQueueMetrics(QueueInfo queueInfo,
      ResourceMetricsChecker checker,
      Function<QueueMetrics, Resource> func,
      MetricsForCustomResource metricsType,
      Map<String, Long> expectedCustomResourceValues) {
    // assert normal resource metrics values
    queueInfo.checkAllQueueSources(qs -> ResourceMetricsChecker
        .createFromChecker(checker).checkAgainst(qs));

    // assert custom resource metrics values
    queueInfo.checkAllQueueMetrics(qm -> {
      assertCustomResourceValue(qm, metricsType, func, CUSTOM_RES_1,
          expectedCustomResourceValues.get(CUSTOM_RES_1));
      assertCustomResourceValue(qm, metricsType, func, CUSTOM_RES_2,
          expectedCustomResourceValues.get(CUSTOM_RES_2));
    });
  }

  private Map<String, Long> computeExpectedCustomResourceValues(
      Map<String, Long> customResourceValues,
      BiFunction<String, Long, Long> func) {
    Map<String, Long> values = Maps.newHashMap();
    for (Map.Entry<String, Long> res : customResourceValues.entrySet()) {
      values.put(res.getKey(), func.apply(res.getKey(), res.getValue()));
    }
    return values;
  }

  @Test
  public void testSetAvailableResourcesToQueue1() {
    String queueName = "single";
    QueueMetrics metrics = QueueMetrics.forQueue(ms, queueName, null,
        false, CONF);
    MetricsSource queueSource = queueSource(ms, queueName);

    metrics.setAvailableResourcesToQueue(newResource(
        GB, 4,
        ImmutableMap.<String, String> builder()
            .put(CUSTOM_RES_1, String.valueOf(5 * GB))
            .put(CUSTOM_RES_2, String.valueOf(6 * GB))
            .build()));
    ResourceMetricsChecker.create()
        .gaugeLong(AVAILABLE_MB, GB)
        .gaugeInt(AVAILABLE_V_CORES, 4)
        .gaugeLong(AVAILABLE_CUSTOM_RES1, 5 * GB)
        .gaugeLong(AVAILABLE_CUSTOM_RES2, 6 * GB)
        .checkAgainst(queueSource);

    assertCustomResourceValue(metrics,
        MetricsForCustomResource.AVAILABLE,
        QueueMetrics::getAvailableResources, CUSTOM_RES_1, 5 * GB);
    assertCustomResourceValue(metrics,
        MetricsForCustomResource.AVAILABLE,
        QueueMetrics::getAvailableResources, CUSTOM_RES_2, 6 * GB);
  }

  @Test
  public void testSetAvailableResourcesToQueue2() {
    String queueName = "single";
    QueueMetrics metrics = QueueMetrics.forQueue(ms, queueName, null,
        false, CONF);
    MetricsSource queueSource = queueSource(ms, queueName);

    metrics.setAvailableResourcesToQueue(null,
        newResource(GB, 4,
        ImmutableMap.<String, String> builder()
            .put(CUSTOM_RES_1, String.valueOf(15 * GB))
            .put(CUSTOM_RES_2, String.valueOf(20 * GB))
            .build()));
    ResourceMetricsChecker.create()
        .gaugeLong(AVAILABLE_MB, GB)
        .gaugeInt(AVAILABLE_V_CORES, 4)
        .gaugeLong(AVAILABLE_CUSTOM_RES1, 15 * GB)
        .gaugeLong(AVAILABLE_CUSTOM_RES2, 20 * GB)
        .checkAgainst(queueSource);

    assertCustomResourceValue(metrics,
        MetricsForCustomResource.AVAILABLE,
        QueueMetrics::getAvailableResources, CUSTOM_RES_1, 15 * GB);
    assertCustomResourceValue(metrics,
        MetricsForCustomResource.AVAILABLE,
        QueueMetrics::getAvailableResources, CUSTOM_RES_2, 20 * GB);
  }

  @Test
  public void testIncreasePendingResources() {
    QueueMetricsTestData testData = createQueueMetricsTestDataWithContainers(5)
        .withLeafQueue(createBasicQueueHierarchy())
        .withResourceToDecrease(
            newResource(GB, 2, getCustomResourcesWithValue(2 * GB)), 2)
        .withResources(defaultResource)
        .build();

    testIncreasePendingResources(testData);
  }

  @Test
  public void testDecreasePendingResources() {
    Resource resourceToDecrease =
        newResource(GB, 2, getCustomResourcesWithValue(2 * GB));
    int containersToDecrease = 2;
    int containers = 5;
    QueueMetricsTestData testData =
        createQueueMetricsTestDataWithContainers(containers)
            .withLeafQueue(createBasicQueueHierarchy())
        .withResourceToDecrease(resourceToDecrease, containers)
        .withResources(defaultResource)
        .build();

    //compute expected values
    final int vCoresToDecrease = resourceToDecrease.getVirtualCores();
    final long memoryMBToDecrease = resourceToDecrease.getMemorySize();
    final int containersAfterDecrease = containers - containersToDecrease;
    final long customRes1ToDecrease =
      resourceToDecrease.getResourceValue(CUSTOM_RES_1);
    final long customRes2ToDecrease =
      resourceToDecrease.getResourceValue(CUSTOM_RES_2);

    final int vcoresAfterDecrease =
        (defaultResource.getVirtualCores() * containers)
            - (vCoresToDecrease * containersToDecrease);
    final long memoryAfterDecrease =
        (defaultResource.getMemorySize() * containers)
            - (memoryMBToDecrease * containersToDecrease);
    final long customResource1AfterDecrease =
      (testData.customResourceValues.get(CUSTOM_RES_1) * containers)
          - (customRes1ToDecrease * containersToDecrease);
    final long customResource2AfterDecrease =
      (testData.customResourceValues.get(CUSTOM_RES_2) * containers)
          - (customRes2ToDecrease * containersToDecrease);

    //first, increase resources to be able to decrease some
    testIncreasePendingResources(testData);

    //decrease resources
    testData.leafQueue.queueMetrics.decrPendingResources(testData.partition,
        testData.user, containersToDecrease,
        ResourceTypesTestHelper.newResource(memoryMBToDecrease,
            vCoresToDecrease,
            extractCustomResourcesAsStrings(resourceToDecrease)));

    //check
    ResourceMetricsChecker checker = ResourceMetricsChecker
        .create()
        .gaugeInt(PENDING_CONTAINERS, containersAfterDecrease)
        .gaugeLong(PENDING_MB, memoryAfterDecrease)
        .gaugeInt(PENDING_V_CORES, vcoresAfterDecrease)
        .gaugeLong(PENDING_CUSTOM_RES1, customResource1AfterDecrease)
        .gaugeLong(PENDING_CUSTOM_RES2, customResource2AfterDecrease)
        .checkAgainst(testData.leafQueue.queueSource);

    assertAllMetrics(testData.leafQueue, checker,
        QueueMetrics::getPendingResources,
        MetricsForCustomResource.PENDING,
        computeExpectedCustomResourceValues(testData.customResourceValues,
            (k, v) -> v * containers - (resourceToDecrease.getResourceValue(k)
                * containersToDecrease)));
  }

  @Test
  public void testAllocateResourcesWithoutDecreasePending() {
    QueueMetricsTestData testData = createQueueMetricsTestDataWithContainers(5)
        .withLeafQueue(createBasicQueueHierarchy())
        .withResources(defaultResource)
        .build();

    testAllocateResources(false, testData);
  }

  @Test
  public void testAllocateResourcesWithDecreasePending() {
    QueueMetricsTestData testData = createQueueMetricsTestDataWithContainers(5)
        .withLeafQueue(createBasicQueueHierarchy())
        .withResourceToDecrease(
            newResource(GB, 2, getCustomResourcesWithValue(2 * GB)), 2)
        .withResources(defaultResource)
        .build();

    //first, increase pending resources to be able to decrease some
    testIncreasePendingResources(testData);

    //then allocate with decrease pending resources
    testAllocateResources(true, testData);
  }

  @Test
  public void testAllocateResourcesWithoutContainer() {
    QueueMetricsTestData testData = createDefaultQueueMetricsTestData()
        .withLeafQueue(createBasicQueueHierarchy())
        .withResources(defaultResource)
        .build();

    //first, increase pending resources
    testIncreasePendingResourcesWithoutContainer(testData);

    Resource resource = testData.resource;
    testData.leafQueue.queueMetrics.allocateResources(testData.partition,
        testData.user, resource);

    ResourceMetricsChecker checker = ResourceMetricsChecker.create()
        .gaugeLong(ALLOCATED_MB, resource.getMemorySize())
        .gaugeInt(ALLOCATED_V_CORES, resource.getVirtualCores())
        .gaugeInt(PENDING_CONTAINERS, 1).gaugeLong(PENDING_MB, 0)
        .gaugeInt(PENDING_V_CORES, 0)
        .gaugeLong(ALLOCATED_CUSTOM_RES1,
            testData.customResourceValues.get(CUSTOM_RES_1))
        .gaugeLong(ALLOCATED_CUSTOM_RES2,
            testData.customResourceValues.get(CUSTOM_RES_2));

    checker.checkAgainst(testData.leafQueue.queueSource);
    checker.checkAgainst(testData.leafQueue.getRoot().queueSource);

    assertAllMetrics(testData.leafQueue, checker,
        QueueMetrics::getPendingResources,
        MetricsForCustomResource.PENDING, computeExpectedCustomResourceValues(
            testData.customResourceValues, (k, v) -> 0L));
    assertAllMetrics(testData.leafQueue, checker,
        QueueMetrics::getAllocatedResources,
        MetricsForCustomResource.ALLOCATED, computeExpectedCustomResourceValues(
            testData.customResourceValues, (k, v) -> v));
  }

  @Test
  public void testReleaseResources() {
    int containers = 5;
    QueueMetricsTestData testData =
        createQueueMetricsTestDataWithContainers(containers)
            .withLeafQueue(createBasicQueueHierarchy())
        .withResourceToDecrease(defaultResource, containers)
        .withResources(defaultResource)
        .build();

    //first, allocate some resources so that we can release some
    testAllocateResources(false, testData);

    testData.leafQueue.queueMetrics.releaseResources(testData.partition,
        testData.user, containers, defaultResource);

    ResourceMetricsChecker checker = ResourceMetricsChecker
        .create()
        .counter(AGGREGATE_CONTAINERS_ALLOCATED, containers)
        .counter(AGGREGATE_CONTAINERS_RELEASED, containers)
        .checkAgainst(testData.leafQueue.queueSource);
    assertAllMetrics(testData.leafQueue, checker,
        QueueMetrics::getAllocatedResources,
        MetricsForCustomResource.ALLOCATED, computeExpectedCustomResourceValues(
            testData.customResourceValues, (k, v) -> 0L));
  }

  @Test
  public void testUpdatePreemptedSecondsForCustomResources() {
    QueueMetricsTestData testData = createQueueMetricsTestDataWithContainers(5)
        .withLeafQueue(createFourLevelQueueHierarchy())
        .withResources(defaultResource)
        .build();

    final int seconds = 1;
    testUpdatePreemptedSeconds(testData, seconds);
  }

  @Test
  public void testUpdatePreemptedSecondsForCustomResourcesMoreSeconds() {
    QueueMetricsTestData testData = createQueueMetricsTestDataWithContainers(5)
        .withLeafQueue(createFourLevelQueueHierarchy())
        .withResources(defaultResource)
        .build();

    final int seconds = 15;
    testUpdatePreemptedSeconds(testData, seconds);
  }

  @Test
  public void testReserveResources() {
    QueueMetricsTestData testData = createQueueMetricsTestDataWithContainers(5)
        .withLeafQueue(createBasicQueueHierarchy())
        .withResources(defaultResource)
        .build();

    testReserveResources(testData);
  }

  @Test
  public void testUnreserveResources() {
    QueueMetricsTestData testData = createQueueMetricsTestDataWithContainers(5)
        .withLeafQueue(createBasicQueueHierarchy())
        .withResources(defaultResource)
        .build();

    testReserveResources(testData);

    testData.leafQueue.queueMetrics.unreserveResource(testData.partition,
        testData.user, defaultResource);

    ResourceMetricsChecker checker = ResourceMetricsChecker
        .create()
        .gaugeInt(RESERVED_CONTAINERS, 0)
        .gaugeLong(RESERVED_MB, 0)
        .gaugeInt(RESERVED_V_CORES, 0)
        .gaugeLong(RESERVED_CUSTOM_RES1, 0).gaugeLong(RESERVED_CUSTOM_RES2, 0)
        .checkAgainst(testData.leafQueue.queueSource);
    assertAllMetrics(testData.leafQueue, checker,
        QueueMetrics::getReservedResources,
        MetricsForCustomResource.RESERVED, computeExpectedCustomResourceValues(
            testData.customResourceValues, (k, v) -> 0L));
  }

  @Test
  public void testGetAllocatedResourcesWithCustomResources() {
    QueueMetricsTestData testData = createQueueMetricsTestDataWithContainers(5)
        .withLeafQueue(createBasicQueueHierarchy())
        .withResources(defaultResource)
        .build();

    testGetAllocatedResources(testData);
  }

  @Test
  public void testGetAllocatedResourcesWithoutCustomResources() {
    QueueMetricsTestData testData = createQueueMetricsTestDataWithContainers(5)
        .withResources(newResource(4 * GB, 4, Collections.emptyMap()))
        .withLeafQueue(createBasicQueueHierarchy())
        .build();

    testGetAllocatedResources(testData);
  }

}
