diff --git a/.github/workflows/test-spark-examples.yaml b/.github/workflows/test-spark-examples.yaml index 8a1251717..c18d307a3 100644 --- a/.github/workflows/test-spark-examples.yaml +++ b/.github/workflows/test-spark-examples.yaml @@ -14,7 +14,7 @@ jobs: strategy: fail-fast: false matrix: - kubernetes-version: ["1.32.0"] + kubernetes-version: ["1.32.11", "1.33.7", "1.34.3", "1.35.0"] steps: - name: Free disk space (Ubuntu runner) @@ -48,41 +48,20 @@ jobs: curl -Lo ./kubectl "https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl" && chmod +x ./kubectl && sudo mv ./kubectl /usr/local/bin/kubectl curl -sSfL https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash - - name: Checkout spark-operator - uses: actions/checkout@v6 - with: - repository: kubeflow/spark-operator - path: spark-operator - fetch-depth: 1 - - - name: Build Spark Operator image - run: | - docker build -t ghcr.io/kubeflow/spark-operator/controller:local ./spark-operator - timeout-minutes: 10 - - name: Setup Kind cluster with Spark Operator run: | make test-e2e-setup-cluster K8S_VERSION=${{ matrix.kubernetes-version }} env: SPARK_TEST_CLUSTER: spark-test SPARK_TEST_NAMESPACE: spark-test - SPARK_OPERATOR_IMAGE_TAG: local timeout-minutes: 15 - - name: Build and load Spark E2E runner image (in-cluster) - run: | - docker build -f hack/Dockerfile.spark-e2e-runner -t spark-e2e-runner:local . - kind load docker-image spark-e2e-runner:local --name spark-test - timeout-minutes: 5 - - name: Run example validation tests run: uv run pytest test/e2e/spark/test_spark_examples.py -v --tb=short env: SPARK_TEST_CLUSTER: spark-test SPARK_TEST_NAMESPACE: spark-test SPARK_E2E_DEBUG: "1" - SPARK_E2E_RUN_IN_CLUSTER: "1" - SPARK_E2E_RUNNER_IMAGE: spark-e2e-runner:local timeout-minutes: 15 - name: Collect logs on failure @@ -90,4 +69,5 @@ jobs: run: | kubectl get pods -n spark-test kubectl get sparkconnect -n spark-test 2>/dev/null || true + kubectl get sparkapplication -n spark-test 2>/dev/null || true kubectl logs -n spark-test -l app.kubernetes.io/component=server --tail=100 || true diff --git a/examples/spark/README.md b/examples/spark/README.md index fc605cd1a..d4a2b9454 100644 --- a/examples/spark/README.md +++ b/examples/spark/README.md @@ -16,7 +16,6 @@ For the full documentation, see [Interactive Sessions](https://sdk.kubeflow.org/ - **spark_connect_simple.py** - Basic SparkClient usage with simple API - **spark_advanced_options.py** - Advanced configuration with Driver/Executor objects - **demo_existing_sparkconnect.py** - Connect to existing SparkConnect cluster -- **connect_existing_session.py** - Connect to an existing Spark Connect session through `base_url` - **test_connect_url.py** - Test URL-based connection to Spark Connect ### Batch Jobs diff --git a/examples/spark/connect_existing_session.py b/examples/spark/connect_existing_session.py deleted file mode 100644 index e08ed6bcd..000000000 --- a/examples/spark/connect_existing_session.py +++ /dev/null @@ -1,123 +0,0 @@ -#!/usr/bin/env python3 -# Copyright 2025 The Kubeflow Authors. -# -# Licensed 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. - -"""E2E Test: Connect to Existing SparkConnect Session (two-client pattern). - -This example demonstrates the "bring your own server" use case where: -1. A setup client creates a SparkConnect server -2. A test client connects to the existing server via base_url - -This validates the connect(base_url="sc://...") codepath which bypasses -session creation and directly connects to an existing Spark Connect server. - -Usage: - # Run in-cluster only (via K8s Job): - SPARK_E2E_RUN_IN_CLUSTER=1 python examples/spark/connect_existing_session.py -""" - -import os -import sys -import uuid - -from kubeflow.common.types import KubernetesBackendConfig -from kubeflow.spark import Name, SparkClient -from kubeflow.spark.backends.kubernetes.utils import build_service_url - - -def _backend_config(): - """Backend config; uses SPARK_TEST_NAMESPACE in CI.""" - return KubernetesBackendConfig(namespace=os.environ.get("SPARK_TEST_NAMESPACE", "spark-test")) - - -def _unique_session_name() -> str: - """Generate unique session name to avoid conflicts in E2E runs.""" - return f"connect-existing-{uuid.uuid4().hex[:8]}" - - -def test_connect_to_existing_session(): - """Test connect(base_url=...) with two clients. - - Two-client pattern: - - Setup client: creates SparkConnect server, stops SparkSession (server stays running) - - Test client: connects via base_url to the existing server - """ - print("=" * 70) - print("E2E: Connect to Existing SparkConnect Session") - print("=" * 70) - - session_name = _unique_session_name() - setup_client = None - test_spark = None - - try: - # Phase 1: Setup client creates SparkConnect server - print("\n[Phase 1] Creating SparkConnect server...") - setup_client = SparkClient(backend_config=_backend_config()) - setup_spark = setup_client.connect(options=[Name(session_name)], timeout=180) - - info = setup_client.get_session(session_name) - service_url = build_service_url(info) - print(f" Session: {session_name}") - print(f" URL: {service_url}") - - setup_spark.stop() - print(" Setup SparkSession stopped (server still running)") - - # Phase 2: Test client connects via base_url - print("\n[Phase 2] Connecting via base_url...") - test_client = SparkClient(backend_config=_backend_config()) - test_spark = test_client.connect(base_url=service_url) - print(" Connected successfully!") - - # Phase 3: Validate with Spark operations - print("\n[Phase 3] Validating...") - count = test_spark.range(100).count() - print(f" spark.range(100).count() = {count}") - assert count == 100, f"Expected 100, got {count}" - - print("\n[SUCCESS] connect(base_url=...) works correctly!") - - finally: - # Phase 4: Cleanup - print("\n[Phase 4] Cleanup...") - if test_spark: - try: - test_spark.stop() - except Exception as e: - print(f" Warning: {e}") - if setup_client: - try: - setup_client.delete_session(session_name) - print(f" Deleted {session_name}") - except Exception as e: - print(f" Warning: {e}") - - -def main(): - """Entry point for E2E test.""" - if os.environ.get("SPARK_E2E_RUN_IN_CLUSTER") != "1": - print("SKIP: Requires in-cluster execution (SPARK_E2E_RUN_IN_CLUSTER=1)") - sys.exit(0) - - try: - test_connect_to_existing_session() - sys.exit(0) - except Exception as e: - print(f"\nFailed: {e}") - sys.exit(1) - - -if __name__ == "__main__": - main() diff --git a/hack/Dockerfile.spark-e2e-runner b/hack/Dockerfile.spark-e2e-runner deleted file mode 100644 index 65a3569b0..000000000 --- a/hack/Dockerfile.spark-e2e-runner +++ /dev/null @@ -1,18 +0,0 @@ -# Copyright 2025 The Kubeflow Authors. -# In-cluster E2E runner: runs Spark example scripts inside the cluster so -# SparkClient uses the in-cluster URL (sc://...svc.cluster.local) and no port-forward. -FROM python:3.11-slim - -WORKDIR /app - -COPY pyproject.toml README.md LICENSE ./ -COPY kubeflow/ kubeflow/ -COPY examples/ examples/ - -RUN pip install --no-cache-dir .[spark] - -ENV SPARK_TEST_NAMESPACE=spark-test -ENV PYTHONUNBUFFERED=1 - -# Override with args to run a different example (e.g. spark_advanced_options.py) -CMD ["python", "examples/spark/spark_connect_simple.py"] diff --git a/hack/crds/sparkoperator.k8s.io_sparkconnects.yaml b/hack/crds/sparkoperator.k8s.io_sparkconnects.yaml deleted file mode 100644 index 3f12bd5dd..000000000 --- a/hack/crds/sparkoperator.k8s.io_sparkconnects.yaml +++ /dev/null @@ -1,809 +0,0 @@ ---- -apiVersion: apiextensions.k8s.io/v1 -kind: CustomResourceDefinition -metadata: - annotations: - api-approved.kubernetes.io: https://github.com/kubeflow/spark-operator/pull/1298 - controller-gen.kubebuilder.io/version: v0.17.1 - name: sparkconnects.sparkoperator.k8s.io -spec: - group: sparkoperator.k8s.io - names: - kind: SparkConnect - listKind: SparkConnectList - plural: sparkconnects - shortNames: - - sparkconn - singular: sparkconnect - scope: Namespaced - versions: - - additionalPrinterColumns: - - jsonPath: .metadata.creationTimestamp - name: Age - type: date - - jsonPath: .status.state - name: Status - type: string - - jsonPath: .status.server.podName - name: PodName - type: string - name: v1alpha1 - schema: - openAPIV3Schema: - description: SparkConnect is the Schema for the sparkconnections API. - properties: - apiVersion: - description: |- - APIVersion defines the versioned schema of this representation of an object. - Servers should convert recognized schemas to the latest internal value, and - may reject unrecognized values. - More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources - type: string - kind: - description: |- - Kind is a string value representing the REST resource this object represents. - Servers may infer this from the endpoint the client submits requests to. - Cannot be updated. - In CamelCase. - More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds - type: string - metadata: - type: object - spec: - description: SparkConnectSpec defines the desired state of SparkConnect. - properties: - dynamicAllocation: - description: |- - DynamicAllocation configures dynamic allocation that becomes available for the Kubernetes - scheduler backend since Spark 3.0. - properties: - enabled: - description: Enabled controls whether dynamic allocation is enabled - or not. - type: boolean - initialExecutors: - description: |- - InitialExecutors is the initial number of executors to request. If .spec.executor.instances - is also set, the initial number of executors is set to the bigger of that and this option. - format: int32 - type: integer - maxExecutors: - description: MaxExecutors is the upper bound for the number of - executors if dynamic allocation is enabled. - format: int32 - type: integer - minExecutors: - description: MinExecutors is the lower bound for the number of - executors if dynamic allocation is enabled. - format: int32 - type: integer - shuffleTrackingEnabled: - description: |- - ShuffleTrackingEnabled enables shuffle file tracking for executors, which allows dynamic allocation without - the need for an external shuffle service. This option will try to keep alive executors that are storing - shuffle data for active jobs. If external shuffle service is enabled, set ShuffleTrackingEnabled to false. - ShuffleTrackingEnabled is true by default if dynamicAllocation.enabled is true. - type: boolean - shuffleTrackingTimeout: - description: |- - ShuffleTrackingTimeout controls the timeout in milliseconds for executors that are holding - shuffle data if shuffle tracking is enabled (true by default if dynamic allocation is enabled). - format: int64 - type: integer - type: object - executor: - description: Executor is the Spark executor specification. - properties: - cores: - description: Cores maps to `spark.driver.cores` or `spark.executor.cores` - for the driver and executors, respectively. - format: int32 - minimum: 1 - type: integer - instances: - description: Instances is the number of executor instances. - format: int32 - minimum: 0 - type: integer - memory: - description: Memory is the amount of memory to request for the - pod. - type: string - template: - description: |- - Template is a pod template that can be used to define the driver or executor pod configurations that Spark configurations do not support. - Spark version >= 3.0.0 is required. - Ref: https://spark.apache.org/docs/latest/running-on-kubernetes.html#pod-template. - type: object - x-kubernetes-preserve-unknown-fields: true - type: object - hadoopConf: - additionalProperties: - type: string - description: |- - HadoopConf carries user-specified Hadoop configuration properties as they would use the "--conf" option - in spark-submit. The SparkApplication controller automatically adds prefix "spark.hadoop." to Hadoop - configuration properties. - type: object - image: - description: |- - Image is the container image for the driver, executor, and init-container. Any custom container images for the - driver, executor, or init-container takes precedence over this. - type: string - server: - description: Server is the Spark connect server specification. - properties: - cores: - description: Cores maps to `spark.driver.cores` or `spark.executor.cores` - for the driver and executors, respectively. - format: int32 - minimum: 1 - type: integer - memory: - description: Memory is the amount of memory to request for the - pod. - type: string - service: - description: Service exposes the Spark connect server. - properties: - apiVersion: - description: |- - APIVersion defines the versioned schema of this representation of an object. - Servers should convert recognized schemas to the latest internal value, and - may reject unrecognized values. - More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources - type: string - kind: - description: |- - Kind is a string value representing the REST resource this object represents. - Servers may infer this from the endpoint the client submits requests to. - Cannot be updated. - In CamelCase. - More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds - type: string - metadata: - description: |- - Standard object's metadata. - More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata - properties: - annotations: - additionalProperties: - type: string - type: object - finalizers: - items: - type: string - type: array - labels: - additionalProperties: - type: string - type: object - name: - type: string - namespace: - type: string - type: object - spec: - description: |- - Spec defines the behavior of a service. - https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#spec-and-status - properties: - allocateLoadBalancerNodePorts: - description: |- - allocateLoadBalancerNodePorts defines if NodePorts will be automatically - allocated for services with type LoadBalancer. Default is "true". It - may be set to "false" if the cluster load-balancer does not rely on - NodePorts. If the caller requests specific NodePorts (by specifying a - value), those requests will be respected, regardless of this field. - This field may only be set for services with type LoadBalancer and will - be cleared if the type is changed to any other type. - type: boolean - clusterIP: - description: |- - clusterIP is the IP address of the service and is usually assigned - randomly. If an address is specified manually, is in-range (as per - system configuration), and is not in use, it will be allocated to the - service; otherwise creation of the service will fail. This field may not - be changed through updates unless the type field is also being changed - to ExternalName (which requires this field to be blank) or the type - field is being changed from ExternalName (in which case this field may - optionally be specified, as describe above). Valid values are "None", - empty string (""), or a valid IP address. Setting this to "None" makes a - "headless service" (no virtual IP), which is useful when direct endpoint - connections are preferred and proxying is not required. Only applies to - types ClusterIP, NodePort, and LoadBalancer. If this field is specified - when creating a Service of type ExternalName, creation will fail. This - field will be wiped when updating a Service to type ExternalName. - More info: https://kubernetes.io/docs/concepts/services-networking/service/#virtual-ips-and-service-proxies - type: string - clusterIPs: - description: |- - ClusterIPs is a list of IP addresses assigned to this service, and are - usually assigned randomly. If an address is specified manually, is - in-range (as per system configuration), and is not in use, it will be - allocated to the service; otherwise creation of the service will fail. - This field may not be changed through updates unless the type field is - also being changed to ExternalName (which requires this field to be - empty) or the type field is being changed from ExternalName (in which - case this field may optionally be specified, as describe above). Valid - values are "None", empty string (""), or a valid IP address. Setting - this to "None" makes a "headless service" (no virtual IP), which is - useful when direct endpoint connections are preferred and proxying is - not required. Only applies to types ClusterIP, NodePort, and - LoadBalancer. If this field is specified when creating a Service of type - ExternalName, creation will fail. This field will be wiped when updating - a Service to type ExternalName. If this field is not specified, it will - be initialized from the clusterIP field. If this field is specified, - clients must ensure that clusterIPs[0] and clusterIP have the same - value. - - This field may hold a maximum of two entries (dual-stack IPs, in either order). - These IPs must correspond to the values of the ipFamilies field. Both - clusterIPs and ipFamilies are governed by the ipFamilyPolicy field. - More info: https://kubernetes.io/docs/concepts/services-networking/service/#virtual-ips-and-service-proxies - items: - type: string - type: array - x-kubernetes-list-type: atomic - externalIPs: - description: |- - externalIPs is a list of IP addresses for which nodes in the cluster - will also accept traffic for this service. These IPs are not managed by - Kubernetes. The user is responsible for ensuring that traffic arrives - at a node with this IP. A common example is external load-balancers - that are not part of the Kubernetes system. - items: - type: string - type: array - x-kubernetes-list-type: atomic - externalName: - description: |- - externalName is the external reference that discovery mechanisms will - return as an alias for this service (e.g. a DNS CNAME record). No - proxying will be involved. Must be a lowercase RFC-1123 hostname - (https://tools.ietf.org/html/rfc1123) and requires `type` to be "ExternalName". - type: string - externalTrafficPolicy: - description: |- - externalTrafficPolicy describes how nodes distribute service traffic they - receive on one of the Service's "externally-facing" addresses (NodePorts, - ExternalIPs, and LoadBalancer IPs). If set to "Local", the proxy will configure - the service in a way that assumes that external load balancers will take care - of balancing the service traffic between nodes, and so each node will deliver - traffic only to the node-local endpoints of the service, without masquerading - the client source IP. (Traffic mistakenly sent to a node with no endpoints will - be dropped.) The default value, "Cluster", uses the standard behavior of - routing to all endpoints evenly (possibly modified by topology and other - features). Note that traffic sent to an External IP or LoadBalancer IP from - within the cluster will always get "Cluster" semantics, but clients sending to - a NodePort from within the cluster may need to take traffic policy into account - when picking a node. - type: string - healthCheckNodePort: - description: |- - healthCheckNodePort specifies the healthcheck nodePort for the service. - This only applies when type is set to LoadBalancer and - externalTrafficPolicy is set to Local. If a value is specified, is - in-range, and is not in use, it will be used. If not specified, a value - will be automatically allocated. External systems (e.g. load-balancers) - can use this port to determine if a given node holds endpoints for this - service or not. If this field is specified when creating a Service - which does not need it, creation will fail. This field will be wiped - when updating a Service to no longer need it (e.g. changing type). - This field cannot be updated once set. - format: int32 - type: integer - internalTrafficPolicy: - description: |- - InternalTrafficPolicy describes how nodes distribute service traffic they - receive on the ClusterIP. If set to "Local", the proxy will assume that pods - only want to talk to endpoints of the service on the same node as the pod, - dropping the traffic if there are no local endpoints. The default value, - "Cluster", uses the standard behavior of routing to all endpoints evenly - (possibly modified by topology and other features). - type: string - ipFamilies: - description: |- - IPFamilies is a list of IP families (e.g. IPv4, IPv6) assigned to this - service. This field is usually assigned automatically based on cluster - configuration and the ipFamilyPolicy field. If this field is specified - manually, the requested family is available in the cluster, - and ipFamilyPolicy allows it, it will be used; otherwise creation of - the service will fail. This field is conditionally mutable: it allows - for adding or removing a secondary IP family, but it does not allow - changing the primary IP family of the Service. Valid values are "IPv4" - and "IPv6". This field only applies to Services of types ClusterIP, - NodePort, and LoadBalancer, and does apply to "headless" services. - This field will be wiped when updating a Service to type ExternalName. - - This field may hold a maximum of two entries (dual-stack families, in - either order). These families must correspond to the values of the - clusterIPs field, if specified. Both clusterIPs and ipFamilies are - governed by the ipFamilyPolicy field. - items: - description: |- - IPFamily represents the IP Family (IPv4 or IPv6). This type is used - to express the family of an IP expressed by a type (e.g. service.spec.ipFamilies). - type: string - type: array - x-kubernetes-list-type: atomic - ipFamilyPolicy: - description: |- - IPFamilyPolicy represents the dual-stack-ness requested or required by - this Service. If there is no value provided, then this field will be set - to SingleStack. Services can be "SingleStack" (a single IP family), - "PreferDualStack" (two IP families on dual-stack configured clusters or - a single IP family on single-stack clusters), or "RequireDualStack" - (two IP families on dual-stack configured clusters, otherwise fail). The - ipFamilies and clusterIPs fields depend on the value of this field. This - field will be wiped when updating a service to type ExternalName. - type: string - loadBalancerClass: - description: |- - loadBalancerClass is the class of the load balancer implementation this Service belongs to. - If specified, the value of this field must be a label-style identifier, with an optional prefix, - e.g. "internal-vip" or "example.com/internal-vip". Unprefixed names are reserved for end-users. - This field can only be set when the Service type is 'LoadBalancer'. If not set, the default load - balancer implementation is used, today this is typically done through the cloud provider integration, - but should apply for any default implementation. If set, it is assumed that a load balancer - implementation is watching for Services with a matching class. Any default load balancer - implementation (e.g. cloud providers) should ignore Services that set this field. - This field can only be set when creating or updating a Service to type 'LoadBalancer'. - Once set, it can not be changed. This field will be wiped when a service is updated to a non 'LoadBalancer' type. - type: string - loadBalancerIP: - description: |- - Only applies to Service Type: LoadBalancer. - This feature depends on whether the underlying cloud-provider supports specifying - the loadBalancerIP when a load balancer is created. - This field will be ignored if the cloud-provider does not support the feature. - Deprecated: This field was under-specified and its meaning varies across implementations. - Using it is non-portable and it may not support dual-stack. - Users are encouraged to use implementation-specific annotations when available. - type: string - loadBalancerSourceRanges: - description: |- - If specified and supported by the platform, this will restrict traffic through the cloud-provider - load-balancer will be restricted to the specified client IPs. This field will be ignored if the - cloud-provider does not support the feature." - More info: https://kubernetes.io/docs/tasks/access-application-cluster/create-external-load-balancer/ - items: - type: string - type: array - x-kubernetes-list-type: atomic - ports: - description: |- - The list of ports that are exposed by this service. - More info: https://kubernetes.io/docs/concepts/services-networking/service/#virtual-ips-and-service-proxies - items: - description: ServicePort contains information on service's - port. - properties: - appProtocol: - description: |- - The application protocol for this port. - This is used as a hint for implementations to offer richer behavior for protocols that they understand. - This field follows standard Kubernetes label syntax. - Valid values are either: - - * Un-prefixed protocol names - reserved for IANA standard service names (as per - RFC-6335 and https://www.iana.org/assignments/service-names). - - * Kubernetes-defined prefixed names: - * 'kubernetes.io/h2c' - HTTP/2 prior knowledge over cleartext as described in https://www.rfc-editor.org/rfc/rfc9113.html#name-starting-http-2-with-prior- - * 'kubernetes.io/ws' - WebSocket over cleartext as described in https://www.rfc-editor.org/rfc/rfc6455 - * 'kubernetes.io/wss' - WebSocket over TLS as described in https://www.rfc-editor.org/rfc/rfc6455 - - * Other protocols should use implementation-defined prefixed names such as - mycompany.com/my-custom-protocol. - type: string - name: - description: |- - The name of this port within the service. This must be a DNS_LABEL. - All ports within a ServiceSpec must have unique names. When considering - the endpoints for a Service, this must match the 'name' field in the - EndpointPort. - Optional if only one ServicePort is defined on this service. - type: string - nodePort: - description: |- - The port on each node on which this service is exposed when type is - NodePort or LoadBalancer. Usually assigned by the system. If a value is - specified, in-range, and not in use it will be used, otherwise the - operation will fail. If not specified, a port will be allocated if this - Service requires one. If this field is specified when creating a - Service which does not need it, creation will fail. This field will be - wiped when updating a Service to no longer need it (e.g. changing type - from NodePort to ClusterIP). - More info: https://kubernetes.io/docs/concepts/services-networking/service/#type-nodeport - format: int32 - type: integer - port: - description: The port that will be exposed by this - service. - format: int32 - type: integer - protocol: - default: TCP - description: |- - The IP protocol for this port. Supports "TCP", "UDP", and "SCTP". - Default is TCP. - type: string - targetPort: - anyOf: - - type: integer - - type: string - description: |- - Number or name of the port to access on the pods targeted by the service. - Number must be in the range 1 to 65535. Name must be an IANA_SVC_NAME. - If this is a string, it will be looked up as a named port in the - target Pod's container ports. If this is not specified, the value - of the 'port' field is used (an identity map). - This field is ignored for services with clusterIP=None, and should be - omitted or set equal to the 'port' field. - More info: https://kubernetes.io/docs/concepts/services-networking/service/#defining-a-service - x-kubernetes-int-or-string: true - required: - - port - type: object - type: array - x-kubernetes-list-map-keys: - - port - - protocol - x-kubernetes-list-type: map - publishNotReadyAddresses: - description: |- - publishNotReadyAddresses indicates that any agent which deals with endpoints for this - Service should disregard any indications of ready/not-ready. - The primary use case for setting this field is for a StatefulSet's Headless Service to - propagate SRV DNS records for its Pods for the purpose of peer discovery. - The Kubernetes controllers that generate Endpoints and EndpointSlice resources for - Services interpret this to mean that all endpoints are considered "ready" even if the - Pods themselves are not. Agents which consume only Kubernetes generated endpoints - through the Endpoints or EndpointSlice resources can safely assume this behavior. - type: boolean - selector: - additionalProperties: - type: string - description: |- - Route service traffic to pods with label keys and values matching this - selector. If empty or not present, the service is assumed to have an - external process managing its endpoints, which Kubernetes will not - modify. Only applies to types ClusterIP, NodePort, and LoadBalancer. - Ignored if type is ExternalName. - More info: https://kubernetes.io/docs/concepts/services-networking/service/ - type: object - x-kubernetes-map-type: atomic - sessionAffinity: - description: |- - Supports "ClientIP" and "None". Used to maintain session affinity. - Enable client IP based session affinity. - Must be ClientIP or None. - Defaults to None. - More info: https://kubernetes.io/docs/concepts/services-networking/service/#virtual-ips-and-service-proxies - type: string - sessionAffinityConfig: - description: sessionAffinityConfig contains the configurations - of session affinity. - properties: - clientIP: - description: clientIP contains the configurations - of Client IP based session affinity. - properties: - timeoutSeconds: - description: |- - timeoutSeconds specifies the seconds of ClientIP type session sticky time. - The value must be >0 && <=86400(for 1 day) if ServiceAffinity == "ClientIP". - Default value is 10800(for 3 hours). - format: int32 - type: integer - type: object - type: object - trafficDistribution: - description: |- - TrafficDistribution offers a way to express preferences for how traffic is - distributed to Service endpoints. Implementations can use this field as a - hint, but are not required to guarantee strict adherence. If the field is - not set, the implementation will apply its default routing strategy. If set - to "PreferClose", implementations should prioritize endpoints that are - topologically close (e.g., same zone). - This is a beta field and requires enabling ServiceTrafficDistribution feature. - type: string - type: - description: |- - type determines how the Service is exposed. Defaults to ClusterIP. Valid - options are ExternalName, ClusterIP, NodePort, and LoadBalancer. - "ClusterIP" allocates a cluster-internal IP address for load-balancing - to endpoints. Endpoints are determined by the selector or if that is not - specified, by manual construction of an Endpoints object or - EndpointSlice objects. If clusterIP is "None", no virtual IP is - allocated and the endpoints are published as a set of endpoints rather - than a virtual IP. - "NodePort" builds on ClusterIP and allocates a port on every node which - routes to the same endpoints as the clusterIP. - "LoadBalancer" builds on NodePort and creates an external load-balancer - (if supported in the current cloud) which routes to the same endpoints - as the clusterIP. - "ExternalName" aliases this service to the specified externalName. - Several other fields do not apply to ExternalName services. - More info: https://kubernetes.io/docs/concepts/services-networking/service/#publishing-services-service-types - type: string - type: object - status: - description: |- - Most recently observed status of the service. - Populated by the system. - Read-only. - More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#spec-and-status - properties: - conditions: - description: Current service state - items: - description: Condition contains details for one aspect - of the current state of this API Resource. - properties: - lastTransitionTime: - description: |- - lastTransitionTime is the last time the condition transitioned from one status to another. - This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable. - format: date-time - type: string - message: - description: |- - message is a human readable message indicating details about the transition. - This may be an empty string. - maxLength: 32768 - type: string - observedGeneration: - description: |- - observedGeneration represents the .metadata.generation that the condition was set based upon. - For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date - with respect to the current state of the instance. - format: int64 - minimum: 0 - type: integer - reason: - description: |- - reason contains a programmatic identifier indicating the reason for the condition's last transition. - Producers of specific condition types may define expected values and meanings for this field, - and whether the values are considered a guaranteed API. - The value should be a CamelCase string. - This field may not be empty. - maxLength: 1024 - minLength: 1 - pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$ - type: string - status: - description: status of the condition, one of True, - False, Unknown. - enum: - - "True" - - "False" - - Unknown - type: string - type: - description: type of condition in CamelCase or in - foo.example.com/CamelCase. - maxLength: 316 - pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ - type: string - required: - - lastTransitionTime - - message - - reason - - status - - type - type: object - type: array - x-kubernetes-list-map-keys: - - type - x-kubernetes-list-type: map - loadBalancer: - description: |- - LoadBalancer contains the current status of the load-balancer, - if one is present. - properties: - ingress: - description: |- - Ingress is a list containing ingress points for the load-balancer. - Traffic intended for the service should be sent to these ingress points. - items: - description: |- - LoadBalancerIngress represents the status of a load-balancer ingress point: - traffic intended for the service should be sent to an ingress point. - properties: - hostname: - description: |- - Hostname is set for load-balancer ingress points that are DNS based - (typically AWS load-balancers) - type: string - ip: - description: |- - IP is set for load-balancer ingress points that are IP based - (typically GCE or OpenStack load-balancers) - type: string - ipMode: - description: |- - IPMode specifies how the load-balancer IP behaves, and may only be specified when the ip field is specified. - Setting this to "VIP" indicates that traffic is delivered to the node with - the destination set to the load-balancer's IP and port. - Setting this to "Proxy" indicates that traffic is delivered to the node or pod with - the destination set to the node's IP and node port or the pod's IP and port. - Service implementations may use this information to adjust traffic routing. - type: string - ports: - description: |- - Ports is a list of records of service ports - If used, every port defined in the service should have an entry in it - items: - description: PortStatus represents the error - condition of a service port - properties: - error: - description: |- - Error is to record the problem with the service port - The format of the error shall comply with the following rules: - - built-in error values shall be specified in this file and those shall use - CamelCase names - - cloud provider specific error values must have names that comply with the - format foo.example.com/CamelCase. - maxLength: 316 - pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ - type: string - port: - description: Port is the port number of - the service port of which status is - recorded here - format: int32 - type: integer - protocol: - description: |- - Protocol is the protocol of the service port of which status is recorded here - The supported values are: "TCP", "UDP", "SCTP" - type: string - required: - - error - - port - - protocol - type: object - type: array - x-kubernetes-list-type: atomic - type: object - type: array - x-kubernetes-list-type: atomic - type: object - type: object - type: object - template: - description: |- - Template is a pod template that can be used to define the driver or executor pod configurations that Spark configurations do not support. - Spark version >= 3.0.0 is required. - Ref: https://spark.apache.org/docs/latest/running-on-kubernetes.html#pod-template. - type: object - x-kubernetes-preserve-unknown-fields: true - type: object - sparkConf: - additionalProperties: - type: string - description: |- - SparkConf carries user-specified Spark configuration properties as they would use the "--conf" option in - spark-submit. - type: object - sparkVersion: - description: SparkVersion is the version of Spark the spark connect - use. - type: string - required: - - executor - - server - - sparkVersion - type: object - status: - description: SparkConnectStatus defines the observed state of SparkConnect. - properties: - conditions: - description: Represents the latest available observations of a SparkConnect's - current state. - items: - description: Condition contains details for one aspect of the current - state of this API Resource. - properties: - lastTransitionTime: - description: |- - lastTransitionTime is the last time the condition transitioned from one status to another. - This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable. - format: date-time - type: string - message: - description: |- - message is a human readable message indicating details about the transition. - This may be an empty string. - maxLength: 32768 - type: string - observedGeneration: - description: |- - observedGeneration represents the .metadata.generation that the condition was set based upon. - For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date - with respect to the current state of the instance. - format: int64 - minimum: 0 - type: integer - reason: - description: |- - reason contains a programmatic identifier indicating the reason for the condition's last transition. - Producers of specific condition types may define expected values and meanings for this field, - and whether the values are considered a guaranteed API. - The value should be a CamelCase string. - This field may not be empty. - maxLength: 1024 - minLength: 1 - pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$ - type: string - status: - description: status of the condition, one of True, False, Unknown. - enum: - - "True" - - "False" - - Unknown - type: string - type: - description: type of condition in CamelCase or in foo.example.com/CamelCase. - maxLength: 316 - pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ - type: string - required: - - lastTransitionTime - - message - - reason - - status - - type - type: object - type: array - x-kubernetes-list-map-keys: - - type - x-kubernetes-list-type: map - executors: - additionalProperties: - type: integer - description: Executors represents the current state of the SparkConnect - executors. - type: object - lastUpdateTime: - description: LastUpdateTime is the time at which the SparkConnect - controller last updated the SparkConnect. - format: date-time - type: string - server: - description: Server represents the current state of the SparkConnect - server. - properties: - podIp: - description: PodIP is the IP address of the pod that is running - the Spark Connect server. - type: string - podName: - description: PodName is the name of the pod that is running the - Spark Connect server. - type: string - serviceName: - description: ServiceName is the name of the service that is exposing - the Spark Connect server. - type: string - type: object - startTime: - description: StartTime is the time at which the SparkConnect controller - started processing the SparkConnect. - format: date-time - type: string - state: - description: State represents the current state of the SparkConnect. - type: string - type: object - required: - - metadata - - spec - type: object - served: true - storage: true - subresources: - status: {} diff --git a/hack/e2e-setup-cluster.sh b/hack/e2e-setup-cluster.sh index 0f37dd238..34ecfd29e 100755 --- a/hack/e2e-setup-cluster.sh +++ b/hack/e2e-setup-cluster.sh @@ -17,7 +17,7 @@ set -euo pipefail CLUSTER_NAME="${SPARK_TEST_CLUSTER:-spark-test}" NAMESPACE="${SPARK_TEST_NAMESPACE:-spark-test}" -SPARK_OPERATOR_VERSION="${SPARK_OPERATOR_VERSION:-2.1.0}" +SPARK_OPERATOR_VERSION="${SPARK_OPERATOR_VERSION:-2.5.0}" SPARK_OPERATOR_IMAGE_TAG="${SPARK_OPERATOR_IMAGE_TAG:-latest}" K8S_VERSION="${K8S_VERSION:-1.32.0}" KIND_BIN="${KIND:-kind}" @@ -136,6 +136,7 @@ install_spark_operator() { --set image.repository=kubeflow/spark-operator/controller --set "image.tag=$SPARK_OPERATOR_IMAGE_TAG" --set "spark.jobNamespaces[0]=$NAMESPACE" + --set "hook.upgradeCrd=true" ) local helm_timeout="${HELM_TIMEOUT:-15m}" log_info "Helm may take up to $helm_timeout. 'context canceled' usually means the process was killed (external timeout or Ctrl+C)." @@ -205,190 +206,24 @@ setup_test_namespace() { phase_end "setup_test_namespace" } -# 1) Operator: grant Spark Operator controller permission to manage SparkConnect and pods in the test namespace. -# 2) Driver: grant default SA in test namespace permission so the Spark Connect server (driver) can create/watch executor pods. -# 3) ClusterRole for endpointslices: Helm chart may not grant discovery.k8s.io/endpointslices; bind to controller so Service/EndpointSlice updates succeed. -ensure_sparkconnect_rbac() { - phase_start "ensure_sparkconnect_rbac" - log_info "Creating Role, RoleBinding, and ClusterRole for SparkConnect (namespace $NAMESPACE)" - - kubectl apply -f - </dev/null || true - local script_dir repo_root crds_dir - script_dir="$(cd "$(dirname "$0")" && pwd)" - repo_root="$(cd "$script_dir/.." && pwd)" - crds_dir="$repo_root/hack/crds" - if [[ ! -d "$crds_dir" ]]; then - log_error "CRDs dir not found: $crds_dir" - exit 1 - fi - for f in "$crds_dir"/*.yaml; do - [[ -f "$f" ]] || continue - log_info "Applying CRD: $(basename "$f")" - kubectl apply -f "$f" - done - phase_end "apply_crd_only" -} - print_status() { echo "" log_info "=== Cluster Status ===" echo "Cluster: $CLUSTER_NAME" echo "Kubernetes version: $K8S_VERSION" echo "Test namespace: $NAMESPACE" - if [[ "${E2E_CRD_ONLY:-0}" == "1" ]]; then - echo "Mode: CRD-only (no Spark Operator controller)" - echo "CRDs:" - kubectl get crd | grep sparkoperator || true - else - echo "Spark Operator version: $SPARK_OPERATOR_VERSION" - echo "Image tag: $SPARK_OPERATOR_IMAGE_TAG" - echo "" - echo "Spark Operator Deployment:" - kubectl get deployment -n spark-operator 2>/dev/null || true - fi + echo "Spark Operator version: $SPARK_OPERATOR_VERSION" + echo "Image tag: $SPARK_OPERATOR_IMAGE_TAG" + echo "" + echo "Spark Operator Deployment:" + kubectl get deployment -n spark-operator 2>/dev/null || true echo "" echo "Test Namespace Pods:" kubectl get pods -n "$NAMESPACE" 2>/dev/null || echo "No pods yet" echo "" log_info "=== Usage ===" - if [[ "${E2E_CRD_ONLY:-0}" == "1" ]]; then - echo "Smoke test: uv run pytest test/e2e/spark/test_spark_examples.py -v -k smoke" - else - echo "To run E2E tests:" - echo " python -m pytest test/e2e/spark/test_spark_examples.py -v" - fi + echo "To run E2E tests:" + echo " python -m pytest test/e2e/spark/test_spark_examples.py -v" echo "" echo "To delete cluster:" echo " make test-e2e-setup-cluster K8S_VERSION=$K8S_VERSION --delete" @@ -406,19 +241,13 @@ main() { check_prerequisites create_cluster setup_test_namespace - if [[ "${E2E_CRD_ONLY:-0}" == "1" ]]; then - apply_crd_only - else - ensure_sparkconnect_rbac - apply_sparkconnect_crd - if [[ "$SPARK_OPERATOR_IMAGE_TAG" == "local" ]]; then - phase_start "kind_load_local_image" - log_info "Loading locally built controller image into Kind..." - "$KIND_BIN" load docker-image "ghcr.io/kubeflow/spark-operator/controller:local" --name "$CLUSTER_NAME" - phase_end "kind_load_local_image" - fi - install_spark_operator + if [[ "$SPARK_OPERATOR_IMAGE_TAG" == "local" ]]; then + phase_start "kind_load_local_image" + log_info "Loading locally built controller image into Kind..." + "$KIND_BIN" load docker-image "ghcr.io/kubeflow/spark-operator/controller:local" --name "$CLUSTER_NAME" + phase_end "kind_load_local_image" fi + install_spark_operator print_status local total_elapsed total_elapsed=$(($(date +%s) - main_start)) diff --git a/test/e2e/spark/README.md b/test/e2e/spark/README.md index 0b6a64bf3..4b88c86cb 100644 --- a/test/e2e/spark/README.md +++ b/test/e2e/spark/README.md @@ -4,13 +4,21 @@ End-to-end tests that validate Spark examples execute correctly with Kubernetes ## Test Files -### **test_spark_examples.py** (3 tests) +### **test_spark_examples.py** -Validates that Spark example scripts execute successfully: +Validates that Spark example scripts execute successfully, run as subprocesses +against the ambient kubeconfig (no in-cluster Job execution): +- `test_spark_connect_crd_smoke` - Smoke test that the SparkConnect CRD is accepted by the API server - `test_spark_connect_simple_example` - Validates spark_connect_simple.py runs without errors - `test_spark_advanced_options_example` - Validates spark_advanced_options.py runs without errors -- `test_demo_existing_sparkconnect_example` - Validates demo_existing_sparkconnect.py structure (SKIPPED - requires manual port-forward) +- `test_batch_job_lifecycle_example` - Validates batch_job_lifecycle.py runs without errors +- `test_batch_failed_job_example` - Validates batch_failed_job.py handles failed Spark jobs +- `test_batch_func_job_lifecycle_example` - Validates batch_func_job_lifecycle.py runs without errors +- `test_batch_job_options_example` - Validates batch_job_options.py runs without errors + +A background cluster watcher (`cluster_watcher.py`) polls `SparkConnect`, +`SparkApplication`, pods, and events for diagnostics on failure. ## Prerequisites @@ -94,7 +102,7 @@ E2E tests are integrated into GitHub Actions and run automatically on pull reque - Manual workflow dispatch **Matrix:** -- Kubernetes versions: 1.30.0, 1.31.0, 1.32.3 +- Kubernetes versions: 1.32.11, 1.33.7, 1.34.3, 1.35.0 - Python version: 3.11 **Tests:** diff --git a/test/e2e/spark/cluster_watcher.py b/test/e2e/spark/cluster_watcher.py index feaf114e5..1321ec79d 100644 --- a/test/e2e/spark/cluster_watcher.py +++ b/test/e2e/spark/cluster_watcher.py @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Cluster watcher for E2E tests: logs SparkConnect, pods, events, driver logs.""" +"""Cluster watcher for E2E tests: logs SparkConnect/SparkApplication, pods, events, driver logs.""" import subprocess import threading @@ -41,6 +41,8 @@ def _snapshot(namespace: str, elapsed_sec: float) -> list[str]: lines = [f"--- T+{elapsed_sec:.0f}s ---"] sc = _run_kubectl(["get", "sparkconnect", "-o", "wide"], namespace) lines.append(f"SparkConnect:\n{sc}" if sc else "SparkConnect: (none)") + sa = _run_kubectl(["get", "sparkapplication", "-o", "wide"], namespace) + lines.append(f"SparkApplication:\n{sa}" if sa else "SparkApplication: (none)") pods = _run_kubectl(["get", "pods", "-o", "wide"], namespace) lines.append(f"Pods:\n{pods}" if pods else "Pods: (none)") events = _run_kubectl( @@ -58,15 +60,27 @@ def _snapshot(namespace: str, elapsed_sec: float) -> list[str]: return lines -def _driver_pod_from_sparkconnect(namespace: str) -> str | None: - out = _run_kubectl( +def _driver_pod_from_cluster(namespace: str) -> str | None: + """Resolve a driver pod name from either SparkConnect or SparkApplication status.""" + sc_out = _run_kubectl( ["get", "sparkconnect", "-o", "jsonpath={.items[*].status.server.podName}"], namespace, ) - if not out or out.startswith("(exit") or out.startswith("(error)"): - return None - pods = [p for p in out.strip().split() if p] - return pods[0] if pods else None + if sc_out and not sc_out.startswith("(exit") and not sc_out.startswith("(error)"): + pods = [p for p in sc_out.strip().split() if p] + if pods: + return pods[0] + + sa_out = _run_kubectl( + ["get", "sparkapplication", "-o", "jsonpath={.items[*].status.driverInfo.podName}"], + namespace, + ) + if sa_out and not sa_out.startswith("(exit") and not sa_out.startswith("(error)"): + pods = [p for p in sa_out.strip().split() if p] + if pods: + return pods[0] + + return None def _driver_logs(namespace: str, pod_name: str, tail: int = 25) -> str: @@ -83,9 +97,9 @@ def run_watcher( ) -> None: """Run cluster watcher until stop_event or max_duration; append to log_out. - Each interval: logs SparkConnect list, pods, events; when a driver pod - appears (from SparkConnect status), appends its logs so we see driver/ - executor startup and where time is spent. + Each interval: logs SparkConnect/SparkApplication list, pods, events; when + a driver pod appears (from SparkConnect or SparkApplication status), + appends its logs so we see driver/executor startup and where time is spent. """ start = time.monotonic() last_driver: str | None = None @@ -96,7 +110,7 @@ def run_watcher( break for line in _snapshot(namespace, elapsed): log_out.append(line) - driver_pod = _driver_pod_from_sparkconnect(namespace) + driver_pod = _driver_pod_from_cluster(namespace) if driver_pod and (log_driver_when_ready or driver_pod != last_driver): last_driver = driver_pod dr_logs = _driver_logs(namespace, driver_pod) diff --git a/test/e2e/spark/run_in_cluster.py b/test/e2e/spark/run_in_cluster.py deleted file mode 100644 index 7deb2c246..000000000 --- a/test/e2e/spark/run_in_cluster.py +++ /dev/null @@ -1,231 +0,0 @@ -# Copyright 2025 The Kubeflow Authors. -# -# Licensed 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. - -"""Run Spark example scripts inside the cluster as a Job (in-cluster URL, no port-forward).""" - -import contextlib -import os -import subprocess -import tempfile - - -def run_example_in_cluster( - example_script_name: str, - namespace: str, - image: str, - timeout_sec: int = 300, -) -> tuple[bool, str, str]: - """Run an example script in-cluster via a Kubernetes Job. - - The Job pod has KUBERNETES_SERVICE_HOST set, so SparkClient uses the - in-cluster URL (sc://...svc.cluster.local) and no port-forward. - - Args: - example_script_name: Script filename (e.g. spark_connect_simple.py). - namespace: Kubernetes namespace for the Job. - image: Container image that has the SDK and examples (e.g. spark-e2e-runner:local). - timeout_sec: Max time to wait for Job completion. - - Returns: - (success, combined_stdout_stderr, job_description). - """ - base = example_script_name.replace(".py", "").replace("_", "-") - job_name = f"spark-e2e-{base}"[:63].rstrip("-") - script_path = f"examples/spark/{example_script_name}" - # Job manifest: one pod, run the example script, default SA (has e2e-sparkconnect-client Role). - manifest = f""" -apiVersion: batch/v1 -kind: Job -metadata: - name: {job_name} - namespace: {namespace} -spec: - backoffLimit: 0 - activeDeadlineSeconds: {timeout_sec} - template: - spec: - restartPolicy: Never - containers: - - name: runner - image: {image} - imagePullPolicy: IfNotPresent - command: - - python - - {script_path} - env: - - name: SPARK_TEST_NAMESPACE - value: "{namespace}" - - name: SPARK_E2E_RUN_IN_CLUSTER - value: "1" -""" - with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: - f.write(manifest) - manifest_path = f.name - - subprocess.run( - ["kubectl", "delete", "job", job_name, "-n", namespace, "--ignore-not-found=true"], - capture_output=True, - timeout=15, - ) - apply_result = subprocess.run( - ["kubectl", "apply", "-f", manifest_path], - capture_output=True, - text=True, - timeout=30, - ) - if apply_result.returncode != 0: - with contextlib.suppress(Exception): - os.unlink(manifest_path) - err = (apply_result.stderr or "").strip() or (apply_result.stdout or "").strip() - return False, "", f"Failed to create Job: {err or apply_result.returncode}" - - wait_result = subprocess.run( - [ - "kubectl", - "wait", - "--for=condition=complete", - f"job/{job_name}", - "-n", - namespace, - f"--timeout={timeout_sec}s", - ], - capture_output=True, - text=True, - timeout=timeout_sec + 30, - ) - wait_stderr = (wait_result.stderr or "").strip() - if wait_result.returncode != 0: - succeeded = False - failed = True - else: - result = subprocess.run( - [ - "kubectl", - "get", - "job", - job_name, - "-n", - namespace, - "-o", - "jsonpath={.status.succeeded},{.status.failed}", - ], - capture_output=True, - text=True, - timeout=10, - ) - out = (result.stdout or "").strip() if result.returncode == 0 else "0,0" - parts = out.split(",") - succeeded = (parts[0] or "0") == "1" - failed = (parts[1] or "0") != "0" - - # Get logs from the Job pod - pod_result = subprocess.run( - [ - "kubectl", - "get", - "pods", - "-n", - namespace, - "-l", - f"job-name={job_name}", - "-o", - "jsonpath={.items[0].metadata.name}", - ], - capture_output=True, - text=True, - timeout=10, - ) - pod_name = (pod_result.stdout or "").strip() if pod_result.returncode == 0 else "" - no_pod_extra = "" - if not pod_name: - pods_list = subprocess.run( - ["kubectl", "get", "pods", "-n", namespace, "-l", f"job-name={job_name}", "-o", "wide"], - capture_output=True, - text=True, - timeout=10, - ) - if pods_list.returncode == 0 and (pods_list.stdout or pods_list.stderr): - no_pod_extra = ( - "\n--- Pods (job-name) ---\n" + (pods_list.stdout or "") + (pods_list.stderr or "") - ) - logs = "" - if pod_name: - log_result = subprocess.run( - ["kubectl", "logs", pod_name, "-n", namespace, "--tail=500"], - capture_output=True, - text=True, - timeout=30, - ) - logs = (log_result.stdout or "") + (log_result.stderr or "") - if not logs.strip(): - prev_result = subprocess.run( - ["kubectl", "logs", pod_name, "-n", namespace, "--tail=500", "--previous"], - capture_output=True, - text=True, - timeout=10, - ) - if prev_result.returncode == 0 and (prev_result.stdout or prev_result.stderr): - logs = ( - "(previous container)\n" - + (prev_result.stdout or "") - + (prev_result.stderr or "") - ) - - # Job description for debugging - desc_result = subprocess.run( - ["kubectl", "describe", "job", job_name, "-n", namespace], - capture_output=True, - text=True, - timeout=15, - ) - job_desc = desc_result.stdout or "" - if wait_stderr: - job_desc = f"--- kubectl wait stderr ---\n{wait_stderr}\n\n{job_desc}" - if pod_name: - pod_desc_result = subprocess.run( - ["kubectl", "describe", "pod", pod_name, "-n", namespace], - capture_output=True, - text=True, - timeout=15, - ) - if pod_desc_result.returncode == 0 and pod_desc_result.stdout: - job_desc = job_desc + "\n--- Pod describe ---\n" + pod_desc_result.stdout - if no_pod_extra: - job_desc = job_desc + no_pod_extra - - with contextlib.suppress(Exception): - os.unlink(manifest_path) - - subprocess.run( - ["kubectl", "delete", "job", job_name, "-n", namespace, "--ignore-not-found=true"], - capture_output=True, - timeout=15, - ) - subprocess.run( - [ - "kubectl", - "delete", - "sparkconnect", - "--all", - "-n", - namespace, - "--ignore-not-found=true", - "--wait=false", - ], - capture_output=True, - timeout=30, - ) - - success = succeeded and not failed - return success, logs, job_desc diff --git a/test/e2e/spark/test_spark_examples.py b/test/e2e/spark/test_spark_examples.py index 3e5efdc2c..3adaa620b 100644 --- a/test/e2e/spark/test_spark_examples.py +++ b/test/e2e/spark/test_spark_examples.py @@ -28,7 +28,6 @@ from kubeflow.spark.types.types import SparkConnectState from .cluster_watcher import run_watcher_in_thread -from .run_in_cluster import run_example_in_cluster # Path to examples directory EXAMPLES_DIR = Path(__file__).parent.parent.parent.parent / "examples" / "spark" @@ -36,10 +35,6 @@ EXAMPLE_TIMEOUT_SEC = 600 WATCHER_INTERVAL_SEC = 5.0 -# In-cluster: run example as K8s Job so client uses sc://...svc.cluster.local (no port-forward). -USE_IN_CLUSTER = os.environ.get("SPARK_E2E_RUN_IN_CLUSTER") == "1" -RUNNER_IMAGE = os.environ.get("SPARK_E2E_RUNNER_IMAGE", "") - def _run_example_with_watcher( example_path: Path, @@ -133,7 +128,9 @@ def _dump_on_failure( """Build failure message with cluster watcher log and example output.""" parts = [msg] if watcher_log: - parts.append("\n--- Cluster watcher (SparkConnect / pods / events / driver logs) ---") + parts.append( + "\n--- Cluster watcher (SparkConnect / SparkApplication / pods / events / driver logs) ---" + ) parts.append("\n".join(watcher_log)) parts.append("\n--- Example stdout ---") parts.append(stdout or "(empty)") @@ -142,30 +139,7 @@ def _dump_on_failure( return "\n".join(parts) def _run_example(self, example_script_name: str, namespace: str): - """Run example: in-cluster Job if SPARK_E2E_RUN_IN_CLUSTER=1 and image set, else subprocess.""" - if USE_IN_CLUSTER and RUNNER_IMAGE: - success, logs, job_desc = run_example_in_cluster( - example_script_name, - namespace, - image=RUNNER_IMAGE, - timeout_sec=EXAMPLE_TIMEOUT_SEC, - ) - if not success: - fail_msg = ( - f"In-cluster example {example_script_name} failed.\n" - f"--- Job describe ---\n{job_desc}\n--- Pod logs ---\n{logs or '(empty)'}" - ) - assert success, fail_msg - assert ( - "Session" in logs - or "complete" in logs.lower() - or "Level" in logs - or "Driver" in logs - or "Executor" in logs - or "EXAMPLE" in logs - or "E2E: Starting" in logs - ), f"In-cluster example produced no expected output. Logs:\n{logs or '(empty)'}" - return + """Run example as a subprocess against the ambient kubeconfig.""" example_path = EXAMPLES_DIR / example_script_name assert example_path.exists(), f"Example not found: {example_path}" returncode, stdout, stderr, watcher_log = _run_example_with_watcher( @@ -184,76 +158,35 @@ def _run_example(self, example_script_name: str, namespace: str): def test_spark_connect_simple_example(self): """EX01: Validate spark_connect_simple.py runs without errors.""" namespace = os.environ.get("SPARK_TEST_NAMESPACE", "spark-test") - if USE_IN_CLUSTER and RUNNER_IMAGE: - self._run_example("spark_connect_simple.py", namespace) - return stdout = self._run_example("spark_connect_simple.py", namespace) assert "SparkConnect session created" in stdout or "Session" in stdout def test_spark_advanced_options_example(self): """EX02: Validate spark_advanced_options.py runs without errors.""" namespace = os.environ.get("SPARK_TEST_NAMESPACE", "spark-test") - if USE_IN_CLUSTER and RUNNER_IMAGE: - self._run_example("spark_advanced_options.py", namespace) - return stdout = self._run_example("spark_advanced_options.py", namespace) assert "Driver" in stdout or "Executor" in stdout - def test_connect_existing_session_example(self): - """EX03: Validate connect_existing_session.py - base_url connect via two-client pattern. - - Runs in-cluster only (K8s Job mode). - """ - namespace = os.environ.get("SPARK_TEST_NAMESPACE", "spark-test") - - if not (USE_IN_CLUSTER and RUNNER_IMAGE): - pytest.skip("Requires in-cluster execution (SPARK_E2E_RUN_IN_CLUSTER=1)") - - self._run_example("connect_existing_session.py", namespace) - def test_batch_job_lifecycle_example(self): """EX04: Validate batch_job_lifecycle.py runs without errors.""" namespace = os.environ.get("SPARK_TEST_NAMESPACE", "spark-test") - - if USE_IN_CLUSTER and RUNNER_IMAGE: - self._run_example("batch_job_lifecycle.py", namespace) - return - stdout = self._run_example("batch_job_lifecycle.py", namespace) - assert "BATCH JOB LIFECYCLE COMPLETE!" in stdout def test_batch_failed_job_example(self): """EX05: Validate batch_failed_job.py handles failed Spark jobs.""" namespace = os.environ.get("SPARK_TEST_NAMESPACE", "spark-test") - - if USE_IN_CLUSTER and RUNNER_IMAGE: - self._run_example("batch_failed_job.py", namespace) - return - stdout = self._run_example("batch_failed_job.py", namespace) assert "Job failed as expected." in stdout def test_batch_func_job_lifecycle_example(self): """EX06: Validate batch_func_job_lifecycle.py runs without errors.""" namespace = os.environ.get("SPARK_TEST_NAMESPACE", "spark-test") - - if USE_IN_CLUSTER and RUNNER_IMAGE: - self._run_example("batch_func_job_lifecycle.py", namespace) - return - stdout = self._run_example("batch_func_job_lifecycle.py", namespace) - assert "FUNCJOB LIFECYCLE COMPLETE!" in stdout def test_batch_job_options_example(self): """EX07: Validate batch_job_options.py runs without errors.""" namespace = os.environ.get("SPARK_TEST_NAMESPACE", "spark-test") - - if USE_IN_CLUSTER and RUNNER_IMAGE: - self._run_example("batch_job_options.py", namespace) - return - stdout = self._run_example("batch_job_options.py", namespace) - assert "BATCH JOB OPTIONS COMPLETE!" in stdout