HDFS-12975. [SBN read] Changes to the NameNode to support reads from standby. Contributed by Chao Sun.
This commit is contained in:
parent
686fcd4db3
commit
3a78e5ffd8
@ -157,7 +157,9 @@ enum StartupOption{
|
|||||||
// only used for StorageDirectory.analyzeStorage() in hot swap drive scenario.
|
// only used for StorageDirectory.analyzeStorage() in hot swap drive scenario.
|
||||||
// TODO refactor StorageDirectory.analyzeStorage() so that we can do away with
|
// TODO refactor StorageDirectory.analyzeStorage() so that we can do away with
|
||||||
// this in StartupOption.
|
// this in StartupOption.
|
||||||
HOTSWAP("-hotswap");
|
HOTSWAP("-hotswap"),
|
||||||
|
// Startup the namenode in observer mode.
|
||||||
|
OBSERVER("-observer");
|
||||||
|
|
||||||
private static final Pattern ENUM_WITH_ROLLING_UPGRADE_OPTION = Pattern.compile(
|
private static final Pattern ENUM_WITH_ROLLING_UPGRADE_OPTION = Pattern.compile(
|
||||||
"(\\w+)\\((\\w+)\\)");
|
"(\\w+)\\((\\w+)\\)");
|
||||||
|
@ -542,7 +542,8 @@ private void logAuditEvent(boolean succeeded,
|
|||||||
private final ReentrantLock cpLock;
|
private final ReentrantLock cpLock;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Used when this NN is in standby state to read from the shared edit log.
|
* Used when this NN is in standby or observer state to read from the
|
||||||
|
* shared edit log.
|
||||||
*/
|
*/
|
||||||
private EditLogTailer editLogTailer = null;
|
private EditLogTailer editLogTailer = null;
|
||||||
|
|
||||||
@ -1380,24 +1381,25 @@ void stopActiveServices() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Start services required in standby state
|
* Start services required in standby or observer state
|
||||||
*
|
*
|
||||||
* @throws IOException
|
* @throws IOException
|
||||||
*/
|
*/
|
||||||
void startStandbyServices(final Configuration conf) throws IOException {
|
void startStandbyServices(final Configuration conf, boolean isObserver)
|
||||||
LOG.info("Starting services required for standby state");
|
throws IOException {
|
||||||
|
LOG.info("Starting services required for " +
|
||||||
|
(isObserver ? "observer" : "standby") + " state");
|
||||||
if (!getFSImage().editLog.isOpenForRead()) {
|
if (!getFSImage().editLog.isOpenForRead()) {
|
||||||
// During startup, we're already open for read.
|
// During startup, we're already open for read.
|
||||||
getFSImage().editLog.initSharedJournalsForRead();
|
getFSImage().editLog.initSharedJournalsForRead();
|
||||||
}
|
}
|
||||||
|
|
||||||
blockManager.setPostponeBlocksFromFuture(true);
|
blockManager.setPostponeBlocksFromFuture(true);
|
||||||
|
|
||||||
// Disable quota checks while in standby.
|
// Disable quota checks while in standby.
|
||||||
dir.disableQuotaChecks();
|
dir.disableQuotaChecks();
|
||||||
editLogTailer = new EditLogTailer(this, conf);
|
editLogTailer = new EditLogTailer(this, conf);
|
||||||
editLogTailer.start();
|
editLogTailer.start();
|
||||||
if (standbyShouldCheckpoint) {
|
if (!isObserver && standbyShouldCheckpoint) {
|
||||||
standbyCheckpointer = new StandbyCheckpointer(conf, this);
|
standbyCheckpointer = new StandbyCheckpointer(conf, this);
|
||||||
standbyCheckpointer.start();
|
standbyCheckpointer.start();
|
||||||
}
|
}
|
||||||
|
@ -365,6 +365,7 @@ public long getProtocolVersion(String protocol,
|
|||||||
LoggerFactory.getLogger("BlockStateChange");
|
LoggerFactory.getLogger("BlockStateChange");
|
||||||
public static final HAState ACTIVE_STATE = new ActiveState();
|
public static final HAState ACTIVE_STATE = new ActiveState();
|
||||||
public static final HAState STANDBY_STATE = new StandbyState();
|
public static final HAState STANDBY_STATE = new StandbyState();
|
||||||
|
public static final HAState OBSERVER_STATE = new StandbyState(true);
|
||||||
|
|
||||||
private static final String NAMENODE_HTRACE_PREFIX = "namenode.htrace.";
|
private static final String NAMENODE_HTRACE_PREFIX = "namenode.htrace.";
|
||||||
|
|
||||||
@ -987,6 +988,8 @@ protected HAState createHAState(StartupOption startOpt) {
|
|||||||
if (!haEnabled || startOpt == StartupOption.UPGRADE
|
if (!haEnabled || startOpt == StartupOption.UPGRADE
|
||||||
|| startOpt == StartupOption.UPGRADEONLY) {
|
|| startOpt == StartupOption.UPGRADEONLY) {
|
||||||
return ACTIVE_STATE;
|
return ACTIVE_STATE;
|
||||||
|
} else if (startOpt == StartupOption.OBSERVER) {
|
||||||
|
return OBSERVER_STATE;
|
||||||
} else {
|
} else {
|
||||||
return STANDBY_STATE;
|
return STANDBY_STATE;
|
||||||
}
|
}
|
||||||
@ -1481,6 +1484,8 @@ static StartupOption parseArguments(String args[]) {
|
|||||||
startOpt = StartupOption.BACKUP;
|
startOpt = StartupOption.BACKUP;
|
||||||
} else if (StartupOption.CHECKPOINT.getName().equalsIgnoreCase(cmd)) {
|
} else if (StartupOption.CHECKPOINT.getName().equalsIgnoreCase(cmd)) {
|
||||||
startOpt = StartupOption.CHECKPOINT;
|
startOpt = StartupOption.CHECKPOINT;
|
||||||
|
} else if (StartupOption.OBSERVER.getName().equalsIgnoreCase(cmd)) {
|
||||||
|
startOpt = StartupOption.OBSERVER;
|
||||||
} else if (StartupOption.UPGRADE.getName().equalsIgnoreCase(cmd)
|
} else if (StartupOption.UPGRADE.getName().equalsIgnoreCase(cmd)
|
||||||
|| StartupOption.UPGRADEONLY.getName().equalsIgnoreCase(cmd)) {
|
|| StartupOption.UPGRADEONLY.getName().equalsIgnoreCase(cmd)) {
|
||||||
startOpt = StartupOption.UPGRADE.getName().equalsIgnoreCase(cmd) ?
|
startOpt = StartupOption.UPGRADE.getName().equalsIgnoreCase(cmd) ?
|
||||||
@ -1794,6 +1799,11 @@ synchronized void transitionToActive()
|
|||||||
if (!haEnabled) {
|
if (!haEnabled) {
|
||||||
throw new ServiceFailedException("HA for namenode is not enabled");
|
throw new ServiceFailedException("HA for namenode is not enabled");
|
||||||
}
|
}
|
||||||
|
if (state == OBSERVER_STATE) {
|
||||||
|
// TODO: we may need to remove this when enabling failover for observer
|
||||||
|
throw new ServiceFailedException(
|
||||||
|
"Cannot transition from Observer to Active");
|
||||||
|
}
|
||||||
state.setState(haContext, ACTIVE_STATE);
|
state.setState(haContext, ACTIVE_STATE);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -1803,6 +1813,11 @@ synchronized void transitionToStandby()
|
|||||||
if (!haEnabled) {
|
if (!haEnabled) {
|
||||||
throw new ServiceFailedException("HA for namenode is not enabled");
|
throw new ServiceFailedException("HA for namenode is not enabled");
|
||||||
}
|
}
|
||||||
|
if (state == OBSERVER_STATE) {
|
||||||
|
// TODO: we may need to remove this when enabling failover for observer
|
||||||
|
throw new ServiceFailedException(
|
||||||
|
"Cannot transition from Observer to Standby");
|
||||||
|
}
|
||||||
state.setState(haContext, STANDBY_STATE);
|
state.setState(haContext, STANDBY_STATE);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -1857,6 +1872,7 @@ public String getNNRole() {
|
|||||||
|
|
||||||
@Override // NameNodeStatusMXBean
|
@Override // NameNodeStatusMXBean
|
||||||
public String getState() {
|
public String getState() {
|
||||||
|
// TODO: maybe we should return a different result for observer namenode?
|
||||||
String servStateStr = "";
|
String servStateStr = "";
|
||||||
HAServiceState servState = getServiceState();
|
HAServiceState servState = getServiceState();
|
||||||
if (null != servState) {
|
if (null != servState) {
|
||||||
@ -1957,7 +1973,8 @@ public void stopActiveServices() throws IOException {
|
|||||||
@Override
|
@Override
|
||||||
public void startStandbyServices() throws IOException {
|
public void startStandbyServices() throws IOException {
|
||||||
try {
|
try {
|
||||||
namesystem.startStandbyServices(getConf());
|
namesystem.startStandbyServices(getConf(),
|
||||||
|
state == NameNode.OBSERVER_STATE);
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
doImmediateShutdown(t);
|
doImmediateShutdown(t);
|
||||||
}
|
}
|
||||||
@ -2004,6 +2021,9 @@ public void checkOperation(final OperationCategory op)
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean allowStaleReads() {
|
public boolean allowStaleReads() {
|
||||||
|
if (state == OBSERVER_STATE) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
return allowStaleStandbyReads;
|
return allowStaleStandbyReads;
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -2017,6 +2037,10 @@ public boolean isActiveState() {
|
|||||||
return (state.equals(ACTIVE_STATE));
|
return (state.equals(ACTIVE_STATE));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public boolean isObserverState() {
|
||||||
|
return state.equals(OBSERVER_STATE);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Returns whether the NameNode is completely started
|
* Returns whether the NameNode is completely started
|
||||||
*/
|
*/
|
||||||
|
@ -1575,7 +1575,7 @@ public Boolean call() throws IOException {
|
|||||||
|
|
||||||
if (nn.getFSImage().isUpgradeFinalized() &&
|
if (nn.getFSImage().isUpgradeFinalized() &&
|
||||||
!namesystem.isRollingUpgrade() &&
|
!namesystem.isRollingUpgrade() &&
|
||||||
!nn.isStandbyState() &&
|
nn.isActiveState() &&
|
||||||
noStaleStorages) {
|
noStaleStorages) {
|
||||||
return new FinalizeCommand(poolId);
|
return new FinalizeCommand(poolId);
|
||||||
}
|
}
|
||||||
|
@ -39,8 +39,15 @@
|
|||||||
*/
|
*/
|
||||||
@InterfaceAudience.Private
|
@InterfaceAudience.Private
|
||||||
public class StandbyState extends HAState {
|
public class StandbyState extends HAState {
|
||||||
|
private final boolean isObserver;
|
||||||
|
|
||||||
public StandbyState() {
|
public StandbyState() {
|
||||||
|
this(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
public StandbyState(boolean isObserver) {
|
||||||
super(HAServiceState.STANDBY);
|
super(HAServiceState.STANDBY);
|
||||||
|
this.isObserver = isObserver;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@ -49,6 +56,11 @@ public void setState(HAContext context, HAState s) throws ServiceFailedException
|
|||||||
setStateInternal(context, s);
|
setStateInternal(context, s);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
if (isObserver && s == NameNode.STANDBY_STATE) {
|
||||||
|
// To guard against the exception in the following super call.
|
||||||
|
// The other case, standby -> observer, should not happen.
|
||||||
|
return;
|
||||||
|
}
|
||||||
super.setState(context, s);
|
super.setState(context, s);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -92,5 +104,10 @@ public void checkOperation(HAContext context, OperationCategory op)
|
|||||||
public boolean shouldPopulateReplQueues() {
|
public boolean shouldPopulateReplQueues() {
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String toString() {
|
||||||
|
return isObserver ? "observer" : "standby";
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user