supervision as supervisord does it: Postfix stopped, give-ups FATAL, reload's exit read
test / test (push) Canceled after 0s
test / test (push) Canceled after 0s
C-P42. postfix-script's start-fg runs master as its child and master calls setsid() (Postfix 3.7.11, conf/postfix-script; src/master/master.c); supervisord signals the pid alone (supervisor 4.2.5, process.py, Subprocess.kill()). A stop of program postfix left master running - a restart then found "already running" - and a container stop cut Postfix off with SIGKILL. The wrapper now stays the parent of "postfix start-fg" and on TERM/INT runs "postfix stop", asking again while start-fg is still in its checks (master -t). stopwaitsecs=15. C-V6. The wrapper's give-up on the milter sockets came after the default startsecs, so it was EXITED and restarted for ever. startsecs is now MILTER_WAIT_TIMEOUT + 10, exported by the entrypoint as POSTFIX_STARTSECS and expanded by supervisord (options.py, processes_from_section(); a non-number MILTER_WAIT_TIMEOUT stops the start): a give-up is "Exited too quickly", BACKOFF, FATAL after startretries, crashexit. For those seconds after each start the program is STARTING (the healthcheck's 90 s start period covers it; the sender check counts no downtime then). C-V5. logrotate had startsecs=0, so a refused config restarted it every second for ever. startsecs=1: FATAL after startretries; crashexit reads the event's processname and leaves logrotate's FATAL on Server > Health. B-U2. "supervisorctl start" returns when a startsecs=0 program is RUNNING, not when it ends, and mapped "already started" to success. The panel now speaks supervisord's XML-RPC itself (internal/supervisor, RunOnce): startProcess without waiting, getProcessInfo until it leaves STARTING/RUNNING, exitstatus 0 or an error; a run in progress is waited out and one of its own follows. A failed reload while program postfix is not RUNNING is logged, not an error: the next master reads the files. C-V10: postfix-reload rests EXITED after a run; the e2e process check takes STOPPED or EXITED. C-P34: the dead "postfix set-permissions" branch goes - postfix check's create-missing builds the queue tree. Stand (2026-10-14, make e2e, exit 0; MILTER_WAIT_TIMEOUT=10 on the stand): new supervision_stops_postfix_and_gives_up - supervisorctl restart postfix writes postfix-script's "stopping the Postfix mail system" and leaves one master; with the panel's sockets removed, start answers "postfix: ERROR (spawn error)" (finish() sets spawnerr "Exited too quickly", onwait() reads it first) and Postfix comes up once the panel is back; a group-writable logrotate config ends FATAL with the container up; the panel's Reload configuration answers only after mail.log has "refreshing the Postfix mail system", postfix-reload EXITED. session_survives_restart sees the stop line for the container's restart. Measured: master's own "terminating on signal 15" never reaches mail.log (master TERMs its group, postlogd included, before it logs). go vet and go test ./... pass on the server (24 packages; new unit tests of RunOnce against a fake supervisord). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,280 @@
|
||||
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 "<NAME>: <detail>").
|
||||
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 version="1.0"?><methodCall><methodName>`)
|
||||
xml.EscapeText(&b, []byte(method))
|
||||
b.WriteString(`</methodName><params>`)
|
||||
for _, p := range params {
|
||||
b.WriteString(`<param><value>`)
|
||||
switch v := p.(type) {
|
||||
case string:
|
||||
b.WriteString(`<string>`)
|
||||
xml.EscapeText(&b, []byte(v))
|
||||
b.WriteString(`</string>`)
|
||||
case bool:
|
||||
if v {
|
||||
b.WriteString(`<boolean>1</boolean>`)
|
||||
} else {
|
||||
b.WriteString(`<boolean>0</boolean>`)
|
||||
}
|
||||
default:
|
||||
return xvalue{}, fmt.Errorf("xml-rpc: parameter of type %T", p)
|
||||
}
|
||||
b.WriteString(`</value></param>`)
|
||||
}
|
||||
b.WriteString(`</params></methodCall>`)
|
||||
|
||||
// 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 <value> 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
|
||||
}
|
||||
@@ -0,0 +1,156 @@
|
||||
package supervisor
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// fakeSupervisord answers startProcess and getProcessInfo the way supervisor
|
||||
// 4.2.5 does over XML-RPC (Python's xmlrpc.client.dumps with methodresponse:
|
||||
// one <param>, or a <fault> struct), from a script of states.
|
||||
type fakeSupervisord struct {
|
||||
mu sync.Mutex
|
||||
// alreadyStarted is how many startProcess calls get fault 60 first.
|
||||
alreadyStarted int
|
||||
// states are the (state, statename, exitstatus) getProcessInfo returns,
|
||||
// one per call; the last one repeats.
|
||||
states []fakeState
|
||||
starts int
|
||||
asked int
|
||||
}
|
||||
|
||||
type fakeState struct {
|
||||
state int
|
||||
name string
|
||||
exitStatus int
|
||||
}
|
||||
|
||||
var methodRe = regexp.MustCompile(`<methodName>([^<]+)</methodName>`)
|
||||
|
||||
func (f *fakeSupervisord) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
m := methodRe.FindSubmatch(body)
|
||||
if r.URL.Path != "/RPC2" || m == nil {
|
||||
http.Error(w, "no", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
w.Header().Set("Content-Type", "text/xml")
|
||||
switch string(m[1]) {
|
||||
case "supervisor.startProcess":
|
||||
f.starts++
|
||||
if !strings.Contains(string(body), "<boolean>0</boolean>") {
|
||||
http.Error(w, "start must not wait", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if f.alreadyStarted > 0 {
|
||||
f.alreadyStarted--
|
||||
io.WriteString(w, "<?xml version='1.0'?>\n<methodResponse>\n<fault>\n<value><struct>\n"+
|
||||
"<member>\n<name>faultCode</name>\n<value><int>60</int></value>\n</member>\n"+
|
||||
"<member>\n<name>faultString</name>\n<value><string>ALREADY_STARTED: postfix-reload</string></value>\n</member>\n"+
|
||||
"</struct></value>\n</fault>\n</methodResponse>\n")
|
||||
return
|
||||
}
|
||||
io.WriteString(w, "<?xml version='1.0'?>\n<methodResponse>\n<params>\n<param>\n<value><boolean>1</boolean></value>\n</param>\n</params>\n</methodResponse>\n")
|
||||
case "supervisor.getProcessInfo":
|
||||
s := f.states[min(f.asked, len(f.states)-1)]
|
||||
f.asked++
|
||||
fmt.Fprintf(w, "<?xml version='1.0'?>\n<methodResponse>\n<params>\n<param>\n<value><struct>\n"+
|
||||
"<member>\n<name>name</name>\n<value><string>postfix-reload</string></value>\n</member>\n"+
|
||||
"<member>\n<name>state</name>\n<value><int>%d</int></value>\n</member>\n"+
|
||||
"<member>\n<name>statename</name>\n<value><string>%s</string></value>\n</member>\n"+
|
||||
"<member>\n<name>exitstatus</name>\n<value><int>%d</int></value>\n</member>\n"+
|
||||
"<member>\n<name>spawnerr</name>\n<value><string></string></value>\n</member>\n"+
|
||||
"</struct></value>\n</param>\n</params>\n</methodResponse>\n", s.state, s.name, s.exitStatus)
|
||||
default:
|
||||
http.Error(w, "unknown method", http.StatusBadRequest)
|
||||
}
|
||||
}
|
||||
|
||||
func serveFake(t *testing.T, f *fakeSupervisord) *rpc {
|
||||
t.Helper()
|
||||
// A short directory: a unix socket's path is limited to about 100 bytes.
|
||||
dir, err := os.MkdirTemp("", "sv")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { os.RemoveAll(dir) })
|
||||
sock := filepath.Join(dir, "s")
|
||||
ln, err := net.Listen("unix", sock)
|
||||
if err != nil {
|
||||
t.Skipf("unix sockets: %v", err)
|
||||
}
|
||||
srv := &http.Server{Handler: f}
|
||||
go srv.Serve(ln)
|
||||
t.Cleanup(func() { srv.Close() })
|
||||
old := pollEvery
|
||||
pollEvery = time.Millisecond
|
||||
t.Cleanup(func() { pollEvery = old })
|
||||
return newRPC(sock)
|
||||
}
|
||||
|
||||
var (
|
||||
running = fakeState{20, "RUNNING", 0}
|
||||
exited0 = fakeState{100, "EXITED", 0}
|
||||
)
|
||||
|
||||
func TestRunOnceWaitsForTheEndAndItsStatus(t *testing.T) {
|
||||
f := &fakeSupervisord{states: []fakeState{running, running, exited0}}
|
||||
if err := serveFake(t, f).runOnce("postfix-reload"); err != nil {
|
||||
t.Fatalf("runOnce: %v", err)
|
||||
}
|
||||
if f.starts != 1 || f.asked != 3 {
|
||||
t.Fatalf("starts %d, asked %d; want 1 start and asked until EXITED (3)", f.starts, f.asked)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunOnceReportsAFailedRun(t *testing.T) {
|
||||
f := &fakeSupervisord{states: []fakeState{running, {100, "EXITED", 1}}}
|
||||
err := serveFake(t, f).runOnce("postfix-reload")
|
||||
if err == nil || !strings.Contains(err.Error(), "ended EXITED, exit status 1") {
|
||||
t.Fatalf("runOnce: %v; want the exit status 1", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A run in progress may have read the files before the caller wrote them:
|
||||
// it is waited out and a run of the caller's own follows.
|
||||
func TestRunOnceStartsItsOwnAfterARunInProgress(t *testing.T) {
|
||||
f := &fakeSupervisord{alreadyStarted: 1, states: []fakeState{running, exited0, running, exited0}}
|
||||
if err := serveFake(t, f).runOnce("postfix-reload"); err != nil {
|
||||
t.Fatalf("runOnce: %v", err)
|
||||
}
|
||||
if f.starts != 2 || f.asked != 4 {
|
||||
t.Fatalf("starts %d, asked %d; want the running one waited out, then a start of its own (2 starts, 4 asks)", f.starts, f.asked)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunOnceGivesUpOnARunThatDoesNotEnd(t *testing.T) {
|
||||
f := &fakeSupervisord{states: []fakeState{running}}
|
||||
c := serveFake(t, f)
|
||||
old := runOnceFor
|
||||
runOnceFor = 20 * time.Millisecond
|
||||
t.Cleanup(func() { runOnceFor = old })
|
||||
err := c.runOnce("postfix-reload")
|
||||
if err == nil || !strings.Contains(err.Error(), "still RUNNING") {
|
||||
t.Fatalf("runOnce: %v; want still RUNNING", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFaultIsReadFromTheAnswer(t *testing.T) {
|
||||
f := &fakeSupervisord{alreadyStarted: 1, states: []fakeState{exited0}}
|
||||
_, err := serveFake(t, f).call("supervisor.startProcess", "postfix-reload", false)
|
||||
fault, ok := err.(*Fault)
|
||||
if !ok || fault.Code != faultAlreadyStarted || fault.String != "ALREADY_STARTED: postfix-reload" {
|
||||
t.Fatalf("call: %v; want fault 60 ALREADY_STARTED", err)
|
||||
}
|
||||
}
|
||||
@@ -10,21 +10,6 @@ import (
|
||||
|
||||
const confPath = "/etc/supervisor/supervisord.conf"
|
||||
|
||||
// Start issues a `supervisorctl start` for the named program. A program that
|
||||
// is already running is not an error (the pending run covers the caller's
|
||||
// intent).
|
||||
func Start(program string) error {
|
||||
cmd := exec.Command("supervisorctl", "-c", confPath, "start", program)
|
||||
out, err := cmd.CombinedOutput()
|
||||
if err != nil {
|
||||
if strings.Contains(string(out), "already started") {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("supervisor start %s: %w: %s", program, err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Signal sends a signal to a supervised program via `supervisorctl signal`.
|
||||
func Signal(sig, program string) error {
|
||||
cmd := exec.Command("supervisorctl", "-c", confPath, "signal", sig, program)
|
||||
|
||||
Reference in New Issue
Block a user