amadan

carlos / platform Public

instancectl client, LayoutJournalPath, ReadBeneathLimit (plan 3, task 1)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev d86427f0eb25535566fbf5b30583403e7ecbf4b1 parent de37683
7 files changed, +353 −4
  • internal/instancectl/client.go +166 −0
  • internal/instancectl/client_test.go +121 −0
  • internal/instancectl/server_test.go +13 −4
  • internal/instancefs/instancefs.go +24 −0
  • internal/instancefs/instancefs_linux_test.go +14 −0
  • internal/provision/provision.go +9 −0
  • internal/provision/provision_test.go +6 −0

CI passed — run details

diff --git a/internal/instancectl/client.go b/internal/instancectl/client.go
new file mode 100644
index 000000000..689b08eb0
--- /dev/null
+++ b/internal/instancectl/client.go
@@ -0,0 +1,166 @@
+package instancectl
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "net/url"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// Deadlines per verb (design §5.3). A start or stop that exceeds its
+// deadline is a failed wake or an unproven stop, never a hang: §6.7 puts
+// one such call on a path that must not stall the activator. The seed's is
+// long because a seed is allowed to be slow.
+const (
+ DeadlineControl = 30 * time.Second
+ DeadlineSeed = 150 * time.Second
+ DeadlineSweep = 60 * time.Second
+ DeadlineReady = 5 * time.Second
+)
+
+// Client is the agent's side of the control channel. It is plain HTTP over
+// the unix socket at Path; unixdial's checks are for UPSTREAMS the edge
+// proxies to, and this socket is the one path they refuse, so the client
+// dials it directly.
+type Client struct {
+ Path string
+ // Dial is the dialer; nil means net.Dialer. Tests do not need to override
+ // it, but a caller composing several clients may.
+ Dial func(ctx context.Context, path string) (net.Conn, error)
+}
+
+func NewClient(path string) *Client { return &Client{Path: path} }
+
+// NotReadyError carries the probe verdict a 503 /ready returned.
+type NotReadyError struct{ Report ProbeReport }
+
+func (e *NotReadyError) Error() string {
+ var fails []string
+ for _, l := range e.Report.Lines {
+ if l.Verdict != "ok" {
+ fails = append(fails, l.Check+" "+l.Verdict+" "+l.Detail)
+ }
+ }
+ if len(fails) == 0 {
+ return "instance control: box not ready (no verdict yet)"
+ }
+ return "instance control: box not ready: " + strings.Join(fails, "; ")
+}
+
+func (c *Client) httpClient() *http.Client {
+ dial := c.Dial
+ if dial == nil {
+ dial = func(ctx context.Context, path string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, "unix", path)
+ }
+ }
+ return &http.Client{Transport: &http.Transport{
+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { return dial(ctx, c.Path) },
+ DisableKeepAlives: true,
+ }}
+}
+
+// do issues one verb with the given deadline and decodes a JSON body when
+// there is one. A non-2xx answer becomes an error carrying the server's
+// message; a dial failure names the socket so a forgotten enable reads as
+// what it is from carlos logs.
+func (c *Client) do(ctx context.Context, deadline time.Duration, method, path string, q url.Values, out any) (int, error) {
+ ctx, cancel := context.WithTimeout(ctx, deadline)
+ defer cancel()
+ u := "http://control" + path
+ if len(q) > 0 {
+ u += "?" + q.Encode()
+ }
+ req, err := http.NewRequestWithContext(ctx, method, u, nil)
+ if err != nil {
+ return 0, err
+ }
+ resp, err := c.httpClient().Do(req)
+ if err != nil {
+ return 0, fmt.Errorf("instance control socket %s: %w", c.Path, err)
+ }
+ defer resp.Body.Close()
+ body, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
+ if resp.StatusCode == http.StatusServiceUnavailable && path == "/ready" {
+ var r ProbeReport
+ _ = json.Unmarshal(body, &r)
+ return resp.StatusCode, &NotReadyError{Report: r}
+ }
+ // A 502 from /sweep still carries the units it did and didn't manage to
+ // stop, so it is decoded into out here even though it is about to be
+ // reported as an error too; the caller needs the partial list alongside
+ // the failure, not instead of it.
+ if out != nil && len(body) > 0 && (resp.StatusCode/100 == 2 || (resp.StatusCode == http.StatusBadGateway && path == "/sweep")) {
+ if err := json.Unmarshal(body, out); err != nil {
+ return resp.StatusCode, fmt.Errorf("instance control %s: %v", path, err)
+ }
+ }
+ if resp.StatusCode/100 != 2 {
+ return resp.StatusCode, fmt.Errorf("instance control %s: %d: %s", path, resp.StatusCode, strings.TrimSpace(string(body)))
+ }
+ return resp.StatusCode, nil
+}
+
+func hostRole(host string, role Role) url.Values {
+ q := url.Values{"host": {host}}
+ if role != "" {
+ q.Set("role", string(role))
+ }
+ return q
+}
+
+// Start starts the run or sidecar unit for host with the memory tier and
+// returns the invocation id the server read after the start.
+func (c *Client) Start(ctx context.Context, host string, role Role, memoryMax int64) (string, error) {
+ q := hostRole(host, role)
+ q.Set("memory_max", strconv.FormatInt(memoryMax, 10))
+ var out StartResponse
+ if _, err := c.do(ctx, DeadlineControl, http.MethodPost, "/start", q, &out); err != nil {
+ return "", err
+ }
+ return out.InvocationID, nil
+}
+
+// Seed runs the seed unit to completion and returns its retained result.
+func (c *Client) Seed(ctx context.Context, host string, memoryMax int64) (SeedResponse, error) {
+ q := hostRole(host, "")
+ q.Set("memory_max", strconv.FormatInt(memoryMax, 10))
+ var out SeedResponse
+ _, err := c.do(ctx, DeadlineSeed, http.MethodPost, "/seed", q, &out)
+ return out, err
+}
+
+func (c *Client) Stop(ctx context.Context, host string, role Role) error {
+ _, err := c.do(ctx, DeadlineControl, http.MethodPost, "/stop", hostRole(host, role), nil)
+ return 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
+}
+
+// Sweep stops every instance unit on the box (design §6.7). A 502 carries
+// the units that were stopped and the ones that would not; both are
+// returned with the error.
+func (c *Client) Sweep(ctx context.Context) (SweepResponse, error) {
+ var out SweepResponse
+ status, err := c.do(ctx, DeadlineSweep, http.MethodPost, "/sweep", nil, &out)
+ if err != nil && status == http.StatusBadGateway {
+ return out, errors.Join(err, errors.New("some instance units would not stop"))
+ }
+ return out, err
+}
+
+// Ready is nil on 204 and a *NotReadyError on 503.
+func (c *Client) Ready(ctx context.Context) error {
+ _, err := c.do(ctx, DeadlineReady, http.MethodGet, "/ready", nil, nil)
+ return err
+}
diff --git a/internal/instancectl/client_test.go b/internal/instancectl/client_test.go
new file mode 100644
index 000000000..dc84c5837
--- /dev/null
+++ b/internal/instancectl/client_test.go
@@ -0,0 +1,121 @@
+package instancectl
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+)
+
+// newClient serves a Server on a real unix socket with the given fake
+// systemctl and returns the agent-side Client for it.
+func newClient(t *testing.T, ctl *fakeCtl, configure ...func(*Server)) (*Server, *Client) {
+ t.Helper()
+ s, sock := serveOnSocket(t, ctl, configure...)
+ return s, NewClient(sock)
+}
+
+func TestClientStartCarriesTierAndReturnsInvocation(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
+ _, c := newClient(t, ctl)
+ inv, err := c.Start(context.Background(), "h.example.com", RoleRun, 536870912)
+ if err != nil || inv != "inv1" {
+ t.Fatalf("Start = %q, %v", inv, err)
+ }
+ if !strings.Contains(ctl.joined(), "MemoryMax=536870912") {
+ t.Errorf("tier not sent: %s", ctl.joined())
+ }
+ if !strings.Contains(ctl.joined(), "start "+UnitName(RoleRun, "h.example.com")) {
+ t.Errorf("run unit not started: %s", ctl.joined())
+ }
+}
+
+func TestClientSeedStopResetSweep(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{
+ "show -p InvocationID,Result,ExecMainStatus,ExecMainCode": "InvocationID=s1\nResult=success\nExecMainStatus=0\nExecMainCode=exited\n",
+ }}
+ _, c := newClient(t, ctl)
+ res, err := c.Seed(context.Background(), "h.example.com", 536870912)
+ if err != nil || res.Result != "success" || res.InvocationID != "s1" || res.Status != "0" || res.Code != "exited" {
+ t.Fatalf("Seed = %+v, %v", res, err)
+ }
+ if err := c.Stop(context.Background(), "h.example.com", RoleSidecar); err != nil {
+ t.Fatalf("Stop: %v", err)
+ }
+ if !strings.Contains(ctl.joined(), "stop "+UnitName(RoleSidecar, "h.example.com")) {
+ t.Errorf("sidecar unit not stopped: %s", ctl.joined())
+ }
+ if err := c.Reset(context.Background(), "h.example.com", RoleRun); err != nil {
+ t.Fatalf("Reset: %v", err)
+ }
+ sw, err := c.Sweep(context.Background())
+ if err != nil || len(sw.Units) != 0 || len(sw.Failed) != 0 {
+ t.Fatalf("Sweep = %+v, %v", sw, err)
+ }
+}
+
+func TestClientReadyIsNilOn204AndCarriesTheReportOn503(t *testing.T) {
+ ctl := &fakeCtl{}
+ s, c := newClient(t, ctl)
+ // Before the first probe the server answers 503 with one line saying
+ // so; the client surfaces it as a NotReadyError whose message names
+ // the failing line.
+ err := c.Ready(context.Background())
+ var nr *NotReadyError
+ if !errors.As(err, &nr) || len(nr.Report.Lines) != 1 || nr.Report.Lines[0].Check != "probe" {
+ t.Fatalf("Ready before a probe = %v", err)
+ }
+ if !strings.Contains(err.Error(), "not yet run") {
+ t.Errorf("message = %q", err.Error())
+ }
+ b, _ := json.Marshal(ProbeReport{Ready: true, Lines: []VerdictLine{{Check: "pid1", Verdict: "ok"}}})
+ if err := os.WriteFile(s.ProbeOut, b, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ s.RunProbe(context.Background())
+ if err := c.Ready(context.Background()); err != nil {
+ t.Fatalf("Ready after a passing probe: %v", err)
+ }
+}
+
+func TestClientSurfacesTheServersMessageOnA502(t *testing.T) {
+ ctl := &fakeCtl{fail: map[string]error{"start": errors.New("boom")}}
+ _, c := newClient(t, ctl)
+ _, err := c.Start(context.Background(), "h.example.com", RoleRun, 536870912)
+ if err == nil || !strings.Contains(err.Error(), "boom") || !strings.Contains(err.Error(), "/start") {
+ t.Fatalf("Start error = %v, want the server's message and the verb", err)
+ }
+}
+
+func TestClientSweepReturnsThePartialListWithTheError(t *testing.T) {
+ // One unit is listed, its stop fails, and the server answers 502 with
+ // the partial result; the client returns both.
+ unit := UnitName(RoleRun, "h.example.com")
+ ctl := &fakeCtl{
+ answers: map[string]string{
+ "list-units": unit + " loaded active running carlos instance\n",
+ },
+ fail: map[string]error{"stop": errors.New("would not stop")},
+ }
+ _, c := newClient(t, ctl)
+ sw, err := c.Sweep(context.Background())
+ if err == nil {
+ t.Fatalf("Sweep = %+v, want an error", sw)
+ }
+ if _, listed := sw.Failed[unit]; !listed {
+ t.Errorf("Failed = %+v, want %s", sw.Failed, unit)
+ }
+}
+
+func TestClientRefusesWhenTheSocketIsAbsent(t *testing.T) {
+ c := NewClient(filepath.Join(t.TempDir(), "absent.sock"))
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ if err := c.Ready(ctx); err == nil || !strings.Contains(err.Error(), "instance control socket") {
+ t.Fatalf("absent socket: %v", err)
+ }
+}
diff --git a/internal/instancectl/server_test.go b/internal/instancectl/server_test.go
index 2ccc339ee..0ed37a3f6 100644
--- a/internal/instancectl/server_test.go
+++ b/internal/instancectl/server_test.go
@@ -74,6 +74,18 @@ func (f *fakeCtl) joined() string {
// PeerOK (or anything else) sets it up front rather than racing Serve's
// first Accept (codex review fix round 2 item 1).
func newServer(t *testing.T, ctl *fakeCtl, configure ...func(*Server)) (*Server, *http.Client) {
+ t.Helper()
+ s, sock := serveOnSocket(t, ctl, configure...)
+ client := &http.Client{Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, "unix", sock)
+ }}}
+ return s, client
+}
+
+// serveOnSocket is newServer without the http.Client: it returns the
+// socket path so a test of the agent-side Client can dial it the way
+// production does.
+func serveOnSocket(t *testing.T, ctl *fakeCtl, configure ...func(*Server)) (*Server, string) {
t.Helper()
root := t.TempDir()
s := &Server{
@@ -102,10 +114,7 @@ func newServer(t *testing.T, ctl *fakeCtl, configure ...func(*Server)) (*Server,
t.Cleanup(cancel)
go s.Serve(ctx, ln) //nolint:errcheck
t.Cleanup(func() { ln.Close() })
- client := &http.Client{Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
- return (&net.Dialer{}).DialContext(ctx, "unix", sock)
- }}}
- return s, client
+ return s, sock
}
func post(t *testing.T, c *http.Client, path string) *http.Response {
diff --git a/internal/instancefs/instancefs.go b/internal/instancefs/instancefs.go
index de8ec6824..acad5dcff 100644
--- a/internal/instancefs/instancefs.go
+++ b/internal/instancefs/instancefs.go
@@ -21,6 +21,7 @@ package instancefs
import (
"errors"
+ "fmt"
"io"
"os"
"path/filepath"
@@ -54,3 +55,26 @@ func ReadBeneath(dir, name string) ([]byte, error) {
defer f.Close()
return io.ReadAll(f)
}
+
+// ErrTooLarge is returned by ReadBeneathLimit when the file exceeds max.
+var ErrTooLarge = errors.New("instancefs: file larger than the limit")
+
+// ReadBeneathLimit is ReadBeneath with a size cap. Files in an instance
+// directory are tenant-writable, and some are read by the agent outside the
+// tenant's memory wall (the exit record, the launch file), so no read there
+// is unbounded.
+func ReadBeneathLimit(dir, name string, max int64) ([]byte, error) {
+ f, err := OpenBeneath(dir, name, os.O_RDONLY, 0)
+ if err != nil {
+ return nil, err
+ }
+ defer f.Close()
+ b, err := io.ReadAll(io.LimitReader(f, max+1))
+ if err != nil {
+ return nil, err
+ }
+ if int64(len(b)) > max {
+ return nil, fmt.Errorf("%w: %s over %d bytes", ErrTooLarge, filepath.Join(dir, name), max)
+ }
+ return b, nil
+}
diff --git a/internal/instancefs/instancefs_linux_test.go b/internal/instancefs/instancefs_linux_test.go
index cf702a578..674a2a73d 100644
--- a/internal/instancefs/instancefs_linux_test.go
+++ b/internal/instancefs/instancefs_linux_test.go
@@ -193,3 +193,17 @@ func TestMountOptionOfIsTheGeneralForm(t *testing.T) {
}
_ = got // / is not reliably ro or rw across test hosts; only the absence of an error matters here.
}
+
+func TestReadBeneathLimitRefusesALargeFile(t *testing.T) {
+ dir := t.TempDir()
+ if err := os.WriteFile(filepath.Join(dir, "big"), make([]byte, 4096), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := ReadBeneathLimit(dir, "big", 1024); !errors.Is(err, ErrTooLarge) {
+ t.Fatalf("err = %v, want ErrTooLarge", err)
+ }
+ got, err := ReadBeneathLimit(dir, "big", 4096)
+ if err != nil || len(got) != 4096 {
+ t.Fatalf("exact limit: %d bytes, %v", len(got), err)
+ }
+}
diff --git a/internal/provision/provision.go b/internal/provision/provision.go
index aa2dcfe96..525201223 100644
--- a/internal/provision/provision.go
+++ b/internal/provision/provision.go
@@ -1091,6 +1091,15 @@ func LogSocket(host, role string) string {
return filepath.Join(SocketDir, "logs", InstanceID(host)+"."+role+".sock")
}
+// LayoutJournalPath is the migration journal for a host in flight
+// (isolated-exec design §8.2): under <dataDir>/layout, outside every tenant
+// mount. Its presence is what makes a host `fenced` (§8.1); plan 4 writes
+// it, and every consumer that must not act on a half-migrated host checks
+// for it here.
+func LayoutJournalPath(dataDir, host string) string {
+ return filepath.Join(dataDir, "layout", host+".json")
+}
+
// IsControlEndpoint reports whether path is the control channel or
// anything else under its directory. The edge and the agent refuse to
// dial such a path as an upstream (design §5.4): a registry row pointing
diff --git a/internal/provision/provision_test.go b/internal/provision/provision_test.go
index 595b52611..1ab77449c 100644
--- a/internal/provision/provision_test.go
+++ b/internal/provision/provision_test.go
@@ -1109,3 +1109,9 @@ func TestIsControlEndpoint(t *testing.T) {
}
}
}
+
+func TestLayoutJournalPath(t *testing.T) {
+ if got := LayoutJournalPath("/data/", "h.example.com"); got != "/data/layout/h.example.com.json" {
+ t.Errorf("LayoutJournalPath = %q", got)
+ }
+}