Proposal: Support Configurable Orchestration of Inference Pipeline#1249
Proposal: Support Configurable Orchestration of Inference Pipeline#1249shivansh-gohem wants to merge 1 commit into
Conversation
volcano-sh-bot
requested review from
LiZhenCheng9527,
YaoZengzeng,
git-malu and
hzxuzhonghu
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements native Encode-Prefill-Decode (EPD) disaggregated routing support in the Kthena Router. It introduces pipelineMode and encodeLabels to the ModelServer CRD and Go API, updates the datastore to track encode pods, adds a ProxyEPD method to the KVConnector interface, and integrates EPD scheduling and routing. The review feedback highlights a critical defensive programming issue in proxyToEPDDisaggregated where potential nil pointer dereferences and missing IP validations on the selected pods could lead to runtime panics or routing failures.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
There was a problem hiding this comment.
Pull request overview
This PR adds user-configurable routing modes for the Kthena Router by introducing spec.pipelineMode on ModelServerSpec and extending PDGroup role selection to include encode pods, with an initial router path intended to support EPD (Encode/Prefill/Decode) disaggregation.
Changes:
- Added
PipelineMode(vllm-epd,sglang-epd,pd-disaggregated) to the ModelServer API, updated CRD schema/docs, and regenerated client/apply config + deepcopy. - Extended PDGroup pod categorization and scheduler context to include
EncodePodsand store lookups for encode pods. - Introduced an EPD routing path (
proxyToEPDDisaggregated) and a newKVConnector.ProxyEPDinterface method (currently stubbed in connectors shown).
Reviewed changes
Copilot reviewed 14 out of 17 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/kthena-router/scheduler/scheduler_impl.go | Fetches/scores encode pods and stores them in scheduling context. |
| pkg/kthena-router/scheduler/framework/interface.go | Adds EncodePods to the scheduler framework context. |
| pkg/kthena-router/router/router.go | Branches routing based on PipelineMode and adds proxyToEPDDisaggregated. |
| pkg/kthena-router/debug/handlers_test.go | Extends mock store with GetEncodePods. |
| pkg/kthena-router/datastore/store.go | Adds GetEncodePods to store interface + implementation. |
| pkg/kthena-router/datastore/pdgroup_pods.go | Tracks encode pods in PDGroup pod sets. |
| pkg/kthena-router/datastore/model_server.go | Categorizes pods into encode/prefill/decode sets and exposes all-encode aggregation. |
| pkg/kthena-router/connectors/interface.go | Adds ProxyEPD to KVConnector. |
| pkg/kthena-router/connectors/http.go | Adds a stub ProxyEPD method (returns not implemented). |
| pkg/kthena-router/connectors/nixl.go | Adds a stub ProxyEPD method (returns not implemented). |
| pkg/kthena-router/connectors/sglang.go | Adds a stub ProxyEPD method (returns not implemented). |
| pkg/apis/networking/v1alpha1/modelserver_types.go | Adds PipelineMode + PDGroup.EncodeLabels to the API types. |
| pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go | Regenerates deepcopy for new fields. |
| docs/kthena/docs/reference/crd/networking.serving.volcano.sh.md | Documents pipelineMode + encodeLabels. |
| client-go/applyconfiguration/networking/v1alpha1/modelserverspec.go | Adds applyconfig builder for PipelineMode. |
| client-go/applyconfiguration/networking/v1alpha1/pdgroup.go | Adds applyconfig builder for EncodeLabels. |
| charts/kthena/charts/networking/crds/networking.serving.volcano.sh_modelservers.yaml | Updates CRD schema for pipelineMode and encodeLabels. |
Files not reviewed (3)
- client-go/applyconfiguration/networking/v1alpha1/modelserverspec.go: Generated file
- client-go/applyconfiguration/networking/v1alpha1/pdgroup.go: Generated file
- pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go: Generated file
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
shivansh-gohem
force-pushed
the
feat/epd-orchestration-clean
branch
2 times, most recently
from
3ecc9e5 to
1b1b86d
Compare
shivansh-gohem
force-pushed
the
feat/epd-orchestration-clean
branch
from
1b1b86d to
6f24ca1
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 16 out of 19 changed files in this pull request and generated 4 comments.
Files not reviewed (3)
- client-go/applyconfiguration/networking/v1alpha1/modelserverspec.go: Generated file
- client-go/applyconfiguration/networking/v1alpha1/pdgroup.go: Generated file
- pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go: Generated file
| func (m *MockStore) GetEncodePods(modelServerName types.NamespacedName) ([]*datastore.PodInfo, error) { | ||
| args := m.Called(modelServerName) | ||
| if args.Get(0) == nil { | ||
| return nil, args.Error(1) | ||
| } | ||
| return args.Get(0).([]*datastore.PodInfo), args.Error(1) | ||
| } | ||
|
|
| // Check if pod matches decode labels | ||
| isDecodePod := matchesLabels(podLabels, pdGroup.DecodeLabels) | ||
| if isDecodePod { | ||
| pdGroupPods.AddDecodePod(podName) | ||
| return | ||
| } | ||
|
|
||
| // Check if pod matches prefill labels | ||
| isPrefillPod := matchesLabels(podLabels, pdGroup.PrefillLabels) | ||
| if isPrefillPod { | ||
| pdGroupPods.AddPrefillPod(podName) | ||
| return | ||
| } | ||
|
|
||
| // Check if pod matches encode labels (for EPD) | ||
| isEncodePod := matchesLabels(podLabels, pdGroup.EncodeLabels) | ||
| if isEncodePod { | ||
| pdGroupPods.AddEncodePod(podName) | ||
| } |
| // ProxyEPD executes the complete encode-prefill-decode flow for HTTP connector | ||
| func (h *HTTPConnector) ProxyEPD(c *gin.Context, reqBody map[string]interface{}, encodeAddr, prefillAddr, decodeAddr string, hooks *OnFlightHooks) (int, error) { | ||
| // For now, EPD is not implemented in the base HTTP connector. | ||
| // You can implement sequential encode -> prefill -> decode here if needed. | ||
| return 0, fmt.Errorf("EPD is not natively implemented for HTTPConnector yet") | ||
| } |
shivansh-gohem
force-pushed
the
feat/epd-orchestration-clean
branch
from
6f24ca1 to
62d17f3
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 16 out of 19 changed files in this pull request and generated 11 comments.
Files not reviewed (3)
- client-go/applyconfiguration/networking/v1alpha1/modelserverspec.go: Generated file
- client-go/applyconfiguration/networking/v1alpha1/pdgroup.go: Generated file
- pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go: Generated file
| modelServer := r.store.GetModelServer(ctx.ModelServerName) | ||
| if modelServer != nil && modelServer.Spec.PipelineMode != nil && | ||
| (*modelServer.Spec.PipelineMode == v1alpha1.PipelineModeEPD || *modelServer.Spec.PipelineMode == v1alpha1.PipelineModeSGLangEPD) { | ||
| // EPD disaggregated mode | ||
| return r.proxyToEPDDisaggregated(c, req, ctx, kvConnector, modelRequest, port) | ||
| } |
| // Check if pod matches decode labels | ||
| isDecodePod := matchesLabels(podLabels, pdGroup.DecodeLabels) | ||
| if isDecodePod { | ||
| pdGroupPods.AddDecodePod(podName) | ||
| return | ||
| } | ||
|
|
||
| // Check if pod matches prefill labels | ||
| isPrefillPod := matchesLabels(podLabels, pdGroup.PrefillLabels) | ||
| if isPrefillPod { | ||
| pdGroupPods.AddPrefillPod(podName) | ||
| } |
| // Get encode pods for the same PD group as the decode pod (O(1) lookup) | ||
| // Only for EPD mode. If encode pods are not found or none available, we just don't populate it for this index. | ||
| selectedEncodePods, err := s.store.GetEncodePodsForDecodeGroup(ctx.ModelServerName, decodePodName) | ||
| if err == nil && len(selectedEncodePods) > 0 { | ||
| klog.V(4).Info("Running score plugins for encode pod") | ||
| encodeScores := s.RunScorePlugins(selectedEncodePods, ctx) | ||
| bestEncodePod := TopNPodInfos(encodeScores, 1) | ||
| if len(bestEncodePod) > 0 { | ||
| // Initialize ctx.EncodePods if not already done | ||
| if len(ctx.EncodePods) != len(topNDecodePods) { | ||
| ctx.EncodePods = make([]*datastore.PodInfo, len(topNDecodePods)) | ||
| } | ||
| ctx.EncodePods[i] = bestEncodePod[0] | ||
| } | ||
| } |
| // ProxyEPD executes the complete encode-prefill-decode flow natively for NIXL. | ||
| func (n *NIXLConnector) ProxyEPD(c *gin.Context, reqBody map[string]interface{}, encodeAddr, prefillAddr, decodeAddr string, hooks *OnFlightHooks) (int, error) { | ||
| return http.StatusNotImplemented, fmt.Errorf("EPD is not natively implemented for NIXLConnector yet") | ||
| } |
| // ProxyEPD executes the complete encode-prefill-decode flow natively for SGLang. | ||
| func (s *SGLangConnector) ProxyEPD(c *gin.Context, reqBody map[string]interface{}, encodeAddr, prefillAddr, decodeAddr string, hooks *OnFlightHooks) (int, error) { | ||
| return http.StatusNotImplemented, fmt.Errorf("EPD is not natively implemented for SGLangConnector yet") | ||
| } |
| // IsEmpty returns true if both decode and prefill pod sets are empty | ||
| func (p *PDGroupPods) IsEmpty() bool { | ||
| p.mutex.RLock() | ||
| defer p.mutex.RUnlock() | ||
| return p.decodePods.Len() == 0 && p.prefillPods.Len() == 0 | ||
| return p.decodePods.Len() == 0 && p.prefillPods.Len() == 0 && p.encodePods.Len() == 0 | ||
| } |
| modelServer := r.store.GetModelServer(ctx.ModelServerName) | ||
| if modelServer != nil && modelServer.Spec.PipelineMode != nil && | ||
| (*modelServer.Spec.PipelineMode == v1alpha1.PipelineModeEPD || *modelServer.Spec.PipelineMode == v1alpha1.PipelineModeSGLangEPD) { | ||
| // EPD disaggregated mode | ||
| return r.proxyToEPDDisaggregated(c, req, ctx, kvConnector, modelRequest, port) |
YaoZengzeng
left a comment
YaoZengzeng
left a comment
There was a problem hiding this comment.
Thanks for the work on EPD orchestration. I went through the full diff. The CRD/datastore/scheduler plumbing is generally clean and consistent with the existing PD design, but there are a few blocking issues before this can go in.
Blocking
1. The feature is wired into the request path but is non-functional.
ProxyEPD is stubbed to return http.StatusNotImplemented for every connector (http.go, nixl.go, sglang.go). Since proxyModelEndpoint now routes vllm-epd/sglang-epd to proxyToEPDDisaggregated, which calls kvConnector.ProxyEPD(...), any user who sets pipelineMode: vllm-epd|sglang-epd will get a 500 on every request. We shouldn't merge a user-selectable mode that always fails. Please either implement at least one real connector (vLLM or SGLang), or gate the mode as experimental/behind validation so it can't be selected yet.
2. Semantic bug in the stub return value.
The ProxyEPD stubs return http.StatusNotImplemented (501) as the first return value, but the KVConnector interface defines that value as "the number of output tokens consumed". Returning an HTTP status code where a token count is expected is incorrect (it happens to be masked today only because err != nil short-circuits). It should return 0, err.
3. Title/scope mismatch.
The PR is titled "Proposal: ..." but contains no design doc under docs/proposal/ — it's an implementation PR with executable code. Please either add the proposal doc (if this is meant to be a design proposal) or drop "Proposal:" from the title and describe the actual runtime behavior + tests.
Should fix
4. EPD proxy has no fallback across candidates.
proxyToEPDDisaggregated only uses index [0] of EncodePods/PrefillPods/DecodePods and aborts with 500 if any single slot is empty/nil. Compare with proxyToPDDisaggregated, which iterates over all scheduled pairs and retries on failure. With topN > 1, one unhealthy pod will fail the whole request here. Consider mirroring the PD loop for resilience.
5. Config inconsistency can silently pass scheduling and only fail at proxy time.
EncodePods is populated only inside the existing ctx.PDGroup != nil block, and only when GetEncodePodsForDecodeGroup returns pods. If pipelineMode is EPD but encodeLabels/encode pods are missing, Schedule succeeds, then proxyToEPDDisaggregated 500s at runtime. Please validate consistency early — e.g. a webhook rule "EPD mode requires pdGroup.encodeLabels", or fail fast in the scheduler.
6. RunPostHooks(ctx, 0) hardcodes index 0.
The PD path passes the actual selected index i. Hardcoding 0 will record the wrong pod in the prefix cache once more than one pair is supported. Use the selected index.
Nits
7. ctx.EncodePods lazy allocation (if len(ctx.EncodePods) != len(topNDecodePods)) is fragile. Allocate it once up front like prefillPods for clarity and to avoid index mismatches.
8. Encode in-flight tracking is done ad-hoc with manual IncrPodOnFlightRequests/Decr instead of going through OnFlightHooks like prefill/decode. Either add IncrEncode/DecrEncode hooks or note why it's handled separately. The inline comments ("let's add ProxyEPD to KVConnector and then implement it", "You can implement ... if needed") read like dev notes — please clean them up.
9. PipelineMode includes pd-disaggregated, which is functionally identical to leaving the field unset (it falls through to the PD path). Two ways to express the same behavior is confusing — consider dropping it or documenting the default explicitly.
10. Please confirm the generated artifacts (zz_generated.deepcopy.go, client-go/applyconfiguration/*, CRD yaml, CRD ref md) were produced via make generate/codegen rather than hand-edited. The testdata revision-hash churn is expected given the spec change.
11. No tests cover the new EPD paths (only the mock store was extended). Please add unit tests for EncodePods population in the scheduler and for the success/error branches of the EPD proxy.
Happy to re-review once the connector implementation (or feature gating) and the token-count return are addressed.
YaoZengzeng
left a comment
YaoZengzeng
left a comment
There was a problem hiding this comment.
Posting per-line comments inline (replaces my earlier summary). High-level: the CRD/datastore/scheduler plumbing is clean and consistent with the existing PD design, but the EPD request path is currently non-functional (all connectors stub ProxyEPD to 501) and there are a few correctness issues to address before merge. Also: the title says "Proposal:" but there's no design doc under docs/proposal/ and no tests for the new EPD paths — please reconcile the title/scope and add unit tests for EncodePods population and the EPD proxy success/error branches.
| if modelServer != nil && modelServer.Spec.PipelineMode != nil && | ||
| (*modelServer.Spec.PipelineMode == v1alpha1.PipelineModeEPD || *modelServer.Spec.PipelineMode == v1alpha1.PipelineModeSGLangEPD) { | ||
| // EPD disaggregated mode | ||
| return r.proxyToEPDDisaggregated(c, req, ctx, kvConnector, modelRequest, port) |
There was a problem hiding this comment.
Blocking: this routes vllm-epd/sglang-epd to proxyToEPDDisaggregated, which calls kvConnector.ProxyEPD(...). But every connector's ProxyEPD is currently a stub returning 501 NotImplemented (see http.go/nixl.go/sglang.go). So any user who sets pipelineMode: vllm-epd|sglang-epd gets a 500 on every request. Please either implement at least one real connector (vLLM or SGLang), or gate these modes behind validation so they can't be selected yet.
| // ProxyEPD executes the complete encode-prefill-decode flow natively for vLLM/SGLang. | ||
| ProxyEPD(c *gin.Context, reqBody map[string]interface{}, encodeAddr, prefillAddr, decodeAddr string, hooks *OnFlightHooks) (int, error) |
There was a problem hiding this comment.
The new ProxyEPD shares Proxy's contract (first return = output tokens consumed). Please document that explicitly here, since all three implementations currently violate it by returning an HTTP status code.
| // PipelineModeSGLangEPD indicates an Encode-Prefill-Decode execution mode for SGLang | ||
| PipelineModeSGLangEPD PipelineMode = "sglang-epd" | ||
| // PipelineModePDDisaggregated indicates standard Prefill-Decode disaggregation | ||
| PipelineModePDDisaggregated PipelineMode = "pd-disaggregated" |
There was a problem hiding this comment.
pd-disaggregated is functionally identical to leaving pipelineMode unset (both fall through to proxyToPDDisaggregated). Two ways to express the same behavior is confusing — consider dropping this enum value, or document that it is the explicit form of the default.
shivansh-gohem
force-pushed
the
feat/epd-orchestration-clean
branch
from
62d17f3 to
425f27b
Compare
|
Hi @YaoZengzeng, thanks for the review , please have a look
|
shivansh-gohem
force-pushed
the
feat/epd-orchestration-clean
branch
from
425f27b to
a82a0ab
Compare
shivansh-gohem
force-pushed
the
feat/epd-orchestration-clean
branch
from
a82a0ab to
26835b9
Compare
shivansh-gohem
force-pushed
the
feat/epd-orchestration-clean
branch
from
df8c189 to
aab3d4e
Compare
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
shivansh-gohem
force-pushed
the
feat/epd-orchestration-clean
branch
from
aab3d4e to
6227850
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 22 out of 25 changed files in this pull request and generated 5 comments.
Files not reviewed (3)
- client-go/applyconfiguration/networking/v1alpha1/modelserverspec.go: Generated file
- client-go/applyconfiguration/networking/v1alpha1/pdgroup.go: Generated file
- pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go: Generated file
| // Get encode pods for the same PD group as the decode pod (O(1) lookup) | ||
| // Only for EPD mode. If encode pods are not found or none available, we return an error. | ||
| if isEPDMode { | ||
| selectedEncodePods, err := s.store.GetEncodePodsForDecodeGroup(ctx.ModelServerName, decodePodName) | ||
| if err != nil || len(selectedEncodePods) == 0 { |
| // 1. Setup backend mock server | ||
| encodeReqs := 0 | ||
| prefillReqs := 0 | ||
| decodeReqs := 0 | ||
| backendHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | ||
| body, _ := io.ReadAll(r.Body) | ||
| var reqBody ModelRequest | ||
| json.Unmarshal(body, &reqBody) | ||
|
|
||
| if r.URL.Port() == "8081" { | ||
| encodeReqs++ | ||
| } else if r.URL.Port() == "8082" { | ||
| prefillReqs++ | ||
| } else { | ||
| decodeReqs++ | ||
| } | ||
|
|
||
| w.WriteHeader(http.StatusOK) | ||
| fmt.Fprint(w, `{"id":"epd-resp"}`) | ||
| }) |
| reqBody := `{"model": "test-model-epd", "prompt": "hello epd"}` | ||
| c.Request, _ = http.NewRequest("POST", "/v1/chat/completions", bytes.NewBufferString(reqBody)) | ||
| c.Request.Header.Set("Content-Type", "application/json") | ||
|
|
||
| // 4. Execute handler | ||
| router.HandlerFunc()(c) | ||
|
|
||
| // 5. Assertions | ||
| // Note: HTTPConnector natively implements ProxyEPD, so the proxy loop | ||
| // successfully sends the requests sequentially and returns the final decode response. | ||
| assert.Equal(t, http.StatusOK, w.Code) | ||
| assert.Contains(t, w.Body.String(), "epd-resp") | ||
| } |
| p.encodePods.Delete(podName) | ||
| } | ||
|
|
||
| // RemovePod removes a pod from both decode and prefill sets |
| return p.encodePods.UnsortedList() | ||
| } | ||
|
|
||
| // IsEmpty returns true if both decode and prefill pod sets are empty |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 22 out of 25 changed files in this pull request and generated 3 comments.
Files not reviewed (3)
- client-go/applyconfiguration/networking/v1alpha1/modelserverspec.go: Generated file
- client-go/applyconfiguration/networking/v1alpha1/pdgroup.go: Generated file
- pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go: Generated file
| // Get encode pods for the same PD group as the decode pod (O(1) lookup) | ||
| // Only for EPD mode. If encode pods are not found or none available, we return an error. | ||
| if isEPDMode { |
| decodeBody := cloneReqBody(reqBody) | ||
| h.decodeRequest = BuildDecodeRequest(c, c.Request, decodeBody) | ||
|
|
||
| prefillBody := cloneReqBody(reqBody) | ||
| h.prefillRequest = buildPrefillRequest(c.Request, prefillBody) | ||
|
|
||
| encodeBody := cloneReqBody(reqBody) | ||
| h.encodeRequest = buildEncodeRequest(c.Request, encodeBody) | ||
|
|
| // ProxyEPD executes the complete encode-prefill-decode flow natively for vLLM/SGLang. | ||
| // Returns the number of output tokens consumed, or error if the operation fails. | ||
| ProxyEPD(c *gin.Context, reqBody map[string]interface{}, encodeAddr, prefillAddr, decodeAddr string, hooks *OnFlightHooks) (int, error) |
shivansh-gohem
force-pushed
the
feat/epd-orchestration-clean
branch
from
6227850 to
1dce2d6
Compare
|
@kube-gopher @YaoZengzeng please have a look at this and presnt your views |
|
@kube-gopher i have took at every minor correction you told !! the pr is ready for the review |
|
@shivansh-gohem Sorry, I've been too busy lately and haven't had a chance to look at it yet. It might take me a while, but perhaps another reviewer can help you |
|
@JesseStutler @hzxuzhonghu @YaoZengzeng could you please help by reviewing this as kube is unavailable ! |
|
@shivansh-gohem Sorry, I'll need to look into this a bit more, because I've never encountered a pipeline like that before |
| hooks.IncrEncode() | ||
| } | ||
|
|
||
| err := h.encode(h.encodeRequest, encodeAddr) |
There was a problem hiding this comment.
This always sends the request through the encoder, including text-only requests. The proposal requires skipping E when no images are present, and forwarding the complete chat body also does not create an encoder request per multimodal item. Please extract the multimodal inputs and bypass the encode phase when there are none.
| return | ||
| } | ||
|
|
||
| // Check if pod matches encode labels (for EPD) |
There was a problem hiding this comment.
Changing encodeLabels on an existing ModelServer leaves already-bound pods in the old role indexes. AddOrUpdateModelServer swaps the spec but preserves pdGroups, and the controller then skips those existing bindings, so recategorization only happens after another Pod event or restart. Please rebuild the role indexes when these selectors change.
| klog.Errorf("proxy failed for encode %s, prefill %s, decode %s: %v", | ||
| encodePod.Name, prefillPod.Name, decodePod.Name, err) | ||
| lastErr = err | ||
| continue |
There was a problem hiding this comment.
If the decode stream writes data and then returns a read error, this continues to the next tuple and appends a second response to an already-started 200 stream. The aggregate proxy path stops retrying when c.Writer.Written() is true; the EPD path needs the same guard.
Signed-off-by: Shivansh Sahu <sahushivansh142@gmail.com>
shivansh-gohem
force-pushed
the
feat/epd-orchestration-clean
branch
from
1dce2d6 to
f3a0c21
Compare
|
@hzxuzhonghu @YaoZengzeng I have pushed , resolves all test failures, and addresses the recent Copilot AI review comments (specifically fixing the thread-safety and concurrency state bugs in |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 23 out of 26 changed files in this pull request and generated 3 comments.
Files not reviewed (3)
- client-go/applyconfiguration/networking/v1alpha1/modelserverspec.go: Generated file
- client-go/applyconfiguration/networking/v1alpha1/pdgroup.go: Generated file
- pkg/apis/networking/v1alpha1/zz_generated.deepcopy.go: Generated file
| func buildEncodeRequest(req *http.Request, modelRequest map[string]interface{}) *http.Request { | ||
| // For EPD disaggregated mode, we need to send an encode request to the encode pod with non stream mode. | ||
| preparePrefillBody(modelRequest) | ||
|
|
||
| body, err := json.Marshal(modelRequest) | ||
| if err != nil { | ||
| return nil | ||
| } | ||
|
|
||
| // build request | ||
| reqCopy := req.Clone(req.Context()) | ||
| reqCopy.URL.Scheme = "http" | ||
| reqCopy.Body = io.NopCloser(bytes.NewBuffer(body)) | ||
| reqCopy.ContentLength = int64(len(body)) | ||
|
|
||
| return reqCopy | ||
| } |
| modelServer := r.store.GetModelServer(ctx.ModelServerName) | ||
| if modelServer != nil && modelServer.Spec.PipelineMode != nil && | ||
| (*modelServer.Spec.PipelineMode == v1alpha1.PipelineModeEPD || *modelServer.Spec.PipelineMode == v1alpha1.PipelineModeSGLangEPD) { | ||
| // EPD disaggregated mode | ||
| return r.proxyToEPDDisaggregated(c, req, ctx, kvConnector, modelRequest, port) | ||
| } | ||
|
|
||
| // PD disaggregated mode - use KV connector | ||
| return r.proxyToPDDisaggregated(c, req, ctx, kvConnector, modelRequest, port) | ||
| } |
| // ProxyEPD executes the complete encode-prefill-decode flow natively for SGLang. | ||
| func (s *SGLangConnector) ProxyEPD(c *gin.Context, reqBody map[string]interface{}, encodeAddr, prefillAddr, decodeAddr string, hooks *OnFlightHooks) (int, error) { | ||
| return 0, fmt.Errorf("EPD is not natively implemented for SGLangConnector yet") | ||
| } |
What type of PR is this?
Feature
What this PR does / why we need it:
This PR introduces a configurable and orchestratable inference pipeline framework for Kthena Router that transforms the current hardcoded inference flow processing into a flexible model. By introducing
PipelineModetoModelServerSpec, users can configurevllm-epd,sglang-epd, or standardpd-disaggregatedmodes. The PR explicitly leverages the native EPD logic for inference endpoints, segregating encoding logic cleanly insideKVConnectornatively. It implements theProxyEPDmechanism and hooks it deeply into the proxying and scheduling algorithms, isolating roles likeencodePods,prefillPods, anddecodePodsinto separate logical entities.Which issue(s) this PR fixes:
Fixes #827
Special notes for reviewer:
This PR branch isolates the native EPD logic into a single atomic commit with 17 modified files.