Merge Go frontend (apiserver + xcecli) into meson build

This commit is contained in:
Rongsong Shen 2026-04-14 11:20:17 +08:00
parent 43dc12a36c
commit c9dc8c4676
205 changed files with 30793 additions and 328 deletions

View file

@ -0,0 +1,84 @@
package auth
import (
"fmt"
"gopkg.in/yaml.v3"
)
// AuthFileConfig is the top-level structure of auth.yaml.
type AuthFileConfig struct {
Provider string `yaml:"provider"`
ConfigFile *ConfigFileAuth `yaml:"config_file"`
}
// ConfigFileAuth holds the user entries for config_file provider.
type ConfigFileAuth struct {
Users []UserEntry `yaml:"users"`
}
// UserEntry represents a single user in auth.yaml.
type UserEntry struct {
Username string `yaml:"username"`
Password string `yaml:"password"`
Roles []string `yaml:"roles"`
}
// ConfigFileProvider authenticates users against a YAML config file.
type ConfigFileProvider struct {
users map[string]UserEntry
}
// NewConfigFileProvider creates a provider from raw YAML bytes.
func NewConfigFileProvider(data []byte) (*ConfigFileProvider, error) {
var cfg AuthFileConfig
if err := yaml.Unmarshal(data, &cfg); err != nil {
return nil, fmt.Errorf("failed to parse auth config: %w", err)
}
if len(cfg.ConfigFile.Users) == 0 {
return nil, fmt.Errorf("no users defined in auth config")
}
users := make(map[string]UserEntry, len(cfg.ConfigFile.Users))
for _, u := range cfg.ConfigFile.Users {
if u.Username == "" {
return nil, fmt.Errorf("empty username in auth config")
}
users[u.Username] = u
}
return &ConfigFileProvider{users: users}, nil
}
// Authenticate checks username/password against the config file.
func (p *ConfigFileProvider) Authenticate(username, password string) error {
entry, ok := p.users[username]
if !ok {
return fmt.Errorf("invalid username or password")
}
if entry.Password != password {
return fmt.Errorf("invalid username or password")
}
return nil
}
// GetUserRoles returns roles defined in the config for the given user.
func (p *ConfigFileProvider) GetUserRoles(username string) []string {
if entry, ok := p.users[username]; ok {
return entry.Roles
}
return nil
}
// Name returns the provider identifier.
func (p *ConfigFileProvider) Name() string { return "config_file" }
// AllUsers returns all usernames known to this provider (implements UserRoleSeeder).
func (p *ConfigFileProvider) AllUsers() []string {
users := make([]string, 0, len(p.users))
for u := range p.users {
users = append(users, u)
}
return users
}

View file

@ -0,0 +1,133 @@
package auth
import (
"testing"
)
func TestConfigFileProvider_Authenticate(t *testing.T) {
yaml := `
provider: config_file
config_file:
users:
- username: admin
password: secret
roles: [cluster_admin]
- username: viewer
password: pass
roles: [basic_user]
`
provider, err := NewConfigFileProvider([]byte(yaml))
if err != nil {
t.Fatalf("NewConfigFileProvider: %v", err)
}
tests := []struct {
name string
user string
pass string
wantErr bool
}{
{"valid admin", "admin", "secret", false},
{"valid viewer", "viewer", "pass", false},
{"wrong password", "admin", "wrong", true},
{"unknown user", "nobody", "pass", true},
{"empty user", "", "pass", true},
{"empty password", "admin", "", true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := provider.Authenticate(tt.user, tt.pass)
if (err != nil) != tt.wantErr {
t.Errorf("Authenticate() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
func TestConfigFileProvider_GetUserRoles(t *testing.T) {
yaml := `
provider: config_file
config_file:
users:
- username: admin
password: admin
roles: [cluster_admin, operator]
- username: guest
password: guest
roles: []
`
provider, err := NewConfigFileProvider([]byte(yaml))
if err != nil {
t.Fatalf("NewConfigFileProvider: %v", err)
}
if roles := provider.GetUserRoles("admin"); len(roles) != 2 {
t.Errorf("admin roles = %v, want 2", roles)
}
if roles := provider.GetUserRoles("guest"); len(roles) != 0 {
t.Errorf("guest roles = %v, want empty", roles)
}
if roles := provider.GetUserRoles("unknown"); roles != nil {
t.Errorf("unknown roles = %v, want nil", roles)
}
}
func TestConfigFileProvider_Name(t *testing.T) {
p := &ConfigFileProvider{}
if got := p.Name(); got != "config_file" {
t.Errorf("Name() = %q, want %q", got, "config_file")
}
}
func TestNewConfigFileProvider_Errors(t *testing.T) {
t.Run("invalid yaml", func(t *testing.T) {
_, err := NewConfigFileProvider([]byte("{{invalid"))
if err == nil {
t.Error("expected error for invalid yaml")
}
})
t.Run("no users", func(t *testing.T) {
_, err := NewConfigFileProvider([]byte("provider: config_file\nconfig_file:\n users: []"))
if err == nil {
t.Error("expected error for empty users list")
}
})
t.Run("empty username", func(t *testing.T) {
_, err := NewConfigFileProvider([]byte("provider: config_file\nconfig_file:\n users:\n - username: ''\n password: x"))
if err == nil {
t.Error("expected error for empty username")
}
})
}
func TestPAMProvider_Name(t *testing.T) {
p := &PAMProvider{}
if got := p.Name(); got != "pam" {
t.Errorf("Name() = %q, want %q", got, "pam")
}
}
func TestLDAPProvider_Name(t *testing.T) {
p := &LDAPProvider{}
if got := p.Name(); got != "ldap" {
t.Errorf("Name() = %q, want %q", got, "ldap")
}
if err := p.Authenticate("u", "p"); err == nil {
t.Error("expected not-implemented error")
}
}
func TestOAuth2Provider_Name(t *testing.T) {
p := &OAuth2Provider{}
if got := p.Name(); got != "oauth2" {
t.Errorf("Name() = %q, want %q", got, "oauth2")
}
if err := p.Authenticate("u", "p"); err == nil {
t.Error("expected not-implemented error")
}
}

View file

@ -0,0 +1,44 @@
package auth
import (
"fmt"
"os"
"gopkg.in/yaml.v3"
)
var GlobalProvider AuthProvider
// InitAuthProvider reads auth.yaml and creates the appropriate provider.
func InitAuthProvider(configPath string) error {
data, err := os.ReadFile(configPath)
if err != nil {
return fmt.Errorf("failed to read auth config %s: %w", configPath, err)
}
var cfg struct {
Provider string `yaml:"provider"`
}
if err := yaml.Unmarshal(data, &cfg); err != nil {
return fmt.Errorf("failed to parse auth config: %w", err)
}
switch cfg.Provider {
case "config_file":
provider, err := NewConfigFileProvider(data)
if err != nil {
return fmt.Errorf("failed to create config_file provider: %w", err)
}
GlobalProvider = provider
case "pam":
GlobalProvider = &PAMProvider{}
case "ldap":
GlobalProvider = &LDAPProvider{}
case "oauth2":
GlobalProvider = &OAuth2Provider{}
default:
return fmt.Errorf("unsupported auth provider: %s", cfg.Provider)
}
return nil
}

View file

@ -0,0 +1,17 @@
package auth
import "fmt"
// LDAPProvider is a stub for future LDAP authentication.
type LDAPProvider struct{}
// Authenticate verifies credentials via LDAP (not yet implemented).
func (p *LDAPProvider) Authenticate(username, password string) error {
return fmt.Errorf("LDAP authentication not yet implemented")
}
// GetUserRoles returns nil (not yet implemented).
func (p *LDAPProvider) GetUserRoles(username string) []string { return nil }
// Name returns the provider identifier.
func (p *LDAPProvider) Name() string { return "ldap" }

View file

@ -0,0 +1,17 @@
package auth
import "fmt"
// OAuth2Provider is a stub for future OAuth2/OIDC authentication.
type OAuth2Provider struct{}
// Authenticate verifies credentials via OAuth2 (not yet implemented).
func (p *OAuth2Provider) Authenticate(username, password string) error {
return fmt.Errorf("OAuth2 authentication not yet implemented")
}
// GetUserRoles returns nil (not yet implemented).
func (p *OAuth2Provider) GetUserRoles(username string) []string { return nil }
// Name returns the provider identifier.
func (p *OAuth2Provider) Name() string { return "oauth2" }

View file

@ -0,0 +1,11 @@
package auth
// PAMProvider authenticates users via the system PAM stack.
// Only available on Linux.
type PAMProvider struct{}
// GetUserRoles returns nil — PAM users get roles from RBAC config.
func (p *PAMProvider) GetUserRoles(username string) []string { return nil }
// Name returns the provider identifier.
func (p *PAMProvider) Name() string { return "pam" }

View file

@ -0,0 +1,112 @@
//go:build linux && cgo
package auth
/*
#cgo LDFLAGS: -ldl
#include <dlfcn.h>
#include <stdlib.h>
#include <string.h>
// PAM types and constants (from <security/pam_appl.h>)
#define PAM_SUCCESS 0
#define PAM_PROMPT_ECHO_OFF 1
typedef struct pam_message {
int msg_style;
const char *msg;
} pam_message;
typedef struct pam_response {
char *resp;
int resp_retcode;
} pam_response;
typedef struct pam_conv {
int (*conv)(int, const struct pam_message **, struct pam_response **, void *);
void *appdata_ptr;
} pam_conv;
typedef void* pam_handle_t;
// Function pointer types
typedef int (*pam_start_fn)(const char*, const char*, const pam_conv*, pam_handle_t**);
typedef int (*pam_authenticate_fn)(pam_handle_t*, int);
typedef int (*pam_end_fn)(pam_handle_t*, int);
static void *pam_handle = NULL;
static pam_start_fn fn_pam_start = NULL;
static pam_authenticate_fn fn_pam_authenticate = NULL;
static pam_end_fn fn_pam_end = NULL;
// load_pam opens libpam.so dynamically. Returns 0 on success.
static int load_pam() {
if (pam_handle) return 0;
const char *libs[] = {"libpam.so.0", "libpam.so", NULL};
for (int i = 0; libs[i]; i++) {
pam_handle = dlopen(libs[i], RTLD_LAZY);
if (pam_handle) break;
}
if (!pam_handle) return -1;
fn_pam_start = (pam_start_fn) dlsym(pam_handle, "pam_start");
fn_pam_authenticate = (pam_authenticate_fn) dlsym(pam_handle, "pam_authenticate");
fn_pam_end = (pam_end_fn) dlsym(pam_handle, "pam_end");
if (!fn_pam_start || !fn_pam_authenticate || !fn_pam_end) {
dlclose(pam_handle);
pam_handle = NULL;
return -2;
}
return 0;
}
// conv_fn is the PAM conversation callback — returns the password.
static int conv_fn(int num_msg, const pam_message **msg,
pam_response **resp, void *appdata) {
*resp = (pam_response*)calloc(num_msg, sizeof(pam_response));
if (!*resp) return 4; // PAM_BUF_ERR
for (int i = 0; i < num_msg; i++) {
if (msg[i]->msg_style == PAM_PROMPT_ECHO_OFF) {
(*resp)[i].resp = strdup((const char*)appdata);
}
}
return PAM_SUCCESS;
}
// pam_auth_dynamic authenticates username/password via dlopen'd libpam.
// Returns 0 on success, -1 if libpam not found, PAM error code otherwise.
static int pam_auth_dynamic(const char *username, const char *password) {
if (load_pam() != 0) return -1;
pam_conv conv = { conv_fn, (void*)password };
pam_handle_t *pamh = NULL;
int ret = fn_pam_start("system-auth", username, &conv, &pamh);
if (ret != PAM_SUCCESS) return ret;
ret = fn_pam_authenticate(pamh, 0);
fn_pam_end(pamh, ret);
return ret;
}
*/
import "C"
import (
"fmt"
"unsafe"
)
// Authenticate verifies credentials via dynamically loaded libpam.so.
// The binary does not link against libpam at build time — it is loaded
// at runtime only when this provider is active.
func (p *PAMProvider) Authenticate(username, password string) error {
cu := C.CString(username)
cp := C.CString(password)
defer C.free(unsafe.Pointer(cu))
defer C.free(unsafe.Pointer(cp))
ret := C.pam_auth_dynamic(cu, cp)
switch ret {
case 0:
return nil
case -1:
return fmt.Errorf("PAM library (libpam.so) not found on this system")
default:
return fmt.Errorf("invalid username or password")
}
}

View file

@ -0,0 +1,13 @@
//go:build !linux || !cgo
package auth
import (
"fmt"
"runtime"
)
// Authenticate returns an error when PAM is not available (non-Linux or CGO disabled).
func (p *PAMProvider) Authenticate(username, password string) error {
return fmt.Errorf("PAM authentication is not supported on %s or without CGO; use config_file provider", runtime.GOOS)
}

View file

@ -0,0 +1,14 @@
package auth
// UserRoleSeeder is an optional interface for providers that can enumerate all users.
// Used to pre-populate the role store at startup so permissions work after restart.
type UserRoleSeeder interface {
AllUsers() []string
}
// AuthProvider is the interface for authentication strategies.
type AuthProvider interface {
Authenticate(username, password string) error
GetUserRoles(username string) []string
Name() string
}

View file

@ -0,0 +1,319 @@
//go:build cgo
package backend
/*
#include <stdlib.h>
#include <stdint.h>
#include <errno.h>
#include "client.h"
#include "netio.h"
#include "helper.h"
#include "controller.capnp.h"
// backend_ctx_t wraps xcecli_ctx_t for use as the C handler's arg.
typedef struct {
xcecli_ctx_t ctx;
} backend_ctx_t;
// go_push_handler is exported from Go and called from C for every incoming request.
extern int go_push_handler(xce_request_t*, void*);
static int xce_backend_handler(xce_request_t *req, void *arg) {
return go_push_handler(req, arg);
}
// xce_backend_init initialises the client context, sets the handler,
// starts the network loop, then joins as HostRole_apiserver.
static int xce_backend_init(xce_client_config_t *config, backend_ctx_t *ctx) {
int rc = xcecli_ctx_init(&(ctx->ctx), config);
if (rc < 0) return rc;
xcecli_set_handler(ctx->ctx, (xce_request_handler_t)xce_backend_handler, NULL);
rc = xcecli_start(ctx->ctx);
if (rc < 0) return rc;
return xcecli_join(ctx->ctx, HostRole_apiserver, NULL);
}
// Helpers to extract typed row pointers from a dataset_t.
static xce_job_t **dataset_jobs(dataset_t *d) { return (xce_job_t **)d->rows; }
static xce_worker_t **dataset_workers(dataset_t *d) { return (xce_worker_t **)d->rows; }
// CGO cannot access C union fields by name (they appear as [N]byte).
// These thin helpers set the correct union branch from C so Go callers
// don't need unsafe pointer arithmetic.
static void jobctrl_set_siginfo(xce_jobctrl_t *ctrl, xce_signal_t *sig) {
ctrl->u.siginfo = sig;
}
static void jobctrl_set_ckptoption(xce_jobctrl_t *ctrl, xce_ckpt_options_t *ckpt) {
ctrl->u.ckptoption = ckpt;
}
*/
import "C"
import (
"apiserver/models"
logger "apiserver/utils"
"fmt"
"unsafe"
"go.uber.org/zap"
)
// gCtx holds the C client context for the lifetime of the process.
var gCtx C.backend_ctx_t
// gHandler receives push notifications from the backend.
// Must be set via StartNotificationListener before RegisterToBackend.
var gHandler NotificationHandler
// RegisterToBackend initialises the XCE client library and joins as an API
// server. StartNotificationListener should be called first so that push
// events have a handler ready before the first push can arrive.
func RegisterToBackend(config *BackendConfig) error {
from := config.From
if from == "" {
from = "xce_apiserver"
}
key := config.Key
if key == "" {
key = "keepsecret$"
}
cAddr := C.CString(config.Address)
cFrom := C.CString(from)
cKey := C.CString(key)
defer func() {
C.free(unsafe.Pointer(cAddr))
C.free(unsafe.Pointer(cFrom))
C.free(unsafe.Pointer(cKey))
}()
var cConfig C.xce_client_config_t
cConfig.nmasters = 1
cConfig.masters[0] = cAddr
cConfig.master_ports[0] = C.int(config.Port)
cConfig.from = cFrom
cConfig.key = cKey
rc := C.xce_backend_init(&cConfig, &gCtx)
if rc < 0 {
return fmt.Errorf("xce_backend_init failed: %d", int(rc))
}
logger.Info("Backend connection established",
zap.String("address", config.Address),
zap.Int("port", config.Port))
return nil
}
// StartNotificationListener registers the handler that receives backend push
// events. Call this before RegisterToBackend.
func StartNotificationListener(handler NotificationHandler) error {
gHandler = handler
return nil
}
// go_push_handler is called by the C library on every incoming request.
// It must be exported so the C trampoline can call it from a C thread.
//
//export go_push_handler
func go_push_handler(req *C.xce_request_t, _ unsafe.Pointer) C.int {
if req == nil {
return 0
}
defer C.free_Request(req)
if gHandler == nil {
return 0
}
switch req.opcode {
case C.OPCode_push:
return handlePush(req)
}
return 0
}
// handlePush dispatches an OPCode_push request to the appropriate handler.
func handlePush(req *C.xce_request_t) C.int {
switch req.body_which {
case C.Request_body_datamsg:
dataset := (*C.dataset_t)(unsafe.Pointer(C.xce_request_body(req)))
return handleDataset(dataset)
case C.Request_body_configpushmsg:
cfgdata := (*C.xce_configdata_t)(unsafe.Pointer(C.xce_request_body(req)))
return handleConfigPush(cfgdata)
}
return 0
}
// handleDataset processes a dataset push (jobs or workers).
func handleDataset(d *C.dataset_t) C.int {
switch d.table {
case C.Data_data_jobs:
n := int(d.nrows)
if n == 0 {
return 0
}
jobs := unsafe.Slice(C.dataset_jobs(d), n)
for _, jp := range jobs {
if jp == nil {
continue
}
info := cJobToModel(jp)
gHandler.OnJobUpdated(info)
}
case C.Data_data_workers:
n := int(d.nrows)
if n == 0 {
return 0
}
workers := unsafe.Slice(C.dataset_workers(d), n)
for _, wp := range workers {
if wp == nil {
continue
}
info := cWorkerToModel(wp)
gHandler.OnHostUpdated(info)
}
}
return 0
}
// handleConfigPush processes a cluster configuration push.
func handleConfigPush(cfgdata *C.xce_configdata_t) C.int {
if cfgdata.data_which != C.ConfigData_data_clusterconfig {
return 0
}
conf := (*C.cluster_config_t)(C.xce_configdata_ptr(cfgdata))
if conf == nil {
return 0
}
for _, q := range cClusterConfQueues(conf) {
qi := q
gHandler.OnQueueUpdated(&qi)
}
for _, r := range cClusterConfResources(conf) {
ri := r
gHandler.OnResourceUpdated(&ri)
}
return 0
}
// SubmitJob sends a job submission to the backend via xcecli_jobsubmit.
// The store is updated only when the backend pushes the resulting OnJobCreated
// notification — not immediately here.
func SubmitJob(job *models.JobInfo) error {
js, sd, si := modelJobToSubmit(job)
defer freeJobSubmit(js, sd, si)
var outJobID C.uint64_t
rc := C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID)
if rc < 0 {
return fmt.Errorf("xcecli_jobsubmit failed: %d", int(rc))
}
if outJobID != 0 {
job.Unique_JobID = int64(outJobID)
job.JobID = int64(outJobID)
}
return nil
}
// SubmitArrayJobs submits an array job via xcecli_jobsubmit with array_size > 1.
func SubmitArrayJobs(arrayjob *models.ArrayJobInfo, jobs []models.JobInfo) error {
if len(jobs) == 0 {
return fmt.Errorf("no jobs in array")
}
// Use the first job as the template; set array_size to the total count.
js, sd, si := modelJobToSubmit(&jobs[0])
defer freeJobSubmit(js, sd, si)
js.array_size = C.uint64_t(len(jobs))
var outJobID C.uint64_t
rc := C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID)
if rc < 0 {
return fmt.Errorf("xcecli_jobsubmit (array) failed: %d", int(rc))
}
if outJobID != 0 {
arrayjob.JobID = int64(outJobID)
}
return nil
}
// ControlJob sends a job control action to the backend via xcecli_jobctrl.
// action must be one of the models.JOB_* constants; content is an optional
// signal name (e.g. "TERM") for kill operations.
func ControlJob(jobid uint64, action int, content string) error {
var ctrl C.xce_jobctrl_t
ctrl.jobid = C.uint64_t(jobid)
ctrl.action = C.int(action)
// For signal-based actions set signo; for checkpoint set ckpt options.
// xce_jobctrl_t.u is a C union — we use a static C helper to set the
// right branch safely.
switch action {
case models.JOB_STOP, models.JOB_RESUME, models.JOB_KILL:
var sig C.xce_signal_t
sig.signo = C.int(signalNumber(content, action))
sig.force = 0
C.jobctrl_set_siginfo(&ctrl, &sig)
case models.JOB_CHKPNT:
var ckpt C.xce_ckpt_options_t
ckpt.kill = 0
C.jobctrl_set_ckptoption(&ctrl, &ckpt)
}
rc := C.xcecli_jobctrl(gCtx.ctx, &ctrl)
if rc < 0 {
return fmt.Errorf("xcecli_jobctrl failed: %d", int(rc))
}
return nil
}
// ControlQueue sends a queue open/close/activate/deactivate action to the
// backend via xcecli_queuectrl.
func ControlQueue(queuename string, action int) error {
cQueue := C.CString(queuename)
defer C.free(unsafe.Pointer(cQueue))
ctrl := C.xce_queuectrl_t{
action: C.int(action),
queue: cQueue,
}
rc := C.xcecli_queuectrl(gCtx.ctx, &ctrl)
if rc < 0 {
return fmt.Errorf("xcecli_queuectrl failed: %d", int(rc))
}
return nil
}
// ControlHost is a placeholder; host control via xcecli is not yet exposed
// in client.h. Use the REST API workaround until the C library adds
// xcecli_hostctrl.
func ControlHost(hostname string, action int) error {
return fmt.Errorf("host control not yet supported via C library (host: %s, action: %d)", hostname, action)
}
// signalNumber maps a job action and optional signal name to a POSIX signal
// number, falling back to sensible defaults.
func signalNumber(signame string, action int) int {
if signame != "" {
if n, ok := models.JobSignalMapTable[signame]; ok {
return n
}
}
switch action {
case models.JOB_KILL:
return models.JobSignalMapTable["TERM"] // 15
case models.JOB_STOP:
return models.JobSignalMapTable["TSTP"] // 20
case models.JOB_RESUME:
return models.JobSignalMapTable["CONT"] // 18
}
return 15 // SIGTERM as safe default
}

View file

@ -0,0 +1,10 @@
package backend
// BackendConfig holds connection parameters for the XCE controller.
// Used by RegisterToBackend to establish the C client connection.
type BackendConfig struct {
Address string // controller hostname (e.g. "localhost")
Port int // controller port (e.g. 9990)
From string // identity sent to controller (default: "xce_apiserver")
Key string // authentication key
}

View file

@ -0,0 +1,210 @@
//go:build cgo
package backend
/*
#include <stdlib.h>
#include "client.h"
*/
import "C"
import (
"apiserver/models"
"unsafe"
)
// goString safely converts a C string to a Go string, handling nil.
func goString(s *C.char) string {
if s == nil {
return ""
}
return C.GoString(s)
}
// cJobToModel extracts C xce_job_t fields into a rawJobData and maps it to JobInfo.
func cJobToModel(job *C.xce_job_t) *models.JobInfo {
r := rawJobData{
jobID: int64(job.jobid),
state: int(job.state),
runTime: int(job.run_time),
subTime: int64(job.sub_time),
queueTime: int64(job.queue_time),
startTime: int64(job.start_time),
finishTime: int64(job.finish_time),
}
if job.submit != nil {
s := job.submit
rd := &rawSubmitData{
queue: goString(s.queue),
user: goString(s.user),
jobName: goString(s.job_name),
projectName: goString(s.project_name),
userGroup: goString(s.user_group),
jobDesc: goString(s.job_description),
resReq: goString(s.resreq),
minTasks: int(s.mintasks),
maxTasks: int(s.maxtasks),
}
if s.start_info != nil {
si := s.start_info
rd.startInfo = &rawStartInfo{
cmdline: goString(si.cmdline),
launchType: goString(si.launch_type),
preExec: goString(si.job_pre_exec),
postExec: goString(si.job_post_exec),
execHome: goString(si.exechome),
execCWD: goString(si.execcwd),
execUser: goString(si.execuser),
}
}
r.submit = rd
}
return rawJobToModel(r)
}
// cWorkerToModel extracts C xce_worker_t fields into a rawWorkerData and maps it to HostInfo.
func cWorkerToModel(worker *C.xce_worker_t) *models.HostInfo {
r := rawWorkerData{
name: goString(worker.name),
status: int(worker.status),
}
if worker.attrs != nil {
a := worker.attrs
r.hasAttrs = true
r.hostType = goString(a.hosttype)
r.hostModel = goString(a.hostmodel)
r.cpuFactor = float64(a.cpufactor)
r.maxCPUs = int(a.maxcpus)
r.maxMem = int(a.maxmem)
r.maxSwap = int(a.maxswap)
r.maxTmp = int(a.maxtmp)
r.nDisks = int(a.ndisks)
}
return rawWorkerToModel(r)
}
// cQueueToModel extracts C xce_queue_t fields and maps them to QueueInfo.
func cQueueToModel(queue *C.xce_queue_t) *models.QueueInfo {
return rawQueueToModel(rawQueueData{
name: goString(queue.queue),
status: int(queue.status),
priority: int(queue.priority),
})
}
// cResourceToModel extracts C xce_resource_t fields and maps them to ResourceDefinitionInfo.
func cResourceToModel(res *C.xce_resource_t) *models.ResourceDefinitionInfo {
return rawResourceToModel(rawResourceData{
name: goString(res.name),
desc: goString(res.desc),
valType: int(res.valtype),
ordType: int(res.ordtype),
flags: int(res.flags),
interval: int(res.interval),
})
}
// cClusterConfQueues extracts queue models from a cluster_config_t.
func cClusterConfQueues(conf *C.cluster_config_t) []models.QueueInfo {
n := int(conf.nqueues)
if n == 0 {
return nil
}
qps := unsafe.Slice(conf.queues, n)
out := make([]models.QueueInfo, 0, n)
for _, qp := range qps {
if qp != nil {
out = append(out, *cQueueToModel(qp))
}
}
return out
}
// cClusterConfResources extracts resource definition models from a cluster_config_t.
func cClusterConfResources(conf *C.cluster_config_t) []models.ResourceDefinitionInfo {
n := int(conf.nres)
if n == 0 {
return nil
}
rps := unsafe.Slice(conf.resources, n)
out := make([]models.ResourceDefinitionInfo, 0, n)
for _, rp := range rps {
if rp != nil {
out = append(out, *cResourceToModel(rp))
}
}
return out
}
// modelJobToSubmit converts a Go JobInfo to C xce_job_submit_t heap structs.
// The caller must free all returned C memory by calling freeJobSubmit.
func modelJobToSubmit(job *models.JobInfo) (*C.xce_job_submit_t, *C.job_submit_data_t, *C.job_startinfo_t) {
si := (*C.job_startinfo_t)(C.calloc(1, C.sizeof_job_startinfo_t))
si.cmdline = C.CString(job.Command)
if job.LaunchType != "" {
si.launch_type = C.CString(job.LaunchType)
}
if job.PreExecCmd != "" {
si.job_pre_exec = C.CString(job.PreExecCmd)
}
if job.PostExecCmd != "" {
si.job_post_exec = C.CString(job.PostExecCmd)
}
sd := (*C.job_submit_data_t)(C.calloc(1, C.sizeof_job_submit_data_t))
sd.queue = C.CString(job.QueueName)
sd.user = C.CString(job.SubmitUser)
sd.job_name = C.CString(job.JobName)
if job.ProjectName != "" {
sd.project_name = C.CString(job.ProjectName)
}
if job.UserGroup != "" {
sd.user_group = C.CString(job.UserGroup)
}
if job.JobDescription != "" {
sd.job_description = C.CString(job.JobDescription)
}
if job.ResReq != "" {
sd.resreq = C.CString(job.ResReq)
}
sd.mintasks = C.uint(job.NumProcessors)
if sd.mintasks == 0 {
sd.mintasks = 1
}
sd.maxtasks = C.uint(job.MaxNumProcessors)
if sd.maxtasks < sd.mintasks {
sd.maxtasks = sd.mintasks
}
sd.start_info = si
js := (*C.xce_job_submit_t)(C.calloc(1, C.sizeof_xce_job_submit_t))
js.jobid = C.uint64_t(job.Unique_JobID)
js.array_size = 1
js.submit = sd
return js, sd, si
}
// freeJobSubmit releases C memory allocated by modelJobToSubmit.
func freeJobSubmit(js *C.xce_job_submit_t, sd *C.job_submit_data_t, si *C.job_startinfo_t) {
if si != nil {
C.free(unsafe.Pointer(si.cmdline))
C.free(unsafe.Pointer(si.launch_type))
C.free(unsafe.Pointer(si.job_pre_exec))
C.free(unsafe.Pointer(si.job_post_exec))
C.free(unsafe.Pointer(si))
}
if sd != nil {
C.free(unsafe.Pointer(sd.queue))
C.free(unsafe.Pointer(sd.user))
C.free(unsafe.Pointer(sd.job_name))
C.free(unsafe.Pointer(sd.project_name))
C.free(unsafe.Pointer(sd.user_group))
C.free(unsafe.Pointer(sd.job_description))
C.free(unsafe.Pointer(sd.resreq))
C.free(unsafe.Pointer(sd))
}
if js != nil {
C.free(unsafe.Pointer(js))
}
}

View file

@ -0,0 +1,161 @@
package backend
import (
"apiserver/models"
"time"
)
// rawStartInfo mirrors the C job_startinfo_t fields used by the Go model.
type rawStartInfo struct {
cmdline string
launchType string
preExec string
postExec string
execHome string
execCWD string
execUser string
}
// rawSubmitData mirrors the C job_submit_data_t fields used by the Go model.
type rawSubmitData struct {
queue string
user string
jobName string
projectName string
userGroup string
jobDesc string
resReq string
minTasks int
maxTasks int
startInfo *rawStartInfo
}
// rawJobData mirrors the C xce_job_t fields used by the Go model.
type rawJobData struct {
jobID int64
state int
runTime int
subTime int64
queueTime int64
startTime int64
finishTime int64
submit *rawSubmitData
}
// rawWorkerData mirrors the C xce_worker_t fields used by the Go model.
type rawWorkerData struct {
name string
status int
hasAttrs bool
hostType string
hostModel string
cpuFactor float64
maxCPUs int
maxMem int
maxSwap int
maxTmp int
nDisks int
}
// rawQueueData mirrors the C xce_queue_t fields used by the Go model.
type rawQueueData struct {
name string
status int
priority int
}
// rawResourceData mirrors the C xce_resource_t fields used by the Go model.
type rawResourceData struct {
name string
desc string
valType int
ordType int
flags int
interval int
}
// rawJobToModel maps a rawJobData to a models.JobInfo.
// This pure-Go function contains the field-mapping logic and is tested without CGO.
func rawJobToModel(r rawJobData) *models.JobInfo {
info := &models.JobInfo{
Unique_JobID: r.jobID,
JobID: r.jobID,
JobStatus: r.state,
RunTime: r.runTime,
}
if r.submit != nil {
s := r.submit
info.QueueName = s.queue
info.SubmitUser = s.user
info.JobName = s.jobName
info.ProjectName = s.projectName
info.UserGroup = s.userGroup
info.JobDescription = s.jobDesc
info.ResReq = s.resReq
info.NumProcessors = s.minTasks
info.MaxNumProcessors = s.maxTasks
if s.startInfo != nil {
si := s.startInfo
info.Command = si.cmdline
info.LaunchType = si.launchType
info.PreExecCmd = si.preExec
info.PostExecCmd = si.postExec
info.ExecHome = si.execHome
info.ExecCWD = si.execCWD
info.ExecUser = si.execUser
}
}
if r.subTime > 0 {
info.SubmitTime = time.Unix(r.subTime, 0)
}
if r.queueTime > 0 {
info.QueueTime = time.Unix(r.queueTime, 0)
}
if r.startTime > 0 {
info.StartTime = time.Unix(r.startTime, 0)
}
if r.finishTime > 0 {
info.EndTime = time.Unix(r.finishTime, 0)
}
return info
}
// rawWorkerToModel maps a rawWorkerData to a models.HostInfo.
func rawWorkerToModel(r rawWorkerData) *models.HostInfo {
info := &models.HostInfo{
HostName: r.name,
HostStatus: r.status,
}
if r.hasAttrs {
info.HostType = r.hostType
info.HostModel = r.hostModel
info.CPUFactor = r.cpuFactor
info.NumCPUs = r.maxCPUs
info.MaxMem = r.maxMem
info.MaxSwap = r.maxSwap
info.MaxTmp = r.maxTmp
info.NumDisk = r.nDisks
}
return info
}
// rawQueueToModel maps a rawQueueData to a models.QueueInfo.
func rawQueueToModel(r rawQueueData) *models.QueueInfo {
return &models.QueueInfo{
QueueName: r.name,
QueueStatus: r.status,
Priority: r.priority,
}
}
// rawResourceToModel maps a rawResourceData to a models.ResourceDefinitionInfo.
func rawResourceToModel(r rawResourceData) *models.ResourceDefinitionInfo {
return &models.ResourceDefinitionInfo{
ResourceName: r.name,
Description: r.desc,
ValueType: r.valType,
OrderType: r.ordType,
Flag: r.flags,
Interval: r.interval,
}
}

View file

@ -0,0 +1,199 @@
package backend
import (
"testing"
"time"
)
func TestRawJobToModel_BasicFields(t *testing.T) {
r := rawJobData{
jobID: 123,
state: 2, // JOB_STAT_RUN
runTime: 42,
submit: &rawSubmitData{
queue: "normal",
user: "alice",
jobName: "myjob",
minTasks: 1,
maxTasks: 4,
startInfo: &rawStartInfo{
cmdline: "sleep 60",
},
},
}
info := rawJobToModel(r)
if info.Unique_JobID != 123 {
t.Errorf("Unique_JobID: got %d, want 123", info.Unique_JobID)
}
if info.JobID != 123 {
t.Errorf("JobID: got %d, want 123", info.JobID)
}
if info.QueueName != "normal" {
t.Errorf("QueueName: got %q, want %q", info.QueueName, "normal")
}
if info.SubmitUser != "alice" {
t.Errorf("SubmitUser: got %q, want %q", info.SubmitUser, "alice")
}
if info.JobName != "myjob" {
t.Errorf("JobName: got %q, want %q", info.JobName, "myjob")
}
if info.Command != "sleep 60" {
t.Errorf("Command: got %q, want %q", info.Command, "sleep 60")
}
if info.JobStatus != 2 {
t.Errorf("JobStatus: got %d, want 2", info.JobStatus)
}
if info.NumProcessors != 1 {
t.Errorf("NumProcessors: got %d, want 1", info.NumProcessors)
}
if info.MaxNumProcessors != 4 {
t.Errorf("MaxNumProcessors: got %d, want 4", info.MaxNumProcessors)
}
if info.RunTime != 42 {
t.Errorf("RunTime: got %d, want 42", info.RunTime)
}
}
func TestRawJobToModel_NilSubmit(t *testing.T) {
r := rawJobData{jobID: 99, state: 1 /* JOB_STAT_PEND */}
info := rawJobToModel(r)
if info.Unique_JobID != 99 {
t.Errorf("Unique_JobID: got %d, want 99", info.Unique_JobID)
}
if info.QueueName != "" {
t.Errorf("QueueName should be empty for nil submit, got %q", info.QueueName)
}
}
func TestRawJobToModel_Timestamps(t *testing.T) {
r := rawJobData{
jobID: 1,
subTime: 1700000000,
queueTime: 1700000010,
startTime: 1700000020,
finishTime: 1700000030,
}
info := rawJobToModel(r)
if info.SubmitTime != time.Unix(1700000000, 0) {
t.Errorf("SubmitTime mismatch")
}
if info.QueueTime != time.Unix(1700000010, 0) {
t.Errorf("QueueTime mismatch")
}
if info.StartTime != time.Unix(1700000020, 0) {
t.Errorf("StartTime mismatch")
}
if info.EndTime != time.Unix(1700000030, 0) {
t.Errorf("EndTime mismatch")
}
}
func TestRawWorkerToModel_BasicFields(t *testing.T) {
r := rawWorkerData{
name: "node01",
status: 3, // XCE_WORKER_UNAVAIL
hasAttrs: true,
hostType: "linux",
hostModel: "x86_64",
cpuFactor: 1.5,
maxCPUs: 16,
maxMem: 65536,
maxSwap: 32768,
maxTmp: 1024,
nDisks: 2,
}
info := rawWorkerToModel(r)
if info.HostName != "node01" {
t.Errorf("HostName: got %q, want %q", info.HostName, "node01")
}
if info.HostType != "linux" {
t.Errorf("HostType: got %q, want %q", info.HostType, "linux")
}
if info.HostModel != "x86_64" {
t.Errorf("HostModel: got %q, want %q", info.HostModel, "x86_64")
}
if info.NumCPUs != 16 {
t.Errorf("NumCPUs: got %d, want 16", info.NumCPUs)
}
if info.MaxMem != 65536 {
t.Errorf("MaxMem: got %d, want 65536", info.MaxMem)
}
if info.NumDisk != 2 {
t.Errorf("NumDisk: got %d, want 2", info.NumDisk)
}
if info.HostStatus != 3 {
t.Errorf("HostStatus: got %d, want 3", info.HostStatus)
}
}
func TestRawWorkerToModel_NoAttrs(t *testing.T) {
r := rawWorkerData{name: "node02", status: 1, hasAttrs: false}
info := rawWorkerToModel(r)
if info.HostName != "node02" {
t.Errorf("HostName: got %q, want node02", info.HostName)
}
if info.HostType != "" {
t.Errorf("HostType should be empty when hasAttrs=false, got %q", info.HostType)
}
if info.NumCPUs != 0 {
t.Errorf("NumCPUs should be 0 when hasAttrs=false, got %d", info.NumCPUs)
}
}
func TestRawQueueToModel_BasicFields(t *testing.T) {
r := rawQueueData{
name: "normal",
status: 1, // QUEUE_STAT_OPEN
priority: 30,
}
info := rawQueueToModel(r)
if info.QueueName != "normal" {
t.Errorf("QueueName: got %q, want %q", info.QueueName, "normal")
}
if info.Priority != 30 {
t.Errorf("Priority: got %d, want 30", info.Priority)
}
if info.QueueStatus != 1 {
t.Errorf("QueueStatus: got %d, want 1", info.QueueStatus)
}
}
func TestRawResourceToModel_BasicFields(t *testing.T) {
r := rawResourceData{
name: "mem",
desc: "memory resource",
valType: 1, // XCE_VAL_NUMERIC
flags: 4, // XCE_RESF_CONSUMABLE
interval: 60,
}
info := rawResourceToModel(r)
if info.ResourceName != "mem" {
t.Errorf("ResourceName: got %q, want %q", info.ResourceName, "mem")
}
if info.Description != "memory resource" {
t.Errorf("Description: got %q, want %q", info.Description, "memory resource")
}
if info.ValueType != 1 {
t.Errorf("ValueType: got %d, want 1", info.ValueType)
}
if info.Flag != 4 {
t.Errorf("Flag: got %d, want 4", info.Flag)
}
if info.Interval != 60 {
t.Errorf("Interval: got %d, want 60", info.Interval)
}
}

View file

@ -0,0 +1,38 @@
package backend
import (
"apiserver/models"
"apiserver/store"
)
// DefaultHandler implements NotificationHandler by writing to stores.
type DefaultHandler struct{}
func (h *DefaultHandler) OnJobCreated(job *models.JobInfo) {
store.Jobs.Put(uint64(job.Unique_JobID), *job)
}
func (h *DefaultHandler) OnJobUpdated(job *models.JobInfo) {
store.Jobs.Put(uint64(job.Unique_JobID), *job)
}
func (h *DefaultHandler) OnJobDeleted(jobid uint64) {
store.Jobs.Delete(jobid)
}
func (h *DefaultHandler) OnQueueUpdated(queue *models.QueueInfo) {
store.Queues.Put(queue.QueueName, *queue)
}
func (h *DefaultHandler) OnHostUpdated(host *models.HostInfo) {
store.Hosts.Put(host.HostName, *host)
}
func (h *DefaultHandler) OnResourceUpdated(res *models.ResourceDefinitionInfo) {
store.Resources.Put(res.ResourceName, *res)
}
func (h *DefaultHandler) OnResourceLocationUpdated(res *models.ResourceDistributionInfo) {
key := res.ResourceName + ":" + res.Location
store.ResourceLocations.Put(key, *res)
}

View file

@ -0,0 +1,16 @@
package backend
import "apiserver/models"
// NotificationHandler defines callbacks for backend push events.
// When the backend creates/updates/deletes objects, these callbacks
// are invoked to update the API server's in-memory stores.
type NotificationHandler interface {
OnJobCreated(job *models.JobInfo)
OnJobUpdated(job *models.JobInfo)
OnJobDeleted(jobid uint64)
OnQueueUpdated(queue *models.QueueInfo)
OnHostUpdated(host *models.HostInfo)
OnResourceUpdated(res *models.ResourceDefinitionInfo)
OnResourceLocationUpdated(res *models.ResourceDistributionInfo)
}

View file

@ -0,0 +1,31 @@
//go:build !cgo
package backend
import (
"apiserver/models"
"errors"
)
var errNoCGO = errors.New("backend requires CGO: rebuild with CGO_ENABLED=1 and libxce")
// RegisterToBackend stub — real implementation in clib.go (requires CGO).
func RegisterToBackend(config *BackendConfig) error { return errNoCGO }
// StartNotificationListener stub — real implementation in clib.go (requires CGO).
func StartNotificationListener(handler NotificationHandler) error { return errNoCGO }
// SubmitJob stub — real implementation in clib.go (requires CGO).
func SubmitJob(job *models.JobInfo) error { return errNoCGO }
// SubmitArrayJobs stub — real implementation in clib.go (requires CGO).
func SubmitArrayJobs(arrayjob *models.ArrayJobInfo, jobs []models.JobInfo) error { return errNoCGO }
// ControlJob stub — real implementation in clib.go (requires CGO).
func ControlJob(jobid uint64, action int, content string) error { return errNoCGO }
// ControlQueue stub — real implementation in clib.go (requires CGO).
func ControlQueue(queuename string, action int) error { return errNoCGO }
// ControlHost stub — real implementation in clib.go (requires CGO).
func ControlHost(hostname string, action int) error { return errNoCGO }