forked from lei.chen/myxce
Rongsong Shen
6b06aa02d9
- 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
281 lines
7.9 KiB
C
281 lines
7.9 KiB
C
#define _GNU_SOURCE
|
|
#include <errno.h>
|
|
#include <inttypes.h>
|
|
#include <openssl/sha.h>
|
|
#include <stdint.h>
|
|
#include <stdio.h>
|
|
#include <stdlib.h>
|
|
#include <string.h>
|
|
#include <tlog.h>
|
|
|
|
#include "byteorder.h"
|
|
#include "config_repl.h"
|
|
#include "configobj.h"
|
|
#include "controller.capnp.h"
|
|
#include "myconf.h"
|
|
#include "utils.h"
|
|
#include "xce/cluster_def.h"
|
|
#include "xce/types.h"
|
|
|
|
int add_config(void* c, char* name, char* value)
|
|
{
|
|
cluster_t* cls = (cluster_t*)c;
|
|
xce_config_t cvar = { .name = name, .value = value };
|
|
xce_config_t* cv = NULL;
|
|
|
|
cv = makeConfig(cls->configmgr, &cvar);
|
|
if (cv == NULL) {
|
|
return -errno;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
struct cluster_configuration {
|
|
cluster_config_t* cluster;
|
|
};
|
|
|
|
static int config_fini(int type, int situation, void* data, va_list ap)
|
|
{
|
|
struct cluster_configuration* cf = (struct cluster_configuration*)data;
|
|
|
|
switch (situation) {
|
|
case HOOK_AFTER_EVENT:
|
|
if (cf->cluster != NULL) {
|
|
free_ClusterConfig(cf->cluster);
|
|
FREEUP(cf->cluster);
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
static int cluster_hook_fun(int type, int situation, int detail, void* data,
|
|
va_list ap)
|
|
{
|
|
if (detail == HOOK_DETAIL_CLUSTER_SHUTDOWN) {
|
|
return config_fini(type, situation, data, ap);
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
#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;
|
|
perror("fopen");
|
|
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;
|
|
}
|
|
|
|
int load_cluster_config(cluster_t* c,
|
|
struct cluster_configuration** cluster_config)
|
|
{
|
|
cluster_config_t* cfg;
|
|
int rc;
|
|
|
|
*cluster_config
|
|
= (struct cluster_configuration*)calloc(1, sizeof(*cluster_config));
|
|
if ((*cluster_config) == NULL) {
|
|
return -ENOMEM;
|
|
}
|
|
|
|
rc = load_cluster_config_from_file(c->myconf->conffile,
|
|
&((*cluster_config)->cluster));
|
|
|
|
if (rc < 0) {
|
|
ERROR("failed to load configuration file `%s'", c->myconf->conffile);
|
|
return rc;
|
|
}
|
|
|
|
if ((*cluster_config)->cluster == NULL) {
|
|
return -errno;
|
|
}
|
|
|
|
cfg = (*cluster_config)->cluster;
|
|
|
|
fill_default_cluster_config(cfg);
|
|
|
|
xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER,
|
|
(hook_fn_t)cluster_hook_fun, (*cluster_config));
|
|
|
|
return 0;
|
|
}
|
|
|
|
cluster_config_t* get_default_conf(void* cl)
|
|
{
|
|
cluster_t* cls = (cluster_t*)cl;
|
|
cluster_config_t* cfg = NULL;
|
|
|
|
int rc = load_cluster_config_from_file(cls->myconf->conffile, &cfg);
|
|
if (rc < 0 || cfg == NULL) {
|
|
PERROR("load_cluster_config_from_file");
|
|
return NULL;
|
|
}
|
|
|
|
fill_default_cluster_config(cfg);
|
|
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)
|
|
{
|
|
if (!myconf)
|
|
return;
|
|
free(myconf->conffile);
|
|
myconf->conffile = NULL;
|
|
if (myconf->cluster_config) {
|
|
free_ClusterConfig(myconf->cluster_config);
|
|
free(myconf->cluster_config);
|
|
myconf->cluster_config = NULL;
|
|
}
|
|
}
|