From 6b06aa02d99d4e0cce1d71a4acd0f99eed47f01b Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 13 May 2026 17:06:45 +0000 Subject: [PATCH] feat: split cluster_init() into phase1/phase2; update myapp_run() for two-phase startup with am_i_primary gating MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - cluster_init_phase1(): hooks, HA init (priority from --peers), standby_endpoints zarray, cbuf - cluster_init_phase2(): SSL ctx from ha.ssl_ctx, cluster_config guard, kvmgr sizing, iomgr, endpoint, persist, role promotion, create_lost_and_found, persist_recover - cluster_init(): legacy wrapper calling both phases with inline SSL init for backward compat - myapp_run(): phase1 → SSL init → am_i_primary → load_primary_config/fetch_config_from_primary → phase2 → start → run - myreq_handler(): add OPCode_replConfig and OPCode_upgradeConfig dispatch - handle_endpoint_error(): prune standby_endpoints on HostRole_controller disconnect - cluster_run(): log restart_requested on exit - config_repl.c: add create_lost_and_found, handle_config_request, handle_upgrade_config stubs - config_repl.h: add declarations for new stubs; include netio.h + request.h - cluster.h: declare cluster_init_phase1 and cluster_init_phase2 - myconf.c: fix FATAL → ERROR (FATAL macro undefined); add fetch_config_from_primary stub --- src/controller/cluster.c | 277 ++++++++++++++++++++++--------- src/controller/config_repl.c | 26 +++ src/controller/config_repl.h | 22 +++ src/controller/include/cluster.h | 2 + src/controller/myapp.c | 50 +++++- src/controller/myconf.c | 22 ++- 6 files changed, 307 insertions(+), 92 deletions(-) diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 9859170..d8667d3 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -2,10 +2,12 @@ #include #include +#include #include #include #include "cluster.h" +#include "config_repl.h" #include "dataq.h" #include "htab.h" #include "kvmgr.h" @@ -67,6 +69,19 @@ static void handle_endpoint_error(endpoint_t* ep) handle_scheduler_leave(c); break; case HostRole_controller: + if (c->ha.standby_endpoints) { + pthread_mutex_lock(&c->ha.lock); + for (int idx = 0; + idx < c->ha.standby_endpoints->count; idx++) { + endpoint_t *se = (endpoint_t *)zarray_get( + c->ha.standby_endpoints, idx); + if (se == ep) { + zarray_del(c->ha.standby_endpoints, idx); + break; + } + } + pthread_mutex_unlock(&c->ha.lock); + } break; case HostRole_client: handle_client_leave(c); @@ -201,6 +216,34 @@ static int myreq_handler(vtask_t* task) case OPCode_clrpreason: rc = do_clr_preason(c, ep, &request); break; + case OPCode_replConfig: + rc = handle_config_request(c, ep, &request); + break; + case OPCode_upgradeConfig: +#if CONF_ENABLE_UPGRADE_CONFIG + if (!ha_is_primary(&c->ha)) { + /* Standby cannot handle upgrade — send error reply */ + xce_reply_t ureply; + init_reply_by_req(&ureply, &request); + ureply.status = -EAGAIN; + ureply.data_which = RequestReply_data_nodata; + xce_reply(ep, c->myhost, &ureply); + rc = -EAGAIN; + break; + } + rc = handle_upgrade_config(c, ep, &request); + break; +#else + { + xce_reply_t ureply; + init_reply_by_req(&ureply, &request); + ureply.status = -ENOSYS; + ureply.data_which = RequestReply_data_nodata; + xce_reply(ep, c->myhost, &ureply); + rc = -ENOSYS; + } + break; +#endif default: rc = -E_XCE_BAD_OPCODE; ERROR("Unknown opcode: %d", request.opcode); @@ -510,10 +553,70 @@ static void config_obj_killer(void* arg) } } -int cluster_init(cluster_t* c) +int cluster_init_phase1(cluster_t* c) +{ + int rc; + + if (c == NULL) + return -EINVAL; + + c->myhost = &from_controller; + c->ncontrollers = 1; + c->myidx = 0; + + xce_hook_init(&(c->hooksys), XCE_CONTROLLER_HOOK_MAX); + xce_install_hooks(c); + + /* HA config and priority — derived from --peers list before config loads */ + ha_config_t ha_cfg = { + .heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS, + .heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS, + .election_timeout_ms = HA_DEFAULT_ELECTION_TIMEOUT_MS, + .probe_connect_timeout_ms = HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS, + .max_promotion_retries = HA_DEFAULT_MAX_PROMOTION_RETRIES, + .election_jitter_max_ms = HA_DEFAULT_ELECTION_JITTER_MAX_MS, + .election_priority_step_ms = HA_DEFAULT_ELECTION_PRIORITY_STEP_MS, + }; + + int my_priority = 0; + const xce_cluster_config_t* pcfg = &c->myconf->peers_cfg; + if (pcfg->candidates_count > 0) { + my_priority = pcfg->candidates_count - 1; /* default: last = lowest */ + for (int i = 0; i < pcfg->candidates_count; i++) { + const char* key = pcfg->candidates[i].name[0] + ? pcfg->candidates[i].name + : pcfg->candidates[i].hostname; + if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) { + my_priority = i; + break; + } + } + INFO("HA peers: %d candidates, my_priority=%d (name=%s)", + pcfg->candidates_count, my_priority, c->myconf->name); + } + + rc = ha_init(&c->ha, &ha_cfg, my_priority); + if (rc != 0) { + ERROR("ha_init failed: %d", rc); + return rc; + } + ha_install_sigusr1(&c->ha); + + /* Initialize standby endpoint tracking */ + c->ha.standby_endpoints = (zarray_t *)calloc(1, sizeof(zarray_t)); + if (c->ha.standby_endpoints) { + zarray_init(c->ha.standby_endpoints, 4, NULL); + } + + /* cbuf used by cluster_run() to signal restart/stop */ + cbuf_init(&(c->wbuf), 1, sizeof(int)); + + return 0; +} + +int cluster_init_phase2(cluster_t* c) { int i, rc; - SSL_CTX* ssl_ctx = NULL; kvmgr_params_t config_mgr_params = { .key_type = KVMGR_KEY_STRING, .nslots = 128, @@ -532,34 +635,30 @@ int cluster_init(cluster_t* c) = { .key_type = KVMGR_KEY_STRING, .datakiller = (destroy_fn_t)queue_obj_killer }; - if (c == NULL) { + if (c == NULL) + return -EINVAL; + + if (c->myconf->cluster_config == NULL) { + ERROR("cluster_init_phase2: cluster_config not loaded"); return -EINVAL; } - c->myhost = &from_controller; - /* FIXME: */ - c->ncontrollers = 1; - c->myidx = 0; - - xce_hook_init(&(c->hooksys), XCE_CONTROLLER_HOOK_MAX); - - xce_install_hooks(c); - - rc = xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, - HOOK_BEFORE_EVENT, HOOK_DETAIL_CLUSTER_INIT, c); - if (rc < 0) { - return rc; + SSL_CTX* ssl_ctx = (SSL_CTX*)c->ha.ssl_ctx; + if (ssl_ctx == NULL) { + ERROR("cluster_init_phase2: ssl_ctx not set (myapp_run must set " + "c->ha.ssl_ctx before calling phase2)"); + return -EINVAL; } + /* HOOK_BEFORE_EVENT for CLUSTER_INIT loads cluster_config from disk via + * xce_config_init(). In the two-phase path, cluster_config is already + * populated by load_primary_config() / fetch_config_from_primary() before + * phase2 is called — so we do NOT re-run the hook here. + * The legacy cluster_init() wrapper runs it before calling phase2. */ + mylock_init(&(c->sched.lock)); c->sched.ep = NULL; - initialize_openssl(); - ssl_ctx = create_ssl_context(c->myconf->certfile, c->myconf->keyfile); - if (ssl_ctx == NULL) { - return -errno; - } - resmgr_params.nslots = c->myconf->cluster_config->resmgr->nhash_slots; resmgr_params.delta = c->myconf->cluster_config->resmgr->delta; jobmgr_params.nslots = c->myconf->cluster_config->jobmgr->nhash_slots; @@ -570,9 +669,8 @@ int cluster_init(cluster_t* c) qmgr_params.delta = c->myconf->cluster_config->qmgr->delta; c->configmgr = makeInstance(&xkvmgr_type, &config_mgr_params); - - c->jobmgr = makeInstance(&xkvmgr_type, &jobmgr_params); - c->resmgr = makeInstance(&xkvmgr_type, &resmgr_params); + c->jobmgr = makeInstance(&xkvmgr_type, &jobmgr_params); + c->resmgr = makeInstance(&xkvmgr_type, &resmgr_params); zarray_init(&(c->sharedinst), c->myconf->cluster_config->resmgr->inst_delta, (zarray_item_killer_t)resval_obj_killer); c->workermgr = makeInstance(&xkvmgr_type, &workermgr_params); @@ -586,10 +684,26 @@ int cluster_init(cluster_t* c) zarray_init(&(c->endpoints), 16, endpoint_killer); - rc = endpoint_init(&(c->ep), AF_INET, SOCK_STREAM, - c->myconf->cluster_config->port, 5); + /* Determine listen port: prefer per-candidate port from --peers list, + * fall back to cluster_config->port */ + uint32_t my_port = 0; + { + const xce_cluster_config_t* pcfg = &c->myconf->peers_cfg; + for (int j = 0; j < pcfg->candidates_count; j++) { + const char* key = pcfg->candidates[j].name[0] + ? pcfg->candidates[j].name + : pcfg->candidates[j].hostname; + if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) { + my_port = pcfg->candidates[j].port; + break; + } + } + } + if (my_port == 0) + my_port = (uint32_t)c->myconf->cluster_config->port; + + rc = endpoint_init(&(c->ep), AF_INET, SOCK_STREAM, (int)my_port, 5); if (rc < 0) { - cleanup_openssl(); return rc; } @@ -599,8 +713,6 @@ int cluster_init(cluster_t* c) controller_init(&(c->controllers[i]), c); } - cbuf_init(&(c->wbuf), 1, sizeof(int)); - zarray_init(&(c->configs), 64, NULL /* FIXME */); vtaskmgr_init(&(c->vtaskmgr), c->myconf->cluster_config->vcpus); @@ -628,59 +740,19 @@ int cluster_init(cluster_t* c) } } - /* HA initialisation */ - { - ha_config_t ha_cfg = { - .heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS, - .heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS, - .election_timeout_ms = HA_DEFAULT_ELECTION_TIMEOUT_MS, - .probe_connect_timeout_ms = HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS, - .max_promotion_retries = HA_DEFAULT_MAX_PROMOTION_RETRIES, - }; - - /* Derive priority from position in --peers list. - * Index 0 = highest priority (becomes primary on startup). - * If the controller's name is not found in the list, fall back to - * the last index (lowest priority = standby). */ - int my_priority = 0; - const xce_cluster_config_t* pcfg = &c->myconf->peers_cfg; - if (pcfg->candidates_count > 0) { - my_priority = pcfg->candidates_count - 1; /* default: last = lowest */ - for (int i = 0; i < pcfg->candidates_count; i++) { - /* Match by logical name first (supports multiple controllers - * on the same dev host via name=host:port syntax). Fall back - * to hostname match when no name was specified. */ - const char* key = pcfg->candidates[i].name[0] - ? pcfg->candidates[i].name - : pcfg->candidates[i].hostname; - if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) { - my_priority = i; - break; - } - } - INFO("HA peers: %d candidates, my_priority=%d (name=%s)", - pcfg->candidates_count, my_priority, c->myconf->name); - } - - rc = ha_init(&c->ha, &ha_cfg, my_priority); - if (rc != 0) { - ERROR("ha_init failed: %d", rc); - persist_fini(&c->persist); - return rc; - } - ha_install_sigusr1(&c->ha); - - /* Determine initial role: priority-0 node starts as primary */ - if (am_i_primary(c)) - ha_promote_to_primary(c); - else - atomic_store_explicit(&c->ha.role, (int32_t)CTRL_STANDBY, - memory_order_release); - } + /* Determine initial role based on am_i_primary() result set by myapp_run */ + if (ha_is_primary(&c->ha)) + ha_promote_to_primary(c); + else + atomic_store_explicit(&c->ha.role, (int32_t)CTRL_STANDBY, + memory_order_release); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT, HOOK_DETAIL_CLUSTER_INIT, c); + /* Ensure lost_and_found synthetic queue/host exist before recovery */ + create_lost_and_found(c); + /* Recover WAL/snapshot now that resmgr, jobmgr, and workermgr are * fully populated by cluster_build_resources() above. */ rc = persist_recover(&c->persist); @@ -693,6 +765,42 @@ int cluster_init(cluster_t* c) return 0; } +int cluster_init(cluster_t* c) +{ + int rc; + + /* Legacy wrapper: run both phases with SSL init inline. + * Used by tests and any caller that provides a fully populated myconf_t + * (including cluster_config) before calling cluster_init(). */ + if (c == NULL) + return -EINVAL; + + rc = cluster_init_phase1(c); + if (rc != 0) + return rc; + + /* Create SSL context for the wrapper path */ + initialize_openssl(); + SSL_CTX* ssl_ctx = create_ssl_context(c->myconf->certfile, + c->myconf->keyfile); + if (ssl_ctx == NULL) { + ERROR("create_ssl_context failed"); + return -errno; + } + c->ha.ssl_ctx = ssl_ctx; + + /* For the legacy path, run the before-event hook to populate + * cluster_config via xce_config_init() if it isn't already set */ + if (c->myconf->cluster_config == NULL) { + rc = xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, + HOOK_BEFORE_EVENT, HOOK_DETAIL_CLUSTER_INIT, c); + if (rc < 0) + return rc; + } + + return cluster_init_phase2(c); +} + static int make_heartbeat(cluster_t* c, endpoint_t* ep) { return xce_nop_req(ep); @@ -876,11 +984,22 @@ int cluster_run(cluster_t* c) iomgr_unmonitor(&(c->iomgr), &(c->ep)); iomgr_stop(&(c->iomgr)); + if (atomic_load(&c->restart_requested)) { + INFO("Config upgrade restart requested — exiting for service manager restart"); + } + return 0; } void cluster_fini(cluster_t* c) { + /* Free heap-allocated standby_endpoints array (created in phase1) */ + if (c->ha.standby_endpoints) { + zarray_fini(c->ha.standby_endpoints); + free(c->ha.standby_endpoints); + c->ha.standby_endpoints = NULL; + } + /* Stop HA watchdog thread before tearing down other subsystems — P1-5 */ ha_fini(&c->ha); diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 152907c..f8b132f 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -388,6 +388,32 @@ int load_primary_config_for_test(cluster_t *c) } #endif +/* ── create_lost_and_found ──────────────────────── */ + +void create_lost_and_found(cluster_t *c) { (void)c; /* Task 10 */ } + +/* ── handle_config_request ──────────────────────── */ + +int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req) +{ + (void)c; + (void)ep; + (void)req; + /* Task 9: full implementation — read versioned blob, send reply */ + return -ENOSYS; +} + +/* ── handle_upgrade_config ──────────────────────── */ + +int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req) +{ + (void)c; + (void)ep; + (void)req; + /* Task 9: full implementation — validate, store, fan-out, restart */ + return -ENOSYS; +} + /* ── sanitise_cn ────────────────────────────────── */ char *sanitise_cn(const char *src, size_t maxlen) diff --git a/src/controller/config_repl.h b/src/controller/config_repl.h index 2f2ebd2..69b1eeb 100644 --- a/src/controller/config_repl.h +++ b/src/controller/config_repl.h @@ -7,6 +7,8 @@ #include "myconf.h" #include "clusterconfig.h" +#include "netio.h" +#include "request.h" /* Max config blob size — reject before allocating */ #define CONF_MAX_SIZE (1 * 1024 * 1024) /* 1 MiB */ @@ -84,6 +86,26 @@ void conf_trim_old_versions(const char *workdir, int keep_n); */ int config_contains_reserved_name(cluster_config_t *cfg); +/* + * create_lost_and_found: ensure the synthetic "lost_and_found" queue and + * host exist before persist_recover() runs. Stub for Task 10. + */ +void create_lost_and_found(cluster_t *c); + +/* + * handle_config_request: handle an OP_replConfig request from a standby. + * Reads the current versioned config blob and sends it back. + * Stub for Task 8 (full implementation in Task 9). + */ +int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req); + +/* + * handle_upgrade_config: handle an OP_upgradeConfig request from an admin tool. + * Validates, stores, and fans out the new config blob; triggers restart. + * Stub for Task 9. Only callable when CONF_ENABLE_UPGRADE_CONFIG is set. + */ +int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req); + /* * sanitise_cn: copy at most maxlen printable ASCII chars from src to a * static buffer, replacing non-printable/control chars with '?'. diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index d249004..25b8c75 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -22,6 +22,8 @@ typedef int (*collect_check_fn_t)(void*); extern int check_controller(void*); +extern int cluster_init_phase1(cluster_t*); +extern int cluster_init_phase2(cluster_t*); extern int cluster_init(cluster_t*); extern int cluster_start(cluster_t*); extern int cluster_run(cluster_t*); diff --git a/src/controller/myapp.c b/src/controller/myapp.c index e7d8706..11ff22e 100644 --- a/src/controller/myapp.c +++ b/src/controller/myapp.c @@ -23,17 +23,51 @@ int myapp_run(myconf_t* myconf) int rc; memset(&c, 0x0, sizeof(c)); - c.myconf = myconf; - rc = cluster_init(&c); + /* Phase 1: pre-config init (hooks, HA priority, standby_endpoints, cbuf) */ + rc = cluster_init_phase1(&c); if (rc < 0) { - SET_ERROR(-rc); - - ERROR("cluster_init: %s", strerror(-rc)); - + ERROR("cluster_init_phase1: %s", strerror(-rc)); cluster_fini(&c); + return rc; + } + /* SSL context — created after Phase 1, used by election probing and Phase 2 */ + initialize_openssl(); + SSL_CTX* ssl_ctx = create_ssl_context(myconf->certfile, myconf->keyfile); + if (!ssl_ctx) { + ERROR("create_ssl_context failed"); + cluster_fini(&c); + return -EINVAL; + } + c.ha.ssl_ctx = ssl_ctx; + + /* Election: determine role before loading config */ + if (am_i_primary(&c)) { + /* Primary: load config from conf/current (FATAL if missing) */ + rc = load_primary_config(&c); + if (rc != 0) { + cluster_fini(&c); + myconf_fini(myconf); + return rc; + } + } else { + /* Standby: fetch config from primary before Phase 2 */ + rc = fetch_config_from_primary(&c); + if (rc != 0) { + cluster_fini(&c); + myconf_fini(myconf); + return rc; + } + } + + /* Phase 2: full init (requires cluster_config) */ + rc = cluster_init_phase2(&c); + if (rc < 0) { + ERROR("cluster_init_phase2: %s", strerror(-rc)); + cluster_fini(&c); + myconf_fini(myconf); return rc; } @@ -42,13 +76,13 @@ int myapp_run(myconf_t* myconf) SET_ERROR(-rc); perror("cluster_start"); cluster_fini(&c); - + myconf_fini(myconf); return rc; } rc = cluster_run(&c); cluster_fini(&c); - + myconf_fini(myconf); return rc; } diff --git a/src/controller/myconf.c b/src/controller/myconf.c index 662eb61..bd9905b 100644 --- a/src/controller/myconf.c +++ b/src/controller/myconf.c @@ -166,7 +166,7 @@ int load_primary_config(cluster_t *c) char fname[256] = {0}; int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname)); if (rc != 0) { - FATAL("Primary has no config at '%s/conf/current' (rc=%d) — " + ERROR("Primary has no config at '%s/conf/current' (rc=%d) — " "provision the config before starting", conf_dir, rc); return -ENOENT; @@ -183,7 +183,7 @@ int load_primary_config(cluster_t *c) size_t blen = 0; rc = conf_read_raw(myconf->conffile, &bytes, &blen); if (rc != 0) { - FATAL("Primary config '%s' unreadable: %d", myconf->conffile, rc); + ERROR("Primary config '%s' unreadable: %d", myconf->conffile, rc); free(myconf->conffile); myconf->conffile = NULL; return rc; @@ -205,7 +205,7 @@ int load_primary_config(cluster_t *c) /* Decode Cap'n Proto directly into myconf->cluster_config */ rc = load_cluster_config_from_file(myconf->conffile, &myconf->cluster_config); if (rc != 0) { - FATAL("Primary config '%s' failed Cap'n Proto decode: %d", + ERROR("Primary config '%s' failed Cap'n Proto decode: %d", myconf->conffile, rc); free(bytes); free(myconf->conffile); @@ -214,7 +214,7 @@ int load_primary_config(cluster_t *c) } if (myconf->cluster_config == NULL) { - FATAL("Primary config '%s' decoded to NULL", myconf->conffile); + ERROR("Primary config '%s' decoded to NULL", myconf->conffile); free(bytes); free(myconf->conffile); myconf->conffile = NULL; @@ -227,7 +227,7 @@ int load_primary_config(cluster_t *c) /* Check for reserved lost_and_found name */ if (config_contains_reserved_name(myconf->cluster_config)) { - FATAL("Config '%s' contains reserved name 'lost_and_found'", + ERROR("Config '%s' contains reserved name 'lost_and_found'", myconf->conffile); free_ClusterConfig(myconf->cluster_config); free(myconf->cluster_config); @@ -253,6 +253,18 @@ int load_primary_config(cluster_t *c) return 0; } +/* ── fetch_config_from_primary ──────────────────── */ + +int fetch_config_from_primary(cluster_t *c) +{ + (void)c; + /* Task 9: full implementation — connect to primary, send OP_replConfig, + * receive blob, verify SHA-256, call conf_save_config(), decode into + * myconf->cluster_config. */ + ERROR("fetch_config_from_primary: not yet implemented (Task 9)"); + return -ENOSYS; +} + /* ── myconf_fini ────────────────────────────────── */ void myconf_fini(myconf_t *myconf)