feat(component-failover-adoption): route apiserver's candidate parsing through the real C parser
Replaces apiserver's hand-written Go parsePeer() with a CGO wrapper around the shared C parser+validator pair (xce_parse_peers then xce_cluster_config_validate, both src/lib/cluster.c) that worker and scheduler already use from their own main.c's. Gives apiserver the same [name=]host:port[,...] syntax (including the name= prefix the old Go parser could not express) and identical malformed-candidate rejection (empty host, port <= 0, non-numeric port) -- round-1 review caught the first pass skipping the validate() half, silently accepting inputs worker/scheduler both reject. A malformed entry now rejects the whole candidate list rather than being skipped with a warning (matches xce_parse_peers()'s own all-or-nothing construction and worker/scheduler's precedent) -- surfaced loudly via RegisterToBackend's error, not silently. Parsing only -- RegisterToBackend()'s connect loop and tryConnect() are unchanged; collapsing them into a single connect call is task 4.2's job. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NNB4zGPCsHyDUCDJyMbiQN
This commit is contained in:
parent
d383d04311
commit
8b1e13a52b
7 changed files with 262 additions and 134 deletions
|
|
@ -86,7 +86,7 @@
|
|||
- _Depends: 3.2_
|
||||
|
||||
- [ ] 4. apiserver: unified candidate-list connect via CGO, config parity, rejoin, and fail-fast
|
||||
- [ ] 4.1 (P) Route apiserver's candidate-list parsing through the real C parser
|
||||
- [x] 4.1 (P) Route apiserver's candidate-list parsing through the real C parser
|
||||
- Replace the hand-written Go candidate-list parser with the shared C parser used by worker and scheduler, retiring the Go parser and its call sites entirely
|
||||
- Map validation errors to clear Go error messages naming the malformed entry
|
||||
- Observable completion: the same malformed-candidate-string cases worker/scheduler already reject are rejected identically when passed through apiserver's config path
|
||||
|
|
@ -168,5 +168,6 @@
|
|||
- Task 2.2: fixed the startup-ordering bug by threading `cluster_cfg` by value through `xcecd_config_t` → `xcecd_context_t` → `cc_ctx` (set in `main.c` before `xcecd_init()`/`xcecd_start()` run, instead of after). Safe as a plain struct copy because `xce_candidate_t.hostname` is a fixed `char[]` array, not a pointer — no aliasing hazard here (unlike the `xce_cluster_config_to_client_config()` output struct noted above). Scheduler (3.1) likely has the same ordering gap (`sched_init` loops candidates correctly already per design.md, so may not need this fix — verify against current source, don't assume). Rejoin-retry pattern established: `checking_connection()` checks `needs_rejoin` after every probe return, calls `xcecli_join()` directly (not full re-entry into the join-cluster function), clears the flag only on success — same pattern scheduler/apiserver rejoin tasks (3.1, 4.3) should follow.
|
||||
- Task 2.3 (real E2E caught what unit tests missed — 2 production bugs slipped through tasks 2.1/2.2's approved reviews): (1) `xcecli_start()` (`src/lib/client.c`) unconditionally resets `ctx->failover.rejoin_cb = NULL` as part of its own init — any `xcecli_set_reconnect_handler()` call before `xcecli_start()` is silently wiped. Register the reconnect handler AFTER `xcecli_start()` succeeds, not before. Scheduler/apiserver's own reconnect-handler registration (3.1, 4.3) must follow this same ordering or they'll hit the identical silent-wipe bug. (2) A successful `xcecli_join()`/worker join returns the assigned slot index as `reply.status` (any `rc >= 0`, not always exactly `0`) — checking `rc == 0` for success is wrong; use `rc < 0` as the only failure test (matches `xcecd_join_cluster()`'s own established convention). Any new code checking a join/rejoin call's success must use this same `rc < 0` convention, not `rc == 0`. (3) Security lesson: an env-var runtime bypass of a security check (`XCECD_ALLOW_NONROOT`) added for test/E2E convenience was rejected on review — the correct pattern for skipping a production-only check (root privilege, etc.) in a test/virtual build is a compile-time guard against the project's existing `VIRTUAL_CLUSTER` or `XCE_TESTING` flags, never a runtime environment variable reachable in the shipped production binary.
|
||||
- Task 3.2: implemented as a shared `sched_publish_retry_loop()`/`sched_publish_check_rejoin()` pair called from both `sched_publish_dispatch()` and `sched_publish_pend_reason()`, not literally inlined into `sched_publish_dispatch()` alone as design.md's prose names it — reviewed and judged compliant (same thread, fires once after the whole retry loop's successful return, not per-attempt, not in `sched_allocator`). Testing lesson for any future scheduler.c unit test needing to intercept `xcecli_*` calls: `xcecli_jobdispatch`/`xcecli_set_pend_reason`/`xcecli_join`/`xcecli_start`/`xcecli_set_reconnect_handler` all live in the single `client.c` translation unit inside `libxce.a` alongside symbols scheduler.c genuinely needs — the "test double linked ahead of the archive" trick (used by test_driver_decision_cb.c one layer up) causes duplicate-definition errors here. Use `-Wl,--wrap=<symbol>` linker interposition instead (see `test_scheduler_decision_publish_retry.c`/its `meson.build` target) when a test needs scheduler.c itself linked live alongside intercepted libxce calls. Also confirmed: `sched_admin_request_test`'s `test_recordstart_symlink_escape_replies_error` fails independent of this spec's changes (its white-box `#include` list doesn't pull in `scheduler.c` at all) — a pre-existing, unrelated, environment-specific (root-in-sandbox symlink-permission) failure; do not re-investigate it as a regression in later tasks.
|
||||
- Task 4.1: first apiserver-group (Go+CGO) task. `xce_parse_peers()` alone (`src/lib/cluster.c`) does syntax parsing only — no semantic checks (accepts port 0, non-numeric port via silent `atoi()`-to-0, empty host). Worker/scheduler always pair it with a second `xce_cluster_config_validate()` call in their `main.c`s; round-1 review caught `parseCandidates()` (`candidates.go`) skipping that second call, silently accepting malformed candidates worker/scheduler reject — a reminder that "the C function I called is correct" isn't the same as "I called every C function the precedent calls." Fix: call `xce_cluster_config_validate(&cCfg)` after `xce_parse_peers()`, stamping a local placeholder into `cCfg.identity` first since that function also checks a non-empty top-level `identity` (unrelated to per-candidate validation, never propagated to the real connection identity elsewhere in `clib.go`). `xce_cluster_config_validate()` checks `hostname`/`port`/top-level `identity` only, never per-candidate `name` — so the `name=` prefix stays legitimately optional. Established convention confirmed pre-existing (not invented this task): a `//go:build cgo` test file with no `import "C"` of its own, pairing with a plain-Go-signature function in a separate cgo-preamble file (`_test.go` files can never `import "C"` themselves — a Go-toolchain restriction, not just a project style choice) — see `wire_dispatch_test.go`/`wire_dispatch_support.go` for the pre-existing precedent this task's `candidates_test.go`/`candidates.go` split follows. Task 4.1 is parsing-only by design — `RegisterToBackend()`'s connect loop and `tryConnect()` stay untouched; task 4.2 is the one that collapses them into a single connect call, so don't expect this diff to touch that territory. Behavior change (deliberate, disclosed, not a regression): a malformed entry now rejects the WHOLE candidate list (matches worker/scheduler's all-or-nothing precedent, `xce_parse_peers()` has no partial-success mode by construction) instead of the old Go parser's skip-with-warning leniency — surfaces via `RegisterToBackend`'s error → `cmd/main.go`'s `logger.Error`+`os.Exit(1)`, loud not silent. Tooling gap for future apiserver tasks (4.2-4.6): `make test-backend`'s documented CGO env-var setup doesn't actually export `CGO_CFLAGS`/`CGO_LDFLAGS` to a general shell (they only exist as meson `custom_target` env vars) — reconstruct via `pkg-config` against the deps `apiserver/meson.build` lists, or run the full `xce-build.sh apiserver asan` (rebuilds `libxce.a` fresh too) rather than a bare `go test`.
|
||||
- Task 3.4: pure test-coverage task, zero production changes to `scheduler.c`/`driver.c` (correct outcome — task 3.2 already implemented the retry/shutdown/rejoin behavior correctly; 3.4 proves it holds at multi-decision QUEUE granularity through the real drain loop, which task 3.2's own tests never exercised, all 6 of which call `sched_publish_dispatch()`/`sched_publish_pend_reason()` directly on one decision). New `test_scheduler_decision_redelivery.c` links real `scheduler_srcs` with `-Wl,--wrap=xcecli_jobdispatch` (task 3.2's established technique) and drives the real `xce_sched_driver_on_decision()` (driver.c) over 3 queued, distinct-jobid decisions. Two load-bearing tests, both RED→GREEN-verified against the real code by two independent reviewer rounds (not just the implementer's claim): ordering (decision 1 stuck-then-succeeds fully completes before decision 2's call ever fires) and bounded shutdown latency (decisions queued behind one that triggers `core.stop` mid-retry get ZERO publish calls — the concrete, fast, deterministic proof that `sched_publish_retry_loop()`'s check-`stop`-before-each-attempt placement is correct, verified by reviewers moving the check to bottom-of-loop and confirming only this one test catches it). Test-fixture gotcha for any future test building a bare `xce_sched_core_t` via `xce_sched_core_init()` alone (not full `sched_init()`): do NOT call `xce_sched_core_fini()` on it — `fini()` calls `dataq_fini()` on `pend_updates[0/1]`, which only `sched_init()` (not `xce_sched_core_init()`) ever `dataq_init()`s, so `fini()` on a bare core SEGVs on a NULL dataq; omit the `fini()` call entirely, matching `test_core_events.c`'s established "fixtures aren't torn down" convention for this exact minimal-fixture pattern.
|
||||
- Task 3.3: fixed a real production gap found along the way — `sched_init()`'s two all-candidates-unreachable/join-failure log lines logged only a candidate COUNT, never host:port names (same gap worker's task 2.3 fixed for the worker). Added `sched_format_candidate_list()`, threading `sched->cluster_cfg` through as a parameter rather than a new file-scope global (unlike worker's `g_cluster_cfg`) since `test_scheduler_connect.c` links `scheduler_srcs` as separate objects, not `#include`, so the `static` helper isn't directly callable from tests — coverage requires a real `dup2()`-based stdout capture around a live `sched_init()` call hitting the failure path, not a white-box direct call (see `capture_stdout_during_sched_init()` in `test_scheduler_connect.c` — reusable pattern for any future scheduler.c log-content assertion). Round-1 review caught this diagnostic fix shipping with zero test coverage (proven by reverting the 59-line diff and rebuilding: all existing tests still passed) — a reminder that "the fix works, I verified it manually/via live logs" is not the same as a committed regression test; always add the assertion, don't just observe the correct behavior once. The worst-case-occupancy backpressure test (`test_scheduler_backpressure.c`) only proves Req 3 AC2 (buffering/block/resume-with-no-loss) — it does not exercise task 3.2's `-E_XCE_NOT_READY` retry path, so don't cite it for Req 7 coverage (that's task 3.4's job). E2E kill-primary/reconnect and all-unreachable-at-startup were validated manually against a real 3-node `/xce-deploy-ha` cluster rather than committed as an automated test, following worker's task 2.3 precedent — this repo has no automated live-multi-process-deploy E2E harness for any C daemon (the only one, `test_e2e_live_ha_push.py`, is Python/apiserver-only); reviewer judged this acceptable both rounds. Manual validation confirmed join, primary-death survival, and lazy socket-orphaning-on-idle (matches design), but could not observe a live rejoin firing against real sockets — blocked by bug-010 (`xcecd` ASAN "already joined thread" abort under `unshare -r`, needed to bypass the worker daemon's root-only check in this sandbox; unconfirmed whether it's a genuine bug or a userns artifact, never tried under real root). The exact rejoin-after-publish-success code path already has solid white-box mock coverage from task 3.2's `test_rejoin_composition_retries_across_publish_calls`, so this residual gap is narrowly "proven against mocks, not yet against a live socket" — worth closing opportunistically if bug-010 ever gets resolved, not a blocker for this spec.
|
||||
|
|
|
|||
97
src/frontend/apiserver/backend/candidates.go
Normal file
97
src/frontend/apiserver/backend/candidates.go
Normal file
|
|
@ -0,0 +1,97 @@
|
|||
//go:build cgo
|
||||
|
||||
package backend
|
||||
|
||||
/*
|
||||
#include <stdlib.h>
|
||||
#include "xce/cluster.h"
|
||||
*/
|
||||
import "C"
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"unsafe"
|
||||
)
|
||||
|
||||
// Candidate is one parsed controller candidate: an optional logical name
|
||||
// plus the host:port it resolves to. Mirrors xce_candidate_t (xce/cluster.h).
|
||||
type Candidate struct {
|
||||
Name string
|
||||
Host string
|
||||
Port int
|
||||
}
|
||||
|
||||
// parseCandidates parses a comma-joined "[name=]host:port,..." candidate
|
||||
// list through the same shared C parsing+validation pair (xce_parse_peers
|
||||
// then xce_cluster_config_validate, both src/lib/cluster.c) that worker and
|
||||
// scheduler already call directly, in that same order, from their C
|
||||
// main.c's -- retiring apiserver's own hand-written parsePeer() in favor of
|
||||
// it (Req 6.1: apiserver accepts the identical candidate-list syntax
|
||||
// scheduler and worker already accept).
|
||||
//
|
||||
// xce_parse_peers alone is SYNTAX-only: it splits on ','/'='/':' but never
|
||||
// rejects a zero/non-numeric port (atoi() silently defaults malformed input
|
||||
// to 0) or an empty hostname. xce_cluster_config_validate is what actually
|
||||
// rejects those (src/lib/cluster.c:85-100) -- worker/scheduler always call
|
||||
// it immediately after a successful xce_parse_peers
|
||||
// (src/scheduler/src/main.c:300-309, src/worker/src/main.c:414-422), so
|
||||
// this wrapper must too, or apiserver would silently accept candidate
|
||||
// strings worker/scheduler reject (Req 6.2).
|
||||
// xce_cluster_config_validate also requires cfg->identity to be non-empty;
|
||||
// parseCandidates has no real identity to give it at this layer (clib.go's
|
||||
// RegisterToBackend assembles that separately, after parsing, from
|
||||
// operator-supplied From/Key config) -- a placeholder is stamped here
|
||||
// purely to satisfy that one precondition so the per-candidate
|
||||
// hostname/port checks actually run. It is never read or reused beyond
|
||||
// this call.
|
||||
//
|
||||
// On any malformed entry -- whether xce_parse_peers' own syntax check or
|
||||
// xce_cluster_config_validate's semantic check fails -- the WHOLE string is
|
||||
// rejected; neither function identifies which token was the bad one, and
|
||||
// neither does this wrapper (Req 6.2): that matches worker's and
|
||||
// scheduler's own established behavior, which likewise echoes the full
|
||||
// attempted --peers string back on any parse/validate failure (main.c in
|
||||
// both) rather than pinpointing one entry.
|
||||
//
|
||||
// If candidatesCount exceeds MAX_HA_CANDIDATES, xce_parse_peers silently
|
||||
// stops reading further tokens (no error, no warning at this layer) --
|
||||
// this wrapper does not add stricter behavior than the C functions have.
|
||||
func parseCandidates(peersStr string) ([]Candidate, error) {
|
||||
cPeers := C.CString(peersStr)
|
||||
defer C.free(unsafe.Pointer(cPeers))
|
||||
|
||||
malformedErr := func() error {
|
||||
return fmt.Errorf(
|
||||
"apiserver: --peers parse error: expected \"[name=]host:port,...\" got %q",
|
||||
peersStr)
|
||||
}
|
||||
|
||||
var cCfg C.xce_cluster_config_t
|
||||
if rc := C.xce_parse_peers(cPeers, &cCfg); rc != 0 {
|
||||
return nil, malformedErr()
|
||||
}
|
||||
|
||||
// Placeholder identity: satisfies xce_cluster_config_validate's
|
||||
// non-empty-identity precondition only -- see doc comment above. cCfg is
|
||||
// a fresh zero-valued struct, so the array is already NUL-padded past
|
||||
// what we write here.
|
||||
const placeholderIdentity = "apiserver-candidates-validate"
|
||||
for i := 0; i < len(placeholderIdentity); i++ {
|
||||
cCfg.identity[i] = C.char(placeholderIdentity[i])
|
||||
}
|
||||
if rc := C.xce_cluster_config_validate(&cCfg); rc != 0 {
|
||||
return nil, malformedErr()
|
||||
}
|
||||
|
||||
n := int(cCfg.candidates_count)
|
||||
out := make([]Candidate, n)
|
||||
for i := 0; i < n; i++ {
|
||||
cand := &cCfg.candidates[i]
|
||||
out[i] = Candidate{
|
||||
Name: C.GoString(&cand.name[0]),
|
||||
Host: C.GoString(&cand.hostname[0]),
|
||||
Port: int(cand.port),
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
140
src/frontend/apiserver/backend/candidates_test.go
Normal file
140
src/frontend/apiserver/backend/candidates_test.go
Normal file
|
|
@ -0,0 +1,140 @@
|
|||
//go:build cgo
|
||||
|
||||
package backend
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// candidates_test.go exercises parseCandidates (candidates.go), the CGO
|
||||
// wrapper around the shared C parser xce_parse_peers() that replaces
|
||||
// apiserver's retired hand-written parsePeer() (Req 6.1, 6.2). Kept
|
||||
// //go:build cgo since parseCandidates only exists in the CGO build --
|
||||
// this file itself does not `import "C"` (a _test.go file can never do
|
||||
// that directly, golang/go#13938): it only calls the plain-Go-signature
|
||||
// parseCandidates defined in candidates.go, same split already used by
|
||||
// wire_dispatch_test.go/wire_dispatch_support.go in this package.
|
||||
|
||||
func TestParseCandidates_ValidHostPort(t *testing.T) {
|
||||
got, err := parseCandidates("ctrl01:9990")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
want := []Candidate{{Name: "", Host: "ctrl01", Port: 9990}}
|
||||
if len(got) != len(want) || got[0] != want[0] {
|
||||
t.Errorf("got %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseCandidates_NamePrefixSyntax(t *testing.T) {
|
||||
// The old hand-written parsePeer() could not parse this syntax at all
|
||||
// -- this is the format-parity case that proves real parity with
|
||||
// worker/scheduler's --peers (Req 6.1).
|
||||
got, err := parseCandidates("ctrl2=host2:9991")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
want := []Candidate{{Name: "ctrl2", Host: "host2", Port: 9991}}
|
||||
if len(got) != len(want) || got[0] != want[0] {
|
||||
t.Errorf("got %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseCandidates_MultipleCommaJoined(t *testing.T) {
|
||||
got, err := parseCandidates("ctrl01:9990,ctrl2=host2:9991,ctrl03:9992")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
want := []Candidate{
|
||||
{Name: "", Host: "ctrl01", Port: 9990},
|
||||
{Name: "ctrl2", Host: "host2", Port: 9991},
|
||||
{Name: "", Host: "ctrl03", Port: 9992},
|
||||
}
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("got %d candidates, want %d: %+v", len(got), len(want), got)
|
||||
}
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
t.Errorf("candidate[%d]: got %+v, want %+v", i, got[i], want[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseCandidates_MalformedEntry_MissingColon(t *testing.T) {
|
||||
_, err := parseCandidates("ctrl01")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for missing colon, got nil")
|
||||
}
|
||||
// Req 6.2: reject identifying the malformed entry, consistent with how
|
||||
// worker/scheduler report it -- since xce_parse_peers() itself cannot
|
||||
// identify which token was malformed (same limit worker/scheduler's own
|
||||
// --peers parsing has), the whole attempted string is echoed back.
|
||||
if !strings.Contains(err.Error(), "ctrl01") {
|
||||
t.Errorf("error %q does not identify the malformed input", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseCandidates_EmptyString(t *testing.T) {
|
||||
_, err := parseCandidates("")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for empty peers string, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseCandidates_MalformedEntry_MidList(t *testing.T) {
|
||||
// A malformed second entry rejects the WHOLE list -- xce_parse_peers()
|
||||
// does not skip just the bad token and keep the rest (Req 6.2, matching
|
||||
// worker/scheduler's own all-or-nothing --peers parsing).
|
||||
_, err := parseCandidates("ctrl01:9990,badentry,ctrl03:9992")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for malformed mid-list entry, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseCandidates_MalformedEntry_ZeroPort(t *testing.T) {
|
||||
// xce_parse_peers() alone would happily atoi() "0" into port=0 -- it is
|
||||
// xce_cluster_config_validate()'s port<=0 check (src/lib/cluster.c:94)
|
||||
// that rejects it, same as worker/scheduler (Req 6.2).
|
||||
_, err := parseCandidates("ctrl01:0")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for port 0, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseCandidates_MalformedEntry_NonNumericPort(t *testing.T) {
|
||||
// atoi("abc") silently returns 0 -- caught by the same port<=0 check as
|
||||
// the explicit-zero case above, not by xce_parse_peers() itself.
|
||||
_, err := parseCandidates("ctrl01:abc")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for non-numeric port, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseCandidates_MalformedEntry_EmptyHost(t *testing.T) {
|
||||
// xce_cluster_config_validate()'s hostname[0]=='\0' check
|
||||
// (src/lib/cluster.c:92) rejects an empty host, same as worker/scheduler.
|
||||
_, err := parseCandidates(":9990")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for empty host, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseCandidates_OverCapSilentlyTruncates(t *testing.T) {
|
||||
// MAX_HA_CANDIDATES (src/include/config.h) is 4. xce_parse_peers()
|
||||
// silently stops reading further tokens once the cap is reached --
|
||||
// no error, no warning at this layer (the WARN-on-truncation behavior
|
||||
// belongs to xce_cluster_config_to_client_config(), a different
|
||||
// function, not exercised by this task). Don't invent stricter
|
||||
// behavior than the C function actually has.
|
||||
got, err := parseCandidates("c1:1,c2:2,c3:3,c4:4,c5:5")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if len(got) != 4 {
|
||||
t.Fatalf("got %d candidates, want 4 (MAX_HA_CANDIDATES cap)", len(got))
|
||||
}
|
||||
if got[3] != (Candidate{Name: "", Host: "c4", Port: 4}) {
|
||||
t.Errorf("candidate[3] = %+v, want the 4th entry (c5 silently dropped)", got[3])
|
||||
}
|
||||
}
|
||||
|
|
@ -59,6 +59,7 @@ import (
|
|||
logger "apiserver/utils"
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"unsafe"
|
||||
|
||||
|
|
@ -114,22 +115,27 @@ func RegisterToBackend(config *BackendConfig) error {
|
|||
}
|
||||
|
||||
if len(config.Peers) > 0 {
|
||||
// Parsing is routed through the same C parser (xce_parse_peers) worker
|
||||
// and scheduler already call directly -- Req 6.1/6.2. A malformed
|
||||
// entry rejects the whole candidate list up front (matching
|
||||
// worker's/scheduler's own all-or-nothing --peers parsing) rather
|
||||
// than being skipped with a warning as the old hand-written
|
||||
// parsePeer()-per-entry loop did.
|
||||
candidates, err := parseCandidates(strings.Join(config.Peers, ","))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var lastErr error
|
||||
for _, peer := range config.Peers {
|
||||
host, port, err := parsePeer(peer)
|
||||
if err != nil {
|
||||
logger.Warn("Skipping invalid peer", zap.String("peer", peer), zap.Error(err))
|
||||
lastErr = err
|
||||
continue
|
||||
}
|
||||
if err := tryConnect(host, port, from, key); err != nil {
|
||||
for _, cand := range candidates {
|
||||
if err := tryConnect(cand.Host, cand.Port, from, key); err != nil {
|
||||
logger.Warn("Candidate unreachable, trying next",
|
||||
zap.String("host", host), zap.Int("port", port), zap.Error(err))
|
||||
zap.String("host", cand.Host), zap.Int("port", cand.Port), zap.Error(err))
|
||||
lastErr = err
|
||||
continue
|
||||
}
|
||||
logger.Info("Backend connection established (HA)",
|
||||
zap.String("host", host), zap.Int("port", port))
|
||||
zap.String("host", cand.Host), zap.Int("port", cand.Port))
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("all HA candidates failed: %w", lastErr)
|
||||
|
|
|
|||
|
|
@ -8,10 +8,13 @@ type BackendConfig struct {
|
|||
From string // identity sent to controller (default: "xce_apiserver")
|
||||
Key string // authentication key
|
||||
|
||||
// Peers lists additional controller candidates for HA mode in "host:port"
|
||||
// format. When non-empty, RegisterToBackend tries each candidate in order
|
||||
// and connects to whichever accepts the join as primary. Address+Port
|
||||
// serve as a single-host fallback when Peers is empty.
|
||||
// Peers lists additional controller candidates for HA mode, each entry in
|
||||
// "[name=]host:port" format -- the same syntax scheduler's and worker's
|
||||
// --peers flag already accept (Req 6.1), parsed via the shared C parser
|
||||
// xce_parse_peers (candidates.go). When non-empty, RegisterToBackend
|
||||
// tries each candidate in order and connects to whichever accepts the
|
||||
// join as primary. Address+Port serve as a single-host fallback when
|
||||
// Peers is empty.
|
||||
Peers []string
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,24 +0,0 @@
|
|||
package backend
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// parsePeer splits a "host:port" string. Uses net.SplitHostPort so IPv6
|
||||
// bracket notation (e.g. "[::1]:9990") is handled correctly.
|
||||
func parsePeer(peer string) (host string, port int, err error) {
|
||||
h, portStr, err := net.SplitHostPort(peer)
|
||||
if err != nil {
|
||||
return "", 0, fmt.Errorf("invalid peer %q: %w", peer, err)
|
||||
}
|
||||
if h == "" {
|
||||
return "", 0, fmt.Errorf("invalid peer %q: empty host", peer)
|
||||
}
|
||||
port, err = strconv.Atoi(portStr)
|
||||
if err != nil || port <= 0 || port > 65535 {
|
||||
return "", 0, fmt.Errorf("invalid port in peer %q", peer)
|
||||
}
|
||||
return h, port, nil
|
||||
}
|
||||
|
|
@ -1,95 +0,0 @@
|
|||
package backend
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestParsePeer(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
peer string
|
||||
wantHost string
|
||||
wantPort int
|
||||
wantError bool
|
||||
}{
|
||||
{
|
||||
name: "valid simple host:port",
|
||||
peer: "ctrl01:9990",
|
||||
wantHost: "ctrl01",
|
||||
wantPort: 9990,
|
||||
},
|
||||
{
|
||||
name: "valid IPv4",
|
||||
peer: "192.168.1.10:9990",
|
||||
wantHost: "192.168.1.10",
|
||||
wantPort: 9990,
|
||||
},
|
||||
{
|
||||
name: "valid FQDN",
|
||||
peer: "ctrl01.cluster.local:9990",
|
||||
wantHost: "ctrl01.cluster.local",
|
||||
wantPort: 9990,
|
||||
},
|
||||
{
|
||||
name: "min valid port",
|
||||
peer: "host:1",
|
||||
wantHost: "host",
|
||||
wantPort: 1,
|
||||
},
|
||||
{
|
||||
name: "max valid port",
|
||||
peer: "host:65535",
|
||||
wantHost: "host",
|
||||
wantPort: 65535,
|
||||
},
|
||||
{
|
||||
name: "missing colon",
|
||||
peer: "ctrl019990",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "port zero",
|
||||
peer: "ctrl01:0",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "port out of range",
|
||||
peer: "ctrl01:65536",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "non-numeric port",
|
||||
peer: "ctrl01:abc",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "colon at start (empty host)",
|
||||
peer: ":9990",
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "empty string",
|
||||
peer: "",
|
||||
wantError: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
host, port, err := parsePeer(tc.peer)
|
||||
if tc.wantError {
|
||||
if err == nil {
|
||||
t.Errorf("parsePeer(%q): expected error, got host=%q port=%d", tc.peer, host, port)
|
||||
}
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("parsePeer(%q): unexpected error: %v", tc.peer, err)
|
||||
}
|
||||
if host != tc.wantHost {
|
||||
t.Errorf("host: got %q, want %q", host, tc.wantHost)
|
||||
}
|
||||
if port != tc.wantPort {
|
||||
t.Errorf("port: got %d, want %d", port, tc.wantPort)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue