VGA console: one console operation per VM, a data websocket per SPICE channel
spice-html5 opens a websocket per channel (main, display, inputs, cursor, playback), and each one was making Ham Lab open a new console operation, which Incus refuses as already connected. Now the first channel opens the operation and holds its control websocket; every further channel is one more data websocket on the same operation, which Incus allows; the operation is let go when the last channel closes. A stale operation from a previous Ham Lab process is taken over with force. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5.1
parent
a5ebfdc9ce
commit
80a5b4b464
+2
-13
@@ -535,24 +535,14 @@ func (s *Server) virtVGA(w http.ResponseWriter, r *http.Request) {
|
|||||||
conn.SetReadLimit(64 << 20)
|
conn.SetReadLimit(64 << 20)
|
||||||
ctx, cancel := context.WithCancel(r.Context())
|
ctx, cancel := context.WithCancel(r.Context())
|
||||||
defer cancel()
|
defer cancel()
|
||||||
display, control, err := s.virt.VGA(ctx, name)
|
display, release, err := s.virt.VGAChannel(ctx, name)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("virt: vga %s refused: %v (from %s)", name, err, clientIP(r))
|
log.Printf("virt: vga %s refused: %v (from %s)", name, err, clientIP(r))
|
||||||
conn.Close(websocket.StatusInternalError, truncate(err.Error(), 120))
|
conn.Close(websocket.StatusInternalError, truncate(err.Error(), 120))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
log.Printf("virt: vga console of %s opened (from %s)", name, clientIP(r))
|
defer release()
|
||||||
defer display.Close(websocket.StatusNormalClosure, "")
|
defer display.Close(websocket.StatusNormalClosure, "")
|
||||||
if control != nil {
|
|
||||||
defer control.Close(websocket.StatusNormalClosure, "")
|
|
||||||
go func() { // the control channel carries nothing we send; keep it open and drain it
|
|
||||||
for {
|
|
||||||
if _, _, err := control.Read(ctx); err != nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
go func() {
|
go func() {
|
||||||
defer cancel()
|
defer cancel()
|
||||||
for {
|
for {
|
||||||
@@ -576,7 +566,6 @@ func (s *Server) virtVGA(w http.ResponseWriter, r *http.Request) {
|
|||||||
}
|
}
|
||||||
cancel()
|
cancel()
|
||||||
conn.Close(websocket.StatusNormalClosure, "")
|
conn.Close(websocket.StatusNormalClosure, "")
|
||||||
log.Printf("virt: vga console of %s closed", name)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func truncate(s string, n int) string {
|
func truncate(s string, n int) string {
|
||||||
|
|||||||
@@ -58,6 +58,7 @@ type Manager struct {
|
|||||||
|
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
cpuPrev map[string]map[string]cpuMark // caller → instance → last reading, so two readers do not spoil each other's deltas
|
cpuPrev map[string]map[string]cpuMark // caller → instance → last reading, so two readers do not spoil each other's deltas
|
||||||
|
vga map[string]*vgaSession // instance → the console operation its browser channels share
|
||||||
remote struct {
|
remote struct {
|
||||||
at time.Time
|
at time.Time
|
||||||
list []Image
|
list []Image
|
||||||
|
|||||||
+102
-26
@@ -14,6 +14,7 @@ import (
|
|||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
"os"
|
"os"
|
||||||
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/coder/websocket"
|
"github.com/coder/websocket"
|
||||||
@@ -44,28 +45,98 @@ func (m *Manager) apiClient() *http.Client {
|
|||||||
}}
|
}}
|
||||||
}
|
}
|
||||||
|
|
||||||
// VGA opens the display and control websockets for a running VM. The caller closes both.
|
// A SPICE client opens one connection per channel — main, display, inputs, cursor, playback
|
||||||
func (m *Manager) VGA(ctx context.Context, name string) (display, control *websocket.Conn, err error) {
|
// — and spice-html5 does the same with websockets, all to one URL. Incus allows exactly one
|
||||||
|
// console operation per VM at a time, but takes any number of data websockets on it. So a VGA
|
||||||
|
// session is one operation, held for as long as any channel is open, and every browser
|
||||||
|
// websocket becomes one more data websocket on it.
|
||||||
|
type vgaSession struct {
|
||||||
|
op string
|
||||||
|
secret string
|
||||||
|
control *websocket.Conn
|
||||||
|
cancel context.CancelFunc
|
||||||
|
refs int
|
||||||
|
}
|
||||||
|
|
||||||
|
// VGAChannel returns a fresh data websocket on the VM's console operation, opening the
|
||||||
|
// operation first when none is held. release must be called when the channel is done.
|
||||||
|
func (m *Manager) VGAChannel(ctx context.Context, name string) (data *websocket.Conn, release func(), err error) {
|
||||||
if !ValidName(name) {
|
if !ValidName(name) {
|
||||||
return nil, nil, errors.New("bad name")
|
return nil, nil, errors.New("bad name")
|
||||||
}
|
}
|
||||||
client := m.apiClient()
|
client := m.apiClient()
|
||||||
body, _ := json.Marshal(map[string]any{"type": "vga", "width": 0, "height": 0})
|
m.mu.Lock()
|
||||||
|
defer m.mu.Unlock()
|
||||||
|
if m.vga == nil {
|
||||||
|
m.vga = map[string]*vgaSession{}
|
||||||
|
}
|
||||||
|
dial := func(s *vgaSession) (*websocket.Conn, error) {
|
||||||
|
c, cancel := context.WithTimeout(ctx, 20*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
u := fmt.Sprintf("ws://incus/1.0/operations/%s/websocket?secret=%s", s.op, s.secret)
|
||||||
|
ws, _, err := websocket.Dial(c, u, &websocket.DialOptions{HTTPClient: client, CompressionMode: websocket.CompressionDisabled})
|
||||||
|
return ws, err
|
||||||
|
}
|
||||||
|
s := m.vga[name]
|
||||||
|
if s != nil {
|
||||||
|
if ws, err := dial(s); err == nil {
|
||||||
|
s.refs++
|
||||||
|
ws.SetReadLimit(64 << 20)
|
||||||
|
return ws, m.releaser(name, s), nil
|
||||||
|
}
|
||||||
|
// the operation is gone (the VM restarted, or Incus did): start over
|
||||||
|
s.cancel()
|
||||||
|
delete(m.vga, name)
|
||||||
|
}
|
||||||
|
s, err = m.openVGA(ctx, client, name, false)
|
||||||
|
if err != nil && strings.Contains(err.Error(), "already connected") {
|
||||||
|
// a session a previous Ham Lab process held and never let go of; take it over
|
||||||
|
s, err = m.openVGA(ctx, client, name, true)
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return nil, nil, err
|
||||||
|
}
|
||||||
|
ws, err := dial(s)
|
||||||
|
if err != nil {
|
||||||
|
s.cancel()
|
||||||
|
return nil, nil, err
|
||||||
|
}
|
||||||
|
ws.SetReadLimit(64 << 20)
|
||||||
|
s.refs = 1
|
||||||
|
m.vga[name] = s
|
||||||
|
return ws, m.releaser(name, s), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *Manager) releaser(name string, s *vgaSession) func() {
|
||||||
|
return func() {
|
||||||
|
m.mu.Lock()
|
||||||
|
defer m.mu.Unlock()
|
||||||
|
s.refs--
|
||||||
|
if s.refs <= 0 && m.vga[name] == s {
|
||||||
|
s.cancel()
|
||||||
|
delete(m.vga, name)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// openVGA asks Incus for the console operation and holds its control websocket, which is
|
||||||
|
// what keeps the operation alive.
|
||||||
|
func (m *Manager) openVGA(ctx context.Context, client *http.Client, name string, force bool) (*vgaSession, error) {
|
||||||
|
body, _ := json.Marshal(map[string]any{"type": "vga", "width": 0, "height": 0, "force": force})
|
||||||
c, cancel := context.WithTimeout(ctx, 20*time.Second)
|
c, cancel := context.WithTimeout(ctx, 20*time.Second)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
req, _ := http.NewRequestWithContext(c, "POST", "http://incus/1.0/instances/"+name+"/console", bytes.NewReader(body))
|
req, _ := http.NewRequestWithContext(c, "POST", "http://incus/1.0/instances/"+name+"/console", bytes.NewReader(body))
|
||||||
req.Header.Set("Content-Type", "application/json")
|
req.Header.Set("Content-Type", "application/json")
|
||||||
resp, err := client.Do(req)
|
resp, err := client.Do(req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if os.IsPermission(errors.Unwrap(err)) || isPermission(err) {
|
if isPermission(err) {
|
||||||
return nil, nil, errors.New("Ham Lab's account can't reach Incus's socket; Finish setup on the Virtualization tab puts it in incus-admin")
|
return nil, errors.New("Ham Lab's account can't reach Incus's socket; Finish setup on the Virtualization tab puts it in incus-admin")
|
||||||
}
|
}
|
||||||
return nil, nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
defer resp.Body.Close()
|
defer resp.Body.Close()
|
||||||
var op struct {
|
var op struct {
|
||||||
Error string `json:"error"`
|
Error string `json:"error"`
|
||||||
Operation string `json:"operation"`
|
|
||||||
Metadata struct {
|
Metadata struct {
|
||||||
ID string `json:"id"`
|
ID string `json:"id"`
|
||||||
Metadata struct {
|
Metadata struct {
|
||||||
@@ -74,33 +145,38 @@ func (m *Manager) VGA(ctx context.Context, name string) (display, control *webso
|
|||||||
} `json:"metadata"`
|
} `json:"metadata"`
|
||||||
}
|
}
|
||||||
if err := json.NewDecoder(resp.Body).Decode(&op); err != nil {
|
if err := json.NewDecoder(resp.Body).Decode(&op); err != nil {
|
||||||
return nil, nil, errors.New("unreadable answer from incus")
|
return nil, errors.New("unreadable answer from incus")
|
||||||
}
|
}
|
||||||
if op.Error != "" {
|
if op.Error != "" {
|
||||||
return nil, nil, errors.New(op.Error)
|
return nil, errors.New(op.Error)
|
||||||
}
|
}
|
||||||
data, control0 := op.Metadata.Metadata.FDs["0"], op.Metadata.Metadata.FDs["control"]
|
secret, controlSecret := op.Metadata.Metadata.FDs["0"], op.Metadata.Metadata.FDs["control"]
|
||||||
if op.Metadata.ID == "" || data == "" {
|
if op.Metadata.ID == "" || secret == "" {
|
||||||
return nil, nil, errors.New("incus gave no console; is the VM running?")
|
return nil, errors.New("incus gave no console; is the VM running?")
|
||||||
}
|
}
|
||||||
dial := func(secret string) (*websocket.Conn, error) {
|
s := &vgaSession{op: op.Metadata.ID, secret: secret}
|
||||||
u := fmt.Sprintf("ws://incus/1.0/operations/%s/websocket?secret=%s", op.Metadata.ID, secret)
|
// the control websocket lives outside any one request: it is the session
|
||||||
ws, _, err := websocket.Dial(ctx, u, &websocket.DialOptions{HTTPClient: client, CompressionMode: websocket.CompressionDisabled})
|
sctx, scancel := context.WithCancel(context.Background())
|
||||||
return ws, err
|
s.cancel = scancel
|
||||||
|
if controlSecret != "" {
|
||||||
|
u := fmt.Sprintf("ws://incus/1.0/operations/%s/websocket?secret=%s", s.op, controlSecret)
|
||||||
|
control, _, err := websocket.Dial(sctx, u, &websocket.DialOptions{HTTPClient: client, CompressionMode: websocket.CompressionDisabled})
|
||||||
|
if err != nil {
|
||||||
|
scancel()
|
||||||
|
return nil, err
|
||||||
}
|
}
|
||||||
if control0 != "" {
|
s.control = control
|
||||||
if control, err = dial(control0); err != nil {
|
go func() { // drain it; when it ends, so has the operation
|
||||||
return nil, nil, err
|
for {
|
||||||
|
if _, _, err := control.Read(sctx); err != nil {
|
||||||
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if display, err = dial(data); err != nil {
|
control.Close(websocket.StatusNormalClosure, "")
|
||||||
if control != nil {
|
}()
|
||||||
control.Close(websocket.StatusInternalError, "")
|
go func() { <-sctx.Done(); control.Close(websocket.StatusNormalClosure, "") }()
|
||||||
}
|
}
|
||||||
return nil, nil, err
|
return s, nil
|
||||||
}
|
|
||||||
display.SetReadLimit(64 << 20) // a SPICE frame can carry a whole screen
|
|
||||||
return display, control, nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func isPermission(err error) bool {
|
func isPermission(err error) bool {
|
||||||
|
|||||||
Reference in New Issue
Block a user