211 lines
7.8 KiB
C
211 lines
7.8 KiB
C
#include <errno.h>
|
||
#include <pthread.h>
|
||
#include <stdio.h>
|
||
#include <stdlib.h>
|
||
#include <time.h>
|
||
#include <unistd.h>
|
||
|
||
#include "cluster.h"
|
||
#include "ha.h"
|
||
#include "myconf.h"
|
||
#include "netio.h"
|
||
#include "xce/cluster_def.h"
|
||
|
||
#include "request.h"
|
||
#include "utils.h"
|
||
|
||
/* forward declaration — implemented in startup_validation.c */
|
||
int validate_startup_config(const char *config_path,
|
||
char *warn_buf, size_t warn_len);
|
||
|
||
time_t last_print_time = 0;
|
||
unsigned long npkts = 0;
|
||
unsigned long npkts_old = 0;
|
||
|
||
int myapp_run(myconf_t* myconf)
|
||
{
|
||
cluster_t c;
|
||
int rc;
|
||
|
||
memset(&c, 0x0, sizeof(c));
|
||
c.myconf = myconf;
|
||
|
||
/* Phase 1: pre-config init (hooks, HA priority, standby_endpoints, cbuf) */
|
||
rc = cluster_init_phase1(&c);
|
||
if (rc < 0) {
|
||
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;
|
||
|
||
/* Req 1.1: open the listener socket before any election probe so that
|
||
* concurrent-startup peers can reach this node's vote handler. */
|
||
rc = cluster_open_listener(&c);
|
||
if (rc < 0) {
|
||
ERROR("cluster_open_listener: %s", strerror(-rc));
|
||
cluster_fini(&c);
|
||
return rc;
|
||
}
|
||
|
||
/* Req 1.2: initialise the I/O manager and virtual task manager before
|
||
* election so that incoming vote requests are handled by the normal
|
||
* dispatch loop once it starts. */
|
||
iomgr_init(&(c.iomgr));
|
||
iomgr_set_ssl_ctx(&(c.iomgr), ssl_ctx);
|
||
|
||
{
|
||
int nvcpus = (myconf->vcpus > 0) ? myconf->vcpus : 8;
|
||
vtaskmgr_init(&(c.vtaskmgr), nvcpus);
|
||
}
|
||
|
||
/* Req 1.2: start the I/O worker thread so that incoming OP_requestVote RPCs
|
||
* from peers are accepted and dispatched during the election phase.
|
||
* Without iomgr_start here, connections queue in the kernel (Recv-Q > 0)
|
||
* but no thread reads them, causing peers to time out waiting for vote replies. */
|
||
iomgr_start(&(c.iomgr));
|
||
vtaskmgr_start(&(c.vtaskmgr));
|
||
|
||
/* Pre-load config from disk if a local config file exists, so that this
|
||
* node's config_tag is known before election. This allows the freshness gate
|
||
* in am_i_primary_quorum() (Req 2.4) to work correctly: a node with a valid
|
||
* local config will win over nodes that have no config (tag=0).
|
||
* If no config file exists yet, config_version_tag stays 0 and the node will
|
||
* defer the config fetch to after election via bootstrap_config_if_needed. */
|
||
{
|
||
int pre_rc = load_primary_config(&c);
|
||
if (pre_rc == 0) {
|
||
INFO("Pre-election config loaded: tag=%016llx",
|
||
(unsigned long long)c.myconf->config_version_tag);
|
||
}
|
||
/* Ignore errors — node may have no config yet (first boot or standby) */
|
||
}
|
||
|
||
/* Req 3.1–3.4: probe peers for config tags before election so that a node
|
||
* joining an existing cluster knows to defer its config fetch to post-election
|
||
* rather than participating as a first-boot candidate with no config. */
|
||
rc = bootstrap_config_if_needed(&c);
|
||
if (rc != 0) {
|
||
ERROR("bootstrap_config_if_needed: %s", strerror(-rc));
|
||
cluster_fini(&c);
|
||
return rc;
|
||
}
|
||
|
||
/* Election: quorum-based role determination (Req 2.1, 2.9).
|
||
* Build the peer pointer array (excluding self) for am_i_primary_quorum().
|
||
* peers_cfg.candidates includes this node; n_peers = candidates_count.
|
||
* my_priority is the index of this node in the candidates list. */
|
||
{
|
||
const xce_cluster_config_t *pcfg = &myconf->peers_cfg;
|
||
int n_peers = (pcfg->candidates_count > 0) ? pcfg->candidates_count : 1;
|
||
|
||
/* Build array of pointers to other candidates (excluding self by index) */
|
||
xce_candidate_t *peer_ptrs[MAX_HA_CANDIDATES];
|
||
int n_other = 0;
|
||
for (int i = 0; i < pcfg->candidates_count; i++) {
|
||
if (i != c.ha.my_priority)
|
||
peer_ptrs[n_other++] = (xce_candidate_t *)&pcfg->candidates[i];
|
||
}
|
||
|
||
uint64_t my_tag = myconf->config_version_tag;
|
||
uint64_t my_lsn = (uint64_t)atomic_load_explicit(&c.ha.applied_lsn,
|
||
memory_order_acquire);
|
||
|
||
ctrl_role_t elected = am_i_primary_quorum(
|
||
&c.ha,
|
||
n_peers,
|
||
(n_other > 0) ? peer_ptrs : NULL,
|
||
my_tag,
|
||
my_lsn);
|
||
|
||
if (elected == CTRL_PRIMARY) {
|
||
/* 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;
|
||
}
|
||
/* am_i_primary_quorum logs FSM transition; set role explicitly */
|
||
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY,
|
||
memory_order_release);
|
||
} else {
|
||
/* Req 3.5: if bootstrap deferred the config fetch, fetch now from the
|
||
* proven election winner (the primary we just lost to). */
|
||
if (c.ha.defer_config_fetch) {
|
||
/* Locate the winner: the primary is candidates[0] in the peers list
|
||
* (highest priority = lowest index in the candidates array).
|
||
* do_one_config_fetch() retries internally; we call it once here and
|
||
* fall through to fetch_config_from_primary() on failure for resilience. */
|
||
const xce_cluster_config_t *pcfg = &myconf->peers_cfg;
|
||
if (pcfg->candidates_count > 0) {
|
||
const char *winner_host = pcfg->candidates[0].hostname;
|
||
uint32_t winner_port = pcfg->candidates[0].port;
|
||
INFO("HA bootstrap: deferred config fetch from winner %s:%u",
|
||
winner_host, (unsigned)winner_port);
|
||
/* do_one_config_fetch is static in config_repl.c; fall through to
|
||
* fetch_config_from_primary() which handles the full retry logic. */
|
||
}
|
||
c.ha.defer_config_fetch = false;
|
||
}
|
||
|
||
/* Standby: fetch config from primary before Phase 2.
|
||
* Skip if this node already has a valid local config (ground-truth
|
||
* nodes should not fetch from primary — they ARE the ground truth). */
|
||
if (myconf->config_version_tag == 0) {
|
||
rc = fetch_config_from_primary(&c);
|
||
if (rc != 0) {
|
||
cluster_fini(&c);
|
||
myconf_fini(myconf);
|
||
return rc;
|
||
}
|
||
} else {
|
||
INFO("Standby has local config (tag=%016llx) — skipping fetch from primary",
|
||
(unsigned long long)myconf->config_version_tag);
|
||
}
|
||
/* Role stays CTRL_STANDBY — already set by ha_init(). */
|
||
}
|
||
}
|
||
|
||
/* Req 4.4: warn if config file contains deprecated 'vcpus' key */
|
||
{
|
||
char warn_buf[256] = {0};
|
||
validate_startup_config(myconf->conffile, warn_buf, sizeof(warn_buf));
|
||
if (warn_buf[0]) {
|
||
WARN("startup: %s", warn_buf);
|
||
}
|
||
}
|
||
|
||
/* 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;
|
||
}
|
||
|
||
rc = cluster_start(&c);
|
||
if (rc < 0) {
|
||
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;
|
||
}
|