amadan

carlos / platform Public

instancectl: POST /cmd and the GET /cmd/{host} long-poll (run spec §5.4, codex 4)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude c8a2639fc313dc4f9fb569bcf8627575493c2566 parent e3b80ec
6 files changed, +393 −1
  • internal/instancectl/claim_test.go +1 −1
  • internal/instancectl/client.go +21 −0
  • internal/instancectl/cmd.go +138 −0
  • internal/instancectl/cmd_test.go +205 −0
  • internal/instancectl/names.go +19 −0
  • internal/instancectl/server.go +9 −0
diff --git a/internal/instancectl/claim_test.go b/internal/instancectl/claim_test.go
index d106f53c8..be54b9562 100644
--- a/internal/instancectl/claim_test.go
+++ b/internal/instancectl/claim_test.go
@@ -11,7 +11,7 @@ import (
func TestClaimRootRefusesLegacyProcessMutations(t *testing.T) {
called := false
server := &Server{ClaimStop: http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), Systemctl: func(context.Context, ...string) (string, error) { called = true; return "", nil }}
- for _, verb := range []string{"start", "seed", "stop", "reset", "sweep"} {
+ for _, verb := range []string{"start", "seed", "stop", "reset", "sweep", "cmd"} {
response := httptest.NewRecorder()
server.Handler().ServeHTTP(response, httptest.NewRequest(http.MethodPost, "/"+verb+"?host=app.example&role=run&memory_max=536870912", strings.NewReader("")))
if response.Code != http.StatusServiceUnavailable {
diff --git a/internal/instancectl/client.go b/internal/instancectl/client.go
index cae0ca4cb..444b9e941 100644
--- a/internal/instancectl/client.go
+++ b/internal/instancectl/client.go
@@ -25,6 +25,9 @@ const (
DeadlineSeed = 150 * time.Second
DeadlineSweep = 60 * time.Second
DeadlineReady = 5 * time.Second
+
+ // DeadlineCmdPoll bounds GET /cmd/{host}; the server answers within 30 s.
+ DeadlineCmdPoll = 35 * time.Second
)
// Client is the agent's side of the control channel. It is plain HTTP over
@@ -165,6 +168,24 @@ func (c *Client) Stop(ctx context.Context, host string, role Role) error {
return err
}
+// Cmd starts host's cmd unit and returns its invocation (run spec §5.4).
+func (c *Client) Cmd(ctx context.Context, host string, memoryMax int64) (string, error) {
+ q := hostRole(host, "")
+ q.Set("memory_max", strconv.FormatInt(memoryMax, 10))
+ var out StartResponse
+ if _, _, err := c.do(ctx, DeadlineControl, http.MethodPost, "/cmd", q, &out); err != nil {
+ return "", err
+ }
+ return out.InvocationID, nil
+}
+
+// CmdStatus long-polls the named invocation of host's cmd unit.
+func (c *Client) CmdStatus(ctx context.Context, host, invocation string) (CmdStatus, error) {
+ var out CmdStatus
+ _, _, err := c.do(ctx, DeadlineCmdPoll, http.MethodGet, "/cmd/"+url.PathEscape(host), url.Values{"invocation": {invocation}}, &out)
+ return out, err
+}
+
func (c *Client) Reset(ctx context.Context, host string, role Role) error {
_, _, err := c.do(ctx, DeadlineControl, http.MethodPost, "/reset", hostRole(host, role), nil)
return err
diff --git a/internal/instancectl/cmd.go b/internal/instancectl/cmd.go
new file mode 100644
index 000000000..5e3294ee2
--- /dev/null
+++ b/internal/instancectl/cmd.go
@@ -0,0 +1,138 @@
+package instancectl
+
+import (
+ "context"
+ "fmt"
+ "net/http"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+func (s *Server) cmdPoll() time.Duration {
+ if s.cmdPollEvery > 0 {
+ return s.cmdPollEvery
+ }
+ return 250 * time.Millisecond
+}
+
+func (s *Server) cmdWait() time.Duration {
+ if s.cmdWaitBound > 0 {
+ return s.cmdWaitBound
+ }
+ return 30 * time.Second
+}
+
+func (s *Server) cmdStartWait() time.Duration {
+ if s.cmdStartBound > 0 {
+ return s.cmdStartBound
+ }
+ return 10 * time.Second
+}
+
+// handleCmd starts host's cmd unit and answers at once with the new
+// invocation (run spec §5.4); /seed blocks, this never does. The agent
+// stops a cmd unit after every run, so a unit still loaded (active,
+// activating, deactivating) belongs to a run this request must not end: it
+// is refused, never restarted. A failed one is reset first, or start would
+// keep its old result.
+func (s *Server) handleCmd(w http.ResponseWriter, r *http.Request) {
+ req, err := parseRequest(r, nil, true)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ ctx := r.Context()
+ unit := UnitName(RoleCmd, req.host)
+ switch state := s.showValue(ctx, "ActiveState", unit); state {
+ case "active", "activating", "deactivating", "reloading", "refreshing":
+ http.Error(w, "cmd: "+unit+" is "+state, http.StatusConflict)
+ return
+ case "failed":
+ if out, err := s.Systemctl(ctx, "reset-failed", unit); err != nil {
+ http.Error(w, "reset-failed: "+strings.TrimSpace(out+" "+err.Error()), http.StatusBadGateway)
+ return
+ }
+ }
+ if s.refuseStart(w, req.host, unit) {
+ return
+ }
+ if err := s.setTier(ctx, req.host, req.tier); err != nil {
+ http.Error(w, "set-property: "+err.Error(), http.StatusBadGateway)
+ return
+ }
+ before := s.showValue(ctx, "InvocationID", unit)
+ if out, err := s.Systemctl(ctx, "start", "--no-block", unit); err != nil {
+ http.Error(w, "start: "+strings.TrimSpace(out+" "+err.Error()), http.StatusBadGateway)
+ return
+ }
+ // --no-block returns once the job is queued; the invocation exists
+ // once the job runs.
+ deadline := time.Now().Add(s.cmdStartWait())
+ for {
+ if inv := s.showValue(ctx, "InvocationID", unit); inv != "" && inv != before {
+ s.logf("instance-control: %s: started %s memory.max=%d", req.host, unit, req.tier.MemoryMax)
+ writeJSON(w, http.StatusOK, StartResponse{InvocationID: inv})
+ return
+ }
+ if time.Now().After(deadline) || ctx.Err() != nil {
+ dctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
+ s.stopUnit(dctx, unit)
+ cancel()
+ http.Error(w, "cmd: "+unit+" showed no new invocation", http.StatusBadGateway)
+ return
+ }
+ time.Sleep(s.cmdPoll())
+ }
+}
+
+// handleCmdStatus is the bounded long-poll: it answers as soon as the named
+// invocation is done or gone, and running at the bound.
+func (s *Server) handleCmdStatus(w http.ResponseWriter, r *http.Request) {
+ host := r.PathValue("host")
+ if !provision.ValidHostShape(host) {
+ http.Error(w, fmt.Sprintf("host %q is not a host", host), http.StatusBadRequest)
+ return
+ }
+ want := r.URL.Query().Get("invocation")
+ if want == "" {
+ http.Error(w, "invocation is required", http.StatusBadRequest)
+ return
+ }
+ ctx := r.Context()
+ unit := UnitName(RoleCmd, host)
+ deadline := time.Now().Add(s.cmdWait())
+ for {
+ out, err := s.Systemctl(ctx, "show", "-p", "InvocationID,ActiveState,SubState,Result,ExecMainStatus,ExecMainCode", unit)
+ if err != nil {
+ http.Error(w, "show: "+err.Error(), http.StatusBadGateway)
+ return
+ }
+ st := cmdStatusOf(parseShow(out), want)
+ if st.State != CmdRunning || !time.Now().Before(deadline) {
+ writeJSON(w, http.StatusOK, st)
+ return
+ }
+ select {
+ case <-ctx.Done():
+ return
+ case <-time.After(s.cmdPoll()):
+ }
+ }
+}
+
+// cmdStatusOf reads one show for the named invocation.
+func cmdStatusOf(m map[string]string, want string) CmdStatus {
+ st := CmdStatus{InvocationID: m["InvocationID"], Result: m["Result"], Status: m["ExecMainStatus"], Code: m["ExecMainCode"]}
+ active, sub := m["ActiveState"], m["SubState"]
+ switch {
+ case st.InvocationID != want:
+ st.State = CmdGone
+ case active == "failed", active == "inactive", active == "active" && sub == "exited":
+ st.State = CmdDone
+ default:
+ st.State = CmdRunning
+ }
+ return st
+}
diff --git a/internal/instancectl/cmd_test.go b/internal/instancectl/cmd_test.go
new file mode 100644
index 000000000..790b86c94
--- /dev/null
+++ b/internal/instancectl/cmd_test.go
@@ -0,0 +1,205 @@
+package instancectl
+
+import (
+ "context"
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+ "time"
+)
+
+// set changes one answer under the fake's lock, for a test whose unit
+// moves on while a handler polls.
+func (f *fakeCtl) set(prefix, out string) {
+ f.mu.Lock()
+ f.answers[prefix] = out
+ f.mu.Unlock()
+}
+
+const cmdShowAll = "show -p InvocationID,ActiveState,SubState,Result,ExecMainStatus,ExecMainCode"
+
+func fastCmd(s *Server) {
+ s.cmdPollEvery = 5 * time.Millisecond
+ s.cmdWaitBound = 300 * time.Millisecond
+ s.cmdStartBound = 300 * time.Millisecond
+}
+
+func TestCmdStartsWithoutBlockingAndAnswersTheNewInvocation(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{
+ "show -p ActiveState --value": "inactive\n",
+ "show -p InvocationID --value": "old\n",
+ }}
+ ctl.onStart = func() { ctl.set("show -p InvocationID --value", "new\n") }
+ _, c := newServer(t, ctl, fastCmd)
+ resp := post(t, c, "/cmd?host=h.example.com&memory_max=536870912")
+ if resp.StatusCode != 200 {
+ t.Fatalf("status %d", resp.StatusCode)
+ }
+ var out StartResponse
+ if err := json.NewDecoder(resp.Body).Decode(&out); err != nil || out.InvocationID != "new" {
+ t.Fatalf("response %+v, %v", out, err)
+ }
+ calls := ctl.joined()
+ for _, want := range []string{"set-property --runtime carlos-instances-h.example.com.slice", "start --no-block carlos-instance-cmd@h.example.com.service"} {
+ if !strings.Contains(calls, want) {
+ t.Errorf("missing %q in\n%s", want, calls)
+ }
+ }
+ if strings.Contains(calls, "stop ") {
+ t.Errorf("a started command was stopped:\n%s", calls)
+ }
+}
+
+func TestCmdRefusesAUnitStillLoaded(t *testing.T) {
+ for _, state := range []string{"active", "activating", "deactivating"} {
+ ctl := &fakeCtl{answers: map[string]string{"show -p ActiveState --value": state + "\n"}}
+ _, c := newServer(t, ctl, fastCmd)
+ resp := post(t, c, "/cmd?host=h.example.com&memory_max=536870912")
+ if resp.StatusCode != http.StatusConflict {
+ t.Errorf("%s: status %d, want 409", state, resp.StatusCode)
+ }
+ if strings.Contains(ctl.joined(), "start ") {
+ t.Errorf("%s: a loaded unit was started again:\n%s", state, ctl.joined())
+ }
+ }
+}
+
+func TestCmdResetsAFailedUnitBeforeTheStart(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{
+ "show -p ActiveState --value": "failed\n",
+ "show -p InvocationID --value": "old\n",
+ }}
+ ctl.onStart = func() { ctl.set("show -p InvocationID --value", "new\n") }
+ _, c := newServer(t, ctl, fastCmd)
+ if resp := post(t, c, "/cmd?host=h.example.com&memory_max=536870912"); resp.StatusCode != 200 {
+ t.Fatalf("status %d", resp.StatusCode)
+ }
+ calls := ctl.joined()
+ if r, s := strings.Index(calls, "reset-failed carlos-instance-cmd@"), strings.Index(calls, "start --no-block"); r < 0 || s < 0 || r > s {
+ t.Errorf("reset-failed must come before the start:\n%s", calls)
+ }
+}
+
+func TestCmdStopsAUnitThatNeverReportsAnInvocation(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{
+ "show -p ActiveState --value": "inactive\n",
+ "show -p InvocationID --value": "old\n",
+ }}
+ _, c := newServer(t, ctl, fastCmd)
+ if resp := post(t, c, "/cmd?host=h.example.com&memory_max=536870912"); resp.StatusCode != http.StatusBadGateway {
+ t.Fatalf("status %d, want 502", resp.StatusCode)
+ }
+ if !strings.Contains(ctl.joined(), "stop carlos-instance-cmd@h.example.com.service") {
+ t.Errorf("the unit was not stopped:\n%s", ctl.joined())
+ }
+}
+
+func getStatus(t *testing.T, c *http.Client, path string) CmdStatus {
+ t.Helper()
+ resp, err := c.Get("http://unix" + path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != 200 {
+ t.Fatalf("GET %s: status %d", path, resp.StatusCode)
+ }
+ var st CmdStatus
+ if err := json.NewDecoder(resp.Body).Decode(&st); err != nil {
+ t.Fatal(err)
+ }
+ return st
+}
+
+func TestCmdStatusLongPollsUntilDone(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{
+ cmdShowAll: "InvocationID=inv1\nActiveState=activating\nSubState=start\nResult=success\nExecMainStatus=0\nExecMainCode=0\n",
+ }}
+ _, c := newServer(t, ctl, fastCmd)
+ go func() {
+ time.Sleep(50 * time.Millisecond)
+ ctl.set(cmdShowAll, "InvocationID=inv1\nActiveState=active\nSubState=exited\nResult=success\nExecMainStatus=3\nExecMainCode=1\n")
+ }()
+ st := getStatus(t, c, "/cmd/h.example.com?invocation=inv1")
+ if st.State != CmdDone || st.Status != "3" || st.Code != "1" || st.InvocationID != "inv1" {
+ t.Fatalf("status = %+v", st)
+ }
+}
+
+func TestCmdStatusAnswersRunningAtTheBound(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{
+ cmdShowAll: "InvocationID=inv1\nActiveState=activating\nSubState=start\nResult=success\nExecMainStatus=0\nExecMainCode=0\n",
+ }}
+ _, c := newServer(t, ctl, fastCmd)
+ start := time.Now()
+ if st := getStatus(t, c, "/cmd/h.example.com?invocation=inv1"); st.State != CmdRunning {
+ t.Fatalf("status = %+v", st)
+ }
+ if took := time.Since(start); took < 250*time.Millisecond {
+ t.Errorf("answered running after %s, before the bound", took)
+ }
+}
+
+func TestCmdStatusDoneCoversFailedAndInactive(t *testing.T) {
+ for _, show := range []string{
+ "InvocationID=inv1\nActiveState=failed\nSubState=failed\nResult=exit-code\nExecMainStatus=1\nExecMainCode=1\n",
+ "InvocationID=inv1\nActiveState=inactive\nSubState=dead\nResult=success\nExecMainStatus=0\nExecMainCode=1\n",
+ } {
+ ctl := &fakeCtl{answers: map[string]string{cmdShowAll: show}}
+ _, c := newServer(t, ctl, fastCmd)
+ if st := getStatus(t, c, "/cmd/h.example.com?invocation=inv1"); st.State != CmdDone {
+ t.Errorf("%q: state %s, want done", show, st.State)
+ }
+ }
+}
+
+func TestCmdStatusIsGoneForAnotherInvocation(t *testing.T) {
+ for _, show := range []string{
+ "InvocationID=inv2\nActiveState=active\nSubState=exited\nResult=success\nExecMainStatus=0\nExecMainCode=1\n",
+ "InvocationID=\nActiveState=inactive\nSubState=dead\nResult=success\nExecMainStatus=0\nExecMainCode=0\n",
+ } {
+ ctl := &fakeCtl{answers: map[string]string{cmdShowAll: show}}
+ _, c := newServer(t, ctl, fastCmd)
+ if st := getStatus(t, c, "/cmd/h.example.com?invocation=inv1"); st.State != CmdGone {
+ t.Errorf("%q: state %s, want gone", show, st.State)
+ }
+ }
+}
+
+func TestCmdRoutesRefuseWithClaimControlArmed(t *testing.T) {
+ called := false
+ s := &Server{ClaimStop: http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), Systemctl: func(context.Context, ...string) (string, error) { called = true; return "", nil }}
+ for _, req := range []*http.Request{
+ httptest.NewRequest(http.MethodPost, "/cmd?host=app.example&memory_max=536870912", nil),
+ httptest.NewRequest(http.MethodGet, "/cmd/app.example?invocation=x", nil),
+ } {
+ w := httptest.NewRecorder()
+ s.Handler().ServeHTTP(w, req)
+ if w.Code != http.StatusServiceUnavailable {
+ t.Errorf("%s %s: status %d", req.Method, req.URL, w.Code)
+ }
+ }
+ if called {
+ t.Fatal("a cmd route reached systemctl with claim control armed")
+ }
+}
+
+func TestClientCmdAndCmdStatus(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{
+ "show -p ActiveState --value": "inactive\n",
+ "show -p InvocationID --value": "old\n",
+ cmdShowAll: "InvocationID=new\nActiveState=failed\nSubState=failed\nResult=exit-code\nExecMainStatus=2\nExecMainCode=1\n",
+ }}
+ ctl.onStart = func() { ctl.set("show -p InvocationID --value", "new\n") }
+ _, cl := newClient(t, ctl, fastCmd)
+ inv, err := cl.Cmd(context.Background(), "h.example.com", 536870912)
+ if err != nil || inv != "new" {
+ t.Fatalf("Cmd = %q, %v", inv, err)
+ }
+ st, err := cl.CmdStatus(context.Background(), "h.example.com", inv)
+ if err != nil || st.State != CmdDone || st.Status != "2" {
+ t.Fatalf("CmdStatus = %+v, %v", st, err)
+ }
+}
diff --git a/internal/instancectl/names.go b/internal/instancectl/names.go
index f69ccc8fa..c2a2717f4 100644
--- a/internal/instancectl/names.go
+++ b/internal/instancectl/names.go
@@ -194,6 +194,25 @@ type SeedResponse struct {
Code string `json:"code"`
}
+// CmdStatus is GET /cmd/{host}'s answer (run spec §5.4). State is running,
+// done once the named invocation's main process has exited (active
+// (exited), failed, or inactive after a stop), or gone when the unit's
+// invocation is another one or none. Result, Status and Code are systemd's
+// Result, ExecMainStatus and ExecMainCode as `systemctl show` prints them.
+type CmdStatus struct {
+ State string `json:"state"`
+ InvocationID string `json:"invocation_id"`
+ Result string `json:"result"`
+ Status string `json:"status"`
+ Code string `json:"code"`
+}
+
+const (
+ CmdRunning = "running"
+ CmdDone = "done"
+ CmdGone = "gone"
+)
+
type SweepResponse struct {
Units []string `json:"units"`
Failed map[string]string `json:"failed,omitempty"`
diff --git a/internal/instancectl/server.go b/internal/instancectl/server.go
index 474b8f584..cd037ab6d 100644
--- a/internal/instancectl/server.go
+++ b/internal/instancectl/server.go
@@ -73,6 +73,13 @@ type Server struct {
Statfs func(path string) (avail, total uint64, err error)
DiskEvery time.Duration
+ // cmdPollEvery, cmdWaitBound and cmdStartBound pace the cmd routes:
+ // how often GET /cmd/{host} re-reads the unit (250 ms), how long it
+ // holds a request (30 s, run spec §5.4), and how long POST /cmd waits
+ // for the queued start to show a new invocation (10 s). Zero is the
+ // default; tests shorten them.
+ cmdPollEvery, cmdWaitBound, cmdStartBound time.Duration
+
// Run executes one command as root and returns its combined output;
// production is exec.CommandContext, tests record. Only
// inspectMountpoints uses it, and every argument it passes is a
@@ -258,6 +265,8 @@ func (s *Server) Handler() http.Handler {
mux.HandleFunc("POST /stop", s.claimAwareLegacy(s.handleStop))
mux.HandleFunc("POST /reset", s.claimAwareLegacy(s.handleReset))
mux.HandleFunc("POST /sweep", s.claimAwareLegacy(s.handleSweep))
+ mux.HandleFunc("POST /cmd", s.claimAwareLegacy(s.handleCmd))
+ mux.HandleFunc("GET /cmd/{host}", s.claimAwareLegacy(s.handleCmdStatus))
mux.HandleFunc("GET /ready", s.handleReady)
return mux
}