Skip to content

feat: e2e dispatcher integration #458

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lioraron merged 10 commits into from
Jul 3, 2026
Merged

feat: e2e dispatcher integration #458

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
1e0aa25
feat: e2e dispatcher integration
evacchi May 27, 2026
0da222f
bump llm-d-async from v0.7.0-rc.6 to v0.7.2
evacchi Jun 29, 2026
bd37da6
address comments
evacchi Jul 1, 2026
21e186e
address comments
evacchi Jul 1, 2026
9b51646
address comments
evacchi Jul 1, 2026
640125b
fix: ensure per-queue config is used
evacchi Jul 2, 2026
7da991a
consolidate testing scripts
evacchi Jul 2, 2026
9b2a183
enable tests on CI
evacchi Jul 2, 2026
f092e6d
address comments
evacchi Jul 3, 2026
fa335e6
Merge branch 'main' into e2e-dispatcher
lioraron Jul 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion .github/workflows/ci-integration-tests.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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

Expand All @@ -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
Expand Down
3 changes: 3 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
bin/
.build/
.vscode/
.claude/
.cursor/
Expand All @@ -8,3 +9,5 @@ __pycache__/
*.pyc
benchmarks/results/*
!benchmarks/results/.gitkeep
.dispatcher-port-forward.pid
.dispatcher-sim-port-forward.pid
22 changes: 19 additions & 3 deletions charts/batch-gateway/templates/processor-configmap.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 }}

@yizhaodev yizhaodev Jul 2, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do you need inference_pool_name for global_inference_gateway?

because i saw this in below

return fmt.Errorf("global_inference_gateway is not supported with dispatch_mode %q; use model_gateways with inference_pool_name", DispatchModeAsync)

@evacchi evacchi Jul 3, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

no, global_inference_gateway doesn't need inference_pool_name. Async mode explicitly rejects global_inference_gateway, validation fails with that error message.

inference_pool_name is only needed on model_gateways entries, which should be where the template renders it. The error message is telling users to switch from global_inference_gateway to model_gateways when using async mode

we can do amendments to the config format in another PR anyway

{{- end }}
request_timeout: {{ .requestTimeout | quote }}
max_retries: {{ .maxRetries }}
initial_backoff: {{ .initialBackoff | quote }}
Expand All @@ -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 }}
Expand Down
16 changes: 13 additions & 3 deletions cmd/batch-processor/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down
11 changes: 8 additions & 3 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -30,18 +32,21 @@ require (
)

require (
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
github.com/felixge/httpsnoop v1.0.4 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
github.com/jackc/puddle/v2 v2.2.2 // indirect
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
github.com/redis/go-redis/extra/rediscmd/v9 v9.21.0 // indirect
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect
go.opentelemetry.io/otel/metric v1.44.0 // indirect
go.opentelemetry.io/proto/otlp v1.10.0 // indirect
golang.org/x/text v0.37.0 // indirect
golang.org/x/time v0.15.0 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
google.golang.org/grpc v1.81.1 // indirect
Expand All @@ -67,11 +72,11 @@ require (
github.com/go-logr/logr v1.4.3
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/prometheus/client_model v0.6.2
github.com/prometheus/common v0.66.1 // indirect
github.com/prometheus/procfs v0.16.1 // indirect
github.com/prometheus/common v0.67.5 // indirect
github.com/prometheus/procfs v0.17.0 // indirect
github.com/yuin/gopher-lua v1.1.1 // indirect
go.uber.org/atomic v1.11.0 // indirect
go.yaml.in/yaml/v2 v2.4.2 // indirect
go.yaml.in/yaml/v2 v2.4.3 // indirect
golang.org/x/net v0.55.0 // indirect
golang.org/x/sys v0.45.0 // indirect
google.golang.org/protobuf v1.36.11 // indirect
Expand Down
27 changes: 16 additions & 11 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,8 @@ github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F9
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/exaring/otelpgx v0.11.1 h1:pE79fIg/qh/Lpu00kvswFC5dKfqyJJhMJ4Y4N3w5Lj4=
github.com/exaring/otelpgx v0.11.1/go.mod h1:3OojrUKhhy3lTbYIMBijP3YjMey/jo14eHAW5cXcUdk=
github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
Expand Down Expand Up @@ -88,20 +88,25 @@ 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=
github.com/pashagolub/pgxmock/v4 v4.9.0/go.mod h1:9L57pC193h2aKRHVyiiE817avasIPZnPwPlw3JczWvM=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o=
github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg=
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs=
github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
github.com/prometheus/common v0.67.5 h1:pIgK94WWlQt1WLwAC5j2ynLaBRDiinoAb86HZHTUGI4=
github.com/prometheus/common v0.67.5/go.mod h1:SjE/0MzDEEAyrdr5Gqc6G+sXI67maCxzaT3A2+HqjUw=
github.com/prometheus/procfs v0.17.0 h1:FuLQ+05u4ZI+SS/w9+BWEM2TXiHKsUQ9TADiRH7DuK0=
github.com/prometheus/procfs v0.17.0/go.mod h1:oPQLaDAMRbA+u8H5Pbfq+dl3VDAvHxMUOVhe0wYB2zw=
github.com/quasilyte/go-ruleguard/dsl v0.3.23 h1:lxjt5B6ZCiBeeNO8/oQsegE6fLeCzuMRoVWSkXC4uvY=
github.com/quasilyte/go-ruleguard/dsl v0.3.23/go.mod h1:KeCP03KrjuSO0H1kTuZQCWlQPulDV6YMIXmpQss17rU=
github.com/redis/go-redis/extra/rediscmd/v9 v9.21.0 h1:jsV3tyMeJrEoc2f3EhNf7qoBW3NEZW7l/4ziT3M+OJI=
Expand Down Expand Up @@ -145,8 +150,8 @@ go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0=
go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8=
golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8=
golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww=
golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM=
Expand All @@ -155,8 +160,8 @@ golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE=
golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8=
Expand Down
61 changes: 26 additions & 35 deletions internal/processor/config/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Comment thread
evacchi marked this conversation as resolved.
// IsAsync returns true when the processor is configured for async dispatch.
func (c *ProcessorConfig) IsAsync() bool {
return c.DispatchMode == DispatchModeAsync
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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 {

@acardace acardace Jun 16, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: don't you think all of this is still gateways validation? Maybe all the logic should be in that single function validateGateways and maybe we could even drop the switch case.

@evacchi evacchi Jun 19, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I folded the logic for sync/async+validateDispatchMode into validateGateways, I hope that's what you meant.

return err
return fmt.Errorf("async_dispatch.result_poll_timeout must be > 0")
Comment thread
evacchi marked this conversation as resolved.
}
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 {

@acardace acardace Jun 16, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wasn't introduced by this PR but I think having InferencePoolName as the only useful field for async buried in ModelGateways where all the other fields are effectively ignored is not the best API design, maybe we should have a separate config for Async?

@evacchi evacchi Jun 19, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

agreed, but this should be addressed separately, let's create an issue

@evacchi evacchi Jun 30, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

created #526

return fmt.Errorf("model_gateways must be configured when dispatch_mode is %q", DispatchModeAsync)
}
for model, gw := range c.ModelGateways {
if gw.InferencePoolName == "" {
Expand Down Expand Up @@ -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
Expand All @@ -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 {
Expand Down
Loading
Loading