|
|
@@ -1,981 +0,0 @@
|
|
|
-/**
|
|
|
- * 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.dfs;
|
|
|
-
|
|
|
-import java.io.*;
|
|
|
-import java.util.*;
|
|
|
-import java.util.concurrent.ExecutorService;
|
|
|
-import java.util.concurrent.Executors;
|
|
|
-import java.util.concurrent.TimeUnit;
|
|
|
-import java.util.concurrent.atomic.AtomicInteger;
|
|
|
-import java.util.concurrent.atomic.AtomicBoolean;
|
|
|
-import java.util.concurrent.atomic.AtomicLong;
|
|
|
-import java.net.SocketTimeoutException;
|
|
|
-import java.net.InetSocketAddress;
|
|
|
-
|
|
|
-import org.apache.commons.logging.*;
|
|
|
-import org.apache.hadoop.io.Text;
|
|
|
-import org.apache.hadoop.io.retry.*;
|
|
|
-import org.apache.hadoop.util.StringUtils;
|
|
|
-import org.apache.hadoop.util.Daemon;
|
|
|
-import org.apache.hadoop.conf.Configuration;
|
|
|
-import org.apache.hadoop.ipc.RPC;
|
|
|
-
|
|
|
-/**
|
|
|
- * This class associates a block generation stamp with with block. This
|
|
|
- * generation stamp is written to each metadata file. Please see
|
|
|
- * HADOOP-1700 for details.
|
|
|
- */
|
|
|
-class GenerationStampUpgradeDatanode extends UpgradeObjectDatanode {
|
|
|
-
|
|
|
- public static final Log LOG =
|
|
|
- LogFactory.getLog("org.apache.hadoop.dfs.GenerationStampUpgrade");
|
|
|
-
|
|
|
- DatanodeProtocol namenode;
|
|
|
- InetSocketAddress namenodeAddr;
|
|
|
-
|
|
|
- // stats
|
|
|
- private AtomicInteger blocksPreviouslyUpgraded = new AtomicInteger(0);
|
|
|
- private AtomicInteger blocksToUpgrade = new AtomicInteger(0);
|
|
|
- private AtomicInteger blocksUpgraded = new AtomicInteger(0);
|
|
|
- private AtomicInteger errors = new AtomicInteger(0);
|
|
|
-
|
|
|
- // process the upgrade using a pool of threads.
|
|
|
- static private final int poolSize = 4;
|
|
|
-
|
|
|
- // If no progress has occured during this time, print warnings message.
|
|
|
- static private final int LONG_TIMEOUT_MILLISEC = 1*60*1000; // 1 minute
|
|
|
-
|
|
|
- // This object is needed to indicate that namenode is not running upgrade.
|
|
|
- static UpgradeCommand noUpgradeOnNamenode = new UpgradeCommand();
|
|
|
-
|
|
|
- private List<UpgradeExecutor> completedList = new LinkedList<UpgradeExecutor>();
|
|
|
-
|
|
|
- /* This is set when the datanode misses the regular upgrade.
|
|
|
- * When this is set, it upgrades the block but stops heartbeating
|
|
|
- * to the namenode.
|
|
|
- */
|
|
|
- private AtomicBoolean offlineUpgrade = new AtomicBoolean(false);
|
|
|
- private AtomicBoolean upgradeCompleted = new AtomicBoolean(false);
|
|
|
-
|
|
|
- // Implement the common interfaces required by UpgradeObjectDatanode
|
|
|
-
|
|
|
- public int getVersion() {
|
|
|
- return GenerationStampUpgradeNamenode.PRE_GENERATIONSTAMP_LAYOUT_VERSION;
|
|
|
- }
|
|
|
-
|
|
|
- /*
|
|
|
- * Start upgrade if it not already running. It sends status to
|
|
|
- * namenode even if an upgrade is already in progress.
|
|
|
- */
|
|
|
- public synchronized UpgradeCommand startUpgrade() throws IOException {
|
|
|
- if (offlineUpgrade.get()) {
|
|
|
- doUpgrade();
|
|
|
- }
|
|
|
- return null;
|
|
|
- }
|
|
|
-
|
|
|
- public String getDescription() {
|
|
|
- return "Block Generation Stamp Upgrade at Datanode";
|
|
|
- }
|
|
|
-
|
|
|
- public short getUpgradeStatus() {
|
|
|
- return (blocksToUpgrade.get() == blocksUpgraded.get()) ? 100 :
|
|
|
- (short) Math.floor(blocksUpgraded.get()*100.0/blocksToUpgrade.get());
|
|
|
- }
|
|
|
-
|
|
|
- public UpgradeCommand completeUpgrade() throws IOException {
|
|
|
- // return latest stats command.
|
|
|
- assert getUpgradeStatus() == 100;
|
|
|
- return new DatanodeStatsCommand(getUpgradeStatus(),
|
|
|
- getDatanode().dnRegistration,
|
|
|
- blocksPreviouslyUpgraded.get() + blocksUpgraded.get(),
|
|
|
- blocksToUpgrade.get()-blocksUpgraded.get(),
|
|
|
- errors.get(),
|
|
|
- GenerationStampUpgradeNamenode.PRE_GENERATIONSTAMP_LAYOUT_VERSION);
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- boolean preUpgradeAction(NamespaceInfo nsInfo) throws IOException {
|
|
|
- int nsUpgradeVersion = nsInfo.getDistributedUpgradeVersion();
|
|
|
- if(nsUpgradeVersion >= getVersion()) {
|
|
|
- return false; // Normal upgrade.
|
|
|
- }
|
|
|
-
|
|
|
- LOG.info("\n This Datanode has missed a cluster wide Block generation Stamp Upgrade." +
|
|
|
- "\n Will perform an 'offline' upgrade of the blocks." +
|
|
|
- "\n During this time, Datanode does not heartbeat.");
|
|
|
-
|
|
|
-
|
|
|
- // Namenode removes this node from the registered nodes
|
|
|
- try {
|
|
|
- getDatanode().namenode.errorReport(getDatanode().dnRegistration,
|
|
|
- DatanodeProtocol.NOTIFY,
|
|
|
- "Performing an offline generation stamp " +
|
|
|
- "upgrade. " +
|
|
|
- "Will be back online once the ugprade " +
|
|
|
- "completes. Please see datanode logs.");
|
|
|
-
|
|
|
- } catch(IOException ignored) {
|
|
|
- LOG.info("\n This Datanode was unable to send error report to namenode.");
|
|
|
- }
|
|
|
- offlineUpgrade.set(true);
|
|
|
- return true;
|
|
|
- }
|
|
|
-
|
|
|
- public GenerationStampUpgradeDatanode() {
|
|
|
- blocksPreviouslyUpgraded.set(0);
|
|
|
- blocksToUpgrade.set(0);
|
|
|
- blocksUpgraded.set(0);
|
|
|
- errors.set(0);
|
|
|
- }
|
|
|
-
|
|
|
- static File getPreGenerationMetaFile(File f) {
|
|
|
- return new File(f.getAbsolutePath() + FSDataset.METADATA_EXTENSION);
|
|
|
- }
|
|
|
-
|
|
|
- // This class is invoked by the worker thread to convert the
|
|
|
- // metafile into the new format
|
|
|
- //
|
|
|
- class UpgradeExecutor implements Runnable {
|
|
|
- Block block;
|
|
|
- Throwable throwable;
|
|
|
-
|
|
|
- UpgradeExecutor(Block b) {
|
|
|
- block = b;
|
|
|
- }
|
|
|
-
|
|
|
- public void run() {
|
|
|
- try {
|
|
|
- // do the real work here
|
|
|
- FSDataset dataset = (FSDataset) getDatanode().data;
|
|
|
- upgradeToCurVersion(dataset, block);
|
|
|
- } catch (Throwable t) {
|
|
|
- throwable = t;
|
|
|
- }
|
|
|
- synchronized (completedList) {
|
|
|
- completedList.add(this);
|
|
|
- completedList.notify();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * Upgrades the metadata file to current version if required.
|
|
|
- * @param dataset
|
|
|
- * @param block
|
|
|
- */
|
|
|
- void upgradeToCurVersion(FSDataset dataset, Block block)
|
|
|
- throws IOException {
|
|
|
- File blockFile = dataset.getBlockFile(block);
|
|
|
- if (blockFile == null) {
|
|
|
- throw new IOException("Could find file for " + block);
|
|
|
- }
|
|
|
-
|
|
|
- File metadataFile = dataset.getMetaFile(block);
|
|
|
- File oldmetadataFile = getPreGenerationMetaFile(blockFile);
|
|
|
-
|
|
|
- if (metadataFile.exists() && oldmetadataFile.exists()) {
|
|
|
- //
|
|
|
- // If both file exists and are of the same size,
|
|
|
- // then delete the old one. If the sizes are not same then
|
|
|
- // leave both of them and consider the upgrade as successful.
|
|
|
- //
|
|
|
- if (metadataFile.length() == oldmetadataFile.length()) {
|
|
|
- if (!oldmetadataFile.delete()) {
|
|
|
- LOG.info("Unable to delete old metadata file " + oldmetadataFile);
|
|
|
- }
|
|
|
- }
|
|
|
- } else if (metadataFile.exists()) {
|
|
|
- //
|
|
|
- // Only the new file exists, nothing more to do.
|
|
|
- //
|
|
|
- return;
|
|
|
- } else if (oldmetadataFile.exists()) {
|
|
|
- //
|
|
|
- // The old file exists but the new one is missing. Rename
|
|
|
- // old one to new name.
|
|
|
- //
|
|
|
- if (!oldmetadataFile.renameTo(metadataFile)) {
|
|
|
- throw new IOException("Could find rename " + oldmetadataFile +
|
|
|
- " to " + metadataFile);
|
|
|
- }
|
|
|
- } else {
|
|
|
- throw new IOException("Could find any metadata file for " + block);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // This method iterates through all the blocks on a datanode and
|
|
|
- // do the upgrade.
|
|
|
- //
|
|
|
- void doUpgrade() throws IOException {
|
|
|
-
|
|
|
- if (upgradeCompleted.get()) {
|
|
|
- assert offlineUpgrade.get() :
|
|
|
- ("Multiple calls to doUpgrade is expected only during " +
|
|
|
- "offline upgrade");
|
|
|
- return;
|
|
|
- }
|
|
|
-
|
|
|
- FSDataset dataset = (FSDataset) getDatanode().data;
|
|
|
-
|
|
|
- // Set up the retry policy so that each attempt waits for one minute.
|
|
|
- Configuration conf = new Configuration();
|
|
|
- // set rpc timeout to one minute.
|
|
|
- conf.set("ipc.client.timeout", "60000");
|
|
|
-
|
|
|
- RetryPolicy timeoutPolicy =
|
|
|
- RetryPolicies.retryUpToMaximumCountWithFixedSleep(
|
|
|
- LONG_TIMEOUT_MILLISEC/1000,
|
|
|
- 1, TimeUnit.MILLISECONDS);
|
|
|
-
|
|
|
- Map<Class<? extends Exception>,RetryPolicy> exceptionToPolicyMap =
|
|
|
- new HashMap<Class<? extends Exception>, RetryPolicy>();
|
|
|
- exceptionToPolicyMap.put(SocketTimeoutException.class, timeoutPolicy);
|
|
|
- RetryPolicy methodPolicy = RetryPolicies.retryByException(
|
|
|
- RetryPolicies.TRY_ONCE_THEN_FAIL, exceptionToPolicyMap);
|
|
|
- Map<String,RetryPolicy> methodNameToPolicyMap =
|
|
|
- new HashMap<String, RetryPolicy>();
|
|
|
- // do we need to set the policy for connection failures also?
|
|
|
- methodNameToPolicyMap.put("processUpgradeCommand", methodPolicy);
|
|
|
-
|
|
|
- LOG.info("Starting Block Generation Stamp Upgrade on datanode " +
|
|
|
- getDatanode());
|
|
|
-
|
|
|
- for (;;) {
|
|
|
- try {
|
|
|
- namenodeAddr = getDatanode().getNameNodeAddr();
|
|
|
- namenode = (DatanodeProtocol) RetryProxy.create(
|
|
|
- DatanodeProtocol.class,
|
|
|
- RPC.waitForProxy(DatanodeProtocol.class,
|
|
|
- DatanodeProtocol.versionID,
|
|
|
- namenodeAddr,
|
|
|
- conf),
|
|
|
- methodNameToPolicyMap);
|
|
|
- break;
|
|
|
- } catch (IOException e) {
|
|
|
- LOG.warn("Generation Stamp Upgrade Exception " +
|
|
|
- "while trying to connect to NameNode at " +
|
|
|
- getDatanode().getNameNodeAddr().toString() + " : " +
|
|
|
- StringUtils.stringifyException(e));
|
|
|
- try {
|
|
|
- Thread.sleep(10*1000);
|
|
|
- } catch (InterruptedException e1) {
|
|
|
- throw new IOException("Interrupted Sleep while creating RPC proxy." +
|
|
|
- e1);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- LOG.info("Block Generation Stamp Upgrade Datanode connected to " +
|
|
|
- "namenode at " + namenodeAddr);
|
|
|
-
|
|
|
- // Get a list of all the blocks :
|
|
|
- LinkedList<UpgradeExecutor> blockList = new LinkedList<UpgradeExecutor>();
|
|
|
-
|
|
|
- //Fill blockList with blocks to be upgraded.
|
|
|
- Block [] blockArr = dataset.getBlockReport();
|
|
|
-
|
|
|
- for (Block b : blockArr) {
|
|
|
- File blockFile = null;
|
|
|
- try {
|
|
|
- blockFile = dataset.getBlockFile(b);
|
|
|
- } catch (IOException e) {
|
|
|
- //The block might just be deleted. ignore it.
|
|
|
- LOG.warn("Could not find file location for " + b +
|
|
|
- ". It might already be deleted. Exception : " +
|
|
|
- StringUtils.stringifyException(e));
|
|
|
- errors.getAndIncrement();
|
|
|
- continue;
|
|
|
- }
|
|
|
- if (!blockFile.exists()) {
|
|
|
- errors.getAndIncrement();
|
|
|
- LOG.error("could not find block file " + blockFile);
|
|
|
- continue;
|
|
|
- }
|
|
|
- File metaFile = dataset.getMetaFile(b);
|
|
|
- File oldMetaFile = getPreGenerationMetaFile(blockFile);
|
|
|
- if (metaFile.exists()) {
|
|
|
- blocksPreviouslyUpgraded.getAndIncrement();
|
|
|
- continue;
|
|
|
- }
|
|
|
- blocksToUpgrade.getAndIncrement();
|
|
|
- blockList.add(new UpgradeExecutor(b));
|
|
|
- }
|
|
|
- blockArr = null;
|
|
|
- int nLeft = blockList.size();
|
|
|
-
|
|
|
- LOG.info("Starting upgrade of " + blocksToUpgrade.get() + " blocks out of " +
|
|
|
- (blocksToUpgrade.get() + blocksPreviouslyUpgraded.get()));
|
|
|
-
|
|
|
- // Start the pool of upgrade workers
|
|
|
- ExecutorService pool = Executors.newFixedThreadPool(poolSize);
|
|
|
- for (Iterator<UpgradeExecutor> it = blockList.iterator(); it.hasNext();) {
|
|
|
- pool.submit(it.next());
|
|
|
- }
|
|
|
-
|
|
|
- // Inform the namenode
|
|
|
- sendStatus();
|
|
|
-
|
|
|
- // Report status to namenode every so many seconds:
|
|
|
- long now = System.currentTimeMillis();
|
|
|
- long statusReportIntervalMilliSec = 30*1000;
|
|
|
- long lastStatusReportTime = now;
|
|
|
- long lastUpdateTime = now;
|
|
|
- long lastWarnTime = now;
|
|
|
-
|
|
|
- // Now wait for the tasks to complete.
|
|
|
- //
|
|
|
- while (nLeft > 0) {
|
|
|
- synchronized (completedList) {
|
|
|
- if (completedList.size() <= 0) {
|
|
|
- try {
|
|
|
- completedList.wait(1000);
|
|
|
- } catch (InterruptedException ignored) {}
|
|
|
- }
|
|
|
-
|
|
|
- now = System.currentTimeMillis();
|
|
|
-
|
|
|
- if (completedList.size()> 0) {
|
|
|
- UpgradeExecutor exe = completedList.remove(0);
|
|
|
- nLeft--;
|
|
|
- if (exe.throwable != null) {
|
|
|
- errors.getAndIncrement();
|
|
|
- LOG.error("Got an exception during generation stamp upgrade of " +
|
|
|
- exe.block + ": " +
|
|
|
- StringUtils.stringifyException(exe.throwable));
|
|
|
- }
|
|
|
- blocksUpgraded.getAndIncrement();
|
|
|
- lastUpdateTime = now;
|
|
|
- } else {
|
|
|
- if ((now - lastUpdateTime) >= LONG_TIMEOUT_MILLISEC &&
|
|
|
- (now - lastWarnTime) >= LONG_TIMEOUT_MILLISEC) {
|
|
|
- lastWarnTime = now;
|
|
|
- LOG.warn("No block was updated in last " +
|
|
|
- (LONG_TIMEOUT_MILLISEC/(60*1000)) +
|
|
|
- " minutes! will keep waiting... ");
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- if ((now-lastStatusReportTime) > statusReportIntervalMilliSec) {
|
|
|
- sendStatus();
|
|
|
- lastStatusReportTime = System.currentTimeMillis();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- pool.shutdown();
|
|
|
- upgradeCompleted.set(true);
|
|
|
-
|
|
|
- LOG.info("Completed Block Generation Stamp Upgrade. Total of " +
|
|
|
- (blocksPreviouslyUpgraded.get() + blocksToUpgrade.get()) +
|
|
|
- " blocks : " + blocksPreviouslyUpgraded.get() + " blocks previously " +
|
|
|
- "upgraded, " + blocksUpgraded.get() + " blocks upgraded this time " +
|
|
|
- "with " + errors.get() + " errors.");
|
|
|
-
|
|
|
- // now inform the name node about the completion.
|
|
|
- // What if there is no upgrade running on Namenode now?
|
|
|
- while (!sendStatus());
|
|
|
-
|
|
|
- }
|
|
|
-
|
|
|
- /** Sends current status and stats to namenode and logs it to local log*/
|
|
|
- boolean sendStatus() {
|
|
|
- LOG.info((offlineUpgrade.get() ? "Offline " : "") +
|
|
|
- "Block Generation Stamp Upgrade : " +
|
|
|
- getUpgradeStatus() + "% completed.");
|
|
|
- if (offlineUpgrade.get()) {
|
|
|
- return true;
|
|
|
- }
|
|
|
-
|
|
|
- DatanodeStatsCommand cmd = null;
|
|
|
- synchronized (this) {
|
|
|
- cmd = new DatanodeStatsCommand(getUpgradeStatus(),
|
|
|
- getDatanode().dnRegistration,
|
|
|
- blocksPreviouslyUpgraded.get() + blocksUpgraded.get(),
|
|
|
- blocksToUpgrade.get()-blocksUpgraded.get(),
|
|
|
- errors.get(),
|
|
|
- GenerationStampUpgradeNamenode.PRE_GENERATIONSTAMP_LAYOUT_VERSION);
|
|
|
- }
|
|
|
- UpgradeCommand reply = sendCommand(namenodeAddr, namenode, cmd, 0);
|
|
|
- if (reply == null) {
|
|
|
- LOG.warn("Could not send status to Namenode. Namenode might be " +
|
|
|
- "over loaded or down.");
|
|
|
- }
|
|
|
- return reply != null;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- // Sends a command to the namenode
|
|
|
- static UpgradeCommand sendCommand(InetSocketAddress namenodeAddr,
|
|
|
- DatanodeProtocol namenode,
|
|
|
- UpgradeCommand cmd, int retries) {
|
|
|
- for(int i=0; i<=retries || retries<0; i++) {
|
|
|
- try {
|
|
|
- UpgradeCommand reply = namenode.processUpgradeCommand(cmd);
|
|
|
- if (reply == null) {
|
|
|
- /* namenode might not be running upgrade or finished
|
|
|
- * an upgrade. We just return a static object */
|
|
|
- return noUpgradeOnNamenode;
|
|
|
- }
|
|
|
- return reply;
|
|
|
- } catch (IOException e) {
|
|
|
- // print the stack trace only for the last retry.
|
|
|
- LOG.warn("Exception to " + namenodeAddr +
|
|
|
- " while sending command " +
|
|
|
- cmd.getAction() + ": " + e +
|
|
|
- ((retries<0 || i>=retries)? "... will retry ..." :
|
|
|
- ": " + StringUtils.stringifyException(e)));
|
|
|
- }
|
|
|
- }
|
|
|
- return null;
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-/**
|
|
|
- * Once an upgrade starts at the namenode , this class manages the upgrade
|
|
|
- * process.
|
|
|
- */
|
|
|
-class GenerationStampUpgradeNamenode extends UpgradeObjectNamenode {
|
|
|
-
|
|
|
- public static final Log LOG =
|
|
|
- LogFactory.getLog("org.apache.hadoop.dfs.GenerationStampUpgradeNamenode");
|
|
|
-
|
|
|
- static final long inactivityExtension = 10*1000; // 10 seconds
|
|
|
- AtomicLong lastNodeCompletionTime = new AtomicLong(0);
|
|
|
-
|
|
|
- // The layout version before the generation stamp upgrade.
|
|
|
- static final int PRE_GENERATIONSTAMP_LAYOUT_VERSION = -13;
|
|
|
-
|
|
|
- static final int DN_CMD_STATS = 300;
|
|
|
-
|
|
|
- enum UpgradeStatus {
|
|
|
- INITIALIZED,
|
|
|
- STARTED,
|
|
|
- DATANODES_DONE,
|
|
|
- COMPLETED,
|
|
|
- }
|
|
|
-
|
|
|
- UpgradeStatus upgradeStatus = UpgradeStatus.INITIALIZED;
|
|
|
-
|
|
|
- class DnInfo {
|
|
|
- short percentCompleted = 0;
|
|
|
- long blocksUpgraded = 0;
|
|
|
- long blocksRemaining = 0;
|
|
|
- long errors = 0;
|
|
|
-
|
|
|
- DnInfo(short pcCompleted) {
|
|
|
- percentCompleted = status;
|
|
|
- }
|
|
|
- DnInfo() {}
|
|
|
-
|
|
|
- void setStats(DatanodeStatsCommand cmd) {
|
|
|
- percentCompleted = cmd.getCurrentStatus();
|
|
|
- blocksUpgraded = cmd.blocksUpgraded;
|
|
|
- blocksRemaining = cmd.blocksRemaining;
|
|
|
- errors = cmd.errors;
|
|
|
- }
|
|
|
-
|
|
|
- boolean isDone() {
|
|
|
- return percentCompleted >= 100;
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- /* We should track only the storageIDs and not DatanodeID, which
|
|
|
- * includes datanode name and storage id.
|
|
|
- */
|
|
|
- HashMap<DatanodeID, DnInfo> dnMap = new HashMap<DatanodeID, DnInfo>();
|
|
|
- HashMap<DatanodeID, DnInfo> unfinishedDnMap =
|
|
|
- new HashMap<DatanodeID, DnInfo>();
|
|
|
-
|
|
|
- Daemon monitorThread;
|
|
|
- double avgDatanodeCompletionPct = 0;
|
|
|
- boolean forceDnCompletion = false;
|
|
|
-
|
|
|
- //Upgrade object interface:
|
|
|
-
|
|
|
- public int getVersion() {
|
|
|
- return PRE_GENERATIONSTAMP_LAYOUT_VERSION;
|
|
|
- }
|
|
|
-
|
|
|
- public UpgradeCommand completeUpgrade() throws IOException {
|
|
|
- return null;
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public String getDescription() {
|
|
|
- return "Block Generation Stamp Upgrade at Namenode";
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public synchronized short getUpgradeStatus() {
|
|
|
- // Reserve 10% for deleting files.
|
|
|
- if (upgradeStatus == UpgradeStatus.COMPLETED) {
|
|
|
- return 100;
|
|
|
- }
|
|
|
- return (short) avgDatanodeCompletionPct;
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public UpgradeCommand startUpgrade() throws IOException {
|
|
|
-
|
|
|
- assert monitorThread == null;
|
|
|
- lastNodeCompletionTime.set(System.currentTimeMillis());
|
|
|
-
|
|
|
- monitorThread = new Daemon(new UpgradeMonitor());
|
|
|
- monitorThread.start();
|
|
|
- return super.startUpgrade();
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public synchronized void forceProceed() throws IOException {
|
|
|
- if (forceDnCompletion) {
|
|
|
- LOG.warn("forceProceed is already set for this upgrade. It can take " +
|
|
|
- "a short while to take affect. Please wait.");
|
|
|
- return;
|
|
|
- }
|
|
|
-
|
|
|
- LOG.info("got forceProceed request for this upgrade. Datanodes upgrade " +
|
|
|
- "will be considered done. It can take a few seconds to take " +
|
|
|
- "effect.");
|
|
|
- forceDnCompletion = true;
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- UpgradeCommand processUpgradeCommand(UpgradeCommand command)
|
|
|
- throws IOException {
|
|
|
- switch (command.getAction()) {
|
|
|
-
|
|
|
- case GenerationStampUpgradeNamenode.DN_CMD_STATS :
|
|
|
- return handleStatsCmd(command);
|
|
|
-
|
|
|
- default:
|
|
|
- throw new IOException("Unknown Command for Generation Stamp Upgrade : " +
|
|
|
- command.getAction());
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public UpgradeStatusReport getUpgradeStatusReport(boolean details)
|
|
|
- throws IOException {
|
|
|
-
|
|
|
- /* If 'details' is true should we update block level status?
|
|
|
- * It could take multiple minutes
|
|
|
- * updateBlckLevelStats()?
|
|
|
- */
|
|
|
-
|
|
|
- String replyString = "";
|
|
|
-
|
|
|
- short status = 0;
|
|
|
-
|
|
|
- synchronized (this) {
|
|
|
-
|
|
|
- status = getUpgradeStatus();
|
|
|
-
|
|
|
- replyString = String.format(
|
|
|
- ((monitorThread == null) ? "\tUpgrade has not been started yet.\n" : "")+
|
|
|
- ((forceDnCompletion) ? "\tForce Proceed is ON\n" : "") +
|
|
|
- "\tLast Block Level Stats updated at : %tc\n" +
|
|
|
- "\tLast Block Level Stats : %s\n" +
|
|
|
- "\tBrief Datanode Status : %s\n" +
|
|
|
- "%s",
|
|
|
- latestBlockLevelStats.updatedAt,
|
|
|
- latestBlockLevelStats.statusString("\n\t "),
|
|
|
- printStatus("\n\t "),
|
|
|
- ((status < 100 && upgradeStatus == UpgradeStatus.DATANODES_DONE) ?
|
|
|
- "\tNOTE: Upgrade at the Datanodes has finished. Deleteing \".crc\" " +
|
|
|
- "files\n\tcan take longer than status implies.\n" : "")
|
|
|
- );
|
|
|
-
|
|
|
- if (details) {
|
|
|
- // list all the known data nodes
|
|
|
- StringBuilder str = null;
|
|
|
- Iterator<DatanodeID> keys = dnMap.keySet().iterator();
|
|
|
- Iterator<DnInfo> values = dnMap.values().iterator();
|
|
|
-
|
|
|
- for(; keys.hasNext() && values.hasNext() ;) {
|
|
|
- DatanodeID dn = keys.next();
|
|
|
- DnInfo info = values.next();
|
|
|
- String dnStr = "\t\t" + dn.getName() + "\t : " +
|
|
|
- info.percentCompleted + " % \t" +
|
|
|
- info.blocksUpgraded + " u \t" +
|
|
|
- info.blocksRemaining + " r \t" +
|
|
|
- info.errors + " e\n";
|
|
|
- if ( str == null ) {
|
|
|
- str = new StringBuilder(dnStr.length()*
|
|
|
- (dnMap.size() + (dnMap.size()+7)/8));
|
|
|
- }
|
|
|
- str.append(dnStr);
|
|
|
- }
|
|
|
-
|
|
|
- replyString += "\n\tDatanode Stats (total: " + dnMap.size() + "): " +
|
|
|
- "pct Completion(%) blocks upgraded (u) " +
|
|
|
- "blocks remaining (r) errors (e)\n\n" +
|
|
|
- (( str == null ) ?
|
|
|
- "\t\tThere are no known Datanodes\n" : str);
|
|
|
- }
|
|
|
- }
|
|
|
- return new GenerationStampUpgradeStatusReport(
|
|
|
- PRE_GENERATIONSTAMP_LAYOUT_VERSION,
|
|
|
- status, replyString);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- /**
|
|
|
- * The namenode process a periodic statistics message from the datanode.
|
|
|
- */
|
|
|
- private synchronized UpgradeCommand handleStatsCmd(UpgradeCommand cmd) {
|
|
|
-
|
|
|
- DatanodeStatsCommand stats = (DatanodeStatsCommand)cmd;
|
|
|
-
|
|
|
- DatanodeID dn = stats.datanodeId;
|
|
|
- DnInfo dnInfo = dnMap.get(dn);
|
|
|
- boolean alreadyCompleted = (dnInfo != null && dnInfo.isDone());
|
|
|
-
|
|
|
- if (dnInfo == null) {
|
|
|
- dnInfo = new DnInfo();
|
|
|
- dnMap.put(dn, dnInfo);
|
|
|
- LOG.info("Upgrade started/resumed at datanode " + dn.getName());
|
|
|
- }
|
|
|
-
|
|
|
- dnInfo.setStats(stats);
|
|
|
- if (!dnInfo.isDone()) {
|
|
|
- unfinishedDnMap.put(dn, dnInfo);
|
|
|
- }
|
|
|
-
|
|
|
- if (dnInfo.isDone() && !alreadyCompleted) {
|
|
|
- LOG.info("upgrade completed on datanode " + dn.getName());
|
|
|
- unfinishedDnMap.remove(dn);
|
|
|
- if (unfinishedDnMap.size() == 0) {
|
|
|
- lastNodeCompletionTime.set(System.currentTimeMillis());
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- //Should we send any more info?
|
|
|
- return new UpgradeCommand();
|
|
|
- }
|
|
|
-
|
|
|
- public GenerationStampUpgradeNamenode() {
|
|
|
- }
|
|
|
-
|
|
|
- // For now we will wait for all the nodes to complete upgrade.
|
|
|
- synchronized boolean isUpgradeDone() {
|
|
|
- return upgradeStatus == UpgradeStatus.COMPLETED;
|
|
|
- }
|
|
|
-
|
|
|
- synchronized String printStatus(String spacing) {
|
|
|
- //NOTE: iterates on all the datanodes.
|
|
|
-
|
|
|
- // Calculate % completion on all the data nodes.
|
|
|
- long errors = 0;
|
|
|
- long totalCompletion = 0;
|
|
|
- for( Iterator<DnInfo> it = dnMap.values().iterator(); it.hasNext(); ) {
|
|
|
- DnInfo dnInfo = it.next();
|
|
|
- totalCompletion += dnInfo.percentCompleted;
|
|
|
- errors += dnInfo.errors;
|
|
|
- }
|
|
|
-
|
|
|
- avgDatanodeCompletionPct = totalCompletion/(dnMap.size() + 1e-20);
|
|
|
-
|
|
|
- String msg = "Avg completion of all Datanodes: " +
|
|
|
- String.format("%.2f%%", avgDatanodeCompletionPct) +
|
|
|
- " with " + errors + " errors. " +
|
|
|
- ((unfinishedDnMap.size() > 0) ? spacing +
|
|
|
- unfinishedDnMap.size() + " out of " + dnMap.size() +
|
|
|
- " nodes are not done." : "");
|
|
|
-
|
|
|
- LOG.info("Generation Stamp Upgrade is " + (isUpgradeDone() ?
|
|
|
- "complete. " : "still running. ") + spacing + msg);
|
|
|
- return msg;
|
|
|
- }
|
|
|
-
|
|
|
- private synchronized void setStatus(UpgradeStatus status) {
|
|
|
- upgradeStatus = status;
|
|
|
- }
|
|
|
-
|
|
|
- /* Checks if upgrade completed based on datanode's status and/or
|
|
|
- * if all the blocks are upgraded.
|
|
|
- */
|
|
|
- private synchronized UpgradeStatus checkOverallCompletion() {
|
|
|
-
|
|
|
- if (upgradeStatus == UpgradeStatus.COMPLETED ||
|
|
|
- upgradeStatus == UpgradeStatus.DATANODES_DONE) {
|
|
|
- return upgradeStatus;
|
|
|
- }
|
|
|
-
|
|
|
- if (upgradeStatus != UpgradeStatus.DATANODES_DONE) {
|
|
|
- boolean datanodesDone =
|
|
|
- (dnMap.size() > 0 && unfinishedDnMap.size() == 0 &&
|
|
|
- ( System.currentTimeMillis() - lastNodeCompletionTime.get() ) >
|
|
|
- inactivityExtension) || forceDnCompletion ;
|
|
|
-
|
|
|
- if ( datanodesDone ) {
|
|
|
- LOG.info("Upgrade of DataNode blocks is complete. " +
|
|
|
- ((forceDnCompletion) ? "(ForceDnCompletion is on.)" : ""));
|
|
|
- upgradeStatus = UpgradeStatus.DATANODES_DONE;
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- if (upgradeStatus != UpgradeStatus.DATANODES_DONE &&
|
|
|
- latestBlockLevelStats.updatedAt > 0) {
|
|
|
- // check if last block report marked all
|
|
|
- if (latestBlockLevelStats.minimallyReplicatedBlocks == 0 &&
|
|
|
- latestBlockLevelStats.underReplicatedBlocks == 0) {
|
|
|
-
|
|
|
- LOG.info("Marking datanode upgrade complete since all the blocks are " +
|
|
|
- "upgraded (even though some datanodes may not have " +
|
|
|
- "reported completion. Block level stats :\n\t" +
|
|
|
- latestBlockLevelStats.statusString("\n\t"));
|
|
|
- upgradeStatus = UpgradeStatus.DATANODES_DONE;
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return upgradeStatus;
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * This class monitors the upgrade progress and periodically prints
|
|
|
- * status message to log.
|
|
|
- */
|
|
|
- class UpgradeMonitor implements Runnable {
|
|
|
-
|
|
|
- static final long statusReportIntervalMillis = 1*60*1000;
|
|
|
- static final long blockReportIntervalMillis = 5*60*1000;
|
|
|
- static final int sleepTimeSec = 5;
|
|
|
-
|
|
|
- public void run() {
|
|
|
- long lastReportTime = System.currentTimeMillis();
|
|
|
- long lastBlockReportTime = lastReportTime;
|
|
|
-
|
|
|
- while ( !isUpgradeDone() ) {
|
|
|
- UpgradeStatus status = checkOverallCompletion();
|
|
|
-
|
|
|
- if (status == UpgradeStatus.DATANODES_DONE) {
|
|
|
- setStatus(UpgradeStatus.COMPLETED);
|
|
|
- }
|
|
|
-
|
|
|
- long now = System.currentTimeMillis();
|
|
|
-
|
|
|
-
|
|
|
- if (now-lastBlockReportTime >= blockReportIntervalMillis) {
|
|
|
- updateBlockLevelStats();
|
|
|
- // Check if all the blocks have been upgraded.
|
|
|
- lastBlockReportTime = now;
|
|
|
- }
|
|
|
-
|
|
|
- if ((now - lastReportTime) >= statusReportIntervalMillis ||
|
|
|
- isUpgradeDone()) {
|
|
|
- printStatus("\n\t");
|
|
|
- lastReportTime = now;
|
|
|
- }
|
|
|
-
|
|
|
- if (isUpgradeDone()) {
|
|
|
- break;
|
|
|
- }
|
|
|
-
|
|
|
- try {
|
|
|
- Thread.sleep(sleepTimeSec*1000);
|
|
|
- } catch (InterruptedException e) {
|
|
|
- break;
|
|
|
- }
|
|
|
- }
|
|
|
- LOG.info("Leaving the Generation Stamp Upgrade Namenode monitor thread");
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- private BlockLevelStats latestBlockLevelStats = new BlockLevelStats();
|
|
|
- // internal class to hold the stats.
|
|
|
- private static class BlockLevelStats {
|
|
|
- long fullyReplicatedBlocks = 0;
|
|
|
- long minimallyReplicatedBlocks = 0;
|
|
|
- long underReplicatedBlocks = 0; // includes unReplicatedBlocks
|
|
|
- long unReplicatedBlocks = 0; // zero replicas upgraded
|
|
|
- long errors;
|
|
|
- long updatedAt;
|
|
|
-
|
|
|
- String statusString(String spacing) {
|
|
|
- long totalBlocks = fullyReplicatedBlocks +
|
|
|
- minimallyReplicatedBlocks +
|
|
|
- underReplicatedBlocks;
|
|
|
- double multiplier = 100/(totalBlocks + 1e-20);
|
|
|
-
|
|
|
- if (spacing.equals("")) {
|
|
|
- spacing = ", ";
|
|
|
- }
|
|
|
-
|
|
|
- return String.format(
|
|
|
- "Total Blocks : %d" +
|
|
|
- "%sFully Upgragraded : %.2f%%" +
|
|
|
- "%sMinimally Upgraded : %.2f%%" +
|
|
|
- "%sUnder Upgraded : %.2f%% (includes Un-upgraded blocks)" +
|
|
|
- "%sUn-upgraded : %.2f%%" +
|
|
|
- "%sErrors : %d", totalBlocks,
|
|
|
- spacing, (fullyReplicatedBlocks * multiplier),
|
|
|
- spacing, (minimallyReplicatedBlocks * multiplier),
|
|
|
- spacing, (underReplicatedBlocks * multiplier),
|
|
|
- spacing, (unReplicatedBlocks * multiplier),
|
|
|
- spacing, errors);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- void updateBlockLevelStats(String path, BlockLevelStats stats) {
|
|
|
- DFSFileInfo[] fileArr = getFSNamesystem().dir.getListing(path);
|
|
|
-
|
|
|
- for (DFSFileInfo file:fileArr) {
|
|
|
- if (file.isDir()) {
|
|
|
- updateBlockLevelStats(file.getPath().toString(), stats);
|
|
|
- } else {
|
|
|
- // Get the all the blocks.
|
|
|
- LocatedBlocks blockLoc = null;
|
|
|
- try {
|
|
|
- blockLoc = getFSNamesystem().getBlockLocations(
|
|
|
- file.getPath().toString(), 0, file.getLen());
|
|
|
- int numBlocks = blockLoc.locatedBlockCount();
|
|
|
- for (int i=0; i<numBlocks; i++) {
|
|
|
- LocatedBlock loc = blockLoc.get(i);
|
|
|
- DatanodeInfo[] dnArr = loc.getLocations();
|
|
|
- int numUpgraded = 0;
|
|
|
- synchronized (this) {
|
|
|
- for (DatanodeInfo dn:dnArr) {
|
|
|
- DnInfo dnInfo = dnMap.get(dn);
|
|
|
- if (dnInfo != null && dnInfo.isDone()) {
|
|
|
- numUpgraded++;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- if (numUpgraded >= file.getReplication()) {
|
|
|
- stats.fullyReplicatedBlocks++;
|
|
|
- } else if (numUpgraded >= getFSNamesystem().getMinReplication()) {
|
|
|
- stats.minimallyReplicatedBlocks++;
|
|
|
- } else {
|
|
|
- stats.underReplicatedBlocks++;
|
|
|
- }
|
|
|
- if (numUpgraded == 0) {
|
|
|
- stats.unReplicatedBlocks++;
|
|
|
- }
|
|
|
- }
|
|
|
- } catch (IOException e) {
|
|
|
- LOG.error("BlockGenerationStampUpgrade: could not get block locations for " +
|
|
|
- file.getPath().toString() + " : " +
|
|
|
- StringUtils.stringifyException(e));
|
|
|
- stats.errors++;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- void updateBlockLevelStats() {
|
|
|
- /* This iterates over all the blocks and updates various
|
|
|
- * counts.
|
|
|
- * Since iterating over all the blocks at once would be quite
|
|
|
- * large operation under lock, we iterate over all the files
|
|
|
- * and update the counts for blocks that belong to a file.
|
|
|
- */
|
|
|
-
|
|
|
- LOG.info("Starting update of block level stats. " +
|
|
|
- "This could take a few minutes");
|
|
|
- BlockLevelStats stats = new BlockLevelStats();
|
|
|
- updateBlockLevelStats("/", stats);
|
|
|
- stats.updatedAt = System.currentTimeMillis();
|
|
|
-
|
|
|
- LOG.info("Block level stats:\n\t" + stats.statusString("\n\t"));
|
|
|
- synchronized (this) {
|
|
|
- latestBlockLevelStats = stats;
|
|
|
- }
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-/**
|
|
|
- * The Datanode sends this statistics object to the Namenode periodically.
|
|
|
- */
|
|
|
-class DatanodeStatsCommand extends UpgradeCommand {
|
|
|
- DatanodeID datanodeId;
|
|
|
- int blocksUpgraded;
|
|
|
- int blocksRemaining;
|
|
|
- int errors;
|
|
|
-
|
|
|
- DatanodeStatsCommand() {
|
|
|
- super(GenerationStampUpgradeNamenode.DN_CMD_STATS, 0, (short)0);
|
|
|
- datanodeId = new DatanodeID();
|
|
|
- }
|
|
|
-
|
|
|
- public DatanodeStatsCommand(short status, DatanodeID dn,
|
|
|
- int blocksUpgraded, int blocksRemaining,
|
|
|
- int errors, int version) {
|
|
|
- super(GenerationStampUpgradeNamenode.DN_CMD_STATS, version, status);
|
|
|
- //copy so that only ID part gets serialized
|
|
|
- datanodeId = new DatanodeID(dn);
|
|
|
- this.blocksUpgraded = blocksUpgraded;
|
|
|
- this.blocksRemaining = blocksRemaining;
|
|
|
- this.errors = errors;
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public void readFields(DataInput in) throws IOException {
|
|
|
- super.readFields(in);
|
|
|
- datanodeId.readFields(in);
|
|
|
- blocksUpgraded = in.readInt();
|
|
|
- blocksRemaining = in.readInt();
|
|
|
- errors = in.readInt();
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public void write(DataOutput out) throws IOException {
|
|
|
- super.write(out);
|
|
|
- datanodeId.write(out);
|
|
|
- out.writeInt(blocksUpgraded);
|
|
|
- out.writeInt(blocksRemaining);
|
|
|
- out.writeInt(errors);
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-
|
|
|
-/**
|
|
|
- * A status report object for Generation Stamp Upgrades
|
|
|
- */
|
|
|
-class GenerationStampUpgradeStatusReport extends UpgradeStatusReport {
|
|
|
-
|
|
|
- String extraText = "";
|
|
|
-
|
|
|
- public GenerationStampUpgradeStatusReport() {
|
|
|
- }
|
|
|
-
|
|
|
- public GenerationStampUpgradeStatusReport(int version, short status,
|
|
|
- String extraText) {
|
|
|
- super(version, status, false);
|
|
|
- this.extraText = extraText;
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public String getStatusText(boolean details) {
|
|
|
- return super.getStatusText(details) + "\n\n" + extraText;
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public void readFields(DataInput in) throws IOException {
|
|
|
- super.readFields(in);
|
|
|
- extraText = Text.readString(in);
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public void write(DataOutput out) throws IOException {
|
|
|
- super.write(out);
|
|
|
- Text.writeString(out, extraText);
|
|
|
- }
|
|
|
-}
|
|
|
-
|
Raghu Angadi