You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/01/12 19:40:32 UTC
[flink] 04/05: [FLINK-20852][webui] Provide more detailed per
subtask backpressure stats
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f43fc0429095f8532d85ce785d023a61cea95700
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Jan 11 18:29:03 2021 +0100
[FLINK-20852][webui] Provide more detailed per subtask backpressure stats
---
.../src/test/resources/rest_api_v1.snapshot | 24 ++++++++-----
.../src/app/interfaces/job-backpressure.ts | 2 ++
...job-overview-drawer-backpressure.component.html | 6 ++--
.../job-overview-drawer-backpressure.component.ts | 9 +++++
.../handler/job/JobVertexBackPressureHandler.java | 24 ++++++++++---
.../rest/messages/JobVertexBackPressureInfo.java | 41 +++++++++++++++++-----
.../job/JobVertexBackPressureHandlerTest.java | 27 +++++++++++---
.../messages/JobVertexBackPressureInfoTest.java | 6 ++--
8 files changed, 106 insertions(+), 33 deletions(-)
diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
index b2450e3..173eed4 100644
--- a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
+++ b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
@@ -984,10 +984,10 @@
"type" : "integer"
},
"processed_data" : {
- "type" : "integer"
+ "type" : "integer"
},
"persisted_data" : {
- "type" : "integer"
+ "type" : "integer"
},
"num_subtasks" : {
"type" : "integer"
@@ -1081,10 +1081,10 @@
"type" : "integer"
},
"processed_data" : {
- "type" : "integer"
+ "type" : "integer"
},
"persisted_data" : {
- "type" : "integer"
+ "type" : "integer"
},
"num_subtasks" : {
"type" : "integer"
@@ -1296,10 +1296,10 @@
"type" : "integer"
},
"processed_data" : {
- "type" : "integer"
+ "type" : "integer"
},
"persisted_data" : {
- "type" : "integer"
+ "type" : "integer"
},
"num_subtasks" : {
"type" : "integer"
@@ -1397,10 +1397,10 @@
"type" : "integer"
},
"processed_data" : {
- "type" : "integer"
+ "type" : "integer"
},
"persisted_data" : {
- "type" : "integer"
+ "type" : "integer"
},
"num_subtasks" : {
"type" : "integer"
@@ -2049,6 +2049,12 @@
},
"ratio" : {
"type" : "number"
+ },
+ "idleRatio" : {
+ "type" : "number"
+ },
+ "busyRatio" : {
+ "type" : "number"
}
}
}
@@ -3183,4 +3189,4 @@
}
}
} ]
-}
+}
\ No newline at end of file
diff --git a/flink-runtime-web/web-dashboard/src/app/interfaces/job-backpressure.ts b/flink-runtime-web/web-dashboard/src/app/interfaces/job-backpressure.ts
index b91234f..7e4163a 100644
--- a/flink-runtime-web/web-dashboard/src/app/interfaces/job-backpressure.ts
+++ b/flink-runtime-web/web-dashboard/src/app/interfaces/job-backpressure.ts
@@ -27,4 +27,6 @@ export interface JobBackpressureSubtaskInterface {
subtask: number;
'backpressure-level': string;
ratio: number;
+ idleRatio: number;
+ busyRatio: number;
}
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.html b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.html
index 12b6e62..41377e1 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.html
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.html
@@ -52,14 +52,14 @@
</tr>
<tr>
<th>SubTask</th>
- <th>Ratio</th>
- <th>Status</th>
+ <th>Backpressured / Idle / Busy</th>
+ <th>Backpressure Status</th>
</tr>
</thead>
<tbody>
<tr *ngFor="let subtask of listOfSubTaskBackpressure;trackBy:trackBackPressureBy;">
<td>{{ subtask['subtask'] }}</td>
- <td>{{ subtask['ratio'] }}</td>
+ <td>{{ this.prettyPrint(subtask['ratio']) }} / {{ this.prettyPrint(subtask['idleRatio']) }} / {{ this.prettyPrint(subtask['busyRatio']) }}</td>
<td>
<flink-backpressure-badge [state]="backpressure['backpressure-level']"></flink-backpressure-badge>
</td>
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.ts b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.ts
index f9768d2..2068006 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.ts
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.ts
@@ -41,6 +41,15 @@ export class JobOverviewDrawerBackpressureComponent implements OnInit, OnDestroy
return node.subtask;
}
+ prettyPrint(value: number): string {
+ if (isNaN(value)) {
+ return "N/A"
+ }
+ else {
+ return Math.round(value * 100) + "%";
+ }
+ }
+
ngOnInit() {
this.jobService.jobWithVertex$
.pipe(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandler.java
index 47b9bda..edeacf1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandler.java
@@ -107,11 +107,15 @@ public class JobVertexBackPressureHandler
int subtaskIndex = entry.getKey();
ComponentMetricStore subtaskMetricStore = entry.getValue();
double backPressureRatio = getBackPressureRatio(subtaskMetricStore);
+ double idleRatio = getIdleRatio(subtaskMetricStore);
+ double busyRatio = getBusyRatio(subtaskMetricStore);
result.add(
new SubtaskBackPressureInfo(
subtaskIndex,
getBackPressureLevel(backPressureRatio),
- backPressureRatio));
+ backPressureRatio,
+ idleRatio,
+ busyRatio));
}
result.sort(Comparator.comparingInt(SubtaskBackPressureInfo::getSubtask));
return result;
@@ -119,14 +123,26 @@ public class JobVertexBackPressureHandler
private double getMaxBackPressureRatio(List<SubtaskBackPressureInfo> subtaskBackPressureInfos) {
return subtaskBackPressureInfos.stream()
- .mapToDouble(backPressureInfo -> backPressureInfo.getRatio())
+ .mapToDouble(backPressureInfo -> backPressureInfo.getBackPressuredRatio())
.max()
.getAsDouble();
}
private double getBackPressureRatio(ComponentMetricStore metricStore) {
- return Double.valueOf(metricStore.getMetric(MetricNames.TASK_BACK_PRESSURED_TIME, "0"))
- / 1_000;
+ return getMsPerSecondMetricAsRatio(metricStore, MetricNames.TASK_BACK_PRESSURED_TIME);
+ }
+
+ private double getIdleRatio(ComponentMetricStore metricStore) {
+ return getMsPerSecondMetricAsRatio(metricStore, MetricNames.TASK_IDLE_TIME);
+ }
+
+ private double getBusyRatio(ComponentMetricStore metricStore) {
+ return getMsPerSecondMetricAsRatio(metricStore, MetricNames.TASK_BUSY_TIME);
+ }
+
+ private double getMsPerSecondMetricAsRatio(
+ ComponentMetricStore metricStore, String metricName) {
+ return Double.valueOf(metricStore.getMetric(metricName, "0")) / 1_000;
}
/**
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexBackPressureInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexBackPressureInfo.java
index de54683..5415c29 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexBackPressureInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexBackPressureInfo.java
@@ -122,7 +122,9 @@ public class JobVertexBackPressureInfo implements ResponseBody {
public static final String FIELD_NAME_SUBTASK = "subtask";
public static final String FIELD_NAME_BACKPRESSURE_LEVEL = "backpressure-level";
- public static final String FIELD_NAME_RATIO = "ratio";
+ public static final String FIELD_NAME_BACK_PRESSURED_RATIO = "ratio";
+ public static final String FIELD_NAME_IDLE_RATIO = "idleRatio";
+ public static final String FIELD_NAME_BUSY_RATIO = "busyRatio";
@JsonProperty(FIELD_NAME_SUBTASK)
private final int subtask;
@@ -130,17 +132,27 @@ public class JobVertexBackPressureInfo implements ResponseBody {
@JsonProperty(FIELD_NAME_BACKPRESSURE_LEVEL)
private final VertexBackPressureLevel backpressureLevel;
- @JsonProperty(FIELD_NAME_RATIO)
- private final double ratio;
+ @JsonProperty(FIELD_NAME_BACK_PRESSURED_RATIO)
+ private final double backPressuredRatio;
+
+ @JsonProperty(FIELD_NAME_IDLE_RATIO)
+ private final double idleRatio;
+
+ @JsonProperty(FIELD_NAME_BUSY_RATIO)
+ private final double busyRatio;
public SubtaskBackPressureInfo(
@JsonProperty(FIELD_NAME_SUBTASK) int subtask,
@JsonProperty(FIELD_NAME_BACKPRESSURE_LEVEL)
VertexBackPressureLevel backpressureLevel,
- @JsonProperty(FIELD_NAME_RATIO) double ratio) {
+ @JsonProperty(FIELD_NAME_BACK_PRESSURED_RATIO) double backPressuredRatio,
+ @JsonProperty(FIELD_NAME_IDLE_RATIO) double idleRatio,
+ @JsonProperty(FIELD_NAME_BUSY_RATIO) double busyRatio) {
this.subtask = subtask;
this.backpressureLevel = checkNotNull(backpressureLevel);
- this.ratio = ratio;
+ this.backPressuredRatio = backPressuredRatio;
+ this.idleRatio = idleRatio;
+ this.busyRatio = busyRatio;
}
@Override
@@ -153,13 +165,16 @@ public class JobVertexBackPressureInfo implements ResponseBody {
}
SubtaskBackPressureInfo that = (SubtaskBackPressureInfo) o;
return subtask == that.subtask
- && ratio == that.ratio
+ && backPressuredRatio == that.backPressuredRatio
+ && idleRatio == that.idleRatio
+ && busyRatio == that.busyRatio
&& Objects.equals(backpressureLevel, that.backpressureLevel);
}
@Override
public int hashCode() {
- return Objects.hash(subtask, backpressureLevel, ratio);
+ return Objects.hash(
+ subtask, backpressureLevel, backPressuredRatio, idleRatio, busyRatio);
}
public int getSubtask() {
@@ -170,8 +185,16 @@ public class JobVertexBackPressureInfo implements ResponseBody {
return backpressureLevel;
}
- public double getRatio() {
- return ratio;
+ public double getBackPressuredRatio() {
+ return backPressuredRatio;
+ }
+
+ public double getIdleRatio() {
+ return idleRatio;
+ }
+
+ public double getBusyRatio() {
+ return busyRatio;
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java
index 89dbaaf..faca126 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java
@@ -32,6 +32,7 @@ import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
import org.apache.flink.runtime.rest.messages.JobVertexBackPressureHeaders;
import org.apache.flink.runtime.rest.messages.JobVertexBackPressureInfo;
+import org.apache.flink.runtime.rest.messages.JobVertexBackPressureInfo.SubtaskBackPressureInfo;
import org.apache.flink.runtime.rest.messages.JobVertexBackPressureInfo.VertexBackPressureStatus;
import org.apache.flink.runtime.rest.messages.JobVertexIdPathParameter;
import org.apache.flink.runtime.rest.messages.JobVertexMessageParameters;
@@ -80,6 +81,8 @@ public class JobVertexBackPressureHandlerTest {
TEST_JOB_VERTEX_ID.toString(),
0);
dumps.add(new GaugeDump(task0, MetricNames.TASK_BACK_PRESSURED_TIME, "1000"));
+ dumps.add(new GaugeDump(task0, MetricNames.TASK_IDLE_TIME, "0"));
+ dumps.add(new GaugeDump(task0, MetricNames.TASK_BUSY_TIME, "0"));
TaskQueryScopeInfo task1 =
new TaskQueryScopeInfo(
@@ -87,6 +90,8 @@ public class JobVertexBackPressureHandlerTest {
TEST_JOB_VERTEX_ID.toString(),
1);
dumps.add(new GaugeDump(task1, MetricNames.TASK_BACK_PRESSURED_TIME, "500"));
+ dumps.add(new GaugeDump(task1, MetricNames.TASK_IDLE_TIME, "100"));
+ dumps.add(new GaugeDump(task1, MetricNames.TASK_BUSY_TIME, "900"));
// missing task2
@@ -96,6 +101,8 @@ public class JobVertexBackPressureHandlerTest {
TEST_JOB_VERTEX_ID.toString(),
3);
dumps.add(new GaugeDump(task3, MetricNames.TASK_BACK_PRESSURED_TIME, "100"));
+ dumps.add(new GaugeDump(task3, MetricNames.TASK_IDLE_TIME, "200"));
+ dumps.add(new GaugeDump(task3, MetricNames.TASK_BUSY_TIME, "700"));
return dumps;
}
@@ -158,21 +165,31 @@ public class JobVertexBackPressureHandlerTest {
assertThat(
jobVertexBackPressureInfo.getSubtasks().stream()
- .map(JobVertexBackPressureInfo.SubtaskBackPressureInfo::getRatio)
+ .map(SubtaskBackPressureInfo::getBackPressuredRatio)
.collect(Collectors.toList()),
contains(1.0, 0.5, 0.1));
assertThat(
jobVertexBackPressureInfo.getSubtasks().stream()
- .map(
- JobVertexBackPressureInfo.SubtaskBackPressureInfo
- ::getBackpressureLevel)
+ .map(SubtaskBackPressureInfo::getIdleRatio)
+ .collect(Collectors.toList()),
+ contains(0.0, 0.1, 0.2));
+
+ assertThat(
+ jobVertexBackPressureInfo.getSubtasks().stream()
+ .map(SubtaskBackPressureInfo::getBusyRatio)
+ .collect(Collectors.toList()),
+ contains(0.0, 0.9, 0.7));
+
+ assertThat(
+ jobVertexBackPressureInfo.getSubtasks().stream()
+ .map(SubtaskBackPressureInfo::getBackpressureLevel)
.collect(Collectors.toList()),
contains(HIGH, LOW, OK));
assertThat(
jobVertexBackPressureInfo.getSubtasks().stream()
- .map(JobVertexBackPressureInfo.SubtaskBackPressureInfo::getSubtask)
+ .map(SubtaskBackPressureInfo::getSubtask)
.collect(Collectors.toList()),
contains(0, 1, 3));
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/JobVertexBackPressureInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/JobVertexBackPressureInfoTest.java
index 7c1477b..0dc78e4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/JobVertexBackPressureInfoTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/JobVertexBackPressureInfoTest.java
@@ -34,13 +34,13 @@ public class JobVertexBackPressureInfoTest
List<JobVertexBackPressureInfo.SubtaskBackPressureInfo> subtaskList = new ArrayList<>();
subtaskList.add(
new JobVertexBackPressureInfo.SubtaskBackPressureInfo(
- 0, JobVertexBackPressureInfo.VertexBackPressureLevel.LOW, 0.1));
+ 0, JobVertexBackPressureInfo.VertexBackPressureLevel.LOW, 0.1, 0.5, 0.4));
subtaskList.add(
new JobVertexBackPressureInfo.SubtaskBackPressureInfo(
- 1, JobVertexBackPressureInfo.VertexBackPressureLevel.OK, 0.4));
+ 1, JobVertexBackPressureInfo.VertexBackPressureLevel.OK, 0.4, 0.3, 0.3));
subtaskList.add(
new JobVertexBackPressureInfo.SubtaskBackPressureInfo(
- 2, JobVertexBackPressureInfo.VertexBackPressureLevel.HIGH, 0.9));
+ 2, JobVertexBackPressureInfo.VertexBackPressureLevel.HIGH, 0.9, 0.0, 0.1));
return new JobVertexBackPressureInfo(
JobVertexBackPressureInfo.VertexBackPressureStatus.OK,
JobVertexBackPressureInfo.VertexBackPressureLevel.LOW,