wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge 038d67cd90 into 27dd94f291

This commit is contained in:
Jonathan Lafleche 2026-08-03 14:30:04 +08:00 committed by GitHub
parent 27dd94f291 038d67cd90
commit a99f03dced
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 595 additions and 201 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
fdbclient/DatabaseContext.cpp
View File

@ -248,8 +248,11 @@ void DatabaseContext::getLatestCommitVersions(const Reference<LocationInfo>& loc
latestCommitVersions.clear();
if (info->readOptions.present() && info->readOptions.get().debugID.present()) {
g_traceBatch.addEvent(
"TransactionDebug", info->readOptions.get().debugID.get().first(), "NativeAPI.getLatestCommitVersions");
g_traceBatch.addEvent("TransactionDebug",
info->readOptions.get().debugID.get().first(),
"NativeAPI.getLatestCommitVersions",
info->spanContext.traceID,
info->spanContext.spanID);
}
if (!info->readVersionObtainedFromGrvProxy) {
@ -855,8 +858,12 @@ Future<Void> assertFailure(GrvProxyInterface remote, Future<ErrorOr<GetReadVersi
Future<Void> attemptGRVFromOldProxies(std::vector<GrvProxyInterface> oldProxies,
std::vector<GrvProxyInterface> newProxies) {
auto debugID = nondeterministicRandom()->randomUniqueID();
g_traceBatch.addEvent("AttemptGRVFromOldProxyDebug", debugID.first(), "NativeAPI.attemptGRVFromOldProxies.Start");
Span span("NAPI:VerifyCausalReadRisky"_loc);
g_traceBatch.addEvent("AttemptGRVFromOldProxyDebug",
debugID.first(),
"NativeAPI.attemptGRVFromOldProxies.Start",
span.context.traceID,
span.context.spanID);
std::vector<Future<Void>> replies;
replies.reserve(oldProxies.size());
GetReadVersionRequest req(

248
fdbclient/NativeAPI.actor.cpp
View File

@ -1195,7 +1195,11 @@ Future<KeyRangeLocationInfo> getKeyLocation_internal(Database cx,
}
if (debugID.present())
g_traceBatch.addEvent("TransactionDebug", debugID.get().first(), "NativeAPI.getKeyLocation.Before");
g_traceBatch.addEvent("TransactionDebug",
debugID.get().first(),
"NativeAPI.getKeyLocation.Before",
spanContext.traceID,
spanContext.spanID);
while (true) {
try {
@ -1210,7 +1214,11 @@ Future<KeyRangeLocationInfo> getKeyLocation_internal(Database cx,
TaskPriority::DefaultPromiseEndpoint);
++cx->transactionKeyServerLocationRequestsCompleted;
if (debugID.present())
g_traceBatch.addEvent("TransactionDebug", debugID.get().first(), "NativeAPI.getKeyLocation.After");
g_traceBatch.addEvent("TransactionDebug",
debugID.get().first(),
"NativeAPI.getKeyLocation.After",
spanContext.traceID,
spanContext.spanID);
ASSERT(rep.results.size() == 1);
auto locationInfo = cx->setCachedLocation(rep.results[0].first, rep.results[0].second);
@ -1344,7 +1352,11 @@ Future<std::vector<KeyRangeLocationInfo>> getKeyRangeLocations_internal(Database
Version version) {
Span span("NAPI:getKeyRangeLocations"_loc, spanContext);
if (debugID.present())
g_traceBatch.addEvent("TransactionDebug", debugID.get().first(), "NativeAPI.getKeyLocations.Before");
g_traceBatch.addEvent("TransactionDebug",
debugID.get().first(),
"NativeAPI.getKeyLocations.Before",
spanContext.traceID,
spanContext.spanID);
while (true) {
try {
@ -1359,7 +1371,11 @@ Future<std::vector<KeyRangeLocationInfo>> getKeyRangeLocations_internal(Database
TaskPriority::DefaultPromiseEndpoint);
++cx->transactionKeyServerLocationRequestsCompleted;
if (debugID.present())
g_traceBatch.addEvent("TransactionDebug", debugID.get().first(), "NativeAPI.getKeyLocations.After");
g_traceBatch.addEvent("TransactionDebug",
debugID.get().first(),
"NativeAPI.getKeyLocations.After",
spanContext.traceID,
spanContext.spanID);
ASSERT(rep.results.size());
std::vector<KeyRangeLocationInfo> results;
@ -1590,11 +1606,16 @@ ACTOR Future<Optional<Value>> getValue(Reference<TransactionState> trState,
getValueID = nondeterministicRandom()->randomUniqueID();
readOptions.get().debugID = getValueID;
g_traceBatch.addAttach(
"GetValueAttachID", trState->readOptions.get().debugID.get().first(), getValueID.get().first());
g_traceBatch.addAttach("GetValueAttachID",
trState->readOptions.get().debugID.get().first(),
getValueID.get().first(),
trState->spanContext.traceID,
trState->spanContext.spanID);
g_traceBatch.addEvent("GetValueDebug",
getValueID.get().first(),
"NativeAPI.getValue.Before"); //.detail("TaskID", g_network->getCurrentTask());
"NativeAPI.getValue.Before",
trState->spanContext.traceID,
trState->spanContext.spanID); //.detail("TaskID", g_network->getCurrentTask());
/*TraceEvent("TransactionDebugGetValueInfo", getValueID.get())
.detail("Key", key)
.detail("ReqVersion", ver)
@ -1645,7 +1666,8 @@ ACTOR Future<Optional<Value>> getValue(Reference<TransactionState> trState,
if (trState->trLogInfo && recordLogInfo) {
int valueSize = reply.value.present() ? reply.value.get().size() : 0;
trState->trLogInfo->addLog(FdbClientLogEvents::EventGet(
startTimeD, trState->cx->clientLocality.dcId(), latency, valueSize, key));
startTimeD, trState->cx->clientLocality.dcId(), latency, valueSize, key),
trState->spanContext);
}
trState->cx->getValueCompleted->latency = timer_int() - startTime;
trState->cx->getValueCompleted->log();
@ -1655,7 +1677,9 @@ ACTOR Future<Optional<Value>> getValue(Reference<TransactionState> trState,
if (getValueID.present()) {
g_traceBatch.addEvent("GetValueDebug",
getValueID.get().first(),
"NativeAPI.getValue.After"); //.detail("TaskID", g_network->getCurrentTask());
"NativeAPI.getValue.After",
trState->spanContext.traceID,
trState->spanContext.spanID); //.detail("TaskID", g_network->getCurrentTask());
/*TraceEvent("TransactionDebugGetValueDone", getValueID.get())
.detail("Key", key)
.detail("ReqVersion", ver)
@ -1669,15 +1693,21 @@ ACTOR Future<Optional<Value>> getValue(Reference<TransactionState> trState,
trState->cx->getValueCompleted->latency = timer_int() - startTime;
trState->cx->getValueCompleted->log();
if (getValueID.present()) {
g_traceBatch.addEvent("GetValueDebug", getValueID.get().first(), "NativeAPI.getValue.Error");
g_traceBatch.addEvent("GetValueDebug",
getValueID.get().first(),
"NativeAPI.getValue.Error",
trState->spanContext.traceID,
trState->spanContext.spanID);
}
if (e.code() == error_code_wrong_shard_server || e.code() == error_code_all_alternatives_failed) {
trState->cx->invalidateCache(key);
wait(delay(CLIENT_KNOBS->WRONG_SHARD_SERVER_DELAY, trState->taskID));
} else {
if (trState->trLogInfo && recordLogInfo)
trState->trLogInfo->addLog(FdbClientLogEvents::EventGetError(
startTimeD, trState->cx->clientLocality.dcId(), static_cast<int>(e.code()), key));
trState->trLogInfo->addLog(
FdbClientLogEvents::EventGetError(
startTimeD, trState->cx->clientLocality.dcId(), static_cast<int>(e.code()), key),
trState->spanContext);
throw e;
}
}
@ -1695,13 +1725,18 @@ ACTOR Future<Key> getKey(Reference<TransactionState> trState, KeySelector k) {
getKeyID = nondeterministicRandom()->randomUniqueID();
readOptions.get().debugID = getKeyID;
g_traceBatch.addAttach(
"GetKeyAttachID", trState->readOptions.get().debugID.get().first(), getKeyID.get().first());
g_traceBatch.addAttach("GetKeyAttachID",
trState->readOptions.get().debugID.get().first(),
getKeyID.get().first(),
trState->spanContext.traceID,
trState->spanContext.spanID);
g_traceBatch.addEvent(
"GetKeyDebug",
getKeyID.get().first(),
"NativeAPI.getKey.AfterVersion"); //.detail("StartKey",
// k.getKey()).detail("Offset",k.offset).detail("OrEqual",k.orEqual);
"NativeAPI.getKey.AfterVersion",
trState->spanContext.traceID,
trState->spanContext.spanID); //.detail("StartKey",
// k.getKey()).detail("Offset",k.offset).detail("OrEqual",k.orEqual);
}
loop {
@ -1726,8 +1761,10 @@ ACTOR Future<Key> getKey(Reference<TransactionState> trState, KeySelector k) {
g_traceBatch.addEvent(
"GetKeyDebug",
getKeyID.get().first(),
"NativeAPI.getKey.Before"); //.detail("StartKey",
// k.getKey()).detail("Offset",k.offset).detail("OrEqual",k.orEqual);
"NativeAPI.getKey.Before",
trState->spanContext.traceID,
trState->spanContext.spanID); //.detail("StartKey",
// k.getKey()).detail("Offset",k.offset).detail("OrEqual",k.orEqual);
++trState->cx->transactionPhysicalReads;
GetKeyRequest req(span.context,
@ -1764,15 +1801,21 @@ ACTOR Future<Key> getKey(Reference<TransactionState> trState, KeySelector k) {
if (getKeyID.present())
g_traceBatch.addEvent("GetKeyDebug",
getKeyID.get().first(),
"NativeAPI.getKey.After"); //.detail("NextKey",reply.sel.key).detail("Offset",
// reply.sel.offset).detail("OrEqual", k.orEqual);
"NativeAPI.getKey.After",
trState->spanContext.traceID,
trState->spanContext.spanID); //.detail("NextKey",reply.sel.key).detail("Offset",
// reply.sel.offset).detail("OrEqual", k.orEqual);
k = reply.sel;
if (!k.offset && k.orEqual) {
return k.getKey();
}
} catch (Error& e) {
if (getKeyID.present())
g_traceBatch.addEvent("GetKeyDebug", getKeyID.get().first(), "NativeAPI.getKey.Error");
g_traceBatch.addEvent("GetKeyDebug",
getKeyID.get().first(),
"NativeAPI.getKey.Error",
trState->spanContext.traceID,
trState->spanContext.spanID);
if (e.code() == error_code_wrong_shard_server || e.code() == error_code_all_alternatives_failed) {
trState->cx->invalidateCache(k.getKey(), Reverse{ k.isBackward() });
@ -1878,11 +1921,16 @@ ACTOR Future<Version> watchValue(Database cx, Reference<const WatchParameters> p
if (parameters->debugID.present()) {
watchValueID = nondeterministicRandom()->randomUniqueID();
g_traceBatch.addAttach(
"WatchValueAttachID", parameters->debugID.get().first(), watchValueID.get().first());
g_traceBatch.addAttach("WatchValueAttachID",
parameters->debugID.get().first(),
watchValueID.get().first(),
parameters->spanContext.traceID,
parameters->spanContext.spanID);
g_traceBatch.addEvent("WatchValueDebug",
watchValueID.get().first(),
"NativeAPI.watchValue.Before"); //.detail("TaskID", g_network->getCurrentTask());
"NativeAPI.watchValue.Before",
parameters->spanContext.traceID,
parameters->spanContext.spanID); //.detail("TaskID", g_network->getCurrentTask());
}
state WatchValueReply resp;
choose {
@ -1903,7 +1951,11 @@ ACTOR Future<Version> watchValue(Database cx, Reference<const WatchParameters> p
}
}
if (watchValueID.present()) {
g_traceBatch.addEvent("WatchValueDebug", watchValueID.get().first(), "NativeAPI.watchValue.After");
g_traceBatch.addEvent("WatchValueDebug",
watchValueID.get().first(),
"NativeAPI.watchValue.After",
parameters->spanContext.traceID,
parameters->spanContext.spanID);
}
// FIXME: wait for known committed version on the storage server before replying,
@ -1925,7 +1977,11 @@ ACTOR Future<Version> watchValue(Database cx, Reference<const WatchParameters> p
ver = v;
if (watchValueID.present()) {
g_traceBatch.addEvent("WatchValueDebug", watchValueID.get().first(), "NativeAPI.watchValue.Retry");
g_traceBatch.addEvent("WatchValueDebug",
watchValueID.get().first(),
"NativeAPI.watchValue.Retry",
parameters->spanContext.traceID,
parameters->spanContext.spanID);
}
} catch (Error& e) {
if (e.code() == error_code_wrong_shard_server || e.code() == error_code_all_alternatives_failed) {
@ -2221,7 +2277,9 @@ Future<RangeResultFamily> getExactRange(Reference<TransactionState> trState,
if (trState->readOptions.present() && trState->readOptions.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
trState->readOptions.get().debugID.get().first(),
"NativeAPI.getExactRange.Before");
"NativeAPI.getExactRange.Before",
trState->spanContext.traceID,
trState->spanContext.spanID);
/*TraceEvent("TransactionDebugGetExactRangeInfo", trState->readOptions.get().debugID.get())
.detail("ReqBeginKey", req.begin.getKey())
.detail("ReqEndKey", req.end.getKey())
@ -2258,7 +2316,9 @@ Future<RangeResultFamily> getExactRange(Reference<TransactionState> trState,
if (trState->readOptions.present() && trState->readOptions.get().debugID.present())
g_traceBatch.addEvent("TransactionDebug",
trState->readOptions.get().debugID.get().first(),
"NativeAPI.getExactRange.After");
"NativeAPI.getExactRange.After",
trState->spanContext.traceID,
trState->spanContext.spanID);
output.arena().dependsOn(rep.arena);
output.append(output.arena(), rep.data.begin(), rep.data.size());
@ -2461,8 +2521,10 @@ void getRangeFinished(Reference<TransactionState> trState,
trState->cx->transactionKeysRead += result.size();
if (trState->trLogInfo) {
trState->trLogInfo->addLog(FdbClientLogEvents::EventGetRange(
startTime, trState->cx->clientLocality.dcId(), now() - startTime, bytes, begin.getKey(), end.getKey()));
trState->trLogInfo->addLog(
FdbClientLogEvents::EventGetRange(
startTime, trState->cx->clientLocality.dcId(), now() - startTime, bytes, begin.getKey(), end.getKey()),
trState->spanContext);
}
if (!snapshot) {
@ -2590,12 +2652,19 @@ Future<RangeResultFamily> getRange(Reference<TransactionState> trState,
req.spanContext = span.context;
if (trState->readOptions.present() && trState->readOptions.get().debugID.present()) {
getRangeID = nondeterministicRandom()->randomUniqueID();
g_traceBatch.addAttach(
"TransactionAttachID", trState->readOptions.get().debugID.get().first(), getRangeID.get().first());
g_traceBatch.addAttach("TransactionAttachID",
trState->readOptions.get().debugID.get().first(),
getRangeID.get().first(),
trState->spanContext.traceID,
trState->spanContext.spanID);
}
try {
if (getRangeID.present()) {
g_traceBatch.addEvent("TransactionDebug", getRangeID.get().first(), "NativeAPI.getRange.Before");
g_traceBatch.addEvent("TransactionDebug",
getRangeID.get().first(),
"NativeAPI.getRange.Before",
trState->spanContext.traceID,
trState->spanContext.spanID);
/*
if (trState->readOptions.present() && trState->readOptions.get().debugID.present()) {
TraceEvent("TransactionDebugGetRangeInfo", trState->readOptions.get().debugID.get())
@ -2642,7 +2711,9 @@ Future<RangeResultFamily> getRange(Reference<TransactionState> trState,
if (getRangeID.present()) {
g_traceBatch.addEvent("TransactionDebug",
getRangeID.get().first(),
"NativeAPI.getRange.After"); //.detail("SizeOf", rep.data.size());
"NativeAPI.getRange.After",
trState->spanContext.traceID,
trState->spanContext.spanID); //.detail("SizeOf", rep.data.size());
/*
if (trState->readOptions.present() && trState->readOptions.get().debugID.present()) {
TraceEvent("TransactionDebugGetRangeDone", trState->readOptions.get().debugID.get())
@ -2758,7 +2829,11 @@ Future<RangeResultFamily> getRange(Reference<TransactionState> trState,
} catch (Error& e) {
if (getRangeID.present()) {
g_traceBatch.addEvent("TransactionDebug", getRangeID.get().first(), "NativeAPI.getRange.Error");
g_traceBatch.addEvent("TransactionDebug",
getRangeID.get().first(),
"NativeAPI.getRange.Error",
trState->spanContext.traceID,
trState->spanContext.spanID);
TraceEvent("TransactionDebugError", getRangeID.get()).error(e);
}
if (e.code() == error_code_wrong_shard_server || e.code() == error_code_all_alternatives_failed) {
@ -2782,7 +2857,8 @@ Future<RangeResultFamily> getRange(Reference<TransactionState> trState,
trState->cx->clientLocality.dcId(),
static_cast<int>(e.code()),
begin.getKey(),
end.getKey()));
end.getKey()),
trState->spanContext);
throw e;
}
}
@ -3006,7 +3082,9 @@ ACTOR Future<Void> getRangeStreamImpl(Reference<TransactionState> trState,
if (trState->readOptions.present() && trState->readOptions.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
trState->readOptions.get().debugID.get().first(),
"NativeAPI.RangeStream.Before");
"NativeAPI.RangeStream.Before",
trState->spanContext.traceID,
trState->spanContext.spanID);
}
++trState->cx->transactionPhysicalReads;
state GetKeyValuesStreamReply rep;
@ -3105,7 +3183,9 @@ ACTOR Future<Void> getRangeStreamImpl(Reference<TransactionState> trState,
if (trState->readOptions.present() && trState->readOptions.get().debugID.present())
g_traceBatch.addEvent("TransactionDebug",
trState->readOptions.get().debugID.get().first(),
"NativeAPI.getExactRange.After");
"NativeAPI.getExactRange.After",
trState->spanContext.traceID,
trState->spanContext.spanID);
RangeResult output(RangeResultRef(rep.data, rep.more), rep.arena);
if (tssDuplicateStream.present() && !tssDuplicateStream.get().done()) {
@ -4357,8 +4437,16 @@ ACTOR static Future<Void> tryCommit(Reference<TransactionState> trState, CommitT
if (debugID.present()) {
commitID = nondeterministicRandom()->randomUniqueID();
g_traceBatch.addAttach("CommitAttachID", debugID.get().first(), commitID.get().first());
g_traceBatch.addEvent("CommitDebug", commitID.get().first(), "NativeAPI.commit.Before");
g_traceBatch.addAttach("CommitAttachID",
debugID.get().first(),
commitID.get().first(),
trState->spanContext.traceID,
trState->spanContext.spanID);
g_traceBatch.addEvent("CommitDebug",
commitID.get().first(),
"NativeAPI.commit.Before",
trState->spanContext.traceID,
trState->spanContext.spanID);
}
req.debugID = commitID;
@ -4420,7 +4508,11 @@ ACTOR static Future<Void> tryCommit(Reference<TransactionState> trState, CommitT
trState->cx->transactionCommittedMutationBytes += req.transaction.mutations.expectedSize();
if (commitID.present())
g_traceBatch.addEvent("CommitDebug", commitID.get().first(), "NativeAPI.commit.After");
g_traceBatch.addEvent("CommitDebug",
commitID.get().first(),
"NativeAPI.commit.After",
trState->spanContext.traceID,
trState->spanContext.spanID);
double latency = now() - startTime;
trState->cx->commitLatencies.addSample(latency);
@ -4433,7 +4525,8 @@ ACTOR static Future<Void> tryCommit(Reference<TransactionState> trState, CommitT
req.transaction.mutations.size(),
req.transaction.mutations.expectedSize(),
ci.version,
req));
req),
trState->spanContext);
if (trState->automaticIdempotency && alternativeChosen >= 0) {
// Automatic idempotency means we're responsible for best effort idempotency id clean up
proxiesUsed->getInterface(alternativeChosen)
@ -4464,7 +4557,11 @@ ACTOR static Future<Void> tryCommit(Reference<TransactionState> trState, CommitT
TraceEvent(interval.end()).detail("Conflict", 1);
if (commitID.present())
g_traceBatch.addEvent("CommitDebug", commitID.get().first(), "NativeAPI.commit.After");
g_traceBatch.addEvent("CommitDebug",
commitID.get().first(),
"NativeAPI.commit.After",
trState->spanContext.traceID,
trState->spanContext.spanID);
throw not_committed();
}
@ -4526,8 +4623,10 @@ ACTOR static Future<Void> tryCommit(Reference<TransactionState> trState, CommitT
TraceEvent(SevError, "TryCommitError").error(e);
}
if (trState->trLogInfo)
trState->trLogInfo->addLog(FdbClientLogEvents::EventCommitError(
startTime, trState->cx->clientLocality.dcId(), static_cast<int>(e.code()), req));
trState->trLogInfo->addLog(
FdbClientLogEvents::EventCommitError(
startTime, trState->cx->clientLocality.dcId(), static_cast<int>(e.code()), req),
trState->spanContext);
throw;
}
}
@ -4688,6 +4787,23 @@ Future<Void> Transaction::commit() {
return committing;
}
static void traceTransactionBeingTraced(Reference<TransactionState> trState) {
if (!trState->trLogInfo || trState->trLogInfo->identifier.empty() || !trState->readOptions.present() ||
!trState->readOptions.get().debugID.present()) {
return;
}
TraceEvent event(SevInfo, "TransactionBeingTraced");
event.detail("DebugTransactionID", trState->trLogInfo->identifier)
.detail("ServerTraceID", trState->readOptions.get().debugID.get());
if (trState->spanContext.traceID.isValid()) {
event.detail("TraceID", trState->spanContext.traceID.toString());
}
if (trState->spanContext.spanID != 0) {
event.detail("SpanID", format("%016" PRIx64, trState->spanContext.spanID));
}
}
// Returns a thread-local mt19937_64 seeded once with 32 bytes of OS entropy.
// Used for AUTOMATIC_IDEMPOTENCY ID generation in non-simulation runs.
static std::mt19937_64& getIdempotencyRng() {
@ -4771,11 +4887,7 @@ void Transaction::setOption(FDBTransactionOptions::Option option, Optional<Strin
makeReference<TransactionLogInfo>(value.get().printable(), TransactionLogInfo::DONT_LOG);
trState->trLogInfo->maxFieldLength = trState->options.maxTransactionLoggingFieldLength;
}
if (trState->readOptions.present() && trState->readOptions.get().debugID.present()) {
TraceEvent(SevInfo, "TransactionBeingTraced")
.detail("DebugTransactionID", trState->trLogInfo->identifier)
.detail("ServerTraceID", trState->readOptions.get().debugID.get());
}
traceTransactionBeingTraced(trState);
break;
case FDBTransactionOptions::LOG_TRANSACTION:
@ -4806,12 +4918,7 @@ void Transaction::setOption(FDBTransactionOptions::Option option, Optional<Strin
case FDBTransactionOptions::SERVER_REQUEST_TRACING:
validateOptionValueNotPresent(value);
debugTransaction(deterministicRandom()->randomUniqueID());
if (trState->trLogInfo && !trState->trLogInfo->identifier.empty() && trState->readOptions.present() &&
trState->readOptions.get().debugID.present()) {
TraceEvent(SevInfo, "TransactionBeingTraced")
.detail("DebugTransactionID", trState->trLogInfo->identifier)
.detail("ServerTraceID", trState->readOptions.get().debugID.get());
}
traceTransactionBeingTraced(trState);
break;
case FDBTransactionOptions::MAX_RETRY_DELAY:
@ -5005,7 +5112,11 @@ ACTOR Future<GetReadVersionReply> getConsistentReadVersion(SpanContext parentSpa
++cx->transactionReadVersionBatches;
if (debugID.present())
g_traceBatch.addEvent("TransactionDebug", debugID.get().first(), "NativeAPI.getConsistentReadVersion.Before");
g_traceBatch.addEvent("TransactionDebug",
debugID.get().first(),
"NativeAPI.getConsistentReadVersion.Before",
parentSpan.traceID,
parentSpan.spanID);
loop {
try {
state GetReadVersionRequest req(span.context,
@ -5046,8 +5157,11 @@ ACTOR Future<GetReadVersionReply> getConsistentReadVersion(SpanContext parentSpa
}
if (debugID.present())
g_traceBatch.addEvent(
"TransactionDebug", debugID.get().first(), "NativeAPI.getConsistentReadVersion.After");
g_traceBatch.addEvent("TransactionDebug",
debugID.get().first(),
"NativeAPI.getConsistentReadVersion.After",
parentSpan.traceID,
parentSpan.spanID);
ASSERT(v.version > 0);
cx->minAcceptableReadVersion = std::min(cx->minAcceptableReadVersion, v.version);
if (cx->versionVectorCacheActive(v.ssVersionVectorDelta)) {
@ -5107,7 +5221,11 @@ ACTOR Future<Void> readVersionBatcher(DatabaseContext* cx,
if (!debugID.present()) {
debugID = nondeterministicRandom()->randomUniqueID();
}
g_traceBatch.addAttach("TransactionAttachID", req.debugID.get().first(), debugID.get().first());
g_traceBatch.addAttach("TransactionAttachID",
req.debugID.get().first(),
debugID.get().first(),
req.spanContext.traceID,
req.spanContext.spanID);
}
span.addLink(req.spanContext);
requests.push_back(req.reply);
@ -5186,8 +5304,12 @@ ACTOR Future<Version> extractReadVersion(Reference<TransactionState> trState,
}
trState->cx->GRVLatencies.addSample(latency);
if (trState->trLogInfo)
trState->trLogInfo->addLog(FdbClientLogEvents::EventGetVersion_V3(
trState->startTime, trState->cx->clientLocality.dcId(), latency, trState->options.priority, rep.version));
trState->trLogInfo->addLog(FdbClientLogEvents::EventGetVersion_V3(trState->startTime,
trState->cx->clientLocality.dcId(),
latency,
trState->options.priority,
rep.version),
trState->spanContext);
if (rep.locked && !trState->options.lockAware)
throw database_locked();
@ -6476,6 +6598,8 @@ void Transaction::setTransactionID(UID id) {
void Transaction::setToken(uint64_t token) {
ASSERT(getSize() == 0);
trState->spanContext = SpanContext(trState->spanContext.traceID, token);
tr.spanContext = trState->spanContext;
span.context = trState->spanContext;
}
void enableClientInfoLogging() {

50
fdbclient/include/fdbclient/ClientLogEvents.h
View File

@ -24,6 +24,7 @@
#include "fdbclient/FDBTypes.h"
#include "fdbclient/CommitProxyInterface.h"
#include <inttypes.h>
// NOTE: data structures defined in this file and serialization thereof represent
// persistent format state internal to FDB. Changes that render old values unreadable
@ -65,7 +66,15 @@ struct Event {
Key dcId{};
Optional<LegacyTenantName> legacyEmptyTenant{};
void logEvent(std::string id, int maxFieldLength) const {}
void logEvent(std::string id, int maxFieldLength, SpanContext spanContext) const {}
void augmentTraceEvent(TraceEvent& event, SpanContext spanContext) const {
if (spanContext.traceID.isValid()) {
event.detail("TraceID", spanContext.traceID.toString());
}
if (spanContext.spanID != 0) {
event.detail("SpanID", format("%016" PRIx64, spanContext.spanID));
}
}
};
struct EventGetVersion : public Event {
@ -81,9 +90,10 @@ struct EventGetVersion : public Event {
double latency;
void logEvent(std::string id, int maxFieldLength) const {
void logEvent(std::string id, int maxFieldLength, SpanContext spanContext) const {
TraceEvent event("TransactionTrace_GetVersion");
event.detail("TransactionID", id).detail("Latency", latency);
augmentTraceEvent(event, spanContext);
}
};
@ -102,9 +112,10 @@ struct EventGetVersion_V2 : public Event {
double latency;
TransactionPriorityType priorityType{ TransactionPriorityType::UNSET };
void logEvent(std::string id, int maxFieldLength) const {
void logEvent(std::string id, int maxFieldLength, SpanContext spanContext) const {
TraceEvent event("TransactionTrace_GetVersion");
event.detail("TransactionID", id).detail("Latency", latency).detail("PriorityType", priorityType);
augmentTraceEvent(event, spanContext);
}
};
@ -146,12 +157,13 @@ struct EventGetVersion_V3 : public Event {
TransactionPriorityType priorityType{ TransactionPriorityType::UNSET };
Version readVersion;
void logEvent(std::string id, int maxFieldLength) const {
void logEvent(std::string id, int maxFieldLength, SpanContext spanContext) const {
TraceEvent event("TransactionTrace_GetVersion");
event.detail("TransactionID", id)
.detail("Latency", latency)
.detail("PriorityType", priorityType)
.detail("ReadVersion", readVersion);
augmentTraceEvent(event, spanContext);
}
};
@ -172,7 +184,7 @@ struct EventGet : public Event {
int valueSize;
Key key;
void logEvent(std::string id, int maxFieldLength) const {
void logEvent(std::string id, int maxFieldLength, SpanContext spanContext) const {
TraceEvent event("TransactionTrace_Get");
event.setMaxEventLength(-1)
.detail("TransactionID", id)
@ -180,6 +192,7 @@ struct EventGet : public Event {
.detail("ValueSizeBytes", valueSize)
.setMaxFieldLength(maxFieldLength)
.detail("Key", key);
augmentTraceEvent(event, spanContext);
}
};
@ -207,7 +220,7 @@ struct EventGetRange : public Event {
Key startKey;
Key endKey;
void logEvent(std::string id, int maxFieldLength) const {
void logEvent(std::string id, int maxFieldLength, SpanContext spanContext) const {
TraceEvent event("TransactionTrace_GetRange");
event.setMaxEventLength(-1)
.detail("TransactionID", id)
@ -216,6 +229,7 @@ struct EventGetRange : public Event {
.setMaxFieldLength(maxFieldLength)
.detail("StartKey", startKey)
.detail("EndKey", endKey);
augmentTraceEvent(event, spanContext);
}
};
@ -236,7 +250,7 @@ struct EventCommit : public Event {
CommitTransactionRequest
req; // Only CommitTransactionRef and Arena object within CommitTransactionRequest is serialized
void logEvent(std::string id, int maxFieldLength) const {
void logEvent(std::string id, int maxFieldLength, SpanContext spanContext) const {
for (auto& read_range : req.transaction.read_conflict_ranges) {
TraceEvent ev1("TransactionTrace_Commit_ReadConflictRange");
ev1.setMaxEventLength(-1)
@ -244,6 +258,7 @@ struct EventCommit : public Event {
.setMaxFieldLength(maxFieldLength)
.detail("Begin", read_range.begin)
.detail("End", read_range.end);
augmentTraceEvent(ev1, spanContext);
}
for (auto& write_range : req.transaction.write_conflict_ranges) {
@ -253,6 +268,7 @@ struct EventCommit : public Event {
.setMaxFieldLength(maxFieldLength)
.detail("Begin", write_range.begin)
.detail("End", write_range.end);
augmentTraceEvent(ev2, spanContext);
}
for (auto& mutation : req.transaction.mutations) {
@ -261,6 +277,7 @@ struct EventCommit : public Event {
.detail("TransactionID", id)
.setMaxFieldLength(maxFieldLength)
.detail("Mutation", mutation);
augmentTraceEvent(ev3, spanContext);
}
TraceEvent ev4("TransactionTrace_Commit");
@ -268,6 +285,7 @@ struct EventCommit : public Event {
.detail("Latency", latency)
.detail("NumMutations", numMutations)
.detail("CommitSizeBytes", commitBytes);
augmentTraceEvent(ev4, spanContext);
}
};
@ -301,7 +319,7 @@ struct EventCommit_V2 : public Event {
CommitTransactionRequest
req; // Only CommitTransactionRef and Arena object within CommitTransactionRequest is serialized
void logEvent(std::string id, int maxFieldLength) const {
void logEvent(std::string id, int maxFieldLength, SpanContext spanContext) const {
for (auto& read_range : req.transaction.read_conflict_ranges) {
TraceEvent ev1("TransactionTrace_Commit_ReadConflictRange");
ev1.setMaxEventLength(-1)
@ -309,6 +327,7 @@ struct EventCommit_V2 : public Event {
.setMaxFieldLength(maxFieldLength)
.detail("Begin", read_range.begin)
.detail("End", read_range.end);
augmentTraceEvent(ev1, spanContext);
}
for (auto& write_range : req.transaction.write_conflict_ranges) {
@ -318,6 +337,7 @@ struct EventCommit_V2 : public Event {
.setMaxFieldLength(maxFieldLength)
.detail("Begin", write_range.begin)
.detail("End", write_range.end);
augmentTraceEvent(ev2, spanContext);
}
for (auto& mutation : req.transaction.mutations) {
@ -326,6 +346,7 @@ struct EventCommit_V2 : public Event {
.detail("TransactionID", id)
.setMaxFieldLength(maxFieldLength)
.detail("Mutation", mutation);
augmentTraceEvent(ev3, spanContext);
}
TraceEvent ev4("TransactionTrace_Commit");
@ -334,6 +355,7 @@ struct EventCommit_V2 : public Event {
.detail("Latency", latency)
.detail("NumMutations", numMutations)
.detail("CommitSizeBytes", commitBytes);
augmentTraceEvent(ev4, spanContext);
}
};
@ -353,13 +375,14 @@ struct EventGetError : public Event {
int errCode;
Key key;
void logEvent(std::string id, int maxFieldLength) const {
void logEvent(std::string id, int maxFieldLength, SpanContext spanContext) const {
TraceEvent event("TransactionTrace_GetError");
event.setMaxEventLength(-1)
.detail("TransactionID", id)
.detail("ErrCode", errCode)
.setMaxFieldLength(maxFieldLength)
.detail("Key", key);
augmentTraceEvent(event, spanContext);
}
};
@ -384,7 +407,7 @@ struct EventGetRangeError : public Event {
Key startKey;
Key endKey;
void logEvent(std::string id, int maxFieldLength) const {
void logEvent(std::string id, int maxFieldLength, SpanContext spanContext) const {
TraceEvent event("TransactionTrace_GetRangeError");
event.setMaxEventLength(-1)
.detail("TransactionID", id)
@ -392,6 +415,7 @@ struct EventGetRangeError : public Event {
.setMaxFieldLength(maxFieldLength)
.detail("StartKey", startKey)
.detail("EndKey", endKey);
augmentTraceEvent(event, spanContext);
}
};
@ -415,7 +439,7 @@ struct EventCommitError : public Event {
CommitTransactionRequest
req; // Only CommitTransactionRef and Arena object within CommitTransactionRequest is serialized
void logEvent(std::string id, int maxFieldLength) const {
void logEvent(std::string id, int maxFieldLength, SpanContext spanContext) const {
for (auto& read_range : req.transaction.read_conflict_ranges) {
TraceEvent ev1("TransactionTrace_CommitError_ReadConflictRange");
ev1.setMaxEventLength(-1)
@ -423,6 +447,7 @@ struct EventCommitError : public Event {
.setMaxFieldLength(maxFieldLength)
.detail("Begin", read_range.begin)
.detail("End", read_range.end);
augmentTraceEvent(ev1, spanContext);
}
for (auto& write_range : req.transaction.write_conflict_ranges) {
@ -432,6 +457,7 @@ struct EventCommitError : public Event {
.setMaxFieldLength(maxFieldLength)
.detail("Begin", write_range.begin)
.detail("End", write_range.end);
augmentTraceEvent(ev2, spanContext);
}
for (auto& mutation : req.transaction.mutations) {
@ -440,10 +466,12 @@ struct EventCommitError : public Event {
.detail("TransactionID", id)
.setMaxFieldLength(maxFieldLength)
.detail("Mutation", mutation);
augmentTraceEvent(ev3, spanContext);
}
TraceEvent ev4("TransactionTrace_CommitError");
ev4.detail("TransactionID", id).detail("ErrCode", errCode);
augmentTraceEvent(ev4, spanContext);
}
};
} // namespace FdbClientLogEvents

4
fdbclient/include/fdbclient/NativeAPI.actor.h
View File

@ -209,10 +209,10 @@ struct TransactionLogInfo : public ReferenceCounted<TransactionLogInfo>, NonCopy
void logTo(LoggingLocation loc) { logLocation = logLocation | loc; }
template <typename T>
void addLog(const T& event) {
void addLog(const T& event, SpanContext spanContext) {
if (logLocation & TRACE_LOG) {
ASSERT(!identifier.empty());
event.logEvent(identifier, maxFieldLength);
event.logEvent(identifier, maxFieldLength, spanContext);
}
if (flushed) {

130
fdbserver/commitproxy/CommitProxyServer.cpp
View File

@ -290,7 +290,11 @@ Future<Void> commitBatcher(ProxyCommitData* commitData,
commitData->stats.uniqueClients.insert(req.reply.getEndpoint().getPrimaryAddress());
if (req.debugID.present()) {
g_traceBatch.addEvent("CommitDebug", req.debugID.get().first(), "CommitProxyServer.batcher");
g_traceBatch.addEvent("CommitDebug",
req.debugID.get().first(),
"CommitProxyServer.batcher",
req.spanContext.traceID,
req.spanContext.spanID);
}
if (batch.empty()) {
@ -505,7 +509,7 @@ struct CommitBatchContext {
// The current stage of batch commit
std::string_view stage = UNSET;
Optional<UID> debugID;
Optional<BatchDebugIDs> debugIDs;
bool forceRecovery = false;
bool rejected = false; // If rejected due to long queue length
@ -583,6 +587,8 @@ struct CommitBatchContext {
CommitBatchContext(ProxyCommitData*, const std::vector<CommitTransactionRequest>*, const int);
Optional<UID> getDebugID() const;
void setupTraceBatch();
std::set<Tag> getWrittenTagsPreResolution();
@ -760,20 +766,36 @@ CommitBatchContext::CommitBatchContext(ProxyCommitData* const pProxyCommitData_,
void CommitBatchContext::setupTraceBatch() {
for (const auto& tr : trs) {
if (tr.debugID.present()) {
if (!debugID.present()) {
debugID = nondeterministicRandom()->randomUniqueID();
if (!debugIDs.present()) {
debugIDs = BatchDebugIDs{ nondeterministicRandom()->randomUniqueID(),
tr.spanContext.traceID,
tr.spanContext.spanID };
}
g_traceBatch.addAttach("CommitAttachID", tr.debugID.get().first(), debugID.get().first());
Optional<UID> debugID = getDebugID();
g_traceBatch.addAttach("CommitAttachID",
tr.debugID.get().first(),
debugID.get().first(),
tr.spanContext.traceID,
tr.spanContext.spanID);
}
span.addLink(tr.spanContext);
}
if (debugID.present()) {
g_traceBatch.addEvent("CommitDebug", debugID.get().first(), "CommitProxyServer.commitBatch.Before");
if (debugIDs.present()) {
Optional<UID> debugID = getDebugID();
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.Before",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
}
Optional<UID> CommitBatchContext::getDebugID() const {
return debugIDs.present() ? Optional<UID>(debugIDs.get().debugID) : Optional<UID>();
}
void CommitBatchContext::evaluateBatchSize() {
for (const auto& tr : trs) {
const auto& mutations = tr.transaction.mutations;
@ -807,7 +829,8 @@ Future<Void> preresolutionProcessing(CommitBatchContext* self) {
std::vector<CommitTransactionRequest>& trs = self->trs;
const int64_t localBatchNumber = self->localBatchNumber;
const int latencyBucket = self->latencyBucket;
const Optional<UID>& debugID = self->debugID;
Optional<BatchDebugIDs> debugIDs = self->debugIDs;
Optional<UID> debugID = self->getDebugID();
Span span("MP:preresolutionProcessing"_loc, self->span.context);
double startTime = g_network->timer_monotonic();
@ -856,8 +879,11 @@ Future<Void> preresolutionProcessing(CommitBatchContext* self) {
self->releaseDelay = delay(computeReleaseDelay(self, latencyBucket), TaskPriority::ProxyMasterVersionReply);
if (debugID.present()) {
g_traceBatch.addEvent(
"CommitDebug", debugID.get().first(), "CommitProxyServer.commitBatch.GettingCommitVersion");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.GettingCommitVersion",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
if (SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST) {
@ -897,7 +923,11 @@ Future<Void> preresolutionProcessing(CommitBatchContext* self) {
//TraceEvent("ProxyGotVer", pProxyContext->dbgid).detail("Commit", commitVersion).detail("Prev", prevVersion);
if (debugID.present()) {
g_traceBatch.addEvent("CommitDebug", debugID.get().first(), "CommitProxyServer.commitBatch.GotCommitVersion");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.GotCommitVersion",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
}
@ -936,14 +966,14 @@ Future<Void> getResolution(CommitBatchContext* self) {
Future<ResolveTransactionBatchReply> singleResolverReply;
double singleResolverStart = 0;
if (pProxyCommitData->resolvers.size() == 1) {
requests.requests[0].debugID = self->debugID;
requests.requests[0].debugID = self->getDebugID();
requests.requests[0].writtenTags = self->writtenTagsPreResolution;
singleResolverStart = g_network->timer_monotonic();
singleResolverReply = brokenPromiseToNever(
pProxyCommitData->resolvers[0].resolve.getReply(requests.requests[0], TaskPriority::ProxyResolverReply));
} else {
for (int r = 0; r < pProxyCommitData->resolvers.size(); r++) {
requests.requests[r].debugID = self->debugID;
requests.requests[r].debugID = self->getDebugID();
requests.requests[r].writtenTags = self->writtenTagsPreResolution;
replies.push_back(
trackResolutionMetrics(pProxyCommitData->stats.resolverDist[r],
@ -984,9 +1014,13 @@ Future<Void> getResolution(CommitBatchContext* self) {
}
self->pProxyCommitData->stats.resolutionDist->sampleSeconds(g_network->timer_monotonic() - resolutionStart);
if (self->debugID.present()) {
g_traceBatch.addEvent(
"CommitDebug", self->debugID.get().first(), "CommitProxyServer.commitBatch.AfterResolution");
if (self->debugIDs.present()) {
Optional<UID> debugID = self->getDebugID();
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.AfterResolution",
self->debugIDs.get().debugTraceID,
self->debugIDs.get().debugSpanID);
}
}
@ -1546,7 +1580,8 @@ Future<Void> postResolution(CommitBatchContext* self) {
ProxyCommitData* const pProxyCommitData = self->pProxyCommitData;
std::vector<CommitTransactionRequest>& trs = self->trs;
const int64_t localBatchNumber = self->localBatchNumber;
const Optional<UID>& debugID = self->debugID;
Optional<BatchDebugIDs> debugIDs = self->debugIDs;
Optional<UID> debugID = self->getDebugID();
Span span("MP:postResolution"_loc, self->span.context);
bool queuedCommits = pProxyCommitData->latestLocalCommitBatchLogging.get() < localBatchNumber - 1;
@ -1561,8 +1596,11 @@ Future<Void> postResolution(CommitBatchContext* self) {
pProxyCommitData->stats.txnCommitResolved += trs.size();
if (debugID.present()) {
g_traceBatch.addEvent(
"CommitDebug", debugID.get().first(), "CommitProxyServer.commitBatch.ProcessingMutations");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.ProcessingMutations",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
self->isMyFirstBatch = !pProxyCommitData->version.get();
@ -1573,15 +1611,21 @@ Future<Void> postResolution(CommitBatchContext* self) {
applyMetadataEffect(self);
if (debugID.present()) {
g_traceBatch.addEvent(
"CommitDebug", debugID.get().first(), "CommitProxyServer.commitBatch.ApplyMetadataEffect");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.ApplyMetadataEffect",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
determineCommittedTransactions(self);
if (debugID.present()) {
g_traceBatch.addEvent(
"CommitDebug", debugID.get().first(), "CommitProxyServer.commitBatch.DetermineCommittedTransactions");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.DetermineCommittedTransactions",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
if (self->forceRecovery) {
@ -1592,8 +1636,11 @@ Future<Void> postResolution(CommitBatchContext* self) {
co_await applyMetadataToCommittedTransactions(self);
if (debugID.present()) {
g_traceBatch.addEvent(
"CommitDebug", debugID.get().first(), "CommitProxyServer.commitBatch.ApplyMetadataToCommittedTxn");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.ApplyMetadataToCommittedTxn",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
// After applyed metadata change, this commit proxy has the latest view of locked ranges.
@ -1607,7 +1654,11 @@ Future<Void> postResolution(CommitBatchContext* self) {
co_await assignMutationsToStorageServers(self);
if (debugID.present()) {
g_traceBatch.addEvent("CommitDebug", debugID.get().first(), "CommitProxyServer.commitBatch.AssignMutationToSS");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.AssignMutationToSS",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
// Serialize and backup the mutations as a single mutation
@ -1730,9 +1781,12 @@ Future<Void> postResolution(CommitBatchContext* self) {
self->msg = self->storeCommits.back().first.get();
if (self->debugID.present())
g_traceBatch.addEvent(
"CommitDebug", self->debugID.get().first(), "CommitProxyServer.commitBatch.AfterStoreCommits");
if (self->debugIDs.present())
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.AfterStoreCommits",
self->debugIDs.get().debugTraceID,
self->debugIDs.get().debugSpanID);
// txnState (transaction subsystem state) tag: message extracted from log adapter
bool firstMessage = true;
@ -1774,7 +1828,7 @@ Future<Void> postResolution(CommitBatchContext* self) {
pProxyCommitData->committedVersion.get(),
pProxyCommitData->minKnownCommittedVersion };
self->loggingComplete =
pProxyCommitData->logSystem->push(versionSet, self->toCommit, span.context, self->debugID, tpcvMap);
pProxyCommitData->logSystem->push(versionSet, self->toCommit, span.context, self->getDebugID(), tpcvMap);
float ratio = self->toCommit.getEmptyMessageRatio();
pProxyCommitData->stats.commitBatchingEmptyMessageRatio.addMeasurement(ratio);
@ -1868,7 +1922,8 @@ Future<Void> reply(CommitBatchContext* self) {
ProxyCommitData* const pProxyCommitData = self->pProxyCommitData;
Span span("MP:reply"_loc, self->span.context);
const Optional<UID>& debugID = self->debugID;
Optional<BatchDebugIDs> debugIDs = self->debugIDs;
Optional<UID> debugID = self->getDebugID();
if (!SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST) {
// Version vector/unicast is disabled: Logging completed, so the current version (and all versions prior to
@ -1888,7 +1943,11 @@ Future<Void> reply(CommitBatchContext* self) {
// .detail("PrevVersion", self->prevVersion)
// .detail("Version", self->commitVersion);
if (debugID.present())
g_traceBatch.addEvent("CommitDebug", debugID.get().first(), "CommitProxyServer.commitBatch.AfterLogPush");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.AfterLogPush",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
// After logging finishes, we report the commit version to master so that every other proxy can get the most
// up-to-date live committed version. We also maintain the invariant that master's committed version >=
@ -1915,8 +1974,11 @@ Future<Void> reply(CommitBatchContext* self) {
}
if (debugID.present()) {
g_traceBatch.addEvent(
"CommitDebug", debugID.get().first(), "CommitProxyServer.commitBatch.AfterReportRawCommittedVersion");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"CommitProxyServer.commitBatch.AfterReportRawCommittedVersion",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
if (SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST) {

49
fdbserver/grvproxy/GrvProxyServer.cpp
View File

@ -598,7 +598,9 @@ Future<Void> queueGetReadVersionRequests(Reference<AsyncVar<ServerDBInfo> const>
if (req.debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
req.debugID.get().first(),
"GrvProxyServer.queueTransactionStartRequests.Before");
"GrvProxyServer.queueTransactionStartRequests.Before",
req.spanContext.traceID,
req.spanContext.spanID);
}
if (systemQueue->empty() && defaultQueue->empty() && batchQueue->empty()) {
@ -685,10 +687,14 @@ Future<Void> lastCommitUpdater(GrvProxyData* self, PromiseStream<Future<Void>> a
}
}
Optional<UID> getDebugID(Optional<BatchDebugIDs> debugIDs) {
return debugIDs.present() ? Optional<UID>(debugIDs.get().debugID) : Optional<UID>();
}
Future<GetReadVersionReply> getLiveCommittedVersion(std::vector<SpanContext> spanContexts,
GrvProxyData* grvProxyData,
uint32_t flags,
Optional<UID> debugID,
Optional<BatchDebugIDs> debugIDs,
int transactionCount,
int systemTransactionCount,
int defaultPriTransactionCount,
@ -705,6 +711,7 @@ Future<GetReadVersionReply> getLiveCommittedVersion(std::vector<SpanContext> spa
++grvProxyData->stats.txnStartBatch;
double grvStart = now();
Optional<UID> debugID = getDebugID(debugIDs);
Future<GetRawCommittedVersionReply> replyFromMasterFuture;
replyFromMasterFuture = grvProxyData->master.getLiveCommittedVersion.getReply(
GetRawCommittedVersionRequest(span.context, debugID, grvProxyData->ssVersionVectorCache.getMaxVersion()),
@ -720,8 +727,11 @@ Future<GetReadVersionReply> getLiveCommittedVersion(std::vector<SpanContext> spa
double grvConfirmEpochLive = now();
grvProxyData->stats.grvConfirmEpochLiveDist->sampleSeconds(grvConfirmEpochLive - grvStart);
if (debugID.present()) {
g_traceBatch.addEvent(
"TransactionDebug", debugID.get().first(), "GrvProxyServer.getLiveCommittedVersion.confirmEpochLive");
g_traceBatch.addEvent("TransactionDebug",
debugID.get().first(),
"GrvProxyServer.getLiveCommittedVersion.confirmEpochLive",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
GetRawCommittedVersionReply repFromMaster =
@ -748,8 +758,11 @@ Future<GetReadVersionReply> getLiveCommittedVersion(std::vector<SpanContext> spa
: g_network->networkInfo.metrics.lastRunLoopBusyness);
if (debugID.present()) {
g_traceBatch.addEvent(
"TransactionDebug", debugID.get().first(), "GrvProxyServer.getLiveCommittedVersion.After");
g_traceBatch.addEvent("TransactionDebug",
debugID.get().first(),
"GrvProxyServer.getLiveCommittedVersion.After",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
grvProxyData->stats.txnStartOut += transactionCount;
@ -1016,7 +1029,7 @@ static Future<Void> transactionStarter(GrvProxyInterface proxy,
std::vector<std::vector<GetReadVersionRequest>> start(
2); // start[0] is transactions starting with !(flags&CAUSAL_READ_RISKY), start[1] is transactions starting
// with flags&CAUSAL_READ_RISKY
Optional<UID> debugID;
Optional<BatchDebugIDs> debugIDs;
int requestsToStart = 0;
@ -1046,10 +1059,17 @@ static Future<Void> transactionStarter(GrvProxyInterface proxy,
}
if (req.debugID.present()) {
if (!debugID.present()) {
debugID = nondeterministicRandom()->randomUniqueID();
if (!debugIDs.present()) {
debugIDs = BatchDebugIDs{ nondeterministicRandom()->randomUniqueID(),
req.spanContext.traceID,
req.spanContext.spanID };
}
g_traceBatch.addAttach("TransactionAttachID", req.debugID.get().first(), debugID.get().first());
Optional<UID> debugID = getDebugID(debugIDs);
g_traceBatch.addAttach("TransactionAttachID",
req.debugID.get().first(),
debugID.get().first(),
req.spanContext.traceID,
req.spanContext.spanID);
}
transactionsStarted[req.flags & 1] += tc;
@ -1124,10 +1144,13 @@ static Future<Void> transactionStarter(GrvProxyInterface proxy,
systemQueue.empty() && defaultQueue.empty() && batchQueue.empty(),
elapsed);
if (debugID.present()) {
if (debugIDs.present()) {
Optional<UID> debugID = getDebugID(debugIDs);
g_traceBatch.addEvent("TransactionDebug",
debugID.get().first(),
"GrvProxyServer.transactionStarter.AskLiveCommittedVersionFromMaster");
"GrvProxyServer.transactionStarter.AskLiveCommittedVersionFromMaster",
debugIDs.get().debugTraceID,
debugIDs.get().debugSpanID);
}
int defaultGRVProcessed = 0;
@ -1143,7 +1166,7 @@ static Future<Void> transactionStarter(GrvProxyInterface proxy,
Future<GetReadVersionReply> readVersionReply = getLiveCommittedVersion(spanContexts,
grvProxyData,
i,
debugID,
debugIDs,
transactionsStarted[i],
systemTransactionsStarted[i],
defaultPriTransactionsStarted[i],

30
fdbserver/resolver/Resolver.cpp
View File

@ -272,8 +272,16 @@ Future<Void> resolveBatch(Reference<Resolver> self, ResolveTransactionBatchReque
if (req.debugID.present()) {
debugID = nondeterministicRandom()->randomUniqueID();
g_traceBatch.addAttach("CommitAttachID", req.debugID.get().first(), debugID.get().first());
g_traceBatch.addEvent("CommitDebug", debugID.get().first(), "Resolver.resolveBatch.Before");
g_traceBatch.addAttach("CommitAttachID",
req.debugID.get().first(),
debugID.get().first(),
req.spanContext.traceID,
req.spanContext.spanID);
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"Resolver.resolveBatch.Before",
req.spanContext.traceID,
req.spanContext.spanID);
}
/* TraceEvent("ResolveBatchStart", self->dbgid).detail("From", proxyAddress).detail("Version",
@ -295,7 +303,11 @@ Future<Void> resolveBatch(Reference<Resolver> self, ResolveTransactionBatchReque
}
if (debugID.present()) {
g_traceBatch.addEvent("CommitDebug", debugID.get().first(), "Resolver.resolveBatch.AfterQueueSizeCheck");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"Resolver.resolveBatch.AfterQueueSizeCheck",
req.spanContext.traceID,
req.spanContext.spanID);
}
co_await versionReady(self.getPtr(), &proxyInfo, req.prevVersion);
@ -328,7 +340,11 @@ Future<Void> resolveBatch(Reference<Resolver> self, ResolveTransactionBatchReque
proxyInfo.lastVersion = req.version;
if (req.debugID.present())
g_traceBatch.addEvent("CommitDebug", debugID.get().first(), "Resolver.resolveBatch.AfterOrderer");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"Resolver.resolveBatch.AfterOrderer",
req.spanContext.traceID,
req.spanContext.spanID);
ResolveTransactionBatchReply& reply = proxyInfo.outstandingBatches[req.version];
reply.writtenTags = req.writtenTags;
@ -530,7 +546,11 @@ Future<Void> resolveBatch(Reference<Resolver> self, ResolveTransactionBatchReque
self->computeTimeDist->sampleSeconds(endComputeTime - beginComputeTime);
if (req.debugID.present())
g_traceBatch.addEvent("CommitDebug", debugID.get().first(), "Resolver.resolveBatch.After");
g_traceBatch.addEvent("CommitDebug",
debugID.get().first(),
"Resolver.resolveBatch.After",
req.spanContext.traceID,
req.spanContext.spanID);
} else {
CODE_PROBE(true, "Duplicate resolve batch request");
//TraceEvent("DupResolveBatchReq", self->dbgid).detail("From", proxyAddress);

4
fdbserver/sequencer/masterserver.cpp
View File

@ -255,7 +255,9 @@ class LiveCommittedVersionServer {
if (req.debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
req.debugID.get().first(),
"MasterServer.serveLiveCommittedVersion.GetRawCommittedVersion");
"MasterServer.serveLiveCommittedVersion.GetRawCommittedVersion",
req.spanContext.traceID,
req.spanContext.spanID);
}
if (self->liveCommittedVersion.get() == invalidVersion) {

184
fdbserver/storageserver/storageserver.cpp
View File

@ -812,9 +812,15 @@ public:
int watchReplyCount{ 0 };
Optional<TagSet> tags;
Optional<UID> debugID;
SpanContext spanContext;
ServerWatchMetadata(Key key, Optional<Value> value, Version version, Optional<TagSet> tags, Optional<UID> debugID)
: key(key), value(value), version(version), tags(tags), debugID(debugID) {}
ServerWatchMetadata(Key key,
Optional<Value> value,
Version version,
Optional<TagSet> tags,
Optional<UID> debugID,
SpanContext spanContext)
: key(key), value(value), version(version), tags(tags), debugID(debugID), spanContext(spanContext) {}
};
struct BusiestWriteTagContext {
@ -2146,7 +2152,9 @@ Future<Void> getValueQ(StorageServer* data, GetValueRequest req) {
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("GetValueDebug",
req.options.get().debugID.get().first(),
"getValueQ.DoRead"); //.detail("TaskID", g_network->getCurrentTask());
"getValueQ.DoRead",
req.spanContext.traceID,
req.spanContext.spanID); //.detail("TaskID", g_network->getCurrentTask());
}
Optional<Value> v;
@ -2158,7 +2166,9 @@ Future<Void> getValueQ(StorageServer* data, GetValueRequest req) {
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("GetValueDebug",
req.options.get().debugID.get().first(),
"getValueQ.AfterVersion"); //.detail("TaskID", g_network->getCurrentTask());
"getValueQ.AfterVersion",
req.spanContext.traceID,
req.spanContext.spanID); //.detail("TaskID", g_network->getCurrentTask());
}
uint64_t changeCounter = data->shardChangeCounter;
@ -2225,7 +2235,9 @@ Future<Void> getValueQ(StorageServer* data, GetValueRequest req) {
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("GetValueDebug",
req.options.get().debugID.get().first(),
"getValueQ.AfterRead"); //.detail("TaskID", g_network->getCurrentTask());
"getValueQ.AfterRead",
req.spanContext.traceID,
req.spanContext.spanID); //.detail("TaskID", g_network->getCurrentTask());
}
GetValueReply reply(v, /*cached=*/false);
@ -2266,7 +2278,9 @@ Future<Version> watchWaitForValueChange(coro::FrameSizeRecorder, StorageServer*
if (metadata->debugID.present()) {
g_traceBatch.addEvent("WatchValueDebug",
metadata->debugID.get().first(),
"watchValueSendReply.Before"); //.detail("TaskID", g_network->getCurrentTask());
"watchValueSendReply.Before",
metadata->spanContext.traceID,
metadata->spanContext.spanID); //.detail("TaskID", g_network->getCurrentTask());
}
Version originalMetadataVersion = metadata->version;
@ -2274,7 +2288,9 @@ Future<Version> watchWaitForValueChange(coro::FrameSizeRecorder, StorageServer*
if (metadata->debugID.present()) {
g_traceBatch.addEvent("WatchValueDebug",
metadata->debugID.get().first(),
"watchValueSendReply.AfterVersion"); //.detail("TaskID", g_network->getCurrentTask());
"watchValueSendReply.AfterVersion",
metadata->spanContext.traceID,
metadata->spanContext.spanID); //.detail("TaskID", g_network->getCurrentTask());
}
Version minVersion = data->data().latestVersion;
@ -2311,10 +2327,11 @@ Future<Version> watchWaitForValueChange(coro::FrameSizeRecorder, StorageServer*
data->thisServerID);
if (metadata->debugID.present()) {
g_traceBatch.addEvent(
"WatchValueDebug",
metadata->debugID.get().first(),
"watchValueSendReply.AfterRead"); //.detail("TaskID", g_network->getCurrentTask());
g_traceBatch.addEvent("WatchValueDebug",
metadata->debugID.get().first(),
"watchValueSendReply.AfterRead",
metadata->spanContext.traceID,
metadata->spanContext.spanID); //.detail("TaskID", g_network->getCurrentTask());
}
// If the version we read is less than minVersion, then we may fail to be notified of any changes that occur
@ -2355,8 +2372,11 @@ Future<Version> watchWaitForValueChange(coro::FrameSizeRecorder, StorageServer*
}
if (metadata->debugID.present())
g_traceBatch.addEvent(
"WatchValueDebug", metadata->debugID.get().first(), "watchValueSendReply.WaitChange");
g_traceBatch.addEvent("WatchValueDebug",
metadata->debugID.get().first(),
"watchValueSendReply.WaitChange",
metadata->spanContext.traceID,
metadata->spanContext.spanID);
co_await watchFuture;
data->watchBytes -= watchBytes;
} catch (Error& e) {
@ -3353,8 +3373,11 @@ Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
try {
if (req.options.present() && req.options.get().debugID.present())
g_traceBatch.addEvent(
"TransactionDebug", req.options.get().debugID.get().first(), "storageserver.getKeyValues.Before");
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getKeyValues.Before",
req.spanContext.traceID,
req.spanContext.spanID);
Version commitVersion = getLatestCommitVersion(req.ssLatestCommitVersions, data->tag);
Version version = co_await waitForVersion(data, commitVersion, req.version, span.context);
@ -3374,8 +3397,11 @@ Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
KeyRange shard = getShardKeyRange(data, req.begin);
if (req.options.present() && req.options.get().debugID.present())
g_traceBatch.addEvent(
"TransactionDebug", req.options.get().debugID.get().first(), "storageserver.getKeyValues.AfterVersion");
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getKeyValues.AfterVersion",
req.spanContext.traceID,
req.spanContext.spanID);
//.detail("ShardBegin", shard.begin).detail("ShardEnd", shard.end);
//} catch (Error& e) { TraceEvent("WrongShardServer", data->thisServerID).detail("Begin",
// req.begin.toString()).detail("End", req.end.toString()).detail("Version", version).detail("Shard",
@ -3404,8 +3430,11 @@ Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
Key end = co_await fEnd;
if (req.options.present() && req.options.get().debugID.present())
g_traceBatch.addEvent(
"TransactionDebug", req.options.get().debugID.get().first(), "storageserver.getKeyValues.AfterKeys");
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getKeyValues.AfterKeys",
req.spanContext.traceID,
req.spanContext.spanID);
//.detail("Off1",offset1).detail("Off2",offset2).detail("ReqBegin",req.begin.getKey()).detail("ReqEnd",req.end.getKey());
// Offsets of zero indicate begin/end keys in this shard, which obviously means we can answer the query
@ -3434,8 +3463,11 @@ Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
if (begin >= end) {
if (req.options.present() && req.options.get().debugID.present())
g_traceBatch.addEvent(
"TransactionDebug", req.options.get().debugID.get().first(), "storageserver.getKeyValues.Send");
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getKeyValues.Send",
req.spanContext.traceID,
req.spanContext.spanID);
//.detail("Begin",begin).detail("End",end);
GetKeyValuesReply none;
@ -3464,7 +3496,9 @@ Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getKeyValues.AfterReadRange");
"storageserver.getKeyValues.AfterReadRange",
req.spanContext.traceID,
req.spanContext.spanID);
}
//.detail("Begin",begin).detail("End",end).detail("SizeOf",r.data.size());
data->checkChangeCounter(
@ -3547,7 +3581,9 @@ Future<GetRangeReqAndResultRef> quickGetKeyValues(StorageServer* data,
if (pOriginalReq->options.present() && pOriginalReq->options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
pOriginalReq->options.get().debugID.get().first(),
"storageserver.quickGetKeyValues.Before");
"storageserver.quickGetKeyValues.Before",
pOriginalReq->spanContext.traceID,
pOriginalReq->spanContext.spanID);
}
try {
// TODO: Use a lower level API may be better?
@ -3583,7 +3619,9 @@ Future<GetRangeReqAndResultRef> quickGetKeyValues(StorageServer* data,
if (pOriginalReq->options.present() && pOriginalReq->options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
pOriginalReq->options.get().debugID.get().first(),
"storageserver.quickGetKeyValues.AfterLocalFetch");
"storageserver.quickGetKeyValues.AfterLocalFetch",
pOriginalReq->spanContext.traceID,
pOriginalReq->spanContext.spanID);
}
co_return getRange;
}
@ -3613,7 +3651,9 @@ Future<GetRangeReqAndResultRef> quickGetKeyValues(StorageServer* data,
if (pOriginalReq->options.present() && pOriginalReq->options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
pOriginalReq->options.get().debugID.get().first(),
"storageserver.quickGetKeyValues.AfterRemoteFetch");
"storageserver.quickGetKeyValues.AfterRemoteFetch",
pOriginalReq->spanContext.traceID,
pOriginalReq->spanContext.spanID);
}
co_return getRange;
} else {
@ -5544,8 +5584,11 @@ Future<GetMappedKeyValuesReply> mapKeyValues(StorageServer* data,
result.data.reserve(result.arena, input.data.size());
if (pOriginalReq->options.present() && pOriginalReq->options.get().debugID.present())
g_traceBatch.addEvent(
"TransactionDebug", pOriginalReq->options.get().debugID.get().first(), "storageserver.mapKeyValues.Start");
g_traceBatch.addEvent("TransactionDebug",
pOriginalReq->options.get().debugID.get().first(),
"storageserver.mapKeyValues.Start",
pOriginalReq->spanContext.traceID,
pOriginalReq->spanContext.spanID);
Tuple mappedKeyFormatTuple;
try {
@ -5566,7 +5609,9 @@ Future<GetMappedKeyValuesReply> mapKeyValues(StorageServer* data,
if (pOriginalReq->options.present() && pOriginalReq->options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
pOriginalReq->options.get().debugID.get().first(),
"storageserver.mapKeyValues.BeforeLoop");
"storageserver.mapKeyValues.BeforeLoop",
pOriginalReq->spanContext.traceID,
pOriginalReq->spanContext.spanID);
}
for (; (offset < sz) && (*remainingLimitBytes > 0); offset += SERVER_KNOBS->MAX_PARALLEL_QUICK_GET_VALUE) {
@ -5591,7 +5636,9 @@ Future<GetMappedKeyValuesReply> mapKeyValues(StorageServer* data,
if (pOriginalReq->options.present() && pOriginalReq->options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
pOriginalReq->options.get().debugID.get().first(),
"storageserver.mapKeyValues.AfterBatch");
"storageserver.mapKeyValues.AfterBatch",
pOriginalReq->spanContext.traceID,
pOriginalReq->spanContext.spanID);
}
subqueries.clear();
for (int i = 0; i + offset < sz && i < SERVER_KNOBS->MAX_PARALLEL_QUICK_GET_VALUE; i++) {
@ -5619,7 +5666,9 @@ Future<GetMappedKeyValuesReply> mapKeyValues(StorageServer* data,
if (pOriginalReq->options.present() && pOriginalReq->options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
pOriginalReq->options.get().debugID.get().first(),
"storageserver.mapKeyValues.AfterAll");
"storageserver.mapKeyValues.AfterAll",
pOriginalReq->spanContext.traceID,
pOriginalReq->spanContext.spanID);
}
co_return result;
}
@ -5655,8 +5704,11 @@ Future<Void> getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRequest
try {
if (req.options.present() && req.options.get().debugID.present())
g_traceBatch.addEvent(
"TransactionDebug", req.options.get().debugID.get().first(), "storageserver.getMappedKeyValues.Before");
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getMappedKeyValues.Before",
req.spanContext.traceID,
req.spanContext.spanID);
// VERSION_VECTOR change
Version commitVersion = getLatestCommitVersion(req.ssLatestCommitVersions, data->tag);
Version version = co_await waitForVersion(data, commitVersion, req.version, span.context);
@ -5670,7 +5722,9 @@ Future<Void> getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRequest
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getMappedKeyValues.AfterVersion");
"storageserver.getMappedKeyValues.AfterVersion",
req.spanContext.traceID,
req.spanContext.spanID);
}
//.detail("ShardBegin", shard.begin).detail("ShardEnd", shard.end);
//} catch (Error& e) { TraceEvent("WrongShardServer", data->thisServerID).detail("Begin",
@ -5698,7 +5752,9 @@ Future<Void> getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRequest
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getMappedKeyValues.AfterKeys");
"storageserver.getMappedKeyValues.AfterKeys",
req.spanContext.traceID,
req.spanContext.spanID);
}
//.detail("Off1",offset1).detail("Off2",offset2).detail("ReqBegin",req.begin.getKey()).detail("ReqEnd",req.end.getKey());
@ -5720,7 +5776,9 @@ Future<Void> getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRequest
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getMappedKeyValues.Send");
"storageserver.getMappedKeyValues.Send",
req.spanContext.traceID,
req.spanContext.spanID);
}
//.detail("Begin",begin).detail("End",end);
@ -5762,7 +5820,9 @@ Future<Void> getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRequest
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getMappedKeyValues.AfterReadRange");
"storageserver.getMappedKeyValues.AfterReadRange",
req.spanContext.traceID,
req.spanContext.spanID);
}
//.detail("Begin",begin).detail("End",end).detail("SizeOf",r.data.size());
data->checkChangeCounter(
@ -5832,8 +5892,11 @@ Future<Void> getKeyValuesStreamQ(StorageServer* data, GetKeyValuesStreamRequest
try {
if (req.options.present() && req.options.get().debugID.present())
g_traceBatch.addEvent(
"TransactionDebug", req.options.get().debugID.get().first(), "storageserver.getKeyValuesStream.Before");
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getKeyValuesStream.Before",
req.spanContext.traceID,
req.spanContext.spanID);
Version commitVersion = getLatestCommitVersion(req.ssLatestCommitVersions, data->tag);
Version version = co_await waitForVersion(data, commitVersion, req.version, span.context);
@ -5845,7 +5908,9 @@ Future<Void> getKeyValuesStreamQ(StorageServer* data, GetKeyValuesStreamRequest
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getKeyValuesStream.AfterVersion");
"storageserver.getKeyValuesStream.AfterVersion",
req.spanContext.traceID,
req.spanContext.spanID);
}
//.detail("ShardBegin", shard.begin).detail("ShardEnd", shard.end);
//} catch (Error& e) { TraceEvent("WrongShardServer", data->thisServerID).detail("Begin",
@ -5872,7 +5937,9 @@ Future<Void> getKeyValuesStreamQ(StorageServer* data, GetKeyValuesStreamRequest
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getKeyValuesStream.AfterKeys");
"storageserver.getKeyValuesStream.AfterKeys",
req.spanContext.traceID,
req.spanContext.spanID);
}
//.detail("Off1",offset1).detail("Off2",offset2).detail("ReqBegin",req.begin.getKey()).detail("ReqEnd",req.end.getKey());
@ -5894,7 +5961,9 @@ Future<Void> getKeyValuesStreamQ(StorageServer* data, GetKeyValuesStreamRequest
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getKeyValuesStream.Send");
"storageserver.getKeyValuesStream.Send",
req.spanContext.traceID,
req.spanContext.spanID);
}
//.detail("Begin",begin).detail("End",end);
@ -5938,7 +6007,9 @@ Future<Void> getKeyValuesStreamQ(StorageServer* data, GetKeyValuesStreamRequest
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("TransactionDebug",
req.options.get().debugID.get().first(),
"storageserver.getKeyValuesStream.AfterReadRange");
"storageserver.getKeyValuesStream.AfterReadRange",
req.spanContext.traceID,
req.spanContext.spanID);
}
//.detail("Begin",begin).detail("End",end).detail("SizeOf",r.data.size());
data->checkChangeCounter(
@ -11579,7 +11650,9 @@ Future<Void> serveGetValueRequests(StorageServer* self, FutureStream<GetValueReq
if (req.options.present() && req.options.get().debugID.present()) {
g_traceBatch.addEvent("GetValueDebug",
req.options.get().debugID.get().first(),
"storageServer.received"); //.detail("TaskID", g_network->getCurrentTask());
"storageServer.received",
req.spanContext.traceID,
req.spanContext.spanID); //.detail("TaskID", g_network->getCurrentTask());
}
if (SHORT_CIRCUT_ACTUAL_STORAGE && normalKeys.contains(req.key))
@ -11657,7 +11730,8 @@ Future<Void> serveWatchValueRequestsImpl(StorageServer* self, FutureStream<Watch
// case 1: no watch set for the current key
if (!metadata.isValid()) {
metadata = makeReference<ServerWatchMetadata>(req.key, req.value, req.version, req.tags, req.debugID);
metadata = makeReference<ServerWatchMetadata>(
req.key, req.value, req.version, req.tags, req.debugID, req.spanContext);
KeyRef key = self->setWatchMetadata(metadata);
metadata->watch_impl = forward(watchWaitForValueChange(self, span.context, key), metadata->versionPromise);
addWatchValueReply(self, req, metadata, span.context);
@ -11666,11 +11740,17 @@ Future<Void> serveWatchValueRequestsImpl(StorageServer* self, FutureStream<Watch
else if (metadata->value == req.value) {
if (req.debugID.present()) {
if (metadata->debugID.present()) {
g_traceBatch.addAttach(
"WatchRequestCase2", req.debugID.get().first(), metadata->debugID.get().first());
g_traceBatch.addAttach("WatchRequestCase2",
req.debugID.get().first(),
metadata->debugID.get().first(),
req.spanContext.traceID,
req.spanContext.spanID);
} else {
g_traceBatch.addEvent(
"WatchValueDebug", metadata->debugID.get().first(), "watchValueSendReply.Case2");
g_traceBatch.addEvent("WatchValueDebug",
req.debugID.get().first(),
"watchValueSendReply.Case2",
req.spanContext.traceID,
req.spanContext.spanID);
}
}
@ -11680,6 +11760,7 @@ Future<Void> serveWatchValueRequestsImpl(StorageServer* self, FutureStream<Watch
if (req.debugID.present()) {
metadata->debugID = req.debugID;
}
metadata->spanContext = req.spanContext;
}
addWatchValueReply(self, req, metadata, span.context);
@ -11690,7 +11771,8 @@ Future<Void> serveWatchValueRequestsImpl(StorageServer* self, FutureStream<Watch
metadata->versionPromise.send(req.version);
metadata->watch_impl.cancel();
metadata = makeReference<ServerWatchMetadata>(req.key, req.value, req.version, req.tags, req.debugID);
metadata = makeReference<ServerWatchMetadata>(
req.key, req.value, req.version, req.tags, req.debugID, req.spanContext);
KeyRef key = self->setWatchMetadata(metadata);
metadata->watch_impl = forward(watchWaitForValueChange(self, span.context, key), metadata->versionPromise);
@ -11722,8 +11804,8 @@ Future<Void> serveWatchValueRequestsImpl(StorageServer* self, FutureStream<Watch
}
if (reply.value == req.value) { // valSS == valreq
metadata =
makeReference<ServerWatchMetadata>(req.key, req.value, req.version, req.tags, req.debugID);
metadata = makeReference<ServerWatchMetadata>(
req.key, req.value, req.version, req.tags, req.debugID, req.spanContext);
KeyRef key = self->setWatchMetadata(metadata);
metadata->watch_impl =
forward(watchWaitForValueChange(self, span.context, key), metadata->versionPromise);

30
fdbserver/tlog/TLogServer.cpp
View File

@ -2837,8 +2837,16 @@ Future<Void> tLogCommit(TLogData* self,
Optional<UID> tlogDebugID;
if (req.debugID.present()) {
tlogDebugID = nondeterministicRandom()->randomUniqueID();
g_traceBatch.addAttach("CommitAttachID", req.debugID.get().first(), tlogDebugID.get().first());
g_traceBatch.addEvent("CommitDebug", tlogDebugID.get().first(), "TLog.tLogCommit.BeforeWaitForVersion");
g_traceBatch.addAttach("CommitAttachID",
req.debugID.get().first(),
tlogDebugID.get().first(),
req.spanContext.traceID,
req.spanContext.spanID);
g_traceBatch.addEvent("CommitDebug",
tlogDebugID.get().first(),
"TLog.tLogCommit.BeforeWaitForVersion",
req.spanContext.traceID,
req.spanContext.spanID);
}
logData->minKnownCommittedVersion = std::max(logData->minKnownCommittedVersion, req.minKnownCommittedVersion);
@ -2874,7 +2882,11 @@ Future<Void> tLogCommit(TLogData* self,
bool isNotDuplicate = (logData->version.get() == req.prevVersion);
if (isNotDuplicate) {
if (req.debugID.present())
g_traceBatch.addEvent("CommitDebug", tlogDebugID.get().first(), "TLog.tLogCommit.Before");
g_traceBatch.addEvent("CommitDebug",
tlogDebugID.get().first(),
"TLog.tLogCommit.Before",
req.spanContext.traceID,
req.spanContext.spanID);
//TraceEvent("TLogCommit", logData->logId).detail("Version", req.version);
commitMessages(self, logData, req.version, req.arena, req.messages);
@ -2901,7 +2913,11 @@ Future<Void> tLogCommit(TLogData* self,
}
if (req.debugID.present())
g_traceBatch.addEvent("CommitDebug", tlogDebugID.get().first(), "TLog.tLogCommit.AfterTLogCommit");
g_traceBatch.addEvent("CommitDebug",
tlogDebugID.get().first(),
"TLog.tLogCommit.AfterTLogCommit",
req.spanContext.traceID,
req.spanContext.spanID);
}
// Send replies only once all prior messages have been received and committed.
Future<Void> stopped = logData->stopCommit.onTrigger();
@ -2918,7 +2934,11 @@ Future<Void> tLogCommit(TLogData* self,
}
if (req.debugID.present())
g_traceBatch.addEvent("CommitDebug", tlogDebugID.get().first(), "TLog.tLogCommit.After");
g_traceBatch.addEvent("CommitDebug",
tlogDebugID.get().first(),
"TLog.tLogCommit.After",
req.spanContext.traceID,
req.spanContext.spanID);
req.reply.send(logData->durableKnownCommittedVersion);

30
flow/Trace.cpp
View File

@ -1473,23 +1473,24 @@ bool TraceBatch::dumpImmediately() {
return (g_network->isSimulated() || FLOW_KNOBS->AUTOMATIC_TRACE_DUMP);
}
void TraceBatch::addEvent(const char* name, uint64_t id, const char* location) {
void TraceBatch::addEvent(const char* name, uint64_t id, const char* location, UID traceID, uint64_t spanID) {
if (FLOW_KNOBS->MIN_TRACE_SEVERITY > TRACE_BATCH_IMPLICIT_SEVERITY) {
return;
}
auto& eventInfo =
eventBatch.emplace_back(EventInfo(TraceEvent::getCurrentTime(), ::timer_monotonic(), name, id, location));
auto& eventInfo = eventBatch.emplace_back(
EventInfo(TraceEvent::getCurrentTime(), ::timer_monotonic(), name, id, location, traceID, spanID));
if (dumpImmediately())
dump();
else
g_traceLog.annotateEvent(eventInfo.fields);
}
void TraceBatch::addAttach(const char* name, uint64_t id, uint64_t to) {
void TraceBatch::addAttach(const char* name, uint64_t id, uint64_t to, UID traceID, uint64_t spanID) {
if (FLOW_KNOBS->MIN_TRACE_SEVERITY > TRACE_BATCH_IMPLICIT_SEVERITY) {
return;
}
auto& attachInfo = attachBatch.emplace_back(AttachInfo(TraceEvent::getCurrentTime(), name, id, to));
auto& attachInfo =
attachBatch.emplace_back(AttachInfo(TraceEvent::getCurrentTime(), name, id, to, traceID, spanID));
if (dumpImmediately())
dump();
else
@ -1551,7 +1552,9 @@ TraceBatch::EventInfo::EventInfo(double time,
double monotonicTime,
const char* name,
uint64_t id,
const char* location) {
const char* location,
UID traceID,
uint64_t spanID) {
fields.addField("Severity", format("%d", (int)TRACE_BATCH_IMPLICIT_SEVERITY));
fields.addField("Time", format("%.6f", time));
// Include monotonic time for computing elapsed time between events on the same machine.
@ -1563,9 +1566,18 @@ TraceBatch::EventInfo::EventInfo(double time,
fields.addField("Type", name);
fields.addField("ID", format("%016" PRIx64, id));
fields.addField("Location", location);
if (traceID != UID() && spanID != 0) {
fields.addField("TraceID", traceID.toString());
fields.addField("SpanID", format("%016" PRIx64, spanID));
}
}
TraceBatch::AttachInfo::AttachInfo(double time, const char* name, uint64_t id, uint64_t to) {
TraceBatch::AttachInfo::AttachInfo(double time,
const char* name,
uint64_t id,
uint64_t to,
UID traceID,
uint64_t spanID) {
fields.addField("Severity", format("%d", (int)TRACE_BATCH_IMPLICIT_SEVERITY));
fields.addField("Time", format("%.6f", time));
if (FLOW_KNOBS && FLOW_KNOBS->TRACE_DATETIME_ENABLED) {
@ -1574,6 +1586,10 @@ TraceBatch::AttachInfo::AttachInfo(double time, const char* name, uint64_t id, u
fields.addField("Type", name);
fields.addField("ID", format("%016" PRIx64, id));
fields.addField("To", format("%016" PRIx64, to));
if (traceID != UID() && spanID != 0) {
fields.addField("TraceID", traceID.toString());
fields.addField("SpanID", format("%016" PRIx64, spanID));
}
}
TraceBatch::BuggifyInfo::BuggifyInfo(double time, int activated, int line, std::string file) {

24
flow/include/flow/Trace.h
View File

@ -180,20 +180,26 @@ inline void save(Archive& ar, const TraceEventFields& value) {
class TraceBatch {
public:
void addEvent(const char* name, uint64_t id, const char* location);
void addAttach(const char* name, uint64_t id, uint64_t to);
void addEvent(const char* name, uint64_t id, const char* location, UID traceID = UID(), uint64_t spanID = 0);
void addAttach(const char* name, uint64_t id, uint64_t to, UID traceID = UID(), uint64_t spanID = 0);
void addBuggify(int activated, int line, std::string file);
void dump();
private:
struct EventInfo {
TraceEventFields fields;
EventInfo(double time, double monotonicTime, const char* name, uint64_t id, const char* location);
EventInfo(double time,
double monotonicTime,
const char* name,
uint64_t id,
const char* location,
UID traceID = UID(),
uint64_t spanID = 0);
};
struct AttachInfo {
TraceEventFields fields;
AttachInfo(double time, const char* name, uint64_t id, uint64_t to);
AttachInfo(double time, const char* name, uint64_t id, uint64_t to, UID traceID = UID(), uint64_t spanID = 0);
};
struct BuggifyInfo {
@ -207,6 +213,12 @@ private:
static bool dumpImmediately();
};
struct BatchDebugIDs {
UID debugID;
UID debugTraceID;
uint64_t debugSpanID;
};
struct DynamicEventMetric;
template <class T>
@ -218,9 +230,7 @@ struct SpecialTraceMetricType
#define TRACE_METRIC_TYPE(from, to) \
template <> \
struct SpecialTraceMetricType<from> : std::true_type { \
static to getValue(from v) { \
return v; \
} \
static to getValue(from v) { return v; } \
}
TRACE_METRIC_TYPE(double, double);