[Bug] Sidecar mode shouldn't restart head pod when head pod is deleted

[400Ping]

Why are these changes needed?

When using sidecar mode, the head pod should not be recreated after it is deleted. The RayJob should be marked as Failed.

Related issue number

Closes #4130

Checks

• I've made sure the tests are passing.
• Testing Strategy
  • Unit tests
  • Manual tests
  • This PR is not tested :(

[rueian]

@400Ping, the change should be made in the raycluster_controller. We need to make the raycluster_controller not recreate the head pod if the cluster belongs to a RayJob, so that we can avoid races where the raycluster_controller recreates the head before the rayjob_controller checks it.

[400Ping]

@400Ping, the change should be made in the raycluster_controller. We need to make the raycluster_controller not recreate the head pod if the cluster belongs to a RayJob, so that we can avoid races where the raycluster_controller recreates the head before the rayjob_controller checks it.

ok, thanks

[rueian]

Won't this cause no head pod to be created at all? We still need to create the first head pod. I think you can check the RayClusterProvisioned condition to decide whether to create one or not.

[Future-Outlier]

I think this is related to the flaky test in CI now.
https://buildkite.com/ray-project/ray-ecosystem-ci-kuberay-ci/builds/11427#019a2849-e063-47f4-8aef-9143855d8976

[Future-Outlier]

let's try to fix this, super important.

[400Ping]

sorry, I am a bit busy taking midterm this week so I am a bit slow to respond

[Future-Outlier]

sorry, I am a bit busy taking midterm this week so I am a bit slow to respond

good luck for your midterm exam, thank you!

[Future-Outlier]

I tested this in my local laptop, and @machichima did it too.

reproduce step

1. create a RayJob

apiVersion: ray.io/v1
kind: RayJob
metadata:
  name: rayjob-sidecar-mode-flaky
spec:
  # In SidecarMode, the KubeRay operator injects a container into the Ray head Pod to submit the Ray job and tail logs.
  # This will avoid inter-Pod communication, which may cause network issues. For example, some users face WebSocket hangs.
  # For more details, see https://github.com/ray-project/kuberay/issues/3928#issuecomment-3187164736.
  submissionMode: "SidecarMode"
  entrypoint: python -c  "import time; time.sleep(60)"
  runtimeEnvYAML: |
    pip:
      - requests==2.26.0
      - pendulum==2.1.2
    env_vars:
      counter_name: "test_counter"

  rayClusterSpec:
    rayVersion: '2.46.0'
    headGroupSpec:
      rayStartParams: {}
      template:
        spec:
          containers:
          - name: ray-head
            image: rayproject/ray:2.46.0
            ports:
            - containerPort: 6379
              name: gcs-server
            - containerPort: 8265
              name: dashboard
            - containerPort: 10001
              name: client
            resources:
              limits:
                cpu: "1"
              requests:
                cpu: "200m"
            volumeMounts:
            - mountPath: /home/ray/samples
              name: code-sample
          volumes:
          - name: code-sample
            configMap:
              name: ray-job-code-sample
              items:
              - key: sample_code.py
                path: sample_code.py
    workerGroupSpecs:
    - replicas: 1
      minReplicas: 1
      maxReplicas: 5
      groupName: small-group
      rayStartParams: {}
      template:
        spec:
          containers:
          - name: ray-worker
            image: rayproject/ray:2.46.0
            resources:
              limits:
                cpu: "1"
              requests:
                cpu: "200m"

---
apiVersion: v1
kind: ConfigMap
metadata:
  name: ray-job-code-sample
data:
  sample_code.py: |
    import ray
    import os
    import requests

    ray.init()

    @ray.remote
    class Counter:
        def __init__(self):
            # Used to verify runtimeEnv
            self.name = os.getenv("counter_name")
            assert self.name == "test_counter"
            self.counter = 0

        def inc(self):
            self.counter += 1

        def get_counter(self):
            return "{} got {}".format(self.name, self.counter)

    counter = Counter.remote()

    for _ in range(5):
        ray.get(counter.inc.remote())
        print(ray.get(counter.get_counter.remote()))

    # Verify that the correct runtime env was used for the job.
    assert requests.__version__ == "2.26.0"

2. delete the head pod once the head container and submitter container are both running

3. check the rayjob CR's status (should be failure both)

4. you might wonder should we delete the worker pod? the answer is no, since we can't know if we use cluster selector from rayjob in raycluster.
   but 1 thing we can do is we can use deletion policy to delete the whole raycluster if needed.

[Future-Outlier]

cc @rueian @andrewsykim for merge, thank you!

[rueian]

@400Ping, The RayJob_fails_when_head_Pod_is_deleted_when_job_is_running test is failing

[Future-Outlier]

The latest change works in my local environment.
I downloaded the log from buildkite and found that k8s job mode's e2e test failed.
based on the log, I think this is a temporary fix.
however, I do found that I can refactor rayjob controller's logic better, I will open a follow-up PR recently.

cc @rueian @andrewsykim