mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-24 05:04:03 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| ab93d1fdcb | |||
| 3a79c1b866 | |||
| 8eefbc405f |
@@ -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) {
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
@@ -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")
|
||||||
|
|||||||
@@ -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{
|
||||||
|
|||||||
@@ -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 {
|
||||||
|
|||||||
@@ -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
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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",
|
||||||
|
|||||||
@@ -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
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
Reference in New Issue
Block a user