diff --git a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java b/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java index 940d7fa05ec6..7d2329a1464f 100644 --- a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java +++ b/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java @@ -293,7 +293,7 @@ protected static String fullTableName(String catalogName, TableIdentifier identi protected MetricsReporter metricsReporter() { if (metricsReporter == null) { - metricsReporter = CatalogUtil.loadMetricsReporter(properties()); + metricsReporter = CatalogUtil.loadMetricsReporter(name(), properties()); } return metricsReporter; diff --git a/core/src/main/java/org/apache/iceberg/CatalogProperties.java b/core/src/main/java/org/apache/iceberg/CatalogProperties.java index 6b85ccbc87bc..ab36a93aefa3 100644 --- a/core/src/main/java/org/apache/iceberg/CatalogProperties.java +++ b/core/src/main/java/org/apache/iceberg/CatalogProperties.java @@ -33,6 +33,63 @@ private CatalogProperties() {} public static final String VIEW_OVERRIDE_PREFIX = "view-override."; public static final String METRICS_REPORTER_IMPL = "metrics-reporter-impl"; + /** + * Comma-separated list of Java regexes applied to {@code tableName()} of {@link + * org.apache.iceberg.metrics.ScanReport} and {@link org.apache.iceberg.metrics.CommitReport}. + * When set, only reports whose table name matches at least one pattern are forwarded to the + * configured {@link org.apache.iceberg.metrics.MetricsReporter}. + * + *

Each pattern is matched against the entire table name rather than any substring of it, so + * {@code prod\..*} matches {@code prod.db.table} but not {@code production.db.table}. Empty + * values are treated as not set. + */ + public static final String METRICS_REPORTER_TABLE_NAME_INCLUDE = + "metrics-reporter.table-name.include"; + + /** + * Comma-separated list of Java regexes applied to {@code tableName()} of {@link + * org.apache.iceberg.metrics.ScanReport} and {@link org.apache.iceberg.metrics.CommitReport}. + * When set, reports whose table name matches any pattern are dropped before reaching the + * configured {@link org.apache.iceberg.metrics.MetricsReporter}. An exclude match wins over an + * include match. + * + *

Each pattern is matched against the entire table name rather than any substring of it. Empty + * values are treated as not set. + */ + public static final String METRICS_REPORTER_TABLE_NAME_EXCLUDE = + "metrics-reporter.table-name.exclude"; + + /** + * Comma-separated list of Java regexes applied to the namespace of the table a {@link + * org.apache.iceberg.metrics.ScanReport} or {@link org.apache.iceberg.metrics.CommitReport} was + * produced for. When set, only reports for tables whose namespace matches at least one pattern + * are forwarded to the configured {@link org.apache.iceberg.metrics.MetricsReporter}. + * + *

The namespace is the dot-joined form of the table's namespace levels, with the catalog name + * removed, so {@code db} matches a table reported as {@code prod.db.table} in a catalog named + * {@code prod}. Filtering on the namespace is less prone to unintended matches than filtering on + * the table name, because a namespace is part of the table's identity rather than a naming + * convention. + * + *

Each pattern is matched against the entire namespace rather than any substring of it. Empty + * values are treated as not set. + */ + public static final String METRICS_REPORTER_NAMESPACE_INCLUDE = + "metrics-reporter.namespace.include"; + + /** + * Comma-separated list of Java regexes applied to the namespace of the table a {@link + * org.apache.iceberg.metrics.ScanReport} or {@link org.apache.iceberg.metrics.CommitReport} was + * produced for. When set, reports for tables whose namespace matches any pattern are dropped + * before reaching the configured {@link org.apache.iceberg.metrics.MetricsReporter}. An exclude + * match wins over an include match. + * + *

Each pattern is matched against the entire namespace rather than any substring of it. Empty + * values are treated as not set. + */ + public static final String METRICS_REPORTER_NAMESPACE_EXCLUDE = + "metrics-reporter.namespace.exclude"; + /** * Controls whether the catalog will cache table entries upon load. * diff --git a/core/src/main/java/org/apache/iceberg/CatalogUtil.java b/core/src/main/java/org/apache/iceberg/CatalogUtil.java index 4fa9fc30f1d0..96c6b0848b86 100644 --- a/core/src/main/java/org/apache/iceberg/CatalogUtil.java +++ b/core/src/main/java/org/apache/iceberg/CatalogUtil.java @@ -40,6 +40,7 @@ import org.apache.iceberg.io.StorageCredential; import org.apache.iceberg.io.SupportsBulkOperations; import org.apache.iceberg.io.SupportsStorageCredentials; +import org.apache.iceberg.metrics.FilteringMetricsReporter; import org.apache.iceberg.metrics.LoggingMetricsReporter; import org.apache.iceberg.metrics.MetricsReporter; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; @@ -513,38 +514,57 @@ public static void configureHadoopConf(Object maybeConfigurable, Object conf) { * loaded class cannot be cast to the given interface type */ public static MetricsReporter loadMetricsReporter(Map properties) { + return loadMetricsReporter(null, properties); + } + + /** + * Load a custom {@link MetricsReporter} implementation. + * + *

The implementation must have a no-arg constructor. + * + * @param catalogName name of the catalog the reporter is loaded for, used to derive a table's + * namespace when filtering by namespace is configured + * @param properties catalog properties which contains class name of a custom {@link + * MetricsReporter} implementation + * @return An initialized {@link MetricsReporter}. + * @throws IllegalArgumentException if class path not found or right constructor not found or the + * loaded class cannot be cast to the given interface type + */ + public static MetricsReporter loadMetricsReporter( + String catalogName, Map properties) { String impl = properties.get(CatalogProperties.METRICS_REPORTER_IMPL); + MetricsReporter reporter; if (impl == null) { - return LoggingMetricsReporter.instance(); - } + reporter = LoggingMetricsReporter.instance(); + } else { + LOG.info("Loading custom MetricsReporter implementation: {}", impl); + DynConstructors.Ctor ctor; + try { + ctor = + DynConstructors.builder(MetricsReporter.class) + .loader(CatalogUtil.class.getClassLoader()) + .impl(impl) + .buildChecked(); + } catch (NoSuchMethodException e) { + throw new IllegalArgumentException( + String.format( + "Cannot initialize MetricsReporter, missing no-arg constructor: %s", impl), + e); + } - LOG.info("Loading custom MetricsReporter implementation: {}", impl); - DynConstructors.Ctor ctor; - try { - ctor = - DynConstructors.builder(MetricsReporter.class) - .loader(CatalogUtil.class.getClassLoader()) - .impl(impl) - .buildChecked(); - } catch (NoSuchMethodException e) { - throw new IllegalArgumentException( - String.format("Cannot initialize MetricsReporter, missing no-arg constructor: %s", impl), - e); - } + try { + reporter = ctor.newInstance(); + } catch (ClassCastException e) { + throw new IllegalArgumentException( + String.format( + "Cannot initialize MetricsReporter, %s does not implement MetricsReporter.", impl), + e); + } - MetricsReporter reporter; - try { - reporter = ctor.newInstance(); - } catch (ClassCastException e) { - throw new IllegalArgumentException( - String.format( - "Cannot initialize MetricsReporter, %s does not implement MetricsReporter.", impl), - e); + reporter.initialize(properties); } - reporter.initialize(properties); - - return reporter; + return FilteringMetricsReporter.wrap(reporter, catalogName, properties); } public static String fullTableName(String catalogName, TableIdentifier identifier) { diff --git a/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java b/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java new file mode 100644 index 000000000000..de023d0422ff --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java @@ -0,0 +1,231 @@ +/* + * 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.iceberg.metrics; + +import java.util.List; +import java.util.Map; +import java.util.regex.Pattern; +import java.util.regex.PatternSyntaxException; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; + +/** + * A {@link MetricsReporter} wrapper that drops {@link ScanReport} and {@link CommitReport} + * instances that do not pass the configured namespace and table-name filters before forwarding to a + * delegate reporter. + * + *

Filtering happens on two levels, either of which may be configured on its own: + * + *

+ * + *

Each property holds a comma-separated list of Java regular expressions. A report is forwarded + * unless a filter drops it, and the levels are applied independently: an exclude match on either + * level drops the report, and when a level has an include list configured the report must match it. + * An exclude match always wins over an include match. When no property is set, {@link + * #wrap(MetricsReporter, String, Map)} returns the delegate unchanged so the default path incurs no + * overhead. + * + *

Patterns are matched against the entire namespace or table name rather than any substring of + * it, so {@code prod\..*} matches {@code prod.db.table} but not {@code production.db.table}. + * + *

{@link MetricsReport} subtypes other than {@link ScanReport} and {@link CommitReport} are + * forwarded without filtering, since they do not identify a table. + */ +public class FilteringMetricsReporter implements MetricsReporter { + + private final MetricsReporter delegate; + private final String catalogName; + private final List namespaceInclude; + private final List namespaceExclude; + private final List tableNameInclude; + private final List tableNameExclude; + + private FilteringMetricsReporter( + MetricsReporter delegate, + String catalogName, + List namespaceInclude, + List namespaceExclude, + List tableNameInclude, + List tableNameExclude) { + this.delegate = delegate; + this.catalogName = catalogName; + this.namespaceInclude = namespaceInclude; + this.namespaceExclude = namespaceExclude; + this.tableNameInclude = tableNameInclude; + this.tableNameExclude = tableNameExclude; + } + + /** + * Wraps the given delegate in a {@code FilteringMetricsReporter} when any of the namespace or + * table-name filters is configured in {@code properties}; otherwise returns the delegate + * unchanged so the default case incurs no runtime overhead. + * + * @param delegate the underlying reporter that receives forwarded reports + * @param catalogName name of the catalog the reports originate from, used to derive a table's + * namespace from the reported table name. When null, namespace filters cannot be applied and + * configuring them is rejected. + * @param properties catalog properties; consulted for the namespace and table-name filters + * @return either the delegate unchanged, or a new filtering wrapper around it + */ + public static MetricsReporter wrap( + MetricsReporter delegate, String catalogName, Map properties) { + List namespaceInclude = + compilePatterns(properties, CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE); + List namespaceExclude = + compilePatterns(properties, CatalogProperties.METRICS_REPORTER_NAMESPACE_EXCLUDE); + List tableNameInclude = + compilePatterns(properties, CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE); + List tableNameExclude = + compilePatterns(properties, CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE); + + if (namespaceInclude.isEmpty() + && namespaceExclude.isEmpty() + && tableNameInclude.isEmpty() + && tableNameExclude.isEmpty()) { + return delegate; + } + + if (catalogName == null && !(namespaceInclude.isEmpty() && namespaceExclude.isEmpty())) { + throw new IllegalArgumentException( + String.format( + "Cannot filter metrics by namespace without a catalog name: %s, %s", + CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, + CatalogProperties.METRICS_REPORTER_NAMESPACE_EXCLUDE)); + } + + return new FilteringMetricsReporter( + delegate, + catalogName, + namespaceInclude, + namespaceExclude, + tableNameInclude, + tableNameExclude); + } + + private static List compilePatterns( + Map properties, String propertyName) { + String value = properties.get(propertyName); + if (value == null || value.trim().isEmpty()) { + return ImmutableList.of(); + } + + ImmutableList.Builder patterns = ImmutableList.builder(); + for (String pattern : value.split(",", -1)) { + String trimmed = pattern.trim(); + if (trimmed.isEmpty()) { + continue; + } + + try { + patterns.add(Pattern.compile(trimmed)); + } catch (PatternSyntaxException e) { + throw new IllegalArgumentException( + String.format("Invalid regex for %s: %s", propertyName, trimmed), e); + } + } + + return patterns.build(); + } + + @Override + public void report(MetricsReport report) { + String tableName = tableName(report); + if (tableName == null) { + delegate.report(report); + return; + } + + if (!passes(tableName, tableNameInclude, tableNameExclude)) { + return; + } + + if (!namespaceInclude.isEmpty() || !namespaceExclude.isEmpty()) { + String namespace = namespace(tableName); + if (namespace == null || !passes(namespace, namespaceInclude, namespaceExclude)) { + return; + } + } + + delegate.report(report); + } + + private static boolean passes(String value, List include, List exclude) { + if (matchesAny(value, exclude)) { + return false; + } + + return include.isEmpty() || matchesAny(value, include); + } + + private static boolean matchesAny(String value, List patterns) { + for (Pattern pattern : patterns) { + if (pattern.matcher(value).matches()) { + return true; + } + } + + return false; + } + + /** + * Derives the namespace of a reported table by removing the catalog name prefix and the table + * name, mirroring how {@code CatalogUtil#fullTableName} builds the reported name. Returns an + * empty string for a table directly under the catalog, or null when the name does not carry the + * expected catalog prefix. + */ + private String namespace(String tableName) { + String prefix; + if (catalogName.contains("/") || catalogName.contains(":")) { + // URI-like catalog names are joined with /, as in thrift://host:port/db.table + prefix = catalogName.endsWith("/") ? catalogName : catalogName + "/"; + } else { + prefix = catalogName + "."; + } + + if (!tableName.startsWith(prefix)) { + return null; + } + + String withoutCatalog = tableName.substring(prefix.length()); + int lastDot = withoutCatalog.lastIndexOf('.'); + return lastDot < 0 ? "" : withoutCatalog.substring(0, lastDot); + } + + private static String tableName(MetricsReport report) { + if (report instanceof ScanReport) { + return ((ScanReport) report).tableName(); + } + + if (report instanceof CommitReport) { + return ((CommitReport) report).tableName(); + } + + return null; + } + + @Override + public void close() { + delegate.close(); + } +} diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java index 1d8e9da6d43c..002ff99bc1fb 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java @@ -62,6 +62,7 @@ import org.apache.iceberg.io.FileIOTracker; import org.apache.iceberg.io.StorageCredential; import org.apache.iceberg.io.SupportsStorageCredentials; +import org.apache.iceberg.metrics.FilteringMetricsReporter; import org.apache.iceberg.metrics.MetricsReporter; import org.apache.iceberg.metrics.MetricsReporters; import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; @@ -169,6 +170,8 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog private FileIO io = null; private MetricsReporter reporter = null; private ExecutorService metricsExecutor = null; + private Map reporterFilterProperties = ImmutableMap.of(); + private String reporterCatalogName = null; private boolean reportingViaRestEnabled; private Integer pageSize = null; private CloseableGroup closeables = null; @@ -273,7 +276,10 @@ public void initialize(String name, Map unresolved) { RESTCatalogProperties.SNAPSHOT_LOADING_MODE_DEFAULT.name()) .toUpperCase(Locale.US)); - this.reporter = CatalogUtil.loadMetricsReporter(mergedProps); + // name() is only available after super.initialize below, so use the name argument here + this.reporter = CatalogUtil.loadMetricsReporter(name, mergedProps); + this.reporterCatalogName = name; + this.reporterFilterProperties = mergedProps; this.closeables.addCloseable(reporter); this.reportingViaRestEnabled = @@ -666,11 +672,15 @@ private void trackFileIO(RESTTableOperations ops) { } } - private MetricsReporter metricsReporter(String metricsEndpoint, RESTClient restClient) { + @VisibleForTesting + MetricsReporter metricsReporter(String metricsEndpoint, RESTClient restClient) { if (reportingViaRestEnabled && endpoints.contains(Endpoint.V1_REPORT_METRICS)) { RESTMetricsReporter restMetricsReporter = new RESTMetricsReporter(restClient, metricsEndpoint, Map::of, metricsExecutor); - return MetricsReporters.combine(reporter, restMetricsReporter); + return MetricsReporters.combine( + reporter, + FilteringMetricsReporter.wrap( + restMetricsReporter, reporterCatalogName, reporterFilterProperties)); } else { return this.reporter; } diff --git a/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java b/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java index 84e79e35c9b5..6a5a8dc52d99 100644 --- a/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java +++ b/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java @@ -43,6 +43,7 @@ import org.apache.iceberg.io.StorageCredential; import org.apache.iceberg.io.SupportsBulkOperations; import org.apache.iceberg.io.SupportsStorageCredentials; +import org.apache.iceberg.metrics.FilteringMetricsReporter; import org.apache.iceberg.metrics.MetricsReport; import org.apache.iceberg.metrics.MetricsReporter; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; @@ -266,6 +267,28 @@ public void loadCustomMetricsReporter_badClass() { .hasMessageContaining("does not implement MetricsReporter"); } + @Test + public void loadMetricsReporter_wrappedWhenTableNameFilterPresent() { + MetricsReporter metricsReporter = + CatalogUtil.loadMetricsReporter( + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_IMPL, + TestMetricsReporterDefault.class.getName(), + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, + "prod\\..*")); + assertThat(metricsReporter).isInstanceOf(FilteringMetricsReporter.class); + } + + @Test + public void loadMetricsReporter_notWrappedWhenFilterAbsent() { + MetricsReporter metricsReporter = + CatalogUtil.loadMetricsReporter( + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_IMPL, + TestMetricsReporterDefault.class.getName())); + assertThat(metricsReporter).isInstanceOf(TestMetricsReporterDefault.class); + } + @Test public void fullTableNameWithDifferentValues() { String uriTypeCatalogName = "thrift://host:port/db.table"; diff --git a/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java b/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java new file mode 100644 index 000000000000..f5ad404d7d16 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java @@ -0,0 +1,434 @@ +/* + * 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.iceberg.metrics; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.List; +import java.util.regex.PatternSyntaxException; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.junit.jupiter.api.Test; + +public class TestFilteringMetricsReporter { + + private static final ScanReport SCAN_PROD = newScanReport("prod_db.orders"); + private static final ScanReport SCAN_TMP = newScanReport("prod_db.tmp_staging"); + private static final ScanReport SCAN_DEV = newScanReport("dev_db.orders"); + private static final CommitReport COMMIT_PROD = newCommitReport("prod_db.orders"); + + @Test + public void wrapReturnsDelegateWhenNoPropertiesSet() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = FilteringMetricsReporter.wrap(delegate, null, ImmutableMap.of()); + assertThat(wrapped).isSameAs(delegate); + } + + @Test + public void wrapReturnsDelegateWhenPropertiesAreEmpty() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + null, + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "", + CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, "")); + assertThat(wrapped).isSameAs(delegate); + } + + @Test + public void includeOnlyForwardsMatchingTableNames() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + null, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod_db\\..*")); + + wrapped.report(SCAN_PROD); + wrapped.report(SCAN_DEV); + wrapped.report(COMMIT_PROD); + + assertThat(delegate.reports).containsExactly(SCAN_PROD, COMMIT_PROD); + } + + @Test + public void excludeOnlyDropsMatchingTableNames() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + null, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, ".*\\.tmp_.*")); + + wrapped.report(SCAN_PROD); + wrapped.report(SCAN_TMP); + + assertThat(delegate.reports).containsExactly(SCAN_PROD); + } + + @Test + public void excludeWinsOverInclude() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + null, + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod_db\\..*", + CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, ".*\\.tmp_.*")); + + wrapped.report(SCAN_PROD); + wrapped.report(SCAN_TMP); + wrapped.report(SCAN_DEV); + + assertThat(delegate.reports).containsExactly(SCAN_PROD); + } + + @Test + public void unknownReportSubtypeIsForwardedWithoutFiltering() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + null, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "no_such\\..*")); + + MetricsReport unknown = new MetricsReport() {}; + wrapped.report(unknown); + + assertThat(delegate.reports).containsExactly(unknown); + } + + @Test + public void wrapThrowsClearErrorForInvalidRegex() { + assertThatThrownBy( + () -> + FilteringMetricsReporter.wrap( + new CapturingMetricsReporter(), + null, + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "[invalid"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE) + .hasMessageContaining("[invalid") + .hasCauseInstanceOf(PatternSyntaxException.class); + } + + @Test + public void loadMetricsReporterFiltersThroughUserConfiguredReporter() { + StaticCapturingReporter.REPORTS.clear(); + MetricsReporter reporter = + CatalogUtil.loadMetricsReporter( + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_IMPL, + StaticCapturingReporter.class.getName(), + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, + "prod_db\\..*")); + + reporter.report(SCAN_PROD); + reporter.report(SCAN_DEV); + reporter.report(COMMIT_PROD); + + assertThat(StaticCapturingReporter.REPORTS) + .as( + "Reports configured via metrics-reporter-impl receive only table names that pass the" + + " include filter") + .containsExactly(SCAN_PROD, COMMIT_PROD); + } + + @Test + public void closeIsDelegated() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + null, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, ".*")); + + wrapped.close(); + + assertThat(delegate.closed).isTrue(); + } + + @Test + public void includeAcceptsCommaSeparatedPatterns() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + null, + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, + "prod_db\\..*, analytics_db\\..*")); + + ScanReport analytics = newScanReport("analytics_db.events"); + wrapped.report(SCAN_PROD); + wrapped.report(analytics); + wrapped.report(SCAN_DEV); + + assertThat(delegate.reports).containsExactly(SCAN_PROD, analytics); + } + + @Test + public void excludeAcceptsCommaSeparatedPatterns() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + null, + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, ".*\\.tmp_.*,dev_db\\..*")); + + wrapped.report(SCAN_PROD); + wrapped.report(SCAN_TMP); + wrapped.report(SCAN_DEV); + + assertThat(delegate.reports).containsExactly(SCAN_PROD); + } + + @Test + public void patternsMatchWholeNameNotSubstring() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + null, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod\\..*")); + + ScanReport prod = newScanReport("prod.orders"); + // names that a substring match would wrongly accept + wrapped.report(prod); + wrapped.report(newScanReport("production.orders")); + wrapped.report(newScanReport("prod_sandbox.orders")); + wrapped.report(newScanReport("staging.prod.orders")); + + assertThat(delegate.reports).containsExactly(prod); + } + + @Test + public void namespaceIncludeFiltersOnNamespaceOnly() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "cat", + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "prod,analytics")); + + ScanReport prod = newScanReport("cat.prod.orders"); + ScanReport analytics = newScanReport("cat.analytics.events"); + wrapped.report(prod); + wrapped.report(analytics); + wrapped.report(newScanReport("cat.staging.orders")); + // a namespace that only shares a prefix must not match + wrapped.report(newScanReport("cat.production.orders")); + + assertThat(delegate.reports).containsExactly(prod, analytics); + } + + @Test + public void namespaceExcludeDropsMatchingNamespaces() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "cat", + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_NAMESPACE_EXCLUDE, "staging,sandbox")); + + ScanReport prod = newScanReport("cat.prod.orders"); + wrapped.report(prod); + wrapped.report(newScanReport("cat.staging.orders")); + wrapped.report(newScanReport("cat.sandbox.orders")); + + assertThat(delegate.reports).containsExactly(prod); + } + + @Test + public void namespaceAndTableNameFiltersCombine() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "cat", + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "prod", + CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, ".*\\.bench_.*")); + + ScanReport prod = newScanReport("cat.prod.orders"); + wrapped.report(prod); + // in the included namespace, but excluded by table name + wrapped.report(newScanReport("cat.prod.bench_scratch")); + // outside the included namespace + wrapped.report(newScanReport("cat.staging.orders")); + + assertThat(delegate.reports).containsExactly(prod); + } + + @Test + public void namespaceFilterHandlesMultiLevelAndEmptyNamespaces() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "cat", + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "a\\.b")); + + ScanReport nested = newScanReport("cat.a.b.orders"); + wrapped.report(nested); + wrapped.report(newScanReport("cat.a.orders")); + // table directly under the catalog has an empty namespace + wrapped.report(newScanReport("cat.orders")); + + assertThat(delegate.reports).containsExactly(nested); + } + + @Test + public void namespaceFilterHandlesCatalogNameContainingDots() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "my.cat", + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "db")); + + ScanReport report = newScanReport("my.cat.db.orders"); + wrapped.report(report); + wrapped.report(newScanReport("my.cat.other.orders")); + + assertThat(delegate.reports).containsExactly(report); + } + + @Test + public void namespaceFilterHandlesUriStyleCatalogNames() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "thrift://localhost:9083", + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "db")); + + ScanReport report = newScanReport("thrift://localhost:9083/db.orders"); + wrapped.report(report); + wrapped.report(newScanReport("thrift://localhost:9083/other.orders")); + + assertThat(delegate.reports).containsExactly(report); + } + + @Test + public void namespaceFilterDropsReportsWithoutExpectedCatalogPrefix() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "cat", + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, ".*")); + + // the namespace cannot be derived, so the report cannot be shown to pass the filter + wrapped.report(newScanReport("other.db.orders")); + + assertThat(delegate.reports).isEmpty(); + } + + @Test + public void namespaceFilterWithoutCatalogNameIsRejected() { + assertThatThrownBy( + () -> + FilteringMetricsReporter.wrap( + new CapturingMetricsReporter(), + null, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "prod"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE); + } + + @Test + public void wrapThrowsClearErrorForInvalidNamespaceRegex() { + assertThatThrownBy( + () -> + FilteringMetricsReporter.wrap( + new CapturingMetricsReporter(), + "cat", + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_NAMESPACE_EXCLUDE, "[invalid"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(CatalogProperties.METRICS_REPORTER_NAMESPACE_EXCLUDE) + .hasMessageContaining("[invalid"); + } + + private static ScanReport newScanReport(String tableName) { + return ImmutableScanReport.builder() + .tableName(tableName) + .snapshotId(1L) + .filter(Expressions.alwaysTrue()) + .schemaId(1) + .projectedFieldIds(ImmutableList.of()) + .projectedFieldNames(ImmutableList.of()) + .scanMetrics(ImmutableScanMetricsResult.builder().build()) + .metadata(ImmutableMap.of()) + .build(); + } + + private static CommitReport newCommitReport(String tableName) { + return ImmutableCommitReport.builder() + .tableName(tableName) + .snapshotId(1L) + .sequenceNumber(1L) + .operation("append") + .commitMetrics(ImmutableCommitMetricsResult.builder().build()) + .metadata(ImmutableMap.of()) + .build(); + } + + private static class CapturingMetricsReporter implements MetricsReporter { + private final List reports = Lists.newArrayList(); + private boolean closed = false; + + @Override + public void report(MetricsReport report) { + reports.add(report); + } + + @Override + public void close() { + this.closed = true; + } + } + + /** + * Public no-arg reporter usable via {@code metrics-reporter-impl}. Captured reports live on a + * static list so the test can inspect what reached the underlying reporter after CatalogUtil + * instantiated it via reflection. + */ + public static class StaticCapturingReporter implements MetricsReporter { + static final List REPORTS = Lists.newCopyOnWriteArrayList(); + + @Override + public void report(MetricsReport report) { + REPORTS.add(report); + } + } +} diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java index d2d45c6173f5..276d8bc2f0df 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java +++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java @@ -30,6 +30,7 @@ import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -93,6 +94,10 @@ import org.apache.iceberg.io.StorageCredential; import org.apache.iceberg.io.SupportsStorageCredentials; import org.apache.iceberg.metrics.CommitReport; +import org.apache.iceberg.metrics.ImmutableScanMetricsResult; +import org.apache.iceberg.metrics.ImmutableScanReport; +import org.apache.iceberg.metrics.MetricsReporter; +import org.apache.iceberg.metrics.ScanReport; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; @@ -3990,6 +3995,49 @@ public Table registerTable( } } + @Test + public void metricsFilterAppliesToRestMetricsReporter() throws Exception { + RESTCatalog catalog = + initCatalog( + "withFilter", + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod_db\\..*")); + + String metricsEndpoint = "/v1/prefix/namespaces/x/tables/y/metrics"; + RESTClient mockClient = mock(RESTClient.class); + MetricsReporter combined = + catalog.sessionCatalog().metricsReporter(metricsEndpoint, mockClient); + + ScanReport excluded = scanReport("dev_db.scratch"); + ScanReport included = scanReport("prod_db.orders"); + + combined.report(excluded); + combined.report(included); + + // RESTMetricsReporter dispatches each report to a single-threaded executor asynchronously, so + // the resulting client.post() happens off-thread. The excluded report is submitted first and + // dropped by the FilteringMetricsReporter before it reaches the REST client; the included one + // is submitted next and produces exactly one post. Because the executor is single-threaded and + // FIFO, observing the included post (waited for via timeout) guarantees the excluded report has + // already been processed, so times(1) confirms the filter suppressed it. + verify(mockClient, timeout(5000).times(1)) + .post(eq(metricsEndpoint), any(RESTRequest.class), any(), any(Supplier.class), any()); + + catalog.close(); + } + + private static ScanReport scanReport(String tableName) { + return ImmutableScanReport.builder() + .tableName(tableName) + .snapshotId(1L) + .filter(Expressions.alwaysTrue()) + .schemaId(1) + .projectedFieldIds(ImmutableList.of()) + .projectedFieldNames(ImmutableList.of()) + .scanMetrics(ImmutableScanMetricsResult.builder().build()) + .metadata(ImmutableMap.of()) + .build(); + } + private RESTCatalog catalog(RESTCatalogAdapter adapter) { RESTCatalog catalog = new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter); diff --git a/docs/docs/metrics-reporting.md b/docs/docs/metrics-reporting.md index 4ca452b0d503..eb0b042befdd 100644 --- a/docs/docs/metrics-reporting.md +++ b/docs/docs/metrics-reporting.md @@ -147,6 +147,37 @@ public class InMemoryMetricsReporter implements MetricsReporter { The [catalog property](catalog-properties.md) `metrics-reporter-impl` allows registering a given [`MetricsReporter`](https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/metrics/MetricsReporter.java) by specifying its fully-qualified class name, e.g. `metrics-reporter-impl=org.apache.iceberg.metrics.InMemoryMetricsReporter`. +### Filtering which tables are reported + +Reports forwarded to the configured `MetricsReporter` can be filtered on two levels, either of which may be used on its own: + +| Property | Effect | +|---|---| +| `metrics-reporter.namespace.include` | Forward only reports for tables whose namespace matches; drop the rest. | +| `metrics-reporter.namespace.exclude` | Drop reports for tables whose namespace matches; forward the rest. | +| `metrics-reporter.table-name.include` | Forward only reports whose table name matches; drop the rest. | +| `metrics-reporter.table-name.exclude` | Drop reports whose table name matches; forward the rest. | + +Each property accepts a comma-separated list of Java regular expressions. Namespace patterns are matched against the table's namespace levels joined by dots, with the catalog name removed; table-name patterns are matched against `ScanReport.tableName()` and `CommitReport.tableName()`, which include the catalog name. + +Patterns are matched against the **entire** namespace or table name rather than any substring of it. This matters in practice: `prod\..*` matches `prod.db.table` but not `production.db.table` or `prod_sandbox.db.table`, which a substring match would wrongly accept. + +An `exclude` match always wins over an `include` match, and the two levels are applied independently — a report must survive both to be forwarded. When no property is set, behavior is identical to today: every report is forwarded, and no wrapper is instantiated. Empty values are treated as not set, to avoid accidentally silencing all metrics on misconfiguration. + +Filtering by namespace is less error-prone than filtering by table name, because a namespace is part of a table's identity rather than a naming convention: a table added to an included namespace later is picked up automatically, and no table outside that namespace can match by accident. Table-name patterns remain available for cases a namespace cannot express, such as excluding a few noisy tables inside an otherwise interesting namespace. + +For example, to report on the `prod` and `analytics` namespaces while dropping benchmark tables anywhere: + +``` +metrics-reporter-impl=org.apache.iceberg.metrics.LoggingMetricsReporter +metrics-reporter.namespace.include=prod,analytics +metrics-reporter.table-name.exclude=.*\.bench_.* +``` + +The filter applies uniformly to all `MetricsReporter` implementations (`LoggingMetricsReporter`, `RESTMetricsReporter`, and custom user-supplied ones). Reports whose subtype does not identify a table (i.e. anything other than `ScanReport` and `CommitReport`) are forwarded without filtering. + +Namespace filtering requires the catalog name, which the catalog supplies when it loads the reporter. Configuring a namespace filter where no catalog name is available fails at initialization with a clear error rather than silently dropping reports. + ### Via the Java API during Scan planning Independently of the [`MetricsReporter`](https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/metrics/MetricsReporter.java) being registered at the catalog level via the `metrics-reporter-impl` property, it is also possible to supply additional reporters during scan planning as shown below: