[draft pr][RayJob] Use timeout to prevent RayCluster leak

[Future-Outlier]

Why are these changes needed?

Related issue number

#3860 (comment)

Checks

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

[Future-Outlier]

cc @machichima

apiVersion: ray.io/v1
kind: RayJob
metadata:
  name: rayjob-sample-external-redis-2-49-0-2
spec:
  # submissionMode specifies how RayJob submits the Ray job to the RayCluster.
  # The default value is "K8sJobMode", meaning RayJob will submit the Ray job via a submitter Kubernetes Job.
  # The alternative value is "HTTPMode", indicating that KubeRay will submit the Ray job by sending an HTTP request to the RayCluster.
  # submissionMode: "K8sJobMode"
  entrypoint: python -c "import os, time; print(os.environ.get('HOSTNAME')); [time.sleep(1) or print(i) for i in range(1000)]"
  # shutdownAfterJobFinishes specifies whether the RayCluster should be deleted after the RayJob finishes. Default is false.
  # shutdownAfterJobFinishes: false

  # ttlSecondsAfterFinished specifies the number of seconds after which the RayCluster will be deleted after the RayJob finishes.
  # ttlSecondsAfterFinished: 10

  # activeDeadlineSeconds is the duration in seconds that the RayJob may be active before
  # KubeRay actively tries to terminate the RayJob; value must be positive integer.
  # activeDeadlineSeconds: 120

  # RuntimeEnvYAML represents the runtime environment configuration provided as a multi-line YAML string.
  # See https://docs.ray.io/en/latest/ray-core/handling-dependencies.html for details.
  # (New in KubeRay version 1.0.)
  runtimeEnvYAML: |
    pip:
      - requests==2.26.0
      - pendulum==2.1.2
    env_vars:
      counter_name: "test_counter"

  # Suspend specifies whether the RayJob controller should create a RayCluster instance.
  # If a job is applied with the suspend field set to true, the RayCluster will not be created and we will wait for the transition to false.
  # If the RayCluster is already created, it will be deleted. In the case of transition to false, a new RayCluster will be created.
  # suspend: false

  # rayClusterSpec specifies the RayCluster instance to be created by the RayJob controller.
  rayClusterSpec:
    rayVersion: '2.46.0' # should match the Ray version in the image of the containers
    gcsFaultToleranceOptions:
    # In most cases, you don't need to set `externalStorageNamespace` because KubeRay will
    # automatically set it to the UID of RayCluster. Only modify this annotation if you fully understand
    # the behaviors of the Ray GCS FT and RayService to avoid misconfiguration.
    # [Example]:
    # externalStorageNamespace: "my-raycluster-storage"
      redisAddress: "redis:6379"
      redisPassword:
        valueFrom:
          secretKeyRef:
            name: redis-password-secret
            key: password
    # Ray head pod template
    headGroupSpec:
      # The `rayStartParams` are used to configure the `ray start` command.
      # See https://github.com/ray-project/kuberay/blob/master/docs/guidance/rayStartParams.md for the default settings of `rayStartParams` in KubeRay.
      # See https://docs.ray.io/en/latest/cluster/cli.html#ray-start for all available options in `rayStartParams`.
      rayStartParams: {}
      #pod template
      template:
        spec:
          # terminationGracePeriodSeconds: 1203
          containers:
          - name: ray-head
            image: rayproject/ray:2.46.0
            resources:
              limits:
                cpu: "1"
              requests:
                cpu: "1"
            
            lifecycle:
              preStop:
                exec:
                  command: ["python -c 'import ray; ray.shutdown()'"]
            ports:
            - containerPort: 6379
              name: redis
            - containerPort: 8265
              name: dashboard
            - containerPort: 10001
              name: client
            volumeMounts:
            - mountPath: /tmp/ray
              name: ray-logs
            - mountPath: /home/ray/samples
              name: ray-example-configmap
          volumes:
          - name: ray-logs
            emptyDir: {}
          - name: ray-example-configmap
            configMap:
              name: ray-example
              defaultMode: 0777
              items:
              - key: detached_actor.py
                path: detached_actor.py
              - key: increment_counter.py
                path: increment_counter.py
    workerGroupSpecs:
    # the pod replicas in this group typed worker
    - replicas: 1
      minReplicas: 1
      maxReplicas: 10
      groupName: small-group
      # The `rayStartParams` are used to configure the `ray start` command.
      # See https://github.com/ray-project/kuberay/blob/master/docs/guidance/rayStartParams.md for the default settings of `rayStartParams` in KubeRay.
      # See https://docs.ray.io/en/latest/cluster/cli.html#ray-start for all available options in `rayStartParams`.
      rayStartParams: {}
      # Pod template
      template:
        spec:
          containers:
          - name: ray-worker
            image: rayproject/ray:2.46.0
            volumeMounts:
            - mountPath: /tmp/ray
              name: ray-logs
            resources:
              limits:
                cpu: "1"
              requests:
                cpu: "1"
          volumes:
          - name: ray-logs
            emptyDir: {}
---
kind: ConfigMap
apiVersion: v1
metadata:
  name: redis-config
  labels:
    app: redis
data:
  redis.conf: |-
    dir /data
    port 6379
    bind 0.0.0.0
    appendonly yes
    protected-mode no
    requirepass 5241590000000000
    pidfile /data/redis-6379.pid
---
apiVersion: v1
kind: Service
metadata:
  name: redis
  labels:
    app: redis
spec:
  type: ClusterIP
  ports:
  - name: redis
    port: 6379
  selector:
    app: redis
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: redis
  labels:
    app: redis
spec:
  replicas: 1
  selector:
    matchLabels:
      app: redis
  template:
    metadata:
      labels:
        app: redis
    spec:
      containers:
      - name: redis
        image: redis:7.4.0
        command:
        - "sh"
        - "-c"
        - "redis-server /usr/local/etc/redis/redis.conf"
        ports:
        - containerPort: 6379
        volumeMounts:
        - name: config
          mountPath: /usr/local/etc/redis/redis.conf
          subPath: redis.conf
      volumes:
      - name: config
        configMap:
          name: redis-config
---
# Redis password
apiVersion: v1
kind: Secret
metadata:
  name: redis-password-secret
type: Opaque
data:
  # echo -n "5241590000000000" | base64
  password: NTI0MTU5MDAwMDAwMDAwMA==
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: ray-example
data:
  detached_actor.py: |
    import ray

    @ray.remote(num_cpus=1)
    class Counter:
      def __init__(self):
          self.value = 0

      def increment(self):
          self.value += 1
          return self.value

    ray.init(namespace="default_namespace")
    Counter.options(name="counter_actor", lifetime="detached").remote()
  increment_counter.py: |
    import ray

    ray.init(namespace="default_namespace")
    counter = ray.get_actor("counter_actor")
    print(ray.get(counter.increment.remote()))