Idea for RFC to add a new Service entity, like Environment but providing a network endpoint

[mwiebe]

Background

Many parallel workloads depend on a shared service to provide functionality like coordination or cached data. While you can deploy such a shared service and provide access by deploying infrastructure for it as part of your compute cluster, that puts the endpoint the Tasks depend on outside the tooling parseable job template. It would be useful to specify these services directly in job templates instead. A minimal specification for a service looks very similar to the already-defined step or environment script, so providing a new entity similar to the Environment looks like a promising way to do this.

Some concrete workloads that motivate this:

• A Valkey in-memory data store that a batch of Tasks read from and write to for caching computed data or coordinating shared state.
• A coordination service that a set of Tasks use to divide work, share intermediate results, or barrier between phases.
• An application "server mode" process, such as a DCC or renderer that accepts commands over a port, that is expensive to start and is reused across many Tasks.

This idea also relates to an active discussion, #136: Idea to add a [Monitoring] extension, which observes that the runtime currently cannot tell whether a running Action's process is still healthy. A long-lived Service endpoint would benefit from that same health monitoring, so it's another consumer of whatever that idea produces.

The RFC Idea

The idea is to define a new Service entity that is like the <Environment> entity, but that opens a network port that other entities within its scope, like Environments, Steps, and other Services, can access. Like Environment has jobEnvironments and stepEnvironments, we define jobServices and stepServices.

A Service differs from an Environment in four ways that the schema needs to make first-class:

1. It can run on a different host. The scheduler starts the service before scheduling any of the Tasks that depend on it, then schedules those Tasks, potentially spread across many hosts, configured to point at the service's endpoint.
2. It is long-lived. The service process runs for the lifetime of its scope (the Job, for a jobService; the set of Tasks for that Step, for a stepService) rather than running to completion up front. The scheduler starts it, keeps it alive while dependent Tasks run, and stops it once the scope completes.
3. It publishes an endpoint. The service declares one or more named ports. The runtime allocates a concrete address/port on the service's host and makes it discoverable both to the service process itself and, through a new Service.* format-string scope, to every dependent entity, so a Task on another host knows where to connect.
4. It has a readiness signal. The scheduler must know when the service is ready to accept connections before it schedules dependent Tasks, and must react if the service dies, either restarting it or failing the dependent scope, depending on the service's restartable setting.

Everything else, such as name, description, embedded files, variables, the let/expression scoping, and cancelation methods, is intended to mirror <Environment> so the entity is immediately familiar and reuses existing machinery.

Lifetime and placement

The defining characteristic of a Service is that it isn't tied to the host running the Tasks that use it, and it outlives any single Session:

• A jobService is scoped to the Job. The scheduler brings it up before scheduling any Task of the Job, keeps it running while the Job's Tasks execute (across however many hosts and Sessions those Tasks occupy), and tears it down once the Job no longer needs it.
• A stepService is scoped to a single Step. It is available only to that Step's own Tasks, not to any other Step (including Steps that depend on it), and it is torn down once that Step's Tasks are done.

Because the consuming Tasks may run on other hosts, the service's endpoint must be reachable over the network between Worker Hosts, and the scheduler is responsible for keeping the service alive for the duration of the scope. The scheduler is responsible for wiring up the connectivity, by running hosts on the same network or by using port forwarding.

Placement is up to the running system, though. A distributed render manager might dedicate a host to the service, whereas the OpenJD CLI's openjd run command runs an entire job on a single host and would simply start the service alongside the Tasks there.

Example: a Valkey store shared across a Job's Tasks

   Service Host                          Worker Hosts running Tasks
   ┌──────────────────────────┐          ┌───────────────────────────┐
   │ jobService: Cache        │          │  Session A                │
   │   run: valkey-server     │◄─────────┼─ Task 1 ─ onRun connects  │
   │     --port {{...port...}}│          │  Task 2 ─ to Valkey       │
   │   (stays up for the      │          └───────────────────────────┘
   │    Job's lifetime)       │          ┌───────────────────────────┐
   │                          │◄─────────┼─ Session B (other host)   │
   │                          │  connect │  Task 3 ─ onRun connects  │
   └──────────────────────────┘  to      │  Task 4 ─ to Valkey       │
        ▲                        Service.└───────────────────────────┘
        │                        Cache.main.{connectAddress,port}
        └─ scheduler starts this host before, and stops it after,
           the Tasks that depend on it

A sketch of how a Service could look

name: "Frame Processing With Shared Valkey Store"
specificationVersion: "jobtemplate-2023-09"
extensions: [SERVICE, EXPR]
parameterDefinitions:
  - name: FrameEnd
    type: INT
# A jobService is started before any Task of the Job is scheduled, stays up
# while the Job's Tasks run (across however many hosts and Sessions they
# occupy), and is stopped once the Job no longer needs it. Its endpoint is
# available to every Step's script and Tasks via the Service.* scope. A
# stepService is declared the same way on a Step, but scoped to that one Step.
jobServices:
  - name: Cache
    description: "A Valkey store the Job's Tasks share to cache and coordinate."
    # restartable: true lets the scheduler restart the service if it dies,
    # and stop/restart it around scheduling churn (e.g. to yield capacity to
    # a higher-priority job). Suitable here because the Tasks can repopulate
    # the cache. Defaults to false, which keeps the service pinned for the
    # whole scope and requeues completed Tasks if it dies.
    restartable: true
    # A service can carry its own host requirements, evaluated when the
    # scheduler places it. Requiring a non-preemptible host keeps the
    # service off Spot-style capacity that could be reclaimed mid-scope.
    hostRequirements:
      attributes:
        - name: attr.worker.preemptible
          anyOf: ["false"]
    # Named ports the service exposes. The runtime allocates a port and
    # makes the value available to the service process (below) and via the
    # Service.* scope to every dependent entity.
    ports:
      - name: main
        # readiness tells the scheduler when the service is ready to accept
        # connections. Dependent Tasks are not scheduled until it passes.
        readiness:
          # A TCP-connect probe against the allocated port. Alternatives
          # discussed below (command probe, stdout signal).
          type: TCP_CONNECT
          timeoutSeconds: 60
    script:
      actions:
        # 'run' is the long-lived action: its process IS the service. The
        # runtime keeps it running for the scope's lifetime and cancels it
        # on exit (default cancelation is TERMINATE, as for other actions).
        run:
          command: "valkey-server"
          # The allocated port and the interface to bind are injected via
          # the new Service scope, keyed by the service and port name. The
          # runtime decides the bind interface: broad (reachable from other
          # hosts) for a distributed scheduler, or loopback for a single-host
          # runner like `openjd run`. The service just binds what it's told.
          args:
            - "--port"
            - "{{ Service.Cache.main.port }}"
            - "--bind"
            - "{{ Service.Cache.main.bindAddress }}"
steps:
  - name: ProcessFrames
    parameterSpace:
      taskParameterDefinitions:
        - name: Frame
          type: INT
          range: "1-{{ Param.FrameEnd }}"
    script:
      actions:
        onRun:
          command: bash
          args: ["{{ Task.File.run }}"]
      embeddedFiles:
        - name: run
          type: TEXT
          data: |
            #!/usr/bin/env bash
            # The same Service.* scope is available to the Job's Tasks, so a
            # Task on any host connects to Valkey by its allocated endpoint.
            process-frame \
              --frame {{ Task.Param.Frame }} \
              --valkey-host "{{ Service.Cache.main.connectAddress }}" \
              --valkey-port "{{ Service.Cache.main.port }}"

The new pieces

Most of the entity is illustrated in the sketch above. A few more details:

1. jobServices / stepServices and <Service>. Ordered lists on the Job Template and Step Template, mirroring jobEnvironments/stepEnvironments; a stepService is scoped to its one Step. Names must be unique within their scope, and a jobService name must not collide with a stepService name in the same Job. A <Service> mirrors <Environment> (name, description, variables, embedded files) but adds ports, restartable, and hostRequirements, and uses a long-lived run action in place of onEnter/onExit.

2. ports and readiness. Ports are runtime-allocated by default, or an author may request a specific port. Each declares a readiness probe the runtime uses to gate dependent Tasks; sketched options are TCP_CONNECT, a COMMAND probe (run a command until it exits 0, à la a Docker docker ps check), or an openjd_service_ready: stdout line consistent with the existing openjd_* protocol.

3. attr.worker.preemptible. A new standard boolean-valued attribute capability, so a Service (or any Step) can require attr.worker.preemptible anyOf ["false"] to stay off capacity that could be reclaimed mid-scope, like EC2 Spot Instances. Workers advertise the value; this proposal just reserves the name and meaning.

4. The Service.* format-string scope. New @fmtstring[host] value references (resolved at execution time, since the endpoint isn't known until the scheduler places the service, and it may change across a restart):

   • Service.<name>.<port>.port and Service.<name>.<port>.bindAddress, for the service process, telling it the port and interface to listen on.
   • Service.<name>.<port>.connectAddress, for dependent Environments, scripts, and Tasks, telling them where to connect.