diff --git a/docs/README.md b/docs/README.md index e8f429e..acf14b9 100644 --- a/docs/README.md +++ b/docs/README.md @@ -7,6 +7,8 @@ Guides and references for running and integrating **microinit**. | [Operator guide](operator.md) | Linux admins / device operators | Managing services from the shell, writing JSON config, dependencies, boot sequence | | [Control socket API](api.md) | Integrators / UI / scripts | Unix socket framing, request/response JSON | | [Architecture](architecture.md) | Developers | Design overview: `init` vs `supervise`, reload, OTel, distribution | +| [Developer index](developer.md) | Contributors / embedders | Doc map + Go SDK pointer | +| [Go SDK](sdk/golang.md) | Go integrators | `client` / `config` / `supervise` with examples | Also see the man pages in the repository: diff --git a/docs/api.md b/docs/api.md index 1ab9cb7..fb74ac4 100644 --- a/docs/api.md +++ b/docs/api.md @@ -174,13 +174,14 @@ Operators normally use the companion `shutdown` binary (`shutdown -r now`, …) "pid": 1234, "restarts": 0, "liveness_failures": 0, - "enabled": true + "enabled": true, + "labels": { "created-by": "bigfred" } } ] } ``` -`pid` may be `null` when not tracked. `liveness_failures` counts how many times `livenessProbe` failed since boot (or since the service was added on reload). +`pid` may be `null` when not tracked. `liveness_failures` counts how many times `livenessProbe` failed since boot (or since the service was added on reload). `labels` is omitted when empty; keys come from the service config / drop-in. ### `status` @@ -193,7 +194,8 @@ Operators normally use the companion `shutdown` binary (`shutdown -r now`, …) "pid": 1234, "restarts": 0, "liveness_failures": 0, - "enabled": true + "enabled": true, + "labels": { "created-by": "bigfred" } } } ``` diff --git a/docs/developer.md b/docs/developer.md new file mode 100644 index 0000000..c26a244 --- /dev/null +++ b/docs/developer.md @@ -0,0 +1,37 @@ +# Developer documentation + +Index of docs for people changing or integrating **microinit**. + +## Guides + +| Document | Contents | +|----------|----------| +| [Architecture](architecture.md) | `init` vs `supervise`, reload, OTel, distribution | +| [Control socket API](api.md) | Unix socket framing and JSON messages | +| [Operator guide](operator.md) | Shell usage, JSON config, boot sequence | +| [Documentation index](README.md) | Operator-oriented entry point | + +## SDKs + +| Document | Language | Contents | +|----------|----------|----------| +| [Go SDK](sdk/golang.md) | Go | `client`, `config`, `supervise` — embed or control microinit | + +## Source layout (Go) + +``` +go/ + go.mod # module github.com/dcc-bigfred/microinit/go + client/ # IPC client + config/ # ServiceDef + drop-ins + supervise/ # EnsureRunning / Shutdown host + README.md +``` + +Version tags: `go/vX.Y.Z`. See [Go SDK](sdk/golang.md) for import examples. + +## Man pages / examples + +- `man/man8/microinit.8.mdoc` — CLI +- `man/man5/microinit.json.5.mdoc` — config fields +- [`examples/microinit.json.example`](../examples/microinit.json.example) diff --git a/docs/operator.md b/docs/operator.md index d4c5a92..b37a9bd 100644 --- a/docs/operator.md +++ b/docs/operator.md @@ -24,6 +24,9 @@ The control socket defaults to `$DATA_DIR/run/microinit.sock` (hub: `/data/run/m ```bash microinit list # name, state, pid, restarts, enabled, live_fail +microinit list --show-labels # same + LABELS column +microinit list -l created-by=bigfred # filter (AND if -l repeated) +microinit describe redis # deps, events, labels microinit describe redis # deps, reverse deps, graph, recent events microinit start redis microinit start --force alloy # start even if dependsOn are not ready @@ -91,7 +94,7 @@ Minimal long-running service (foreground binary — preferred so microinit can t "name": "myapp", "enabled": true, "daemon": true, - "restart": true, + "restartPolicy": "onError", "restartBackoff": 2, "startWaitSecs": 1, "shutdownWaitSecs": 5, @@ -114,14 +117,14 @@ Or set explicit commands: "stopCmd": "killall myapp" ``` -If `startCmd` is set, it is used instead of `cmd start`. Prefer **`exec` of the real process in the foreground** in the start script so `killall` / crashes are visible to microinit and `restart: true` works. +If `startCmd` is set, it is used instead of `cmd start`. Prefer **`exec` of the real process in the foreground** in the start script so `killall` / crashes are visible to microinit and `restartPolicy` works. ### Important fields (plain language) | Field | Role | |-------|------| | `daemon` | `true` = long-lived; `false` = one-shot job | -| `restart` | Restart after crash (daemons only) | +| `restartPolicy` | `always` / `onError` (default) / `none` — auto-restart (daemons only) | | `restartBackoff` | Seconds to wait before restarting | | `startWaitSecs` | After start, wait this long; if the process dies in that window → `failed`. Use `1` (or more) when the start command **stays** as the service process | | `shutdownWaitSecs` | After stop, wait then `SIGKILL` | @@ -241,7 +244,7 @@ Edit `/data/etc/microinit.json` (or a drop-in), save — wait a moment for reloa **Service dies and stays dead** -Check `restart: true` and that microinit is tracking a real PID (`list` shows a PID). Scripts that background with `start-stop-daemon -b` and exit leave microinit thinking the service is fine with no PID — prefer foreground `exec`. +Check `restartPolicy` and that microinit is tracking a real PID (`list` shows a PID). Scripts that background with `start-stop-daemon -b` and exit leave microinit thinking the service is fine with no PID — prefer foreground `exec`. --- diff --git a/docs/sdk/golang.md b/docs/sdk/golang.md new file mode 100644 index 0000000..7338849 --- /dev/null +++ b/docs/sdk/golang.md @@ -0,0 +1,205 @@ +# Go SDK + +Embed or control [microinit](https://github.com/dcc-bigfred/microinit) from Go. + +## Module + +``` +github.com/dcc-bigfred/microinit/go +``` + +Tag releases as `go/vX.Y.Z` (required because the module path ends with `/go`). + +```bash +go get github.com/dcc-bigfred/microinit/go@go/v0.3.0 +``` + +Private repos: `GOPRIVATE=github.com/dcc-bigfred/*`. + +Local monorepo: + +```go +replace github.com/dcc-bigfred/microinit/go => ../microinit/go +``` + +## Packages + +| Package | Import path | Role | +|---------|-------------|------| +| **client** | `…/go/client` | IPC to a running daemon | +| **config** | `…/go/config` | `ServiceDef`, drop-in read/write, labels helpers | +| **supervise** | `…/go/supervise` | Join or spawn `microinit supervise` inside your process | + +Default control socket: `/data/run/microinit.sock` (`client.DefaultSocket`). + +## Labels + +Service configs may include `labels` (`map[string]string`). Convention for embedders: + +```go +svc := config.WithCreatedBy(config.ServiceDef{ + Name: "worker", StartCmd: "exec /usr/bin/worker", +}, "my-app") +// writes labels: {"created-by":"my-app"} + +// Filter a List() result: +for _, s := range list { + if config.MatchLabels(s.Labels, map[string]string{config.LabelCreatedBy: "my-app"}) { + fmt.Println(s.Name) + } +} +``` + +CLI: `microinit list -l created-by=my-app` and `microinit list --show-labels`. + +## Design: process vs product policy + +`supervise.Host` only manages the **daemon process**: + +- join an existing socket, or spawn one supervise instance +- `Shutdown` only if **this** Host spawned the process + +Stopping services, tracking “owned” drop-ins, refusing system service names, Redis/Alloy templates — that stays in the application (e.g. bigfred). + +```mermaid +flowchart LR + app[Your app] + host[supervise.Host] + cfg[config drop-ins] + cli[client IPC] + mi[microinit process] + app --> host + app --> cfg + host --> cli + host -->|spawn or join| mi + cfg -->|JSON files| mi + cli --> mi +``` + +## Example: IPC only (admin UI) + +```go +package main + +import ( + "fmt" + "log" + + "github.com/dcc-bigfred/microinit/go/client" +) + +func main() { + c := &client.Client{Socket: client.DefaultSocket} + list, err := c.List() + if err != nil { + log.Fatal(err) + } + for _, s := range list { + fmt.Printf("%s %s\n", s.Name, s.State) + } + if err := c.Control("redis", "restart"); err != nil { + log.Fatal(err) + } +} +``` + +## Example: embed microinit in your process + +```go +package main + +import ( + "context" + "log" + "os" + "os/signal" + + "github.com/dcc-bigfred/microinit/go/config" + "github.com/dcc-bigfred/microinit/go/supervise" +) + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + defer stop() + + data := "/data" // or your DATA_DIR + h := supervise.New( + data+"/run/microinit.sock", + "microinit", + data+"/etc/microinit.json", + data+"/etc/microinit.d/services", + ) + + joined, err := h.EnsureRunning(ctx) + if err != nil { + log.Fatal(err) + } + log.Printf("microinit ready (joined=%v)", joined) + + // Application policy: write only your drop-ins. + _ = config.WriteDropin(h.DropinDir, "app", "worker", config.WithCreatedBy(config.ServiceDef{ + Name: "worker", + Enabled: config.BoolPtr(true), + Daemon: config.BoolPtr(true), + RestartPolicy: config.RestartOnError, + StartCmd: "exec /usr/bin/my-worker", + }, "my-app")) + + <-ctx.Done() + + // Application policy: stop services you started (optional). + _ = h.Client().Control("worker", "stop") + + // SDK: tear down the process only if we spawned it. + if err := h.Shutdown(context.Background()); err != nil { + log.Fatal(err) + } +} +``` + +## Example: respect system services from base config + +```go +system, err := config.BaseConfigServiceNames("/data/etc/microinit.json") +if err != nil { + log.Fatal(err) +} +if _, ok := system["redis"]; ok { + log.Fatal("refusing to overwrite system service redis") +} +err = config.WriteDropin(dropinDir, "infra", "redis", svc) +``` + +## client API (summary) + +| Method | Description | +|--------|-------------| +| `List()` | All services | +| `Status(name)` | One service | +| `Control(name, start\|stop\|restart)` | Lifecycle | +| `Shutdown()` | Halt-mode shutdown (IPC) | +| `FollowLogs` / `ReadResponse` | Log stream | +| `ValidateName` / `FormatLogLine` | Helpers | + +## config API (summary) + +| Function | Description | +|----------|-------------| +| `WriteDropin` / `RemoveDropin` | Single file under `{dir}/{group}/{name}.json` | +| `SyncGroup` / `ListGroup` | Reconcile a group directory | +| `DropinExists` | Presence check | +| `BaseConfigServiceNames` | Names from main `microinit.json` | +| `WithCreatedBy` / `MatchLabels` | Label helpers (`created-by`) | +| `BoolPtr` / `IntPtr` | Optional JSON helpers | + +## supervise API (summary) + +| Method | Description | +|--------|-------------| +| `New(socket, bin, configPath, dropinDir)` | Construct host | +| `EnsureRunning(ctx) (joined, err)` | Join or spawn + wait for IPC | +| `Client()` | Bound IPC client | +| `Spawned()` | Whether this host owns the process | +| `Shutdown(ctx)` | Stop process **only if spawned** | + +Also see [module README](../../go/README.md) and [Control socket API](../api.md). diff --git a/examples/microinit.json.example b/examples/microinit.json.example index 225d969..d553443 100644 --- a/examples/microinit.json.example +++ b/examples/microinit.json.example @@ -14,7 +14,7 @@ "name": "network", "enabled": true, "daemon": false, - "restart": false, + "restartPolicy": "none", "restartBackoff": 2, "successExitCodes": [ 0 @@ -34,7 +34,7 @@ "name": "redis", "enabled": true, "daemon": true, - "restart": true, + "restartPolicy": "onError", "restartBackoff": 2, "successExitCodes": [ 0 @@ -56,7 +56,7 @@ "name": "remote-icmp", "enabled": true, "daemon": true, - "restart": true, + "restartPolicy": "onError", "restartBackoff": 5, "successExitCodes": [ 0 diff --git a/go/README.md b/go/README.md new file mode 100644 index 0000000..7093abc --- /dev/null +++ b/go/README.md @@ -0,0 +1,46 @@ +# Go SDK for microinit + +Module: `github.com/dcc-bigfred/microinit/go` + +| Package | Import | Role | +|---------|--------|------| +| client | `github.com/dcc-bigfred/microinit/go/client` | IPC (list/control/logs) | +| config | `github.com/dcc-bigfred/microinit/go/config` | ServiceDef + drop-ins | +| supervise | `github.com/dcc-bigfred/microinit/go/supervise` | Join/spawn daemon in-process | + +Full guide with examples: **[docs/sdk/golang.md](../docs/sdk/golang.md)**. Developer index: **[docs/developer.md](../docs/developer.md)**. + +## Install + +```bash +go get github.com/dcc-bigfred/microinit/go@go/v0.3.0 +``` + +Tag Go releases as **`go/vX.Y.Z`**. Private: `GOPRIVATE=github.com/dcc-bigfred/*`. + +```go +// local monorepo +replace github.com/dcc-bigfred/microinit/go => ../microinit/go +``` + +## Quick start + +```go +import ( + "github.com/dcc-bigfred/microinit/go/client" + "github.com/dcc-bigfred/microinit/go/config" + "github.com/dcc-bigfred/microinit/go/supervise" +) + +c := &client.Client{Socket: client.DefaultSocket} +list, err := c.List() + +svc := config.WithCreatedBy(config.ServiceDef{Name: "worker", StartCmd: "exec worker"}, "my-app") + +h := supervise.New(socket, "microinit", configPath, dropinDir) +joined, err := h.EnsureRunning(ctx) +// … app writes drop-ins / stops its services … +err = h.Shutdown(ctx) // no-op if joined +``` + +Default socket: `/data/run/microinit.sock`. diff --git a/go/client/client.go b/go/client/client.go new file mode 100644 index 0000000..088f832 --- /dev/null +++ b/go/client/client.go @@ -0,0 +1,362 @@ +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"` + LivenessFailures uint32 `json:"liveness_failures,omitempty"` + Labels map[string]string `json:"labels,omitempty"` +} + +// 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). +// +// Prefer [Client.ReadFrame] for follow streams: it enforces a per-frame idle +// deadline (default 30s) and skips server heartbeats so a quiet but live +// service does not time out. +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). +// Server heartbeats (type "heartbeat") are skipped so a quiet-but-live +// follow stream stays open; a dead server is detected via deadline or EOF. +func (c *Client) ReadFrame(conn net.Conn) (Response, error) { + for { + _ = conn.SetReadDeadline(time.Now().Add(c.readTimeout())) + var resp Response + if err := readFrame(conn, &resp); err != nil { + return Response{}, err + } + if resp.Type == "heartbeat" { + continue + } + 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 := writeFull(w, hdr[:]); err != nil { + return err + } + return writeFull(w, payload) +} + +func writeFull(w io.Writer, p []byte) error { + for len(p) > 0 { + n, err := w.Write(p) + if n > 0 { + p = p[n:] + } + if err != nil { + return err + } + if n == 0 { + return io.ErrShortWrite + } + } + return nil +} + +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) + } + // Read exactly n bytes so the next frame stays aligned, then decode. + // Size is capped by maxFrameBytes above. + buf := make([]byte, n) + if _, err := io.ReadFull(r, buf); err != nil { + return err + } + return json.Unmarshal(buf, dest) +} + +// responseError maps an IPC error response to a typed error. It prefers the +// stable `code` field (populated by 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/client/frame_test.go b/go/client/frame_test.go new file mode 100644 index 0000000..9b15b0d --- /dev/null +++ b/go/client/frame_test.go @@ -0,0 +1,57 @@ +package client + +import ( + "bytes" + "testing" +) + +func TestWriteFullShortWrites(t *testing.T) { + var buf bytes.Buffer + w := &shortWriter{w: &buf, max: 3} + if err := writeFrame(w, map[string]string{"type": "ok"}); err != nil { + t.Fatal(err) + } + var resp Response + if err := readFrame(bytes.NewReader(buf.Bytes()), &resp); err != nil { + t.Fatal(err) + } + if resp.Type != "ok" { + t.Fatalf("got %+v", resp) + } +} + +type shortWriter struct { + w *bytes.Buffer + max int +} + +func (s *shortWriter) Write(p []byte) (int, error) { + if len(p) > s.max { + p = p[:s.max] + } + return s.w.Write(p) +} + +func TestReadFrameExactLength(t *testing.T) { + var buf bytes.Buffer + if err := writeFrame(&buf, map[string]any{"type": "ok"}); err != nil { + t.Fatal(err) + } + if err := writeFrame(&buf, map[string]any{"type": "list", "services": []any{}}); err != nil { + t.Fatal(err) + } + r := bytes.NewReader(buf.Bytes()) + var a, b Response + if err := readFrame(r, &a); err != nil { + t.Fatal(err) + } + if a.Type != "ok" { + t.Fatalf("first: %+v", a) + } + if err := readFrame(r, &b); err != nil { + t.Fatal(err) + } + if b.Type != "list" { + t.Fatalf("second: %+v", b) + } +} diff --git a/go/config/baseconfig.go b/go/config/baseconfig.go new file mode 100644 index 0000000..7c60262 --- /dev/null +++ b/go/config/baseconfig.go @@ -0,0 +1,39 @@ +package config + +import ( + "encoding/json" + "fmt" + "os" +) + +// baseConfigFile is the subset of microinit.json needed to list declared services. +type baseConfigFile struct { + Services []struct { + Name string `json:"name"` + } `json:"services"` +} + +// BaseConfigServiceNames returns service names declared in the main +// microinit.json (configPath). Callers often treat these as system-owned and +// refuse to overwrite them with drop-ins. +func BaseConfigServiceNames(configPath string) (map[string]struct{}, error) { + data, err := os.ReadFile(configPath) + if err != nil { + if os.IsNotExist(err) { + return map[string]struct{}{}, nil + } + return nil, fmt.Errorf("read microinit config %s: %w", configPath, err) + } + var cfg baseConfigFile + if err := json.Unmarshal(data, &cfg); err != nil { + return nil, fmt.Errorf("parse microinit config %s: %w", configPath, err) + } + out := make(map[string]struct{}, len(cfg.Services)) + for _, svc := range cfg.Services { + if svc.Name == "" { + continue + } + out[svc.Name] = struct{}{} + } + return out, nil +} diff --git a/go/config/doc.go b/go/config/doc.go new file mode 100644 index 0000000..9f3698e --- /dev/null +++ b/go/config/doc.go @@ -0,0 +1,2 @@ +// Package config provides microinit service definition types and drop-in helpers. +package config diff --git a/go/config/dropin.go b/go/config/dropin.go new file mode 100644 index 0000000..6299baa --- /dev/null +++ b/go/config/dropin.go @@ -0,0 +1,151 @@ +package config + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + + "github.com/dcc-bigfred/microinit/go/client" +) + +// WriteDropin writes a single-service drop-in at dir/group/name.json. +func WriteDropin(dir, group, name string, svc ServiceDef) error { + if err := client.ValidateName(group); err != nil { + return fmt.Errorf("drop-in group: %w", err) + } + if err := client.ValidateName(name); err != nil { + return fmt.Errorf("drop-in name: %w", err) + } + if svc.Name == "" { + svc.Name = name + } + if svc.Name != name { + return fmt.Errorf("drop-in name %q does not match service %q", name, svc.Name) + } + content, err := json.MarshalIndent(DropinFile{Services: []ServiceDef{svc}}, "", " ") + if err != nil { + return err + } + return WriteFileAtomically(filepath.Join(dir, group, name+".json"), append(content, '\n')) +} + +// RemoveDropin deletes dir/group/name.json (no-op if missing). +func RemoveDropin(dir, group, name string) error { + if err := client.ValidateName(group); err != nil { + return err + } + if err := client.ValidateName(name); err != nil { + return err + } + err := os.Remove(filepath.Join(dir, group, name+".json")) + if os.IsNotExist(err) { + return nil + } + return err +} + +// SyncGroup makes dir/group contain exactly the services in desired. +func SyncGroup(dir, group string, desired map[string]ServiceDef) error { + if err := client.ValidateName(group); err != nil { + return err + } + groupDir := filepath.Join(dir, group) + if err := os.MkdirAll(groupDir, 0o755); err != nil { + return err + } + entries, err := os.ReadDir(groupDir) + if err != nil { + return err + } + for _, entry := range entries { + if entry.IsDir() || filepath.Ext(entry.Name()) != ".json" { + continue + } + name := entry.Name()[:len(entry.Name())-len(".json")] + if _, ok := desired[name]; !ok { + if err := os.Remove(filepath.Join(groupDir, entry.Name())); err != nil { + return err + } + } + } + names := make([]string, 0, len(desired)) + for name := range desired { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + if err := WriteDropin(dir, group, name, desired[name]); err != nil { + return err + } + } + return nil +} + +// ListGroup reads all drop-ins under dir/group. +func ListGroup(dir, group string) (map[string]ServiceDef, error) { + groupDir := filepath.Join(dir, group) + entries, err := os.ReadDir(groupDir) + if os.IsNotExist(err) { + return map[string]ServiceDef{}, nil + } + if err != nil { + return nil, err + } + out := make(map[string]ServiceDef) + for _, entry := range entries { + if entry.IsDir() || filepath.Ext(entry.Name()) != ".json" { + continue + } + data, err := os.ReadFile(filepath.Join(groupDir, entry.Name())) + if err != nil { + return nil, err + } + var dropin DropinFile + if err := json.Unmarshal(data, &dropin); err != nil { + return nil, err + } + for _, svc := range dropin.Services { + out[svc.Name] = svc + } + } + return out, nil +} + +// DropinExists reports whether a drop-in file is present for group/name. +func DropinExists(dir, group, name string) bool { + if err := client.ValidateName(group); err != nil { + return false + } + if err := client.ValidateName(name); err != nil { + return false + } + _, err := os.Stat(filepath.Join(dir, group, name+".json")) + return err == nil +} + +// WriteFileAtomically creates parent dirs and renames into place. +func WriteFileAtomically(path string, content []byte) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + tmp, err := os.CreateTemp(filepath.Dir(path), ".tmp-*") + if err != nil { + return err + } + tmpName := tmp.Name() + defer os.Remove(tmpName) + if err := tmp.Chmod(0o600); err != nil { + _ = tmp.Close() + return err + } + if _, err := tmp.Write(content); err != nil { + _ = tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(tmpName, path) +} diff --git a/go/config/dropin_test.go b/go/config/dropin_test.go new file mode 100644 index 0000000..017e3fc --- /dev/null +++ b/go/config/dropin_test.go @@ -0,0 +1,74 @@ +package config_test + +import ( + "os" + "path/filepath" + "testing" + + "github.com/dcc-bigfred/microinit/go/config" +) + +func TestWriteDropinAndListGroup(t *testing.T) { + dir := t.TempDir() + svc := config.ServiceDef{ + Name: "redis", + Enabled: config.BoolPtr(true), + StartCmd: "exec redis-server", + } + if err := config.WriteDropin(dir, "infra", "redis", svc); err != nil { + t.Fatal(err) + } + if !config.DropinExists(dir, "infra", "redis") { + t.Fatal("expected drop-in") + } + got, err := config.ListGroup(dir, "infra") + if err != nil { + t.Fatal(err) + } + if got["redis"].StartCmd != "exec redis-server" { + t.Fatalf("%+v", got["redis"]) + } + if err := config.SyncGroup(dir, "infra", map[string]config.ServiceDef{}); err != nil { + t.Fatal(err) + } + if config.DropinExists(dir, "infra", "redis") { + t.Fatal("expected removed") + } +} + +func TestBaseConfigServiceNames(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "microinit.json") + if err := os.WriteFile(path, []byte(`{ + "services": [ + {"name": "redis", "cmd": "/etc/init.d/redis"}, + {"name": "alloy", "cmd": "/etc/init.d/alloy"}, + {"name": ""} + ] +}`), 0o644); err != nil { + t.Fatal(err) + } + names, err := config.BaseConfigServiceNames(path) + if err != nil { + t.Fatal(err) + } + if _, ok := names["redis"]; !ok { + t.Fatal("expected redis") + } + if _, ok := names["alloy"]; !ok { + t.Fatal("expected alloy") + } + if len(names) != 2 { + t.Fatalf("len=%d want 2", len(names)) + } +} + +func TestBaseConfigServiceNamesMissingFile(t *testing.T) { + names, err := config.BaseConfigServiceNames(filepath.Join(t.TempDir(), "missing.json")) + if err != nil { + t.Fatal(err) + } + if len(names) != 0 { + t.Fatalf("len=%d", len(names)) + } +} diff --git a/go/config/labels.go b/go/config/labels.go new file mode 100644 index 0000000..deb3383 --- /dev/null +++ b/go/config/labels.go @@ -0,0 +1,32 @@ +package config + +// LabelCreatedBy is the conventional label for the embedding application. +const LabelCreatedBy = "created-by" + +// WithCreatedBy returns a copy of svc with labels["created-by"]=who. +// Existing labels are preserved; created-by is set/overwritten. +func WithCreatedBy(svc ServiceDef, who string) ServiceDef { + out := svc + if out.Labels == nil { + out.Labels = map[string]string{} + } else { + cp := make(map[string]string, len(out.Labels)+1) + for k, v := range out.Labels { + cp[k] = v + } + out.Labels = cp + } + out.Labels[LabelCreatedBy] = who + return out +} + +// MatchLabels reports whether have contains every key=value in want (AND). +// An empty want always matches. +func MatchLabels(have, want map[string]string) bool { + for k, v := range want { + if have[k] != v { + return false + } + } + return true +} diff --git a/go/config/labels_test.go b/go/config/labels_test.go new file mode 100644 index 0000000..a131fbd --- /dev/null +++ b/go/config/labels_test.go @@ -0,0 +1,33 @@ +package config_test + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/dcc-bigfred/microinit/go/config" +) + +func TestWithCreatedByAndMatchLabels(t *testing.T) { + svc := config.WithCreatedBy(config.ServiceDef{ + Name: "redis", + StartCmd: "exec redis-server", + Labels: map[string]string{"env": "prod"}, + }, "bigfred") + if svc.Labels[config.LabelCreatedBy] != "bigfred" || svc.Labels["env"] != "prod" { + t.Fatalf("%+v", svc.Labels) + } + raw, err := json.Marshal(svc) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(raw), `"created-by":"bigfred"`) { + t.Fatalf("json: %s", raw) + } + if !config.MatchLabels(svc.Labels, map[string]string{config.LabelCreatedBy: "bigfred"}) { + t.Fatal("expected match") + } + if config.MatchLabels(svc.Labels, map[string]string{config.LabelCreatedBy: "other"}) { + t.Fatal("expected no match") + } +} diff --git a/go/config/shell.go b/go/config/shell.go new file mode 100644 index 0000000..de2b43e --- /dev/null +++ b/go/config/shell.go @@ -0,0 +1,21 @@ +package config + +import "strings" + +// ShellQuote wraps value in single quotes, escaping any embedded single +// quotes so the result is safe to interpolate into a POSIX shell command +// line. Use it for every argv token assembled into ServiceDef.StartCmd. +func ShellQuote(value string) string { + return "'" + strings.ReplaceAll(value, "'", "'\\''") + "'" +} + +// BuildStartCmd joins args into a single shell command line, shell-quoting +// each token and prefixing with "exec " so microinit's spawn shell replaces +// itself with the daemon (no extra shell PID lingering). +func BuildStartCmd(args []string) string { + quoted := make([]string, len(args)) + for i, a := range args { + quoted[i] = ShellQuote(a) + } + return "exec " + strings.Join(quoted, " ") +} diff --git a/go/config/types.go b/go/config/types.go new file mode 100644 index 0000000..44a3399 --- /dev/null +++ b/go/config/types.go @@ -0,0 +1,49 @@ +package config + +// ServiceDef is one service entry in microinit.json or a drop-in file. +type ServiceDef struct { + Name string `json:"name"` + Enabled *bool `json:"enabled,omitempty"` + Daemon *bool `json:"daemon,omitempty"` + // RestartPolicy is "always", "onError" (default), or "none". + RestartPolicy string `json:"restartPolicy,omitempty"` + RestartBackoff *int `json:"restartBackoff,omitempty"` + StartWaitSecs *int `json:"startWaitSecs,omitempty"` + ShutdownWaitSecs *int `json:"shutdownWaitSecs,omitempty"` + DependsOn []string `json:"dependsOn,omitempty"` + StartCmd string `json:"startCmd,omitempty"` + StopCmd string `json:"stopCmd,omitempty"` + Cmd string `json:"cmd,omitempty"` + Cwd string `json:"cwd,omitempty"` + LivenessProbe *LivenessProbe `json:"livenessProbe,omitempty"` + Labels map[string]string `json:"labels,omitempty"` +} + +// Restart policy values for ServiceDef.RestartPolicy. +const ( + RestartAlways = "always" + RestartOnError = "onError" + RestartNone = "none" +) + +// LivenessProbe mirrors microinit JSON probe fields. +type LivenessProbe struct { + HTTPUrl string `json:"httpUrl,omitempty"` + HTTPAcceptedCodes []int `json:"httpAcceptedCodes,omitempty"` + TCPAddr string `json:"tcpAddr,omitempty"` + Cmd string `json:"cmd,omitempty"` + SuccessExitCodes []int `json:"successExitCodes,omitempty"` + Interval int `json:"interval,omitempty"` + Timeout int `json:"timeout,omitempty"` +} + +// DropinFile is the JSON envelope for files under microinit.d/services/. +type DropinFile struct { + Services []ServiceDef `json:"services"` +} + +// BoolPtr returns a pointer to v (for optional JSON bool fields). +func BoolPtr(v bool) *bool { return &v } + +// IntPtr returns a pointer to v (for optional JSON int fields). +func IntPtr(v int) *int { return &v } 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 diff --git a/go/supervise/doc.go b/go/supervise/doc.go new file mode 100644 index 0000000..df6544b --- /dev/null +++ b/go/supervise/doc.go @@ -0,0 +1,6 @@ +// Package supervise embeds a microinit daemon inside a client process. +// +// It only manages process lifecycle (join existing socket or spawn +// `microinit supervise`). Stopping individual services, drop-in ownership, and +// product policies belong in the caller. +package supervise diff --git a/go/supervise/host.go b/go/supervise/host.go new file mode 100644 index 0000000..b94e563 --- /dev/null +++ b/go/supervise/host.go @@ -0,0 +1,439 @@ +package supervise + +import ( + "context" + "encoding/json" + "fmt" + "io" + "os" + "os/exec" + "os/signal" + "path/filepath" + "strings" + "sync" + "syscall" + "time" + + "github.com/dcc-bigfred/microinit/go/client" + "github.com/dcc-bigfred/microinit/go/config" +) + +const ( + defaultReadyTimeout = 10 * time.Second + defaultShutdownTimeout = 15 * time.Second + // hardKillGrace is how long to wait after SIGTERM to the process group + // before escalating to SIGKILL, within the overall ShutdownTimeout budget. + hardKillGrace = 5 * time.Second + // spawnLogCap bounds the captured stdout/stderr of the spawned microinit + // process so a chatty daemon cannot exhaust memory on embedded hosts. + spawnLogCap = 256 * 1024 +) + +// Host joins or spawns a microinit supervise instance. +type Host struct { + Socket string + Bin string + ConfigPath string + // DropinDir is created during EnsureRunning when set (microinit loads it + // from the config directory layout; callers still write drop-ins themselves). + DropinDir string + // PidFile is written after a successful spawn (pid + /proc starttime) so + // embedders can reap orphans safely. Empty defaults to /microinit.pid. + PidFile string + + // ReadyTimeout waits for IPC after spawn (default 10s). + ReadyTimeout time.Duration + // ShutdownTimeout waits after Shutdown IPC (default 15s). + ShutdownTimeout time.Duration + + client *client.Client + spawned bool + cmd *exec.Cmd + waitCh <-chan error + mu sync.Mutex +} + +// New returns a Host. Empty Socket defaults to client.DefaultSocket; empty Bin +// defaults to "microinit". +func New(socket, bin, configPath, dropinDir string) *Host { + if socket == "" { + socket = client.DefaultSocket + } + if bin == "" { + bin = "microinit" + } + return &Host{ + Socket: socket, + Bin: bin, + ConfigPath: configPath, + DropinDir: dropinDir, + client: &client.Client{Socket: socket}, + } +} + +// Client returns the IPC client bound to this host's socket. +func (h *Host) Client() *client.Client { return h.client } + +// Spawned reports whether this Host started the microinit process. +func (h *Host) Spawned() bool { + h.mu.Lock() + defer h.mu.Unlock() + return h.spawned +} + +// EnsureRunning joins an existing microinit when its socket responds. +// Otherwise it launches one `microinit supervise` and waits for IPC. +// +// The spawned process is NOT tied to ctx: cancelling ctx triggers a graceful +// SIGTERM (soft-kill) so microinit can stop its services, then escalates to +// SIGKILL only after ShutdownTimeout. This prevents orphaned managed +// processes (redis, dcc-bus, …) when the embedding process receives SIGTERM. +// +// joined is true when an already-running daemon was used (this Host must not +// Shutdown that process). +func (h *Host) EnsureRunning(ctx context.Context) (joined bool, err error) { + h.mu.Lock() + defer h.mu.Unlock() + if _, err := h.client.List(); err == nil { + return true, nil + } + if h.spawned && h.cmd != nil && h.cmd.Process != nil { + return false, fmt.Errorf("microinit process is running but IPC is unavailable (socket %s)", h.Socket) + } + if h.DropinDir != "" { + if err := os.MkdirAll(h.DropinDir, 0o755); err != nil { + return false, fmt.Errorf("create microinit drop-in dir %s: %w", h.DropinDir, err) + } + } + if h.ConfigPath != "" { + if err := os.MkdirAll(filepath.Dir(h.ConfigPath), 0o755); err != nil { + return false, fmt.Errorf("create microinit config dir %s: %w", filepath.Dir(h.ConfigPath), err) + } + } + if sockDir := filepath.Dir(h.Socket); sockDir != "" && sockDir != "." { + if err := os.MkdirAll(sockDir, 0o755); err != nil { + return false, fmt.Errorf("create microinit socket dir %s: %w", sockDir, err) + } + } + if h.ConfigPath != "" { + if _, err := os.Stat(h.ConfigPath); os.IsNotExist(err) { + content, marshalErr := json.Marshal(map[string]any{"services": []any{}, "socket": h.Socket}) + if marshalErr != nil { + return false, marshalErr + } + if err := config.WriteFileAtomically(h.ConfigPath, append(content, '\n')); err != nil { + return false, fmt.Errorf("write microinit config %s: %w", h.ConfigPath, err) + } + } else if err != nil { + return false, fmt.Errorf("stat microinit config %s: %w", h.ConfigPath, err) + } + } + if h.ConfigPath == "" { + return false, fmt.Errorf("ConfigPath is required to spawn microinit") + } + + logBuf := newBoundedBuffer(spawnLogCap) + // exec.Command (not CommandContext): ctx cancellation is handled below as a + // graceful SIGTERM so microinit can stop managed services before exit. + cmd := exec.Command(h.Bin, "--socket", h.Socket, "supervise", "--config", h.ConfigPath) + cmd.Stdout = logBuf + cmd.Stderr = logBuf + // Run microinit in its own process group so a SIGTERM to the group reaches + // the daemon and (optionally) its tracked children, not the embedder. + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + if err := cmd.Start(); err != nil { + return false, fmt.Errorf("start microinit (%s --socket %s supervise --config %s): %w", h.Bin, h.Socket, h.ConfigPath, err) + } + if err := h.writePidFile(cmd.Process.Pid); err != nil { + _ = signalProcessGroup(cmd, syscall.SIGKILL) + _, _ = cmd.Process.Wait() + return false, fmt.Errorf("write microinit pid file: %w", err) + } + waitCh := make(chan error, 1) + go func() { waitCh <- cmd.Wait() }() + h.cmd, h.spawned, h.waitCh = cmd, true, waitCh + + ready := h.ReadyTimeout + if ready <= 0 { + ready = defaultReadyTimeout + } + deadline := time.NewTimer(ready) + defer deadline.Stop() + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + for { + if _, err := h.client.List(); err == nil { + return false, nil + } + select { + case <-ctx.Done(): + h.spawned, h.cmd, h.waitCh = false, nil, nil + return false, fmt.Errorf("microinit startup cancelled: %w", h.terminateSoft(ctx, cmd, waitCh, logBuf)) + case waitErr := <-waitCh: + h.spawned, h.cmd, h.waitCh = false, nil, nil + h.removePidFile() + detail := strings.TrimSpace(logBuf.String()) + if detail == "" { + detail = fmt.Sprintf("exit: %v", waitErr) + } + return false, fmt.Errorf("microinit exited before ready (bin=%s socket=%s config=%s): %s", h.Bin, h.Socket, h.ConfigPath, detail) + case <-deadline.C: + h.spawned, h.cmd, h.waitCh = false, nil, nil + return false, fmt.Errorf("microinit did not become ready within %s (bin=%s socket=%s config=%s): %s", ready, h.Bin, h.Socket, h.ConfigPath, h.softKillDetail(cmd, waitCh, logBuf)) + case <-ticker.C: + } + } +} + +// terminateSoft sends SIGTERM to the spawned microinit process group and waits +// for it to exit gracefully (up to ShutdownTimeout), then escalates to SIGKILL. +// Used when ctx is cancelled during startup. +func (h *Host) terminateSoft(ctx context.Context, cmd *exec.Cmd, waitCh <-chan error, logBuf *boundedBuffer) error { + timeout := h.shutdownTimeout() + _ = signalProcessGroup(cmd, syscall.SIGTERM) + select { + case <-waitCh: + h.removePidFile() + return ctx.Err() + case <-time.After(timeout): + _ = signalProcessGroup(cmd, syscall.SIGKILL) + <-waitCh + h.removePidFile() + return fmt.Errorf("%w (microinit did not exit after SIGTERM within %s)", ctx.Err(), timeout) + } +} + +// softKillDetail SIGTERM-waits then SIGKILLs the process group and returns the captured log. +func (h *Host) softKillDetail(cmd *exec.Cmd, waitCh <-chan error, logBuf *boundedBuffer) string { + timeout := h.shutdownTimeout() + _ = signalProcessGroup(cmd, syscall.SIGTERM) + select { + case <-waitCh: + case <-time.After(timeout): + _ = signalProcessGroup(cmd, syscall.SIGKILL) + <-waitCh + } + h.removePidFile() + detail := strings.TrimSpace(logBuf.String()) + if detail == "" { + detail = fmt.Sprintf("killed after %s", timeout) + } + return detail +} + +// Shutdown stops the microinit process only if this Host spawned it. +// When EnsureRunning joined an existing daemon, this is a no-op. +// Callers that need to stop their own services must do so themselves first. +// +// Sequence within ShutdownTimeout (default 15s): IPC halt → wait → SIGTERM to +// the process group → wait remaining budget (capped) → SIGKILL to the group. +func (h *Host) Shutdown(ctx context.Context) error { + h.mu.Lock() + spawned, cmd, waitCh := h.spawned, h.cmd, h.waitCh + h.mu.Unlock() + if !spawned || cmd == nil || cmd.Process == nil { + return nil + } + timeout := h.shutdownTimeout() + deadline := time.Now().Add(timeout) + if dl, ok := ctx.Deadline(); ok && dl.Before(deadline) { + deadline = dl + } + + // Ask microinit to stop its services and exit cleanly. + _ = h.client.Shutdown() + if waitCh == nil { + done := make(chan error, 1) + go func() { done <- cmd.Wait() }() + waitCh = done + } + remaining := time.Until(deadline) + if remaining < 0 { + remaining = 0 + } + select { + case <-ctx.Done(): + h.terminateHard(cmd, waitCh, time.Until(deadline)) + h.clearSpawn() + return ctx.Err() + case <-time.After(remaining): + h.terminateHard(cmd, waitCh, hardKillGrace) + h.clearSpawn() + return fmt.Errorf("microinit shutdown timed out (socket %s)", h.Socket) + case <-waitCh: + } + h.clearSpawn() + return nil +} + +func (h *Host) terminateHard(cmd *exec.Cmd, waitCh <-chan error, grace time.Duration) { + _ = signalProcessGroup(cmd, syscall.SIGTERM) + if grace <= 0 { + grace = hardKillGrace + } + select { + case <-waitCh: + case <-time.After(grace): + _ = signalProcessGroup(cmd, syscall.SIGKILL) + <-waitCh + } +} + +// signalProcessGroup delivers sig to the process group of cmd (Setpgid). +// Falls back to signaling the process itself when the group kill fails. +func signalProcessGroup(cmd *exec.Cmd, sig syscall.Signal) error { + if cmd == nil || cmd.Process == nil { + return nil + } + pid := cmd.Process.Pid + if err := syscall.Kill(-pid, sig); err != nil { + return cmd.Process.Signal(sig) + } + return nil +} + +func (h *Host) pidFilePath() string { + if h.PidFile != "" { + return h.PidFile + } + dir := filepath.Dir(h.Socket) + if dir == "" || dir == "." { + dir = "." + } + return filepath.Join(dir, "microinit.pid") +} + +func (h *Host) writePidFile(pid int) error { + path := h.pidFilePath() + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + starttime, err := procStartTime(pid) + if err != nil { + // Best-effort: still write the PID so orphan cleanup has something. + starttime = 0 + } + content := fmt.Sprintf("%d\n%d\n", pid, starttime) + tmp := path + ".tmp" + if err := os.WriteFile(tmp, []byte(content), 0o644); err != nil { + return err + } + return os.Rename(tmp, path) +} + +func (h *Host) removePidFile() { + _ = os.Remove(h.pidFilePath()) +} + +// procStartTime reads field 22 (starttime) from /proc//stat. +func procStartTime(pid int) (uint64, error) { + data, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid)) + if err != nil { + return 0, err + } + // Format: pid (comm) state ppid ... — comm may contain spaces/parens. + s := string(data) + idx := strings.LastIndex(s, ") ") + if idx < 0 { + return 0, fmt.Errorf("parse /proc/%d/stat: no comm", pid) + } + fields := strings.Fields(s[idx+2:]) + // After ") ": field[0]=state (stat field 3). starttime is field 22 → index 19. + if len(fields) < 20 { + return 0, fmt.Errorf("parse /proc/%d/stat: too few fields", pid) + } + var starttime uint64 + _, err = fmt.Sscanf(fields[19], "%d", &starttime) + return starttime, err +} + +func (h *Host) shutdownTimeout() time.Duration { + if h.ShutdownTimeout > 0 { + return h.ShutdownTimeout + } + return defaultShutdownTimeout +} + +func (h *Host) clearSpawn() { + h.removePidFile() + h.mu.Lock() + h.spawned, h.cmd, h.waitCh = false, nil, nil + h.mu.Unlock() +} + +// ForwardSignals subscribes the Host to SIGTERM/SIGINT on the embedder +// process and triggers a graceful Shutdown when either arrives. Call once +// after EnsureRunning when the embedder wants OS signals to tear down +// microinit (e.g. loco-server). The returned stop function restores the +// default signal behavior; call it before a subsequent EnsureRunning. +func (h *Host) ForwardSignals() (stop func()) { + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) + done := make(chan struct{}) + go func() { + select { + case <-sigCh: + _ = h.Shutdown(context.Background()) + case <-done: + } + }() + return func() { + signal.Stop(sigCh) + close(done) + } +} + +// boundedBuffer is a small io.Writer that keeps at most cap bytes of the most +// recent output (ring-style: once full, new writes overwrite the oldest). +// Used to capture spawn diagnostics without unbounded memory growth. +type boundedBuffer struct { + cap int + buf []byte + pos int + full bool + mu sync.Mutex +} + +func newBoundedBuffer(cap int) *boundedBuffer { + return &boundedBuffer{cap: cap, buf: make([]byte, 0, cap)} +} + +func (b *boundedBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + n := len(p) + for len(p) > 0 { + free := b.cap - len(b.buf) + if b.full { + free = 0 + } + if free > len(p) { + free = len(p) + } + if free > 0 { + b.buf = append(b.buf, p[:free]...) + p = p[free:] + } + if len(p) == 0 { + break + } + // Buffer full: wrap around and overwrite oldest. + copied := copy(b.buf[b.pos:], p) + b.pos = (b.pos + copied) % b.cap + p = p[copied:] + b.full = true + } + return n, nil +} + +func (b *boundedBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + if !b.full { + return string(b.buf) + } + return string(b.buf[b.pos:]) + string(b.buf[:b.pos]) +} + +// Ensure interface compliance. +var _ io.Writer = (*boundedBuffer)(nil) diff --git a/go/supervise/host_buffer_test.go b/go/supervise/host_buffer_test.go new file mode 100644 index 0000000..ee1439e --- /dev/null +++ b/go/supervise/host_buffer_test.go @@ -0,0 +1,21 @@ +package supervise + +import ( + "bytes" + "testing" +) + +func TestBoundedBufferWriteReturnsInputLength(t *testing.T) { + b := newBoundedBuffer(8) + in := bytes.Repeat([]byte("x"), 32) + n, err := b.Write(in) + if err != nil { + t.Fatal(err) + } + if n != len(in) { + t.Fatalf("Write returned %d, want %d (os/exec requires full length)", n, len(in)) + } + if got := b.String(); len(got) != 8 { + t.Fatalf("captured len=%d want 8", len(got)) + } +} diff --git a/go/supervise/host_extra_test.go b/go/supervise/host_extra_test.go new file mode 100644 index 0000000..f26b3db --- /dev/null +++ b/go/supervise/host_extra_test.go @@ -0,0 +1,130 @@ +package supervise_test + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/dcc-bigfred/microinit/go/client" + "github.com/dcc-bigfred/microinit/go/supervise" +) + +// sleepScript writes a tiny shell wrapper that ignores its argv (microinit +// passes --socket/supervise/--config) and stays alive for `dur`. Returns the +// executable path to use as the microinit binary in tests. +func sleepScript(t *testing.T, dur time.Duration) string { + t.Helper() + dir := t.TempDir() + p := filepath.Join(dir, "fake-microinit") + if err := os.WriteFile(p, []byte("#!/bin/sh\nexec sleep "+dur.String()+"\n"), 0o755); err != nil { + t.Fatal(err) + } + return p +} + +// TestEnsureRunningSpawnFail verifies a missing binary surfaces a clear error +// and does not leave the host in a spawned state (6). +func TestEnsureRunningSpawnFail(t *testing.T) { + dir := t.TempDir() + sock := filepath.Join(dir, "microinit.sock") + h := supervise.New(sock, "/nonexistent/microinit-bin", filepath.Join(dir, "microinit.json"), "") + joined, err := h.EnsureRunning(context.Background()) + if err == nil { + t.Fatal("expected spawn error, got nil") + } + if joined { + t.Fatal("must not report joined on spawn failure") + } + if h.Spawned() { + t.Fatal("must not be spawned after spawn failure") + } + if !strings.Contains(err.Error(), "start microinit") { + t.Fatalf("expected start error, got %v", err) + } +} + +// TestEnsureRunningSpawnExitsBeforeReady verifies a binary that exits +// immediately is reported as a pre-ready exit with captured output (6). +func TestEnsureRunningSpawnExitsBeforeReady(t *testing.T) { + dir := t.TempDir() + sock := filepath.Join(dir, "microinit.sock") + // "false" exits 1 immediately and prints nothing. + h := supervise.New(sock, "false", filepath.Join(dir, "microinit.json"), "") + h.ReadyTimeout = 2 * time.Second + _, err := h.EnsureRunning(context.Background()) + if err == nil { + t.Fatal("expected pre-ready exit error, got nil") + } + if !strings.Contains(err.Error(), "exited before ready") { + t.Fatalf("expected 'exited before ready', got %v", err) + } +} + +// TestEnsureRunningReadyTimeout verifies a binary that stays alive but never +// opens the socket is killed after ReadyTimeout and reported (6). +func TestEnsureRunningReadyTimeout(t *testing.T) { + dir := t.TempDir() + sock := filepath.Join(dir, "microinit.sock") + // "sleep 30" stays alive but never opens the socket. + h := supervise.New(sock, sleepScript(t, 30*time.Second), filepath.Join(dir, "microinit.json"), "") + h.ReadyTimeout = 500 * time.Millisecond + _, err := h.EnsureRunning(context.Background()) + if err == nil { + t.Fatal("expected ready timeout, got nil") + } + if !strings.Contains(err.Error(), "did not become ready") { + t.Fatalf("expected 'did not become ready', got %v", err) + } + if h.Spawned() { + t.Fatal("must clear spawned after timeout kill") + } +} + +// TestEnsureRunningConfigRequired verifies spawning without ConfigPath fails +// fast instead of launching a daemon with no config (6). +func TestEnsureRunningConfigRequired(t *testing.T) { + dir := t.TempDir() + sock := filepath.Join(dir, "microinit.sock") + h := supervise.New(sock, "sleep", "", "") + _, err := h.EnsureRunning(context.Background()) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "ConfigPath is required") { + t.Fatalf("expected ConfigPath error, got %v", err) + } +} + +// TestEnsureRunningCtxCancelledSendsSIGTERM verifies cancelling ctx during +// startup sends SIGTERM (soft-kill) and returns the ctx error (2.3). +func TestEnsureRunningCtxCancelledSendsSIGTERM(t *testing.T) { + dir := t.TempDir() + sock := filepath.Join(dir, "microinit.sock") + // fake-microinit stays alive and will be terminated by our SIGTERM during startup. + h := supervise.New(sock, sleepScript(t, 30*time.Second), filepath.Join(dir, "microinit.json"), "") + h.ReadyTimeout = 30 * time.Second + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(300 * time.Millisecond) + cancel() + }() + _, err := h.EnsureRunning(ctx) + if err == nil { + t.Fatal("expected cancellation error, got nil") + } + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } +} + +// TestClientErrNotFoundSentinel verifies the exported sentinel exists and is +// usable for errors.Is checks by downstream callers (6). +func TestClientErrNotFoundSentinel(t *testing.T) { + if !errors.Is(client.ErrNotFound, client.ErrNotFound) { + t.Fatal("ErrNotFound must satisfy errors.Is with itself") + } +} diff --git a/go/supervise/host_test.go b/go/supervise/host_test.go new file mode 100644 index 0000000..582454a --- /dev/null +++ b/go/supervise/host_test.go @@ -0,0 +1,92 @@ +package supervise_test + +import ( + "context" + "encoding/binary" + "encoding/json" + "io" + "net" + "path/filepath" + "testing" + "time" + + "github.com/dcc-bigfred/microinit/go/supervise" +) + +func TestEnsureRunningJoinsExisting(t *testing.T) { + dir := t.TempDir() + sock := filepath.Join(dir, "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 func(c net.Conn) { + defer c.Close() + if _, err := readReq(c); err != nil { + return + } + _ = writeResp(c, map[string]any{"type": "list", "services": []any{}}) + }(conn) + } + }() + + h := supervise.New(sock, "false", filepath.Join(dir, "microinit.json"), filepath.Join(dir, "dropins")) + joined, err := h.EnsureRunning(context.Background()) + if err != nil { + t.Fatal(err) + } + if !joined { + t.Fatal("expected joined") + } + if h.Spawned() { + t.Fatal("must not spawn when joining") + } + if err := h.Shutdown(context.Background()); err != nil { + t.Fatal(err) + } +} + +func TestShutdownNoopWhenNotSpawned(t *testing.T) { + h := supervise.New(filepath.Join(t.TempDir(), "missing.sock"), "microinit", "", "") + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := h.Shutdown(ctx); err != nil { + t.Fatal(err) + } +} + +func readReq(r io.Reader) (map[string]any, error) { + var hdr [4]byte + if _, err := io.ReadFull(r, hdr[:]); err != nil { + return nil, err + } + n := binary.LittleEndian.Uint32(hdr[:]) + buf := make([]byte, n) + if _, err := io.ReadFull(r, buf); err != nil { + return nil, err + } + var req map[string]any + 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/man/man5/microinit.json.5.mdoc b/man/man5/microinit.json.5.mdoc index 72d1d9e..e7f8cb6 100644 --- a/man/man5/microinit.json.5.mdoc +++ b/man/man5/microinit.json.5.mdoc @@ -126,6 +126,20 @@ when specific cmds are null Explicit shell commands .It Cm env , cwd Environment and working directory +.It Cm labels +Optional object of string key/value pairs +.Pq for example Cm created-by=bigfred . +Keys must match +.Li [A-Za-z0-9][A-Za-z0-9._-]* +.Pq max 63 characters ; +values are non-empty +.Pq max 253 characters . +Shown by +.Cm describe +and by +.Cm list Fl -show-labels ; +filter with +.Cm list Fl l . .It Cm livenessProbe Optional object. Exactly one of .Cm cmd , diff --git a/man/man8/microinit.8.mdoc b/man/man8/microinit.8.mdoc index 227e7e2..abd4e34 100644 --- a/man/man8/microinit.8.mdoc +++ b/man/man8/microinit.8.mdoc @@ -33,6 +33,13 @@ .Nm .Op Fl -socket Ns = Ns Ar path .Cm list +.Op Fl -show-labels +.Op Fl -selector Ns = Ns Ar key=value +.Op Fl l Ar key=value +.Nm +.Op Fl -socket Ns = Ns Ar path +.Cm describe +.Ar name .Nm .Op Fl -socket Ns = Ns Ar path .Cm logs @@ -173,6 +180,9 @@ Control socket .Bd -literal microinit init --logs-tty=/dev/tty2 --init-logs-tty=/dev/tty3 microinit list +microinit list --show-labels +microinit list -l created-by=bigfred +microinit describe redis microinit start redis microinit start --force alloy microinit restart redis @@ -180,6 +190,19 @@ microinit logs alloy --follow microinit disable dropbear .Ed .Pp +.Cm list +supports +.Fl -show-labels +to print a +.Cm LABELS +column, and repeatable +.Fl l / Fl -selector +.Pq form Cm key=value +to keep only services matching all selectors +.Pq AND . +.Cm describe +always prints labels. +.Pp .Cm start prints a status line on stdout: either that the service is starting, that it is .Cm waiting_for_dependency , diff --git a/src/cli.rs b/src/cli.rs index 7fb347b..45bae53 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -66,28 +66,57 @@ pub fn cmd_shutdown(socket: &Path, mode: ShutdownMode) -> Result<()> { simple_ok(socket, Request::Shutdown { mode }) } -pub fn cmd_list(socket: &Path) -> Result<()> { +pub fn cmd_list(socket: &Path, show_labels: bool, selectors: &[String]) -> Result<()> { + let want: Vec<(String, String)> = selectors + .iter() + .map(|s| crate::labels::parse_selector(s)) + .collect::>>()?; match request(socket, &Request::List)? { Response::List { services } => { - println!( - "{:<20} {:<22} {:>8} {:>8} {:>8} {:>10}", - "NAME", "STATE", "PID", "RESTARTS", "ENABLED", "LIVE_FAIL" - ); - for s in services { - let pid = s.pid.map(|p| p.to_string()).unwrap_or_else(|| "-".into()); + let services: Vec<_> = services + .into_iter() + .filter(|s| crate::labels::matches_selectors(&s.labels, &want)) + .collect(); + if show_labels { + println!( + "{:<20} {:<22} {:>8} {:>8} {:>8} {:>10} LABELS", + "NAME", "STATE", "PID", "RESTARTS", "ENABLED", "LIVE_FAIL" + ); + } else { println!( "{:<20} {:<22} {:>8} {:>8} {:>8} {:>10}", - s.name, - s.state.to_string(), - pid, - s.restarts, - if s.enabled { "yes" } else { "no" }, - s.liveness_failures + "NAME", "STATE", "PID", "RESTARTS", "ENABLED", "LIVE_FAIL" ); } + for s in services { + let pid = s.pid.map(|p| p.to_string()).unwrap_or_else(|| "-".into()); + if show_labels { + let labels = crate::labels::format_labels(&s.labels); + println!( + "{:<20} {:<22} {:>8} {:>8} {:>8} {:>10} {}", + s.name, + s.state.to_string(), + pid, + s.restarts, + if s.enabled { "yes" } else { "no" }, + s.liveness_failures, + if labels.is_empty() { "-" } else { &labels } + ); + } else { + println!( + "{:<20} {:<22} {:>8} {:>8} {:>8} {:>10}", + s.name, + s.state.to_string(), + pid, + s.restarts, + if s.enabled { "yes" } else { "no" }, + s.liveness_failures + ); + } + } Ok(()) } - Response::Error { message } => Err(Error::Ipc(message)), + Response::Error { message, .. } => Err(Error::Ipc(message)), other => Err(Error::Ipc(format!("unexpected response: {other:?}"))), } } @@ -98,7 +127,7 @@ pub fn cmd_describe(socket: &Path, name: &str) -> Result<()> { print_describe(&describe); Ok(()) } - Response::Error { message } => Err(Error::Ipc(message)), + Response::Error { message, .. } => Err(Error::Ipc(message)), other => Err(Error::Ipc(format!("unexpected response: {other:?}"))), } } @@ -115,6 +144,16 @@ fn print_describe(d: &ServiceDescribe) { println!("Uptime: {}", format_uptime(d.uptime_secs)); println!(); + println!("Labels:"); + if s.labels.is_empty() { + println!(" (none)"); + } else { + for (k, v) in &s.labels { + println!(" {k}={v}"); + } + } + println!(); + println!("Depends on:"); print_dep_list(&d.depends_on); println!(); @@ -302,7 +341,7 @@ pub fn cmd_start(socket: &Path, name: &str, force: bool) -> Result<()> { } Ok(()) } - Response::Error { message } => Err(Error::Ipc(message)), + Response::Error { message, .. } => Err(Error::Ipc(message)), other => Err(Error::Ipc(format!("unexpected response: {other:?}"))), } } @@ -360,8 +399,9 @@ pub fn cmd_logs( writeln!(out, "[{}] {}: {}", line.ts, line.service, line.msg)?; out.flush()?; } + Response::Heartbeat => {} Response::Ok { .. } => break, - Response::Error { message } => return Err(Error::Ipc(message)), + Response::Error { message, .. } => return Err(Error::Ipc(message)), other => return Err(Error::Ipc(format!("unexpected: {other:?}"))), } } @@ -376,7 +416,7 @@ fn simple_ok(socket: &Path, req: Request) -> Result<()> { } Ok(()) } - Response::Error { message } => Err(Error::Ipc(message)), + Response::Error { message, .. } => Err(Error::Ipc(message)), other => Err(Error::Ipc(format!("unexpected response: {other:?}"))), } } diff --git a/src/config.rs b/src/config.rs index 0c66774..b7b5d64 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,6 +1,6 @@ //! Configuration model and load/save for microinit.json + enabled override. -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::fs; use std::path::{Path, PathBuf}; @@ -185,6 +185,41 @@ impl LivenessProbe { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "camelCase")] +#[non_exhaustive] +pub enum RestartPolicy { + /// Restart on every exit, including success (exit 0 / successExitCodes). + Always, + /// Restart only on non-success exits (default). + #[default] + OnError, + /// Never auto-restart. + None, +} + +impl RestartPolicy { + /// Whether an exit with the given success classification should trigger a restart. + #[must_use] + pub fn should_restart(self, success: bool) -> bool { + match self { + Self::Always => true, + Self::OnError => !success, + Self::None => false, + } + } + + /// Policies other than [`Self::None`] require `daemon=true`. + #[must_use] + pub fn requires_daemon(self) -> bool { + !matches!(self, Self::None) + } +} + +fn default_restart_policy() -> RestartPolicy { + RestartPolicy::OnError +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct ServiceConfig { @@ -193,8 +228,10 @@ pub struct ServiceConfig { pub enabled: bool, #[serde(default = "default_true")] pub daemon: bool, - #[serde(default)] - pub restart: bool, + /// Auto-restart policy. Replaces the former `restart` bool + /// (`true` → `onError`, `false` → `none`). + #[serde(default = "default_restart_policy")] + pub restart_policy: RestartPolicy, #[serde(default = "default_backoff")] pub restart_backoff: u64, #[serde(default = "default_success_codes")] @@ -225,6 +262,9 @@ pub struct ServiceConfig { /// Optional periodic health check; on failure the service is restarted. #[serde(default)] pub liveness_probe: Option, + /// Arbitrary key=value labels (e.g. `created-by=bigfred`). Stable order via BTreeMap. + #[serde(default)] + pub labels: BTreeMap, } fn default_true() -> bool { @@ -247,6 +287,50 @@ fn default_cwd() -> String { "/".to_string() } +const LABEL_KEY_MAX: usize = 63; +const LABEL_VALUE_MAX: usize = 253; + +/// Validate service label map (keys/values non-empty, key charset, length limits). +pub fn validate_labels(service: &str, labels: &BTreeMap) -> Result<()> { + for (key, value) in labels { + if key.is_empty() { + return Err(Error::Config(format!( + "service '{service}': label key must not be empty" + ))); + } + if key.len() > LABEL_KEY_MAX { + return Err(Error::Config(format!( + "service '{service}': label key '{key}' exceeds {LABEL_KEY_MAX} characters" + ))); + } + if !is_valid_label_key(key) { + return Err(Error::Config(format!( + "service '{service}': invalid label key '{key}' (want [A-Za-z0-9][A-Za-z0-9._-]*)" + ))); + } + if value.is_empty() { + return Err(Error::Config(format!( + "service '{service}': label '{key}' value must not be empty" + ))); + } + if value.len() > LABEL_VALUE_MAX { + return Err(Error::Config(format!( + "service '{service}': label '{key}' value exceeds {LABEL_VALUE_MAX} characters" + ))); + } + } + Ok(()) +} + +fn is_valid_label_key(key: &str) -> bool { + let mut chars = key.chars(); + match chars.next() { + Some(c) if c.is_ascii_alphanumeric() => {} + _ => return false, + } + chars.all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '_' || c == '-') +} + impl ServiceConfig { /// Resolve start command: startCmd or `cmd start`. pub fn resolve_start(&self) -> Result { @@ -390,9 +474,9 @@ impl Config { svc.name ))); } - if svc.restart && !svc.daemon { + if matches!(svc.restart_policy, RestartPolicy::Always) && !svc.daemon { return Err(Error::Config(format!( - "service '{}': restart=true requires daemon=true", + "service '{}': restartPolicy=always requires daemon=true", svc.name ))); } @@ -442,6 +526,7 @@ impl Config { ))); } } + validate_labels(&svc.name, &svc.labels)?; } for svc in &self.services { for dep in &svc.depends_on { @@ -628,7 +713,7 @@ pub fn example_config() -> Config { name: "network".into(), enabled: true, daemon: false, - restart: false, + restart_policy: RestartPolicy::None, restart_backoff: 2, success_exit_codes: vec![0], start_wait_secs: 0, @@ -651,12 +736,13 @@ pub fn example_config() -> Config { interval: 30, timeout: 5, }), + labels: BTreeMap::new(), }, ServiceConfig { name: "redis".into(), enabled: true, daemon: true, - restart: true, + restart_policy: RestartPolicy::OnError, restart_backoff: 2, success_exit_codes: vec![0], start_wait_secs: 0, @@ -670,12 +756,13 @@ pub fn example_config() -> Config { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), }, ServiceConfig { name: "remote-icmp".into(), enabled: true, daemon: true, - restart: true, + restart_policy: RestartPolicy::OnError, restart_backoff: 5, success_exit_codes: vec![0], start_wait_secs: 0, @@ -692,6 +779,7 @@ pub fn example_config() -> Config { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), }, ], } diff --git a/src/error.rs b/src/error.rs index 0ef69bd..afa3c83 100644 --- a/src/error.rs +++ b/src/error.rs @@ -63,4 +63,16 @@ impl Error { source, } } + + /// Stable machine-readable error code for the IPC `Response::Error`. + /// Clients map on this instead of substring-matching the human message. + /// `None` means "no stable code"; clients fall back to the message. + pub fn code(&self) -> Option<&'static str> { + match self { + Error::UnknownService(_) => Some("not_found"), + Error::Disabled(_) => Some("disabled"), + Error::Cycle(_) => Some("cycle"), + _ => None, + } + } } diff --git a/src/init.rs b/src/init.rs index 8740263..231b233 100644 --- a/src/init.rs +++ b/src/init.rs @@ -352,21 +352,11 @@ fn handle_ipc( } Request::Status { name } => match supervisor.status(&name) { Ok(status) => write_frame(stream, &Response::Status { status })?, - Err(e) => write_frame( - stream, - &Response::Error { - message: e.to_string(), - }, - )?, + Err(e) => write_frame(stream, &error_response(&e))?, }, Request::Describe { name } => match supervisor.describe(&name) { Ok(describe) => write_frame(stream, &Response::Describe { describe })?, - Err(e) => write_frame( - stream, - &Response::Error { - message: e.to_string(), - }, - )?, + Err(e) => write_frame(stream, &error_response(&e))?, }, Request::Start { name, force } => { respond_start(stream, supervisor.start_service(&name, force))?; @@ -395,14 +385,27 @@ fn handle_ipc( } if follow { let rx = hub.subscribe(); - while let Ok(line) = rx.recv() { - if let Some(ref nme) = name { - if &line.service != nme { - continue; + // Heartbeats keep Go/UI clients with idle read deadlines alive + // when a service is healthy but quiet. + const HEARTBEAT: std::time::Duration = std::time::Duration::from_secs(10); + loop { + match rx.recv_timeout(HEARTBEAT) { + Ok(line) => { + if let Some(ref nme) = name { + if &line.service != nme { + continue; + } + } + if write_frame(stream, &Response::Log { line }).is_err() { + break; + } } - } - if write_frame(stream, &Response::Log { line }).is_err() { - break; + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + if write_frame(stream, &Response::Heartbeat).is_err() { + break; + } + } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break, } } } else { @@ -425,12 +428,7 @@ fn respond_start(stream: &mut UnixStream, res: Result) -> Result<()> { message: Some(message), }, )?, - Err(e) => write_frame( - stream, - &Response::Error { - message: e.to_string(), - }, - )?, + Err(e) => write_frame(stream, &error_response(&e))?, } Ok(()) } @@ -438,16 +436,21 @@ fn respond_start(stream: &mut UnixStream, res: Result) -> Result<()> { fn respond_result(stream: &mut UnixStream, res: Result<()>) -> Result<()> { match res { Ok(()) => write_frame(stream, &Response::Ok { message: None })?, - Err(e) => write_frame( - stream, - &Response::Error { - message: e.to_string(), - }, - )?, + Err(e) => write_frame(stream, &error_response(&e))?, } Ok(()) } +/// Build an IPC error response with the stable [Error::code] populated when +/// available, so clients can map on `code` instead of substring-matching +/// the human-readable message. +fn error_response(e: &crate::error::Error) -> Response { + Response::Error { + message: e.to_string(), + code: e.code().map(|s| s.to_string()), + } +} + #[cfg(feature = "init")] fn getty_respawn(console: &str) { let tty = console.trim_start_matches("/dev/"); diff --git a/src/ipc.rs b/src/ipc.rs index 48c085c..dd434e7 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -136,6 +136,7 @@ pub fn serve(socket_path: &Path, handler: Handler) -> Result<()> { &mut stream, &Response::Error { message: "permission denied".into(), + code: Some("permission_denied".into()), }, ); continue; @@ -147,6 +148,7 @@ pub fn serve(socket_path: &Path, handler: Handler) -> Result<()> { message: format!( "too many concurrent IPC clients (max {MAX_IPC_CLIENTS})" ), + code: Some("busy".into()), }, ); continue; @@ -163,6 +165,7 @@ pub fn serve(socket_path: &Path, handler: Handler) -> Result<()> { &mut stream, &Response::Error { message: e.to_string(), + code: e.code().map(|s| s.to_string()), }, ); } diff --git a/src/labels.rs b/src/labels.rs new file mode 100644 index 0000000..7434741 --- /dev/null +++ b/src/labels.rs @@ -0,0 +1,86 @@ +//! Label helpers for CLI selectors and formatting. + +use std::collections::BTreeMap; + +use crate::error::{Error, Result}; + +/// Parse a single `key=value` selector (exactly one `=`). +pub fn parse_selector(s: &str) -> Result<(String, String)> { + let Some((key, value)) = s.split_once('=') else { + return Err(Error::Config(format!( + "invalid label selector {s:?}: want key=value" + ))); + }; + if key.is_empty() || value.is_empty() { + return Err(Error::Config(format!( + "invalid label selector {s:?}: key and value must be non-empty" + ))); + } + if key.contains('=') { + return Err(Error::Config(format!( + "invalid label selector {s:?}: want key=value" + ))); + } + Ok((key.to_string(), value.to_string())) +} + +/// True when `labels` contains every key=value in `want` (AND). +pub fn matches_selectors(labels: &BTreeMap, want: &[(String, String)]) -> bool { + want.iter() + .all(|(k, v)| labels.get(k).map(|have| have == v).unwrap_or(false)) +} + +/// Format labels as `k=v,k=v` (BTreeMap order). +pub fn format_labels(labels: &BTreeMap) -> String { + labels + .iter() + .map(|(k, v)| format!("{k}={v}")) + .collect::>() + .join(",") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_ok() { + assert_eq!( + parse_selector("created-by=bigfred").unwrap(), + ("created-by".into(), "bigfred".into()) + ); + } + + #[test] + fn parse_rejects_bad() { + assert!(parse_selector("noequals").is_err()); + assert!(parse_selector("=v").is_err()); + assert!(parse_selector("k=").is_err()); + } + + #[test] + fn match_and() { + let mut labels = BTreeMap::new(); + labels.insert("created-by".into(), "bigfred".into()); + labels.insert("env".into(), "prod".into()); + assert!(matches_selectors( + &labels, + &[("created-by".into(), "bigfred".into())] + )); + assert!(matches_selectors( + &labels, + &[ + ("created-by".into(), "bigfred".into()), + ("env".into(), "prod".into()) + ] + )); + assert!(!matches_selectors( + &labels, + &[("created-by".into(), "other".into())] + )); + assert!(!matches_selectors( + &labels, + &[("missing".into(), "x".into())] + )); + } +} diff --git a/src/lib.rs b/src/lib.rs index 8a79764..10e2cb6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -26,6 +26,7 @@ pub mod error; pub mod graph; pub mod init; pub mod ipc; +pub mod labels; pub mod liveness; pub mod logs; #[cfg(feature = "otel")] diff --git a/src/liveness.rs b/src/liveness.rs index ec5d8d1..b5ab207 100644 --- a/src/liveness.rs +++ b/src/liveness.rs @@ -127,17 +127,19 @@ fn run_http_probe(url: &str, method: &str, accepted: &[u16], timeout: Duration) #[cfg(test)] mod tests { use super::*; - use std::collections::HashMap; + use std::collections::{BTreeMap, HashMap}; use std::io::{Read, Write}; use std::net::TcpListener; use std::thread; + use crate::config::RestartPolicy; + fn cfg() -> ServiceConfig { ServiceConfig { name: "t".into(), enabled: true, daemon: false, - restart: false, + restart_policy: RestartPolicy::None, restart_backoff: 1, success_exit_codes: vec![0], start_wait_secs: 0, @@ -151,6 +153,7 @@ mod tests { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), } } diff --git a/src/main.rs b/src/main.rs index 6baa297..c1f7dd7 100644 --- a/src/main.rs +++ b/src/main.rs @@ -87,7 +87,14 @@ enum Commands { /// Disable a service (persist override + stop) Disable { name: String }, /// List services and their state - List, + List { + /// Show labels column + #[arg(long)] + show_labels: bool, + /// Label selector `key=value` (repeatable; AND). Short `-l`. + #[arg(short = 'l', long = "selector")] + selector: Vec, + }, /// Show detailed status, dependencies, and recent lifecycle events Describe { name: String }, /// Show service logs (or mixed if name omitted) @@ -241,7 +248,10 @@ fn main() -> ExitCode { Commands::Restart { name } => cli::cmd_restart(&cli.socket, &name), Commands::Enable { name } => cli::cmd_enable(&cli.socket, &name), Commands::Disable { name } => cli::cmd_disable(&cli.socket, &name), - Commands::List => cli::cmd_list(&cli.socket), + Commands::List { + show_labels, + selector, + } => cli::cmd_list(&cli.socket, show_labels, &selector), Commands::Describe { name } => cli::cmd_describe(&cli.socket, &name), Commands::Logs { name, diff --git a/src/protocol.rs b/src/protocol.rs index b758161..0391e94 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -1,5 +1,7 @@ //! IPC protocol messages (length-prefixed JSON frames). +use std::collections::BTreeMap; + use serde::{Deserialize, Serialize}; /// Runtime state of a service as reported over IPC. @@ -49,6 +51,9 @@ pub struct ServiceStatus { #[serde(default)] pub liveness_failures: u32, pub enabled: bool, + /// Service labels from config (stable key order). + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub labels: BTreeMap, } /// Kind of lifecycle event retained for `describe`. @@ -222,6 +227,11 @@ pub enum Response { }, Error { message: String, + /// Stable machine-readable code (e.g. "not_found", "disabled"). + /// Absent for errors without a canonical code; clients fall back to + /// substring-matching `message` for backward compatibility. + #[serde(default, skip_serializing_if = "Option::is_none")] + code: Option, }, List { services: Vec, @@ -236,4 +246,7 @@ pub enum Response { Log { line: LogLine, }, + /// Keepalive on follow log streams so clients with idle read deadlines + /// do not disconnect a quiet but healthy service. + Heartbeat, } diff --git a/src/supervisor.rs b/src/supervisor.rs index 785327f..450a10e 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -3,7 +3,7 @@ //! Child process waits are owned by the central PID 1 reaper ([`ExitRegistry`]), //! not by `std::process::Child::wait`, to avoid racing `waitpid(-1)`. -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Condvar, Mutex}; @@ -330,6 +330,7 @@ impl Supervisor { restarts: rt.restarts, liveness_failures: rt.liveness_failures, enabled: rt.enabled, + labels: s.labels.clone(), }) }) .collect() @@ -337,9 +338,16 @@ impl Supervisor { pub fn status(&self, name: &str) -> Result { let map = mutex_lock(&self.shared.runtimes); + let cfg = mutex_lock(&self.config); let rt = map .get(name) .ok_or_else(|| Error::UnknownService(name.to_string()))?; + let labels = cfg + .services + .iter() + .find(|s| s.name == name) + .map(|s| s.labels.clone()) + .unwrap_or_default(); Ok(ServiceStatus { name: name.to_string(), state: rt.state, @@ -347,6 +355,7 @@ impl Supervisor { restarts: rt.restarts, liveness_failures: rt.liveness_failures, enabled: rt.enabled, + labels, }) } @@ -356,7 +365,7 @@ impl Supervisor { /// `config`), then builds the graph and formats event timestamps unlocked. pub fn describe(&self, name: &str) -> Result { // --- Snapshot under runtimes (released before config / graph work) --- - let (status, uptime_secs, events, states) = { + let (mut status, uptime_secs, events, states) = { let map = mutex_lock(&self.shared.runtimes); let rt = map .get(name) @@ -369,6 +378,7 @@ impl Supervisor { restarts: rt.restarts, liveness_failures: rt.liveness_failures, enabled: rt.enabled, + labels: BTreeMap::new(), }; // Uptime only while currently `Running` (`running_since` is cleared otherwise). let uptime_secs = if matches!(rt.state, ServiceState::Running) { @@ -399,6 +409,7 @@ impl Supervisor { .iter() .find(|s| s.name == name) .ok_or_else(|| Error::UnknownService(name.to_string()))?; + status.labels = svc.labels.clone(); let depends_on_names = svc.depends_on.clone(); let services_deps: Vec<(String, Vec)> = cfg .services @@ -1015,29 +1026,32 @@ impl Supervisor { return; } - if success { - self.shared.set_state(name, ServiceState::Succeeded, None); + let stop_all = self.shared.stop_all.load(Ordering::SeqCst); + let should_restart = cfg.restart_policy.should_restart(success) && enabled && !stop_all; + if !should_restart { + let st = if success { + ServiceState::Succeeded + } else { + ServiceState::Failed + }; + self.shared.set_state(name, st, None); return; } - if cfg.restart && enabled && !self.shared.stop_all.load(Ordering::SeqCst) { - self.shared.set_state(name, ServiceState::Restarting, None); - self.shared.bump_restarts(name); - self.hub.emit( - INIT_SERVICE, - LogLevel::Info, - format!( - "{name}: exited {code}, restarting in {}s", - cfg.restart_backoff - ), - ); - thread::sleep(Duration::from_secs(cfg.restart_backoff)); - if let Err(e) = self.do_start(cfg, tracked, false) { - self.hub - .emit(INIT_SERVICE, LogLevel::Error, format!("{name}: {e}")); - self.shared.set_state(name, ServiceState::Failed, None); - } - } else { + self.shared.set_state(name, ServiceState::Restarting, None); + self.shared.bump_restarts(name); + self.hub.emit( + INIT_SERVICE, + LogLevel::Info, + format!( + "{name}: exited {code}, restarting in {}s", + cfg.restart_backoff + ), + ); + thread::sleep(Duration::from_secs(cfg.restart_backoff)); + if let Err(e) = self.do_start(cfg, tracked, false) { + self.hub + .emit(INIT_SERVICE, LogLevel::Error, format!("{name}: {e}")); self.shared.set_state(name, ServiceState::Failed, None); } } diff --git a/tests/config_test.rs b/tests/config_test.rs index 92ea784..93e8b50 100644 --- a/tests/config_test.rs +++ b/tests/config_test.rs @@ -1,6 +1,6 @@ //! Unit/integration tests for microinit::config -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::fs; use std::path::PathBuf; @@ -11,7 +11,7 @@ fn minimal_svc(name: &str) -> ServiceConfig { name: name.into(), enabled: true, daemon: true, - restart: false, + restart_policy: RestartPolicy::None, restart_backoff: 2, success_exit_codes: vec![0], start_wait_secs: 0, @@ -25,6 +25,7 @@ fn minimal_svc(name: &str) -> ServiceConfig { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), } } @@ -49,7 +50,7 @@ fn resolve_cmd_fallback() { name: "x".into(), enabled: true, daemon: true, - restart: false, + restart_policy: RestartPolicy::None, restart_backoff: 2, success_exit_codes: vec![0], start_wait_secs: 0, @@ -63,6 +64,7 @@ fn resolve_cmd_fallback() { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), }; assert_eq!(svc.resolve_start().unwrap(), "/etc/init.d/redis start"); assert_eq!(svc.resolve_stop().unwrap(), "/etc/init.d/redis stop"); @@ -75,7 +77,7 @@ fn resolve_explicit_cmds_prefer_over_cmd() { name: "x".into(), enabled: true, daemon: true, - restart: false, + restart_policy: RestartPolicy::None, restart_backoff: 2, success_exit_codes: vec![0], start_wait_secs: 0, @@ -89,6 +91,7 @@ fn resolve_explicit_cmds_prefer_over_cmd() { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), }; assert_eq!(svc.resolve_start().unwrap(), "start-me"); assert_eq!(svc.resolve_stop().unwrap(), "stop-me"); @@ -101,7 +104,7 @@ fn resolve_restart_falls_back_to_stop_and_start() { name: "x".into(), enabled: true, daemon: true, - restart: false, + restart_policy: RestartPolicy::None, restart_backoff: 2, success_exit_codes: vec![0], start_wait_secs: 0, @@ -115,6 +118,7 @@ fn resolve_restart_falls_back_to_stop_and_start() { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), }; assert_eq!(svc.resolve_restart().unwrap(), "do-stop && do-start"); } @@ -125,7 +129,7 @@ fn resolve_start_errors_without_cmds() { name: "x".into(), enabled: true, daemon: true, - restart: false, + restart_policy: RestartPolicy::None, restart_backoff: 2, success_exit_codes: vec![0], start_wait_secs: 0, @@ -139,6 +143,7 @@ fn resolve_start_errors_without_cmds() { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), }; assert!(svc.resolve_start().is_err()); assert!(svc.resolve_stop().is_err()); @@ -153,6 +158,21 @@ fn is_success_custom_codes() { assert!(!svc.is_success(1)); } +#[test] +fn validate_rejects_bad_labels() { + let mut cfg = Config::default(); + let mut s = minimal_svc("a"); + s.labels.insert("bad key".into(), "x".into()); + cfg.services.push(s); + assert!(cfg.validate().is_err()); + + let mut cfg = Config::default(); + let mut s = minimal_svc("a"); + s.labels.insert("created-by".into(), "bigfred".into()); + cfg.services.push(s); + cfg.validate().unwrap(); +} + #[test] fn example_validates() { let cfg = example_config(); @@ -181,7 +201,7 @@ fn validate_rejects_restart_without_daemon() { let mut cfg = Config::default(); let mut s = minimal_svc("job"); s.daemon = false; - s.restart = true; + s.restart_policy = RestartPolicy::Always; cfg.services.push(s); assert!(cfg.validate().is_err()); } @@ -271,7 +291,8 @@ fn load_or_create_writes_defaults_and_example() { let config = dir.join("microinit.json"); let example = dir.join("microinit.json.example"); let override_f = dir.join("override.json"); - let cfg = load_or_create(&config, &example, &override_f).unwrap(); + let dropins = dir.join("dropins"); + let cfg = load_or_create_with_dropins(&config, &example, &override_f, &dropins).unwrap(); assert!(config.is_file()); assert!(example.is_file()); assert!(cfg.services.is_empty()); // default config has empty services @@ -283,7 +304,7 @@ fn load_or_create_writes_defaults_and_example() { save_config(&config, &example_config()).unwrap(); map.insert("redis".into(), false); save_override(&override_f, &map).unwrap(); - let cfg2 = load_or_create(&config, &example, &override_f).unwrap(); + let cfg2 = load_or_create_with_dropins(&config, &example, &override_f, &dropins).unwrap(); assert!(!cfg2.get("redis").unwrap().enabled); let _ = fs::remove_dir_all(&dir); } @@ -393,7 +414,7 @@ fn rejects_empty_liveness_probe_cmd() { let mut cfg = Config::default(); let mut svc = minimal_svc("net"); svc.daemon = false; - svc.restart = false; + svc.restart_policy = RestartPolicy::None; svc.liveness_probe = Some(LivenessProbe { cmd: Some(" ".into()), http_url: None, diff --git a/tests/graph_test.rs b/tests/graph_test.rs index 849d0c7..692331e 100644 --- a/tests/graph_test.rs +++ b/tests/graph_test.rs @@ -1,8 +1,8 @@ //! Unit/integration tests for microinit::graph -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; -use microinit::config::ServiceConfig; +use microinit::config::{RestartPolicy, ServiceConfig}; use microinit::error::Error; use microinit::graph::*; @@ -11,7 +11,7 @@ fn svc(name: &str, deps: &[&str], bg: bool) -> ServiceConfig { name: name.into(), enabled: true, daemon: true, - restart: false, + restart_policy: RestartPolicy::None, restart_backoff: 2, success_exit_codes: vec![0], start_wait_secs: 0, @@ -25,6 +25,7 @@ fn svc(name: &str, deps: &[&str], bg: bool) -> ServiceConfig { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), } } diff --git a/tests/ipc_test.rs b/tests/ipc_test.rs index ca0098c..a1c82e8 100644 --- a/tests/ipc_test.rs +++ b/tests/ipc_test.rs @@ -1,5 +1,6 @@ //! Unit/integration tests for microinit::ipc +use std::collections::BTreeMap; use std::io::Cursor; use std::os::unix::net::UnixStream; use std::sync::Arc; @@ -54,6 +55,7 @@ fn frame_roundtrip_unix_pair() { restarts: 0, liveness_failures: 0, enabled: true, + labels: BTreeMap::new(), }], }, ) @@ -88,6 +90,7 @@ fn serve_list_roundtrip() { stream, &Response::Error { message: "no".into(), + code: None, }, )?, } diff --git a/tests/protocol_test.rs b/tests/protocol_test.rs index 440ac68..c52a07a 100644 --- a/tests/protocol_test.rs +++ b/tests/protocol_test.rs @@ -1,5 +1,7 @@ //! Unit/integration tests for microinit::protocol +use std::collections::BTreeMap; + use microinit::protocol::*; #[test] @@ -51,6 +53,7 @@ fn request_response_serde_roundtrip() { restarts: 2, liveness_failures: 1, enabled: true, + labels: BTreeMap::new(), }, }; let json = serde_json::to_string(&resp).unwrap(); @@ -67,6 +70,7 @@ fn request_response_serde_roundtrip() { restarts: 0, liveness_failures: 0, enabled: true, + labels: BTreeMap::new(), }, uptime_secs: Some(10), depends_on: vec![DepNode { @@ -155,6 +159,7 @@ fn describe_event_kinds_serde_roundtrip() { restarts: 0, liveness_failures: 0, enabled: true, + labels: BTreeMap::new(), }, uptime_secs: None, depends_on: vec![], diff --git a/tests/service_test.rs b/tests/service_test.rs index e623584..5b1adcb 100644 --- a/tests/service_test.rs +++ b/tests/service_test.rs @@ -1,8 +1,8 @@ //! Unit/integration tests for microinit::service -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; -use microinit::config::ServiceConfig; +use microinit::config::{RestartPolicy, ServiceConfig}; use microinit::service::*; fn cfg() -> ServiceConfig { @@ -10,7 +10,7 @@ fn cfg() -> ServiceConfig { name: "t".into(), enabled: true, daemon: false, - restart: false, + restart_policy: RestartPolicy::None, restart_backoff: 2, success_exit_codes: vec![0], start_wait_secs: 0, @@ -24,6 +24,7 @@ fn cfg() -> ServiceConfig { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), } } diff --git a/tests/supervisor_test.rs b/tests/supervisor_test.rs index 05d5d96..1ac328a 100644 --- a/tests/supervisor_test.rs +++ b/tests/supervisor_test.rs @@ -1,12 +1,12 @@ //! Unit/integration tests for microinit::supervisor -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::io::Write; use std::sync::{Arc, Mutex}; use std::thread; use std::time::Duration; -use microinit::config::{Config, LogsConfig, ServiceConfig}; +use microinit::config::{Config, LogsConfig, RestartPolicy, ServiceConfig}; use microinit::console::Console; use microinit::error::Error; use microinit::logs::LogHub; @@ -30,7 +30,7 @@ fn job(name: &str, start: &str, deps: &[&str], enabled: bool) -> ServiceConfig { name: name.into(), enabled, daemon: false, - restart: false, + restart_policy: RestartPolicy::None, restart_backoff: 1, success_exit_codes: vec![0], start_wait_secs: 0, @@ -44,6 +44,7 @@ fn job(name: &str, start: &str, deps: &[&str], enabled: bool) -> ServiceConfig { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), } } @@ -52,7 +53,7 @@ fn daemon_cfg(name: &str, start: &str) -> ServiceConfig { name: name.into(), enabled: true, daemon: true, - restart: false, + restart_policy: RestartPolicy::None, restart_backoff: 1, success_exit_codes: vec![0], start_wait_secs: 0, @@ -66,6 +67,7 @@ fn daemon_cfg(name: &str, start: &str) -> ServiceConfig { env: HashMap::new(), cwd: "/".into(), liveness_probe: None, + labels: BTreeMap::new(), } }