alluxio-master-2.x/job/server/src/main/java/alluxio/master/job/metrics/DistributedCmdMetrics.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.master.job.metrics;

import alluxio.AlluxioURI;
import alluxio.client.file.FileSystem;
import alluxio.exception.AlluxioException;
import alluxio.job.JobConfig;
import alluxio.job.plan.BatchedJobConfig;
import alluxio.job.plan.load.LoadConfig;
import alluxio.job.plan.migrate.MigrateConfig;
import alluxio.job.plan.persist.PersistConfig;
import alluxio.metrics.MetricKey;
import alluxio.metrics.MetricsSystem;
import alluxio.retry.RetryPolicy;

import com.codahale.metrics.Counter;
import com.codahale.metrics.Meter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.Map;

/**
 * DistributedCmdMetrics has definitions for distributed command metrics and help methods.
 */
public class DistributedCmdMetrics {
  private static final Logger LOG = LoggerFactory.getLogger(DistributedCmdMetrics.class);

  private static final Counter JOB_DISTRIBUTED_LOAD_SUCCESS =
          MetricsSystem.counter(MetricKey.MASTER_JOB_DISTRIBUTED_LOAD_SUCCESS.getName());
  private static final Counter JOB_DISTRIBUTED_LOAD_FAIL =
          MetricsSystem.counter(MetricKey.MASTER_JOB_DISTRIBUTED_LOAD_FAIL.getName());
  private static final Counter JOB_DISTRIBUTED_LOAD_CANCEL =
          MetricsSystem.counter(MetricKey.MASTER_JOB_DISTRIBUTED_LOAD_CANCEL.getName());
  private static final Counter JOB_DISTRIBUTED_LOAD_FILE_COUNT =
          MetricsSystem.counter(MetricKey.MASTER_JOB_DISTRIBUTED_LOAD_FILE_COUNT.getName());
  private static final Counter JOB_DISTRIBUTED_LOAD_FILE_SIZE =
          MetricsSystem.counter(MetricKey.MASTER_JOB_DISTRIBUTED_LOAD_FILE_SIZE.getName());
  private static final Meter JOB_DISTRIBUTED_LOAD_RATE =
          MetricsSystem.meter(MetricKey.MASTER_JOB_DISTRIBUTED_LOAD_RATE.getName());

  private static final Counter MIGRATE_JOB_SUCCESS =
          MetricsSystem.counter(MetricKey.MASTER_MIGRATE_JOB_SUCCESS.getName());
  private static final Counter MIGRATE_JOB_FAIL =
          MetricsSystem.counter(MetricKey.MASTER_MIGRATE_JOB_FAIL.getName());
  private static final Counter MIGRATE_JOB_CANCEL =
          MetricsSystem.counter(MetricKey.MASTER_MIGRATE_JOB_CANCEL.getName());
  private static final Counter MIGRATE_JOB_FILE_COUNT =
          MetricsSystem.counter(MetricKey.MASTER_MIGRATE_JOB_FILE_COUNT.getName());
  private static final Counter MIGRATE_JOB_FILE_SIZE =
          MetricsSystem.counter(MetricKey.MASTER_MIGRATE_JOB_FILE_SIZE.getName());

  private static final Counter ASYNC_PERSIST_SUCCESS =
          MetricsSystem.counter(MetricKey.MASTER_ASYNC_PERSIST_SUCCESS.getName());
  private static final Counter ASYNC_PERSIST_FAIL =
          MetricsSystem.counter(MetricKey.MASTER_ASYNC_PERSIST_FAIL.getName());
  private static final Counter ASYNC_PERSIST_CANCEL =
          MetricsSystem.counter(MetricKey.MASTER_ASYNC_PERSIST_CANCEL.getName());
  private static final Counter ASYNC_PERSIST_FILE_COUNT =
          MetricsSystem.counter(MetricKey.MASTER_ASYNC_PERSIST_FILE_COUNT.getName());
  private static final Counter ASYNC_PERSIST_FILE_SIZE =
          MetricsSystem.counter(MetricKey.MASTER_ASYNC_PERSIST_FILE_SIZE.getName());

  private static final long DEFAULT_INCREMENT_VALUE = 1;

/**
 * Increment failed job count.
 * @param jobType job type
 * @param count count to increment
 */
  public static void incrementForFailStatus(String jobType, long count) {
    switch (jobType) {
      case LoadConfig.NAME:
        JOB_DISTRIBUTED_LOAD_FAIL.inc(count);
        break;
      case MigrateConfig.NAME:
        MIGRATE_JOB_FAIL.inc(count);
        break;
      case PersistConfig.NAME:
        ASYNC_PERSIST_FAIL.inc(count);
        break;
      default:
        LOG.warn("JobType does not belong to Load, Migrate and Persist");
        break;
    }
  }

/**
 * Increment cancelled job count.
 * @param jobType job type
 * @param count count to increment
 */
  public static void incrementForCancelStatus(String jobType, long count) {
    switch (jobType) {
      case LoadConfig.NAME:
        JOB_DISTRIBUTED_LOAD_CANCEL.inc(count);
        break;
      case MigrateConfig.NAME:
        MIGRATE_JOB_CANCEL.inc(count);
        break;
      case PersistConfig.NAME:
        ASYNC_PERSIST_CANCEL.inc(count);
        break;
      default:
        LOG.warn("JobType does not belong to Load, Migrate and Persist");
        break;
    }
  }

  /**
   * get file size for a given filePath, return 0 if exceeding retries.
   * @param filePath file path
   * @param fileSystem file system
   * @param retryPolicy retry policy
   * @return file size
   */
  public static long getFileSize(String filePath, FileSystem fileSystem, RetryPolicy retryPolicy) {
    while (retryPolicy.attempt()) {
      try {
        return fileSystem.getStatus(new AlluxioURI(filePath)).getLength();
      } catch (IOException | AlluxioException | RuntimeException e) {
        LOG.warn("Retry getStatus for URI {} for {}-th time, {}",
                filePath, retryPolicy.getAttemptCount(), Arrays.toString(e.getStackTrace()));
      }
    }
    return 0;
  }

  /**
   * Increment for non-batch config for complete status.
   * @param config the job config
   * @param fileSystem filesystem
   * @param retryPolicy the retry policy used for retry operations
   */
  public static void incrementForCompleteStatusWithRetry(
          JobConfig config, FileSystem fileSystem, RetryPolicy retryPolicy) {
    String filePath;
    long fileSize;
    if (config instanceof LoadConfig) {
      filePath = ((LoadConfig) config).getFilePath();
      JOB_DISTRIBUTED_LOAD_SUCCESS.inc(DEFAULT_INCREMENT_VALUE);
      JOB_DISTRIBUTED_LOAD_FILE_COUNT.inc(DEFAULT_INCREMENT_VALUE);

      fileSize = getFileSize(filePath, fileSystem, retryPolicy);
      JOB_DISTRIBUTED_LOAD_FILE_SIZE.inc(fileSize);
      JOB_DISTRIBUTED_LOAD_RATE.mark(fileSize);
    } else if (config instanceof MigrateConfig) {
      filePath = ((MigrateConfig) config).getSource();
      MIGRATE_JOB_SUCCESS.inc(DEFAULT_INCREMENT_VALUE);
      MIGRATE_JOB_FILE_COUNT.inc(DEFAULT_INCREMENT_VALUE);

      fileSize = getFileSize(filePath, fileSystem, retryPolicy);
      MIGRATE_JOB_FILE_SIZE.inc(fileSize);
    } else if (config instanceof PersistConfig) {
      filePath = ((PersistConfig) config).getFilePath();
      ASYNC_PERSIST_SUCCESS.inc(DEFAULT_INCREMENT_VALUE);
      ASYNC_PERSIST_FILE_COUNT.inc(DEFAULT_INCREMENT_VALUE);

      fileSize = getFileSize(filePath, fileSystem, retryPolicy);
      ASYNC_PERSIST_FILE_SIZE.inc(fileSize);
    } else {
      LOG.warn("JobType does not belong to Load, Migrate and Persist");
    }
  }

/**
 * Increment for batch config for complete status.
 * @param config the job config
 * @param fileSystem filesystem
 * @param retryPolicy the retry policy used for retry operations
 */
  public static void batchIncrementForCompleteStatusWithRetry(
          BatchedJobConfig config, FileSystem fileSystem, RetryPolicy retryPolicy) {
    String jobType = config.getJobType();
    long count = config.getJobConfigs().size(); //count of files

    switch (jobType) {
      case LoadConfig.NAME:
        JOB_DISTRIBUTED_LOAD_SUCCESS.inc(count);
        JOB_DISTRIBUTED_LOAD_FILE_COUNT.inc(count);

        for (Map<String, String> jobConfig : config.getJobConfigs()) {
          //Look up actual filePath for LoadConfig from the map, key is given "filePath"
          String filePath = jobConfig.get("filePath");
          long fileSize = getFileSize(filePath, fileSystem, retryPolicy);
          JOB_DISTRIBUTED_LOAD_FILE_SIZE.inc(fileSize);
          JOB_DISTRIBUTED_LOAD_RATE.mark(fileSize);
        }
        return;
      case MigrateConfig.NAME:
        MIGRATE_JOB_SUCCESS.inc(count);
        MIGRATE_JOB_FILE_COUNT.inc(count);

        for (Map<String, String> jobConfig : config.getJobConfigs()) {
          //Look up actual filePath for MigrateConfig from the map, key is given "source"
          String filePath = jobConfig.get("source");
          long fileSize = getFileSize(filePath, fileSystem, retryPolicy);
          MIGRATE_JOB_FILE_SIZE.inc(fileSize);
        }
        return;
      case PersistConfig.NAME:
        ASYNC_PERSIST_SUCCESS.inc(count);
        ASYNC_PERSIST_FILE_COUNT.inc(count);

        for (Map<String, String> jobConfig : config.getJobConfigs()) {
          //Look up actual filePath for PersistConfig from the map, key is given "filePath"
          String filePath = jobConfig.get("filePath");
          long fileSize = getFileSize(filePath, fileSystem, retryPolicy);
          ASYNC_PERSIST_FILE_SIZE.inc(fileSize);
        }
        return;
      default:
        LOG.warn("JobType does not belong to Load, Migrate and Persist");
        return;
    }
  }

/**
 *  Increment for Fail status for batch job.
 * @param config job config
 */
  public static void batchIncrementForFailStatus(BatchedJobConfig config) {
    String jobType = config.getJobType();
    long count = config.getJobConfigs().size();
    incrementForFailStatus(jobType, count);
  }

/**
 *  Increment for Cancel status for batch job.
 * @param config job config
 */
  public static void batchIncrementForCancelStatus(BatchedJobConfig config) {
    String jobType = config.getJobType();
    long count = config.getJobConfigs().size();
    incrementForCancelStatus(jobType, count);
  }

  /**
   * Increment for both batch and non-batch configs for COMPLETE status.
   * @param config job config
   * @param fileSystem filesystem
   * @param retryPolicy the retry policy used for retry operations
   */
  public static void incrementForAllConfigsCompleteStatus(
          JobConfig config, FileSystem fileSystem, RetryPolicy retryPolicy) {
    if (config instanceof BatchedJobConfig) {
      batchIncrementForCompleteStatusWithRetry((BatchedJobConfig) config, fileSystem, retryPolicy);
    } else {
      incrementForCompleteStatusWithRetry(config, fileSystem, retryPolicy);
    }
  }

/**
 * Increment for both batch and non-batch configs for Cancel status.
  * @param config job config
 */
  public static void incrementForAllConfigsCancelStatus(JobConfig config) {
    if (config instanceof BatchedJobConfig) {
      batchIncrementForCancelStatus((BatchedJobConfig) config);
    } else {
      incrementForCancelStatus(config.getName(), DEFAULT_INCREMENT_VALUE);
    }
  }

/**
 * Increment for both batch and non-batch configs for Fail status.
 * @param config job config
 */
  public static void incrementForAllConfigsFailStatus(JobConfig config) {
    if (config instanceof BatchedJobConfig) {
      batchIncrementForFailStatus((BatchedJobConfig) config);
    } else {
      incrementForFailStatus(config.getName(), DEFAULT_INCREMENT_VALUE);
    }
  }
}
