feat: e2e dispatcher integration

.github/workflows/ci-integration-tests.yml

@@ -19,7 +19,7 @@ permissions:
 
 jobs:
   integration-tests:
-    name: Integration Tests (${{ matrix.file_client_type }}, ${{ matrix.db_client_type }}, ${{ matrix.exchange_client_type }}${{ matrix.enable_gie == 'true' && ', gie' || '' }})
+    name: Integration Tests (${{ matrix.file_client_type }}, ${{ matrix.db_client_type }}, ${{ matrix.exchange_client_type }}${{ matrix.enable_gie == 'true' && ', gie' || '' }}${{ matrix.enable_dispatcher == 'true' && ', dispatcher' || '' }})
     runs-on: ubuntu-latest
     timeout-minutes: 30
     strategy:
@@ -45,6 +45,10 @@ jobs:
             db_client_type: postgresql
             exchange_client_type: redis
             enable_gie: "true"
+          - file_client_type: s3
+            db_client_type: postgresql
+            exchange_client_type: redis
+            enable_dispatcher: "true"
     steps:
       - uses: actions/checkout@v7
 
@@ -65,6 +69,7 @@ jobs:
           DB_CLIENT_TYPE: ${{ matrix.db_client_type }}
           EXCHANGE_CLIENT_TYPE: ${{ matrix.exchange_client_type }}
           ENABLE_GIE: ${{ matrix.enable_gie || 'false' }}
+          ENABLE_DISPATCHER: ${{ matrix.enable_dispatcher || 'false' }}
 
       - name: Run integration tests
         run: make test-e2e

.gitignore

@@ -1,4 +1,5 @@
 bin/
+.build/
 .vscode/
 .claude/
 .cursor/
@@ -8,3 +9,5 @@ __pycache__/
 *.pyc
 benchmarks/results/*
 !benchmarks/results/.gitkeep
+.dispatcher-port-forward.pid
+.dispatcher-sim-port-forward.pid

charts/batch-gateway/templates/processor-configmap.yaml

@@ -42,12 +42,23 @@ data:
         conn_max_idle_time: {{ .Values.global.dbClient.redis.connMaxIdleTime | quote }}
         conn_max_lifetime: {{ .Values.global.dbClient.redis.connMaxLifetime | quote }}
 
+    {{- if .Values.processor.config.dispatchMode }}
+    dispatch_mode: {{ .Values.processor.config.dispatchMode | quote }}
+    {{- end }}
+    {{- if .Values.processor.config.asyncDispatch }}
+    async_dispatch:
+      result_poll_timeout: {{ .Values.processor.config.asyncDispatch.resultPollTimeout | quote }}
+    {{- end }}
+
     {{- with .Values.processor.config.globalInferenceGateway }}
     global_inference_gateway:
       url: {{ .url | quote }}
       {{- if .inferenceObjective }}
       inference_objective: {{ .inferenceObjective | quote }}
       {{- end }}
+      {{- if .inferencePoolName }}
+      inference_pool_name: {{ .inferencePoolName | quote }}
+      {{- end }}
       request_timeout: {{ .requestTimeout | quote }}
       max_retries: {{ .maxRetries }}
       initial_backoff: {{ .initialBackoff | quote }}
@@ -74,15 +85,20 @@ data:
     model_gateways:
       {{- range $model, $cfg := .Values.processor.config.modelGateways }}
       {{ $model | quote }}:
-        url: {{ $cfg.url | quote }}
-        {{- if $cfg.inferenceObjective }}
-        inference_objective: {{ $cfg.inferenceObjective | quote }}
+        {{- if $cfg.inferencePoolName }}
+        inference_pool_name: {{ $cfg.inferencePoolName | quote }}
         {{- end }}
+        {{- if $cfg.url }}
+        url: {{ $cfg.url | quote }}
         request_timeout: {{ $cfg.requestTimeout | quote }}
         max_retries: {{ $cfg.maxRetries }}
         initial_backoff: {{ $cfg.initialBackoff | quote }}
         max_backoff: {{ $cfg.maxBackoff | quote }}
         tls_insecure_skip_verify: {{ $cfg.tlsInsecureSkipVerify | default false }}
+        {{- end }}
+        {{- if $cfg.inferenceObjective }}
+        inference_objective: {{ $cfg.inferenceObjective | quote }}
+        {{- end }}
         {{- if $cfg.apiKeyName }}
         api_key_name: {{ $cfg.apiKeyName | quote }}
         {{- end }}

cmd/batch-processor/main.go

@@ -296,19 +296,29 @@ func buildProcessorClients(ctx context.Context, cfg *config.ProcessorConfig) (*c
 	if len(resolved.PerModel) > 0 {
 		opts = append(opts, clientset.WithPerModelInference(resolved.PerModel))
 	}
+	if resolved.Async != nil {
+		opts = append(opts, clientset.WithAsyncInference(*resolved.Async))
+	}
 	clients, err := clientset.NewClientset(ctx, ucom.ComponentProcessor, opts...)
 	if err != nil {
 		logger.Error(err, "Failed to create clients")
 		return nil, err
 	}
 
-	// Validate() guarantees exactly one of resolved.Global or resolved.PerModel is set.
-	if resolved.Global != nil {
+	// ResolveModelGateways populates exactly one of Async, Global, or PerModel
+	// based on the dispatch mode validated by Validate().
+	switch {
+	case resolved.Async != nil:
+		logger.V(logging.INFO).Info("Processor clients initialized",
+			"mode", "async",
+			"numModels", len(resolved.Async.Models),
+			"fileClientType", cfg.FileClientCfg.Type)
+	case resolved.Global != nil:
 		logger.V(logging.INFO).Info("Processor clients initialized",
 			"mode", "global",
 			"gatewayURL", resolved.Global.URL,
 			"fileClientType", cfg.FileClientCfg.Type)
-	} else {
+	default:
 		logger.V(logging.INFO).Info("Processor clients initialized",
 			"mode", "per-model",
 			"numModelGateways", len(resolved.PerModel),

go.mod

@@ -14,6 +14,8 @@ require (
 	github.com/go-resty/resty/v2 v2.17.2
 	github.com/google/uuid v1.6.0
 	github.com/jackc/pgx/v5 v5.10.0
+	github.com/llm-d-incubation/llm-d-async/api v0.7.2
+	github.com/llm-d-incubation/llm-d-async/producer v0.7.2
 	github.com/pashagolub/pgxmock/v4 v4.9.0
 	github.com/prometheus/client_golang v1.23.2
 	github.com/quasilyte/go-ruleguard/dsl v0.3.23

go.sum

@@ -88,6 +88,10 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/llm-d-incubation/llm-d-async/api v0.7.2 h1:sf6iFDa5LpVoKDYiOOlsbnv+6Ykj5TA22yRqMdIbOfY=
+github.com/llm-d-incubation/llm-d-async/api v0.7.2/go.mod h1:m2zJUwD/AZypJv8RPws3uvCnfQoNW7iv+hH9wPk/Xw4=
+github.com/llm-d-incubation/llm-d-async/producer v0.7.2 h1:hhnLqi+MXq8kBjsQhnUCWYtTCG6uFTZLCH296CZMlpY=
+github.com/llm-d-incubation/llm-d-async/producer v0.7.2/go.mod h1:IrClO4XwMtgLRnlkAqO1LDsuqmwtQjELDabT2f+5d40=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/pashagolub/pgxmock/v4 v4.9.0 h1:itlO8nrVRnzkdMBXLs8pWUyyB2PC3Gku0WGIj/gGl7I=

internal/processor/config/config.go

@@ -230,18 +230,6 @@ type BucketConfig struct {
 	BucketCount  int     `yaml:"count"`
 }
 
-const asyncTenantID = "$batch"
-
-// RequestQueueName returns the Redis sorted-set name for submitting async requests to the given pool.
-func RequestQueueName(poolName string) string {
-	return "llm-d-async:requests:" + poolName
-}
-
-// ResultQueueName returns the Redis list name for collecting async results from the given pool.
-func ResultQueueName(poolName string) string {
-	return "llm-d-async:results:" + poolName + ":" + asyncTenantID
-}
-
 // IsAsync returns true when the processor is configured for async dispatch.
 func (c *ProcessorConfig) IsAsync() bool {
 	return c.DispatchMode == DispatchModeAsync
@@ -386,14 +374,21 @@ func (c *ProcessorConfig) Validate() error {
 		return fmt.Errorf("progress_ttl_seconds must be > 0")
 	}
 
-	if err := c.validateDispatchMode(); err != nil {
+	if err := c.validateGateways(); err != nil {
 		return err
 	}
 
 	return nil
 }
 
-func (c *ProcessorConfig) validateDispatchMode() error {
+func (c *ProcessorConfig) validateGateways() error {
+	if c.GlobalInferenceGateway == nil && len(c.ModelGateways) == 0 {
+		return fmt.Errorf("either global_inference_gateway or model_gateways must be configured")
+	}
+	if c.GlobalInferenceGateway != nil && len(c.ModelGateways) > 0 {
+		return fmt.Errorf("global_inference_gateway and model_gateways are mutually exclusive")
+	}
+
 	switch c.DispatchMode {
 	case DispatchModeSync, DispatchMode(""):
 		c.DispatchMode = DispatchModeSync
@@ -405,21 +400,7 @@ func (c *ProcessorConfig) validateDispatchMode() error {
 	}
 }
 
-func (c *ProcessorConfig) validateGateways() error {
-	if c.GlobalInferenceGateway == nil && len(c.ModelGateways) == 0 {
-		return fmt.Errorf("either global_inference_gateway or model_gateways must be configured")
-	}
-	if c.GlobalInferenceGateway != nil && len(c.ModelGateways) > 0 {
-		return fmt.Errorf("global_inference_gateway and model_gateways are mutually exclusive")
-	}
-	return nil
-}
-
 func (c *ProcessorConfig) validateSyncDispatchConfig() error {
-	if err := c.validateGateways(); err != nil {
-		return err
-	}
-
 	if c.GlobalInferenceGateway != nil {
 		if err := validateGatewayConfig("global_inference_gateway", *c.GlobalInferenceGateway); err != nil {
 			return err
@@ -435,15 +416,13 @@ func (c *ProcessorConfig) validateSyncDispatchConfig() error {
 
 func (c *ProcessorConfig) validateAsyncDispatchConfig() error {
 	if c.AsyncDispatchConfig.ResultPollTimeout <= 0 {
-		return fmt.Errorf("async.result_poll_timeout must be > 0")
-	}
-	if err := c.validateGateways(); err != nil {
-		return err
+		return fmt.Errorf("async_dispatch.result_poll_timeout must be > 0")
 	}
 	if c.GlobalInferenceGateway != nil {
-		if c.GlobalInferenceGateway.InferencePoolName == "" {
-			return fmt.Errorf("global_inference_gateway.inference_pool_name must be set when dispatch_mode is %q", DispatchModeAsync)
-		}
+		return fmt.Errorf("global_inference_gateway is not supported with dispatch_mode %q; use model_gateways with inference_pool_name", DispatchModeAsync)
+	}
+	if len(c.ModelGateways) == 0 {
+		return fmt.Errorf("model_gateways must be configured when dispatch_mode is %q", DispatchModeAsync)
 	}
 	for model, gw := range c.ModelGateways {
 		if gw.InferencePoolName == "" {
@@ -580,6 +559,7 @@ func toGatewayClientConfig(gw ModelGatewayConfig, apiKey string) inference.Gatew
 type ResolvedGateways struct {
 	Global   *inference.GatewayClientConfig
 	PerModel map[string]inference.GatewayClientConfig
+	Async    *inference.AsyncClientConfig
 }
 
 // ResolveModelGateways resolves API keys for all configured gateways and returns
@@ -588,6 +568,17 @@ type ResolvedGateways struct {
 func ResolveModelGateways(cfg *ProcessorConfig) (*ResolvedGateways, error) {
 	result := &ResolvedGateways{}
 
+	if cfg.IsAsync() {
+		models := make(map[string]string, len(cfg.ModelGateways))
+		for model, gw := range cfg.ModelGateways {
+			models[model] = gw.InferencePoolName
+		}
+		result.Async = &inference.AsyncClientConfig{
+			Models: models,
+		}
+		return result, nil
+	}
+
 	if cfg.GlobalInferenceGateway != nil {
 		apiKey, err := resolveGatewayAPIKey("global_inference_gateway", *cfg.GlobalInferenceGateway)
 		if err != nil {

internal/processor/config/config_test.go

@@ -715,12 +715,12 @@ func TestProcessorConfig_Validate_AsyncDispatch(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "async valid global gateway with inference_pool_name",
+			name: "async global gateway rejected",
 			mutate: func(c *ProcessorConfig) {
 				c.ModelGateways = nil
 				c.GlobalInferenceGateway = &ModelGatewayConfig{URL: "http://gw:8000", InferencePoolName: "default-pool"}
 			},
-			wantErr: false,
+			wantErr: true,
 		},
 		{
 			name: "async no gateways configured",
@@ -784,15 +784,6 @@ func TestProcessorConfig_Validate_AsyncDispatch(t *testing.T) {
 	}
 }
 
-func TestQueueNameHelpers(t *testing.T) {
-	if got := RequestQueueName("pool-a"); got != "llm-d-async:requests:pool-a" {
-		t.Fatalf("RequestQueueName(\"pool-a\") = %q, want %q", got, "llm-d-async:requests:pool-a")
-	}
-	if got := ResultQueueName("pool-a"); got != "llm-d-async:results:pool-a:$batch" {
-		t.Fatalf("ResultQueueName(\"pool-a\") = %q, want %q", got, "llm-d-async:results:pool-a:$batch")
-	}
-}
-
 func TestValidate_NormalizesEmptyDispatchMode(t *testing.T) {
 	c := NewConfig()
 	c.ModelGateways = validPerModelConfig()
@@ -919,3 +910,58 @@ func TestProcessorConfig_InferenceObjectiveFor(t *testing.T) {
 		})
 	}
 }
+
+func TestResolveModelGateways_Async(t *testing.T) {
+	t.Run("populates Async field", func(t *testing.T) {
+		cfg := NewConfig()
+		cfg.DispatchMode = DispatchModeAsync
+		cfg.AsyncDispatchConfig = AsyncDispatchConfig{
+			ResultPollTimeout: 10 * time.Second,
+		}
+		cfg.ModelGateways = map[string]ModelGatewayConfig{
+			"model-a": {InferencePoolName: "pool-a"},
+			"model-b": {InferencePoolName: "pool-b"},
+		}
+
+		resolved, err := ResolveModelGateways(cfg)
+		if err != nil {
+			t.Fatalf("ResolveModelGateways() error: %v", err)
+		}
+
+		if resolved.Async == nil {
+			t.Fatal("expected Async to be set")
+		}
+		if resolved.Global != nil {
+			t.Error("expected Global to be nil in async mode")
+		}
+		if resolved.PerModel != nil {
+			t.Error("expected PerModel to be nil in async mode")
+		}
+		if len(resolved.Async.Models) != 2 {
+			t.Fatalf("Models count = %d, want 2", len(resolved.Async.Models))
+		}
+		if resolved.Async.Models["model-a"] != "pool-a" {
+			t.Errorf("Models[model-a] = %q, want %q", resolved.Async.Models["model-a"], "pool-a")
+		}
+		if resolved.Async.Models["model-b"] != "pool-b" {
+			t.Errorf("Models[model-b] = %q, want %q", resolved.Async.Models["model-b"], "pool-b")
+		}
+	})
+
+	t.Run("sync mode does not populate Async", func(t *testing.T) {
+		cfg := NewConfig()
+		cfg.ModelGateways = validPerModelConfig()
+
+		resolved, err := ResolveModelGateways(cfg)
+		if err != nil {
+			t.Fatalf("ResolveModelGateways() error: %v", err)
+		}
+
+		if resolved.Async != nil {
+			t.Error("expected Async to be nil in sync mode")
+		}
+		if len(resolved.PerModel) == 0 {
+			t.Error("expected PerModel to be populated in sync mode")
+		}
+	})
+}