Render -X jobs' windows on the user's own X server with zero client install, mirroring the ssh -X experience: the job command starts with DISPLAY pointed at a per-job TCP listener on the worker, and the user's xcecli relays that listener's byte stream to their local X server. - xcecli: new -X/--x11 flag (interactive=2 + exdata "x11":1). After submit it fetches the per-job MIT-MAGIC-COOKIE-1, registers it in the user's Xauthority, then bridges one xattach websocket to the local X server address parsed from $DISPLAY (unix socket or TCP). The cookie entry is removed when the session ends. Pre-flight checks report missing DISPLAY/xauth before submit. - apiserver: GET /xce/v1/jobs/:id/xattach relays websocket binary frames to the worker's forwarding port (raw TCP, no ttyd). Close codes match the ttyd attach (4000 unreachable, 4001 ended). The job-finished event ends the relay. Teardown closes both hops as soon as either pump exits, so a client disconnect propagates instead of deadlocking in a blocking read. - worker: XJOB jobs with exdata x11=1 skip VNC, listen on 6100-6199, generate a per-job xauth file (/tmp/job_<id>_xauth, worker-side since the worker has no session), export DISPLAY/XAUTHORITY to the command, and bridge the app connection (loopback) with the relay connection (non-loopback; roles classified by source address, not arrival order). A 60 s no-client watchdog reaps jobs nothing ever attached to; the v1 single-connection boundary refuses a second X connection loudly. t1-verified per the plan matrix: window rendering (xclock/xeyes on Xvfb), normal end -> DONE + xauth removed, no-client 60 s reap (137), kill -> immediate teardown with no residue, -I/-I --browse regression, daemon health, missing-DISPLAY error, second-connection refusal. (--xjob VNC on t1 fails for lack of a vncserver binary; pre-existing.) Co-Authored-By: Claude Code <noreply@anthropic.com>
938 lines
33 KiB
Go
938 lines
33 KiB
Go
package backend
|
|
|
|
import (
|
|
"apiserver/models"
|
|
"apiserver/utils"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"regexp"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// jobCtrlDiscriminant maps a models.JOB_* action to the capnp JobControl union
|
|
// discriminant carried in xce_jobctrl_t.action: JobControl_u_siginfo (0) for
|
|
// signal-based actions (kill/stop/resume), JobControl_u_ckptoption (1) for
|
|
// checkpoint. The models.JOB_* constants are a DIFFERENT enum — using them as
|
|
// the discriminant made the controller's job_ctrl() select the wrong union
|
|
// branch (or none) and reject every control request with -EINVAL (bug-1490).
|
|
// Returns -1 for an unknown action (clib.go never calls ControlJob with one).
|
|
func jobCtrlDiscriminant(action int) int {
|
|
switch action {
|
|
case models.JOB_STOP, models.JOB_RESUME, models.JOB_KILL:
|
|
return 0 // JobControl_u_siginfo
|
|
case models.JOB_CHKPNT:
|
|
return 1 // JobControl_u_ckptoption
|
|
}
|
|
return -1
|
|
}
|
|
|
|
// JobID bit layout constants — mirrors src/include/config.h.
|
|
//
|
|
// |63 62|61 56|55 52|51 20|19 15|14 0|
|
|
// | TAG | CLUSTER_ID | APISERVER_ID| TIMESTAMP | SEQ_HIGH | ARRAY_IDX |
|
|
// | 2 bits | 6 bits | 4 bits | 32 bits | 5 bits | 15 bits |
|
|
const (
|
|
nTagBits = 2
|
|
tagShift = 62
|
|
nClusterIDBits = 6
|
|
clusterIDShift = 56
|
|
nApiserverIDBits = 4
|
|
apiserverIDShift = 52
|
|
nTimestampBits = 32
|
|
timestampShift = 20
|
|
nArrayIdxBits = 15
|
|
sequenceHighShift = 15 // mirrors SEQUENCE_HIGH_SHIFT in config.h (== N_ARRAY_IDX_BITS)
|
|
nSequenceBits = 20 // N_JOBID_BITS(64) - N_CLUSTER_ID_BITS(6) - N_APISERVER_ID_BITS(4) - N_TIMESTAMP_BITS(32) - N_TAG_BITS(2) = 20
|
|
tagJobArray = 0x01 // mirrors TAG_JOB_ARRAY in config.h
|
|
)
|
|
|
|
// extractBits extracts a bit field from id at the given shift position.
|
|
// Mirrors EXTRACT_BITS(id, shift, bits) in src/include/utils.h.
|
|
func extractBits(id uint64, shift, bits int) uint64 {
|
|
mask := uint64((1 << bits) - 1)
|
|
return (id >> shift) & mask
|
|
}
|
|
|
|
// extractIDTag extracts the 2-bit tag field (bits 62-63).
|
|
// Mirrors EXTRACT_ID_TAG in src/include/utils.h.
|
|
func extractIDTag(id uint64) uint64 {
|
|
return extractBits(id, tagShift, nTagBits)
|
|
}
|
|
|
|
// ExtractArrayBaseID extracts the base (family) job ID from a 64-bit encoded
|
|
// jobid. Mirrors C's array_jobid() in src/lib/utils.c and EXTRACT_ARRAY_BASEID
|
|
// in src/include/utils.h.
|
|
// - For non-array jobs (tag=0x00): returns id unchanged.
|
|
// - For array children (tag=0x01): returns the shared base ID (tag=0, idx=0).
|
|
func ExtractArrayBaseID(id uint64) uint64 {
|
|
if extractIDTag(id) == 0x00 {
|
|
return id
|
|
}
|
|
clusterID := extractBits(id, clusterIDShift, nClusterIDBits)
|
|
apiserverID := extractBits(id, apiserverIDShift, nApiserverIDBits)
|
|
timestamp := extractBits(id, timestampShift, nTimestampBits)
|
|
// EXTRACT_ARRAY_JOB: high bits of sequence (mirrors config.h SEQUENCE_HIGH_SHIFT)
|
|
arrayJob := extractBits(id, sequenceHighShift, nSequenceBits-nArrayIdxBits) << sequenceHighShift
|
|
return (clusterID << clusterIDShift) |
|
|
(apiserverID << apiserverIDShift) |
|
|
(timestamp << timestampShift) |
|
|
arrayJob // tag=0, idx=0
|
|
}
|
|
|
|
// ExtractArrayIdx extracts the array index from a 64-bit encoded jobid.
|
|
// Mirrors C's array_idx() in src/lib/utils.c and EXTRACT_ARRAY_IDX in
|
|
// src/include/utils.h.
|
|
// Returns 0 for non-array jobs (tag=0x00).
|
|
func ExtractArrayIdx(id uint64) int {
|
|
if extractIDTag(id) == 0x00 {
|
|
return 0
|
|
}
|
|
return int(extractBits(id, 0, nArrayIdxBits))
|
|
}
|
|
|
|
// IsArrayJob reports whether the 64-bit encoded jobid belongs to an array job
|
|
// (tag == TAG_JOB_ARRAY, 0x01). Returns false for non-array jobs (tag=0x00).
|
|
func IsArrayJob(id uint64) bool {
|
|
return extractIDTag(id) == tagJobArray
|
|
}
|
|
|
|
// FormatJobID formats a packed 64-bit job ID as a human-readable string,
|
|
// matching the C fmt_jobid() output in src/lib/utils.c.
|
|
// - Regular job: "{cluster}-{apiserver}-{timestamp}-{sequence}"
|
|
// - Array child: "{cluster}-{apiserver}-{timestamp}-{base}[{idx}]"
|
|
func FormatJobID(uniqueJobID int64) string {
|
|
id := uint64(uniqueJobID)
|
|
clusterID := extractBits(id, clusterIDShift, nClusterIDBits)
|
|
apiserverID := extractBits(id, apiserverIDShift, nApiserverIDBits)
|
|
timestamp := extractBits(id, timestampShift, nTimestampBits)
|
|
|
|
if extractIDTag(id) == tagJobArray {
|
|
arrayBase := extractBits(id, sequenceHighShift, nSequenceBits-nArrayIdxBits) << uint(sequenceHighShift)
|
|
arrayIdx := extractBits(id, 0, nArrayIdxBits)
|
|
return fmt.Sprintf("%d-%d-%d-%d[%d]",
|
|
clusterID, apiserverID, timestamp, arrayBase, arrayIdx)
|
|
}
|
|
sequence := extractBits(id, 0, nSequenceBits)
|
|
return fmt.Sprintf("%d-%d-%d-%d",
|
|
clusterID, apiserverID, timestamp, sequence)
|
|
}
|
|
|
|
// humanJobIDPattern matches a plain human-readable job ID, "{cluster}-
|
|
// {apiserver}-{timestamp}-{sequence}" (no array-index brackets).
|
|
var humanJobIDPattern = regexp.MustCompile(`^(\d+)-(\d+)-(\d+)-(\d+)$`)
|
|
|
|
// ParseHumanJobID parses a plain human-readable job ID (as produced by
|
|
// FormatJobID, brackets already stripped by the caller) back into the
|
|
// packed numeric job ID the job_id/unique_job_id DB columns store. Returns
|
|
// ok=false if s doesn't match that shape (e.g. it's already numeric).
|
|
//
|
|
// Deliberately does not itself handle an "[idx]" array-child suffix:
|
|
// controllers/jobs.go's parseArrayName already splits that off (for the
|
|
// API's own array-index filter syntax) before this ever runs, leaving a
|
|
// plain quad whose 4th number is the packed ID's low sequence bits with
|
|
// tag=0 -- true for a real non-array job's own ID, and also true for an
|
|
// array child's dash-quad prefix (FormatJobID's arrayBase is those same
|
|
// bits already shifted into position with tag/idx zeroed, i.e. exactly
|
|
// ExtractArrayBaseID(original)).
|
|
func ParseHumanJobID(s string) (uint64, bool) {
|
|
m := humanJobIDPattern.FindStringSubmatch(s)
|
|
if m == nil {
|
|
return 0, false
|
|
}
|
|
clusterID, _ := strconv.ParseUint(m[1], 10, 64)
|
|
apiserverID, _ := strconv.ParseUint(m[2], 10, 64)
|
|
timestamp, _ := strconv.ParseUint(m[3], 10, 64)
|
|
sequence, _ := strconv.ParseUint(m[4], 10, 64)
|
|
return (clusterID << clusterIDShift) |
|
|
(apiserverID << apiserverIDShift) |
|
|
(timestamp << timestampShift) |
|
|
sequence, true
|
|
}
|
|
|
|
// Errors returned by ParseHumanJobIDStrict. Callers map them to their own
|
|
// messages: ErrJobIDSyntax is the written-form syntax rule, ErrJobIDWidth the
|
|
// field-width rule (job-array-one-based-index 5.14).
|
|
var (
|
|
ErrJobIDSyntax = errors.New("job ID is not of the form cluster-apiserver-timestamp-sequence")
|
|
ErrJobIDWidth = errors.New("job ID field out of range")
|
|
)
|
|
|
|
// ParseHumanJobIDStrict parses a written job identity "c-a-ts-seq" (decimal
|
|
// digits only, leading zeros allowed, no sign, no whitespace, no "[...]")
|
|
// into its packed, untagged ID. Unlike ParseHumanJobID it checks every field
|
|
// against its bit width -- cluster <= 63, apiserver <= 15, timestamp < 2^32,
|
|
// sequence < 2^20 -- so an oversized field is rejected (ErrJobIDWidth)
|
|
// instead of spilling into a neighbouring field. A field too large even for
|
|
// uint64 is also ErrJobIDWidth. Anything not of that shape is ErrJobIDSyntax.
|
|
func ParseHumanJobIDStrict(s string) (uint64, error) {
|
|
m := humanJobIDPattern.FindStringSubmatch(s)
|
|
if m == nil {
|
|
return 0, ErrJobIDSyntax
|
|
}
|
|
widths := [4]int{nClusterIDBits, nApiserverIDBits, nTimestampBits, nSequenceBits}
|
|
var f [4]uint64
|
|
for i, w := range widths {
|
|
v, err := strconv.ParseUint(m[i+1], 10, 64)
|
|
if err != nil {
|
|
if errors.Is(err, strconv.ErrRange) {
|
|
return 0, ErrJobIDWidth
|
|
}
|
|
return 0, ErrJobIDSyntax
|
|
}
|
|
if v >= uint64(1)<<w {
|
|
return 0, ErrJobIDWidth
|
|
}
|
|
f[i] = v
|
|
}
|
|
return (f[0] << clusterIDShift) |
|
|
(f[1] << apiserverIDShift) |
|
|
(f[2] << timestampShift) |
|
|
f[3], nil
|
|
}
|
|
|
|
// arrayIdxMask covers the 15-bit array index field (bits 0-14).
|
|
const arrayIdxMask = uint64(1)<<nArrayIdxBits - 1
|
|
|
|
// IsArrayCapableKey reports whether key can name an array (an "<array-id>"
|
|
// in "<array-id>[k]"): tag 0 and the low 15 (index) bits clear, i.e. the
|
|
// sequence is a multiple of 32768 (job-array-one-based-index 5.8).
|
|
func IsArrayCapableKey(key uint64) bool {
|
|
return extractIDTag(key) == 0 && key&arrayIdxMask == 0
|
|
}
|
|
|
|
// MakeElementID composes the ID of element k of the array identified by key:
|
|
// key's cluster, apiserver, timestamp and sequence-high bits, tag
|
|
// TAG_JOB_ARRAY and index k (masked to 15 bits). key's own tag and index bits
|
|
// are ignored. Mirrors jobsel_element_id() in src/lib/utils.c.
|
|
func MakeElementID(key uint64, k int) uint64 {
|
|
base := key &^ (uint64(1<<nTagBits-1)<<tagShift | arrayIdxMask)
|
|
return uint64(tagJobArray)<<tagShift | base | uint64(k)&arrayIdxMask
|
|
}
|
|
|
|
// WorkerDefaultOutputPath returns the output file path the worker uses by default
|
|
// for the given job ID and task index, matching uv_proc.c make_job_output_path().
|
|
func WorkerDefaultOutputPath(uniqueJobID int64, taskIdx int) string {
|
|
return fmt.Sprintf("/tmp/job_%s_task%d.out", FormatJobID(uniqueJobID), taskIdx)
|
|
}
|
|
|
|
// jobCtrlUnionDiscriminant returns the value xce_jobctrl_t.action must carry
|
|
// for the given job_ctrl_type_t action (models.JOB_STOP/RESUME/KILL/CHKPNT):
|
|
// the capnp JobControl_u_which union discriminant (JobControl_u_siginfo=0,
|
|
// JobControl_u_ckptoption=1 -- src/lib/controller.capnp.h), NOT the action
|
|
// code itself. xce_jobctrl_req() (src/lib/request.c) switches on this field
|
|
// and rejects anything else with -EINVAL; ControlJob (clib.go) previously
|
|
// stored the raw action code there instead, so kill(3)/resume(2) always hit
|
|
// that -EINVAL default case (gitea issue #45) while stop(1) happened to
|
|
// collide with ckptoption(1) and read through the wrong union member.
|
|
func jobCtrlUnionDiscriminant(action int) int {
|
|
if action == models.JOB_CHKPNT {
|
|
return 1 // JobControl_u_ckptoption
|
|
}
|
|
return 0 // JobControl_u_siginfo (stop/resume/kill all signal the job)
|
|
}
|
|
|
|
// jobCtrlArgs is what ControlJob (clib.go) puts into xce_jobctrl_t's union
|
|
// branch: Signo/Force fill xce_signal_t (stop/resume/kill), CkptKill/Method
|
|
// fill xce_ckpt_options_t (checkpoint).
|
|
type jobCtrlArgs struct {
|
|
Signo int
|
|
Force int
|
|
Grace int // seconds; > 0: graceful kill/stop sequence (bug-1957)
|
|
CkptKill int
|
|
Method string
|
|
}
|
|
|
|
// jobCtrlParams maps a job action and ControlJob's content argument to the
|
|
// values sent to the controller. content is either the JSON-encoded
|
|
// models.JobControlEvent the job control handlers pass (controllers/jobs.go)
|
|
// or, for older callers, a bare signal name or number; empty means the
|
|
// action's default signal.
|
|
//
|
|
// bug-1940: content used to be looked up only as a signal name, so the JSON
|
|
// event always missed and every kill sent TERM with force 0, and checkpoint
|
|
// kill/method were never set. An unknown signal is now an error instead of a
|
|
// silent fallback to the default.
|
|
func jobCtrlParams(action int, content string) (jobCtrlArgs, error) {
|
|
var (
|
|
args jobCtrlArgs
|
|
signo int
|
|
forceKill bool
|
|
)
|
|
|
|
content = strings.TrimSpace(content)
|
|
if strings.HasPrefix(content, "{") {
|
|
var evt models.JobControlEvent
|
|
|
|
if err := json.Unmarshal([]byte(content), &evt); err != nil {
|
|
return args, fmt.Errorf("invalid job control content: %w", err)
|
|
}
|
|
if evt.Signal != 0 && !isKnownSignal(evt.Signal) {
|
|
return args, fmt.Errorf("unsupported signal: %d", evt.Signal)
|
|
}
|
|
signo, forceKill, args.Method = evt.Signal, evt.ForceKill, evt.Method
|
|
} else if content != "" {
|
|
n, err := parseSignal(content)
|
|
if err != nil {
|
|
return args, err
|
|
}
|
|
signo = n
|
|
}
|
|
|
|
switch action {
|
|
case models.JOB_STOP, models.JOB_RESUME, models.JOB_KILL:
|
|
if signo == 0 {
|
|
signo = defaultActionSignal(action)
|
|
// bug-1957: a plain kill or stop is graceful, as LSF's
|
|
// TERMINATE and SUSPEND actions; a force kill is immediate
|
|
if action == models.JOB_STOP || (action == models.JOB_KILL && !forceKill) {
|
|
args.Grace = jobTerminateInterval()
|
|
}
|
|
}
|
|
args.Signo = signo
|
|
if action == models.JOB_KILL && forceKill {
|
|
args.Force = 1
|
|
}
|
|
case models.JOB_CHKPNT:
|
|
if forceKill {
|
|
args.CkptKill = 1
|
|
}
|
|
}
|
|
return args, nil
|
|
}
|
|
|
|
// defaultActionSignal is the signal a stop/resume/kill sends when the caller
|
|
// gave none.
|
|
//
|
|
// bug-1957: kill and stop name the sequence's last signal (SIGKILL,
|
|
// SIGSTOP); with a grace period the worker sends SIGINT/SIGTERM or SIGTSTP
|
|
// first, so the job can clean up or prepare to stop.
|
|
func defaultActionSignal(action int) int {
|
|
switch action {
|
|
case models.JOB_STOP:
|
|
return models.JobSignalMapTable["STOP"] // 19
|
|
case models.JOB_RESUME:
|
|
return models.JobSignalMapTable["CONT"] // 18
|
|
}
|
|
return models.JobSignalMapTable["KILL"] // 9
|
|
}
|
|
|
|
// defaultJobTerminateInterval is LSF's JOB_TERMINATE_INTERVAL default.
|
|
const defaultJobTerminateInterval = 10
|
|
|
|
// jobTerminateInterval is job.terminate_interval, or the default when it is
|
|
// unset or 0.
|
|
func jobTerminateInterval() int {
|
|
if utils.JobYamlConfig != nil && utils.JobYamlConfig.TerminateInterval > 0 {
|
|
return utils.JobYamlConfig.TerminateInterval
|
|
}
|
|
return defaultJobTerminateInterval
|
|
}
|
|
|
|
// isKnownSignal reports whether n is one of models.JobSignalMapTable's values.
|
|
func isKnownSignal(n int) bool {
|
|
for _, v := range models.JobSignalMapTable {
|
|
if v == n {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// parseSignal accepts a signal number or name (case-insensitive, with or
|
|
// without the SIG prefix) and returns its number.
|
|
func parseSignal(s string) (int, error) {
|
|
if n, err := strconv.Atoi(s); err == nil {
|
|
if !isKnownSignal(n) {
|
|
return 0, fmt.Errorf("unsupported signal: %s", s)
|
|
}
|
|
return n, nil
|
|
}
|
|
name := strings.TrimPrefix(strings.ToUpper(s), "SIG")
|
|
if n, ok := models.JobSignalMapTable[name]; ok {
|
|
return n, nil
|
|
}
|
|
return 0, fmt.Errorf("unsupported signal: %s", s)
|
|
}
|
|
|
|
// resolveExecUser returns the effective exec user for a job submission: the
|
|
// caller-specified ExecUser if present, otherwise the submitting user
|
|
// (bug-1780: an unset execuser in the startjob request reaches xcecd as
|
|
// empty, which controller_connector.c defaults to the literal Linux user
|
|
// 'nobody' -- a user with no home/shell that then fails to exec the job).
|
|
func resolveExecUser(execUser, submitUser string) string {
|
|
if execUser != "" {
|
|
return execUser
|
|
}
|
|
return submitUser
|
|
}
|
|
|
|
// 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
|
|
exdata string
|
|
}
|
|
|
|
// submitStartInfo maps a Go JobInfo to the job_startinfo_t fields a
|
|
// submission carries to the controller (and from there, unchanged, to the
|
|
// worker's startjob request). modelJobToSubmit (convert.go) is the CGO
|
|
// adapter that copies the result into the C struct.
|
|
//
|
|
// bug-1936: exdata (env, limits, transfers, interactive mode, built by the
|
|
// submit handler into ExtendData), the exec cwd and the exec home were
|
|
// never set, so none of them reached the worker.
|
|
func submitStartInfo(job *models.JobInfo) rawStartInfo {
|
|
execCWD := job.ExecCWD
|
|
if execCWD == "" {
|
|
execCWD = job.CWD
|
|
}
|
|
return rawStartInfo{
|
|
cmdline: job.Command,
|
|
launchType: job.LaunchType,
|
|
preExec: job.PreExecCmd,
|
|
postExec: job.PostExecCmd,
|
|
execHome: job.ExecHome,
|
|
execCWD: execCWD,
|
|
execUser: resolveExecUser(job.ExecUser, job.SubmitUser),
|
|
exdata: job.ExtendData.String,
|
|
}
|
|
}
|
|
|
|
// maxExdataLen is the longest exdata the worker can take: xcecd hands it to
|
|
// xcecd_worker in job_assignment_t.exdata[EXDATA_LENGTH] (16 KiB,
|
|
// src/worker/include/uv_ipc.h), NUL included.
|
|
const maxExdataLen = 16*1024 - 1
|
|
|
|
// ValidateExdata checks a job's extended data before it is submitted: the
|
|
// worker's parse_exdata() fails the job at launch on anything but a JSON
|
|
// object, and a longer string than maxExdataLen would be cut short on its
|
|
// way to the job process.
|
|
func ValidateExdata(exdata string) error {
|
|
if exdata == "" {
|
|
return nil
|
|
}
|
|
if len(exdata) > maxExdataLen {
|
|
return fmt.Errorf("extended job data is %d bytes, more than the limit of %d", len(exdata), maxExdataLen)
|
|
}
|
|
var obj map[string]interface{}
|
|
if err := json.Unmarshal([]byte(exdata), &obj); err != nil {
|
|
return errors.New("extended job data must be a JSON object")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// 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
|
|
fromHost string
|
|
}
|
|
|
|
// rawHostReason mirrors the C xce_host_reason_t (one host-level PEND reason).
|
|
type rawHostReason struct {
|
|
Host string `json:"host"`
|
|
Reason int `json:"reason"`
|
|
ReasonName string `json:"reason_name"`
|
|
}
|
|
|
|
// rawResReason mirrors the C xce_res_reason_t (one resource-level PEND reason).
|
|
type rawResReason struct {
|
|
Resource string `json:"resource"`
|
|
Host string `json:"host"`
|
|
Reason int `json:"reason"`
|
|
ReasonName string `json:"reason_name"`
|
|
}
|
|
|
|
// rawPendReason mirrors the C xce_job_preason_t attached to a PEND job by the
|
|
// controller (scheduler → setpreason → job->pend_reasons). JSON-tagged so the
|
|
// marshaled value lands in models.JobInfo.PendingReason as a JSONB object —
|
|
// processJobData's JSONBJFields pass then surfaces it in the API response.
|
|
type rawPendReason struct {
|
|
MainReason int `json:"main_reason"`
|
|
MainReasonName string `json:"main_reason_name"`
|
|
HostReasons []rawHostReason `json:"host_reasons,omitempty"`
|
|
ResReasons []rawResReason `json:"res_reasons,omitempty"`
|
|
}
|
|
|
|
// pendReasonName maps a PREASON_* main_reason code to a readable name. The
|
|
// codes are the PEND reason bitfield of xce_job_preason_t (src/include/
|
|
// preason.h). Unknown codes are reported verbatim so a new scheduler reason
|
|
// never silently disappears.
|
|
func pendReasonName(code int) string {
|
|
switch code {
|
|
case 1: // PREASON_BAD_RESREQ
|
|
return "bad_resreq"
|
|
case 2: // PREASON_NO_RESOURCE
|
|
return "no_resource"
|
|
case 3: // PREASON_MATCH_FAIL
|
|
return "match_fail"
|
|
case 4: // PREASON_LIMIT_GLOBAL
|
|
return "limit_global"
|
|
case 5: // PREASON_LIMIT_HOST
|
|
return "limit_host"
|
|
case 6: // PREASON_QUEUE_CLOSED
|
|
return "queue_closed"
|
|
case 7: // PREASON_QUEUE_DRAINING
|
|
return "queue_draining"
|
|
case 8: // PREASON_TIME_WINDOW
|
|
return "time_window"
|
|
case 9: // PREASON_HOST_UNAVAIL
|
|
return "host_unavail"
|
|
default:
|
|
return fmt.Sprintf("unknown(%d)", code)
|
|
}
|
|
}
|
|
|
|
// rawHostSlot mirrors one xce_host_alloc_t's host name plus its "slots"
|
|
// resource allocation count (xce_host_alloc_t.allocs[], scanned for
|
|
// resname=="slots" -- mirrors the C scheduler's own get_nslots_in_alloc()/
|
|
// repl_worker_alloc_nslots() convention, defaulting to 1 when no explicit
|
|
// slots entry is present).
|
|
type rawHostSlot struct {
|
|
host string
|
|
slots int
|
|
}
|
|
|
|
// rawRunUsage mirrors the C xce_job_runusage_t job-level fields used by the
|
|
// Go model (bug-1736): accumulated CPU time and memory, and the CPU peak and
|
|
// efficiency values the controller computes with LSF bjobs -l definitions
|
|
// (efficiencies in percent, memory in MB as the worker reports it).
|
|
type rawRunUsage struct {
|
|
maxMem float64
|
|
avgMem float64
|
|
usedMem float64
|
|
cpuTime int
|
|
peakCPUTime int
|
|
peakCPUDuration int
|
|
memEfficiency float64
|
|
avgCPUEfficiency float64
|
|
peakCPUEfficiency float64
|
|
peakCPU float64
|
|
}
|
|
|
|
// rawPidInfo mirrors the C job_pidinfo_t fields used by the Go model —
|
|
// the worker-reported process bookkeeping a job start carries. Exdata is
|
|
// the worker's start_exdata JSON; for interactive jobs (TTY/XJOB/DISPLAY,
|
|
// xce_job_type in src/include/xce/types.h) the worker packs the session
|
|
// URL into it as {"url": ...} (generate_exdata, xcecd_exdata.c).
|
|
type rawPidInfo struct {
|
|
pids []int
|
|
pgids []int
|
|
exdata string
|
|
}
|
|
|
|
// 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
|
|
execHosts []rawHostSlot
|
|
pendReason *rawPendReason
|
|
runUsage *rawRunUsage // nil until the job's first usage report
|
|
pidInfo *rawPidInfo // nil until the worker's start report
|
|
}
|
|
|
|
// 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
|
|
resvals []models.ResourceDistributionInfo
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// interactiveURLFromExdata extracts the interactive session URL from a job
|
|
// start's exdata JSON. The worker packs it as {"url": ...}
|
|
// (generate_exdata, xcecd_exdata.c); anything else — non-JSON, other keys,
|
|
// an empty/non-string url — yields "". The URL reaches the apiserver ONLY
|
|
// via this path: job_pidinfo_t.exdata rides the capnp Job wire (pidinfo@9)
|
|
// but was previously dropped by the conversion, leaving the API's
|
|
// interactive_job_url permanently empty and xcecli's openBrowser flow dead.
|
|
func interactiveURLFromExdata(exdata string) string {
|
|
if exdata == "" {
|
|
return ""
|
|
}
|
|
var obj struct {
|
|
URL string `json:"url"`
|
|
}
|
|
if err := json.Unmarshal([]byte(exdata), &obj); err != nil {
|
|
return ""
|
|
}
|
|
return obj.URL
|
|
}
|
|
|
|
// x11CookieFromExdata extracts the per-job X11 auth cookie of an -X job
|
|
// from the same start-report exdata (the worker packs it as
|
|
// {"x11cookie": ...} next to the x11:// URL). The xcecli -X client needs it
|
|
// to register the session's MIT-MAGIC-COOKIE-1 with the user's Xauthority
|
|
// before the app's first X connection is relayed through.
|
|
func x11CookieFromExdata(exdata string) string {
|
|
if exdata == "" {
|
|
return ""
|
|
}
|
|
var obj struct {
|
|
Cookie string `json:"x11cookie"`
|
|
}
|
|
if err := json.Unmarshal([]byte(exdata), &obj); err != nil {
|
|
return ""
|
|
}
|
|
return obj.Cookie
|
|
}
|
|
|
|
// pidInfoJSONB marshals a start report's pids/pgids into the PIDInfo Jsonb
|
|
// the API exposes (processJobData parses the JSONB string before the row
|
|
// leaves the API, so clients see an object with pids/pgids). Shared by
|
|
// rawJobToModel (the xce_job_t conversion, which carries the full
|
|
// start report) and OnJobPidInfo (the standalone jobPidInfo notify, whose
|
|
// only extra payload is the interactive URL).
|
|
func pidInfoJSONB(pids, pgids []int) (models.Jsonb, bool) {
|
|
data, err := json.Marshal(struct {
|
|
Pids []int `json:"pids,omitempty"`
|
|
Pgids []int `json:"pgids,omitempty"`
|
|
}{Pids: pids, Pgids: pgids})
|
|
if err != nil {
|
|
return models.Jsonb{}, false
|
|
}
|
|
return models.Jsonb{String: string(data)}, true
|
|
}
|
|
|
|
// 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 {
|
|
raw := uint64(r.jobID)
|
|
info := &models.JobInfo{
|
|
Unique_JobID: r.jobID,
|
|
JobID: int64(ExtractArrayBaseID(raw)),
|
|
ArrayIdx: ExtractArrayIdx(raw),
|
|
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
|
|
info.FromHost = s.fromHost
|
|
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.CWD = si.execCWD // the submit cwd is sent as the exec cwd (bug-1936)
|
|
info.ExecUser = si.execUser
|
|
info.ExtendData = models.Jsonb{String: si.exdata}
|
|
}
|
|
}
|
|
info.SubmitTime = r.subTime
|
|
info.QueueTime = r.queueTime
|
|
info.StartTime = r.startTime
|
|
info.EndTime = r.finishTime
|
|
if len(r.execHosts) > 0 {
|
|
type hostSlot struct {
|
|
HostName string `json:"host_name"`
|
|
NumSlots int `json:"num_slots"`
|
|
}
|
|
type execHostsJSON struct {
|
|
HostSlots []hostSlot `json:"host_slots"`
|
|
NumHosts int `json:"num_hosts"`
|
|
}
|
|
slots := make([]hostSlot, len(r.execHosts))
|
|
for i, h := range r.execHosts {
|
|
slots[i] = hostSlot{HostName: h.host, NumSlots: h.slots}
|
|
}
|
|
data, _ := json.Marshal(execHostsJSON{HostSlots: slots, NumHosts: len(slots)})
|
|
info.ExecHosts = models.Jsonb{String: string(data)}
|
|
}
|
|
if r.pendReason != nil {
|
|
// main_reason is only set by an explicit add_job_preason(reasons,
|
|
// code, NULL, NULL) (which erases the detail lists); the allocator's
|
|
// ordinary shortfalls attach per-host/per-res DETAIL rows and leave
|
|
// main_reason 0. A detail-only record promotes its first detail row
|
|
// to main_reason_name here — the API-visible "overall reason"
|
|
// (pend-reason P0: main_reason_name used to read unknown(0)) — while
|
|
// the C record's main_reason value itself marshals unchanged (the
|
|
// scheduler-framework trace/verify stack counts main_reason as its
|
|
// own occurrence and its committed .replay baselines encode
|
|
// main_reason=0 detail-only PENDs, so the promotion stays an
|
|
// API-layer view).
|
|
if r.pendReason.MainReasonName == "" && r.pendReason.MainReason != 0 {
|
|
r.pendReason.MainReasonName = pendReasonName(r.pendReason.MainReason)
|
|
} else if r.pendReason.MainReasonName == "" && r.pendReason.MainReason == 0 {
|
|
for _, h := range r.pendReason.HostReasons {
|
|
if h.Reason != 0 {
|
|
r.pendReason.MainReasonName = pendReasonName(h.Reason)
|
|
break
|
|
}
|
|
}
|
|
if r.pendReason.MainReasonName == "" {
|
|
for _, rr := range r.pendReason.ResReasons {
|
|
if rr.Reason != 0 {
|
|
r.pendReason.MainReasonName = pendReasonName(rr.Reason)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if data, err := json.Marshal(r.pendReason); err == nil {
|
|
info.PendingReason = models.Jsonb{String: string(data)}
|
|
}
|
|
}
|
|
if u := r.runUsage; u != nil {
|
|
info.MaxMem = u.maxMem
|
|
info.AvgMem = u.avgMem
|
|
info.UsedMem = u.usedMem
|
|
info.CPUTime = u.cpuTime
|
|
info.PeakCPU = u.peakCPU
|
|
info.PeakCPUTime = u.peakCPUTime
|
|
info.PeakCPUDuration = float64(u.peakCPUDuration)
|
|
info.MemEfficiency = u.memEfficiency
|
|
info.AvgCPUEfficiency = u.avgCPUEfficiency
|
|
info.PeakCPUEfficiency = u.peakCPUEfficiency
|
|
}
|
|
if p := r.pidInfo; p != nil {
|
|
// pid_info is a JSONBJFields entry: processJobData parses the JSONB
|
|
// string before the row leaves the API, so clients see an object
|
|
// with pids/pgids. The interactive URL is promoted to its own
|
|
// top-level field — it is what xcecli's waitForInteractiveURL and
|
|
// the --browse flow consume. An -X job's auth cookie rides the same
|
|
// report and gets the same treatment.
|
|
if jb, ok := pidInfoJSONB(p.pids, p.pgids); ok {
|
|
info.PIDInfo = jb
|
|
}
|
|
info.InteractiveJobURL = interactiveURLFromExdata(p.exdata)
|
|
info.X11Cookie = x11CookieFromExdata(p.exdata)
|
|
}
|
|
return info
|
|
}
|
|
|
|
// rawJobArrayToModels expands one compact (base jobid, array_size, one
|
|
// shared spec) array descriptor into N models.JobInfo entries locally, one
|
|
// per index in [0, array_size), using base_jobid + i arithmetic — the same
|
|
// arithmetic the controller's restore_job_array() (job.c) and the
|
|
// scheduler's sched_on_job_array() (handler.c) use to derive each
|
|
// element's identity (job-array-single-wal-record task 5.4, Req 6.2, 6.5).
|
|
// A freshly-received array element mirrors sched_on_job_array's
|
|
// make_array_element: JOB_STAT_PEND, receive-time sub_time — the compact
|
|
// wire descriptor carries no per-element timestamps to recover.
|
|
// Returns nil for a zero array_size or a nil submit (nothing to expand).
|
|
// Pure Go (no CGO) — cJobArrayToModels (convert.go) is the CGO adapter
|
|
// that decodes a *C.xce_job_submit_t into the rawSubmitData this consumes.
|
|
func rawJobArrayToModels(baseJobID uint64, arraySize uint64, submit *rawSubmitData) []*models.JobInfo {
|
|
if arraySize == 0 || submit == nil {
|
|
return nil
|
|
}
|
|
|
|
now := time.Now().Unix()
|
|
out := make([]*models.JobInfo, 0, arraySize)
|
|
for i := uint64(0); i < arraySize; i++ {
|
|
r := rawJobData{
|
|
jobID: int64(baseJobID + i),
|
|
state: models.JOBSTAT["PEND"],
|
|
subTime: now,
|
|
submit: submit,
|
|
}
|
|
out = append(out, rawJobToModel(r))
|
|
}
|
|
return out
|
|
}
|
|
|
|
// rawWorkerToModel maps a rawWorkerData to a models.HostInfo.
|
|
func rawWorkerToModel(r rawWorkerData) *models.HostInfo {
|
|
info := &models.HostInfo{
|
|
HostName: r.name,
|
|
HostStatus: r.status,
|
|
ResourceValues: r.resvals,
|
|
}
|
|
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,
|
|
}
|
|
}
|
|
|
|
// rawHostConfigData mirrors the C xce_hostconfig_t fields used by the Go model.
|
|
// Only fields that map to models.HostInfo are included:
|
|
// - name → HostName, hostmodel → HostModel, hosttype → HostType, mxj → MaxSlot
|
|
//
|
|
// resources (xce_config_t**) and thresholds (xce_res_threshold_t**) have no HostInfo field and are excluded.
|
|
// dispatchwindow (xce_timewindow_config_t*) is deferred — DisptWindow is left as empty string.
|
|
type rawHostConfigData struct {
|
|
name string
|
|
hostmodel string
|
|
hosttype string
|
|
mxj int
|
|
}
|
|
|
|
// rawHostConfigToModel maps a rawHostConfigData to a models.HostInfo.
|
|
// Config-sourced fields only; worker-sourced fields remain zero.
|
|
func rawHostConfigToModel(r rawHostConfigData) *models.HostInfo {
|
|
return &models.HostInfo{
|
|
HostName: r.name,
|
|
HostModel: r.hostmodel,
|
|
HostType: r.hosttype,
|
|
MaxSlot: r.mxj,
|
|
// TODO(dispatchwindow): xce_hostconfig_t.dispatchwindow is xce_timewindow_config_t*
|
|
// (nsegs int + segs[] with open/close time fields). Serializing it requires a Go
|
|
// mirror struct and json.Marshal. Deferred: no API endpoint currently consumes
|
|
// DisptWindow. Implement when /xce/v1/hosts returns dispatch window data.
|
|
}
|
|
}
|
|
|
|
// rawResvalData mirrors the C xce_resval_t fields used by the Go model.
|
|
// IsDynamic is always true for push-sourced values (controller sends live measurements).
|
|
type rawResvalData struct {
|
|
name string
|
|
host string
|
|
valWhich int // mirrors C xce_val_t.which (Value_u_* discriminant)
|
|
i32 int32
|
|
i64 int64
|
|
u32 uint32
|
|
u64 uint64
|
|
fval float64
|
|
sval string
|
|
bval bool
|
|
valNil bool // true when the C xce_val_t* was NULL
|
|
}
|
|
|
|
// rawResvalToDistribution maps a rawResvalData to a models.ResourceDistributionInfo.
|
|
// Returns nil when valNil is true (C val pointer was NULL) or when valWhich
|
|
// is an unknown discriminant — callers should skip nil results.
|
|
func rawResvalToDistribution(r rawResvalData) *models.ResourceDistributionInfo {
|
|
rd := &models.ResourceDistributionInfo{
|
|
ResourceName: r.name,
|
|
Location: r.host,
|
|
IsDynamic: true, // resource values pushed from workers are always dynamic
|
|
}
|
|
if r.valNil {
|
|
return nil
|
|
}
|
|
switch r.valWhich {
|
|
case 0: // Value_u_ivalue32
|
|
rd.ValueType = int(models.NUMERIC)
|
|
rd.NumericValue = float64(r.i32)
|
|
case 2: // Value_u_ivalue64
|
|
rd.ValueType = int(models.NUMERIC)
|
|
rd.NumericValue = float64(r.i64)
|
|
case 1: // Value_u_uvalue32
|
|
rd.ValueType = int(models.NUMERIC)
|
|
rd.NumericValue = float64(r.u32)
|
|
case 3: // Value_u_uvalue64
|
|
rd.ValueType = int(models.NUMERIC)
|
|
rd.NumericValue = float64(r.u64)
|
|
case 5: // Value_u_fvalue
|
|
rd.ValueType = int(models.NUMERIC)
|
|
rd.NumericValue = r.fval
|
|
case 4: // Value_u_svalue
|
|
rd.ValueType = int(models.STRING)
|
|
rd.StringValue = r.sval
|
|
case 6: // Value_u_bvalue
|
|
rd.ValueType = int(models.BOOLEAN)
|
|
rd.BooleanValue = r.bval
|
|
default:
|
|
return nil
|
|
}
|
|
return rd
|
|
}
|
|
|
|
// rawClusterData mirrors the C cluster_config_t fields used by the Go model.
|
|
// Only config-sourced fields are included (heartbeat_interval, vcpus, inactive_heartbeat_count).
|
|
// Built-in/sourced fields (Version, Status, ControllerName, etc.) come from LoadBuiltinData
|
|
// and are not part of the config push conversion.
|
|
type rawClusterData struct {
|
|
name string
|
|
heartbeatInterval int
|
|
vcpus int
|
|
inactiveHeartbeatCount int
|
|
}
|
|
|
|
// rawClusterToModel maps a rawClusterData to a models.ClusterInfo.
|
|
// Config-sourced fields only; built-in fields remain zero.
|
|
func rawClusterToModel(r rawClusterData) *models.ClusterInfo {
|
|
return &models.ClusterInfo{
|
|
ClusterName: r.name,
|
|
HeartbeatInterval: r.heartbeatInterval,
|
|
Vcpus: r.vcpus,
|
|
InactiveHeartbeatCount: r.inactiveHeartbeatCount,
|
|
}
|
|
}
|