mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-16 01:22:00 +00:00
feat: complete EC2 instance health checks (Phase C + E) (#438)
Wire crash detection and host-resource signals through to the AWS-faithful DescribeInstanceStatus API and the spx CLI, closing out the EC2 health monitoring / self-healing work. Phase C — QMP health monitoring: - qmpHeartbeat now tracks consecutive query-status failures; at QMPMaxConsecutiveFailures it checks process liveness, triggering HandleCrash on a dead QEMU and marking an alive-but-wedged process impaired. - InstanceHealthState gains QMPConsecutiveFailures, LastQMPSuccess, ImpairedSince. Phase E — DescribeInstanceStatus mapping: - buildInstanceStatus maps running VMs to ok / impaired (QMP-unresponsive, with ImpairedSince) / initializing (2-min launch grace), non-running to not-applicable. - SystemStatus reflects node memory pressure via ResourceManager HostUnderMemoryPressure (live MemAvailable below the host reserve). Phase C2 — observability: - spx get vms gains HEALTH and CRASHES columns derived from vmHealthLabel. Tests cover crash/QMP helpers, status mapping, memory-pressure system status, and the health label.
This commit is contained in:
@@ -289,7 +289,7 @@ func runGetVMs(cmd *cobra.Command, args []string) {
|
||||
})
|
||||
|
||||
tableData := pterm.TableData{
|
||||
{"INSTANCE", "STATUS", "TYPE", "VCPU", "MEM", "NODE", "IP", "AGE"},
|
||||
{"INSTANCE", "STATUS", "HEALTH", "CRASHES", "TYPE", "VCPU", "MEM", "NODE", "IP", "AGE"},
|
||||
}
|
||||
|
||||
for _, v := range allVMs {
|
||||
@@ -297,9 +297,15 @@ func runGetVMs(cmd *cobra.Command, args []string) {
|
||||
if v.LaunchTime > 0 {
|
||||
age = formatUptime(time.Now().Unix() - v.LaunchTime)
|
||||
}
|
||||
health := v.Health
|
||||
if health == "" {
|
||||
health = "-"
|
||||
}
|
||||
tableData = append(tableData, []string{
|
||||
v.InstanceID,
|
||||
v.Status,
|
||||
health,
|
||||
strconv.Itoa(v.CrashCount),
|
||||
v.InstanceType,
|
||||
strconv.Itoa(v.VCPU),
|
||||
formatMemGB(v.MemoryGB),
|
||||
|
||||
@@ -448,6 +448,8 @@ func (d *Daemon) handleNodeVMs(msg *nats.Msg) {
|
||||
Status: string(v.Status),
|
||||
InstanceType: v.InstanceType,
|
||||
ManagedBy: v.ManagedBy,
|
||||
Health: vmHealthLabel(v),
|
||||
CrashCount: v.Health.CrashCount,
|
||||
}
|
||||
if it, ok := d.resourceMgr.instanceTypes[v.InstanceType]; ok {
|
||||
info.VCPU = int(instanceTypeVCPUs(it))
|
||||
@@ -470,3 +472,19 @@ func (d *Daemon) handleNodeVMs(msg *nats.Msg) {
|
||||
|
||||
respondWithJSON(msg, resp)
|
||||
}
|
||||
|
||||
// vmHealthLabel derives the display health for spx get vms. Only running VMs
|
||||
// carry health: QMP-unresponsive past the failure gate is impaired, a VM that
|
||||
// has crashed before but is running again is recovering, otherwise ok.
|
||||
func vmHealthLabel(v *vm.VM) string {
|
||||
if v.Status != vm.StateRunning {
|
||||
return "-"
|
||||
}
|
||||
if v.Health.QMPConsecutiveFailures >= vm.QMPMaxConsecutiveFailures {
|
||||
return "impaired"
|
||||
}
|
||||
if v.Health.CrashCount > 0 {
|
||||
return "recovering"
|
||||
}
|
||||
return "ok"
|
||||
}
|
||||
|
||||
@@ -4133,3 +4133,24 @@ func TestRespondWithJSON_RespondFailureLogs(t *testing.T) {
|
||||
assert.Contains(t, logged, "Failed to respond to NATS request")
|
||||
assert.NotContains(t, logged, "Failed to marshal response")
|
||||
}
|
||||
|
||||
func TestVMHealthLabel(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
status vm.InstanceState
|
||||
health vm.InstanceHealthState
|
||||
want string
|
||||
}{
|
||||
{"stopped", vm.StateStopped, vm.InstanceHealthState{}, "-"},
|
||||
{"running clean", vm.StateRunning, vm.InstanceHealthState{}, "ok"},
|
||||
{"running after crash", vm.StateRunning, vm.InstanceHealthState{CrashCount: 2}, "recovering"},
|
||||
{"qmp impaired", vm.StateRunning, vm.InstanceHealthState{QMPConsecutiveFailures: vm.QMPMaxConsecutiveFailures}, "impaired"},
|
||||
{"qmp impaired wins over crash", vm.StateRunning, vm.InstanceHealthState{CrashCount: 5, QMPConsecutiveFailures: vm.QMPMaxConsecutiveFailures}, "impaired"},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
v := &vm.VM{ID: "i-x", Status: tc.status, Health: tc.health}
|
||||
assert.Equal(t, tc.want, vmHealthLabel(v))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -118,6 +118,25 @@ func canAllocateCount(availVCPU, allocVCPU int, availMem, allocMem float64,
|
||||
return max(result, 0)
|
||||
}
|
||||
|
||||
// HostUnderMemoryPressure reports whether live MemAvailable has fallen below the
|
||||
// host reserve, i.e. the node has eaten into the headroom kept for the daemon
|
||||
// and co-located services. Surfaced as DescribeInstanceStatus SystemStatus.
|
||||
// Returns false when the live gate is disabled or unreadable (fail open — the
|
||||
// accounting gate still bounds admission).
|
||||
func (rm *ResourceManager) HostUnderMemoryPressure() bool {
|
||||
if rm.readMemAvailableGB == nil {
|
||||
return false
|
||||
}
|
||||
avail, ok := rm.readMemAvailableGB()
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
rm.mu.RLock()
|
||||
reserve := rm.reservedMem
|
||||
rm.mu.RUnlock()
|
||||
return avail < reserve
|
||||
}
|
||||
|
||||
// liveMemCount clamps n by how many guests of memGB fit in live headroom
|
||||
// (availMemGB − reservedMemGB). Negative headroom yields 0.
|
||||
func liveMemCount(n int, availMemGB, reservedMemGB, memGB float64) int {
|
||||
|
||||
@@ -573,6 +573,28 @@ func TestLiveMemGate(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestHostUnderMemoryPressure(t *testing.T) {
|
||||
t.Run("nil reader fails open", func(t *testing.T) {
|
||||
rm := &ResourceManager{reservedMem: 2}
|
||||
assert.False(t, rm.HostUnderMemoryPressure())
|
||||
})
|
||||
|
||||
t.Run("read failure fails open", func(t *testing.T) {
|
||||
rm := &ResourceManager{reservedMem: 2, readMemAvailableGB: func() (float64, bool) { return 0, false }}
|
||||
assert.False(t, rm.HostUnderMemoryPressure())
|
||||
})
|
||||
|
||||
t.Run("available above reserve is healthy", func(t *testing.T) {
|
||||
rm := &ResourceManager{reservedMem: 2, readMemAvailableGB: func() (float64, bool) { return 4, true }}
|
||||
assert.False(t, rm.HostUnderMemoryPressure())
|
||||
})
|
||||
|
||||
t.Run("available below reserve is pressure", func(t *testing.T) {
|
||||
rm := &ResourceManager{reservedMem: 2, readMemAvailableGB: func() (float64, bool) { return 1, true }}
|
||||
assert.True(t, rm.HostUnderMemoryPressure())
|
||||
})
|
||||
}
|
||||
|
||||
func TestResolveNbdkitCharge(t *testing.T) {
|
||||
t.Run("defaults when unset", func(t *testing.T) {
|
||||
main, aux := resolveNbdkitCharge(func(string) string { return "" })
|
||||
|
||||
@@ -2429,7 +2429,23 @@ var (
|
||||
}
|
||||
)
|
||||
|
||||
func (s *InstanceServiceImpl) buildInstanceStatus(v *vm.VM) *ec2.InstanceStatus {
|
||||
// hostHealthReporter is an optional capability on the resource manager that
|
||||
// reports node-level memory pressure, surfaced as DescribeInstanceStatus
|
||||
// SystemStatus. Implemented by daemon.ResourceManager.
|
||||
type hostHealthReporter interface {
|
||||
HostUnderMemoryPressure() bool
|
||||
}
|
||||
|
||||
// hostUnderMemoryPressure reports whether the node is under memory pressure when
|
||||
// the resource manager supports the check, false otherwise (fail open).
|
||||
func (s *InstanceServiceImpl) hostUnderMemoryPressure() bool {
|
||||
if hr, ok := s.resourceMgr.(hostHealthReporter); ok {
|
||||
return hr.HostUnderMemoryPressure()
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (s *InstanceServiceImpl) buildInstanceStatus(v *vm.VM, systemImpaired bool) *ec2.InstanceStatus {
|
||||
state := &ec2.InstanceState{}
|
||||
if info, ok := vm.EC2StateCodes[v.Status]; ok {
|
||||
state.SetCode(info.Code)
|
||||
@@ -2439,11 +2455,25 @@ func (s *InstanceServiceImpl) buildInstanceStatus(v *vm.VM) *ec2.InstanceStatus
|
||||
state.SetName(vm.EC2StateCodes[vm.StatePending].Name)
|
||||
}
|
||||
|
||||
status := instanceStatusNotApplicable
|
||||
reachability := instanceStatusNotApplicable
|
||||
if v.Status == vm.StateRunning {
|
||||
status = instanceStatusOK
|
||||
reachability = instanceStatusPassed
|
||||
status, reachability, impairedSince := instanceHealthSummary(v)
|
||||
|
||||
instDetail := &ec2.InstanceStatusDetails{
|
||||
Name: aws.String(reachabilityDetailName),
|
||||
Status: aws.String(reachability),
|
||||
}
|
||||
if impairedSince != nil {
|
||||
instDetail.ImpairedSince = impairedSince
|
||||
}
|
||||
|
||||
// SystemStatus reflects host/node health, independent of the VM process: a
|
||||
// running VM's host is reachable unless under memory pressure; non-running
|
||||
// instances are not-applicable.
|
||||
systemStatus, systemReach := instanceStatusOK, instanceStatusPassed
|
||||
switch {
|
||||
case v.Status != vm.StateRunning:
|
||||
systemStatus, systemReach = instanceStatusNotApplicable, instanceStatusNotApplicable
|
||||
case systemImpaired:
|
||||
systemStatus, systemReach = instanceStatusImpaired, instanceStatusFailed
|
||||
}
|
||||
|
||||
return &ec2.InstanceStatus{
|
||||
@@ -2451,26 +2481,58 @@ func (s *InstanceServiceImpl) buildInstanceStatus(v *vm.VM) *ec2.InstanceStatus
|
||||
InstanceId: aws.String(v.ID),
|
||||
InstanceState: state,
|
||||
InstanceStatus: &ec2.InstanceStatusSummary{
|
||||
Status: aws.String(status),
|
||||
Details: []*ec2.InstanceStatusDetails{{
|
||||
Name: aws.String(reachabilityDetailName),
|
||||
Status: aws.String(reachability),
|
||||
}},
|
||||
Status: aws.String(status),
|
||||
Details: []*ec2.InstanceStatusDetails{instDetail},
|
||||
},
|
||||
SystemStatus: &ec2.InstanceStatusSummary{
|
||||
Status: aws.String(status),
|
||||
Status: aws.String(systemStatus),
|
||||
Details: []*ec2.InstanceStatusDetails{{
|
||||
Name: aws.String(reachabilityDetailName),
|
||||
Status: aws.String(reachability),
|
||||
Status: aws.String(systemReach),
|
||||
}},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// instanceInitializingGrace is the post-launch window during which a running VM
|
||||
// reports initializing, matching AWS's status-check grace period.
|
||||
const instanceInitializingGrace = 2 * time.Minute
|
||||
|
||||
// instanceHealthSummary maps a VM's runtime and QMP health into AWS
|
||||
// InstanceStatus fields: the status label, the reachability detail status, and
|
||||
// an optional ImpairedSince timestamp. Only running VMs carry real health; all
|
||||
// other states are not-applicable per AWS behavior.
|
||||
func instanceHealthSummary(v *vm.VM) (status, reachability string, impairedSince *time.Time) {
|
||||
if v.Status != vm.StateRunning {
|
||||
return instanceStatusNotApplicable, instanceStatusNotApplicable, nil
|
||||
}
|
||||
|
||||
// QMP unresponsive past the failure gate → impaired/failed.
|
||||
if v.Health.QMPConsecutiveFailures >= vm.QMPMaxConsecutiveFailures {
|
||||
var impPtr *time.Time
|
||||
if !v.Health.ImpairedSince.IsZero() {
|
||||
since := v.Health.ImpairedSince
|
||||
impPtr = &since
|
||||
}
|
||||
return instanceStatusImpaired, instanceStatusFailed, impPtr
|
||||
}
|
||||
|
||||
// Grace period: freshly launched VMs report initializing until reachable.
|
||||
if v.Instance != nil && v.Instance.LaunchTime != nil &&
|
||||
time.Since(*v.Instance.LaunchTime) < instanceInitializingGrace {
|
||||
return instanceStatusInitializing, instanceStatusInitializing, nil
|
||||
}
|
||||
|
||||
return instanceStatusOK, instanceStatusPassed, nil
|
||||
}
|
||||
|
||||
const (
|
||||
instanceStatusOK = "ok"
|
||||
instanceStatusPassed = "passed"
|
||||
instanceStatusNotApplicable = "not-applicable"
|
||||
instanceStatusImpaired = "impaired"
|
||||
instanceStatusFailed = "failed"
|
||||
instanceStatusInitializing = "initializing"
|
||||
reachabilityDetailName = "reachability"
|
||||
)
|
||||
|
||||
@@ -2536,6 +2598,10 @@ func (s *InstanceServiceImpl) DescribeInstanceStatus(input *ec2.DescribeInstance
|
||||
includedStates = describeInstanceStatusAllIncluded
|
||||
}
|
||||
|
||||
// SystemStatus is node-wide: evaluate host memory pressure once per request
|
||||
// rather than per instance.
|
||||
systemImpaired := s.hostUnderMemoryPressure()
|
||||
|
||||
var statuses []*ec2.InstanceStatus
|
||||
s.vmMgr.View(func(vms map[string]*vm.VM) {
|
||||
for _, v := range vms {
|
||||
@@ -2548,7 +2614,7 @@ func (s *InstanceServiceImpl) DescribeInstanceStatus(input *ec2.DescribeInstance
|
||||
if !includedStates[v.Status] {
|
||||
continue
|
||||
}
|
||||
is := s.buildInstanceStatus(v)
|
||||
is := s.buildInstanceStatus(v, systemImpaired)
|
||||
if len(parsedFilters) > 0 && !instanceStatusMatchesFilters(v, is, parsedFilters) {
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -530,8 +530,12 @@ type fakeResourceCapacityProvider struct {
|
||||
reservationAllocated []*ec2.InstanceTypeInfo
|
||||
reservationReleased []*ec2.InstanceTypeInfo
|
||||
reservationAvailFn func(string, string, *ec2.InstanceTypeInfo) int
|
||||
|
||||
memPressure bool
|
||||
}
|
||||
|
||||
func (f *fakeResourceCapacityProvider) HostUnderMemoryPressure() bool { return f.memPressure }
|
||||
|
||||
func (f *fakeResourceCapacityProvider) GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo {
|
||||
f.calls++
|
||||
f.gotShowCap = showCapacity
|
||||
@@ -3188,6 +3192,104 @@ func TestDescribeInstanceStatus_TagFilter(t *testing.T) {
|
||||
assert.Equal(t, "i-tag", *out.InstanceStatuses[0].InstanceId)
|
||||
}
|
||||
|
||||
func TestDescribeInstanceStatus_QMPUnresponsiveImpaired(t *testing.T) {
|
||||
owner := "111122223333"
|
||||
v := runningVM("i-imp", owner)
|
||||
since := time.Now().Add(-90 * time.Second)
|
||||
v.Health.QMPConsecutiveFailures = vm.QMPMaxConsecutiveFailures
|
||||
v.Health.ImpairedSince = since
|
||||
svc := instanceStatusService(t, "az-a", map[string]*vm.VM{v.ID: v})
|
||||
|
||||
out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, owner)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, out.InstanceStatuses, 1)
|
||||
|
||||
s := out.InstanceStatuses[0]
|
||||
assert.Equal(t, "running", *s.InstanceState.Name)
|
||||
assert.Equal(t, "impaired", *s.InstanceStatus.Status)
|
||||
require.Len(t, s.InstanceStatus.Details, 1)
|
||||
assert.Equal(t, "failed", *s.InstanceStatus.Details[0].Status)
|
||||
require.NotNil(t, s.InstanceStatus.Details[0].ImpairedSince)
|
||||
assert.WithinDuration(t, since, *s.InstanceStatus.Details[0].ImpairedSince, time.Second)
|
||||
// SystemStatus is host-level: the node is still reachable.
|
||||
assert.Equal(t, "ok", *s.SystemStatus.Status)
|
||||
}
|
||||
|
||||
func TestDescribeInstanceStatus_BelowThresholdStaysOK(t *testing.T) {
|
||||
owner := "111122223333"
|
||||
v := runningVM("i-ok", owner)
|
||||
v.Health.QMPConsecutiveFailures = vm.QMPMaxConsecutiveFailures - 1
|
||||
svc := instanceStatusService(t, "az-a", map[string]*vm.VM{v.ID: v})
|
||||
|
||||
out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, owner)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, out.InstanceStatuses, 1)
|
||||
assert.Equal(t, "ok", *out.InstanceStatuses[0].InstanceStatus.Status)
|
||||
assert.Equal(t, "passed", *out.InstanceStatuses[0].InstanceStatus.Details[0].Status)
|
||||
}
|
||||
|
||||
func TestDescribeInstanceStatus_RecentLaunchInitializing(t *testing.T) {
|
||||
owner := "111122223333"
|
||||
v := runningVM("i-new", owner)
|
||||
now := time.Now()
|
||||
v.Instance.LaunchTime = &now
|
||||
svc := instanceStatusService(t, "az-a", map[string]*vm.VM{v.ID: v})
|
||||
|
||||
out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, owner)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, out.InstanceStatuses, 1)
|
||||
assert.Equal(t, "initializing", *out.InstanceStatuses[0].InstanceStatus.Status)
|
||||
assert.Equal(t, "initializing", *out.InstanceStatuses[0].InstanceStatus.Details[0].Status)
|
||||
}
|
||||
|
||||
func TestDescribeInstanceStatus_PastGraceOK(t *testing.T) {
|
||||
owner := "111122223333"
|
||||
v := runningVM("i-old", owner)
|
||||
old := time.Now().Add(-5 * time.Minute)
|
||||
v.Instance.LaunchTime = &old
|
||||
svc := instanceStatusService(t, "az-a", map[string]*vm.VM{v.ID: v})
|
||||
|
||||
out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, owner)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, out.InstanceStatuses, 1)
|
||||
assert.Equal(t, "ok", *out.InstanceStatuses[0].InstanceStatus.Status)
|
||||
}
|
||||
|
||||
func TestDescribeInstanceStatus_MemoryPressureSystemImpaired(t *testing.T) {
|
||||
owner := "111122223333"
|
||||
v := runningVM("i-press", owner)
|
||||
svc := &InstanceServiceImpl{
|
||||
config: &config.Config{AZ: "az-a"},
|
||||
vmMgr: mgrWith(map[string]*vm.VM{v.ID: v}),
|
||||
resourceMgr: &fakeResourceCapacityProvider{memPressure: true},
|
||||
}
|
||||
|
||||
out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, owner)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, out.InstanceStatuses, 1)
|
||||
|
||||
s := out.InstanceStatuses[0]
|
||||
// Instance itself is healthy; only the host system status is impaired.
|
||||
assert.Equal(t, "ok", *s.InstanceStatus.Status)
|
||||
assert.Equal(t, "impaired", *s.SystemStatus.Status)
|
||||
assert.Equal(t, "failed", *s.SystemStatus.Details[0].Status)
|
||||
}
|
||||
|
||||
func TestDescribeInstanceStatus_NoPressureSystemOK(t *testing.T) {
|
||||
owner := "111122223333"
|
||||
v := runningVM("i-fine", owner)
|
||||
svc := &InstanceServiceImpl{
|
||||
config: &config.Config{AZ: "az-a"},
|
||||
vmMgr: mgrWith(map[string]*vm.VM{v.ID: v}),
|
||||
resourceMgr: &fakeResourceCapacityProvider{memPressure: false},
|
||||
}
|
||||
|
||||
out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, owner)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, out.InstanceStatuses, 1)
|
||||
assert.Equal(t, "ok", *out.InstanceStatuses[0].SystemStatus.Status)
|
||||
}
|
||||
|
||||
// TestInstanceArchitecture pins the safe-extraction contract: malformed
|
||||
// InstanceTypeInfo returns "" rather than panicking, and the firmware probe
|
||||
// surfaces "" as a clear error on the launch path.
|
||||
|
||||
@@ -87,6 +87,11 @@ type VMInfo struct {
|
||||
// system-managed resources out of customer-facing listings.
|
||||
ManagedBy string `json:"managed_by,omitempty"`
|
||||
GPU *VMGPUInfo `json:"gpu,omitempty"`
|
||||
// Health is a display label for instance health: "ok", "impaired",
|
||||
// "recovering", or "-" for non-running VMs. CrashCount is the lifetime
|
||||
// crash tally within the current restart window.
|
||||
Health string `json:"health,omitempty"`
|
||||
CrashCount int `json:"crash_count,omitempty"`
|
||||
}
|
||||
|
||||
// NodeVMsResponse is returned by the spinifex.node.vms NATS topic (fan-out).
|
||||
|
||||
@@ -508,3 +508,37 @@ func TestRestartCrashedInstance_RunFailureRollback(t *testing.T) {
|
||||
assert.Equal(t, StateError, m.Status(instance),
|
||||
"instance must end in StateError after Run-failure rollback")
|
||||
}
|
||||
|
||||
func TestRecordQMPFailure_StampsImpairedAtThreshold(t *testing.T) {
|
||||
m := NewManager()
|
||||
instance := &VM{ID: "i-qmp", Status: StateRunning}
|
||||
m.Insert(instance)
|
||||
|
||||
for i := 1; i < QMPMaxConsecutiveFailures; i++ {
|
||||
count := m.recordQMPFailure(instance)
|
||||
assert.Equal(t, i, count)
|
||||
assert.True(t, instance.Health.ImpairedSince.IsZero(),
|
||||
"must not stamp ImpairedSince before threshold")
|
||||
}
|
||||
|
||||
count := m.recordQMPFailure(instance)
|
||||
assert.Equal(t, QMPMaxConsecutiveFailures, count)
|
||||
assert.False(t, instance.Health.ImpairedSince.IsZero(),
|
||||
"must stamp ImpairedSince at threshold")
|
||||
}
|
||||
|
||||
func TestRecordQMPSuccess_ClearsFailureState(t *testing.T) {
|
||||
m := NewManager()
|
||||
instance := &VM{ID: "i-qmp", Status: StateRunning}
|
||||
m.Insert(instance)
|
||||
|
||||
for range QMPMaxConsecutiveFailures {
|
||||
m.recordQMPFailure(instance)
|
||||
}
|
||||
require.False(t, instance.Health.ImpairedSince.IsZero())
|
||||
|
||||
m.recordQMPSuccess(instance)
|
||||
assert.Zero(t, instance.Health.QMPConsecutiveFailures)
|
||||
assert.True(t, instance.Health.ImpairedSince.IsZero(), "success must clear ImpairedSince")
|
||||
assert.False(t, instance.Health.LastQMPSuccess.IsZero(), "success must stamp LastQMPSuccess")
|
||||
}
|
||||
|
||||
@@ -585,11 +585,23 @@ func newQMPClientWithHandshake(v *VM) (*qmp.QMPClient, error) {
|
||||
return client, nil
|
||||
}
|
||||
|
||||
// qmpHeartbeat sends query-status every 30s. Exits and closes the QMP
|
||||
// connection when the instance reaches a terminal or transitional state.
|
||||
const (
|
||||
// qmpHeartbeatInterval is the QMP query-status poll period.
|
||||
qmpHeartbeatInterval = 30 * time.Second
|
||||
// QMPMaxConsecutiveFailures is how many back-to-back query-status failures
|
||||
// mark an instance impaired and trigger a process-liveness check (90s of
|
||||
// unresponsiveness). Exported so DescribeInstanceStatus uses the same gate.
|
||||
QMPMaxConsecutiveFailures = 3
|
||||
)
|
||||
|
||||
// qmpHeartbeat polls query-status every qmpHeartbeatInterval and acts on
|
||||
// failures: it tracks consecutive failures, and once they reach
|
||||
// QMPMaxConsecutiveFailures it checks process liveness. A dead process triggers
|
||||
// crash recovery; an alive-but-wedged QEMU stays impaired for an operator. The
|
||||
// goroutine exits and closes the QMP connection on any terminal/transitional state.
|
||||
func (m *Manager) qmpHeartbeat(instance *VM) {
|
||||
for {
|
||||
time.Sleep(30 * time.Second)
|
||||
time.Sleep(qmpHeartbeatInterval)
|
||||
|
||||
status := m.Status(instance)
|
||||
|
||||
@@ -607,13 +619,51 @@ func (m *Manager) qmpHeartbeat(instance *VM) {
|
||||
slog.Debug("QMP heartbeat", "instance", instance.ID)
|
||||
qmpStatus, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{Execute: "query-status"}, instance.ID)
|
||||
if err != nil {
|
||||
slog.Warn("QMP heartbeat failed", "instance", instance.ID, "err", err)
|
||||
failures := m.recordQMPFailure(instance)
|
||||
slog.Warn("QMP heartbeat failed", "instance", instance.ID, "consecutiveFailures", failures, "err", err)
|
||||
if failures < QMPMaxConsecutiveFailures {
|
||||
continue
|
||||
}
|
||||
if !isInstanceProcessRunning(instance) {
|
||||
slog.Error("QEMU process dead and QMP unresponsive, triggering crash recovery",
|
||||
"instance", instance.ID, "consecutiveFailures", failures)
|
||||
m.HandleCrash(instance, fmt.Errorf("qmp unresponsive (%d failures), process dead", failures))
|
||||
return
|
||||
}
|
||||
slog.Error("QEMU process alive but QMP unresponsive", "instance", instance.ID, "consecutiveFailures", failures)
|
||||
continue
|
||||
}
|
||||
|
||||
m.recordQMPSuccess(instance)
|
||||
slog.Debug("QMP status", "instance", instance.ID, "status", string(qmpStatus.Return))
|
||||
}
|
||||
}
|
||||
|
||||
// recordQMPFailure increments the consecutive QMP failure counter, stamping
|
||||
// ImpairedSince when the count first reaches QMPMaxConsecutiveFailures. Returns
|
||||
// the new count.
|
||||
func (m *Manager) recordQMPFailure(instance *VM) int {
|
||||
var count int
|
||||
m.UpdateState(instance.ID, func(v *VM) {
|
||||
v.Health.QMPConsecutiveFailures++
|
||||
count = v.Health.QMPConsecutiveFailures
|
||||
if count == QMPMaxConsecutiveFailures {
|
||||
v.Health.ImpairedSince = time.Now()
|
||||
}
|
||||
})
|
||||
return count
|
||||
}
|
||||
|
||||
// recordQMPSuccess clears the failure counter and impaired marker after a
|
||||
// healthy poll and records the success time.
|
||||
func (m *Manager) recordQMPSuccess(instance *VM) {
|
||||
m.UpdateState(instance.ID, func(v *VM) {
|
||||
v.Health.QMPConsecutiveFailures = 0
|
||||
v.Health.ImpairedSince = time.Time{}
|
||||
v.Health.LastQMPSuccess = time.Now()
|
||||
})
|
||||
}
|
||||
|
||||
// sendQMPCommand encodes cmd and decodes the response, skipping event messages.
|
||||
func sendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceID string) (*qmp.QMPResponse, error) {
|
||||
if q == nil || q.Encoder == nil || q.Decoder == nil {
|
||||
|
||||
@@ -24,6 +24,13 @@ type InstanceHealthState struct {
|
||||
LastCrashReason string `json:"last_crash_reason,omitempty"`
|
||||
RestartCount int `json:"restart_count"`
|
||||
FirstCrashTime time.Time `json:"first_crash_time"`
|
||||
|
||||
// QMP health monitoring. QMPConsecutiveFailures counts back-to-back
|
||||
// query-status failures; at QMPMaxConsecutiveFailures the instance is
|
||||
// impaired and ImpairedSince is stamped. A successful poll clears all three.
|
||||
QMPConsecutiveFailures int `json:"qmp_consecutive_failures"`
|
||||
LastQMPSuccess time.Time `json:"last_qmp_success,omitzero"`
|
||||
ImpairedSince time.Time `json:"impaired_since,omitzero"`
|
||||
}
|
||||
|
||||
// ExtraENI describes an additional VPC network interface attached to a VM
|
||||
|
||||
Reference in New Issue
Block a user