Split internal/daemon vm.go and guest_sessions.go by concern
vm.go (1529 LOC) splits into vm_create, vm_lifecycle, vm_set, vm_stats, vm_disk, vm_authsync; firecracker/DNS/helpers stay in vm.go. guest_sessions.go (1266 LOC) splits into session_controller, session_lifecycle, session_attach, session_stream; scripts and helpers stay in guest_sessions.go. Mechanical move only. No behavior change. Adds doc.go and ARCHITECTURE.md capturing subsystem map and current lock ordering as the baseline for the upcoming subsystem extraction. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
09590cbaa0
commit
ea0db1e17e
14 changed files with 2101 additions and 1846 deletions
|
|
@ -13,14 +13,11 @@ import (
|
|||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"banger/internal/api"
|
||||
"banger/internal/guest"
|
||||
"banger/internal/model"
|
||||
"banger/internal/sessionstream"
|
||||
"banger/internal/system"
|
||||
|
||||
"golang.org/x/crypto/ssh"
|
||||
|
|
@ -80,622 +77,6 @@ func (d *Daemon) waitForGuestSessionReadyHook(ctx context.Context, vm model.VMRe
|
|||
return d.waitForGuestSessionReadyDefault(ctx, vm, session)
|
||||
}
|
||||
|
||||
type guestSessionController struct {
|
||||
stream *guest.StreamSession
|
||||
streams []*guest.StreamSession
|
||||
stdin io.WriteCloser
|
||||
attachMu sync.Mutex
|
||||
attach net.Conn
|
||||
writeMu sync.Mutex
|
||||
closeOnce sync.Once
|
||||
}
|
||||
|
||||
func (c *guestSessionController) setAttach(conn net.Conn) error {
|
||||
c.attachMu.Lock()
|
||||
defer c.attachMu.Unlock()
|
||||
if c.attach != nil {
|
||||
return errors.New("session already has an active attach")
|
||||
}
|
||||
c.attach = conn
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *guestSessionController) clearAttach(conn net.Conn) {
|
||||
c.attachMu.Lock()
|
||||
defer c.attachMu.Unlock()
|
||||
if c.attach == conn {
|
||||
c.attach = nil
|
||||
}
|
||||
}
|
||||
|
||||
func (c *guestSessionController) writeFrame(channel byte, payload []byte) {
|
||||
c.attachMu.Lock()
|
||||
conn := c.attach
|
||||
c.attachMu.Unlock()
|
||||
if conn == nil {
|
||||
return
|
||||
}
|
||||
c.writeMu.Lock()
|
||||
err := sessionstream.WriteFrame(conn, channel, payload)
|
||||
c.writeMu.Unlock()
|
||||
if err != nil {
|
||||
_ = conn.Close()
|
||||
c.clearAttach(conn)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *guestSessionController) writeControl(message sessionstream.ControlMessage) {
|
||||
c.attachMu.Lock()
|
||||
conn := c.attach
|
||||
c.attachMu.Unlock()
|
||||
if conn == nil {
|
||||
return
|
||||
}
|
||||
c.writeMu.Lock()
|
||||
err := sessionstream.WriteControl(conn, message)
|
||||
c.writeMu.Unlock()
|
||||
if err != nil {
|
||||
_ = conn.Close()
|
||||
c.clearAttach(conn)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *guestSessionController) close() error {
|
||||
if c == nil {
|
||||
return nil
|
||||
}
|
||||
var err error
|
||||
c.closeOnce.Do(func() {
|
||||
c.attachMu.Lock()
|
||||
conn := c.attach
|
||||
c.attach = nil
|
||||
c.attachMu.Unlock()
|
||||
if conn != nil {
|
||||
err = errors.Join(err, conn.Close())
|
||||
}
|
||||
if c.stdin != nil {
|
||||
err = errors.Join(err, c.stdin.Close())
|
||||
}
|
||||
if c.stream != nil {
|
||||
err = errors.Join(err, c.stream.Close())
|
||||
}
|
||||
for _, stream := range c.streams {
|
||||
if stream != nil {
|
||||
err = errors.Join(err, stream.Close())
|
||||
}
|
||||
}
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
type guestSessionStateSnapshot struct {
|
||||
Status string
|
||||
GuestPID int
|
||||
ExitCode *int
|
||||
Alive bool
|
||||
LastError string
|
||||
}
|
||||
|
||||
func (d *Daemon) StartGuestSession(ctx context.Context, params api.GuestSessionStartParams) (model.GuestSession, error) {
|
||||
stdinMode := model.GuestSessionStdinMode(strings.TrimSpace(params.StdinMode))
|
||||
if stdinMode == "" {
|
||||
stdinMode = model.GuestSessionStdinClosed
|
||||
}
|
||||
if stdinMode != model.GuestSessionStdinClosed && stdinMode != model.GuestSessionStdinPipe {
|
||||
return model.GuestSession{}, fmt.Errorf("unsupported stdin mode %q", params.StdinMode)
|
||||
}
|
||||
if strings.TrimSpace(params.Command) == "" {
|
||||
return model.GuestSession{}, errors.New("session command is required")
|
||||
}
|
||||
var created model.GuestSession
|
||||
_, err := d.withVMLockByRef(ctx, params.VMIDOrName, func(vm model.VMRecord) (model.VMRecord, error) {
|
||||
if vm.State != model.VMStateRunning || !system.ProcessRunning(vm.Runtime.PID, vm.Runtime.APISockPath) {
|
||||
return model.VMRecord{}, fmt.Errorf("vm %q is not running", vm.Name)
|
||||
}
|
||||
session, err := d.startGuestSessionLocked(ctx, vm, params, stdinMode)
|
||||
if err != nil {
|
||||
return model.VMRecord{}, err
|
||||
}
|
||||
created = session
|
||||
return vm, nil
|
||||
})
|
||||
return created, err
|
||||
}
|
||||
|
||||
func (d *Daemon) startGuestSessionLocked(ctx context.Context, vm model.VMRecord, params api.GuestSessionStartParams, stdinMode model.GuestSessionStdinMode) (model.GuestSession, error) {
|
||||
id, err := model.NewID()
|
||||
if err != nil {
|
||||
return model.GuestSession{}, err
|
||||
}
|
||||
now := model.Now()
|
||||
session := model.GuestSession{
|
||||
ID: id,
|
||||
VMID: vm.ID,
|
||||
Name: defaultGuestSessionName(id, params.Command, params.Name),
|
||||
Backend: guestSessionBackendSSH,
|
||||
Command: params.Command,
|
||||
Args: append([]string(nil), params.Args...),
|
||||
CWD: strings.TrimSpace(params.CWD),
|
||||
Env: cloneStringMap(params.Env),
|
||||
StdinMode: stdinMode,
|
||||
Status: model.GuestSessionStatusStarting,
|
||||
GuestStateDir: guestSessionStateDir(id),
|
||||
StdoutLogPath: guestSessionStdoutLogPath(id),
|
||||
StderrLogPath: guestSessionStderrLogPath(id),
|
||||
Tags: cloneStringMap(params.Tags),
|
||||
Attachable: stdinMode == model.GuestSessionStdinPipe,
|
||||
Reattachable: stdinMode == model.GuestSessionStdinPipe,
|
||||
CreatedAt: now,
|
||||
UpdatedAt: now,
|
||||
}
|
||||
if session.Attachable {
|
||||
session.AttachBackend = guestSessionAttachBackendSSHBridge
|
||||
session.AttachMode = guestSessionAttachModeExclusive
|
||||
} else {
|
||||
session.AttachBackend = guestSessionAttachBackendNone
|
||||
}
|
||||
if err := d.store.UpsertGuestSession(ctx, session); err != nil {
|
||||
return model.GuestSession{}, err
|
||||
}
|
||||
fail := func(stage, message, rawLog string) (model.GuestSession, error) {
|
||||
session = failGuestSessionLaunch(session, stage, message, rawLog)
|
||||
if err := d.store.UpsertGuestSession(ctx, session); err != nil {
|
||||
return model.GuestSession{}, err
|
||||
}
|
||||
return session, nil
|
||||
}
|
||||
address := net.JoinHostPort(vm.Runtime.GuestIP, "22")
|
||||
if err := d.waitForGuestSSH(ctx, address, 250*time.Millisecond); err != nil {
|
||||
return fail("ssh_unavailable", fmt.Sprintf("guest ssh unavailable: %v", err), "")
|
||||
}
|
||||
client, err := d.dialGuest(ctx, address)
|
||||
if err != nil {
|
||||
return fail("dial_guest", fmt.Sprintf("dial guest ssh: %v", err), "")
|
||||
}
|
||||
defer client.Close()
|
||||
var preflightLog bytes.Buffer
|
||||
if err := client.RunScript(ctx, guestSessionCWDPreflightScript(session.CWD), &preflightLog); err != nil {
|
||||
return fail("preflight_cwd", fmt.Sprintf("guest working directory is unavailable: %s", defaultGuestSessionCWD(session.CWD)), preflightLog.String())
|
||||
}
|
||||
preflightLog.Reset()
|
||||
requiredCommands := normalizeGuestSessionRequiredCommands(params.Command, params.RequiredCommands)
|
||||
if err := client.RunScript(ctx, guestSessionCommandPreflightScript(requiredCommands), &preflightLog); err != nil {
|
||||
return fail("preflight_command", fmt.Sprintf("required guest command is unavailable: %s", strings.TrimSpace(preflightLog.String())), preflightLog.String())
|
||||
}
|
||||
var uploadLog bytes.Buffer
|
||||
if err := client.UploadFile(ctx, guestSessionScriptPath(id), 0o755, []byte(guestSessionScript(session)), &uploadLog); err != nil {
|
||||
return fail("upload_script", "upload guest session script failed", uploadLog.String())
|
||||
}
|
||||
var launchLog bytes.Buffer
|
||||
launchScript := fmt.Sprintf("set -euo pipefail\nnohup bash %s >/dev/null 2>&1 </dev/null &\ndisown || true\n", guestShellQuote(guestSessionScriptPath(id)))
|
||||
if err := client.RunScript(ctx, launchScript, &launchLog); err != nil {
|
||||
return fail("launch", "launch guest session failed", launchLog.String())
|
||||
}
|
||||
readyCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
updated, err := d.waitForGuestSessionReadyHook(readyCtx, vm, session)
|
||||
if err != nil {
|
||||
return fail("ready_wait", "guest session did not report ready state", err.Error())
|
||||
}
|
||||
session = updated
|
||||
if session.Status == model.GuestSessionStatusStarting {
|
||||
session.Status = model.GuestSessionStatusRunning
|
||||
session.StartedAt = model.Now()
|
||||
session.UpdatedAt = model.Now()
|
||||
}
|
||||
session.LaunchStage = ""
|
||||
session.LaunchMessage = ""
|
||||
session.LaunchRawLog = ""
|
||||
session.LastError = ""
|
||||
if err := d.store.UpsertGuestSession(ctx, session); err != nil {
|
||||
return model.GuestSession{}, err
|
||||
}
|
||||
return session, nil
|
||||
}
|
||||
|
||||
func (d *Daemon) GetGuestSession(ctx context.Context, params api.GuestSessionRefParams) (model.GuestSession, error) {
|
||||
vm, err := d.FindVM(ctx, params.VMIDOrName)
|
||||
if err != nil {
|
||||
return model.GuestSession{}, err
|
||||
}
|
||||
session, err := d.findGuestSession(ctx, vm.ID, params.SessionIDOrName)
|
||||
if err != nil {
|
||||
return model.GuestSession{}, err
|
||||
}
|
||||
return d.refreshGuestSession(ctx, vm, session)
|
||||
}
|
||||
|
||||
func (d *Daemon) ListGuestSessions(ctx context.Context, params api.VMRefParams) ([]model.GuestSession, error) {
|
||||
vm, err := d.FindVM(ctx, params.IDOrName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sessions, err := d.store.ListGuestSessionsByVM(ctx, vm.ID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for index := range sessions {
|
||||
refreshed, refreshErr := d.refreshGuestSession(ctx, vm, sessions[index])
|
||||
if refreshErr == nil {
|
||||
sessions[index] = refreshed
|
||||
}
|
||||
}
|
||||
return sessions, nil
|
||||
}
|
||||
|
||||
func (d *Daemon) StopGuestSession(ctx context.Context, params api.GuestSessionRefParams) (model.GuestSession, error) {
|
||||
return d.signalGuestSession(ctx, params, "TERM")
|
||||
}
|
||||
|
||||
func (d *Daemon) KillGuestSession(ctx context.Context, params api.GuestSessionRefParams) (model.GuestSession, error) {
|
||||
return d.signalGuestSession(ctx, params, "KILL")
|
||||
}
|
||||
|
||||
func (d *Daemon) signalGuestSession(ctx context.Context, params api.GuestSessionRefParams, signal string) (model.GuestSession, error) {
|
||||
vm, err := d.FindVM(ctx, params.VMIDOrName)
|
||||
if err != nil {
|
||||
return model.GuestSession{}, err
|
||||
}
|
||||
session, err := d.findGuestSession(ctx, vm.ID, params.SessionIDOrName)
|
||||
if err != nil {
|
||||
return model.GuestSession{}, err
|
||||
}
|
||||
session, _ = d.refreshGuestSession(ctx, vm, session)
|
||||
if session.Status == model.GuestSessionStatusExited || session.Status == model.GuestSessionStatusFailed {
|
||||
return session, nil
|
||||
}
|
||||
if vm.State != model.VMStateRunning || !system.ProcessRunning(vm.Runtime.PID, vm.Runtime.APISockPath) {
|
||||
session.Status = model.GuestSessionStatusFailed
|
||||
session.LastError = "vm is not running"
|
||||
now := model.Now()
|
||||
session.UpdatedAt = now
|
||||
session.EndedAt = now
|
||||
session.Attachable = false
|
||||
if err := d.store.UpsertGuestSession(ctx, session); err != nil {
|
||||
return model.GuestSession{}, err
|
||||
}
|
||||
return session, nil
|
||||
}
|
||||
client, err := guest.Dial(ctx, net.JoinHostPort(vm.Runtime.GuestIP, "22"), d.config.SSHKeyPath)
|
||||
if err != nil {
|
||||
return model.GuestSession{}, err
|
||||
}
|
||||
defer client.Close()
|
||||
var log bytes.Buffer
|
||||
if err := client.RunScript(ctx, guestSessionSignalScript(session.ID, signal), &log); err != nil {
|
||||
return model.GuestSession{}, formatGuestSessionStepError("signal guest session", err, log.String())
|
||||
}
|
||||
session.Status = model.GuestSessionStatusStopping
|
||||
session.UpdatedAt = model.Now()
|
||||
if err := d.store.UpsertGuestSession(ctx, session); err != nil {
|
||||
return model.GuestSession{}, err
|
||||
}
|
||||
return session, nil
|
||||
}
|
||||
|
||||
func (d *Daemon) GuestSessionLogs(ctx context.Context, params api.GuestSessionLogsParams) (api.GuestSessionLogsResult, error) {
|
||||
vm, err := d.FindVM(ctx, params.VMIDOrName)
|
||||
if err != nil {
|
||||
return api.GuestSessionLogsResult{}, err
|
||||
}
|
||||
session, err := d.findGuestSession(ctx, vm.ID, params.SessionIDOrName)
|
||||
if err != nil {
|
||||
return api.GuestSessionLogsResult{}, err
|
||||
}
|
||||
streamName := strings.TrimSpace(params.Stream)
|
||||
if streamName == "" {
|
||||
streamName = "stdout"
|
||||
}
|
||||
tailLines := params.TailLines
|
||||
if tailLines <= 0 {
|
||||
tailLines = guestSessionLogTailLine
|
||||
}
|
||||
path := session.StdoutLogPath
|
||||
if streamName == "stderr" {
|
||||
path = session.StderrLogPath
|
||||
}
|
||||
content, err := d.readGuestSessionLog(ctx, vm, session, streamName, tailLines)
|
||||
if err != nil {
|
||||
return api.GuestSessionLogsResult{}, err
|
||||
}
|
||||
return api.GuestSessionLogsResult{Session: session, Stream: streamName, Path: path, Content: content}, nil
|
||||
}
|
||||
|
||||
func (d *Daemon) SendToGuestSession(ctx context.Context, params api.GuestSessionSendParams) (api.GuestSessionSendResult, error) {
|
||||
vm, err := d.FindVM(ctx, params.VMIDOrName)
|
||||
if err != nil {
|
||||
return api.GuestSessionSendResult{}, err
|
||||
}
|
||||
session, err := d.findGuestSession(ctx, vm.ID, params.SessionIDOrName)
|
||||
if err != nil {
|
||||
return api.GuestSessionSendResult{}, err
|
||||
}
|
||||
if session.StdinMode != model.GuestSessionStdinPipe {
|
||||
return api.GuestSessionSendResult{}, errors.New("session does not have a stdin pipe")
|
||||
}
|
||||
if session.Status != model.GuestSessionStatusRunning {
|
||||
return api.GuestSessionSendResult{}, fmt.Errorf("session is not running (status=%s)", session.Status)
|
||||
}
|
||||
if vm.State != model.VMStateRunning || !system.ProcessRunning(vm.Runtime.PID, vm.Runtime.APISockPath) {
|
||||
return api.GuestSessionSendResult{}, fmt.Errorf("vm %q is not running", vm.Name)
|
||||
}
|
||||
if len(params.Payload) == 0 {
|
||||
return api.GuestSessionSendResult{Session: session}, nil
|
||||
}
|
||||
client, err := d.dialGuest(ctx, net.JoinHostPort(vm.Runtime.GuestIP, "22"))
|
||||
if err != nil {
|
||||
return api.GuestSessionSendResult{}, fmt.Errorf("dial guest: %w", err)
|
||||
}
|
||||
defer client.Close()
|
||||
tmpPath := fmt.Sprintf("/tmp/banger-send-%s.bin", session.ID[:8])
|
||||
var uploadLog bytes.Buffer
|
||||
if err := client.UploadFile(ctx, tmpPath, 0o600, params.Payload, &uploadLog); err != nil {
|
||||
return api.GuestSessionSendResult{}, fmt.Errorf("upload payload: %w", err)
|
||||
}
|
||||
sendScript := fmt.Sprintf(
|
||||
"set -euo pipefail\ncat %s >> %s\nrm -f %s\n",
|
||||
guestShellQuote(tmpPath),
|
||||
guestShellQuote(guestSessionStdinPipePath(session.ID)),
|
||||
guestShellQuote(tmpPath),
|
||||
)
|
||||
var sendLog bytes.Buffer
|
||||
if err := client.RunScript(ctx, sendScript, &sendLog); err != nil {
|
||||
return api.GuestSessionSendResult{}, fmt.Errorf("send to session: %w: %s", err, strings.TrimSpace(sendLog.String()))
|
||||
}
|
||||
return api.GuestSessionSendResult{Session: session, BytesWritten: len(params.Payload)}, nil
|
||||
}
|
||||
|
||||
func (d *Daemon) BeginGuestSessionAttach(ctx context.Context, params api.GuestSessionAttachBeginParams) (api.GuestSessionAttachBeginResult, error) {
|
||||
vm, err := d.FindVM(ctx, params.VMIDOrName)
|
||||
if err != nil {
|
||||
return api.GuestSessionAttachBeginResult{}, err
|
||||
}
|
||||
session, err := d.findGuestSession(ctx, vm.ID, params.SessionIDOrName)
|
||||
if err != nil {
|
||||
return api.GuestSessionAttachBeginResult{}, err
|
||||
}
|
||||
session, _ = d.refreshGuestSession(ctx, vm, session)
|
||||
if !session.Attachable {
|
||||
return api.GuestSessionAttachBeginResult{}, errors.New("session is not attachable")
|
||||
}
|
||||
controller := &guestSessionController{}
|
||||
if !d.claimGuestSessionController(session.ID, controller) {
|
||||
return api.GuestSessionAttachBeginResult{}, errors.New("session already has an active attach")
|
||||
}
|
||||
attachID, err := model.NewID()
|
||||
if err != nil {
|
||||
d.clearGuestSessionController(session.ID)
|
||||
return api.GuestSessionAttachBeginResult{}, err
|
||||
}
|
||||
socketPath := filepath.Join(d.layout.RuntimeDir, "guest-session-attach-"+attachID[:12]+".sock")
|
||||
_ = os.Remove(socketPath)
|
||||
listener, err := net.Listen("unix", socketPath)
|
||||
if err != nil {
|
||||
d.clearGuestSessionController(session.ID)
|
||||
return api.GuestSessionAttachBeginResult{}, err
|
||||
}
|
||||
if err := os.Chmod(socketPath, 0o600); err != nil {
|
||||
_ = listener.Close()
|
||||
_ = os.Remove(socketPath)
|
||||
d.clearGuestSessionController(session.ID)
|
||||
return api.GuestSessionAttachBeginResult{}, err
|
||||
}
|
||||
go d.serveGuestSessionAttach(session, controller, attachID, socketPath, listener)
|
||||
return api.GuestSessionAttachBeginResult{
|
||||
Session: session,
|
||||
AttachID: attachID,
|
||||
TransportKind: guestSessionTransportUnixSocket,
|
||||
TransportTarget: socketPath,
|
||||
SocketPath: socketPath,
|
||||
StreamFormat: sessionstream.FormatV1,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (d *Daemon) setGuestSessionController(id string, controller *guestSessionController) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
d.sessionControllers[id] = controller
|
||||
}
|
||||
|
||||
func (d *Daemon) claimGuestSessionController(id string, controller *guestSessionController) bool {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
if d.sessionControllers[id] != nil {
|
||||
return false
|
||||
}
|
||||
d.sessionControllers[id] = controller
|
||||
return true
|
||||
}
|
||||
|
||||
func (d *Daemon) getGuestSessionController(id string) *guestSessionController {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
return d.sessionControllers[id]
|
||||
}
|
||||
|
||||
func (d *Daemon) clearGuestSessionController(id string) *guestSessionController {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
controller := d.sessionControllers[id]
|
||||
delete(d.sessionControllers, id)
|
||||
return controller
|
||||
}
|
||||
|
||||
func (d *Daemon) closeGuestSessionControllers() error {
|
||||
d.mu.Lock()
|
||||
controllers := make([]*guestSessionController, 0, len(d.sessionControllers))
|
||||
for _, controller := range d.sessionControllers {
|
||||
controllers = append(controllers, controller)
|
||||
}
|
||||
d.sessionControllers = nil
|
||||
d.mu.Unlock()
|
||||
var err error
|
||||
for _, controller := range controllers {
|
||||
err = errors.Join(err, controller.close())
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (d *Daemon) forwardGuestSessionOutput(_ string, controller *guestSessionController, channel byte, reader io.Reader) {
|
||||
buffer := make([]byte, 32*1024)
|
||||
for {
|
||||
n, err := reader.Read(buffer)
|
||||
if n > 0 {
|
||||
controller.writeFrame(channel, buffer[:n])
|
||||
}
|
||||
if err != nil {
|
||||
if !errors.Is(err, io.EOF) {
|
||||
controller.writeControl(sessionstream.ControlMessage{Type: "error", Error: err.Error()})
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Daemon) waitForGuestSessionExit(id string, controller *guestSessionController, session model.GuestSession) {
|
||||
err := controller.stream.Wait()
|
||||
updated := session
|
||||
updated.Attachable = false
|
||||
now := model.Now()
|
||||
updated.UpdatedAt = now
|
||||
updated.EndedAt = now
|
||||
if exitCode, ok := guestSessionExitCode(err); ok {
|
||||
updated.ExitCode = &exitCode
|
||||
if exitCode == 0 {
|
||||
updated.Status = model.GuestSessionStatusExited
|
||||
} else {
|
||||
updated.Status = model.GuestSessionStatusFailed
|
||||
}
|
||||
}
|
||||
if err != nil && updated.LastError == "" {
|
||||
updated.LastError = err.Error()
|
||||
}
|
||||
if vm, getErr := d.store.GetVMByID(context.Background(), updated.VMID); getErr == nil {
|
||||
if refreshed, refreshErr := d.refreshGuestSession(context.Background(), vm, updated); refreshErr == nil {
|
||||
updated = refreshed
|
||||
}
|
||||
}
|
||||
_ = d.store.UpsertGuestSession(context.Background(), updated)
|
||||
controller.writeControl(sessionstream.ControlMessage{Type: "exit", ExitCode: updated.ExitCode})
|
||||
_ = controller.close()
|
||||
d.clearGuestSessionController(id)
|
||||
}
|
||||
|
||||
func (d *Daemon) serveGuestSessionAttach(session model.GuestSession, controller *guestSessionController, _ string, socketPath string, listener net.Listener) {
|
||||
defer func() {
|
||||
_ = listener.Close()
|
||||
_ = os.Remove(socketPath)
|
||||
_ = controller.close()
|
||||
d.clearGuestSessionController(session.ID)
|
||||
}()
|
||||
conn, err := listener.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer conn.Close()
|
||||
if err := controller.setAttach(conn); err != nil {
|
||||
_ = sessionstream.WriteControl(conn, sessionstream.ControlMessage{Type: "error", Error: err.Error()})
|
||||
return
|
||||
}
|
||||
defer controller.clearAttach(conn)
|
||||
if err := d.attachGuestSessionBridge(session, controller); err != nil {
|
||||
_ = sessionstream.WriteControl(conn, sessionstream.ControlMessage{Type: "error", Error: err.Error()})
|
||||
return
|
||||
}
|
||||
for {
|
||||
channel, payload, err := sessionstream.ReadFrame(conn)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
switch channel {
|
||||
case sessionstream.ChannelStdin:
|
||||
if controller.stdin == nil {
|
||||
continue
|
||||
}
|
||||
if _, err := controller.stdin.Write(payload); err != nil {
|
||||
_ = sessionstream.WriteControl(conn, sessionstream.ControlMessage{Type: "error", Error: err.Error()})
|
||||
return
|
||||
}
|
||||
case sessionstream.ChannelControl:
|
||||
message, err := sessionstream.ReadControl(payload)
|
||||
if err != nil {
|
||||
_ = sessionstream.WriteControl(conn, sessionstream.ControlMessage{Type: "error", Error: err.Error()})
|
||||
return
|
||||
}
|
||||
if message.Type == "eof" && controller.stdin != nil {
|
||||
_ = controller.stdin.Close()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Daemon) attachGuestSessionBridge(session model.GuestSession, controller *guestSessionController) error {
|
||||
vm, err := d.store.GetVMByID(context.Background(), session.VMID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if vm.State != model.VMStateRunning || !system.ProcessRunning(vm.Runtime.PID, vm.Runtime.APISockPath) {
|
||||
return fmt.Errorf("vm %q is not running", vm.Name)
|
||||
}
|
||||
address := net.JoinHostPort(vm.Runtime.GuestIP, "22")
|
||||
stdinStream, err := d.openGuestSessionAttachStream(address, guestSessionAttachInputCommand(session.ID))
|
||||
if err != nil {
|
||||
return fmt.Errorf("open guest session stdin stream: %w", err)
|
||||
}
|
||||
stdoutStream, err := d.openGuestSessionAttachStream(address, guestSessionAttachTailCommand(session.StdoutLogPath))
|
||||
if err != nil {
|
||||
_ = stdinStream.Close()
|
||||
return fmt.Errorf("open guest session stdout stream: %w", err)
|
||||
}
|
||||
stderrStream, err := d.openGuestSessionAttachStream(address, guestSessionAttachTailCommand(session.StderrLogPath))
|
||||
if err != nil {
|
||||
_ = stdinStream.Close()
|
||||
_ = stdoutStream.Close()
|
||||
return fmt.Errorf("open guest session stderr stream: %w", err)
|
||||
}
|
||||
controller.streams = append(controller.streams, stdinStream, stdoutStream, stderrStream)
|
||||
controller.stdin = stdinStream.Stdin()
|
||||
go d.forwardGuestSessionOutput(session.ID, controller, sessionstream.ChannelStdout, stdoutStream.Stdout())
|
||||
go d.forwardGuestSessionOutput(session.ID, controller, sessionstream.ChannelStderr, stderrStream.Stdout())
|
||||
go d.watchGuestSessionAttach(session.ID, controller, session)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *Daemon) openGuestSessionAttachStream(address, command string) (*guest.StreamSession, error) {
|
||||
client, err := guest.Dial(context.Background(), address, d.config.SSHKeyPath)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stream, err := client.StartCommand(context.Background(), command)
|
||||
if err != nil {
|
||||
_ = client.Close()
|
||||
return nil, err
|
||||
}
|
||||
return stream, nil
|
||||
}
|
||||
|
||||
func (d *Daemon) watchGuestSessionAttach(id string, controller *guestSessionController, session model.GuestSession) {
|
||||
ticker := time.NewTicker(250 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
vm, err := d.store.GetVMByID(context.Background(), session.VMID)
|
||||
if err != nil {
|
||||
controller.writeControl(sessionstream.ControlMessage{Type: "error", Error: err.Error()})
|
||||
_ = controller.close()
|
||||
return
|
||||
}
|
||||
refreshed, err := d.refreshGuestSession(context.Background(), vm, session)
|
||||
if err == nil {
|
||||
session = refreshed
|
||||
}
|
||||
if session.Status == model.GuestSessionStatusExited || session.Status == model.GuestSessionStatusFailed {
|
||||
controller.writeControl(sessionstream.ControlMessage{Type: "exit", ExitCode: session.ExitCode})
|
||||
_ = controller.close()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Daemon) waitForGuestSessionReadyDefault(ctx context.Context, vm model.VMRecord, session model.GuestSession) (model.GuestSession, error) {
|
||||
for {
|
||||
updated, err := d.refreshGuestSession(ctx, vm, session)
|
||||
|
|
@ -846,37 +227,6 @@ func inspectGuestSessionStateFromDir(stateDir string) (guestSessionStateSnapshot
|
|||
return snapshot, nil
|
||||
}
|
||||
|
||||
func (d *Daemon) readGuestSessionLog(ctx context.Context, vm model.VMRecord, session model.GuestSession, stream string, tailLines int) (string, error) {
|
||||
if vm.State == model.VMStateRunning && system.ProcessRunning(vm.Runtime.PID, vm.Runtime.APISockPath) {
|
||||
client, err := guest.Dial(ctx, net.JoinHostPort(vm.Runtime.GuestIP, "22"), d.config.SSHKeyPath)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer client.Close()
|
||||
path := session.StdoutLogPath
|
||||
if stream == "stderr" {
|
||||
path = session.StderrLogPath
|
||||
}
|
||||
var output bytes.Buffer
|
||||
script := fmt.Sprintf("set -euo pipefail\nif [ -f %s ]; then tail -n %d %s; fi\n", guestShellQuote(path), tailLines, guestShellQuote(path))
|
||||
if err := client.RunScript(ctx, script, &output); err != nil {
|
||||
return "", formatGuestSessionStepError("read guest session log", err, output.String())
|
||||
}
|
||||
return output.String(), nil
|
||||
}
|
||||
runner := d.runner
|
||||
if runner == nil {
|
||||
runner = system.NewRunner()
|
||||
}
|
||||
workMount, cleanup, err := system.MountTempDir(ctx, runner, vm.Runtime.WorkDiskPath, false)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer cleanup()
|
||||
logPath := filepath.Join(workMount, guestSessionRelativeStateDir(session.ID), stream+".log")
|
||||
return tailFileContent(logPath, tailLines)
|
||||
}
|
||||
|
||||
func (d *Daemon) findGuestSession(ctx context.Context, vmID, idOrName string) (model.GuestSession, error) {
|
||||
if strings.TrimSpace(idOrName) == "" {
|
||||
return model.GuestSession{}, errors.New("session id or name is required")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue