Jamesmoore/arch 224 silo support device groups

pkg/storage/config/silo.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"errors"
 	"fmt"
 	"os"
 	"strconv"
@@ -123,6 +124,25 @@ func (ds *DeviceSchema) Encode() []byte {
 	return f.Bytes()
 }
 
+func (ds *DeviceSchema) EncodeAsBlock() []byte {
+	f := hclwrite.NewEmptyFile()
+	block := gohcl.EncodeAsBlock(ds, "device")
+	f.Body().AppendBlock(block)
+	return f.Bytes()
+}
+
+func DecodeDeviceFromBlock(schema string) (*DeviceSchema, error) {
+	sf := &SiloSchema{}
+	err := sf.Decode([]byte(schema))
+	if err != nil {
+		return nil, err
+	}
+	if len(sf.Device) != 1 {
+		return nil, errors.New("more than one device in schema")
+	}
+	return sf.Device[0], nil
+}
+
 func (ds *DeviceSchema) Decode(schema string) error {
 	file, diag := hclsyntax.ParseConfig([]byte(schema), "", hcl.Pos{Line: 1, Column: 1})
 	if diag.HasErrors() {

pkg/storage/config/silo_test.go

@@ -66,3 +66,31 @@ func TestSiloConfig(t *testing.T) {
 	assert.NoError(t, err)
 	// TODO: Check data is as expected
 }
+
+func TestSiloConfigBlock(t *testing.T) {
+
+	schema := `device Disk0 {
+		size = "1G"
+		expose = true
+		system = "memory"
+	}
+
+	device Disk1 {
+		size = "2M"
+		system = "memory"
+	}
+	`
+
+	s := new(SiloSchema)
+	err := s.Decode([]byte(schema))
+	assert.NoError(t, err)
+
+	block0 := s.Device[0].EncodeAsBlock()
+
+	ds := &SiloSchema{}
+	err = ds.Decode(block0)
+	assert.NoError(t, err)
+
+	// Make sure it used the label
+	assert.Equal(t, ds.Device[0].Name, s.Device[0].Name)
+}

pkg/storage/devicegroup/README.md

@@ -0,0 +1,75 @@
+# Device Group
+
+The `DeviceGroup` combines some number of Silo devices into a single unit, which can then be migrated to another Silo instance.
+All internal concerns such as volatilityMonitor, waitingCache, as well as the new S3 assist, are now hidden from the consumer.
+
+## Creation
+
+There are two methods to create a `DeviceGroup`.
+
+### NewFromSchema
+
+This takes in an array of Silo device configs, and creates the devices. If `expose==true` then a corresponding NBD device will be created and attached.
+
+### NewFromProtocol
+
+This takes in a `protocol` and creates the devices as they are received from a sender.
+
+## Usage (Sending devices)
+
+Devices in a `DeviceGroup` are sent together, which allows Silo to optimize all aspects of the transfer.
+
+    // Create a device group from schema
+	dg, err := devicegroup.NewFromSchema(devices, log, siloMetrics)
+
+    // Start a migration
+	err := dg.StartMigrationTo(protocol)
+
+    // Migrate the data with max total concurrency 100
+	err = dg.MigrateAll(100, pHandler)
+
+    // Migrate any dirty blocks
+    // hooks gives some control over the dirty loop
+	err = dg.MigrateDirty(hooks)
+
+    // Send completion events for all devices
+    err = dg.Completed()
+
+    // Close everything
+    dg.CloseAll()
+
+Within the `MigrateDirty` there are a number of hooks we can use to control things. MigrateDirty will return once all devices have no more dirty data. You can of course then call MigrateDirty again eg for continuous sync.
+
+    type MigrateDirtyHooks struct {
+        PreGetDirty      func(name string) error
+        PostGetDirty     func(name string, blocks []uint) (bool, error)
+        PostMigrateDirty func(name string, blocks []uint) (bool, error)
+        Completed        func(name string)
+    }
+
+
+There is also support for sending global custom data. This would typically be done either in `pHandler` (The progress handler), or in one of the `MigrateDirty` hooks.
+
+    pHandler := func(ps []*migrator.MigrationProgress) {
+        // Do some test here to see if enough data migrated
+
+        // If so, send a custom Authority Transfer event.
+        dg.SendCustomData(authorityTransferPacket)
+    }
+
+## Usage (Receiving devices)
+
+    // Create a DeviceGroup from protocol
+    // tweak func allows us to modify the schema, eg pathnames
+	dg, err = NewFromProtocol(ctx, protocol, tweak, nil, nil)
+
+    // Handle any custom data events
+    // For example resume the VM here.
+	go dg.HandleCustomData(func(data []byte) {
+        // We got sent some custom data!
+    })
+
+    // Wait for migration completion
+	dg.WaitForCompletion()
+
+Once a `DeviceGroup` is has been created and migration is completed, you can then send the devices somewhere else with `StartMigration(protocol)`.

pkg/storage/devicegroup/device_group.go

@@ -0,0 +1,139 @@
+package devicegroup
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/loopholelabs/logging/types"
+	"github.com/loopholelabs/silo/pkg/storage"
+	"github.com/loopholelabs/silo/pkg/storage/blocks"
+	"github.com/loopholelabs/silo/pkg/storage/config"
+	"github.com/loopholelabs/silo/pkg/storage/dirtytracker"
+	"github.com/loopholelabs/silo/pkg/storage/metrics"
+	"github.com/loopholelabs/silo/pkg/storage/migrator"
+	"github.com/loopholelabs/silo/pkg/storage/protocol"
+	"github.com/loopholelabs/silo/pkg/storage/protocol/packets"
+	"github.com/loopholelabs/silo/pkg/storage/volatilitymonitor"
+	"github.com/loopholelabs/silo/pkg/storage/waitingcache"
+)
+
+const volatilityExpiry = 30 * time.Minute
+const defaultBlockSize = 1024 * 1024
+
+var errNotSetup = errors.New("toProtocol not setup")
+
+type DeviceGroup struct {
+	log               types.Logger
+	met               metrics.SiloMetrics
+	ctx               context.Context
+	devices           []*DeviceInformation
+	controlProtocol   protocol.Protocol
+	incomingDevicesCh chan bool
+	progressLock      sync.Mutex
+	progress          map[string]*migrator.MigrationProgress
+}
+
+type DeviceInformation struct {
+	Size               uint64
+	BlockSize          uint64
+	NumBlocks          int
+	Schema             *config.DeviceSchema
+	Prov               storage.Provider
+	Storage            storage.LockableProvider
+	Exp                storage.ExposedStorage
+	Volatility         *volatilitymonitor.VolatilityMonitor
+	DirtyLocal         *dirtytracker.Local
+	DirtyRemote        *dirtytracker.Remote
+	To                 *protocol.ToProtocol
+	Orderer            *blocks.PriorityBlockOrder
+	Migrator           *migrator.Migrator
+	migrationError     chan error
+	WaitingCacheLocal  *waitingcache.Local
+	WaitingCacheRemote *waitingcache.Remote
+	EventHandler       func(e *packets.Event)
+}
+
+func (dg *DeviceGroup) GetDeviceSchema() []*config.DeviceSchema {
+	s := make([]*config.DeviceSchema, 0)
+	for _, di := range dg.devices {
+		s = append(s, di.Schema)
+	}
+	return s
+}
+
+func (dg *DeviceGroup) GetAllNames() []string {
+	names := make([]string, 0)
+	for _, di := range dg.devices {
+		names = append(names, di.Schema.Name)
+	}
+	return names
+}
+
+func (dg *DeviceGroup) GetDeviceInformationByName(name string) *DeviceInformation {
+	for _, di := range dg.devices {
+		if di.Schema.Name == name {
+			return di
+		}
+	}
+	return nil
+}
+
+func (dg *DeviceGroup) GetExposedDeviceByName(name string) storage.ExposedStorage {
+	for _, di := range dg.devices {
+		if di.Schema.Name == name && di.Exp != nil {
+			return di.Exp
+		}
+	}
+	return nil
+}
+
+func (dg *DeviceGroup) GetProviderByName(name string) storage.Provider {
+	for _, di := range dg.devices {
+		if di.Schema.Name == name {
+			return di.Prov
+		}
+	}
+	return nil
+}
+
+func (dg *DeviceGroup) GetBlockSizeByName(name string) int {
+	for _, di := range dg.devices {
+		if di.Schema.Name == name {
+			return int(di.BlockSize)
+		}
+	}
+	return -1
+}
+
+func (dg *DeviceGroup) CloseAll() error {
+	if dg.log != nil {
+		dg.log.Debug().Int("devices", len(dg.devices)).Msg("close device group")
+	}
+
+	var e error
+	for _, d := range dg.devices {
+		// Unlock the storage so nothing blocks here...
+		// If we don't unlock there may be pending nbd writes that can't be completed.
+		d.Storage.Unlock()
+
+		err := d.Prov.Close()
+		if err != nil {
+			if dg.log != nil {
+				dg.log.Error().Err(err).Msg("error closing device group storage provider")
+			}
+			e = errors.Join(e, err)
+		}
+		if d.Exp != nil {
+			err = d.Exp.Shutdown()
+			if err != nil {
+				if dg.log != nil {
+					dg.log.Error().Err(err).Msg("error closing device group exposed storage")
+				}
+				e = errors.Join(e, err)
+			}
+		}
+	}
+	return e
+}