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

import static alluxio.worker.block.BlockMetadataManager.WORKER_STORAGE_TIER_ASSOC;

import alluxio.metrics.MetricKey;
import alluxio.metrics.MetricsSystem;

import com.codahale.metrics.Counter;
import com.codahale.metrics.Meter;

import javax.annotation.concurrent.ThreadSafe;

/**
 * This class listens on block events and increases the metrics counters.
 */
@ThreadSafe
public final class BlockMetricsReporter extends AbstractBlockStoreEventListener {
  private static final Counter BLOCKS_ACCESSED
      = MetricsSystem.counter(MetricKey.WORKER_BLOCKS_ACCESSED.getName());
  private static final Counter BLOCKS_PROMOTED
      = MetricsSystem.counter(MetricKey.WORKER_BLOCKS_PROMOTED.getName());
  private static final Counter BLOCKS_DELETED
      = MetricsSystem.counter(MetricKey.WORKER_BLOCKS_DELETED.getName());
  private static final Counter BLOCKS_EVICTED
      = MetricsSystem.counter(MetricKey.WORKER_BLOCKS_EVICTED.getName());
  private static final Counter BLOCKS_CANCELLED
      = MetricsSystem.counter(MetricKey.WORKER_BLOCKS_CANCELLED.getName());
  private static final Counter BLOCKS_LOST
      = MetricsSystem.counter(MetricKey.WORKER_BLOCKS_LOST.getName());

  private static final Meter BLOCKS_EVICTION_RATE =
      MetricsSystem.meterWithTags(MetricKey.WORKER_BLOCKS_EVICTION_RATE.getName(),
        MetricKey.WORKER_BLOCKS_EVICTION_RATE.isClusterAggregated());

  @Override
  public void onAccessBlock(long blockId) {
    BLOCKS_ACCESSED.inc();
  }

  @Override
  public void onMoveBlockByClient(long blockId, BlockStoreLocation oldLocation,
      BlockStoreLocation newLocation) {
    int oldTierOrdinal = WORKER_STORAGE_TIER_ASSOC.getOrdinal(oldLocation.tierAlias());
    int newTierOrdinal = WORKER_STORAGE_TIER_ASSOC.getOrdinal(newLocation.tierAlias());
    if (newTierOrdinal == 0 && oldTierOrdinal != newTierOrdinal) {
      BLOCKS_PROMOTED.inc();
    }
  }

  @Override
  public void onRemoveBlockByClient(long blockId) {
    BLOCKS_DELETED.inc();
  }

  @Override
  public void onMoveBlockByWorker(long blockId, BlockStoreLocation oldLocation,
      BlockStoreLocation newLocation) {
    int oldTierOrdinal = WORKER_STORAGE_TIER_ASSOC.getOrdinal(oldLocation.tierAlias());
    int newTierOrdinal = WORKER_STORAGE_TIER_ASSOC.getOrdinal(newLocation.tierAlias());
    if (newTierOrdinal == 0 && oldTierOrdinal != newTierOrdinal) {
      BLOCKS_PROMOTED.inc();
    }
  }

  @Override
  public void onRemoveBlockByWorker(long blockId) {
    BLOCKS_EVICTED.inc();
    BLOCKS_EVICTION_RATE.mark();
  }

  @Override
  public void onAbortBlock(long blockId) {
    BLOCKS_CANCELLED.inc();
  }

  @Override
  public void onBlockLost(long blockId) {
    BLOCKS_LOST.inc();
  }
}
