hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FileJournalManager.java - hadoop - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.hadoop.hdfs.server.namenode;

 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;

 import java.io.File;
 import java.io.IOException;
 import java.util.List;
 import java.util.Comparator;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;

 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.hdfs.server.common.Storage.StorageDirectory;
 import org.apache.hadoop.hdfs.server.namenode.NNStorageRetentionManager.StoragePurger;
 import org.apache.hadoop.hdfs.server.namenode.FSEditLogLoader.EditLogValidation;
 import org.apache.hadoop.hdfs.server.namenode.NNStorage.NameNodeFile;
 import org.apache.hadoop.hdfs.server.protocol.RemoteEditLog;

 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import com.google.common.collect.ComparisonChain;

 /**
  * Journal manager for the common case of edits files being written
  * to a storage directory.
  *
  * Note: this class is not thread-safe and should be externally
  * synchronized.
  */
 class FileJournalManager implements JournalManager {
   private static final Log LOG = LogFactory.getLog(FileJournalManager.class);

   private final StorageDirectory sd;
   private int outputBufferCapacity = 512*1024;

   private static final Pattern EDITS_REGEX = Pattern.compile(
     NameNodeFile.EDITS.getName() + "_(\\d+)-(\\d+)");
   private static final Pattern EDITS_INPROGRESS_REGEX = Pattern.compile(
     NameNodeFile.EDITS_INPROGRESS.getName() + "_(\\d+)");

   @VisibleForTesting
   StoragePurger purger
     = new NNStorageRetentionManager.DeletionStoragePurger();

   public FileJournalManager(StorageDirectory sd) {
     this.sd = sd;
   }

   @Override
   public EditLogOutputStream startLogSegment(long txid) throws IOException {
     File newInProgress = NNStorage.getInProgressEditsFile(sd, txid);
     EditLogOutputStream stm = new EditLogFileOutputStream(newInProgress,
         outputBufferCapacity);
     stm.create();
     return stm;
   }

   @Override
   public void finalizeLogSegment(long firstTxId, long lastTxId)
       throws IOException {
     File inprogressFile = NNStorage.getInProgressEditsFile(
         sd, firstTxId);
     File dstFile = NNStorage.getFinalizedEditsFile(
         sd, firstTxId, lastTxId);
     LOG.debug("Finalizing edits file " + inprogressFile + " -> " + dstFile);

     Preconditions.checkState(!dstFile.exists(),
         "Can't finalize edits file " + inprogressFile + " since finalized file " +
         "already exists");
     if (!inprogressFile.renameTo(dstFile)) {
       throw new IOException("Unable to finalize edits file " + inprogressFile);
     }
   }

   @VisibleForTesting
   public StorageDirectory getStorageDirectory() {
     return sd;
   }

   @Override
   public String toString() {
     return "FileJournalManager for storage directory " + sd;
   }

   @Override
   public void setOutputBufferCapacity(int size) {
     this.outputBufferCapacity = size;
   }

   @Override
   public void purgeLogsOlderThan(long minTxIdToKeep)
       throws IOException {
     File[] files = FileUtil.listFiles(sd.getCurrentDir());
     List<EditLogFile> editLogs =
       FileJournalManager.matchEditLogs(files);
     for (EditLogFile log : editLogs) {
       if (log.getFirstTxId() < minTxIdToKeep &&
           log.getLastTxId() < minTxIdToKeep) {
         purger.purgeLog(log);
       }
     }
   }

   @Override
   public EditLogInputStream getInProgressInputStream(long segmentStartsAtTxId)
       throws IOException {
     File f = NNStorage.getInProgressEditsFile(sd, segmentStartsAtTxId);
     return new EditLogFileInputStream(f);
   }

   /**
    * Find all editlog segments starting at or above the given txid.
    * @param fromTxId the txnid which to start looking
    * @return a list of remote edit logs
    * @throws IOException if edit logs cannot be listed.
    */
   List<RemoteEditLog> getRemoteEditLogs(long firstTxId) throws IOException {
     File currentDir = sd.getCurrentDir();
     List<EditLogFile> allLogFiles = matchEditLogs(
         FileUtil.listFiles(currentDir));
     List<RemoteEditLog> ret = Lists.newArrayListWithCapacity(
         allLogFiles.size());

     for (EditLogFile elf : allLogFiles) {
       if (elf.isCorrupt() || elf.isInProgress()) continue;
       if (elf.getFirstTxId() >= firstTxId) {
         ret.add(new RemoteEditLog(elf.firstTxId, elf.lastTxId));
       } else if ((firstTxId > elf.getFirstTxId()) &&
                  (firstTxId <= elf.getLastTxId())) {
         throw new IOException("Asked for firstTxId " + firstTxId
             + " which is in the middle of file " + elf.file);
       }
     }

     return ret;
   }

   static List<EditLogFile> matchEditLogs(File[] filesInStorage) {
     List<EditLogFile> ret = Lists.newArrayList();
     for (File f : filesInStorage) {
       String name = f.getName();
       // Check for edits
       Matcher editsMatch = EDITS_REGEX.matcher(name);
       if (editsMatch.matches()) {
         try {
           long startTxId = Long.valueOf(editsMatch.group(1));
           long endTxId = Long.valueOf(editsMatch.group(2));
           ret.add(new EditLogFile(f, startTxId, endTxId));
         } catch (NumberFormatException nfe) {
           LOG.error("Edits file " + f + " has improperly formatted " +
                     "transaction ID");
           // skip
         }
       }

       // Check for in-progress edits
       Matcher inProgressEditsMatch = EDITS_INPROGRESS_REGEX.matcher(name);
       if (inProgressEditsMatch.matches()) {
         try {
           long startTxId = Long.valueOf(inProgressEditsMatch.group(1));
           ret.add(
             new EditLogFile(f, startTxId, EditLogFile.UNKNOWN_END));
         } catch (NumberFormatException nfe) {
           LOG.error("In-progress edits file " + f + " has improperly " +
                     "formatted transaction ID");
           // skip
         }
       }
     }
     return ret;
   }

   /**
    * Record of an edit log that has been located and had its filename parsed.
    */
   static class EditLogFile {
     private File file;
     private final long firstTxId;
     private long lastTxId;

     private EditLogValidation cachedValidation = null;
     private boolean isCorrupt = false;

     static final long UNKNOWN_END = -1;

     final static Comparator<EditLogFile> COMPARE_BY_START_TXID
       = new Comparator<EditLogFile>() {
       public int compare(EditLogFile a, EditLogFile b) {
         return ComparisonChain.start()
         .compare(a.getFirstTxId(), b.getFirstTxId())
         .compare(a.getLastTxId(), b.getLastTxId())
         .result();
       }
     };

     EditLogFile(File file,
         long firstTxId, long lastTxId) {
       assert lastTxId == UNKNOWN_END || lastTxId >= firstTxId;
       assert firstTxId > 0;
       assert file != null;

       this.firstTxId = firstTxId;
       this.lastTxId = lastTxId;
       this.file = file;
     }

     public void finalizeLog() throws IOException {
       long numTransactions = validateLog().numTransactions;
       long lastTxId = firstTxId + numTransactions - 1;
       File dst = new File(file.getParentFile(),
           NNStorage.getFinalizedEditsFileName(firstTxId, lastTxId));
       LOG.info("Finalizing edits log " + file + " by renaming to "
           + dst.getName());
       if (!file.renameTo(dst)) {
         throw new IOException("Couldn't finalize log " +
             file + " to " + dst);
       }
       this.lastTxId = lastTxId;
       file = dst;
     }

     long getFirstTxId() {
       return firstTxId;
     }

     long getLastTxId() {
       return lastTxId;
     }

     EditLogValidation validateLog() throws IOException {
       if (cachedValidation == null) {
         cachedValidation = EditLogFileInputStream.validateEditLog(file);
       }
       return cachedValidation;
     }

     boolean isInProgress() {
       return (lastTxId == UNKNOWN_END);
     }

     File getFile() {
       return file;
     }

     void markCorrupt() {
       isCorrupt = true;
     }

     boolean isCorrupt() {
       return isCorrupt;
     }

     void moveAsideCorruptFile() throws IOException {
       assert isCorrupt;

       File src = file;
       File dst = new File(src.getParent(), src.getName() + ".corrupt");
       boolean success = src.renameTo(dst);
       if (!success) {
         throw new IOException(
           "Couldn't rename corrupt log " + src + " to " + dst);
       }
       file = dst;
     }

     @Override
     public String toString() {
       return String.format("EditLogFile(file=%s,first=%019d,last=%019d,"
                            +"inProgress=%b,corrupt=%b)", file.toString(),
                            firstTxId, lastTxId, isInProgress(), isCorrupt);
     }
   }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.hadoop.hdfs.server.namenode;

	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;

	import java.io.File;
	import java.io.IOException;
	import java.util.List;
	import java.util.Comparator;
	import java.util.regex.Matcher;
	import java.util.regex.Pattern;

	import org.apache.hadoop.fs.FileUtil;
	import org.apache.hadoop.hdfs.server.common.Storage.StorageDirectory;
	import org.apache.hadoop.hdfs.server.namenode.NNStorageRetentionManager.StoragePurger;
	import org.apache.hadoop.hdfs.server.namenode.FSEditLogLoader.EditLogValidation;
	import org.apache.hadoop.hdfs.server.namenode.NNStorage.NameNodeFile;
	import org.apache.hadoop.hdfs.server.protocol.RemoteEditLog;

	import com.google.common.annotations.VisibleForTesting;
	import com.google.common.base.Preconditions;
	import com.google.common.collect.Lists;
	import com.google.common.collect.ComparisonChain;

	/**
	* Journal manager for the common case of edits files being written
	* to a storage directory.
	*
	* Note: this class is not thread-safe and should be externally
	* synchronized.
	*/
	class FileJournalManager implements JournalManager {
	private static final Log LOG = LogFactory.getLog(FileJournalManager.class);

	private final StorageDirectory sd;
	private int outputBufferCapacity = 512*1024;

	private static final Pattern EDITS_REGEX = Pattern.compile(
	NameNodeFile.EDITS.getName() + "_(\\d+)-(\\d+)");
	private static final Pattern EDITS_INPROGRESS_REGEX = Pattern.compile(
	NameNodeFile.EDITS_INPROGRESS.getName() + "_(\\d+)");

	@VisibleForTesting
	StoragePurger purger
	= new NNStorageRetentionManager.DeletionStoragePurger();

	public FileJournalManager(StorageDirectory sd) {
	this.sd = sd;
	}

	@Override
	public EditLogOutputStream startLogSegment(long txid) throws IOException {
	File newInProgress = NNStorage.getInProgressEditsFile(sd, txid);
	EditLogOutputStream stm = new EditLogFileOutputStream(newInProgress,
	outputBufferCapacity);
	stm.create();
	return stm;
	}

	@Override
	public void finalizeLogSegment(long firstTxId, long lastTxId)
	throws IOException {
	File inprogressFile = NNStorage.getInProgressEditsFile(
	sd, firstTxId);
	File dstFile = NNStorage.getFinalizedEditsFile(
	sd, firstTxId, lastTxId);
	LOG.debug("Finalizing edits file " + inprogressFile + " -> " + dstFile);

	Preconditions.checkState(!dstFile.exists(),
	"Can't finalize edits file " + inprogressFile + " since finalized file " +
	"already exists");
	if (!inprogressFile.renameTo(dstFile)) {
	throw new IOException("Unable to finalize edits file " + inprogressFile);
	}
	}

	@VisibleForTesting
	public StorageDirectory getStorageDirectory() {
	return sd;
	}

	@Override
	public String toString() {
	return "FileJournalManager for storage directory " + sd;
	}

	@Override
	public void setOutputBufferCapacity(int size) {
	this.outputBufferCapacity = size;
	}

	@Override
	public void purgeLogsOlderThan(long minTxIdToKeep)
	throws IOException {
	File[] files = FileUtil.listFiles(sd.getCurrentDir());
	List<EditLogFile> editLogs =
	FileJournalManager.matchEditLogs(files);
	for (EditLogFile log : editLogs) {
	if (log.getFirstTxId() < minTxIdToKeep &&
	log.getLastTxId() < minTxIdToKeep) {
	purger.purgeLog(log);
	}
	}
	}

	@Override
	public EditLogInputStream getInProgressInputStream(long segmentStartsAtTxId)
	throws IOException {
	File f = NNStorage.getInProgressEditsFile(sd, segmentStartsAtTxId);
	return new EditLogFileInputStream(f);
	}

	/**
	* Find all editlog segments starting at or above the given txid.
	* @param fromTxId the txnid which to start looking
	* @return a list of remote edit logs
	* @throws IOException if edit logs cannot be listed.
	*/
	List<RemoteEditLog> getRemoteEditLogs(long firstTxId) throws IOException {
	File currentDir = sd.getCurrentDir();
	List<EditLogFile> allLogFiles = matchEditLogs(
	FileUtil.listFiles(currentDir));
	List<RemoteEditLog> ret = Lists.newArrayListWithCapacity(
	allLogFiles.size());

	for (EditLogFile elf : allLogFiles) {
	if (elf.isCorrupt() \|\| elf.isInProgress()) continue;
	if (elf.getFirstTxId() >= firstTxId) {
	ret.add(new RemoteEditLog(elf.firstTxId, elf.lastTxId));
	} else if ((firstTxId > elf.getFirstTxId()) &&
	(firstTxId <= elf.getLastTxId())) {
	throw new IOException("Asked for firstTxId " + firstTxId
	+ " which is in the middle of file " + elf.file);
	}
	}

	return ret;
	}

	static List<EditLogFile> matchEditLogs(File[] filesInStorage) {
	List<EditLogFile> ret = Lists.newArrayList();
	for (File f : filesInStorage) {
	String name = f.getName();
	// Check for edits
	Matcher editsMatch = EDITS_REGEX.matcher(name);
	if (editsMatch.matches()) {
	try {
	long startTxId = Long.valueOf(editsMatch.group(1));
	long endTxId = Long.valueOf(editsMatch.group(2));
	ret.add(new EditLogFile(f, startTxId, endTxId));
	} catch (NumberFormatException nfe) {
	LOG.error("Edits file " + f + " has improperly formatted " +
	"transaction ID");
	// skip
	}
	}

	// Check for in-progress edits
	Matcher inProgressEditsMatch = EDITS_INPROGRESS_REGEX.matcher(name);
	if (inProgressEditsMatch.matches()) {
	try {
	long startTxId = Long.valueOf(inProgressEditsMatch.group(1));
	ret.add(
	new EditLogFile(f, startTxId, EditLogFile.UNKNOWN_END));
	} catch (NumberFormatException nfe) {
	LOG.error("In-progress edits file " + f + " has improperly " +
	"formatted transaction ID");
	// skip
	}
	}
	}
	return ret;
	}

	/**
	* Record of an edit log that has been located and had its filename parsed.
	*/
	static class EditLogFile {
	private File file;
	private final long firstTxId;
	private long lastTxId;

	private EditLogValidation cachedValidation = null;
	private boolean isCorrupt = false;

	static final long UNKNOWN_END = -1;

	final static Comparator<EditLogFile> COMPARE_BY_START_TXID
	= new Comparator<EditLogFile>() {
	public int compare(EditLogFile a, EditLogFile b) {
	return ComparisonChain.start()
	.compare(a.getFirstTxId(), b.getFirstTxId())
	.compare(a.getLastTxId(), b.getLastTxId())
	.result();
	}
	};

	EditLogFile(File file,
	long firstTxId, long lastTxId) {
	assert lastTxId == UNKNOWN_END \|\| lastTxId >= firstTxId;
	assert firstTxId > 0;
	assert file != null;

	this.firstTxId = firstTxId;
	this.lastTxId = lastTxId;
	this.file = file;
	}

	public void finalizeLog() throws IOException {
	long numTransactions = validateLog().numTransactions;
	long lastTxId = firstTxId + numTransactions - 1;
	File dst = new File(file.getParentFile(),
	NNStorage.getFinalizedEditsFileName(firstTxId, lastTxId));
	LOG.info("Finalizing edits log " + file + " by renaming to "
	+ dst.getName());
	if (!file.renameTo(dst)) {
	throw new IOException("Couldn't finalize log " +
	file + " to " + dst);
	}
	this.lastTxId = lastTxId;
	file = dst;
	}

	long getFirstTxId() {
	return firstTxId;
	}

	long getLastTxId() {
	return lastTxId;
	}

	EditLogValidation validateLog() throws IOException {
	if (cachedValidation == null) {
	cachedValidation = EditLogFileInputStream.validateEditLog(file);
	}
	return cachedValidation;
	}

	boolean isInProgress() {
	return (lastTxId == UNKNOWN_END);
	}

	File getFile() {
	return file;
	}

	void markCorrupt() {
	isCorrupt = true;
	}

	boolean isCorrupt() {
	return isCorrupt;
	}

	void moveAsideCorruptFile() throws IOException {
	assert isCorrupt;

	File src = file;
	File dst = new File(src.getParent(), src.getName() + ".corrupt");
	boolean success = src.renameTo(dst);
	if (!success) {
	throw new IOException(
	"Couldn't rename corrupt log " + src + " to " + dst);
	}
	file = dst;
	}

	@Override
	public String toString() {
	return String.format("EditLogFile(file=%s,first=%019d,last=%019d,"
	+"inProgress=%b,corrupt=%b)", file.toString(),
	firstTxId, lastTxId, isInProgress(), isCorrupt);
	}
	}
	}