First stab at refactoring mounter

Signed-off-by: Jimmy Moore <[email protected]>

cmd/drafter-mounter/main.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"flag"
+	"fmt"
 	"io"
 	"log"
 	"net"
@@ -15,9 +16,11 @@ import (
 	"sync"
 	"time"
 
+	"github.com/loopholelabs/drafter/pkg/common"
 	"github.com/loopholelabs/drafter/pkg/mounter"
 	"github.com/loopholelabs/drafter/pkg/packager"
 	"github.com/loopholelabs/goroutine-manager/pkg/manager"
+	"github.com/loopholelabs/silo/pkg/storage/migrator"
 )
 
 type CompositeDevices struct {
@@ -195,7 +198,7 @@ func main() {
 	)
 	defer goroutineManager.Wait()
 	defer goroutineManager.StopAllGoroutines()
-	defer goroutineManager.CreateBackgroundPanicCollector()()
+	//	defer goroutineManager.CreateBackgroundPanicCollector()()
 
 	bubbleSignals := false
 
@@ -445,13 +448,13 @@ l:
 
 			defer migratableMounter.Close()
 
-			migrateToDevices := []mounter.MigrateToDevice{}
+			migrateToDevices := []common.MigrateToDevice{}
 			for _, device := range devices {
 				if !device.MakeMigratable {
 					continue
 				}
 
-				migrateToDevices = append(migrateToDevices, mounter.MigrateToDevice{
+				migrateToDevices = append(migrateToDevices, common.MigrateToDevice{
 					Name: device.Name,
 
 					MaxDirtyBlocks: device.MaxDirtyBlocks,
@@ -530,9 +533,31 @@ l:
 					OnAllMigrationsCompleted: func() {
 						log.Println("Completed all device migrations")
 					},
+					OnProgress: func(p map[string]*migrator.MigrationProgress) {
+						totalSize := 0
+						totalDone := 0
+						for _, prog := range p {
+							totalSize += (prog.TotalBlocks * prog.BlockSize)
+							totalDone += (prog.ReadyBlocks * prog.BlockSize)
+						}
+
+						perc := float64(0.0)
+						if totalSize > 0 {
+							perc = float64(totalDone) * 100 / float64(totalSize)
+						}
+						// Report overall migration progress
+						log.Printf("# Overall migration Progress # (%d / %d) %.1f%%\n", totalDone, totalSize, perc)
+
+						// Report individual devices
+						for name, prog := range p {
+							log.Printf(" [%s] Progress Migrated Blocks (%d/%d)  %d ready %d total\n", name, prog.MigratedBlocks, prog.TotalBlocks, prog.ReadyBlocks, prog.TotalMigratedBlocks)
+						}
+
+					},
 				},
 			)
 		}(); err != nil {
+			fmt.Printf("ERROR %v\n", err)
 			panic(err)
 		}
 	}

cmd/drafter-peer/main.go

@@ -558,26 +558,13 @@ func main() {
 
 	log.Println("Migrating to", conn.RemoteAddr())
 
-	makeMigratableDevices := []mounter.MakeMigratableDevice{}
+	migrateToDevices := []common.MigrateToDevice{}
 	for _, device := range devices {
 		if !device.MakeMigratable || device.Shared {
 			continue
 		}
 
-		makeMigratableDevices = append(makeMigratableDevices, mounter.MakeMigratableDevice{
-			Name: device.Name,
-
-			Expiry: device.Expiry,
-		})
-	}
-
-	migrateToDevices := []mounter.MigrateToDevice{}
-	for _, device := range devices {
-		if !device.MakeMigratable || device.Shared {
-			continue
-		}
-
-		migrateToDevices = append(migrateToDevices, mounter.MigrateToDevice{
+		migrateToDevices = append(migrateToDevices, common.MigrateToDevice{
 			Name: device.Name,
 
 			MaxDirtyBlocks: device.MaxDirtyBlocks,

hack/migrate_mount.sh

@@ -0,0 +1,14 @@
+sudo drafter-mounter --laddr '' --raddr 'localhost:1337' --devices '[
+  {
+    "name": "testdata",
+    "base": "testdata2",
+    "blockSize": 1048576,
+    "expiry": 1000000000,
+    "maxDirtyBlocks": 200,
+    "minCycles": 5,
+    "maxCycles": 20,
+    "cycleThrottle": 500000000,
+    "makeMigratable": true,
+    "shared": false
+  }
+]'

hack/mount.sh

@@ -0,0 +1,14 @@
+sudo drafter-mounter --raddr '' --laddr 'localhost:1337' --devices '[
+  {
+    "name": "testdata",
+    "base": "testdata",
+    "blockSize": 1048576,
+    "expiry": 1000000000,
+    "maxDirtyBlocks": 200,
+    "minCycles": 5,
+    "maxCycles": 20,
+    "cycleThrottle": 500000000,
+    "makeMigratable": true,
+    "shared": false
+  }
+]'

pkg/common/dirty_manager.go

@@ -5,10 +5,13 @@ import (
 	"sync"
 	"time"
 
-	"github.com/loopholelabs/drafter/pkg/mounter"
 	"github.com/loopholelabs/drafter/pkg/packager"
 )
 
+var (
+	ErrCouldNotSuspendAndMsyncVM = errors.New("could not suspend and msync VM")
+)
+
 type DeviceStatus struct {
 	TotalCycles                   int
 	CycleThrottle                 time.Duration
@@ -168,7 +171,7 @@ func (dm *DirtyManager) PostMigrateDirty(name string, blocks []uint) (bool, erro
 			dm.suspendLock.Unlock()
 
 			if err != nil {
-				return true, errors.Join(mounter.ErrCouldNotSuspendAndMsyncVM, err)
+				return true, errors.Join(ErrCouldNotSuspendAndMsyncVM, err)
 			}
 		} else {
 			dm.suspendLock.Unlock()

pkg/common/migrations.go

@@ -9,8 +9,8 @@ import (
 	"path/filepath"
 	"strings"
 	"syscall"
+	"time"
 
-	"github.com/loopholelabs/drafter/pkg/mounter"
 	"github.com/loopholelabs/drafter/pkg/snapshotter"
 	"github.com/loopholelabs/logging/types"
 	"github.com/loopholelabs/silo/pkg/storage/config"
@@ -22,6 +22,23 @@ import (
 	"golang.org/x/sys/unix"
 )
 
+var (
+	ErrCouldNotCreateDeviceDirectory  = errors.New("could not create device directory")
+	ErrCouldNotGetBaseDeviceStat      = errors.New("could not get base device statistics")
+	ErrCouldNotCreateOverlayDirectory = errors.New("could not create overlay directory")
+	ErrCouldNotCreateStateDirectory   = errors.New("could not create state directory")
+)
+
+type MigrateToDevice struct {
+	Name string `json:"name"`
+
+	MaxDirtyBlocks int `json:"maxDirtyBlocks"`
+	MinCycles      int `json:"minCycles"`
+	MaxCycles      int `json:"maxCycles"`
+
+	CycleThrottle time.Duration `json:"cycleThrottle"`
+}
+
 type MigrateFromDevice struct {
 	Name      string `json:"name"`
 	Base      string `json:"base"`
@@ -38,6 +55,9 @@ var (
 
 // expose a Silo Device as a file within the vm directory
 func ExposeSiloDeviceAsFile(vmpath string, name string, devicePath string) error {
+	if vmpath == "" {
+		return nil
+	}
 	deviceInfo, err := os.Stat(devicePath)
 	if err != nil {
 		return errors.Join(snapshotter.ErrCouldNotGetDeviceStat, err)
@@ -64,7 +84,7 @@ func ExposeSiloDeviceAsFile(vmpath string, name string, devicePath string) error
 func CreateSiloDevSchema(i *MigrateFromDevice) (*config.DeviceSchema, error) {
 	stat, err := os.Stat(i.Base)
 	if err != nil {
-		return nil, errors.Join(mounter.ErrCouldNotGetBaseDeviceStat, err)
+		return nil, errors.Join(ErrCouldNotGetBaseDeviceStat, err)
 	}
 
 	ds := &config.DeviceSchema{
@@ -76,19 +96,19 @@ func CreateSiloDevSchema(i *MigrateFromDevice) (*config.DeviceSchema, error) {
 	if strings.TrimSpace(i.Overlay) == "" || strings.TrimSpace(i.State) == "" {
 		err := os.MkdirAll(filepath.Dir(i.Base), os.ModePerm)
 		if err != nil {
-			return nil, errors.Join(mounter.ErrCouldNotCreateDeviceDirectory, err)
+			return nil, errors.Join(ErrCouldNotCreateDeviceDirectory, err)
 		}
 		ds.System = "file"
 		ds.Location = i.Base
 	} else {
 		err := os.MkdirAll(filepath.Dir(i.Overlay), os.ModePerm)
 		if err != nil {
-			return nil, errors.Join(mounter.ErrCouldNotCreateOverlayDirectory, err)
+			return nil, errors.Join(ErrCouldNotCreateOverlayDirectory, err)
 		}
 
 		err = os.MkdirAll(filepath.Dir(i.State), os.ModePerm)
 		if err != nil {
-			return nil, errors.Join(mounter.ErrCouldNotCreateStateDirectory, err)
+			return nil, errors.Join(ErrCouldNotCreateStateDirectory, err)
 		}
 
 		ds.System = "sparsefile"
@@ -114,19 +134,19 @@ func CreateIncomingSiloDevSchema(i *MigrateFromDevice, schema *config.DeviceSche
 	if strings.TrimSpace(i.Overlay) == "" || strings.TrimSpace(i.State) == "" {
 		err := os.MkdirAll(filepath.Dir(i.Base), os.ModePerm)
 		if err != nil {
-			return nil, errors.Join(mounter.ErrCouldNotCreateDeviceDirectory, err)
+			return nil, errors.Join(ErrCouldNotCreateDeviceDirectory, err)
 		}
 		ds.System = "file"
 		ds.Location = i.Base
 	} else {
 		err := os.MkdirAll(filepath.Dir(i.Overlay), os.ModePerm)
 		if err != nil {
-			return nil, errors.Join(mounter.ErrCouldNotCreateOverlayDirectory, err)
+			return nil, errors.Join(ErrCouldNotCreateOverlayDirectory, err)
 		}
 
 		err = os.MkdirAll(filepath.Dir(i.State), os.ModePerm)
 		if err != nil {
-			return nil, errors.Join(mounter.ErrCouldNotCreateStateDirectory, err)
+			return nil, errors.Join(ErrCouldNotCreateStateDirectory, err)
 		}
 
 		ds.System = "sparsefile"
@@ -276,7 +296,7 @@ func MigrateFromPipe(log types.Logger, met metrics.SiloMetrics, vmpath string,
  */
 func MigrateToPipe(ctx context.Context, readers []io.Reader, writers []io.Writer,
 	dg *devicegroup.DeviceGroup, concurrency int, onProgress func(p map[string]*migrator.MigrationProgress),
-	vmState *VMStateMgr, devices []mounter.MigrateToDevice, getCustomPayload func() []byte) error {
+	vmState *VMStateMgr, devices []MigrateToDevice, getCustomPayload func() []byte) error {
 
 	// Create a protocol for use by Silo
 	pro := protocol.NewRW(ctx, readers, writers, nil)

pkg/common/migrations_test.go

@@ -14,7 +14,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/loopholelabs/drafter/pkg/mounter"
 	"github.com/loopholelabs/silo/pkg/storage/config"
 	"github.com/loopholelabs/silo/pkg/storage/devicegroup"
 	"github.com/loopholelabs/silo/pkg/storage/migrator"
@@ -144,7 +143,7 @@ func TestMigrateFromFsThenBetween(t *testing.T) {
 			fmt.Printf("Progress...\n")
 		}
 
-		devices := []mounter.MigrateToDevice{
+		devices := []MigrateToDevice{
 			{
 				Name:           "test",
 				MaxDirtyBlocks: 10,

pkg/common/vm_state.go

@@ -41,6 +41,18 @@ func NewVMStateMgr(ctx context.Context,
 	}
 }
 
+func NewDummyVMStateMgr(ctx context.Context) *VMStateMgr {
+	return &VMStateMgr{
+		ctx:             ctx,
+		suspendFunc:     func(context.Context, time.Duration) error { return nil },
+		suspendTimeout:  10 * time.Second,
+		msyncFunc:       func(context.Context) error { return nil },
+		onBeforeSuspend: func() {},
+		onAfterSuspend:  func() {},
+		suspendedCh:     make(chan struct{}),
+	}
+}
+
 func (sm *VMStateMgr) GetSuspsnededVMCh() chan struct{} {
 	return sm.suspendedCh
 }