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

import alluxio.Constants;
import alluxio.DefaultStorageTierAssoc;
import alluxio.StorageTierAssoc;
import alluxio.collections.Pair;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.worker.block.BlockStoreLocation;
import alluxio.worker.block.BlockStoreMeta;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Streams;

import java.util.List;
import java.util.Map;
import java.util.Optional;
import javax.annotation.Nullable;

/**
 * Metadata of the paged block store.
 */
public class PagedBlockStoreMeta implements BlockStoreMeta {
  // Paged block store does not support multiple tiers, so assume everything is in the top tier
  public static final String DEFAULT_TIER =
      Configuration.getString(PropertyKey.MASTER_TIERED_STORE_GLOBAL_LEVEL0_ALIAS);
  // top tier is conventionally ramdisk
  public static final String DEFAULT_MEDIUM = Constants.MEDIUM_MEM;
  public static final DefaultStorageTierAssoc DEFAULT_STORAGE_TIER_ASSOC =
      new DefaultStorageTierAssoc(ImmutableList.of(DEFAULT_MEDIUM));

  private final long mCapacity;
  private final long mUsed;
  private final int mNumBlocks;
  private final List<String> mDirPaths;
  private final List<Long> mCapacityBytesOnDirs;
  private final List<Long> mUsedBytesOnDirs;
  private final Optional<Map<BlockStoreLocation, List<Long>>> mBlockLocations;
  private final Optional<List<Long>> mBlocks;

  PagedBlockStoreMeta(List<String> storageDirPaths, List<Long> capacityBytesOnDirs,
      List<Long> usedBytesOnDirs, int numBlocks) {
    mDirPaths = storageDirPaths;
    mCapacityBytesOnDirs = capacityBytesOnDirs;
    mUsedBytesOnDirs = usedBytesOnDirs;
    mCapacity = capacityBytesOnDirs.stream().reduce(0L, Long::sum);
    mUsed = usedBytesOnDirs.stream().reduce(0L, Long::sum);
    mNumBlocks = numBlocks;
    mBlocks = Optional.empty();
    mBlockLocations = Optional.empty();
  }

  PagedBlockStoreMeta(List<String> storageDirPaths, List<Long> capacityBytesOnDirs,
      List<Long> usedBytesOnDirs, Map<BlockStoreLocation, List<Long>> blockLocations) {
    mDirPaths = storageDirPaths;
    mCapacityBytesOnDirs = capacityBytesOnDirs;
    mUsedBytesOnDirs = usedBytesOnDirs;
    mCapacity = capacityBytesOnDirs.stream().reduce(0L, Long::sum);
    mUsed = usedBytesOnDirs.stream().reduce(0L, Long::sum);

    mBlockLocations = Optional.of(blockLocations);
    ImmutableList.Builder<Long> blockIdsListBuilder = new ImmutableList.Builder<>();
    blockLocations.values().forEach(blockIdsListBuilder::addAll);
    List<Long> blockIds = blockIdsListBuilder.build();
    mNumBlocks = blockIds.size();
    mBlocks = Optional.of(blockIds);
  }

  @Nullable
  @Override
  public Map<String, List<Long>> getBlockList() {
    return mBlocks.map(blocks -> ImmutableMap.of(DEFAULT_TIER, blocks)).orElse(null);
  }

  @Nullable
  @Override
  public Map<BlockStoreLocation, List<Long>> getBlockListByStorageLocation() {
    return mBlockLocations.orElse(null);
  }

  @Override
  public long getCapacityBytes() {
    return mCapacity;
  }

  @Override
  public Map<String, Long> getCapacityBytesOnTiers() {
    return ImmutableMap.of(DEFAULT_TIER, mCapacity);
  }

  @Override
  public Map<Pair<String, String>, Long> getCapacityBytesOnDirs() {
    return Streams.zip(mDirPaths.stream(), mCapacityBytesOnDirs.stream(), Pair::new)
        .collect(ImmutableMap.toImmutableMap(
           pair -> new Pair<>(DEFAULT_TIER, pair.getFirst()),
           Pair::getSecond
        ));
  }

  @Override
  public Map<String, List<String>> getDirectoryPathsOnTiers() {
    return ImmutableMap.of(DEFAULT_TIER, mDirPaths);
  }

  @Override
  public Map<String, List<String>> getLostStorage() {
    return ImmutableMap.of();
  }

  @Override
  public int getNumberOfBlocks() {
    return mNumBlocks;
  }

  @Override
  public long getUsedBytes() {
    return mUsed;
  }

  @Override
  public Map<String, Long> getUsedBytesOnTiers() {
    return ImmutableMap.of(DEFAULT_TIER, mUsed);
  }

  @Override
  public Map<Pair<String, String>, Long> getUsedBytesOnDirs() {
    return Streams.zip(mDirPaths.stream(), mUsedBytesOnDirs.stream(), Pair::new)
        .collect(ImmutableMap.toImmutableMap(
            pair -> new Pair<>(DEFAULT_TIER, pair.getFirst()),
            Pair::getSecond
        ));
  }

  @Override
  public StorageTierAssoc getStorageTierAssoc() {
    return DEFAULT_STORAGE_TIER_ASSOC;
  }
}
