gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-15 09:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: move nbd endpoint helpers to compute runtime

Browse Source
This commit is contained in:
joshsiv-mulga
2026-10-02 15:40:47 +10:00
parent 14ce3e228f
commit e712c6a7f8
13 changed files with 280 additions and 266 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spinifex/admin/image_import.go
+2 -2
View File
@@ -11,7 +11,7 @@ import (
"strconv"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
)
// ImportOpts describes one raw disk image to import into a new volume.
@@ -143,7 +143,7 @@ func writeImageToNBD(ctx context.Context, sourcePath, nbdURI string, progress io
// qemuNBDTarget renders a provider NBD URI in the form qemu-img accepts. The
// contract's unix form is not qemu's, so it is rebuilt rather than passed on.
func qemuNBDTarget(nbdURI string) (string, error) {
serverType, socketPath, host, port, err := utils.ParseNBDURI(nbdURI)
serverType, socketPath, host, port, err := nbd.ParseNBDURI(nbdURI)
if err != nil {
return "", fmt.Errorf("parse NBD URI %q: %w", nbdURI, err)
}
spinifex/providers/ebs/conformance/nbdclient.go
+4 -4
View File
@@ -17,7 +17,7 @@ import (
"time"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -94,13 +94,13 @@ func assertNBDURI(t *testing.T, nbdURI string) {
// tested rather than every subtest failing on the same string.
func dialURI(t *testing.T, nbdURI string) string {
t.Helper()
serverType, socketPath, host, port, err := utils.ParseNBDURI(nbdURI)
serverType, socketPath, host, port, err := nbd.ParseNBDURI(nbdURI)
require.NoErrorf(t, err, "parse published NBD URI %q", nbdURI)
if serverType == "unix" {
return utils.FormatNBDSocketURI(socketPath)
return nbd.FormatNBDSocketURI(socketPath)
}
return utils.FormatNBDTCPURI(host, port)
return nbd.FormatNBDTCPURI(host, port)
}
// RequireNBDTools skips when libnbd's client tools are absent, so a host
spinifex/utils/nbd.go → spinifex/runtime/compute/nbd/endpoint.go
+13 -1
View File
@@ -1,10 +1,11 @@
package utils
package nbd
import (
"errors"
"fmt"
"net"
"net/url"
"os"
"path/filepath"
"strconv"
"strings"
@@ -147,3 +148,14 @@ func ParseNBDURI(nbdURI string) (serverType, path, host string, port int, err er
return "", "", "", 0, fmt.Errorf("unsupported NBD URI format: %s", nbdURI)
}
func dirExists(path string) bool {
info, err := os.Stat(path)
if os.IsNotExist(err) {
return false
}
if err != nil {
return false
}
return info.IsDir()
}
spinifex/runtime/compute/nbd/endpoint_test.go
+241
View File
@@ -0,0 +1,241 @@
package nbd
import (
"fmt"
"net"
"os"
"path/filepath"
"strconv"
"strings"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestGenerateSocketFile(t *testing.T) {
socketPath := fmt.Sprintf("%s/%s", t.TempDir(), "utilsunittest")
name, err := GenerateSocketFile(socketPath)
assert.NoError(t, err)
assert.True(t, strings.HasSuffix(name, "utilsunittest.sock"))
// Test empty socket path
_, err = GenerateSocketFile("")
assert.Error(t, err)
}
func TestWaitForNBDReady(t *testing.T) {
t.Run("unix socket ready", func(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "nbd.sock")
ln, err := net.Listen("unix", path)
require.NoError(t, err)
t.Cleanup(func() { _ = ln.Close() })
require.NoError(t, WaitForNBDReady(FormatNBDSocketURI(path), time.Second))
})
t.Run("unix socket times out", func(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "missing.sock")
err := WaitForNBDReady(FormatNBDSocketURI(path), 100*time.Millisecond)
require.Error(t, err)
})
t.Run("tcp listener ready", func(t *testing.T) {
ln, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
t.Cleanup(func() { _ = ln.Close() })
_, portStr, err := net.SplitHostPort(ln.Addr().String())
require.NoError(t, err)
port, err := strconv.Atoi(portStr)
require.NoError(t, err)
require.NoError(t, WaitForNBDReady(FormatNBDTCPURI("127.0.0.1", port), time.Second))
})
t.Run("tcp listener times out", func(t *testing.T) {
// Port 1 is unprivileged-bind reserved; nothing will be listening.
err := WaitForNBDReady(FormatNBDTCPURI("127.0.0.1", 1), 100*time.Millisecond)
require.Error(t, err)
})
t.Run("rejects malformed uri", func(t *testing.T) {
err := WaitForNBDReady("garbage://nope", time.Second)
require.Error(t, err)
})
}
func TestParseNBDURI(t *testing.T) {
tests := []struct {
name string
uri string
wantType string
wantPath string
wantHost string
wantPort int
wantErr bool
}{
{
name: "Unix socket URI",
uri: "nbd+unix:///?socket=/run/user/1000/nbd-vol-123.sock",
wantType: "unix",
wantPath: "/run/user/1000/nbd-vol-123.sock",
},
{
name: "Unix socket URI without the empty export path",
uri: "nbd+unix://?socket=/run/nbd-vol.sock",
wantType: "unix",
wantPath: "/run/nbd-vol.sock",
},
{
name: "legacy QEMU nbd:unix: filename form",
uri: "nbd:unix:/run/user/1000/nbd-vol-123.sock",
wantType: "unix",
wantPath: "/run/user/1000/nbd-vol-123.sock",
},
{
name: "TCP address",
uri: "nbd://127.0.0.1:34305",
wantType: "inet",
wantHost: "127.0.0.1",
wantPort: 34305,
},
{
name: "TCP with hostname",
uri: "nbd://storage.local:9000",
wantType: "inet",
wantHost: "storage.local",
wantPort: 9000,
},
{
name: "Empty socket path",
uri: "nbd:unix:",
wantErr: true,
},
{
name: "Unix socket URI with no socket parameter",
uri: "nbd+unix:///",
wantErr: true,
},
{
name: "Missing port in TCP",
uri: "nbd://127.0.0.1",
wantErr: true,
},
{
name: "Invalid port",
uri: "nbd://127.0.0.1:notaport",
wantErr: true,
},
{
name: "Unsupported format",
uri: "http://example.com",
wantErr: true,
},
{
name: "Empty string",
uri: "",
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
serverType, path, host, port, err := ParseNBDURI(tt.uri)
if tt.wantErr {
assert.Error(t, err)
return
}
assert.NoError(t, err)
assert.Equal(t, tt.wantType, serverType)
assert.Equal(t, tt.wantPath, path)
assert.Equal(t, tt.wantHost, host)
assert.Equal(t, tt.wantPort, port)
})
}
}
func TestIsSocketURI(t *testing.T) {
tests := []struct {
name string
uri string
want bool
}{
{"Socket suffix", "/run/nbd-vol.sock", true},
{"Unix prefix", "unix:/run/nbd-vol", true},
{"Both", "unix:/run/nbd-vol.sock", true},
{"TCP URI", "nbd://127.0.0.1:9000", false},
{"Empty", "", false},
{"Random path", "/tmp/somefile.txt", false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert.Equal(t, tt.want, IsSocketURI(tt.uri))
})
}
}
func TestFormatNBDSocketURI(t *testing.T) {
assert.Equal(t, "nbd+unix:///?socket=/run/nbd-vol.sock", FormatNBDSocketURI("/run/nbd-vol.sock"))
assert.Equal(t, "nbd+unix:///?socket=/tmp/test.sock", FormatNBDSocketURI("/tmp/test.sock"))
}
// TestFormatNBDSocketURIRoundTrips guards the pairing that a third-party NBD
// client depends on: what we publish must be what we can parse back.
func TestFormatNBDSocketURIRoundTrips(t *testing.T) {
const path = "/run/spinifex/nbd/nbd-vol-abc123.sock"
serverType, parsed, _, _, err := ParseNBDURI(FormatNBDSocketURI(path))
require.NoError(t, err)
assert.Equal(t, "unix", serverType)
assert.Equal(t, path, parsed)
}
func TestFormatNBDTCPURI(t *testing.T) {
assert.Equal(t, "nbd://127.0.0.1:9000", FormatNBDTCPURI("127.0.0.1", 9000))
assert.Equal(t, "nbd://storage.local:34305", FormatNBDTCPURI("storage.local", 34305))
}
func TestGenerateUniqueSocketFile(t *testing.T) {
path1, err := GenerateUniqueSocketFile("vol-123")
require.NoError(t, err)
assert.Contains(t, path1, "nbd-vol-123-")
assert.True(t, strings.HasSuffix(path1, ".sock"))
// Two calls should produce different paths (different timestamps)
time.Sleep(time.Nanosecond)
path2, err := GenerateUniqueSocketFile("vol-123")
require.NoError(t, err)
assert.NotEqual(t, path1, path2)
// Empty volume name
_, err = GenerateUniqueSocketFile("")
assert.Error(t, err)
}
func TestDirExists(t *testing.T) {
// Existing directory
assert.True(t, dirExists(os.TempDir()))
// Non-existent path
assert.False(t, dirExists("/nonexistent/path/should/not/exist"))
// File (not a directory)
tmpFile, err := os.CreateTemp(t.TempDir(), "direxists-test-*")
require.NoError(t, err)
tmpFile.Close()
assert.False(t, dirExists(tmpFile.Name()))
}
func TestGenerateSocketFile_EmptyName(t *testing.T) {
_, err := GenerateSocketFile("")
assert.Error(t, err)
}
spinifex/runtime/compute/vm/lifecycle.go
+5 -5
View File
@@ -21,9 +21,9 @@ import (
telemetryv1 "github.com/mulgadc/spinifex/contracts/telemetry/v1"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
"github.com/mulgadc/spinifex/spinifex/utils"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
@@ -579,7 +579,7 @@ func (m *Manager) startQEMU(instance *VM) error {
instance.Config.Devices = append(instance.Config.Devices, Device{Value: devSpec})
}
qmpSocket, err := utils.GenerateSocketFile(fmt.Sprintf("qmp-%s", instance.ID))
qmpSocket, err := nbd.GenerateSocketFile(fmt.Sprintf("qmp-%s", instance.ID))
if err != nil {
slog.Error("Failed to generate QMP socket", "err", err)
return err
@@ -598,7 +598,7 @@ func (m *Manager) startQEMU(instance *VM) error {
// Second QMP monitor for the metrics collector; a stale socket from a
// SIGKILLed QEMU is unlinked so the fresh process can bind. Telemetry
// never blocks a launch — failures degrade to no metrics for this VM.
if telemetrySocket, terr := utils.GenerateSocketFile(telemetryv1.QMPTelemetryPrefix + instance.ID); terr != nil {
if telemetrySocket, terr := nbd.GenerateSocketFile(telemetryv1.QMPTelemetryPrefix + instance.ID); terr != nil {
slog.Warn("Failed to generate telemetry QMP socket", "instanceId", instance.ID, "err", terr)
} else {
_ = os.Remove(telemetrySocket)
@@ -615,7 +615,7 @@ func (m *Manager) startQEMU(instance *VM) error {
}
instance.EBSRequests.Mu.Unlock()
for _, ep := range nbdEndpoints {
if err := utils.WaitForNBDReady(ep.uri, nbdReadyTimeout); err != nil {
if err := nbd.WaitForNBDReady(ep.uri, nbdReadyTimeout); err != nil {
return fmt.Errorf("nbd endpoint not ready for %s: %w", ep.name, err)
}
}
@@ -1524,7 +1524,7 @@ func buildDrives(requests []EBSRequest, cpuCount int, machineType string) (drive
v.HotplugPort = port
}
serverType, socketPath, nbdHost, nbdPort, err := utils.ParseNBDURI(v.NBDURI)
serverType, socketPath, nbdHost, nbdPort, err := nbd.ParseNBDURI(v.NBDURI)
if err != nil {
return driveConfig{}, fmt.Errorf("parse NBDURI for volume %s: %w", v.Name, err)
}
spinifex/runtime/compute/vm/restore.go
+2 -2
View File
@@ -13,9 +13,9 @@ import (
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
hostprocess "github.com/mulgadc/spinifex/spinifex/runtime/host/process"
"github.com/mulgadc/spinifex/spinifex/utils"
)
// maxConcurrentRecovery bounds the recovery fan-out; cold-AMI clones are I/O-heavy.
@@ -524,7 +524,7 @@ func AreVolumeSocketsValid(instance *VM) bool {
if req.NBDURI == "" {
continue
}
serverType, sockPath, _, _, err := utils.ParseNBDURI(req.NBDURI)
serverType, sockPath, _, _, err := nbd.ParseNBDURI(req.NBDURI)
if err != nil || serverType != "unix" {
continue
}
spinifex/runtime/compute/vm/volumes.go
+2 -2
View File
@@ -11,8 +11,8 @@ import (
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
"github.com/mulgadc/spinifex/spinifex/utils"
)
// blockdevDelMaxAttempts caps the bounded retry on "node is in use" as a
@@ -119,7 +119,7 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string)
return "", fmt.Errorf("mount volume %s: %w", volumeID, err)
}
serverType, socketPath, nbdHost, nbdPort, err := utils.ParseNBDURI(ebsRequest.NBDURI)
serverType, socketPath, nbdHost, nbdPort, err := nbd.ParseNBDURI(ebsRequest.NBDURI)
if err != nil {
slog.ErrorContext(ctx, "AttachVolume: failed to parse NBDURI", "uri", ebsRequest.NBDURI, "err", err)
_ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest)
spinifex/runtime/roles/qemunbdd/live_lifecycle_test.go
+2 -1
View File
@@ -16,6 +16,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/stretchr/testify/require"
)
@@ -155,7 +156,7 @@ func cleanupBestEffort(t *testing.T, step string, fn func() error) {
// rather than by slicing the string.
func requireUnixSocketPath(t *testing.T, nbdURI string) string {
t.Helper()
serverType, path, _, _, err := utils.ParseNBDURI(nbdURI)
serverType, path, _, _, err := nbd.ParseNBDURI(nbdURI)
require.NoError(t, err, "parse NBD URI %q", nbdURI)
require.Equal(t, "unix", serverType, "published volume must be a nbd:unix: URI, got %q", nbdURI)
require.NotEmpty(t, path, "NBD URI %q has an empty socket path", nbdURI)
spinifex/runtime/roles/qemunbdd/provider.go
+3 -3
View File
@@ -16,7 +16,7 @@ import (
"syscall"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
)
// nbdSharedClients bounds how many simultaneous NBD connections qemu-nbd
@@ -688,7 +688,7 @@ func (p *Provider) PublishVolume(ctx context.Context, req ebsprovider.PublishVol
if pub.nodeID != req.NodeID {
return nil, fmt.Errorf("%w: volume %s is published to %s", ebsprovider.ErrVolumeInUse, req.VolumeID, pub.nodeID)
}
return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: pub.nodeID, NBDURI: utils.FormatNBDSocketURI(pub.socketPath)}, nil
return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: pub.nodeID, NBDURI: nbd.FormatNBDSocketURI(pub.socketPath)}, nil
}
sockPath := p.socketPath(req.VolumeID)
@@ -718,7 +718,7 @@ func (p *Provider) PublishVolume(ctx context.Context, req ebsprovider.PublishVol
pid := readPID(pidPath)
p.published[req.VolumeID] = &publication{nodeID: req.NodeID, socketPath: sockPath, pid: pid}
return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, NBDURI: utils.FormatNBDSocketURI(sockPath)}, nil
return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, NBDURI: nbd.FormatNBDSocketURI(sockPath)}, nil
}
// readPID best-effort reads the PID qemu-nbd's --fork wrote to pidPath. A
spinifex/services/viperblockd/nbdkit_recovery.go
+3 -3
View File
@@ -19,7 +19,7 @@ import (
"time"
viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/nbd"
"github.com/mulgadc/viperblock/viperblock"
"github.com/nats-io/nats.go"
)
@@ -170,9 +170,9 @@ func rebuildMountedVolume(ctx context.Context, cfg *Config, nc *nats.Conn, disc
var nbdURI string
if disc.Socket != "" {
nbdURI = utils.FormatNBDSocketURI(disc.Socket)
nbdURI = nbd.FormatNBDSocketURI(disc.Socket)
} else {
nbdURI = utils.FormatNBDTCPURI("127.0.0.1", disc.Port)
nbdURI = nbd.FormatNBDTCPURI("127.0.0.1", disc.Port)
}
configSub, err := nc.Subscribe(viperblocklegacyv1.VolumeConfigUpdateSubject(disc.Volume), makeConfigUpdateHandler(vb, disc.Volume))
spinifex/services/viperblockd/provider_handlers.go
+3 -3
View File
@@ -1426,17 +1426,17 @@ func mountVolume(ctx context.Context, cfg *Config, nc *nats.Conn, volumeName str
return ebsResponse, err
}
nbdURI = utils.FormatNBDTCPURI("127.0.0.1", nbdPort)
nbdURI = nbd.FormatNBDTCPURI("127.0.0.1", nbdPort)
slog.InfoContext(ctx, "Mounting volume (TCP)", "name", volumeName, "port", nbdPort, "uri", nbdURI)
} else {
// Unix socket transport (default) - generate unique socket path
nbdSocket, err = utils.GenerateUniqueSocketFile(volumeName)
nbdSocket, err = nbd.GenerateUniqueSocketFile(volumeName)
if err != nil {
ebsResponse.Error = err.Error()
return ebsResponse, err
}
nbdURI = utils.FormatNBDSocketURI(nbdSocket)
nbdURI = nbd.FormatNBDSocketURI(nbdSocket)
slog.InfoContext(ctx, "Mounting volume (socket)", "name", volumeName, "socket", nbdSocket, "uri", nbdURI)
}
spinifex/utils/utils.go
-12
View File
@@ -7,7 +7,6 @@ import (
"encoding/json"
"errors"
"log/slog"
"os"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
@@ -73,14 +72,3 @@ func UnmarshalJsonPayload(input any, jsonData []byte) []byte {
return nil
}
func dirExists(path string) bool {
info, err := os.Stat(path)
if os.IsNotExist(err) {
return false
}
if err != nil {
return false
}
return info.IsDir()
}
spinifex/utils/utils_test.go
-228
View File
@@ -6,13 +6,11 @@ import (
"crypto/x509"
"encoding/hex"
"fmt"
"net"
"net/http"
"net/http/httptest"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"sync/atomic"
"testing"
@@ -49,63 +47,6 @@ func TestGenerateResourceID(t *testing.T) {
}
}
func TestGenerateSocketFile(t *testing.T) {
socketPath := fmt.Sprintf("%s/%s", t.TempDir(), "utilsunittest")
name, err := GenerateSocketFile(socketPath)
assert.NoError(t, err)
assert.True(t, strings.HasSuffix(name, "utilsunittest.sock"))
// Test empty socket path
_, err = GenerateSocketFile("")
assert.Error(t, err)
}
func TestWaitForNBDReady(t *testing.T) {
t.Run("unix socket ready", func(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "nbd.sock")
ln, err := net.Listen("unix", path)
require.NoError(t, err)
t.Cleanup(func() { _ = ln.Close() })
require.NoError(t, WaitForNBDReady(FormatNBDSocketURI(path), time.Second))
})
t.Run("unix socket times out", func(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "missing.sock")
err := WaitForNBDReady(FormatNBDSocketURI(path), 100*time.Millisecond)
require.Error(t, err)
})
t.Run("tcp listener ready", func(t *testing.T) {
ln, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
t.Cleanup(func() { _ = ln.Close() })
_, portStr, err := net.SplitHostPort(ln.Addr().String())
require.NoError(t, err)
port, err := strconv.Atoi(portStr)
require.NoError(t, err)
require.NoError(t, WaitForNBDReady(FormatNBDTCPURI("127.0.0.1", port), time.Second))
})
t.Run("tcp listener times out", func(t *testing.T) {
// Port 1 is unprivileged-bind reserved; nothing will be listening.
err := WaitForNBDReady(FormatNBDTCPURI("127.0.0.1", 1), 100*time.Millisecond)
require.Error(t, err)
})
t.Run("rejects malformed uri", func(t *testing.T) {
err := WaitForNBDReady("garbage://nope", time.Second)
require.Error(t, err)
})
}
func TestUnmarshalJsonPayload(t *testing.T) {
type TestStruct struct {
Name string `json:"name"`
@@ -259,98 +200,6 @@ func TestValidateErrorPayload(t *testing.T) {
}
}
func TestParseNBDURI(t *testing.T) {
tests := []struct {
name string
uri string
wantType string
wantPath string
wantHost string
wantPort int
wantErr bool
}{
{
name: "Unix socket URI",
uri: "nbd+unix:///?socket=/run/user/1000/nbd-vol-123.sock",
wantType: "unix",
wantPath: "/run/user/1000/nbd-vol-123.sock",
},
{
name: "Unix socket URI without the empty export path",
uri: "nbd+unix://?socket=/run/nbd-vol.sock",
wantType: "unix",
wantPath: "/run/nbd-vol.sock",
},
{
name: "legacy QEMU nbd:unix: filename form",
uri: "nbd:unix:/run/user/1000/nbd-vol-123.sock",
wantType: "unix",
wantPath: "/run/user/1000/nbd-vol-123.sock",
},
{
name: "TCP address",
uri: "nbd://127.0.0.1:34305",
wantType: "inet",
wantHost: "127.0.0.1",
wantPort: 34305,
},
{
name: "TCP with hostname",
uri: "nbd://storage.local:9000",
wantType: "inet",
wantHost: "storage.local",
wantPort: 9000,
},
{
name: "Empty socket path",
uri: "nbd:unix:",
wantErr: true,
},
{
name: "Unix socket URI with no socket parameter",
uri: "nbd+unix:///",
wantErr: true,
},
{
name: "Missing port in TCP",
uri: "nbd://127.0.0.1",
wantErr: true,
},
{
name: "Invalid port",
uri: "nbd://127.0.0.1:notaport",
wantErr: true,
},
{
name: "Unsupported format",
uri: "http://example.com",
wantErr: true,
},
{
name: "Empty string",
uri: "",
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
serverType, path, host, port, err := ParseNBDURI(tt.uri)
if tt.wantErr {
assert.Error(t, err)
return
}
assert.NoError(t, err)
assert.Equal(t, tt.wantType, serverType)
assert.Equal(t, tt.wantPath, path)
assert.Equal(t, tt.wantHost, host)
assert.Equal(t, tt.wantPort, port)
})
}
}
// Test file extraction process
func TestExtractDiskImageFromFile(t *testing.T) {
@@ -436,83 +285,6 @@ func TestExtractDiskImageFromFile(t *testing.T) {
//assert.NoError(t, err, "Could not remove temp dir")
}
func TestIsSocketURI(t *testing.T) {
tests := []struct {
name string
uri string
want bool
}{
{"Socket suffix", "/run/nbd-vol.sock", true},
{"Unix prefix", "unix:/run/nbd-vol", true},
{"Both", "unix:/run/nbd-vol.sock", true},
{"TCP URI", "nbd://127.0.0.1:9000", false},
{"Empty", "", false},
{"Random path", "/tmp/somefile.txt", false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert.Equal(t, tt.want, IsSocketURI(tt.uri))
})
}
}
func TestFormatNBDSocketURI(t *testing.T) {
assert.Equal(t, "nbd+unix:///?socket=/run/nbd-vol.sock", FormatNBDSocketURI("/run/nbd-vol.sock"))
assert.Equal(t, "nbd+unix:///?socket=/tmp/test.sock", FormatNBDSocketURI("/tmp/test.sock"))
}
// TestFormatNBDSocketURIRoundTrips guards the pairing that a third-party NBD
// client depends on: what we publish must be what we can parse back.
func TestFormatNBDSocketURIRoundTrips(t *testing.T) {
const path = "/run/spinifex/nbd/nbd-vol-abc123.sock"
serverType, parsed, _, _, err := ParseNBDURI(FormatNBDSocketURI(path))
require.NoError(t, err)
assert.Equal(t, "unix", serverType)
assert.Equal(t, path, parsed)
}
func TestFormatNBDTCPURI(t *testing.T) {
assert.Equal(t, "nbd://127.0.0.1:9000", FormatNBDTCPURI("127.0.0.1", 9000))
assert.Equal(t, "nbd://storage.local:34305", FormatNBDTCPURI("storage.local", 34305))
}
func TestGenerateUniqueSocketFile(t *testing.T) {
path1, err := GenerateUniqueSocketFile("vol-123")
require.NoError(t, err)
assert.Contains(t, path1, "nbd-vol-123-")
assert.True(t, strings.HasSuffix(path1, ".sock"))
// Two calls should produce different paths (different timestamps)
time.Sleep(time.Nanosecond)
path2, err := GenerateUniqueSocketFile("vol-123")
require.NoError(t, err)
assert.NotEqual(t, path1, path2)
// Empty volume name
_, err = GenerateUniqueSocketFile("")
assert.Error(t, err)
}
func TestDirExists(t *testing.T) {
// Existing directory
assert.True(t, dirExists(os.TempDir()))
// Non-existent path
assert.False(t, dirExists("/nonexistent/path/should/not/exist"))
// File (not a directory)
tmpFile, err := os.CreateTemp(t.TempDir(), "direxists-test-*")
require.NoError(t, err)
tmpFile.Close()
assert.False(t, dirExists(tmpFile.Name()))
}
func TestGenerateSocketFile_EmptyName(t *testing.T) {
_, err := GenerateSocketFile("")
assert.Error(t, err)
}
func TestExtractDiskImagePath_NoMatch(t *testing.T) {
output := []byte("somefile.txt\nanotherfile.conf\n")
diskimage, err := extractDiskImagePath(t.TempDir(), output)