[Neha-Nimsatkar]

What this PR does / why we need it

Adds end-to-end integration and an example workflow notebook (examples/spark/batch_job_s3a.ipynb) demonstrating Spark batch job submissions with scripts hosted in S3-compatible object stores using the s3a:// URI scheme. It also introduces a modular, backend-agnostic object store testing framework for the E2E CI test suite.

Key Changes

• Generic S3A Notebook:
  • Converted the legacy script into an interactive Jupyter notebook (examples/spark/batch_job_s3a.ipynb) parameterized via Papermill tags.
  • Replaced plaintext credential arguments with secure in-cluster Kubernetes secret bindings (spark.kubernetes.{role}.secretKeyRef.{key}).
  • Stripped redundant JVM/SSL properties, relying on standard Hadoop-AWS S3A resolution and modular AWS SDK v2 dependencies.
• Pluggable Object Store E2E Architecture:
  • Decoupled generic object store orchestration (hack/e2e/object-store/setup.sh) from specific storage backends (hack/e2e/object-store/backends/seaweedfs.sh).
  • Added declarative S3 artifact seeding via hack/e2e/object-store/seed-job.yaml driven by E2E_S3_SEED.
  • Replaced hardcoded install_seaweedfs in hack/e2e-setup-cluster.sh with a configurable setup_object_store() hook gated on E2E_OBJECT_STORE.
• CI Test Suite Integration:
  • Wired notebook execution into .github/workflows/test-spark-examples.yaml with dynamic PAPERMILL_PARAMS (S3_ENDPOINT, S3_REGION, S3_BUCKET, S3_PREFIX, S3_SECRET_NAME).
  • Configured cluster setup environment variables for SeaweedFS provisioning and test script seeding.
• Documentation:
  • Updated docs/source/spark/batch-jobs.rst to reflect verified s3a:// remote script submission support and linked the example notebook.
  • Indexed the notebook in examples/spark/README.md.

Testing Done

• Validated full local setup and notebook execution using hack/e2e-setup-cluster.sh and make test-e2e-notebook on Kind.
• Verified Secret replication across namespaces and successful object seeding via seed-job.yaml.
• Verified clean driver and executor pod lifecycle, secret mounting, and log retrieval.

Which issue(s) this PR fixes

Fixes #595

Checklist

• Docs included if any changes are user facing
• All PR review comments addressed and conversations resolved

[google-oss-prow]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign electronic-waste for approval. For more information see the Kubernetes Code Review Process.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[tariq-hasan]

/ok-to-test

[review-notebook-app]

Check out this pull request on 

See visual diffs & provide feedback on Jupyter Notebooks.

---

Powered by ReviewNB

[tariq-hasan]

Thanks for the PR @Neha-Nimsatkar!

The local:// and https:// schemes are already covered in other notebooks so s3a:// was the missing gap that this PR covers.

Let's also refactor the PR to use the following layout. That way we keep the SeaweedFS-specific implementation details decoupled from S3A and generic object store setup for E2E tests.

• hack/e2e/object-store/setup.sh: Secret/ConfigMap per client namespace, buckets + seed Job, verify, write outputs
• hack/e2e/object-store/seed-job.yaml: aws-cli Job that creates buckets, uploads and verifies
• hack/e2e/object-store/backends/seaweedfs.sh: deploy, wait for ready, create the user, print the endpoint
• hack/e2e/object-store/backends/seaweedfs.yaml: Deployment + Service seaweedfs on 8333

Let's also use the following as the input interface that are read as environment variables setup.sh:

• E2E_OBJECT_STORE: which server to deploy (seaweedfs)
• E2E_S3_NAMESPACE: namespace the storage server runs in, separate from any test namespace
• E2E_S3_CLIENT_NAMESPACES: client namespaces that get the Secret and ConfigMap
• E2E_S3_BUCKETS: buckets created through the S3 API
• E2E_S3_SEED: Job artifact paths uploaded through the S3 API
• E2E_S3_ACCESS_KEY / E2E_S3_SECRET_KEY: credentials for the single test user

You can set these outputs from hack/e2e/object-store/setup.sh:

• Secret e2e-s3-credentials in each client namespace with keys AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY
• ConfigMap e2e-s3-config in each client namespace with AWS_ENDPOINT_URL and AWS_REGION
• Env file (${E2E_S3_ENV_FILE:-.e2e/object-store.env}) containing S3_ENDPOINT, S3_REGION, S3_SECRET_NAME and S3_BUCKETS

I have added review comments inline on the PR aligning with the above design so we can get this completed.

[tariq-hasan]

hack/manifests/ suggests product manifests, but this is test infrastructure.

Please move it to hack/e2e/object-store/ and split it into a generic setup.sh (that is responsible for Secret/ConfigMap per client namespace, buckets and seeding through the S3 API, verification, and writing outputs) and backends/seaweedfs.sh (that deploys the server, waits for it and create the user).

The inline comments below say which part each piece moves to.

[tariq-hasan]

Using a single argument here is likely to mix up the server's namespace with the test namespace.

Insead of using a single argument here as the client namespace that receives the secret and hardcoding the the server namespace as seaweedfs let's read the inputs from the interface instead: E2E_S3_NAMESPACE for the server and E2E_S3_CLIENT_NAMESPACES for the namespaces that get the Secret and ConfigMap.

[tariq-hasan]

This secret is not used anywhere.

Please have setup.sh generate credentials that are random per run unless E2E_S3_ACCESS_KEY and E2E_S3_SECRET_KEY are set.

Then in each client namespace create the neutral Secret e2e-s3-credentials with the keys AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY and also create the ConfigMap e2e-s3-config with AWS_ENDPOINT_URL and AWS_REGION.

for ns in ${E2E_S3_CLIENT_NAMESPACES}; do
  kubectl -n "$ns" create secret generic e2e-s3-credentials \
    --from-literal=AWS_ACCESS_KEY_ID="$AK" --from-literal=AWS_SECRET_ACCESS_KEY="$SK" \
    --dry-run=client -o yaml | kubectl apply -f -
  kubectl -n "$ns" create configmap e2e-s3-config \
    --from-literal=AWS_ENDPOINT_URL="$ENDPOINT" --from-literal=AWS_REGION=us-east-1 \
    --dry-run=client -o yaml | kubectl apply -f -
done

[tariq-hasan]

The bucket creation here relies on authentication being off. The unsigned PUT only works because SeaweedFS has no users configured, so it accepts anyone. Once authentication is on, it gets 403 and the bucket is silently never created.

The result is also not checked.

Let's split this into two:

• Creating the user is server-specific and goes in backends/seaweedfs.sh, using SeaweedFS's admin shell and with retries until the server is ready:

  kubectl -n "$E2E_S3_NAMESPACE" exec deploy/seaweedfs -- sh -c "
    for i in \$(seq 1 30); do
      echo 's3.configure -user e2e -access_key $AK -secret_key $SK -actions Read,Write,List,Admin -apply' | weed shell && exit 0
      sleep 2
    done; exit 1"

• Creating buckets is generic and belongs in the seed Job, using aws s3 mb. That way callers can ask for buckets the same way whichever server is behind it.

[tariq-hasan]

Instead of having a second fixture let's reuse examples/spark/spark_job.py.

Also let's drive seeding from E2E_S3_SEED with a generic Job:

• Build a ConfigMap from the listed repo files.
• Run an amazon/aws-cli Job (pinned by digest) with credentials from e2e-s3-credentials.
• The Job creates each bucket in E2E_S3_BUCKETS, uploads each entry, and checks each one with aws s3api head-object.
• setup.sh runs kubectl wait --for=condition=complete job/e2e-s3-seed --timeout=120s and fails on error.

The Spark workflow then passes E2E_S3_SEED="e2e/spark/spark_job.py=examples/spark/spark_job.py". Nothing in the fixture names the file.

[tariq-hasan]

Please remove these settings and verify if the CI passes. If so let's remove them.

[tariq-hasan]

Please also remove these settings and verify if the CI passes. If so let's remove them.

[tariq-hasan]

Let's replace with the following. The goal is to make the setup configurable and extensible beyond seaweedfs.

setup_object_store() {
    [[ -n "${E2E_OBJECT_STORE:-}" ]] || return 0
    phase_start "setup_object_store"
    local script_dir
    script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
    E2E_S3_CLIENT_NAMESPACES="${E2E_S3_CLIENT_NAMESPACES:-$NAMESPACE}" \
        "${script_dir}/e2e/object-store/setup.sh"
    phase_end "setup_object_store"
}

[tariq-hasan]

Let's add the following as well to the list of environment variables.

E2E_OBJECT_STORE: seaweedfs
E2E_S3_SEED: "e2e/spark/spark_job.py=examples/spark/spark_job.py"

[tariq-hasan]

Let's add the configurations as well.

Suggested change

	make test-e2e-notebook \
	NOTEBOOK_INPUT=./examples/spark/batch_job_s3a_seaweedfs.ipynb \
	NOTEBOOK_OUTPUT=./artifacts/notebooks/batch_job_s3a_seaweedfs.ipynb \
	PAPERMILL_TIMEOUT=600
	make test-e2e-notebook \
	NOTEBOOK_INPUT=./examples/spark/batch_job_s3a.ipynb \
	NOTEBOOK_OUTPUT=./artifacts/notebooks/batch_job_s3a.ipynb \
	PAPERMILL_PARAMS="-p S3_ENDPOINT $S3_ENDPOINT -p S3_REGION $S3_REGION -p S3_BUCKET e2e -p S3_PREFIX spark -p S3_SECRET_NAME $S3_SECRET_NAME" \
	PAPERMILL_TIMEOUT=600

[tariq-hasan]

/ok-to-test

[tariq-hasan]

/ok-to-test
/retest

[tariq-hasan]

Rebasing on master once #930 merges should help with this since it pins the latest commit, updates the Python models and also updates the E2E tests to more accurately reflect upstream operator behavior.

[Neha-Nimsatkar]

Hi @tariq-hasan, I've rebased on top of #930 . Could you please run /ok-to-test so we can verify if the operator fix resolves the CI run?

[tariq-hasan]

/ok-to-test

[tariq-hasan]

/ok-to-test
/retest

[Neha-Nimsatkar]

Hi @tariq-hasan, could you please click Approve and run on the workflow checks?

[tariq-hasan]

I have added a few more comments.

[tariq-hasan]

Let's move to a current SeaweedFS release, pinned by digest.

Suggested change

	image: chrislusf/seaweedfs:3.79@sha256:21425305e04b355904ef350ff75a4e81b8afc5124f63e0485b946164b58562d3
	image: ghcr.io/chrislusf/seaweedfs:4.50@sha256:<digest>

[tariq-hasan]

Please add a readiness probe. Without one, kubectl rollout status in seaweedfs.sh returns as soon as the container starts, and the seed Job (with backoffLimit: 0) can hit 8333 before the gateway is listening. Please add the following after ports::

          readinessProbe:
            httpGet:
              path: /status
              port: 8333
            periodSeconds: 2
            failureThreshold: 30

[tariq-hasan]

Even with the readiness probe, there's a window after s3.configure -apply where SeaweedFS hasn't loaded the identity yet and still accepts anonymous requests. Seeding (and later the Spark job) can then pass without credentials ever being checked.

Let's wait until signed requests work and anonymous requests are rejected. That covers readiness and also proves auth is enforced:

Suggested change

	set -euo pipefail
	echo "Creating buckets: ${E2E_S3_BUCKETS}"
	set -euo pipefail
	echo "Waiting for S3 API with authentication enforced..."
	ready=0
	for i in $(seq 1 30); do
	if aws --endpoint-url="${AWS_ENDPOINT_URL}" s3api list-buckets >/dev/null 2>&1 \
	&& ! aws --endpoint-url="${AWS_ENDPOINT_URL}" --no-sign-request s3api list-buckets >/dev/null 2>&1; then
	ready=1; break
	fi
	sleep 2
	done
	[ "${ready}" = 1 ] || { echo "S3 API not ready or anonymous access is allowed" >&2; exit 1; }
	echo "Creating buckets: ${E2E_S3_BUCKETS}"

[tariq-hasan]

Let's bump amazon/aws-cli:2.15.15 as it is an older version.

[tariq-hasan]

The original interface was E2E_S3_SEED="<bucket>/<key>=<repo path>", e.g. e2e/spark/spark_job.py=examples/spark/spark_job.py. Uploading everything to FIRST_BUCKET means callers can't seed a second bucket, and the target bucket is implied rather than stated.

Mounting by basename also collides when two entries share a filename (e.g. two spark_job.py fixtures). Let's key the ConfigMap by index; setup.sh iterates in the same order:

Suggested change

	FIRST_BUCKET=$(echo "${E2E_S3_BUCKETS}" | awk '{print $1}')
	if [ -n "${E2E_S3_SEED:-}" ]; then
	for item in ${E2E_S3_SEED}; do
	s3_key="${item%%=*}"
	local_name="${item#*=}"
	filename="$(basename "${local_name}")"
	echo "Uploading /seed/${filename} to s3://${FIRST_BUCKET}/${s3_key}..."
	aws --endpoint-url="${AWS_ENDPOINT_URL}" s3 cp "/seed/${filename}" "s3://${FIRST_BUCKET}/${s3_key}"
	echo "Verifying s3://${FIRST_BUCKET}/${s3_key}..."
	aws --endpoint-url="${AWS_ENDPOINT_URL}" s3api head-object --bucket "${FIRST_BUCKET}" --key "${s3_key}"
	done
	fi
	if [ -n "${E2E_S3_SEED:-}" ]; then
	i=0
	for item in ${E2E_S3_SEED}; do
	dest="${item%%=*}"
	case "${dest}" in */*) ;; *) echo "E2E_S3_SEED entry '${item}' must be <bucket>/<key>=<path>" >&2; exit 1 ;; esac
	bucket="${dest%%/*}"
	key="${dest#*/}"
	echo "Uploading seed-${i} to s3://${bucket}/${key}..."
	aws --endpoint-url="${AWS_ENDPOINT_URL}" s3 cp "/seed/seed-${i}" "s3://${bucket}/${key}"
	aws --endpoint-url="${AWS_ENDPOINT_URL}" s3api head-object --bucket "${bucket}" --key "${key}"
	i=$((i + 1))
	done
	fi

[tariq-hasan]

Since backoffLimit: 0, a failed seed Job never becomes complete, so this waits the full 120 s before printing logs. Let's stop on either outcome:

Suggested change

	kubectl wait --for=condition=complete job/e2e-s3-seed -n "${PRIMARY_CLIENT_NS}" --timeout=120s || (kubectl logs -n "${PRIMARY_CLIENT_NS}" -l job-name=e2e-s3-seed --tail=50 && kubectl describe job/e2e-s3-seed -n "${PRIMARY_CLIENT_NS}" && exit 1)
	for _ in $(seq 1 60); do
	succeeded="$(kubectl get job/e2e-s3-seed -n "${PRIMARY_CLIENT_NS}" -o jsonpath='{.status.succeeded}')"
	failed="$(kubectl get job/e2e-s3-seed -n "${PRIMARY_CLIENT_NS}" -o jsonpath='{.status.failed}')"
	[[ "${succeeded:-0}" -ge 1 ]] && break
	if [[ "${failed:-0}" -ge 1 || "${_}" -eq 60 ]]; then
	kubectl logs -n "${PRIMARY_CLIENT_NS}" -l job-name=e2e-s3-seed --tail=100 || true
	kubectl describe job/e2e-s3-seed -n "${PRIMARY_CLIENT_NS}" || true
	exit 1
	fi
	sleep 2
	done

[tariq-hasan]

With the <bucket>/<key> seed format:

Suggested change

	E2E_OBJECT_STORE: seaweedfs
	E2E_S3_BUCKETS: e2e
	E2E_S3_SEED: "spark/spark_job.py=examples/spark/spark_job.py"
	E2E_OBJECT_STORE: seaweedfs
	E2E_S3_BUCKETS: e2e
	E2E_S3_SEED: "e2e/spark/spark_job.py=examples/spark/spark_job.py"

[tariq-hasan]

spark_job.py doesn't read sys.argv, so args=["10"] is a leftover from the earlier fixture. Let's drop it.

[tariq-hasan]

The test currently passes on status alone. Let's assert that the driver actually ran our script from S3:

logs = list(client.get_job_logs(job_name))
for line in logs:
    print(line.rstrip())
assert any("Row count: 10" in line for line in logs), "Expected output from spark_job.py not found in driver logs"

[tariq-hasan]

Suggested change

	- **batch_job_lifecycle.ipynb** - Submit a `FileJob` and exercise the batch job lifecycle (`get_job`, `list_jobs`, `get_job_logs`, `delete_job`)
	- **batch_func_job_lifecycle.ipynb** - Submit a `FuncJob` and exercise the batch job lifecycle
	- **batch_failed_job.ipynb** - Submit a `FileJob` expected to fail and inspect the `FAILED` state
	- **batch_job_options.ipynb** - Submit a batch job with Kubernetes options (labels, annotations, node selector, tolerations, custom name)
	- **batch_job_s3a.ipynb** - Submit a `FileJob` referencing an application script stored in S3 (`s3a://` URI)

[RobuRishabh]

/ok-to-test