[SPARK-53325] Support Prometheus 2.0 text-based-format and best practices for metrics naming

### What changes were proposed in this pull request?

This PR adds support for Prometheus text-based-format and best practices for metrics naming

Existing format

```
metrics_jvm_bufferPool_direct_capacity_Number{type="gauges"} 98348
metrics_jvm_bufferPool_direct_capacity_Value{type="gauges"} 98348
metrics_jvm_bufferPool_direct_count_Number{type="gauges"} 41
metrics_jvm_bufferPool_direct_count_Value{type="gauges"} 41
metrics_kubernetes_client_http_response_latency_nanos_Count{type="histograms"} 26910
metrics_kubernetes_client_http_response_latency_nanos_Max{type="histograms"} 232417143
metrics_kubernetes_client_http_response_latency_nanos_Mean{type="histograms"} 1.1410164260725182E7
metrics_kubernetes_client_http_response_latency_nanos_Min{type="histograms"} 2931711
metrics_kubernetes_client_http_response_latency_nanos_50thPercentile{type="histograms"} 7559152.0
metrics_kubernetes_client_http_response_latency_nanos_75thPercentile{type="histograms"} 9440850.0
metrics_kubernetes_client_http_response_latency_nanos_95thPercentile{type="histograms"} 1.2576766E7
metrics_kubernetes_client_http_response_latency_nanos_98thPercentile{type="histograms"} 1.34034482E8
metrics_kubernetes_client_http_response_latency_nanos_99thPercentile{type="histograms"} 1.34034482E8
metrics_kubernetes_client_http_response_latency_nanos_999thPercentile{type="histograms"} 1.34034482E8
metrics_kubernetes_client_http_response_latency_nanos_StdDev{type="histograms"} 2.177784612259799E7
metrics_kubernetes_client_pods_get_Count{type="counters"} 8967
metrics_kubernetes_client_pods_get_MeanRate{type="counters"} 0.02678169644780033
metrics_kubernetes_client_pods_get_OneMinuteRate{type="counters"} 0.049758750361204154
metrics_kubernetes_client_pods_get_FiveMinuteRate{type="counters"} 0.035255140329213855
metrics_kubernetes_client_pods_get_FifteenMinuteRate{type="counters"} 0.02931221844089468
```

with this patch, operator would be able to export format matching Prometheus 2.0 recommended practice like

```
# HELP jvm.bufferPool.direct.capacity Gauge metric
# TYPE jvm.bufferPool.direct.capacity gauge
jvm_bufferpool_direct_capacity 99011

# HELP jvm.bufferPool.direct.count Gauge metric
# TYPE jvm.bufferPool.direct.count gauge
jvm_bufferpool_direct_count 53

# HELP kubernetes_client_1xx_total Meter count
# TYPE kubernetes_client_1xx_total counter
kubernetes_client_1xx_total 36

# HELP kubernetes_client_http_response_latency_nanos Histogram metric
# TYPE kubernetes_client_http_response_latency_nanos histogram
kubernetes_client_http_response_latency_nanos_bucket{le="0.5"} 72
kubernetes_client_http_response_latency_nanos_bucket{le="0.75"} 72
kubernetes_client_http_response_latency_nanos_bucket{le="0.95"} 72
kubernetes_client_http_response_latency_nanos_bucket{le="0.98"} 72
kubernetes_client_http_response_latency_nanos_bucket{le="0.99"} 72
kubernetes_client_http_response_latency_nanos_bucket{le="+Inf"} 72
kubernetes_client_http_response_latency_nanos_count 72
kubernetes_client_http_response_latency_nanos_sum 4.656066964000001E9
```

### Why are the changes needed?

It's Prometheus 2.0 best practice for using the next format with necessary comments.
Also, some common scrapers (like Datadog) rely on these metadata (e.g. # HELP and # TYPE) to parse metrics correctly. They may skip metrics if these are missing.

### Does this PR introduce _any_ user-facing change?

New functionalities becomes available (for metrics format)

### How was this patch tested?

CIs / curl on :19090/prometheus to validate the format

### Was this patch authored or co-authored using generative AI tooling?

No

Author: jiangzho

docs/config_properties.md

@@ -29,6 +29,8 @@
  | spark.kubernetes.operator.metrics.clientMetricsEnabled | Boolean | true | false | Enable KubernetesClient metrics for measuring the HTTP traffic to the Kubernetes API Server. Since the metrics is collected via Okhttp interceptors, can be disabled when opt in customized interceptors. | 
  | spark.kubernetes.operator.metrics.clientMetricsGroupByResponseCodeEnabled | Boolean | true | false | When enabled, additional metrics group by http response code group(1xx, 2xx, 3xx, 4xx, 5xx) received from API server will be added. Users can disable it when their monitoring system can combine lower level kubernetes.client.http.response.<3-digit-response-code> metrics. | 
  | spark.kubernetes.operator.metrics.port | Integer | 19090 | false | The port used for checking metrics | 
+ | spark.kubernetes.operator.enablePrometheusTextBasedFormat | Boolean | true | false | Whether or not to enable text-based format for Prometheus 2.0, as recommended by https://prometheus.io/docs/instrumenting/exposition_formats/#text-based-format | 
+ | spark.kubernetes.operator.enableSanitizePrometheusMetricsName | Boolean | true | false | Whether or not to enable automatic name sanitizing for all metrics based on best-practice guide from Prometheus https://prometheus.io/docs/practices/naming/ | 
  | spark.kubernetes.operator.health.probePort | Integer | 19091 | false | The port used for health/readiness check probe status. | 
  | spark.kubernetes.operator.health.sentinelExecutorPoolSize | Integer | 3 | false | Size of executor service in Sentinel Managers to check the health of sentinel resources. | 
  | spark.kubernetes.operator.health.sentinelResourceReconciliationDelaySeconds | Integer | 60 | true | Allowed max time(seconds) between spec update and reconciliation for sentinel resources. | 

spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java

@@ -334,6 +334,30 @@ public final class SparkOperatorConf {
           .defaultValue(19090)
           .build();
 
+  public static final ConfigOption<Boolean> EnablePrometheusTextBasedFormat =
+      ConfigOption.<Boolean>builder()
+          .key("spark.kubernetes.operator.enablePrometheusTextBasedFormat")
+          .enableDynamicOverride(false)
+          .description(
+              "Whether or not to enable text-based format for Prometheus 2.0, as "
+                  + "recommended by "
+                  + "https://prometheus.io/docs/instrumenting/exposition_formats/#text-based-format")
+          .typeParameterClass(Boolean.class)
+          .defaultValue(true)
+          .build();
+
+  public static final ConfigOption<Boolean> EnableSanitizePrometheusMetricsName =
+      ConfigOption.<Boolean>builder()
+          .key("spark.kubernetes.operator.enableSanitizePrometheusMetricsName")
+          .enableDynamicOverride(false)
+          .description(
+              "Whether or not to enable automatic name sanitizing for all metrics based on "
+                  + "best-practice guide from Prometheus "
+                  + "https://prometheus.io/docs/practices/naming/")
+          .typeParameterClass(Boolean.class)
+          .defaultValue(true)
+          .build();
+
   public static final ConfigOption<Integer> OPERATOR_PROBE_PORT =
       ConfigOption.<Integer>builder()
           .key("spark.kubernetes.operator.health.probePort")

spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/PrometheusPullModelHandler.java

@@ -28,22 +28,39 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.TimeUnit;
 
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Histogram;
+import com.codahale.metrics.Meter;
 import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.Snapshot;
+import com.codahale.metrics.Timer;
 import com.sun.net.httpserver.HttpExchange;
 import com.sun.net.httpserver.HttpHandler;
 import jakarta.servlet.http.HttpServletRequest;
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
+import org.apache.spark.k8s.operator.config.SparkOperatorConf;
 import org.apache.spark.metrics.sink.PrometheusServlet;
 
 /** Serves as simple Prometheus sink (pull model), presenting metrics snapshot as HttpHandler. */
 @Slf4j
 public class PrometheusPullModelHandler extends PrometheusServlet implements HttpHandler {
   private static final String EMPTY_RECORD_VALUE = "[]";
+  @Getter private final MetricRegistry registry;
+  @Getter private final boolean enablePrometheusTextBasedFormat;
+  @Getter private final boolean enableSanitizePrometheusMetricsName;
 
   public PrometheusPullModelHandler(Properties properties, MetricRegistry registry) {
     super(properties, registry);
+    this.registry = registry;
+    this.enablePrometheusTextBasedFormat =
+        SparkOperatorConf.EnablePrometheusTextBasedFormat.getValue();
+    this.enableSanitizePrometheusMetricsName =
+        SparkOperatorConf.EnableSanitizePrometheusMetricsName.getValue();
   }
 
   @Override
@@ -58,13 +75,21 @@ public void stop() {
 
   @Override
   public void handle(HttpExchange exchange) throws IOException {
-    HttpServletRequest httpServletRequest = null;
-    String value = getMetricsSnapshot(httpServletRequest);
-    sendMessage(
-        exchange,
-        HTTP_OK,
-        String.join("\n", filterNonEmptyRecords(value)),
-        Map.of("Content-Type", Collections.singletonList("text/plain;version=0.0.4")));
+    if (SparkOperatorConf.EnablePrometheusTextBasedFormat.getValue()) {
+      sendMessage(
+          exchange,
+          HTTP_OK,
+          formatMetricsSnapshot(),
+          Map.of("Content-Type", Collections.singletonList("text/plain;version=0.0.4")));
+    } else {
+      HttpServletRequest httpServletRequest = null;
+      String value = getMetricsSnapshot(httpServletRequest);
+      sendMessage(
+          exchange,
+          HTTP_OK,
+          String.join("\n", filterNonEmptyRecords(value)),
+          Map.of("Content-Type", Collections.singletonList("text/plain;version=0.0.4")));
+    }
   }
 
   protected List<String> filterNonEmptyRecords(String metricsSnapshot) {
@@ -82,4 +107,199 @@ protected List<String> filterNonEmptyRecords(String metricsSnapshot) {
     }
     return filteredRecords;
   }
+
+  protected String formatMetricsSnapshot() {
+    Map<String, Gauge> gauges = registry.getGauges();
+    Map<String, Counter> counters = registry.getCounters();
+    Map<String, Histogram> histograms = registry.getHistograms();
+    Map<String, Meter> meters = registry.getMeters();
+    Map<String, Timer> timers = registry.getTimers();
+
+    StringBuilder stringBuilder = new StringBuilder();
+
+    for (Map.Entry<String, Gauge> entry : gauges.entrySet()) {
+      stringBuilder.append(formatGauge(entry.getKey(), entry.getValue()));
+    }
+
+    // Counters
+    for (Map.Entry<String, Counter> entry : counters.entrySet()) {
+      String name = sanitize(entry.getKey()) + "_total";
+      Counter counter = entry.getValue();
+      stringBuilder.append(formatCounter(name, counter));
+    }
+
+    // Histograms
+    for (Map.Entry<String, Histogram> entry : histograms.entrySet()) {
+      stringBuilder.append(formatHistogram(entry.getKey(), entry.getValue()));
+    }
+
+    // Meters
+    for (Map.Entry<String, Meter> entry : meters.entrySet()) {
+      stringBuilder.append(formatMeter(entry.getKey(), entry.getValue()));
+    }
+
+    // Timers (Meter + Histogram in nanoseconds)
+    for (Map.Entry<String, Timer> entry : timers.entrySet()) {
+      stringBuilder.append(formatTimer(entry.getKey(), entry.getValue()));
+    }
+    return stringBuilder.toString();
+  }
+
+  protected String formatGauge(String name, Gauge gauge) {
+    if (gauge != null
+        && gauge.getValue() != null
+        && !EMPTY_RECORD_VALUE.equals(gauge.getValue())
+        && gauge.getValue() instanceof Number) {
+      return "# HELP "
+          + name
+          + " Gauge metric\n"
+          + "# TYPE "
+          + name
+          + " gauge\n"
+          + sanitize(name)
+          + ' '
+          + gauge.getValue()
+          + "\n\n";
+    }
+    return null;
+  }
+
+  protected String formatCounter(String name, Counter counter) {
+    if (counter != null) {
+      return "# HELP "
+          + name
+          + " Counter metric\n"
+          + "# TYPE "
+          + name
+          + " counter\n"
+          + name
+          + " "
+          + counter.getCount()
+          + "\n\n";
+    }
+    return null;
+  }
+
+  protected String formatHistogram(String name, Histogram histogram) {
+    if (histogram != null && histogram.getSnapshot() != null) {
+      StringBuilder stringBuilder = new StringBuilder(300);
+      String baseName = sanitize(name);
+      Snapshot snap = histogram.getSnapshot();
+      long count = histogram.getCount();
+      stringBuilder
+          .append("# HELP ")
+          .append(baseName)
+          .append(" Histogram metric\n# TYPE ")
+          .append(baseName)
+          .append(" histogram\n");
+
+      appendBucket(stringBuilder, baseName, "le=\"0.5\"", count); // approximated
+      appendBucket(stringBuilder, baseName, "le=\"0.75\"", count);
+      appendBucket(stringBuilder, baseName, "le=\"0.95\"", count);
+      appendBucket(stringBuilder, baseName, "le=\"0.98\"", count);
+      appendBucket(stringBuilder, baseName, "le=\"0.99\"", count);
+      appendBucket(stringBuilder, baseName, "le=\"+Inf\"", count);
+      stringBuilder
+          .append(baseName)
+          .append("_count ")
+          .append(count)
+          .append('\n')
+          .append(baseName)
+          .append("_sum ")
+          .append(snap.getMean() * count)
+          .append("\n\n");
+      return stringBuilder.toString();
+    }
+    return null;
+  }
+
+  protected String formatMeter(String name, Meter meter) {
+    if (meter != null) {
+      StringBuilder stringBuilder = new StringBuilder(200);
+      String baseName = sanitize(name);
+      stringBuilder
+          .append("# HELP ")
+          .append(baseName)
+          .append("_total Meter count\n# TYPE ")
+          .append(baseName)
+          .append("_total counter\n")
+          .append(baseName)
+          .append("_total ")
+          .append(meter.getCount())
+          .append("\n\n# TYPE ")
+          .append(baseName)
+          .append("_rate gauge\n")
+          .append(baseName)
+          .append("_rate{interval=\"1m\"} ")
+          .append(meter.getOneMinuteRate())
+          .append('\n')
+          .append(baseName)
+          .append("_rate{interval=\"5m\"} ")
+          .append(meter.getFiveMinuteRate())
+          .append('\n')
+          .append(baseName)
+          .append("_rate{interval=\"15m\"} ")
+          .append(meter.getFifteenMinuteRate())
+          .append("\n\n");
+      return stringBuilder.toString();
+    }
+    return null;
+  }
+
+  protected String formatTimer(String name, Timer timer) {
+    if (timer != null && timer.getSnapshot() != null) {
+      StringBuilder stringBuilder = new StringBuilder(300);
+      String baseName = sanitize(name);
+      Snapshot snap = timer.getSnapshot();
+      long count = timer.getCount();
+      stringBuilder
+          .append("# HELP ")
+          .append(baseName)
+          .append("_duration_seconds Timer histogram\n# TYPE ")
+          .append(baseName)
+          .append("_duration_seconds histogram\n");
+      appendBucket(stringBuilder, baseName + "_duration_seconds", "le=\"0.5\"", count);
+      appendBucket(stringBuilder, baseName + "_duration_seconds", "le=\"0.75\"", count);
+      appendBucket(stringBuilder, baseName + "_duration_seconds", "le=\"0.95\"", count);
+      appendBucket(stringBuilder, baseName + "_duration_seconds", "le=\"+Inf\"", count);
+      stringBuilder
+          .append(baseName)
+          .append("_duration_seconds_count ")
+          .append(count)
+          .append('\n')
+          .append(baseName)
+          .append("_duration_seconds_sum ")
+          .append(nanosToSeconds(snap.getMean() * count))
+          .append("\n\n# TYPE ")
+          .append(baseName)
+          .append("_calls_total counter\n")
+          .append(baseName)
+          .append("_calls_total ")
+          .append(count)
+          .append("\n\n");
+      return stringBuilder.toString();
+    }
+    return null;
+  }
+
+  protected void appendBucket(StringBuilder builder, String baseName, String leLabel, long value) {
+    builder
+        .append(baseName)
+        .append("_bucket{")
+        .append(leLabel)
+        .append("} ")
+        .append(value)
+        .append('\n');
+  }
+
+  protected double nanosToSeconds(double nanos) {
+    return TimeUnit.SECONDS.convert((long) nanos, TimeUnit.NANOSECONDS);
+  }
+
+  protected String sanitize(String name) {
+    if (enableSanitizePrometheusMetricsName) {
+      return name.replaceAll("[^a-zA-Z0-9_:]", "_").toLowerCase();
+    }
+    return name;
+  }
 }