Skip to content

[Fix]Add audit log event queue size limit #37786

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wangbo merged 1 commit into from
Jul 16, 2024
Merged

[Fix]Add audit log event queue size limit #37786

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions fe/fe-common/src/main/java/org/apache/doris/common/Config.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2899,6 +2899,10 @@ public static int metaServiceRpcRetryTimes() {
@ConfField(mutable = true)
public static boolean enable_cloud_running_txn_check = true;

//* audit_event_log_queue_size = qps * query_audit_log_timeout_ms
@ConfField(mutable = true)
public static int audit_event_log_queue_size = 250000;

@ConfField(description = {"存算分离模式下streamload导入使用的转发策略, 可选值为public-private或者空",
"streamload route policy in cloud mode, availale options are public-private and empty string"})
public static String streamload_redirect_policy = "";
Expand Down
13 changes: 11 additions & 2 deletions fe/fe-core/src/main/java/org/apache/doris/qe/AuditEventProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,12 +70,21 @@ public void stop() {
}
}

public void handleAuditEvent(AuditEvent auditEvent) {
public boolean handleAuditEvent(AuditEvent auditEvent) {
return handleAuditEvent(auditEvent, false);
}

public boolean handleAuditEvent(AuditEvent auditEvent, boolean ignoreQueueFullLog) {
boolean isAddSucc = true;
try {
eventQueue.add(auditEvent);
} catch (Exception e) {
LOG.warn("encounter exception when handle audit event, ignore", e);
isAddSucc = false;
if (!ignoreQueueFullLog) {
LOG.warn("encounter exception when handle audit event {}, ignore", auditEvent.type, e);
}
}
return isAddSucc;
}

public class Worker implements Runnable {
Expand Down
47 changes: 34 additions & 13 deletions ...src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,20 +71,35 @@ protected void runAfterCatalogReady() {
Map<String, TQueryStatistics> queryStatisticsMap = getQueryStatisticsMap();

// 2 log query audit
List<AuditEvent> auditEventList = getQueryNeedAudit();
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;
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;
}
boolean ret = Env.getCurrentAuditEventProcessor().handleAuditEvent(auditEvent, true);
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);
}
Env.getCurrentAuditEventProcessor().handleAuditEvent(auditEvent);
} catch (Throwable t) {
LOG.warn("exception happens when handleAuditEvent, ", t);
}

// 3 clear beToQueryStatsMap when be report timeout
Expand All @@ -94,6 +109,12 @@ protected void runAfterCatalogReady() {
public void submitFinishQueryToAudit(AuditEvent event) {
queryAuditEventLogWriteLock();
try {
if (queryAuditEventList.size() >= Config.audit_event_log_queue_size) {
LOG.warn("audit log event queue size {} is full, this may cause audit log missed."
+ "you can check whether qps is too high or reset audit_event_log_queue_size",
queryAuditEventList.size());
return;
}
event.pushToAuditLogQueueTime = System.currentTimeMillis();
queryAuditEventList.add(event);
} finally {
Expand Down