alluxio-master-2.x/core/client/fs/src/test/java/alluxio/client/block/policy/LocalFirstPolicyTest.java
/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.client.block.policy;

import static alluxio.client.util.ClientTestUtils.worker;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import alluxio.Constants;
import alluxio.client.block.BlockWorkerInfo;
import alluxio.client.block.policy.options.GetWorkerOptions;
import alluxio.conf.AlluxioConfiguration;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.network.TieredIdentityFactory;
import alluxio.util.network.NetworkAddressUtils;
import alluxio.wire.BlockInfo;
import alluxio.wire.WorkerNetAddress;

import com.google.common.testing.EqualsTester;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;

/**
 * Tests {@link LocalFirstPolicy}.
 */
public final class LocalFirstPolicyTest {

  private static final AlluxioConfiguration S_CONF = Configuration.global();
  private static final int S_RESOLUTION_TIMEOUT =
      (int) Configuration.getMs(PropertyKey.NETWORK_HOST_RESOLUTION_TIMEOUT_MS);

  /**
   * Tests that the local host is returned first.
   */
  @Test
  public void getLocalFirst() {
    String localhostName =
        NetworkAddressUtils.getLocalHostName(S_RESOLUTION_TIMEOUT);
    LocalFirstPolicy policy = new LocalFirstPolicy(S_CONF);
    List<BlockWorkerInfo> workers = new ArrayList<>();
    workers.add(worker(Constants.GB, "worker1", ""));
    workers.add(worker(Constants.GB, localhostName, ""));
    GetWorkerOptions options = GetWorkerOptions.defaults()
        .setBlockWorkerInfos(workers).setBlockInfo(new BlockInfo().setLength(Constants.MB));
    assertEquals(localhostName, policy.getWorker(options)
        .orElseThrow(() -> new IllegalStateException("Expected worker")).getHost());
  }

  /**
   * Tests that another worker is picked in case the local host does not have enough capacity.
   */
  @Test
  public void getOthersWhenNotEnoughCapacityOnLocal() {
    String localhostName = NetworkAddressUtils.getLocalHostName(S_RESOLUTION_TIMEOUT);
    LocalFirstPolicy policy = new LocalFirstPolicy(S_CONF);
    List<BlockWorkerInfo> workers = new ArrayList<>();
    workers.add(worker(Constants.GB, "worker1", ""));
    workers.add(worker(Constants.MB, localhostName, ""));
    GetWorkerOptions options = GetWorkerOptions.defaults()
        .setBlockWorkerInfos(workers).setBlockInfo(new BlockInfo().setLength(Constants.GB));
    assertEquals("worker1", policy.getWorker(options)
        .orElseThrow(() -> new IllegalStateException("Expected worker")).getHost());
  }

  /**
   * Tests that non-local workers are randomly selected.
   */
  @Test
  public void getOthersRandomly() {
    LocalFirstPolicy policy = new LocalFirstPolicy(S_CONF);
    List<BlockWorkerInfo> workers = new ArrayList<>();
    workers.add(worker(Constants.GB, "worker1", ""));
    workers.add(worker(Constants.GB, "worker2", ""));

    boolean success = false;
    GetWorkerOptions options = GetWorkerOptions.defaults()
        .setBlockWorkerInfos(workers).setBlockInfo(new BlockInfo().setLength(Constants.MB));
    for (int i = 0; i < 100; i++) {
      String host = policy.getWorker(options)
          .orElseThrow(() -> new IllegalStateException("Expected worker")).getHost();
      if (!host.equals(policy.getWorker(options)
          .orElseThrow(() -> new IllegalStateException("Expected worker")).getHost())) {
        success = true;
        break;
      }
    }
    assertTrue(success);
  }

  @Test
  public void chooseClosestTier() throws Exception {
    List<BlockWorkerInfo> workers = new ArrayList<>();
    workers.add(worker(Constants.GB, "node2", "rack3"));
    workers.add(worker(Constants.GB, "node3", "rack2"));
    workers.add(worker(Constants.GB, "node4", "rack3"));
    LocalFirstPolicy policy;
    WorkerNetAddress chosen;
    // local rack
    policy = new LocalFirstPolicy(TieredIdentityFactory.fromString("node=node1,rack=rack2",
        S_CONF), S_CONF);
    GetWorkerOptions options = GetWorkerOptions.defaults()
        .setBlockWorkerInfos(workers).setBlockInfo(new BlockInfo().setLength(Constants.GB));
    chosen = policy.getWorker(options)
        .orElseThrow(() -> new IllegalStateException("Expected worker"));
    assertEquals("rack2", chosen.getTieredIdentity().getTier(1).getValue());

    // local node
    policy = new LocalFirstPolicy(TieredIdentityFactory.fromString("node=node4,rack=rack3",
        S_CONF),
        S_CONF);
    chosen = policy.getWorker(options)
        .orElseThrow(() -> new IllegalStateException("Expected worker"));
    assertEquals("node4", chosen.getTieredIdentity().getTier(0).getValue());
  }

  @Test
  public void tieredLocalityEnoughSpace() throws Exception {
    List<BlockWorkerInfo> workers = new ArrayList<>();
    // Local node doesn't have enough space
    workers.add(worker(Constants.MB, "node2", "rack3"));
    workers.add(worker(Constants.GB, "node3", "rack2"));
    // Local rack has enough space
    workers.add(worker(Constants.GB, "node4", "rack3"));
    LocalFirstPolicy policy = new LocalFirstPolicy(TieredIdentityFactory
        .fromString("node=node2,rack=rack3", S_CONF), S_CONF);
    GetWorkerOptions options = GetWorkerOptions.defaults()
        .setBlockWorkerInfos(workers).setBlockInfo(new BlockInfo().setLength(Constants.GB));
    WorkerNetAddress chosen = policy.getWorker(options)
        .orElseThrow(() -> new IllegalStateException("Expected worker"));
    assertEquals(workers.get(2).getNetAddress(), chosen);
  }

  @Test
  public void equalsTest() throws Exception {
    new EqualsTester()
        .addEqualityGroup(new LocalFirstPolicy(TieredIdentityFactory.fromString("node=x,rack=y",
            S_CONF), S_CONF))
        .addEqualityGroup(new LocalFirstPolicy(TieredIdentityFactory.fromString("node=x,rack=z",
            S_CONF), S_CONF))
        .testEquals();
  }
}
