feat: edge runtime model queue and status dispatch updates
- Update control-plane edge wire and edge-node runtime wire contracts - Refactor model queue service with admission control - Update chat handler and responses handler for edge - Modify run dispatch and status provider logic - Add/modify runtime proto definitions - Move G07 status logs to archive
This commit is contained in:
parent
35eb21959f
commit
5b95208f4c
15 changed files with 440 additions and 70 deletions
|
|
@ -36,7 +36,7 @@ Control Plane은 Edge 연결 view와 제어 요청/결과를 관리하고, Edge
|
|||
- `EdgeHelloRequest.edge_id`: Control Plane connection registry의 Edge identity다. 비어 있으면 hello는 거부된다.
|
||||
- `EdgeStatusResponse.nodes`: Edge가 소유한 node snapshot view다.
|
||||
- `EdgeNodeSnapshot.config`: Node에 내려간 config payload의 관찰용 요약이다.
|
||||
- `EdgeNodeSnapshot.provider_snapshots`: runtime `ProviderSnapshot` wire name을 재사용한 Node resource/provider snapshot이다. `category`가 CLI/API/local inference resource kind를 구분하며, Node address, token, transport internals는 싣지 않는다.
|
||||
- `EdgeNodeSnapshot.provider_snapshots`: runtime `ProviderSnapshot` wire name을 재사용한 Node resource/provider snapshot이다. `category`가 CLI/API/local inference resource kind를 구분하며, Node address, token, transport internals는 싣지 않는다. `long_context_capacity`, `long_in_flight`, `long_queued` 필드를 포함해 long-context admission 상태를 함께 전달한다.
|
||||
- `EdgeCommandRequest.operation`: Edge-owned operation 이름이다. Node 직접 scheduling 명령으로 사용하지 않는다.
|
||||
- `EdgeCommandRequest.target_selector`: Edge 내부 operation이 해석할 대상 selector다. Node address나 token을 외부화하지 않는다.
|
||||
- `metadata`: 필요한 Edge identity 또는 운영 보조 정보만 담는다. secret과 private endpoint 원문은 tracked 계약에 쓰지 않는다.
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ Edge는 Node 연결을 수락하고, Node는 연결 직후 등록 요청을 보
|
|||
- `NodeConfigPayload.adapters`: Edge가 Node에 내려주는 adapter instance 설정이다.
|
||||
- `AdapterConfig.name`: node 내부 stable adapter instance identity다. 비어 있으면 legacy single-instance type 이름과 동등하다.
|
||||
- `NodeRuntimeConfig.concurrency`: legacy compatibility runtime metadata다. 실행 admission은 이 값을 node-wide global gate로 사용하지 않고 provider/resource capacity를 기준으로 한다. Node store 위치나 CLI 실행 작업 디렉터리는 이 runtime payload에 싣지 않는다.
|
||||
- `ProviderSnapshot`: legacy wire name을 유지하지만 Node 아래 resource/provider 상태 snapshot으로 해석한다. `category`가 `api`, `cli`, `local_inference` resource kind를 나타내며, provider-pool dispatch 대상은 Edge config `models[].providers`가 참조한 resource뿐이다.
|
||||
- `ProviderSnapshot`: legacy wire name을 유지하지만 Node 아래 resource/provider 상태 snapshot으로 해석한다. `category`가 `api`, `cli`, `local_inference` resource kind를 나타내며, provider-pool dispatch 대상은 Edge config `models[].providers`가 참조한 resource뿐이다. 또한 `long_context_capacity`, `long_in_flight`, `long_queued` 필드는 long-context admission 제어를 위해 사용되며, provider가 확보한 long slot 용량과 현재 점유 중인 long slot 수, 대기 중인 long 요청 수를 나타낸다.
|
||||
|
||||
## 금지 사항
|
||||
|
||||
|
|
|
|||
|
|
@ -24,32 +24,36 @@ task=m-model-group-long-context-admission/04+01,02_status_logs, plan=0, tag=STAT
|
|||
|
||||
| 항목 | 완료 여부 |
|
||||
|------|---------|
|
||||
| [STATUS_LOGS-1] ProviderSnapshot long fields | [ ] |
|
||||
| [STATUS_LOGS-2] Admission log and queue reason evidence | [ ] |
|
||||
| [STATUS_LOGS-1] ProviderSnapshot long fields | [x] |
|
||||
| [STATUS_LOGS-2] Admission log and queue reason evidence | [x] |
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [ ] `ProviderSnapshot`와 관련 inner contract에 long capacity/in-flight 노출 필드를 추가하고 generated Go proto를 `make proto`로 갱신한다.
|
||||
- [ ] service status snapshot이 `long_context_capacity`, `long_in_flight`, 필요한 queued/reason 정보를 채우게 한다.
|
||||
- [ ] dispatch/queue log 또는 event metadata에 `estimated_input_tokens`, `context_class`, queue reason을 남긴다. 검증: status 또는 log sample로 admission 판단을 재구성할 수 있다.
|
||||
- [ ] control-plane relay tests가 새 provider snapshot fields를 보존하는지 검증한다.
|
||||
- [ ] proto preflight, `make proto`, 관련 Go tests를 실행한다.
|
||||
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
- [x] `ProviderSnapshot`와 관련 inner contract에 long capacity/in-flight 노출 필드를 추가하고 generated Go proto를 `make proto`로 갱신한다.
|
||||
- [x] service status snapshot이 `long_context_capacity`, `long_in_flight`, 필요한 queued/reason 정보를 채우게 한다.
|
||||
- [x] dispatch/queue log 또는 event metadata에 `estimated_input_tokens`, `context_class`, queue reason을 남긴다. 검증: status 또는 log sample로 admission 판단을 재구성할 수 있다.
|
||||
- [x] control-plane relay tests가 새 provider snapshot fields를 보존하는지 검증한다.
|
||||
- [x] proto preflight, `make proto`, 관련 Go tests를 실행한다.
|
||||
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
## 코드리뷰 전용 체크리스트
|
||||
|
||||
- [ ] 판정을 append한다.
|
||||
- [ ] active review/plan을 `code_review_local_G07_0.log`, `plan_local_G07_0.log`로 아카이브한다.
|
||||
- [ ] PASS이면 `complete.log` 작성 후 archive로 이동한다.
|
||||
- [x] 판정을 append한다.
|
||||
- [x] active review/plan을 `code_review_local_G07_0.log`, `plan_local_G07_0.log`로 아카이브한다.
|
||||
- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다.
|
||||
- [x] PASS이면 `complete.log` 작성 후 archive로 이동한다.
|
||||
- [x] PASS이고 task group이 `m-model-group-long-context-admission`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
|
||||
- [x] PASS split 작업이며 active parent `agent-task/m-model-group-long-context-admission/`는 sibling 작업이 남아 유지한다.
|
||||
- [ ] WARN/FAIL이면 다음 active plan/review 또는 `USER_REVIEW.md`를 작성한다.
|
||||
|
||||
## 계획 대비 변경 사항
|
||||
|
||||
_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._
|
||||
없음. 계획된 proto/status/log/test 범위 안에서 구현했다.
|
||||
|
||||
## 주요 설계 결정
|
||||
|
||||
_구현 에이전트가 주요 설계 결정 사항을 기록한다._
|
||||
- `ProviderSnapshot`에는 `long_context_capacity`, `long_in_flight`, `long_queued`를 추가하고, live queue snapshot에서 provider별 long slot 상태를 계산한다.
|
||||
- OpenAI-compatible chat/responses dispatch 로그에는 `estimated_input_tokens`, `context_class`, `queue_reason`을 함께 남겨 status/log evidence로 admission 판단을 재구성할 수 있게 했다.
|
||||
|
||||
## 사용자 리뷰 요청
|
||||
|
||||
|
|
@ -76,35 +80,76 @@ _기본값은 `없음`이다. 구현 중 새 결정이 필요해 보여도 직
|
|||
|
||||
```bash
|
||||
$ command -v protoc
|
||||
(output)
|
||||
/config/.local/bin/protoc
|
||||
$ command -v protoc-gen-go
|
||||
(output)
|
||||
/config/.local/bin/protoc-gen-go
|
||||
$ make proto
|
||||
(output)
|
||||
protoc \
|
||||
--go_out=. \
|
||||
--go_opt=module=iop \
|
||||
--proto_path=. \
|
||||
proto/iop/runtime.proto \
|
||||
proto/iop/node.proto \
|
||||
proto/iop/control.proto \
|
||||
proto/iop/job.proto
|
||||
$ go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/controlplane
|
||||
(output)
|
||||
ok iop/apps/edge/internal/service 0.573s
|
||||
ok iop/apps/edge/internal/controlplane 4.448s
|
||||
```
|
||||
|
||||
### STATUS_LOGS-2 중간 검증
|
||||
|
||||
```bash
|
||||
$ go test -count=1 ./apps/edge/internal/service
|
||||
(output)
|
||||
ok iop/apps/edge/internal/service 0.573s
|
||||
```
|
||||
|
||||
### 최종 검증
|
||||
|
||||
```bash
|
||||
$ command -v protoc
|
||||
(output)
|
||||
/config/.local/bin/protoc
|
||||
$ command -v protoc-gen-go
|
||||
(output)
|
||||
/config/.local/bin/protoc-gen-go
|
||||
$ make proto
|
||||
(output)
|
||||
protoc \
|
||||
--go_out=. \
|
||||
--go_opt=module=iop \
|
||||
--proto_path=. \
|
||||
proto/iop/runtime.proto \
|
||||
proto/iop/node.proto \
|
||||
proto/iop/control.proto \
|
||||
proto/iop/job.proto
|
||||
$ go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/controlplane ./apps/edge/internal/transport ./apps/node/internal/transport ./apps/control-plane/internal/wire
|
||||
(output)
|
||||
ok iop/apps/edge/internal/service 0.566s
|
||||
ok iop/apps/edge/internal/controlplane 4.446s
|
||||
ok iop/apps/edge/internal/transport 2.038s
|
||||
ok iop/apps/node/internal/transport 5.342s
|
||||
ok iop/apps/control-plane/internal/wire 1.472s
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
> **[IMPLEMENTING AGENT — BEFORE SAVING] Fill every implementation-owned section before saving.**
|
||||
|
||||
## 코드리뷰 결과
|
||||
|
||||
- 종합 판정: PASS
|
||||
- 차원별 평가:
|
||||
- correctness: Pass
|
||||
- completeness: Pass
|
||||
- test coverage: Pass
|
||||
- API contract: Pass
|
||||
- code quality: Pass
|
||||
- implementation deviation: Pass
|
||||
- verification trust: Pass
|
||||
- spec conformance: Pass
|
||||
- 발견된 문제: 없음
|
||||
- 검증:
|
||||
- `command -v protoc` - PASS (`/config/.local/bin/protoc`)
|
||||
- `command -v protoc-gen-go` - PASS (`/config/.local/bin/protoc-gen-go`)
|
||||
- `make proto` - PASS
|
||||
- `go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/controlplane ./apps/edge/internal/transport ./apps/node/internal/transport ./apps/control-plane/internal/wire ./apps/edge/internal/openai` - PASS
|
||||
- `go test -count=1 ./apps/edge/... ./apps/node/internal/transport ./apps/control-plane/internal/wire ./proto/gen/...` - PASS
|
||||
- `git diff --check` - PASS
|
||||
- 다음 단계: PASS이므로 active review/plan을 로그로 아카이브하고 `complete.log` 작성 후 task directory를 archive로 이동한다.
|
||||
|
|
@ -0,0 +1,48 @@
|
|||
# Complete - m-model-group-long-context-admission/04+01,02_status_logs
|
||||
|
||||
## 완료 일시
|
||||
|
||||
2026-07-05
|
||||
|
||||
## 요약
|
||||
|
||||
Model Group Long-Context Admission의 status/log observability 작업을 1회 루프로 완료했다. 최종 판정: PASS.
|
||||
|
||||
## 루프 이력
|
||||
|
||||
| Plan | Review | Verdict | 메모 |
|
||||
|------|--------|---------|------|
|
||||
| `plan_local_G07_0.log` | `code_review_local_G07_0.log` | PASS | ProviderSnapshot long fields, Control Plane relay preservation, dispatch log metadata, queue reason evidence 검증 완료 |
|
||||
|
||||
## 구현/정리 내용
|
||||
|
||||
- `ProviderSnapshot`에 `long_context_capacity`, `long_in_flight`, `long_queued` 필드를 추가하고 `make proto`로 Go 생성물을 갱신했다.
|
||||
- Edge service status snapshot이 provider별 long capacity, long in-flight, long queued 상태를 채우도록 했다.
|
||||
- OpenAI-compatible chat/responses dispatch 로그에 `estimated_input_tokens`, `context_class`, `queue_reason`을 남기도록 했다.
|
||||
- queue reason 계산과 Control Plane provider snapshot relay 보존 테스트를 추가했다.
|
||||
- 리뷰 중 비어 있던 review artifact의 구현 소유 요약 섹션을 실제 구현 내용 기준으로 정리했다.
|
||||
|
||||
## 최종 검증
|
||||
|
||||
- `command -v protoc` - PASS; `/config/.local/bin/protoc`
|
||||
- `command -v protoc-gen-go` - PASS; `/config/.local/bin/protoc-gen-go`
|
||||
- `make proto` - PASS; `runtime.proto`, `node.proto`, `control.proto`, `job.proto` Go 생성 완료
|
||||
- `go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/controlplane ./apps/edge/internal/transport ./apps/node/internal/transport ./apps/control-plane/internal/wire ./apps/edge/internal/openai` - PASS; 모든 대상 패키지 `ok`
|
||||
- `go test -count=1 ./apps/edge/... ./apps/node/internal/transport ./apps/control-plane/internal/wire ./proto/gen/...` - PASS; 모든 대상 패키지 `ok`, `iop/proto/gen/iop`은 test files 없음
|
||||
- `git diff --check` - PASS; whitespace error 없음
|
||||
- repo 내부 edge-node full-cycle smoke - 미수행; 이번 split task의 dev smoke evidence는 후속 `06+01,02,03,04,05_capacity_smoke`에서 수집하도록 계획되어 있다.
|
||||
|
||||
## Roadmap Completion
|
||||
|
||||
- Milestone: `agent-roadmap/phase/operational-observability-provider-management/milestones/model-group-long-context-admission.md`
|
||||
- Completed task ids:
|
||||
- `status-logs`: PASS; evidence=`agent-task/archive/2026/07/m-model-group-long-context-admission/04+01,02_status_logs/plan_local_G07_0.log`, `agent-task/archive/2026/07/m-model-group-long-context-admission/04+01,02_status_logs/code_review_local_G07_0.log`; verification=`make proto`, `go test -count=1 ./apps/edge/... ./apps/node/internal/transport ./apps/control-plane/internal/wire ./proto/gen/...`
|
||||
- Not completed task ids: 없음
|
||||
|
||||
## 잔여 Nit
|
||||
|
||||
- 없음
|
||||
|
||||
## 후속 작업
|
||||
|
||||
- 없음
|
||||
|
|
@ -226,11 +226,14 @@ func TestConnectorRespondsToStatusRequestFromProvider(t *testing.T) {
|
|||
Label: "node0",
|
||||
ProviderSnapshots: []*iop.ProviderSnapshot{
|
||||
{
|
||||
Adapter: "cli",
|
||||
Status: "available",
|
||||
Capacity: 4,
|
||||
InFlight: 2,
|
||||
Queued: 3,
|
||||
Adapter: "cli",
|
||||
Status: "available",
|
||||
Capacity: 4,
|
||||
InFlight: 2,
|
||||
Queued: 3,
|
||||
LongContextCapacity: 2,
|
||||
LongInFlight: 1,
|
||||
LongQueued: 1,
|
||||
},
|
||||
},
|
||||
},
|
||||
|
|
@ -297,7 +300,8 @@ func TestConnectorRespondsToStatusRequestFromProvider(t *testing.T) {
|
|||
t.Fatalf("expected 1 provider snapshot, got %d", len(n.GetProviderSnapshots()))
|
||||
}
|
||||
snap := n.GetProviderSnapshots()[0]
|
||||
if snap.GetAdapter() != "cli" || snap.GetStatus() != "available" || snap.GetCapacity() != 4 || snap.GetInFlight() != 2 || snap.GetQueued() != 3 {
|
||||
if snap.GetAdapter() != "cli" || snap.GetStatus() != "available" || snap.GetCapacity() != 4 || snap.GetInFlight() != 2 || snap.GetQueued() != 3 ||
|
||||
snap.GetLongContextCapacity() != 2 || snap.GetLongInFlight() != 1 || snap.GetLongQueued() != 1 {
|
||||
t.Errorf("unexpected snapshot contents: %+v", snap)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -92,6 +92,16 @@ func (s *Server) handleChatCompletions(w http.ResponseWriter, r *http.Request) {
|
|||
writeError(w, http.StatusBadGateway, "node_dispatch_error", err.Error())
|
||||
return
|
||||
}
|
||||
s.logger.Info("openai chat completion dispatch",
|
||||
zap.String("run_id", handle.Dispatch().RunID),
|
||||
zap.String("node_id", handle.Dispatch().NodeID),
|
||||
zap.String("model_group", handle.Dispatch().ModelGroupKey),
|
||||
zap.String("adapter", handle.Dispatch().Adapter),
|
||||
zap.String("target", handle.Dispatch().Target),
|
||||
zap.Int("estimated_input_tokens", handle.Dispatch().EstimatedInputTokens),
|
||||
zap.String("context_class", handle.Dispatch().ContextClass),
|
||||
zap.String("queue_reason", handle.Dispatch().QueueReason),
|
||||
)
|
||||
|
||||
if req.Stream {
|
||||
s.streamChatCompletion(w, r, req, submitReq, handle, outputPolicy, validation)
|
||||
|
|
|
|||
|
|
@ -122,6 +122,17 @@ func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) {
|
|||
}
|
||||
defer handle.Close()
|
||||
|
||||
s.logger.Info("openai responses dispatch",
|
||||
zap.String("run_id", handle.Dispatch().RunID),
|
||||
zap.String("node_id", handle.Dispatch().NodeID),
|
||||
zap.String("model_group", handle.Dispatch().ModelGroupKey),
|
||||
zap.String("adapter", handle.Dispatch().Adapter),
|
||||
zap.String("target", handle.Dispatch().Target),
|
||||
zap.Int("estimated_input_tokens", handle.Dispatch().EstimatedInputTokens),
|
||||
zap.String("context_class", handle.Dispatch().ContextClass),
|
||||
zap.String("queue_reason", handle.Dispatch().QueueReason),
|
||||
)
|
||||
|
||||
s.completeResponse(w, r, req, handle, outputPolicy)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -77,6 +77,7 @@ type queueItem struct {
|
|||
waitCh chan admitResult
|
||||
deadline time.Time
|
||||
long bool // true when the queued request is a long-context request
|
||||
reason string
|
||||
}
|
||||
|
||||
type modelQueueGroup struct {
|
||||
|
|
@ -268,6 +269,11 @@ func candidateLess(a, b *candidateNode) bool {
|
|||
// The optional long flag marks the request as long-context so it is gated on
|
||||
// provider long slots and reserves a long slot on dispatch.
|
||||
func (m *modelQueueManager) admit(ctx context.Context, groupKey, adapter, target string, candidates []candidateNode, policy groupPolicy, longFlag ...bool) (*candidateNode, error) {
|
||||
candidate, _, err := m.admitWithReason(ctx, groupKey, adapter, target, candidates, policy, longFlag...)
|
||||
return candidate, err
|
||||
}
|
||||
|
||||
func (m *modelQueueManager) admitWithReason(ctx context.Context, groupKey, adapter, target string, candidates []candidateNode, policy groupPolicy, longFlag ...bool) (*candidateNode, string, error) {
|
||||
long := len(longFlag) > 0 && longFlag[0]
|
||||
m.mu.Lock()
|
||||
|
||||
|
|
@ -292,12 +298,34 @@ func (m *modelQueueManager) admit(ctx context.Context, groupKey, adapter, target
|
|||
}
|
||||
group.lastSelectedSlot = slot
|
||||
m.mu.Unlock()
|
||||
return candidate, nil
|
||||
return candidate, "dispatched", nil
|
||||
}
|
||||
|
||||
if len(group.queue) >= group.policy.maxQueue {
|
||||
m.mu.Unlock()
|
||||
return nil, fmt.Errorf("model group %q: %w", groupKey, errQueueFull)
|
||||
return nil, "", fmt.Errorf("model group %q: %w", groupKey, errQueueFull)
|
||||
}
|
||||
|
||||
// Determine queue reason.
|
||||
reason := "capacity_full"
|
||||
if long {
|
||||
longBlocked := false
|
||||
for _, c := range candidates {
|
||||
if c.capacity <= 0 {
|
||||
continue
|
||||
}
|
||||
slot := c.slotKey()
|
||||
inflight := group.inflight[slot]
|
||||
if inflight < c.capacity {
|
||||
if c.longContextCapacity > 0 && group.longInflight[slot] >= c.longContextCapacity {
|
||||
longBlocked = true
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
if longBlocked {
|
||||
reason = "long_context_capacity_full"
|
||||
}
|
||||
}
|
||||
|
||||
item := &queueItem{
|
||||
|
|
@ -305,6 +333,7 @@ func (m *modelQueueManager) admit(ctx context.Context, groupKey, adapter, target
|
|||
waitCh: make(chan admitResult, 1),
|
||||
deadline: time.Now().Add(group.policy.queueTimeout),
|
||||
long: long,
|
||||
reason: reason,
|
||||
}
|
||||
group.queue = append(group.queue, item)
|
||||
timeout := group.policy.queueTimeout
|
||||
|
|
@ -315,7 +344,7 @@ func (m *modelQueueManager) admit(ctx context.Context, groupKey, adapter, target
|
|||
|
||||
select {
|
||||
case res := <-item.waitCh:
|
||||
return res.candidate, res.err
|
||||
return res.candidate, item.reason, res.err
|
||||
case <-timer.C:
|
||||
m.mu.Lock()
|
||||
m.removeItemLocked(groupKey, item)
|
||||
|
|
@ -341,7 +370,7 @@ func (m *modelQueueManager) admit(ctx context.Context, groupKey, adapter, target
|
|||
}
|
||||
default:
|
||||
}
|
||||
return nil, fmt.Errorf("model group %q: %w", groupKey, errQueueTimeout)
|
||||
return nil, "", fmt.Errorf("model group %q: %w", groupKey, errQueueTimeout)
|
||||
case <-ctx.Done():
|
||||
m.mu.Lock()
|
||||
m.removeItemLocked(groupKey, item)
|
||||
|
|
@ -365,7 +394,7 @@ func (m *modelQueueManager) admit(ctx context.Context, groupKey, adapter, target
|
|||
}
|
||||
default:
|
||||
}
|
||||
return nil, ctx.Err()
|
||||
return nil, "", ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -611,12 +640,16 @@ func (m *modelQueueManager) getSnapshotForNode(nodeID string, rec *edgenode.Node
|
|||
Health: "disabled",
|
||||
LoadRatio: 0,
|
||||
LifecycleCapabilities: lifecycleCaps,
|
||||
LongContextCapacity: 0,
|
||||
LongInFlight: 0,
|
||||
LongQueued: 0,
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
capVal := prov.Capacity
|
||||
inflight, queued := m.getStatsForProviderLocked(nodeID, prov.ID)
|
||||
longInflight, longQueued := m.getLongStatsForProviderLocked(nodeID, prov.ID)
|
||||
|
||||
var loadRatio float32
|
||||
if capVal > 0 {
|
||||
|
|
@ -636,6 +669,9 @@ func (m *modelQueueManager) getSnapshotForNode(nodeID string, rec *edgenode.Node
|
|||
Health: prov.Health,
|
||||
LoadRatio: loadRatio,
|
||||
LifecycleCapabilities: lifecycleCaps,
|
||||
LongContextCapacity: int32(prov.LongContextCapacity),
|
||||
LongInFlight: int32(longInflight),
|
||||
LongQueued: int32(longQueued),
|
||||
})
|
||||
}
|
||||
return snaps
|
||||
|
|
@ -754,15 +790,27 @@ func (m *modelQueueManager) getStatsForProviderLocked(nodeID, providerID string)
|
|||
return
|
||||
}
|
||||
|
||||
// getLongStatsForProviderLocked returns the long-context in-flight count for a
|
||||
// getLongStatsForProviderLocked returns the long-context in-flight and queued count for a
|
||||
// provider-pool provider identified by (nodeID, providerID), derived from tracked
|
||||
// runs. Must be called with m.mu held.
|
||||
func (m *modelQueueManager) getLongStatsForProviderLocked(nodeID, providerID string) (longInFlight int) {
|
||||
func (m *modelQueueManager) getLongStatsForProviderLocked(nodeID, providerID string) (longInFlight, longQueued int) {
|
||||
for _, rec := range m.inflightByRun {
|
||||
if rec.nodeID == nodeID && rec.providerID == providerID && rec.long {
|
||||
longInFlight++
|
||||
}
|
||||
}
|
||||
for _, group := range m.groups {
|
||||
for _, item := range group.queue {
|
||||
if item.long {
|
||||
for _, c := range item.candidates {
|
||||
if c.entry.NodeID == nodeID && c.providerID == providerID {
|
||||
longQueued++
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -771,7 +819,8 @@ func (m *modelQueueManager) getLongStatsForProviderLocked(nodeID, providerID str
|
|||
func (m *modelQueueManager) longInFlightForProvider(nodeID, providerID string) int {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.getLongStatsForProviderLocked(nodeID, providerID)
|
||||
longInFlight, _ := m.getLongStatsForProviderLocked(nodeID, providerID)
|
||||
return longInFlight
|
||||
}
|
||||
|
||||
func (m *modelQueueManager) getStatsForAdapterLocked(nodeID string, rec *edgenode.NodeRecord, adapterName string) (inFlight, queued int) {
|
||||
|
|
|
|||
|
|
@ -1954,3 +1954,86 @@ func TestModelQueueLongSlotFullExcludesLongKeepsNormal(t *testing.T) {
|
|||
t.Fatalf("long request on long-full provider: expected queue timeout, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestModelQueueReasonLongContextCapacityFull(t *testing.T) {
|
||||
entryA := &edgenode.NodeEntry{NodeID: "node-a"}
|
||||
cands := []candidateNode{
|
||||
{entry: entryA, capacity: 3, priority: 0, providerID: "prov-a", longContextCapacity: 1},
|
||||
}
|
||||
m := newModelQueueManager(nil)
|
||||
|
||||
// Long #1 → prov-a, filling prov-a's single long slot.
|
||||
_, _, err := m.admitWithReason(context.Background(), "g-x", "", "", cands, groupPolicy{}, true)
|
||||
if err != nil {
|
||||
t.Fatalf("first admit: %v", err)
|
||||
}
|
||||
|
||||
// Long #2 → prov-a. It should block because longContextCapacity is full.
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
_, _, err := m.admitWithReason(context.Background(), "g-x", "", "", cands, groupPolicy{maxQueue: 1, queueTimeout: 50 * time.Millisecond}, true)
|
||||
if !errors.Is(err, errQueueTimeout) {
|
||||
t.Errorf("expected timeout, got %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
// Wait a bit for goroutine to queue the item.
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
m.mu.Lock()
|
||||
g, ok := m.groups["g-x"]
|
||||
if !ok || len(g.queue) != 1 {
|
||||
m.mu.Unlock()
|
||||
t.Fatalf("expected 1 queued item, got group ok=%t, len=%d", ok, len(g.queue))
|
||||
}
|
||||
reason := g.queue[0].reason
|
||||
m.mu.Unlock()
|
||||
|
||||
if reason != "long_context_capacity_full" {
|
||||
t.Errorf("expected queue reason 'long_context_capacity_full', got %q", reason)
|
||||
}
|
||||
|
||||
<-done
|
||||
}
|
||||
|
||||
func TestModelQueueReasonCapacityFull(t *testing.T) {
|
||||
entryA := &edgenode.NodeEntry{NodeID: "node-a"}
|
||||
cands := []candidateNode{
|
||||
{entry: entryA, capacity: 1, priority: 0, providerID: "prov-a"},
|
||||
}
|
||||
m := newModelQueueManager(nil)
|
||||
|
||||
// #1 → prov-a, filling capacity.
|
||||
_, _, err := m.admitWithReason(context.Background(), "g-y", "", "", cands, groupPolicy{}, false)
|
||||
if err != nil {
|
||||
t.Fatalf("first admit: %v", err)
|
||||
}
|
||||
|
||||
// #2 → prov-a. It should block because capacity is full.
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
_, _, err := m.admitWithReason(context.Background(), "g-y", "", "", cands, groupPolicy{maxQueue: 1, queueTimeout: 50 * time.Millisecond}, false)
|
||||
if !errors.Is(err, errQueueTimeout) {
|
||||
t.Errorf("expected timeout, got %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
m.mu.Lock()
|
||||
g, ok := m.groups["g-y"]
|
||||
if !ok || len(g.queue) != 1 {
|
||||
m.mu.Unlock()
|
||||
t.Fatalf("expected 1 queued item")
|
||||
}
|
||||
reason := g.queue[0].reason
|
||||
m.mu.Unlock()
|
||||
|
||||
if reason != "capacity_full" {
|
||||
t.Errorf("expected queue reason 'capacity_full', got %q", reason)
|
||||
}
|
||||
|
||||
<-done
|
||||
}
|
||||
|
|
|
|||
|
|
@ -51,15 +51,18 @@ type SubmitRunRequest struct {
|
|||
// RunDispatch describes a dispatched run in surface-neutral terms. It is the
|
||||
// metadata any caller (console, HTTP, future RPC) needs after submission.
|
||||
type RunDispatch struct {
|
||||
RunID string
|
||||
NodeID string
|
||||
NodeLabel string
|
||||
ModelGroupKey string
|
||||
Adapter string
|
||||
Target string
|
||||
SessionID string
|
||||
Background bool
|
||||
TimeoutSec int
|
||||
RunID string
|
||||
NodeID string
|
||||
NodeLabel string
|
||||
ModelGroupKey string
|
||||
Adapter string
|
||||
Target string
|
||||
SessionID string
|
||||
Background bool
|
||||
TimeoutSec int
|
||||
EstimatedInputTokens int
|
||||
ContextClass string
|
||||
QueueReason string
|
||||
}
|
||||
|
||||
// RunStream carries asynchronous events for a dispatched foreground run.
|
||||
|
|
@ -137,7 +140,7 @@ func (s *Service) submitRunQueued(ctx context.Context, req SubmitRunRequest) (Ru
|
|||
}
|
||||
|
||||
long := req.ContextClass == contextClassLong
|
||||
selected, err := s.queue.admit(ctx, req.ModelGroupKey, req.Adapter, req.Target, candidates, policy, long)
|
||||
selected, queueReason, err := s.queue.admitWithReason(ctx, req.ModelGroupKey, req.Adapter, req.Target, candidates, policy, long)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -191,15 +194,18 @@ func (s *Service) submitRunQueued(ctx context.Context, req SubmitRunRequest) (Ru
|
|||
|
||||
return &RunHandle{
|
||||
RunDispatch: RunDispatch{
|
||||
RunID: runID,
|
||||
NodeID: selected.entry.NodeID,
|
||||
NodeLabel: nodeLabel(selected.entry),
|
||||
ModelGroupKey: req.ModelGroupKey,
|
||||
Adapter: runReq.GetAdapter(),
|
||||
Target: runReq.GetTarget(),
|
||||
SessionID: runReq.GetSessionId(),
|
||||
Background: runReq.GetBackground(),
|
||||
TimeoutSec: int(runReq.GetTimeoutSec()),
|
||||
RunID: runID,
|
||||
NodeID: selected.entry.NodeID,
|
||||
NodeLabel: nodeLabel(selected.entry),
|
||||
ModelGroupKey: req.ModelGroupKey,
|
||||
Adapter: runReq.GetAdapter(),
|
||||
Target: runReq.GetTarget(),
|
||||
SessionID: runReq.GetSessionId(),
|
||||
Background: runReq.GetBackground(),
|
||||
TimeoutSec: int(runReq.GetTimeoutSec()),
|
||||
EstimatedInputTokens: req.EstimatedInputTokens,
|
||||
ContextClass: req.ContextClass,
|
||||
QueueReason: queueReason,
|
||||
},
|
||||
RunStream: RunStream{
|
||||
Events: runEvents,
|
||||
|
|
@ -681,15 +687,18 @@ func (s *Service) dispatchToEntry(entry *edgenode.NodeEntry, req SubmitRunReques
|
|||
|
||||
return &RunHandle{
|
||||
RunDispatch: RunDispatch{
|
||||
RunID: runID,
|
||||
NodeID: entry.NodeID,
|
||||
NodeLabel: nodeLabel(entry),
|
||||
ModelGroupKey: req.ModelGroupKey,
|
||||
Adapter: runReq.GetAdapter(),
|
||||
Target: runReq.GetTarget(),
|
||||
SessionID: runReq.GetSessionId(),
|
||||
Background: runReq.GetBackground(),
|
||||
TimeoutSec: int(runReq.GetTimeoutSec()),
|
||||
RunID: runID,
|
||||
NodeID: entry.NodeID,
|
||||
NodeLabel: nodeLabel(entry),
|
||||
ModelGroupKey: req.ModelGroupKey,
|
||||
Adapter: runReq.GetAdapter(),
|
||||
Target: runReq.GetTarget(),
|
||||
SessionID: runReq.GetSessionId(),
|
||||
Background: runReq.GetBackground(),
|
||||
TimeoutSec: int(runReq.GetTimeoutSec()),
|
||||
EstimatedInputTokens: req.EstimatedInputTokens,
|
||||
ContextClass: req.ContextClass,
|
||||
QueueReason: "dispatched",
|
||||
},
|
||||
RunStream: RunStream{
|
||||
Events: runEvents,
|
||||
|
|
|
|||
|
|
@ -94,6 +94,9 @@ func staticProviderCatalogSnapshots(rec *edgenode.NodeRecord) []*iop.ProviderSna
|
|||
ServedModels: servedModels,
|
||||
Health: "disabled",
|
||||
LifecycleCapabilities: lifecycleCaps,
|
||||
LongContextCapacity: 0,
|
||||
LongInFlight: 0,
|
||||
LongQueued: 0,
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
|
@ -108,6 +111,9 @@ func staticProviderCatalogSnapshots(rec *edgenode.NodeRecord) []*iop.ProviderSna
|
|||
ServedModels: servedModels,
|
||||
Health: prov.Health,
|
||||
LifecycleCapabilities: lifecycleCaps,
|
||||
LongContextCapacity: int32(prov.LongContextCapacity),
|
||||
LongInFlight: 0,
|
||||
LongQueued: 0,
|
||||
})
|
||||
}
|
||||
return snaps
|
||||
|
|
|
|||
|
|
@ -475,3 +475,74 @@ func TestListNodeSnapshotsProviderIdDiffersFromAdapterKey(t *testing.T) {
|
|||
t.Fatalf("expected provider snapshot with id=provider-vllm-primary, got %d provider snapshots", len(snap.ProviderSnapshots))
|
||||
}
|
||||
}
|
||||
|
||||
func TestListNodeSnapshotsLongContextFields(t *testing.T) {
|
||||
reg := edgenode.NewRegistry()
|
||||
reg.Register(&edgenode.NodeEntry{NodeID: "node-long-1", Alias: "node-long-1"})
|
||||
|
||||
store := edgenode.NewNodeStore()
|
||||
rec := &edgenode.NodeRecord{
|
||||
ID: "node-long-1",
|
||||
Providers: []config.NodeProviderConf{
|
||||
{
|
||||
ID: "long-provider",
|
||||
Type: "vllm",
|
||||
Category: config.CategoryAPI,
|
||||
Adapter: "vllm-gpu",
|
||||
Models: []string{"long-model"},
|
||||
Capacity: 4,
|
||||
LongContextCapacity: 2,
|
||||
},
|
||||
},
|
||||
Runtime: config.RuntimeConf{Concurrency: 2},
|
||||
}
|
||||
store.Add(rec)
|
||||
|
||||
bus := edgeevents.NewBus()
|
||||
svc := New(reg, bus)
|
||||
svc.SetNodeStore(store)
|
||||
|
||||
svc.queue.mu.Lock()
|
||||
svc.queue.inflightByRun["run-lp-001"] = inflightRec{
|
||||
groupKey: "test-group",
|
||||
nodeID: "node-long-1",
|
||||
providerID: "long-provider",
|
||||
long: true,
|
||||
}
|
||||
svc.queue.groups["test-group"] = &modelQueueGroup{
|
||||
key: "test-group",
|
||||
adapter: "vllm-gpu",
|
||||
inflight: map[string]int{"node-long-1": 1},
|
||||
queue: []*queueItem{
|
||||
{
|
||||
candidates: []candidateNode{
|
||||
{entry: reg.All()[0], capacity: 4, providerID: "long-provider"},
|
||||
},
|
||||
long: true,
|
||||
},
|
||||
},
|
||||
}
|
||||
svc.queue.mu.Unlock()
|
||||
|
||||
snaps := svc.ListNodeSnapshots()
|
||||
if len(snaps) != 1 {
|
||||
t.Fatalf("expected 1 node snapshot, got %d", len(snaps))
|
||||
}
|
||||
snap := snaps[0]
|
||||
if len(snap.ProviderSnapshots) != 1 {
|
||||
t.Fatalf("expected 1 provider snapshot, got %d", len(snap.ProviderSnapshots))
|
||||
}
|
||||
p := snap.ProviderSnapshots[0]
|
||||
if p.Id != "long-provider" {
|
||||
t.Errorf("expected long-provider, got %s", p.Id)
|
||||
}
|
||||
if p.LongContextCapacity != 2 {
|
||||
t.Errorf("expected LongContextCapacity 2, got %d", p.LongContextCapacity)
|
||||
}
|
||||
if p.LongInFlight != 1 {
|
||||
t.Errorf("expected LongInFlight 1, got %d", p.LongInFlight)
|
||||
}
|
||||
if p.LongQueued != 1 {
|
||||
t.Errorf("expected LongQueued 1, got %d", p.LongQueued)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -998,8 +998,12 @@ type ProviderSnapshot struct {
|
|||
// lifecycle_capabilities lists coarse lifecycle capability flags (e.g.
|
||||
// "list_models", "load_model", "unload_model", "pull_model", "delete_model").
|
||||
LifecycleCapabilities []string `protobuf:"bytes,12,rep,name=lifecycle_capabilities,json=lifecycleCapabilities,proto3" json:"lifecycle_capabilities,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
// Long-context fields.
|
||||
LongContextCapacity int32 `protobuf:"varint,13,opt,name=long_context_capacity,json=longContextCapacity,proto3" json:"long_context_capacity,omitempty"`
|
||||
LongInFlight int32 `protobuf:"varint,14,opt,name=long_in_flight,json=longInFlight,proto3" json:"long_in_flight,omitempty"`
|
||||
LongQueued int32 `protobuf:"varint,15,opt,name=long_queued,json=longQueued,proto3" json:"long_queued,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *ProviderSnapshot) Reset() {
|
||||
|
|
@ -1116,6 +1120,27 @@ func (x *ProviderSnapshot) GetLifecycleCapabilities() []string {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (x *ProviderSnapshot) GetLongContextCapacity() int32 {
|
||||
if x != nil {
|
||||
return x.LongContextCapacity
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *ProviderSnapshot) GetLongInFlight() int32 {
|
||||
if x != nil {
|
||||
return x.LongInFlight
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *ProviderSnapshot) GetLongQueued() int32 {
|
||||
if x != nil {
|
||||
return x.LongQueued
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
type AgentUsageStatus struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
RawOutput string `protobuf:"bytes,1,opt,name=raw_output,json=rawOutput,proto3" json:"raw_output,omitempty"`
|
||||
|
|
@ -2387,7 +2412,7 @@ const file_proto_iop_runtime_proto_rawDesc = "" +
|
|||
"\x12provider_snapshots\x18\t \x03(\v2\x15.iop.ProviderSnapshotR\x11providerSnapshots\x1a9\n" +
|
||||
"\vResultEntry\x12\x10\n" +
|
||||
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
|
||||
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xe8\x02\n" +
|
||||
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xe3\x03\n" +
|
||||
"\x10ProviderSnapshot\x12\x18\n" +
|
||||
"\aadapter\x18\x01 \x01(\tR\aadapter\x12\x16\n" +
|
||||
"\x06status\x18\x02 \x01(\tR\x06status\x12\x1a\n" +
|
||||
|
|
@ -2402,7 +2427,11 @@ const file_proto_iop_runtime_proto_rawDesc = "" +
|
|||
" \x01(\tR\x06health\x12\x1d\n" +
|
||||
"\n" +
|
||||
"load_ratio\x18\v \x01(\x02R\tloadRatio\x125\n" +
|
||||
"\x16lifecycle_capabilities\x18\f \x03(\tR\x15lifecycleCapabilities\"\xc9\x02\n" +
|
||||
"\x16lifecycle_capabilities\x18\f \x03(\tR\x15lifecycleCapabilities\x122\n" +
|
||||
"\x15long_context_capacity\x18\r \x01(\x05R\x13longContextCapacity\x12$\n" +
|
||||
"\x0elong_in_flight\x18\x0e \x01(\x05R\flongInFlight\x12\x1f\n" +
|
||||
"\vlong_queued\x18\x0f \x01(\x05R\n" +
|
||||
"longQueued\"\xc9\x02\n" +
|
||||
"\x10AgentUsageStatus\x12\x1d\n" +
|
||||
"\n" +
|
||||
"raw_output\x18\x01 \x01(\tR\trawOutput\x12\x1f\n" +
|
||||
|
|
|
|||
|
|
@ -143,6 +143,11 @@ message ProviderSnapshot {
|
|||
// lifecycle_capabilities lists coarse lifecycle capability flags (e.g.
|
||||
// "list_models", "load_model", "unload_model", "pull_model", "delete_model").
|
||||
repeated string lifecycle_capabilities = 12;
|
||||
|
||||
// Long-context fields.
|
||||
int32 long_context_capacity = 13;
|
||||
int32 long_in_flight = 14;
|
||||
int32 long_queued = 15;
|
||||
}
|
||||
|
||||
message AgentUsageStatus {
|
||||
|
|
|
|||
Loading…
Reference in a new issue