diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/metrics/UdfMetrics.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/metrics/UdfMetrics.java new file mode 100644 index 0000000000000..3b04315d956e5 --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/metrics/UdfMetrics.java @@ -0,0 +1,98 @@ +/* + * 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.table.runtime.operators.metrics; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.ThreadSafeSimpleCounter; +import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram; +import org.apache.flink.util.Preconditions; + +/** + * Per-operator metrics for a single user-defined function, registered under {@code + * .udf.}: {@code udfProcessingTime} (a latency histogram) and {@code + * udfExceptionCount} (a counter of exceptions escaping the function). + * + *

Timing is sampled: only one invocation out of every {@code sampleInterval} is measured, so a + * hot function does not pay {@code System.nanoTime()} on every record. The sample decision advances + * a plain {@code int} and is only ever taken on the task thread, so it needs no synchronization. + * The two registered metrics are safe for the async completion thread to touch: the histogram + * synchronizes internally and the exception counter is {@link ThreadSafeSimpleCounter}. + */ +public final class UdfMetrics { + + private static final int HISTORY_SIZE = 128; + + private final int sampleInterval; + private final Histogram processingTime; + private final Counter exceptionCount; + + // Sample counter; only touched on the task thread, hence a plain int with no synchronization. + private int invocationCount = 0; + + private UdfMetrics(int sampleInterval, Histogram processingTime, Counter exceptionCount) { + this.sampleInterval = sampleInterval; + this.processingTime = processingTime; + this.exceptionCount = exceptionCount; + } + + /** + * Registers the {@code udfProcessingTime} histogram and {@code udfExceptionCount} counter under + * {@code .udf.}. + * + * @param sampleInterval measure one invocation out of every {@code sampleInterval}; must be + * {@code >= 1} ({@code 1} measures every invocation) + */ + public static UdfMetrics register( + MetricGroup operatorMetricGroup, String udfName, int sampleInterval) { + Preconditions.checkArgument( + sampleInterval >= 1, + "UDF metric sample interval must be >= 1, but was %s.", + sampleInterval); + MetricGroup group = operatorMetricGroup.addGroup("udf", udfName); + return new UdfMetrics( + sampleInterval, + group.histogram( + "udfProcessingTime", new DescriptiveStatisticsHistogram(HISTORY_SIZE)), + group.counter("udfExceptionCount", new ThreadSafeSimpleCounter())); + } + + /** + * Returns {@code true} for the one invocation in every {@code sampleInterval} whose processing + * time should be measured. Must be called once per invocation, on the task thread only. + */ + public boolean shouldSample() { + if (sampleInterval == 1) { + return true; + } + invocationCount = (invocationCount + 1 < sampleInterval) ? invocationCount + 1 : 0; + return invocationCount == 1; + } + + /** Records an elapsed processing time (nanoseconds) for a sampled invocation. */ + public void update(long elapsedNanos) { + processingTime.update(elapsedNanos); + } + + /** Counts one exception escaping the user-defined function. */ + public void markException() { + exceptionCount.inc(); + } +} diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/metrics/UdfMetricsTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/metrics/UdfMetricsTest.java new file mode 100644 index 0000000000000..49a5c78714479 --- /dev/null +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/metrics/UdfMetricsTest.java @@ -0,0 +1,140 @@ +/* + * 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.table.runtime.operators.metrics; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.ThreadSafeSimpleCounter; +import org.apache.flink.metrics.testutils.MetricListener; +import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link UdfMetrics}. */ +class UdfMetricsTest { + + private MetricListener metricListener; + + @BeforeEach + void setUp() { + metricListener = new MetricListener(); + } + + @Test + void sampleEveryNth() { + UdfMetrics metrics = register("myUdf", 100); + + List sampledCalls = new ArrayList<>(); + for (int call = 1; call <= 300; call++) { + if (metrics.shouldSample()) { + sampledCalls.add(call); + } + } + + // With interval 100 exactly one call in every 100 is sampled, on the 1st, 101st, 201st. + assertThat(sampledCalls).containsExactly(1, 101, 201); + } + + @Test + void sampleEveryCallWhenIntervalOne() { + UdfMetrics metrics = register("myUdf", 1); + + for (int call = 0; call < 10; call++) { + assertThat(metrics.shouldSample()).isTrue(); + } + } + + @Test + void rejectsNonPositiveInterval() { + assertThatThrownBy(() -> register("myUdf", 0)).isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> register("myUdf", -1)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void registrationNamesAndTypes() { + register("myUdf", 100); + + assertThat(metricListener.getHistogram("udf", "myUdf", "udfProcessingTime")) + .get() + .isInstanceOf(DescriptiveStatisticsHistogram.class); + assertThat(metricListener.getCounter("udf", "myUdf", "udfExceptionCount")) + .get() + .isInstanceOf(ThreadSafeSimpleCounter.class); + } + + @Test + void updateFeedsHistogram() { + UdfMetrics metrics = register("myUdf", 1); + Histogram histogram = + metricListener.getHistogram("udf", "myUdf", "udfProcessingTime").get(); + + metrics.update(10L); + metrics.update(20L); + metrics.update(30L); + + assertThat(histogram.getCount()).isEqualTo(3L); + assertThat(histogram.getStatistics().getMin()).isEqualTo(10L); + assertThat(histogram.getStatistics().getMax()).isEqualTo(30L); + } + + @Test + void markExceptionCounts() { + UdfMetrics metrics = register("myUdf", 1); + Counter counter = metricListener.getCounter("udf", "myUdf", "udfExceptionCount").get(); + + metrics.markException(); + metrics.markException(); + + assertThat(counter.getCount()).isEqualTo(2L); + } + + @Test + void metricsAreIndependentPerUdf() { + UdfMetrics first = register("firstUdf", 1); + register("secondUdf", 1); + Histogram firstHistogram = + metricListener.getHistogram("udf", "firstUdf", "udfProcessingTime").get(); + Counter firstCounter = + metricListener.getCounter("udf", "firstUdf", "udfExceptionCount").get(); + Histogram secondHistogram = + metricListener.getHistogram("udf", "secondUdf", "udfProcessingTime").get(); + Counter secondCounter = + metricListener.getCounter("udf", "secondUdf", "udfExceptionCount").get(); + + first.update(10L); + first.markException(); + + assertThat(firstHistogram.getCount()).isEqualTo(1L); + assertThat(firstCounter.getCount()).isEqualTo(1L); + assertThat(secondHistogram.getCount()).isZero(); + assertThat(secondCounter.getCount()).isZero(); + } + + private UdfMetrics register(String udfName, int sampleInterval) { + return UdfMetrics.register(metricListener.getMetricGroup(), udfName, sampleInterval); + } +}