You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/07/23 06:05:22 UTC
(spark-kubernetes-operator) branch main updated: [SPARK-58160] Add Gateway API HTTPRoute/GRPCRoute support for driver endpoints
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 7466027 [SPARK-58160] Add Gateway API HTTPRoute/GRPCRoute support for driver endpoints
7466027 is described below
commit 746602722928ce7963d4de37997ea82852682d6b
Author: Chandni Singh <cs...@cloudera.com>
AuthorDate: Wed Jul 22 23:05:15 2026 -0700
[SPARK-58160] Add Gateway API HTTPRoute/GRPCRoute support for driver endpoints
### What changes were proposed in this pull request?
Adds opt-in native support for Kubernetes Gateway API v1 routes on SparkApplication, parallel to the existing driverServiceIngressList. Two new optional lists on ApplicationSpec — driverHttpRouteList and driverGrpcRouteList — each pair a Service (fronting the driver pod) with an HTTPRoute or GRPCRoute. When the user omits the route rules, a default rule targeting the Service's first port is emitted.
No behavior change when unset. Not tied to a specific Gateway controller — Istio, Envoy Gateway, and Cilium are all GA on HTTPRoute/GRPCRoute v1.
Includes:
- DriverHttpRouteSpec / DriverGrpcRouteSpec CRD types
- DriverGatewayRouteUtils in spark-submission-worker
- Wiring through SparkAppSubmissionWorker / SparkAppResourceSpec
- Helm RBAC for gateway.networking.k8s.io httproutes/grpcroutes
- Regenerated v1 CRD
- Unit tests for default and user-provided rule shapes
### Why are the changes needed?
Kubernetes Gateway API v1 is the successor to Ingress and is GA across major implementations (Istio, Envoy Gateway, Cilium, etc.). Today SparkApplication only supports Ingress via driverServiceIngressList; users on Gateway-based clusters have to create HTTPRoute / GRPCRoute objects out-of-band, which don't share the driver pod's lifecycle and can leak on cleanup. Native support lets the operator emit and GC these routes alongside the driver, on par with the existing Ingress path, and [...]
### Does this PR introduce _any_ user-facing change?
Yes. Adds two optional fields on SparkApplication.spec:
- driverHttpRouteList — emits a Service + Gateway API v1 HTTPRoute per entry for the driver.
- driverGrpcRouteList — emits a Service + Gateway API v1 GRPCRoute per entry.
Both are opt-in and parallel to the existing driverServiceIngressList. The emitted resources are owned by the driver pod, so they GC automatically. If a route list is set, httpRouteSpec.parentRefs / grpcRouteSpec.parentRefs and the service/route metadata names are required — otherwise the app moves to Failed with a message naming the offending field. No behavior change when both fields are unset.
### How was this patch tested?
Unit tests and on an internal dev cluster
### Was this patch authored or co-authored using generative AI tooling?
Yes.
Generated-by: Claude Code (Claude Opus 4.7)
Closes #763 from otterc/http_grpc_routes.
Authored-by: Chandni Singh <cs...@cloudera.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../sparkapplications.spark.apache.org-v1.yaml | 1231 ++++++++++++++++++++
.../templates/operator-rbac.yaml | 13 +
docs/operations.md | 11 +
docs/spark_custom_resources.md | 47 +
gradle/libs.versions.toml | 1 +
spark-operator-api/build.gradle | 1 +
.../spark/k8s/operator/spec/ApplicationSpec.java | 3 +
...plicationSpec.java => DriverGrpcRouteSpec.java} | 43 +-
...plicationSpec.java => DriverHttpRouteSpec.java} | 43 +-
.../reconciler/reconcilesteps/AppValidateStep.java | 96 +-
.../reconcilesteps/AppValidateStepTest.java | 268 +++++
spark-submission-worker/build.gradle | 1 +
.../spark/k8s/operator/SparkAppResourceSpec.java | 45 +
.../k8s/operator/SparkAppSubmissionWorker.java | 15 +-
.../operator/utils/DriverGatewayRouteUtils.java | 155 +++
.../k8s/operator/SparkAppResourceSpecTest.java | 4 +-
.../utils/DriverGatewayRouteUtilsTest.java | 269 +++++
17 files changed, 2182 insertions(+), 64 deletions(-)
diff --git a/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml b/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml
index 22795a8..1579369 100644
--- a/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml
+++ b/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml
@@ -8811,6 +8811,1237 @@ spec:
items:
type: string
type: array
+ driverGrpcRouteList:
+ items:
+ properties:
+ grpcRouteMetadata:
+ properties:
+ annotations:
+ additionalProperties:
+ type: string
+ type: object
+ creationTimestamp:
+ type: string
+ deletionGracePeriodSeconds:
+ type: integer
+ deletionTimestamp:
+ type: string
+ finalizers:
+ items:
+ type: string
+ type: array
+ generateName:
+ type: string
+ generation:
+ type: integer
+ labels:
+ additionalProperties:
+ type: string
+ type: object
+ managedFields:
+ items:
+ properties:
+ apiVersion:
+ type: string
+ fieldsType:
+ type: string
+ fieldsV1:
+ type: object
+ manager:
+ type: string
+ operation:
+ type: string
+ subresource:
+ type: string
+ time:
+ type: string
+ type: object
+ type: array
+ name:
+ type: string
+ namespace:
+ type: string
+ ownerReferences:
+ items:
+ properties:
+ apiVersion:
+ type: string
+ blockOwnerDeletion:
+ type: boolean
+ controller:
+ type: boolean
+ kind:
+ type: string
+ name:
+ type: string
+ uid:
+ type: string
+ type: object
+ type: array
+ resourceVersion:
+ type: string
+ selfLink:
+ type: string
+ uid:
+ type: string
+ type: object
+ grpcRouteSpec:
+ properties:
+ hostnames:
+ items:
+ type: string
+ type: array
+ parentRefs:
+ items:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ namespace:
+ type: string
+ port:
+ type: integer
+ sectionName:
+ type: string
+ type: object
+ type: array
+ rules:
+ items:
+ properties:
+ backendRefs:
+ items:
+ properties:
+ filters:
+ items:
+ properties:
+ extensionRef:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ type: object
+ requestHeaderModifier:
+ properties:
+ add:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ remove:
+ items:
+ type: string
+ type: array
+ set:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ type: object
+ requestMirror:
+ properties:
+ backendRef:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ namespace:
+ type: string
+ port:
+ type: integer
+ type: object
+ fraction:
+ properties:
+ denominator:
+ type: integer
+ numerator:
+ type: integer
+ type: object
+ percent:
+ type: integer
+ type: object
+ responseHeaderModifier:
+ properties:
+ add:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ remove:
+ items:
+ type: string
+ type: array
+ set:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ type: object
+ type:
+ type: string
+ type: object
+ type: array
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ namespace:
+ type: string
+ port:
+ type: integer
+ weight:
+ type: integer
+ type: object
+ type: array
+ filters:
+ items:
+ properties:
+ extensionRef:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ type: object
+ requestHeaderModifier:
+ properties:
+ add:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ remove:
+ items:
+ type: string
+ type: array
+ set:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ type: object
+ requestMirror:
+ properties:
+ backendRef:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ namespace:
+ type: string
+ port:
+ type: integer
+ type: object
+ fraction:
+ properties:
+ denominator:
+ type: integer
+ numerator:
+ type: integer
+ type: object
+ percent:
+ type: integer
+ type: object
+ responseHeaderModifier:
+ properties:
+ add:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ remove:
+ items:
+ type: string
+ type: array
+ set:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ type: object
+ type:
+ type: string
+ type: object
+ type: array
+ matches:
+ items:
+ properties:
+ headers:
+ items:
+ properties:
+ name:
+ type: string
+ type:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ method:
+ properties:
+ method:
+ type: string
+ service:
+ type: string
+ type:
+ type: string
+ type: object
+ type: object
+ type: array
+ name:
+ type: string
+ sessionPersistence:
+ properties:
+ absoluteTimeout:
+ type: string
+ cookieConfig:
+ properties:
+ lifetimeType:
+ type: string
+ type: object
+ idleTimeout:
+ type: string
+ sessionName:
+ type: string
+ type:
+ type: string
+ type: object
+ type: object
+ type: array
+ useDefaultGateways:
+ type: string
+ type: object
+ serviceMetadata:
+ properties:
+ annotations:
+ additionalProperties:
+ type: string
+ type: object
+ creationTimestamp:
+ type: string
+ deletionGracePeriodSeconds:
+ type: integer
+ deletionTimestamp:
+ type: string
+ finalizers:
+ items:
+ type: string
+ type: array
+ generateName:
+ type: string
+ generation:
+ type: integer
+ labels:
+ additionalProperties:
+ type: string
+ type: object
+ managedFields:
+ items:
+ properties:
+ apiVersion:
+ type: string
+ fieldsType:
+ type: string
+ fieldsV1:
+ type: object
+ manager:
+ type: string
+ operation:
+ type: string
+ subresource:
+ type: string
+ time:
+ type: string
+ type: object
+ type: array
+ name:
+ type: string
+ namespace:
+ type: string
+ ownerReferences:
+ items:
+ properties:
+ apiVersion:
+ type: string
+ blockOwnerDeletion:
+ type: boolean
+ controller:
+ type: boolean
+ kind:
+ type: string
+ name:
+ type: string
+ uid:
+ type: string
+ type: object
+ type: array
+ resourceVersion:
+ type: string
+ selfLink:
+ type: string
+ uid:
+ type: string
+ type: object
+ serviceSpec:
+ properties:
+ allocateLoadBalancerNodePorts:
+ type: boolean
+ clusterIP:
+ type: string
+ clusterIPs:
+ items:
+ type: string
+ type: array
+ externalIPs:
+ items:
+ type: string
+ type: array
+ externalName:
+ type: string
+ externalTrafficPolicy:
+ type: string
+ healthCheckNodePort:
+ type: integer
+ internalTrafficPolicy:
+ type: string
+ ipFamilies:
+ items:
+ type: string
+ type: array
+ ipFamilyPolicy:
+ type: string
+ loadBalancerClass:
+ type: string
+ loadBalancerIP:
+ type: string
+ loadBalancerSourceRanges:
+ items:
+ type: string
+ type: array
+ ports:
+ items:
+ properties:
+ appProtocol:
+ type: string
+ name:
+ type: string
+ nodePort:
+ type: integer
+ port:
+ type: integer
+ protocol:
+ type: string
+ targetPort:
+ anyOf:
+ - type: integer
+ - type: string
+ x-kubernetes-int-or-string: true
+ type: object
+ type: array
+ publishNotReadyAddresses:
+ type: boolean
+ selector:
+ additionalProperties:
+ type: string
+ type: object
+ sessionAffinity:
+ type: string
+ sessionAffinityConfig:
+ properties:
+ clientIP:
+ properties:
+ timeoutSeconds:
+ type: integer
+ type: object
+ type: object
+ trafficDistribution:
+ type: string
+ type:
+ type: string
+ type: object
+ required:
+ - grpcRouteMetadata
+ - serviceMetadata
+ - serviceSpec
+ type: object
+ type: array
+ driverHttpRouteList:
+ items:
+ properties:
+ httpRouteMetadata:
+ properties:
+ annotations:
+ additionalProperties:
+ type: string
+ type: object
+ creationTimestamp:
+ type: string
+ deletionGracePeriodSeconds:
+ type: integer
+ deletionTimestamp:
+ type: string
+ finalizers:
+ items:
+ type: string
+ type: array
+ generateName:
+ type: string
+ generation:
+ type: integer
+ labels:
+ additionalProperties:
+ type: string
+ type: object
+ managedFields:
+ items:
+ properties:
+ apiVersion:
+ type: string
+ fieldsType:
+ type: string
+ fieldsV1:
+ type: object
+ manager:
+ type: string
+ operation:
+ type: string
+ subresource:
+ type: string
+ time:
+ type: string
+ type: object
+ type: array
+ name:
+ type: string
+ namespace:
+ type: string
+ ownerReferences:
+ items:
+ properties:
+ apiVersion:
+ type: string
+ blockOwnerDeletion:
+ type: boolean
+ controller:
+ type: boolean
+ kind:
+ type: string
+ name:
+ type: string
+ uid:
+ type: string
+ type: object
+ type: array
+ resourceVersion:
+ type: string
+ selfLink:
+ type: string
+ uid:
+ type: string
+ type: object
+ httpRouteSpec:
+ properties:
+ hostnames:
+ items:
+ type: string
+ type: array
+ parentRefs:
+ items:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ namespace:
+ type: string
+ port:
+ type: integer
+ sectionName:
+ type: string
+ type: object
+ type: array
+ rules:
+ items:
+ properties:
+ backendRefs:
+ items:
+ properties:
+ filters:
+ items:
+ properties:
+ cors:
+ properties:
+ allowCredentials:
+ type: boolean
+ allowHeaders:
+ items:
+ type: string
+ type: array
+ allowMethods:
+ items:
+ type: string
+ type: array
+ allowOrigins:
+ items:
+ type: string
+ type: array
+ exposeHeaders:
+ items:
+ type: string
+ type: array
+ maxAge:
+ type: integer
+ type: object
+ extensionRef:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ type: object
+ externalAuth:
+ properties:
+ backendRef:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ namespace:
+ type: string
+ port:
+ type: integer
+ type: object
+ forwardBody:
+ properties:
+ maxSize:
+ type: integer
+ type: object
+ grpc:
+ properties:
+ allowedHeaders:
+ items:
+ type: string
+ type: array
+ type: object
+ http:
+ properties:
+ allowedHeaders:
+ items:
+ type: string
+ type: array
+ allowedResponseHeaders:
+ items:
+ type: string
+ type: array
+ path:
+ type: string
+ type: object
+ protocol:
+ type: string
+ type: object
+ requestHeaderModifier:
+ properties:
+ add:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ remove:
+ items:
+ type: string
+ type: array
+ set:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ type: object
+ requestMirror:
+ properties:
+ backendRef:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ namespace:
+ type: string
+ port:
+ type: integer
+ type: object
+ fraction:
+ properties:
+ denominator:
+ type: integer
+ numerator:
+ type: integer
+ type: object
+ percent:
+ type: integer
+ type: object
+ requestRedirect:
+ properties:
+ hostname:
+ type: string
+ path:
+ properties:
+ replaceFullPath:
+ type: string
+ replacePrefixMatch:
+ type: string
+ type:
+ type: string
+ type: object
+ port:
+ type: integer
+ scheme:
+ type: string
+ statusCode:
+ type: integer
+ type: object
+ responseHeaderModifier:
+ properties:
+ add:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ remove:
+ items:
+ type: string
+ type: array
+ set:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ type: object
+ type:
+ type: string
+ urlRewrite:
+ properties:
+ hostname:
+ type: string
+ path:
+ properties:
+ replaceFullPath:
+ type: string
+ replacePrefixMatch:
+ type: string
+ type:
+ type: string
+ type: object
+ type: object
+ type: object
+ type: array
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ namespace:
+ type: string
+ port:
+ type: integer
+ weight:
+ type: integer
+ type: object
+ type: array
+ filters:
+ items:
+ properties:
+ cors:
+ properties:
+ allowCredentials:
+ type: boolean
+ allowHeaders:
+ items:
+ type: string
+ type: array
+ allowMethods:
+ items:
+ type: string
+ type: array
+ allowOrigins:
+ items:
+ type: string
+ type: array
+ exposeHeaders:
+ items:
+ type: string
+ type: array
+ maxAge:
+ type: integer
+ type: object
+ extensionRef:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ type: object
+ externalAuth:
+ properties:
+ backendRef:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ namespace:
+ type: string
+ port:
+ type: integer
+ type: object
+ forwardBody:
+ properties:
+ maxSize:
+ type: integer
+ type: object
+ grpc:
+ properties:
+ allowedHeaders:
+ items:
+ type: string
+ type: array
+ type: object
+ http:
+ properties:
+ allowedHeaders:
+ items:
+ type: string
+ type: array
+ allowedResponseHeaders:
+ items:
+ type: string
+ type: array
+ path:
+ type: string
+ type: object
+ protocol:
+ type: string
+ type: object
+ requestHeaderModifier:
+ properties:
+ add:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ remove:
+ items:
+ type: string
+ type: array
+ set:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ type: object
+ requestMirror:
+ properties:
+ backendRef:
+ properties:
+ group:
+ type: string
+ kind:
+ type: string
+ name:
+ type: string
+ namespace:
+ type: string
+ port:
+ type: integer
+ type: object
+ fraction:
+ properties:
+ denominator:
+ type: integer
+ numerator:
+ type: integer
+ type: object
+ percent:
+ type: integer
+ type: object
+ requestRedirect:
+ properties:
+ hostname:
+ type: string
+ path:
+ properties:
+ replaceFullPath:
+ type: string
+ replacePrefixMatch:
+ type: string
+ type:
+ type: string
+ type: object
+ port:
+ type: integer
+ scheme:
+ type: string
+ statusCode:
+ type: integer
+ type: object
+ responseHeaderModifier:
+ properties:
+ add:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ remove:
+ items:
+ type: string
+ type: array
+ set:
+ items:
+ properties:
+ name:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ type: object
+ type:
+ type: string
+ urlRewrite:
+ properties:
+ hostname:
+ type: string
+ path:
+ properties:
+ replaceFullPath:
+ type: string
+ replacePrefixMatch:
+ type: string
+ type:
+ type: string
+ type: object
+ type: object
+ type: object
+ type: array
+ matches:
+ items:
+ properties:
+ headers:
+ items:
+ properties:
+ name:
+ type: string
+ type:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ method:
+ type: string
+ path:
+ properties:
+ type:
+ type: string
+ value:
+ type: string
+ type: object
+ queryParams:
+ items:
+ properties:
+ name:
+ type: string
+ type:
+ type: string
+ value:
+ type: string
+ type: object
+ type: array
+ type: object
+ type: array
+ name:
+ type: string
+ retry:
+ properties:
+ attempts:
+ type: integer
+ backoff:
+ type: string
+ codes:
+ items:
+ type: integer
+ type: array
+ type: object
+ sessionPersistence:
+ properties:
+ absoluteTimeout:
+ type: string
+ cookieConfig:
+ properties:
+ lifetimeType:
+ type: string
+ type: object
+ idleTimeout:
+ type: string
+ sessionName:
+ type: string
+ type:
+ type: string
+ type: object
+ timeouts:
+ properties:
+ backendRequest:
+ type: string
+ request:
+ type: string
+ type: object
+ type: object
+ type: array
+ useDefaultGateways:
+ type: string
+ type: object
+ serviceMetadata:
+ properties:
+ annotations:
+ additionalProperties:
+ type: string
+ type: object
+ creationTimestamp:
+ type: string
+ deletionGracePeriodSeconds:
+ type: integer
+ deletionTimestamp:
+ type: string
+ finalizers:
+ items:
+ type: string
+ type: array
+ generateName:
+ type: string
+ generation:
+ type: integer
+ labels:
+ additionalProperties:
+ type: string
+ type: object
+ managedFields:
+ items:
+ properties:
+ apiVersion:
+ type: string
+ fieldsType:
+ type: string
+ fieldsV1:
+ type: object
+ manager:
+ type: string
+ operation:
+ type: string
+ subresource:
+ type: string
+ time:
+ type: string
+ type: object
+ type: array
+ name:
+ type: string
+ namespace:
+ type: string
+ ownerReferences:
+ items:
+ properties:
+ apiVersion:
+ type: string
+ blockOwnerDeletion:
+ type: boolean
+ controller:
+ type: boolean
+ kind:
+ type: string
+ name:
+ type: string
+ uid:
+ type: string
+ type: object
+ type: array
+ resourceVersion:
+ type: string
+ selfLink:
+ type: string
+ uid:
+ type: string
+ type: object
+ serviceSpec:
+ properties:
+ allocateLoadBalancerNodePorts:
+ type: boolean
+ clusterIP:
+ type: string
+ clusterIPs:
+ items:
+ type: string
+ type: array
+ externalIPs:
+ items:
+ type: string
+ type: array
+ externalName:
+ type: string
+ externalTrafficPolicy:
+ type: string
+ healthCheckNodePort:
+ type: integer
+ internalTrafficPolicy:
+ type: string
+ ipFamilies:
+ items:
+ type: string
+ type: array
+ ipFamilyPolicy:
+ type: string
+ loadBalancerClass:
+ type: string
+ loadBalancerIP:
+ type: string
+ loadBalancerSourceRanges:
+ items:
+ type: string
+ type: array
+ ports:
+ items:
+ properties:
+ appProtocol:
+ type: string
+ name:
+ type: string
+ nodePort:
+ type: integer
+ port:
+ type: integer
+ protocol:
+ type: string
+ targetPort:
+ anyOf:
+ - type: integer
+ - type: string
+ x-kubernetes-int-or-string: true
+ type: object
+ type: array
+ publishNotReadyAddresses:
+ type: boolean
+ selector:
+ additionalProperties:
+ type: string
+ type: object
+ sessionAffinity:
+ type: string
+ sessionAffinityConfig:
+ properties:
+ clientIP:
+ properties:
+ timeoutSeconds:
+ type: integer
+ type: object
+ type: object
+ trafficDistribution:
+ type: string
+ type:
+ type: string
+ type: object
+ required:
+ - httpRouteMetadata
+ - serviceMetadata
+ - serviceSpec
+ type: object
+ type: array
driverServiceIngressList:
items:
properties:
diff --git a/build-tools/helm/spark-kubernetes-operator/templates/operator-rbac.yaml b/build-tools/helm/spark-kubernetes-operator/templates/operator-rbac.yaml
index 46494d5..ec43357 100644
--- a/build-tools/helm/spark-kubernetes-operator/templates/operator-rbac.yaml
+++ b/build-tools/helm/spark-kubernetes-operator/templates/operator-rbac.yaml
@@ -83,6 +83,19 @@ rules:
- update
- patch
- delete
+ - apiGroups:
+ - "gateway.networking.k8s.io"
+ resources:
+ - httproutes
+ - grpcroutes
+ verbs:
+ - get
+ - list
+ - watch
+ - create
+ - update
+ - patch
+ - delete
- apiGroups:
- "policy"
resources:
diff --git a/docs/operations.md b/docs/operations.md
index 8ff47d9..35ac801 100644
--- a/docs/operations.md
+++ b/docs/operations.md
@@ -27,6 +27,17 @@ under the License.
patch support will not be performed on k8s versions that reached EOL.
- Spark versions 4.0 or above.
+## Optional Prerequisites
+
+- **Gateway API v1 CRDs** (`httproutes.gateway.networking.k8s.io`,
+ `grpcroutes.gateway.networking.k8s.io`) — required only when SparkApplications use
+ `spec.driverHttpRouteList` or `spec.driverGrpcRouteList`. These CRDs are not bundled with the
+ operator; install them from
+ [sigs.k8s.io/gateway-api](https://gateway-api.sigs.k8s.io/guides/) (or a Gateway API
+ implementation such as Istio) before submitting SparkApplications that request Gateway API
+ routing. If the CRDs are missing, reconciliation of such SparkApplications will fail with a
+ `no matches for kind "HTTPRoute"` error from the Kubernetes API.
+
## Spark Application Namespaces
By default, Spark applications are created in the same namespace as the operator deployment.
diff --git a/docs/spark_custom_resources.md b/docs/spark_custom_resources.md
index 2472f72..8a1ae69 100644
--- a/docs/spark_custom_resources.md
+++ b/docs/spark_custom_resources.md
@@ -142,6 +142,53 @@ default rule backed by the associated Service. It's recommended to always provid
to make sure it's compatible with your
[IngressController](https://kubernetes.io/docs/concepts/services-networking/ingress-controllers/).
+## Enable Gateway API Route for Driver
+
+As an alternative to `Ingress`, the operator can expose driver endpoints via the Kubernetes
+[Gateway API](https://gateway-api.sigs.k8s.io/) by creating `HTTPRoute` or `GRPCRoute` resources
+alongside a backing Service. Use `driverHttpRouteList` for HTTP endpoints (for example the Spark
+UI) and `driverGrpcRouteList` for gRPC endpoints (for example Spark Connect).
+
+```yaml
+spec:
+ driverHttpRouteList:
+ - serviceMetadata:
+ name: "spark-ui-service"
+ serviceSpec:
+ ports:
+ - protocol: TCP
+ port: 80
+ targetPort: 4040
+ httpRouteMetadata:
+ name: "spark-ui-route"
+ httpRouteSpec:
+ parentRefs:
+ - group: gateway.networking.k8s.io
+ kind: Gateway
+ name: my-gateway
+ rules:
+ - matches:
+ - path:
+ type: PathPrefix
+ value: "/"
+ backendRefs:
+ - name: spark-ui-service
+ port: 80
+```
+
+Prerequisites and behavior:
+
+- The Gateway API v1 CRDs (`httproutes.gateway.networking.k8s.io`,
+ `grpcroutes.gateway.networking.k8s.io`) must be installed on the cluster. If a CRD is missing,
+ reconciliation of a SparkApplication that references the corresponding route list will fail
+ with a `no matches for kind "HTTPRoute"` (or `"GRPCRoute"`) error from the Kubernetes API.
+- `httpRouteSpec.parentRefs` (and `grpcRouteSpec.parentRefs`) is required — without at least one
+ `parentRef` the route object would be orphaned from any `Gateway` and receive no traffic. The
+ operator rejects SparkApplications where any route-list entry omits `parentRefs`.
+- As with Ingress, the Service's `.spec.selector` defaults to the driver labels, and if
+ `httpRouteSpec.rules` / `grpcRouteSpec.rules` is not provided the operator populates a single
+ default rule that backends to the first port of the associated Service.
+
## Create and Mount ConfigMap
It is possible to ask operator to create configmap so they can be used by driver and/or executor
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 1bcce86..c35f9df 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -45,6 +45,7 @@ versions-plugin = "0.54.0"
[libraries]
kubernetes-client = { group = "io.fabric8", name = "kubernetes-client", version.ref = "fabric8" }
+kubernetes-model-gatewayapi = { group = "io.fabric8", name = "kubernetes-model-gatewayapi", version.ref = "fabric8" }
kubernetes-httpclient-vertx = { group = "io.fabric8", name = "kubernetes-httpclient-vertx", version.ref = "fabric8" }
kubernetes-server-mock = { group = "io.fabric8", name = "kubernetes-server-mock", version.ref = "fabric8" }
kube-api-test-client-inject = {group = "io.fabric8", name = "kube-api-test-client-inject", version.ref = "fabric8"}
diff --git a/spark-operator-api/build.gradle b/spark-operator-api/build.gradle
index 4a3416e..acdf834 100644
--- a/spark-operator-api/build.gradle
+++ b/spark-operator-api/build.gradle
@@ -19,6 +19,7 @@
dependencies {
// fabric8
implementation(libs.kubernetes.client)
+ implementation(libs.kubernetes.model.gatewayapi)
compileOnly(libs.crd.generator.annotations)
testImplementation(platform(libs.junit.bom))
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
index 56599b8..36ed271 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
@@ -43,6 +43,7 @@ import lombok.NoArgsConstructor;
@EqualsAndHashCode(callSuper = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
@JsonIgnoreProperties(ignoreUnknown = true)
+@SuppressWarnings("PMD.TooManyFields")
public class ApplicationSpec extends BaseSpec {
protected String mainClass;
@Required protected RuntimeVersions runtimeVersions;
@@ -60,5 +61,7 @@ public class ApplicationSpec extends BaseSpec {
protected BaseApplicationTemplateSpec driverSpec;
protected BaseApplicationTemplateSpec executorSpec;
protected List<DriverServiceIngressSpec> driverServiceIngressList;
+ protected List<DriverHttpRouteSpec> driverHttpRouteList;
+ protected List<DriverGrpcRouteSpec> driverGrpcRouteList;
protected List<ConfigMapSpec> configMapSpecs;
}
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/DriverGrpcRouteSpec.java
similarity index 57%
copy from spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
copy to spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/DriverGrpcRouteSpec.java
index 56599b8..c0ee5f0 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/DriverGrpcRouteSpec.java
@@ -19,46 +19,37 @@
package org.apache.spark.k8s.operator.spec;
-import java.util.ArrayList;
-import java.util.List;
-
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonInclude;
import io.fabric8.generator.annotation.Required;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.ServiceSpec;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRouteSpec;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
-import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
/**
- * Spec for a Spark application.
+ * Spec for a driver service exposed via a Gateway API {@code GRPCRoute}.
+ *
+ * <p>Intended for gRPC endpoints such as Spark Connect. Emits a {@code GRPCRoute}
+ * ({@code gateway.networking.k8s.io/v1}) alongside a backing Service. If
+ * {@code grpcRouteSpec.rules} is not provided, a single default rule is populated that routes all
+ * traffic to the first port of the backing Service.
*
- * @since 0.1.0
+ * @since 1.0.0
*/
@Data
-@NoArgsConstructor
-@AllArgsConstructor
@Builder
-@EqualsAndHashCode(callSuper = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
+@NoArgsConstructor
+@AllArgsConstructor
@JsonIgnoreProperties(ignoreUnknown = true)
-public class ApplicationSpec extends BaseSpec {
- protected String mainClass;
- @Required protected RuntimeVersions runtimeVersions;
- protected String jars;
- protected String pyFiles;
- protected String sparkRFiles;
- protected String files;
- @Builder.Default protected DeploymentMode deploymentMode = DeploymentMode.ClusterMode;
- protected String proxyUser;
- @Builder.Default protected List<String> driverArgs = new ArrayList<>();
-
- @Builder.Default
- protected ApplicationTolerations applicationTolerations = new ApplicationTolerations();
+public class DriverGrpcRouteSpec {
+ @Required protected ObjectMeta serviceMetadata;
+ @Required protected ServiceSpec serviceSpec;
- protected BaseApplicationTemplateSpec driverSpec;
- protected BaseApplicationTemplateSpec executorSpec;
- protected List<DriverServiceIngressSpec> driverServiceIngressList;
- protected List<ConfigMapSpec> configMapSpecs;
+ @Required protected ObjectMeta grpcRouteMetadata;
+ protected GRPCRouteSpec grpcRouteSpec;
}
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/DriverHttpRouteSpec.java
similarity index 57%
copy from spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
copy to spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/DriverHttpRouteSpec.java
index 56599b8..fdac410 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/DriverHttpRouteSpec.java
@@ -19,46 +19,37 @@
package org.apache.spark.k8s.operator.spec;
-import java.util.ArrayList;
-import java.util.List;
-
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonInclude;
import io.fabric8.generator.annotation.Required;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.ServiceSpec;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRouteSpec;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
-import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
/**
- * Spec for a Spark application.
+ * Spec for a driver service exposed via a Gateway API {@code HTTPRoute}.
+ *
+ * <p>Parallel to {@link DriverServiceIngressSpec} but emits a {@code HTTPRoute}
+ * ({@code gateway.networking.k8s.io/v1}) instead of a {@code networking.k8s.io/v1 Ingress}.
+ * If {@code httpRouteSpec.rules} is not provided, a single default rule is populated that routes
+ * all traffic to the first port of the backing Service.
*
- * @since 0.1.0
+ * @since 1.0.0
*/
@Data
-@NoArgsConstructor
-@AllArgsConstructor
@Builder
-@EqualsAndHashCode(callSuper = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
+@NoArgsConstructor
+@AllArgsConstructor
@JsonIgnoreProperties(ignoreUnknown = true)
-public class ApplicationSpec extends BaseSpec {
- protected String mainClass;
- @Required protected RuntimeVersions runtimeVersions;
- protected String jars;
- protected String pyFiles;
- protected String sparkRFiles;
- protected String files;
- @Builder.Default protected DeploymentMode deploymentMode = DeploymentMode.ClusterMode;
- protected String proxyUser;
- @Builder.Default protected List<String> driverArgs = new ArrayList<>();
-
- @Builder.Default
- protected ApplicationTolerations applicationTolerations = new ApplicationTolerations();
+public class DriverHttpRouteSpec {
+ @Required protected ObjectMeta serviceMetadata;
+ @Required protected ServiceSpec serviceSpec;
- protected BaseApplicationTemplateSpec driverSpec;
- protected BaseApplicationTemplateSpec executorSpec;
- protected List<DriverServiceIngressSpec> driverServiceIngressList;
- protected List<ConfigMapSpec> configMapSpecs;
+ @Required protected ObjectMeta httpRouteMetadata;
+ protected HTTPRouteSpec httpRouteSpec;
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStep.java
index 23c66b6..a9d2a19 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStep.java
@@ -24,10 +24,18 @@ import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.proceed
import static org.apache.spark.k8s.operator.utils.ModelUtils.isClientMode;
import static org.apache.spark.k8s.operator.utils.SparkAppStatusUtils.isValidApplicationStatus;
+import java.util.List;
+
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRouteSpec;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRouteSpec;
import lombok.extern.slf4j.Slf4j;
+import org.apache.spark.k8s.operator.SparkApplication;
import org.apache.spark.k8s.operator.context.SparkAppContext;
import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
+import org.apache.spark.k8s.operator.spec.ApplicationSpec;
+import org.apache.spark.k8s.operator.spec.DriverGrpcRouteSpec;
+import org.apache.spark.k8s.operator.spec.DriverHttpRouteSpec;
import org.apache.spark.k8s.operator.status.ApplicationState;
import org.apache.spark.k8s.operator.status.ApplicationStateSummary;
import org.apache.spark.k8s.operator.status.ApplicationStatus;
@@ -46,19 +54,89 @@ public final class AppValidateStep extends AppReconcileStep {
@Override
public ReconcileProgress reconcile(
SparkAppContext context, SparkAppStatusRecorder statusRecorder) {
- if (!isValidApplicationStatus(context.getResource())) {
+ SparkApplication app = context.getResource();
+ if (!isValidApplicationStatus(app)) {
log.warn("Spark application found with empty status. Resetting to initial state.");
return attemptStatusUpdate(context, statusRecorder, new ApplicationStatus(), proceed());
}
- if (isClientMode(context.getResource())) {
- ApplicationState failure =
- new ApplicationState(ApplicationStateSummary.Failed, "Client mode is not supported yet.");
- return attemptStatusUpdate(
- context,
- statusRecorder,
- context.getResource().getStatus().appendNewState(failure),
- completeAndImmediateRequeue());
+ if (isClientMode(app)) {
+ return failWith(context, statusRecorder, "Client mode is not supported yet.");
+ }
+ if (app.getStatus().getCurrentState().getCurrentStateSummary().isInitializing()) {
+ String routeFailure = validateGatewayRoutes(app.getSpec());
+ if (routeFailure != null) {
+ return failWith(context, statusRecorder, routeFailure);
+ }
}
return proceed();
}
+
+ private ReconcileProgress failWith(
+ SparkAppContext context, SparkAppStatusRecorder statusRecorder, String message) {
+ ApplicationState failure = new ApplicationState(ApplicationStateSummary.Failed, message);
+ return attemptStatusUpdate(
+ context,
+ statusRecorder,
+ context.getResource().getStatus().appendNewState(failure),
+ completeAndImmediateRequeue());
+ }
+
+ static String validateGatewayRoutes(ApplicationSpec spec) {
+ if (spec == null) {
+ return null;
+ }
+ List<DriverHttpRouteSpec> httpList = spec.getDriverHttpRouteList();
+ if (httpList != null) {
+ for (int i = 0; i < httpList.size(); i++) {
+ DriverHttpRouteSpec entry = httpList.get(i);
+ String prefix = "driverHttpRouteList[" + i + "]";
+ if (entry.getServiceSpec() == null) {
+ return prefix + ".serviceSpec is required.";
+ }
+ if (entry.getServiceMetadata() == null
+ || entry.getServiceMetadata().getName() == null
+ || entry.getServiceMetadata().getName().isEmpty()) {
+ return prefix + ".serviceMetadata.name is required.";
+ }
+ if (entry.getHttpRouteMetadata() == null
+ || entry.getHttpRouteMetadata().getName() == null
+ || entry.getHttpRouteMetadata().getName().isEmpty()) {
+ return prefix + ".httpRouteMetadata.name is required.";
+ }
+ HTTPRouteSpec routeSpec = entry.getHttpRouteSpec();
+ if (routeSpec == null
+ || routeSpec.getParentRefs() == null
+ || routeSpec.getParentRefs().isEmpty()) {
+ return prefix + ".httpRouteSpec.parentRefs is required.";
+ }
+ }
+ }
+ List<DriverGrpcRouteSpec> grpcList = spec.getDriverGrpcRouteList();
+ if (grpcList != null) {
+ for (int i = 0; i < grpcList.size(); i++) {
+ DriverGrpcRouteSpec entry = grpcList.get(i);
+ String prefix = "driverGrpcRouteList[" + i + "]";
+ if (entry.getServiceSpec() == null) {
+ return prefix + ".serviceSpec is required.";
+ }
+ if (entry.getServiceMetadata() == null
+ || entry.getServiceMetadata().getName() == null
+ || entry.getServiceMetadata().getName().isEmpty()) {
+ return prefix + ".serviceMetadata.name is required.";
+ }
+ if (entry.getGrpcRouteMetadata() == null
+ || entry.getGrpcRouteMetadata().getName() == null
+ || entry.getGrpcRouteMetadata().getName().isEmpty()) {
+ return prefix + ".grpcRouteMetadata.name is required.";
+ }
+ GRPCRouteSpec routeSpec = entry.getGrpcRouteSpec();
+ if (routeSpec == null
+ || routeSpec.getParentRefs() == null
+ || routeSpec.getParentRefs().isEmpty()) {
+ return prefix + ".grpcRouteSpec.parentRefs is required.";
+ }
+ }
+ }
+ return null;
+ }
}
diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStepTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStepTest.java
new file mode 100644
index 0000000..7819b09
--- /dev/null
+++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStepTest.java
@@ -0,0 +1,268 @@
+/*
+ * 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.reconciler.reconcilesteps;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.ServiceSpec;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRouteSpec;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRouteSpecBuilder;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRouteSpec;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRouteSpecBuilder;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.ParentReference;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.ParentReferenceBuilder;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import org.apache.spark.k8s.operator.SparkApplication;
+import org.apache.spark.k8s.operator.context.SparkAppContext;
+import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
+import org.apache.spark.k8s.operator.spec.ApplicationSpec;
+import org.apache.spark.k8s.operator.spec.DriverGrpcRouteSpec;
+import org.apache.spark.k8s.operator.spec.DriverHttpRouteSpec;
+import org.apache.spark.k8s.operator.status.ApplicationState;
+import org.apache.spark.k8s.operator.status.ApplicationStateSummary;
+import org.apache.spark.k8s.operator.status.ApplicationStatus;
+import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
+
+class AppValidateStepTest {
+
+ private static final ParentReference PARENT_REF =
+ new ParentReferenceBuilder().withGroup("gateway.networking.k8s.io").withKind("Gateway")
+ .withName("dex-gw").build();
+
+ private static SparkApplication appWith(ApplicationSpec spec) {
+ SparkApplication app = new SparkApplication();
+ ObjectMeta meta = new ObjectMeta();
+ meta.setName("app1");
+ meta.setNamespace("default");
+ app.setMetadata(meta);
+ app.setSpec(spec);
+ ApplicationStatus status =
+ new ApplicationStatus()
+ .appendNewState(new ApplicationState(ApplicationStateSummary.Submitted, "ok"));
+ app.setStatus(status);
+ return app;
+ }
+
+ private static DriverHttpRouteSpec httpEntry(HTTPRouteSpec routeSpec) {
+ DriverHttpRouteSpec s = new DriverHttpRouteSpec();
+ s.setServiceMetadata(new ObjectMetaBuilder().withName("svc").build());
+ s.setServiceSpec(new ServiceSpec());
+ s.setHttpRouteMetadata(new ObjectMetaBuilder().withName("route").build());
+ s.setHttpRouteSpec(routeSpec);
+ return s;
+ }
+
+ private static DriverGrpcRouteSpec grpcEntry(GRPCRouteSpec routeSpec) {
+ DriverGrpcRouteSpec s = new DriverGrpcRouteSpec();
+ s.setServiceMetadata(new ObjectMetaBuilder().withName("svc").build());
+ s.setServiceSpec(new ServiceSpec());
+ s.setGrpcRouteMetadata(new ObjectMetaBuilder().withName("route").build());
+ s.setGrpcRouteSpec(routeSpec);
+ return s;
+ }
+
+ // Tests below cover SPARK-58160 Gateway API HTTPRoute/GRPCRoute validation.
+
+ @Test
+ void nullSpecReturnsNoError() {
+ Assertions.assertNull(AppValidateStep.validateGatewayRoutes(null));
+ }
+
+ @Test
+ void emptyRouteListsReturnNoError() {
+ Assertions.assertNull(AppValidateStep.validateGatewayRoutes(new ApplicationSpec()));
+ }
+
+ @Test
+ void httpRouteWithParentRefsPasses() {
+ HTTPRouteSpec routeSpec = new HTTPRouteSpecBuilder().addToParentRefs(PARENT_REF).build();
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverHttpRouteList(List.of(httpEntry(routeSpec)));
+ Assertions.assertNull(AppValidateStep.validateGatewayRoutes(spec));
+ }
+
+ @Test
+ void grpcRouteWithParentRefsPasses() {
+ GRPCRouteSpec routeSpec = new GRPCRouteSpecBuilder().addToParentRefs(PARENT_REF).build();
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverGrpcRouteList(List.of(grpcEntry(routeSpec)));
+ Assertions.assertNull(AppValidateStep.validateGatewayRoutes(spec));
+ }
+
+ @Test
+ void httpRouteMissingParentRefsIsRejected() {
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverHttpRouteList(List.of(httpEntry(new HTTPRouteSpec())));
+ String err = AppValidateStep.validateGatewayRoutes(spec);
+ Assertions.assertNotNull(err);
+ Assertions.assertTrue(err.contains("driverHttpRouteList[0].httpRouteSpec.parentRefs"), err);
+ }
+
+ @Test
+ void httpRouteNullSpecIsRejected() {
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverHttpRouteList(List.of(httpEntry(null)));
+ String err = AppValidateStep.validateGatewayRoutes(spec);
+ Assertions.assertNotNull(err);
+ Assertions.assertTrue(err.contains("driverHttpRouteList[0]"), err);
+ }
+
+ @Test
+ void httpEntryMissingServiceSpecIsRejected() {
+ HTTPRouteSpec routeSpec = new HTTPRouteSpecBuilder().addToParentRefs(PARENT_REF).build();
+ DriverHttpRouteSpec entry = httpEntry(routeSpec);
+ entry.setServiceSpec(null);
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverHttpRouteList(List.of(entry));
+ String err = AppValidateStep.validateGatewayRoutes(spec);
+ Assertions.assertNotNull(err);
+ Assertions.assertTrue(err.contains("driverHttpRouteList[0].serviceSpec"), err);
+ }
+
+ @Test
+ void httpEntryMissingServiceMetadataNameIsRejected() {
+ HTTPRouteSpec routeSpec = new HTTPRouteSpecBuilder().addToParentRefs(PARENT_REF).build();
+ DriverHttpRouteSpec entry = httpEntry(routeSpec);
+ entry.setServiceMetadata(new ObjectMeta());
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverHttpRouteList(List.of(entry));
+ String err = AppValidateStep.validateGatewayRoutes(spec);
+ Assertions.assertNotNull(err);
+ Assertions.assertTrue(err.contains("driverHttpRouteList[0].serviceMetadata.name"), err);
+ }
+
+ @Test
+ void httpEntryMissingRouteMetadataNameIsRejected() {
+ HTTPRouteSpec routeSpec = new HTTPRouteSpecBuilder().addToParentRefs(PARENT_REF).build();
+ DriverHttpRouteSpec entry = httpEntry(routeSpec);
+ entry.setHttpRouteMetadata(new ObjectMeta());
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverHttpRouteList(List.of(entry));
+ String err = AppValidateStep.validateGatewayRoutes(spec);
+ Assertions.assertNotNull(err);
+ Assertions.assertTrue(err.contains("driverHttpRouteList[0].httpRouteMetadata.name"), err);
+ }
+
+ @Test
+ void grpcEntryMissingServiceSpecIsRejected() {
+ GRPCRouteSpec routeSpec = new GRPCRouteSpecBuilder().addToParentRefs(PARENT_REF).build();
+ DriverGrpcRouteSpec entry = grpcEntry(routeSpec);
+ entry.setServiceSpec(null);
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverGrpcRouteList(List.of(entry));
+ String err = AppValidateStep.validateGatewayRoutes(spec);
+ Assertions.assertNotNull(err);
+ Assertions.assertTrue(err.contains("driverGrpcRouteList[0].serviceSpec"), err);
+ }
+
+ @Test
+ void grpcEntryMissingRouteMetadataNameIsRejected() {
+ GRPCRouteSpec routeSpec = new GRPCRouteSpecBuilder().addToParentRefs(PARENT_REF).build();
+ DriverGrpcRouteSpec entry = grpcEntry(routeSpec);
+ entry.setGrpcRouteMetadata(new ObjectMeta());
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverGrpcRouteList(List.of(entry));
+ String err = AppValidateStep.validateGatewayRoutes(spec);
+ Assertions.assertNotNull(err);
+ Assertions.assertTrue(err.contains("driverGrpcRouteList[0].grpcRouteMetadata.name"), err);
+ }
+
+ @Test
+ void grpcRouteMissingParentRefsIsRejected() {
+ HTTPRouteSpec http = new HTTPRouteSpecBuilder().addToParentRefs(PARENT_REF).build();
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverHttpRouteList(List.of(httpEntry(http)));
+ spec.setDriverGrpcRouteList(List.of(grpcEntry(new GRPCRouteSpec())));
+ String err = AppValidateStep.validateGatewayRoutes(spec);
+ Assertions.assertNotNull(err);
+ Assertions.assertTrue(err.contains("driverGrpcRouteList[0].grpcRouteSpec.parentRefs"), err);
+ }
+
+ @Test
+ void reconcileMovesAppToFailedWhenParentRefsMissing() {
+ AppValidateStep step = new AppValidateStep();
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverHttpRouteList(List.of(httpEntry(new HTTPRouteSpec())));
+ SparkApplication app = appWith(spec);
+
+ SparkAppContext ctx = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+ when(ctx.getResource()).thenReturn(app);
+ when(recorder.persistStatus(any(), any())).thenReturn(true);
+
+ ReconcileProgress progress = step.reconcile(ctx, recorder);
+ Assertions.assertEquals(ReconcileProgress.completeAndImmediateRequeue(), progress);
+
+ ArgumentCaptor<ApplicationStatus> captor = ArgumentCaptor.forClass(ApplicationStatus.class);
+ Mockito.verify(recorder).persistStatus(any(), captor.capture());
+ ApplicationStatus persisted = captor.getValue();
+ Assertions.assertEquals(
+ ApplicationStateSummary.Failed, persisted.getCurrentState().getCurrentStateSummary());
+ Assertions.assertTrue(
+ persisted.getCurrentState().getMessage().contains("parentRefs"),
+ persisted.getCurrentState().getMessage());
+ }
+
+ @Test
+ void reconcileSkipsRouteValidationForRunningApp() {
+ AppValidateStep step = new AppValidateStep();
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverHttpRouteList(List.of(httpEntry(new HTTPRouteSpec())));
+ SparkApplication app = appWith(spec);
+ app.setStatus(
+ app.getStatus()
+ .appendNewState(new ApplicationState(ApplicationStateSummary.RunningHealthy, "ok")));
+
+ SparkAppContext ctx = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+ when(ctx.getResource()).thenReturn(app);
+
+ ReconcileProgress progress = step.reconcile(ctx, recorder);
+ Assertions.assertEquals(ReconcileProgress.proceed(), progress);
+ Mockito.verifyNoInteractions(recorder);
+ }
+
+ @Test
+ void reconcileProceedsWhenRoutesValid() {
+ AppValidateStep step = new AppValidateStep();
+ HTTPRouteSpec routeSpec = new HTTPRouteSpecBuilder().addToParentRefs(PARENT_REF).build();
+ ApplicationSpec spec = new ApplicationSpec();
+ spec.setDriverHttpRouteList(List.of(httpEntry(routeSpec)));
+ SparkApplication app = appWith(spec);
+
+ SparkAppContext ctx = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+ when(ctx.getResource()).thenReturn(app);
+
+ ReconcileProgress progress = step.reconcile(ctx, recorder);
+ Assertions.assertEquals(ReconcileProgress.proceed(), progress);
+ }
+
+}
diff --git a/spark-submission-worker/build.gradle b/spark-submission-worker/build.gradle
index b338fe4..c6a7e13 100644
--- a/spark-submission-worker/build.gradle
+++ b/spark-submission-worker/build.gradle
@@ -19,6 +19,7 @@
dependencies {
implementation project(":spark-operator-api")
implementation(libs.kubernetes.client)
+ implementation(libs.kubernetes.model.gatewayapi)
implementation(libs.log4j.api)
implementation(libs.log4j.core)
implementation(libs.log4j.slf4j2.impl)
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java
index 4a70c3d..5b50d54 100644
--- a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java
@@ -37,8 +37,11 @@ import org.apache.spark.deploy.k8s.KubernetesDriverSpec;
import org.apache.spark.deploy.k8s.SparkPod;
import org.apache.spark.deploy.k8s.submit.KubernetesClientUtils;
import org.apache.spark.k8s.operator.spec.ConfigMapSpec;
+import org.apache.spark.k8s.operator.spec.DriverGrpcRouteSpec;
+import org.apache.spark.k8s.operator.spec.DriverHttpRouteSpec;
import org.apache.spark.k8s.operator.spec.DriverServiceIngressSpec;
import org.apache.spark.k8s.operator.utils.ConfigMapSpecUtils;
+import org.apache.spark.k8s.operator.utils.DriverGatewayRouteUtils;
import org.apache.spark.k8s.operator.utils.DriverServiceIngressUtils;
import org.apache.spark.k8s.operator.utils.StringUtils;
@@ -65,12 +68,16 @@ public class SparkAppResourceSpec {
* @param kubernetesDriverConf The KubernetesDriverConf for the application.
* @param kubernetesDriverSpec The KubernetesDriverSpec for the application.
* @param driverServiceIngressList A list of DriverServiceIngressSpec for the driver service.
+ * @param driverHttpRouteList A list of DriverHttpRouteSpec for Gateway API HTTPRoutes.
+ * @param driverGrpcRouteList A list of DriverGrpcRouteSpec for Gateway API GRPCRoutes.
* @param configMapSpecs A list of ConfigMapSpec for additional config maps.
*/
public SparkAppResourceSpec(
SparkAppDriverConf kubernetesDriverConf,
KubernetesDriverSpec kubernetesDriverSpec,
List<DriverServiceIngressSpec> driverServiceIngressList,
+ List<DriverHttpRouteSpec> driverHttpRouteList,
+ List<DriverGrpcRouteSpec> driverGrpcRouteList,
List<ConfigMapSpec> configMapSpecs) {
this.kubernetesDriverConf = kubernetesDriverConf;
String namespace = kubernetesDriverConf.sparkConf().get(Config.KUBERNETES_NAMESPACE().key());
@@ -98,6 +105,8 @@ public class SparkAppResourceSpec {
this.driverPreResources.addAll(ConfigMapSpecUtils.buildConfigMaps(configMapSpecs));
this.driverPreResources.add(buildNetworkPolicy(kubernetesDriverConf.appId(), namespace));
this.driverResources.addAll(configureDriverServerIngress(sparkPod, driverServiceIngressList));
+ this.driverResources.addAll(configureDriverHttpRoutes(sparkPod, driverHttpRouteList));
+ this.driverResources.addAll(configureDriverGrpcRoutes(sparkPod, driverGrpcRouteList));
this.driverPreResources.forEach(r -> setNamespaceIfMissing(r, namespace));
this.driverResources.forEach(r -> setNamespaceIfMissing(r, namespace));
}
@@ -167,6 +176,42 @@ public class SparkAppResourceSpec {
.collect(Collectors.toList());
}
+ /**
+ * Configures driver Gateway API HTTPRoute resources.
+ *
+ * @param pod The SparkPod for the driver.
+ * @param driverHttpRouteList A list of DriverHttpRouteSpec.
+ * @return A List of HasMetadata objects representing the HTTPRoute + Service resources.
+ */
+ private List<HasMetadata> configureDriverHttpRoutes(
+ SparkPod pod, List<DriverHttpRouteSpec> driverHttpRouteList) {
+ if (driverHttpRouteList == null || driverHttpRouteList.isEmpty()) {
+ return List.of();
+ }
+ return driverHttpRouteList.stream()
+ .map(spec -> DriverGatewayRouteUtils.buildHttpRouteService(spec, pod.pod().getMetadata()))
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Configures driver Gateway API GRPCRoute resources.
+ *
+ * @param pod The SparkPod for the driver.
+ * @param driverGrpcRouteList A list of DriverGrpcRouteSpec.
+ * @return A List of HasMetadata objects representing the GRPCRoute + Service resources.
+ */
+ private List<HasMetadata> configureDriverGrpcRoutes(
+ SparkPod pod, List<DriverGrpcRouteSpec> driverGrpcRouteList) {
+ if (driverGrpcRouteList == null || driverGrpcRouteList.isEmpty()) {
+ return List.of();
+ }
+ return driverGrpcRouteList.stream()
+ .map(spec -> DriverGatewayRouteUtils.buildGrpcRouteService(spec, pod.pod().getMetadata()))
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
+
/**
* Builds the NetworkPolicy for the SparkApplication.
*
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
index 708f36d..012f67c 100644
--- a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
@@ -39,6 +39,8 @@ import org.apache.spark.deploy.k8s.submit.PythonMainAppResource;
import org.apache.spark.deploy.k8s.submit.RMainAppResource;
import org.apache.spark.k8s.operator.spec.ApplicationSpec;
import org.apache.spark.k8s.operator.spec.ConfigMapSpec;
+import org.apache.spark.k8s.operator.spec.DriverGrpcRouteSpec;
+import org.apache.spark.k8s.operator.spec.DriverHttpRouteSpec;
import org.apache.spark.k8s.operator.spec.DriverServiceIngressSpec;
import org.apache.spark.k8s.operator.spec.RestartConfig;
import org.apache.spark.k8s.operator.spec.RestartPolicy;
@@ -111,6 +113,8 @@ public class SparkAppSubmissionWorker {
return buildResourceSpec(
appDriverConf,
app.getSpec().getDriverServiceIngressList(),
+ app.getSpec().getDriverHttpRouteList(),
+ app.getSpec().getDriverGrpcRouteList(),
app.getSpec().getConfigMapSpecs(),
client);
}
@@ -193,6 +197,8 @@ public class SparkAppSubmissionWorker {
*
* @param kubernetesDriverConf The SparkAppDriverConf.
* @param driverServiceIngressList A list of DriverServiceIngressSpec.
+ * @param driverHttpRouteList A list of DriverHttpRouteSpec.
+ * @param driverGrpcRouteList A list of DriverGrpcRouteSpec.
* @param configMapSpecs A list of ConfigMapSpec.
* @param client The KubernetesClient.
* @return A SparkAppResourceSpec instance.
@@ -200,13 +206,20 @@ public class SparkAppSubmissionWorker {
protected SparkAppResourceSpec buildResourceSpec(
SparkAppDriverConf kubernetesDriverConf,
List<DriverServiceIngressSpec> driverServiceIngressList,
+ List<DriverHttpRouteSpec> driverHttpRouteList,
+ List<DriverGrpcRouteSpec> driverGrpcRouteList,
List<ConfigMapSpec> configMapSpecs,
KubernetesClient client) {
KubernetesDriverBuilder builder = new KubernetesDriverBuilder();
KubernetesDriverSpec kubernetesDriverSpec =
builder.buildFromFeatures(kubernetesDriverConf, client);
return new SparkAppResourceSpec(
- kubernetesDriverConf, kubernetesDriverSpec, driverServiceIngressList, configMapSpecs);
+ kubernetesDriverConf,
+ kubernetesDriverSpec,
+ driverServiceIngressList,
+ driverHttpRouteList,
+ driverGrpcRouteList,
+ configMapSpecs);
}
/**
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/DriverGatewayRouteUtils.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/DriverGatewayRouteUtils.java
new file mode 100644
index 0000000..f880f7b
--- /dev/null
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/DriverGatewayRouteUtils.java
@@ -0,0 +1,155 @@
+/*
+ * 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.utils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.Service;
+import io.fabric8.kubernetes.api.model.ServiceBuilder;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRoute;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRouteBuilder;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRouteSpec;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRouteSpecBuilder;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRoute;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRouteBuilder;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRouteSpec;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRouteSpecBuilder;
+
+import org.apache.spark.k8s.operator.spec.DriverGrpcRouteSpec;
+import org.apache.spark.k8s.operator.spec.DriverHttpRouteSpec;
+
+/**
+ * Utility class for building Gateway API {@code HTTPRoute} / {@code GRPCRoute} resources for a
+ * driver.
+ */
+public final class DriverGatewayRouteUtils {
+ private DriverGatewayRouteUtils() {}
+
+ /**
+ * Builds the Service + {@code HTTPRoute} resources for a driver.
+ *
+ * @param spec the {@link DriverHttpRouteSpec}.
+ * @param driverPodMetaData the {@link ObjectMeta} of the driver pod.
+ * @return a {@link List} containing the Service followed by the HTTPRoute.
+ */
+ public static List<HasMetadata> buildHttpRouteService(
+ DriverHttpRouteSpec spec, ObjectMeta driverPodMetaData) {
+ List<HasMetadata> resources = new ArrayList<>(2);
+ Service service =
+ buildService(spec.getServiceMetadata(), spec.getServiceSpec(), driverPodMetaData);
+ resources.add(service);
+ resources.add(buildHttpRoute(spec, service));
+ return resources;
+ }
+
+ /**
+ * Builds the Service + {@code GRPCRoute} resources for a driver.
+ *
+ * @param spec the {@link DriverGrpcRouteSpec}.
+ * @param driverPodMetaData the {@link ObjectMeta} of the driver pod.
+ * @return a {@link List} containing the Service followed by the GRPCRoute.
+ */
+ public static List<HasMetadata> buildGrpcRouteService(
+ DriverGrpcRouteSpec spec, ObjectMeta driverPodMetaData) {
+ List<HasMetadata> resources = new ArrayList<>(2);
+ Service service =
+ buildService(spec.getServiceMetadata(), spec.getServiceSpec(), driverPodMetaData);
+ resources.add(service);
+ resources.add(buildGrpcRoute(spec, service));
+ return resources;
+ }
+
+ private static Service buildService(
+ ObjectMeta serviceMetadata,
+ io.fabric8.kubernetes.api.model.ServiceSpec serviceSpec,
+ ObjectMeta driverPodMetaData) {
+ ObjectMeta serviceMeta = new ObjectMetaBuilder(serviceMetadata).build();
+ serviceMeta.setNamespace(driverPodMetaData.getNamespace());
+ Map<String, String> selectors = serviceSpec.getSelector();
+ if (selectors == null || selectors.isEmpty()) {
+ selectors = driverPodMetaData.getLabels();
+ }
+ return new ServiceBuilder()
+ .withMetadata(serviceMeta)
+ .withNewSpecLike(serviceSpec)
+ .withSelector(selectors)
+ .endSpec()
+ .build();
+ }
+
+ private static HTTPRoute buildHttpRoute(DriverHttpRouteSpec spec, Service service) {
+ ObjectMeta metadata = new ObjectMetaBuilder(spec.getHttpRouteMetadata()).build();
+ HTTPRouteSpec httpRouteSpec =
+ spec.getHttpRouteSpec() == null
+ ? new HTTPRouteSpec()
+ : new HTTPRouteSpecBuilder(spec.getHttpRouteSpec()).build();
+ if ((httpRouteSpec.getRules() == null || httpRouteSpec.getRules().isEmpty())
+ && service.getSpec().getPorts() != null
+ && !service.getSpec().getPorts().isEmpty()) {
+ httpRouteSpec =
+ new HTTPRouteSpecBuilder(httpRouteSpec)
+ .addNewRule()
+ .addNewMatch()
+ .withNewPath()
+ .withType("PathPrefix")
+ .withValue("/")
+ .endPath()
+ .endMatch()
+ .addNewBackendRef()
+ .withGroup("")
+ .withKind("Service")
+ .withName(service.getMetadata().getName())
+ .withPort(service.getSpec().getPorts().get(0).getPort())
+ .endBackendRef()
+ .endRule()
+ .build();
+ }
+ return new HTTPRouteBuilder().withMetadata(metadata).withSpec(httpRouteSpec).build();
+ }
+
+ private static GRPCRoute buildGrpcRoute(DriverGrpcRouteSpec spec, Service service) {
+ ObjectMeta metadata = new ObjectMetaBuilder(spec.getGrpcRouteMetadata()).build();
+ GRPCRouteSpec grpcRouteSpec =
+ spec.getGrpcRouteSpec() == null
+ ? new GRPCRouteSpec()
+ : new GRPCRouteSpecBuilder(spec.getGrpcRouteSpec()).build();
+ if ((grpcRouteSpec.getRules() == null || grpcRouteSpec.getRules().isEmpty())
+ && service.getSpec().getPorts() != null
+ && !service.getSpec().getPorts().isEmpty()) {
+ grpcRouteSpec =
+ new GRPCRouteSpecBuilder(grpcRouteSpec)
+ .addNewRule()
+ .addNewBackendRef()
+ .withGroup("")
+ .withKind("Service")
+ .withName(service.getMetadata().getName())
+ .withPort(service.getSpec().getPorts().get(0).getPort())
+ .endBackendRef()
+ .endRule()
+ .build();
+ }
+ return new GRPCRouteBuilder().withMetadata(metadata).withSpec(grpcRouteSpec).build();
+ }
+}
diff --git a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppResourceSpecTest.java b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppResourceSpecTest.java
index 4ef3476..b241eaf 100644
--- a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppResourceSpecTest.java
+++ b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppResourceSpecTest.java
@@ -63,7 +63,7 @@ class SparkAppResourceSpecTest {
KubernetesDriverSpec.create(sparkPod, preResourceList, resourceList, Map.of());
SparkAppResourceSpec appResourceSpec =
- new SparkAppResourceSpec(mockConf, spec, List.of(), List.of());
+ new SparkAppResourceSpec(mockConf, spec, List.of(), List.of(), List.of(), List.of());
Assertions.assertEquals(2, appResourceSpec.getDriverResources().size());
Assertions.assertEquals(2, appResourceSpec.getDriverPreResources().size());
@@ -119,7 +119,7 @@ class SparkAppResourceSpecTest {
KubernetesDriverSpec.create(sparkPod, List.of(), List.of(), Map.of());
SparkAppResourceSpec appResourceSpec =
- new SparkAppResourceSpec(mockConf, spec, List.of(), List.of());
+ new SparkAppResourceSpec(mockConf, spec, List.of(), List.of(), List.of(), List.of());
Assertions.assertEquals(1, appResourceSpec.getDriverPreResources().size());
Assertions.assertEquals(NetworkPolicy.class,
diff --git a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/utils/DriverGatewayRouteUtilsTest.java b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/utils/DriverGatewayRouteUtilsTest.java
new file mode 100644
index 0000000..f182a01
--- /dev/null
+++ b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/utils/DriverGatewayRouteUtilsTest.java
@@ -0,0 +1,269 @@
+/*
+ * 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.utils;
+
+import java.util.List;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.IntOrString;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.Service;
+import io.fabric8.kubernetes.api.model.ServiceSpec;
+import io.fabric8.kubernetes.api.model.ServiceSpecBuilder;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCBackendRef;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRoute;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRouteRule;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRouteSpec;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.GRPCRouteSpecBuilder;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPBackendRef;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRoute;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRouteRule;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRouteSpec;
+import io.fabric8.kubernetes.api.model.gatewayapi.v1.HTTPRouteSpecBuilder;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import org.apache.spark.k8s.operator.spec.DriverGrpcRouteSpec;
+import org.apache.spark.k8s.operator.spec.DriverHttpRouteSpec;
+
+class DriverGatewayRouteUtilsTest {
+
+ @Test
+ void buildHttpRouteServiceWithDefaultRule() {
+ // SPARK-58160: default rule targets the Service's first port when user omits rules.
+ ObjectMeta driverMeta =
+ new ObjectMetaBuilder()
+ .withName("foo-driver")
+ .withNamespace("foo-ns")
+ .addToLabels("foo", "bar")
+ .build();
+ DriverHttpRouteSpec spec = buildHttpSpecWithSinglePortService();
+
+ List<HasMetadata> resources = DriverGatewayRouteUtils.buildHttpRouteService(spec, driverMeta);
+
+ Assertions.assertEquals(2, resources.size());
+ Assertions.assertInstanceOf(Service.class, resources.get(0));
+ Assertions.assertInstanceOf(HTTPRoute.class, resources.get(1));
+
+ Service service = (Service) resources.get(0);
+ Assertions.assertEquals(driverMeta.getLabels(), service.getSpec().getSelector());
+ Assertions.assertEquals(driverMeta.getNamespace(), service.getMetadata().getNamespace());
+
+ HTTPRoute route = (HTTPRoute) resources.get(1);
+ Assertions.assertEquals(1, route.getSpec().getRules().size());
+ HTTPRouteRule rule = route.getSpec().getRules().get(0);
+ Assertions.assertEquals(1, rule.getMatches().size());
+ Assertions.assertEquals("PathPrefix", rule.getMatches().get(0).getPath().getType());
+ Assertions.assertEquals("/", rule.getMatches().get(0).getPath().getValue());
+ Assertions.assertEquals(1, rule.getBackendRefs().size());
+ HTTPBackendRef backend = rule.getBackendRefs().get(0);
+ Assertions.assertEquals(service.getMetadata().getName(), backend.getName());
+ Assertions.assertEquals("Service", backend.getKind());
+ Assertions.assertEquals(service.getSpec().getPorts().get(0).getPort(), backend.getPort());
+ }
+
+ @Test
+ void buildHttpRouteServiceWithUserProvidedRules() {
+ // SPARK-58160: user-provided HTTPRouteSpec rules pass through unchanged.
+ ObjectMeta driverMeta =
+ new ObjectMetaBuilder().withName("foo-driver").withNamespace("foo-ns").build();
+ DriverHttpRouteSpec spec = buildHttpSpecWithRules();
+
+ List<HasMetadata> resources = DriverGatewayRouteUtils.buildHttpRouteService(spec, driverMeta);
+
+ Assertions.assertEquals(2, resources.size());
+ HTTPRoute route = (HTTPRoute) resources.get(1);
+ // User-provided rules pass through unchanged.
+ Assertions.assertEquals(spec.getHttpRouteSpec().getRules(), route.getSpec().getRules());
+ Assertions.assertEquals(spec.getHttpRouteSpec().getHostnames(), route.getSpec().getHostnames());
+ Assertions.assertEquals(
+ spec.getHttpRouteSpec().getParentRefs(), route.getSpec().getParentRefs());
+ }
+
+ @Test
+ void buildGrpcRouteServiceWithDefaultRule() {
+ // SPARK-58160: GRPCRoute default rule targets the Service's first port.
+ ObjectMeta driverMeta =
+ new ObjectMetaBuilder()
+ .withName("foo-driver")
+ .withNamespace("foo-ns")
+ .addToLabels("foo", "bar")
+ .build();
+ DriverGrpcRouteSpec spec = buildGrpcSpecWithSinglePortService();
+
+ List<HasMetadata> resources = DriverGatewayRouteUtils.buildGrpcRouteService(spec, driverMeta);
+
+ Assertions.assertEquals(2, resources.size());
+ Assertions.assertInstanceOf(Service.class, resources.get(0));
+ Assertions.assertInstanceOf(GRPCRoute.class, resources.get(1));
+
+ Service service = (Service) resources.get(0);
+ Assertions.assertEquals(driverMeta.getLabels(), service.getSpec().getSelector());
+ Assertions.assertEquals(driverMeta.getNamespace(), service.getMetadata().getNamespace());
+
+ GRPCRoute route = (GRPCRoute) resources.get(1);
+ Assertions.assertEquals(1, route.getSpec().getRules().size());
+ GRPCRouteRule rule = route.getSpec().getRules().get(0);
+ Assertions.assertEquals(1, rule.getBackendRefs().size());
+ GRPCBackendRef backend = rule.getBackendRefs().get(0);
+ Assertions.assertEquals(service.getMetadata().getName(), backend.getName());
+ Assertions.assertEquals("Service", backend.getKind());
+ Assertions.assertEquals(service.getSpec().getPorts().get(0).getPort(), backend.getPort());
+ }
+
+ @Test
+ void buildGrpcRouteServiceWithUserProvidedRules() {
+ // SPARK-58160: user-provided GRPCRouteSpec rules pass through unchanged.
+ ObjectMeta driverMeta =
+ new ObjectMetaBuilder().withName("foo-driver").withNamespace("foo-ns").build();
+ DriverGrpcRouteSpec spec = buildGrpcSpecWithRules();
+
+ List<HasMetadata> resources = DriverGatewayRouteUtils.buildGrpcRouteService(spec, driverMeta);
+
+ Assertions.assertEquals(2, resources.size());
+ GRPCRoute route = (GRPCRoute) resources.get(1);
+ Assertions.assertEquals(spec.getGrpcRouteSpec().getRules(), route.getSpec().getRules());
+ Assertions.assertEquals(
+ spec.getGrpcRouteSpec().getParentRefs(), route.getSpec().getParentRefs());
+ }
+
+ private DriverHttpRouteSpec buildHttpSpecWithSinglePortService() {
+ ServiceSpec serviceSpec =
+ new ServiceSpecBuilder()
+ .addNewPort()
+ .withPort(80)
+ .withProtocol("TCP")
+ .withTargetPort(new IntOrString(4040))
+ .endPort()
+ .withType("ClusterIP")
+ .build();
+ ObjectMeta serviceMeta = new ObjectMetaBuilder().withName("foo-ui-svc").build();
+ ObjectMeta routeMeta = new ObjectMetaBuilder().withName("foo-ui-route").build();
+ return DriverHttpRouteSpec.builder()
+ .serviceMetadata(serviceMeta)
+ .serviceSpec(serviceSpec)
+ .httpRouteMetadata(routeMeta)
+ .build();
+ }
+
+ private DriverHttpRouteSpec buildHttpSpecWithRules() {
+ ServiceSpec serviceSpec =
+ new ServiceSpecBuilder()
+ .addNewPort()
+ .withPort(80)
+ .withProtocol("TCP")
+ .withTargetPort(new IntOrString(4040))
+ .endPort()
+ .withType("ClusterIP")
+ .build();
+ HTTPRouteSpec httpRouteSpec =
+ new HTTPRouteSpecBuilder()
+ .withHostnames("spark.example.com")
+ .addNewParentRef()
+ .withName("shared-gateway")
+ .withNamespace("gateway-system")
+ .withSectionName("https")
+ .endParentRef()
+ .addNewRule()
+ .addNewMatch()
+ .withNewPath()
+ .withType("PathPrefix")
+ .withValue("/spark-app")
+ .endPath()
+ .endMatch()
+ .addNewBackendRef()
+ .withGroup("")
+ .withKind("Service")
+ .withName("foo-ui-svc")
+ .withPort(80)
+ .endBackendRef()
+ .endRule()
+ .build();
+ ObjectMeta serviceMeta = new ObjectMetaBuilder().withName("foo-ui-svc").build();
+ ObjectMeta routeMeta = new ObjectMetaBuilder().withName("foo-ui-route").build();
+ return DriverHttpRouteSpec.builder()
+ .serviceMetadata(serviceMeta)
+ .serviceSpec(serviceSpec)
+ .httpRouteMetadata(routeMeta)
+ .httpRouteSpec(httpRouteSpec)
+ .build();
+ }
+
+ private DriverGrpcRouteSpec buildGrpcSpecWithSinglePortService() {
+ ServiceSpec serviceSpec =
+ new ServiceSpecBuilder()
+ .addNewPort()
+ .withPort(15002)
+ .withProtocol("TCP")
+ .withTargetPort(new IntOrString(15002))
+ .endPort()
+ .withType("ClusterIP")
+ .build();
+ ObjectMeta serviceMeta = new ObjectMetaBuilder().withName("foo-connect-svc").build();
+ ObjectMeta routeMeta = new ObjectMetaBuilder().withName("foo-connect-route").build();
+ return DriverGrpcRouteSpec.builder()
+ .serviceMetadata(serviceMeta)
+ .serviceSpec(serviceSpec)
+ .grpcRouteMetadata(routeMeta)
+ .build();
+ }
+
+ private DriverGrpcRouteSpec buildGrpcSpecWithRules() {
+ ServiceSpec serviceSpec =
+ new ServiceSpecBuilder()
+ .addNewPort()
+ .withPort(15002)
+ .withProtocol("TCP")
+ .withTargetPort(new IntOrString(15002))
+ .endPort()
+ .withType("ClusterIP")
+ .build();
+ GRPCRouteSpec grpcRouteSpec =
+ new GRPCRouteSpecBuilder()
+ .addNewParentRef()
+ .withName("shared-gateway")
+ .withNamespace("gateway-system")
+ .withSectionName("grpc")
+ .endParentRef()
+ .addNewRule()
+ .addNewMatch()
+ .editOrNewMethod()
+ .withService("spark.connect.SparkConnectService")
+ .endMethod()
+ .endMatch()
+ .addNewBackendRef()
+ .withGroup("")
+ .withKind("Service")
+ .withName("foo-connect-svc")
+ .withPort(15002)
+ .endBackendRef()
+ .endRule()
+ .build();
+ ObjectMeta serviceMeta = new ObjectMetaBuilder().withName("foo-connect-svc").build();
+ ObjectMeta routeMeta = new ObjectMetaBuilder().withName("foo-connect-route").build();
+ return DriverGrpcRouteSpec.builder()
+ .serviceMetadata(serviceMeta)
+ .serviceSpec(serviceSpec)
+ .grpcRouteMetadata(routeMeta)
+ .grpcRouteSpec(grpcRouteSpec)
+ .build();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org