diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index f0fc3a5..facda86 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -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*); diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index 6e686e5..2e07e30 100644 --- a/src/scheduler/src/main.c +++ b/src/scheduler/src/main.c @@ -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"); } diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 4749860..812a611 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -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; diff --git a/src/worker/include/controller_connector.h b/src/worker/include/controller_connector.h index b60625c..3be9650 100644 --- a/src/worker/include/controller_connector.h +++ b/src/worker/include/controller_connector.h @@ -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; diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index 6ccc3f1..fed44dd 100755 --- a/src/worker/src/controller_connector.c +++ b/src/worker/src/controller_connector.c @@ -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", diff --git a/src/worker/src/main.c b/src/worker/src/main.c index d1d0fee..18161c1 100644 --- a/src/worker/src/main.c +++ b/src/worker/src/main.c @@ -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.");