dongjoon-hyun commented on code in PR #794: URL: https://github.com/apache/spark-kubernetes-operator/pull/794#discussion_r4036153420
########## examples/cluster-with-jmx-exporter.yaml: ########## @@ -0,0 +1,78 @@ +# 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. +apiVersion: v1 +kind: ConfigMap +metadata: + name: jmx-exporter-config +data: + jmx-exporter-config.yaml: | + rules: + - pattern: 'java.lang<type=Memory><HeapMemoryUsage>(\w+)' + name: jvm_memory_heap_$1 + type: GAUGE + - pattern: 'java.lang<type=GarbageCollector, name=(\w+)><>CollectionCount' + name: jvm_gc_collection_count + labels: + gc: "$1" + type: COUNTER + - pattern: 'metrics<name=(\S+)><>Value' Review Comment: This rule still never matches, and my previous comment was incomplete: `JmxSink` alone does not fix it. Spark bundles Dropwizard `metrics-jmx` 4.2.37, whose `DefaultObjectNameFactory` puts both `name` and `type` into the ObjectName, e.g. `metrics:name=worker.coresFree,type=gauges`. jmx_exporter renders that as `metrics<name=worker.coresFree, type=gauges><>Value` and wraps the rule as `^.*(?:pattern).*$`, so `(\S+)>` cannot get past the `, `. I checked the regex against both property orders on a local JVM and it is false for both. With a non-empty `rules` list unmatched beans are dropped, so the scrape has no `spark_worker_*` series at all. Suggested: ```yaml - pattern: 'metrics<name=worker\.(\w+), type=gauges><>Value' name: spark_worker_$1 type: GAUGE ``` A plain `(\S+), type=gauges` would produce `spark_worker_worker_coresFree`, because `WorkerSource.sourceName` is already `worker`. The sentence in `docs/operations.md` ("Without this sink, the exporter only exposes JVM metrics") is right, but please make sure the example actually delivers the other half. ########## examples/cluster-with-jmx-exporter.yaml: ########## @@ -0,0 +1,78 @@ +# 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. +apiVersion: v1 +kind: ConfigMap +metadata: + name: jmx-exporter-config +data: + jmx-exporter-config.yaml: | + rules: + - pattern: 'java.lang<type=Memory><HeapMemoryUsage>(\w+)' + name: jvm_memory_heap_$1 + type: GAUGE + - pattern: 'java.lang<type=GarbageCollector, name=(\w+)><>CollectionCount' Review Comment: `name=(\w+)` does not match the default collectors. On a JVM with `-XX:+UseG1GC` the beans are `G1 Young Generation`, `G1 Concurrent GC`, and `G1 Old Generation`; the spaces make this rule fail, and it only matches SerialGC names like `Copy`. G1 is the default on the JDK 17/21 `apache/spark` images once the pod has 2+ CPUs and about 1.8 GB, so `jvm_gc_collection_count` is never emitted. Either use `name=([^>]+)`, or drop the two JVM rules altogether: the javaagent already exports `jvm_memory_*` and `jvm_gc_*` from its built-in collectors. ########## examples/cluster-with-jmx-exporter.yaml: ########## @@ -0,0 +1,78 @@ +# 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. +apiVersion: v1 +kind: ConfigMap +metadata: + name: jmx-exporter-config +data: + jmx-exporter-config.yaml: | + rules: + - pattern: 'java.lang<type=Memory><HeapMemoryUsage>(\w+)' + name: jvm_memory_heap_$1 + type: GAUGE + - pattern: 'java.lang<type=GarbageCollector, name=(\w+)><>CollectionCount' + name: jvm_gc_collection_count + labels: + gc: "$1" + type: COUNTER + - pattern: 'metrics<name=(\S+)><>Value' + name: spark_worker_$1 + type: GAUGE +--- +apiVersion: spark.apache.org/v1 +kind: SparkCluster +metadata: + name: cluster-with-jmx-exporter +spec: + runtimeVersions: + sparkVersion: "4.2.0" + clusterTolerations: + instanceConfig: + initWorkers: 1 + minWorkers: 1 + maxWorkers: 1 + workerSpec: + networkPolicy: + metricsPort: 9404 + # Allow scraper pods in "monitoring", in addition to the default cluster/driver peers. + metricsIngress: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: "monitoring" + statefulSetSpec: + template: + spec: + containers: + - name: worker + # Requires a custom image that has jmx_prometheus_javaagent pre-baked at this path + # (the stock apache/spark image does not include it). + env: + - name: SPARK_DAEMON_JAVA_OPTS + value: "-javaagent:/opt/jmx_exporter/jmx_prometheus_javaagent.jar=9404:/etc/metrics/jmx-exporter-config.yaml" + ports: + - name: jmx-metrics + containerPort: 9404 + volumeMounts: + - name: jmx-exporter-config + mountPath: /etc/metrics + readOnly: true + volumes: + - name: jmx-exporter-config + configMap: + name: jmx-exporter-config + sparkConf: + spark.metrics.conf.*.sink.jmx.class: "org.apache.spark.metrics.sink.JmxSink" Review Comment: Two small things on the example config: - `sparkConf` entries are forwarded as `-D` to both `SPARK_MASTER_OPTS` and `SPARK_WORKER_OPTS`, so the `*` instance also starts `JmxSink` for the master's `master` and `applications` metrics systems, where no agent reads the MBeans. `spark.metrics.conf.worker.sink.jmx.class` is enough. - Rules only filter output. Without `includeObjectNames` the agent queries every MBean on every scrape and then discards almost all of it. Something like `includeObjectNames: ["metrics:type=gauges,*"]` (plus the `java.lang` entries if you keep the JVM rules) keeps the scrape cheap. Also, `docs/spark_custom_resources.md` already documents a `JmxSink` recipe via a mounted `metrics.properties`. A cross-link from the new section would avoid having two unconnected recipes. ########## spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterResourceSpec.java: ########## @@ -467,35 +469,53 @@ private static Optional<PodDisruptionBudget> buildPodDisruptionBudget( * resource and does not carry the cluster label, yet the driver must reach the executors' block * manager to fetch task results larger than {@code spark.task.maxDirectResultSize}. * + * <p>If the worker network policy is configured with metrics ingress peers, a separate rule + * admits those peers on the configured port only. This should be a dedicated exporter port, not + * the worker web UI port; the operator does not verify this. + * * @param clusterName The name of the SparkCluster. * @param namespace The namespace of the SparkApplication. + * @param workerSpec The WorkerSpec, used to look up the optional metrics port and its peers. * @return A NetworkPolicy object. */ - private NetworkPolicy buildWorkerNetworkPolicy(String clusterName, String namespace) { - return new NetworkPolicyBuilder() - .withNewMetadata() - .withName(clusterName + "-worker") - .withNamespace(namespace) - .addToLabels(LABEL_SPARK_CLUSTER_NAME, clusterName) - .endMetadata() - .withNewSpec() - .withNewPodSelector() - .addToMatchLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE) - .addToMatchLabels(LABEL_SPARK_CLUSTER_NAME, clusterName) - .endPodSelector() + private NetworkPolicy buildWorkerNetworkPolicy( + String clusterName, String namespace, WorkerSpec workerSpec) { + var builder = + new NetworkPolicyBuilder() + .withNewMetadata() + .withName(clusterName + "-worker") + .withNamespace(namespace) + .addToLabels(LABEL_SPARK_CLUSTER_NAME, clusterName) + .endMetadata() + .withNewSpec() + .withNewPodSelector() + .addToMatchLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_WORKER_VALUE) + .addToMatchLabels(LABEL_SPARK_CLUSTER_NAME, clusterName) + .endPodSelector() + .addNewIngress() + .addNewFrom() + .withNewPodSelector() + .addToMatchLabels(LABEL_SPARK_CLUSTER_NAME, clusterName) + .endPodSelector() + .endFrom() + .addNewFrom() + .withNewPodSelector() + .addToMatchLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_DRIVER_VALUE) + .endPodSelector() + .endFrom() + .endIngress(); + WorkerNetworkPolicySpec networkPolicy = workerSpec.getNetworkPolicy(); + if (networkPolicy == null || networkPolicy.getMetricsIngress().isEmpty()) { Review Comment: Optional. I agree nulls cannot arrive through the API server: an old CRD or a `v1beta1` write prunes the whole block, and an explicit `null` is dropped and then fails `required`. So I am not asking for the null checks back. Two notes though: - This guard is security-relevant, not cosmetic: an ingress rule with an empty or absent `from` admits **all** sources, and fabric8 omits an empty list when serializing. A one-line comment saying so would stop a future cleanup from removing it on the grounds that the schema guarantees a non-empty list. - For Java callers, `new IntOrString((Integer) null)` is accepted by fabric8 and yields `ports: [{protocol: TCP}]`, i.e. every TCP port for the listed peers. That is fail-open, and nothing pins the `required` list today, so it would become reachable if `@Required` were ever dropped. ########## spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/WorkerNetworkPolicySpec.java: ########## @@ -0,0 +1,62 @@ +/* + * 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.k8s.operator.spec; + +import java.util.List; + +import com.fasterxml.jackson.annotation.JsonInclude; +import io.fabric8.generator.annotation.Max; +import io.fabric8.generator.annotation.Min; +import io.fabric8.generator.annotation.Required; +import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicyPeer; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * Network policy for the Spark workers. + * + * @since 1.1.0 + */ +@Data +@NoArgsConstructor +@AllArgsConstructor +@Builder +@JsonInclude(JsonInclude.Include.NON_NULL) +public class WorkerNetworkPolicySpec { + /** + * Port of a metrics endpoint (e.g. a JMX-to-Prometheus exporter agent) running in the worker + * container. This should be a dedicated exporter port, not the worker web UI port; the operator + * does not verify this. The caller is responsible for making the container listen on this port. + */ + @Required + @Min(1) + @Max(65535) + protected Integer metricsPort; + + /** + * Sources allowed to scrape {@link #metricsPort} on the worker pods, in addition to the sources + * the generated worker NetworkPolicy admits by default. When this list is empty, no metrics + * ingress rule is generated. + */ + @Required + protected List<NetworkPolicyPeer> metricsIngress; Review Comment: `{metricsPort: 9404, metricsIngress: []}` still passes the schema and silently does nothing, which is the same silent no-op we removed for partial blocks. In the Helm chart an empty list means "deny" around a metrics rule that always exists; here the block's only purpose is to add a rule, so an empty list has no meaning. `@Size(min = 1)` (it is in the same `generator-annotations` jar) emits `minItems: 1`, so the block is either effective or rejected at apply time. Please keep the Java `isEmpty()` guard regardless, see the comment there. ########## docs/operations.md: ########## @@ -174,6 +174,55 @@ Note that this requires a CNI plugin that enforces NetworkPolicy; on clusters wi a plugin the policy is silently ignored. Egress traffic of the operator (Kubernetes API server, DNS) is not restricted by this policy. +## Exposing SparkCluster Worker Metrics + +Every `SparkCluster` gets a generated worker `NetworkPolicy` that only admits ingress from pods +carrying the cluster label or the driver-role label, so a Prometheus scraper is locked out by +default. Opening the worker web UI port (`8081` by default) is not a safe fix: Spark's built-in +`PrometheusServlet` metrics endpoint is served by the same embedded HTTP server as the web UI, so +admitting that port to any source would expose the whole UI, not just metrics. + +Use the [Prometheus JMX Exporter](https://github.com/prometheus/jmx_exporter) +(`jmx_prometheus_javaagent`) to serve metrics on a dedicated HTTP port. See +[examples/cluster-with-jmx-exporter.yaml](../examples/cluster-with-jmx-exporter.yaml) for the full +`ConfigMap` and `SparkCluster` configuration. Replace the example's image with a custom Spark image +containing the exporter jar at `/opt/jmx_exporter/jmx_prometheus_javaagent.jar`; the stock +`apache/spark` image does not include it. + +The example mounts the exporter rules and attaches the agent through `SPARK_DAEMON_JAVA_OPTS`. +The operator sets `SPARK_WORKER_OPTS` itself, so a value there would be overwritten. It also sets +`spark.metrics.conf.*.sink.jmx.class` to `org.apache.spark.metrics.sink.JmxSink` in `sparkConf` to +register Spark metrics as MBeans. Without this sink, the exporter only exposes JVM metrics. + +Set `workerSpec.networkPolicy.metricsPort` to the exporter port and list the scraper's peers under +`metricsIngress`. The generated policy adds ingress on that port for those peers, alongside the +existing cluster/driver label allow-list. This mirrors +`operatorDeployment.networkPolicy.metricsIngress` +([above](#restricting-network-access-to-the-operator)) and accepts the same +[`NetworkPolicyPeer`](https://kubernetes.io/docs/reference/kubernetes-api/policy-resources/network-policy-v1/#NetworkPolicyPeer) +entries: + +```yaml +spec: + workerSpec: + networkPolicy: + metricsPort: 9404 + metricsIngress: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: "monitoring" +``` + +The `networkPolicy` block is optional. When present, the API server requires both fields and a +port between 1 and 65535. An empty `metricsIngress` list adds no rule. Choose a dedicated exporter Review Comment: Please add one sentence that the policy is generated when the cluster is created. The `NetworkPolicy` is applied only in `ClusterInitStep` (state `Submitted`) and nothing is reconciled once the cluster is `RunningHealthy`, so a `kubectl edit` that adds or removes a peer is accepted by the API server and then never takes effect. That is existing behavior for every `SparkCluster` field, but scraper peers are the one knob people will edit on a long-running cluster. Changing them means recreating the `SparkCluster`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
