diff --git a/benchmarks/setup.sh b/benchmarks/setup.sh index 8efcf91d1..cbdb9c114 100755 --- a/benchmarks/setup.sh +++ b/benchmarks/setup.sh @@ -171,8 +171,8 @@ GIE_UPSTREAM_REPO="https://github.com/kubernetes-sigs/gateway-api-inference-exte # Async-processor settings for scenario 5 DISPATCHER_VERSION="${DISPATCHER_VERSION:-v0.7.3}" -DISPATCHER_IMAGE="${DISPATCHER_IMAGE:-ghcr.io/llm-d-incubation/llm-d-async:${DISPATCHER_VERSION}}" -DISPATCHER_CHART="${DISPATCHER_CHART:-oci://ghcr.io/llm-d-incubation/charts/async-processor}" +DISPATCHER_IMAGE="${DISPATCHER_IMAGE:-ghcr.io/llm-d/llm-d-async:${DISPATCHER_VERSION}}" +DISPATCHER_CHART="${DISPATCHER_CHART:-oci://ghcr.io/llm-d/charts/llm-d-async}" DISPATCHER_CHART_VERSION="${DISPATCHER_CHART_VERSION:-0.7.3}" # --- Inference backend --- @@ -618,13 +618,9 @@ EOVLLMSVC -n "${NAMESPACE}" \ --set "ap.image.repository=${DISPATCHER_IMAGE_REPO}" \ --set "ap.image.tag=${DISPATCHER_IMAGE_TAG}" \ - --set ap.messageQueueImpl=redis-sortedset \ + --set ap.transport=redis-sortedset \ --set ap.concurrency=1 \ - --set ap.redis.enabled=true \ - --set "ap.redis.url=redis://redis-master.${NAMESPACE}.svc.cluster.local:6379" \ - --set ap.redis.pollIntervalMs=500 \ - --set ap.redis.batchSize=10 \ - --set-json "ap.redis.queuesConfig=[{\"queue_name\":\"llm-d-async:requests:${ASYNC_POOL_NAME}\",\"result_queue_name\":\"llm-d-async:results:${ASYNC_POOL_NAME}\",\"request_path_url\":\"/v1/chat/completions\",\"igw_base_url\":\"${ASYNC_IGW_URL}\",\"gate_type\":\"endpoint-scrape\",\"gate_params\":{\"url\":\"${ASYNC_METRICS_URL}\",\"metric\":\"vllm:num_requests_waiting\",\"max_count_per_pod\":\"5\",\"fallback\":\"1.0\"}}]" \ + --set-json "ap.transportConfig={\"urlSecret\":{\"url\":\"redis://redis-master.${NAMESPACE}.svc.cluster.local:6379\"},\"result_queue_name\":\"result-list\",\"poll_interval_ms\":500,\"batch_size\":10,\"queues\":[{\"queue_name\":\"llm-d-async:requests:${ASYNC_POOL_NAME}\",\"result_queue_name\":\"llm-d-async:results:${ASYNC_POOL_NAME}\",\"request_path_url\":\"/v1/chat/completions\",\"igw_base_url\":\"${ASYNC_IGW_URL}\",\"gate_type\":\"endpoint-scrape\",\"gate_params\":{\"url\":\"${ASYNC_METRICS_URL}\",\"metric\":\"vllm:num_requests_waiting\",\"max_count_per_pod\":\"5\",\"fallback\":\"1.0\"}}]}" \ --set ap.modelServerMonitor.enabled=false \ --set ap.metrics.enabled=true \ --set ap.metrics.port=9091 \ @@ -632,7 +628,7 @@ EOVLLMSVC --wait --timeout=120s >/dev/null log " Waiting for async-processor to be ready..." - ${K} -n "${NAMESPACE}" wait --for=condition=available deployment/async-processor --timeout=120s >/dev/null + ${K} -n "${NAMESPACE}" wait --for=condition=available deployment/async-processor-llm-d-async --timeout=120s >/dev/null log " Async-processor deployed (pool: ${ASYNC_POOL_NAME}, gate: endpoint-scrape)" fi diff --git a/docs/design/batch-dispatcher-queue-design.md b/docs/design/batch-dispatcher-queue-design.md index 6af4045a7..bc51d0a30 100644 --- a/docs/design/batch-dispatcher-queue-design.md +++ b/docs/design/batch-dispatcher-queue-design.md @@ -83,26 +83,25 @@ The Redis URL is read from a mounted secret at runtime (not stored in the config ### Dispatcher Configuration -The dispatcher (llm-d-async) already supports the Redis sorted-set flow with dispatch budget gating. The request queue is configured via the [JSON queues config file](https://github.com/llm-d/llm-d-async/blob/main/README.md#redis-sorted-set-persisted) (`--redis.ss.queues-config-file`); the result queue is configured via `--redis.ss.result-queue-name`: +The dispatcher (llm-d-async) already supports the Redis sorted-set flow with dispatch budget gating. It is configured with a single [transport configuration document](https://github.com/llm-d/llm-d-async/blob/main/README.md#transport-configuration), passed via `--transport redis-sortedset` and `--transport-config` (or `--transport-config-file`). The request queue is defined by the `queues` array; the result queue is set via the top-level `result_queue_name`: ```json -[ - { - "queue_name": "llm-d-async:requests:optimized-baseline", - "igw_base_url": "http://llm-d-inference-gateway-istio:80", - "request_path_url": "/v1/completions", - "gate_type": "prometheus-budget", - "gate_params": { - "pool": "optimized-baseline", - "max_concurrency": "100", - "baseline": "0.05" +{ + "result_queue_name": "llm-d-async:results:optimized-baseline:$batch", + "queues": [ + { + "queue_name": "llm-d-async:requests:optimized-baseline", + "igw_base_url": "http://llm-d-inference-gateway-istio:80", + "request_path_url": "/v1/completions", + "gate_type": "prometheus-budget", + "gate_params": { + "pool": "optimized-baseline", + "max_concurrency": "100", + "baseline": "0.05" + } } - } -] -``` - -``` ---redis.ss.result-queue-name llm-d-async:results:optimized-baseline:$batch + ] +} ``` The queue names must match those derived by the batch-processor's `RequestQueueName()` and `ResultQueueName()` functions. diff --git a/docs/guides/deploy-k8s.md b/docs/guides/deploy-k8s.md index 82b495a7a..a73f2aae3 100644 --- a/docs/guides/deploy-k8s.md +++ b/docs/guides/deploy-k8s.md @@ -1050,8 +1050,8 @@ kubectl rollout status deployment/${PROMETHEUS_NAME} -n ${LLM_NAMESPACE} --timeo ```bash DISPATCHER_RELEASE=dispatcher DISPATCHER_VERSION=${DISPATCHER_VERSION:-v0.7.3} -DISPATCHER_IMAGE="ghcr.io/llm-d-incubation/llm-d-async:${DISPATCHER_VERSION}" -DISPATCHER_CHART="oci://ghcr.io/llm-d-incubation/charts/async-processor" +DISPATCHER_IMAGE="ghcr.io/llm-d/llm-d-async:${DISPATCHER_VERSION}" +DISPATCHER_CHART="oci://ghcr.io/llm-d/charts/llm-d-async" REDIS_SVC=redis-master # or redis-valkey-primary for Valkey REDIS_HOST="${REDIS_SVC}.${BATCH_NAMESPACE}.svc.cluster.local" @@ -1065,16 +1065,17 @@ PROMETHEUS_URL="http://prometheus.${LLM_NAMESPACE}.svc.cluster.local:9090" cat > /tmp/dispatcher-values.yaml < "${values_file}" </dev/null || echo "") @@ -1230,7 +1231,7 @@ EOF || test_failures=$? if [ "${ENABLE_DISPATCHER}" = "true" ]; then - verify_dispatcher_runtime "${DISPATCHER_RELEASE}-async-processor" "${LLMD_POOL_NAME}" + verify_dispatcher_runtime "${DISPATCHER_RELEASE}-llm-d-async" "${LLMD_POOL_NAME}" fi if [ "${ENABLE_FLOW_CONTROL}" = "true" ]; then diff --git a/scripts/dev-deploy-dispatcher.sh b/scripts/dev-deploy-dispatcher.sh index c119c4ea5..e5e951582 100755 --- a/scripts/dev-deploy-dispatcher.sh +++ b/scripts/dev-deploy-dispatcher.sh @@ -13,7 +13,7 @@ KIND_CLUSTER_NAME="${KIND_CLUSTER_NAME:-batch-gateway-dev}" DISPATCHER_RELEASE="${DISPATCHER_RELEASE:-dispatcher}" DISPATCHER_VERSION="${DISPATCHER_VERSION:-v0.7.4}" DISPATCHER_IMAGE="${DISPATCHER_IMAGE:-ghcr.io/llm-d/llm-d-async:${DISPATCHER_VERSION}}" -DISPATCHER_CHART="${DISPATCHER_CHART:-oci://ghcr.io/llm-d/charts/async-processor}" +DISPATCHER_CHART="${DISPATCHER_CHART:-oci://ghcr.io/llm-d/charts/llm-d-async}" DISPATCHER_CHART_VERSION="${DISPATCHER_CHART_VERSION:-0.7.4}" DISPATCHER_REDIS_PORT="${DISPATCHER_REDIS_PORT:-6399}" PID_FILE="${REPO_ROOT}/.dispatcher-port-forward.pid" @@ -49,7 +49,7 @@ if [[ -n "${DISPATCHER_SOURCE}" ]]; then die "DISPATCHER_SOURCE directory not found: ${DISPATCHER_SOURCE}" fi DISPATCHER_IMAGE="ghcr.io/llm-d/llm-d-async:dev-local" - DISPATCHER_CHART="${DISPATCHER_SOURCE}/charts/async-processor" + DISPATCHER_CHART="${DISPATCHER_SOURCE}/charts/llm-d-async" unset DISPATCHER_CHART_VERSION step "Building async-processor image from ${DISPATCHER_SOURCE}..." ${CONTAINER_TOOL} build -t "${DISPATCHER_IMAGE}" "${DISPATCHER_SOURCE}" @@ -124,11 +124,11 @@ log "Dispatchers deployed." # ── Verify dispatchers ─────────────────────────────────────────────────────── step "Waiting for dispatcher pods to be ready..." -kubectl wait --for=condition=available deployment/"${DISPATCHER_RELEASE}-async-processor" \ +kubectl wait --for=condition=available deployment/"${DISPATCHER_RELEASE}-llm-d-async" \ --namespace "${NAMESPACE}" --timeout=60s -kubectl wait --for=condition=available deployment/"${DISPATCHER_SCRAPE_RELEASE}-async-processor" \ +kubectl wait --for=condition=available deployment/"${DISPATCHER_SCRAPE_RELEASE}-llm-d-async" \ --namespace "${NAMESPACE}" --timeout=60s -kubectl wait --for=condition=available deployment/"${DISPATCHER_PROM_RELEASE}-async-processor" \ +kubectl wait --for=condition=available deployment/"${DISPATCHER_PROM_RELEASE}-llm-d-async" \ --namespace "${NAMESPACE}" --timeout=60s # ── Add vllm-sim to Prometheus scrape targets ──────────────────────────────── diff --git a/test/e2e/dispatcher/helm-values-prometheus.yaml b/test/e2e/dispatcher/helm-values-prometheus.yaml index 4243c1f91..690641a31 100644 --- a/test/e2e/dispatcher/helm-values-prometheus.yaml +++ b/test/e2e/dispatcher/helm-values-prometheus.yaml @@ -1,6 +1,6 @@ ap: imagePullPolicy: Never - messageQueueImpl: "redis-sortedset" + transport: "redis-sortedset" concurrency: 1 prometheusURL: "http://prometheus.default.svc.cluster.local:9090" prometheusCacheTTL: "0s" @@ -9,17 +9,17 @@ ap: insecure: true sampler: "always_on" samplerArg: "1.0" - redisTracing: false metrics: enabled: true port: 9092 secure: false - redis: - enabled: true - url: "redis://redis-master.default.svc.cluster.local:6379" - pollIntervalMs: 500 - batchSize: 10 - queuesConfig: + transportConfig: + urlSecret: + url: "redis://redis-master.default.svc.cluster.local:6379" + result_queue_name: "result-list" + poll_interval_ms: 500 + batch_size: 10 + queues: - queue_name: "llm-d-async:requests:sim-pool-prom" result_queue_name: "llm-d-async:results:sim-pool-prom" request_path_url: "/v1/completions" diff --git a/test/e2e/dispatcher/helm-values-scrape.yaml b/test/e2e/dispatcher/helm-values-scrape.yaml index fc8b566b4..0f1158127 100644 --- a/test/e2e/dispatcher/helm-values-scrape.yaml +++ b/test/e2e/dispatcher/helm-values-scrape.yaml @@ -1,6 +1,6 @@ ap: imagePullPolicy: Never - messageQueueImpl: "redis-sortedset" + transport: "redis-sortedset" concurrency: 1 prometheusCacheTTL: "0s" otel: @@ -8,17 +8,17 @@ ap: insecure: true sampler: "always_on" samplerArg: "1.0" - redisTracing: false metrics: enabled: true port: 9091 secure: false - redis: - enabled: true - url: "redis://redis-master.default.svc.cluster.local:6379" - pollIntervalMs: 500 - batchSize: 10 - queuesConfig: + transportConfig: + urlSecret: + url: "redis://redis-master.default.svc.cluster.local:6379" + result_queue_name: "result-list" + poll_interval_ms: 500 + batch_size: 10 + queues: - queue_name: "llm-d-async:requests:sim-pool-scrape" result_queue_name: "llm-d-async:results:sim-pool-scrape" request_path_url: "/v1/completions" diff --git a/test/e2e/dispatcher/helm-values.yaml b/test/e2e/dispatcher/helm-values.yaml index 70df79ade..8e19a4223 100644 --- a/test/e2e/dispatcher/helm-values.yaml +++ b/test/e2e/dispatcher/helm-values.yaml @@ -1,6 +1,6 @@ ap: imagePullPolicy: Never - messageQueueImpl: "redis-sortedset" + transport: "redis-sortedset" concurrency: 1 prometheusCacheTTL: "0s" otel: @@ -8,17 +8,17 @@ ap: insecure: true sampler: "always_on" samplerArg: "1.0" - redisTracing: false metrics: enabled: true port: 9090 secure: false - redis: - enabled: true - url: "redis://redis-master.default.svc.cluster.local:6379" - pollIntervalMs: 500 - batchSize: 10 - queuesConfig: + transportConfig: + urlSecret: + url: "redis://redis-master.default.svc.cluster.local:6379" + result_queue_name: "result-list" + poll_interval_ms: 500 + batch_size: 10 + queues: - queue_name: "llm-d-async:requests:sim-pool" result_queue_name: "llm-d-async:results:sim-pool" request_path_url: "/v1/completions" diff --git a/test/e2e/dispatcher_otel_test.go b/test/e2e/dispatcher_otel_test.go index 8e73de61b..c5d018031 100644 --- a/test/e2e/dispatcher_otel_test.go +++ b/test/e2e/dispatcher_otel_test.go @@ -45,7 +45,7 @@ func TestDispatcherOTelTraces(t *testing.T) { // testCrossServiceTracePropagation verifies that a batch processed through the // async dispatcher produces a connected trace spanning both batch-gateway and -// async-processor. The batch-gateway injects trace context into +// llm-d-async. The batch-gateway injects trace context into // RequestMessage.Metadata, the async processor extracts it, and both services // export spans to the same Jaeger instance under the same trace ID. func testCrossServiceTracePropagation(t *testing.T, jaegerClient *http.Client) { @@ -62,8 +62,8 @@ func testCrossServiceTracePropagation(t *testing.T, jaegerClient *http.Client) { t.Fatalf("Expected 1 completed request, got %d", batch.RequestCounts.Completed) } - // Poll Jaeger for traces from batch-gateway that contain async-processor spans. - // The batch-gateway creates a "process-batch" span, and the async-processor + // Poll Jaeger for traces from batch-gateway that contain llm-d-async spans. + // The batch-gateway creates a "process-batch" span, and llm-d-async // creates a "process-request" child span under the same trace ID. type jaegerSpan struct { OperationName string `json:"operationName"` @@ -89,7 +89,7 @@ func testCrossServiceTracePropagation(t *testing.T, jaegerClient *http.Client) { default: } - resp, err := jaegerClient.Get(testJaegerURL + "/api/traces?service=async-processor&limit=20&lookback=5m") + resp, err := jaegerClient.Get(testJaegerURL + "/api/traces?service=llm-d-async&limit=20&lookback=5m") if err != nil { t.Logf("Jaeger query failed (retrying): %v", err) time.Sleep(2 * time.Second) @@ -114,7 +114,7 @@ func testCrossServiceTracePropagation(t *testing.T, jaegerClient *http.Client) { services[proc.ServiceName] = true } } - if services["batch-gateway"] && services["async-processor"] { + if services["batch-gateway"] && services["llm-d-async"] { found = trace break } @@ -132,7 +132,7 @@ func testCrossServiceTracePropagation(t *testing.T, jaegerClient *http.Client) { } if !spanOps["process-request"] { - t.Error("Expected 'process-request' span from async-processor in trace") + t.Error("Expected 'process-request' span from llm-d-async in trace") } // Collect service names for logging diff --git a/test/e2e/dispatcher_test.go b/test/e2e/dispatcher_test.go index cc153b7ea..5b0ee8cb0 100644 --- a/test/e2e/dispatcher_test.go +++ b/test/e2e/dispatcher_test.go @@ -88,7 +88,7 @@ func newDispatcherProducer(t *testing.T, rdb *redis.Client, poolName string) *pr return p } -// detectDispatcherDeployed checks whether at least one async-processor +// detectDispatcherDeployed checks whether at least one llm-d-async // deployment exists in the test namespace. func detectDispatcherDeployed(t *testing.T) bool { t.Helper() @@ -102,7 +102,7 @@ func detectDispatcherDeployed(t *testing.T) bool { return false } for _, line := range strings.Split(string(out), "\n") { - if strings.Contains(line, "async-processor") { + if strings.Contains(line, "llm-d-async") { return true } } @@ -160,7 +160,7 @@ func TestDispatcher(t *testing.T) { // from the async ResultMessage (e.g. 403 with an empty body) is preserved in // the batch output file instead of being collapsed into parse_error. // -// Uses the consumer-less "sim-pool-inject" pool: no async-processor subscribes +// Uses the consumer-less "sim-pool-inject" pool: no llm-d-async subscribes // to it, so the request stays in the queue deterministically until the test // removes it and injects a synthetic ResultMessage. func testDispatcherHTTPErrorStatusPreserved(t *testing.T, rdb *redis.Client) {