push more to api server

This commit is contained in:
Rongsong Shen 2026-01-22 17:23:18 +08:00
parent 00d161aa11
commit 44fc5b23e9
4 changed files with 501 additions and 34 deletions

View file

@ -3,6 +3,7 @@
#include "cluster.h"
#include "controller.capnp.h"
#include "objmgr.h"
#include "request.h"
#include "xce/types.h"
@ -50,7 +51,7 @@ static int xce_apiserver_push_workers(cluster_t* c, endpoint_t* ep)
deref_objs(workers.n_total, (void**)workers.workers);
FREEUP(workers.workers);
return 0;
return rc;
}
static int check_job_(xce_job_t* j)
@ -81,30 +82,430 @@ static int xce_apiserver_push_jobs(cluster_t* c, endpoint_t* ep)
deref_objs(jobs.n_total, (void**)jobs.jobs);
FREEUP(jobs.jobs);
return rc;
}
static int apiserver_push_joblist(cluster_t* c, xlist_t* l)
{
int i;
for (i = 0; i < N_APISERVER_MAX; i++) {
if ((c->apiservers[i].name != NULL) && c->apiservers[i].ep != NULL) {
int rc;
xce_data_t data;
xce_jobs_data_t jobs = { .n_total = l->count,
.pos = 0,
.count = l->count,
.jobs = (xce_job_t**)l->data };
memset(&data, 0x0, sizeof(data));
data.table = Data_data_schedjobs;
data.data.jobs = &jobs;
rc = xce_push_data_req(c->apiservers[i].ep, c->myhost, &data,
check_controller, c, 100);
if (rc < 0) {
WARN("%s push job list to api server %s has problem, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
}
return 0;
}
static int apiserver_job_new_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
xlist_t* l = va_arg(ap, xlist_t*);
int rc = 0;
switch (situation) {
case HOOK_AFTER_EVENT:
rc = -EINVAL;
if (l != NULL) {
rc = apiserver_push_joblist(c, l);
}
break;
default:
break;
}
return rc;
}
static int apiserver_job_start_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
job_pidinfo_t* info = va_arg(ap, job_pidinfo_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_schedStartJob,
.data = { .sched_start_job = info->jobid } };
/* FIXME: add job pid to API server? */
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN("%s notify job start to api server %s fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
}
break;
default:
break;
}
return 0;
}
static int apiserver_job_exit_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
xce_jobfinish_t* info = va_arg(ap, xce_jobfinish_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_schedExitJob,
.data = { .sched_exit_job = info->jobid } };
/* FIXME: we need to send exit code of job to API server */
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN("%s notify job exit to api server %s fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
}
break;
default:
break;
}
return 0;
}
static int apiserver_job_fail_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
job_failinfo_t* info = va_arg(ap, job_failinfo_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_schedFailJob,
.data = { .sched_fail_job = info->jobid } };
/* FIXME: we need to add fail reason of job to api server */
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN("%s notify job fail to api server %s fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
}
break;
default:
break;
}
return 0;
}
static int apiserver_job_update_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
job_t* job = va_arg(ap, job_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_schedUpdateJob,
.data = { .sched_update_job = &(job->job) } };
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN("%s notify job update to api server %s fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
}
break;
default:
break;
}
return 0;
}
static int apiserver_job_rusage_update_fun(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
xce_job_rusages_t* rusages = va_arg(ap, xce_job_rusages_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_schedJobRusages,
.data = { .sched_job_rusages = rusages } };
rc = notify(c, c->apiservers[i].ep, &xdata);
if (rc < 0) {
WARN("%s notify rusage update to api server %s fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
}
break;
default:
break;
}
return 0;
}
static int apiserver_job_hook_fun(int type, int situation, int detail,
void* data, va_list ap)
{
switch (detail) {
case HOOK_DETAIL_JOB_NEW:
return apiserver_job_new_fun(type, situation, data, ap);
case HOOK_DETAIL_JOB_START:
return apiserver_job_start_fun(type, situation, data, ap);
case HOOK_DETAIL_JOB_EXIT:
return apiserver_job_exit_fun(type, situation, data, ap);
case HOOK_DETAIL_JOB_UPDATE:
return apiserver_job_update_fun(type, situation, data, ap);
case HOOK_DETAIL_JOB_FAIL:
return apiserver_job_fail_fun(type, situation, data, ap);
case HOOK_DETAIL_JOB_RUSAGE_UPDATE:
return apiserver_job_rusage_update_fun(type, situation, data, ap);
}
return 0;
}
static int apiserver_push_workerlist(cluster_t* c, endpoint_t* ep, xlist_t* l)
{
xce_data_t data;
xce_workers_data_t workers = { .n_total = l->count,
.pos = 0,
.count = l->count,
.workers = (xce_worker_t**)l->data };
int rc = 0;
memset(&data, 0x0, sizeof(data));
data.table = Data_data_schedworkers;
data.data.workers = &workers;
rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100);
return rc;
}
static int apiserver_worker_change(int type, int situation, void* data,
cluster_t* c, worker_t* w)
{
int i;
TRACE("%s worker %s idx %d situation %d", __func__, w->host.name, w->idx,
situation);
collect_resval_with_idx(&(w->host.resvals), &(w->host.nresvals_to_sched),
&(w->host.resvals_to_sched));
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
xlist_t wlist;
int rc;
xlist_init(&wlist, 2);
xlist_store(&wlist, w);
rc = apiserver_push_workerlist(c, c->apiservers[i].ep, &wlist);
if (rc < 0) {
WARN("%s notify worker change failed to api server %s, rc "
"%d",
__func__, c->apiservers[i].name, rc);
}
xlist_fini(&wlist);
}
}
break;
default:
break;
}
free_resval_with_idx_list(&(w->host.resvals_to_sched),
w->host.nresvals_to_sched);
return 0;
}
static int apiserver_worker_join(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
char* name = va_arg(ap, char*);
worker_t* w = NULL;
int rc;
if (situation == HOOK_BEFORE_EVENT) {
/* nothing to do */
return 0;
}
w = workermgr_lock(&(c->workermgr), name, OBJREF_READ);
rc = apiserver_worker_change(type, situation, data, c, w);
/* FIXME: need to mark the jobs in this worker */
obj_deref(w);
return rc;
}
static int apiserver_worker_dead(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
worker_t* w = va_arg(ap, worker_t*);
/* FIXME: need to handle job status in this worker */
if (situation == HOOK_BEFORE_EVENT) {
return 0;
}
return apiserver_worker_change(type, situation, data, c, w);
}
static int apiserver_worker_update(int type, int situation, void* data,
va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
worker_t* w = va_arg(ap, worker_t*);
if (situation == HOOK_BEFORE_EVENT) {
return 0;
}
return apiserver_worker_change(type, situation, data, c, w);
}
static int apiserver_worker_hook_fun(int type, int situation, int detail,
void* data, va_list ap)
{
switch (detail) {
case HOOK_DETAIL_WORKER_JOIN:
return apiserver_worker_join(type, situation, data, ap);
case HOOK_DETAIL_WORKER_DEAD:
return apiserver_worker_dead(type, situation, data, ap);
case HOOK_DETAIL_WORKER_UPDATE:
return apiserver_worker_update(type, situation, data, ap);
}
return 0;
}
static int apiserver_resource_hook_fun(int type, int situation, int detail,
void* data, va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
xce_resval_list_t* resvals = va_arg(ap, xce_resval_list_t*);
int i;
if (detail != HOOK_DETAIL_RESOURCE_UPDATE) {
return 0;
}
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_schedUpdateResvals,
.data = { .sched_update_resvals = resvals } };
rc = notify(c, c->sched.ep, &xdata);
if (rc < 0) {
WARN("%s notify api server %s resource update fail, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
}
break;
default:
break;
}
return 0;
}
static int apiserver_queue_hook_fun(int type, int situation, int detail,
void* data, va_list ap)
{
cluster_t* c = va_arg(ap, cluster_t*);
queue_t* q = va_arg(ap, queue_t*);
int i;
switch (situation) {
case HOOK_AFTER_EVENT:
for (i = 0; i < N_APISERVER_MAX; i++) {
if ((c->apiservers[i].name != NULL)
&& (c->apiservers[i].ep != NULL)) {
int rc;
xce_notify_data_t xdata
= { .event = NotifyData_data_schedUpdateQueue,
.data = { .sched_update_queue = &(q->queue) } };
rc = notify(c, c->sched.ep, &xdata);
if (rc < 0) {
WARN(
"%s notify queue change to api server %s failed, rc %d",
__func__, c->apiservers[i].name, rc);
}
}
}
break;
default:
break;
}
return 0;
}
@ -134,6 +535,53 @@ int on_apiserver_join(cluster_t* c, endpoint_t* ep)
int handle_apiserver_leave(cluster_t* c, endpoint_t* ep)
{
/* FIXME */
ep_private_t* priv;
int i;
if (ep == NULL) {
return -EINVAL;
}
priv = (ep_private_t*)ep->data;
if (priv == NULL) {
return -EINVAL;
}
i = priv->u.idx;
if (c->apiservers[i].name != NULL) {
obj_deref(c->apiservers[i].name);
c->apiservers[i].ep = NULL;
}
/* FIXME: kill ep? */
return 0;
}
int add_apiserver(cluster_t* c, char* name, endpoint_t* ep)
{
int i;
for (i = 0; i < N_APISERVER_MAX; i++) {
if ((c->apiservers[i].name != NULL)
&& (strcmp(c->apiservers[i].name, name) == 0)) {
if (c->apiservers[i].ep != NULL) {
/* FIXME: we need to delete 1 */
}
c->apiservers[i].ep = ep;
return i;
}
}
for (i = 0; i < N_APISERVER_MAX; i++) {
if (c->apiservers[i].name == NULL) {
c->apiservers[i].name
= create_obj_by_typename(&(c->objmgr), "xstring", name);
c->apiservers[i].ep = ep;
return i;
}
}
/* better error code */
return -1;
}

View file

@ -137,6 +137,6 @@ extern int collect_resval_with_idx(zarray_t*, int*, xce_val_with_idx_t***);
extern int do_set_preason(cluster_t*, endpoint_t*, xce_request_t*);
extern int do_clr_preason(cluster_t*, endpoint_t*, xce_request_t*);
extern int handle_apiserver_leave(cluster_t*, endpoint_t*);
extern int add_apiserver(cluster_t*, char*, endpoint_t*);
#endif

View file

@ -117,21 +117,33 @@ static int handle_controller_join(cluster_t* c, endpoint_t* ep,
return rc;
}
static int handle_apiserver_join(cluster_t *c, endpoint_t *ep, xce_request_t *request) {
xce_reply_t reply;
ep_private_t *priv = (ep_private_t *)ep->data;
int rc = 0;
static int handle_apiserver_join(cluster_t* c, endpoint_t* ep,
xce_request_t* request)
{
xce_reply_t reply;
xce_peerinfo_t* from = request->from;
int rc = 0;
TRACE("%s api server join, ep %p", __func__, ep);
init_reply_by_req(&reply, request);
rc = on_apiserver_join(c, ep);
rc = add_apiserver(c, from->name, ep);
if (rc >= 0) {
ep_private_t* priv = (ep_private_t*)ep->data;
reply.status = rc;
priv->u.idx = rc;
on_apiserver_join(c, ep);
}
reply.status = rc;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
if (rc < 0) {
ERROR("api server %s join failed, rc %d", from->name, rc);
return rc;
}
return 0;
}

View file

@ -20,6 +20,7 @@
#include "zarray.h"
#define N_CONTROLLER_MAX (16)
#define N_APISERVER_MAX (16)
enum {
XCE_CONTROLLER_HOOK_JOB = 0,
@ -82,31 +83,37 @@ typedef struct {
} scheduler_t;
typedef struct {
myconf_t* myconf;
xce_peerinfo_t* myhost;
hooksys_t hooksys;
iomgr_t iomgr;
objmgr_t objmgr;
zarray_t endpoints;
endpoint_t ep;
zarray_t configs;
zarray_t sharedinst;
controller_t controllers[N_CONTROLLER_MAX];
int myidx;
int master;
int ncontrollers;
delaymgr_t delaymgr;
scheduler_t sched;
jobmgr_t jobmgr;
resmgr_t resmgr;
workermgr_t workermgr;
qmgr_t qmgr;
kvmgr_t configmgr;
kvmgr_t ugmgr;
kvmgr_t hgmgr;
vtaskmgr_t vtaskmgr;
cbuf_t wbuf;
int running;
char* name;
endpoint_t* ep;
} cluster_apiserver_t;
typedef struct {
myconf_t* myconf;
xce_peerinfo_t* myhost;
hooksys_t hooksys;
iomgr_t iomgr;
objmgr_t objmgr;
zarray_t endpoints;
endpoint_t ep;
zarray_t configs;
zarray_t sharedinst;
controller_t controllers[N_CONTROLLER_MAX];
cluster_apiserver_t apiservers[N_APISERVER_MAX];
int myidx;
int master;
int ncontrollers;
delaymgr_t delaymgr;
scheduler_t sched;
jobmgr_t jobmgr;
resmgr_t resmgr;
workermgr_t workermgr;
qmgr_t qmgr;
kvmgr_t configmgr;
kvmgr_t ugmgr;
kvmgr_t hgmgr;
vtaskmgr_t vtaskmgr;
cbuf_t wbuf;
int running;
} cluster_t;
#endif /* XCE_CLUSTER_DEF_H */