diff --git a/spinifex/admin/image_import.go b/spinifex/admin/image_import.go index 8a732309d..58db2286d 100644 --- a/spinifex/admin/image_import.go +++ b/spinifex/admin/image_import.go @@ -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) } diff --git a/spinifex/providers/ebs/conformance/nbdclient.go b/spinifex/providers/ebs/conformance/nbdclient.go index 92e371066..efc47de7e 100644 --- a/spinifex/providers/ebs/conformance/nbdclient.go +++ b/spinifex/providers/ebs/conformance/nbdclient.go @@ -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 diff --git a/spinifex/utils/nbd.go b/spinifex/runtime/compute/nbd/endpoint.go similarity index 96% rename from spinifex/utils/nbd.go rename to spinifex/runtime/compute/nbd/endpoint.go index 77aa1a2bc..bb097adaa 100644 --- a/spinifex/utils/nbd.go +++ b/spinifex/runtime/compute/nbd/endpoint.go @@ -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() +} diff --git a/spinifex/runtime/compute/nbd/endpoint_test.go b/spinifex/runtime/compute/nbd/endpoint_test.go new file mode 100644 index 000000000..0e408e02c --- /dev/null +++ b/spinifex/runtime/compute/nbd/endpoint_test.go @@ -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) +} diff --git a/spinifex/runtime/compute/vm/lifecycle.go b/spinifex/runtime/compute/vm/lifecycle.go index 69ad0d8ec..fba446dfb 100644 --- a/spinifex/runtime/compute/vm/lifecycle.go +++ b/spinifex/runtime/compute/vm/lifecycle.go @@ -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) } diff --git a/spinifex/runtime/compute/vm/restore.go b/spinifex/runtime/compute/vm/restore.go index c5886fc3b..9f64fd5cc 100644 --- a/spinifex/runtime/compute/vm/restore.go +++ b/spinifex/runtime/compute/vm/restore.go @@ -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 } diff --git a/spinifex/runtime/compute/vm/volumes.go b/spinifex/runtime/compute/vm/volumes.go index 404ea53fe..40b101eca 100644 --- a/spinifex/runtime/compute/vm/volumes.go +++ b/spinifex/runtime/compute/vm/volumes.go @@ -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) diff --git a/spinifex/runtime/roles/qemunbdd/live_lifecycle_test.go b/spinifex/runtime/roles/qemunbdd/live_lifecycle_test.go index 583e33f3d..d2dfd6200 100644 --- a/spinifex/runtime/roles/qemunbdd/live_lifecycle_test.go +++ b/spinifex/runtime/roles/qemunbdd/live_lifecycle_test.go @@ -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) diff --git a/spinifex/runtime/roles/qemunbdd/provider.go b/spinifex/runtime/roles/qemunbdd/provider.go index 0a1f1a996..44d49ae3f 100644 --- a/spinifex/runtime/roles/qemunbdd/provider.go +++ b/spinifex/runtime/roles/qemunbdd/provider.go @@ -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 diff --git a/spinifex/services/viperblockd/nbdkit_recovery.go b/spinifex/services/viperblockd/nbdkit_recovery.go index 4fbda8fec..21195039a 100644 --- a/spinifex/services/viperblockd/nbdkit_recovery.go +++ b/spinifex/services/viperblockd/nbdkit_recovery.go @@ -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)) diff --git a/spinifex/services/viperblockd/provider_handlers.go b/spinifex/services/viperblockd/provider_handlers.go index 10a1ec723..2083d51f8 100644 --- a/spinifex/services/viperblockd/provider_handlers.go +++ b/spinifex/services/viperblockd/provider_handlers.go @@ -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) } diff --git a/spinifex/utils/utils.go b/spinifex/utils/utils.go index 9ccc43359..e9c19351a 100644 --- a/spinifex/utils/utils.go +++ b/spinifex/utils/utils.go @@ -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() -} diff --git a/spinifex/utils/utils_test.go b/spinifex/utils/utils_test.go index 7743adf52..1bb2520bc 100644 --- a/spinifex/utils/utils_test.go +++ b/spinifex/utils/utils_test.go @@ -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)