diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index f8b132f8..6598b4f4 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -16,11 +16,155 @@ #include #include +#include + #include "byteorder.h" #include "clusterconfig.h" +#include "ha.h" #include "utils.h" #include "xce/cluster_def.h" +/* ── config load helpers (shared with myconf.c logic) ─────── */ + +#define SET_DEFAULT_IF_UNSET(var, unset, val) \ + do { \ + if ((var) == (unset)) { \ + (var) = (val); \ + } \ + } while (0) + +static void fill_default_cluster_config(cluster_config_t *c) +{ + SET_DEFAULT_IF_UNSET(c->port, 0, 9990); + SET_DEFAULT_IF_UNSET(c->heartbeat_interval, 0, 5); + SET_DEFAULT_IF_UNSET(c->vcpus, 0, 8); + SET_DEFAULT_IF_UNSET(c->fetch_job_interval, 0, 10); +} + +static int load_cluster_config_from_file(char *file, cluster_config_t **config) +{ + struct capn c; + ClusterConfig_ptr p; + FILE *fp; + int rc = 0; + + fp = fopen(file, "rb"); + if (fp == NULL) { + rc = -errno; + return rc; + } + + capn_init_fp(&c, fp, 0); + p.p = capn_getp(capn_root(&c), 0, 1); + decode_ClusterConfig_ptr(config, p); + capn_free(&c); + fclose(fp); + return 0; +} + +/* ── load_primary_config ────────────────────────── */ + +int load_primary_config(cluster_t *c) +{ + myconf_t *myconf = c->myconf; + + /* Build conf dir path: / */ + char conf_dir[4096]; + snprintf(conf_dir, sizeof(conf_dir), "%s/%s", + myconf->workdir, myconf->name); + + /* Resolve conf/current → versioned filename */ + char fname[256] = {0}; + int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname)); + if (rc != 0) { + ERROR("Primary has no config at '%s/conf/current' (rc=%d) — " + "provision the config before starting", + conf_dir, rc); + return -ENOENT; + } + + /* Heap-allocate conffile so it survives past this stack frame */ + free(myconf->conffile); + myconf->conffile = NULL; + if (asprintf(&myconf->conffile, "%s/conf/%s", conf_dir, fname) < 0) + return -ENOMEM; + + /* Read raw bytes and check size */ + uint8_t *bytes = NULL; + size_t blen = 0; + rc = conf_read_raw(myconf->conffile, &bytes, &blen); + if (rc != 0) { + ERROR("Primary config '%s' unreadable: %d", myconf->conffile, rc); + free(myconf->conffile); + myconf->conffile = NULL; + return rc; + } + + /* Compute and cache SHA-256 */ + SHA256(bytes, blen, myconf->config_sha256); + uint64_t tag; + memcpy(&tag, myconf->config_sha256, 8); + myconf->config_version_tag = be64toh(tag); + + /* Free any prior decoded config before overwriting the pointer */ + if (myconf->cluster_config) { + free_ClusterConfig(myconf->cluster_config); + free(myconf->cluster_config); + myconf->cluster_config = NULL; + } + + /* Decode Cap'n Proto directly into myconf->cluster_config */ + rc = load_cluster_config_from_file(myconf->conffile, + &myconf->cluster_config); + if (rc != 0) { + ERROR("Primary config '%s' failed Cap'n Proto decode: %d", + myconf->conffile, rc); + free(bytes); + free(myconf->conffile); + myconf->conffile = NULL; + return rc; + } + + if (myconf->cluster_config == NULL) { + ERROR("Primary config '%s' decoded to NULL", myconf->conffile); + free(bytes); + free(myconf->conffile); + myconf->conffile = NULL; + return -EINVAL; + } + + free(bytes); + + fill_default_cluster_config(myconf->cluster_config); + + /* Check for reserved lost_and_found name */ + if (config_contains_reserved_name(myconf->cluster_config)) { + ERROR("Config '%s' contains reserved name 'lost_and_found'", + myconf->conffile); + free_ClusterConfig(myconf->cluster_config); + free(myconf->cluster_config); + myconf->cluster_config = NULL; + free(myconf->conffile); + myconf->conffile = NULL; + return -EINVAL; + } + + /* Default generation to 1 if not set */ + if (myconf->cluster_config->config_generation == 0) + myconf->cluster_config->config_generation = 1; + myconf->config_generation = myconf->cluster_config->config_generation; + + /* Check staleness */ + myconf->conf_stale = conf_check_stale(conf_dir, myconf->conf_stale_timeout); + if (myconf->conf_stale) + WARN("Primary config appears stale (age > %lds)", + (long)myconf->conf_stale_timeout); + + INFO("Primary config loaded: tag=%016" PRIx64 " gen=%" PRIu64, + myconf->config_version_tag, myconf->config_generation); + return 0; +} + /* ── internal helpers ──────────────────────────── */ static void make_conf_dir(const char *workdir, char *out, size_t sz) @@ -395,14 +539,181 @@ 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) +{ + uint64_t their_tag = req->body.replconfigrequestmsg->version_tag; + uint64_t my_tag = c->myconf->config_version_tag; + + xce_reply_t reply; + init_reply_by_req(&reply, req); + reply.data_which = RequestReply_data_replconfigreplymsg; + + repl_config_reply_t *rmsg = (repl_config_reply_t *)calloc( + 1, sizeof(repl_config_reply_t)); + if (!rmsg) { + reply.status = -ENOMEM; + reply.data_which = RequestReply_data_nodata; + xce_reply(ep, c->myhost, &reply); + return -ENOMEM; + } + reply.data.replconfigreplymsg = rmsg; + + if (their_tag == my_tag) { + reply.status = 0; + rmsg->up_to_date = 1; + rmsg->version_tag = 0; + rmsg->sha256 = NULL; + rmsg->bytes = NULL; + xce_reply(ep, c->myhost, &reply); + free(rmsg); + reply.data.replconfigreplymsg = NULL; + return 0; + } + + /* Read raw config bytes */ + uint8_t *bytes = NULL; + size_t blen = 0; + int rc = conf_read_raw(c->myconf->conffile, &bytes, &blen); + if (rc != 0) { + ERROR("handle_config_request: conf_read_raw failed: %d", rc); + reply.status = rc; + reply.data_which = RequestReply_data_nodata; + free(rmsg); + reply.data.replconfigreplymsg = NULL; + xce_reply(ep, c->myhost, &reply); + return rc; + } + + /* Verify SHA-256 matches what we cached at startup */ + uint8_t recomputed[32]; + SHA256(bytes, blen, recomputed); + if (memcmp(recomputed, c->myconf->config_sha256, 32) != 0) { + ERROR("handle_config_request: config file changed since startup" + " — refusing to send (hash mismatch)"); + free(bytes); + free(rmsg); + reply.data.replconfigreplymsg = NULL; + reply.status = -EIO; + reply.data_which = RequestReply_data_nodata; + xce_reply(ep, c->myhost, &reply); + return -EIO; + } + + reply.status = 0; + rmsg->up_to_date = 0; + rmsg->version_tag = my_tag; + /* sha256 and bytes are xce_data_t* — raw-byte encoding is wired in Task 11. + * For now set to NULL so encode_Data_ptr emits capn_null pointers. + * The standby will fall through to its saved config on this path until + * Task 11 completes the typemap shims. */ + rmsg->sha256 = NULL; + rmsg->bytes = NULL; + + xce_reply(ep, c->myhost, &reply); + free(bytes); + free(rmsg); + reply.data.replconfigreplymsg = NULL; + return 0; +} + +/* ── fetch_config_from_primary helpers ─────────────────────── */ + +/* + * do_one_config_fetch: attempt one network fetch of config from the primary. + * + * The full wire protocol (capnp encode/decode over a standalone SSL socket) + * depends on the typemap shims being complete (Task 11). Until then this + * function returns -ENOSYS so that fetch_config_from_primary() falls back to + * the locally saved config. + */ +static int do_one_config_fetch(cluster_t *c, const char *host, uint32_t port, + uint64_t our_tag) { (void)c; - (void)ep; - (void)req; - /* Task 9: full implementation — read versioned blob, send reply */ + (void)host; + (void)port; + (void)our_tag; + /* Task 11: wire full capnp encode/decode over standalone SSL socket */ return -ENOSYS; } +/* ── fetch_config_from_primary ──────────────────── */ + +int fetch_config_from_primary(cluster_t *c) +{ + const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg; + + /* Build conf dir path: / */ + char conf_dir[4096]; + snprintf(conf_dir, sizeof(conf_dir), "%s/%s", + c->myconf->workdir, c->myconf->name); + + uint64_t saved_tag = conf_get_saved_tag(conf_dir); + + if (pcfg->candidates_count == 0) { + /* Single-node mode — no primary to contact; load from saved file */ + goto use_saved; + } + + { + /* Primary is candidates[0] (highest priority = lowest index) */ + const char *primary_host = pcfg->candidates[0].hostname; + uint32_t primary_port = pcfg->candidates[0].port; + + INFO("HA: requesting config from primary %s:%u (my_tag=%016" PRIx64 ")", + primary_host, primary_port, saved_tag); + + uint32_t delay_ms = 500; + const uint32_t max_delay_ms = 5000; + const int max_attempts = 20; + + for (int attempt = 0; attempt < max_attempts; attempt++) { + /* If promoted during retry, switch to primary path */ + if (ha_is_primary(&c->ha)) { + INFO("HA: promoted to primary during config fetch" + " — switching to primary path"); + return load_primary_config(c); + } + + int rc = do_one_config_fetch(c, primary_host, primary_port, + saved_tag); + if (rc == 0) { + INFO("HA: config received from primary" + " (tag=%016" PRIx64 ")", + c->myconf->config_version_tag); + return 0; + } + + if (attempt < max_attempts - 1) { + WARN("HA: config fetch attempt %d/%d failed (rc=%d)" + " — retrying in %ums", + attempt + 1, max_attempts, rc, delay_ms); + usleep((useconds_t)delay_ms * 1000); + delay_ms = delay_ms * 2 < max_delay_ms + ? delay_ms * 2 : max_delay_ms; + } + } + } + +use_saved: + { + char fname[256] = {0}; + int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname)); + if (rc != 0) { + ERROR("Standby (priority=%d) could not fetch config from primary" + " and has no saved config at '%s/conf/current'" + " — start the primary first or provision config manually", + c->ha.my_priority, conf_dir); + return -ENOENT; + } + WARN("HA: using saved config (tag=%016" PRIx64 ")" + " — config may be stale", saved_tag); + /* load_primary_config resolves conf/current → versioned file, + * decodes Cap'n Proto, computes SHA-256, and checks staleness. + * It is reused here since the saved-config load path is identical. */ + return load_primary_config(c); + } +} + /* ── handle_upgrade_config ──────────────────────── */ int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req) diff --git a/src/controller/join.c b/src/controller/join.c index 4eade071..c5cf8043 100644 --- a/src/controller/join.c +++ b/src/controller/join.c @@ -126,17 +126,28 @@ static int handle_controller_join(cluster_t* c, endpoint_t* ep, xce_request_t* request) { xce_reply_t reply; - int rc = -ENOSYS; - - ERROR("we still have not support join of controller"); init_reply_by_req(&reply, request); - reply.status = rc; + reply.status = 0; reply.data_which = RequestReply_data_nodata; - xce_reply(ep, c->myhost, &reply); - return rc; + /* Register this endpoint for OP_upgradeConfig fan-out to standbys */ + if (c->ha.standby_endpoints) { + ep_private_t *priv = (ep_private_t *)ep->data; + if (priv) + priv->role = HostRole_controller; + + pthread_mutex_lock(&c->ha.lock); + zarray_store(c->ha.standby_endpoints, ep); + pthread_mutex_unlock(&c->ha.lock); + + INFO("controller %s joined — %d standby(s) tracked", + request->from ? request->from->name : "?", + c->ha.standby_endpoints->count); + } + + return 0; } static int handle_apiserver_join(cluster_t* c, endpoint_t* ep, diff --git a/src/controller/myconf.c b/src/controller/myconf.c index bd9905bd..f9eed048 100644 --- a/src/controller/myconf.c +++ b/src/controller/myconf.c @@ -152,119 +152,6 @@ cluster_config_t* get_default_conf(void* cl) return cfg; } -/* ── load_primary_config ────────────────────────── */ - -int load_primary_config(cluster_t *c) -{ - myconf_t *myconf = c->myconf; - - /* Build conf dir path: / */ - char conf_dir[4096]; - snprintf(conf_dir, sizeof(conf_dir), "%s/%s", myconf->workdir, myconf->name); - - /* Resolve conf/current → versioned filename */ - char fname[256] = {0}; - int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname)); - if (rc != 0) { - ERROR("Primary has no config at '%s/conf/current' (rc=%d) — " - "provision the config before starting", - conf_dir, rc); - return -ENOENT; - } - - /* Heap-allocate conffile so it survives past this stack frame */ - free(myconf->conffile); - myconf->conffile = NULL; - if (asprintf(&myconf->conffile, "%s/conf/%s", conf_dir, fname) < 0) - return -ENOMEM; - - /* Read raw bytes and check size */ - uint8_t *bytes = NULL; - size_t blen = 0; - rc = conf_read_raw(myconf->conffile, &bytes, &blen); - if (rc != 0) { - ERROR("Primary config '%s' unreadable: %d", myconf->conffile, rc); - free(myconf->conffile); - myconf->conffile = NULL; - return rc; - } - - /* Compute and cache SHA-256 */ - SHA256(bytes, blen, myconf->config_sha256); - uint64_t tag; - memcpy(&tag, myconf->config_sha256, 8); - myconf->config_version_tag = be64toh(tag); - - /* Free any prior decoded config before overwriting the pointer */ - if (myconf->cluster_config) { - free_ClusterConfig(myconf->cluster_config); - free(myconf->cluster_config); - myconf->cluster_config = NULL; - } - - /* Decode Cap'n Proto directly into myconf->cluster_config */ - rc = load_cluster_config_from_file(myconf->conffile, &myconf->cluster_config); - if (rc != 0) { - ERROR("Primary config '%s' failed Cap'n Proto decode: %d", - myconf->conffile, rc); - free(bytes); - free(myconf->conffile); - myconf->conffile = NULL; - return rc; - } - - if (myconf->cluster_config == NULL) { - ERROR("Primary config '%s' decoded to NULL", myconf->conffile); - free(bytes); - free(myconf->conffile); - myconf->conffile = NULL; - return -EINVAL; - } - - free(bytes); - - fill_default_cluster_config(myconf->cluster_config); - - /* Check for reserved lost_and_found name */ - if (config_contains_reserved_name(myconf->cluster_config)) { - ERROR("Config '%s' contains reserved name 'lost_and_found'", - myconf->conffile); - free_ClusterConfig(myconf->cluster_config); - free(myconf->cluster_config); - myconf->cluster_config = NULL; - free(myconf->conffile); - myconf->conffile = NULL; - return -EINVAL; - } - - /* Default generation to 1 if not set */ - if (myconf->cluster_config->config_generation == 0) - myconf->cluster_config->config_generation = 1; - myconf->config_generation = myconf->cluster_config->config_generation; - - /* Check staleness */ - myconf->conf_stale = conf_check_stale(conf_dir, myconf->conf_stale_timeout); - if (myconf->conf_stale) - WARN("Primary config appears stale (age > %lds)", - (long)myconf->conf_stale_timeout); - - INFO("Primary config loaded: tag=%016" PRIx64 " gen=%" PRIu64, - myconf->config_version_tag, myconf->config_generation); - 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)