feat: establish NekoNest Cloud control and relay
This commit is contained in:
@@ -0,0 +1,175 @@
|
||||
package migration
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/klarkxy/nekonest-cloud/relay/internal/controlplane"
|
||||
"github.com/klarkxy/nekonest-cloud/relay/internal/tenantbackup"
|
||||
"github.com/klarkxy/nekonest-cloud/relay/internal/tenantfs"
|
||||
)
|
||||
|
||||
type Registry interface {
|
||||
Quiesce(tenantID string, generation int64) (tenantfs.Paths, error)
|
||||
}
|
||||
|
||||
type ControlPlane interface {
|
||||
AdvanceMigration(context.Context, controlplane.MigrationAdvance) error
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
DataRoot string
|
||||
BackupRoot string
|
||||
Registry Registry
|
||||
ControlPlane ControlPlane
|
||||
MaxConcurrent int
|
||||
Now func() time.Time
|
||||
Logger *slog.Logger
|
||||
}
|
||||
|
||||
type Manager struct {
|
||||
config Config
|
||||
mu sync.Mutex
|
||||
inFlight map[string]struct{}
|
||||
semaphore chan struct{}
|
||||
wait sync.WaitGroup
|
||||
}
|
||||
|
||||
func New(config Config) (*Manager, error) {
|
||||
if config.DataRoot == "" || config.BackupRoot == "" || config.Registry == nil || config.ControlPlane == nil {
|
||||
return nil, errors.New("migration manager requires data, backup, registry, and control-plane ports")
|
||||
}
|
||||
if config.MaxConcurrent <= 0 {
|
||||
config.MaxConcurrent = 2
|
||||
}
|
||||
if config.MaxConcurrent > 8 {
|
||||
return nil, errors.New("migration concurrency is unreasonably high")
|
||||
}
|
||||
if config.Now == nil {
|
||||
config.Now = time.Now
|
||||
}
|
||||
if config.Logger == nil {
|
||||
config.Logger = slog.Default()
|
||||
}
|
||||
return &Manager{
|
||||
config: config, inFlight: make(map[string]struct{}),
|
||||
semaphore: make(chan struct{}, config.MaxConcurrent),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Handle starts bounded, de-duplicated migration work returned by a trusted
|
||||
// heartbeat. A later heartbeat safely retries any stage that did not commit.
|
||||
func (manager *Manager) Handle(ctx context.Context, assignments []controlplane.MigrationAssignment) {
|
||||
for _, assignment := range assignments {
|
||||
manager.mu.Lock()
|
||||
if _, exists := manager.inFlight[assignment.MigrationID]; exists {
|
||||
manager.mu.Unlock()
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case manager.semaphore <- struct{}{}:
|
||||
manager.inFlight[assignment.MigrationID] = struct{}{}
|
||||
manager.wait.Add(1)
|
||||
manager.mu.Unlock()
|
||||
go func(assignment controlplane.MigrationAssignment) {
|
||||
defer func() {
|
||||
<-manager.semaphore
|
||||
manager.mu.Lock()
|
||||
delete(manager.inFlight, assignment.MigrationID)
|
||||
manager.mu.Unlock()
|
||||
manager.wait.Done()
|
||||
}()
|
||||
if err := manager.process(ctx, assignment); err != nil && ctx.Err() == nil {
|
||||
manager.config.Logger.Error("Relay migration stage failed",
|
||||
"migration_id", assignment.MigrationID, "tenant_id", assignment.TenantID,
|
||||
"state", assignment.State, "error", err)
|
||||
}
|
||||
}(assignment)
|
||||
default:
|
||||
manager.mu.Unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (manager *Manager) process(ctx context.Context, assignment controlplane.MigrationAssignment) error {
|
||||
if assignment.MigrationID == "" || assignment.TenantID == "" {
|
||||
return errors.New("migration assignment is incomplete")
|
||||
}
|
||||
switch assignment.State {
|
||||
case "quiescing":
|
||||
if assignment.Role != "source" {
|
||||
return errors.New("quiescing assignment did not target the source")
|
||||
}
|
||||
if _, err := manager.config.Registry.Quiesce(assignment.TenantID, assignment.SourceGeneration); err != nil {
|
||||
return manager.fail(ctx, assignment, "relay_quiesce_failed", err)
|
||||
}
|
||||
backup, err := tenantbackup.Create(
|
||||
ctx, manager.config.DataRoot, manager.config.BackupRoot,
|
||||
assignment.TenantID, assignment.SourceGeneration, manager.config.Now(),
|
||||
)
|
||||
if err != nil {
|
||||
return manager.fail(ctx, assignment, "relay_backup_failed", err)
|
||||
}
|
||||
return manager.config.ControlPlane.AdvanceMigration(ctx, controlplane.MigrationAdvance{
|
||||
MigrationID: assignment.MigrationID, Action: "quiesced",
|
||||
BackupRef: backup.BackupRef, ManifestSHA256: backup.ManifestSHA256,
|
||||
})
|
||||
case "copying":
|
||||
if assignment.Role != "target" || assignment.BackupRef == "" || assignment.ManifestSHA256 == "" {
|
||||
return errors.New("copying assignment is incomplete")
|
||||
}
|
||||
backupPath, err := tenantbackup.ResolveReference(manager.config.BackupRoot, assignment.BackupRef)
|
||||
if err != nil {
|
||||
return manager.fail(ctx, assignment, "relay_backup_unavailable", err)
|
||||
}
|
||||
if _, err := tenantbackup.Restore(
|
||||
ctx, backupPath, manager.config.DataRoot, assignment.TenantID,
|
||||
assignment.SourceGeneration, assignment.ManifestSHA256,
|
||||
); err != nil {
|
||||
return manager.fail(ctx, assignment, "relay_restore_failed", err)
|
||||
}
|
||||
return manager.config.ControlPlane.AdvanceMigration(ctx, controlplane.MigrationAdvance{
|
||||
MigrationID: assignment.MigrationID, Action: "copied",
|
||||
BackupRef: assignment.BackupRef, ManifestSHA256: assignment.ManifestSHA256,
|
||||
})
|
||||
case "switching":
|
||||
if assignment.Role != "target" {
|
||||
return errors.New("switching assignment did not target the destination")
|
||||
}
|
||||
return manager.config.ControlPlane.AdvanceMigration(ctx, controlplane.MigrationAdvance{
|
||||
MigrationID: assignment.MigrationID, Action: "switched",
|
||||
})
|
||||
case "draining":
|
||||
if assignment.Role != "target" || assignment.FinalizeAfter == "" {
|
||||
return errors.New("draining assignment is incomplete")
|
||||
}
|
||||
finalizeAfter, err := time.Parse(time.RFC3339Nano, assignment.FinalizeAfter)
|
||||
if err != nil {
|
||||
return fmt.Errorf("invalid finalize fence: %w", err)
|
||||
}
|
||||
if manager.config.Now().Before(finalizeAfter) {
|
||||
return nil
|
||||
}
|
||||
return manager.config.ControlPlane.AdvanceMigration(ctx, controlplane.MigrationAdvance{
|
||||
MigrationID: assignment.MigrationID, Action: "finalized",
|
||||
})
|
||||
default:
|
||||
return fmt.Errorf("unsupported migration state %q", assignment.State)
|
||||
}
|
||||
}
|
||||
|
||||
func (manager *Manager) fail(ctx context.Context, assignment controlplane.MigrationAssignment, code string, cause error) error {
|
||||
advanceErr := manager.config.ControlPlane.AdvanceMigration(ctx, controlplane.MigrationAdvance{
|
||||
MigrationID: assignment.MigrationID, Action: "failed", ErrorCode: code,
|
||||
})
|
||||
if advanceErr != nil {
|
||||
return errors.Join(cause, fmt.Errorf("report migration failure: %w", advanceErr))
|
||||
}
|
||||
return cause
|
||||
}
|
||||
|
||||
func (manager *Manager) Wait() { manager.wait.Wait() }
|
||||
@@ -0,0 +1,103 @@
|
||||
package migration
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/klarkxy/nekonest-cloud/relay/internal/controlplane"
|
||||
"github.com/klarkxy/nekonest-cloud/relay/internal/tenantbackup"
|
||||
"github.com/klarkxy/nekonest-cloud/relay/internal/tenantfs"
|
||||
"github.com/klarkxy/nekonest-cloud/relay/internal/tenantstore"
|
||||
"github.com/klarkxy/nekonest/relaycore/protocol"
|
||||
)
|
||||
|
||||
type fakeRegistry struct {
|
||||
paths tenantfs.Paths
|
||||
err error
|
||||
}
|
||||
|
||||
func (registry fakeRegistry) Quiesce(string, int64) (tenantfs.Paths, error) {
|
||||
return registry.paths, registry.err
|
||||
}
|
||||
|
||||
type fakeControl struct {
|
||||
updates []controlplane.MigrationAdvance
|
||||
}
|
||||
|
||||
func (control *fakeControl) AdvanceMigration(_ context.Context, update controlplane.MigrationAdvance) error {
|
||||
control.updates = append(control.updates, update)
|
||||
return nil
|
||||
}
|
||||
|
||||
func createSource(t *testing.T, root, tenantID string) tenantfs.Paths {
|
||||
t.Helper()
|
||||
paths, err := tenantfs.Resolve(root, tenantID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
store, err := tenantstore.NewWithTransportMode(paths.Database, string(protocol.TransportSealed))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := store.RegisterDevice("host_0123456789abcdef0123456789abcdef", "Host", "linux"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return paths
|
||||
}
|
||||
|
||||
func TestManagerCreatesFencedBackupAndRestoresTarget(t *testing.T) {
|
||||
const tenantID = "tenant_0123456789abcdef0123456789abcdef"
|
||||
sourceRoot := t.TempDir()
|
||||
paths := createSource(t, sourceRoot, tenantID)
|
||||
backupRoot := t.TempDir()
|
||||
control := &fakeControl{}
|
||||
now := time.Date(2026, 8, 12, 12, 0, 0, 0, time.UTC)
|
||||
source, err := New(Config{
|
||||
DataRoot: sourceRoot, BackupRoot: backupRoot,
|
||||
Registry: fakeRegistry{paths: paths}, ControlPlane: control, Now: func() time.Time { return now },
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assignment := controlplane.MigrationAssignment{
|
||||
MigrationID: "migration_0123456789abcdef0123456789abcdef", TenantID: tenantID,
|
||||
Role: "source", State: "quiescing", SourceGeneration: 3, TargetGeneration: 4,
|
||||
}
|
||||
if err := source.process(context.Background(), assignment); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(control.updates) != 1 || control.updates[0].Action != "quiesced" || control.updates[0].BackupRef == "" {
|
||||
t.Fatalf("source update = %#v", control.updates)
|
||||
}
|
||||
backupPath, err := tenantbackup.ResolveReference(backupRoot, control.updates[0].BackupRef)
|
||||
if err != nil || filepath.Base(backupPath) == "" {
|
||||
t.Fatalf("backup path = %q err=%v", backupPath, err)
|
||||
}
|
||||
targetRoot := t.TempDir()
|
||||
targetControl := &fakeControl{}
|
||||
target, err := New(Config{
|
||||
DataRoot: targetRoot, BackupRoot: backupRoot,
|
||||
Registry: fakeRegistry{}, ControlPlane: targetControl, Now: func() time.Time { return now },
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assignment.Role = "target"
|
||||
assignment.State = "copying"
|
||||
assignment.BackupRef = control.updates[0].BackupRef
|
||||
assignment.ManifestSHA256 = control.updates[0].ManifestSHA256
|
||||
if err := target.process(context.Background(), assignment); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := target.process(context.Background(), assignment); err != nil {
|
||||
t.Fatal("copy retry was not idempotent:", err)
|
||||
}
|
||||
if len(targetControl.updates) != 2 || targetControl.updates[0].Action != "copied" {
|
||||
t.Fatalf("target updates = %#v", targetControl.updates)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user