forked from lei.chen/myxce
feat(apiserver): fail-fast circuit breaker when backend controller is down (Req 4)
component-failover-adoption Req 4: when the apiserver's controller connection is down, backend-requiring API requests must return a clear service-unavailable error rather than hanging. Adds: - health.go: an atomic circuit breaker (IsDegraded/MarkDegraded/MarkHealthy). - callBackend: the four write calls (jobsubmit/jobctrl/queuectrl/hostctrl) now route through it — degraded ⇒ immediate "backend controller unavailable" error without invoking the lib; a transport OR XCE not-ready/masterless/ standby failure (E_XCE_NOMASTER/REDIRECT/NOT_READY/STANDBY) trips the breaker so the NEXT call fails fast. Semantic rejections (-EINVAL/-ENOENT/E_XCE_BAD_*) never trip it. - StartHeartbeat: a failed tick marks degraded; recovery re-joins (rejoinBackend) and marks healthy. Verified live (t1 cluster): kill controller → first submit returns xcecli_jobsubmit failed: -10013 (bounded, no hang), second submit returns "backend controller unavailable (reconnecting)" in ~11ms. 7 unit tests. Recovery-via-heartbeat after a FULL controller death is NOT yet reliable — the lib transport reconnect does not always re-establish to the restarted controller (same class as the scheduler reconnect; tracked under task 13). Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
d4b98aab01
commit
2bbe894f99
3 changed files with 195 additions and 12 deletions
|
|
@ -224,10 +224,12 @@ func StartHeartbeat(ctx context.Context) {
|
||||||
rc := C.xcecli_heartbeat(gCtx.ctx, C.int(secs))
|
rc := C.xcecli_heartbeat(gCtx.ctx, C.int(secs))
|
||||||
if rc < 0 {
|
if rc < 0 {
|
||||||
prevFailed.Store(true)
|
prevFailed.Store(true)
|
||||||
|
MarkDegraded()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if prevFailed.Swap(false) {
|
if prevFailed.Swap(false) {
|
||||||
rejoinBackend()
|
rejoinBackend()
|
||||||
|
MarkHealthy()
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
@ -404,6 +406,49 @@ func handleNotifyData(nd *C.xce_notify_data_t) C.int {
|
||||||
return 0
|
return 0
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// isBackendUnavailable reports whether a negative xcecli_* return means the
|
||||||
|
// backend controller cannot serve the op (transport failure OR an XCE
|
||||||
|
// not-ready/masterless/standby code) rather than a semantic rejection. Only
|
||||||
|
// these trip the Req 4 circuit breaker — a semantic rejection (-EINVAL/
|
||||||
|
// -ENOENT/...) must never mark the backend "down".
|
||||||
|
func isBackendUnavailable(rc int) bool {
|
||||||
|
switch rc {
|
||||||
|
case -int(C.ECONNRESET), -int(C.ECONNREFUSED), -int(C.EHOSTUNREACH),
|
||||||
|
-int(C.ENETUNREACH), -int(C.ENETDOWN), -int(C.ETIMEDOUT),
|
||||||
|
-int(C.EPIPE), -int(C.EIO), -int(C.ENOTCONN), -int(C.ESHUTDOWN),
|
||||||
|
-int(C.ENODEV), -int(C.ENOLINK), -int(C.ECONNABORTED):
|
||||||
|
return true
|
||||||
|
// XCE controller role codes (errnox.h): the controller is reachable but
|
||||||
|
// cannot authoritatively serve the op — no master, redirecting, initializing/
|
||||||
|
// electing, or a standby. All mean "backend unavailable right now".
|
||||||
|
case -(int(C.E_XCE_NOMASTER)), -(int(C.E_XCE_REDIRECT)),
|
||||||
|
-(int(C.E_XCE_NOT_READY)), -(int(C.E_XCE_STANDBY)):
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
// callBackend guards a C.xcecli_* write call with the fail-fast circuit
|
||||||
|
// breaker (component-failover-adoption Req 4): when degraded, return a
|
||||||
|
// service-unavailable error immediately rather than invoking a call that would
|
||||||
|
// hang on a dead controller. A transport failure trips the breaker so the NEXT
|
||||||
|
// call fails fast (the lib's retry budget already bounded this call's block).
|
||||||
|
// fn returns the (negative) xcecli return code converted to a plain int so the
|
||||||
|
// breaker logic is unit-testable without cgo.
|
||||||
|
func callBackend(name string, fn func() int) error {
|
||||||
|
if IsDegraded() {
|
||||||
|
return fmt.Errorf("%s: backend controller unavailable (reconnecting)", name)
|
||||||
|
}
|
||||||
|
rc := fn()
|
||||||
|
if rc < 0 {
|
||||||
|
if isBackendUnavailable(rc) {
|
||||||
|
MarkDegraded()
|
||||||
|
}
|
||||||
|
return fmt.Errorf("%s failed: %d", name, rc)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// SubmitJob sends a job submission to the backend via xcecli_jobsubmit.
|
// SubmitJob sends a job submission to the backend via xcecli_jobsubmit.
|
||||||
// arraySize=1 for normal jobs, >1 for array jobs.
|
// arraySize=1 for normal jobs, >1 for array jobs.
|
||||||
// The store is updated only when the backend pushes the resulting OnJobCreated
|
// The store is updated only when the backend pushes the resulting OnJobCreated
|
||||||
|
|
@ -415,9 +460,10 @@ func SubmitJob(job *models.JobInfo, arraySize int) error {
|
||||||
js.array_size = C.uint64_t(arraySize)
|
js.array_size = C.uint64_t(arraySize)
|
||||||
|
|
||||||
var outJobID C.uint64_t
|
var outJobID C.uint64_t
|
||||||
rc := C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID)
|
if err := callBackend("xcecli_jobsubmit", func() int {
|
||||||
if rc < 0 {
|
return int(C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID))
|
||||||
return fmt.Errorf("xcecli_jobsubmit failed: %d", int(rc))
|
}); err != nil {
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
if outJobID != 0 {
|
if outJobID != 0 {
|
||||||
job.Unique_JobID = int64(outJobID)
|
job.Unique_JobID = int64(outJobID)
|
||||||
|
|
@ -455,9 +501,10 @@ func ControlJob(jobid uint64, action int, content string) error {
|
||||||
C.jobctrl_set_ckptoption(&ctrl, &ckpt)
|
C.jobctrl_set_ckptoption(&ctrl, &ckpt)
|
||||||
}
|
}
|
||||||
|
|
||||||
rc := C.xcecli_jobctrl(gCtx.ctx, &ctrl)
|
if err := callBackend("xcecli_jobctrl", func() int {
|
||||||
if rc < 0 {
|
return int(C.xcecli_jobctrl(gCtx.ctx, &ctrl))
|
||||||
return fmt.Errorf("xcecli_jobctrl failed: %d", int(rc))
|
}); err != nil {
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
@ -473,9 +520,10 @@ func ControlQueue(queuename string, action int) error {
|
||||||
queue: cQueue,
|
queue: cQueue,
|
||||||
}
|
}
|
||||||
|
|
||||||
rc := C.xcecli_queuectrl(gCtx.ctx, &ctrl)
|
if err := callBackend("xcecli_queuectrl", func() int {
|
||||||
if rc < 0 {
|
return int(C.xcecli_queuectrl(gCtx.ctx, &ctrl))
|
||||||
return fmt.Errorf("xcecli_queuectrl failed: %d", int(rc))
|
}); err != nil {
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
@ -490,9 +538,10 @@ func ControlHost(hostname string, action int) error {
|
||||||
hostname: cHost,
|
hostname: cHost,
|
||||||
}
|
}
|
||||||
|
|
||||||
rc := C.xcecli_hostctrl(gCtx.ctx, &ctrl)
|
if err := callBackend("xcecli_hostctrl", func() int {
|
||||||
if rc < 0 {
|
return int(C.xcecli_hostctrl(gCtx.ctx, &ctrl))
|
||||||
return fmt.Errorf("xcecli_hostctrl failed: %d", int(rc))
|
}); err != nil {
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
|
||||||
24
src/frontend/apiserver/backend/health.go
Normal file
24
src/frontend/apiserver/backend/health.go
Normal file
|
|
@ -0,0 +1,24 @@
|
||||||
|
package backend
|
||||||
|
|
||||||
|
import "sync/atomic"
|
||||||
|
|
||||||
|
// backendDegraded is the fail-fast circuit breaker (component-failover-adoption
|
||||||
|
// Req 4): when the controller connection is down, backend-requiring API
|
||||||
|
// requests return a clear "backend unavailable" error immediately instead of
|
||||||
|
// hanging until the connection is restored.
|
||||||
|
//
|
||||||
|
// Lifecycle:
|
||||||
|
// - tripped by a write-path transport failure (callBackend) and by a failed
|
||||||
|
// heartbeat tick (StartHeartbeat);
|
||||||
|
// - cleared by a successful heartbeat recovery (StartHeartbeat → rejoinBackend
|
||||||
|
// → MarkHealthy), which also re-joins the apiserver session so pushes resume.
|
||||||
|
var backendDegraded atomic.Bool
|
||||||
|
|
||||||
|
// IsDegraded reports whether the backend controller is currently unreachable.
|
||||||
|
func IsDegraded() bool { return backendDegraded.Load() }
|
||||||
|
|
||||||
|
// MarkDegraded trips the circuit breaker — backend calls now fail fast.
|
||||||
|
func MarkDegraded() { backendDegraded.Store(true) }
|
||||||
|
|
||||||
|
// MarkHealthy clears the circuit breaker after a successful reconnect/rejoin.
|
||||||
|
func MarkHealthy() { backendDegraded.Store(false) }
|
||||||
110
src/frontend/apiserver/backend/health_test.go
Normal file
110
src/frontend/apiserver/backend/health_test.go
Normal file
|
|
@ -0,0 +1,110 @@
|
||||||
|
package backend
|
||||||
|
|
||||||
|
import "testing"
|
||||||
|
|
||||||
|
// Linux errno + XCE error values. Plain ints keep this test cgo-free;
|
||||||
|
// callBackend's fn returns a plain int by design for exactly this reason.
|
||||||
|
const (
|
||||||
|
ecnreset = -104 // ECONNRESET
|
||||||
|
einval = -22 // EINVAL
|
||||||
|
xceNotReady = -10013 // E_XCE_NOT_READY (controller init/electing)
|
||||||
|
xceBadQueue = -10008 // E_XCE_BAD_QUEUE (semantic — must NOT trip)
|
||||||
|
)
|
||||||
|
|
||||||
|
func resetBreaker() { MarkHealthy() }
|
||||||
|
|
||||||
|
// ── state transitions ────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
func TestCircuitBreakerStateTransitions(t *testing.T) {
|
||||||
|
resetBreaker()
|
||||||
|
if IsDegraded() {
|
||||||
|
t.Fatal("fresh breaker should be healthy")
|
||||||
|
}
|
||||||
|
MarkDegraded()
|
||||||
|
if !IsDegraded() {
|
||||||
|
t.Fatal("MarkDegraded should trip the breaker")
|
||||||
|
}
|
||||||
|
MarkHealthy()
|
||||||
|
if IsDegraded() {
|
||||||
|
t.Fatal("MarkHealthy should clear the breaker")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ── callBackend ──────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
func TestCallBackend_Degraded_FailsFastWithoutCalling(t *testing.T) {
|
||||||
|
resetBreaker()
|
||||||
|
MarkDegraded()
|
||||||
|
|
||||||
|
called := false
|
||||||
|
err := callBackend("test", func() int {
|
||||||
|
called = true
|
||||||
|
return 0
|
||||||
|
})
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("degraded call should return an error")
|
||||||
|
}
|
||||||
|
if called {
|
||||||
|
t.Fatal("degraded call must NOT invoke the backend fn")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCallBackend_TransportError_TripsBreaker(t *testing.T) {
|
||||||
|
resetBreaker()
|
||||||
|
|
||||||
|
err := callBackend("test", func() int { return ecnreset })
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("transport error should return an error")
|
||||||
|
}
|
||||||
|
if !IsDegraded() {
|
||||||
|
t.Fatal("transport error should trip the circuit breaker")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCallBackend_XceNotReady_TripsBreaker(t *testing.T) {
|
||||||
|
resetBreaker()
|
||||||
|
|
||||||
|
err := callBackend("test", func() int { return xceNotReady })
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("XCE not-ready should return an error")
|
||||||
|
}
|
||||||
|
if !IsDegraded() {
|
||||||
|
t.Fatal("XCE not-ready (controller unavailable) should trip the circuit breaker")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCallBackend_SemanticError_DoesNotTripBreaker(t *testing.T) {
|
||||||
|
resetBreaker()
|
||||||
|
|
||||||
|
err := callBackend("test", func() int { return einval })
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("semantic error should return an error")
|
||||||
|
}
|
||||||
|
if IsDegraded() {
|
||||||
|
t.Fatal("semantic error must NOT trip the circuit breaker")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCallBackend_XceSemanticError_DoesNotTripBreaker(t *testing.T) {
|
||||||
|
resetBreaker()
|
||||||
|
|
||||||
|
err := callBackend("test", func() int { return xceBadQueue })
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("XCE semantic error should return an error")
|
||||||
|
}
|
||||||
|
if IsDegraded() {
|
||||||
|
t.Fatal("XCE semantic rejection must NOT trip the circuit breaker")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCallBackend_Success_StaysHealthy(t *testing.T) {
|
||||||
|
resetBreaker()
|
||||||
|
|
||||||
|
err := callBackend("test", func() int { return 0 })
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("success should return nil, got %v", err)
|
||||||
|
}
|
||||||
|
if IsDegraded() {
|
||||||
|
t.Fatal("success must not trip the circuit breaker")
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Add table
Add a link
Reference in a new issue