[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 256092

# HELP jvm_bufferpool_direct_count Gauge metric
# TYPE jvm_bufferpool_direct_count gauge
jvm_bufferpool_direct_count 44

# HELP kubernetes_client_2xx_total Meter count
# TYPE kubernetes_client_2xx_total counter
kubernetes_client_2xx_total 130

# HELP kubernetes_client_http_response_latency Histogram metric
# TYPE kubernetes_client_http_response_latency histogram
kubernetes_client_http_response_latency_seconds_bucket{le="0.5"} 0.000104422
kubernetes_client_http_response_latency_seconds_bucket{le="0.75"} 0.000128422
kubernetes_client_http_response_latency_seconds_bucket{le="0.95"} 0.000139544
kubernetes_client_http_response_latency_seconds_bucket{le="0.98"} 0.000169124
kubernetes_client_http_response_latency_seconds_bucket{le="0.99"} 0.066452639
kubernetes_client_http_response_latency_seconds_count 2000
kubernetes_client_http_response_latency_seconds_sum 0.456670434
```

### 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

@@ -29,21 +29,38 @@
 import java.util.Map;
 import java.util.Properties;
 
+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.commons.lang3.StringUtils;
 
+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,256 @@ 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()) {
+      appendIfNotEmpty(stringBuilder, formatGauge(entry.getKey(), entry.getValue()));
+    }
+
+    // Counters
+    for (Map.Entry<String, Counter> entry : counters.entrySet()) {
+      String name = sanitize(entry.getKey()) + "_total";
+      Counter counter = entry.getValue();
+      appendIfNotEmpty(stringBuilder, formatCounter(name, counter));
+    }
+
+    // Histograms
+    for (Map.Entry<String, Histogram> entry : histograms.entrySet()) {
+      appendIfNotEmpty(stringBuilder, formatHistogram(entry.getKey(), entry.getValue()));
+    }
+
+    // Meters
+    for (Map.Entry<String, Meter> entry : meters.entrySet()) {
+      appendIfNotEmpty(stringBuilder, formatMeter(entry.getKey(), entry.getValue()));
+    }
+
+    // Timers (Meter + Histogram in nanoseconds)
+    for (Map.Entry<String, Timer> entry : timers.entrySet()) {
+      appendIfNotEmpty(stringBuilder, formatTimer(entry.getKey(), entry.getValue()));
+    }
+    return stringBuilder.toString();
+  }
+
+  protected void appendIfNotEmpty(StringBuilder stringBuilder, String value) {
+    if (StringUtils.isNotEmpty(value)) {
+      stringBuilder.append(value);
+    }
+  }
+
+  protected String formatGauge(String name, Gauge gauge) {
+    if (gauge != null
+        && gauge.getValue() != null
+        && !EMPTY_RECORD_VALUE.equals(gauge.getValue())
+        && gauge.getValue() instanceof Number) {
+      String formattedName = sanitize(name);
+      return "# HELP "
+          + formattedName
+          + " Gauge metric\n"
+          + "# TYPE "
+          + formattedName
+          + " gauge\n"
+          + sanitize(formattedName)
+          + ' '
+          + 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");
+      boolean isNanosHistogram = baseName.contains("nanos");
+      if (isNanosHistogram) {
+        baseName = nanosMetricsNameToSeconds(baseName);
+      }
+      appendBucket(
+          stringBuilder,
+          baseName,
+          "le=\"0.5\"",
+          isNanosHistogram ? nanosToSeconds(snap.getMedian()) : snap.getMean());
+      appendBucket(
+          stringBuilder,
+          baseName,
+          "le=\"0.75\"",
+          isNanosHistogram ? nanosToSeconds(snap.get75thPercentile()) : snap.get75thPercentile());
+      appendBucket(
+          stringBuilder,
+          baseName,
+          "le=\"0.95\"",
+          isNanosHistogram ? nanosToSeconds(snap.get95thPercentile()) : snap.get95thPercentile());
+      appendBucket(
+          stringBuilder,
+          baseName,
+          "le=\"0.98\"",
+          isNanosHistogram ? nanosToSeconds(snap.get98thPercentile()) : snap.get98thPercentile());
+      appendBucket(
+          stringBuilder,
+          baseName,
+          "le=\"0.99\"",
+          isNanosHistogram ? nanosToSeconds(snap.get99thPercentile()) : snap.get99thPercentile());
+      double sum =
+          isNanosHistogram ? nanosToSeconds(snap.getMean() * count) : snap.getMean() * count;
+      stringBuilder
+          .append(baseName)
+          .append("_count ")
+          .append(count)
+          .append('\n')
+          .append(baseName)
+          .append("_sum ")
+          .append(sum)
+          .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\"",
+          nanosToSeconds(snap.getMedian()));
+      appendBucket(
+          stringBuilder,
+          baseName + "_duration_seconds",
+          "le=\"0.75\"",
+          nanosToSeconds(snap.get75thPercentile()));
+      appendBucket(
+          stringBuilder,
+          baseName + "_duration_seconds",
+          "le=\"0.95\"",
+          nanosToSeconds(snap.get95thPercentile()));
+      appendBucket(
+          stringBuilder,
+          baseName + "_duration_seconds",
+          "le=\"0.98\"",
+          nanosToSeconds(snap.get98thPercentile()));
+      appendBucket(
+          stringBuilder,
+          baseName + "_duration_seconds",
+          "le=\"0.99\"",
+          nanosToSeconds(snap.get99thPercentile()));
+      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, double value) {
+    builder
+        .append(baseName)
+        .append("_bucket{")
+        .append(leLabel)
+        .append("} ")
+        .append(value)
+        .append('\n');
+  }
+
+  protected double nanosToSeconds(double nanos) {
+    return nanos / 1_000_000_000.0;
+  }
+
+  protected String sanitize(String name) {
+    if (enableSanitizePrometheusMetricsName) {
+      return name.replaceAll("[^a-zA-Z0-9_:]", "_").toLowerCase();
+    }
+    return name;
+  }
+
+  protected String nanosMetricsNameToSeconds(String name) {
+    return name.replaceAll("_nanos", "_seconds");
+  }
 }