forked from lei.chen/myxce
feat: wire HA candidate-list reconnect into worker+scheduler join paths
Worker (xcecd): - Add xce_cluster_config_t cluster_cfg field to controller_connector_ctx_t (controller_connector.h includes xce/cluster.h) - Populate cc_ctx->cluster_cfg from parsed cluster_cfg after xcecd_start() sets cc_ctx (main.c, removes the P3 placeholder) - Replace xcecd_join_cluster() single-shot connect with a candidate-iterating loop: tries each candidate, calls xcecli_stop() before the next attempt, falls back to the original single-host path when candidates_count == 0 Scheduler (xce_scheduler): - Add xce_cluster_config_t cluster_cfg field to xce_scheduler_t (scheduler.h includes xce/cluster.h) - Add sched_set_cluster_cfg() setter - Replace the hardcoded xcecli_start/xcecli_join pair in sched_init() with the same candidate-iterating pattern; original single-host path preserved - main.c calls sched_set_cluster_cfg() before sched_init() when --peers given (removes the P3 placeholder comment) checking_connection() already calls xcecd_join_cluster() on retry, giving the worker free reconnect-after-failover behaviour for no additional code.
This commit is contained in:
parent
fd9cfce1c1
commit
b75703ba6b
6 changed files with 131 additions and 20 deletions
|
|
@ -15,6 +15,7 @@
|
|||
#include "kvmgr.h"
|
||||
#include "mylock.h"
|
||||
#include "plugin.h"
|
||||
#include "xce/cluster.h"
|
||||
#include "xce/types.h"
|
||||
#include "xlist.h"
|
||||
#include "xvec.h"
|
||||
|
|
@ -88,10 +89,12 @@ typedef struct {
|
|||
_Atomic int eval_in_progress; /* 1 while eval_select loop runs */
|
||||
_Atomic int staging_merge_complete; /* 1 after staging merge completes */
|
||||
select_eval_ctx_t* sel_eval_ctx; /* session-level memoisation context; NULL between sessions */
|
||||
xce_cluster_config_t cluster_cfg; /* HA candidate list — empty means single-host mode */
|
||||
} xce_scheduler_t;
|
||||
|
||||
extern int sched_init(xce_scheduler_t*, xce_client_config_t*,
|
||||
xce_request_handler_t);
|
||||
extern void sched_set_cluster_cfg(xce_scheduler_t*, const xce_cluster_config_t*);
|
||||
extern void sched_fini(xce_scheduler_t*);
|
||||
extern int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t*);
|
||||
extern int sched_update(xce_scheduler_t*);
|
||||
|
|
|
|||
|
|
@ -188,7 +188,7 @@ int main(int argc, char* argv[])
|
|||
INFO("HA cluster config: %d candidate(s), identity=%s, ca_cert=%s",
|
||||
cluster_cfg.candidates_count, cluster_cfg.identity,
|
||||
cluster_cfg.ca_cert_path);
|
||||
/* P3 will call xce_connect_cluster(&cluster_cfg) here */
|
||||
sched_set_cluster_cfg(&sched, &cluster_cfg);
|
||||
} else {
|
||||
INFO("No --peers flag; using single-controller path");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -426,14 +426,52 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
|
|||
|
||||
xcecli_set_handler(sched->apictx, handler, sched);
|
||||
|
||||
rc = xcecli_start(sched->apictx);
|
||||
if (rc < 0) {
|
||||
return rc;
|
||||
}
|
||||
/* HA mode: iterate candidates until one accepts as primary */
|
||||
if (sched->cluster_cfg.candidates_count > 0) {
|
||||
int joined = 0;
|
||||
for (int i = 0; i < sched->cluster_cfg.candidates_count; i++) {
|
||||
const char* host = sched->cluster_cfg.candidates[i].hostname;
|
||||
int port = sched->cluster_cfg.candidates[i].port;
|
||||
|
||||
rc = xcecli_join(sched->apictx, HostRole_scheduler, NULL);
|
||||
if (rc < 0) {
|
||||
return rc;
|
||||
config->nmasters = 1;
|
||||
config->masters[0] = (char*)host;
|
||||
config->master_ports[0] = port;
|
||||
|
||||
rc = xcecli_ctx_init(&(sched->apictx), config);
|
||||
if (rc < 0) {
|
||||
continue;
|
||||
}
|
||||
xcecli_set_handler(sched->apictx, handler, sched);
|
||||
|
||||
rc = xcecli_start(sched->apictx);
|
||||
if (rc < 0) {
|
||||
continue;
|
||||
}
|
||||
|
||||
rc = xcecli_join(sched->apictx, HostRole_scheduler, NULL);
|
||||
if (rc == 0) {
|
||||
joined = 1;
|
||||
break;
|
||||
}
|
||||
|
||||
/* Standby replied — try next candidate */
|
||||
xcecli_stop(sched->apictx);
|
||||
}
|
||||
|
||||
if (!joined) {
|
||||
return -1;
|
||||
}
|
||||
} else {
|
||||
/* Single-host path (original) */
|
||||
rc = xcecli_start(sched->apictx);
|
||||
if (rc < 0) {
|
||||
return rc;
|
||||
}
|
||||
|
||||
rc = xcecli_join(sched->apictx, HostRole_scheduler, NULL);
|
||||
if (rc < 0) {
|
||||
return rc;
|
||||
}
|
||||
}
|
||||
|
||||
rc = pthread_create(&(sched->jobpicker), NULL, &sched_job_picker, sched);
|
||||
|
|
@ -446,6 +484,15 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
|
|||
return rc;
|
||||
}
|
||||
|
||||
void sched_set_cluster_cfg(xce_scheduler_t* sched,
|
||||
const xce_cluster_config_t* cfg)
|
||||
{
|
||||
if (sched == NULL || cfg == NULL) {
|
||||
return;
|
||||
}
|
||||
sched->cluster_cfg = *cfg;
|
||||
}
|
||||
|
||||
void sched_fini(xce_scheduler_t* sched)
|
||||
{
|
||||
void* res;
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@
|
|||
#include "client.h"
|
||||
#include "controller.capnp.h"
|
||||
#include "thread_pool.h"
|
||||
#include "xce/cluster.h"
|
||||
#include "xcecd_utils.h"
|
||||
#include "xcectx.h"
|
||||
|
||||
|
|
@ -13,14 +14,15 @@ typedef struct {
|
|||
} api_ctx_t;
|
||||
|
||||
typedef struct controller_connector_ctx_s {
|
||||
void* thread_pool;
|
||||
void* process_manager;
|
||||
char* remote_host;
|
||||
char* virtual_name;
|
||||
void** worker_list;
|
||||
int remote_port;
|
||||
int join_interval;
|
||||
bool running;
|
||||
void* thread_pool;
|
||||
void* process_manager;
|
||||
char* remote_host;
|
||||
char* virtual_name;
|
||||
void** worker_list;
|
||||
int remote_port;
|
||||
int join_interval;
|
||||
bool running;
|
||||
xce_cluster_config_t cluster_cfg; /* HA candidate list — empty means single-host mode */
|
||||
} controller_connector_ctx_t;
|
||||
extern controller_connector_ctx_t* cc_ctx;
|
||||
|
||||
|
|
|
|||
|
|
@ -227,6 +227,10 @@ int checking_connection()
|
|||
/**
|
||||
* @brief Joins the cluster using the provided host attributes.
|
||||
*
|
||||
* In HA mode (cc_ctx->cluster_cfg.candidates_count > 0), iterates through
|
||||
* all candidates until one accepts the join as primary. In single-host mode
|
||||
* the original path is taken unchanged.
|
||||
*
|
||||
* @param attrs Pointer to the host attributes structure.
|
||||
* @return int 0 on success, -1 on failure.
|
||||
*/
|
||||
|
|
@ -236,15 +240,63 @@ int xcecd_join_cluster(hostattrs_t* attrs)
|
|||
return -1;
|
||||
}
|
||||
|
||||
int rc;
|
||||
int role = HostRole_worker;
|
||||
|
||||
/* HA mode: try each candidate until we find the primary */
|
||||
if (cc_ctx != NULL && cc_ctx->cluster_cfg.candidates_count > 0) {
|
||||
for (int i = 0; i < cc_ctx->cluster_cfg.candidates_count; i++) {
|
||||
const char* host = cc_ctx->cluster_cfg.candidates[i].hostname;
|
||||
int port = cc_ctx->cluster_cfg.candidates[i].port;
|
||||
|
||||
/* Point cc_ctx at this candidate so init_client_config() picks it up */
|
||||
cc_ctx->remote_host = (char*)host;
|
||||
cc_ctx->remote_port = port;
|
||||
|
||||
XCECD_LOG(XCE_LOG_INFO,
|
||||
"xcecd_join_cluster: trying candidate %d: %s:%d",
|
||||
i, host, port);
|
||||
|
||||
init_client_config(&xce_client_cfg);
|
||||
xcecli_ctx_init(&xcecli_ctx, &xce_client_cfg);
|
||||
xcecli_set_handler(xcecli_ctx, &myreq_handler, NULL);
|
||||
|
||||
int rc = xcecli_start(xcecli_ctx);
|
||||
if (rc < 0) {
|
||||
XCECD_LOG(XCE_LOG_WARN,
|
||||
"xcecd_join_cluster: connect to %s:%d failed (rc=%d)",
|
||||
host, port, rc);
|
||||
continue;
|
||||
}
|
||||
|
||||
rc = xcecli_join(xcecli_ctx, role, attrs);
|
||||
if (rc == 0) {
|
||||
XCECD_LOG(XCE_LOG_INFO,
|
||||
"xcecd_join_cluster: joined primary at %s:%d",
|
||||
host, port);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Standby replied — try next candidate */
|
||||
XCECD_LOG(XCE_LOG_INFO,
|
||||
"xcecd_join_cluster: candidate %s:%d returned %d "
|
||||
"(likely standby redirect), trying next",
|
||||
host, port, rc);
|
||||
xcecli_stop(xcecli_ctx);
|
||||
}
|
||||
|
||||
XCECD_LOG(XCE_LOG_ERROR,
|
||||
"xcecd_join_cluster: all %d candidates failed",
|
||||
cc_ctx->cluster_cfg.candidates_count);
|
||||
return -1;
|
||||
}
|
||||
|
||||
/* Single-host fallback (original path) */
|
||||
init_client_config(&xce_client_cfg);
|
||||
xcecli_ctx_init(&xcecli_ctx, &xce_client_cfg);
|
||||
xcecli_set_handler(xcecli_ctx, &myreq_handler, NULL);
|
||||
xcecli_start(xcecli_ctx);
|
||||
|
||||
rc = xcecli_join(xcecli_ctx, role, attrs);
|
||||
|
||||
int rc = xcecli_join(xcecli_ctx, role, attrs);
|
||||
if (rc < 0) {
|
||||
errno = -rc;
|
||||
XCECD_LOG(XCE_LOG_ERROR, "Call xcecli_join() failure: %s",
|
||||
|
|
|
|||
|
|
@ -534,7 +534,7 @@ int main(int argc, char* argv[])
|
|||
"(host=%s port=%d)",
|
||||
config.remote_host, config.remote_port);
|
||||
}
|
||||
(void)use_ha; /* P3 will consume this to call xce_connect_cluster() */
|
||||
/* use_ha and cluster_cfg are consumed after xcecd_start() sets cc_ctx */
|
||||
|
||||
struct stat st = { 0 };
|
||||
|
||||
|
|
@ -641,6 +641,13 @@ int main(int argc, char* argv[])
|
|||
return 1;
|
||||
}
|
||||
|
||||
if (use_ha && cc_ctx != NULL) {
|
||||
cc_ctx->cluster_cfg = cluster_cfg;
|
||||
XCECD_LOG(XCE_LOG_INFO,
|
||||
"HA mode active: %d candidates, primary will be auto-discovered",
|
||||
cluster_cfg.candidates_count);
|
||||
}
|
||||
|
||||
#ifndef VIRTUAL_CLUSTER
|
||||
if (!ipc_master_init(config.domain_sock_file)) {
|
||||
XCECD_LOG(XCE_LOG_ERROR, "Failed to initialize XCECD IPC.");
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue