Add X11 forwarding for interactive jobs (xcecli -I -X, ssh -X style)
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>
This commit is contained in:
parent
4f188310ba
commit
85a22271c1
19 changed files with 1744 additions and 50 deletions
|
|
@ -549,8 +549,8 @@ type rawJobData struct {
|
||||||
submit *rawSubmitData
|
submit *rawSubmitData
|
||||||
execHosts []rawHostSlot
|
execHosts []rawHostSlot
|
||||||
pendReason *rawPendReason
|
pendReason *rawPendReason
|
||||||
runUsage *rawRunUsage // nil until the job's first usage report
|
runUsage *rawRunUsage // nil until the job's first usage report
|
||||||
pidInfo *rawPidInfo // nil until the worker's start report
|
pidInfo *rawPidInfo // nil until the worker's start report
|
||||||
}
|
}
|
||||||
|
|
||||||
// rawWorkerData mirrors the C xce_worker_t fields used by the Go model.
|
// rawWorkerData mirrors the C xce_worker_t fields used by the Go model.
|
||||||
|
|
@ -606,6 +606,24 @@ func interactiveURLFromExdata(exdata string) string {
|
||||||
return obj.URL
|
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
|
// pidInfoJSONB marshals a start report's pids/pgids into the PIDInfo Jsonb
|
||||||
// the API exposes (processJobData parses the JSONB string before the row
|
// the API exposes (processJobData parses the JSONB string before the row
|
||||||
// leaves the API, so clients see an object with pids/pgids). Shared by
|
// leaves the API, so clients see an object with pids/pgids). Shared by
|
||||||
|
|
@ -730,11 +748,13 @@ func rawJobToModel(r rawJobData) *models.JobInfo {
|
||||||
// string before the row leaves the API, so clients see an object
|
// string before the row leaves the API, so clients see an object
|
||||||
// with pids/pgids. The interactive URL is promoted to its own
|
// with pids/pgids. The interactive URL is promoted to its own
|
||||||
// top-level field — it is what xcecli's waitForInteractiveURL and
|
// top-level field — it is what xcecli's waitForInteractiveURL and
|
||||||
// the --browse flow consume.
|
// 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 {
|
if jb, ok := pidInfoJSONB(p.pids, p.pgids); ok {
|
||||||
info.PIDInfo = jb
|
info.PIDInfo = jb
|
||||||
}
|
}
|
||||||
info.InteractiveJobURL = interactiveURLFromExdata(p.exdata)
|
info.InteractiveJobURL = interactiveURLFromExdata(p.exdata)
|
||||||
|
info.X11Cookie = x11CookieFromExdata(p.exdata)
|
||||||
}
|
}
|
||||||
return info
|
return info
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -455,7 +455,7 @@ func TestRawJobToModel_PendReasonDetailOnlyPromotesName(t *testing.T) {
|
||||||
jobID: 9,
|
jobID: 9,
|
||||||
state: 1, // JOB_STAT_PEND
|
state: 1, // JOB_STAT_PEND
|
||||||
pendReason: &rawPendReason{
|
pendReason: &rawPendReason{
|
||||||
MainReason: 0, // detail-only: no explicit main reason was set
|
MainReason: 0, // detail-only: no explicit main reason was set
|
||||||
HostReasons: []rawHostReason{{Host: "t2", Reason: 2}, {Host: "t1", Reason: 3}},
|
HostReasons: []rawHostReason{{Host: "t2", Reason: 2}, {Host: "t1", Reason: 3}},
|
||||||
ResReasons: []rawResReason{{Resource: "slots", Host: "", Reason: 4}},
|
ResReasons: []rawResReason{{Resource: "slots", Host: "", Reason: 4}},
|
||||||
},
|
},
|
||||||
|
|
@ -608,6 +608,57 @@ func TestRawJobToModel_NilPidInfo(t *testing.T) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// An -X job's start report carries the per-job X11 auth cookie next to the
|
||||||
|
// x11:// URL; both must surface on the job row (the cookie is what the
|
||||||
|
// xcecli -X client registers with the user's Xauthority).
|
||||||
|
func TestRawJobToModel_X11CookieFromPidInfoExdata(t *testing.T) {
|
||||||
|
r := rawJobData{
|
||||||
|
jobID: 14,
|
||||||
|
state: 2,
|
||||||
|
pidInfo: &rawPidInfo{
|
||||||
|
pids: []int{4321},
|
||||||
|
exdata: `{"url": "x11://10.0.1.5:6100", "x11": 1, "x11cookie": "aabbccdd00112233445566778899eeff"}`,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
info := rawJobToModel(r)
|
||||||
|
|
||||||
|
if want := "x11://10.0.1.5:6100"; info.InteractiveJobURL != want {
|
||||||
|
t.Errorf("InteractiveJobURL: got %q, want %q", info.InteractiveJobURL, want)
|
||||||
|
}
|
||||||
|
if want := "aabbccdd00112233445566778899eeff"; info.X11Cookie != want {
|
||||||
|
t.Errorf("X11Cookie: got %q, want %q", info.X11Cookie, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Non-X jobs never carry a cookie: ttyd/VNC exdata must not leak into
|
||||||
|
// X11Cookie (xcecli refuses to attach -X without one, so a wrong cookie
|
||||||
|
// would break the auth model silently).
|
||||||
|
func TestRawJobToModel_X11CookieAbsentForOtherJobs(t *testing.T) {
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
exdata string
|
||||||
|
}{
|
||||||
|
{"ttyd attach", `{"url": "http://10.0.1.5:9000"}`},
|
||||||
|
{"vnc xjob", `{"url": "10.0.1.5::5901"}`},
|
||||||
|
{"x11 without cookie key", `{"url": "x11://10.0.1.5:6100", "x11": 1}`},
|
||||||
|
{"malformed", "not json"},
|
||||||
|
{"empty", ""},
|
||||||
|
}
|
||||||
|
for _, tc := range cases {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
info := rawJobToModel(rawJobData{
|
||||||
|
jobID: 15,
|
||||||
|
state: 2,
|
||||||
|
pidInfo: &rawPidInfo{pids: []int{1}, exdata: tc.exdata},
|
||||||
|
})
|
||||||
|
if info.X11Cookie != "" {
|
||||||
|
t.Errorf("X11Cookie = %q, want empty for exdata %q", info.X11Cookie, tc.exdata)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestRawJobToModel_Timestamps(t *testing.T) {
|
func TestRawJobToModel_Timestamps(t *testing.T) {
|
||||||
r := rawJobData{
|
r := rawJobData{
|
||||||
jobID: 1,
|
jobID: 1,
|
||||||
|
|
|
||||||
|
|
@ -142,6 +142,7 @@ func (h *DefaultHandler) OnJobPidInfo(jobid uint64, pidInfo *rawPidInfo) {
|
||||||
existing.PIDInfo = jb
|
existing.PIDInfo = jb
|
||||||
}
|
}
|
||||||
existing.InteractiveJobURL = interactiveURLFromExdata(pidInfo.exdata)
|
existing.InteractiveJobURL = interactiveURLFromExdata(pidInfo.exdata)
|
||||||
|
existing.X11Cookie = x11CookieFromExdata(pidInfo.exdata)
|
||||||
}
|
}
|
||||||
store.Jobs.Put(jobid, existing)
|
store.Jobs.Put(jobid, existing)
|
||||||
DefaultBroker.Publish(Event{Type: EventJobUpdated, JobID: jobid,
|
DefaultBroker.Publish(Event{Type: EventJobUpdated, JobID: jobid,
|
||||||
|
|
|
||||||
261
src/frontend/apiserver/controllers/job_xattach.go
Normal file
261
src/frontend/apiserver/controllers/job_xattach.go
Normal file
|
|
@ -0,0 +1,261 @@
|
||||||
|
package controllers
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"net"
|
||||||
|
"net/http"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"apiserver/backend"
|
||||||
|
"apiserver/common"
|
||||||
|
"apiserver/store"
|
||||||
|
|
||||||
|
logger "apiserver/utils"
|
||||||
|
|
||||||
|
"github.com/gin-gonic/gin"
|
||||||
|
"github.com/gorilla/websocket"
|
||||||
|
"go.uber.org/zap"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Interactive XJOB X11-forwarding attach (websocket relay).
|
||||||
|
//
|
||||||
|
// An -X job (interactive=2 + exdata x11=1) runs its command with DISPLAY
|
||||||
|
// pointed at a forwarding port on the worker (inter_job_utils.c, 6100-6199):
|
||||||
|
// the app's X connections arrive there, and this relay bridges each of them
|
||||||
|
// to one websocket so the app's windows render on the user's own X server,
|
||||||
|
// ssh -X style. Worker ports are typically unreachable from user clients in
|
||||||
|
// production, so the apiserver sits in between exactly as for ttyd attach.
|
||||||
|
//
|
||||||
|
// Unlike the ttyd attach, the worker side is a RAW TCP listener -- no
|
||||||
|
// websocket upgrade, no subprotocol, no ttyd framing. One websocket
|
||||||
|
// connection carries one X connection byte-for-byte in binary frames
|
||||||
|
// (X11 needs no framing of its own; text frames are ignored). Close codes
|
||||||
|
// match the ttyd attach: 4000 = cannot reach the forwarding port, 4001 =
|
||||||
|
// the session (or its job) ended. Auth and job ownership are checked once
|
||||||
|
// at relay setup; the worker port is never exposed to the user.
|
||||||
|
|
||||||
|
const (
|
||||||
|
// xattachDialTimeout bounds the apiserver→worker TCP dial.
|
||||||
|
xattachDialTimeout = 10 * time.Second
|
||||||
|
|
||||||
|
// xattachWriteTimeout bounds any single write on either hop.
|
||||||
|
xattachWriteTimeout = 10 * time.Second
|
||||||
|
)
|
||||||
|
|
||||||
|
// xattachUpgrader upgrades the client-facing hop (same shape as the ttyd
|
||||||
|
// attach: the route sits behind JWT auth, so CheckOrigin can stay open).
|
||||||
|
var xattachUpgrader = websocket.Upgrader{
|
||||||
|
ReadBufferSize: 32 * 1024,
|
||||||
|
WriteBufferSize: 32 * 1024,
|
||||||
|
CheckOrigin: func(r *http.Request) bool { return true },
|
||||||
|
}
|
||||||
|
|
||||||
|
// XAttachJob relays the caller's websocket to one X connection of an -X
|
||||||
|
// job's forwarding port. GET /xce/v1/jobs/:jobids/xattach (websocket
|
||||||
|
// upgrade). Auth mirrors AttachJob: Authorization header or ?token= query.
|
||||||
|
func (j JobController) XAttachJob(ctx *gin.Context) {
|
||||||
|
logger.Debug("Enter XAttachJob() function from ", zap.String("client_ip", ctx.ClientIP()))
|
||||||
|
|
||||||
|
token := ctx.GetHeader("Authorization")
|
||||||
|
if token == "" {
|
||||||
|
if qt := ctx.Query("token"); qt != "" {
|
||||||
|
token = "Bearer " + qt
|
||||||
|
}
|
||||||
|
}
|
||||||
|
logonUser, err := GetUsernameFromToken(token)
|
||||||
|
if err != nil {
|
||||||
|
ReturnError(ctx, common.ERROR_CODE, err.Error(), nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
jobInfo, err := parseJobIDParam(ctx.Param("jobids"))
|
||||||
|
if err != nil {
|
||||||
|
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
job, ok := store.Jobs.Get(uint64(jobInfo.Unique_JobID))
|
||||||
|
if !ok {
|
||||||
|
ReturnError(ctx, common.ERROR_NOT_FOUND_CODE, common.ERROR_JOB_NOT_FOUND_MSG, nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Same ownership rule as the ttyd attach: the session is job control.
|
||||||
|
if job.SubmitUser != logonUser {
|
||||||
|
scope, perr := userJobScope(logonUser, "job:control")
|
||||||
|
if perr != nil || !scope.covers(job.SubmitUser) {
|
||||||
|
ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if job.InteractiveJobURL == "" {
|
||||||
|
ReturnError(ctx, common.INVALID_PARAMS_CODE,
|
||||||
|
fmt.Sprintf("job <%s> has no interactive session URL (not an interactive job, not running, or started before URL reporting)",
|
||||||
|
backend.FormatJobID(job.Unique_JobID)), nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
workerAddr, err := x11ForwardAddr(job.InteractiveJobURL)
|
||||||
|
if err != nil {
|
||||||
|
ReturnError(ctx, common.INVALID_PARAMS_CODE, "invalid X11 session URL: "+err.Error(), nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Upgrade the client hop. After this point errors go out as websocket
|
||||||
|
// close frames, not HTTP responses.
|
||||||
|
clientConn, err := xattachUpgrader.Upgrade(ctx.Writer, ctx.Request, nil)
|
||||||
|
if err != nil {
|
||||||
|
logger.Error("xattach: client upgrade failed", zap.Error(err))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer clientConn.Close()
|
||||||
|
|
||||||
|
workerConn, err := net.DialTimeout("tcp", workerAddr, xattachDialTimeout)
|
||||||
|
if err != nil {
|
||||||
|
logger.Warn("xattach: worker dial failed", zap.String("addr", workerAddr), zap.Error(err))
|
||||||
|
attachClose(clientConn, attachCloseJobGone, "cannot reach the X11 forwarding port on the worker: "+err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer workerConn.Close()
|
||||||
|
|
||||||
|
logger.Info("xattach: relaying X11 connection",
|
||||||
|
zap.Int64("jobid", job.Unique_JobID), zap.String("user", logonUser),
|
||||||
|
zap.String("worker_addr", workerAddr))
|
||||||
|
|
||||||
|
xattachRelay(uint64(job.Unique_JobID), clientConn, workerConn)
|
||||||
|
logger.Info("xattach: relay finished", zap.Int64("jobid", job.Unique_JobID))
|
||||||
|
}
|
||||||
|
|
||||||
|
// x11ForwardAddr turns the reported session URL (x11://host:port) into the
|
||||||
|
// worker TCP address to relay to.
|
||||||
|
func x11ForwardAddr(raw string) (string, error) {
|
||||||
|
raw = strings.TrimSpace(raw)
|
||||||
|
if !strings.HasPrefix(raw, "x11://") {
|
||||||
|
return "", fmt.Errorf("not an X11 forwarding URL: %q", raw)
|
||||||
|
}
|
||||||
|
addr := strings.TrimPrefix(raw, "x11://")
|
||||||
|
if addr == "" || strings.Contains(addr, "/") {
|
||||||
|
return "", fmt.Errorf("malformed X11 forwarding URL %q", raw)
|
||||||
|
}
|
||||||
|
return addr, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// xattachRelay copies bytes between the websocket and the worker TCP
|
||||||
|
// connection until either side goes away. The job's terminal state ends the
|
||||||
|
// relay regardless (EventJobFinished watcher, same as the ttyd attach).
|
||||||
|
func xattachRelay(jobID uint64, client *websocket.Conn, worker net.Conn) {
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
wg.Add(2)
|
||||||
|
done := make(chan struct{})
|
||||||
|
// teardown closes both hops once; each pump's blocking read unblocks on
|
||||||
|
// the close of the socket it is reading, so one close ends both pumps.
|
||||||
|
// Closing must happen the moment either pump exits -- waiting for
|
||||||
|
// wg.Wait() here would deadlock (the remaining pump blocks in a read
|
||||||
|
// nothing else would ever interrupt). The eventJobFinished watcher and
|
||||||
|
// the late teardown below also call Close; net.Conn and gorilla
|
||||||
|
// websocket Close are both idempotent.
|
||||||
|
var closeOnce sync.Once
|
||||||
|
teardown := func() {
|
||||||
|
closeOnce.Do(func() {
|
||||||
|
close(done)
|
||||||
|
client.Close()
|
||||||
|
worker.Close()
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
defer teardown()
|
||||||
|
if !xattachPumpTCP(worker, client, done) {
|
||||||
|
// worker side ended on its own: the X connection (or its job)
|
||||||
|
// ended. Send the policy close before the teardown closes the
|
||||||
|
// socket so the client reliably reads 4001 instead of 1006.
|
||||||
|
attachClose(client, attachCloseJobEnded, "X11 session ended")
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
defer teardown()
|
||||||
|
// client-initiated end: teardown closes the worker conn so the
|
||||||
|
// TCP pump's blocking read unblocks.
|
||||||
|
xattachPumpWS(client, worker, done)
|
||||||
|
}()
|
||||||
|
|
||||||
|
// The job ending is a session end regardless of the worker hop.
|
||||||
|
go func() {
|
||||||
|
ch := backend.DefaultBroker.Subscribe(4)
|
||||||
|
defer backend.DefaultBroker.Unsubscribe(ch)
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-done:
|
||||||
|
return
|
||||||
|
case ev, ok := <-ch:
|
||||||
|
if !ok {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if ev.JobID == jobID && ev.Type == backend.EventJobFinished {
|
||||||
|
select {
|
||||||
|
case <-done:
|
||||||
|
default:
|
||||||
|
logger.Info("xattach: job finished, ending relayed session",
|
||||||
|
zap.Uint64("jobid", jobID))
|
||||||
|
_ = client.WriteControl(websocket.CloseMessage,
|
||||||
|
websocket.FormatCloseMessage(attachCloseJobEnded,
|
||||||
|
"X11 session ended (job finished)"),
|
||||||
|
time.Now().Add(xattachWriteTimeout))
|
||||||
|
}
|
||||||
|
go teardown()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
// Either pump's exit tears both sockets down (teardown above), which
|
||||||
|
// unblocks the other pump's read; wg.Wait below just reaps them.
|
||||||
|
<-done
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
|
||||||
|
// xattachPumpTCP copies worker TCP bytes into websocket binary frames.
|
||||||
|
// Returns false when the TCP side ended on its own (its peer outlived it,
|
||||||
|
// and the caller may want to send it a policy close).
|
||||||
|
func xattachPumpTCP(src net.Conn, dst *websocket.Conn, done <-chan struct{}) bool {
|
||||||
|
buf := make([]byte, 32*1024)
|
||||||
|
for {
|
||||||
|
n, err := src.Read(buf)
|
||||||
|
if n > 0 {
|
||||||
|
_ = dst.SetWriteDeadline(time.Now().Add(xattachWriteTimeout))
|
||||||
|
if werr := dst.WriteMessage(websocket.BinaryMessage, buf[:n]); werr != nil {
|
||||||
|
return true // dst went away first; teardown unblocks us
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
select {
|
||||||
|
case <-done:
|
||||||
|
return true
|
||||||
|
default:
|
||||||
|
return err == net.ErrClosed
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// xattachPumpWS copies websocket binary frames into the worker TCP
|
||||||
|
// connection. Text frames are not part of the X11 wire and are ignored.
|
||||||
|
func xattachPumpWS(src *websocket.Conn, dst net.Conn, done <-chan struct{}) {
|
||||||
|
for {
|
||||||
|
msgType, data, err := src.ReadMessage()
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if msgType != websocket.BinaryMessage {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
_ = dst.SetWriteDeadline(time.Now().Add(xattachWriteTimeout))
|
||||||
|
if _, err := dst.Write(data); err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
47
src/frontend/apiserver/controllers/job_xattach_test.go
Normal file
47
src/frontend/apiserver/controllers/job_xattach_test.go
Normal file
|
|
@ -0,0 +1,47 @@
|
||||||
|
package controllers
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
// x11ForwardAddr turns the worker's reported session URL (x11://host:port,
|
||||||
|
// generate_url's X11 branch) into the raw TCP address the relay dials.
|
||||||
|
func TestX11ForwardAddr(t *testing.T) {
|
||||||
|
cases := []struct {
|
||||||
|
raw string
|
||||||
|
want string
|
||||||
|
}{
|
||||||
|
{"x11://10.0.1.5:6100", "10.0.1.5:6100"},
|
||||||
|
{"x11://worker.example:6199", "worker.example:6199"},
|
||||||
|
{" x11://10.0.1.5:6100 ", "10.0.1.5:6100"},
|
||||||
|
}
|
||||||
|
for _, c := range cases {
|
||||||
|
got, err := x11ForwardAddr(c.raw)
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("x11ForwardAddr(%q): unexpected error %v", c.raw, err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if got != c.want {
|
||||||
|
t.Errorf("x11ForwardAddr(%q): got %q, want %q", c.raw, got, c.want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A ttyd or VNC URL routed at the xattach relay is a bug elsewhere in the
|
||||||
|
// chain (the URL decides the backend) -- reject it instead of dialing a
|
||||||
|
// websocket endpoint as raw TCP.
|
||||||
|
func TestX11ForwardAddr_NotX11(t *testing.T) {
|
||||||
|
cases := []string{
|
||||||
|
"", " ",
|
||||||
|
"http://10.0.1.5:9000",
|
||||||
|
"10.0.1.5::5901",
|
||||||
|
"x11://",
|
||||||
|
"x11://10.0.1.5:6100/ws",
|
||||||
|
"x11://10.0.1.5:6100/path?x=1",
|
||||||
|
}
|
||||||
|
for _, raw := range cases {
|
||||||
|
if got, err := x11ForwardAddr(raw); err == nil {
|
||||||
|
t.Errorf("x11ForwardAddr(%q): got %q, want error", raw, got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -2056,6 +2056,13 @@ func processJobRequest(ctx *gin.Context, req *models.JobSubmitRequest, defaultVa
|
||||||
extendDataMap["attach"] = req.Attach
|
extendDataMap["attach"] = req.Attach
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// x11=1: an XJOB job running as X11 forwarding (xcecli -I -X). The
|
||||||
|
// worker exports DISPLAY to its forwarding port and reports a per-job
|
||||||
|
// xauth cookie instead of starting VNC.
|
||||||
|
if req.X11 != 0 {
|
||||||
|
extendDataMap["x11"] = req.X11
|
||||||
|
}
|
||||||
|
|
||||||
if len(extendDataMap) > 0 {
|
if len(extendDataMap) > 0 {
|
||||||
updatedExtendDataBytes, err := json.Marshal(extendDataMap)
|
updatedExtendDataBytes, err := json.Marshal(extendDataMap)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
|
||||||
|
|
@ -478,6 +478,52 @@ func TestJobHandler_SubmitInteractiveAttachExdata(t *testing.T) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestJobHandler_SubmitX11Exdata checks the submit handler packs x11 into
|
||||||
|
// the job's ExtendData for -X jobs: the key selects the X11-forwarding
|
||||||
|
// backend worker-side (DISPLAY to the forwarding port instead of VNC).
|
||||||
|
func TestJobHandler_SubmitX11Exdata(t *testing.T) {
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
req map[string]interface{}
|
||||||
|
wantX11 bool
|
||||||
|
}{
|
||||||
|
{"x11 job", map[string]interface{}{
|
||||||
|
"job_name": "a", "command": "xclock", "interactive": 2, "x11": 1,
|
||||||
|
}, true},
|
||||||
|
{"plain xjob (VNC backend)", map[string]interface{}{
|
||||||
|
"job_name": "a", "command": "xclock", "interactive": 2,
|
||||||
|
}, false},
|
||||||
|
{"tty job without x11", map[string]interface{}{
|
||||||
|
"job_name": "a", "command": "sleep 1", "interactive": 1,
|
||||||
|
}, false},
|
||||||
|
}
|
||||||
|
for _, tc := range cases {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
f := newHandlerFixture(t)
|
||||||
|
f.seedQueue("normal")
|
||||||
|
|
||||||
|
w := f.submit(tc.req)
|
||||||
|
if w.Code != http.StatusOK {
|
||||||
|
t.Fatalf("status = %d, want 200; body: %s", w.Code, w.Body.String())
|
||||||
|
}
|
||||||
|
subs := f.submissions()
|
||||||
|
if len(subs) != 1 {
|
||||||
|
t.Fatalf("submitJob calls = %d, want 1", len(subs))
|
||||||
|
}
|
||||||
|
var ex map[string]interface{}
|
||||||
|
if raw := subs[0].Job.ExtendData.String; raw != "" {
|
||||||
|
if err := json.Unmarshal([]byte(raw), &ex); err != nil {
|
||||||
|
t.Fatalf("ExtendData %q is not JSON: %v", raw, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_, hasX11 := ex["x11"]
|
||||||
|
if hasX11 != tc.wantX11 {
|
||||||
|
t.Errorf("ExtendData x11 present = %v, want %v (exdata: %s)", hasX11, tc.wantX11, subs[0].Job.ExtendData.String)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestJobHandler_SmokeControlByID controls one seeded ordinary job by ID
|
// TestJobHandler_SmokeControlByID controls one seeded ordinary job by ID
|
||||||
// and checks the dispatch is recorded once with its unique_job_id.
|
// and checks the dispatch is recorded once with its unique_job_id.
|
||||||
func TestJobHandler_SmokeControlByID(t *testing.T) {
|
func TestJobHandler_SmokeControlByID(t *testing.T) {
|
||||||
|
|
|
||||||
|
|
@ -580,6 +580,12 @@ type JobInfo struct {
|
||||||
// JobAdditionalFields (the query-field whitelist) must stay in sync with
|
// JobAdditionalFields (the query-field whitelist) must stay in sync with
|
||||||
// this name.
|
// this name.
|
||||||
InteractiveJobURL string `json:"interactive_job_url"`
|
InteractiveJobURL string `json:"interactive_job_url"`
|
||||||
|
// X11Cookie is the per-job MIT-MAGIC-COOKIE-1 of an -X job (XJOB with
|
||||||
|
// exdata x11=1), extracted from the same start-report exdata as
|
||||||
|
// InteractiveJobURL. The xcecli -X client registers it with the user's
|
||||||
|
// Xauthority before the app's first relayed X connection. Empty for
|
||||||
|
// every other kind of job.
|
||||||
|
X11Cookie string `json:"x11_cookie"`
|
||||||
}
|
}
|
||||||
|
|
||||||
type JobExecHosts struct {
|
type JobExecHosts struct {
|
||||||
|
|
@ -632,6 +638,7 @@ type JobAdditionalFields struct {
|
||||||
Pids []int `json:"pids"` // List of process IDs associated with the job
|
Pids []int `json:"pids"` // List of process IDs associated with the job
|
||||||
Pgids []int `json:"pgids"` // List of process group IDs associated with the job
|
Pgids []int `json:"pgids"` // List of process group IDs associated with the job
|
||||||
InteractiveJobURL string `json:"interactive_job_url"`
|
InteractiveJobURL string `json:"interactive_job_url"`
|
||||||
|
X11Cookie string `json:"x11_cookie"` // -X job's per-session auth cookie (from start exdata)
|
||||||
// CoreLimit/MemLimit: JobInfo has no equivalent field (these are
|
// CoreLimit/MemLimit: JobInfo has no equivalent field (these are
|
||||||
// submission-time resource limits, not job-state fields), but xcecli's
|
// submission-time resource limits, not job-state fields), but xcecli's
|
||||||
// JobFieldsMap accepts them as valid -o/-l display fields (bug-1674) --
|
// JobFieldsMap accepts them as valid -o/-l display fields (bug-1674) --
|
||||||
|
|
@ -642,38 +649,43 @@ type JobAdditionalFields struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
type JobSubmitRequest struct {
|
type JobSubmitRequest struct {
|
||||||
JobName string `json:"job_name,omitempty" `
|
JobName string `json:"job_name,omitempty" `
|
||||||
Command string `json:"command" binding:"required"`
|
Command string `json:"command" binding:"required"`
|
||||||
QueueName string `json:"queue_name,omitempty"` // set to DEFAULT_QUEUE in config
|
QueueName string `json:"queue_name,omitempty"` // set to DEFAULT_QUEUE in config
|
||||||
ProjectName string `json:"project_name,omitempty"` // default project: "default"
|
ProjectName string `json:"project_name,omitempty"` // default project: "default"
|
||||||
UserGroup string `json:"user_group,omitempty"`
|
UserGroup string `json:"user_group,omitempty"`
|
||||||
LicenseProject string `json:"license_project,omitempty"`
|
LicenseProject string `json:"license_project,omitempty"`
|
||||||
JobDescription string `json:"job_description,omitempty"`
|
JobDescription string `json:"job_description,omitempty"`
|
||||||
Resreq string `json:"resreq,omitempty"` // default: "type == local"
|
Resreq string `json:"resreq,omitempty"` // default: "type == local"
|
||||||
Affinity string `json:"affinity,omitempty"`
|
Affinity string `json:"affinity,omitempty"`
|
||||||
Gpureq string `json:"gpureq,omitempty"`
|
Gpureq string `json:"gpureq,omitempty"`
|
||||||
RsvID string `json:"rsv_id,omitempty"`
|
RsvID string `json:"rsv_id,omitempty"`
|
||||||
SLA string `json:"sla,omitempty"`
|
SLA string `json:"sla,omitempty"`
|
||||||
MaxNumProcessors int `json:"maxnum_processors,omitempty"`
|
MaxNumProcessors int `json:"maxnum_processors,omitempty"`
|
||||||
NumProcessors int `json:"num_processors,omitempty"`
|
NumProcessors int `json:"num_processors,omitempty"`
|
||||||
AskedHosts string `json:"asked_hosts,omitempty"`
|
AskedHosts string `json:"asked_hosts,omitempty"`
|
||||||
LaunchType string `json:"launch_type,omitempty"`
|
LaunchType string `json:"launch_type,omitempty"`
|
||||||
BeginTime CustomTime `json:"begin_time,omitempty"` //k change type
|
BeginTime CustomTime `json:"begin_time,omitempty"` //k change type
|
||||||
TermTime CustomTime `json:"term_time,omitempty"` //k change type
|
TermTime CustomTime `json:"term_time,omitempty"` //k change type
|
||||||
JobPriority int `json:"job_priority,omitempty"`
|
JobPriority int `json:"job_priority,omitempty"`
|
||||||
DependCond string `json:"depend_cond,omitempty"`
|
DependCond string `json:"depend_cond,omitempty"`
|
||||||
JobGroup string `json:"job_group,omitempty"`
|
JobGroup string `json:"job_group,omitempty"`
|
||||||
HostSpec string `json:"host_spec,omitempty"`
|
HostSpec string `json:"host_spec,omitempty"`
|
||||||
Env []string `json:"env,omitempty"`
|
Env []string `json:"env,omitempty"`
|
||||||
Tags []string `json:"tags,omitempty"`
|
Tags []string `json:"tags,omitempty"`
|
||||||
Exclusive bool `json:"exclusive,omitempty"`
|
Exclusive bool `json:"exclusive,omitempty"`
|
||||||
Rerunable bool `json:"rerunable,omitempty"`
|
Rerunable bool `json:"rerunable,omitempty"`
|
||||||
Interactive int `json:"interactive,omitempty"`
|
Interactive int `json:"interactive,omitempty"`
|
||||||
// Attach marks an interactive TTY job as the terminal-attach kind
|
// Attach marks an interactive TTY job as the terminal-attach kind
|
||||||
// (xcecli -I without --browse): the session ends with the command, so
|
// (xcecli -I without --browse): the session ends with the command, so
|
||||||
// the worker may reap the ttyd when no client ever connects. 0 keeps
|
// the worker may reap the ttyd when no client ever connects. 0 keeps
|
||||||
// the web-viewed semantics.
|
// the web-viewed semantics.
|
||||||
Attach int `json:"attach,omitempty"`
|
Attach int `json:"attach,omitempty"`
|
||||||
|
// X11 (exdata key "x11") selects the X11-forwarding backend for an XJOB
|
||||||
|
// (interactive=2) job: the worker exports DISPLAY to its forwarding port
|
||||||
|
// instead of starting VNC, and the app's X connection is relayed to the
|
||||||
|
// user's local X server through the xattach websocket.
|
||||||
|
X11 int `json:"x11,omitempty"`
|
||||||
WarningTimePeriod int `json:"warning_time_period,omitempty"`
|
WarningTimePeriod int `json:"warning_time_period,omitempty"`
|
||||||
WarningAction string `json:"warning_action,omitempty"`
|
WarningAction string `json:"warning_action,omitempty"`
|
||||||
DataSpecFile string `json:"data_spec_file,omitempty"`
|
DataSpecFile string `json:"data_spec_file,omitempty"`
|
||||||
|
|
|
||||||
|
|
@ -147,6 +147,9 @@ func Router() *gin.Engine {
|
||||||
// Interactive-job websocket relay (attach): long-lived connection to
|
// Interactive-job websocket relay (attach): long-lived connection to
|
||||||
// the job's ttyd session on the worker, proxied by the apiserver.
|
// the job's ttyd session on the worker, proxied by the apiserver.
|
||||||
jobsApi.GET("/:jobids/attach", controllers.RequirePermission("job:control"), jobHandler.AttachJob)
|
jobsApi.GET("/:jobids/attach", controllers.RequirePermission("job:control"), jobHandler.AttachJob)
|
||||||
|
// X11-forwarding relay (-X jobs): one websocket per X connection of
|
||||||
|
// the job's forwarding port, raw TCP on the worker side.
|
||||||
|
jobsApi.GET("/:jobids/xattach", controllers.RequirePermission("job:control"), jobHandler.XAttachJob)
|
||||||
jobsApi.GET("", controllers.RequirePermission("job:view"), jobHandler.GetJobsInfo)
|
jobsApi.GET("", controllers.RequirePermission("job:view"), jobHandler.GetJobsInfo)
|
||||||
jobsApi.GET("/:jobids", controllers.RequirePermission("job:view"), jobHandler.GetJobByIDs)
|
jobsApi.GET("/:jobids", controllers.RequirePermission("job:view"), jobHandler.GetJobByIDs)
|
||||||
jobsApi.POST("/create", controllers.RequirePermission("queue:submit"), jobHandler.CreateJob)
|
jobsApi.POST("/create", controllers.RequirePermission("queue:submit"), jobHandler.CreateJob)
|
||||||
|
|
|
||||||
|
|
@ -57,6 +57,7 @@ type JobSubmitter struct {
|
||||||
interactive bool
|
interactive bool
|
||||||
useTTY bool
|
useTTY bool
|
||||||
useXJob bool
|
useXJob bool
|
||||||
|
useX11 bool
|
||||||
useBrowse bool
|
useBrowse bool
|
||||||
reportedStart bool
|
reportedStart bool
|
||||||
projectName string
|
projectName string
|
||||||
|
|
@ -89,16 +90,8 @@ func (js *JobSubmitter) ParseCmdArgs(cmd *cobra.Command, args []string) error {
|
||||||
}
|
}
|
||||||
|
|
||||||
if js.interactive {
|
if js.interactive {
|
||||||
if js.useTTY && js.useXJob {
|
if err := js.applyInteractiveDefaults(); err != nil {
|
||||||
return fmt.Errorf("error: cannot use both --tty and --xjob")
|
return err
|
||||||
}
|
|
||||||
// Bare -I is the attach interactive session: -I without --xjob is
|
|
||||||
// equivalent to -I --tty. The old mapping (interactive=3, readonly
|
|
||||||
// DISPLAY web view) had no interactive input and no user hint; the
|
|
||||||
// API's interactive=3 semantics are unchanged -- only this CLI's
|
|
||||||
// default remaps to TTY attach (use --xjob for VNC).
|
|
||||||
if !js.useXJob {
|
|
||||||
js.useTTY = true
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -243,6 +236,31 @@ func (js *JobSubmitter) parseConfigFile() error {
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// applyInteractiveDefaults resolves the -I flag family into the concrete
|
||||||
|
// interactive kind. Bare -I is the attach interactive session: -I without
|
||||||
|
// --xjob is equivalent to -I --tty. The old mapping (interactive=3, readonly
|
||||||
|
// DISPLAY web view) had no interactive input and no user hint; the API's
|
||||||
|
// interactive=3 semantics are unchanged -- only this CLI's default remaps to
|
||||||
|
// TTY attach (use --xjob for VNC). -I -X runs a graphical X11 app in a popup
|
||||||
|
// window on this display (ssh -X experience): interactive=2 (XJOB) with the
|
||||||
|
// x11 backend selected worker-side by exdata; it suppresses the TTY remap.
|
||||||
|
func (js *JobSubmitter) applyInteractiveDefaults() error {
|
||||||
|
if js.useTTY && js.useXJob {
|
||||||
|
return fmt.Errorf("error: cannot use both --tty and --xjob")
|
||||||
|
}
|
||||||
|
if js.useX11 && js.useBrowse {
|
||||||
|
return fmt.Errorf("error: cannot use both --x11 and --browse")
|
||||||
|
}
|
||||||
|
if !js.useXJob {
|
||||||
|
js.useTTY = true
|
||||||
|
}
|
||||||
|
if js.useX11 {
|
||||||
|
js.useTTY = false
|
||||||
|
js.useXJob = true
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// BuildJobRequest
|
// BuildJobRequest
|
||||||
func (js *JobSubmitter) BuildJobRequest() (*models.JobSubmitRequest, error) {
|
func (js *JobSubmitter) BuildJobRequest() (*models.JobSubmitRequest, error) {
|
||||||
// create the job request with only required fields
|
// create the job request with only required fields
|
||||||
|
|
@ -261,6 +279,12 @@ func (js *JobSubmitter) BuildJobRequest() (*models.JobSubmitRequest, error) {
|
||||||
}
|
}
|
||||||
} else if js.useXJob {
|
} else if js.useXJob {
|
||||||
jobReq.Interactive = 2 // XJob mode
|
jobReq.Interactive = 2 // XJob mode
|
||||||
|
// -X: the X11-forwarding backend (popup window on the user's own
|
||||||
|
// display) instead of VNC; the worker picks the backend from this
|
||||||
|
// exdata key, so the interactive enum stays untouched.
|
||||||
|
if js.useX11 {
|
||||||
|
jobReq.X11 = 1
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
jobReq.Interactive = 3 // only interactive, no special mode
|
jobReq.Interactive = 3 // only interactive, no special mode
|
||||||
}
|
}
|
||||||
|
|
@ -540,6 +564,14 @@ func (js *JobSubmitter) RunJobsSubmit() error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// -X (X11 forwarding): the app's X connection is relayed through the
|
||||||
|
// apiserver to this display -- windows pop up here like ssh -X. The
|
||||||
|
// URL is a worker-internal detail; the session ends with the command
|
||||||
|
// and the CLI pumps the relay until the job finishes.
|
||||||
|
if js.useX11 {
|
||||||
|
return runX11Session(serverInfo.URL, serverInfo.Token, jobID)
|
||||||
|
}
|
||||||
|
|
||||||
// XJOB (VNC) has no terminal rendering -- browser remains the path
|
// XJOB (VNC) has no terminal rendering -- browser remains the path
|
||||||
fmt.Printf("interactive session URL: %s\n", url)
|
fmt.Printf("interactive session URL: %s\n", url)
|
||||||
if err := openBrowser(url); err != nil {
|
if err := openBrowser(url); err != nil {
|
||||||
|
|
@ -765,6 +797,7 @@ Example:
|
||||||
xce jobs submit -I --tty sleep 10 # submit interactive job with tty and attach in this terminal
|
xce jobs submit -I --tty sleep 10 # submit interactive job with tty and attach in this terminal
|
||||||
xce jobs submit -I --tty --browse sleep 10 # submit interactive tty job, open the web URL in a browser
|
xce jobs submit -I --tty --browse sleep 10 # submit interactive tty job, open the web URL in a browser
|
||||||
xce jobs submit -I --xjob sleep 10 # submit interactive job with xjob (VNC)
|
xce jobs submit -I --xjob sleep 10 # submit interactive job with xjob (VNC)
|
||||||
|
xce jobs submit -I -X xclock # run xclock on the cluster, window pops up on this display (ssh -X style)
|
||||||
xce jobs submit -J 'name[1-N]' sleep 10 # submit a job array of N elements named "name"
|
xce jobs submit -J 'name[1-N]' sleep 10 # submit a job array of N elements named "name"
|
||||||
xce jobs submit -J 'name[1]' sleep 10 # submit a job array of one element
|
xce jobs submit -J 'name[1]' sleep 10 # submit a job array of one element
|
||||||
|
|
||||||
|
|
@ -829,9 +862,10 @@ Job arrays:
|
||||||
flags.StringVarP(&jobSubmitter.outdir, "outdir", "d", "", "specify the output directory")
|
flags.StringVarP(&jobSubmitter.outdir, "outdir", "d", "", "specify the output directory")
|
||||||
flags.StringVarP(&jobSubmitter.projectName, "project_name", "P", "", "specify the project name")
|
flags.StringVarP(&jobSubmitter.projectName, "project_name", "P", "", "specify the project name")
|
||||||
flags.StringVar(&jobSubmitter.configFile, "configfile", "", "specify the configuration file for job submission")
|
flags.StringVar(&jobSubmitter.configFile, "configfile", "", "specify the configuration file for job submission")
|
||||||
flags.BoolVarP(&jobSubmitter.interactive, "interactive", "I", false, "submit interactive job (attach in this terminal; same as -I --tty; add --xjob for VNC)")
|
flags.BoolVarP(&jobSubmitter.interactive, "interactive", "I", false, "submit interactive job (attach in this terminal; same as -I --tty; add --xjob for VNC; add -X for X11 popup)")
|
||||||
flags.BoolVar(&jobSubmitter.useTTY, "tty", false, "use TTY for interactive job (attach in this terminal after dispatch; --browse opens the web URL instead; default for -I)")
|
flags.BoolVar(&jobSubmitter.useTTY, "tty", false, "use TTY for interactive job (attach in this terminal after dispatch; --browse opens the web URL instead; default for -I)")
|
||||||
flags.BoolVar(&jobSubmitter.useXJob, "xjob", false, "use XJob (VNC) for interactive job")
|
flags.BoolVar(&jobSubmitter.useXJob, "xjob", false, "use XJob (VNC) for interactive job")
|
||||||
|
flags.BoolVarP(&jobSubmitter.useX11, "x11", "X", false, "run a graphical X11 app and show its windows on this display (ssh -X style; requires a local X server and xauth)")
|
||||||
flags.BoolVar(&jobSubmitter.useBrowse, "browse", false, "open the interactive session URL in a browser instead of attaching (TTY jobs attach by default)")
|
flags.BoolVar(&jobSubmitter.useBrowse, "browse", false, "open the interactive session URL in a browser instead of attaching (TTY jobs attach by default)")
|
||||||
flags.StringVarP(&jobSubmitter.askedHosts, "m", "m", "", "specify the running hosts, with space for multiple values")
|
flags.StringVarP(&jobSubmitter.askedHosts, "m", "m", "", "specify the running hosts, with space for multiple values")
|
||||||
flags.StringVarP(&jobSubmitter.launchType, "launch_type", "", "", "specify the launch type")
|
flags.StringVarP(&jobSubmitter.launchType, "launch_type", "", "", "specify the launch type")
|
||||||
|
|
|
||||||
311
src/frontend/xcecli/cmd/jobs/x11.go
Normal file
311
src/frontend/xcecli/cmd/jobs/x11.go
Normal file
|
|
@ -0,0 +1,311 @@
|
||||||
|
package jobs
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"net"
|
||||||
|
"os"
|
||||||
|
"os/exec"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/gorilla/websocket"
|
||||||
|
|
||||||
|
"xcecli/common/request"
|
||||||
|
"xcecli/models"
|
||||||
|
)
|
||||||
|
|
||||||
|
// X11-forwarding session for an interactive XJOB job (-I -X).
|
||||||
|
//
|
||||||
|
// The worker exported DISPLAY to its forwarding port and starts the app
|
||||||
|
// immediately (no daemon waits for a client). The app connects to that port;
|
||||||
|
// the apiserver's /xattach relay bridges each worker-side TCP connection to
|
||||||
|
// one websocket; this side pumps each websocket into the user's local X
|
||||||
|
// server, so the app's windows appear on the user's display exactly like
|
||||||
|
// ssh -X. Rendering happens client-side: no VNC server, no browser.
|
||||||
|
//
|
||||||
|
// One websocket = one X connection = one worker-side accept. An X app opens
|
||||||
|
// one connection on XOpenDisplay (secondary ones, e.g. xim, are rare and not
|
||||||
|
// supported in v1 -- the worker refuses them loudly rather than dropping
|
||||||
|
// events silently).
|
||||||
|
//
|
||||||
|
// Auth is ssh -X's model: the worker generates a per-job MIT-MAGIC-COOKIE-1
|
||||||
|
// and reports it in the job's exdata; we add it to the user's Xauthority
|
||||||
|
// before the app's first connection can arrive and remove it when the
|
||||||
|
// session ends. The job detail API carries the cookie next to the URL.
|
||||||
|
|
||||||
|
// x11DisplayAddr resolves $DISPLAY into a dialable address of the local X
|
||||||
|
// server: a unix socket path or a TCP host:port. Only local displays are
|
||||||
|
// supported -- the whole point of -X is popping windows up here.
|
||||||
|
func x11DisplayAddr(display string) (string, bool, error) {
|
||||||
|
display = strings.TrimSpace(display)
|
||||||
|
if display == "" {
|
||||||
|
return "", false, fmt.Errorf("DISPLAY is not set: -X shows windows on this machine's X server, run from a graphical session (or inside ssh -X)")
|
||||||
|
}
|
||||||
|
// X syntax: [host]:N[.S]; an empty host means the local unix socket.
|
||||||
|
idx := strings.LastIndex(display, ":")
|
||||||
|
if idx < 0 {
|
||||||
|
return "", false, fmt.Errorf("cannot parse DISPLAY %q (expected [host]:N)", display)
|
||||||
|
}
|
||||||
|
host := display[:idx]
|
||||||
|
num := display[idx+1:]
|
||||||
|
if dot := strings.Index(num, "."); dot >= 0 {
|
||||||
|
num = num[:dot]
|
||||||
|
}
|
||||||
|
n := 0
|
||||||
|
if _, err := fmt.Sscanf(num, "%d", &n); err != nil {
|
||||||
|
return "", false, fmt.Errorf("cannot parse display number in DISPLAY %q", display)
|
||||||
|
}
|
||||||
|
if host == "" || host == "unix" {
|
||||||
|
return fmt.Sprintf("/tmp/.X11-unix/X%d", n), true, nil
|
||||||
|
}
|
||||||
|
// A remote-looking host is only valid when it is us (X server bound to
|
||||||
|
// an interface) -- TCP to host:6000+N either way; ssh -X users normally
|
||||||
|
// have "localhost:10.0" (their own ssh-forwarded display), which is a
|
||||||
|
// local TCP listener owned by sshd and works the same.
|
||||||
|
return fmt.Sprintf("%s:%d", host, 6000+n), false, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// x11XauthFile picks the Xauthority file xauth should touch: $XAUTHORITY if
|
||||||
|
// set, else ~/.Xauthority. ssh -X sessions set XAUTHORITY to a per-session
|
||||||
|
// file, which is exactly the one the local X server honors.
|
||||||
|
func x11XauthFile() (string, error) {
|
||||||
|
if f := os.Getenv("XAUTHORITY"); f != "" {
|
||||||
|
return f, nil
|
||||||
|
}
|
||||||
|
home, err := os.UserHomeDir()
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("cannot locate the Xauthority file (no XAUTHORITY, no home): %s", err)
|
||||||
|
}
|
||||||
|
return filepath.Join(home, ".Xauthority"), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// x11Xauth runs xauth with the given arguments, scoped to our auth file.
|
||||||
|
func x11Xauth(authFile string, args ...string) error {
|
||||||
|
cmd := exec.Command("xauth", append([]string{"-f", authFile}, args...)...)
|
||||||
|
if out, err := cmd.CombinedOutput(); err != nil {
|
||||||
|
return fmt.Errorf("xauth %s failed: %s (%s)", strings.Join(args, " "),
|
||||||
|
attachErrText(err), strings.TrimSpace(string(out)))
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// x11Cookie fetches the job's X11 auth cookie reported by the worker
|
||||||
|
// (exdata "x11cookie") alongside the session URL.
|
||||||
|
func x11Cookie(serverURL, token, jobID string) (string, error) {
|
||||||
|
srv, err := x11ServerBase(serverURL)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
var resp models.JobsResponse
|
||||||
|
client := request.NewAPIClient(srv)
|
||||||
|
err = client.GetRequestFromAPIServer(
|
||||||
|
"Bearer "+token,
|
||||||
|
"",
|
||||||
|
fmt.Sprintf("/xce/v1/jobs/%s", jobID),
|
||||||
|
&resp,
|
||||||
|
map[string]string{"fields": "x11_cookie"},
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("failed to get job details: [%s]", request.ExtractErrorMessage(err))
|
||||||
|
}
|
||||||
|
for _, row := range resp.Data.Rows {
|
||||||
|
if row.X11Cookie != "" {
|
||||||
|
return row.X11Cookie, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return "", fmt.Errorf("job <%s> has no X11 session cookie (not an -X job, not running yet, or an older worker without X11 support)", jobID)
|
||||||
|
}
|
||||||
|
|
||||||
|
// x11ServerBase normalizes a configured server URL to the http(s) API base
|
||||||
|
// (attachEndpoint's host extraction, minus the websocket scheme swap).
|
||||||
|
func x11ServerBase(serverURL string) (string, error) {
|
||||||
|
base := strings.TrimRight(strings.TrimSpace(serverURL), "/")
|
||||||
|
if base == "" {
|
||||||
|
return "", fmt.Errorf("no API server URL configured")
|
||||||
|
}
|
||||||
|
if idx := strings.Index(base, "/xce/"); idx != -1 {
|
||||||
|
base = base[:idx]
|
||||||
|
}
|
||||||
|
if !strings.HasPrefix(base, "http://") && !strings.HasPrefix(base, "https://") {
|
||||||
|
return "", fmt.Errorf("unsupported API server URL scheme: %s", serverURL)
|
||||||
|
}
|
||||||
|
return base, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// runX11Session wires the relayed X connection(s) of one job into the local
|
||||||
|
// X server until the session (its job) ends, then cleans the cookie up.
|
||||||
|
func runX11Session(serverURL, token, jobID string) error {
|
||||||
|
cookie, err := x11Cookie(serverURL, token, jobID)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
display := os.Getenv("DISPLAY")
|
||||||
|
addr, isUnix, err := x11DisplayAddr(display)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
authFile, err := x11XauthFile()
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if _, err := exec.LookPath("xauth"); err != nil {
|
||||||
|
return fmt.Errorf("xauth not found in PATH: X11 forwarding needs it to register the session's auth cookie (an X server is present, as DISPLAY=%s suggests)", display)
|
||||||
|
}
|
||||||
|
|
||||||
|
// The worker's X server-side auth check happens against the cookie we
|
||||||
|
// are about to install; the app may connect the moment we open the
|
||||||
|
// relay, so the cookie must be in place first. xauth takes the display
|
||||||
|
// name, auth name and cookie as three separate arguments -- one combined
|
||||||
|
// argument is a "bad add command line".
|
||||||
|
if err := x11Xauth(authFile, "add", display, ".", cookie); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
removed := false
|
||||||
|
removeCookie := func() {
|
||||||
|
if removed {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
removed = true
|
||||||
|
// Best-effort: a leftover entry expires only in that it stops
|
||||||
|
// matching anything once the job's display is gone.
|
||||||
|
_ = x11Xauth(authFile, "remove", display)
|
||||||
|
}
|
||||||
|
defer removeCookie()
|
||||||
|
|
||||||
|
fmt.Printf("attaching X11 session of job <%s> to display %s ...\n", jobID, display)
|
||||||
|
endpoint, err := attachEndpoint(serverURL, jobID)
|
||||||
|
endpoint = strings.Replace(endpoint, "/attach", "/xattach", 1)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
sess := &x11Session{
|
||||||
|
endpoint: endpoint,
|
||||||
|
token: token,
|
||||||
|
addr: addr,
|
||||||
|
isUnix: isUnix,
|
||||||
|
}
|
||||||
|
err = sess.run()
|
||||||
|
|
||||||
|
// The job ending is the normal end of an -X session (the command's exit
|
||||||
|
// tears the relay down); a connection error before any data flowed is
|
||||||
|
// the interesting failure -- report it.
|
||||||
|
if err != nil {
|
||||||
|
if ce, ok := err.(*websocket.CloseError); ok && (ce.Code == 4001 || ce.Code == websocket.CloseNormalClosure) {
|
||||||
|
fmt.Println("<<X11 session ended>>")
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
fmt.Println("<<X11 session ended>>")
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type x11Session struct {
|
||||||
|
endpoint string
|
||||||
|
token string
|
||||||
|
addr string
|
||||||
|
isUnix bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// run dials the relay once and pumps it into the local X server. The first
|
||||||
|
// worker-side accept (the app's XOpenDisplay) rides this connection.
|
||||||
|
func (s *x11Session) run() error {
|
||||||
|
xc, err := x11DialLocal(s.addr, s.isUnix)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("cannot reach the local X server at %s: %s (is the display still alive?)", s.addr, attachErrText(err))
|
||||||
|
}
|
||||||
|
defer xc.Close()
|
||||||
|
|
||||||
|
ws, _, err := websocket.DefaultDialer.Dial(s.endpoint,
|
||||||
|
map[string][]string{"Authorization": {"Bearer " + s.token}})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("xattach failed: [%s]", attachErrText(err))
|
||||||
|
}
|
||||||
|
defer ws.Close()
|
||||||
|
|
||||||
|
done := make(chan struct{})
|
||||||
|
errCh := make(chan error, 2)
|
||||||
|
|
||||||
|
// local X server -> relay: the app talks to its (remote) display here
|
||||||
|
go func() {
|
||||||
|
buf := make([]byte, 32*1024)
|
||||||
|
for {
|
||||||
|
n, err := xc.Read(buf)
|
||||||
|
if n > 0 {
|
||||||
|
if werr := ws.WriteMessage(websocket.BinaryMessage, buf[:n]); werr != nil {
|
||||||
|
errCh <- werr
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
if n == 0 {
|
||||||
|
errCh <- err // X connection closed: the app is done with it
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
// relay -> local X server
|
||||||
|
go func() {
|
||||||
|
for {
|
||||||
|
mt, data, err := ws.ReadMessage()
|
||||||
|
if err != nil {
|
||||||
|
select {
|
||||||
|
case <-done:
|
||||||
|
default:
|
||||||
|
errCh <- err
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if mt != websocket.BinaryMessage {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, err := xc.Write(data); err != nil {
|
||||||
|
errCh <- err
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
// Either side ending tears both down; report the first real error (the
|
||||||
|
// websocket close usually follows the app exiting, so a bare 1006 with
|
||||||
|
// the X side already gone is a normal end).
|
||||||
|
select {
|
||||||
|
case err := <-errCh:
|
||||||
|
close(done)
|
||||||
|
if isX11NormalEnd(err) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
case <-done:
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// isX11NormalEnd reports whether err is the session ending the way an
|
||||||
|
// ssh -X session does: the app exited (X connection EOF -> relay torn down
|
||||||
|
// -> abrupt websocket close) or the job finished (policy close 4001).
|
||||||
|
func isX11NormalEnd(err error) bool {
|
||||||
|
if err == nil {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
if websocket.IsCloseError(err, 4001, websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
// A bare "unexpected EOF"/io.ErrUnexpectedEOF from a torn-down TCP leg
|
||||||
|
// is the app-exit race, same as attach's 1006-after-output handling.
|
||||||
|
return err == net.ErrClosed || strings.Contains(err.Error(), "unexpected EOF")
|
||||||
|
}
|
||||||
|
|
||||||
|
// x11DialLocal opens one connection to the local X server (unix socket or
|
||||||
|
// TCP depending on how DISPLAY points at it).
|
||||||
|
func x11DialLocal(addr string, isUnix bool) (net.Conn, error) {
|
||||||
|
if isUnix {
|
||||||
|
return net.DialTimeout("unix", addr, 5*time.Second)
|
||||||
|
}
|
||||||
|
return net.DialTimeout("tcp", addr, 5*time.Second)
|
||||||
|
}
|
||||||
182
src/frontend/xcecli/cmd/jobs/x11_test.go
Normal file
182
src/frontend/xcecli/cmd/jobs/x11_test.go
Normal file
|
|
@ -0,0 +1,182 @@
|
||||||
|
package jobs
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"xcecli/models"
|
||||||
|
)
|
||||||
|
|
||||||
|
// -X maps to interactive=2 (XJOB) with the x11 exdata key selecting the
|
||||||
|
// forwarding backend worker-side; the interactive enum itself stays
|
||||||
|
// untouched (master/worker ABI). The matrix runs the same interactive
|
||||||
|
// defaults the cobra flow applies before BuildJobRequest.
|
||||||
|
func TestBuildJobRequest_X11FlagMatrix(t *testing.T) {
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
interactive bool
|
||||||
|
useTTY bool
|
||||||
|
useX11 bool
|
||||||
|
useXJob bool
|
||||||
|
useBrowse bool
|
||||||
|
wantInter int
|
||||||
|
wantX11 int
|
||||||
|
}{
|
||||||
|
{"x11 job (-I -X)", true, false, true, false, false, 2, 1},
|
||||||
|
{"bare -I unchanged", true, false, false, false, false, 1, 0},
|
||||||
|
{"tty unchanged", true, true, false, false, false, 1, 0},
|
||||||
|
{"xjob unchanged", true, false, false, true, false, 2, 0},
|
||||||
|
}
|
||||||
|
for _, tc := range cases {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
js := NewJobSubmitter(nil)
|
||||||
|
js.command = "xclock"
|
||||||
|
js.interactive = tc.interactive
|
||||||
|
js.useTTY = tc.useTTY
|
||||||
|
js.useX11 = tc.useX11
|
||||||
|
js.useXJob = tc.useXJob
|
||||||
|
js.useBrowse = tc.useBrowse
|
||||||
|
if tc.interactive {
|
||||||
|
if err := js.applyInteractiveDefaults(); err != nil {
|
||||||
|
t.Fatalf("applyInteractiveDefaults: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
req, err := js.BuildJobRequest()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("BuildJobRequest: %v", err)
|
||||||
|
}
|
||||||
|
if req.Interactive != tc.wantInter {
|
||||||
|
t.Errorf("Interactive = %d, want %d", req.Interactive, tc.wantInter)
|
||||||
|
}
|
||||||
|
if req.X11 != tc.wantX11 {
|
||||||
|
t.Errorf("X11 = %d, want %d", req.X11, tc.wantX11)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// -I -X conflicts with the TTY remap and with --browse: the interactive
|
||||||
|
// defaults must reject the combinations before anything is submitted.
|
||||||
|
func TestInteractiveDefaults_X11Conflicts(t *testing.T) {
|
||||||
|
js := NewJobSubmitter(nil)
|
||||||
|
js.interactive = true
|
||||||
|
js.useX11 = true
|
||||||
|
js.useBrowse = true
|
||||||
|
if err := js.applyInteractiveDefaults(); err == nil {
|
||||||
|
t.Fatal("--x11 with --browse should be rejected")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestInteractiveDefaults_X11Remap(t *testing.T) {
|
||||||
|
js := NewJobSubmitter(nil)
|
||||||
|
js.interactive = true
|
||||||
|
js.useX11 = true
|
||||||
|
if err := js.applyInteractiveDefaults(); err != nil {
|
||||||
|
t.Fatalf("applyInteractiveDefaults: %v", err)
|
||||||
|
}
|
||||||
|
if js.useTTY {
|
||||||
|
t.Error("useTTY should be cleared for -X (not a terminal session)")
|
||||||
|
}
|
||||||
|
if !js.useXJob {
|
||||||
|
t.Error("useXJob should be set for -X (interactive=2)")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Bare -I keeps the TTY attach remap when -X is absent.
|
||||||
|
func TestInteractiveDefaults_BareTTYUnchanged(t *testing.T) {
|
||||||
|
js := NewJobSubmitter(nil)
|
||||||
|
js.interactive = true
|
||||||
|
if err := js.applyInteractiveDefaults(); err != nil {
|
||||||
|
t.Fatalf("applyInteractiveDefaults: %v", err)
|
||||||
|
}
|
||||||
|
if !js.useTTY || js.useXJob {
|
||||||
|
t.Errorf("bare -I = (tty=%v, xjob=%v), want (true, false)", js.useTTY, js.useXJob)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// DISPLAY parsing: local unix socket forms, TCP forms, and the errors that
|
||||||
|
// keep -X from silently pointing at something that is not this display.
|
||||||
|
func TestX11DisplayAddr(t *testing.T) {
|
||||||
|
cases := []struct {
|
||||||
|
display string
|
||||||
|
wantAddr string
|
||||||
|
wantUnix bool
|
||||||
|
wantErr bool
|
||||||
|
}{
|
||||||
|
{":0", "/tmp/.X11-unix/X0", true, false},
|
||||||
|
{":12.0", "/tmp/.X11-unix/X12", true, false},
|
||||||
|
{"unix:0", "/tmp/.X11-unix/X0", true, false},
|
||||||
|
{"localhost:0", "localhost:6000", false, false},
|
||||||
|
{"localhost:10.0", "localhost:6010", false, false}, // ssh -X forwarded display
|
||||||
|
{"", "", false, true},
|
||||||
|
{"bogus", "", false, true},
|
||||||
|
{":", "", false, true},
|
||||||
|
{":x", "", false, true},
|
||||||
|
}
|
||||||
|
for _, tc := range cases {
|
||||||
|
t.Run(tc.display, func(t *testing.T) {
|
||||||
|
addr, isUnix, err := x11DisplayAddr(tc.display)
|
||||||
|
if tc.wantErr {
|
||||||
|
if err == nil {
|
||||||
|
t.Fatalf("DISPLAY %q: want error, got %q", tc.display, addr)
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("DISPLAY %q: %v", tc.display, err)
|
||||||
|
}
|
||||||
|
if addr != tc.wantAddr || isUnix != tc.wantUnix {
|
||||||
|
t.Errorf("DISPLAY %q = (%q, %v), want (%q, %v)", tc.display, addr, isUnix, tc.wantAddr, tc.wantUnix)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The cookie fetch reads the job detail's x11_cookie field (the apiserver
|
||||||
|
// extracts it from the worker's start-report exdata).
|
||||||
|
func TestX11Cookie_FromJobDetail(t *testing.T) {
|
||||||
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
if r.URL.Query().Get("fields") != "x11_cookie" {
|
||||||
|
t.Errorf("fields = %q, want x11_cookie", r.URL.Query().Get("fields"))
|
||||||
|
}
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
_ = json.NewEncoder(w).Encode(models.JobsResponse{
|
||||||
|
Code: 0,
|
||||||
|
Data: struct {
|
||||||
|
Count int `json:"count"`
|
||||||
|
Rows []models.QueryJob `json:"rows"`
|
||||||
|
Page_token string `json:"page_token"`
|
||||||
|
}{Rows: []models.QueryJob{{JobStatus: "RUN", X11Cookie: "aabbccdd00112233"}}},
|
||||||
|
})
|
||||||
|
}))
|
||||||
|
t.Cleanup(srv.Close)
|
||||||
|
cookie, err := x11Cookie(srv.URL, "tok", "123")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("x11Cookie: %v", err)
|
||||||
|
}
|
||||||
|
if cookie != "aabbccdd00112233" {
|
||||||
|
t.Errorf("cookie = %q", cookie)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A job without a cookie (wrong kind, older worker) must fail with a hint,
|
||||||
|
// not with an empty cookie that xauth would choke on.
|
||||||
|
func TestX11Cookie_Missing(t *testing.T) {
|
||||||
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
_ = json.NewEncoder(w).Encode(models.JobsResponse{
|
||||||
|
Code: 0,
|
||||||
|
Data: struct {
|
||||||
|
Count int `json:"count"`
|
||||||
|
Rows []models.QueryJob `json:"rows"`
|
||||||
|
Page_token string `json:"page_token"`
|
||||||
|
}{Rows: []models.QueryJob{{JobStatus: "RUN"}}},
|
||||||
|
})
|
||||||
|
}))
|
||||||
|
t.Cleanup(srv.Close)
|
||||||
|
if _, err := x11Cookie(srv.URL, "tok", "123"); err == nil {
|
||||||
|
t.Fatal("want error for a job without x11_cookie")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -191,6 +191,9 @@ type QueryJob struct {
|
||||||
UniqueJobID int64 `json:"unique_job_id"` //
|
UniqueJobID int64 `json:"unique_job_id"` //
|
||||||
ExtendData JobExtendData `json:"extend_data"`
|
ExtendData JobExtendData `json:"extend_data"`
|
||||||
InteractiveJobURL string `json:"interactive_job_url"`
|
InteractiveJobURL string `json:"interactive_job_url"`
|
||||||
|
// X11Cookie carries the per-job MIT-MAGIC-COOKIE-1 of an -X job,
|
||||||
|
// extracted from the worker's start-report exdata by the apiserver.
|
||||||
|
X11Cookie string `json:"x11_cookie"`
|
||||||
PGIDs IntSlice `json:"pgids"`
|
PGIDs IntSlice `json:"pgids"`
|
||||||
PID IntSlice `json:"pids"`
|
PID IntSlice `json:"pids"`
|
||||||
LaunchType string `json:"launch_type"`
|
LaunchType string `json:"launch_type"`
|
||||||
|
|
@ -348,6 +351,11 @@ type JobSubmitRequest struct {
|
||||||
// may reap the ttyd when no client ever connects. 0 (default) keeps the
|
// may reap the ttyd when no client ever connects. 0 (default) keeps the
|
||||||
// web-viewed semantics.
|
// web-viewed semantics.
|
||||||
Attach int `json:"attach,omitempty"`
|
Attach int `json:"attach,omitempty"`
|
||||||
|
// X11 (exdata key "x11") selects the X11-forwarding backend for an XJOB
|
||||||
|
// (interactive=2) job: the worker exports DISPLAY to its forwarding port
|
||||||
|
// instead of starting VNC, and the app's X connection is relayed to the
|
||||||
|
// user's local X server through the apiserver xattach websocket.
|
||||||
|
X11 int `json:"x11,omitempty"`
|
||||||
WarningTimePeriod int `json:"warning_time_period,omitempty"`
|
WarningTimePeriod int `json:"warning_time_period,omitempty"`
|
||||||
WarningAction string `json:"warning_action,omitempty"`
|
WarningAction string `json:"warning_action,omitempty"`
|
||||||
DataSpecFile string `json:"data_spec_file,omitempty"`
|
DataSpecFile string `json:"data_spec_file,omitempty"`
|
||||||
|
|
|
||||||
|
|
@ -5,6 +5,11 @@
|
||||||
#define URL_MAX_LENGTH 256
|
#define URL_MAX_LENGTH 256
|
||||||
#define TTY_END_PORT 10000
|
#define TTY_END_PORT 10000
|
||||||
#define XJOB_END_PORT 5999
|
#define XJOB_END_PORT 5999
|
||||||
|
/* X11 forwarding (-X jobs, exdata "x11":1): the worker listens on this
|
||||||
|
* segment and exports DISPLAY=:<port-6000> to the job; the app's X
|
||||||
|
* connections are relayed to the user's X server through the apiserver. */
|
||||||
|
#define X11_START_PORT 6100
|
||||||
|
#define X11_END_PORT 6199
|
||||||
|
|
||||||
typedef struct {
|
typedef struct {
|
||||||
int tty_port;
|
int tty_port;
|
||||||
|
|
@ -17,5 +22,6 @@ int check_port(port_info_t port, int interactive_mode);
|
||||||
char* generate_cmd(char* job_cmd, port_info_t port, int interactive_mode);
|
char* generate_cmd(char* job_cmd, port_info_t port, int interactive_mode);
|
||||||
char* generate_url(port_info_t port, int interactive_mode);
|
char* generate_url(port_info_t port, int interactive_mode);
|
||||||
int export_display_env(int port);
|
int export_display_env(int port);
|
||||||
|
int export_x11_display_env(int port);
|
||||||
|
|
||||||
#endif /* _INTER_JOB_UTILS_H */
|
#endif /* _INTER_JOB_UTILS_H */
|
||||||
|
|
|
||||||
|
|
@ -28,11 +28,19 @@ typedef struct {
|
||||||
* command (xcecli attach path). The worker may reap the ttyd when no
|
* command (xcecli attach path). The worker may reap the ttyd when no
|
||||||
* client ever connects; web-viewed jobs (0) keep it for a later browser. */
|
* client ever connects; web-viewed jobs (0) keep it for a later browser. */
|
||||||
int attach_mode;
|
int attach_mode;
|
||||||
|
/* x11=1 (exdata key "x11"): XJOB job running as X11 forwarding -- the
|
||||||
|
* worker exports DISPLAY to its forwarding port and reports a per-job
|
||||||
|
* xauth cookie instead of starting VNC (xcecli -I -X). */
|
||||||
|
int x11_mode;
|
||||||
int count;
|
int count;
|
||||||
} transfer_info_t;
|
} transfer_info_t;
|
||||||
|
|
||||||
typedef struct {
|
typedef struct {
|
||||||
char* url;
|
char* url;
|
||||||
|
/* Per-job MIT-MAGIC-COOKIE-1 (hex) for X11-forwarding jobs, generated
|
||||||
|
* with xauth and reported to the apiserver so the user's client can
|
||||||
|
* register it with their Xauthority. NULL otherwise. */
|
||||||
|
char* x11cookie;
|
||||||
} start_exdata_t;
|
} start_exdata_t;
|
||||||
|
|
||||||
bool parse_exdata(const char* exdata, env_var_t** env_vars,
|
bool parse_exdata(const char* exdata, env_var_t** env_vars,
|
||||||
|
|
|
||||||
|
|
@ -326,3 +326,35 @@ int export_display_env(int port)
|
||||||
"Exported DISPLAY environment variable: %s", display_env);
|
"Exported DISPLAY environment variable: %s", display_env);
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @brief Export DISPLAY for an X11-forwarding job: the app's X connections
|
||||||
|
* go to the worker's forwarding port (rendering happens on the user's
|
||||||
|
* own X server via the apiserver relay, ssh -X style).
|
||||||
|
* @param port The forwarding port number (X11_START_PORT..X11_END_PORT).
|
||||||
|
* @return 0 on success, -1 on failure.
|
||||||
|
*/
|
||||||
|
int export_x11_display_env(int port)
|
||||||
|
{
|
||||||
|
if (port < 6000 || port - 6000 > 255) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
|
"Invalid X11 forwarding port %d for DISPLAY", port);
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
char display_env[32];
|
||||||
|
snprintf(display_env, sizeof(display_env), ":%d", port - 6000);
|
||||||
|
|
||||||
|
if (setenv("DISPLAY", display_env, 1) != 0) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
|
"Failed to set DISPLAY environment variable for "
|
||||||
|
"X11 forwarding");
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
|
||||||
|
"Exported DISPLAY environment variable for X11 "
|
||||||
|
"forwarding: %s",
|
||||||
|
display_env);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -252,6 +252,22 @@ bool parse_exdata(const char* exdata, env_var_t** env_vars,
|
||||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Parsed attach mode: %d",
|
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Parsed attach mode: %d",
|
||||||
transfer_infos->attach_mode);
|
transfer_infos->attach_mode);
|
||||||
}
|
}
|
||||||
|
// --- Handle 'x11' ---
|
||||||
|
else if (strcmp(key, "x11") == 0) {
|
||||||
|
if (cJSON_IsString(item) && item->valuestring != NULL) {
|
||||||
|
transfer_infos->x11_mode = atoi(item->valuestring);
|
||||||
|
} else if (cJSON_IsNumber(item)) {
|
||||||
|
transfer_infos->x11_mode = item->valueint;
|
||||||
|
} else {
|
||||||
|
XCECD_WORKER_LOG(
|
||||||
|
XCE_LOG_ERROR,
|
||||||
|
"Error: 'x11' is neither a string nor a number.");
|
||||||
|
success = false;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Parsed x11 mode: %d",
|
||||||
|
transfer_infos->x11_mode);
|
||||||
|
}
|
||||||
// --- Skip 'limits' ---
|
// --- Skip 'limits' ---
|
||||||
else if (strcmp(key, "limits") == 0) {
|
else if (strcmp(key, "limits") == 0) {
|
||||||
// limits will be handled in parse_limit
|
// limits will be handled in parse_limit
|
||||||
|
|
@ -525,6 +541,9 @@ char* generate_exdata(const start_exdata_t* start_info)
|
||||||
}
|
}
|
||||||
|
|
||||||
cJSON_AddStringToObject(root, "url", start_info->url);
|
cJSON_AddStringToObject(root, "url", start_info->url);
|
||||||
|
if (start_info->x11cookie) {
|
||||||
|
cJSON_AddStringToObject(root, "x11cookie", start_info->x11cookie);
|
||||||
|
}
|
||||||
|
|
||||||
char* exdata_str = cJSON_PrintUnformatted(root);
|
char* exdata_str = cJSON_PrintUnformatted(root);
|
||||||
cJSON_Delete(root);
|
cJSON_Delete(root);
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,5 @@
|
||||||
#include "job_signal.h"
|
#include "job_signal.h"
|
||||||
|
#include "job_output_path.h"
|
||||||
#include "inter_job_utils.h"
|
#include "inter_job_utils.h"
|
||||||
#include "process_manager.h"
|
#include "process_manager.h"
|
||||||
#include "static_array.h"
|
#include "static_array.h"
|
||||||
|
|
@ -8,12 +9,18 @@
|
||||||
#include "xcecd_exdata.h"
|
#include "xcecd_exdata.h"
|
||||||
#include "xcecd_utils.h"
|
#include "xcecd_utils.h"
|
||||||
#include "xcelog.h"
|
#include "xcelog.h"
|
||||||
|
#include <arpa/inet.h>
|
||||||
#include <dirent.h>
|
#include <dirent.h>
|
||||||
|
#include <errno.h>
|
||||||
|
#include <fcntl.h>
|
||||||
|
#include <netinet/in.h>
|
||||||
|
#include <netinet/tcp.h>
|
||||||
#include <signal.h>
|
#include <signal.h>
|
||||||
#include <stdbool.h>
|
#include <stdbool.h>
|
||||||
#include <stdio.h>
|
#include <stdio.h>
|
||||||
#include <stdlib.h>
|
#include <stdlib.h>
|
||||||
#include <string.h>
|
#include <string.h>
|
||||||
|
#include <sys/socket.h>
|
||||||
#include <sys/types.h>
|
#include <sys/types.h>
|
||||||
#include <sys/wait.h>
|
#include <sys/wait.h>
|
||||||
#include <unistd.h>
|
#include <unistd.h>
|
||||||
|
|
@ -26,6 +33,12 @@
|
||||||
#define LOG_SIZE (1024 * 1024 * 100) // 100 MB
|
#define LOG_SIZE (1024 * 1024 * 100) // 100 MB
|
||||||
#define LOG_LEVEL 3
|
#define LOG_LEVEL 3
|
||||||
|
|
||||||
|
/* X11 forwarding (-X jobs): the worker listens for the app's X connections
|
||||||
|
* and exports DISPLAY to the listening port; rendering happens on the
|
||||||
|
* user's own X server via the apiserver xattach relay (ssh -X model). One
|
||||||
|
* relayed websocket carries one X connection. */
|
||||||
|
#define X11_NO_CLIENT_MS 60000 /* reap when no X connection ever arrived */
|
||||||
|
|
||||||
typedef struct exdata {
|
typedef struct exdata {
|
||||||
env_var_t* env_vars;
|
env_var_t* env_vars;
|
||||||
transfer_info_t* transfer_info;
|
transfer_info_t* transfer_info;
|
||||||
|
|
@ -69,6 +82,37 @@ static bool tty_attach_saw_child = false;
|
||||||
static int64_t tty_attach_child_gone_ms = 0;
|
static int64_t tty_attach_child_gone_ms = 0;
|
||||||
static const char* tty_reap_reason = NULL;
|
static const char* tty_reap_reason = NULL;
|
||||||
|
|
||||||
|
/* One end of a bridged X11 session: either the job app's X connection or
|
||||||
|
* the apiserver xattach relay connection. The two are told apart by source
|
||||||
|
* address, not arrival order: the app always runs on this host and its
|
||||||
|
* DISPLAY is always ":N", so app connections come from loopback, while the
|
||||||
|
* relay dials the reported x11://<local_ip>:<port> URL from the apiserver
|
||||||
|
* host (which is never loopback, even when apiserver and worker share a
|
||||||
|
* host). Byte pumps are identical for both roles. */
|
||||||
|
#define X11_STASH_SIZE 32768
|
||||||
|
typedef struct {
|
||||||
|
int fd; /* connection fd, -1 = open slot */
|
||||||
|
bool relay; /* true = apiserver relay leg */
|
||||||
|
int len; /* bytes stashed awaiting the peer */
|
||||||
|
char stash[X11_STASH_SIZE];
|
||||||
|
} x11_leg_t;
|
||||||
|
|
||||||
|
/* X11 forwarding session state (one per-job worker process, so a single
|
||||||
|
* static session is enough -- same shape as the VNC branch). */
|
||||||
|
typedef struct {
|
||||||
|
int listen_fd; /* forwarding-port listener, -1 = off */
|
||||||
|
int x11_port; /* TCP port DISPLAY points at */
|
||||||
|
volatile sig_atomic_t accepts; /* total accepted X connections */
|
||||||
|
bool saw_client; /* app+relay bridge ever established */
|
||||||
|
uv_timer_t no_client_timer;
|
||||||
|
bool timer_ready;
|
||||||
|
char display[32]; /* ":N" exported to the job */
|
||||||
|
char cookie_hex[65]; /* per-job MIT-MAGIC-COOKIE-1 */
|
||||||
|
char auth_file[256]; /* per-job Xauthority file, "" = none */
|
||||||
|
x11_leg_t legs[2]; /* the bridged pair */
|
||||||
|
} x11_forward_t;
|
||||||
|
static x11_forward_t x11_fwd = { .listen_fd = -1 };
|
||||||
|
|
||||||
// process manager will share the same event loop with ipc
|
// process manager will share the same event loop with ipc
|
||||||
uv_loop_t* uv_loop = NULL;
|
uv_loop_t* uv_loop = NULL;
|
||||||
|
|
||||||
|
|
@ -77,6 +121,12 @@ static void worker_reconnect_callback(bool connected);
|
||||||
static int xjob_start_vnc();
|
static int xjob_start_vnc();
|
||||||
static void xjob_stop_vnc();
|
static void xjob_stop_vnc();
|
||||||
static void tty_stop_attach_session();
|
static void tty_stop_attach_session();
|
||||||
|
static int x11_start_forwarding();
|
||||||
|
static void x11_stop_forwarding();
|
||||||
|
static void x11_accept_and_pump();
|
||||||
|
static void x11_no_client_callback(uv_timer_t* timer);
|
||||||
|
static void x11_no_client_timer_start();
|
||||||
|
static void x11_no_client_timer_stop();
|
||||||
static bool check_process_status(pid_t target_pid, int expected_signal,
|
static bool check_process_status(pid_t target_pid, int expected_signal,
|
||||||
int timeout_seconds);
|
int timeout_seconds);
|
||||||
static pid_t ttyd_real_pid(pid_t monitored_pid);
|
static pid_t ttyd_real_pid(pid_t monitored_pid);
|
||||||
|
|
@ -537,7 +587,12 @@ static void process_exit_callback(int pid, int exit_status, int term_signal,
|
||||||
|
|
||||||
/* Job post-exec start here (blocked) */
|
/* Job post-exec start here (blocked) */
|
||||||
if (XJOB == mydata.transfer_info->interactive_mode) {
|
if (XJOB == mydata.transfer_info->interactive_mode) {
|
||||||
xjob_stop_vnc();
|
if (1 == mydata.transfer_info->x11_mode) {
|
||||||
|
x11_no_client_timer_stop();
|
||||||
|
x11_stop_forwarding();
|
||||||
|
} else {
|
||||||
|
xjob_stop_vnc();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/* attach-mode TTY job: the session ends with the command (web-viewed
|
/* attach-mode TTY job: the session ends with the command (web-viewed
|
||||||
|
|
@ -630,6 +685,450 @@ static void check_and_reconnect()
|
||||||
reconnect_retry_limit = 5;
|
reconnect_retry_limit = 5;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* ---- X11 forwarding (XJOB + exdata x11=1, the xcecli -I -X path) ----
|
||||||
|
*
|
||||||
|
* The job command starts immediately (no daemon waits for a client) with
|
||||||
|
* DISPLAY pointing at the forwarding port. The app's XOpenDisplay connects
|
||||||
|
* to the listener; the apiserver's /xattach relay picks each accepted
|
||||||
|
* connection up as one websocket and pumps the bytes to the user's X
|
||||||
|
* server, where the windows render. Auth follows ssh -X: a per-job
|
||||||
|
* MIT-MAGIC-COOKIE-1 is generated with xauth (the only X-aware helper
|
||||||
|
* guaranteed around an X11 user setup) and reported through the start
|
||||||
|
* exdata so the user's client can register it with their Xauthority.
|
||||||
|
*
|
||||||
|
* The no-client timer covers the one residue shape an -X job has: a job
|
||||||
|
* whose X connection never arrives (the CLI died before relaying) would
|
||||||
|
* otherwise sit in RUN forever -- the app blocks inside XOpenDisplay and
|
||||||
|
* never exits on its own. Once any connection was accepted the session is
|
||||||
|
* the app's and X's to end (a dropped relay surfaces as an X IO error and
|
||||||
|
* the app exits), so the timer is cancelled on the first accept. */
|
||||||
|
|
||||||
|
/* 32 hex chars = 128 bits of cookie, exactly xauth's MIT-MAGIC-COOKIE-1
|
||||||
|
* format. Reads /dev/urandom; retries a few times before giving up. */
|
||||||
|
static bool x11_generate_cookie(char* out, size_t outlen)
|
||||||
|
{
|
||||||
|
unsigned char raw[16];
|
||||||
|
static const char hex[] = "0123456789abcdef";
|
||||||
|
|
||||||
|
if (outlen < sizeof(raw) * 2 + 1) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
int fd = open("/dev/urandom", O_RDONLY);
|
||||||
|
if (fd < 0) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
size_t got = 0;
|
||||||
|
bool ok = true;
|
||||||
|
while (got < sizeof(raw)) {
|
||||||
|
ssize_t n = read(fd, (char*)raw + got, sizeof(raw) - got);
|
||||||
|
if (n <= 0) {
|
||||||
|
if (n < 0 && errno == EINTR) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
ok = false;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
got += (size_t)n;
|
||||||
|
}
|
||||||
|
close(fd);
|
||||||
|
if (!ok) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
for (size_t i = 0; i < sizeof(raw); i++) {
|
||||||
|
out[i * 2] = hex[raw[i] >> 4];
|
||||||
|
out[i * 2 + 1] = hex[raw[i] & 0xF];
|
||||||
|
}
|
||||||
|
out[sizeof(raw) * 2] = '\0';
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* Create a fresh Xauthority file for this job and register the generated
|
||||||
|
* cookie in it under our display number. The per-job worker runs on a headless
|
||||||
|
* node -- there is no session Xauthority to inherit (getenv("XAUTHORITY") is
|
||||||
|
* normally empty here), so the file is ours from scratch, named after the job
|
||||||
|
* output dir convention in /tmp. The job processes inherit the worker's
|
||||||
|
* environment, so exporting XAUTHORITY alongside DISPLAY (the same channel
|
||||||
|
* export_x11_display_env uses) makes the app's XOpenDisplay present exactly
|
||||||
|
* this cookie. */
|
||||||
|
static int x11_install_cookie(uint64_t jobid)
|
||||||
|
{
|
||||||
|
char cmd[MAXLINELEN];
|
||||||
|
char jobidStr[64];
|
||||||
|
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
|
||||||
|
|
||||||
|
snprintf(x11_fwd.auth_file, sizeof(x11_fwd.auth_file),
|
||||||
|
XCECD_JOB_OUTPUT_DIR "/job_%s_xauth", jobidStr);
|
||||||
|
unlink(x11_fwd.auth_file); /* a re-run job must not inherit an old entry */
|
||||||
|
|
||||||
|
snprintf(cmd, sizeof(cmd),
|
||||||
|
"xauth -f '%s' add ':%d' . %s 2>/dev/null && chmod 600 '%s'",
|
||||||
|
x11_fwd.auth_file, x11_fwd.x11_port - 6000, x11_fwd.cookie_hex,
|
||||||
|
x11_fwd.auth_file);
|
||||||
|
/* xce-fork-ok: xcecd_worker is single-threaded (tech.md) */
|
||||||
|
int rc = system(cmd);
|
||||||
|
if (rc != 0 || access(x11_fwd.auth_file, R_OK) != 0) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
|
"xauth add failed for X11 forwarding display :%d "
|
||||||
|
"(rc=%d)",
|
||||||
|
x11_fwd.x11_port - 6000, rc);
|
||||||
|
unlink(x11_fwd.auth_file);
|
||||||
|
x11_fwd.auth_file[0] = '\0';
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (setenv("XAUTHORITY", x11_fwd.auth_file, 1) != 0) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
|
"Failed to set XAUTHORITY for X11 forwarding: %s",
|
||||||
|
strerror(errno));
|
||||||
|
unlink(x11_fwd.auth_file);
|
||||||
|
x11_fwd.auth_file[0] = '\0';
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
static void x11_remove_cookie()
|
||||||
|
{
|
||||||
|
if (x11_fwd.auth_file[0] != '\0') {
|
||||||
|
unlink(x11_fwd.auth_file);
|
||||||
|
x11_fwd.auth_file[0] = '\0';
|
||||||
|
}
|
||||||
|
x11_fwd.cookie_hex[0] = '\0';
|
||||||
|
}
|
||||||
|
|
||||||
|
static int x11_start_forwarding()
|
||||||
|
{
|
||||||
|
char jobidStr[64];
|
||||||
|
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
|
||||||
|
|
||||||
|
x11_fwd.x11_port = port_info.xjob_port; /* X11 rides the XJOB slot */
|
||||||
|
x11_fwd.listen_fd = -1;
|
||||||
|
x11_fwd.accepts = 0;
|
||||||
|
x11_fwd.saw_client = false;
|
||||||
|
x11_fwd.cookie_hex[0] = '\0';
|
||||||
|
x11_fwd.auth_file[0] = '\0';
|
||||||
|
x11_fwd.legs[0].fd = -1;
|
||||||
|
x11_fwd.legs[1].fd = -1;
|
||||||
|
x11_fwd.legs[0].relay = false;
|
||||||
|
x11_fwd.legs[1].relay = false;
|
||||||
|
x11_fwd.legs[0].len = 0;
|
||||||
|
x11_fwd.legs[1].len = 0;
|
||||||
|
|
||||||
|
if (!x11_generate_cookie(x11_fwd.cookie_hex,
|
||||||
|
sizeof(x11_fwd.cookie_hex))) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
|
"Failed to generate the X11 auth cookie for job %s",
|
||||||
|
jobidStr);
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* The listen socket stays bound for the job's lifetime; the accept
|
||||||
|
* handshake for the app's X connection must already succeed before the
|
||||||
|
* apiserver relay dials (X11 needs no protocol on accept). */
|
||||||
|
int fd = socket(AF_INET, SOCK_STREAM, 0);
|
||||||
|
if (fd < 0) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_ERROR, "X11 forwarding socket failed: %s",
|
||||||
|
strerror(errno));
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
int one = 1;
|
||||||
|
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
|
||||||
|
/* accept() must never block: the main loop shares this fd with the uv
|
||||||
|
* loop and a 1 ms cadence, so a blocking accept would freeze the whole
|
||||||
|
* worker (timers included) the moment the backlog runs dry */
|
||||||
|
int fl = fcntl(fd, F_GETFL, 0);
|
||||||
|
if (fl < 0 || fcntl(fd, F_SETFL, fl | O_NONBLOCK) != 0) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
|
"X11 forwarding listener fcntl failed: %s",
|
||||||
|
strerror(errno));
|
||||||
|
close(fd);
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
struct sockaddr_in addr;
|
||||||
|
memset(&addr, 0, sizeof(addr));
|
||||||
|
addr.sin_family = AF_INET;
|
||||||
|
addr.sin_port = htons(x11_fwd.x11_port);
|
||||||
|
addr.sin_addr.s_addr = INADDR_ANY;
|
||||||
|
if (bind(fd, (struct sockaddr*)&addr, sizeof(addr)) != 0 || listen(fd, 8) != 0) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
|
"X11 forwarding listen on port %d failed: %s",
|
||||||
|
x11_fwd.x11_port, strerror(errno));
|
||||||
|
close(fd);
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
x11_fwd.listen_fd = fd;
|
||||||
|
|
||||||
|
if (x11_install_cookie(jobid) != 0) {
|
||||||
|
x11_stop_forwarding();
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (export_x11_display_env(x11_fwd.x11_port) != 0) {
|
||||||
|
x11_stop_forwarding();
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_INFO,
|
||||||
|
"X11 forwarding ready for job %s: port %d, display :%d, "
|
||||||
|
"authority %s",
|
||||||
|
jobidStr, x11_fwd.x11_port, x11_fwd.x11_port - 6000,
|
||||||
|
x11_fwd.auth_file);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
static void x11_stop_forwarding()
|
||||||
|
{
|
||||||
|
if (x11_fwd.listen_fd >= 0) {
|
||||||
|
close(x11_fwd.listen_fd);
|
||||||
|
x11_fwd.listen_fd = -1;
|
||||||
|
}
|
||||||
|
for (int i = 0; i < 2; i++) {
|
||||||
|
if (x11_fwd.legs[i].fd >= 0) {
|
||||||
|
close(x11_fwd.legs[i].fd);
|
||||||
|
x11_fwd.legs[i].fd = -1;
|
||||||
|
}
|
||||||
|
x11_fwd.legs[i].len = 0;
|
||||||
|
}
|
||||||
|
x11_remove_cookie();
|
||||||
|
}
|
||||||
|
|
||||||
|
/* One pump step for a bridged connection: drain its stash toward the peer
|
||||||
|
* and pull new bytes into it. Non-blocking; short counts and EAGAIN are
|
||||||
|
* normal -- the stash absorbs bursts the peer cannot take right away, and
|
||||||
|
* a full stash applies backpressure (we simply stop recv'ing until the
|
||||||
|
* peer drains it) instead of dropping or killing the session.
|
||||||
|
* Returns false once the peer is gone (EOF or error), which ends the
|
||||||
|
* session: X11 has no notion of a half-closed connection, so when either
|
||||||
|
* leg dies the whole bridge dies with it (the job then exits on IO error,
|
||||||
|
* or the relay teardown reports the end to the CLI). */
|
||||||
|
static bool x11_pump_leg(x11_leg_t* leg, x11_leg_t* peer)
|
||||||
|
{
|
||||||
|
if (leg->fd < 0) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* flush stashed bytes toward the peer first */
|
||||||
|
if (leg->len > 0) {
|
||||||
|
ssize_t n = send(peer->fd, leg->stash, leg->len,
|
||||||
|
MSG_DONTWAIT | MSG_NOSIGNAL);
|
||||||
|
if (n < 0) {
|
||||||
|
if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_DEBUG,
|
||||||
|
"X11 bridge leg %d send failed: %s", leg->fd,
|
||||||
|
strerror(errno));
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
} else if (n > 0) {
|
||||||
|
if (n < leg->len) {
|
||||||
|
memmove(leg->stash, leg->stash + n, leg->len - n);
|
||||||
|
}
|
||||||
|
leg->len -= n;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
char buf[8192];
|
||||||
|
for (;;) {
|
||||||
|
if (leg->len >= X11_STASH_SIZE) {
|
||||||
|
return true; /* backpressure: wait for the peer to drain */
|
||||||
|
}
|
||||||
|
ssize_t n = recv(leg->fd, buf, sizeof(buf), MSG_DONTWAIT);
|
||||||
|
if (n < 0) {
|
||||||
|
if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "X11 bridge leg %d recv failed: %s",
|
||||||
|
leg->fd, strerror(errno));
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
if (n == 0) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "X11 bridge leg %d closed by peer",
|
||||||
|
leg->fd);
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
int off = 0;
|
||||||
|
while (off < n) {
|
||||||
|
int room = X11_STASH_SIZE - leg->len;
|
||||||
|
if (room <= 0) {
|
||||||
|
break; /* stash full -- drain on a later tick */
|
||||||
|
}
|
||||||
|
int take = (n - off) < room ? (n - off) : room;
|
||||||
|
memcpy(leg->stash + leg->len, buf + off, take);
|
||||||
|
leg->len += take;
|
||||||
|
off += take;
|
||||||
|
|
||||||
|
ssize_t w = send(peer->fd, leg->stash, leg->len,
|
||||||
|
MSG_DONTWAIT | MSG_NOSIGNAL);
|
||||||
|
if (w < 0) {
|
||||||
|
if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_DEBUG,
|
||||||
|
"X11 bridge leg %d send failed: %s", leg->fd,
|
||||||
|
strerror(errno));
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
if (w < leg->len) {
|
||||||
|
memmove(leg->stash, leg->stash + w, leg->len - w);
|
||||||
|
}
|
||||||
|
leg->len -= w;
|
||||||
|
}
|
||||||
|
if (off < n) {
|
||||||
|
return true; /* stash filled mid-buffer -- drain on a later tick */
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/* Classify an accepted connection by source address: the app runs on this
|
||||||
|
* host and its DISPLAY is ":N" (loopback), the apiserver relay dials the
|
||||||
|
* worker's non-loopback IP from the apiserver host. */
|
||||||
|
static bool x11_conn_is_relay(int fd)
|
||||||
|
{
|
||||||
|
struct sockaddr_in sa;
|
||||||
|
socklen_t sl = sizeof(sa);
|
||||||
|
if (getpeername(fd, (struct sockaddr*)&sa, &sl) != 0) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
uint32_t ip = ntohl(sa.sin_addr.s_addr);
|
||||||
|
return ip != INADDR_LOOPBACK;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* Non-blocking accept/pump step for the X11 listener, called from the main
|
||||||
|
* loop. v1 carries exactly one X connection: one app leg (loopback) plus
|
||||||
|
* one apiserver relay leg, bridged once both are in. Anything beyond that
|
||||||
|
* is refused loudly -- xim/clipboard-style secondary connections are out of
|
||||||
|
* scope and silent window-event loss would be worse than a clear failure
|
||||||
|
* (plan 2.4). Classification is by source address so connection arrival
|
||||||
|
* order cannot mispair two racing app connections with each other. */
|
||||||
|
static void x11_accept_and_pump()
|
||||||
|
{
|
||||||
|
if (x11_fwd.listen_fd < 0) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
bool bridged = (x11_fwd.legs[0].fd >= 0 && x11_fwd.legs[1].fd >= 0);
|
||||||
|
|
||||||
|
for (;;) {
|
||||||
|
int fd = accept(x11_fwd.listen_fd, NULL, NULL);
|
||||||
|
if (fd < 0) {
|
||||||
|
if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR
|
||||||
|
&& errno != ECONNABORTED) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
|
"X11 forwarding accept failed: %s",
|
||||||
|
strerror(errno));
|
||||||
|
}
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
int fl = fcntl(fd, F_GETFL, 0);
|
||||||
|
if (fl >= 0) {
|
||||||
|
fcntl(fd, F_SETFL, fl | O_NONBLOCK);
|
||||||
|
}
|
||||||
|
/* TCP_NODELAY: X11 turns interactive (cursor moves, key events) --
|
||||||
|
* tiny packets must not wait for Nagle on either leg */
|
||||||
|
int one = 1;
|
||||||
|
setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
|
||||||
|
|
||||||
|
if (bridged) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_NOTICE,
|
||||||
|
"X11 job app opened a second X connection; "
|
||||||
|
"v1 supports one -- closing it");
|
||||||
|
close(fd);
|
||||||
|
break; /* one refusal per tick keeps the loop responsive */
|
||||||
|
}
|
||||||
|
|
||||||
|
bool relay = x11_conn_is_relay(fd);
|
||||||
|
/* slot 0 = app leg, slot 1 = relay leg; a duplicate of a role we
|
||||||
|
* already hold (e.g. a second app racing in) is refused */
|
||||||
|
int slot = relay ? 1 : 0;
|
||||||
|
if (x11_fwd.legs[slot].fd >= 0) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_NOTICE,
|
||||||
|
"X11 job app opened a second X connection; "
|
||||||
|
"v1 supports one -- closing it");
|
||||||
|
close(fd);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
x11_fwd.legs[slot].fd = fd;
|
||||||
|
x11_fwd.legs[slot].relay = relay;
|
||||||
|
x11_fwd.legs[slot].len = 0;
|
||||||
|
x11_fwd.accepts++;
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "X11 forwarding %s connection accepted",
|
||||||
|
relay ? "relay" : "app");
|
||||||
|
|
||||||
|
if (x11_fwd.legs[0].fd >= 0 && x11_fwd.legs[1].fd >= 0) {
|
||||||
|
bridged = true;
|
||||||
|
x11_fwd.saw_client = true;
|
||||||
|
x11_no_client_timer_stop();
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_INFO,
|
||||||
|
"X11 bridge established on port %d",
|
||||||
|
x11_fwd.x11_port);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (bridged) {
|
||||||
|
if (!x11_pump_leg(&x11_fwd.legs[0], &x11_fwd.legs[1])) {
|
||||||
|
x11_stop_forwarding();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if (!x11_pump_leg(&x11_fwd.legs[1], &x11_fwd.legs[0])) {
|
||||||
|
x11_stop_forwarding();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
static void x11_no_client_callback(uv_timer_t* timer)
|
||||||
|
{
|
||||||
|
(void)timer;
|
||||||
|
|
||||||
|
if (worker_exit_flag
|
||||||
|
|| XJOB != mydata.transfer_info->interactive_mode
|
||||||
|
|| 1 != mydata.transfer_info->x11_mode) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if (x11_fwd.saw_client) {
|
||||||
|
/* an X connection arrived: the session belongs to the app now */
|
||||||
|
x11_no_client_timer_stop();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
pid_t pid = uv_proc_get_pid(&uv_proc_manager);
|
||||||
|
if (pid <= 0) {
|
||||||
|
x11_no_client_timer_stop();
|
||||||
|
return; /* process_exit_callback owns the job */
|
||||||
|
}
|
||||||
|
|
||||||
|
char jobidStr[64];
|
||||||
|
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_NOTICE,
|
||||||
|
"X11 job %s: no X client connected within %d s, reaping "
|
||||||
|
"the job",
|
||||||
|
jobidStr, (int)(X11_NO_CLIENT_MS / 1000));
|
||||||
|
|
||||||
|
tty_reap_reason = "No X11 client connected within 60 s";
|
||||||
|
x11_no_client_timer_stop();
|
||||||
|
/* the listener is not the app's parent: killing the job tree directly
|
||||||
|
* lands us in the regular process_exit_callback path */
|
||||||
|
signal_children_recursive(SIGKILL, pid);
|
||||||
|
kill(pid, SIGKILL);
|
||||||
|
}
|
||||||
|
|
||||||
|
static void x11_no_client_timer_start()
|
||||||
|
{
|
||||||
|
if (!x11_fwd.timer_ready) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
uv_timer_start(&x11_fwd.no_client_timer, x11_no_client_callback,
|
||||||
|
X11_NO_CLIENT_MS, X11_NO_CLIENT_MS);
|
||||||
|
}
|
||||||
|
|
||||||
|
static void x11_no_client_timer_stop()
|
||||||
|
{
|
||||||
|
if (x11_fwd.timer_ready) {
|
||||||
|
uv_timer_stop(&x11_fwd.no_client_timer);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
static int xjob_start_vnc()
|
static int xjob_start_vnc()
|
||||||
{
|
{
|
||||||
char jobidStr[64];
|
char jobidStr[64];
|
||||||
|
|
@ -1090,8 +1589,19 @@ static void worker_job_callback(job_assignment_t* info)
|
||||||
"Parsed exdata for job %s: interactive=%d", jobidStr,
|
"Parsed exdata for job %s: interactive=%d", jobidStr,
|
||||||
mydata.transfer_info->interactive_mode);
|
mydata.transfer_info->interactive_mode);
|
||||||
|
|
||||||
int job_port
|
/* X11-forwarding jobs are interactive=XJOB with exdata x11=1: they take
|
||||||
= check_port(port_info, mydata.transfer_info->interactive_mode);
|
* their port from the forwarding segment (6100-6199) instead of the VNC
|
||||||
|
* one, and generate_cmd passes the raw command through (the job needs no
|
||||||
|
* ttyd wrapper). */
|
||||||
|
int x11_backend = (mydata.transfer_info->interactive_mode == XJOB)
|
||||||
|
&& (1 == mydata.transfer_info->x11_mode);
|
||||||
|
|
||||||
|
int job_port;
|
||||||
|
if (x11_backend) {
|
||||||
|
job_port = find_available_port(X11_START_PORT, X11_END_PORT);
|
||||||
|
} else {
|
||||||
|
job_port = check_port(port_info, mydata.transfer_info->interactive_mode);
|
||||||
|
}
|
||||||
if (job_port < 0) {
|
if (job_port < 0) {
|
||||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
"Failed to check port availability for job %s",
|
"Failed to check port availability for job %s",
|
||||||
|
|
@ -1102,6 +1612,8 @@ static void worker_job_callback(job_assignment_t* info)
|
||||||
|| mydata.transfer_info->interactive_mode == DISPLAY) {
|
|| mydata.transfer_info->interactive_mode == DISPLAY) {
|
||||||
port_info.tty_port = job_port;
|
port_info.tty_port = job_port;
|
||||||
} else if (mydata.transfer_info->interactive_mode == XJOB) {
|
} else if (mydata.transfer_info->interactive_mode == XJOB) {
|
||||||
|
/* X11 forwarding and VNC share the xjob_port slot; the backend
|
||||||
|
* branch decides what listens on it */
|
||||||
port_info.xjob_port = job_port;
|
port_info.xjob_port = job_port;
|
||||||
}
|
}
|
||||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Assigned port %d for job %s",
|
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Assigned port %d for job %s",
|
||||||
|
|
@ -1114,8 +1626,13 @@ static void worker_job_callback(job_assignment_t* info)
|
||||||
exit(1);
|
exit(1);
|
||||||
}
|
}
|
||||||
|
|
||||||
job_cmd = generate_cmd((char*)info->cmd, port_info,
|
if (x11_backend) {
|
||||||
mydata.transfer_info->interactive_mode);
|
/* no ttyd wrapper: the command runs directly with DISPLAY inherited */
|
||||||
|
job_cmd = strdup((char*)info->cmd);
|
||||||
|
} else {
|
||||||
|
job_cmd = generate_cmd((char*)info->cmd, port_info,
|
||||||
|
mydata.transfer_info->interactive_mode);
|
||||||
|
}
|
||||||
|
|
||||||
/* bug-1936: run pre-exec and the job in the submission's cwd, with HOME
|
/* bug-1936: run pre-exec and the job in the submission's cwd, with HOME
|
||||||
* = its exec home; before apply_env_vars() so an explicit --env HOME
|
* = its exec home; before apply_env_vars() so an explicit --env HOME
|
||||||
|
|
@ -1139,7 +1656,33 @@ static void worker_job_callback(job_assignment_t* info)
|
||||||
* for the next job (previously exit(1) here took down the whole
|
* for the next job (previously exit(1) here took down the whole
|
||||||
* per-job worker with no user-visible reason). */
|
* per-job worker with no user-visible reason). */
|
||||||
if (XJOB == mydata.transfer_info->interactive_mode) {
|
if (XJOB == mydata.transfer_info->interactive_mode) {
|
||||||
if (xjob_start_vnc() != 0) {
|
if (1 == mydata.transfer_info->x11_mode) {
|
||||||
|
/* X11 forwarding backend (xcecli -I -X): listen for the app's X
|
||||||
|
* connections and point DISPLAY at the listener instead of
|
||||||
|
* running VNC. */
|
||||||
|
if (x11_start_forwarding() != 0) {
|
||||||
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
|
"Failed to start X11 forwarding for job %s",
|
||||||
|
jobidStr);
|
||||||
|
x11_stop_forwarding(); /* best-effort cleanup of partial start */
|
||||||
|
|
||||||
|
check_and_reconnect();
|
||||||
|
|
||||||
|
job_runtime_info.job_pid = 0;
|
||||||
|
job_runtime_info.job_status = JOB_FAILED;
|
||||||
|
job_runtime_info.worker_status = WORKER_EXITED;
|
||||||
|
job_runtime_info.exit_code = 1;
|
||||||
|
|
||||||
|
ipc_worker_send_status(
|
||||||
|
&job_runtime_info,
|
||||||
|
"Failed to start X11 forwarding (check worker log)");
|
||||||
|
worker_exit_flag = 1;
|
||||||
|
return_code = 1;
|
||||||
|
FREEUP(job_cmd);
|
||||||
|
ipc_worker_run(UV_RUN_NOWAIT);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
} else if (xjob_start_vnc() != 0) {
|
||||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
"Failed to start VNC server for job %s",
|
"Failed to start VNC server for job %s",
|
||||||
jobidStr);
|
jobidStr);
|
||||||
|
|
@ -1204,6 +1747,13 @@ static void worker_job_callback(job_assignment_t* info)
|
||||||
tty_attach_watchdog_start();
|
tty_attach_watchdog_start();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* X11-forwarding jobs: reap when no X connection ever arrives (the
|
||||||
|
* app blocks inside XOpenDisplay and never exits on its own) */
|
||||||
|
if (XJOB == mydata.transfer_info->interactive_mode
|
||||||
|
&& 1 == mydata.transfer_info->x11_mode) {
|
||||||
|
x11_no_client_timer_start();
|
||||||
|
}
|
||||||
|
|
||||||
if (mydata.transfer_info->interactive_mode > 0) {
|
if (mydata.transfer_info->interactive_mode > 0) {
|
||||||
url = generate_url(port_info,
|
url = generate_url(port_info,
|
||||||
mydata.transfer_info->interactive_mode);
|
mydata.transfer_info->interactive_mode);
|
||||||
|
|
@ -1213,6 +1763,23 @@ static void worker_job_callback(job_assignment_t* info)
|
||||||
url);
|
url);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (x11_backend && url) {
|
||||||
|
/* rewrite the XJOB URL ("ip::port") to the forwarding form
|
||||||
|
* the apiserver's xattach relay and the xcecli -X client
|
||||||
|
* expect ("x11://ip:port") -- the double colon is the VNC
|
||||||
|
* websockify convention and not a dialable address */
|
||||||
|
char* dcolon = strstr(url, "::");
|
||||||
|
char* x11url = calloc(1, URL_MAX_LENGTH);
|
||||||
|
if (x11url && dcolon) {
|
||||||
|
snprintf(x11url, URL_MAX_LENGTH, "x11://%.*s:%s",
|
||||||
|
(int)(dcolon - url), url, dcolon + 2);
|
||||||
|
FREEUP(url);
|
||||||
|
url = x11url;
|
||||||
|
} else {
|
||||||
|
FREEUP(x11url);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
start_exdata = calloc(1, sizeof(start_exdata_t));
|
start_exdata = calloc(1, sizeof(start_exdata_t));
|
||||||
if (!start_exdata) {
|
if (!start_exdata) {
|
||||||
XCECD_WORKER_LOG(
|
XCECD_WORKER_LOG(
|
||||||
|
|
@ -1221,6 +1788,13 @@ static void worker_job_callback(job_assignment_t* info)
|
||||||
exit(1);
|
exit(1);
|
||||||
}
|
}
|
||||||
start_exdata->url = url;
|
start_exdata->url = url;
|
||||||
|
/* -X jobs report the per-job X11 auth cookie next to the URL so
|
||||||
|
* the user's client can register it with their Xauthority */
|
||||||
|
if (XJOB == mydata.transfer_info->interactive_mode
|
||||||
|
&& 1 == mydata.transfer_info->x11_mode
|
||||||
|
&& x11_fwd.cookie_hex[0] != '\0') {
|
||||||
|
start_exdata->x11cookie = strdup(x11_fwd.cookie_hex);
|
||||||
|
}
|
||||||
|
|
||||||
start_exdata_info = generate_exdata(start_exdata);
|
start_exdata_info = generate_exdata(start_exdata);
|
||||||
if (!start_exdata_info) {
|
if (!start_exdata_info) {
|
||||||
|
|
@ -1252,7 +1826,10 @@ static void worker_job_callback(job_assignment_t* info)
|
||||||
|
|
||||||
FREEUP(job_cmd);
|
FREEUP(job_cmd);
|
||||||
FREEUP(url);
|
FREEUP(url);
|
||||||
FREEUP(start_exdata);
|
if (start_exdata) {
|
||||||
|
FREEUP(start_exdata->x11cookie);
|
||||||
|
FREEUP(start_exdata);
|
||||||
|
}
|
||||||
|
|
||||||
} else {
|
} else {
|
||||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||||
|
|
@ -1881,10 +2458,18 @@ int main(int argc, char* argv[])
|
||||||
uv_timer_init(uv_loop, &tty_attach_timer);
|
uv_timer_init(uv_loop, &tty_attach_timer);
|
||||||
tty_attach_timer_ready = true;
|
tty_attach_timer_ready = true;
|
||||||
|
|
||||||
|
/* X11-forwarding no-client reaper (x11_no_client_callback) */
|
||||||
|
uv_timer_init(uv_loop, &x11_fwd.no_client_timer);
|
||||||
|
x11_fwd.timer_ready = true;
|
||||||
|
|
||||||
do {
|
do {
|
||||||
|
|
||||||
ipc_worker_run(UV_RUN_NOWAIT);
|
ipc_worker_run(UV_RUN_NOWAIT);
|
||||||
|
|
||||||
|
/* X11 forwarding: accept the app's X connection and the apiserver
|
||||||
|
* relay's connection, then bridge them (non-blocking step) */
|
||||||
|
x11_accept_and_pump();
|
||||||
|
|
||||||
usleep(1000);
|
usleep(1000);
|
||||||
|
|
||||||
} while (worker_exit_flag == 0);
|
} while (worker_exit_flag == 0);
|
||||||
|
|
|
||||||
|
|
@ -149,6 +149,62 @@ static void test_apiserver_nofile_limit_parsed(void)
|
||||||
TEST_ASSERT_EQUAL_UINT64(60, limits.run_limit);
|
TEST_ASSERT_EQUAL_UINT64(60, limits.run_limit);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* x11=1 (exdata key "x11", xcecli -I -X): selects the X11-forwarding
|
||||||
|
* backend for an XJOB job -- DISPLAY goes to the worker's forwarding port,
|
||||||
|
* no VNC. Must tolerate the string form like attach/interactive do. */
|
||||||
|
static void test_x11_key_parsed(void)
|
||||||
|
{
|
||||||
|
TEST_ASSERT_TRUE(parse_exdata(
|
||||||
|
"{\"interactive\":2,\"x11\":1}", &env, &tinfo, 1));
|
||||||
|
TEST_ASSERT_EQUAL_INT(XJOB, tinfo.interactive_mode);
|
||||||
|
TEST_ASSERT_EQUAL_INT(1, tinfo.x11_mode);
|
||||||
|
}
|
||||||
|
|
||||||
|
static void test_x11_key_string_form_parsed(void)
|
||||||
|
{
|
||||||
|
TEST_ASSERT_TRUE(parse_exdata(
|
||||||
|
"{\"interactive\":\"2\",\"x11\":\"1\"}", &env, &tinfo, 1));
|
||||||
|
TEST_ASSERT_EQUAL_INT(XJOB, tinfo.interactive_mode);
|
||||||
|
TEST_ASSERT_EQUAL_INT(1, tinfo.x11_mode);
|
||||||
|
}
|
||||||
|
|
||||||
|
/* x11 on a non-XJOB job is recorded as parsed (the lifecycle branch keys
|
||||||
|
* off both fields together, so it stays inert there) but the parse itself
|
||||||
|
* must not fail the job. */
|
||||||
|
static void test_x11_key_absent_defaults_zero(void)
|
||||||
|
{
|
||||||
|
TEST_ASSERT_TRUE(parse_exdata(
|
||||||
|
"{\"interactive\":2}", &env, &tinfo, 1));
|
||||||
|
TEST_ASSERT_EQUAL_INT(XJOB, tinfo.interactive_mode);
|
||||||
|
TEST_ASSERT_EQUAL_INT(0, tinfo.x11_mode);
|
||||||
|
}
|
||||||
|
|
||||||
|
/* generate_exdata packs the -X job's auth cookie next to the URL; without
|
||||||
|
* a cookie (ttyd/VNC jobs) the exdata stays exactly {"url": ...}. */
|
||||||
|
static void test_generate_exdata_x11cookie(void)
|
||||||
|
{
|
||||||
|
start_exdata_t start = {.url = "x11://10.0.1.5:6100",
|
||||||
|
.x11cookie = "aabbccdd00112233"};
|
||||||
|
char* json = generate_exdata(&start);
|
||||||
|
|
||||||
|
TEST_ASSERT_NOT_NULL(json);
|
||||||
|
TEST_ASSERT_TRUE(strstr(json, "\"url\":\"x11://10.0.1.5:6100\"") != NULL);
|
||||||
|
TEST_ASSERT_TRUE(
|
||||||
|
strstr(json, "\"x11cookie\":\"aabbccdd00112233\"") != NULL);
|
||||||
|
free(json);
|
||||||
|
}
|
||||||
|
|
||||||
|
static void test_generate_exdata_without_cookie(void)
|
||||||
|
{
|
||||||
|
start_exdata_t start = {.url = "http://10.0.1.5:9000",
|
||||||
|
.x11cookie = NULL};
|
||||||
|
char* json = generate_exdata(&start);
|
||||||
|
|
||||||
|
TEST_ASSERT_NOT_NULL(json);
|
||||||
|
TEST_ASSERT_EQUAL_STRING("{\"url\":\"http://10.0.1.5:9000\"}", json);
|
||||||
|
free(json);
|
||||||
|
}
|
||||||
|
|
||||||
int main(void)
|
int main(void)
|
||||||
{
|
{
|
||||||
UNITY_BEGIN();
|
UNITY_BEGIN();
|
||||||
|
|
@ -158,5 +214,10 @@ int main(void)
|
||||||
RUN_TEST(test_both_keys_keep_first_list);
|
RUN_TEST(test_both_keys_keep_first_list);
|
||||||
RUN_TEST(test_apiserver_limits_parsed);
|
RUN_TEST(test_apiserver_limits_parsed);
|
||||||
RUN_TEST(test_apiserver_nofile_limit_parsed);
|
RUN_TEST(test_apiserver_nofile_limit_parsed);
|
||||||
|
RUN_TEST(test_x11_key_parsed);
|
||||||
|
RUN_TEST(test_x11_key_string_form_parsed);
|
||||||
|
RUN_TEST(test_x11_key_absent_defaults_zero);
|
||||||
|
RUN_TEST(test_generate_exdata_x11cookie);
|
||||||
|
RUN_TEST(test_generate_exdata_without_cookie);
|
||||||
return UNITY_END();
|
return UNITY_END();
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue