WorkloadRuntimeStatusMgr.java
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
package org.apache.doris.resource.workloadschedpolicy;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.Config;
import org.apache.doris.common.Pair;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.common.util.MasterDaemon;
import org.apache.doris.plugin.AuditEvent;
import org.apache.doris.qe.QeProcessorImpl;
import org.apache.doris.thrift.TQueryStatistics;
import org.apache.doris.thrift.TQueryStatisticsResult;
import org.apache.doris.thrift.TReportWorkloadRuntimeStatusParams;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReentrantLock;
public class WorkloadRuntimeStatusMgr extends MasterDaemon {
private static final Logger LOG = LogManager.getLogger(WorkloadRuntimeStatusMgr.class);
// backend id --> {query id --> (query last report time, query stats)}
private final ConcurrentMap<Long, BeReportInfo> beToQueryStatsMap = Maps.newConcurrentMap();
// Publish an immutable snapshot for synchronous proc/REST readers.
private volatile Map<String, TQueryStatistics> queryStatisticsSnapshot = ImmutableMap.of();
private final ReentrantLock queryAuditEventLock = new ReentrantLock();
private List<AuditEvent> queryAuditEventList = Lists.newLinkedList();
private volatile long lastWarnTime;
private class BeReportInfo {
volatile long beLastReportTime;
BeReportInfo(long beLastReportTime) {
this.beLastReportTime = beLastReportTime;
}
// query id --> (query last report time, query stats)
final ConcurrentMap<String, Pair<Long, TQueryStatisticsResult>> queryStatsMap
= Maps.newConcurrentMap();
}
public WorkloadRuntimeStatusMgr() {
super("workload-runtime-stats-thread", Config.workload_runtime_status_thread_interval_ms);
}
@Override
protected void runAfterCatalogReady() {
// 1 rebuild and publish query statistics snapshot
rebuildQueryStatisticsSnapshot();
// 2 read the latest immutable snapshot for downstream processing
Map<String, TQueryStatistics> queryStatisticsMap = getQueryStatisticsMap();
// 3 log query audit
try {
List<AuditEvent> auditEventList = getQueryNeedAudit();
int missedLogCount = 0;
int succLogCount = 0;
for (AuditEvent auditEvent : auditEventList) {
TQueryStatistics queryStats = queryStatisticsMap.get(auditEvent.queryId);
if (queryStats != null) {
auditEvent.scanRows = queryStats.scan_rows;
auditEvent.scanBytes = queryStats.scan_bytes;
auditEvent.scanBytesFromLocalStorage = queryStats.scan_bytes_from_local_storage;
auditEvent.scanBytesFromRemoteStorage = queryStats.scan_bytes_from_remote_storage;
auditEvent.peakMemoryBytes = queryStats.max_peak_memory_bytes;
auditEvent.cpuTimeMs = queryStats.cpu_ms;
auditEvent.shuffleSendBytes = queryStats.shuffle_send_bytes;
auditEvent.shuffleSendRows = queryStats.shuffle_send_rows;
auditEvent.spillWriteBytesToLocalStorage = queryStats.spill_write_bytes_to_local_storage;
auditEvent.spillReadBytesFromLocalStorage = queryStats.spill_read_bytes_from_local_storage;
}
boolean ret = Env.getCurrentAuditEventProcessor().handleAuditEvent(auditEvent);
if (!ret) {
missedLogCount++;
} else {
succLogCount++;
}
}
if (missedLogCount > 0) {
LOG.warn("discard audit event because of log queue is full, discard num : {}, succ num : {}",
missedLogCount, succLogCount);
}
} catch (Throwable t) {
LOG.warn("exception happens when handleAuditEvent, ", t);
}
// 4 clear beToQueryStatsMap when be report timeout
clearReportTimeoutBeStatistics();
}
// After the query or insert finished, FE will not audit immediately, it will send an audit
// event to this queue. And the worker thread will handle it. If the queue is full, the event
// will be handled immediately and may miss some statistic info. So the statistic info of audit
// event may be not accurate, but it can avoid the case that FE OOM because of too many audit
// events in queue when QPS is high. The event will be logged directly if the queue is full.
// And the worker thread will get an event from the queue and get the statistic info for this
// event from queryStatisticsMap.
public void submitFinishQueryToAudit(AuditEvent event) {
queryAuditEventLogWriteLock();
try {
if (queryAuditEventList.size() > Config.audit_event_log_queue_size) {
long now = System.currentTimeMillis();
if (now - lastWarnTime >= 1000) {
lastWarnTime = now;
// if queryAuditEventList is full, we don't put the event to queryAuditEventList.
// so that the statistic info of this audit event will be ignored,
// and event will be logged directly.
LOG.warn("audit log event queue size {} is full, this may cause audit log missing "
+ "statistics. you can check whether qps is too high or set "
+ "audit_event_log_queue_size to a larger value in fe.conf. query id: {}",
queryAuditEventList.size(), event.queryId);
}
Env.getCurrentAuditEventProcessor().handleAuditEvent(event);
} else {
// 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);
}
} finally {
queryAuditEventLogWriteUnlock();
}
}
private List<AuditEvent> getQueryNeedAudit() {
List<AuditEvent> ret = new ArrayList<>();
long currentTime = System.currentTimeMillis();
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 {
break;
}
}
} finally {
queryAuditEventLogWriteUnlock();
}
return ret;
}
public void updateBeQueryStats(TReportWorkloadRuntimeStatusParams params) {
if (!params.isSetBackendId()) {
LOG.warn("be report workload runtime status but without beid");
return;
}
if (!params.isSetQueryStatisticsResultMap()) {
LOG.warn("be report workload runtime status but without query stats map");
return;
}
long beId = params.backend_id;
long currentTime = System.currentTimeMillis();
// Serialize updates and expiry only per BE. A final update acknowledged here must not be
// removed by a cleaner that made its timeout decision on an older value.
beToQueryStatsMap.compute(beId, (ignored, previousInfo) -> {
BeReportInfo beReportInfo = previousInfo == null
? new BeReportInfo(currentTime) : previousInfo;
beReportInfo.beLastReportTime = currentTime;
for (Map.Entry<String, TQueryStatisticsResult> entry
: params.query_statistics_result_map.entrySet()) {
Pair<Long, TQueryStatisticsResult> incoming = Pair.of(currentTime, entry.getValue());
beReportInfo.queryStatsMap.compute(entry.getKey(), (queryId, previous) ->
previous == null || shouldReplaceQueryStatistics(previous.second, incoming.second)
? incoming : previous);
}
return beReportInfo;
});
}
private boolean shouldReplaceQueryStatistics(TQueryStatisticsResult previous,
TQueryStatisticsResult incoming) {
long previousGeneration = previous.isSetQueryStatisticsGeneration()
? previous.getQueryStatisticsGeneration() : 0;
long incomingGeneration = incoming.isSetQueryStatisticsGeneration()
? incoming.getQueryStatisticsGeneration() : 0;
if (previousGeneration != incomingGeneration) {
return incomingGeneration > previousGeneration;
}
long previousSequence = previous.isSetQueryStatisticsSequence()
? previous.getQueryStatisticsSequence() : 0;
long incomingSequence = incoming.isSetQueryStatisticsSequence()
? incoming.getQueryStatisticsSequence() : 0;
if (previousSequence != incomingSequence) {
return incomingSequence > previousSequence;
}
// Legacy BEs have no ordering fields; once their terminal snapshot arrives, an older
// in-flight periodic snapshot must not make the same query look unfinished again.
return !previous.isSetQueryFinished() || !previous.isQueryFinished()
|| (incoming.isSetQueryFinished() && incoming.isQueryFinished());
}
void clearReportTimeoutBeStatistics() {
// 1 clear report timeout be
Set<Long> currentBeIdSet = beToQueryStatsMap.keySet();
Long currentTime = System.currentTimeMillis();
for (Long beId : currentBeIdSet) {
beToQueryStatsMap.computeIfPresent(beId, (ignored, beReportInfo) -> {
if (currentTime - beReportInfo.beLastReportTime
> Config.be_report_query_statistics_timeout_ms) {
return null;
}
Set<String> queryIdSet = beReportInfo.queryStatsMap.keySet();
for (String queryId : queryIdSet) {
beReportInfo.queryStatsMap.computeIfPresent(queryId, (ignoredQueryId, pair) -> {
long queryLastReportTime = pair.first;
boolean timeout = currentTime - queryLastReportTime
> Config.be_report_query_statistics_timeout_ms;
// Keep timed-out statistics while FE still owns the query. The outer
// per-BE compute makes this decision atomic with a concurrent final update.
return timeout && isQueryNotExistInFe(queryId) ? null : pair;
});
}
return beReportInfo;
});
}
}
boolean isQueryNotExistInFe(String queryId) {
try {
return QeProcessorImpl.INSTANCE.getCoordinator(DebugUtil.parseTUniqueIdFromString(queryId)) == null;
} catch (NumberFormatException e) {
return true;
}
}
// Rebuild query statistics from concurrent runtime maps and publish an immutable snapshot.
// This method is intentionally called by daemon thread and unit tests only.
void rebuildQueryStatisticsSnapshot() {
queryStatisticsSnapshot = ImmutableMap.copyOf(buildQueryStatisticsMapUnsafe());
}
// Return the latest published snapshot for synchronous readers such as proc/REST paths.
public Map<String, TQueryStatistics> getQueryStatisticsMap() {
return queryStatisticsSnapshot;
}
// Build a merged map by traversing concurrent runtime structures.
private Map<String, TQueryStatistics> buildQueryStatisticsMapUnsafe() {
// 1 merge query stats in all be
Set<Long> beIdSet = beToQueryStatsMap.keySet();
Map<String, TQueryStatistics> resultQueryMap = Maps.newHashMap();
for (Long beId : beIdSet) {
BeReportInfo beReportInfo = beToQueryStatsMap.get(beId);
if (beReportInfo == null) {
continue;
}
Set<String> queryIdSet = beReportInfo.queryStatsMap.keySet();
for (String queryId : queryIdSet) {
Pair<Long, TQueryStatisticsResult> queryStatsPair =
beReportInfo.queryStatsMap.get(queryId);
if (queryStatsPair == null || queryStatsPair.second == null) {
continue;
}
TQueryStatisticsResult curQueryStats = queryStatsPair.second;
TQueryStatistics retQuery = resultQueryMap.get(queryId);
if (retQuery == null) {
retQuery = new TQueryStatistics();
resultQueryMap.put(queryId, retQuery);
}
mergeQueryStatistics(retQuery, curQueryStats);
}
}
return resultQueryMap;
}
public Map<Long, TQueryStatisticsResult> getQueryStatistics(String queryId) {
Map<Long, TQueryStatisticsResult> result = Maps.newHashMap();
for (Map.Entry<Long, BeReportInfo> entry : beToQueryStatsMap.entrySet()) {
Pair<Long, TQueryStatisticsResult> pair = entry.getValue().queryStatsMap.get(queryId);
if (pair != null) {
result.put(entry.getKey(), pair.second);
}
}
return result;
}
private void mergeQueryStatistics(TQueryStatistics dst, TQueryStatisticsResult src) {
TQueryStatistics srcStats = src.getStatistics();
if (srcStats == null) {
return;
}
dst.setScanRows(dst.scan_rows + srcStats.scan_rows);
dst.setScanBytes(dst.scan_bytes + srcStats.scan_bytes);
dst.setScanBytesFromLocalStorage(dst.scan_bytes_from_local_storage
+ srcStats.scan_bytes_from_local_storage);
dst.setScanBytesFromRemoteStorage(dst.scan_bytes_from_remote_storage
+ srcStats.scan_bytes_from_remote_storage);
dst.setCpuMs(dst.cpu_ms + srcStats.cpu_ms);
dst.setShuffleSendBytes(dst.shuffle_send_bytes + srcStats.shuffle_send_bytes);
dst.setShuffleSendRows(dst.shuffle_send_rows + srcStats.shuffle_send_rows);
dst.setProcessRows(dst.process_rows + srcStats.process_rows);
dst.setReturnedRows(dst.returned_rows + srcStats.returned_rows);
if (srcStats.isSetTotalTasksNum()) {
dst.setTotalTasksNum(dst.total_tasks_num + srcStats.total_tasks_num);
}
if (srcStats.isSetFinishedTasksNum()) {
dst.setFinishedTasksNum(dst.finished_tasks_num + srcStats.finished_tasks_num);
}
if (dst.current_used_memory_bytes < srcStats.current_used_memory_bytes) {
dst.setCurrentUsedMemoryBytes(srcStats.current_used_memory_bytes);
}
if (dst.workload_group_id <= 0 && srcStats.workload_group_id > 0) {
dst.setWorkloadGroupId(srcStats.workload_group_id);
}
if (dst.max_peak_memory_bytes < srcStats.max_peak_memory_bytes) {
dst.setMaxPeakMemoryBytes(srcStats.max_peak_memory_bytes);
}
dst.setSpillWriteBytesToLocalStorage(dst.spill_write_bytes_to_local_storage
+ srcStats.spill_write_bytes_to_local_storage);
dst.setSpillReadBytesFromLocalStorage(dst.spill_read_bytes_from_local_storage
+ srcStats.spill_read_bytes_from_local_storage);
dst.setBytesWriteIntoCache(dst.bytes_write_into_cache + srcStats.bytes_write_into_cache);
}
private void queryAuditEventLogWriteLock() {
queryAuditEventLock.lock();
}
private void queryAuditEventLogWriteUnlock() {
queryAuditEventLock.unlock();
}
}