package supervisor import ( "bytes" "context" "encoding/xml" "errors" "fmt" "io" "net" "net/http" "strconv" "strings" "time" ) // socketPath is supervisord's control socket: build/supervisord.conf, // [unix_http_server] file=, group selfpost, which the panel is in. const socketPath = "/run/supervisor.sock" // runOnceFor bounds RunOnce: a reload that is still running after it is // reported as a failure. "postfix reload" checks that master runs, moves the // incoming queue to active and sends master SIGHUP (Postfix 3.7.11, // conf/postfix-script, "reload") - well under a second on the stand. var runOnceFor = time.Minute // pollEvery is how often RunOnce asks how the run is going; supervisord's own // deferred start answers on the same beat (supervisor 4.2.5, // supervisor/rpcinterface.py, startProcess(): onwait.delay = 0.05). var pollEvery = 50 * time.Millisecond // supervisord's process states (supervisor 4.2.5, supervisor/states.py, // ProcessStates) as getProcessInfo returns them in 'state'. const ( stateStopped = 0 stateStarting = 10 stateRunning = 20 stateBackoff = 30 stateStopping = 40 stateExited = 100 ) // faultAlreadyStarted is the fault startProcess raises for a program in // STARTING, RUNNING or BACKOFF (supervisor 4.2.5, supervisor/xmlrpc.py, // Faults.ALREADY_STARTED; rpcinterface.py, startProcess(): RUNNING_STATES). const faultAlreadyStarted = 60 // RunOnce runs a one-shot program to its end and reports how it ended: nil // only when it exited with status 0. // // "supervisorctl start" cannot tell that. For a program with startsecs=0 it // returns as soon as the program is RUNNING - spawned, not finished - and its // exit status is never reported (supervisor 4.2.5, supervisor/rpcinterface.py, // startProcess(): process.transition() moves STARTING to RUNNING at once, // then "return True"; assumptions B-U2). So RunOnce speaks supervisord's // XML-RPC interface itself: startProcess without waiting, then // getProcessInfo until the program leaves STARTING and RUNNING, and the // 'exitstatus' supervisord kept (process.py, Subprocess.finish(): with // startsecs=0 every end is EXITED, with its status). // // A run already in progress started before the caller's change was written, // and may have read the old files: RunOnce waits for it to end and then starts // its own. A run someone else starts after this one ended is newer than the // caller's change, so its end serves just as well. func RunOnce(program string) error { return newRPC(socketPath).runOnce(program) } // State is the state supervisord gives a program, by its name: RUNNING, // STARTING, EXITED ... (supervisor 4.2.5, supervisor/states.py). func State(program string) (string, error) { v, err := newRPC(socketPath).call("supervisor.getProcessInfo", program) if err != nil { return "", fmt.Errorf("supervisor: state of %s: %w", program, err) } return v.members()["statename"].str(), nil } type rpc struct { client *http.Client } func newRPC(sock string) *rpc { return &rpc{client: &http.Client{ Timeout: 10 * time.Second, Transport: &http.Transport{ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { var d net.Dialer return d.DialContext(ctx, "unix", sock) }, DisableKeepAlives: true, }, }} } func (c *rpc) runOnce(program string) error { deadline := time.Now().Add(runOnceFor) for { _, err := c.call("supervisor.startProcess", program, false) var f *Fault if errors.As(err, &f) && f.Code == faultAlreadyStarted { info, err := c.waitEnd(program, deadline) if err != nil { return err } if info.state == stateBackoff { // Never for startsecs=0 (Subprocess.finish()), so no // start of ours could follow it. return fmt.Errorf("supervisor: %s is in BACKOFF: %s", program, info.spawnErr) } if time.Now().After(deadline) { return fmt.Errorf("supervisor: %s did not come free within %s", program, runOnceFor) } continue } if err != nil { return fmt.Errorf("supervisor: start %s: %w", program, err) } break } info, err := c.waitEnd(program, deadline) if err != nil { return err } if info.state == stateExited && info.exitStatus == 0 { return nil } msg := fmt.Sprintf("supervisor: %s ended %s, exit status %d", program, info.stateName, info.exitStatus) if info.spawnErr != "" { msg += ": " + info.spawnErr } return errors.New(msg) } // processInfo is the part of getProcessInfo's struct RunOnce reads. type processInfo struct { state int stateName string exitStatus int spawnErr string } // waitEnd asks for the program's state until it is out of STARTING, RUNNING // and STOPPING. func (c *rpc) waitEnd(program string, deadline time.Time) (processInfo, error) { for { v, err := c.call("supervisor.getProcessInfo", program) if err != nil { return processInfo{}, fmt.Errorf("supervisor: state of %s: %w", program, err) } m := v.members() info := processInfo{ state: m["state"].int(), stateName: m["statename"].str(), exitStatus: m["exitstatus"].int(), spawnErr: m["spawnerr"].str(), } switch info.state { case stateStarting, stateRunning, stateStopping: default: return info, nil } if time.Now().After(deadline) { return info, fmt.Errorf("supervisor: %s still %s after %s", program, info.stateName, runOnceFor) } time.Sleep(pollEvery) } } // Fault is an XML-RPC fault supervisord answered with (supervisor 4.2.5, // supervisor/xmlrpc.py, Faults; the string is ": "). type Fault struct { Code int String string } func (f *Fault) Error() string { return fmt.Sprintf("fault %d: %s", f.Code, f.String) } // call makes one XML-RPC call; params are strings and booleans, all the // methods used here take. func (c *rpc) call(method string, params ...any) (xvalue, error) { var b bytes.Buffer b.WriteString(``) xml.EscapeText(&b, []byte(method)) b.WriteString(``) for _, p := range params { b.WriteString(``) switch v := p.(type) { case string: b.WriteString(``) xml.EscapeText(&b, []byte(v)) b.WriteString(``) case bool: if v { b.WriteString(`1`) } else { b.WriteString(`0`) } default: return xvalue{}, fmt.Errorf("xml-rpc: parameter of type %T", p) } b.WriteString(``) } b.WriteString(``) // supervisor 4.2.5, supervisor/xmlrpc.py, supervisor_xmlrpc_handler: // path '/RPC2'; the host is not looked at. resp, err := c.client.Post("http://localhost/RPC2", "text/xml", &b) if err != nil { return xvalue{}, err } defer resp.Body.Close() body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) if err != nil { return xvalue{}, err } if resp.StatusCode != http.StatusOK { return xvalue{}, fmt.Errorf("xml-rpc %s: HTTP %d", method, resp.StatusCode) } var r struct { Params []xvalue `xml:"params>param>value"` Fault *xvalue `xml:"fault>value"` } if err := xml.Unmarshal(body, &r); err != nil { return xvalue{}, fmt.Errorf("xml-rpc %s: %w", method, err) } if r.Fault != nil { m := r.Fault.members() return xvalue{}, &Fault{Code: m["faultCode"].int(), String: m["faultString"].str()} } if len(r.Params) != 1 { return xvalue{}, fmt.Errorf("xml-rpc %s: %d values in the answer", method, len(r.Params)) } return r.Params[0], nil } // xvalue is an XML-RPC as Python's xmlrpc.client writes it: a typed // element, or bare text for a string. type xvalue struct { Int *string `xml:"int"` I4 *string `xml:"i4"` Boolean *string `xml:"boolean"` String *string `xml:"string"` Struct *struct { Members []struct { Name string `xml:"name"` Value xvalue `xml:"value"` } `xml:"member"` } `xml:"struct"` Text string `xml:",chardata"` } func (v xvalue) members() map[string]xvalue { m := map[string]xvalue{} if v.Struct != nil { for _, mb := range v.Struct.Members { m[mb.Name] = mb.Value } } return m } func (v xvalue) int() int { for _, s := range []*string{v.Int, v.I4, v.Boolean} { if s != nil { n, _ := strconv.Atoi(strings.TrimSpace(*s)) return n } } return 0 } func (v xvalue) str() string { if v.String != nil { return *v.String } return v.Text }