Gabriel39 commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3849276875
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -147,92 +182,261 @@ public void submitFinishQueryToAudit(AuditEvent event) {
// put the event to queryAuditEventList and let the worker
thread to handle it.
// the worker thread will try best to wait for the statistic
info before logging this event.
event.pushToAuditLogQueueTime = System.currentTimeMillis();
- queryAuditEventList.add(event);
+ PendingAuditEvent pending = new PendingAuditEvent(event,
expectedBackendIds,
+ bindAndRetainBackendIncarnations(event.queryId,
expectedBackendIds));
+ queryAuditEventList.add(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
}
- private List<AuditEvent> getQueryNeedAudit() {
+ @VisibleForTesting
+ List<AuditEvent> getQueryNeedAudit() {
+ RuntimeStatisticsSnapshot runtimeSnapshot =
buildRuntimeStatisticsSnapshot();
+ queryStatisticsSnapshot =
ImmutableMap.copyOf(runtimeSnapshot.queryStatistics);
+
List<AuditEvent> ret = new ArrayList<>();
long currentTime = System.currentTimeMillis();
+ int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
+ long maximumWaitMs = Math.max(queryAuditLogTimeout,
+ Config.be_report_query_statistics_timeout_ms);
queryAuditEventLogWriteLock();
try {
- int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
- Iterator<AuditEvent> iter = queryAuditEventList.iterator();
- while (iter.hasNext()) {
- AuditEvent ae = iter.next();
- if (currentTime - ae.pushToAuditLogQueueTime >
queryAuditLogTimeout) {
- ret.add(ae);
- iter.remove();
- } else {
+ Iterator<PendingAuditEvent> iterator =
queryAuditEventList.iterator();
+ while (iterator.hasNext()) {
+ PendingAuditEvent pending = iterator.next();
+ long elapsed = currentTime -
pending.event.pushToAuditLogQueueTime;
+ if (elapsed <= queryAuditLogTimeout) {
+ // Events are appended in timestamp order, so later
entries cannot be due yet.
break;
}
+ boolean allFinalSnapshotsReceived = true;
+ TQueryStatistics auditStatistics =
pending.expectedBackendIds.length == 0
+ ?
runtimeSnapshot.queryStatistics.get(pending.event.queryId)
+ : new TQueryStatistics();
+ for (int i = 0; i < pending.expectedBackendIds.length; i++) {
+ long backendStartTime =
pending.expectedBackendStartTimes[i];
+ if (backendStartTime == 0) {
+ backendStartTime =
getBackendStartTime(pending.expectedBackendIds[i]);
+ }
+ TQueryStatisticsResult statistics =
findStatisticsForBackend(
+ pending.event.queryId,
pending.expectedBackendIds[i], backendStartTime);
+ if (statistics == null) {
+ allFinalSnapshotsReceived = false;
+ continue;
+ }
+ mergeQueryStatistics(auditStatistics, statistics);
+ if (!statistics.isSetQueryFinished() ||
!statistics.isQueryFinished()) {
+ allFinalSnapshotsReceived = false;
+ }
+ }
+ if (!allFinalSnapshotsReceived && elapsed <= maximumWaitMs) {
+ continue;
+ }
+ // A DML audit must not pass the queue until every scheduled
BE has published its
+ // query-level final snapshot. The upper bound only protects
the queue if a BE dies.
+ applyQueryStatisticsToAuditEvent(pending.event,
auditStatistics);
+ ret.add(pending.event);
+ iterator.remove();
+ releaseBoundIncarnations(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
return ret;
}
- public void updateBeQueryStats(TReportWorkloadRuntimeStatusParams params) {
+ public boolean updateBeQueryStats(TReportWorkloadRuntimeStatusParams
params) {
if (!params.isSetBackendId()) {
LOG.warn("be report workload runtime status but without beid");
- return;
+ return false;
}
if (!params.isSetQueryStatisticsResultMap()) {
LOG.warn("be report workload runtime status but without query
stats map");
- return;
+ return false;
+ }
+ if (!params.isSetBackendStartTime()) {
+ LOG.warn("be report workload runtime status without backend start
time");
+ return false;
}
long beId = params.backend_id;
- // NOTE(wb) one be sends update request one by one,
- // so there is no need a global lock for beToQueryStatsMap here,
- // just keep one be's put/remove/get is atomic operation is enough
- long currentTime = System.currentTimeMillis();
- BeReportInfo beReportInfo = beToQueryStatsMap.get(beId);
- if (beReportInfo == null) {
- beReportInfo = new BeReportInfo(currentTime);
- beToQueryStatsMap.put(beId, beReportInfo);
+ long backendStartTime = params.backend_start_time;
+ long currentBackendStartTime = getBackendStartTime(beId);
+ if (currentBackendStartTime > 0 && currentBackendStartTime !=
backendStartTime) {
+ LOG.info("ignore stale workload runtime status from backend {},
report start time {}, current {}",
+ beId, backendStartTime, currentBackendStartTime);
+ return false;
+ }
+ if (currentBackendStartTime == 0) {
Review Comment:
Fixed. Every non-positive heartbeat start time is now treated as unknown, so
the first accepted report latches the process epoch for both 0 and the
production -1 sentinel. verifies that a competing epoch is rejected.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -147,92 +182,261 @@ public void submitFinishQueryToAudit(AuditEvent event) {
// put the event to queryAuditEventList and let the worker
thread to handle it.
// the worker thread will try best to wait for the statistic
info before logging this event.
event.pushToAuditLogQueueTime = System.currentTimeMillis();
- queryAuditEventList.add(event);
+ PendingAuditEvent pending = new PendingAuditEvent(event,
expectedBackendIds,
+ bindAndRetainBackendIncarnations(event.queryId,
expectedBackendIds));
+ queryAuditEventList.add(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
}
- private List<AuditEvent> getQueryNeedAudit() {
+ @VisibleForTesting
+ List<AuditEvent> getQueryNeedAudit() {
+ RuntimeStatisticsSnapshot runtimeSnapshot =
buildRuntimeStatisticsSnapshot();
+ queryStatisticsSnapshot =
ImmutableMap.copyOf(runtimeSnapshot.queryStatistics);
+
List<AuditEvent> ret = new ArrayList<>();
long currentTime = System.currentTimeMillis();
+ int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
+ long maximumWaitMs = Math.max(queryAuditLogTimeout,
+ Config.be_report_query_statistics_timeout_ms);
queryAuditEventLogWriteLock();
try {
- int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
- Iterator<AuditEvent> iter = queryAuditEventList.iterator();
- while (iter.hasNext()) {
- AuditEvent ae = iter.next();
- if (currentTime - ae.pushToAuditLogQueueTime >
queryAuditLogTimeout) {
- ret.add(ae);
- iter.remove();
- } else {
+ Iterator<PendingAuditEvent> iterator =
queryAuditEventList.iterator();
+ while (iterator.hasNext()) {
+ PendingAuditEvent pending = iterator.next();
+ long elapsed = currentTime -
pending.event.pushToAuditLogQueueTime;
+ if (elapsed <= queryAuditLogTimeout) {
+ // Events are appended in timestamp order, so later
entries cannot be due yet.
break;
}
+ boolean allFinalSnapshotsReceived = true;
+ TQueryStatistics auditStatistics =
pending.expectedBackendIds.length == 0
+ ?
runtimeSnapshot.queryStatistics.get(pending.event.queryId)
+ : new TQueryStatistics();
+ for (int i = 0; i < pending.expectedBackendIds.length; i++) {
Review Comment:
Fixed. The daemon snapshots only due event identities under the audit queue
lock, performs participant binding, lookup, and statistics merging outside it,
then reacquires the lock only for identity-based removals. blocks hydration
with a latch and verifies that a concurrent submission completes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
