initial work of api server

src/apiserver/cmd/main.go

@@ -0,0 +1,136 @@
+package main
+
+/*
+   #include <stdlib.h>
+   #include <stdint.h>
+   #include <errno.h>
+   
+   #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()
+}

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')
+

src/controller/apiserver.c

@@ -0,0 +1,140 @@
+#include <errno.h>
+#include <stdlib.h>
+
+#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;
+}

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;

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

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);

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',

src/include/client.h

@@ -1,6 +1,6 @@
 #if !defined(_CLIENT_H_)
 
-#define _CLIENT_H_
+#define _CLIENT_H_ 1
 
 #include <stdint.h>
 

src/include/typemap.h

@@ -4,6 +4,7 @@
 #include <stdint.h>
 
 #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;
 

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_shared = shared_library('xce',
-                               libxce_src,
+#                               libxce_src,
-                               include_directories: [incdir, build_inc],
+#                               include_directories: [incdir, build_inc],
-                               dependencies: libxce_deps,
+#                               dependencies: libxce_deps,
-                               install: true)
+#                               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',

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;
     }

src/meson.build

@@ -58,3 +58,4 @@ subdir('controller')
 subdir('worker')
 subdir('autoscaler')
 subdir('scheduler')
+subdir('apiserver')

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");
+#  distributions  @8: List(ResourceDistribution) $C.maplistcount("ndistributions");
-  hostBased      @9: Value $C.mapname("hbased_defined");
+#  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"); 
   }
 }