expose prometheus metrics

Author: razvan

docs/modules/spark-k8s/pages/usage-guide/spark-connect.adoc

@@ -26,6 +26,25 @@ include::example$example-spark-connect.yaml[]
 <7> Customize the driver properties in the `server` role. The number of cores here is not related to Kubernetes cores!
 <8> Customize `spark.executor.\*` and `spark.kubernetes.executor.*` in the `executor` role.
 
+== Metrics
+
+The server pod exposes Prometheus metrics at the following endpoints:
+
+* `/metrics/prometheus` for driver instances.
+* `/metrics/executors/prometheus` for executor instances.
+
+To customize the metrics configuration use the `spec.server.configOverrides' like this:
+
+```
+spec:
+  server:
+    configOverrides:
+      metrics.properties:
+        applications.sink.prometheusServlet.path: "/metrics/applications/prometheus"
+```
+
+The example above adds a new endpoint for application metrics.
+
 == Notable Omissions
 
 The following features are not supported by the Stackable Spark operator yet

rust/operator-binary/src/connect/common.rs

@@ -118,3 +118,29 @@ pub fn security_properties(
 
     to_java_properties_string(result.iter()).context(JvmSecurityPropertiesSnafu)
 }
+
+pub fn metrics_properties(
+    config_overrides: Option<&HashMap<String, String>>,
+) -> Result<String, Error> {
+    let mut result: BTreeMap<String, Option<String>> = [
+        (
+            "*.sink.prometheusServlet.class".to_string(),
+            Some("org.apache.spark.metrics.sink.PrometheusServlet".to_string()),
+        ),
+        (
+            "*.sink.prometheusServlet.path".to_string(),
+            Some("/metrics/prometheus".to_string()),
+        ),
+    ]
+    .into();
+
+    if let Some(user_config) = config_overrides {
+        result.extend(
+            user_config
+                .iter()
+                .map(|(k, v)| (k.clone(), Some(v.clone()))),
+        );
+    }
+
+    to_java_properties_string(result.iter()).context(JvmSecurityPropertiesSnafu)
+}

rust/operator-binary/src/connect/executor.rs

@@ -29,9 +29,9 @@ use crate::{
     connect::{common, crd::v1alpha1},
     crd::constants::{
         JVM_SECURITY_PROPERTIES_FILE, LOG4J2_CONFIG_FILE, MAX_SPARK_LOG_FILES_SIZE,
-        POD_TEMPLATE_FILE, SPARK_DEFAULTS_FILE_NAME, VOLUME_MOUNT_NAME_CONFIG,
-        VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_NAME_LOG_CONFIG, VOLUME_MOUNT_PATH_CONFIG,
-        VOLUME_MOUNT_PATH_LOG, VOLUME_MOUNT_PATH_LOG_CONFIG,
+        METRICS_PROPERTIES_FILE, POD_TEMPLATE_FILE, SPARK_DEFAULTS_FILE_NAME,
+        VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_NAME_LOG_CONFIG,
+        VOLUME_MOUNT_PATH_CONFIG, VOLUME_MOUNT_PATH_LOG, VOLUME_MOUNT_PATH_LOG_CONFIG,
     },
     product_logging,
 };
@@ -78,6 +78,9 @@ pub enum Error {
     #[snafu(display("failed build connect executor security properties"))]
     ExecutorJvmSecurityProperties { source: common::Error },
 
+    #[snafu(display("executor metrics properties for spark connect {name}",))]
+    MetricsProperties { source: common::Error, name: String },
+
     #[snafu(display("failed build connect executor config map metadata"))]
     ConfigMapMetadataBuild { source: builder::meta::Error },
 
@@ -331,6 +334,16 @@ pub fn executor_config_map(
     )
     .context(ExecutorJvmSecurityPropertiesSnafu)?;
 
+    let metrics_props = common::metrics_properties(
+        scs.spec
+            .executor
+            .as_ref()
+            .and_then(|s| s.config_overrides.get(METRICS_PROPERTIES_FILE)),
+    )
+    .context(MetricsPropertiesSnafu {
+        name: scs.name_unchecked(),
+    })?;
+
     let mut cm_builder = ConfigMapBuilder::new();
 
     cm_builder
@@ -348,7 +361,8 @@ pub fn executor_config_map(
                 .context(ConfigMapMetadataBuildSnafu)?
                 .build(),
         )
-        .add_data(JVM_SECURITY_PROPERTIES_FILE, jvm_sec_props);
+        .add_data(JVM_SECURITY_PROPERTIES_FILE, jvm_sec_props)
+        .add_data(METRICS_PROPERTIES_FILE, metrics_props);
 
     let role_group_ref = RoleGroupRef {
         cluster: ObjectRef::from_obj(scs),

rust/operator-binary/src/connect/server.rs

@@ -39,7 +39,7 @@ use crate::{
     },
     crd::constants::{
         APP_NAME, JVM_SECURITY_PROPERTIES_FILE, LOG4J2_CONFIG_FILE, MAX_SPARK_LOG_FILES_SIZE,
-        METRICS_PORT, POD_TEMPLATE_FILE, SPARK_DEFAULTS_FILE_NAME, SPARK_UID,
+        METRICS_PROPERTIES_FILE, POD_TEMPLATE_FILE, SPARK_DEFAULTS_FILE_NAME, SPARK_UID,
         VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_NAME_LOG_CONFIG,
         VOLUME_MOUNT_PATH_CONFIG, VOLUME_MOUNT_PATH_LOG, VOLUME_MOUNT_PATH_LOG_CONFIG,
     },
@@ -83,6 +83,9 @@ pub enum Error {
     #[snafu(display("server jvm security properties for spark connect {name}",))]
     ServerJvmSecurityProperties { source: common::Error, name: String },
 
+    #[snafu(display("server metrics properties for spark connect {name}",))]
+    MetricsProperties { source: common::Error, name: String },
+
     #[snafu(display("failed to serialize [{SPARK_DEFAULTS_FILE_NAME}] for the connect server",))]
     SparkDefaultsProperties {
         source: product_config::writer::PropertiesWriterError,
@@ -138,6 +141,16 @@ pub fn server_config_map(
         name: scs.name_unchecked(),
     })?;
 
+    let metrics_props = common::metrics_properties(
+        scs.spec
+            .server
+            .as_ref()
+            .and_then(|s| s.config_overrides.get(METRICS_PROPERTIES_FILE)),
+    )
+    .context(MetricsPropertiesSnafu {
+        name: scs.name_unchecked(),
+    })?;
+
     let mut cm_builder = ConfigMapBuilder::new();
 
     cm_builder
@@ -157,7 +170,8 @@ pub fn server_config_map(
         )
         .add_data(SPARK_DEFAULTS_FILE_NAME, spark_properties)
         .add_data(POD_TEMPLATE_FILE, executor_pod_template_spec)
-        .add_data(JVM_SECURITY_PROPERTIES_FILE, jvm_sec_props);
+        .add_data(JVM_SECURITY_PROPERTIES_FILE, jvm_sec_props)
+        .add_data(METRICS_PROPERTIES_FILE, metrics_props);
 
     let role_group_ref = RoleGroupRef {
         cluster: ObjectRef::from_obj(scs),
@@ -197,6 +211,7 @@ pub fn build_deployment(
             &SparkConnectRole::Server.to_string(),
         ))
         .context(MetadataBuildSnafu)?
+        .with_label(Label::try_from(("prometheus.io/scrape", "true")).context(LabelBuildSnafu)?)
         .build();
 
     let mut pb = PodBuilder::new();
@@ -248,7 +263,6 @@ pub fn build_deployment(
         .args(args)
         .add_container_port("grpc", CONNECT_GRPC_PORT)
         .add_container_port("http", CONNECT_UI_PORT)
-        .add_container_port("metrics", METRICS_PORT.into())
         .add_env_vars(container_env)
         .add_volume_mount(VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_PATH_CONFIG)
         .context(AddVolumeMountSnafu)?
@@ -397,11 +411,6 @@ pub fn build_service(
                     port: CONNECT_UI_PORT,
                     ..ServicePort::default()
                 },
-                ServicePort {
-                    name: Some(String::from("metrics")),
-                    port: METRICS_PORT.into(),
-                    ..ServicePort::default()
-                },
             ]),
             selector: Some(selector),
             publish_not_ready_addresses,
@@ -512,6 +521,18 @@ pub fn server_properties(
             "spark.driver.extraClassPath".to_string(),
             Some(format!("/stackable/spark/extra-jars/*:/stackable/spark/connect/spark-connect_2.12-{spark_version}.jar")),
         ),
+        (
+            "spark.metrics.conf".to_string(),
+            Some(format!("{VOLUME_MOUNT_PATH_CONFIG}/{METRICS_PROPERTIES_FILE}")),
+        ),
+        // This enables the "/metrics/executors/prometheus" endpoint on the server pod.
+        // The driver collects metrics from the executors and makes them available here.
+        // The "/metrics/prometheus" endpoint delievers the driver metrics.
+        (
+            "spark.ui.prometheus.enabled".to_string(),
+            Some("true".to_string()),
+        ),
+
     ]
     .into();
 
@@ -556,7 +577,7 @@ fn probe() -> Probe {
         http_get: Some(HTTPGetAction {
             port: IntOrString::Int(CONNECT_UI_PORT),
             scheme: Some("HTTP".to_string()),
-            path: Some("/metrics".to_string()),
+            path: Some("/metrics/prometheus".to_string()),
             ..Default::default()
         }),
         failure_threshold: Some(10),

rust/operator-binary/src/crd/constants.rs

@@ -33,6 +33,7 @@ pub const VOLUME_MOUNT_PATH_LOG: &str = "/stackable/log";
 pub const LOG4J2_CONFIG_FILE: &str = "log4j2.properties";
 
 pub const JVM_SECURITY_PROPERTIES_FILE: &str = "security.properties";
+pub const METRICS_PROPERTIES_FILE: &str = "metrics.properties";
 
 pub const ACCESS_KEY_ID: &str = "accessKey";
 pub const SECRET_ACCESS_KEY: &str = "secretKey";

tests/test-definition.yaml

@@ -118,7 +118,6 @@ tests:
     dimensions:
       - spark-connect
       - spark-connect-client
-      - s3-use-tls
       - openshift
 
 suites: