mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
feat: use state machine for daemon (#16)
This commit is contained in:
+38
-57
@@ -575,7 +575,7 @@ func (d *Daemon) restoreInstances() {
|
||||
|
||||
instance := d.Instances.VMS[i]
|
||||
|
||||
if instance.Status == "terminated" {
|
||||
if instance.Status == vm.StateTerminated {
|
||||
slog.Info("Instance state is terminated, skipping", "instance", instance.ID)
|
||||
continue
|
||||
}
|
||||
@@ -813,7 +813,8 @@ func (d *Daemon) ClusterManager() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// WriteState writes the instance state to JetStream KV store (required)
|
||||
// WriteState writes the instance state to JetStream KV store (required).
|
||||
// It acquires d.Instances.Mu internally.
|
||||
func (d *Daemon) WriteState() error {
|
||||
if d.jsManager == nil {
|
||||
return fmt.Errorf("JetStream manager not initialized - cannot write state")
|
||||
@@ -825,6 +826,7 @@ func (d *Daemon) WriteState() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
// LoadState loads the instance state from JetStream KV store (required)
|
||||
func (d *Daemon) LoadState() error {
|
||||
if d.jsManager == nil {
|
||||
@@ -867,36 +869,36 @@ func (d *Daemon) SendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceId
|
||||
if _, ok := msg["event"]; ok {
|
||||
slog.Info("QMP event", "event", msg["event"])
|
||||
|
||||
var updatedStatus string
|
||||
// Look up the instance to handle state transitions
|
||||
d.Instances.Mu.Lock()
|
||||
instance, found := d.Instances.VMS[instanceId]
|
||||
d.Instances.Mu.Unlock()
|
||||
if !found {
|
||||
slog.Info("QMP event - instance not found", "id", instanceId)
|
||||
continue
|
||||
}
|
||||
|
||||
// Handle QMP events with state machine transitions.
|
||||
// TransitionState validates the current state atomically under lock,
|
||||
// so no pre-check is needed here.
|
||||
// RESET indicates a reboot but the VM remains running — no state change.
|
||||
switch msg["event"] {
|
||||
case "STOP":
|
||||
updatedStatus = "stopped"
|
||||
case "RESUME":
|
||||
updatedStatus = "resuming"
|
||||
case "RESET":
|
||||
updatedStatus = "restarting"
|
||||
case "POWERDOWN":
|
||||
updatedStatus = "powering_down"
|
||||
}
|
||||
|
||||
if updatedStatus != "" {
|
||||
|
||||
// Update the instance status
|
||||
d.Instances.Mu.Lock()
|
||||
instance, ok := d.Instances.VMS[instanceId]
|
||||
if !ok {
|
||||
slog.Info("QMP Status - Instance not found", "id", instanceId)
|
||||
continue
|
||||
if err := d.TransitionState(instance, vm.StateStopped); err != nil {
|
||||
slog.Error("QMP STOP event: transition failed", "instanceId", instanceId, "err", err)
|
||||
}
|
||||
|
||||
instance.Status = updatedStatus
|
||||
|
||||
d.Instances.VMS[instanceId] = instance
|
||||
d.Instances.Mu.Unlock()
|
||||
case "RESUME":
|
||||
if err := d.TransitionState(instance, vm.StateRunning); err != nil {
|
||||
slog.Error("QMP RESUME event: transition failed", "instanceId", instanceId, "err", err)
|
||||
}
|
||||
case "POWERDOWN":
|
||||
if err := d.TransitionState(instance, vm.StateStopping); err != nil {
|
||||
slog.Error("QMP POWERDOWN event: transition failed", "instanceId", instanceId, "err", err)
|
||||
}
|
||||
case "RESET":
|
||||
slog.Info("QMP RESET event", "instanceId", instanceId)
|
||||
}
|
||||
|
||||
// Optional: handle events here
|
||||
continue
|
||||
}
|
||||
if errObj, ok := msg["error"].(map[string]any); ok {
|
||||
@@ -1099,7 +1101,7 @@ func (d *Daemon) CreateQMPClient(instance *vm.VM) (err error) {
|
||||
status := instance.Status
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
if status == "stopping" || status == "stopped" || status == "shutting-down" || status == "terminated" || status == "error" {
|
||||
if status == vm.StateStopping || status == vm.StateStopped || status == vm.StateShuttingDown || status == vm.StateTerminated || status == vm.StateError {
|
||||
slog.Info("QMP heartbeat exiting - instance not running", "instance", instance.ID, "status", status)
|
||||
|
||||
// Close the QMP client connection if it exists
|
||||
@@ -1186,21 +1188,15 @@ func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) {
|
||||
}
|
||||
|
||||
// Step 9: Update the instance metadata for running state and volume attached
|
||||
// Marshal to a JSON file
|
||||
// Update state
|
||||
d.Instances.Mu.Lock()
|
||||
// Update to running state
|
||||
instance.Status = "running"
|
||||
|
||||
// Update EC2 Instance state for API compatibility
|
||||
if instance.Instance != nil {
|
||||
instance.Instance.State.SetCode(16) // 16 = running
|
||||
instance.Instance.State.SetName("running")
|
||||
}
|
||||
|
||||
d.Instances.VMS[instance.ID] = instance
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
if err := d.TransitionState(instance, vm.StateRunning); err != nil {
|
||||
slog.Error("Failed to transition instance to running", "instanceId", instance.ID, "err", err)
|
||||
return err
|
||||
}
|
||||
|
||||
// Step 10: Mark boot volumes as "in-use" now that instance is confirmed running
|
||||
instance.EBSRequests.Mu.Lock()
|
||||
for _, ebsReq := range instance.EBSRequests.Requests {
|
||||
@@ -1212,37 +1208,22 @@ func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) {
|
||||
}
|
||||
instance.EBSRequests.Mu.Unlock()
|
||||
|
||||
err = d.WriteState()
|
||||
|
||||
if err != nil {
|
||||
slog.Error("Failed to marshal launchVm", "err", err)
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// markInstanceFailed updates an instance status to indicate a failure during launch
|
||||
func (d *Daemon) markInstanceFailed(instance *vm.VM, reason string) {
|
||||
// Set state reason before transition (requires lock)
|
||||
d.Instances.Mu.Lock()
|
||||
defer d.Instances.Mu.Unlock()
|
||||
|
||||
instance.Status = "shutting-down"
|
||||
|
||||
// Update EC2 Instance state for API compatibility
|
||||
if instance.Instance != nil {
|
||||
instance.Instance.State.SetCode(32) // 32 = shutting-down
|
||||
instance.Instance.State.SetName("shutting-down")
|
||||
// Add state reason
|
||||
instance.Instance.StateReason = &ec2.StateReason{}
|
||||
instance.Instance.StateReason.SetCode("Server.InternalError")
|
||||
instance.Instance.StateReason.SetMessage(reason)
|
||||
}
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
d.Instances.VMS[instance.ID] = instance
|
||||
|
||||
if err := d.WriteState(); err != nil {
|
||||
slog.Error("markInstanceFailed failed to write state", "err", err)
|
||||
if err := d.TransitionState(instance, vm.StateShuttingDown); err != nil {
|
||||
slog.Error("markInstanceFailed transition failed", "instanceId", instance.ID, "err", err)
|
||||
}
|
||||
|
||||
slog.Info("Instance marked as failed", "instanceId", instance.ID, "reason", reason)
|
||||
|
||||
@@ -33,9 +33,8 @@ func (d *Daemon) handleEC2StartInstances(msg *nats.Msg) {
|
||||
|
||||
// Check if the instance is running on this node
|
||||
d.Instances.Mu.Lock()
|
||||
defer d.Instances.Mu.Unlock()
|
||||
|
||||
instance, ok := d.Instances.VMS[ec2StartInstance.InstanceID]
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
if !ok {
|
||||
slog.Error("EC2 Start Request - Instance not found", "instanceId", ec2StartInstanceResponse.InstanceID)
|
||||
@@ -57,7 +56,7 @@ func (d *Daemon) handleEC2StartInstances(msg *nats.Msg) {
|
||||
}
|
||||
}
|
||||
|
||||
// Launch the instance
|
||||
// Launch the instance (acquires d.Instances.Mu internally via TransitionState)
|
||||
err := d.LaunchInstance(instance)
|
||||
|
||||
if err != nil {
|
||||
@@ -68,7 +67,9 @@ func (d *Daemon) handleEC2StartInstances(msg *nats.Msg) {
|
||||
ec2StartInstanceResponse.Error = err.Error()
|
||||
} else {
|
||||
ec2StartInstanceResponse.InstanceID = instance.ID
|
||||
ec2StartInstanceResponse.Status = instance.Status
|
||||
d.Instances.Mu.Lock()
|
||||
ec2StartInstanceResponse.Status = string(instance.Status)
|
||||
d.Instances.Mu.Unlock()
|
||||
}
|
||||
|
||||
ec2StartInstanceResponse.Respond(msg)
|
||||
@@ -132,23 +133,13 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) {
|
||||
return
|
||||
}
|
||||
|
||||
// Update instance state
|
||||
// Update instance attributes (LaunchInstance already transitions to StateRunning)
|
||||
d.Instances.Mu.Lock()
|
||||
instance.Status = "running"
|
||||
instance.Attributes = command.Attributes
|
||||
if instance.Instance != nil {
|
||||
instance.Instance.State.SetCode(16) // 16 = running
|
||||
instance.Instance.State.SetName("running")
|
||||
}
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
slog.Info("Instance started", "instanceId", instance.ID)
|
||||
|
||||
// Write state to disk
|
||||
if writeErr := d.WriteState(); writeErr != nil {
|
||||
slog.Error("Failed to write state to disk", "err", writeErr)
|
||||
}
|
||||
|
||||
msg.Respond(fmt.Appendf(nil, `{"status":"running","instanceId":"%s"}`, instance.ID))
|
||||
return
|
||||
}
|
||||
@@ -157,31 +148,22 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) {
|
||||
if command.Attributes.StopInstance || command.Attributes.TerminateInstance {
|
||||
isTerminate := command.Attributes.TerminateInstance
|
||||
action := "Stopping"
|
||||
initialStatus := "stopping"
|
||||
finalStatus := "stopped"
|
||||
finalCode := int64(80)
|
||||
initialState := vm.StateStopping
|
||||
finalState := vm.StateStopped
|
||||
if isTerminate {
|
||||
action = "Terminating"
|
||||
initialStatus = "shutting-down"
|
||||
finalStatus = "terminated"
|
||||
finalCode = 48
|
||||
initialState = vm.StateShuttingDown
|
||||
finalState = vm.StateTerminated
|
||||
}
|
||||
|
||||
slog.Info(action+" instance", "id", command.ID)
|
||||
|
||||
// Update status to transitional state
|
||||
d.Instances.Mu.Lock()
|
||||
instance.Status = initialStatus
|
||||
if instance.Instance != nil {
|
||||
if isTerminate {
|
||||
instance.Instance.State.SetCode(32) // 32 = shutting-down
|
||||
instance.Instance.State.SetName("shutting-down")
|
||||
} else {
|
||||
instance.Instance.State.SetCode(64) // 64 = stopping
|
||||
instance.Instance.State.SetName("stopping")
|
||||
}
|
||||
// Transition to the initial transitional state
|
||||
if err := d.TransitionState(instance, initialState); err != nil {
|
||||
slog.Error("Failed to transition to "+string(initialState), "instanceId", instance.ID, "err", err)
|
||||
respondWithError(awserrors.ErrorServerInternal)
|
||||
return
|
||||
}
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
// Respond immediately - operation will complete in background
|
||||
// stopInstance() handles the QMP shutdown command, so we don't send it here
|
||||
@@ -191,28 +173,20 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) {
|
||||
go func(inst *vm.VM, attrs qmp.Attributes) {
|
||||
stopErr := d.stopInstance(map[string]*vm.VM{inst.ID: inst}, isTerminate)
|
||||
|
||||
d.Instances.Mu.Lock()
|
||||
if stopErr != nil {
|
||||
slog.Error("Failed to "+strings.ToLower(action)+" instance", "err", stopErr, "id", inst.ID)
|
||||
// On error, revert to previous state or mark as error
|
||||
inst.Status = "error"
|
||||
if inst.Instance != nil {
|
||||
inst.Instance.State.SetCode(0)
|
||||
inst.Instance.State.SetName("error")
|
||||
if err := d.TransitionState(inst, vm.StateError); err != nil {
|
||||
slog.Error("Failed to transition to error state", "instanceId", inst.ID, "err", err)
|
||||
}
|
||||
} else {
|
||||
inst.Status = finalStatus
|
||||
d.Instances.Mu.Lock()
|
||||
inst.Attributes = attrs
|
||||
if inst.Instance != nil {
|
||||
inst.Instance.State.SetCode(finalCode)
|
||||
inst.Instance.State.SetName(finalStatus)
|
||||
}
|
||||
slog.Info("Instance "+finalStatus, "id", inst.ID)
|
||||
}
|
||||
d.Instances.Mu.Unlock()
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
if writeErr := d.WriteState(); writeErr != nil {
|
||||
slog.Error("Failed to write state to disk", "err", writeErr)
|
||||
if err := d.TransitionState(inst, finalState); err != nil {
|
||||
slog.Error("Failed to transition to final state", "instanceId", inst.ID, "err", err)
|
||||
}
|
||||
slog.Info("Instance "+string(finalState), "id", inst.ID)
|
||||
}
|
||||
}(instance, command.Attributes)
|
||||
|
||||
@@ -915,27 +889,13 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) {
|
||||
instanceCopy := *instance.Instance
|
||||
instanceCopy.State = &ec2.InstanceState{}
|
||||
|
||||
// Map internal status to EC2 state codes
|
||||
switch instance.Status {
|
||||
case "pending", "provisioning":
|
||||
instanceCopy.State.SetCode(0)
|
||||
instanceCopy.State.SetName("pending")
|
||||
case "running":
|
||||
instanceCopy.State.SetCode(16)
|
||||
instanceCopy.State.SetName("running")
|
||||
case "stopping":
|
||||
instanceCopy.State.SetCode(64)
|
||||
instanceCopy.State.SetName("stopping")
|
||||
case "stopped":
|
||||
instanceCopy.State.SetCode(80)
|
||||
instanceCopy.State.SetName("stopped")
|
||||
case "shutting-down":
|
||||
instanceCopy.State.SetCode(32)
|
||||
instanceCopy.State.SetName("shutting-down")
|
||||
case "terminated":
|
||||
instanceCopy.State.SetCode(48)
|
||||
instanceCopy.State.SetName("terminated")
|
||||
default:
|
||||
// Map internal status to EC2 state codes using the centralized mapping
|
||||
if info, ok := vm.EC2StateCodes[instance.Status]; ok {
|
||||
instanceCopy.State.SetCode(info.Code)
|
||||
instanceCopy.State.SetName(info.Name)
|
||||
} else {
|
||||
slog.Warn("Instance has unmapped status, reporting as pending",
|
||||
"instanceId", instance.ID, "status", string(instance.Status))
|
||||
instanceCopy.State.SetCode(0)
|
||||
instanceCopy.State.SetName("pending")
|
||||
}
|
||||
|
||||
@@ -941,19 +941,19 @@ func TestDaemon_BootAllocation(t *testing.T) {
|
||||
"i-running": {
|
||||
ID: "i-running",
|
||||
InstanceType: getTestInstanceType(),
|
||||
Status: "running",
|
||||
Status: vm.StateRunning,
|
||||
Attributes: qmp.Attributes{StopInstance: false},
|
||||
},
|
||||
"i-stopped": {
|
||||
ID: "i-stopped",
|
||||
InstanceType: getTestInstanceType(),
|
||||
Status: "stopped",
|
||||
Status: vm.StateStopped,
|
||||
Attributes: qmp.Attributes{StopInstance: true},
|
||||
},
|
||||
"i-terminated": {
|
||||
ID: "i-terminated",
|
||||
InstanceType: getTestInstanceType(),
|
||||
Status: "terminated",
|
||||
Status: vm.StateTerminated,
|
||||
Attributes: qmp.Attributes{StopInstance: false},
|
||||
},
|
||||
}
|
||||
@@ -988,7 +988,7 @@ func TestDaemon_BootAllocation(t *testing.T) {
|
||||
|
||||
// Simulate the allocation loop in Start()
|
||||
for _, instance := range daemon.Instances.VMS {
|
||||
if instance.Status != "terminated" && !instance.Attributes.StopInstance {
|
||||
if instance.Status != vm.StateTerminated && !instance.Attributes.StopInstance {
|
||||
instanceType, ok := daemon.resourceMgr.instanceTypes[instance.InstanceType]
|
||||
if ok {
|
||||
err := daemon.resourceMgr.allocate(instanceType)
|
||||
@@ -1024,7 +1024,7 @@ func TestHandleEC2StartInstances_Allocation(t *testing.T) {
|
||||
daemon.Instances.VMS[instanceId] = &vm.VM{
|
||||
ID: instanceId,
|
||||
InstanceType: instanceType,
|
||||
Status: "stopped",
|
||||
Status: vm.StateStopped,
|
||||
}
|
||||
|
||||
// Subscribe to ec2.startinstances
|
||||
@@ -1070,7 +1070,7 @@ func TestStopInstance_Deallocation(t *testing.T) {
|
||||
daemon.Instances.VMS[instanceId] = &vm.VM{
|
||||
ID: instanceId,
|
||||
InstanceType: instanceTypeStr,
|
||||
Status: "running",
|
||||
Status: vm.StateRunning,
|
||||
}
|
||||
|
||||
err := daemon.resourceMgr.allocate(instanceType)
|
||||
@@ -1254,7 +1254,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
|
||||
|
||||
daemon.Instances.VMS[instanceID] = &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: "running",
|
||||
Status: vm.StateRunning,
|
||||
Reservation: reservation1,
|
||||
Instance: ec2Instance,
|
||||
}
|
||||
@@ -1273,7 +1273,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
|
||||
|
||||
daemon.Instances.VMS[instanceID] = &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: "running",
|
||||
Status: vm.StateRunning,
|
||||
Reservation: reservation2,
|
||||
Instance: ec2Instance,
|
||||
}
|
||||
@@ -1290,7 +1290,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
|
||||
|
||||
daemon.Instances.VMS["i-single-001"] = &vm.VM{
|
||||
ID: "i-single-001",
|
||||
Status: "stopped",
|
||||
Status: vm.StateStopped,
|
||||
Reservation: reservation3,
|
||||
Instance: ec2Instance,
|
||||
}
|
||||
|
||||
@@ -65,15 +65,16 @@ func (m *JetStreamManager) InitKVBucket() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// WriteState writes the instance state to the KV store for the given node
|
||||
// WriteState writes the instance state to the KV store for the given node.
|
||||
// It acquires instances.Mu internally.
|
||||
func (m *JetStreamManager) WriteState(nodeID string, instances *vm.Instances) error {
|
||||
instances.Mu.Lock()
|
||||
defer instances.Mu.Unlock()
|
||||
|
||||
if m.kv == nil {
|
||||
return errors.New("KV bucket not initialized")
|
||||
}
|
||||
|
||||
instances.Mu.Lock()
|
||||
defer instances.Mu.Unlock()
|
||||
|
||||
// Create a struct without the mutex to avoid copying the lock
|
||||
state := struct {
|
||||
VMS map[string]*vm.VM `json:"vms"`
|
||||
|
||||
@@ -73,12 +73,12 @@ func TestJetStreamManager_WriteAndLoadState(t *testing.T) {
|
||||
VMS: map[string]*vm.VM{
|
||||
"i-test-001": {
|
||||
ID: "i-test-001",
|
||||
Status: "running",
|
||||
Status: vm.StateRunning,
|
||||
InstanceType: "t3.micro",
|
||||
},
|
||||
"i-test-002": {
|
||||
ID: "i-test-002",
|
||||
Status: "stopped",
|
||||
Status: vm.StateStopped,
|
||||
InstanceType: "t3.small",
|
||||
},
|
||||
},
|
||||
@@ -97,8 +97,8 @@ func TestJetStreamManager_WriteAndLoadState(t *testing.T) {
|
||||
assert.Len(t, loadedInstances.VMS, 2, "Should have 2 instances")
|
||||
assert.NotNil(t, loadedInstances.VMS["i-test-001"], "Should have i-test-001")
|
||||
assert.NotNil(t, loadedInstances.VMS["i-test-002"], "Should have i-test-002")
|
||||
assert.Equal(t, "running", loadedInstances.VMS["i-test-001"].Status)
|
||||
assert.Equal(t, "stopped", loadedInstances.VMS["i-test-002"].Status)
|
||||
assert.Equal(t, vm.StateRunning, loadedInstances.VMS["i-test-001"].Status)
|
||||
assert.Equal(t, vm.StateStopped, loadedInstances.VMS["i-test-002"].Status)
|
||||
assert.Equal(t, "t3.micro", loadedInstances.VMS["i-test-001"].InstanceType)
|
||||
assert.Equal(t, "t3.small", loadedInstances.VMS["i-test-002"].InstanceType)
|
||||
}
|
||||
@@ -165,7 +165,7 @@ func TestJetStreamManager_BucketReconnection(t *testing.T) {
|
||||
VMS: map[string]*vm.VM{
|
||||
"i-persist": {
|
||||
ID: "i-persist",
|
||||
Status: "running",
|
||||
Status: vm.StateRunning,
|
||||
},
|
||||
},
|
||||
}
|
||||
@@ -190,7 +190,7 @@ func TestJetStreamManager_BucketReconnection(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
assert.NotEmpty(t, loadedInstances.VMS, "Should have persisted instances")
|
||||
assert.NotNil(t, loadedInstances.VMS["i-persist"], "Should have i-persist")
|
||||
assert.Equal(t, "running", loadedInstances.VMS["i-persist"].Status)
|
||||
assert.Equal(t, vm.StateRunning, loadedInstances.VMS["i-persist"].Status)
|
||||
}
|
||||
|
||||
// TestJetStreamManager_DeleteState tests deleting state from the KV store
|
||||
@@ -214,7 +214,7 @@ func TestJetStreamManager_DeleteState(t *testing.T) {
|
||||
VMS: map[string]*vm.VM{
|
||||
"i-delete-me": {
|
||||
ID: "i-delete-me",
|
||||
Status: "running",
|
||||
Status: vm.StateRunning,
|
||||
},
|
||||
},
|
||||
}
|
||||
@@ -278,7 +278,7 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) {
|
||||
VMS: map[string]*vm.VM{
|
||||
"i-initial": {
|
||||
ID: "i-initial",
|
||||
Status: "running",
|
||||
Status: vm.StateRunning,
|
||||
},
|
||||
},
|
||||
}
|
||||
@@ -290,11 +290,11 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) {
|
||||
VMS: map[string]*vm.VM{
|
||||
"i-initial": {
|
||||
ID: "i-initial",
|
||||
Status: "stopped", // Changed status
|
||||
Status: vm.StateStopped, // Changed status
|
||||
},
|
||||
"i-new": { // Added new instance
|
||||
ID: "i-new",
|
||||
Status: "running",
|
||||
Status: vm.StateRunning,
|
||||
},
|
||||
},
|
||||
}
|
||||
@@ -305,7 +305,7 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) {
|
||||
loadedInstances, err := jsm.LoadState(testNodeID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, loadedInstances.VMS, 2, "Should have 2 instances")
|
||||
assert.Equal(t, "stopped", loadedInstances.VMS["i-initial"].Status, "Status should be updated")
|
||||
assert.Equal(t, vm.StateStopped, loadedInstances.VMS["i-initial"].Status, "Status should be updated")
|
||||
assert.NotNil(t, loadedInstances.VMS["i-new"], "Should have new instance")
|
||||
}
|
||||
|
||||
@@ -327,7 +327,7 @@ func TestJetStreamManager_MultipleNodes(t *testing.T) {
|
||||
// Write state for node-1
|
||||
node1Instances := &vm.Instances{
|
||||
VMS: map[string]*vm.VM{
|
||||
"i-node1-001": {ID: "i-node1-001", Status: "running"},
|
||||
"i-node1-001": {ID: "i-node1-001", Status: vm.StateRunning},
|
||||
},
|
||||
}
|
||||
err = jsm.WriteState("node-1", node1Instances)
|
||||
@@ -336,8 +336,8 @@ func TestJetStreamManager_MultipleNodes(t *testing.T) {
|
||||
// Write state for node-2
|
||||
node2Instances := &vm.Instances{
|
||||
VMS: map[string]*vm.VM{
|
||||
"i-node2-001": {ID: "i-node2-001", Status: "stopped"},
|
||||
"i-node2-002": {ID: "i-node2-002", Status: "running"},
|
||||
"i-node2-001": {ID: "i-node2-001", Status: vm.StateStopped},
|
||||
"i-node2-002": {ID: "i-node2-002", Status: vm.StateRunning},
|
||||
},
|
||||
}
|
||||
err = jsm.WriteState("node-2", node2Instances)
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
"github.com/mulgadc/hive/hive/vm"
|
||||
)
|
||||
|
||||
// TransitionState validates and applies a state transition on the given instance.
|
||||
// It sets instance.Status, persists via WriteState, and logs.
|
||||
// On validation failure, the instance status is unchanged and an error is returned.
|
||||
// If WriteState fails, the in-memory status retains the new value (the VM has
|
||||
// physically changed state regardless) and an error is returned.
|
||||
// The caller must NOT hold d.Instances.Mu; this method acquires it internally.
|
||||
func (d *Daemon) TransitionState(instance *vm.VM, target vm.InstanceState) error {
|
||||
d.Instances.Mu.Lock()
|
||||
current := instance.Status
|
||||
if !vm.IsValidTransition(current, target) {
|
||||
d.Instances.Mu.Unlock()
|
||||
return fmt.Errorf("invalid state transition: %s -> %s for instance %s", current, target, instance.ID)
|
||||
}
|
||||
instance.Status = target
|
||||
d.Instances.Mu.Unlock()
|
||||
|
||||
slog.Info("Instance state transition", "instanceId", instance.ID, "from", string(current), "to", string(target))
|
||||
|
||||
if err := d.WriteState(); err != nil {
|
||||
slog.Error("Failed to persist state after transition", "instanceId", instance.ID,
|
||||
"from", string(current), "to", string(target), "err", err)
|
||||
return fmt.Errorf("state transition applied but write failed: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,170 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/hive/hive/config"
|
||||
"github.com/mulgadc/hive/hive/vm"
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// createDaemonWithJetStream creates a daemon backed by an in-process NATS+JetStream server
|
||||
// so that TransitionState (which calls WriteState) works end-to-end.
|
||||
func createDaemonWithJetStream(t *testing.T) *Daemon {
|
||||
t.Helper()
|
||||
|
||||
jsTmpDir, err := os.MkdirTemp("", "nats-js-state-test-*")
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { os.RemoveAll(jsTmpDir) })
|
||||
|
||||
opts := &server.Options{
|
||||
Host: "127.0.0.1",
|
||||
Port: -1,
|
||||
JetStream: true,
|
||||
StoreDir: jsTmpDir,
|
||||
NoLog: true,
|
||||
NoSigs: true,
|
||||
}
|
||||
ns, err := server.NewServer(opts)
|
||||
require.NoError(t, err)
|
||||
go ns.Start()
|
||||
require.True(t, ns.ReadyForConnections(5*time.Second), "NATS server failed to start")
|
||||
t.Cleanup(func() { ns.Shutdown() })
|
||||
|
||||
natsURL := ns.ClientURL()
|
||||
|
||||
tmpDir, err := os.MkdirTemp("", "hive-state-test-*")
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { os.RemoveAll(tmpDir) })
|
||||
|
||||
clusterCfg := &config.ClusterConfig{
|
||||
Node: "node-1",
|
||||
Nodes: map[string]config.Config{"node-1": {BaseDir: tmpDir}},
|
||||
}
|
||||
daemon := NewDaemon(clusterCfg)
|
||||
daemon.config = &config.Config{BaseDir: tmpDir}
|
||||
|
||||
nc, err := nats.Connect(natsURL)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { nc.Close() })
|
||||
|
||||
daemon.natsConn = nc
|
||||
daemon.jsManager, err = NewJetStreamManager(nc, 1)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, daemon.jsManager.InitKVBucket())
|
||||
|
||||
return daemon
|
||||
}
|
||||
|
||||
func TestTransitionState_ValidTransitions(t *testing.T) {
|
||||
daemon := createDaemonWithJetStream(t)
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
from vm.InstanceState
|
||||
to vm.InstanceState
|
||||
}{
|
||||
{"provisioning->running", vm.StateProvisioning, vm.StateRunning},
|
||||
{"provisioning->error", vm.StateProvisioning, vm.StateError},
|
||||
{"provisioning->shutting-down", vm.StateProvisioning, vm.StateShuttingDown},
|
||||
{"pending->running", vm.StatePending, vm.StateRunning},
|
||||
{"pending->error", vm.StatePending, vm.StateError},
|
||||
{"pending->shutting-down", vm.StatePending, vm.StateShuttingDown},
|
||||
{"running->stopping", vm.StateRunning, vm.StateStopping},
|
||||
{"running->shutting-down", vm.StateRunning, vm.StateShuttingDown},
|
||||
{"running->error", vm.StateRunning, vm.StateError},
|
||||
{"stopping->stopped", vm.StateStopping, vm.StateStopped},
|
||||
{"stopping->shutting-down", vm.StateStopping, vm.StateShuttingDown},
|
||||
{"stopping->error", vm.StateStopping, vm.StateError},
|
||||
{"stopped->running", vm.StateStopped, vm.StateRunning},
|
||||
{"stopped->shutting-down", vm.StateStopped, vm.StateShuttingDown},
|
||||
{"stopped->error", vm.StateStopped, vm.StateError},
|
||||
{"shutting-down->terminated", vm.StateShuttingDown, vm.StateTerminated},
|
||||
{"shutting-down->error", vm.StateShuttingDown, vm.StateError},
|
||||
{"error->running", vm.StateError, vm.StateRunning},
|
||||
{"error->shutting-down", vm.StateError, vm.StateShuttingDown},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
instance := &vm.VM{
|
||||
ID: "i-test-valid",
|
||||
Status: tt.from,
|
||||
}
|
||||
|
||||
daemon.Instances.Mu.Lock()
|
||||
daemon.Instances.VMS[instance.ID] = instance
|
||||
daemon.Instances.Mu.Unlock()
|
||||
|
||||
err := daemon.TransitionState(instance, tt.to)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, tt.to, instance.Status)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestTransitionState_InvalidTransitions(t *testing.T) {
|
||||
daemon := createDaemonWithJetStream(t)
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
from vm.InstanceState
|
||||
to vm.InstanceState
|
||||
}{
|
||||
{"running->running", vm.StateRunning, vm.StateRunning},
|
||||
{"running->pending", vm.StateRunning, vm.StatePending},
|
||||
{"running->stopped", vm.StateRunning, vm.StateStopped},
|
||||
{"running->terminated", vm.StateRunning, vm.StateTerminated},
|
||||
{"stopped->stopping", vm.StateStopped, vm.StateStopping},
|
||||
{"stopped->terminated", vm.StateStopped, vm.StateTerminated},
|
||||
{"terminated->running", vm.StateTerminated, vm.StateRunning},
|
||||
{"terminated->stopped", vm.StateTerminated, vm.StateStopped},
|
||||
{"stopping->running", vm.StateStopping, vm.StateRunning},
|
||||
{"shutting-down->running", vm.StateShuttingDown, vm.StateRunning},
|
||||
{"shutting-down->stopped", vm.StateShuttingDown, vm.StateStopped},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
instance := &vm.VM{
|
||||
ID: "i-test-invalid",
|
||||
Status: tt.from,
|
||||
}
|
||||
|
||||
daemon.Instances.Mu.Lock()
|
||||
daemon.Instances.VMS[instance.ID] = instance
|
||||
daemon.Instances.Mu.Unlock()
|
||||
|
||||
err := daemon.TransitionState(instance, tt.to)
|
||||
assert.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "invalid state transition")
|
||||
|
||||
// Status should remain unchanged
|
||||
assert.Equal(t, tt.from, instance.Status)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestTransitionState_NilEC2Instance(t *testing.T) {
|
||||
daemon := createDaemonWithJetStream(t)
|
||||
|
||||
instance := &vm.VM{
|
||||
ID: "i-test-nil-ec2",
|
||||
Status: vm.StateProvisioning,
|
||||
Instance: nil, // no EC2 instance metadata
|
||||
}
|
||||
|
||||
daemon.Instances.Mu.Lock()
|
||||
daemon.Instances.VMS[instance.ID] = instance
|
||||
daemon.Instances.Mu.Unlock()
|
||||
|
||||
err := daemon.TransitionState(instance, vm.StateRunning)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, vm.StateRunning, instance.Status)
|
||||
}
|
||||
@@ -116,7 +116,7 @@ func (s *InstanceServiceImpl) RunInstance(input *ec2.RunInstancesInput) (*vm.VM,
|
||||
// Create new instance structure
|
||||
instance := &vm.VM{
|
||||
ID: instanceId,
|
||||
Status: "provisioning",
|
||||
Status: vm.StateProvisioning,
|
||||
InstanceType: *input.InstanceType,
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
package vm
|
||||
|
||||
// InstanceState represents a typed VM lifecycle state.
|
||||
type InstanceState string
|
||||
|
||||
const (
|
||||
StateProvisioning InstanceState = "provisioning"
|
||||
StatePending InstanceState = "pending"
|
||||
StateRunning InstanceState = "running"
|
||||
StateStopping InstanceState = "stopping"
|
||||
StateStopped InstanceState = "stopped"
|
||||
StateShuttingDown InstanceState = "shutting-down"
|
||||
StateTerminated InstanceState = "terminated"
|
||||
StateError InstanceState = "error"
|
||||
)
|
||||
|
||||
// EC2StateInfo holds the EC2 API code and name for a given instance state.
|
||||
type EC2StateInfo struct {
|
||||
Code int64
|
||||
Name string
|
||||
}
|
||||
|
||||
// EC2StateCodes maps each InstanceState to its EC2 API code and name.
|
||||
// Note: StateError and StateProvisioning are Hive-specific states with no direct
|
||||
// AWS EC2 equivalent. Their Code/Name values are best-effort mappings.
|
||||
var EC2StateCodes = map[InstanceState]EC2StateInfo{
|
||||
StateProvisioning: {Code: 0, Name: "pending"},
|
||||
StatePending: {Code: 0, Name: "pending"},
|
||||
StateRunning: {Code: 16, Name: "running"},
|
||||
StateStopping: {Code: 64, Name: "stopping"},
|
||||
StateStopped: {Code: 80, Name: "stopped"},
|
||||
StateShuttingDown: {Code: 32, Name: "shutting-down"},
|
||||
StateTerminated: {Code: 48, Name: "terminated"},
|
||||
StateError: {Code: 0, Name: "error"},
|
||||
}
|
||||
|
||||
// ValidTransitions defines the allowed state transitions for an instance.
|
||||
// StateTerminated is intentionally absent — it is a terminal state with no valid transitions.
|
||||
var ValidTransitions = map[InstanceState][]InstanceState{
|
||||
StateProvisioning: {StateRunning, StateError, StateShuttingDown},
|
||||
StatePending: {StateRunning, StateError, StateShuttingDown},
|
||||
StateRunning: {StateStopping, StateShuttingDown, StateError},
|
||||
StateStopping: {StateStopped, StateShuttingDown, StateError},
|
||||
StateStopped: {StateRunning, StateShuttingDown, StateError},
|
||||
StateShuttingDown: {StateTerminated, StateError},
|
||||
StateError: {StateRunning, StateShuttingDown},
|
||||
}
|
||||
|
||||
// IsValidTransition checks whether moving from current to target is allowed.
|
||||
func IsValidTransition(current, target InstanceState) bool {
|
||||
allowed, ok := ValidTransitions[current]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
for _, s := range allowed {
|
||||
if s == target {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package vm
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestIsValidTransition(t *testing.T) {
|
||||
assert.True(t, IsValidTransition(StateRunning, StateStopping))
|
||||
assert.True(t, IsValidTransition(StateRunning, StateShuttingDown))
|
||||
assert.True(t, IsValidTransition(StateProvisioning, StateRunning))
|
||||
assert.True(t, IsValidTransition(StateStopped, StateRunning))
|
||||
assert.True(t, IsValidTransition(StateShuttingDown, StateTerminated))
|
||||
assert.True(t, IsValidTransition(StateError, StateRunning))
|
||||
|
||||
assert.False(t, IsValidTransition(StateRunning, StateTerminated))
|
||||
assert.False(t, IsValidTransition(StateTerminated, StateRunning))
|
||||
assert.False(t, IsValidTransition(StateRunning, StateRunning))
|
||||
assert.False(t, IsValidTransition(StateStopped, StateStopping))
|
||||
}
|
||||
|
||||
func TestEC2StateCodes_AllStatesHaveMapping(t *testing.T) {
|
||||
allStates := []InstanceState{
|
||||
StateProvisioning,
|
||||
StatePending,
|
||||
StateRunning,
|
||||
StateStopping,
|
||||
StateStopped,
|
||||
StateShuttingDown,
|
||||
StateTerminated,
|
||||
StateError,
|
||||
}
|
||||
|
||||
for _, s := range allStates {
|
||||
info, ok := EC2StateCodes[s]
|
||||
assert.True(t, ok, "State %s should have an EC2 mapping", s)
|
||||
assert.NotEmpty(t, info.Name, "State %s EC2 name should not be empty", s)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEC2StateCodes_CorrectValues(t *testing.T) {
|
||||
expected := map[InstanceState]EC2StateInfo{
|
||||
StateProvisioning: {Code: 0, Name: "pending"},
|
||||
StatePending: {Code: 0, Name: "pending"},
|
||||
StateRunning: {Code: 16, Name: "running"},
|
||||
StateStopping: {Code: 64, Name: "stopping"},
|
||||
StateStopped: {Code: 80, Name: "stopped"},
|
||||
StateShuttingDown: {Code: 32, Name: "shutting-down"},
|
||||
StateTerminated: {Code: 48, Name: "terminated"},
|
||||
StateError: {Code: 0, Name: "error"},
|
||||
}
|
||||
|
||||
for state, exp := range expected {
|
||||
actual := EC2StateCodes[state]
|
||||
assert.Equal(t, exp.Code, actual.Code, "State %s code mismatch", state)
|
||||
assert.Equal(t, exp.Name, actual.Name, "State %s name mismatch", state)
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -36,7 +36,7 @@ type VM struct {
|
||||
PID int `json:"pid"`
|
||||
PTS int `json:"pts"`
|
||||
Running bool `json:"running"`
|
||||
Status string `json:"status"`
|
||||
Status InstanceState `json:"status"`
|
||||
InstanceType string `json:"instance_type"`
|
||||
Config Config `json:"config"`
|
||||
|
||||
|
||||
Reference in New Issue
Block a user