toki
672b4cdb5d
원격 터미널 브리지 POC 전에 Client HTTP lifecycle과 Edge run result surface 계약을 고정해야 하므로 관련 구현, 테스트, 로드맵 상태를 함께 정리한다.
332 lines
10 KiB
Go
332 lines
10 KiB
Go
package opsconsole
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
"time"
|
|
|
|
"iop/apps/edge/internal/bootstrap"
|
|
edgeservice "iop/apps/edge/internal/service"
|
|
"iop/packages/go/config"
|
|
)
|
|
|
|
// TargetState holds the current console target: node selection, adapter execution
|
|
// target, session, and background mode. It mirrors what a single console session
|
|
// is operating against.
|
|
type TargetState struct {
|
|
NodeRef string
|
|
Adapter string
|
|
Target string
|
|
SessionID string
|
|
Background bool
|
|
TimeoutSec int
|
|
}
|
|
|
|
// Run boots an edge runtime and drives the interactive ops console loop.
|
|
// It is the body of `edge console` and is the only entry point exposed to cmd.
|
|
func Run(ctx context.Context, cfg *config.EdgeConfig, in io.Reader, out io.Writer) error {
|
|
rt, err := bootstrap.NewRuntime(cfg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := rt.Start(ctx); err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = rt.Stop() }()
|
|
|
|
edgeSvc := rt.Service
|
|
eventBus := rt.EventBus
|
|
events := NewEventRouter(out, rt.Registry, rt.Logger)
|
|
|
|
consoleCtx, stopConsoleEvents := context.WithCancel(ctx)
|
|
defer stopConsoleEvents()
|
|
runEvents, unsubscribeRuns := eventBus.SubscribeAllRuns(4096)
|
|
defer unsubscribeRuns()
|
|
nodeEvents, unsubscribeNodes := eventBus.SubscribeAllNodes(64)
|
|
defer unsubscribeNodes()
|
|
go events.Drain(consoleCtx, runEvents, nodeEvents)
|
|
|
|
target := &TargetState{
|
|
Adapter: cfg.Console.Adapter,
|
|
Target: cfg.Console.ResolveTarget(),
|
|
SessionID: NormalizeSessionID(cfg.Console.SessionID),
|
|
Background: cfg.Console.Background,
|
|
TimeoutSec: cfg.Console.TimeoutSec,
|
|
}
|
|
|
|
fmt.Fprintf(out, "IOP Edge console listening on %s\n", cfg.Server.Listen)
|
|
fmt.Fprintf(out, "Console target node=%s adapter=%s target=%s session=%s background=%v\n",
|
|
target.NodeRef, target.Adapter, target.Target, target.SessionID, target.Background)
|
|
fmt.Fprintln(out, "Start node.sh on another host, then type a message here.")
|
|
fmt.Fprintln(out, "Commands: /nodes, /node <id|alias>, /session <id>, /background on|off, /terminate-session, /status, /capabilities, /sessions, /transport, /exit")
|
|
|
|
scanner := bufio.NewScanner(in)
|
|
for {
|
|
fmt.Fprint(out, "edge> ")
|
|
if !scanner.Scan() {
|
|
fmt.Fprintln(out)
|
|
return scanner.Err()
|
|
}
|
|
|
|
cmd := ParseCommand(scanner.Text())
|
|
switch cmd.Kind {
|
|
case CommandEmpty:
|
|
continue
|
|
case CommandExit:
|
|
fmt.Fprintln(out, "bye")
|
|
return nil
|
|
case CommandNodes:
|
|
PrintNodes(out, edgeSvc.ListNodeSnapshots(), target.NodeRef)
|
|
case CommandSetNode:
|
|
if cmd.UsageErr != "" {
|
|
fmt.Fprintln(out, cmd.UsageErr)
|
|
continue
|
|
}
|
|
if _, err := edgeSvc.ResolveNodeSnapshot(cmd.Arg); err != nil {
|
|
fmt.Fprintf(out, "error: %v\n", err)
|
|
continue
|
|
}
|
|
target.NodeRef = cmd.Arg
|
|
fmt.Fprintf(out, "node → %s\n", target.NodeRef)
|
|
case CommandSetSession:
|
|
if cmd.UsageErr != "" {
|
|
fmt.Fprintln(out, cmd.UsageErr)
|
|
continue
|
|
}
|
|
target.SessionID = cmd.Arg
|
|
fmt.Fprintf(out, "session → %s\n", target.SessionID)
|
|
case CommandSetBackground:
|
|
if cmd.UsageErr != "" {
|
|
fmt.Fprintln(out, cmd.UsageErr)
|
|
continue
|
|
}
|
|
target.Background = cmd.Arg == "on"
|
|
if target.Background {
|
|
fmt.Fprintln(out, "background → on")
|
|
} else {
|
|
fmt.Fprintln(out, "background → off")
|
|
}
|
|
case CommandTerminateSession:
|
|
HandleTerminateSession(ctx, edgeSvc, out, target)
|
|
case CommandStatus:
|
|
if err := SendStatus(ctx, edgeSvc, out, target); err != nil {
|
|
fmt.Fprintf(out, "error: %v\n", err)
|
|
}
|
|
case CommandCapabilities:
|
|
if err := SendCapabilities(ctx, edgeSvc, out, target); err != nil {
|
|
fmt.Fprintf(out, "error: %v\n", err)
|
|
}
|
|
case CommandSessionList:
|
|
if err := SendSessionList(ctx, edgeSvc, out, target); err != nil {
|
|
fmt.Fprintf(out, "error: %v\n", err)
|
|
}
|
|
case CommandTransportStatus:
|
|
if err := SendTransportStatus(ctx, edgeSvc, out, target); err != nil {
|
|
fmt.Fprintf(out, "error: %v\n", err)
|
|
}
|
|
case CommandPrompt:
|
|
if err := SendRun(ctx, edgeSvc, events, out, target, cmd.Arg); err != nil {
|
|
fmt.Fprintf(out, "error: %v\n", err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// CommandKind enumerates the console slash commands and the implicit "prompt"
|
|
// command (anything not starting with `/`).
|
|
type CommandKind int
|
|
|
|
const (
|
|
CommandEmpty CommandKind = iota
|
|
CommandExit
|
|
CommandNodes
|
|
CommandSetNode
|
|
CommandSetSession
|
|
CommandSetBackground
|
|
CommandTerminateSession
|
|
CommandStatus
|
|
CommandCapabilities
|
|
CommandSessionList
|
|
CommandTransportStatus
|
|
CommandPrompt
|
|
)
|
|
|
|
// Command is the parsed result of a single console input line.
|
|
type Command struct {
|
|
Kind CommandKind
|
|
Arg string
|
|
UsageErr string
|
|
}
|
|
|
|
func ParseCommand(line string) Command {
|
|
msg := strings.TrimSpace(line)
|
|
if msg == "" {
|
|
return Command{Kind: CommandEmpty}
|
|
}
|
|
lower := strings.ToLower(msg)
|
|
switch {
|
|
case lower == "/exit" || lower == "/quit" || lower == "exit" || lower == "quit":
|
|
return Command{Kind: CommandExit}
|
|
case lower == "/nodes":
|
|
return Command{Kind: CommandNodes}
|
|
case lower == "/terminate-session":
|
|
return Command{Kind: CommandTerminateSession}
|
|
case lower == "/status":
|
|
return Command{Kind: CommandStatus}
|
|
case lower == "/capabilities":
|
|
return Command{Kind: CommandCapabilities}
|
|
case lower == "/sessions":
|
|
return Command{Kind: CommandSessionList}
|
|
case lower == "/transport":
|
|
return Command{Kind: CommandTransportStatus}
|
|
case lower == "/node" || strings.HasPrefix(lower, "/node "):
|
|
parts := strings.Fields(msg)
|
|
if len(parts) < 2 || parts[1] == "" {
|
|
return Command{Kind: CommandSetNode, UsageErr: "usage: /node <id|alias>"}
|
|
}
|
|
return Command{Kind: CommandSetNode, Arg: parts[1]}
|
|
case lower == "/session" || strings.HasPrefix(lower, "/session "):
|
|
parts := strings.Fields(msg)
|
|
if len(parts) < 2 || parts[1] == "" {
|
|
return Command{Kind: CommandSetSession, UsageErr: "usage: /session <id>"}
|
|
}
|
|
return Command{Kind: CommandSetSession, Arg: parts[1]}
|
|
case lower == "/background" || strings.HasPrefix(lower, "/background "):
|
|
parts := strings.Fields(lower)
|
|
if len(parts) < 2 || (parts[1] != "on" && parts[1] != "off") {
|
|
return Command{Kind: CommandSetBackground, UsageErr: "usage: /background on|off"}
|
|
}
|
|
return Command{Kind: CommandSetBackground, Arg: parts[1]}
|
|
default:
|
|
return Command{Kind: CommandPrompt, Arg: msg}
|
|
}
|
|
}
|
|
|
|
func PrintNodes(out io.Writer, nodes []edgeservice.NodeSnapshot, selectedRef string) {
|
|
if len(nodes) == 0 {
|
|
fmt.Fprintln(out, "no nodes connected")
|
|
return
|
|
}
|
|
for _, node := range nodes {
|
|
marker := " "
|
|
if selectedRef != "" && (node.NodeID == selectedRef || node.Alias == selectedRef || node.Label == selectedRef) {
|
|
marker = "* "
|
|
}
|
|
fmt.Fprintf(out, "%s%s = %s (%s)\n", marker, node.Label, node.NodeID, node.Alias)
|
|
}
|
|
}
|
|
|
|
func SendRun(ctx context.Context, edgeSvc *edgeservice.Service, events *EventRouter, out io.Writer, target *TargetState, message string) error {
|
|
runID := edgeservice.NewRunID()
|
|
var untrack func()
|
|
if !target.Background {
|
|
untrack = events.TrackForeground(runID)
|
|
defer untrack()
|
|
}
|
|
|
|
handle, err := edgeSvc.SubmitRun(ctx, edgeservice.SubmitRunRequest{
|
|
NodeRef: target.NodeRef,
|
|
RunID: runID,
|
|
Adapter: target.Adapter,
|
|
Target: target.Target,
|
|
SessionID: target.SessionID,
|
|
Prompt: message,
|
|
Background: target.Background,
|
|
TimeoutSec: target.TimeoutSec,
|
|
Metadata: map[string]string{
|
|
"source": "edge-ops-console",
|
|
},
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer handle.Close()
|
|
|
|
dispatch := handle.Dispatch()
|
|
fmt.Fprintf(out, "[edge] sent run_id=%s node=%s adapter=%s target=%s session=%s background=%v\n",
|
|
dispatch.RunID, dispatch.NodeLabel, dispatch.Adapter, dispatch.Target, dispatch.SessionID, dispatch.Background)
|
|
|
|
if target.Background {
|
|
fmt.Fprintf(out, "[edge] background run dispatched, events will arrive asynchronously\n")
|
|
return nil
|
|
}
|
|
|
|
timer := time.NewTimer(handle.WaitTimeout())
|
|
defer timer.Stop()
|
|
|
|
var response *ResponseStream
|
|
stream := handle.Stream()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-timer.C:
|
|
return fmt.Errorf("timed out waiting for node response")
|
|
case event := <-stream.NodeEvents:
|
|
if edgeservice.IsNodeDisconnected(event) {
|
|
return fmt.Errorf("node disconnected: %s", event.GetReason())
|
|
}
|
|
case event := <-stream.Events:
|
|
if event.GetRunId() != dispatch.RunID {
|
|
continue
|
|
}
|
|
label := events.NodeLabel(event)
|
|
if response == nil {
|
|
response = NewResponseStream(out, nodeMessagePrefix(label))
|
|
}
|
|
|
|
switch event.GetType() {
|
|
case "start":
|
|
fmt.Fprintf(out, "%s start run_id=%s\n", nodeEventPrefix(label), dispatch.RunID)
|
|
case "delta":
|
|
response.Write(event.GetDelta())
|
|
case "complete":
|
|
response.Finish()
|
|
fmt.Fprintf(out, "%s complete run_id=%s detail=%q\n", nodeEventPrefix(label), dispatch.RunID, event.GetMessage())
|
|
return nil
|
|
case "cancelled":
|
|
response.FinishIfStarted()
|
|
fmt.Fprintf(out, "%s cancelled run_id=%s\n", nodeEventPrefix(label), dispatch.RunID)
|
|
return nil
|
|
case "error":
|
|
response.FinishIfStarted()
|
|
fmt.Fprintf(out, "%s error run_id=%s detail=%q\n", nodeEventPrefix(label), dispatch.RunID, event.GetError())
|
|
return fmt.Errorf("node reported error")
|
|
default:
|
|
fmt.Fprintf(out, "%s %s run_id=%s detail=%q\n", nodeEventPrefix(label), event.GetType(), dispatch.RunID, event.GetMessage())
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// SendTerminateSessionFunc is overridable for tests of HandleTerminateSession.
|
|
var SendTerminateSessionFunc = SendTerminateSession
|
|
|
|
func HandleTerminateSession(ctx context.Context, edgeSvc *edgeservice.Service, out io.Writer, target *TargetState) {
|
|
if label, err := SendTerminateSessionFunc(ctx, edgeSvc, target); err != nil {
|
|
fmt.Fprintf(out, "error: %v\n", err)
|
|
} else {
|
|
fmt.Fprintf(out, "terminated session %s node=%s\n", target.SessionID, label)
|
|
}
|
|
}
|
|
|
|
func SendTerminateSession(ctx context.Context, edgeSvc *edgeservice.Service, target *TargetState) (string, error) {
|
|
result, err := edgeSvc.TerminateSession(ctx, edgeservice.TerminateSessionRequest{
|
|
NodeRef: target.NodeRef,
|
|
Adapter: target.Adapter,
|
|
Target: target.Target,
|
|
SessionID: target.SessionID,
|
|
})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return result.NodeLabel, nil
|
|
}
|
|
|
|
func NormalizeSessionID(id string) string {
|
|
return edgeservice.NormalizeSessionID(id)
|
|
}
|