feat: implement handle_config_request(), fetch_config_from_primary(), handle_controller_join()
- handle_config_request(): primary-side handler sends up-to-date flag or version_tag; SHA-256 integrity check guards against in-place file changes; sha256/bytes fields left NULL pending Task 11 typemap shims - fetch_config_from_primary(): standby fetch with exponential backoff retry (20 attempts, 500ms..5s); falls back to saved config via load_primary_config(); do_one_config_fetch() stubbed (-ENOSYS) until Task 11 wires capnp over SSL - handle_controller_join(): registers standby endpoint in ha.standby_endpoints via zarray_store() for OP_upgradeConfig fan-out; sends success reply - Moved load_primary_config() from myconf.c to config_repl.c so the unit test binary (which links config_repl.c but not myconf.c) can resolve it
This commit is contained in:
parent
93b8616f11
commit
8f4db308da
3 changed files with 331 additions and 122 deletions
|
|
@ -16,11 +16,155 @@
|
|||
#include <utime.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <inttypes.h>
|
||||
|
||||
#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: <workdir>/<name> */
|
||||
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: <workdir>/<name> */
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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: <workdir>/<name> */
|
||||
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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue