From f9917dd2fddbc0965a5034d50c395b5620c46e1e Mon Sep 17 00:00:00 2001 From: toki Date: Thu, 4 Jun 2026 07:42:57 +0900 Subject: [PATCH] feat(control-plane): edge registry and server refactoring - Refactor edge_registry.go and edge_server.go - Update edge_server_test.go with new test cases - Update main.go and main_test.go for edge integration - Move completed tasks to archive --- .../code_review_cloud_G06_0.log | 61 +++ .../complete.log | 39 ++ .../plan_cloud_G06_0.log} | 0 .../CODE_REVIEW-cloud-G06.md | 33 -- apps/control-plane/cmd/control-plane/main.go | 409 +++++++++++++++--- .../cmd/control-plane/main_test.go | 242 ++++++++++- .../internal/wire/edge_registry.go | 195 +++++++++ .../internal/wire/edge_server.go | 57 +++ .../internal/wire/edge_server_test.go | 177 ++++++++ 9 files changed, 1125 insertions(+), 88 deletions(-) create mode 100644 agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/code_review_cloud_G06_0.log create mode 100644 agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/complete.log rename agent-task/{m-multi-edge-operations/03+01,02_control_plane_fleet_api/PLAN-cloud-G06.md => archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/plan_cloud_G06_0.log} (100%) delete mode 100644 agent-task/m-multi-edge-operations/03+01,02_control_plane_fleet_api/CODE_REVIEW-cloud-G06.md diff --git a/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/code_review_cloud_G06_0.log b/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/code_review_cloud_G06_0.log new file mode 100644 index 0000000..172469e --- /dev/null +++ b/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/code_review_cloud_G06_0.log @@ -0,0 +1,61 @@ + +# Code Review - Control Plane Fleet API + +## Review Scope + +- `apps/control-plane/internal/wire/edge_server.go` +- `apps/control-plane/internal/wire/edge_registry.go` +- `apps/control-plane/cmd/control-plane/main.go` +- Control Plane API/wire tests + +## Dependency Check + +- [x] `agent-task/m-multi-edge-operations/01_native_contract/complete.log` 확인 +- [x] `agent-task/m-multi-edge-operations/02+01_edge_fleet_ops/complete.log` 확인 + +## Reviewer Checklist + +- [x] Control Plane 저장 모델이 connection/control/audit view에 머문다. +- [x] Edge 설정/Node registry/runtime 원본을 복제하지 않는다. +- [x] command routing이 edge별 active client로 정확히 분리된다. +- [x] offline/reconnect/timeout 응답이 명확하다. +- [x] bounded audit/event 저장 정책이 있다. +- [x] 기존 Edge 목록/status/events API가 회귀하지 않았다. + +## Verification Result + +- [x] `go test -count=1 ./apps/control-plane/internal/wire ./apps/control-plane/cmd/control-plane` +- [x] `go test -count=1 ./apps/control-plane/...` +- [x] `go test -count=1 ./apps/edge/internal/controlplane` +- [x] `git diff --check` +- [x] `make test-control-plane-edge-wire` + +## User Review + +검토 기본값: 이 리뷰는 사용자가 명시적으로 보류하지 않으면 구현자가 자체 검증 후 완료 처리한다. + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - Correctness: Pass + - Completeness: Pass + - Test coverage: Pass + - API contract: Pass + - Code quality: Pass + - Plan deviation: Pass + - Verification trust: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS이므로 active plan/review를 로그로 아카이브하고 `complete.log` 작성 후 task directory를 archive로 이동한다. + +## 코드리뷰 전용 체크리스트 + +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append했다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-cloud-G06.md`를 `code_review_cloud_G06_0.log`로 아카이브했다. +- [x] active `PLAN-cloud-G06.md`를 `plan_cloud_G06_0.log`로 아카이브했다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인했다. +- [x] PASS이므로 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않았다. +- [x] PASS이므로 active task 디렉터리 `agent-task/m-multi-edge-operations/03+01,02_control_plane_fleet_api/`를 `agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/`로 이동했다. +- [x] PASS이고 task group이 `m-multi-edge-operations`이므로 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이며 active parent `agent-task/m-multi-edge-operations/`에는 sibling 작업이 남아 유지했다. diff --git a/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/complete.log b/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/complete.log new file mode 100644 index 0000000..003cefd --- /dev/null +++ b/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/complete.log @@ -0,0 +1,39 @@ +# Complete - m-multi-edge-operations/03+01,02_control_plane_fleet_api + +## 완료 일시 + +2026-06-04 + +## 요약 + +Control Plane Fleet API 구현을 1회 리뷰 루프로 검토했고 최종 판정은 PASS다. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G06_0.log` | `code_review_cloud_G06_0.log` | PASS | Multi-edge command API, command result/event relay, fleet status summary API, bounded audit 저장, edge별 command routing이 계획 범위를 충족했다. | + +## 구현/정리 내용 + +- Control Plane HTTP API에 `/edges/{id}/status`, `/edges/{id}/operations`, `POST /edges/{id}/commands`, `/fleet/status`, `POST /fleet/commands`를 추가했다. +- `EdgeServer.SendCommand`와 `EdgeRegistry` command audit/event view를 추가해 target Edge의 active client로만 command를 보내고 result/event를 bounded in-memory view로 보관한다. +- API 응답은 connection/control/audit summary와 Edge-provided capability/domain-agent summary로 제한하고 Edge 설정, Node registry, runtime 원본 복제를 피했다. +- command routing, offline error, command lifecycle event relay, bounded audit copy, fleet status/command fan-out, 기존 Edge list/status/events 회귀 테스트를 보강했다. + +## 최종 검증 + +- `git diff --check` - PASS; 출력 없음. +- `go test -count=1 ./apps/control-plane/internal/wire ./apps/control-plane/cmd/control-plane` - PASS; `iop/apps/control-plane/internal/wire` 1.494s, `iop/apps/control-plane/cmd/control-plane` 0.016s. +- `go test -count=1 ./apps/control-plane/...` - PASS; `cmd/control-plane`와 `internal/wire` 통과. +- `go test -count=1 ./apps/edge/internal/controlplane` - PASS; `iop/apps/edge/internal/controlplane` 4.453s. +- `go test -count=1 ./apps/control-plane/internal/wire -run 'TestEdgeServer(SendsCommandToTargetEdgeOnly|RecordsCommandLifecycleEventsFromEdge|SendCommandFailsWhenEdgeNotConnected)|TestEdgeRegistryCommandsAreBoundedAndCopied'` - PASS; `iop/apps/control-plane/internal/wire` 0.187s. +- `make test-control-plane-edge-wire` - PASS; Control Plane-Edge wire smoke에서 hello accepted, connected to control plane, edge disconnected marker를 확인했다. + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-multi-edge-operations/03+01,02_control_plane_fleet_api/PLAN-cloud-G06.md b/agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/plan_cloud_G06_0.log similarity index 100% rename from agent-task/m-multi-edge-operations/03+01,02_control_plane_fleet_api/PLAN-cloud-G06.md rename to agent-task/archive/2026/06/m-multi-edge-operations/03+01,02_control_plane_fleet_api/plan_cloud_G06_0.log diff --git a/agent-task/m-multi-edge-operations/03+01,02_control_plane_fleet_api/CODE_REVIEW-cloud-G06.md b/agent-task/m-multi-edge-operations/03+01,02_control_plane_fleet_api/CODE_REVIEW-cloud-G06.md deleted file mode 100644 index 7054d78..0000000 --- a/agent-task/m-multi-edge-operations/03+01,02_control_plane_fleet_api/CODE_REVIEW-cloud-G06.md +++ /dev/null @@ -1,33 +0,0 @@ - -# Code Review - Control Plane Fleet API - -## Review Scope - -- `apps/control-plane/internal/wire/edge_server.go` -- `apps/control-plane/internal/wire/edge_registry.go` -- `apps/control-plane/cmd/control-plane/main.go` -- Control Plane API/wire tests - -## Dependency Check - -- [ ] `agent-task/m-multi-edge-operations/01_native_contract/complete.log` 확인 -- [ ] `agent-task/m-multi-edge-operations/02+01_edge_fleet_ops/complete.log` 확인 - -## Reviewer Checklist - -- [ ] Control Plane 저장 모델이 connection/control/audit view에 머문다. -- [ ] Edge 설정/Node registry/runtime 원본을 복제하지 않는다. -- [ ] command routing이 edge별 active client로 정확히 분리된다. -- [ ] offline/reconnect/timeout 응답이 명확하다. -- [ ] bounded audit/event 저장 정책이 있다. -- [ ] 기존 Edge 목록/status/events API가 회귀하지 않았다. - -## Verification Result - -- [ ] `go test ./apps/control-plane/internal/wire ./apps/control-plane/cmd/control-plane` -- [ ] 필요한 경우 `go test ./apps/control-plane/...` -- [ ] 존재 시 `make test-control-plane-edge-wire` - -## User Review - -검토 기본값: 이 리뷰는 사용자가 명시적으로 보류하지 않으면 구현자가 자체 검증 후 완료 처리한다. diff --git a/apps/control-plane/cmd/control-plane/main.go b/apps/control-plane/cmd/control-plane/main.go index c42b1dc..284a05b 100644 --- a/apps/control-plane/cmd/control-plane/main.go +++ b/apps/control-plane/cmd/control-plane/main.go @@ -179,7 +179,8 @@ func run(ctx context.Context, cfg controlPlaneConfig, logger *zap.Logger) error return fmt.Errorf("start edge wire server: %w", err) } defer func() { _ = edgeServer.Stop() }() - registerEdgeRegistryHandlers(mux, edgeServer.Registry(), edgeServer.RequestStatus) + registerEdgeRegistryHandlers(mux, edgeServer.Registry(), edgeServer.RequestStatus, edgeServer.SendCommand) + registerFleetHandlers(mux, edgeServer.Registry(), edgeServer.RequestStatus, edgeServer.SendCommand) server := &http.Server{ Addr: cfg.Server.Listen, @@ -240,13 +241,96 @@ type edgeNodeEventView struct { } type edgeStatusResponseView struct { - RequestID string `json:"request_id"` - EdgeID string `json:"edge_id"` - EdgeName string `json:"edge_name"` - ObservedTimeUnixNano int64 `json:"observed_time_unix_nano"` - Nodes []edgeNodeSnapshotView `json:"nodes"` - Metadata map[string]string `json:"metadata,omitempty"` - Error string `json:"error,omitempty"` + RequestID string `json:"request_id"` + EdgeID string `json:"edge_id"` + EdgeName string `json:"edge_name"` + ObservedTimeUnixNano int64 `json:"observed_time_unix_nano"` + Nodes []edgeNodeSnapshotView `json:"nodes"` + Capabilities []edgeCapabilitySummaryView `json:"capabilities,omitempty"` + DomainAgents []edgeDomainAgentSummaryView `json:"domain_agents,omitempty"` + Metadata map[string]string `json:"metadata,omitempty"` + Error string `json:"error,omitempty"` +} + +type edgeCapabilitySummaryView struct { + Kind string `json:"kind"` + Available bool `json:"available"` + Status string `json:"status,omitempty"` + Summary string `json:"summary,omitempty"` +} + +type edgeDomainAgentSummaryView struct { + AgentKind string `json:"agent_kind"` + Available bool `json:"available"` + LifecycleState string `json:"lifecycle_state,omitempty"` + ActiveCommandID string `json:"active_command_id,omitempty"` + Summary string `json:"summary,omitempty"` +} + +// edgeCommandRequestBody is the JSON body for POST /edges/{id}/commands and +// POST /fleet/commands. It carries the surface-neutral command intent only. +type edgeCommandRequestBody struct { + Operation string `json:"operation"` + TargetSelector string `json:"target_selector,omitempty"` + Parameters map[string]string `json:"parameters,omitempty"` +} + +type edgeCommandResponseView struct { + RequestID string `json:"request_id,omitempty"` + CommandID string `json:"command_id"` + EdgeID string `json:"edge_id"` + Status string `json:"status"` + Summary string `json:"summary,omitempty"` + Error string `json:"error,omitempty"` +} + +type edgeOperationsResponse struct { + EdgeID string `json:"edge_id"` + Operations []edgeCommandRecordView `json:"operations"` +} + +type edgeCommandRecordView struct { + EdgeID string `json:"edge_id"` + CommandID string `json:"command_id"` + Operation string `json:"operation,omitempty"` + TargetSelector string `json:"target_selector,omitempty"` + Status string `json:"status"` + Summary string `json:"summary,omitempty"` + Error string `json:"error,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + Events []edgeCommandEventView `json:"events,omitempty"` +} + +type edgeCommandEventView struct { + Phase string `json:"phase,omitempty"` + Summary string `json:"summary,omitempty"` + OccurredAt time.Time `json:"occurred_at,omitempty"` + ReceivedAt time.Time `json:"received_at"` +} + +type fleetStatusResponse struct { + GeneratedAt time.Time `json:"generated_at"` + Edges []fleetEdgeView `json:"edges"` +} + +type fleetEdgeView struct { + EdgeID string `json:"edge_id"` + EdgeName string `json:"edge_name,omitempty"` + Version string `json:"version,omitempty"` + Protocol string `json:"protocol,omitempty"` + Connected bool `json:"connected"` + Health string `json:"health"` + LastSeen time.Time `json:"last_seen"` + NodeCount int `json:"node_count"` + Capabilities []edgeCapabilitySummaryView `json:"capabilities,omitempty"` + DomainAgents []edgeDomainAgentSummaryView `json:"domain_agents,omitempty"` + Error string `json:"error,omitempty"` +} + +type fleetCommandResponse struct { + Operation string `json:"operation"` + Results []edgeCommandResponseView `json:"results"` } type edgeNodeSnapshotView struct { @@ -288,15 +372,21 @@ func newNodeConfigSummaryView(payload *iop.NodeConfigPayload) *nodeConfigSummary } } -func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistry, requestStatus func(string, time.Duration) (*iop.EdgeStatusResponse, error)) { +// edgeStatusRequester asks a connected Edge for its Edge-owned node snapshot. +type edgeStatusRequester func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) + +// edgeCommandSender dispatches a command to a connected Edge and returns its +// typed response. +type edgeCommandSender func(edgeID, operation, targetSelector string, parameters map[string]string, timeout time.Duration) (*iop.EdgeCommandResponse, error) + +func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistry, requestStatus edgeStatusRequester, sendCommand edgeCommandSender) { mux.HandleFunc("/edges", func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/edges" { http.NotFound(w, r) return } if r.Method != http.MethodGet { - w.Header().Set("Allow", http.MethodGet) - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + methodNotAllowed(w, http.MethodGet) return } states := registry.Snapshots() @@ -307,52 +397,48 @@ func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistr writeJSON(w, http.StatusOK, edgeRegistryResponse{Edges: views}) }) mux.HandleFunc("/edges/", func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - w.Header().Set("Allow", http.MethodGet) - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) - return - } edgePath := strings.TrimPrefix(r.URL.Path, "/edges/") edgeID, rest, hasRest := strings.Cut(edgePath, "/") if edgeID == "" { http.NotFound(w, r) return } - if hasRest { - if rest != "events" && rest != "status" { + if !hasRest { + if r.Method != http.MethodGet { + methodNotAllowed(w, http.MethodGet) + return + } + state, ok := registry.Snapshot(edgeID) + if !ok { http.NotFound(w, r) return } + writeJSON(w, http.StatusOK, edgeRegistryViewFromState(state)) + return + } + switch rest { + case "status": + if r.Method != http.MethodGet { + methodNotAllowed(w, http.MethodGet) + return + } if _, ok := registry.Snapshot(edgeID); !ok { http.NotFound(w, r) return } - if rest == "status" { - resp, err := requestStatus(edgeID, 5*time.Second) - if err != nil { - writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()}) - return - } - nodes := make([]edgeNodeSnapshotView, 0, len(resp.GetNodes())) - for _, n := range resp.GetNodes() { - nodes = append(nodes, edgeNodeSnapshotView{ - NodeID: n.GetNodeId(), - Alias: n.GetAlias(), - Label: n.GetLabel(), - Connected: n.GetConnected(), - Config: newNodeConfigSummaryView(n.GetConfig()), - }) - } - view := edgeStatusResponseView{ - RequestID: resp.GetRequestId(), - EdgeID: resp.GetEdgeId(), - EdgeName: resp.GetEdgeName(), - ObservedTimeUnixNano: resp.GetObservedTimeUnixNano(), - Nodes: nodes, - Metadata: resp.GetMetadata(), - Error: resp.GetError(), - } - writeJSON(w, http.StatusOK, view) + resp, err := requestStatus(edgeID, 5*time.Second) + if err != nil { + writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, http.StatusOK, edgeStatusResponseViewFromProto(resp)) + case "events": + if r.Method != http.MethodGet { + methodNotAllowed(w, http.MethodGet) + return + } + if _, ok := registry.Snapshot(edgeID); !ok { + http.NotFound(w, r) return } records := registry.NodeEvents(edgeID, r.URL.Query().Get("node_id")) @@ -361,15 +447,111 @@ func registerEdgeRegistryHandlers(mux *http.ServeMux, registry *wire.EdgeRegistr views = append(views, edgeNodeEventViewFromRecord(record)) } writeJSON(w, http.StatusOK, edgeNodeEventsResponse{EdgeID: edgeID, Events: views}) - return - } - state, ok := registry.Snapshot(edgeID) - if !ok { + case "operations": + if r.Method != http.MethodGet { + methodNotAllowed(w, http.MethodGet) + return + } + if _, ok := registry.Snapshot(edgeID); !ok { + http.NotFound(w, r) + return + } + records := registry.Commands(edgeID) + views := make([]edgeCommandRecordView, 0, len(records)) + for _, record := range records { + views = append(views, edgeCommandRecordViewFromRecord(record)) + } + writeJSON(w, http.StatusOK, edgeOperationsResponse{EdgeID: edgeID, Operations: views}) + case "commands": + if r.Method != http.MethodPost { + methodNotAllowed(w, http.MethodPost) + return + } + handleEdgeCommandPost(w, r, edgeID, sendCommand) + default: http.NotFound(w, r) + } + }) +} + +// handleEdgeCommandPost decodes a command request body and dispatches it to a +// single target Edge. A not-connected Edge yields a 502 with the wire error. +func handleEdgeCommandPost(w http.ResponseWriter, r *http.Request, edgeID string, sendCommand edgeCommandSender) { + var body edgeCommandRequestBody + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"}) + return + } + if body.Operation == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "operation is required"}) + return + } + if sendCommand == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "command dispatch is not configured"}) + return + } + resp, err := sendCommand(edgeID, body.Operation, body.TargetSelector, body.Parameters, 10*time.Second) + if err != nil { + writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, http.StatusAccepted, edgeCommandResponseViewFromProto(resp)) +} + +func registerFleetHandlers(mux *http.ServeMux, registry *wire.EdgeRegistry, requestStatus edgeStatusRequester, sendCommand edgeCommandSender) { + mux.HandleFunc("/fleet/status", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + methodNotAllowed(w, http.MethodGet) return } - writeJSON(w, http.StatusOK, edgeRegistryViewFromState(state)) + states := registry.Snapshots() + views := make([]fleetEdgeView, 0, len(states)) + for _, state := range states { + views = append(views, fleetEdgeViewFromState(state, requestStatus)) + } + writeJSON(w, http.StatusOK, fleetStatusResponse{GeneratedAt: time.Now().UTC(), Edges: views}) }) + mux.HandleFunc("/fleet/commands", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + methodNotAllowed(w, http.MethodPost) + return + } + var body edgeCommandRequestBody + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"}) + return + } + if body.Operation == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "operation is required"}) + return + } + if sendCommand == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "command dispatch is not configured"}) + return + } + results := make([]edgeCommandResponseView, 0) + for _, state := range registry.Snapshots() { + if !state.Connected { + continue + } + resp, err := sendCommand(state.EdgeID, body.Operation, body.TargetSelector, body.Parameters, 10*time.Second) + if err != nil { + results = append(results, edgeCommandResponseView{ + EdgeID: state.EdgeID, + Status: "error", + Error: err.Error(), + }) + continue + } + results = append(results, edgeCommandResponseViewFromProto(resp)) + } + writeJSON(w, http.StatusAccepted, fleetCommandResponse{Operation: body.Operation, Results: results}) + }) +} + +func methodNotAllowed(w http.ResponseWriter, allow ...string) { + w.Header().Set("Allow", strings.Join(allow, ", ")) + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } func edgeNodeEventViewFromRecord(record wire.EdgeNodeEventRecord) edgeNodeEventView { @@ -411,6 +593,135 @@ func edgeRegistryViewFromState(state wire.EdgeConnectionState) edgeRegistryView } } +func edgeStatusResponseViewFromProto(resp *iop.EdgeStatusResponse) edgeStatusResponseView { + nodes := make([]edgeNodeSnapshotView, 0, len(resp.GetNodes())) + for _, n := range resp.GetNodes() { + nodes = append(nodes, edgeNodeSnapshotView{ + NodeID: n.GetNodeId(), + Alias: n.GetAlias(), + Label: n.GetLabel(), + Connected: n.GetConnected(), + Config: newNodeConfigSummaryView(n.GetConfig()), + }) + } + return edgeStatusResponseView{ + RequestID: resp.GetRequestId(), + EdgeID: resp.GetEdgeId(), + EdgeName: resp.GetEdgeName(), + ObservedTimeUnixNano: resp.GetObservedTimeUnixNano(), + Nodes: nodes, + Capabilities: edgeCapabilityViews(resp.GetCapabilities()), + DomainAgents: edgeDomainAgentViews(resp.GetDomainAgents()), + Metadata: resp.GetMetadata(), + Error: resp.GetError(), + } +} + +func edgeCapabilityViews(caps []*iop.EdgeCapabilitySummary) []edgeCapabilitySummaryView { + if len(caps) == 0 { + return nil + } + views := make([]edgeCapabilitySummaryView, 0, len(caps)) + for _, c := range caps { + views = append(views, edgeCapabilitySummaryView{ + Kind: c.GetKind(), + Available: c.GetAvailable(), + Status: c.GetStatus(), + Summary: c.GetSummary(), + }) + } + return views +} + +func edgeDomainAgentViews(agents []*iop.EdgeDomainAgentSummary) []edgeDomainAgentSummaryView { + if len(agents) == 0 { + return nil + } + views := make([]edgeDomainAgentSummaryView, 0, len(agents)) + for _, a := range agents { + views = append(views, edgeDomainAgentSummaryView{ + AgentKind: a.GetAgentKind(), + Available: a.GetAvailable(), + LifecycleState: a.GetLifecycleState(), + ActiveCommandID: a.GetActiveCommandId(), + Summary: a.GetSummary(), + }) + } + return views +} + +func edgeCommandResponseViewFromProto(resp *iop.EdgeCommandResponse) edgeCommandResponseView { + return edgeCommandResponseView{ + RequestID: resp.GetRequestId(), + CommandID: resp.GetCommandId(), + EdgeID: resp.GetEdgeId(), + Status: resp.GetStatus(), + Summary: resp.GetSummary(), + Error: resp.GetError(), + } +} + +func edgeCommandRecordViewFromRecord(record wire.EdgeCommandRecord) edgeCommandRecordView { + view := edgeCommandRecordView{ + EdgeID: record.EdgeID, + CommandID: record.CommandID, + Operation: record.Operation, + TargetSelector: record.TargetSelector, + Status: record.Status, + Summary: record.Summary, + Error: record.Error, + CreatedAt: record.CreatedAt, + UpdatedAt: record.UpdatedAt, + } + if len(record.Events) > 0 { + view.Events = make([]edgeCommandEventView, 0, len(record.Events)) + for _, e := range record.Events { + view.Events = append(view.Events, edgeCommandEventView{ + Phase: e.Phase, + Summary: e.Summary, + OccurredAt: e.OccurredAt, + ReceivedAt: e.ReceivedAt, + }) + } + } + return view +} + +func fleetEdgeViewFromState(state wire.EdgeConnectionState, requestStatus edgeStatusRequester) fleetEdgeView { + view := fleetEdgeView{ + EdgeID: state.EdgeID, + EdgeName: state.EdgeName, + Version: state.Version, + Protocol: state.Protocol, + Connected: state.Connected, + LastSeen: state.LastSeen, + } + if !state.Connected { + view.Health = "offline" + return view + } + if requestStatus == nil { + view.Health = "online" + return view + } + resp, err := requestStatus(state.EdgeID, 5*time.Second) + if err != nil { + view.Health = "degraded" + view.Error = err.Error() + return view + } + view.NodeCount = len(resp.GetNodes()) + view.Capabilities = edgeCapabilityViews(resp.GetCapabilities()) + view.DomainAgents = edgeDomainAgentViews(resp.GetDomainAgents()) + if resp.GetError() != "" { + view.Health = "degraded" + view.Error = resp.GetError() + return view + } + view.Health = "online" + return view +} + func writeJSON(w http.ResponseWriter, status int, v any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) diff --git a/apps/control-plane/cmd/control-plane/main_test.go b/apps/control-plane/cmd/control-plane/main_test.go index 5a927d6..dc2df0b 100644 --- a/apps/control-plane/cmd/control-plane/main_test.go +++ b/apps/control-plane/cmd/control-plane/main_test.go @@ -11,6 +11,7 @@ import ( "testing" "time" + proto_socket "git.toki-labs.com/toki/proto-socket/go" "google.golang.org/protobuf/types/known/structpb" "iop/apps/control-plane/internal/wire" @@ -256,7 +257,7 @@ func TestEdgeRegistryHTTPHandlersListAndGetEdge(t *testing.T) { }, at.Add(time.Second)) mux := http.NewServeMux() - registerEdgeRegistryHandlers(mux, registry, nil) + registerEdgeRegistryHandlers(mux, registry, nil, nil) listResp := httptest.NewRecorder() mux.ServeHTTP(listResp, httptest.NewRequest(http.MethodGet, "/edges", nil)) @@ -334,7 +335,7 @@ func TestEdgeRegistryHTTPHandlersListNodeEvents(t *testing.T) { }, receivedAt) mux := http.NewServeMux() - registerEdgeRegistryHandlers(mux, registry, nil) + registerEdgeRegistryHandlers(mux, registry, nil, nil) resp := httptest.NewRecorder() mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/events", nil)) @@ -397,7 +398,7 @@ func TestEdgeRegistryHTTPHandlersListNodeEvents(t *testing.T) { func TestEdgeRegistryHTTPHandlersRejectUnsupportedCases(t *testing.T) { registry := wire.NewEdgeRegistry() mux := http.NewServeMux() - registerEdgeRegistryHandlers(mux, registry, nil) + registerEdgeRegistryHandlers(mux, registry, nil, nil) postResp := httptest.NewRecorder() mux.ServeHTTP(postResp, httptest.NewRequest(http.MethodPost, "/edges", nil)) @@ -463,7 +464,7 @@ func TestEdgeRegistryHTTPHandlersGetLiveStatus(t *testing.T) { }, nil } - registerEdgeRegistryHandlers(mux, registry, requestStatus) + registerEdgeRegistryHandlers(mux, registry, requestStatus, nil) resp := httptest.NewRecorder() mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/status", nil)) @@ -511,7 +512,7 @@ func TestEdgeRegistryHTTPHandlersGetLiveStatus(t *testing.T) { requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) { return nil, fmt.Errorf("should not be called") } - registerEdgeRegistryHandlers(mux, registry, requestStatus) + registerEdgeRegistryHandlers(mux, registry, requestStatus, nil) resp := httptest.NewRecorder() mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/missing-edge/status", nil)) @@ -526,7 +527,7 @@ func TestEdgeRegistryHTTPHandlersGetLiveStatus(t *testing.T) { requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) { return nil, fmt.Errorf("upstream timeout") } - registerEdgeRegistryHandlers(mux, registry, requestStatus) + registerEdgeRegistryHandlers(mux, registry, requestStatus, nil) resp := httptest.NewRecorder() mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/status", nil)) @@ -544,3 +545,232 @@ func TestEdgeRegistryHTTPHandlersGetLiveStatus(t *testing.T) { } }) } + +func TestEdgeCommandHTTPHandlerDispatchesToTargetEdge(t *testing.T) { + registry := wire.NewEdgeRegistry() + registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-a", EdgeName: "Edge A"}, time.Unix(1780142200, 0).UTC()) + + t.Run("Dispatch", func(t *testing.T) { + var ( + gotEdge string + gotOperation string + gotSelector string + gotParams map[string]string + ) + sendCommand := func(edgeID, operation, targetSelector string, parameters map[string]string, timeout time.Duration) (*iop.EdgeCommandResponse, error) { + gotEdge = edgeID + gotOperation = operation + gotSelector = targetSelector + gotParams = parameters + return &iop.EdgeCommandResponse{ + RequestId: "req-1", + CommandId: "cmd-1", + EdgeId: edgeID, + Status: "ok", + Summary: "applied", + }, nil + } + mux := http.NewServeMux() + registerEdgeRegistryHandlers(mux, registry, nil, sendCommand) + + body := strings.NewReader(`{"operation":"restart-node","target_selector":"node-1","parameters":{"force":"true"}}`) + resp := httptest.NewRecorder() + mux.ServeHTTP(resp, httptest.NewRequest(http.MethodPost, "/edges/edge-a/commands", body)) + + if resp.Code != http.StatusAccepted { + t.Fatalf("POST commands status=%d body=%s", resp.Code, resp.Body.String()) + } + if gotEdge != "edge-a" || gotOperation != "restart-node" || gotSelector != "node-1" { + t.Fatalf("unexpected dispatch args: edge=%q op=%q selector=%q", gotEdge, gotOperation, gotSelector) + } + if gotParams["force"] != "true" { + t.Fatalf("unexpected parameters: %+v", gotParams) + } + var view edgeCommandResponseView + if err := json.Unmarshal(resp.Body.Bytes(), &view); err != nil { + t.Fatalf("decode command response: %v", err) + } + if view.CommandID != "cmd-1" || view.EdgeID != "edge-a" || view.Status != "ok" { + t.Fatalf("unexpected command view: %+v", view) + } + }) + + t.Run("MissingOperation", func(t *testing.T) { + mux := http.NewServeMux() + registerEdgeRegistryHandlers(mux, registry, nil, func(string, string, string, map[string]string, time.Duration) (*iop.EdgeCommandResponse, error) { + t.Fatal("sendCommand should not be called without an operation") + return nil, nil + }) + resp := httptest.NewRecorder() + mux.ServeHTTP(resp, httptest.NewRequest(http.MethodPost, "/edges/edge-a/commands", strings.NewReader(`{}`))) + if resp.Code != http.StatusBadRequest { + t.Fatalf("expected 400 for missing operation, got %d", resp.Code) + } + }) + + t.Run("OfflineEdgeReturnsError", func(t *testing.T) { + sendCommand := func(string, string, string, map[string]string, time.Duration) (*iop.EdgeCommandResponse, error) { + return nil, fmt.Errorf("edge \"edge-a\" is not connected") + } + mux := http.NewServeMux() + registerEdgeRegistryHandlers(mux, registry, nil, sendCommand) + + resp := httptest.NewRecorder() + mux.ServeHTTP(resp, httptest.NewRequest(http.MethodPost, "/edges/edge-a/commands", strings.NewReader(`{"operation":"drain"}`))) + if resp.Code != http.StatusBadGateway { + t.Fatalf("expected 502 for offline edge, got %d body=%s", resp.Code, resp.Body.String()) + } + var errResp map[string]string + if err := json.Unmarshal(resp.Body.Bytes(), &errResp); err != nil { + t.Fatalf("decode error: %v", err) + } + if !strings.Contains(errResp["error"], "not connected") { + t.Fatalf("unexpected error: %+v", errResp) + } + }) + + t.Run("GetMethodNotAllowed", func(t *testing.T) { + mux := http.NewServeMux() + registerEdgeRegistryHandlers(mux, registry, nil, nil) + resp := httptest.NewRecorder() + mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/commands", nil)) + if resp.Code != http.StatusMethodNotAllowed { + t.Fatalf("expected 405 for GET on commands, got %d", resp.Code) + } + }) +} + +func TestEdgeOperationsHTTPHandlerListsAudit(t *testing.T) { + registry := wire.NewEdgeRegistry() + at := time.Unix(1780142200, 0).UTC() + registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-a", EdgeName: "Edge A"}, at) + registry.RecordCommandRequest("edge-a", &iop.EdgeCommandRequest{CommandId: "cmd-1", Operation: "restart-node", TargetSelector: "node-1"}, at) + registry.RecordCommandEvent("edge-a", &iop.EdgeCommandEvent{CommandId: "cmd-1", Phase: "running", Summary: "executing"}, at.Add(time.Second)) + registry.RecordCommandResult("edge-a", &iop.EdgeCommandResponse{CommandId: "cmd-1", Status: "ok", Summary: "done"}, at.Add(2*time.Second)) + + mux := http.NewServeMux() + registerEdgeRegistryHandlers(mux, registry, nil, nil) + + resp := httptest.NewRecorder() + mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/edges/edge-a/operations", nil)) + if resp.Code != http.StatusOK { + t.Fatalf("GET operations status=%d body=%s", resp.Code, resp.Body.String()) + } + var got edgeOperationsResponse + if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil { + t.Fatalf("decode operations: %v", err) + } + if got.EdgeID != "edge-a" || len(got.Operations) != 1 { + t.Fatalf("unexpected operations response: %+v", got) + } + op := got.Operations[0] + if op.CommandID != "cmd-1" || op.Operation != "restart-node" || op.Status != "ok" || op.Summary != "done" { + t.Fatalf("unexpected operation record: %+v", op) + } + if len(op.Events) != 1 || op.Events[0].Phase != "running" { + t.Fatalf("unexpected operation events: %+v", op.Events) + } +} + +func TestFleetStatusHTTPHandlerCombinesConnectionAndCapabilities(t *testing.T) { + registry := wire.NewEdgeRegistry() + at := time.Unix(1780142200, 0).UTC() + registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-online", EdgeName: "Online", Version: "1.0.0"}, at) + registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-degraded", EdgeName: "Degraded"}, at) + registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-offline", EdgeName: "Offline"}, at) + registry.MarkDisconnected("edge-offline", 3, at.Add(time.Second), proto_socket.DisconnectInfo{Reason: "closed"}) + + requestStatus := func(edgeID string, timeout time.Duration) (*iop.EdgeStatusResponse, error) { + switch edgeID { + case "edge-online": + return &iop.EdgeStatusResponse{ + EdgeId: edgeID, + Nodes: []*iop.EdgeNodeSnapshot{{NodeId: "node-1", Alias: "alpha", Connected: true}}, + Capabilities: []*iop.EdgeCapabilitySummary{ + {Kind: "run-dispatch", Available: true, Status: "ready"}, + }, + }, nil + case "edge-degraded": + return nil, fmt.Errorf("status timeout") + default: + return nil, fmt.Errorf("unexpected status request for %q", edgeID) + } + } + + mux := http.NewServeMux() + registerFleetHandlers(mux, registry, requestStatus, nil) + + resp := httptest.NewRecorder() + mux.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/fleet/status", nil)) + if resp.Code != http.StatusOK { + t.Fatalf("GET /fleet/status status=%d body=%s", resp.Code, resp.Body.String()) + } + var got fleetStatusResponse + if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil { + t.Fatalf("decode fleet status: %v", err) + } + if len(got.Edges) != 3 { + t.Fatalf("expected 3 fleet edges, got %d", len(got.Edges)) + } + health := map[string]fleetEdgeView{} + for _, e := range got.Edges { + health[e.EdgeID] = e + } + if e := health["edge-online"]; e.Health != "online" || e.NodeCount != 1 || len(e.Capabilities) != 1 { + t.Fatalf("unexpected online edge view: %+v", e) + } + if e := health["edge-degraded"]; e.Health != "degraded" || e.Error == "" { + t.Fatalf("unexpected degraded edge view: %+v", e) + } + if e := health["edge-offline"]; e.Health != "offline" || e.Connected { + t.Fatalf("unexpected offline edge view: %+v", e) + } + + // Fleet status must not leak Node-direct internals (addresses, tokens) or + // raw node snapshots; it surfaces a connection + capability summary only. + bodyStr := resp.Body.String() + for _, forbidden := range []string{"node_id", "token", "address", "\"nodes\""} { + if strings.Contains(bodyStr, forbidden) { + t.Fatalf("fleet status leaked %q: %s", forbidden, bodyStr) + } + } +} + +func TestFleetCommandsHTTPHandlerFansOutToConnectedEdgesOnly(t *testing.T) { + registry := wire.NewEdgeRegistry() + at := time.Unix(1780142200, 0).UTC() + registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-a"}, at) + registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-b"}, at) + registry.MarkConnected(&iop.EdgeHelloRequest{EdgeId: "edge-off"}, at) + registry.MarkDisconnected("edge-off", 3, at.Add(time.Second), proto_socket.DisconnectInfo{Reason: "closed"}) + + var dispatched []string + sendCommand := func(edgeID, operation, targetSelector string, parameters map[string]string, timeout time.Duration) (*iop.EdgeCommandResponse, error) { + dispatched = append(dispatched, edgeID) + return &iop.EdgeCommandResponse{CommandId: "cmd-" + edgeID, EdgeId: edgeID, Status: "ok"}, nil + } + + mux := http.NewServeMux() + registerFleetHandlers(mux, registry, nil, sendCommand) + + resp := httptest.NewRecorder() + mux.ServeHTTP(resp, httptest.NewRequest(http.MethodPost, "/fleet/commands", strings.NewReader(`{"operation":"drain"}`))) + if resp.Code != http.StatusAccepted { + t.Fatalf("POST /fleet/commands status=%d body=%s", resp.Code, resp.Body.String()) + } + var got fleetCommandResponse + if err := json.Unmarshal(resp.Body.Bytes(), &got); err != nil { + t.Fatalf("decode fleet command response: %v", err) + } + if got.Operation != "drain" || len(got.Results) != 2 { + t.Fatalf("expected fan-out to 2 connected edges, got %+v", got) + } + if len(dispatched) != 2 { + t.Fatalf("expected dispatch to 2 connected edges, got %+v", dispatched) + } + for _, edge := range dispatched { + if edge == "edge-off" { + t.Fatalf("command was dispatched to a disconnected edge: %+v", dispatched) + } + } +} diff --git a/apps/control-plane/internal/wire/edge_registry.go b/apps/control-plane/internal/wire/edge_registry.go index f5613ee..1da445f 100644 --- a/apps/control-plane/internal/wire/edge_registry.go +++ b/apps/control-plane/internal/wire/edge_registry.go @@ -11,6 +11,8 @@ import ( ) const recentNodeEventCapacity = 128 +const recentCommandCapacity = 128 +const commandEventCapacity = 64 // EdgeConnectionState captures the Control Plane's internal view of a single // Edge wire connection. @@ -38,12 +40,39 @@ type EdgeNodeEventRecord struct { ReceivedAt time.Time } +// EdgeCommandRecord is the Control Plane's bounded audit view of one command +// dispatched to an Edge: the request summary, the latest result, and a bounded +// list of relayed lifecycle events. It is a control/audit view; durable command +// history is intentionally outside this in-memory baseline. +type EdgeCommandRecord struct { + EdgeID string + CommandID string + Operation string + TargetSelector string + Status string + Summary string + Error string + CreatedAt time.Time + UpdatedAt time.Time + Events []EdgeCommandEventRecord +} + +// EdgeCommandEventRecord is the Control Plane's copied view of one command +// lifecycle event relayed from an Edge during command execution. +type EdgeCommandEventRecord struct { + Phase string + Summary string + OccurredAt time.Time + ReceivedAt time.Time +} + // EdgeRegistry stores Edge connection state keyed by edge_id behind a mutex. type EdgeRegistry struct { mu sync.Mutex edges map[string]*EdgeConnectionState nextToken uint64 recentNodeEvents []EdgeNodeEventRecord + commands []*EdgeCommandRecord } // NewEdgeRegistry returns an empty Edge connection registry. @@ -194,6 +223,172 @@ func (r *EdgeRegistry) NodeEvents(edgeID, nodeID string) []EdgeNodeEventRecord { return out } +// RecordCommandRequest stores or refreshes a pending command audit record for a +// command the Control Plane is dispatching to an Edge. It is bounded; the oldest +// record is evicted past capacity. +func (r *EdgeRegistry) RecordCommandRequest(edgeID string, req *iop.EdgeCommandRequest, at time.Time) { + if edgeID == "" || req == nil || req.GetCommandId() == "" { + return + } + if at.IsZero() { + at = time.Now() + } + r.mu.Lock() + defer r.mu.Unlock() + if rec := r.findCommandLocked(edgeID, req.GetCommandId()); rec != nil { + rec.Operation = req.GetOperation() + rec.TargetSelector = req.GetTargetSelector() + rec.UpdatedAt = at + return + } + rec := r.appendCommandLocked(edgeID, req.GetCommandId(), at) + rec.Operation = req.GetOperation() + rec.TargetSelector = req.GetTargetSelector() + rec.Status = "pending" +} + +// RecordCommandResult updates the audit record for a command with the Edge's +// response status/summary. A missing record is created so a result is never +// silently dropped. +func (r *EdgeRegistry) RecordCommandResult(edgeID string, resp *iop.EdgeCommandResponse, at time.Time) { + if edgeID == "" || resp == nil || resp.GetCommandId() == "" { + return + } + if at.IsZero() { + at = time.Now() + } + r.mu.Lock() + defer r.mu.Unlock() + rec := r.findCommandLocked(edgeID, resp.GetCommandId()) + if rec == nil { + rec = r.appendCommandLocked(edgeID, resp.GetCommandId(), at) + } + status := resp.GetStatus() + if status == "" { + status = "ok" + } + rec.Status = status + rec.Summary = resp.GetSummary() + rec.Error = resp.GetError() + rec.UpdatedAt = at +} + +// RecordCommandEventFromConnection appends a bounded command lifecycle event to +// the matching command audit record, but only when token still identifies the +// current connection for edgeID. It returns false for stale superseded +// connections. +func (r *EdgeRegistry) RecordCommandEventFromConnection(edgeID string, token uint64, event *iop.EdgeCommandEvent, receivedAt time.Time) bool { + return r.recordCommandEvent(edgeID, token, true, event, receivedAt) +} + +// RecordCommandEvent stores a command lifecycle event without a connection +// token. It is intended for tests and non-connection seed paths. +func (r *EdgeRegistry) RecordCommandEvent(edgeID string, event *iop.EdgeCommandEvent, receivedAt time.Time) { + r.recordCommandEvent(edgeID, 0, false, event, receivedAt) +} + +func (r *EdgeRegistry) recordCommandEvent(edgeID string, token uint64, requireToken bool, event *iop.EdgeCommandEvent, receivedAt time.Time) bool { + if edgeID == "" || event == nil || event.GetCommandId() == "" { + return false + } + if receivedAt.IsZero() { + receivedAt = time.Now() + } + occurredAt := receivedAt + if event.GetOccurredAt() != 0 { + occurredAt = time.Unix(0, event.GetOccurredAt()) + } + + r.mu.Lock() + defer r.mu.Unlock() + if requireToken { + state, ok := r.edges[edgeID] + if !ok || state.token != token { + return false + } + } + rec := r.findCommandLocked(edgeID, event.GetCommandId()) + if rec == nil { + rec = r.appendCommandLocked(edgeID, event.GetCommandId(), receivedAt) + rec.Status = "running" + } + if len(rec.Events) >= commandEventCapacity { + rec.Events = append(rec.Events[:0], rec.Events[1:]...) + } + rec.Events = append(rec.Events, EdgeCommandEventRecord{ + Phase: event.GetPhase(), + Summary: event.GetSummary(), + OccurredAt: occurredAt, + ReceivedAt: receivedAt, + }) + rec.UpdatedAt = receivedAt + return true +} + +// Commands returns copied command audit records for an Edge, oldest first. +func (r *EdgeRegistry) Commands(edgeID string) []EdgeCommandRecord { + if edgeID == "" { + return nil + } + r.mu.Lock() + defer r.mu.Unlock() + out := make([]EdgeCommandRecord, 0, len(r.commands)) + for _, rec := range r.commands { + if rec.EdgeID != edgeID { + continue + } + out = append(out, cloneEdgeCommandRecord(rec)) + } + if len(out) == 0 { + return nil + } + return out +} + +// Command returns a single copied command audit record. +func (r *EdgeRegistry) Command(edgeID, commandID string) (EdgeCommandRecord, bool) { + r.mu.Lock() + defer r.mu.Unlock() + rec := r.findCommandLocked(edgeID, commandID) + if rec == nil { + return EdgeCommandRecord{}, false + } + return cloneEdgeCommandRecord(rec), true +} + +// findCommandLocked returns the stored command record for an edge_id/command_id +// pair. Callers must hold r.mu. +func (r *EdgeRegistry) findCommandLocked(edgeID, commandID string) *EdgeCommandRecord { + for _, rec := range r.commands { + if rec.EdgeID == edgeID && rec.CommandID == commandID { + return rec + } + } + return nil +} + +// appendCommandLocked appends a new bounded command record. Callers must hold +// r.mu. +func (r *EdgeRegistry) appendCommandLocked(edgeID, commandID string, at time.Time) *EdgeCommandRecord { + if len(r.commands) >= recentCommandCapacity { + r.commands = append(r.commands[:0], r.commands[1:]...) + } + rec := &EdgeCommandRecord{ + EdgeID: edgeID, + CommandID: commandID, + CreatedAt: at, + UpdatedAt: at, + } + r.commands = append(r.commands, rec) + return rec +} + +func cloneEdgeCommandRecord(rec *EdgeCommandRecord) EdgeCommandRecord { + clone := *rec + clone.Events = append([]EdgeCommandEventRecord(nil), rec.Events...) + return clone +} + // Len returns the number of tracked Edge connections. func (r *EdgeRegistry) Len() int { r.mu.Lock() diff --git a/apps/control-plane/internal/wire/edge_server.go b/apps/control-plane/internal/wire/edge_server.go index 6ceb3bc..92e4355 100644 --- a/apps/control-plane/internal/wire/edge_server.go +++ b/apps/control-plane/internal/wire/edge_server.go @@ -83,6 +83,17 @@ func NewEdgeServer(listenAddr string, logger *zap.Logger) (*EdgeServer, error) { registry.RecordNodeEventFromConnection(id, tok, event, time.Now()) }) + proto_socket.AddListenerTyped[*iop.EdgeCommandEvent](&client.Communicator, func(event *iop.EdgeCommandEvent) { + idMu.Lock() + id := edgeID + tok := token + idMu.Unlock() + if id == "" { + return + } + registry.RecordCommandEventFromConnection(id, tok, event, time.Now()) + }) + proto_socket.AddRequestListenerTyped(&client.Communicator, func(req *iop.EdgeHelloRequest) (*iop.EdgeHelloResponse, error) { now := time.Now() if req.GetEdgeId() == "" { @@ -205,6 +216,52 @@ func (s *EdgeServer) RequestStatus(edgeID string, timeout time.Duration) (*iop.E ) } +// SendCommand sends a command request to a connected Edge over the wire, records +// the request and result in the audit registry, and returns the Edge's typed +// response. It routes only to the target Edge's active connection and fails when +// that Edge is not currently connected. Command lifecycle events the Edge relays +// during execution are recorded asynchronously by the connection's event +// listener. It observes and controls through the Edge; it does not reach Node +// directly. +func (s *EdgeServer) SendCommand(edgeID, operation, targetSelector string, parameters map[string]string, timeout time.Duration) (*iop.EdgeCommandResponse, error) { + client, ok := s.activeClient(edgeID) + if !ok { + return nil, fmt.Errorf("edge %q is not connected", edgeID) + } + now := time.Now() + req := &iop.EdgeCommandRequest{ + RequestId: fmt.Sprintf("edge-command-%d", now.UnixNano()), + CommandId: fmt.Sprintf("cmd-%d", now.UnixNano()), + Operation: operation, + TargetSelector: targetSelector, + Parameters: parameters, + } + s.registry.RecordCommandRequest(edgeID, req, now) + + resp, err := proto_socket.SendRequestTyped[*iop.EdgeCommandRequest, *iop.EdgeCommandResponse]( + &client.Communicator, + req, + timeout, + ) + if err != nil { + s.registry.RecordCommandResult(edgeID, &iop.EdgeCommandResponse{ + CommandId: req.GetCommandId(), + EdgeId: edgeID, + Status: "error", + Error: err.Error(), + }, time.Now()) + return nil, err + } + if resp.GetCommandId() == "" { + resp.CommandId = req.GetCommandId() + } + if resp.GetEdgeId() == "" { + resp.EdgeId = edgeID + } + s.registry.RecordCommandResult(edgeID, resp, time.Now()) + return resp, nil +} + // Host returns the host the server is listening on. func (s *EdgeServer) Host() string { return s.host } diff --git a/apps/control-plane/internal/wire/edge_server_test.go b/apps/control-plane/internal/wire/edge_server_test.go index d4af94c..0acc94e 100644 --- a/apps/control-plane/internal/wire/edge_server_test.go +++ b/apps/control-plane/internal/wire/edge_server_test.go @@ -3,6 +3,7 @@ package wire import ( "context" "fmt" + "sync" "testing" "time" @@ -404,6 +405,182 @@ func TestEdgeServerRecordsNodeLifecycleEventFromEdge(t *testing.T) { } } +type commandEdgeRecorder struct { + mu sync.Mutex + received []string +} + +func (r *commandEdgeRecorder) operations() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.received...) +} + +// dialCommandEdge dials the server as an Edge that answers command requests with +// an "ok" response and records the operations it was asked to run. +func dialCommandEdge(t *testing.T, ctx context.Context, port int, edgeID string) *commandEdgeRecorder { + t.Helper() + client := dialEdge(t, ctx, port) + t.Cleanup(func() { _ = client.Close() }) + rec := &commandEdgeRecorder{} + proto_socket.AddRequestListenerTyped(&client.Communicator, func(req *iop.EdgeCommandRequest) (*iop.EdgeCommandResponse, error) { + rec.mu.Lock() + rec.received = append(rec.received, req.GetOperation()) + rec.mu.Unlock() + return &iop.EdgeCommandResponse{ + RequestId: req.GetRequestId(), + CommandId: req.GetCommandId(), + EdgeId: edgeID, + Status: "ok", + Summary: "applied " + req.GetOperation(), + }, nil + }) + sendEdgeHello(t, client, edgeID) + return rec +} + +func TestEdgeServerSendsCommandToTargetEdgeOnly(t *testing.T) { + server, port := startEdgeServer(t) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + recA := dialCommandEdge(t, ctx, port, "edge-a") + recB := dialCommandEdge(t, ctx, port, "edge-b") + + resp, err := server.SendCommand("edge-a", "restart-node", "node-1", map[string]string{"force": "true"}, 2*time.Second) + if err != nil { + t.Fatalf("SendCommand: %v", err) + } + if resp.GetStatus() != "ok" || resp.GetEdgeId() != "edge-a" { + t.Fatalf("unexpected response: %+v", resp) + } + if resp.GetCommandId() == "" { + t.Error("expected a non-empty command_id correlation") + } + + if got := recA.operations(); len(got) != 1 || got[0] != "restart-node" { + t.Fatalf("edge-a received: %+v", got) + } + if got := recB.operations(); len(got) != 0 { + t.Fatalf("edge-b should not receive a targeted command, got: %+v", got) + } + + cmds := server.Registry().Commands("edge-a") + if len(cmds) != 1 { + t.Fatalf("edge-a audit: got %d want 1", len(cmds)) + } + if cmds[0].Operation != "restart-node" || cmds[0].Status != "ok" { + t.Fatalf("unexpected audit record: %+v", cmds[0]) + } + if got := server.Registry().Commands("edge-b"); len(got) != 0 { + t.Fatalf("edge-b audit should be empty, got %+v", got) + } +} + +func TestEdgeServerSendCommandFailsWhenEdgeNotConnected(t *testing.T) { + server, _ := startEdgeServer(t) + if _, err := server.SendCommand("missing-edge", "restart-node", "", nil, time.Second); err == nil { + t.Fatal("expected error sending command to a non-connected edge") + } + if got := server.Registry().Commands("missing-edge"); len(got) != 0 { + t.Fatalf("expected no audit record for a never-connected edge, got %+v", got) + } +} + +func TestEdgeServerRecordsCommandLifecycleEventsFromEdge(t *testing.T) { + server, port := startEdgeServer(t) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + client := dialEdge(t, ctx, port) + defer client.Close() + const edgeID = "edge-dgx-group" + + proto_socket.AddRequestListenerTyped(&client.Communicator, func(req *iop.EdgeCommandRequest) (*iop.EdgeCommandResponse, error) { + _ = client.Send(&iop.EdgeCommandEvent{CommandId: req.GetCommandId(), EdgeId: edgeID, Phase: "accepted", Summary: "queued"}) + _ = client.Send(&iop.EdgeCommandEvent{CommandId: req.GetCommandId(), EdgeId: edgeID, Phase: "running", Summary: "executing"}) + return &iop.EdgeCommandResponse{ + RequestId: req.GetRequestId(), + CommandId: req.GetCommandId(), + EdgeId: edgeID, + Status: "ok", + Summary: "done", + }, nil + }) + + sendEdgeHello(t, client, edgeID) + + resp, err := server.SendCommand(edgeID, "drain", "", nil, 2*time.Second) + if err != nil { + t.Fatalf("SendCommand: %v", err) + } + + var rec EdgeCommandRecord + deadline := time.Now().Add(2 * time.Second) + for { + got, ok := server.Registry().Command(edgeID, resp.GetCommandId()) + if ok && len(got.Events) == 2 { + rec = got + break + } + if time.Now().After(deadline) { + t.Fatalf("command events not recorded in time; record=%+v", got) + } + time.Sleep(20 * time.Millisecond) + } + if rec.Status != "ok" || rec.Summary != "done" { + t.Fatalf("unexpected final command record: %+v", rec) + } + if rec.Events[0].Phase != "accepted" || rec.Events[1].Phase != "running" { + t.Fatalf("unexpected event phases: %+v", rec.Events) + } +} + +func TestEdgeRegistryCommandsAreBoundedAndCopied(t *testing.T) { + registry := NewEdgeRegistry() + at := time.Unix(1780142500, 0).UTC() + + for i := 0; i < recentCommandCapacity+2; i++ { + cmdID := fmt.Sprintf("cmd-%03d", i) + registry.RecordCommandRequest("edge-a", &iop.EdgeCommandRequest{CommandId: cmdID, Operation: "op"}, at.Add(time.Duration(i)*time.Second)) + registry.RecordCommandResult("edge-a", &iop.EdgeCommandResponse{CommandId: cmdID, Status: "ok", Summary: "done"}, at.Add(time.Duration(i)*time.Second).Add(time.Millisecond)) + } + + cmds := registry.Commands("edge-a") + if len(cmds) != recentCommandCapacity { + t.Fatalf("commands len: got %d want %d", len(cmds), recentCommandCapacity) + } + if cmds[0].CommandID != "cmd-002" { + t.Fatalf("expected oldest retained cmd-002, got %q", cmds[0].CommandID) + } + if cmds[0].Status != "ok" { + t.Fatalf("expected result status recorded, got %q", cmds[0].Status) + } + + registry.RecordCommandRequest("edge-b", &iop.EdgeCommandRequest{CommandId: "cmd-b", Operation: "drain"}, at) + for i := 0; i < commandEventCapacity+3; i++ { + registry.RecordCommandEvent("edge-b", &iop.EdgeCommandEvent{CommandId: "cmd-b", Phase: fmt.Sprintf("phase-%03d", i)}, at.Add(time.Duration(i)*time.Nanosecond)) + } + rec, ok := registry.Command("edge-b", "cmd-b") + if !ok { + t.Fatal("edge-b cmd-b missing") + } + if len(rec.Events) != commandEventCapacity { + t.Fatalf("events len: got %d want %d", len(rec.Events), commandEventCapacity) + } + if rec.Events[0].Phase != "phase-003" { + t.Fatalf("expected oldest retained phase-003, got %q", rec.Events[0].Phase) + } + + rec.Events[0].Phase = "mutated" + again, _ := registry.Command("edge-b", "cmd-b") + if again.Events[0].Phase == "mutated" { + t.Fatal("Command returned mutable registry storage") + } +} + func TestEdgeServerRequestStatusFailsWhenEdgeNotConnected(t *testing.T) { server, _ := startEdgeServer(t) if _, err := server.RequestStatus("missing-edge", time.Second); err == nil {