hadoop-tools/hadoop-resourceestimator/src/test/java/org/apache/hadoop/resourceestimator/skylinestore/impl/TestSkylineStore.java

/*
 *
 * 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.resourceestimator.skylinestore.impl;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import org.apache.hadoop.resourceestimator.common.api.RecurrenceId;
import org.apache.hadoop.resourceestimator.common.api.ResourceSkyline;
import org.apache.hadoop.resourceestimator.skylinestore.api.SkylineStore;
import org.apache.hadoop.resourceestimator.skylinestore.exceptions.DuplicateRecurrenceIdException;
import org.apache.hadoop.resourceestimator.skylinestore.exceptions.EmptyResourceSkylineException;
import org.apache.hadoop.resourceestimator.skylinestore.exceptions.NullPipelineIdException;
import org.apache.hadoop.resourceestimator.skylinestore.exceptions.NullRLESparseResourceAllocationException;
import org.apache.hadoop.resourceestimator.skylinestore.exceptions.NullRecurrenceIdException;
import org.apache.hadoop.resourceestimator.skylinestore.exceptions.NullResourceSkylineException;
import org.apache.hadoop.resourceestimator.skylinestore.exceptions.RecurrenceIdNotFoundException;
import org.apache.hadoop.resourceestimator.skylinestore.exceptions.SkylineStoreException;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.RLESparseResourceAllocation;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationInterval;
import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

/**
 * Test {@link SkylineStore} class.
 */
public abstract class TestSkylineStore {
  /**
   * Testing variables.
   */
  private SkylineStore skylineStore;

  private TreeMap<Long, Resource> resourceOverTime;
  private RLESparseResourceAllocation skylineList;
  private ReservationInterval riAdd;
  private Resource resource;

  protected abstract SkylineStore createSkylineStore();

  @Before public final void setup() {
    skylineStore = createSkylineStore();
    resourceOverTime = new TreeMap<>();
    resource = Resource.newInstance(1024 * 100, 100);
  }

  private void compare(final ResourceSkyline skyline1,
      final ResourceSkyline skyline2) {
    Assert.assertEquals(skyline1.getJobId(), skyline2.getJobId());
    Assert.assertEquals(skyline1.getJobInputDataSize(),
        skyline2.getJobInputDataSize(), 0);
    Assert.assertEquals(skyline1.getJobSubmissionTime(),
        skyline2.getJobSubmissionTime());
    Assert
        .assertEquals(skyline1.getJobFinishTime(), skyline2.getJobFinishTime());
    Assert.assertEquals(skyline1.getContainerSpec().getMemorySize(),
        skyline2.getContainerSpec().getMemorySize());
    Assert.assertEquals(skyline1.getContainerSpec().getVirtualCores(),
        skyline2.getContainerSpec().getVirtualCores());
    Assert.assertEquals(true,
        skyline2.getSkylineList().equals(skyline1.getSkylineList()));
  }

  private void addToStore(final RecurrenceId recurrenceId,
      final ResourceSkyline resourceSkyline) throws SkylineStoreException {
    final List<ResourceSkyline> resourceSkylines = new ArrayList<>();
    resourceSkylines.add(resourceSkyline);
    skylineStore.addHistory(recurrenceId, resourceSkylines);
    final List<ResourceSkyline> resourceSkylinesGet =
        skylineStore.getHistory(recurrenceId).get(recurrenceId);
    Assert.assertTrue(resourceSkylinesGet.contains(resourceSkyline));
  }

  private ResourceSkyline getSkyline(final int n) {
    skylineList = new RLESparseResourceAllocation(resourceOverTime,
        new DefaultResourceCalculator());
    for (int i = 0; i < n; i++) {
      riAdd = new ReservationInterval(i * 10, (i + 1) * 10);
      skylineList.addInterval(riAdd, resource);
    }
    final ResourceSkyline resourceSkyline =
        new ResourceSkyline(Integer.toString(n), 1024.5, 0, 20, resource,
            skylineList);

    return resourceSkyline;
  }

  @Test public final void testGetHistory() throws SkylineStoreException {
    // addHistory first recurring pipeline
    final RecurrenceId recurrenceId1 =
        new RecurrenceId("FraudDetection", "17/06/20 00:00:00");
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    addToStore(recurrenceId1, resourceSkyline1);
    final ResourceSkyline resourceSkyline2 = getSkyline(2);
    addToStore(recurrenceId1, resourceSkyline2);
    final RecurrenceId recurrenceId2 =
        new RecurrenceId("FraudDetection", "17/06/21 00:00:00");
    final ResourceSkyline resourceSkyline3 = getSkyline(3);
    addToStore(recurrenceId2, resourceSkyline3);
    final ResourceSkyline resourceSkyline4 = getSkyline(4);
    addToStore(recurrenceId2, resourceSkyline4);
    // addHistory second recurring pipeline
    final RecurrenceId recurrenceId3 =
        new RecurrenceId("Random", "17/06/20 00:00:00");
    addToStore(recurrenceId3, resourceSkyline1);
    addToStore(recurrenceId3, resourceSkyline2);
    // test getHistory {pipelineId, runId}
    Map<RecurrenceId, List<ResourceSkyline>> jobHistory =
        skylineStore.getHistory(recurrenceId1);
    Assert.assertEquals(1, jobHistory.size());
    for (final Map.Entry<RecurrenceId, List<ResourceSkyline>> entry : jobHistory
        .entrySet()) {
      Assert.assertEquals(recurrenceId1, entry.getKey());
      final List<ResourceSkyline> getSkylines = entry.getValue();
      Assert.assertEquals(2, getSkylines.size());
      compare(resourceSkyline1, getSkylines.get(0));
      compare(resourceSkyline2, getSkylines.get(1));
    }
    // test getHistory {pipelineId, *}
    RecurrenceId recurrenceIdTest = new RecurrenceId("FraudDetection", "*");
    jobHistory = skylineStore.getHistory(recurrenceIdTest);
    Assert.assertEquals(2, jobHistory.size());
    for (final Map.Entry<RecurrenceId, List<ResourceSkyline>> entry : jobHistory
        .entrySet()) {
      Assert.assertEquals(recurrenceId1.getPipelineId(),
          entry.getKey().getPipelineId());
      final List<ResourceSkyline> getSkylines = entry.getValue();
      if (entry.getKey().getRunId().equals("17/06/20 00:00:00")) {
        Assert.assertEquals(2, getSkylines.size());
        compare(resourceSkyline1, getSkylines.get(0));
        compare(resourceSkyline2, getSkylines.get(1));
      } else {
        Assert.assertEquals(entry.getKey().getRunId(), "17/06/21 00:00:00");
        Assert.assertEquals(2, getSkylines.size());
        compare(resourceSkyline3, getSkylines.get(0));
        compare(resourceSkyline4, getSkylines.get(1));
      }
    }
    // test getHistory {*, runId}
    recurrenceIdTest = new RecurrenceId("*", "some random runId");
    jobHistory = skylineStore.getHistory(recurrenceIdTest);
    Assert.assertEquals(3, jobHistory.size());
    for (final Map.Entry<RecurrenceId, List<ResourceSkyline>> entry : jobHistory
        .entrySet()) {
      if (entry.getKey().getPipelineId().equals("FraudDetection")) {
        final List<ResourceSkyline> getSkylines = entry.getValue();
        if (entry.getKey().getRunId().equals("17/06/20 00:00:00")) {
          Assert.assertEquals(2, getSkylines.size());
          compare(resourceSkyline1, getSkylines.get(0));
          compare(resourceSkyline2, getSkylines.get(1));
        } else {
          Assert.assertEquals(entry.getKey().getRunId(), "17/06/21 00:00:00");
          Assert.assertEquals(2, getSkylines.size());
          compare(resourceSkyline3, getSkylines.get(0));
          compare(resourceSkyline4, getSkylines.get(1));
        }
      } else {
        Assert.assertEquals("Random", entry.getKey().getPipelineId());
        Assert.assertEquals(entry.getKey().getRunId(), "17/06/20 00:00:00");
        final List<ResourceSkyline> getSkylines = entry.getValue();
        Assert.assertEquals(2, getSkylines.size());
        compare(resourceSkyline1, getSkylines.get(0));
        compare(resourceSkyline2, getSkylines.get(1));
      }
    }
    // test getHistory with wrong RecurrenceId
    recurrenceIdTest =
        new RecurrenceId("some random pipelineId", "some random runId");
    Assert.assertNull(skylineStore.getHistory(recurrenceIdTest));
  }

  @Test public final void testGetEstimation() throws SkylineStoreException {
    // first, add estimation to the skyline store
    final RLESparseResourceAllocation skylineList2 =
        new RLESparseResourceAllocation(resourceOverTime,
            new DefaultResourceCalculator());
    for (int i = 0; i < 5; i++) {
      riAdd = new ReservationInterval(i * 10, (i + 1) * 10);
      skylineList2.addInterval(riAdd, resource);
    }
    skylineStore.addEstimation("FraudDetection", skylineList2);
    // then, try to get the estimation
    final RLESparseResourceAllocation estimation =
        skylineStore.getEstimation("FraudDetection");
    for (int i = 0; i < 50; i++) {
      Assert.assertEquals(skylineList2.getCapacityAtTime(i),
          estimation.getCapacityAtTime(i));
    }
  }

  @Test(expected = NullRecurrenceIdException.class)
  public final void testGetNullRecurrenceId()
      throws SkylineStoreException {
    // addHistory first recurring pipeline
    final RecurrenceId recurrenceId1 =
        new RecurrenceId("FraudDetection", "17/06/20 00:00:00");
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    addToStore(recurrenceId1, resourceSkyline1);
    final ResourceSkyline resourceSkyline2 = getSkyline(2);
    addToStore(recurrenceId1, resourceSkyline2);
    final RecurrenceId recurrenceId2 =
        new RecurrenceId("FraudDetection", "17/06/21 00:00:00");
    final ResourceSkyline resourceSkyline3 = getSkyline(3);
    addToStore(recurrenceId2, resourceSkyline3);
    final ResourceSkyline resourceSkyline4 = getSkyline(4);
    addToStore(recurrenceId2, resourceSkyline4);
    // addHistory second recurring pipeline
    final RecurrenceId recurrenceId3 =
        new RecurrenceId("Random", "17/06/20 00:00:00");
    addToStore(recurrenceId3, resourceSkyline1);
    addToStore(recurrenceId3, resourceSkyline2);
    // try to getHistory with null recurringId
    skylineStore.getHistory(null);
  }

  @Test(expected = NullPipelineIdException.class)
  public final void testGetNullPipelineIdException()
      throws SkylineStoreException {
    skylineStore.getEstimation(null);
  }

  @Test public final void testAddNormal() throws SkylineStoreException {
    // addHistory resource skylines to the in-memory store
    final RecurrenceId recurrenceId =
        new RecurrenceId("FraudDetection", "17/06/20 00:00:00");
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    addToStore(recurrenceId, resourceSkyline1);
    final ArrayList<ResourceSkyline> resourceSkylines =
        new ArrayList<ResourceSkyline>();
    // the resource skylines to be added contain null
    resourceSkylines.add(null);
    final ResourceSkyline resourceSkyline2 = getSkyline(2);
    resourceSkylines.add(resourceSkyline2);
    skylineStore.addHistory(recurrenceId, resourceSkylines);
    // query the in-memory store
    final Map<RecurrenceId, List<ResourceSkyline>> jobHistory =
        skylineStore.getHistory(recurrenceId);
    Assert.assertEquals(1, jobHistory.size());
    for (final Map.Entry<RecurrenceId, List<ResourceSkyline>> entry : jobHistory
        .entrySet()) {
      Assert.assertEquals(recurrenceId, entry.getKey());
      final List<ResourceSkyline> getSkylines = entry.getValue();
      Assert.assertEquals(2, getSkylines.size());
      compare(resourceSkyline1, getSkylines.get(0));
      compare(resourceSkyline2, getSkylines.get(1));
    }
  }

  @Test(expected = NullRecurrenceIdException.class)
  public final void testAddNullRecurrenceId()
      throws SkylineStoreException {
    // recurrenceId is null
    final RecurrenceId recurrenceIdNull = null;
    final ArrayList<ResourceSkyline> resourceSkylines =
        new ArrayList<ResourceSkyline>();
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    resourceSkylines.add(resourceSkyline1);
    skylineStore.addHistory(recurrenceIdNull, resourceSkylines);
  }

  @Test(expected = NullResourceSkylineException.class)
  public final void testAddNullResourceSkyline()
      throws SkylineStoreException {
    final RecurrenceId recurrenceId =
        new RecurrenceId("FraudDetection", "17/06/20 00:00:00");
    final ArrayList<ResourceSkyline> resourceSkylines =
        new ArrayList<ResourceSkyline>();
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    resourceSkylines.add(resourceSkyline1);
    // resourceSkylines is null
    skylineStore.addHistory(recurrenceId, null);
  }

  @Test(expected = DuplicateRecurrenceIdException.class)
  public final void testAddDuplicateRecurrenceId()
      throws SkylineStoreException {
    final RecurrenceId recurrenceId =
        new RecurrenceId("FraudDetection", "17/06/20 00:00:00");
    final ArrayList<ResourceSkyline> resourceSkylines =
        new ArrayList<ResourceSkyline>();
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    resourceSkylines.add(resourceSkyline1);
    // trying to addHistory duplicate resource skylines
    skylineStore.addHistory(recurrenceId, resourceSkylines);
    skylineStore.addHistory(recurrenceId, resourceSkylines);
  }

  @Test(expected = NullPipelineIdException.class)
  public final void testAddNullPipelineIdException()
      throws SkylineStoreException {
    final RLESparseResourceAllocation skylineList2 =
        new RLESparseResourceAllocation(resourceOverTime,
            new DefaultResourceCalculator());
    for (int i = 0; i < 5; i++) {
      riAdd = new ReservationInterval(i * 10, (i + 1) * 10);
      skylineList2.addInterval(riAdd, resource);
    }
    skylineStore.addEstimation(null, skylineList2);
  }

  @Test(expected = NullRLESparseResourceAllocationException.class)
  public final void testAddNullRLESparseResourceAllocationExceptionException()
      throws SkylineStoreException {
    skylineStore.addEstimation("FraudDetection", null);
  }

  @Test public final void testDeleteNormal() throws SkylineStoreException {
    // addHistory first recurring pipeline
    final RecurrenceId recurrenceId1 =
        new RecurrenceId("FraudDetection", "17/06/20 00:00:00");
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    addToStore(recurrenceId1, resourceSkyline1);
    final ResourceSkyline resourceSkyline2 = getSkyline(2);
    addToStore(recurrenceId1, resourceSkyline2);
    // test deleteHistory function of the in-memory store
    skylineStore.deleteHistory(recurrenceId1);
  }

  @Test(expected = NullRecurrenceIdException.class)
  public final void testDeleteNullRecurrenceId()
      throws SkylineStoreException {
    final RecurrenceId recurrenceId1 =
        new RecurrenceId("FraudDetection", "17/06/20 00:00:00");
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    addToStore(recurrenceId1, resourceSkyline1);
    // try to deleteHistory with null recurringId
    skylineStore.deleteHistory(null);
  }

  @Test(expected = RecurrenceIdNotFoundException.class)
  public final void testDeleteRecurrenceIdNotFound()
      throws SkylineStoreException {
    final RecurrenceId recurrenceId1 =
        new RecurrenceId("FraudDetection", "17/06/20 00:00:00");
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    addToStore(recurrenceId1, resourceSkyline1);
    final RecurrenceId recurrenceIdInvalid =
        new RecurrenceId("Some random pipelineId", "Some random runId");
    // try to deleteHistory non-existing recurringId
    skylineStore.deleteHistory(recurrenceIdInvalid);
  }

  @Test public final void testUpdateNormal() throws SkylineStoreException {
    // addHistory first recurring pipeline
    final RecurrenceId recurrenceId1 =
        new RecurrenceId("FraudDetection", "17/06/20 00:00:00");
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    addToStore(recurrenceId1, resourceSkyline1);
    final ArrayList<ResourceSkyline> resourceSkylines =
        new ArrayList<ResourceSkyline>();
    final ResourceSkyline resourceSkyline2 = getSkyline(2);
    resourceSkylines.add(resourceSkyline1);
    resourceSkylines.add(resourceSkyline2);
    skylineStore.updateHistory(recurrenceId1, resourceSkylines);
    // query the in-memory store
    final Map<RecurrenceId, List<ResourceSkyline>> jobHistory =
        skylineStore.getHistory(recurrenceId1);
    Assert.assertEquals(1, jobHistory.size());
    for (final Map.Entry<RecurrenceId, List<ResourceSkyline>> entry : jobHistory
        .entrySet()) {
      Assert.assertEquals(recurrenceId1, entry.getKey());
      final List<ResourceSkyline> getSkylines = entry.getValue();
      Assert.assertEquals(2, getSkylines.size());
      compare(resourceSkyline1, getSkylines.get(0));
      compare(resourceSkyline2, getSkylines.get(1));
    }
  }

  @Test(expected = NullRecurrenceIdException.class)
  public final void testUpdateNullRecurrenceId()
      throws SkylineStoreException {
    final ArrayList<ResourceSkyline> resourceSkylines =
        new ArrayList<ResourceSkyline>();
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    resourceSkylines.add(resourceSkyline1);
    final ArrayList<ResourceSkyline> resourceSkylinesInvalid =
        new ArrayList<ResourceSkyline>();
    resourceSkylinesInvalid.add(null);
    // try to updateHistory with null recurringId
    skylineStore.updateHistory(null, resourceSkylines);
  }

  @Test(expected = NullResourceSkylineException.class)
  public final void testUpdateNullResourceSkyline()
      throws SkylineStoreException {
    final RecurrenceId recurrenceId =
        new RecurrenceId("FraudDetection", "17/06/20 00:00:00");
    final ArrayList<ResourceSkyline> resourceSkylines =
        new ArrayList<ResourceSkyline>();
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    resourceSkylines.add(resourceSkyline1);
    final ArrayList<ResourceSkyline> resourceSkylinesInvalid =
        new ArrayList<ResourceSkyline>();
    resourceSkylinesInvalid.add(null);
    // try to updateHistory with null resourceSkylines
    skylineStore.addHistory(recurrenceId, resourceSkylines);
    skylineStore.updateHistory(recurrenceId, null);
  }

  @Test(expected = EmptyResourceSkylineException.class)
  public final void testUpdateEmptyRecurrenceId()
      throws SkylineStoreException {
    final RecurrenceId recurrenceId =
        new RecurrenceId("FraudDetection", "17/06/20 00:00:00");
    final ArrayList<ResourceSkyline> resourceSkylines =
        new ArrayList<ResourceSkyline>();
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    resourceSkylines.add(resourceSkyline1);
    final ArrayList<ResourceSkyline> resourceSkylinesInvalid =
        new ArrayList<ResourceSkyline>();
    resourceSkylinesInvalid.add(null);
    skylineStore.addHistory(recurrenceId, resourceSkylines);
    // try to updateHistory with empty resourceSkyline
    skylineStore.updateHistory(recurrenceId, resourceSkylinesInvalid);
  }

  @Test(expected = RecurrenceIdNotFoundException.class)
  public final void testUpdateRecurrenceIdNotFound()
      throws SkylineStoreException {
    final ArrayList<ResourceSkyline> resourceSkylines =
        new ArrayList<ResourceSkyline>();
    final ResourceSkyline resourceSkyline1 = getSkyline(1);
    resourceSkylines.add(resourceSkyline1);
    final RecurrenceId recurrenceIdInvalid =
        new RecurrenceId("Some random pipelineId", "Some random runId");
    final ArrayList<ResourceSkyline> resourceSkylinesInvalid =
        new ArrayList<ResourceSkyline>();
    resourceSkylinesInvalid.add(null);
    // try to updateHistory with non-existing recurringId
    skylineStore.updateHistory(recurrenceIdInvalid, resourceSkylines);
  }

  @After public final void cleanUp() {
    skylineStore = null;
    resourceOverTime.clear();
    resourceOverTime = null;
    skylineList = null;
    riAdd = null;
    resource = null;
  }
}