You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "pnowojski (via GitHub)" <gi...@apache.org> on 2023/09/12 10:13:26 UTC
[GitHub] [flink] pnowojski commented on a diff in pull request #22800: [FLINK-23411][checkpoint] Expose Checkpoint metrics in task level
pnowojski commented on code in PR #22800:
URL: https://github.com/apache/flink/pull/22800#discussion_r1322777799
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/AsyncCheckpointMetricsGroup.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.flink.runtime.checkpoint;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.groups.ProxyMetricGroup;
+
+import javax.annotation.Nonnull;
+
+/** A metric group for {@link CheckpointMetrics} related to the async phase. */
+public class AsyncCheckpointMetricsGroup extends ProxyMetricGroup<MetricGroup> {
+
+ private volatile CheckpointMetrics checkpointMetrics;
+
+ public AsyncCheckpointMetricsGroup(MetricGroup parent) {
+ super(parent);
+ this.checkpointMetrics = new CheckpointMetrics();
+ registerMetrics();
+ }
+
+ @VisibleForTesting
+ public static final String BYTES_PERSISTED_DURING_ALIGNMENT =
+ "checkpointBytesPersistedDuringAlignment";
+
+ @VisibleForTesting
+ public static final String ASYNC_DURATION_MILLIS = "checkpointAsyncDurationMillis";
+
+ @VisibleForTesting
+ public static final String BYTES_PERSISTED_OF_THIS_CHECKPOINT =
+ "checkpointBytesPersistedOfThisCheckpoint";
+
+ @VisibleForTesting
+ public static final String TOTAL_BYTES_PERSISTED = "checkpointTotalBytesPersisted";
Review Comment:
Could you explain why this is added here and not in the `MetricNames`?
##########
docs/content.zh/docs/ops/metrics.md:
##########
@@ -1348,16 +1348,46 @@ Note that for failed checkpoints, metrics are updated on a best efforts basis an
<td>Gauge</td>
</tr>
<tr>
- <th rowspan="2"><strong>Task</strong></th>
+ <th rowspan="8"><strong>Task/Operator</strong></th>
<td>checkpointAlignmentTime</td>
- <td>The time in nanoseconds that the last barrier alignment took to complete, or how long the current alignment has taken so far (in nanoseconds). This is the time between receiving first and the last checkpoint barrier. You can find more information in the [Monitoring State and Checkpoints section]({{< ref "docs/ops/state/large_state_tuning" >}}#monitoring-state-and-checkpoints)</td>
+ <td>The time in nanoseconds that the last barrier alignment took to complete, or how long the current alignment has taken so far (in nanoseconds). This is the time between receiving first and the last checkpoint barrier. You can find more information in the <a href="docs/ops/state/large_state_tuning">#monitoring-state-and-checkpoints">Monitoring State and Checkpoints section</a>.</td>
<td>Gauge</td>
</tr>
<tr>
<td>checkpointStartDelayNanos</td>
<td>The time in nanoseconds that elapsed between the creation of the last checkpoint and the time when the checkpointing process has started by this Task. This delay shows how long it takes for the first checkpoint barrier to reach the task. A high value indicates back-pressure. If only a specific task has a long start delay, the most likely reason is data skew.</td>
<td>Gauge</td>
</tr>
+ <tr>
+ <td>checkpointBytesProcessedDuringAlignment</td>
+ <td>The approximate number of bytes processed during the alignment (time between receiving the first and the last checkpoint barrier) over all acknowledged subtasks.</td>
+ <td>Gauge</td>
+ </tr>
+ <tr>
+ <td>checkpointBytesPersistedDuringAlignment</td>
+ <td>The approximate number of bytes persisted during the alignment (time between receiving the first and the last checkpoint barrier) over all acknowledged subtasks. It could be larger than zero only if the unaligned checkpoints are enabled.</td>
+ <td>Gauge</td>
+ </tr>
+ <tr>
+ <td>checkpointSyncDurationMillis</td>
+ <td>The duration of the synchronous part of the checkpoint. This includes snapshotting state of the operators and blocks all other activity on the subtask (processing records, firing timers, etc).</td>
+ <td>Gauge</td>
+ </tr>
+ <tr>
+ <td>checkpointAsyncDurationMillis</td>
+ <td>The duration of the asynchronous part of the checkpoint. This includes time it took to write the checkpoint on to the selected filesystem. For unaligned checkpoints this also includes also the time the subtask had to wait for last of the checkpoint barriers to arrive (alignment duration) and the time it took to persist the in-flight data.</td>
+ <td>Gauge</td>
+ </tr>
+ <tr>
+ <td>checkpointBytesPersistedOfThisCheckpoint</td>
+ <td>The persisted data size during the sync and async phases of that checkpoint, the value could be different from totalBytesPersisted if incremental checkpoint or changelog is enabled.</td>
+ <td>Gauge</td>
+ </tr>
+ <tr>
+ <td>checkpointTotalBytesPersisted</td>
+ <td>The total persisted data size of that checkpoint.</td>
+ <td>Gauge</td>
+ </tr>
Review Comment:
I think all of those metrics should be prefixed `last`.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointBarrierHandler.java:
##########
@@ -118,6 +125,18 @@ public long getCheckpointStartDelayNanos() {
return latestCheckpointStartDelayNanos;
}
+ public long getBytesProcessedDuringAlignment() {
+ if (isDuringAlignment()) {
+ return bytesProcessedDuringAlignment;
+ } else {
+ return FutureUtils.getOrDefault(latestBytesProcessedDuringAlignment, 0L);
+ }
+ }
+
+ public long getCheckpointSyncDurationMillis() {
Review Comment:
Both of those metrics are untested as far as I can tell.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointBarrierHandler.java:
##########
@@ -118,6 +125,18 @@ public long getCheckpointStartDelayNanos() {
return latestCheckpointStartDelayNanos;
}
+ public long getBytesProcessedDuringAlignment() {
+ if (isDuringAlignment()) {
+ return bytesProcessedDuringAlignment;
+ } else {
+ return FutureUtils.getOrDefault(latestBytesProcessedDuringAlignment, 0L);
+ }
+ }
+
+ public long getCheckpointSyncDurationMillis() {
+ return latestCheckpointSyncDurationMillis;
Review Comment:
nit: rename both field and getter to `last/getLastCheckpointSyncDurationMillis`?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/AsyncCheckpointMetricsGroupTest.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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.flink.runtime.checkpoint;
+
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the {@link AsyncCheckpointMetricsGroup} class. */
+public class AsyncCheckpointMetricsGroupTest {
+
+ @Test
+ public void testCheckpointMetricsRegistration() {
+ // Construct a CheckpointMetricsGroup.
+ final Map<String, Gauge<?>> registeredGauges = new HashMap<>();
+ MetricGroup metricGroup =
+ new UnregisteredMetricsGroup() {
+ @Override
+ public <T, G extends Gauge<T>> G gauge(String name, G gauge) {
+ registeredGauges.put(name, gauge);
+ return gauge;
+ }
+ };
+ AsyncCheckpointMetricsGroup asyncCheckpointMetricsGroup =
+ new AsyncCheckpointMetricsGroup(metricGroup);
+
+ // Construct a CheckpointMetrics.
+ CheckpointMetrics checkpointMetrics =
+ new CheckpointMetricsBuilder()
+ .setBytesProcessedDuringAlignment(12L)
+ .setAlignmentDurationNanos(23L)
+ .setBytesPersistedOfThisCheckpoint(34L)
+ .setTotalBytesPersisted(45L)
+ .setAsyncDurationMillis(56L)
+ .setSyncDurationMillis(67L)
+ .setCheckpointStartDelayNanos(78L)
+ .build();
+
+ // Update and Verify the status of AsyncCheckpointMetricsGroup.
+ asyncCheckpointMetricsGroup.reportCheckpointMetrics(checkpointMetrics);
+ assertThat(
Review Comment:
This test doesn't fully cover newly added code. For example one might remove `asyncCheckpointMetricsGroup.reportCheckpointMetrics()` call from the `AsyncCheckpointRunnable` and this test would be still passing.
To have full test coverage I think you should try to write a metrics reporting test in `SubtaskCheckpointCoordinatorTest`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org