[SPARK-57934] Support opt-in NetworkPolicy for operator pod in Helm chart

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

This PR aims to support an opt-in `NetworkPolicy` for the operator pod in the Helm chart.

When `operatorDeployment.networkPolicy.enable` is `true` (default: `false`), all ingress to the
operator pod is denied except the health probe port (open to any source, for kubelet probes) and
the metrics port, which is reachable only from the peers listed in
`operatorDeployment.networkPolicy.metricsIngress`.

```yaml
operatorDeployment:
  networkPolicy:
    enable: true
    metricsIngress:
      - namespaceSelector:
          matchLabels:
            kubernetes.io/metadata.name: "monitoring"
```

### Why are the changes needed?

The operator serves unauthenticated plain-HTTP probe and metrics endpoints, and the chart
provided no way to restrict in-cluster access to them.

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

No, because this is disabled by default.

### How was this patch tested?

Pass the CIs.

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

Generated-by: Claude Fable 5

Closes #742 from dongjoon-hyun/SPARK-57934.

Authored-by: Dongjoon Hyun <dongjoon@apache.org>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: dongjoon-hyun

.github/workflows/build_and_test.yml

@@ -233,6 +233,7 @@ jobs:
           - "1.36.0"
         test-group:
           - configmap-metadata
+          - network-policy
     steps:
       - name: Checkout repository
         uses: actions/checkout@v6

build-tools/helm/spark-kubernetes-operator/templates/network-policy.yaml

@@ -0,0 +1,44 @@
+# 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.
+
+{{- if .Values.operatorDeployment.networkPolicy.enable }}
+---
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: {{ include "spark-operator.name" . }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    {{- include "spark-operator.commonLabels" . | nindent 4 }}
+spec:
+  podSelector:
+    matchLabels:
+      {{- include "spark-operator.deploymentSelectorLabels" . | nindent 6 }}
+  policyTypes:
+    - Ingress
+  ingress:
+    # Health probe port must stay reachable by the kubelet, whose traffic
+    # originates from the node and cannot be selected by pod/namespace selectors.
+    - ports:
+        - port: {{ include "spark-operator.probePort" . }}
+          protocol: TCP
+    {{- with .Values.operatorDeployment.networkPolicy.metricsIngress }}
+    - from:
+        {{- toYaml . | nindent 8 }}
+      ports:
+        - port: {{ include "spark-operator.metricsPort" $ }}
+          protocol: TCP
+    {{- end }}
+{{- end }}

build-tools/helm/spark-kubernetes-operator/templates/tests/test-network-policy.yaml

@@ -0,0 +1,85 @@
+# 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.
+
+{{- if .Values.operatorDeployment.networkPolicy.enable }}
+apiVersion: v1
+kind: Pod
+metadata:
+  name: "{{ include "spark-operator.name" . }}-test-network-policy"
+  namespace: {{ .Release.Namespace }}
+  labels:
+    {{- include "spark-operator.commonLabels" . | nindent 4 }}
+  annotations:
+    "helm.sh/hook": test
+    "helm.sh/hook-delete-policy": before-hook-creation
+spec:
+  containers:
+    - name: kubectl
+      image: bitnamisecure/kubectl:latest
+      command: ['bash', '-c']
+      args:
+        - |
+          set -e
+          echo "Testing operator NetworkPolicy..."
+
+          NP_JSON=$(kubectl get networkpolicy {{ include "spark-operator.name" . }} -n {{ .Release.Namespace }} -o json)
+
+          echo "Validating policyTypes..."
+          POLICY_TYPES=$(echo "$NP_JSON" | jq -r '.spec.policyTypes | join(",")')
+          if [ "$POLICY_TYPES" != "Ingress" ]; then
+            echo "ERROR: Expected policyTypes [Ingress], Got: $POLICY_TYPES"
+            exit 1
+          fi
+
+          echo "Validating podSelector..."
+          SELECTOR_NAME=$(echo "$NP_JSON" | jq -r '.spec.podSelector.matchLabels["app.kubernetes.io/name"]')
+          SELECTOR_COMPONENT=$(echo "$NP_JSON" | jq -r '.spec.podSelector.matchLabels["app.kubernetes.io/component"]')
+          if [ "$SELECTOR_NAME" != "{{ include "spark-operator.name" . }}" ] || [ "$SELECTOR_COMPONENT" != "operator-deployment" ]; then
+            echo "ERROR: podSelector does not match operator deployment labels. Got: $SELECTOR_NAME/$SELECTOR_COMPONENT"
+            exit 1
+          fi
+
+          echo "Validating probe port ingress rule..."
+          PROBE_PORT=$(echo "$NP_JSON" | jq -r '[.spec.ingress[] | select(has("from") | not)][0].ports[0].port')
+          if [ "$PROBE_PORT" != "{{ include "spark-operator.probePort" . }}" ]; then
+            echo "ERROR: Expected probe port {{ include "spark-operator.probePort" . }} open to all sources, Got: $PROBE_PORT"
+            exit 1
+          fi
+
+          {{- if .Values.operatorDeployment.networkPolicy.metricsIngress }}
+          echo "Validating metrics port ingress rule..."
+          METRICS_PORT=$(echo "$NP_JSON" | jq -r '[.spec.ingress[] | select(has("from"))][0].ports[0].port')
+          if [ "$METRICS_PORT" != "{{ include "spark-operator.metricsPort" . }}" ]; then
+            echo "ERROR: Expected metrics port {{ include "spark-operator.metricsPort" . }} in the restricted rule, Got: $METRICS_PORT"
+            exit 1
+          fi
+          FROM_MATCHES=$(echo "$NP_JSON" | jq --argjson expected '{{ toJson .Values.operatorDeployment.networkPolicy.metricsIngress }}' '[.spec.ingress[] | select(has("from"))][0].from == $expected')
+          if [ "$FROM_MATCHES" != "true" ]; then
+            echo "ERROR: Metrics ingress sources do not match values. Got: $(echo "$NP_JSON" | jq -c '[.spec.ingress[] | select(has("from"))][0].from')"
+            exit 1
+          fi
+          {{- else }}
+          echo "Validating that the metrics port is denied..."
+          RESTRICTED_RULES=$(echo "$NP_JSON" | jq '[.spec.ingress[] | select(has("from"))] | length')
+          if [ "$RESTRICTED_RULES" != "0" ]; then
+            echo "ERROR: Expected no ingress rule with sources when metricsIngress is empty, Got: $RESTRICTED_RULES"
+            exit 1
+          fi
+          {{- end }}
+
+          echo "All NetworkPolicy tests passed!"
+  restartPolicy: Never
+  serviceAccountName: {{ .Values.operatorRbac.serviceAccount.name }}
+{{- end }}

build-tools/helm/spark-kubernetes-operator/values.schema.json

@@ -391,6 +391,27 @@
               "additionalProperties": true
             }
           }
+        },
+        "networkPolicy": {
+          "type": "object",
+          "description": "NetworkPolicy for the operator pod",
+          "required": [
+            "enable"
+          ],
+          "properties": {
+            "enable": {
+              "type": "boolean",
+              "description": "Whether to create a NetworkPolicy for the operator pod"
+            },
+            "metricsIngress": {
+              "type": ["array", "null"],
+              "description": "List of NetworkPolicyPeer(s) allowed to reach the metrics port; when empty, ingress to the metrics port is denied",
+              "items": {
+                "type": "object",
+                "additionalProperties": true
+              }
+            }
+          }
         }
       }
     },

build-tools/helm/spark-kubernetes-operator/values.yaml

@@ -87,6 +87,19 @@ operatorDeployment:
     securityContext: { }
     dnsPolicy:
     dnsConfig:
+  # When enabled, a NetworkPolicy is created for the operator pod that allows ingress only to
+  # the health probe port (from any source, for kubelet probes) and to the metrics port from
+  # the sources listed in {networkPolicy.metricsIngress}. All other ingress traffic to the
+  # operator pod is denied. Note that this requires a CNI plugin with NetworkPolicy support,
+  # and that egress traffic is not restricted.
+  networkPolicy:
+    enable: false
+    # List of NetworkPolicyPeer(s) allowed to reach the metrics port, e.g. the Prometheus
+    # scraper. When empty, ingress to the metrics port is denied.
+    metricsIngress: [ ]
+    #  - namespaceSelector:
+    #      matchLabels:
+    #        kubernetes.io/metadata.name: "monitoring"
 
 operatorRbac:
   serviceAccount:

docs/operations.md

@@ -132,6 +132,37 @@ __Notice__: The pod resources should be set as your workload in different enviro
 achieve a matched K8s pod QoS. See
 also [Pod Quality of Service Classes](https://kubernetes.io/docs/concepts/workloads/pods/pod-qos/#quality-of-service-classes).
 
+## Restricting Network Access to the Operator
+
+The operator serves two plain-HTTP endpoints without authentication: the health probes
+(port 19091 by default) and the Prometheus metrics (port 19090 by default). To restrict
+in-cluster access to them, the chart can create a
+[NetworkPolicy](https://kubernetes.io/docs/concepts/services-networking/network-policies/)
+for the operator pod:
+
+```yaml
+operatorDeployment:
+  networkPolicy:
+    enable: true
+    metricsIngress:
+      - namespaceSelector:
+          matchLabels:
+            kubernetes.io/metadata.name: "monitoring"
+```
+
+When enabled, all ingress traffic to the operator pod is denied except:
+
+- the health probe port, reachable from any source — kubelet probe traffic originates from
+  the node and cannot be matched by pod or namespace selectors, and
+- the metrics port, reachable only from the `NetworkPolicyPeer`(s) listed in
+  `metricsIngress`. When the list is empty, ingress to the metrics port is denied, so make
+  sure your metrics scraper is listed before enabling this in an environment that collects
+  operator metrics.
+
+Note that this requires a CNI plugin that enforces NetworkPolicy; on clusters without such
+a plugin the policy is silently ignored. Egress traffic of the operator (Kubernetes API
+server, DNS) is not restricted by this policy.
+
 ## Operator Health(Liveness) Probe with Sentinel Resource
 
 Learning

tests/e2e/helm/helm-test-values/network-policy/values.yaml

@@ -0,0 +1,28 @@
+# 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.
+
+# Test values for validating the operator NetworkPolicy feature
+#
+# Usage:
+#   helm install spark-operator . -f tests/e2e/helm/helm-test-values/network-policy/values.yaml
+#   helm test spark-operator
+
+operatorDeployment:
+  networkPolicy:
+    enable: true
+    metricsIngress:
+      - namespaceSelector:
+          matchLabels:
+            kubernetes.io/metadata.name: "monitoring"