diff --git a/go/README.md b/go/README.md new file mode 100644 index 0000000..419f949 --- /dev/null +++ b/go/README.md @@ -0,0 +1,53 @@ +# Go client for microinit + +IPC client library for the microinit control socket (length-prefixed JSON over Unix domain sockets). + +## Import + +```go +import "github.com/dcc-bigfred/microinit/go/client" + +c := &client.Client{Socket: client.DefaultSocket} // or override +list, err := c.List() +``` + +## Module path / versioning + +``` +module github.com/dcc-bigfred/microinit/go +``` + +Tag Go releases as **`go/vX.Y.Z`** (required because the module path ends with `/go`), for example: + +```bash +git tag go/v0.1.0 +git push origin go/v0.1.0 +``` + +Then consumers: + +```bash +go get github.com/dcc-bigfred/microinit/go@go/v0.1.0 +``` + +For local monorepo development: + +```go +// go.mod +replace github.com/dcc-bigfred/microinit/go => ../microinit/go +``` + +Private repos need `GOPRIVATE=github.com/dcc-bigfred/*`. + +## API + +| Method | Description | +|--------|-------------| +| `List()` | All services | +| `Status(name)` | One service | +| `Control(name, start\|stop\|restart)` | Lifecycle | +| `Shutdown()` | Halt supervise (caller-owned daemon) | +| `FollowLogs` / `ReadResponse` | Log stream | +| `ValidateName` / `FormatLogLine` | Helpers | + +Default socket: `/data/run/microinit.sock` (`client.DefaultSocket`). diff --git a/go/client/client.go b/go/client/client.go new file mode 100644 index 0000000..750b179 --- /dev/null +++ b/go/client/client.go @@ -0,0 +1,346 @@ +package client + +import ( + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "strings" + "time" +) + +const ( + // DefaultSocket is the hub control socket when DATA_DIR is /data. + DefaultSocket = "/data/run/microinit.sock" + maxFrameBytes = 16 * 1024 * 1024 + defaultTimeout = 10 * time.Second +) + +var ( + ErrInvalidName = errors.New("invalid service name") + ErrInvalidAction = errors.New("invalid action") + ErrNotFound = errors.New("service not found") +) + +// ServiceStatus mirrors microinit IPC list/status entries. +type ServiceStatus struct { + Name string `json:"name"` + State string `json:"state"` + PID *int32 `json:"pid"` + Restarts uint32 `json:"restarts"` + Enabled bool `json:"enabled"` +} + +// LogLine is one captured log line from microinit. +type LogLine struct { + TS string `json:"ts"` + Service string `json:"service"` + Level string `json:"level"` + Msg string `json:"msg"` +} + +type request struct { + Type string `json:"type"` + Name string `json:"name,omitempty"` + Follow *bool `json:"follow,omitempty"` + Lines *uint64 `json:"lines,omitempty"` + Mode string `json:"mode,omitempty"` +} + +// Response is one framed IPC reply (exported for streaming callers). +type Response struct { + Type string `json:"type"` + Message string `json:"message,omitempty"` + Code string `json:"code,omitempty"` + Services []ServiceStatus `json:"services,omitempty"` + Status *ServiceStatus `json:"status,omitempty"` + Line *LogLine `json:"line,omitempty"` +} + +// Client dials the microinit Unix socket. +type Client struct { + Socket string + Timeout time.Duration + // ReadTimeout is the per-frame idle timeout for streaming reads + // (FollowLogs). Zero defaults to 30s. Use a larger value on slow + // embedded links; use a smaller one to detect a dead server faster. + ReadTimeout time.Duration + // Dial is overridden in tests. + Dial func(network, address string, timeout time.Duration) (net.Conn, error) +} + +// defaultReadTimeout is the per-frame idle deadline for FollowLogs streams. +const defaultReadTimeout = 30 * time.Second + +func (c *Client) readTimeout() time.Duration { + if c.ReadTimeout > 0 { + return c.ReadTimeout + } + return defaultReadTimeout +} + +func (c *Client) socketPath() string { + if c.Socket != "" { + return c.Socket + } + return DefaultSocket +} + +func (c *Client) timeout() time.Duration { + if c.Timeout > 0 { + return c.Timeout + } + return defaultTimeout +} + +func (c *Client) dial() (net.Conn, error) { + dial := c.Dial + if dial == nil { + dial = net.DialTimeout + } + conn, err := dial("unix", c.socketPath(), c.timeout()) + if err != nil { + return nil, fmt.Errorf("connect %s: %w (is microinit running?)", c.socketPath(), err) + } + _ = conn.SetDeadline(time.Now().Add(c.timeout())) + return conn, nil +} + +// List returns all services known to microinit. +func (c *Client) List() ([]ServiceStatus, error) { + var resp Response + if err := c.roundTrip(request{Type: "list"}, &resp); err != nil { + return nil, err + } + switch resp.Type { + case "list": + if resp.Services == nil { + return []ServiceStatus{}, nil + } + return resp.Services, nil + case "error": + return nil, responseError(resp) + default: + return nil, fmt.Errorf("unexpected response type %q", resp.Type) + } +} + +// Status returns detailed status for one service. +func (c *Client) Status(name string) (*ServiceStatus, error) { + if err := ValidateName(name); err != nil { + return nil, err + } + var resp Response + if err := c.roundTrip(request{Type: "status", Name: name}, &resp); err != nil { + return nil, err + } + if resp.Type == "error" { + return nil, responseError(resp) + } + if resp.Type != "status" || resp.Status == nil { + return nil, fmt.Errorf("unexpected response type %q", resp.Type) + } + return resp.Status, nil +} + +// Control runs start|stop|restart for a service. +func (c *Client) Control(name, action string) error { + if err := ValidateName(name); err != nil { + return err + } + switch action { + case "start", "stop", "restart": + default: + return ErrInvalidAction + } + var resp Response + if err := c.roundTrip(request{Type: action, Name: name}, &resp); err != nil { + return err + } + switch resp.Type { + case "ok": + return nil + case "error": + return responseError(resp) + default: + return fmt.Errorf("unexpected response type %q", resp.Type) + } +} + +// Shutdown requests a halt-mode shutdown (used when stopping a supervise +// instance started by the caller). +func (c *Client) Shutdown() error { + var resp Response + if err := c.roundTrip(request{Type: "shutdown", Mode: "halt"}, &resp); err != nil { + return err + } + switch resp.Type { + case "ok": + return nil + case "error": + return responseError(resp) + default: + return fmt.Errorf("unexpected response type %q", resp.Type) + } +} + +// FollowLogs opens a streaming connection. Caller must Close the conn. +// +// lines < 0 uses the server default buffer size; lines >= 0 requests exactly +// that many historical lines (0 = live-only, no snapshot). +// +// A read deadline is applied per frame by [Client.ReadFrame]; callers using +// [ReadResponse] must manage deadlines themselves. Follow=true resets the +// deadline per frame so a quiet but live service does not trip it; a dead +// server is detected via io.EOF or the read deadline. +func (c *Client) FollowLogs(name string, lines int, follow bool) (net.Conn, error) { + if name != "" { + if err := ValidateName(name); err != nil { + return nil, err + } + } + conn, err := c.dial() + if err != nil { + return nil, err + } + f := follow + req := request{Type: "logs", Follow: &f} + if lines >= 0 { + n := uint64(lines) + req.Lines = &n + } + if name != "" { + req.Name = name + } + if err := writeFrame(conn, req); err != nil { + _ = conn.Close() + return nil, err + } + _ = conn.SetDeadline(time.Time{}) + return conn, nil +} + +// ReadResponse reads one framed response from a FollowLogs connection. +// It does not enforce a read deadline; callers wanting idle-timeout +// protection should use [Client.ReadFrame] instead, or set the deadline +// on the conn themselves before each call. +func ReadResponse(r io.Reader) (Response, error) { + var resp Response + if err := readFrame(r, &resp); err != nil { + return Response{}, err + } + return resp, nil +} + +// ReadFrame reads one framed response from a streaming connection and +// enforces a per-frame idle read deadline (Client.ReadTimeout, default 30s). +// Use this for FollowLogs streams so a dead server is detected within the +// timeout instead of blocking forever. The deadline is reset before each +// frame, so a quiet-but-live service does not trip it. +func (c *Client) ReadFrame(conn net.Conn) (Response, error) { + _ = conn.SetReadDeadline(time.Now().Add(c.readTimeout())) + var resp Response + if err := readFrame(conn, &resp); err != nil { + return Response{}, err + } + return resp, nil +} + +// FormatLogLine renders a LogLine for console / UI output. +func FormatLogLine(line LogLine) string { + if line.TS == "" { + return fmt.Sprintf("%s: %s", line.Service, line.Msg) + } + return fmt.Sprintf("[%s] %s: %s", line.TS, line.Service, line.Msg) +} + +// ValidateName reports whether name is a safe microinit service id. +func ValidateName(name string) error { + if name == "" || strings.ContainsAny(name, `/\`) || strings.Contains(name, "..") { + return ErrInvalidName + } + for _, r := range name { + if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || + r == '-' || r == '_' || r == '.' { + continue + } + return ErrInvalidName + } + return nil +} + +func (c *Client) roundTrip(req request, resp *Response) error { + conn, err := c.dial() + if err != nil { + return err + } + defer conn.Close() + if err := writeFrame(conn, req); err != nil { + return err + } + return readFrame(conn, resp) +} + +func writeFrame(w io.Writer, msg any) error { + payload, err := json.Marshal(msg) + if err != nil { + return err + } + if len(payload) > maxFrameBytes { + return errors.New("frame too large") + } + var hdr [4]byte + binary.LittleEndian.PutUint32(hdr[:], uint32(len(payload))) + if _, err := w.Write(hdr[:]); err != nil { + return err + } + _, err = w.Write(payload) + return err +} + +func readFrame(r io.Reader, dest any) error { + var hdr [4]byte + if _, err := io.ReadFull(r, hdr[:]); err != nil { + return err + } + n := binary.LittleEndian.Uint32(hdr[:]) + if n > maxFrameBytes { + return fmt.Errorf("frame length %d too large", n) + } + // Stream-decode without pre-allocating the full payload: LimitReader caps + // the bytes consumed, and json.Decoder grows its internal buffer on demand + // rather than reserving n bytes up front. A malicious/buggy server claiming + // a huge frame therefore cannot force a 16 MiB allocation in one shot. + // microinit frames are exactly the JSON payload with no trailing padding, + // so the decoder consumes the whole window and leaves nothing behind. + dec := json.NewDecoder(io.LimitReader(r, int64(n))) + if err := dec.Decode(dest); err != nil { + return err + } + return nil +} + +// responseError maps an IPC error response to a typed error. It prefers the +// stable `code` field (populated by newer microinit for Error::UnknownService +// etc.) and falls back to substring-matching the human message for older +// servers that do not send a code. +func responseError(resp Response) error { + switch resp.Code { + case "not_found": + return ErrNotFound + case "disabled": + return fmt.Errorf("%s: %w", resp.Message, ErrNotFound) + case "": + // Legacy server without a code field. + lower := strings.ToLower(resp.Message) + if strings.Contains(lower, "unknown") || strings.Contains(lower, "not found") { + return ErrNotFound + } + } + if resp.Message == "" { + return errors.New("microinit request failed") + } + return errors.New(resp.Message) +} diff --git a/go/client/client_extra_test.go b/go/client/client_extra_test.go new file mode 100644 index 0000000..8a26190 --- /dev/null +++ b/go/client/client_extra_test.go @@ -0,0 +1,192 @@ +package client_test + +import ( + "encoding/binary" + "errors" + "io" + "net" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/dcc-bigfred/microinit/go/client" +) + +// TestReadFrameTooLarge verifies a server claiming a frame larger than +// maxFrameBytes is rejected without allocating the full buffer (3.4). +func TestReadFrameTooLarge(t *testing.T) { + sock := filepath.Join(t.TempDir(), "microinit.sock") + ln, err := net.Listen("unix", sock) + if err != nil { + t.Fatal(err) + } + defer ln.Close() + + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + // Read the request frame so the client gets past the write. + _, _ = readReq(conn) + // Send a header claiming 32 MiB (> 16 MiB cap) with no body. + var hdr [4]byte + binary.LittleEndian.PutUint32(hdr[:], 32*1024*1024) + _, _ = conn.Write(hdr[:]) + // Keep the connection open briefly so the client reads the header. + time.Sleep(500 * time.Millisecond) + }() + + c := &client.Client{Socket: sock, Timeout: 2 * time.Second} + _, err = c.List() + if err == nil { + t.Fatal("expected error for oversized frame, got nil") + } + if !strings.Contains(err.Error(), "too large") { + t.Fatalf("expected 'too large' error, got %v", err) + } +} + +// TestResponseErrorCodeMapping verifies the stable `code` field maps to +// ErrNotFound without relying on substring matching (3.5). +func TestResponseErrorCodeMapping(t *testing.T) { + sock := filepath.Join(t.TempDir(), "microinit.sock") + ln, err := net.Listen("unix", sock) + if err != nil { + t.Fatal(err) + } + defer ln.Close() + + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + _, _ = readReq(conn) + // Stable code, message does NOT contain "unknown"/"not found". + _ = writeResp(conn, map[string]any{ + "type": "error", + "message": "service 'redis' is not registered", + "code": "not_found", + }) + }() + + c := &client.Client{Socket: sock, Timeout: 2 * time.Second} + _, err = c.Status("redis") + if !errors.Is(err, client.ErrNotFound) { + t.Fatalf("expected ErrNotFound via code, got %v", err) + } +} + +// TestResponseErrorLegacyFallback verifies a server without a `code` field +// still maps via the legacy substring heuristic (backward compat, 3.5). +func TestResponseErrorLegacyFallback(t *testing.T) { + sock := filepath.Join(t.TempDir(), "microinit.sock") + ln, err := net.Listen("unix", sock) + if err != nil { + t.Fatal(err) + } + defer ln.Close() + + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + _, _ = readReq(conn) + _ = writeResp(conn, map[string]any{ + "type": "error", + "message": "unknown service 'redis'", + }) + }() + + c := &client.Client{Socket: sock, Timeout: 2 * time.Second} + _, err = c.Status("redis") + if !errors.Is(err, client.ErrNotFound) { + t.Fatalf("expected ErrNotFound via legacy message, got %v", err) + } +} + +// TestFollowLogsReadDeadline verifies ReadFrame enforces a per-frame idle +// deadline so a silent server is detected instead of blocking forever (3.6). +func TestFollowLogsReadDeadline(t *testing.T) { + sock := filepath.Join(t.TempDir(), "microinit.sock") + ln, err := net.Listen("unix", sock) + if err != nil { + t.Fatal(err) + } + defer ln.Close() + + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + _, _ = readReq(conn) + // Never respond. Keep the connection open. + time.Sleep(3 * time.Second) + }() + + c := &client.Client{Socket: sock, Timeout: 2 * time.Second, ReadTimeout: 200 * time.Millisecond} + conn, err := c.FollowLogs("redis", 0, true) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + start := time.Now() + _, err = c.ReadFrame(conn) + elapsed := time.Since(start) + if err == nil { + t.Fatal("expected deadline error, got nil") + } + // Should fire around the 200ms deadline, well under the 2s dial timeout. + if elapsed > 1500*time.Millisecond { + t.Fatalf("read deadline did not fire in time: %v", elapsed) + } +} + +// TestReadFrameStreamingDecode verifies a normal multi-KB frame decodes +// correctly through the streaming decoder (3.4 regression guard). +func TestReadFrameStreamingDecode(t *testing.T) { + sock := filepath.Join(t.TempDir(), "microinit.sock") + ln, err := net.Listen("unix", sock) + if err != nil { + t.Fatal(err) + } + defer ln.Close() + + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + _, _ = readReq(conn) + services := make([]map[string]any, 256) + for i := range services { + services[i] = map[string]any{ + "name": "svc-" + strings.Repeat("x", 60), "state": "running", + "pid": 1, "restarts": 0, "enabled": true, + } + } + _ = writeResp(conn, map[string]any{"type": "list", "services": services}) + }() + + c := &client.Client{Socket: sock, Timeout: 2 * time.Second} + list, err := c.List() + if err != nil { + t.Fatal(err) + } + if len(list) != 256 { + t.Fatalf("expected 256 services, got %d", len(list)) + } +} + +// Ensure the unused io import in this file is referenced (writeResp/readReq use it). +var _ = io.EOF diff --git a/go/client/client_test.go b/go/client/client_test.go new file mode 100644 index 0000000..7ab74ce --- /dev/null +++ b/go/client/client_test.go @@ -0,0 +1,141 @@ +package client_test + +import ( + "encoding/binary" + "encoding/json" + "errors" + "io" + "net" + "path/filepath" + "testing" + "time" + + "github.com/dcc-bigfred/microinit/go/client" +) + +func TestClientListAndControl(t *testing.T) { + sock := filepath.Join(t.TempDir(), "microinit.sock") + ln, err := net.Listen("unix", sock) + if err != nil { + t.Fatal(err) + } + defer ln.Close() + + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go handleFake(conn) + } + }() + + c := &client.Client{Socket: sock, Timeout: 2 * time.Second} + list, err := c.List() + if err != nil { + t.Fatal(err) + } + if len(list) != 1 || list[0].Name != "redis" || list[0].State != "running" { + t.Fatalf("list: %+v", list) + } + st, err := c.Status("redis") + if err != nil { + t.Fatal(err) + } + if st.Name != "redis" || st.State != "running" { + t.Fatalf("status: %+v", st) + } + if err := c.Control("redis", "restart"); err != nil { + t.Fatal(err) + } + if err := c.Control("redis", "pause"); !errors.Is(err, client.ErrInvalidAction) { + t.Fatalf("expected ErrInvalidAction, got %v", err) + } + if err := client.ValidateName("../x"); !errors.Is(err, client.ErrInvalidName) { + t.Fatalf("expected ErrInvalidName, got %v", err) + } + line := client.FormatLogLine(client.LogLine{TS: "t", Service: "redis", Msg: "hi"}) + if line != "[t] redis: hi" { + t.Fatalf("FormatLogLine: %q", line) + } +} + +func handleFake(conn net.Conn) { + defer conn.Close() + req, err := readReq(conn) + if err != nil { + return + } + switch req.Type { + case "list": + _ = writeResp(conn, map[string]any{ + "type": "list", + "services": []map[string]any{{ + "name": "redis", "state": "running", "pid": 42, + "restarts": 0, "enabled": true, + }}, + }) + case "status": + _ = writeResp(conn, map[string]any{ + "type": "status", + "status": map[string]any{ + "name": req.Name, "state": "running", "pid": 42, + "restarts": 0, "enabled": true, + }, + }) + case "start", "stop", "restart": + _ = writeResp(conn, map[string]any{"type": "ok"}) + case "shutdown": + _ = writeResp(conn, map[string]any{"type": "ok"}) + case "logs": + _ = writeResp(conn, map[string]any{ + "type": "log", + "line": map[string]any{ + "ts": "t", "service": req.Name, "level": "stdout", "msg": "hello", + }, + }) + if req.Follow != nil && !*req.Follow { + _ = writeResp(conn, map[string]any{"type": "ok"}) + } + default: + _ = writeResp(conn, map[string]any{"type": "error", "message": "unknown"}) + } +} + +type fakeReq struct { + Type string `json:"type"` + Name string `json:"name"` + Follow *bool `json:"follow"` + Lines *uint64 `json:"lines"` + Mode string `json:"mode"` +} + +func readReq(r io.Reader) (fakeReq, error) { + var hdr [4]byte + if _, err := io.ReadFull(r, hdr[:]); err != nil { + return fakeReq{}, err + } + n := binary.LittleEndian.Uint32(hdr[:]) + buf := make([]byte, n) + if _, err := io.ReadFull(r, buf); err != nil { + return fakeReq{}, err + } + var req fakeReq + err := json.Unmarshal(buf, &req) + return req, err +} + +func writeResp(w io.Writer, msg any) error { + payload, err := json.Marshal(msg) + if err != nil { + return err + } + var hdr [4]byte + binary.LittleEndian.PutUint32(hdr[:], uint32(len(payload))) + if _, err := w.Write(hdr[:]); err != nil { + return err + } + _, err = w.Write(payload) + return err +} diff --git a/go/client/doc.go b/go/client/doc.go new file mode 100644 index 0000000..8e8b13b --- /dev/null +++ b/go/client/doc.go @@ -0,0 +1,4 @@ +// Package client talks to a running microinit daemon over its Unix control socket. +// +// Protocol: 4-byte little-endian length prefix + UTF-8 JSON frame (max 16 MiB). +package client diff --git a/go/go.mod b/go/go.mod new file mode 100644 index 0000000..659531e --- /dev/null +++ b/go/go.mod @@ -0,0 +1,3 @@ +module github.com/dcc-bigfred/microinit/go + +go 1.22