/*
 * 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.tracker;

import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystemContext;
import alluxio.grpc.OperationType;
import alluxio.job.cmd.persist.PersistCmdConfig;
import alluxio.job.wire.JobSource;
import alluxio.master.job.JobMaster;
import alluxio.master.job.common.CmdInfo;
import alluxio.master.job.metrics.DistributedCmdMetrics;
import alluxio.retry.CountingRetry;

import com.google.common.collect.Lists;

import javax.annotation.Nullable;

/**
 * A config runner for a Persist job.
 */
public class PersistRunner {
  static final long DEFAULT_FILE_COUNT = 1;

  private final FileSystem mFileSystem;
  private final JobMaster mJobMaster;

  /**
   * constructor.
   * @param fsContext file system context
   * @param jobMaster job master
   */
  public PersistRunner(@Nullable FileSystemContext fsContext, JobMaster jobMaster) {
    if (fsContext == null) {
      fsContext = FileSystemContext.create();
    }
    mFileSystem = FileSystem.Factory.create(fsContext);
    mJobMaster = jobMaster;
  }

  /**
   * Run a Persist job.
   * @param config the config of persist operation
   * @param jobControlId THe parent level job control ID
   * @return CmdInfo
   */
  public CmdInfo runPersistJob(PersistCmdConfig config, long jobControlId) {
    long submissionTime = System.currentTimeMillis();
    CmdInfo cmdInfo = new CmdInfo(jobControlId, OperationType.PERSIST, JobSource.SYSTEM,
            submissionTime, Lists.newArrayList(config.getFilePath()));

    CmdRunAttempt attempt = new CmdRunAttempt(new CountingRetry(3), mJobMaster);
    setJobConfigAndFileMetrics(config, attempt);

    cmdInfo.addCmdRunAttempt(attempt);
    attempt.run();
    return cmdInfo;
  }

  // Create a JobConfig and set file count and size for the AsyncPersist job.
  private void setJobConfigAndFileMetrics(PersistCmdConfig config, CmdRunAttempt attempt) {
    long fileSize = DistributedCmdMetrics.getFileSize(config.getFilePath(),
            mFileSystem, new CountingRetry(3));
    attempt.setFileCount(DEFAULT_FILE_COUNT);
    attempt.setFileSize(fileSize);
    attempt.setConfig(config.toPersistConfig());
  }
}
