mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
fix: wait for unix socket on qmp client creation (#235)
This commit is contained in:
@@ -220,6 +220,26 @@ func WaitForPidFile(instanceID string, timeout time.Duration) (int, error) {
|
||||
}
|
||||
}
|
||||
|
||||
// WaitForUnixSocket polls until a Unix socket exists at path or the timeout
|
||||
// expires. QEMU binds its QMP socket after writing the pidfile, so callers
|
||||
// that gate on WaitForPidFile can still race the socket bind under recovery
|
||||
// load. Existence is checked via os.Stat rather than a dial probe so we don't
|
||||
// consume the listener's accept queue before the real client connects.
|
||||
func WaitForUnixSocket(path string, timeout time.Duration) error {
|
||||
deadline := time.Now().Add(timeout)
|
||||
for {
|
||||
if _, err := os.Stat(path); err == nil {
|
||||
return nil
|
||||
} else if !errors.Is(err, os.ErrNotExist) {
|
||||
return err
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
return fmt.Errorf("timeout waiting for unix socket %s", path)
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
func WaitForPidFileRemoval(instanceID string, timeout time.Duration) error {
|
||||
timeoutCh := time.After(timeout)
|
||||
ticker := time.NewTicker(100 * time.Millisecond)
|
||||
|
||||
@@ -185,6 +185,42 @@ func TestWaitForPidFile(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestWaitForUnixSocket(t *testing.T) {
|
||||
t.Run("returns nil when socket appears within timeout", func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "appears.sock")
|
||||
|
||||
ln, err := net.Listen("unix", path)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = ln.Close() })
|
||||
|
||||
require.NoError(t, WaitForUnixSocket(path, time.Second))
|
||||
})
|
||||
|
||||
t.Run("returns error when timeout expires", func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "missing.sock")
|
||||
|
||||
err := WaitForUnixSocket(path, 100*time.Millisecond)
|
||||
require.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("waits for socket created after a delay", func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "delayed.sock")
|
||||
|
||||
go func() {
|
||||
time.Sleep(150 * time.Millisecond)
|
||||
ln, err := net.Listen("unix", path)
|
||||
if err == nil {
|
||||
t.Cleanup(func() { _ = ln.Close() })
|
||||
}
|
||||
}()
|
||||
|
||||
require.NoError(t, WaitForUnixSocket(path, time.Second))
|
||||
})
|
||||
}
|
||||
|
||||
func TestUnmarshalJsonPayload(t *testing.T) {
|
||||
type TestStruct struct {
|
||||
Name string `json:"name"`
|
||||
|
||||
@@ -529,6 +529,14 @@ func (m *Manager) AttachQMP(instance *VM) error {
|
||||
// qmp_capabilities handshake. The caller owns starting the heartbeat
|
||||
// goroutine on the returned client.
|
||||
func newQMPClientWithHandshake(v *VM) (*qmp.QMPClient, error) {
|
||||
// QEMU binds the QMP listening socket after writing its pidfile and parsing
|
||||
// args; under post-reboot recovery load the bind can lag behind startQEMU's
|
||||
// pidfile gate, leaving an "ENOENT on dial" race that tears down the VM
|
||||
// before it finishes coming up. Block briefly so the dial only runs once
|
||||
// the socket inode exists.
|
||||
if err := utils.WaitForUnixSocket(v.Config.QMPSocket, 3*time.Second); err != nil {
|
||||
return nil, fmt.Errorf("connect QMP socket %s: %w", v.Config.QMPSocket, err)
|
||||
}
|
||||
client, err := qmp.NewQMPClient(v.Config.QMPSocket)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("connect QMP socket %s: %w", v.Config.QMPSocket, err)
|
||||
|
||||
@@ -258,13 +258,33 @@ printf '%s' "$KEY_MATERIAL" | put_to_node "$NODE_KEY_PATH"
|
||||
node_ssh "chmod 600 $NODE_KEY_PATH" || { fail "chmod key"; exit 1; }
|
||||
pass "key pair: reboot-e2e-key"
|
||||
|
||||
# Stage user-data on the node (HTTP responder for round-robin verification)
|
||||
# Stage user-data on the node (HTTP responder for round-robin verification).
|
||||
# Installed as a systemd unit so the responder survives guest VM reboots —
|
||||
# nohup-from-cloud-init only runs on first boot (cloud-init's per-instance
|
||||
# semaphore) and would leave the app silent after host_reboot, even though
|
||||
# OVN/HAProxy recover correctly.
|
||||
APP_USER_DATA=$(cat <<'USERDATA'
|
||||
#!/bin/bash
|
||||
set -e
|
||||
INSTANCE_ID=$(hostname)
|
||||
mkdir -p /tmp/httpd && cd /tmp/httpd
|
||||
echo "{\"instance_id\": \"${INSTANCE_ID}\"}" > index.html
|
||||
nohup python3 -m http.server 80 --bind 0.0.0.0 > /dev/null 2>&1 &
|
||||
install -d -m 0755 /var/lib/httpd
|
||||
echo "{\"instance_id\": \"${INSTANCE_ID}\"}" > /var/lib/httpd/index.html
|
||||
cat > /etc/systemd/system/reboot-e2e-httpd.service <<'UNIT'
|
||||
[Unit]
|
||||
Description=Reboot E2E HTTP responder
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
|
||||
[Service]
|
||||
ExecStart=/usr/bin/python3 -m http.server 80 --bind 0.0.0.0 --directory /var/lib/httpd
|
||||
Restart=always
|
||||
RestartSec=1
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
UNIT
|
||||
systemctl daemon-reload
|
||||
systemctl enable --now reboot-e2e-httpd.service
|
||||
USERDATA
|
||||
)
|
||||
printf '%s' "$APP_USER_DATA" | put_to_node "$NODE_USERDATA_PATH"
|
||||
|
||||
Reference in New Issue
Block a user