y1chi commented on code in PR #26085:
URL: https://github.com/apache/beam/pull/26085#discussion_r1210569615
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/GrpcWindmillServer.java:
##########
@@ -969,6 +987,76 @@ protected void startThrottleTimer() {
getWorkThrottleTimer.start();
}
+ private class GetWorkTimingInfosTracker {
+ private final Map<State, Duration> getWorkStreamLatencies;
+
+ public GetWorkTimingInfosTracker() {
+ this.getWorkStreamLatencies = new EnumMap<>(State.class);
+ }
+
+ public void addTimingInfo(Collection<GetWorkStreamTimingInfo> infos) {
+ // We want to record duration for each stage and also be reflective on
total work item
+ // processing time. It can be tricky because timings of different
+ // StreamingGetWorkResponseChunks can be interleaved. Current strategy
is to record the
+ // maximum duration in each stage across different chunks, this will
allow us to identify
+ // the slow stage, but note the sum duration of each slowest stages
may be larger than the
+ // duration from first chunk creation to last chunk reception by user
worker.
+ Map<Event, Instant> getWorkStreamTimings = new HashMap<>();
+ for (GetWorkStreamTimingInfo info : infos) {
+ getWorkStreamTimings.putIfAbsent(
+ info.getEvent(), Instant.ofEpochMilli(info.getTimestampUsec() /
1000));
+ }
+
+ for (Cell<Event, Event, State> cell : EVENT_STATE_TABLE.cellSet()) {
+ Event start = cell.getRowKey();
+ Event end = cell.getColumnKey();
+ State state = cell.getValue();
+ if (getWorkStreamTimings.containsKey(start) &&
getWorkStreamTimings.containsKey(end)) {
+ getWorkStreamLatencies.compute(
+ state,
+ (state_key, duration) -> {
+ Duration newDuration =
+ new Duration(getWorkStreamTimings.get(start),
getWorkStreamTimings.get(end));
+ if (duration == null) {
+ return newDuration;
+ }
+ return newDuration.isLongerThan(duration) ? newDuration :
duration;
+ });
+ }
+ }
+ if
(getWorkStreamTimings.containsKey(Event.GET_WORK_RECEIVED_BY_DISPATCHER)) {
+ getWorkStreamLatencies.compute(
+ State.GET_WORK_IN_TRANSIT_TO_USER_WORKER,
+ (state_key, duration) -> {
+ Duration newDuration =
+ new Duration(
+
getWorkStreamTimings.get(Event.GET_WORK_RECEIVED_BY_DISPATCHER),
Review Comment:
Done.
--
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]
