diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java index 5b24b5b6737..28b1451de89 100644 --- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java +++ b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java @@ -303,4 +303,9 @@ public class MetricNames { // for lake tiering metrics - operator level public static final String TIERING_SERVICE_READ_BYTES = "readBytes"; public static final String TIERING_SERVICE_READ_BYTES_RATE = "readBytesPerSecond"; + + // for lake tiering enumerator metrics + public static final String TIERING_HEARTBEAT_FAILURE = "tiering.heartbeat.failure"; + public static final String TIERING_REQUEST_TABLE_EMPTY = "tiering.request-table.empty"; + public static final String TIERING_REQUEST_TABLE_FAILURE = "tiering.request-table.failure"; } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java index f1bf98527af..1c07faec6a5 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java @@ -27,6 +27,7 @@ import org.apache.fluss.flink.tiering.event.FailedTieringEvent; import org.apache.fluss.flink.tiering.event.FinishedTieringEvent; import org.apache.fluss.flink.tiering.event.TieringReachMaxDurationEvent; +import org.apache.fluss.flink.tiering.source.metrics.TieringEnumeratorMetrics; import org.apache.fluss.flink.tiering.source.split.TieringSplit; import org.apache.fluss.flink.tiering.source.split.TieringSplitGenerator; import org.apache.fluss.flink.tiering.source.state.TieringSourceEnumeratorState; @@ -78,7 +79,6 @@ import static org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator.HeartBeatHelper.failedTableHeartBeat; import static org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator.HeartBeatHelper.heartBeatWithRequestNewTieringTable; import static org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator.HeartBeatHelper.tieringTableHeartBeat; -import static org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator.HeartBeatHelper.waitHeartbeatResponse; /** * An implementation of {@link SplitEnumerator} used to request {@link TieringSplit} from Fluss @@ -121,6 +121,8 @@ public class TieringSourceEnumerator private TieringSplitGenerator splitGenerator; private int flussCoordinatorEpoch; + private TieringEnumeratorMetrics tieringMetrics; + private volatile boolean isFailOvering = false; private volatile boolean closed = false; @@ -144,6 +146,17 @@ public TieringSourceEnumerator( this.finishedTables = new ConcurrentHashMap<>(); this.failedTableEpochs = new ConcurrentHashMap<>(); this.tieringReachMaxDurationsTables = Collections.synchronizedSet(new TreeSet<>()); + this.tieringMetrics = new TieringEnumeratorMetrics(enumeratorMetricGroup); + } + + @VisibleForTesting + TieringEnumeratorMetrics getTieringMetrics() { + return tieringMetrics; + } + + @VisibleForTesting + void setCoordinatorGateway(CoordinatorGateway gateway) { + this.coordinatorGateway = gateway; } @Override @@ -171,6 +184,7 @@ public void start() { flussCoordinatorEpoch); } catch (Exception e) { + tieringMetrics.incHeartbeatFailure(); LOG.error("Register Tiering Service failed due to ", e); throw new FlinkRuntimeException("Register Tiering Service failed due to ", e); } @@ -362,6 +376,7 @@ protected void handleTableTieringReachMaxDuration( void generateAndAssignSplits( @Nullable Tuple3 tieringTable, Throwable throwable) { if (throwable != null) { + tieringMetrics.incRequestTableFailure(); ExceptionUtils.rethrow(throwable); } if (tieringTable != null) { @@ -392,7 +407,9 @@ private void assignSplits() { } } - private @Nullable Tuple3 requestTieringTableSplitsViaHeartBeat() { + @VisibleForTesting + @Nullable + Tuple3 requestTieringTableSplitsViaHeartBeat() { if (closed) { return null; } @@ -431,6 +448,7 @@ private void assignSplits() { tieringTableEpochs.put(lakeTieringInfo.f0, lakeTieringInfo.f1); LOG.info("Tiering table {} has been requested.", lakeTieringInfo); } else { + tieringMetrics.incRequestTableEmpty(); LOG.info("No available Tiering table found, will poll later."); } } else { @@ -570,6 +588,7 @@ private void reportFailedTable( LOG.info("Report failed table to Fluss Coordinator success"); } catch (Exception e) { + tieringMetrics.incHeartbeatFailure(); LOG.error("Errors happens when report failed table to Fluss cluster.", e); throw new FlinkRuntimeException( "Errors happens when report failed table to Fluss cluster.", e); @@ -662,15 +681,20 @@ static LakeTieringHeartbeatRequest failedTableHeartBeat( } return lakeTieringHeartbeatRequest; } + } - static LakeTieringHeartbeatResponse waitHeartbeatResponse( - CompletableFuture responseCompletableFuture) { - try { - return responseCompletableFuture.get(3, TimeUnit.MINUTES); - } catch (Exception e) { - LOG.error("Failed to wait heartbeat response due to ", e); - throw new FlinkRuntimeException("Failed to wait heartbeat response due to ", e); - } + /** + * Waits for the heartbeat response from the Fluss coordinator. Increments the heartbeat failure + * counter if the wait fails. + */ + private LakeTieringHeartbeatResponse waitHeartbeatResponse( + CompletableFuture responseCompletableFuture) { + try { + return responseCompletableFuture.get(3, TimeUnit.MINUTES); + } catch (Exception e) { + tieringMetrics.incHeartbeatFailure(); + LOG.error("Failed to wait heartbeat response due to ", e); + throw new FlinkRuntimeException("Failed to wait heartbeat response due to ", e); } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/metrics/TieringEnumeratorMetrics.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/metrics/TieringEnumeratorMetrics.java new file mode 100644 index 00000000000..1f4f6ae9f34 --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/metrics/TieringEnumeratorMetrics.java @@ -0,0 +1,93 @@ +/* + * 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.fluss.flink.tiering.source.metrics; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.metrics.MetricNames; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup; + +/** + * A collection class for handling metrics in {@link + * org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator}. + * + *

All metrics are registered under group "fluss.tieringService", which is a child group of + * {@link SplitEnumeratorMetricGroup}. + * + *

The following metrics are available: + * + *

    + *
  • {@code fluss.tieringService.tiering.heartbeat.failure} - Counter: cumulative number of + * times the heartbeat request to the Fluss coordinator failed. + *
  • {@code fluss.tieringService.tiering.request-table.empty} - Counter: cumulative number of + * times the enumerator requested a new tiering table but the coordinator returned none. + *
  • {@code fluss.tieringService.tiering.request-table.failure} - Counter: cumulative number of + * times the periodic {@code requestTieringTableSplitsViaHeartBeat} call failed. + *
+ */ +@Internal +public class TieringEnumeratorMetrics { + + public static final String FLUSS_METRIC_GROUP = "fluss"; + public static final String TIERING_SERVICE_GROUP = "tieringService"; + + private final Counter heartbeatFailureCounter; + private final Counter requestTableEmptyCounter; + private final Counter requestTableFailureCounter; + + public TieringEnumeratorMetrics(SplitEnumeratorMetricGroup enumeratorMetricGroup) { + MetricGroup tieringServiceGroup = + enumeratorMetricGroup.addGroup(FLUSS_METRIC_GROUP).addGroup(TIERING_SERVICE_GROUP); + + this.heartbeatFailureCounter = + tieringServiceGroup.counter(MetricNames.TIERING_HEARTBEAT_FAILURE); + this.requestTableEmptyCounter = + tieringServiceGroup.counter(MetricNames.TIERING_REQUEST_TABLE_EMPTY); + this.requestTableFailureCounter = + tieringServiceGroup.counter(MetricNames.TIERING_REQUEST_TABLE_FAILURE); + } + + /** Increments when a heartbeat request to the coordinator fails. */ + public void incHeartbeatFailure() { + heartbeatFailureCounter.inc(); + } + + /** Increments when the coordinator has no tiering table to assign. */ + public void incRequestTableEmpty() { + requestTableEmptyCounter.inc(); + } + + /** Increments when the periodic request for tiering table splits fails. */ + public void incRequestTableFailure() { + requestTableFailureCounter.inc(); + } + + public long getHeartbeatFailureCount() { + return heartbeatFailureCounter.getCount(); + } + + public long getRequestTableEmptyCount() { + return requestTableEmptyCounter.getCount(); + } + + public long getRequestTableFailureCount() { + return requestTableFailureCounter.getCount(); + } +} diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java index 8d1f62eb348..3b8e7ea482b 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java @@ -34,7 +34,9 @@ import org.apache.fluss.metadata.TableChange; import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.rpc.gateway.CoordinatorGateway; import org.apache.fluss.rpc.messages.CommitLakeTableSnapshotRequest; +import org.apache.fluss.rpc.messages.LakeTieringHeartbeatResponse; import org.apache.fluss.rpc.messages.PbLakeTableOffsetForBucket; import org.apache.fluss.rpc.messages.PbLakeTableSnapshotInfo; @@ -49,12 +51,14 @@ import javax.annotation.Nullable; import java.io.IOException; +import java.lang.reflect.Proxy; import java.time.Duration; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -937,4 +941,117 @@ void testTableReachMaxTieringDuration() throws Throwable { && !split.shouldSkipCurrentRound()); } } + + @Test + void testHeartbeatFailureMetricIncremented() throws Exception { + try (FlussMockSplitEnumeratorContext context = + new FlussMockSplitEnumeratorContext<>(1)) { + TieringSourceEnumerator enumerator = createTieringSourceEnumerator(flussConf, context); + enumerator.start(); + + // Register a reader so that the "request new table" path is taken + registerSingleReaderAndHandleSplitRequests(context, enumerator, 0, 0); + + long heartbeatFailuresBefore = + enumerator.getTieringMetrics().getHeartbeatFailureCount(); + + // Swap to a failing gateway that throws on lakeTieringHeartbeat + CoordinatorGateway failingGateway = createFailingHeartbeatGateway(); + enumerator.setCoordinatorGateway(failingGateway); + + // Call requestTieringTableSplitsViaHeartBeat; it should throw FlinkRuntimeException + assertThatThrownBy(enumerator::requestTieringTableSplitsViaHeartBeat) + .isInstanceOf(FlinkRuntimeException.class); + + // Verify heartbeat failure counter was incremented + assertThat(enumerator.getTieringMetrics().getHeartbeatFailureCount()) + .isGreaterThan(heartbeatFailuresBefore); + } + } + + @Test + void testRequestTableEmptyMetricIncremented() throws Exception { + try (FlussMockSplitEnumeratorContext context = + new FlussMockSplitEnumeratorContext<>(1)) { + TieringSourceEnumerator enumerator = createTieringSourceEnumerator(flussConf, context); + enumerator.start(); + + // Register a reader so that the "request new table" path is taken + registerSingleReaderAndHandleSplitRequests(context, enumerator, 0, 0); + + long emptyBefore = enumerator.getTieringMetrics().getRequestTableEmptyCount(); + + // Swap to a gateway that returns empty heartbeat response (no tiering table) + CoordinatorGateway emptyResponseGateway = createEmptyHeartbeatResponseGateway(); + enumerator.setCoordinatorGateway(emptyResponseGateway); + + // Call requestTieringTableSplitsViaHeartBeat; it should succeed but find no table + enumerator.requestTieringTableSplitsViaHeartBeat(); + + // Verify request-table.empty counter was incremented + assertThat(enumerator.getTieringMetrics().getRequestTableEmptyCount()) + .isGreaterThan(emptyBefore); + } + } + + @Test + void testRequestTableFailureMetricIncrementedOnGenerateAndAssignSplits() throws Exception { + try (FlussMockSplitEnumeratorContext context = + new FlussMockSplitEnumeratorContext<>(1)) { + TieringSourceEnumerator enumerator = createTieringSourceEnumerator(flussConf, context); + enumerator.start(); + + long failuresBefore = enumerator.getTieringMetrics().getRequestTableFailureCount(); + + // generateAndAssignSplits with a non-null throwable should increment the counter + FlinkRuntimeException testException = + new FlinkRuntimeException("test request-table failure"); + assertThatThrownBy(() -> enumerator.generateAndAssignSplits(null, testException)) + .isSameAs(testException); + + assertThat(enumerator.getTieringMetrics().getRequestTableFailureCount()) + .isEqualTo(failuresBefore + 1); + } + } + + /** + * Creates a {@link CoordinatorGateway} proxy that always returns a failed future for {@code + * lakeTieringHeartbeat}. + */ + private CoordinatorGateway createFailingHeartbeatGateway() { + return (CoordinatorGateway) + Proxy.newProxyInstance( + CoordinatorGateway.class.getClassLoader(), + new Class[] {CoordinatorGateway.class}, + (proxy, method, args) -> { + if (method.getName().equals("lakeTieringHeartbeat")) { + CompletableFuture failed = + new CompletableFuture<>(); + failed.completeExceptionally( + new RuntimeException("simulated heartbeat failure")); + return failed; + } + throw new UnsupportedOperationException( + "Unexpected call: " + method.getName()); + }); + } + + /** + * Creates a {@link CoordinatorGateway} proxy that returns an empty {@link + * LakeTieringHeartbeatResponse} (no tiering table) for {@code lakeTieringHeartbeat}. + */ + private CoordinatorGateway createEmptyHeartbeatResponseGateway() { + return (CoordinatorGateway) + Proxy.newProxyInstance( + CoordinatorGateway.class.getClassLoader(), + new Class[] {CoordinatorGateway.class}, + (proxy, method, args) -> { + if (method.getName().equals("lakeTieringHeartbeat")) { + return CompletableFuture.completedFuture( + new LakeTieringHeartbeatResponse()); + } + throw new UnsupportedOperationException( + "Unexpected call: " + method.getName()); + }); + } }