Compare commits

...

3 Commits

Author SHA1 Message Date
brynmailer-mdc ab93d1fdcb Merge branch 'feat/ebs-export-descriptor-compute' into feat/ebs-export-descriptor 2026-10-23 14:51:37 +10:00
brynmailer-mdc 3a79c1b866 feat(ebsprovider): replace PublishedVolume.NBDURI with a typed export descriptor
PublishedVolume now carries an Export of kind nbd or block, checked by VolumeExport.Validate.
The NATS client rejects an invalid export in a PublishVolume reply, and the server refuses to relay one, reporting it as internal.
SchemaVersion moves to 2.

viperblockd, qemunbdd and the memory provider publish nbd exports exactly as before.
Image import refuses a non-nbd export and still unpublishes.
Conformance requires a valid export, and the NBD-dialling suites fail explicitly on a non-nbd export.
2026-10-23 14:51:08 +10:00
brynmailer-mdc 8eefbc405f feat(vm): attach host block device exports via raw over host_device
EBSRequest gains optional BlockDevice and BlockSizeBytes. A single
classifier decides whether a request is an NBD export, a block export or
invalid, and buildDrives, AttachVolume, launch readiness, the QMP greeting
deadline and restore all use it. A block data volume is attached as a
sized raw node over a host_device file node with the same node name,
iothread and virtio-blk device as an NBD volume; boot and EFI volumes
reject block exports. NBD output is unchanged.
2026-10-23 14:47:17 +10:00
30 changed files with 753 additions and 88 deletions
@@ -48,6 +48,11 @@ type EBSRequest struct {
NBDURI string `json:"NBDURI"` NBDURI string `json:"NBDURI"`
DeviceName string `json:"DeviceName"` DeviceName string `json:"DeviceName"`
HotplugPort int `json:"HotplugPort,omitempty"` HotplugPort int `json:"HotplugPort,omitempty"`
// BlockDevice and BlockSizeBytes describe a host block device export, the
// alternative to NBDURI; the guest sees BlockSizeBytes of it. Exactly one
// of NBDURI and BlockDevice may be set.
BlockDevice string `json:"BlockDevice,omitempty"`
BlockSizeBytes uint64 `json:"BlockSizeBytes,omitempty"`
} }
// EBSMountResponse is the response to an EBSRequest on MountSubject. // EBSMountResponse is the response to an EBSRequest on MountSubject.
@@ -35,6 +35,18 @@ func TestEBSRequestJSONContract(t *testing.T) {
"NBDURI":"nbd:unix:/run/spinifex/vol-123.sock", "NBDURI":"nbd:unix:/run/spinifex/vol-123.sock",
"DeviceName":"/dev/sdf", "HotplugPort":3 "DeviceName":"/dev/sdf", "HotplugPort":3
}`, string(data)) }`, string(data))
block, err := json.Marshal(EBSRequest{
Name: "vol-456",
BlockDevice: "/dev/sdx",
BlockSizeBytes: 10 << 30,
})
require.NoError(t, err)
require.JSONEq(t, `{
"Name":"vol-456", "VolType":"", "Boot":false, "EFI":false,
"DeleteOnTermination":false, "NBDURI":"", "DeviceName":"",
"BlockDevice":"/dev/sdx", "BlockSizeBytes":10737418240
}`, string(block))
} }
func TestVolumeResponseJSONContracts(t *testing.T) { func TestVolumeResponseJSONContracts(t *testing.T) {
+6 -1
View File
@@ -71,7 +71,12 @@ func ImportImage(ctx context.Context, provider ebsprovider.EBSProvider, opts Imp
return fmt.Errorf("import: publish volume %s: %w", opts.VolumeID, err) return fmt.Errorf("import: publish volume %s: %w", opts.VolumeID, err)
} }
writeErr := writeImage(ctx, opts.SourcePath, published.NBDURI, opts.Progress) var writeErr error
if published.Export.Kind == ebsprovider.ExportKindNBD && published.Export.NBD != nil {
writeErr = writeImage(ctx, opts.SourcePath, published.Export.NBD.NBDURI, opts.Progress)
} else {
writeErr = fmt.Errorf("volume is published as a %q export; image import writes only through nbd", published.Export.Kind)
}
// Unpublish regardless: a volume left published cannot be snapshotted or // Unpublish regardless: a volume left published cannot be snapshotted or
// attached, and on the failure path it would strand the export too. // attached, and on the failure path it would strand the export too.
+35 -1
View File
@@ -22,6 +22,10 @@ type recordingProvider struct {
publishes []ebsprovider.PublishVolumeRequest publishes []ebsprovider.PublishVolumeRequest
snapshots []ebsprovider.CreateSnapshotRequest snapshots []ebsprovider.CreateSnapshotRequest
publishURI string publishURI string
// publishExport, when its Kind is set, replaces the nbd export built
// from publishURI.
publishExport ebsprovider.VolumeExport
} }
func newRecordingProvider() *recordingProvider { func newRecordingProvider() *recordingProvider {
@@ -39,7 +43,11 @@ func (p *recordingProvider) PublishVolume(ctx context.Context, req ebsprovider.P
if _, err := p.MemoryProvider.PublishVolume(ctx, req); err != nil { if _, err := p.MemoryProvider.PublishVolume(ctx, req); err != nil {
return nil, err return nil, err
} }
return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, NBDURI: p.publishURI}, nil export := ebsprovider.NewNBDExport(p.publishURI)
if p.publishExport.Kind != "" {
export = p.publishExport
}
return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: export}, nil
} }
func (p *recordingProvider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) error { func (p *recordingProvider) UnpublishVolume(ctx context.Context, req ebsprovider.UnpublishVolumeRequest) error {
@@ -117,6 +125,32 @@ func TestImportImage_WriteFailureStillUnpublishes(t *testing.T) {
"a failed copy must not be snapshotted, and must still be unpublished") "a failed copy must not be snapshotted, and must still be unpublished")
} }
// TestImportImage_BlockExportIsRefused covers a provider that publishes a host
// block device: the import writes only through NBD, so it must fail without
// copying and still release the export.
func TestImportImage_BlockExportIsRefused(t *testing.T) {
var wroteTo string
stubWriteImage(t, nil, &wroteTo)
provider := newRecordingProvider()
provider.publishExport = ebsprovider.VolumeExport{
Kind: ebsprovider.ExportKindBlock,
Block: &ebsprovider.BlockExport{DevicePath: "/dev/sdx", SizeBytes: 1 << 30},
}
err := ImportImage(context.Background(), provider, ImportOpts{
VolumeID: "ami-import005",
NodeID: "node-a",
SizeBytes: 1 << 30,
SourcePath: "/dev/null",
Snapshot: true,
})
require.Error(t, err)
assert.Contains(t, err.Error(), `"block" export`)
assert.Empty(t, wroteTo, "a block export must not reach the NBD writer")
assert.Equal(t, []string{"create", "publish", "unpublish"}, provider.calls,
"a refused export must not be snapshotted, and must still be unpublished")
}
// TestImportImage_NoSnapshotWhenNotAsked covers the weights-style import that // TestImportImage_NoSnapshotWhenNotAsked covers the weights-style import that
// wants the volume but not an AMI snapshot. // wants the volume but not an AMI snapshot.
func TestImportImage_NoSnapshotWhenNotAsked(t *testing.T) { func TestImportImage_NoSnapshotWhenNotAsked(t *testing.T) {
@@ -791,7 +791,10 @@ func RunSuiteWithConfig(t *testing.T, newRawProvider func(t *testing.T) ebsprovi
require.NotNil(t, pub) require.NotNil(t, pub)
assert.Equal(t, cfg.id("vol-pub-ok"), pub.VolumeID) assert.Equal(t, cfg.id("vol-pub-ok"), pub.VolumeID)
assert.Equal(t, cfg.node(), pub.NodeID) assert.Equal(t, cfg.node(), pub.NodeID)
assertNBDURI(t, pub.NBDURI) require.NoError(t, pub.Export.Validate(), "published export is not attachable")
if pub.Export.Kind == ebsprovider.ExportKindNBD {
assertNBDURI(t, pub.Export.NBD.NBDURI)
}
}) })
t.Run("idempotent republish to the same node", func(t *testing.T) { t.Run("idempotent republish to the same node", func(t *testing.T) {
@@ -167,9 +167,9 @@ func RunDataPathSuite(t *testing.T, newProvider func(t *testing.T) ebsprovider.E
provider := newProvider(t) provider := newProvider(t)
volumeID := cfg.VolumePrefix + "datapath" volumeID := cfg.VolumePrefix + "datapath"
pub := publishForNBD(t, provider, cfg, volumeID, false) nbdURI := publishForNBD(t, provider, cfg, volumeID, false)
assertNBDURI(t, pub.NBDURI) assertNBDURI(t, nbdURI)
uri := dialURI(t, pub.NBDURI) uri := dialURI(t, nbdURI)
export := nbdInfo(t, uri) export := nbdInfo(t, uri)
require.Equalf(t, cfg.VolumeBytes, export.Size, "export size does not match the volume that was created") require.Equalf(t, cfg.VolumeBytes, export.Size, "export size does not match the volume that was created")
+30 -20
View File
@@ -177,9 +177,9 @@ func nbdReachable(uri string) bool {
return exec.CommandContext(context.Background(), "nbdinfo", "--size", uri).Run() == nil return exec.CommandContext(context.Background(), "nbdinfo", "--size", uri).Run() == nil
} }
// publishForNBD creates and publishes a volume, registering cleanup so a run // publishForNBD creates and publishes a volume and returns its NBD URI,
// against a real cluster does not leave volumes behind. // registering cleanup so a run against a real cluster leaves nothing behind.
func publishForNBD(t *testing.T, provider ebsprovider.EBSProvider, cfg NBDClientConfig, volumeID string, readOnly bool) *ebsprovider.PublishedVolume { func publishForNBD(t *testing.T, provider ebsprovider.EBSProvider, cfg NBDClientConfig, volumeID string, readOnly bool) string {
t.Helper() t.Helper()
ctx := context.Background() ctx := context.Background()
_, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{ _, err := provider.CreateVolume(ctx, ebsprovider.CreateVolumeRequest{
@@ -198,7 +198,17 @@ func publishForNBD(t *testing.T, provider ebsprovider.EBSProvider, cfg NBDClient
}) })
require.NoError(t, err) require.NoError(t, err)
require.NotNil(t, pub) require.NotNil(t, pub)
return pub return publishedNBDURI(t, pub)
}
// publishedNBDURI fails the test outright when a provider publishes anything
// but an nbd export: the NBD-dialling suites have nothing to dial otherwise.
func publishedNBDURI(t *testing.T, pub *ebsprovider.PublishedVolume) string {
t.Helper()
require.NoErrorf(t, pub.Export.Validate(), "published export for %s is invalid", pub.VolumeID)
require.Equalf(t, ebsprovider.ExportKindNBD, pub.Export.Kind,
"published export for %s is %q; this suite drives only nbd exports", pub.VolumeID, pub.Export.Kind)
return pub.Export.NBD.NBDURI
} }
// cleanupVolume unpublishes and deletes best-effort: cleanup runs after the // cleanupVolume unpublishes and deletes best-effort: cleanup runs after the
@@ -234,21 +244,21 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
t.Run("published URI is a standard NBD URI", func(t *testing.T) { t.Run("published URI is a standard NBD URI", func(t *testing.T) {
provider := newProvider(t) provider := newProvider(t)
pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-uri", false) nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-uri", false)
assertNBDURI(t, pub.NBDURI) assertNBDURI(t, nbdURI)
}) })
t.Run("export is connectable by a standard NBD client", func(t *testing.T) { t.Run("export is connectable by a standard NBD client", func(t *testing.T) {
provider := newProvider(t) provider := newProvider(t)
pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-connect", false) nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-connect", false)
export := nbdInfo(t, dialURI(t, pub.NBDURI)) export := nbdInfo(t, dialURI(t, nbdURI))
assert.Equal(t, cfg.VolumeBytes, export.Size) assert.Equal(t, cfg.VolumeBytes, export.Size)
}) })
t.Run("export advertises usable block size constraints", func(t *testing.T) { t.Run("export advertises usable block size constraints", func(t *testing.T) {
provider := newProvider(t) provider := newProvider(t)
pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-blocksize", false) nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-blocksize", false)
export := nbdInfo(t, dialURI(t, pub.NBDURI)) export := nbdInfo(t, dialURI(t, nbdURI))
require.Positivef(t, export.BlockSizeMinimum, require.Positivef(t, export.BlockSizeMinimum,
"export advertises no minimum block size, so a client cannot know how to align its requests") "export advertises no minimum block size, so a client cannot know how to align its requests")
@@ -261,8 +271,8 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
// export. Publication is what holds the volume, not any one connection. // export. Publication is what holds the volume, not any one connection.
t.Run("export accepts a client reconnecting after the first disconnects", func(t *testing.T) { t.Run("export accepts a client reconnecting after the first disconnects", func(t *testing.T) {
provider := newProvider(t) provider := newProvider(t)
pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-reconnect", false) nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-reconnect", false)
uri := dialURI(t, pub.NBDURI) uri := dialURI(t, nbdURI)
first := nbdInfo(t, uri) first := nbdInfo(t, uri)
@@ -275,7 +285,7 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
t.Run("data written by an external client reads back byte for byte", func(t *testing.T) { t.Run("data written by an external client reads back byte for byte", func(t *testing.T) {
provider := newProvider(t) provider := newProvider(t)
pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-roundtrip", false) nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-roundtrip", false)
dir := t.TempDir() dir := t.TempDir()
in := filepath.Join(dir, "in.bin") in := filepath.Join(dir, "in.bin")
@@ -283,7 +293,7 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
want := patternBytes(int(cfg.VolumeBytes)) want := patternBytes(int(cfg.VolumeBytes))
require.NoError(t, os.WriteFile(in, want, 0o600)) require.NoError(t, os.WriteFile(in, want, 0o600))
uri := dialURI(t, pub.NBDURI) uri := dialURI(t, nbdURI)
runNBDCopy(t, in, uri) runNBDCopy(t, in, uri)
runNBDCopy(t, uri, out) runNBDCopy(t, uri, out)
@@ -301,9 +311,9 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
t.Run("sparse extent reporting matches the advertised capability", func(t *testing.T) { t.Run("sparse extent reporting matches the advertised capability", func(t *testing.T) {
provider := newProvider(t) provider := newProvider(t)
capabilities := capabilitiesOf(t, provider) capabilities := capabilitiesOf(t, provider)
pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-extents", false) nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-extents", false)
holes := slices.ContainsFunc(nbdMap(t, dialURI(t, pub.NBDURI)), func(extent nbdExtent) bool { holes := slices.ContainsFunc(nbdMap(t, dialURI(t, nbdURI)), func(extent nbdExtent) bool {
return strings.Contains(extent.Description, "hole") return strings.Contains(extent.Description, "hole")
}) })
assert.Equalf(t, capabilities.SparseExtentReporting, holes, assert.Equalf(t, capabilities.SparseExtentReporting, holes,
@@ -334,16 +344,16 @@ func RunNBDClientSuiteWithConfig(t *testing.T, newProvider func(t *testing.T) eb
return return
} }
pub := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-readonly", true) nbdURI := publishForNBD(t, provider, cfg, cfg.VolumePrefix+"-readonly", true)
export := nbdInfo(t, dialURI(t, pub.NBDURI)) export := nbdInfo(t, dialURI(t, nbdURI))
assert.True(t, export.ReadOnly, "a volume published ReadOnly must set the NBD read-only transmission flag") assert.True(t, export.ReadOnly, "a volume published ReadOnly must set the NBD read-only transmission flag")
}) })
t.Run("unpublish tears the export down", func(t *testing.T) { t.Run("unpublish tears the export down", func(t *testing.T) {
provider := newProvider(t) provider := newProvider(t)
volumeID := cfg.VolumePrefix + "-unpublish" volumeID := cfg.VolumePrefix + "-unpublish"
pub := publishForNBD(t, provider, cfg, volumeID, false) nbdURI := publishForNBD(t, provider, cfg, volumeID, false)
uri := dialURI(t, pub.NBDURI) uri := dialURI(t, nbdURI)
require.True(t, nbdReachable(uri), "export should be reachable before unpublish") require.True(t, nbdReachable(uri), "export should be reachable before unpublish")
require.NoError(t, provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{ require.NoError(t, provider.UnpublishVolume(context.Background(), ebsprovider.UnpublishVolumeRequest{
+12 -1
View File
@@ -74,13 +74,24 @@ func corruptPublished(published *ebsprovider.PublishedVolume, flavour string) *e
corrupted := *published corrupted := *published
switch flavour { switch flavour {
case "handle", "capacity": case "handle", "capacity":
corrupted.NBDURI = "nbd+unix:///?socket=/hostile/wrong.sock" corrupted.Export = corruptExport(published.Export)
case "state": case "state":
corrupted.NodeID = "hostile-wrong-node" corrupted.NodeID = "hostile-wrong-node"
} }
return &corrupted return &corrupted
} }
// corruptExport points a still-valid export at the wrong place, building new
// variants so the inner provider's own export is never mutated.
func corruptExport(export ebsprovider.VolumeExport) ebsprovider.VolumeExport {
if export.Kind == ebsprovider.ExportKindBlock && export.Block != nil {
block := *export.Block
block.DevicePath = "/dev/hostile-wrong"
return ebsprovider.VolumeExport{Kind: ebsprovider.ExportKindBlock, Block: &block}
}
return ebsprovider.NewNBDExport("nbd+unix:///?socket=/hostile/wrong.sock")
}
func (p *Provider) GetCapabilities(ctx context.Context, req ebsprovider.GetCapabilitiesRequest) (*ebsprovider.GetCapabilitiesResponse, error) { func (p *Provider) GetCapabilities(ctx context.Context, req ebsprovider.GetCapabilitiesRequest) (*ebsprovider.GetCapabilitiesResponse, error) {
injection := p.draw(verbCapabilities, "") injection := p.draw(verbCapabilities, "")
if stop, err := p.apply(ctx, injection); stop || err != nil { if stop, err := p.apply(ctx, injection); stop || err != nil {
+9 -1
View File
@@ -375,7 +375,7 @@ func (m *MemoryProvider) PublishVolume(_ context.Context, req PublishVolumeReque
volume.published = &PublishedVolume{ volume.published = &PublishedVolume{
VolumeID: req.VolumeID, VolumeID: req.VolumeID,
NodeID: req.NodeID, NodeID: req.NodeID,
NBDURI: "nbd+unix:///?socket=/memory/" + req.VolumeID + ".sock", Export: NewNBDExport("nbd+unix:///?socket=/memory/" + req.VolumeID + ".sock"),
} }
volume.volume.State = VolumeStateInUse volume.volume.State = VolumeStateInUse
return clonePublished(volume.published), nil return clonePublished(volume.published), nil
@@ -421,5 +421,13 @@ func clonePublished(published *PublishedVolume) *PublishedVolume {
return nil return nil
} }
clone := *published clone := *published
if published.Export.NBD != nil {
nbd := *published.Export.NBD
clone.Export.NBD = &nbd
}
if published.Export.Block != nil {
block := *published.Export.Block
clone.Export.Block = &block
}
return &clone return &clone
} }
+1 -1
View File
@@ -40,7 +40,7 @@ func TestMemoryProviderLifecycleAndIdempotency(t *testing.T) {
Versioned: NewVersioned(), VolumeID: volume.ID, Handle: volume.Handle, NodeID: "node-1", Versioned: NewVersioned(), VolumeID: volume.ID, Handle: volume.Handle, NodeID: "node-1",
}) })
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, "nbd+unix:///?socket=/memory/vol-1.sock", published.NBDURI) assert.Equal(t, NewNBDExport("nbd+unix:///?socket=/memory/vol-1.sock"), published.Export)
repeatedPublish, err := provider.PublishVolume(ctx, PublishVolumeRequest{ repeatedPublish, err := provider.PublishVolume(ctx, PublishVolumeRequest{
Versioned: NewVersioned(), VolumeID: volume.ID, Handle: volume.Handle, NodeID: "node-1", Versioned: NewVersioned(), VolumeID: volume.ID, Handle: volume.Handle, NodeID: "node-1",
+3
View File
@@ -452,6 +452,9 @@ func (p *NATSProvider) PublishVolume(ctx context.Context, req PublishVolumeReque
if response.Published == nil { if response.Published == nil {
return nil, fmt.Errorf("%s returned no published volume", subject) return nil, fmt.Errorf("%s returned no published volume", subject)
} }
if err := response.Published.Export.Validate(); err != nil {
return nil, fmt.Errorf("%s returned an invalid export: %w", subject, err)
}
return response.Published, nil return response.Published, nil
} }
+44
View File
@@ -432,6 +432,50 @@ func TestNATSProviderRejectsEmptySuccessResponse(t *testing.T) {
}) })
} }
// TestNATSProviderRejectsInvalidExport covers the client-side export check: a
// reply the server marks successful but whose export cannot be attached must
// fail here, not later when QEMU is handed it.
func TestNATSProviderRejectsInvalidExport(t *testing.T) {
_, conn := testutil.StartTestNATS(t)
provider := NewNATSProvider(conn, time.Second)
tests := []struct {
node string
export VolumeExport
wantErr string
}{
{node: "node-nokind", export: VolumeExport{NBD: &NBDExport{NBDURI: "nbd://10.0.0.1:10809"}}, wantErr: `unknown export kind ""`},
{node: "node-noblock", export: VolumeExport{Kind: ExportKindBlock}, wantErr: "block export has no block details"},
{node: "node-emptyuri", export: VolumeExport{Kind: ExportKindNBD, NBD: &NBDExport{}}, wantErr: "nbd export has an empty NBD URI"},
}
for _, tt := range tests {
t.Run(tt.node, func(t *testing.T) {
subject, err := PublishSubject(tt.node)
require.NoError(t, err)
respondWith(t, conn, subject, PublishVolumeResponse{
Versioned: NewVersioned(),
Published: &PublishedVolume{VolumeID: "vol-1", NodeID: tt.node, Export: tt.export},
})
published, err := provider.PublishVolume(t.Context(), PublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: tt.node})
require.EqualError(t, err, subject+" returned an invalid export: "+tt.wantErr)
assert.Nil(t, published)
})
}
t.Run("valid block export passes", func(t *testing.T) {
subject, err := PublishSubject("node-block")
require.NoError(t, err)
export := VolumeExport{Kind: ExportKindBlock, Block: &BlockExport{DevicePath: "/dev/sdb", SizeBytes: 10 << 30}}
respondWith(t, conn, subject, PublishVolumeResponse{
Versioned: NewVersioned(),
Published: &PublishedVolume{VolumeID: "vol-1", NodeID: "node-block", Export: export},
})
published, err := provider.PublishVolume(t.Context(), PublishVolumeRequest{Versioned: NewVersioned(), VolumeID: "vol-1", NodeID: "node-block"})
require.NoError(t, err)
assert.Equal(t, export, published.Export)
})
}
// TestNATSProviderPropagatesProviderErrors covers responseError's populated // TestNATSProviderPropagatesProviderErrors covers responseError's populated
// branch on the enumeration and unpublish verbs, whose errors otherwise reach // branch on the enumeration and unpublish verbs, whose errors otherwise reach
// the caller as an empty page or a silent success. // the caller as an empty page or a silent success.
@@ -276,7 +276,7 @@ func TestServedProviderRoundTrip(t *testing.T) {
Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip", NodeID: "node-1", Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip", NodeID: "node-1",
}) })
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, "nbd+unix:///?socket=/memory/vol-roundtrip.sock", published.NBDURI) assert.Equal(t, ebsprovider.NewNBDExport("nbd+unix:///?socket=/memory/vol-roundtrip.sock"), published.Export)
inUse, err := client.GetVolume(ctx, ebsprovider.GetVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip"}) inUse, err := client.GetVolume(ctx, ebsprovider.GetVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-roundtrip"})
require.NoError(t, err) require.NoError(t, err)
@@ -432,6 +432,42 @@ func (p *countingProvider) PublishVolume(ctx context.Context, req ebsprovider.Pu
return p.EBSProvider.PublishVolume(ctx, req) return p.EBSProvider.PublishVolume(ctx, req)
} }
// exportProvider publishes a fixed export, so the server's own export check
// can be driven with a provider that returns one it must not relay.
type exportProvider struct {
ebsprovider.EBSProvider
export ebsprovider.VolumeExport
}
func (p *exportProvider) PublishVolume(_ context.Context, req ebsprovider.PublishVolumeRequest) (*ebsprovider.PublishedVolume, error) {
return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: p.export}, nil
}
// TestPublishVolumeRejectsInvalidExport covers the server-side export check:
// an unattachable export is the provider's fault and goes out as internal.
func TestPublishVolumeRejectsInvalidExport(t *testing.T) {
provider := &exportProvider{
EBSProvider: ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}),
export: ebsprovider.VolumeExport{Kind: ebsprovider.ExportKindBlock, Block: &ebsprovider.BlockExport{DevicePath: "/dev/sdb"}},
}
conn := serveProvider(t, provider, natsserve.Options{NoQueueGroup: true})
subject, err := ebsprovider.PublishSubject("node-1")
require.NoError(t, err)
payload, err := json.Marshal(ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", NodeID: "node-1"})
require.NoError(t, err)
msg, err := conn.Request(subject, payload, requestTimeout)
require.NoError(t, err)
var resp ebsprovider.PublishVolumeResponse
require.NoError(t, json.Unmarshal(msg.Data, &resp))
assert.Nil(t, resp.Published, "an invalid export must not be relayed")
require.NotNil(t, resp.Error)
assert.Equal(t, ebsprovider.ErrorCodeInternal, resp.Error.Code)
assert.Equal(t, "provider returned an invalid export for vol-1: block export has a zero size", resp.Error.Message)
}
// handlerSettle is how long a handler count is watched after it reaches the // handlerSettle is how long a handler count is watched after it reaches the
// expected value, to catch a delivery that should not have happened arriving // expected value, to catch a delivery that should not have happened arriving
// late. Generous, because a false pass here is a silently broken queue group. // late. Generous, because a false pass here is a silently broken queue group.
@@ -543,7 +579,7 @@ func TestServeRejectsUnroutableNodeID(t *testing.T) {
assert.Contains(t, err.Error(), "build publish subject") assert.Contains(t, err.Error(), "build publish subject")
assert.Nil(t, stop) assert.Nil(t, stop)
_, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(`{"schema_version":1}`), 200*time.Millisecond) _, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(fmt.Sprintf(`{"schema_version":%d}`, ebsprovider.SchemaVersion)), 200*time.Millisecond)
require.ErrorIs(t, err, nats.ErrNoResponders, "a failed Serve must leave no subscription behind") require.ErrorIs(t, err, nats.ErrNoResponders, "a failed Serve must leave no subscription behind")
} }
@@ -571,6 +607,6 @@ func TestServeStopIsIdempotent(t *testing.T) {
stop() stop()
require.NoError(t, conn.Flush()) require.NoError(t, conn.Flush())
_, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(`{"schema_version":1}`), 200*time.Millisecond) _, err = conn.Request(ebsprovider.CapabilitiesSubject, []byte(fmt.Sprintf(`{"schema_version":%d}`, ebsprovider.SchemaVersion)), 200*time.Millisecond)
require.ErrorIs(t, err, nats.ErrNoResponders, "stop must unsubscribe every subject Serve registered") require.ErrorIs(t, err, nats.ErrNoResponders, "stop must unsubscribe every subject Serve registered")
} }
@@ -501,6 +501,16 @@ func handlePublishVolume(provider ebsprovider.EBSProvider) msgHandler {
respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)}) respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: classifyError(err)})
return return
} }
if pub == nil {
respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("provider returned no published volume for %s", req.VolumeID)})
return
}
// An unattachable export is the provider's fault, so it is reported as
// internal rather than relayed for the client to reject.
if err := pub.Export.Validate(); err != nil {
respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("provider returned an invalid export for %s: %v", req.VolumeID, err)})
return
}
respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Published: pub}) respond(ctx, msg, ebsprovider.PublishVolumeResponse{Versioned: ebsprovider.NewVersioned(), Published: pub})
} }
} }
+76 -6
View File
@@ -6,14 +6,16 @@ package ebsprovider
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"path/filepath"
"time" "time"
) )
// SchemaVersion is the only wire-contract version understood by this build. // SchemaVersion is the only wire-contract version understood by this build.
// Every request and response carries it so version skew fails at the boundary // Every request and response carries it so version skew fails at the boundary
// instead of silently decoding a changed payload. // instead of silently decoding a changed payload.
const SchemaVersion uint16 = 1 const SchemaVersion uint16 = 2
// Versioned is embedded in every request and response sent over NATS. // Versioned is embedded in every request and response sent over NATS.
type Versioned struct { type Versioned struct {
@@ -443,12 +445,80 @@ type PublishVolumeRequest struct {
ReadOnly bool `json:"read_only,omitempty"` ReadOnly bool `json:"read_only,omitempty"`
} }
// PublishedVolume is where a published volume is served: the NBD URI on NodeID // ExportKind names how a published volume is reached on its node, and so
// that QEMU connects to. // which VolumeExport variant carries the details.
type ExportKind string
const (
ExportKindNBD ExportKind = "nbd"
ExportKindBlock ExportKind = "block"
)
// VolumeExport is how a published volume is reached; exactly one of NBD and
// Block is set, and it is the one Kind names.
type VolumeExport struct {
Kind ExportKind `json:"kind"`
NBD *NBDExport `json:"nbd,omitempty"`
Block *BlockExport `json:"block,omitempty"`
}
// NBDExport is a volume served over NBD: the URI on the node that QEMU
// connects to.
type NBDExport struct {
NBDURI string `json:"nbd_uri"`
}
// BlockExport is a volume exposed as a host block device. SizeBytes is the
// size the guest sees, which may be smaller than the device itself.
type BlockExport struct {
DevicePath string `json:"device_path"`
SizeBytes uint64 `json:"size_bytes"`
}
// Validate rejects an export a consumer could not attach: an unknown kind, a
// missing or extra variant, or a variant with an unusable field.
func (e VolumeExport) Validate() error {
switch e.Kind {
case ExportKindNBD:
if e.NBD == nil {
return errors.New("nbd export has no nbd details")
}
if e.Block != nil {
return errors.New("nbd export also sets block details")
}
if e.NBD.NBDURI == "" {
return errors.New("nbd export has an empty NBD URI")
}
case ExportKindBlock:
if e.Block == nil {
return errors.New("block export has no block details")
}
if e.NBD != nil {
return errors.New("block export also sets nbd details")
}
if !filepath.IsAbs(e.Block.DevicePath) {
return fmt.Errorf("block export device path %q is not absolute", e.Block.DevicePath)
}
if e.Block.SizeBytes == 0 {
return errors.New("block export has a zero size")
}
default:
return fmt.Errorf("unknown export kind %q", e.Kind)
}
return nil
}
// PublishedVolume is where a published volume is served: the export on NodeID
// that QEMU attaches.
type PublishedVolume struct { type PublishedVolume struct {
VolumeID string `json:"volume_id"` VolumeID string `json:"volume_id"`
NodeID string `json:"node_id"` NodeID string `json:"node_id"`
NBDURI string `json:"nbd_uri"` Export VolumeExport `json:"export"`
}
// NewNBDExport returns an nbd export for uri.
func NewNBDExport(uri string) VolumeExport {
return VolumeExport{Kind: ExportKindNBD, NBD: &NBDExport{NBDURI: uri}}
} }
// PublishVolumeResponse is the reply to PublishVolumeRequest: the mount, or // PublishVolumeResponse is the reply to PublishVolumeRequest: the mount, or
+63
View File
@@ -0,0 +1,63 @@
package ebsprovider
import (
"encoding/json"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestVolumeExportValidate(t *testing.T) {
nbd := &NBDExport{NBDURI: "nbd+unix:///?socket=/run/vol-1.sock"}
block := &BlockExport{DevicePath: "/dev/sdb", SizeBytes: 10 << 30}
tests := []struct {
name string
export VolumeExport
wantErr string
}{
{name: "valid nbd", export: VolumeExport{Kind: ExportKindNBD, NBD: nbd}},
{name: "valid block", export: VolumeExport{Kind: ExportKindBlock, Block: block}},
{name: "empty kind", export: VolumeExport{NBD: nbd}, wantErr: `unknown export kind ""`},
{name: "unknown kind", export: VolumeExport{Kind: "iscsi", NBD: nbd}, wantErr: `unknown export kind "iscsi"`},
{name: "nbd kind without nbd variant", export: VolumeExport{Kind: ExportKindNBD}, wantErr: "nbd export has no nbd details"},
{name: "nbd kind carrying only block variant", export: VolumeExport{Kind: ExportKindNBD, Block: block}, wantErr: "nbd export has no nbd details"},
{name: "nbd kind with block variant", export: VolumeExport{Kind: ExportKindNBD, NBD: nbd, Block: block}, wantErr: "nbd export also sets block details"},
{name: "empty nbd uri", export: VolumeExport{Kind: ExportKindNBD, NBD: &NBDExport{}}, wantErr: "nbd export has an empty NBD URI"},
{name: "block kind without block variant", export: VolumeExport{Kind: ExportKindBlock}, wantErr: "block export has no block details"},
{name: "block kind with nbd variant", export: VolumeExport{Kind: ExportKindBlock, NBD: nbd, Block: block}, wantErr: "block export also sets nbd details"},
{name: "relative device path", export: VolumeExport{Kind: ExportKindBlock, Block: &BlockExport{DevicePath: "dev/sdb", SizeBytes: 1}}, wantErr: `block export device path "dev/sdb" is not absolute`},
{name: "empty device path", export: VolumeExport{Kind: ExportKindBlock, Block: &BlockExport{SizeBytes: 1}}, wantErr: `block export device path "" is not absolute`},
{name: "zero size", export: VolumeExport{Kind: ExportKindBlock, Block: &BlockExport{DevicePath: "/dev/sdb"}}, wantErr: "block export has a zero size"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := tt.export.Validate()
if tt.wantErr == "" {
require.NoError(t, err)
return
}
require.EqualError(t, err, tt.wantErr)
})
}
}
// TestPublishedVolumeWireShape pins the JSON the contract carries: a renamed
// tag would decode as an empty variant on a peer at the same schema version.
func TestPublishedVolumeWireShape(t *testing.T) {
wires := map[string]string{
"nbd": `{"volume_id":"vol-1","node_id":"node-1","export":{"kind":"nbd","nbd":{"nbd_uri":"nbd://10.0.0.1:10809"}}}`,
"block": `{"volume_id":"vol-1","node_id":"node-1","export":{"kind":"block","block":{"device_path":"/dev/sdb","size_bytes":10737418240}}}`,
}
for name, wire := range wires {
t.Run(name, func(t *testing.T) {
var published PublishedVolume
require.NoError(t, json.Unmarshal([]byte(wire), &published))
require.NoError(t, published.Export.Validate())
encoded, err := json.Marshal(published)
require.NoError(t, err)
assert.JSONEq(t, wire, string(encoded))
})
}
}
@@ -1765,7 +1765,7 @@ func handlePublishVolume(ctx context.Context, cfg *Config, nc *nats.Conn, msg *n
Published: &ebsprovider.PublishedVolume{ Published: &ebsprovider.PublishedVolume{
VolumeID: req.VolumeID, VolumeID: req.VolumeID,
NodeID: cfg.NodeName, NodeID: cfg.NodeName,
NBDURI: mv.NBDURI, Export: ebsprovider.NewNBDExport(mv.NBDURI),
}, },
}) })
return return
@@ -1795,7 +1795,7 @@ func handlePublishVolume(ctx context.Context, cfg *Config, nc *nats.Conn, msg *n
Published: &ebsprovider.PublishedVolume{ Published: &ebsprovider.PublishedVolume{
VolumeID: req.VolumeID, VolumeID: req.VolumeID,
NodeID: cfg.NodeName, NodeID: cfg.NodeName,
NBDURI: mountResp.URI, Export: ebsprovider.NewNBDExport(mountResp.URI),
}, },
}) })
} }
@@ -837,7 +837,7 @@ func TestProviderHandlers_PublishVolume_IdempotentRepublish(t *testing.T) {
require.NotNil(t, resp.Published) require.NotNil(t, resp.Published)
assert.Equal(t, volumeName, resp.Published.VolumeID) assert.Equal(t, volumeName, resp.Published.VolumeID)
assert.Equal(t, cfg.NodeName, resp.Published.NodeID) assert.Equal(t, cfg.NodeName, resp.Published.NodeID)
assert.Equal(t, existingURI, resp.Published.NBDURI) assert.Equal(t, ebsprovider.NewNBDExport(existingURI), resp.Published.Export)
cfg.mu.Lock() cfg.mu.Lock()
count := len(cfg.MountedVolumes) count := len(cfg.MountedVolumes)
+32
View File
@@ -355,3 +355,35 @@ func VolumeBlockdev(nodeName string, server NBDServerOpts) Blockdev {
opts = append(opts, "export=", fmt.Sprintf("reconnect-delay=%d", NBDReconnectDelaySeconds)) opts = append(opts, "export=", fmt.Sprintf("reconnect-delay=%d", NBDReconnectDelaySeconds))
return Blockdev{Value: strings.Join(opts, ",")} return Blockdev{Value: strings.Join(opts, ",")}
} }
// HostDeviceOpts is a host block device export. QEMU opens Path through the
// host_device protocol driver with O_DIRECT and native AIO, and a raw format
// node over it exposes only the first SizeBytes to the guest.
type HostDeviceOpts struct {
Path string
SizeBytes uint64
}
// QMPArgs renders the blockdev-add arguments for the raw node and its nested
// host_device file node.
func (o HostDeviceOpts) QMPArgs(nodeName string) map[string]any {
return map[string]any{
"node-name": nodeName,
"driver": "raw",
"size": o.SizeBytes,
"file": map[string]any{
"driver": "host_device",
"filename": o.Path,
"aio": "native",
"cache": map[string]any{"direct": true},
},
}
}
// VolumeHostDeviceBlockdev renders the -blockdev equivalent of QMPArgs for a
// data volume backed by a host block device.
func VolumeHostDeviceBlockdev(nodeName string, dev HostDeviceOpts) Blockdev {
return Blockdev{Value: fmt.Sprintf(
"driver=raw,node-name=%s,size=%d,file.driver=host_device,file.filename=%s,file.aio=native,file.cache.direct=on",
nodeName, dev.SizeBytes, dev.Path)}
}
@@ -1,6 +1,10 @@
package vm package vm
import ( import (
"fmt"
"os"
"path/filepath"
"strings"
"sync" "sync"
viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1" viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
@@ -20,3 +24,45 @@ type EBSRequests struct {
// VM's persisted attachment list. The alias keeps the VM state tied to the // VM's persisted attachment list. The alias keeps the VM state tied to the
// deployed wire shape without making the old catch-all types package an owner. // deployed wire shape without making the old catch-all types package an owner.
type EBSRequest = viperblocklegacyv1.EBSRequest type EBSRequest = viperblocklegacyv1.EBSRequest
// volumeExportKind is how a volume's backing store reaches QEMU.
type volumeExportKind int
const (
volumeExportInvalid volumeExportKind = iota
volumeExportNBD
volumeExportBlock
)
// classifyVolumeExport reports which export a request carries. A block device
// must be absolute, comma-free (QEMU key=value options would split it) and
// sized; both exports set, neither, or a stray size is invalid.
func classifyVolumeExport(req EBSRequest) (volumeExportKind, error) {
switch {
case req.NBDURI != "" && req.BlockDevice != "":
return volumeExportInvalid, fmt.Errorf("volume %s has both NBDURI and BlockDevice set", req.Name)
case req.BlockSizeBytes != 0 && req.BlockDevice == "":
return volumeExportInvalid, fmt.Errorf("volume %s has BlockSizeBytes set without BlockDevice", req.Name)
case req.NBDURI != "":
return volumeExportNBD, nil
case req.BlockDevice != "":
if !filepath.IsAbs(req.BlockDevice) {
return volumeExportInvalid, fmt.Errorf("volume %s BlockDevice %q is not an absolute path", req.Name, req.BlockDevice)
}
if strings.Contains(req.BlockDevice, ",") {
return volumeExportInvalid, fmt.Errorf("volume %s BlockDevice %q contains a comma", req.Name, req.BlockDevice)
}
if req.BlockSizeBytes == 0 {
return volumeExportInvalid, fmt.Errorf("volume %s BlockDevice %q has zero BlockSizeBytes", req.Name, req.BlockDevice)
}
return volumeExportBlock, nil
default:
return volumeExportInvalid, fmt.Errorf("NBDURI not set for volume %s - was volume mounted?", req.Name)
}
}
// checkBlockDevice reports whether a block export's device node is present.
func checkBlockDevice(path string) error {
_, err := os.Stat(path)
return err
}
+49 -20
View File
@@ -607,11 +607,18 @@ func (m *Manager) startQEMU(instance *VM) error {
refreshTelemetryMeta(instance) refreshTelemetryMeta(instance)
} }
// An NBD export is waited on until it accepts connections; a block export
// is local to the host and only has to exist. An invalid request has no
// endpoint to probe (buildDrives rejects it on the command-line path).
instance.EBSRequests.Mu.Lock() instance.EBSRequests.Mu.Lock()
nbdEndpoints := make([]struct{ name, uri string }, 0, len(instance.EBSRequests.Requests)) nbdEndpoints := make([]struct{ name, uri string }, 0, len(instance.EBSRequests.Requests))
var blockDevices []struct{ name, path string }
for _, req := range instance.EBSRequests.Requests { for _, req := range instance.EBSRequests.Requests {
if req.NBDURI != "" { switch kind, _ := classifyVolumeExport(req); kind {
case volumeExportNBD:
nbdEndpoints = append(nbdEndpoints, struct{ name, uri string }{req.Name, req.NBDURI}) nbdEndpoints = append(nbdEndpoints, struct{ name, uri string }{req.Name, req.NBDURI})
case volumeExportBlock:
blockDevices = append(blockDevices, struct{ name, path string }{req.Name, req.BlockDevice})
} }
} }
instance.EBSRequests.Mu.Unlock() instance.EBSRequests.Mu.Unlock()
@@ -620,6 +627,11 @@ func (m *Manager) startQEMU(instance *VM) error {
return fmt.Errorf("nbd endpoint not ready for %s: %w", ep.name, err) return fmt.Errorf("nbd endpoint not ready for %s: %w", ep.name, err)
} }
} }
for _, dev := range blockDevices {
if err := checkBlockDevice(dev.path); err != nil {
return fmt.Errorf("block device not present for %s: %w", dev.name, err)
}
}
processChan := make(chan int, 1) processChan := make(chan int, 1)
exitChan := make(chan int, 1) exitChan := make(chan int, 1)
@@ -900,11 +912,11 @@ func qmpGreetingTimeout(v *VM) time.Duration {
// hasNetworkBackedDrive reports whether any of the guest's drives is served // hasNetworkBackedDrive reports whether any of the guest's drives is served
// over NBD, and so has to be opened across the object store before QEMU's // over NBD, and so has to be opened across the object store before QEMU's
// monitor answers. Every production guest boots from one; a VM without is a // monitor answers. A host block device opens locally and does not count.
// unit test.
func hasNetworkBackedDrive(v *VM) bool { func hasNetworkBackedDrive(v *VM) bool {
return slices.ContainsFunc(v.EBSRequests.Requests, func(req EBSRequest) bool { return slices.ContainsFunc(v.EBSRequests.Requests, func(req EBSRequest) bool {
return req.NBDURI != "" kind, _ := classifyVolumeExport(req)
return kind == volumeExportNBD
}) })
} }
@@ -1445,8 +1457,9 @@ type driveConfig struct {
} }
// buildDrives converts EBS requests into QEMU drive/iothread/device/blockdev // buildDrives converts EBS requests into QEMU drive/iothread/device/blockdev
// configs. Returns an error if any volume is missing its NBDURI, or if the // configs. Returns an error if any volume has no valid export (see
// EBS hot-plug port pool is exhausted while allocating one for a data volume. // classifyVolumeExport), or if the EBS hot-plug port pool is exhausted while
// allocating one for a data volume.
// //
// Boot volumes keep the existing anonymous if=none -drive shape exactly as // Boot volumes keep the existing anonymous if=none -drive shape exactly as
// before — they are not detachable (see ErrVolumeNotDetachable), so the // before — they are not detachable (see ErrVolumeNotDetachable), so the
@@ -1471,11 +1484,17 @@ func buildDrives(requests []EBSRequest, cpuCount int, machineType string) (drive
for i := range requests { for i := range requests {
v := &requests[i] v := &requests[i]
if v.NBDURI == "" { kind, err := classifyVolumeExport(*v)
return driveConfig{}, fmt.Errorf("NBDURI not set for volume %s - was volume mounted?", v.Name) if err != nil {
return driveConfig{}, err
} }
switch { switch {
case kind == volumeExportBlock && (v.Boot || v.EFI):
// Boot and EFI keep the legacy -drive shape, which has no
// host_device form here; only data volumes take a block export.
return driveConfig{}, fmt.Errorf("volume %s: block device export is only supported for data volumes", v.Name)
case v.Boot: case v.Boot:
drive := Drive{ drive := Drive{
File: v.NBDURI, File: v.NBDURI,
@@ -1529,29 +1548,39 @@ func buildDrives(requests []EBSRequest, cpuCount int, machineType string) (drive
v.HotplugPort = port v.HotplugPort = port
} }
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)
}
nodeName := VolumeNodeName(v.Name) nodeName := VolumeNodeName(v.Name)
iothreadID := VolumeIOThreadID(v.Name) iothreadID := VolumeIOThreadID(v.Name)
bus := HotplugEBSBus(v.HotplugPort) bus := HotplugEBSBus(v.HotplugPort)
var blockdev Blockdev
if kind == volumeExportBlock {
blockdev = VolumeHostDeviceBlockdev(nodeName, HostDeviceOpts{Path: v.BlockDevice, SizeBytes: v.BlockSizeBytes})
} else {
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)
}
blockdev = VolumeBlockdev(nodeName, NBDServerOpts{
Type: serverType,
Path: socketPath,
Host: nbdHost,
Port: nbdPort,
})
}
cfg.IOThreads = append(cfg.IOThreads, IOThread{ID: iothreadID}) cfg.IOThreads = append(cfg.IOThreads, IOThread{ID: iothreadID})
cfg.Blockdevs = append(cfg.Blockdevs, VolumeBlockdev(nodeName, NBDServerOpts{ cfg.Blockdevs = append(cfg.Blockdevs, blockdev)
Type: serverType,
Path: socketPath,
Host: nbdHost,
Port: nbdPort,
}))
cfg.Devices = append(cfg.Devices, VolumeBlkDevice(v.Name, nodeName, iothreadID, bus)) cfg.Devices = append(cfg.Devices, VolumeBlkDevice(v.Name, nodeName, iothreadID, bus))
// Deliberately no entry in cfg.Drives: the bare anonymous -drive // Deliberately no entry in cfg.Drives: the bare anonymous -drive
// this case used to emit (no id=, no node-name=) is exactly what // this case used to emit (no id=, no node-name=) is exactly what
// left the volume undetachable. // left the volume undetachable.
} }
slog.Info("Using NBD URI for drive", "volume", v.Name, "uri", v.NBDURI) if kind == volumeExportBlock {
slog.Info("Using block device for drive", "volume", v.Name, "device", v.BlockDevice, "size_bytes", v.BlockSizeBytes)
} else {
slog.Info("Using NBD URI for drive", "volume", v.Name, "uri", v.NBDURI)
}
} }
return cfg, nil return cfg, nil
@@ -273,6 +273,83 @@ func TestBuildDrives(t *testing.T) {
}, },
wantHotplugPorts: []int{0, 0, 3}, wantHotplugPorts: []int{0, 0, 3},
}, },
{
name: "block data volume emits a raw node over host_device with the same device wiring",
requests: []EBSRequest{
{Name: "vol-blk", BlockDevice: "/dev/sdx", BlockSizeBytes: 10 << 30, HotplugPort: 2},
},
cpuCount: 2,
wantIOThreads: []IOThread{{ID: "ioth-vol-blk"}},
wantBlockdevs: []Blockdev{
{Value: "driver=raw,node-name=nbd-vol-blk,size=10737418240,file.driver=host_device,file.filename=/dev/sdx,file.aio=native,file.cache.direct=on"},
},
wantDevices: []Device{
{Value: "virtio-blk-pci,id=vdisk-vol-blk,drive=nbd-vol-blk,iothread=ioth-vol-blk,serial=volblk,bus=hotplug-ebs2,werror=stop,rerror=stop"},
},
wantHotplugPorts: []int{2},
},
{
name: "NBD boot volume alongside a block data volume",
requests: []EBSRequest{
{Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true},
{Name: "vol-blk", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20},
},
cpuCount: 2,
wantDrives: []Drive{
{File: "nbd:unix:/tmp/boot.sock", Format: "raw", If: "none", Media: "disk", ID: "os", Cache: "none", Werror: "stop", Rerror: "stop", ReconnectDelay: NBDReconnectDelaySeconds},
},
wantIOThreads: []IOThread{{ID: "ioth-os"}, {ID: "ioth-vol-blk"}},
wantBlockdevs: []Blockdev{
{Value: "driver=raw,node-name=nbd-vol-blk,size=1048576,file.driver=host_device,file.filename=/dev/sdx,file.aio=native,file.cache.direct=on"},
},
wantDevices: []Device{
{Value: "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=2,bootindex=1"},
{Value: "virtio-blk-pci,id=vdisk-vol-blk,drive=nbd-vol-blk,iothread=ioth-vol-blk,serial=volblk,bus=hotplug-ebs1,werror=stop,rerror=stop"},
},
wantHotplugPorts: []int{0, 1},
},
{
name: "block boot volume is rejected",
requests: []EBSRequest{{Name: "vol-boot", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20, Boot: true}},
cpuCount: 2,
wantErr: "volume vol-boot: block device export is only supported for data volumes",
},
{
name: "block EFI volume is rejected",
requests: []EBSRequest{{Name: "vol-efi", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20, EFI: true}},
cpuCount: 2,
wantErr: "volume vol-efi: block device export is only supported for data volumes",
},
{
name: "both NBDURI and BlockDevice set is rejected",
requests: []EBSRequest{{Name: "vol-both", NBDURI: "nbd:unix:/tmp/x.sock", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20}},
cpuCount: 2,
wantErr: "volume vol-both has both NBDURI and BlockDevice set",
},
{
name: "NBDURI with a stray BlockSizeBytes is rejected",
requests: []EBSRequest{{Name: "vol-stray", NBDURI: "nbd:unix:/tmp/x.sock", BlockSizeBytes: 1 << 20}},
cpuCount: 2,
wantErr: "volume vol-stray has BlockSizeBytes set without BlockDevice",
},
{
name: "block device with zero size is rejected",
requests: []EBSRequest{{Name: "vol-nosize", BlockDevice: "/dev/sdx"}},
cpuCount: 2,
wantErr: `volume vol-nosize BlockDevice "/dev/sdx" has zero BlockSizeBytes`,
},
{
name: "relative block device path is rejected",
requests: []EBSRequest{{Name: "vol-rel", BlockDevice: "dev/sdx", BlockSizeBytes: 1 << 20}},
cpuCount: 2,
wantErr: `volume vol-rel BlockDevice "dev/sdx" is not an absolute path`,
},
{
name: "block device path with a comma is rejected",
requests: []EBSRequest{{Name: "vol-comma", BlockDevice: "/dev/sdx,file.locking=off", BlockSizeBytes: 1 << 20}},
cpuCount: 2,
wantErr: `volume vol-comma BlockDevice "/dev/sdx,file.locking=off" contains a comma`,
},
{ {
name: "empty requests", name: "empty requests",
requests: []EBSRequest{}, requests: []EBSRequest{},
@@ -1008,6 +1085,28 @@ func TestStartQEMU_DirectBoot_InstanceTypeNotFound(t *testing.T) {
assert.Empty(t, plumber.setupCalls) assert.Empty(t, plumber.setupCalls)
} }
// TestStartQEMU_BlockDeviceMissing verifies a block volume's launch readiness
// is the device's presence, not an NBD connect.
func TestStartQEMU_BlockDeviceMissing(t *testing.T) {
m := directBootManager(t, &fakeNetworkPlumber{})
missing := filepath.Join(t.TempDir(), "missing-dev")
instance := &VM{
ID: "i-blk-missing",
InstanceType: "t3.nano",
DirectBoot: true,
}
instance.EBSRequests.Requests = []EBSRequest{
{Name: "vol-blk", BlockDevice: missing, BlockSizeBytes: 1 << 20},
}
err := m.startQEMU(instance)
require.Error(t, err)
assert.Contains(t, err.Error(), "block device not present for vol-blk")
assert.ErrorIs(t, err, os.ErrNotExist)
}
// scriptedNetworkPlumber returns errors from a pre-loaded slice, one per call. // scriptedNetworkPlumber returns errors from a pre-loaded slice, one per call.
type scriptedNetworkPlumber struct { type scriptedNetworkPlumber struct {
calls []TapSpec calls []TapSpec
@@ -1271,6 +1370,12 @@ func TestQMPGreetingTimeout(t *testing.T) {
assert.Greater(t, qmpNBDGreetingTimeout, 35*time.Second, assert.Greater(t, qmpNBDGreetingTimeout, 35*time.Second,
"the deadline must exceed the 35s cold open measured with one node of three away") "the deadline must exceed the 35s cold open measured with one node of three away")
block := &VM{EBSRequests: EBSRequests{Requests: []EBSRequest{
{Name: "vol-data", BlockDevice: "/dev/sdx", BlockSizeBytes: 1 << 20},
}}}
assert.Equal(t, qmp.DefaultGreetingTimeout, qmpGreetingTimeout(block),
"a host block device opens locally, so it keeps the plain default")
// A small GPU guest (<1 GiB scaling contribution) sits at the floor. // A small GPU guest (<1 GiB scaling contribution) sits at the floor.
smallGPU := &VM{ smallGPU := &VM{
GPUAttachments: []gpu.GPUAttachment{{PCIAddress: "0000:5e:00.0"}}, GPUAttachments: []gpu.GPUAttachment{{PCIAddress: "0000:5e:00.0"}},
@@ -142,6 +142,10 @@ type fakeVolumeMounter struct {
unmountErr error unmountErr error
unmountOneErr error unmountOneErr error
mountOneURI string mountOneURI string
// mountOneBlockDevice and mountOneBlockSize, when set, are written onto
// the request by MountOne as a block device export.
mountOneBlockDevice string
mountOneBlockSize uint64
// onMount fires synchronously inside Mount before the configured // onMount fires synchronously inside Mount before the configured
// mountErr is returned. Used by lifecycle tests to simulate a // mountErr is returned. Used by lifecycle tests to simulate a
// concurrent terminate flipping VM.Status while Mount is in flight. // concurrent terminate flipping VM.Status while Mount is in flight.
@@ -181,6 +185,7 @@ func (f *fakeVolumeMounter) MountOne(_ context.Context, _ string, req *EBSReques
f.mountedOne = append(f.mountedOne, req.Name) f.mountedOne = append(f.mountedOne, req.Name)
mountOneErr := f.mountOneErr mountOneErr := f.mountOneErr
uri := f.mountOneURI uri := f.mountOneURI
blockDevice, blockSize := f.mountOneBlockDevice, f.mountOneBlockSize
f.mu.Unlock() f.mu.Unlock()
if mountOneErr != nil { if mountOneErr != nil {
return mountOneErr return mountOneErr
@@ -188,6 +193,10 @@ func (f *fakeVolumeMounter) MountOne(_ context.Context, _ string, req *EBSReques
if uri != "" { if uri != "" {
req.NBDURI = uri req.NBDURI = uri
} }
if blockDevice != "" {
req.BlockDevice = blockDevice
req.BlockSizeBytes = blockSize
}
return nil return nil
} }
+11 -3
View File
@@ -523,14 +523,22 @@ func isInstanceProcessRunning(instance *VM) bool {
} }
// AreVolumeSocketsValid dials each NBD Unix socket to confirm viperblock is // AreVolumeSocketsValid dials each NBD Unix socket to confirm viperblock is
// still listening. A stat-only check is insufficient since viperblock may restart // still listening, and checks each block export's device still exists. TCP,
// leaving stale socket files. TCP and unparseable URIs are treated as valid. // unparseable and invalid exports are treated as valid.
func AreVolumeSocketsValid(instance *VM) bool { func AreVolumeSocketsValid(instance *VM) bool {
instance.EBSRequests.Mu.Lock() instance.EBSRequests.Mu.Lock()
defer instance.EBSRequests.Mu.Unlock() defer instance.EBSRequests.Mu.Unlock()
for _, req := range instance.EBSRequests.Requests { for _, req := range instance.EBSRequests.Requests {
if req.NBDURI == "" { kind, _ := classifyVolumeExport(req)
if kind == volumeExportBlock {
if err := checkBlockDevice(req.BlockDevice); err != nil {
slog.Debug("Block device missing", "volume", req.Name, "device", req.BlockDevice, "err", err)
return false
}
continue
}
if kind != volumeExportNBD {
continue continue
} }
serverType, sockPath, _, _, err := nbd.ParseNBDURI(req.NBDURI) serverType, sockPath, _, _, err := nbd.ParseNBDURI(req.NBDURI)
@@ -82,6 +82,22 @@ func TestAreVolumeSocketsValid_OneOfManyMissing(t *testing.T) {
"a single dead socket is enough to invalidate the whole instance") "a single dead socket is enough to invalidate the whole instance")
} }
// TestAreVolumeSocketsValid_BlockDevice checks a block export is judged by its
// device node's presence, with no NBD probe.
func TestAreVolumeSocketsValid_BlockDevice(t *testing.T) {
present := filepath.Join(t.TempDir(), "dev-present")
require.NoError(t, os.WriteFile(present, nil, 0o600))
missing := filepath.Join(t.TempDir(), "dev-missing")
blockVM := func(path string) *VM {
v := &VM{ID: "i-blk", Status: StateRunning}
v.EBSRequests.Requests = []EBSRequest{{Name: "vol-blk", BlockDevice: path, BlockSizeBytes: 1 << 20}}
return v
}
assert.True(t, AreVolumeSocketsValid(blockVM(present)), "a present block device is valid")
assert.False(t, AreVolumeSocketsValid(blockVM(missing)), "a vanished block device is the orphan-QEMU signal")
}
// classifyTestManager wires a manager with an in-memory StateStore plus // classifyTestManager wires a manager with an in-memory StateStore plus
// the InstanceTypes resolver and Resources controller that // the InstanceTypes resolver and Resources controller that
// classifyRestoredInstances consults. Returns the manager and the // classifyRestoredInstances consults. Returns the manager and the
+29 -17
View File
@@ -119,17 +119,36 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string)
return "", fmt.Errorf("mount volume %s: %w", volumeID, err) return "", fmt.Errorf("mount volume %s: %w", volumeID, err)
} }
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)
return "", fmt.Errorf("parse NBDURI: %w", err)
}
serverArg := NBDServerOpts{Type: serverType, Path: socketPath, Host: nbdHost, Port: nbdPort}.QMPArg()
// Shared with buildDrives' cold-boot path so a relaunched volume's block // Shared with buildDrives' cold-boot path so a relaunched volume's block
// graph uses exactly the same names DetachVolume addresses. // graph uses exactly the same names DetachVolume addresses.
nodeName := VolumeNodeName(volumeID) nodeName := VolumeNodeName(volumeID)
kind, err := classifyVolumeExport(ebsRequest)
if err != nil {
slog.ErrorContext(ctx, "AttachVolume: invalid volume export", "volumeId", volumeID, "err", err)
_ = m.rollbackUnmount(ctx, instance.AccountID, ebsRequest)
return "", fmt.Errorf("volume export: %w", err)
}
var blockdevArgs map[string]any
if kind == volumeExportBlock {
blockdevArgs = HostDeviceOpts{Path: ebsRequest.BlockDevice, SizeBytes: ebsRequest.BlockSizeBytes}.QMPArgs(nodeName)
} else {
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)
return "", fmt.Errorf("parse NBDURI: %w", err)
}
blockdevArgs = map[string]any{
"node-name": nodeName,
"driver": "nbd",
"server": NBDServerOpts{Type: serverType, Path: socketPath, Host: nbdHost, Port: nbdPort}.QMPArg(),
"export": "",
"read-only": false,
"reconnect-delay": NBDReconnectDelaySeconds,
}
}
deviceID := VolumeDeviceID(volumeID) deviceID := VolumeDeviceID(volumeID)
iothreadID := VolumeIOThreadID(volumeID) iothreadID := VolumeIOThreadID(volumeID)
@@ -163,15 +182,8 @@ func (m *Manager) AttachVolume(ctx context.Context, id, volumeID, device string)
} }
if _, err := sendQMPCommand(ctx, instance.QMPClient, qmp.QMPCommand{ if _, err := sendQMPCommand(ctx, instance.QMPClient, qmp.QMPCommand{
Execute: "blockdev-add", Execute: "blockdev-add",
Arguments: map[string]any{ Arguments: blockdevArgs,
"node-name": nodeName,
"driver": "nbd",
"server": serverArg,
"export": "",
"read-only": false,
"reconnect-delay": NBDReconnectDelaySeconds,
},
}, instance.ID); err != nil { }, instance.ID); err != nil {
slog.ErrorContext(ctx, "AttachVolume: QMP blockdev-add failed", "volumeId", volumeID, "err", err) slog.ErrorContext(ctx, "AttachVolume: QMP blockdev-add failed", "volumeId", volumeID, "err", err)
m.delIothreadBestEffort(ctx, instance, iothreadID, volumeID) m.delIothreadBestEffort(ctx, instance, iothreadID, volumeID)
@@ -943,6 +943,91 @@ func TestAttachVolume_NBDURIParseFailure_TriggersUnmountOne(t *testing.T) {
assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne) assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne)
} }
// TestAttachVolume_BlockDeviceBlockdevAdd pins the blockdev-add a block export
// sends: a sized raw node over a host_device file node, under the same node
// name the NBD path uses, followed by the usual device_add.
func TestAttachVolume_BlockDeviceBlockdevAdd(t *testing.T) {
recorder := &qmpRecorder{}
qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
recorder.record(cmd)
return nil
})
defer cancel()
mounter := &fakeVolumeMounter{mountOneBlockDevice: "/dev/sdx", mountOneBlockSize: 10 << 30}
m, v := attachVolumeRunningInstance(t, qmpClient, mounter)
_, err := m.AttachVolume(t.Context(), "i-1", "vol-1", "/dev/sdf")
require.NoError(t, err)
recorder.mu.Lock()
var blockdevAdd *qmp.QMPCommand
for i := range recorder.commands {
if recorder.commands[i].Execute == "blockdev-add" {
blockdevAdd = &recorder.commands[i]
}
}
recorder.mu.Unlock()
require.NotNil(t, blockdevAdd)
got, err := json.Marshal(blockdevAdd.Arguments)
require.NoError(t, err)
assert.JSONEq(t, `{
"node-name":"nbd-vol-1", "driver":"raw", "size":10737418240,
"file":{"driver":"host_device", "filename":"/dev/sdx", "aio":"native", "cache":{"direct":true}}
}`, string(got))
assert.Equal(t, []string{"object-add", "blockdev-add", "device_add"}, recorder.executes()[:3])
v.EBSRequests.Mu.Lock()
defer v.EBSRequests.Mu.Unlock()
require.Len(t, v.EBSRequests.Requests, 1)
assert.Equal(t, "/dev/sdx", v.EBSRequests.Requests[0].BlockDevice)
assert.Equal(t, uint64(10<<30), v.EBSRequests.Requests[0].BlockSizeBytes)
assert.Empty(t, v.EBSRequests.Requests[0].NBDURI)
}
// TestAttachVolume_InvalidExport_TriggersUnmountOne covers a mount that
// returns both exports: AttachVolume must unmount before any QMP traffic.
func TestAttachVolume_InvalidExport_TriggersUnmountOne(t *testing.T) {
recorder := &qmpRecorder{}
qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
recorder.record(cmd)
return nil
})
defer cancel()
mounter := &fakeVolumeMounter{mountOneURI: "nbd:unix:/tmp/test.sock", mountOneBlockDevice: "/dev/sdx", mountOneBlockSize: 1 << 20}
m, _ := attachVolumeRunningInstance(t, qmpClient, mounter)
_, err := m.AttachVolume(t.Context(), "i-1", "vol-1", "/dev/sdf")
require.Error(t, err)
assert.Contains(t, err.Error(), "both NBDURI and BlockDevice set")
assert.Empty(t, recorder.executes())
assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne)
}
// TestAttachVolume_BlockDeviceBlockdevAddFailure_TriggersUnmountOne checks a
// block export unwinds a failed blockdev-add exactly as an NBD one does.
func TestAttachVolume_BlockDeviceBlockdevAddFailure_TriggersUnmountOne(t *testing.T) {
recorder := &qmpRecorder{}
qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
recorder.record(cmd)
if cmd.Execute == "blockdev-add" {
return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "Could not open '/dev/sdx'"}}
}
return nil
})
defer cancel()
mounter := &fakeVolumeMounter{mountOneBlockDevice: "/dev/sdx", mountOneBlockSize: 1 << 20}
m, _ := attachVolumeRunningInstance(t, qmpClient, mounter)
_, err := m.AttachVolume(t.Context(), "i-1", "vol-1", "/dev/sdf")
require.Error(t, err)
assert.Contains(t, err.Error(), "blockdev-add")
assert.Equal(t, []string{"object-add", "blockdev-add", "object-del"}, recorder.executes())
assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne)
}
// TestAttachVolume_ObjectAddFailure_TriggersUnmountOne covers QMP iothread // TestAttachVolume_ObjectAddFailure_TriggersUnmountOne covers QMP iothread
// object-add failure: AttachVolume must unmount and bail before issuing // object-add failure: AttachVolume must unmount and bail before issuing
// blockdev-add (no further QMP traffic past the failed object-add). // blockdev-add (no further QMP traffic past the failed object-add).
@@ -104,8 +104,8 @@ func TestLive_QEMUNBDLifecycle(t *testing.T) {
pub, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: nodeID}) pub, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID, NodeID: nodeID})
require.NoError(t, err, "publish volume") require.NoError(t, err, "publish volume")
t.Logf("published volume %s to node %s: %s", pub.VolumeID, pub.NodeID, pub.NBDURI) t.Logf("published volume %s to node %s: %s", pub.VolumeID, pub.NodeID, liveNBDURI(t, pub))
sockPath := requireUnixSocketPath(t, pub.NBDURI) sockPath := requireUnixSocketPath(t, liveNBDURI(t, pub))
writePattern(t, sockPath, pattern, liveVolumeBytes) writePattern(t, sockPath, pattern, liveVolumeBytes)
readAndVerifyPattern(t, sockPath, pattern, liveVolumeBytes) readAndVerifyPattern(t, sockPath, pattern, liveVolumeBytes)
@@ -129,8 +129,8 @@ func TestLive_QEMUNBDLifecycle(t *testing.T) {
restoredPub, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID, NodeID: nodeID}) restoredPub, err := provider.PublishVolume(ctx, ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewVersioned(), VolumeID: restoredVolumeID, NodeID: nodeID})
require.NoError(t, err, "publish restored volume") require.NoError(t, err, "publish restored volume")
t.Logf("published restored volume %s to node %s: %s", restoredPub.VolumeID, restoredPub.NodeID, restoredPub.NBDURI) t.Logf("published restored volume %s to node %s: %s", restoredPub.VolumeID, restoredPub.NodeID, liveNBDURI(t, restoredPub))
restoredSockPath := requireUnixSocketPath(t, restoredPub.NBDURI) restoredSockPath := requireUnixSocketPath(t, liveNBDURI(t, restoredPub))
// The real proof: bytes written before the snapshot survive both the // The real proof: bytes written before the snapshot survive both the
// snapshot and the restore into an independent volume. // snapshot and the restore into an independent volume.
@@ -151,6 +151,15 @@ func cleanupBestEffort(t *testing.T, step string, fn func() error) {
} }
} }
// liveNBDURI requires pub to carry an nbd export, which is all qemunbdd
// publishes, and returns its URI.
func liveNBDURI(t *testing.T, pub *ebsprovider.PublishedVolume) string {
t.Helper()
require.Equal(t, ebsprovider.ExportKindNBD, pub.Export.Kind, "qemunbdd must publish an nbd export")
require.NotNil(t, pub.Export.NBD, "nbd export has no nbd details")
return pub.Export.NBD.NBDURI
}
// requireUnixSocketPath asserts nbdURI is a nbd:unix: URI and returns its // requireUnixSocketPath asserts nbdURI is a nbd:unix: URI and returns its
// socket path, parsed with the same helper the daemon uses to format it // socket path, parsed with the same helper the daemon uses to format it
// rather than by slicing the string. // rather than by slicing the string.
+2 -2
View File
@@ -688,7 +688,7 @@ func (p *Provider) PublishVolume(ctx context.Context, req ebsprovider.PublishVol
if pub.nodeID != req.NodeID { if pub.nodeID != req.NodeID {
return nil, fmt.Errorf("%w: volume %s is published to %s", ebsprovider.ErrVolumeInUse, req.VolumeID, pub.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: nbd.FormatNBDSocketURI(pub.socketPath)}, nil return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: pub.nodeID, Export: ebsprovider.NewNBDExport(nbd.FormatNBDSocketURI(pub.socketPath))}, nil
} }
sockPath := p.socketPath(req.VolumeID) sockPath := p.socketPath(req.VolumeID)
@@ -718,7 +718,7 @@ func (p *Provider) PublishVolume(ctx context.Context, req ebsprovider.PublishVol
pid := readPID(pidPath) pid := readPID(pidPath)
p.published[req.VolumeID] = &publication{nodeID: req.NodeID, socketPath: sockPath, pid: pid} p.published[req.VolumeID] = &publication{nodeID: req.NodeID, socketPath: sockPath, pid: pid}
return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, NBDURI: nbd.FormatNBDSocketURI(sockPath)}, nil return &ebsprovider.PublishedVolume{VolumeID: req.VolumeID, NodeID: req.NodeID, Export: ebsprovider.NewNBDExport(nbd.FormatNBDSocketURI(sockPath))}, nil
} }
// readPID best-effort reads the PID qemu-nbd's --fork wrote to pidPath. A // readPID best-effort reads the PID qemu-nbd's --fork wrote to pidPath. A
@@ -544,7 +544,7 @@ func TestPublishVolume_ArgvReadWrite(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, "vol-pub", pub.VolumeID) assert.Equal(t, "vol-pub", pub.VolumeID)
assert.Equal(t, "node-1", pub.NodeID) assert.Equal(t, "node-1", pub.NodeID)
assert.Equal(t, "nbd+unix:///?socket="+p.socketPath("vol-pub"), pub.NBDURI) assert.Equal(t, ebsprovider.NewNBDExport("nbd+unix:///?socket="+p.socketPath("vol-pub")), pub.Export)
calls := fr.callsSnapshot() calls := fr.callsSnapshot()
require.Len(t, calls, 1) require.Len(t, calls, 1)