hdfs/src/java/org/apache/hadoop/hdfs/server/namenode/Checkpointer.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 java.io.IOException;
 import java.io.File;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.Collection;

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

 import org.apache.hadoop.hdfs.protocol.FSConstants;
 import org.apache.hadoop.hdfs.server.common.HdfsConstants.NamenodeRole;
 import org.apache.hadoop.hdfs.server.namenode.FSImage.CheckpointStates;
 import org.apache.hadoop.hdfs.server.namenode.FSImage.NameNodeDirType;
 import org.apache.hadoop.hdfs.server.namenode.FSImage.NameNodeFile;
 import org.apache.hadoop.hdfs.server.protocol.CheckpointCommand;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeCommand;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.http.HttpServer;
 import org.apache.hadoop.util.Daemon;

 /**
  * The Checkpointer is responsible for supporting periodic checkpoints
  * of the HDFS metadata.
  *
  * The Checkpointer is a daemon that periodically wakes up
  * up (determined by the schedule specified in the configuration),
  * triggers a periodic checkpoint and then goes back to sleep.
  *
  * The start of a checkpoint is triggered by one of the two factors:
  * (1) time or (2) the size of the edits file.
  */
 class Checkpointer extends Daemon {
   public static final Log LOG =
     LogFactory.getLog(Checkpointer.class.getName());

   private BackupNode backupNode;
   volatile boolean shouldRun;
   private long checkpointPeriod;	// in seconds
   private long checkpointSize;    // size (in MB) of current Edit Log

   private BackupStorage getFSImage() {
     return (BackupStorage)backupNode.getFSImage();
   }

   private NamenodeProtocol getNamenode(){
     return backupNode.namenode;
   }

   /**
    * Create a connection to the primary namenode.
    */
   Checkpointer(Configuration conf, BackupNode bnNode)  throws IOException {
     this.backupNode = bnNode;
     try {
       initialize(conf);
     } catch(IOException e) {
       shutdown();
       throw e;
     }
   }

   /**
    * Initialize checkpoint.
    */
   private void initialize(Configuration conf) throws IOException {
     // Create connection to the namenode.
     shouldRun = true;

     // Initialize other scheduling parameters from the configuration
     checkpointPeriod = conf.getLong(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_PERIOD_KEY,
                                     DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_PERIOD_DEFAULT);
     checkpointSize = conf.getLong(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_SIZE_KEY,
                                   DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_SIZE_DEFAULT);

     HttpServer httpServer = backupNode.httpServer;
     httpServer.setAttribute("name.system.image", getFSImage());
     httpServer.setAttribute("name.conf", conf);
     httpServer.addInternalServlet("getimage", "/getimage", GetImageServlet.class);

     LOG.info("Checkpoint Period : " + checkpointPeriod + " secs " +
              "(" + checkpointPeriod/60 + " min)");
     LOG.info("Log Size Trigger  : " + checkpointSize + " bytes " +
              "(" + checkpointSize/1024 + " KB)");
   }

   /**
    * Shut down the checkpointer.
    */
   void shutdown() {
     shouldRun = false;
     backupNode.stop();
   }

   //
   // The main work loop
   //
   public void run() {
     // Check the size of the edit log once every 5 minutes.
     long periodMSec = 5 * 60;   // 5 minutes
     if(checkpointPeriod < periodMSec) {
       periodMSec = checkpointPeriod;
     }
     periodMSec *= 1000;

     long lastCheckpointTime = 0;
     if(!backupNode.shouldCheckpointAtStartup())
       lastCheckpointTime = FSNamesystem.now();
     while(shouldRun) {
       try {
         long now = FSNamesystem.now();
         boolean shouldCheckpoint = false;
         if(now >= lastCheckpointTime + periodMSec) {
           shouldCheckpoint = true;
         } else {
           long size = getJournalSize();
           if(size >= checkpointSize)
             shouldCheckpoint = true;
         }
         if(shouldCheckpoint) {
           doCheckpoint();
           lastCheckpointTime = now;
         }
       } catch(IOException e) {
         LOG.error("Exception in doCheckpoint: ", e);
       } catch(Throwable e) {
         LOG.error("Throwable Exception in doCheckpoint: ", e);
         shutdown();
         break;
       }
       try {
         Thread.sleep(periodMSec);
       } catch(InterruptedException ie) {
         // do nothing
       }
     }
   }

   private long getJournalSize() throws IOException {
     // If BACKUP node has been loaded
     // get edits size from the local file. ACTIVE has the same.
     if(backupNode.isRole(NamenodeRole.BACKUP)
         && getFSImage().getEditLog().isOpen())
       return backupNode.journalSize();
     // Go to the ACTIVE node for its size
     return getNamenode().journalSize(backupNode.getRegistration());
   }

   /**
    * Download <code>fsimage</code> and <code>edits</code>
    * files from the remote name-node.
    */
   private void downloadCheckpoint(CheckpointSignature sig) throws IOException {
     // Retrieve image file
     String fileid = "getimage=1";
     Collection<File> list = getFSImage().getFiles(NameNodeFile.IMAGE,
         NameNodeDirType.IMAGE);
     File[] files = list.toArray(new File[list.size()]);
     assert files.length > 0 : "No checkpoint targets.";
     String nnHttpAddr = backupNode.nnHttpAddress;
     TransferFsImage.getFileClient(nnHttpAddr, fileid, files);
     LOG.info("Downloaded file " + files[0].getName() + " size " +
              files[0].length() + " bytes.");

     // Retrieve edits file
     fileid = "getedit=1";
     list = getFSImage().getFiles(NameNodeFile.EDITS, NameNodeDirType.EDITS);
     files = list.toArray(new File[list.size()]);
     assert files.length > 0 : "No checkpoint targets.";
     TransferFsImage.getFileClient(nnHttpAddr, fileid, files);
     LOG.info("Downloaded file " + files[0].getName() + " size " +
         files[0].length() + " bytes.");
   }

   /**
    * Copy the new image into remote name-node.
    */
   private void uploadCheckpoint(CheckpointSignature sig) throws IOException {
     InetSocketAddress httpSocAddr = backupNode.getHttpAddress();
     int httpPort = httpSocAddr.getPort();
     String fileid = "putimage=1&port=" + httpPort +
       "&machine=" +
       InetAddress.getLocalHost().getHostAddress() +
       "&token=" + sig.toString();
     LOG.info("Posted URL " + backupNode.nnHttpAddress + fileid);
     TransferFsImage.getFileClient(backupNode.nnHttpAddress, fileid, (File[])null);
   }

   /**
    * Create a new checkpoint
    */
   void doCheckpoint() throws IOException {
     long startTime = FSNamesystem.now();
     NamenodeCommand cmd =
       getNamenode().startCheckpoint(backupNode.getRegistration());
     CheckpointCommand cpCmd = null;
     switch(cmd.getAction()) {
       case NamenodeProtocol.ACT_SHUTDOWN:
         shutdown();
         throw new IOException("Name-node " + backupNode.nnRpcAddress
                                            + " requested shutdown.");
       case NamenodeProtocol.ACT_CHECKPOINT:
         cpCmd = (CheckpointCommand)cmd;
         break;
       default:
         throw new IOException("Unsupported NamenodeCommand: "+cmd.getAction());
     }

     CheckpointSignature sig = cpCmd.getSignature();
     assert FSConstants.LAYOUT_VERSION == sig.getLayoutVersion() :
       "Signature should have current layout version. Expected: "
       + FSConstants.LAYOUT_VERSION + " actual "+ sig.getLayoutVersion();
     assert !backupNode.isRole(NamenodeRole.CHECKPOINT) ||
       cpCmd.isImageObsolete() : "checkpoint node should always download image.";
     backupNode.setCheckpointState(CheckpointStates.UPLOAD_START);
     if(cpCmd.isImageObsolete()) {
       // First reset storage on disk and memory state
       backupNode.resetNamespace();
       downloadCheckpoint(sig);
     }

     BackupStorage bnImage = getFSImage();
     bnImage.loadCheckpoint(sig);
     sig.validateStorageInfo(bnImage);
     bnImage.saveCheckpoint();

     if(cpCmd.needToReturnImage())
       uploadCheckpoint(sig);

     getNamenode().endCheckpoint(backupNode.getRegistration(), sig);

     bnImage.convergeJournalSpool();
     backupNode.setRegistration(); // keep registration up to date
     if(backupNode.isRole(NamenodeRole.CHECKPOINT))
         getFSImage().getEditLog().close();
     LOG.info("Checkpoint completed in "
         + (FSNamesystem.now() - startTime)/1000 + " seconds."
         +	" New Image Size: " + bnImage.getFsImageName().length());
   }
 }
	/**
	* 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 java.io.IOException;
	import java.io.File;
	import java.net.InetAddress;
	import java.net.InetSocketAddress;
	import java.util.Collection;

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

	import org.apache.hadoop.hdfs.protocol.FSConstants;
	import org.apache.hadoop.hdfs.server.common.HdfsConstants.NamenodeRole;
	import org.apache.hadoop.hdfs.server.namenode.FSImage.CheckpointStates;
	import org.apache.hadoop.hdfs.server.namenode.FSImage.NameNodeDirType;
	import org.apache.hadoop.hdfs.server.namenode.FSImage.NameNodeFile;
	import org.apache.hadoop.hdfs.server.protocol.CheckpointCommand;
	import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
	import org.apache.hadoop.hdfs.server.protocol.NamenodeCommand;
	import org.apache.hadoop.hdfs.DFSConfigKeys;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.http.HttpServer;
	import org.apache.hadoop.util.Daemon;

	/**
	* The Checkpointer is responsible for supporting periodic checkpoints
	* of the HDFS metadata.
	*
	* The Checkpointer is a daemon that periodically wakes up
	* up (determined by the schedule specified in the configuration),
	* triggers a periodic checkpoint and then goes back to sleep.
	*
	* The start of a checkpoint is triggered by one of the two factors:
	* (1) time or (2) the size of the edits file.
	*/
	class Checkpointer extends Daemon {
	public static final Log LOG =
	LogFactory.getLog(Checkpointer.class.getName());

	private BackupNode backupNode;
	volatile boolean shouldRun;
	private long checkpointPeriod; // in seconds
	private long checkpointSize; // size (in MB) of current Edit Log

	private BackupStorage getFSImage() {
	return (BackupStorage)backupNode.getFSImage();
	}

	private NamenodeProtocol getNamenode(){
	return backupNode.namenode;
	}

	/**
	* Create a connection to the primary namenode.
	*/
	Checkpointer(Configuration conf, BackupNode bnNode) throws IOException {
	this.backupNode = bnNode;
	try {
	initialize(conf);
	} catch(IOException e) {
	shutdown();
	throw e;
	}
	}

	/**
	* Initialize checkpoint.
	*/
	private void initialize(Configuration conf) throws IOException {
	// Create connection to the namenode.
	shouldRun = true;

	// Initialize other scheduling parameters from the configuration
	checkpointPeriod = conf.getLong(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_PERIOD_KEY,
	DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_PERIOD_DEFAULT);
	checkpointSize = conf.getLong(DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_SIZE_KEY,
	DFSConfigKeys.DFS_NAMENODE_CHECKPOINT_SIZE_DEFAULT);

	HttpServer httpServer = backupNode.httpServer;
	httpServer.setAttribute("name.system.image", getFSImage());
	httpServer.setAttribute("name.conf", conf);
	httpServer.addInternalServlet("getimage", "/getimage", GetImageServlet.class);

	LOG.info("Checkpoint Period : " + checkpointPeriod + " secs " +
	"(" + checkpointPeriod/60 + " min)");
	LOG.info("Log Size Trigger : " + checkpointSize + " bytes " +
	"(" + checkpointSize/1024 + " KB)");
	}

	/**
	* Shut down the checkpointer.
	*/
	void shutdown() {
	shouldRun = false;
	backupNode.stop();
	}

	//
	// The main work loop
	//
	public void run() {
	// Check the size of the edit log once every 5 minutes.
	long periodMSec = 5 * 60; // 5 minutes
	if(checkpointPeriod < periodMSec) {
	periodMSec = checkpointPeriod;
	}
	periodMSec *= 1000;

	long lastCheckpointTime = 0;
	if(!backupNode.shouldCheckpointAtStartup())
	lastCheckpointTime = FSNamesystem.now();
	while(shouldRun) {
	try {
	long now = FSNamesystem.now();
	boolean shouldCheckpoint = false;
	if(now >= lastCheckpointTime + periodMSec) {
	shouldCheckpoint = true;
	} else {
	long size = getJournalSize();
	if(size >= checkpointSize)
	shouldCheckpoint = true;
	}
	if(shouldCheckpoint) {
	doCheckpoint();
	lastCheckpointTime = now;
	}
	} catch(IOException e) {
	LOG.error("Exception in doCheckpoint: ", e);
	} catch(Throwable e) {
	LOG.error("Throwable Exception in doCheckpoint: ", e);
	shutdown();
	break;
	}
	try {
	Thread.sleep(periodMSec);
	} catch(InterruptedException ie) {
	// do nothing
	}
	}
	}

	private long getJournalSize() throws IOException {
	// If BACKUP node has been loaded
	// get edits size from the local file. ACTIVE has the same.
	if(backupNode.isRole(NamenodeRole.BACKUP)
	&& getFSImage().getEditLog().isOpen())
	return backupNode.journalSize();
	// Go to the ACTIVE node for its size
	return getNamenode().journalSize(backupNode.getRegistration());
	}

	/**
	* Download <code>fsimage</code> and <code>edits</code>
	* files from the remote name-node.
	*/
	private void downloadCheckpoint(CheckpointSignature sig) throws IOException {
	// Retrieve image file
	String fileid = "getimage=1";
	Collection<File> list = getFSImage().getFiles(NameNodeFile.IMAGE,
	NameNodeDirType.IMAGE);
	File[] files = list.toArray(new File[list.size()]);
	assert files.length > 0 : "No checkpoint targets.";
	String nnHttpAddr = backupNode.nnHttpAddress;
	TransferFsImage.getFileClient(nnHttpAddr, fileid, files);
	LOG.info("Downloaded file " + files[0].getName() + " size " +
	files[0].length() + " bytes.");

	// Retrieve edits file
	fileid = "getedit=1";
	list = getFSImage().getFiles(NameNodeFile.EDITS, NameNodeDirType.EDITS);
	files = list.toArray(new File[list.size()]);
	assert files.length > 0 : "No checkpoint targets.";
	TransferFsImage.getFileClient(nnHttpAddr, fileid, files);
	LOG.info("Downloaded file " + files[0].getName() + " size " +
	files[0].length() + " bytes.");
	}

	/**
	* Copy the new image into remote name-node.
	*/
	private void uploadCheckpoint(CheckpointSignature sig) throws IOException {
	InetSocketAddress httpSocAddr = backupNode.getHttpAddress();
	int httpPort = httpSocAddr.getPort();
	String fileid = "putimage=1&port=" + httpPort +
	"&machine=" +
	InetAddress.getLocalHost().getHostAddress() +
	"&token=" + sig.toString();
	LOG.info("Posted URL " + backupNode.nnHttpAddress + fileid);
	TransferFsImage.getFileClient(backupNode.nnHttpAddress, fileid, (File[])null);
	}

	/**
	* Create a new checkpoint
	*/
	void doCheckpoint() throws IOException {
	long startTime = FSNamesystem.now();
	NamenodeCommand cmd =
	getNamenode().startCheckpoint(backupNode.getRegistration());
	CheckpointCommand cpCmd = null;
	switch(cmd.getAction()) {
	case NamenodeProtocol.ACT_SHUTDOWN:
	shutdown();
	throw new IOException("Name-node " + backupNode.nnRpcAddress
	+ " requested shutdown.");
	case NamenodeProtocol.ACT_CHECKPOINT:
	cpCmd = (CheckpointCommand)cmd;
	break;
	default:
	throw new IOException("Unsupported NamenodeCommand: "+cmd.getAction());
	}

	CheckpointSignature sig = cpCmd.getSignature();
	assert FSConstants.LAYOUT_VERSION == sig.getLayoutVersion() :
	"Signature should have current layout version. Expected: "
	+ FSConstants.LAYOUT_VERSION + " actual "+ sig.getLayoutVersion();
	assert !backupNode.isRole(NamenodeRole.CHECKPOINT) \|\|
	cpCmd.isImageObsolete() : "checkpoint node should always download image.";
	backupNode.setCheckpointState(CheckpointStates.UPLOAD_START);
	if(cpCmd.isImageObsolete()) {
	// First reset storage on disk and memory state
	backupNode.resetNamespace();
	downloadCheckpoint(sig);
	}

	BackupStorage bnImage = getFSImage();
	bnImage.loadCheckpoint(sig);
	sig.validateStorageInfo(bnImage);
	bnImage.saveCheckpoint();

	if(cpCmd.needToReturnImage())
	uploadCheckpoint(sig);

	getNamenode().endCheckpoint(backupNode.getRegistration(), sig);

	bnImage.convergeJournalSpool();
	backupNode.setRegistration(); // keep registration up to date
	if(backupNode.isRole(NamenodeRole.CHECKPOINT))
	getFSImage().getEditLog().close();
	LOG.info("Checkpoint completed in "
	+ (FSNamesystem.now() - startTime)/1000 + " seconds."
	+ " New Image Size: " + bnImage.getFsImageName().length());
	}
	}