diff --git a/src/apiserver/cmd/main.go b/src/apiserver/cmd/main.go new file mode 100644 index 0000000..8c3c90e --- /dev/null +++ b/src/apiserver/cmd/main.go @@ -0,0 +1,136 @@ +package main + +/* + #include + #include + #include + + #include "client.h" + #include "netio.h" + #include "controller.capnp.h" + + static int init_config(xce_client_config_t *config) { + config->nmasters = 1; + config->masters[0] = "localhost"; + config->master_ports[0] = 9990; + config->from = "xce_apiserver"; + config->key = "keepsecret"; + return 0; + } + + static xce_client_config_t config; + static struct { + xcecli_ctx_t ctx; + } xce_api_server_ctx; + + static uintptr_t go_handler = 0; + + extern int go_callback(uintptr_t, xce_request_t*, void *); + + static int xce_apiserver_handler(xce_request_t *req, void *arg) { + int rc = -EINVAL; + + if (go_handler != 0) { + rc = go_callback(go_handler, req, arg); + } + + return rc; + } + + static int xce_apiserver_init(uintptr_t handler) { + int rc; + + init_config(&config); + + go_handler = handler; + + rc = xcecli_ctx_init(&(xce_api_server_ctx.ctx), &config); + if (rc < 0) { + return rc; + } + + xcecli_set_handler(xce_api_server_ctx.ctx, (xce_request_handler_t)xce_apiserver_handler, NULL); + + rc = xcecli_start(xce_api_server_ctx.ctx); + if (rc < 0) { + return rc; + } + + rc = xcecli_join(xce_api_server_ctx.ctx, HostRole_apiserver, NULL); + + return rc; + } + + static void* xce_request_body(xce_request_t *req) { + switch (req->body_which) { + case Request_body_configpushmsg: + return req->body.configpushmsg; + case Request_body_datamsg: + return req->body.datamsg; + default: + return NULL; + } + } +*/ +import "C" + +import ( + "fmt" + "unsafe" + "runtime/cgo" +) + +//export go_callback +func go_callback(h C.uintptr_t, req *C.xce_request_t, args unsafe.Pointer) C.int { + fn := cgo.Handle(h).Value().(func(*C.xce_request_t,unsafe.Pointer) C.int) + return fn(req, args); +} + +func handle_config(config *C.xce_configdata_t) C.int { + switch (config.data_which) { + case C.ConfigData_data_clusterconfig: + fmt.Printf("cluster configuration received\n") + default: + // + fmt.Printf("unhandled config request %d\n", config.data_which) + } + return 0 +} + +func handle_dataset(dataset *C.dataset_t) C.int { + fmt.Printf("handle dataset table %d\n", dataset.table) + return 0 +} +func go_req_handler(request *C.xce_request_t, arg unsafe.Pointer) C.int { + var from = C.GoString(request.from.name) + var body_which = request.body_which + + fmt.Println("api server handler called") + fmt.Printf("opcode %d body_which %d from %s\n", request.opcode, body_which, from) + + switch (request.opcode) { + case C.OPCode_push: + switch (body_which) { + case C.Request_body_configpushmsg: + // handle push config + return handle_config((*C.xce_configdata_t)(unsafe.Pointer(C.xce_request_body(request)))) + case C.Request_body_datamsg: + // handle pushed data msg + return handle_dataset((*C.dataset_t)(unsafe.Pointer(C.xce_request_body(request)))) + default: + // something wrong + } + default: + } + + C.free_Request(request) + + return 0 +} + +func main() { + h := cgo.NewHandle(go_req_handler) + rc := C.xce_apiserver_init(C.uintptr_t(h)) + fmt.Printf("rc %d\n", rc); + h.Delete() +} diff --git a/src/apiserver/meson.build b/src/apiserver/meson.build new file mode 100644 index 0000000..8506f21 --- /dev/null +++ b/src/apiserver/meson.build @@ -0,0 +1,73 @@ +cgo_cflags ='' +cgo_ldflags = '' + +if with_debug + cgo_cflags += '-fsanitize=address -g -DWITH_DEBUG=1 ' + cgo_ldflags += '-fsanitize=address -g ' +endif + +apiserver_ext_deps = [ + libxce_dep, + c_capnproto_dep, + ssl_dep, + tinylog_dep, + math_dep, + rt_dep, +] + +foreach e: apiserver_ext_deps + cflags_result = run_command(pkgconfig, '--cflags', e.name(), check: false) + + if cflags_result.returncode() == 0 + cgo_cflags += ' ' + cgo_cflags += cflags_result.stdout().strip() + endif + ldflags_result = run_command(pkgconfig, '--libs', e.name(), check: false) + + if ldflags_result.returncode() == 0 + cgo_ldflags += ' ' + cgo_ldflags += ldflags_result.stdout().strip() + endif +endforeach + +apiserver_deps = [ + apiserver_ext_deps, + libxce_dep, +] + +cgo_cflags += ' -I@CURRENT_SOURCE_DIR@/../include -I@CURRENT_SOURCE_DIR@/include -I@BUILD_ROOT@/lib ' +cgo_ldflags += ' -L@BUILD_ROOT@/lib -L@BUILD_ROOT@/apiserver -lxce' + +go_env = environment() +go_env.set('CGO_CFLAGS', cgo_cflags) +go_env.set('CGO_LDFLAGS', cgo_ldflags) +go_env.set('BUILDTYPE', get_option('buildtype')) +go_env.set('FIX_ENVARS','CGO_CFLAGS CGO_LDFLAGS') +go_env.set('DEPENDENCIES','thread math') +go_env.set('ENVARS', 'CURRENT_SOURCE_DIR BUILD_ROOT') + +#libapiserver_src = [ +# 'src/init.c', +#] + +#libapiserver_incdir = include_directories('../include', 'include', '../lib') + +#libapiserver_shared = shared_library('apiserver', +# libapiserver_src, +# include_directories: [libapiserver_incdir], +# dependencies: [libxce_dep], +# install: true) + +apiserver = custom_target('apiserver', + input: 'cmd/main.go', + output: 'apiserver', + command: [env_program, + 'CURRENT_SOURCE_DIR="@CURRENT_SOURCE_DIR@"', + 'BUILD_ROOT="@BUILD_ROOT@"', + go_build_program, + '@INPUT@', '@OUTPUT@'], + env: go_env, + depends: [libxce], + install: true, + install_dir: get_option('prefix') / 'bin') + diff --git a/src/controller/apiserver.c b/src/controller/apiserver.c new file mode 100644 index 0000000..a5d06b3 --- /dev/null +++ b/src/controller/apiserver.c @@ -0,0 +1,140 @@ +#include +#include + +#include "cluster.h" +#include "controller.capnp.h" +#include "request.h" +#include "xce/types.h" + +static int xce_apiserver_push_configs(cluster_t* c, endpoint_t* ep) +{ + xce_configdata_t configdata; + int rc = 0; + + memset(&configdata, 0x0, sizeof(configdata)); + configdata.data.clusterconfig = c->myconf->cluster_config; + configdata.data_which = ConfigData_data_clusterconfig; + + rc = xce_pushconfig_req(ep, c->myhost, &configdata); + + return rc; +} + +static int xce_apiserver_push_workers(cluster_t* c, endpoint_t* ep) +{ + xce_data_t data; + xce_workers_data_t workers + = { .n_total = 0, .pos = 0, .count = 0, .workers = NULL }; + int i, rc = 0; + + memset(&data, 0x0, sizeof(data)); + + data.table = Data_data_schedworkers; + data.data.workers = &workers; + + collect_workers(c, (collect_check_fn_t)check_worker_, &(workers.n_total), + &(workers.workers)); + workers.count = workers.n_total; + + rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100); + + /* free up collected resource value which used + passed to scheduler + */ + for (i = 0; i < workers.n_total; i++) { + xce_worker_t* w = workers.workers[i]; + + free_resval_with_idx_list(&(w->resvals_to_sched), w->nresvals_to_sched); + } + + deref_objs(workers.n_total, (void**)workers.workers); + FREEUP(workers.workers); + + return 0; +} + +static int check_job_(xce_job_t* j) +{ + if (j == NULL) { + return 0; + } + + return 1; +} + +static int xce_apiserver_push_jobs(cluster_t* c, endpoint_t* ep) +{ + xce_data_t data; + xce_jobs_data_t jobs = { .n_total = 0, .pos = 0, .count = 0, .jobs = NULL }; + int rc = 0; + + memset(&data, 0x0, sizeof(data)); + + data.table = Data_data_schedjobs; + data.data.jobs = &jobs; + + collect_jobs(c, (collect_check_fn_t)check_job_, &(jobs.n_total), + &(jobs.jobs)); + jobs.count = jobs.n_total; + + rc = xce_push_data_req(c->sched.ep, c->myhost, &data, check_controller, c, + 100); + deref_objs(jobs.n_total, (void**)jobs.jobs); + FREEUP(jobs.jobs); + + return 0; +} + +static int apiserver_job_hook_fun(int type, int situation, int detail, + void* data, va_list ap) +{ + return 0; +} + +static int apiserver_worker_hook_fun(int type, int situation, int detail, + void* data, va_list ap) +{ + return 0; +} + +static int apiserver_resource_hook_fun(int type, int situation, int detail, + void* data, va_list ap) +{ + return 0; +} + +static int apiserver_queue_hook_fun(int type, int situation, int detail, + void* data, va_list ap) +{ + return 0; +} + +int on_apiserver_join(cluster_t* c, endpoint_t* ep) +{ + /* FIXME: use lock to protect this later */ + static int hook_installed = 0; + + xce_apiserver_push_configs(c, ep); + xce_apiserver_push_workers(c, ep); + xce_apiserver_push_jobs(c, ep); + + if (!hook_installed) { + hook_installed = 1; + + xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, + (hook_fn_t)apiserver_job_hook_fun, NULL); + xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, + (hook_fn_t)apiserver_worker_hook_fun, NULL); + xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_RESOURCE, + (hook_fn_t)apiserver_resource_hook_fun, NULL); + xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_QUEUE, + (hook_fn_t)apiserver_queue_hook_fun, NULL); + } + return 0; +} + +int handle_apiserver_leave(cluster_t* c, endpoint_t* ep) +{ + /* FIXME */ + return 0; +} diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 1fd1d7a..9d8c688 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -67,6 +67,9 @@ static void handle_endpoint_error(endpoint_t* ep) case HostRole_client: handle_client_leave(c); break; + case HostRole_apiserver: + handle_apiserver_leave(c, ep); + break; default: ERROR("unknown role %d for endpoint %p", priv->role, ep); break; diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index b5d5a4e..b1edd54 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -56,6 +56,7 @@ extern queue_t* qmgr_lockq(qmgr_t*, char*, int); extern int xce_update_data(endpoint_t*, cluster_t*, int, int, int*, int, xce_kv_t*); extern int on_scheduler_join(cluster_t*); +extern int on_apiserver_join(cluster_t*, endpoint_t*); extern int xce_push_reslist(endpoint_t*, cluster_t*, int, xce_resource_t**); extern int xce_push_confvars(endpoint_t*, cluster_t*, int, xce_config_t**); @@ -135,5 +136,7 @@ extern int release_resource_from_worker(char*, cluster_t*); 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*); + #endif diff --git a/src/controller/join.c b/src/controller/join.c index d04bdc3..0dc2ffa 100644 --- a/src/controller/join.c +++ b/src/controller/join.c @@ -117,6 +117,24 @@ 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; + + TRACE("%s api server join, ep %p", __func__, ep); + + init_reply_by_req(&reply, request); + + rc = on_apiserver_join(c, ep); + + reply.status = rc; + reply.data_which = RequestReply_data_nodata; + xce_reply(ep, c->myhost, &reply); + + return 0; +} + static int handle_join_(cluster_t* c, endpoint_t* ep, xce_request_t* request) { ep_private_t* priv; @@ -158,6 +176,9 @@ static int handle_join_(cluster_t* c, endpoint_t* ep, xce_request_t* request) case HostRole_controller: rc = handle_controller_join(c, ep, request); break; + case HostRole_apiserver: + rc = handle_apiserver_join(c, ep, request); + break; default: rc = -E_XCE_BAD_ROLE; ERROR("unknown role %d from host %s", role, from->name); diff --git a/src/controller/meson.build b/src/controller/meson.build index d0effc5..ebcd21a 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -15,7 +15,8 @@ controller_src = [ 'task.c', 'worker.c', 'xceobjs.c', - 'myconf.c' + 'apiserver.c', + 'myconf.c', ] incdir = include_directories('.', '../include', '../lib', diff --git a/src/include/client.h b/src/include/client.h index 89c8fe5..39e4ab8 100644 --- a/src/include/client.h +++ b/src/include/client.h @@ -1,6 +1,6 @@ #if !defined(_CLIENT_H_) -#define _CLIENT_H_ +#define _CLIENT_H_ 1 #include diff --git a/src/include/typemap.h b/src/include/typemap.h index 47593fb..b26d61e 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -4,6 +4,7 @@ #include #include "xce/types.h" +#include "clusterconfig.h" typedef struct { char* resname; @@ -64,6 +65,7 @@ typedef struct { union { xce_conf4sched_t* conf4sched; xce_conf4worker_t* conf4worker; + cluster_config_t* clusterconfig; } data; } xce_configdata_t; diff --git a/src/lib/meson.build b/src/lib/meson.build index 2ac5147..8d9bafd 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -53,11 +53,12 @@ ssl_dep = dependency('openssl') libxce_deps = [c_capnproto_dep, ssl_dep, tinylog_dep, thread_dep] -libxce_shared = shared_library('xce', - libxce_src, - include_directories: [incdir, build_inc], - dependencies: libxce_deps, - install: true) +#libxce_shared = shared_library('xce', +# libxce_src, +# include_directories: [incdir, build_inc], +# dependencies: libxce_deps, +# install: true) + libxce = static_library('xce', libxce_src, include_directories: [incdir, build_inc], @@ -66,6 +67,7 @@ libxce = static_library('xce', libxce_dep = declare_dependency(link_with: libxce) +# libxce_shared_dep = declare_dependency(link_with: libxce_shared) libxce_test_src = ['testsuite/test.c', 'testsuite/test_error.c', diff --git a/src/lib/request.c b/src/lib/request.c index 74ca549..80e159c 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -188,6 +188,7 @@ int xce_join_req(endpoint_t* ep, int role, hostattrs_t* attrs) if ((role != HostRole_client) && (role != HostRole_worker) && + (role != HostRole_apiserver) && (role != HostRole_scheduler)) { return -EINVAL; } diff --git a/src/meson.build b/src/meson.build index 68f8f6f..d2c0bfe 100644 --- a/src/meson.build +++ b/src/meson.build @@ -58,3 +58,4 @@ subdir('controller') subdir('worker') subdir('autoscaler') subdir('scheduler') +subdir('apiserver') diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 2896197..622d805 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -127,6 +127,7 @@ enum HostRole { worker @1; scheduler @2; controller @3; + apiserver @4; } struct JoinData $C.mapname("xce_joindata_t") { @@ -179,8 +180,8 @@ struct Resource $C.mapname("xce_resource_t"){ flags @5: Int32; interval @6: Int32; lastReport @7: UInt64 $C.mapname("last_report"); - distributions @8: List(ResourceDistribution) $C.maplistcount("ndistributions"); - hostBased @9: Value $C.mapname("hbased_defined"); +# distributions @8: List(ResourceDistribution) $C.maplistcount("ndistributions"); +# hostBased @9: Value $C.mapname("hbased_defined"); } struct Threshold $C.mapname("xce_res_threshold_t"){ @@ -227,6 +228,7 @@ struct ConfigData $C.mapname("xce_configdata_t") { data :union $C.mapuniontag("data_which") { schedconfig @0: Conf4Sched $C.mapname("conf4sched"); workerconfig @1: Conf4Worker $C.mapname("conf4worker"); + clusterconfig @2: ClusterConfig $C.mapname("clusterconfig"); } }