alluxio-master-2.x/core/server/common/src/main/java/alluxio/master/journal/JournalUtils.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.journal;

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.master.journal.checkpoint.CheckpointInputStream;
import alluxio.master.journal.checkpoint.CheckpointOutputStream;
import alluxio.master.journal.checkpoint.CheckpointType;
import alluxio.master.journal.checkpoint.Checkpointed;
import alluxio.master.journal.checkpoint.CompoundCheckpointFormat.CompoundCheckpointReader;
import alluxio.master.journal.checkpoint.CompoundCheckpointFormat.CompoundCheckpointReader.Entry;
import alluxio.master.journal.sink.JournalSink;
import alluxio.proto.journal.Journal.JournalEntry;
import alluxio.resource.CloseableIterator;
import alluxio.util.StreamUtils;

import com.esotericsoftware.kryo.io.OutputChunked;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.function.Supplier;

/**
 * Utility methods for working with the Alluxio journal.
 */
public final class JournalUtils {
  private static final Logger LOG = LoggerFactory.getLogger(JournalUtils.class);

  /**
   * Returns a URI for the configured location for the specified journal.
   *
   * @return the journal location
   */
  public static URI getJournalLocation() {
    String journalDirectory = Configuration.getString(PropertyKey.MASTER_JOURNAL_FOLDER);
    if (!journalDirectory.endsWith(AlluxioURI.SEPARATOR)) {
      journalDirectory += AlluxioURI.SEPARATOR;
    }
    try {
      return new URI(journalDirectory);
    } catch (URISyntaxException e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * Writes a checkpoint of the entries in the given iterable.
   *
   * This is the complement of
   * {@link #restoreJournalEntryCheckpoint(CheckpointInputStream, Journaled)}.
   *
   * @param output the stream to write to
   * @param iterable the iterable for fetching journal entries
   */
  public static void writeJournalEntryCheckpoint(OutputStream output, JournalEntryIterable iterable)
      throws IOException, InterruptedException {
    output = new CheckpointOutputStream(output, CheckpointType.JOURNAL_ENTRY);
    try (CloseableIterator<JournalEntry> it = iterable.getJournalEntryIterator()) {
      LOG.info("Write journal entry checkpoint");
      while (it.get().hasNext()) {
        if (Thread.interrupted()) {
          throw new InterruptedException();
        }
        it.get().next().writeDelimitedTo(output);
      }
    }
    output.flush();
  }

  /**
   * Restores the given journaled object from the journal entries in the input stream.
   *
   * This is the complement of
   * {@link #writeJournalEntryCheckpoint(OutputStream, JournalEntryIterable)}.
   *
   * @param input the stream to read from
   * @param journaled the object to restore
   */
  public static void restoreJournalEntryCheckpoint(CheckpointInputStream input, Journaled journaled)
      throws IOException {
    Preconditions.checkState(input.getType() == CheckpointType.JOURNAL_ENTRY,
        "Unrecognized checkpoint type when restoring %s: %s", journaled.getCheckpointName(),
        input.getType());
    journaled.resetState();
    LOG.info("Reading journal entries");
    JournalEntryStreamReader reader = new JournalEntryStreamReader(input);
    JournalEntry entry;
    while ((entry = reader.readEntry()) != null) {
      try {
        journaled.processJournalEntry(entry);
      } catch (Throwable t) {
        handleJournalReplayFailure(LOG, t,
            "Failed to process journal entry %s from a journal checkpoint", entry);
      }
    }
  }

  /**
   * Writes a composite checkpoint for the given checkpointed components.
   *
   * This is the complement of {@link #restoreFromCheckpoint(CheckpointInputStream, List)}.
   *
   * @param output the stream to write to
   * @param components the components to checkpoint
   */
  public static void writeToCheckpoint(OutputStream output, List<? extends Checkpointed> components)
      throws IOException, InterruptedException {
    OutputChunked chunked = new OutputChunked(
        new CheckpointOutputStream(output, CheckpointType.COMPOUND), 64 * Constants.KB);
    for (Checkpointed component : components) {
      chunked.writeString(component.getCheckpointName().toString());
      component.writeToCheckpoint(chunked);
      chunked.endChunks();
    }
    chunked.flush();
  }

  /**
   * Restores the given checkpointed components from a composite checkpoint.
   *
   * This is the complement of {@link #writeToCheckpoint(OutputStream, List)}.
   *
   * @param input the stream to read from
   * @param components the components to restore
   */
  public static void restoreFromCheckpoint(CheckpointInputStream input,
      List<? extends Checkpointed> components) throws IOException {
    CompoundCheckpointReader reader = new CompoundCheckpointReader(input);
    Optional<Entry> next;
    while ((next = reader.nextCheckpoint()).isPresent()) {
      Entry nextEntry = next.get();
      boolean found = false;
      for (Checkpointed component : components) {
        if (component.getCheckpointName().equals(nextEntry.getName())) {
          component.restoreFromCheckpoint(nextEntry.getStream());
          found = true;
          break;
        }
      }
      if (!found) {
        throw new RuntimeException(String.format(
            "Unrecognized checkpoint name: %s. Existing components: %s", nextEntry.getName(), Arrays
                .toString(StreamUtils.map(Checkpointed::getCheckpointName, components).toArray())));
      }
    }
  }

  /**
   * Logs a fatal error and exits the system if required.
   *
   * @param logger the logger to log to
   * @param t the throwable causing the fatal error
   * @param format the error message format string
   * @param args args for the format string
   */
  public static void handleJournalReplayFailure(Logger logger, Throwable t,
      String format, Object... args) throws RuntimeException {
    String message = String.format("Journal replay error: " + format, args);
    if (t != null) {
      message += "\n" + Throwables.getStackTraceAsString(t);
    }
    logger.error(message);
    if (!Configuration.getBoolean(PropertyKey.MASTER_JOURNAL_TOLERATE_CORRUPTION)) {
      throw new RuntimeException(t);
    }
  }

  /**
   * Appends a journal entry to all the supplied journal sinks.
   *
   * @param journalSinks a supplier of journal sinks
   * @param entry the journal entry
   */
  public static void sinkAppend(Supplier<Set<JournalSink>> journalSinks, JournalEntry entry) {
    for (JournalSink sink : journalSinks.get()) {
      sink.append(entry);
    }
  }

  /**
   * Appends a flush to all the supplied journal sinks.
   *
   * @param journalSinks a supplier of journal sinks
   */
  public static void sinkFlush(Supplier<Set<JournalSink>> journalSinks) {
    for (JournalSink sink : journalSinks.get()) {
      sink.flush();
    }
  }

  private JournalUtils() {} // prevent instantiation
}
