Merge Release 5.2 into master

build/packages.mk

@@ -26,7 +26,7 @@ PACKAGE_CONTENTS := $(addprefix bin/, $(PACKAGE_BINARIES)) $(addprefix bin/, $(a
 
 packages: TGZ FDBSERVERAPI
 
-TGZ: $(PACKAGE_CONTENTS) versions.target lib/libfdb_java.so
+TGZ: $(PACKAGE_CONTENTS) versions.target lib/libfdb_java.$(DLEXT)
 	@echo "Archiving      tgz"
 	@mkdir -p packages
 	@rm -f packages/FoundationDB-$(PLATFORM)-*.tar.gz

fdbclient/BackupAgent.h

@@ -735,9 +735,8 @@ class BackupConfig : public KeyBackedConfig {
 		// These should not happen
 		if(e.code() == error_code_key_not_found)
 			t.backtrace();
-		std::string msg = format("ERROR: %s %s", e.what(), details.c_str());
 
-		return updateErrorInfo(cx, e, msg);
+		return updateErrorInfo(cx, e, details);
 	}
 };
 #endif

fdbclient/FileBackupAgent.actor.cpp

@@ -35,6 +35,49 @@
 #include <boost/algorithm/string/classification.hpp>
 #include <algorithm>
 
+static std::string boolToYesOrNo(bool val) { return val ? std::string("Yes") : std::string("No"); }
+
+static std::string versionToString(Optional<Version> version) {
+	if (version.present())
+		return std::to_string(version.get());
+	else
+		return "N/A";
+}
+
+static std::string timeStampToString(Optional<int64_t> ts) {
+	if (!ts.present())
+		return "N/A";
+	time_t curTs = ts.get();
+	char buffer[128];
+	struct tm* timeinfo;
+	timeinfo = localtime(&curTs);
+	strftime(buffer, 128, "%D %T", timeinfo);
+	return std::string(buffer);
+}
+
+static Future<Optional<int64_t>> getTimestampFromVersion(Optional<Version> ver, Reference<ReadYourWritesTransaction> tr) {
+	if (!ver.present())
+		return Optional<int64_t>();
+
+	return timeKeeperEpochsFromVersion(ver.get(), tr);
+}
+
+// Time format :
+// <= 59 seconds
+// <= 59.99 minutes
+// <= 23.99 hours
+// N.NN days
+std::string secondsToTimeFormat(int64_t seconds) {
+	if (seconds >= 86400)
+		return format("%.2f day(s)", seconds / 86400.0);
+	else if (seconds >= 3600)
+		return format("%.2f hour(s)", seconds / 3600.0);
+	else if (seconds >= 60)
+		return format("%.2f minute(s)", seconds / 60.0);
+	else
+		return format("%ld second(s)", seconds);
+}
+
 const Key FileBackupAgent::keyLastRestorable = LiteralStringRef("last_restorable");
 
 // For convenience
@@ -177,9 +220,8 @@ class RestoreConfig : public KeyBackedConfig {
 		// These should not happen
 		if(e.code() == error_code_key_not_found)
 			t.backtrace();
-		std::string msg = format("ERROR: %s (%s)", details.c_str(), e.what());
 
-		return updateErrorInfo(cx, e, msg);
+		return updateErrorInfo(cx, e, details);
 	}
 
 	Key mutationLogPrefix() {
@@ -790,7 +832,7 @@ namespace fileBackup {
 	// servers to catch and log to the appropriate config any error that execute/finish didn't catch and log.
 	struct RestoreTaskFuncBase : TaskFuncBase {
 		virtual Future<Void> handleError(Database cx, Reference<Task> task, Error const &error) {
-			return RestoreConfig(task).logError(cx, error, format("Task '%s' UID '%s' %s failed", task->params[Task::reservedTaskParamKeyType].printable().c_str(), task->key.printable().c_str(), toString(task).c_str()));
+			return RestoreConfig(task).logError(cx, error, format("'%s' on '%s'", error.what(), task->params[Task::reservedTaskParamKeyType].printable().c_str()));
 		}
 		virtual std::string toString(Reference<Task> task)
 		{
@@ -800,7 +842,7 @@ namespace fileBackup {
 
 	struct BackupTaskFuncBase : TaskFuncBase {
 		virtual Future<Void> handleError(Database cx, Reference<Task> task, Error const &error) {
-			return BackupConfig(task).logError(cx, error, format("Task '%s' UID '%s' %s failed", task->params[Task::reservedTaskParamKeyType].printable().c_str(), task->key.printable().c_str(), toString(task).c_str()));
+			return BackupConfig(task).logError(cx, error, format("'%s' on '%s'", error.what(), task->params[Task::reservedTaskParamKeyType].printable().c_str()));
 		}
 		virtual std::string toString(Reference<Task> task)
 		{
@@ -3538,17 +3580,50 @@ class FileBackupAgentImpl {
 						state int64_t snapshotInterval;
 						state Version snapshotBeginVersion;
 						state Version snapshotTargetEndVersion;
+						state Optional<Version> latestSnapshotEndVersion;
+						state Optional<Version> latestLogEndVersion;
+						state Optional<int64_t> logBytesWritten;
+						state Optional<int64_t> rangeBytesWritten;
+						state Optional<int64_t> latestSnapshotEndVersionTimestamp;
+						state Optional<int64_t> latestLogEndVersionTimestamp;
+						state Optional<int64_t> snapshotBeginVersionTimestamp;
+						state Optional<int64_t> snapshotTargetEndVersionTimestamp;
+						state bool stopWhenDone;
 
 						Void _ = wait( store(config.snapshotBeginVersion().getOrThrow(tr), snapshotBeginVersion)
 									&& store(config.snapshotTargetEndVersion().getOrThrow(tr), snapshotTargetEndVersion)
 									&& store(config.snapshotIntervalSeconds().getOrThrow(tr), snapshotInterval)
-								);
+									&& store(config.logBytesWritten().get(tr), logBytesWritten)
+									&& store(config.rangeBytesWritten().get(tr), rangeBytesWritten)
+									&& store(config.latestLogEndVersion().get(tr), latestLogEndVersion)
+									&& store(config.latestSnapshotEndVersion().get(tr), latestSnapshotEndVersion)
+									&& store(config.stopWhenDone().getOrThrow(tr), stopWhenDone) 
+									);
+
+						Void _ = wait( store(getTimestampFromVersion(latestSnapshotEndVersion, tr), latestSnapshotEndVersionTimestamp)
+									&& store(getTimestampFromVersion(latestLogEndVersion, tr), latestLogEndVersionTimestamp)
+									&& store(timeKeeperEpochsFromVersion(snapshotBeginVersion, tr), snapshotBeginVersionTimestamp)
+									&& store(timeKeeperEpochsFromVersion(snapshotTargetEndVersion, tr), snapshotTargetEndVersionTimestamp)
+									);
 
 						statusText += format("Snapshot interval is %lld seconds.  ", snapshotInterval);
 						if(backupState == BackupAgentBase::STATE_DIFFERENTIAL)
 							statusText += format("Current snapshot progress target is %3.2f%% (>100%% means the snapshot is supposed to be done)\n", 100.0 * (recentReadVersion - snapshotBeginVersion) / (snapshotTargetEndVersion - snapshotBeginVersion)) ;
 						else
 							statusText += "The initial snapshot is still running.\n";
+
+						statusText += format("\nDetails:\n LogBytes written - %ld\n RangeBytes written - %ld\n "
+											 "Last complete log version and timestamp        - %s, %s\n "
+											 "Last complete snapshot version and timestamp   - %s, %s\n "
+											 "Current Snapshot start version and timestamp   - %s, %s\n "
+											 "Expected snapshot end version and timestamp    - %s, %s\n "
+											 "Backup supposed to stop at next snapshot completion - %s\n",
+											 logBytesWritten.orDefault(0), rangeBytesWritten.orDefault(0),
+											 versionToString(latestLogEndVersion).c_str(), timeStampToString(latestLogEndVersionTimestamp).c_str(),
+											 versionToString(latestSnapshotEndVersion).c_str(), timeStampToString(latestSnapshotEndVersionTimestamp).c_str(),
+											 versionToString(snapshotBeginVersion).c_str(), timeStampToString(snapshotBeginVersionTimestamp).c_str(),
+											 versionToString(snapshotTargetEndVersion).c_str(), timeStampToString(snapshotTargetEndVersionTimestamp).c_str(),
+											 boolToYesOrNo(stopWhenDone).c_str());
 					}
 
 					// Append the errors, if requested
@@ -3559,8 +3634,7 @@ class FileBackupAgentImpl {
 
 						for(auto &e : errors) {
 							Version v = e.second.second;
-							std::string msg = format("%s (%lld seconds ago)\n", e.second.first.c_str(),
-														(recentReadVersion - v) / CLIENT_KNOBS->CORE_VERSIONSPERSECOND);
+							std::string msg = format("%s ago : %s\n", secondsToTimeFormat((recentReadVersion - v) / CLIENT_KNOBS->CORE_VERSIONSPERSECOND).c_str(), e.second.first.c_str());
 
 							// If error version is at or more recent than the latest restorable version then it could be inhibiting progress
 							if(v >= latestRestorableVersion.orDefault(0)) {
@@ -3571,10 +3645,16 @@ class FileBackupAgentImpl {
 							}
 						}
 
-						if(!recentErrors.empty())
-							statusText += "Errors possibly preventing progress:\n" + recentErrors;
+						if (!recentErrors.empty()) {
+							if (latestRestorableVersion.present())
+								statusText += format("Recent Errors (since latest restorable point %s ago)\n",
+														secondsToTimeFormat((recentReadVersion - latestRestorableVersion.get()) / CLIENT_KNOBS->CORE_VERSIONSPERSECOND).c_str())
+											  + recentErrors;
+							else
+								statusText += "Recent Errors (since initialization)\n" + recentErrors;
+						}
 						if(!pastErrors.empty())
-							statusText += "Older errors:\n" + pastErrors;
+							statusText += "Older Errors\n" + pastErrors;
 					}
 				}
 

fdbclient/NativeAPI.actor.cpp

@@ -890,12 +890,12 @@ Reference<ProxyInfo> DatabaseContext::getMasterProxies() {
 }
 
 //Actor which will wait until the ProxyInfo returned by the DatabaseContext cx is not NULL
-ACTOR Future<Reference<ProxyInfo>> getMasterProxiesFuture(DatabaseContext *cx) {
-	loop{
-		Reference<ProxyInfo> proxies = cx->getMasterProxies();
-		if (proxies)
-			return proxies;
-		Void _ = wait( cx->onMasterProxiesChanged() );
+ACTOR Future<Reference<ProxyInfo>> getMasterProxiesFuture(DatabaseContext *cx) {
+	loop{
+		Reference<ProxyInfo> proxies = cx->getMasterProxies();
+		if (proxies)
+			return proxies;
+		Void _ = wait( cx->onMasterProxiesChanged() );
 	}
 }
 

fdbserver/LogSystemConfig.h

@@ -166,6 +166,26 @@ struct LogSystemConfig {
 		return results;
 	}
 
+	std::vector<std::pair<UID, NetworkAddress>> allSharedLogs() const {
+		typedef std::pair<UID, NetworkAddress> IdAddrPair;
+		std::vector<IdAddrPair> results;
+		for (auto &tLog : tLogs) {
+			if (tLog.present())
+				results.push_back(IdAddrPair(tLog.interf().getSharedTLogID(), tLog.interf().address()));
+		}
+
+		for (auto &oldLog : oldTLogs) {
+			for (auto &tLog : oldLog.tLogs) {
+				if (tLog.present())
+					results.push_back(IdAddrPair(tLog.interf().getSharedTLogID(), tLog.interf().address()));
+			}
+		}
+		uniquify(results);
+		// This assert depends on the fact that uniquify will sort the elements based on <UID, NetworkAddr> order
+		ASSERT_WE_THINK(std::unique(results.begin(), results.end(), [](IdAddrPair &x, IdAddrPair &y) { return x.first == y.first; }) == results.end());
+		return results;
+	}
+
 	bool operator == ( const LogSystemConfig& rhs ) const { return isEqual(rhs); }
 
 	bool isEqual(LogSystemConfig const& r) const {

fdbserver/MasterInterface.h

@@ -77,9 +77,10 @@ struct GetRateInfoReply {
 
 struct TLogRejoinRequest {
 	TLogInterface myInterface;
-	DBRecoveryCount recoveryCount;
 	ReplyPromise<bool> reply;   // false means someone else registered, so we should re-register.  true means this master is recovered, so don't send again to the same master.
 
+	TLogRejoinRequest() { }
+	explicit TLogRejoinRequest(const TLogInterface &interf) : myInterface(interf) { }
 	template <class Ar>
 	void serialize(Ar& ar) {
 		ar & myInterface & reply;

fdbserver/TLogInterface.h

@@ -32,6 +32,7 @@ struct TLogInterface {
 	enum { LocationAwareLoadBalance = 1 };
 	LocalityData locality;
 	UID uniqueID;
+	UID sharedTLogID;
 	RequestStream< struct TLogPeekRequest > peekMessages;
 	RequestStream< struct TLogPopRequest > popMessages;
 
@@ -42,8 +43,11 @@ struct TLogInterface {
 	RequestStream<ReplyPromise<Void>> waitFailure;
 	RequestStream< struct TLogRecoveryFinishedRequest > recoveryFinished;
 
-	TLogInterface() : uniqueID( g_random->randomUniqueID() ) {}
+	TLogInterface() { }
+	TLogInterface(UID sharedTLogID, LocalityData locality) : uniqueID( g_random->randomUniqueID() ), sharedTLogID(sharedTLogID), locality(locality) {}
+	TLogInterface(UID uniqueID, UID sharedTLogID, LocalityData locality) : uniqueID(uniqueID), sharedTLogID(sharedTLogID), locality(locality) {}
 	UID id() const { return uniqueID; }
+	UID getSharedTLogID() const { return sharedTLogID; }
 	std::string toString() const { return id().shortString(); }
 	bool operator == ( TLogInterface const& r ) const { return id() == r.id(); }
 	NetworkAddress address() const { return peekMessages.getEndpoint().address; }
@@ -57,7 +61,7 @@ struct TLogInterface {
 
 	template <class Ar> 
 	void serialize( Ar& ar ) {
-		ar & uniqueID & locality & peekMessages & popMessages 
+		ar & uniqueID & sharedTLogID & locality & peekMessages & popMessages
 		   & commit & lock & getQueuingMetrics & confirmRunning & waitFailure & recoveryFinished;
 	}
 };

fdbserver/TLogServer.actor.cpp

@@ -1156,8 +1156,7 @@ ACTOR Future<Void> rejoinMasters( TLogData* self, TLogInterface tli, DBRecoveryC
 		if( registerWithMaster.isReady() ) {
 			if ( self->dbInfo->get().master.id() != lastMasterID) {
 				// The TLogRejoinRequest is needed to establish communications with a new master, which doesn't have our TLogInterface
-				TLogRejoinRequest req;
-				req.myInterface = tli;
+				TLogRejoinRequest req(tli);
 				TraceEvent("TLogRejoining", self->dbgid).detail("Master", self->dbInfo->get().master.id());
 				choose {
 					when ( bool success = wait( brokenPromiseToNever( self->dbInfo->get().master.tlogRejoin.getReply( req ) ) ) ) {
@@ -1564,9 +1563,7 @@ ACTOR Future<Void> restorePersistentState( TLogData* self, LocalityData locality
 		UID id2 = BinaryReader::fromStringRef<UID>( fRecoverCounts.get()[idx].key.removePrefix(persistRecoveryCountKeys.begin), Unversioned() );
 		ASSERT(id1 == id2);
 
-		TLogInterface recruited;
-		recruited.uniqueID = id1;
-		recruited.locality = locality;
+		TLogInterface recruited(id1, self->dbgid, locality);
 		recruited.initEndpoints();
 
 		DUMPTOKEN( recruited.peekMessages );
@@ -1890,7 +1887,7 @@ ACTOR Future<Void> recoverFromLogSystem( TLogData* self, Reference<LogData> logD
 }
 
 ACTOR Future<Void> tLogStart( TLogData* self, InitializeTLogRequest req, LocalityData locality ) {
-	state TLogInterface recruited;
+	state TLogInterface recruited(self->dbgid, locality);
 	recruited.locality = locality;
 	recruited.initEndpoints();
 
@@ -1984,7 +1981,8 @@ ACTOR Future<Void> tLog( IKeyValueStore* persistentData, IDiskQueue* persistentQ
 	state Future<Void> error = actorCollection( self.sharedActors.getFuture() );
 
 	TraceEvent("SharedTlog", tlogId);
-
+	// FIXME: Pass the worker id instead of stubbing it
+	startRole(tlogId, UID(), "SharedTLog");
 	try {
 		if(restoreFromDisk) {
 			Void _ = wait( restorePersistentState( &self, locality, oldLog, recovered, tlogRequests ) );
@@ -2013,6 +2011,7 @@ ACTOR Future<Void> tLog( IKeyValueStore* persistentData, IDiskQueue* persistentQ
 		}
 	} catch (Error& e) {
 		TraceEvent("TLogError", tlogId).error(e, true);
+		endRole(tlogId, "SharedTLog", "Error", true);
 		if(recovered.canBeSet()) recovered.send(Void());
 
 		while(!tlogRequests.isEmpty()) {