alluxio-master-2.x/core/server/worker/src/main/java/alluxio/worker/block/meta/DefaultStorageTier.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.worker.block.meta;

import alluxio.Constants;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.InvalidPathException;
import alluxio.util.CommonUtils;
import alluxio.util.FormatUtils;
import alluxio.util.OSUtils;
import alluxio.util.ShellUtils;
import alluxio.util.UnixMountInfo;
import alluxio.util.io.FileUtils;
import alluxio.util.io.PathUtils;

import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import javax.annotation.Nullable;
import javax.annotation.concurrent.NotThreadSafe;

/**
 * Represents a tier of storage, for example memory or SSD. It serves as a container of
 * {@link StorageDir} which actually contains metadata information about blocks stored and space
 * used/available.
 */
@NotThreadSafe
public final class DefaultStorageTier implements StorageTier {
  private static final Logger LOG = LoggerFactory.getLogger(DefaultStorageTier.class);

  /** Alias value of this tier in tiered storage. */
  private final String mTierAlias;
  /** Ordinal value of this tier in tiered storage, the highest level is 0. */
  private final int mTierOrdinal;
  private final HashMap<Integer, StorageDir> mDirs;
  /** The lost storage paths that are failed to initialize or lost. */
  private final List<String> mLostStorage;

  private DefaultStorageTier(String tierAlias, int tierOrdinal) {
    mTierAlias = tierAlias;
    mTierOrdinal = tierOrdinal;
    mDirs = new HashMap<>();
    mLostStorage = new ArrayList<>();
  }

  private void initStorageTier(boolean isMultiTier) {
    String tmpDir = Configuration.getString(PropertyKey.WORKER_DATA_TMP_FOLDER);
    PropertyKey tierDirPathConf =
        PropertyKey.Template.WORKER_TIERED_STORE_LEVEL_DIRS_PATH.format(mTierOrdinal);
    List<String> dirPaths = Configuration.getList(tierDirPathConf).stream()
        .map(path -> CommonUtils.getWorkerDataDirectory(path, Configuration.global()))
        .collect(ImmutableList.toImmutableList());

    PropertyKey tierDirCapacityConf =
        PropertyKey.Template.WORKER_TIERED_STORE_LEVEL_DIRS_QUOTA.format(mTierOrdinal);
    List<String> dirQuotas = Configuration.getList(tierDirCapacityConf);
    Preconditions.checkState(dirQuotas.size() > 0,
        "Tier capacity configuration should not be blank");

    PropertyKey tierDirMediumConf =
        PropertyKey.Template.WORKER_TIERED_STORE_LEVEL_DIRS_MEDIUMTYPE.format(mTierOrdinal);
    List<String> dirMedium = Configuration.getList(tierDirMediumConf);
    Preconditions.checkState(dirMedium.size() > 0,
        "Tier medium type configuration should not be blank");

    // Set reserved bytes on directories if tier aligning is enabled.
    long reservedBytes = 0;
    if (isMultiTier
        && Configuration.getBoolean(PropertyKey.WORKER_MANAGEMENT_TIER_ALIGN_ENABLED)) {
      reservedBytes =
          Configuration.getBytes(PropertyKey.WORKER_MANAGEMENT_TIER_ALIGN_RESERVED_BYTES);
    }

    for (int i = 0; i < dirPaths.size(); i++) {
      int index = i >= dirQuotas.size() ? dirQuotas.size() - 1 : i;
      int mediumTypeindex = i >= dirMedium.size() ? dirMedium.size() - 1 : i;
      long capacity = FormatUtils.parseSpaceSize(dirQuotas.get(index));
      try {
        StorageDir dir = DefaultStorageDir.newStorageDir(this, i, capacity, reservedBytes,
            dirPaths.get(i), dirMedium.get(mediumTypeindex));
        mDirs.put(i, dir);
      } catch (Exception e) {
        LOG.error("Unable to initialize storage directory at {}", dirPaths.get(i), e);
        mLostStorage.add(dirPaths.get(i));
        continue;
      }

      // Delete tmp directory.
      String tmpDirPath = PathUtils.concatPath(dirPaths.get(i), tmpDir);
      try {
        FileUtils.deletePathRecursively(tmpDirPath);
      } catch (IOException e) {
        if (FileUtils.exists(tmpDirPath)) {
          LOG.error("Failed to clean up temporary directory: {}.", tmpDirPath);
        }
      }
    }
    if (mTierAlias.equals(Constants.MEDIUM_MEM) && mDirs.size() == 1) {
      checkEnoughMemSpace(mDirs.values().iterator().next());
    }
  }

  /**
   * Checks that a tmpfs/ramfs backing the storage directory has enough capacity. If the storage
   * directory is not backed by tmpfs/ramfs or the size of the tmpfs/ramfs cannot be determined, a
   * warning is logged but no exception is thrown.
   *
   * @param storageDir the storage dir to check
   * @throws IllegalStateException if the tmpfs/ramfs is smaller than the configured memory size
   */
  private void checkEnoughMemSpace(StorageDir storageDir) {
    if (!OSUtils.isLinux()) {
      return;
    }
    List<UnixMountInfo> info;
    try {
      info = ShellUtils.getUnixMountInfo();
    } catch (IOException e) {
      LOG.warn("Failed to get mount information for verifying memory capacity: {}", e.toString());
      return;
    }
    boolean foundMountInfo = false;
    for (UnixMountInfo mountInfo : info) {
      Optional<String> mountPointOption = mountInfo.getMountPoint();
      Optional<String> fsTypeOption = mountInfo.getFsType();
      Optional<Long> sizeOption = mountInfo.getOptions().getSize();
      if (!mountPointOption.isPresent() || !fsTypeOption.isPresent() || !sizeOption.isPresent()) {
        continue;
      }
      String mountPoint = mountPointOption.get();
      String fsType = fsTypeOption.get();
      long size = sizeOption.get();
      try {
        // getDirPath gives something like "/mnt/tmpfs/alluxioworker".
        String rootStoragePath = PathUtils.getParent(storageDir.getDirPath());
        if (!PathUtils.cleanPath(mountPoint).equals(rootStoragePath)) {
          continue;
        }
      } catch (InvalidPathException e) {
        continue;
      }
      foundMountInfo = true;
      if (fsType.equalsIgnoreCase("tmpfs") && size < storageDir.getCapacityBytes()) {
        throw new IllegalStateException(String.format(
            "%s is smaller than the configured size: %s size: %s, configured size: %s",
            fsType, fsType, size, storageDir.getCapacityBytes()));
      }
      break;
    }
    if (!foundMountInfo) {
      LOG.warn("Failed to verify memory capacity");
    }
  }

  /**
   * Factory method to create {@link StorageTier}.
   *
   * @param tierAlias the tier alias
   * @param tierOrdinal the tier ordinal
   * @param isMultiTier whether this tier is part of a multi-tier setup
   * @return a new storage tier
   */
  public static StorageTier newStorageTier(String tierAlias, int tierOrdinal, boolean isMultiTier) {
    DefaultStorageTier ret = new DefaultStorageTier(tierAlias, tierOrdinal);
    ret.initStorageTier(isMultiTier);
    return ret;
  }

  @Override
  public int getTierOrdinal() {
    return mTierOrdinal;
  }

  @Override
  public String getTierAlias() {
    return mTierAlias;
  }

  @Override
  public long getCapacityBytes() {
    long totalCapacity = 0;
    for (StorageDir dir : mDirs.values()) {
      totalCapacity += dir.getCapacityBytes();
    }
    return totalCapacity;
  }

  @Override
  public long getAvailableBytes() {
    long availableBytes = 0;
    for (StorageDir dir : mDirs.values()) {
      availableBytes += dir.getAvailableBytes();
    }
    return availableBytes;
  }

  @Override
  @Nullable
  public StorageDir getDir(int dirIndex) {
    return mDirs.get(dirIndex);
  }

  @Override
  public List<StorageDir> getStorageDirs() {
    return new ArrayList<>(mDirs.values());
  }

  @Override
  public List<String> getLostStorage() {
    return new ArrayList<>(mLostStorage);
  }

  /**
   * @throws IllegalArgumentException if dir does not belong to this tier
   */
  @Override
  public void removeStorageDir(StorageDir dir) {
    Preconditions.checkArgument(this == dir.getParentTier(),
        "storage dir %s does not belong to tier %s", dir.getDirPath(), getTierAlias());
    mDirs.remove(dir.getDirIndex());
    mLostStorage.add(dir.getDirPath());
  }
}
