[spark] Expose Paimon metrics via JMX for Prometheus scraping

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/metric/SparkMetricGroup.java

@@ -0,0 +1,153 @@
+/*
+ * 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.paimon.spark.metric;
+
+import org.apache.paimon.metrics.Counter;
+import org.apache.paimon.metrics.Gauge;
+import org.apache.paimon.metrics.Histogram;
+import org.apache.paimon.metrics.MetricGroupImpl;
+
+import com.codahale.metrics.MetricRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A {@link org.apache.paimon.metrics.MetricGroup} that extends {@link MetricGroupImpl} to
+ * additionally register metrics with a Codahale {@link MetricRegistry} for exposure via Spark's
+ * MetricsSystem (JMX, Prometheus, etc.).
+ *
+ * <p>This preserves full backward compatibility with the existing Spark UI integration (which reads
+ * from {@link MetricGroupImpl#getMetrics()}) while also making metrics available to external
+ * monitoring systems.
+ */
+public class SparkMetricGroup extends MetricGroupImpl {
+
+    private static final Logger LOG = LoggerFactory.getLogger(SparkMetricGroup.class);
+
+    private final MetricRegistry codahaleRegistry;
+    private final String metricPrefix;
+    private final Set<String> registeredNames;
+
+    public SparkMetricGroup(
+            String groupName, Map<String, String> variables, MetricRegistry codahaleRegistry) {
+        super(groupName, variables);
+        this.codahaleRegistry = codahaleRegistry;
+        this.metricPrefix = buildPrefix(groupName, variables);
+        this.registeredNames = new HashSet<>();
+    }
+
+    @Override
+    public Counter counter(String name) {
+        Counter paimonCounter = super.counter(name);
+        if (paimonCounter != null) {
+            registerCodahaleGauge(
+                    codahaleName(name), (com.codahale.metrics.Gauge<Long>) paimonCounter::getCount);
+        }
+        return paimonCounter;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public <T> Gauge<T> gauge(String name, Gauge<T> gauge) {
+        Gauge<T> paimonGauge = super.gauge(name, gauge);
+        if (paimonGauge != null) {
+            registerCodahaleGauge(
+                    codahaleName(name), (com.codahale.metrics.Gauge<Object>) paimonGauge::getValue);
+        }
+        return paimonGauge;
+    }
+
+    @Override
+    public Histogram histogram(String name, int windowSize) {
+        Histogram paimonHistogram = super.histogram(name, windowSize);
+        if (paimonHistogram != null) {
+            String base = codahaleName(name);
+            registerCodahaleGauge(
+                    base + ".count", (com.codahale.metrics.Gauge<Long>) paimonHistogram::getCount);
+            registerCodahaleGauge(
+                    base + ".mean",
+                    (com.codahale.metrics.Gauge<Double>)
+                            () -> paimonHistogram.getStatistics().getMean());
+            registerCodahaleGauge(
+                    base + ".min",
+                    (com.codahale.metrics.Gauge<Long>)
+                            () -> paimonHistogram.getStatistics().getMin());
+            registerCodahaleGauge(
+                    base + ".max",
+                    (com.codahale.metrics.Gauge<Long>)
+                            () -> paimonHistogram.getStatistics().getMax());
+            registerCodahaleGauge(
+                    base + ".p50",
+                    (com.codahale.metrics.Gauge<Double>)
+                            () -> paimonHistogram.getStatistics().getQuantile(0.5));
+            registerCodahaleGauge(
+                    base + ".p95",
+                    (com.codahale.metrics.Gauge<Double>)
+                            () -> paimonHistogram.getStatistics().getQuantile(0.95));
+            registerCodahaleGauge(
+                    base + ".p99",
+                    (com.codahale.metrics.Gauge<Double>)
+                            () -> paimonHistogram.getStatistics().getQuantile(0.99));
+        }
+        return paimonHistogram;
+    }
+
+    @Override
+    public void close() {
+        for (String name : registeredNames) {
+            codahaleRegistry.remove(name);
+        }
+        registeredNames.clear();
+        super.close();
+    }
+
+    private String codahaleName(String metricName) {
+        return metricPrefix + "." + metricName;
+    }
+
+    private void registerCodahaleGauge(String name, com.codahale.metrics.Gauge<?> gauge) {
+        // Always remove first to ensure we replace any stale gauge from a previous commit.
+        // This avoids a race window in the try-register/catch-remove/re-register pattern.
+        codahaleRegistry.remove(name);
+        try {
+            codahaleRegistry.register(name, gauge);
+            registeredNames.add(name);
+        } catch (IllegalArgumentException e) {
+            LOG.warn("Failed to register Codahale metric '{}': {}", name, e.getMessage());
+        }
+    }
+
+    private static String buildPrefix(String groupName, Map<String, String> variables) {
+        StringBuilder sb = new StringBuilder("paimon");
+        String table = variables.get("table");
+        if (table != null && !table.isEmpty()) {
+            sb.append('.').append(sanitize(table));
+        }
+        sb.append('.').append(sanitize(groupName));
+        return sb.toString();
+    }
+
+    private static String sanitize(String name) {
+        return name.replaceAll("[^a-zA-Z0-9_]", "_");
+    }
+}

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala

@@ -32,6 +32,7 @@ import org.apache.paimon.io.{CompactIncrement, DataIncrement}
 import org.apache.paimon.manifest.FileKind
 import org.apache.paimon.spark.{SparkRow, SparkTypeUtils}
 import org.apache.paimon.spark.catalog.functions.BucketFunction
+import org.apache.paimon.spark.metric.SparkMetricRegistry
 import org.apache.paimon.spark.schema.SparkSystemColumns.{BUCKET_COL, ROW_KIND_COL}
 import org.apache.paimon.spark.sort.TableSorter
 import org.apache.paimon.spark.util.OptionUtils.paimonExtensionEnabled
@@ -421,7 +422,9 @@ case class PaimonSparkWriter(
     } else {
       writeBuilder
     }
+    val metricRegistry = SparkMetricRegistry()
     val tableCommit = finalWriteBuilder.newCommit()
+    tableCommit.withMetricRegistry(metricRegistry)
     try {
       tableCommit.commit(commitMessages.toList.asJava)
     } catch {

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/metric/SparkMetricRegistry.scala

@@ -18,11 +18,12 @@
 
 package org.apache.paimon.spark.metric
 
-import org.apache.paimon.metrics.{Gauge, Metric, MetricGroup, MetricGroupImpl, MetricRegistry}
+import org.apache.paimon.metrics.{Gauge, Metric, MetricGroup, MetricRegistry}
 import org.apache.paimon.operation.metrics.{CommitMetrics, ScanMetrics, WriterBufferMetric}
 import org.apache.paimon.spark._
 
 import org.apache.spark.sql.connector.metric.CustomTaskMetric
+import org.apache.spark.sql.paimon.PaimonMetricsSource
 
 import java.util.{Map => JMap}
 
@@ -35,7 +36,8 @@ case class SparkMetricRegistry() extends MetricRegistry {
   override def createMetricGroup(
       groupName: String,
       variables: JMap[String, String]): MetricGroup = {
-    val metricGroup = new MetricGroupImpl(groupName, variables)
+    val source = PaimonMetricsSource.getOrCreate()
+    val metricGroup = new SparkMetricGroup(groupName, variables, source.metricRegistry)
     metricGroups.put(groupName, metricGroup)
     metricGroup
   }

paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/PaimonMetricsSource.scala

@@ -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.spark.sql.paimon
+
+import com.codahale.metrics.jmx.JmxReporter
+import org.apache.spark.SparkEnv
+import org.apache.spark.metrics.source.Source
+
+/**
+ * A Spark [[Source]] that holds a shared Codahale [[com.codahale.metrics.MetricRegistry]]. Paimon
+ * metrics registered via [[org.apache.paimon.spark.metric.SparkMetricGroup]] are exposed through
+ * this source.
+ *
+ * A [[JmxReporter]] is started on the registry so that metrics are registered as JMX MBeans
+ * immediately when added. This is necessary because Spark's MetricsSystem takes a snapshot of the
+ * registry at `registerSource` time and does not pick up metrics added later.
+ *
+ * This class must live under `org.apache.spark` because Spark's [[Source]] trait is
+ * package-private.
+ *
+ * Use [[PaimonMetricsSource.getOrCreate()]] to obtain the singleton instance.
+ */
+class PaimonMetricsSource extends Source {
+
+  override val sourceName: String = "paimon"
+
+  override val metricRegistry: com.codahale.metrics.MetricRegistry =
+    new com.codahale.metrics.MetricRegistry()
+
+  private val jmxReporter: JmxReporter = JmxReporter
+    .forRegistry(metricRegistry)
+    .inDomain("paimon")
+    .build()
+
+  jmxReporter.start()
+
+  def stop(): Unit = {
+    jmxReporter.stop()
+  }
+}
+
+object PaimonMetricsSource {
+
+  @volatile private var instance: PaimonMetricsSource = _
+
+  def getOrCreate(): PaimonMetricsSource = {
+    if (instance == null) {
+      synchronized {
+        if (instance == null) {
+          val source = new PaimonMetricsSource()
+          val env = SparkEnv.get
+          if (env != null) {
+            env.metricsSystem.registerSource(source)
+          }
+          Runtime.getRuntime.addShutdownHook(new Thread("paimon-jmx-reporter-shutdown") {
+            override def run(): Unit = source.stop()
+          })
+          instance = source
+        }
+      }
+    }
+    instance
+  }
+}