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

import alluxio.AlluxioURI;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.grpc.BackupPRequest;
import alluxio.master.BackupManager;
import alluxio.master.CoreMasterContext;
import alluxio.master.journal.JournalSystem;
import alluxio.master.transport.GrpcMessagingConnection;
import alluxio.master.transport.GrpcMessagingContext;
import alluxio.resource.CloseableResource;
import alluxio.security.user.UserState;
import alluxio.underfs.UfsManager;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.UnderFileSystemConfiguration;
import alluxio.underfs.options.MkdirsOptions;
import alluxio.util.ThreadFactoryUtils;
import alluxio.util.io.PathUtils;

import com.google.common.io.Closer;
import io.atomix.catalyst.serializer.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Abstract backup role implementation. It provides common infrastructure for leader/worker backup
 * roles.
 */
public abstract class AbstractBackupRole implements BackupRole {
  private static final Logger LOG = LoggerFactory.getLogger(AbstractBackupRole.class);

  // Request timeout for backup messages.
  protected final long mCatalystRequestTimeout;

  /** The manager of all ufs. */
  protected final UfsManager mUfsManager;

  /** Task scheduler. */
  protected ScheduledExecutorService mTaskScheduler;

  /** The executor service. */
  protected ExecutorService mExecutorService;

  /** Catalyst context used for sending/receiving messages. */
  protected GrpcMessagingContext mGrpcMessagingContext;

  /** Journal system. */
  protected JournalSystem mJournalSystem;

  /** Backup Manager. */
  protected BackupManager mBackupManager;

  /** User information for server. */
  protected UserState mServerUserState;

  /** Used to track backup status. */
  protected BackupTracker mBackupTracker;

  /** Whether the role is active. */
  protected boolean mRoleClosed = false;

  /**
   * Creates a BackupMaster.
   *
   * @param masterContext master context
   */
  public AbstractBackupRole(CoreMasterContext masterContext) {
    // Executor service will be used for driving the backup and connections.
    mExecutorService =
        Executors.newCachedThreadPool(ThreadFactoryUtils.build("backup-executor-%d", true));
    // Task scheduler will be used for various backup related tasks.
    mTaskScheduler =
        Executors.newScheduledThreadPool(1, ThreadFactoryUtils.build("backup-task-%d", true));
    // Store master context objects.
    mUfsManager = masterContext.getUfsManager();
    mJournalSystem = masterContext.getJournalSystem();
    mBackupManager = masterContext.getBackupManager();
    mServerUserState = masterContext.getUserState();
    // Initialize messaging infra.
    initializeCatalystContext();
    // Read properties.
    mCatalystRequestTimeout =
        Configuration.getMs(PropertyKey.MASTER_BACKUP_TRANSPORT_TIMEOUT);
    // Initialize backup tracker.
    mBackupTracker = new BackupTracker();
  }

  /**
   * Initializes the catalyst context with known message types.
   */
  private void initializeCatalystContext() {
    Serializer messageSerializer = new Serializer().register(BackupRequestMessage.class)
        .register(BackupHeartbeatMessage.class).register(BackupSuspendMessage.class);
    mGrpcMessagingContext = new GrpcMessagingContext("backup-context-%d", messageSerializer);
  }

  /**
   * Used to send message via connection and wait until response is received. Response is discarded
   * since backup messages are one-way.
   */
  protected void sendMessageBlocking(GrpcMessagingConnection connection, Object message)
      throws IOException {
    try {
      // First get is for the task, second is for the messaging future.
      mGrpcMessagingContext.execute(() -> connection.sendAndReceive(message)).get().get();
    } catch (InterruptedException ie) {
      throw new RuntimeException("Interrupted while waiting for messaging to complete.");
    } catch (ExecutionException ee) {
      throw new IOException("Failed to send and receive message", ee.getCause());
    }
  }

  /**
   * Takes a backup on local master. Note: Master state must have been suspended or locked before
   * calling this.
   *
   * @param request the backup request
   * @param entryCounter counter to receive written entry count
   * @return URI of the backup
   */
  protected AlluxioURI takeBackup(BackupPRequest request, AtomicLong entryCounter)
      throws IOException {
    AlluxioURI backupUri;

    // Acquire the UFS resource under which backup is being created.
    try (Closer closer = Closer.create(); CloseableResource<UnderFileSystem> ufsResource =
        mUfsManager.getRoot().acquireUfsResource()) {
      // Get backup parent directory.
      String backupParentDir = request.hasTargetDirectory() ? request.getTargetDirectory()
          : Configuration.getString(PropertyKey.MASTER_BACKUP_DIRECTORY);
      // Get ufs resource for backup.
      UnderFileSystem ufs = ufsResource.get();
      if (request.getOptions().getLocalFileSystem() && !ufs.getUnderFSType().equals("local")) {
        // TODO(lu) Support getting UFS based on type from UfsManager
        ufs = closer.register(UnderFileSystem.Factory.create("/",
            UnderFileSystemConfiguration.defaults(Configuration.global())));
      }
      // Ensure parent directory for backup.
      if (!ufs.isDirectory(backupParentDir)) {
        if (!ufs.mkdirs(backupParentDir,
            MkdirsOptions.defaults(Configuration.global()).setCreateParent(true))) {
          throw new IOException(String.format("Failed to create directory %s", backupParentDir));
        }
      }
      // Generate backup file path.
      Instant now = Instant.now();
      String backupFileName = String.format(BackupManager.BACKUP_FILE_FORMAT,
          DateTimeFormatter.ISO_LOCAL_DATE.withZone(ZoneId.of("UTC")).format(now),
          now.toEpochMilli());
      String backupFilePath = PathUtils.concatPath(backupParentDir, backupFileName);
      // Calculate URI for the path.
      String rootUfs = Configuration.getString(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS);
      if (request.getOptions().getLocalFileSystem()) {
        rootUfs = "file:///";
      }
      backupUri = new AlluxioURI(new AlluxioURI(rootUfs), new AlluxioURI(backupFilePath));

      // Take the backup.
      try {
        try (OutputStream ufsStream = ufs.create(backupFilePath)) {
          // Create the backup from master state.
          mBackupManager.backup(ufsStream, entryCounter);
        }
        // Add a marker file indicating the file is completed.
        ufs.create(backupFilePath + ".complete").close();
      } catch (IOException e) {
        try {
          ufs.deleteExistingFile(backupFilePath);
        } catch (Exception e2) {
          LOG.error("Failed to clean up failed backup at {}", backupFilePath, e2);
          e.addSuppressed(e2);
        }
        throw new IOException(String.format("Backup failed. BackupUri: %s, LastEntryCount: %d",
            backupUri, entryCounter.get()), e);
      }
    }
    return backupUri;
  }

  @Override
  public void close() throws IOException {
    // Stop task scheduler.
    if (mTaskScheduler != null) {
      mTaskScheduler.shutdownNow();
    }
    // Close messaging context.
    mGrpcMessagingContext.close();
    // Shutdown backup executor.
    mExecutorService.shutdownNow();
    // Reset backup tracker.
    mBackupTracker.reset();
    // Mark the role as closed.
    mRoleClosed = true;
  }
}
