feat: e2e dispatcher integration

[evacchi]

Why is this PR needed?

The batch-gateway processor currently dispatches inference requests via direct HTTP calls to model gateways.
To integrate with the llm-d-async dispatcher we need an async dispatch path that enqueues requests
to Redis sorted sets and reads results from Redis lists, using the llm-d-async wire format.

Test this with:

ENABLE_DISPACHER=true make dev-deploy 
make test-e2e 

As explained below the async interface calls from a more profound refactoring, but I suggest the team to take on this in a follow-up.

What does this PR do?

Originally I used the api package directly, without realizing I should use the producer pkg (neither were published).

Now, if I understand correctly, the batch-gateway processor dispatches multiple requests per model concurrently. The llm-d-async dispatcher also processes requests concurrently (batchSize × budget). So results arrive in the result queue in arbitrary order, not in submission order. A naive SubmitRequest + GetResult per goroutine would return someone else's result.

The AsyncInferenceClient uses a background resultDispatcher goroutine to demux results from the shared result queue by request ID.

A single resultDispatcher per pool reads from the result queue (BRPOP) and routes each result to the correct caller via a sync.Map of requestID -> chan. Each Generate() call registers its channel before submitting, then selects on the channel or context cancellation. On context timeout, the waiter is unregistered to avoid leaks.

Note that because of the current assumptions in the code I had to add a separate
p.processModelAsync(...) alongside the existing p.processModel(...) in executeJob(...). I refactored what I could in helper methods, but in general this should call for a more significant refactoring.

The Async* new set of interfaces is significantly different from the sync versions (they might be be aligned using channels).

More details

Async inference client interface (pkg/clients/inference/async_inference_client_interface.go):

• New AsyncInferenceClient interface with Submit(ctx, req), GetResult(ctx), and Close(), symmetric with the sync InferenceClient.Generate() but non-blocking

Async inference client implementation (pkg/clients/inference/async_inference_client_impl.go):

• asyncProducerClient implements AsyncInferenceClient, backed by an llm-d-async/producer.Producer.
• A shared resultDispatcher goroutine per pool BRPOPs from the result queue and routes
  results to the correct per-job client via a sync.Map of requestID -> channel.
• Each job gets its own client (with its own internal results channel) from
  AsyncGatewayResolver.ClientFor, so concurrent batch jobs on the same model are isolated.
• Close() unregisters all pending waiters from the shared dispatcher.

Async resolver (pkg/clients/inference/async_inference_client_resolver.go):

• AsyncGatewayResolver is separate from GatewayResolver (sync). Stores shared asyncPool
  instances (producer + dispatcher per pool). ClientFor creates a fresh per-job client each call.
• Validates that no two models map to the same pool (single BRPOP reader per result queue).

Async executor path (internal/processor/worker/executor.go):

• processModelAsync — submit/collect pattern: sequential ZADD loop (fast), then
  GetResult loop collecting results as they arrive. No semaphores, no AIMD — the
  llm-d-async dispatcher controls inference concurrency via gates.
• Shared helpers extracted from processModel: readRequestLine, buildOutputLine,
  writeResult, drainAndFinalize, newErrorOutputLine, fairnessID.
• executeJob dispatches to processModelAsync when p.asyncInference != nil.

Processor setup (internal/processor/worker/worker.go):

• initConcurrencyControls extracted from Run(). In async mode, only the job-level
  worker semaphore is created — no global/endpoint semaphores or AIMD.

Clientset (internal/util/clientset/clientset.go):

• AsyncInference *inference.AsyncGatewayResolver field, separate from Inference.
• WithAsyncInference(cfg) sets AsyncInference; Close() closes whichever is set.

Config & resolution (internal/processor/config/config.go):

• ResolveModelGateways populates ResolvedGateways.Async with pool names from
  model_gateways[*].inference_pool_name when dispatch_mode: async.
• Global gateway rejected in async mode (per-model with pool names required).

Helm chart (charts/batch-gateway/templates/processor-configmap.yaml):

• dispatch_mode, async_dispatch.result_poll_timeout, inference_pool_name conditionally rendered.
• HTTP gateway fields rendered only when url is set, fixing a falsy-zero regression.

E2E tests & dev tooling:

• test/e2e/dispatcher_test.go: 4 tests — batch round-trip, multi-request batch,
  Redis dispatch gate, Prometheus-query dispatch gate.
• scripts/dev-deploy-dispatcher.sh: deploys three llm-d-async dispatcher instances
  (redis gate, endpoint-scrape gate, prometheus-query gate), configures Prometheus scrape
  targets, enables fake metrics on vllm-sim, reconfigures processor for async mode.
• make dev-deploy-dispatcher and make test-e2e-dispatcher targets.

How was this tested?

• Unit tests added/updated/verified
  • TestNewAsyncResolver: per-model routing, labels, error on invalid URL
  • TestAsyncInferenceClient_Generate: enqueue/dequeue round-trip via producer, timeout, queue name derivation
  • TestResolveModelGateways_Async: async config resolution, sync mode unaffected
• Integration/e2e tests added/updated/verified
  • TestDispatcher/BatchThroughDispatcher: single request end-to-end via async dispatch
  • TestDispatcher/MultiRequestBatch: 3-request batch completion
  • TestDispatcher/DispatchGate: Redis gate blocks/unblocks dispatch
  • TestDispatcher/PrometheusGate: Prometheus-query gate blocks dispatch when vllm-sim
    reports saturation (fake metrics), unblocks when saturation clears
• Manual testing performed
  • Deployed to Kind cluster via make dev-deploy && make dev-deploy-dispatcher
  • All 4 dispatcher e2e tests pass

Checklist

• Commits are signed off (git commit -s) per DCO
• Code follows project contributing guidelines
• CI checks pass (make ci)
• E2E tests pass (make test-e2e)

[Copilot]

Pull request overview

Note

Copilot was unable to run its full agentic suite in this review.

Adds support for async dispatch via llm-d-async (Redis-backed queues) and introduces deployment + E2E coverage for the dispatcher/gating behaviors.

Changes:

• Introduce async inference client + resolver (llm-d-async producer/Redis) and wire it into Clientset + processor config resolution.
• Extend processor Helm config to support dispatch_mode: async and async_dispatch settings.
• Add dispatcher E2E tests and dev scripts/targets to deploy dispatchers and run the new E2E suite.

Reviewed changes

Copilot reviewed 24 out of 27 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
pkg/clients/inference/async_inference_client.go	New inference client that enqueues requests + polls results via llm-d-async.
pkg/clients/inference/async_inference_client_resolver.go	Resolver creating per-model async clients backed by a shared Redis client + producers.
pkg/clients/inference/inference_client_resolver.go	Adds Close() on resolver to release async resources (Redis/producers).
internal/util/clientset/clientset.go	Adds WithAsyncInference option and constructs async resolver when configured; closes inference resolver on shutdown.
internal/processor/config/config.go / config_test.go	Async mode validation changes + resolves async model→pool mapping into ResolvedGateways.Async.
cmd/batch-processor/main.go	Wires resolved async gateways into clientset and improves initialization logging by mode.
charts/batch-gateway/templates/processor-configmap.yaml	Emits dispatch_mode and async_dispatch config; supports inference_pool_name in model_gateways.
test/e2e/dispatcher_test.go + test/e2e/dispatcher/*.yaml	New E2E tests and Helm values used to validate dispatcher + gate behaviors.
scripts/dev-deploy-dispatcher.sh / scripts/dev-clean.sh / Makefile	Adds dev workflow to install dispatchers, patch Prometheus/vLLM sim, port-forward, and run dispatcher E2E tests.
go.mod / go.sum + Dockerfiles	Adds llm-d-async deps and a local replace, plus Docker build context copying.
test/e2e/go.mod / test/e2e/go.sum	Adds llm-d-async deps + replaces for E2E module.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 26 out of 29 changed files in this pull request and generated 5 comments.

[evacchi]

make dev-deploy
DISPATCHER_SOURCE=/Users/evacchi/Devel/github.com/llm-d-incubation/llm-d-async \
    make dev-deploy-dispatcher
make test-e2e-dispatcher

override using PR at llm-d/llm-d-async#219

jaeger trace:

[evacchi]

squashed+rebased

[evacchi]

rebased+pushed

[evacchi]

addressed comments, rebased, tested e2e locally

[yizhaodev]

I am working on batch-gateway-operator to deploy llm-d-async
what gate_type and gate_params should be default values? or keep it as empty?

[evacchi]

I think prometheus-budget might be the safest, it should cascade automatically to different strategies in case the queries are missing data https://github.com/llm-d-incubation/llm-d-async#per-queue-dispatch-gates

[lioraron]

Thanks @evacchi!