[improve][broker] Trace the asynchronous tasks in logs when loading topics#26163
[improve][broker] Trace the asynchronous tasks in logs when loading topics#26163BewareMyPower wants to merge 8 commits into
Conversation
0cf2bee to
7ed63b7
Compare
void-ptr974
left a comment
There was a problem hiding this comment.
Thanks for the improvement! The additional timing details should help with understanding topic-loading latency. I left a couple of minor comments.
|
Hi, @Denovo1998 @void-ptr974 I've addressed your comments in latest commit with a few improvements, PTAL.
|
Denovo1998
left a comment
There was a problem hiding this comment.
Perhaps the implementation can be simpler?
Each time a log or metric needs to be output, call snapshot() once, take the current time as endNs within it, collect only the checkpoints not later than endNs, and then return an immutable LatencySnapshot. Both logs and metrics take values from the same snapshot.
This approach eliminates the need for intermediate states like -1/-2, and reading latency will not inadvertently permanently end the tracer. Even if timeout logs have been printed, the background topic loading can still continue to record subsequent stages.
Additionally, currently, when a timeout occurs, incomplete stages only show something like done: 29999 ms. It's not clear from the logs whether the actual stall is in the system topic, open-ml, or another future. Since the main goal of this PR is to identify which specific stage is consuming time, it's best to record the current action when a future starts and record the end upon completion. This way, the timeout logs can directly indicate at which step the process is stuck.
|
I agree that freezing the end time in |
| public void trace(String action) { | ||
| timepoints.add(new Timepoint(action, nanoTimeSupplier.getNanos())); | ||
| } | ||
|
|
||
| public Snapshot getLatency() { | ||
| final var timepoints = new ArrayList<>(this.timepoints); | ||
| if (timepoints.isEmpty()) { | ||
| return new Snapshot(startNs, 0, "total: 0 ms"); | ||
| } | ||
| final var sb = new StringBuilder(); | ||
| final var totalLatencyNs = TimeUnit.NANOSECONDS.toMillis(timepoints.get(timepoints.size() - 1).timeInNanos | ||
| - startNs); | ||
| sb.append("total: ").append(totalLatencyNs).append(" ms"); | ||
| long prevNs = startNs; | ||
| for (final var tp : timepoints) { | ||
| sb.append(", ").append(tp.name).append(": "); | ||
| long latencyMs = TimeUnit.NANOSECONDS.toMillis(tp.timeInNanos - prevNs); | ||
| if (latencyMs > 0) { | ||
| sb.append(latencyMs).append(" ms"); | ||
| } else { | ||
| sb.append(TimeUnit.NANOSECONDS.toMicros(tp.timeInNanos - prevNs)).append(" us"); | ||
| } | ||
| prevNs = tp.timeInNanos; | ||
| } | ||
| return new Snapshot(prevNs, totalLatencyNs, sb.toString()); | ||
| } |
There was a problem hiding this comment.
A thread may be paused between acquiring a timestamp and calling add():
-
A acquires a 10 ms timestamp, then pauses.
-
The timeout thread acquires and inserts fail: 20 ms.
-
A resumes, inserts A: 10 ms after fail.
-
getLatency() calculates according to the queue order and treats the last element as the total end time.
The concurrent queue prevents structural corruption, but it does not preserve timestamp order. trace() samples the clock before Queue.add(), so one thread can be descheduled after sampling and insert an older timepoint after a newer one. getLatency() then trusts queue order and uses the last element as the total end time.
Could trace insertion and snapshot creation be serialized, or could the snapshot capture a terminal timestamp and order/filter the copied timepoints by timestamp? A concurrent regression test would also be useful here.
There was a problem hiding this comment.
It's a rare case. Even if it happens, it's still possible to recover the original timestamps from the out-of-order timestamps.
| public <T> CompletableFuture<T> trace(String message, CompletableFuture<T> future) { | ||
| if (future.isDone()) { | ||
| return future; | ||
| } | ||
| return future.whenComplete((__, ___) -> trace(message)); | ||
| } | ||
|
|
||
| public void trace(String action) { | ||
| timepoints.add(new Timepoint(action, nanoTimeSupplier.getNanos())); | ||
| } |
There was a problem hiding this comment.
The timeout total is correct now, but the log still reports the unfinished interval as fail: 29999 ms. Since actions are recorded only after their futures complete, it does not directly identify whether system topic, open ml, or another stage is currently stuck.
If identifying the stalled subtask is part of this PR’s goal, could we also record the action when it starts? This should add negligible overhead because only a few actions are traced, though the implementation should account for overlapping actions and avoid clearing a newer action when an older future completes.
There was a problem hiding this comment.
Yes, but it can record the previous step. For example, if the latency description only includes topic exists: 1 ms, fail: 29999 ms, it's still easy to identify it's stuck at the system topic loading. For debugging purpose, it's enough, because this PR does not aim at introducing a formal tracer, which usually collects spans that have both start time and end time (and something else)
|
@Denovo1998 I pushed a new commit, which unifies the rule of logging when topic loading fails.
|
Motivation
The topic loading time could be high with pressure, but it's hard to measure which sub-task takes the most time. This is a follow-up work of #24785 to improve the logs that show topic loading latency.
Modifications
Add a simple tracing class
LatencyTracerand trace asynchronous futures during topic loading. When tracing a future, if the future is already done, skip tracing it because such tasks are usually not CPU-bounded.Verifying this change
LatencyTracerTestcovers functionality of all public methods with injected timestamps. The topic loading logs can be checked by runningtestConcurrentLoadTopicExceedLimitShouldNotBeAutoCreated:Out of scope
In my previous tests, opening managed ledger could also take much time, this
LatencyTracerclass can be used to track managed ledger open process as well.