mirror of
https://github.com/onyx-dot-app/onyx.git
synced 2026-10-23 11:53:08 -05:00
## Description
Adds `ods stack ls` to list every Onyx dev stack on the machine from any
directory, and lets `stop`/`down --project NAME` act on a stack whose
worktree is gone. The PR also releases port leases when a stack stops
owning them.
**`ods stack ls [--json] [--all] [--orphaned]`**
(`internal/devstack/inventory.go`, `cmd/stack_ls.go`)
- It merges four sources, keyed by Compose project: `docker ps -a` +
`docker inspect` (labels, state, health), `projects/*/owner.json`, every
reachable `state.json` (owner state dirs, plus
`<git-common-dir>/{ods-stack,ods-stack-shared,ods-stack-frontend,worktrees/*/ods-stack,worktrees/*/ods-stack-frontend}`
from owner, lease, Docker and state roots), and `Registry.Leases()`. It
does not call `devstack.Open`, and it takes no locks.
- An Onyx Compose project is one with `relational_db` + `cache`
services, or one that a state or owner record claims.
- Statuses: `active` (an app process is alive, checked with
`groupIsOurs`, so a reused PID does not count), `app-stopped` (infra
runs, phase `stopped`), `infra-only` (infra runs, no app process: a
Compose-only project or dead processes), `stopped`, `orphaned`, and
`stale-lease` (`--all` only).
- Orphan rules: the Compose working dir is gone and no state claims the
project; a state's root is gone; or the root exists but is no longer a
git worktree (state rows, or Docker rows with an owner record whose
state is gone). A row counts as orphaned only when something is left to
clean up: processes, containers, leases or an owner record.
- A missing or corrupt state degrades its row (`errors`). When Docker is
down, `ls` prints a warning and shows rows from saved state only. Exit
codes: 0; 3 when `--orphaned` finds an orphan; 1 when no source can be
read.
**`down`/`stop --project NAME` and `prune`**
(`internal/devstack/project_down.go`)
- The command resolves the stack through `owner.json` → `state_dir`,
then any state that claims the project, then Docker labels. It runs from
any directory.
- It takes the state lock (only if the state dir still exists, so git
metadata is never recreated) and the project lock. It stops recorded
groups with `stopServices`/`stopRecords` and runs `docker compose -p
NAME down` from a neutral dir. Volumes are kept.
- After a successful down it releases the stack's leases and removes
`owner.json` for an orphan.
- It refuses a stack with a live service under an existing git worktree
unless you pass `--force`. If `--project` already matches the current
checkout's stack, the existing checkout path runs.
- `ods stack prune` runs this for every orphaned row after a
confirmation (`--yes` skips it). `--volumes` has its own confirmation,
and `--yes` does not skip it.
**Port lease fixes** (`ports.go`, `lifecycle.go`, `legacy.go`)
- `Stack.Stop` releases app leases (everything except `*_HOST_PORT`).
`Stack.Down` releases app leases after `stopServices`, and every lease
after Compose down. New `Registry.ReleaseNames`.
- Shared adoption (`releaseLegacyLeases`, runs after the legacy guard
and before `leasePorts`): each stopped legacy candidate without live
process groups loses all its leases, or only its app leases while its
containers run.
- Reclaim: `Lease.StateDir` is now recorded (`LeaseFor`). Another
owner's lease is reclaimable when its state is `stopped` or missing,
none of its recorded groups is alive, and, for `*_HOST_PORT` only, its
Compose project has no running container. That last check is lazy, so
app-port leasing stays fast. Leases without `StateDir` are reclaimable
only for shared/frontend keys, which are their own state dirs.
- Drift down: `choosePort` moves a lease to the lowest free port from
`Preferred` when the service is not running. I added
`PortRequest.Running` because `Fixed` alone is not set for a running
service that does not hold its port yet, and moving such a service
changed `web dev`'s full-stack owner
(`TestWebRuntimeUsesTheFullStackWebOwner` caught this).
Docs: `ods stack` help, `tools/ods/README.md`,
`tools/ods/integration/README.md`, and the onyx-dev-stack skill (a new
"Find and stop old stacks" step, and the updated port bullet).
Known limit: `git worktree remove` deletes
`<common>/worktrees/<name>/ods-stack`, so the recorded PIDs are lost.
`down --project` still removes the containers, leases and owner record,
but it cannot find app processes that outlived that removal.
**Discovery-first CLI.** Bare `ods stack` prints a short intro, then
groups its subcommands with the same `AddGroup` pattern as the root
command: Find stacks (`status`, `ls`, `logs`, `plan`), Run this
worktree's stack (`up`, `start`, `restart`, `stop`, `down`), and Clean
up (`prune`). Each `ls` row carries `next`, the exact command that stops
it, and the text output logs it with `log.Info`. Empty results and
refusals name the command to run.
Example on the VM (an isolated stack, then its worktree deleted with `rm
-rf`):
```
$ ods stack ls
PROJECT STATUS APP INFRA WEB AGE ROOT
demo-wt-fd863404 app-stopped - 7/7 - 1m ~/demo-wt
$ rm -rf ~/demo-wt && ods stack ls --orphaned # exit 3
PROJECT STATUS APP INFRA WEB AGE ROOT
demo-wt-fd863404 orphaned - 7/7 - 1m ~/demo-wt (missing)
level=info msg="To stop demo-wt-fd863404 (orphaned), run: ods stack down --project demo-wt-fd863404"
```
After `stack stop`, only the `*_HOST_PORT` leases remained. After `down
--project`, no containers or leases were left.
## How Has This Been Tested?
- Unit tests for the inventory merge (fake containers, owners, states,
leases, liveness): live, app-stopped, infra-only, Compose-only, orphan
with a missing root, a removed worktree whose directory remains, a stale
lease, Docker unavailable, unreadable sources, and sort order.
- Real-process tests for `DownProject`/`StopProject`: an orphan in a
deleted linked worktree is stopped, its leases are released and its
owner record is removed; a live worktree is refused without `--force`.
- Lease tests: reclaim of a stopped owner, live process group skipped,
non-stopped owner skipped, `*_HOST_PORT` with and without running
containers, drift down vs. running, `ReleaseNames`, stop/down releases,
adoption with and without a live legacy group, and the frontend handoff
releasing preview leases.
- Coverage gates: `cmd` is at 90.6% against a 90.4% floor, and
`internal/docker` at 99.3% against 99.2%.
- `go test ./internal/... ./cmd/` passes, except `TestRunCI_failures`,
which also fails on a clean main in this environment. golangci-lint
passes.
- New integration test `TestStackLiveOrphanedWorktree` (`-tags
integration`) passed on the VM (246 s). It starts an isolated stack in a
disposable worktree, deletes the worktree, asserts that `ls --orphaned`
lists the stack with exit 3, then that `down --project` stops its
processes, removes its containers and releases its leases.
## Additional Options
- [ ] [Optional] Please cherry-pick this PR to the latest release
version.
<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit
* **New Features**
* Added `ods stack ls` to view active, stopped, orphaned, and
stale-lease stacks with service, health, port, age, and location
details.
* Added JSON output, filtering, orphan detection, and project-based
shutdown support.
* Added `ods stack prune` to safely clean up orphaned stacks, with
optional volume removal.
* **Bug Fixes**
* Improved port lease reclamation and reassignment for stopped stacks.
* `stop` now releases application ports while preserving infrastructure
ports; `down` releases all ports.
* **Documentation**
* Expanded lifecycle, recovery, listing, and cleanup guidance.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
---------
Co-authored-by: roshan <rohod04@gmail.com>
Co-authored-by: roshan <38771624+rohoswagger@users.noreply.github.com>
927 lines
26 KiB
Go
927 lines
26 KiB
Go
package devstack
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
log "github.com/sirupsen/logrus"
|
|
)
|
|
|
|
// Provider supplies the environment-specific steps of a lifecycle: the tool
|
|
// checks, Compose, migrations and the service plan. The command wires the
|
|
// real one; tests inject fakes.
|
|
type Provider interface {
|
|
// Prepare installs dependencies and checks the host tools.
|
|
Prepare(ctx context.Context) error
|
|
// InfraPorts lists the infrastructure host ports to lease. A request is
|
|
// Fixed when the worktree's container already runs on a port.
|
|
InfraPorts() ([]PortRequest, error)
|
|
// AppPorts lists the application host ports to lease for the plan.
|
|
AppPorts(opts Options) []PortRequest
|
|
// ComposeUp starts the infrastructure containers on the leased ports.
|
|
ComposeUp(ctx context.Context, ports map[string]int, opts Options) error
|
|
// ComposeDown stops the infrastructure containers.
|
|
ComposeDown(ctx context.Context) error
|
|
// Migrate runs the database migrations with env.
|
|
Migrate(ctx context.Context, env []string) error
|
|
MigrationsPending(ctx context.Context, env []string) (bool, error)
|
|
// Plan resolves the services to run. env carries the shell environment
|
|
// and the stack's own overrides, which every service gets.
|
|
Plan(root string, ports map[string]int, env Env, opts Options) ([]Spec, error)
|
|
// InfraEnv maps leased infrastructure ports to the app variables, such
|
|
// as POSTGRES_PORT, written to .vscode/.env and the service env.
|
|
InfraEnv(ports map[string]int) map[string]string
|
|
}
|
|
|
|
// Options tune one lifecycle run.
|
|
type Options struct {
|
|
Profile string
|
|
LocalModelServer bool
|
|
// ReadyTimeout bounds the wait for health checks.
|
|
ReadyTimeout time.Duration
|
|
// LockWait bounds the wait for another lifecycle holder.
|
|
LockWait time.Duration
|
|
// StopTimeout is the grace period before SIGKILL.
|
|
StopTimeout time.Duration
|
|
// SkipPrepare skips dependency installation.
|
|
SkipPrepare bool
|
|
// BindRetries is how many times a service whose leased port turned
|
|
// out busy is moved to a new port.
|
|
BindRetries int
|
|
// SettleTime is how long a service without a health check must stay
|
|
// alive before the stack counts as ready, so a crash on startup is
|
|
// reported rather than hidden by a fast exit.
|
|
SettleTime time.Duration
|
|
}
|
|
|
|
// WithDefaults fills the fields that have no CLI flag. The timeouts keep
|
|
// their value, so an explicit zero means no wait.
|
|
func (o Options) WithDefaults() Options {
|
|
if o.Profile == "" {
|
|
o.Profile = "default"
|
|
}
|
|
if o.BindRetries == 0 {
|
|
o.BindRetries = 3
|
|
}
|
|
if o.SettleTime == 0 {
|
|
o.SettleTime = 5 * time.Second
|
|
}
|
|
return o
|
|
}
|
|
|
|
// Up prepares the worktree, starts infrastructure, migrates, and starts or
|
|
// reconciles the application services, then waits for readiness.
|
|
func (s *Stack) Up(ctx context.Context, p Provider, opts Options) (*Status, error) {
|
|
if s.Mode == Frontend {
|
|
return nil, fmt.Errorf("use 'ods web dev' to launch the frontend")
|
|
}
|
|
return s.run(ctx, p, opts, true)
|
|
}
|
|
|
|
// Start starts or reconciles the application services against the
|
|
// infrastructure that Up already started.
|
|
func (s *Stack) Start(ctx context.Context, p Provider, opts Options) (*Status, error) {
|
|
if s.Mode == Frontend {
|
|
return nil, fmt.Errorf("use 'ods web dev' to launch the frontend")
|
|
}
|
|
return s.run(ctx, p, opts, false)
|
|
}
|
|
|
|
func (s *Stack) run(ctx context.Context, p Provider, opts Options, infra bool) (*Status, error) {
|
|
opts = opts.WithDefaults()
|
|
unlock, err := s.Lock(ctx, "stack up", opts.LockWait)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer unlock()
|
|
return s.runLocked(ctx, p, opts, infra)
|
|
}
|
|
|
|
// runLocked is run with the lifecycle lock already held.
|
|
func (s *Stack) runLocked(ctx context.Context, p Provider, opts Options, infra bool) (*Status, error) {
|
|
st, err := s.ReadState()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if st != nil {
|
|
if st.Project != s.Project {
|
|
if !s.ProjectExplicit {
|
|
s.Project = st.Project
|
|
} else if st.Phase != PhaseStopped || !infra {
|
|
return nil, fmt.Errorf("stop project %s from %s before selecting another project", st.Project, st.Root)
|
|
}
|
|
}
|
|
}
|
|
if st == nil && s.Mode == Shared {
|
|
if err := s.selectLegacyProject(); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
releaseLocal, err := s.lockFrontendCheckout(ctx, opts.LockWait)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer releaseLocal()
|
|
if s.Mode != Frontend {
|
|
release, err := s.LockInfrastructure(ctx, opts.LockWait)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer release()
|
|
}
|
|
if s.Mode == Shared && st != nil && st.Root != s.Root {
|
|
if !infra {
|
|
return nil, s.ownerError(st)
|
|
}
|
|
if err := s.stopServices(opts.StopTimeout); err != nil {
|
|
return nil, err
|
|
}
|
|
st, err = s.ReadState()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
previewPort, err := s.stopFrontendPreview(ctx, opts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if st == nil {
|
|
if s.Mode == Shared {
|
|
lockedProject := s.Project
|
|
if err := s.selectLegacyProject(); err != nil {
|
|
return nil, err
|
|
}
|
|
if s.Project != lockedProject {
|
|
return nil, fmt.Errorf("existing project changed during startup; retry 'ods stack up'")
|
|
}
|
|
if err := s.releaseLegacyLeases(); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
st = &State{Ports: map[string]int{}}
|
|
}
|
|
st.Root = s.Root
|
|
if previewPort > 0 {
|
|
st.Ports["web"] = previewPort
|
|
}
|
|
if st.Profile != "" && st.Profile != opts.Profile {
|
|
log.Infof("Switching profile from %s to %s", st.Profile, opts.Profile)
|
|
}
|
|
st.Profile = opts.Profile
|
|
st.SettleTime = opts.SettleTime
|
|
st.Errors = nil
|
|
st.Phase = PhasePreparing
|
|
if err := s.WriteState(st); err != nil {
|
|
return nil, err
|
|
}
|
|
if _, _, err := PrepareEnvFiles(s.Root); err != nil {
|
|
return nil, s.fail(st, err)
|
|
}
|
|
if !opts.SkipPrepare {
|
|
if err := p.Prepare(ctx); err != nil {
|
|
return nil, s.fail(st, err)
|
|
}
|
|
}
|
|
|
|
var requests []PortRequest
|
|
if infra {
|
|
infraRequests, err := p.InfraPorts()
|
|
if err != nil {
|
|
return nil, s.fail(st, err)
|
|
}
|
|
requests = append(requests, infraRequests...)
|
|
} else if len(st.Ports) == 0 {
|
|
return nil, s.fail(st, errors.New("no infrastructure ports recorded; run `ods stack up` first"))
|
|
}
|
|
appRequests := p.AppPorts(opts)
|
|
for i, req := range appRequests {
|
|
if req.Name == "web" && previewPort > 0 {
|
|
appRequests[i].Preferred = previewPort
|
|
appRequests[i].Replace = true
|
|
}
|
|
}
|
|
requests = append(requests, appRequests...)
|
|
if err := s.leasePorts(st, requests); err != nil {
|
|
return nil, s.fail(st, err)
|
|
}
|
|
|
|
if infra {
|
|
st.Phase = PhaseInfra
|
|
if err := s.WriteState(st); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := p.ComposeUp(ctx, st.Ports, opts); err != nil {
|
|
return nil, s.fail(st, err)
|
|
}
|
|
}
|
|
|
|
env, fileEnv, err := s.serviceEnv(p, st.Ports, opts)
|
|
if err != nil {
|
|
return nil, s.fail(st, err)
|
|
}
|
|
|
|
if infra {
|
|
migrationEnv := env.With(fileEnv)
|
|
pending, err := p.MigrationsPending(ctx, migrationEnv)
|
|
if err != nil {
|
|
return nil, s.fail(st, fmt.Errorf("check migrations: %w", err))
|
|
}
|
|
if pending {
|
|
if err := s.stopServices(opts.StopTimeout); err != nil {
|
|
return nil, err
|
|
}
|
|
st.Services = nil
|
|
st.Phase = PhaseMigrating
|
|
if err := s.WriteState(st); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := p.Migrate(ctx, migrationEnv); err != nil {
|
|
return nil, s.fail(st, fmt.Errorf("migrations: %w", err))
|
|
}
|
|
}
|
|
}
|
|
|
|
st.Phase = PhaseStarting
|
|
if err := s.WriteState(st); err != nil {
|
|
return nil, err
|
|
}
|
|
for attempt := 0; ; attempt++ {
|
|
specs, err := p.Plan(s.Root, st.Ports, env, opts)
|
|
if err != nil {
|
|
return nil, s.fail(st, err)
|
|
}
|
|
var prepare func(Spec) error
|
|
if hook, ok := p.(interface {
|
|
PrepareService(context.Context, Spec) error
|
|
}); ok {
|
|
prepare = func(sp Spec) error { return hook.PrepareService(ctx, sp) }
|
|
}
|
|
busy, err := s.reconcile(st, specs, opts.StopTimeout, prepare)
|
|
if err != nil {
|
|
return nil, s.fail(st, err)
|
|
}
|
|
if len(busy) == 0 {
|
|
status, readyErr := s.waitReady(ctx, st, opts.ReadyTimeout, opts.SettleTime)
|
|
var conflict *bindFailure
|
|
if !errors.As(readyErr, &conflict) {
|
|
return status, readyErr
|
|
}
|
|
busy = conflict.services
|
|
}
|
|
if attempt >= opts.BindRetries {
|
|
return nil, s.fail(st, fmt.Errorf("ports still busy after %d retries: %s", opts.BindRetries, strings.Join(busy, ", ")))
|
|
}
|
|
log.Warnf("Leased ports are busy (%s), leasing new ones", strings.Join(busy, ", "))
|
|
for i, rec := range st.Services {
|
|
for _, name := range busy {
|
|
if rec.Slug != name {
|
|
continue
|
|
}
|
|
if left := stopRecords([]ServiceRecord{rec}, opts.StopTimeout); len(left) > 0 {
|
|
return nil, s.fail(st, fmt.Errorf("cannot stop %s before bind retry", rec.Name))
|
|
}
|
|
st.Services[i].PID = 0
|
|
st.Services[i].PGID = 0
|
|
}
|
|
}
|
|
var retry []PortRequest
|
|
for _, req := range p.AppPorts(opts) {
|
|
for _, name := range busy {
|
|
if req.Name == name {
|
|
delete(st.Ports, name)
|
|
retry = append(retry, req)
|
|
}
|
|
}
|
|
}
|
|
if err := s.leasePorts(st, retry); err != nil {
|
|
return nil, s.fail(st, err)
|
|
}
|
|
if env, _, err = s.serviceEnv(p, st.Ports, opts); err != nil {
|
|
return nil, s.fail(st, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
type bindFailure struct {
|
|
services []string
|
|
}
|
|
|
|
func (e *bindFailure) Error() string {
|
|
return "services failed to bind: " + strings.Join(e.services, ", ")
|
|
}
|
|
|
|
func failedBind(rec ServiceRecord) bool {
|
|
if rec.Port <= 0 || rec.Log == "" {
|
|
return false
|
|
}
|
|
data, err := readLogTail(rec.Log)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
output := strings.ToLower(string(data))
|
|
return strings.Contains(output, "address already in use") || strings.Contains(output, "eaddrinuse")
|
|
}
|
|
|
|
func (s *Stack) fail(st *State, err error) error {
|
|
st.Phase = PhaseFailed
|
|
st.Errors = append(st.Errors, err.Error())
|
|
if werr := s.WriteState(st); werr != nil {
|
|
log.Warnf("Failed to write state: %v", werr)
|
|
}
|
|
return err
|
|
}
|
|
|
|
// leasePorts leases requests. A port that one of this stack's own running
|
|
// services holds is kept as is; the registry rechecks every other lease.
|
|
func (s *Stack) leasePorts(st *State, requests []PortRequest) error {
|
|
if s.Registry == nil {
|
|
return errors.New("no port registry")
|
|
}
|
|
for i, req := range requests {
|
|
for _, rec := range st.Services {
|
|
if !isCurrent(rec) {
|
|
continue
|
|
}
|
|
port := 0
|
|
switch req.Name {
|
|
case rec.Slug:
|
|
port = rec.Port
|
|
case rec.Slug + "-metrics":
|
|
port = st.Ports[req.Name]
|
|
}
|
|
// A port stays fixed only while this stack's own process group
|
|
// holds it; a port lost to another process is leased again.
|
|
if port > 0 {
|
|
requests[i].Running = true
|
|
if portHeldByGroup(port, rec.PGID) {
|
|
requests[i].Fixed = port
|
|
}
|
|
}
|
|
}
|
|
}
|
|
ports, err := s.Registry.LeaseFor(s.leaseRoot(), s.StateDir, requests)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for name, port := range ports {
|
|
st.Ports[name] = port
|
|
}
|
|
return s.WriteState(st)
|
|
}
|
|
|
|
// leaseRoot is the owner key of the stack's port leases: the checkout for
|
|
// an isolated stack, the state directory otherwise.
|
|
func (s *Stack) leaseRoot() string {
|
|
if s.Mode == Isolated {
|
|
return s.Root
|
|
}
|
|
return s.StateDir
|
|
}
|
|
|
|
// releaseLeases drops the stack's application port leases, and with infra
|
|
// also its Compose host port leases.
|
|
func (s *Stack) releaseLeases(st *State, infra bool) error {
|
|
if s.Registry == nil {
|
|
return nil
|
|
}
|
|
if infra {
|
|
return s.Registry.Release(s.leaseRoot())
|
|
}
|
|
return s.Registry.ReleaseNames(s.leaseRoot(), appLeaseNames(st))
|
|
}
|
|
|
|
// appLeaseNames lists the leased application ports of st, such as web, api
|
|
// and *-metrics, without the Compose host ports.
|
|
func appLeaseNames(st *State) []string {
|
|
if st == nil {
|
|
return nil
|
|
}
|
|
var names []string
|
|
for name := range st.Ports {
|
|
if !isInfraLease(name) {
|
|
names = append(names, name)
|
|
}
|
|
}
|
|
sort.Strings(names)
|
|
return names
|
|
}
|
|
|
|
// Env is the layered environment of a service: Base is the shell, and
|
|
// Overrides are the hosts and leased ports of this stack, which win over any
|
|
// env file or launch configuration in between.
|
|
type Env struct {
|
|
Base []string
|
|
Overrides []string
|
|
}
|
|
|
|
// With resolves the environment with layers applied between Base and
|
|
// Overrides, in order.
|
|
func (e Env) With(layers ...[]string) []string {
|
|
env := e.Base
|
|
for _, layer := range layers {
|
|
env = MergeEnv(env, layer)
|
|
}
|
|
return MergeEnv(env, e.Overrides)
|
|
}
|
|
|
|
// serviceEnv builds the environment every service inherits, and returns the
|
|
// entries of .vscode/.env separately. The infrastructure port variables are
|
|
// also written to .vscode/.env for commands run by hand.
|
|
func (s *Stack) serviceEnv(p Provider, ports map[string]int, opts Options) (Env, []string, error) {
|
|
values := p.InfraEnv(ports)
|
|
envPath := filepath.Join(s.Root, ".vscode", ".env")
|
|
if err := UpsertEnvFile(envPath, values); err != nil {
|
|
return Env{}, nil, err
|
|
}
|
|
fileEnv, err := LoadEnvFile(envPath)
|
|
if err != nil {
|
|
return Env{}, nil, err
|
|
}
|
|
envPorts := make(map[string]int, len(ports))
|
|
for name, port := range ports {
|
|
if name != "model" || opts.LocalModelServer {
|
|
envPorts[name] = port
|
|
}
|
|
}
|
|
env := HostServiceEnv(envPorts)
|
|
return env, fileEnv, nil
|
|
}
|
|
|
|
// reconcile makes the running services match specs. A record whose process
|
|
// is current and whose spec hash is unchanged is kept; every other record is
|
|
// stopped and its spec started. It returns the names of ports that a new
|
|
// service could not bind because the port is busy.
|
|
func (s *Stack) reconcile(st *State, specs []Spec, stopTimeout time.Duration, prepare func(Spec) error) ([]string, error) {
|
|
wanted := map[string]Spec{}
|
|
for _, sp := range specs {
|
|
wanted[sp.Slug] = sp
|
|
}
|
|
|
|
var kept []ServiceRecord
|
|
var stale []ServiceRecord
|
|
for _, rec := range st.Services {
|
|
sp, ok := wanted[rec.Slug]
|
|
if ok && sp.Skip == "" && rec.Skipped == "" && rec.SpecHash == sp.Hash() && isCurrent(rec) {
|
|
kept = append(kept, rec)
|
|
continue
|
|
}
|
|
log.Debugf("%s is stale: wanted=%v hash=%s/%s current=%v", rec.Slug, ok, rec.SpecHash, sp.Hash(), isCurrent(rec))
|
|
if rec.PGID > 0 {
|
|
stale = append(stale, rec)
|
|
}
|
|
}
|
|
if len(stale) > 0 {
|
|
names := make([]string, 0, len(stale))
|
|
for _, rec := range stale {
|
|
names = append(names, rec.Name)
|
|
}
|
|
log.Infof("Stopping stale services: %s", strings.Join(names, ", "))
|
|
if left := stopRecords(stale, stopTimeout); len(left) > 0 {
|
|
for _, rec := range left {
|
|
return nil, fmt.Errorf("could not stop %s (pgid %d)", rec.Name, rec.PGID)
|
|
}
|
|
}
|
|
}
|
|
st.Services = kept
|
|
if err := s.WriteState(st); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
keptSlugs := map[string]bool{}
|
|
for _, rec := range kept {
|
|
keptSlugs[rec.Slug] = true
|
|
}
|
|
var busy []string
|
|
for _, sp := range specs {
|
|
if keptSlugs[sp.Slug] {
|
|
continue
|
|
}
|
|
if sp.Skip == "" && sp.FixedPort > 0 && !bindable(sp.FixedPort) {
|
|
sp.Skip = fmt.Sprintf("its port %d is in use", sp.FixedPort)
|
|
}
|
|
if sp.Skip != "" {
|
|
log.Infof("Skipping %s: %s", sp.Name, sp.Skip)
|
|
st.Services = append(st.Services, ServiceRecord{Name: sp.Name, Slug: sp.Slug, Skipped: sp.Skip, Optional: sp.Optional, SpecHash: sp.Hash()})
|
|
continue
|
|
}
|
|
if sp.Port > 0 && !bindable(sp.Port) {
|
|
busy = append(busy, sp.Slug)
|
|
continue
|
|
}
|
|
if prepare != nil {
|
|
if err := prepare(sp); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
rec, err := startService(sp, filepath.Join(s.LogDir(), sp.Slug+".log"))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
log.Infof("Started %s (pid %d), log: %s", sp.Name, rec.PID, rec.Log)
|
|
st.Services = append(st.Services, rec)
|
|
if err := s.WriteState(st); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
sortRecords(st.Services, specs)
|
|
return busy, s.WriteState(st)
|
|
}
|
|
|
|
func sortRecords(recs []ServiceRecord, specs []Spec) {
|
|
order := map[string]int{}
|
|
for i, sp := range specs {
|
|
order[sp.Slug] = i
|
|
}
|
|
sort.SliceStable(recs, func(i, j int) bool { return order[recs[i].Slug] < order[recs[j].Slug] })
|
|
}
|
|
|
|
// Stop stops the managed services and keeps the infrastructure running.
|
|
func (s *Stack) Stop(ctx context.Context, opts Options) error {
|
|
opts = opts.WithDefaults()
|
|
unlock, err := s.Lock(ctx, "stack stop", opts.LockWait)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer unlock()
|
|
if err := s.checkOwner(); err != nil {
|
|
return err
|
|
}
|
|
if err := s.stopServices(opts.StopTimeout); err != nil {
|
|
return err
|
|
}
|
|
st, err := s.ReadState()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return s.releaseLeases(st, false)
|
|
}
|
|
|
|
func (s *Stack) stopServices(timeout time.Duration) error {
|
|
st, err := s.ReadState()
|
|
if err != nil || st == nil {
|
|
return err
|
|
}
|
|
left := stopRecords(st.Services, timeout)
|
|
st.Services = left
|
|
st.Phase = PhaseStopped
|
|
if len(left) > 0 {
|
|
st.Phase = PhaseFailed
|
|
names := make([]string, 0, len(left))
|
|
for _, rec := range left {
|
|
names = append(names, rec.Name)
|
|
}
|
|
err = fmt.Errorf("could not stop: %s", strings.Join(names, ", "))
|
|
st.Errors = append(st.Errors, err.Error())
|
|
}
|
|
if werr := s.WriteState(st); werr != nil {
|
|
return werr
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (s *Stack) ownerError(st *State) error {
|
|
return fmt.Errorf("shared stack is owned by %s; run this command there, or run 'ods stack up' here to switch checkouts", st.Root)
|
|
}
|
|
|
|
func (s *Stack) checkOwner() error {
|
|
st, err := s.ReadState()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if st != nil && st.Project != s.Project {
|
|
return fmt.Errorf("recorded project is %s, not %s; stop the recorded stack before switching projects with 'ods stack up'", st.Project, s.Project)
|
|
}
|
|
if s.Mode == Shared && st != nil && st.Root != s.Root {
|
|
return s.ownerError(st)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Down stops the services and the infrastructure. Volumes and port leases
|
|
// are kept so the next Up reuses them.
|
|
func (s *Stack) Down(ctx context.Context, p Provider, opts Options) error {
|
|
opts = opts.WithDefaults()
|
|
unlock, err := s.Lock(ctx, "stack down", opts.LockWait)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer unlock()
|
|
if err := s.checkOwner(); err != nil {
|
|
return err
|
|
}
|
|
st, err := s.ReadState()
|
|
if err != nil || st == nil {
|
|
return err
|
|
}
|
|
if s.Mode != Frontend {
|
|
release, err := s.LockInfrastructure(ctx, opts.LockWait)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer release()
|
|
}
|
|
if err := s.stopServices(opts.StopTimeout); err != nil {
|
|
return err
|
|
}
|
|
if err := s.releaseLeases(st, false); err != nil {
|
|
return err
|
|
}
|
|
if s.Mode == Frontend {
|
|
return nil
|
|
}
|
|
if err := p.ComposeDown(ctx); err != nil {
|
|
return err
|
|
}
|
|
return s.releaseLeases(st, true)
|
|
}
|
|
|
|
// Restart stops the services and starts them again against the running
|
|
// infrastructure.
|
|
func (s *Stack) Restart(ctx context.Context, p Provider, opts Options) (*Status, error) {
|
|
if s.Mode == Frontend {
|
|
return nil, fmt.Errorf("use 'ods web dev' to launch the frontend")
|
|
}
|
|
opts = opts.WithDefaults()
|
|
unlock, err := s.Lock(ctx, "stack restart", opts.LockWait)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer unlock()
|
|
if err := s.checkOwner(); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.stopServices(opts.StopTimeout); err != nil {
|
|
return nil, err
|
|
}
|
|
return s.runLocked(ctx, p, opts, false)
|
|
}
|
|
|
|
// ServiceStatus is one service as reported by Status.
|
|
type ServiceStatus struct {
|
|
Name string `json:"name"`
|
|
Slug string `json:"slug"`
|
|
State string `json:"state"`
|
|
PID int `json:"pid,omitempty"`
|
|
Port int `json:"port,omitempty"`
|
|
Health string `json:"health,omitempty"`
|
|
Healthy *bool `json:"healthy,omitempty"`
|
|
Log string `json:"log,omitempty"`
|
|
Reason string `json:"reason,omitempty"`
|
|
Optional bool `json:"optional,omitempty"`
|
|
// LogTail holds the last lines of the log of a service that exited or
|
|
// is unhealthy, so a failure can be read from the status alone.
|
|
LogTail string `json:"log_tail,omitempty"`
|
|
}
|
|
|
|
// Service states.
|
|
const (
|
|
StateRunning = "running"
|
|
StateExited = "exited"
|
|
StateSkipped = "skipped"
|
|
)
|
|
|
|
// Status is the live view of a stack.
|
|
type Status struct {
|
|
Root string `json:"root"`
|
|
Project string `json:"project"`
|
|
Profile string `json:"profile"`
|
|
Phase string `json:"phase"`
|
|
Ready bool `json:"ready"`
|
|
Ports map[string]int `json:"ports"`
|
|
URLs map[string]string `json:"urls"`
|
|
Services []ServiceStatus `json:"services"`
|
|
Errors []string `json:"errors"`
|
|
StateDir string `json:"state_dir"`
|
|
|
|
// healthyNow is Ready without the phase condition, for the readiness
|
|
// wait while the phase is still starting.
|
|
healthyNow bool
|
|
}
|
|
|
|
// Status reports the current state without taking the lock. It returns nil
|
|
// when the stack has never been started.
|
|
func (s *Stack) Status() (*Status, error) {
|
|
st, err := s.ReadState()
|
|
if err != nil || st == nil {
|
|
return nil, err
|
|
}
|
|
return s.status(st, true), nil
|
|
}
|
|
|
|
// status is statusCtx without a deadline, for calls outside a readiness
|
|
// wait.
|
|
func (s *Stack) status(st *State, probe bool) *Status {
|
|
return s.statusCtx(context.Background(), st, probe)
|
|
}
|
|
|
|
// statusCtx computes the live status. Health probes are bounded by ctx.
|
|
func (s *Stack) statusCtx(ctx context.Context, st *State, probe bool) *Status {
|
|
out := &Status{
|
|
Root: st.Root,
|
|
Project: st.Project,
|
|
Profile: st.Profile,
|
|
Phase: st.Phase,
|
|
Ports: st.Ports,
|
|
URLs: URLs(st.Ports),
|
|
Errors: st.Errors,
|
|
StateDir: s.StateDir,
|
|
}
|
|
if out.Errors == nil {
|
|
out.Errors = []string{}
|
|
}
|
|
ready := len(st.Services) > 0
|
|
for _, rec := range st.Services {
|
|
ss := ServiceStatus{Name: rec.Name, Slug: rec.Slug, PID: rec.PID, Port: rec.Port, Health: rec.Health, Log: rec.Log, Optional: rec.Optional}
|
|
switch {
|
|
case rec.Skipped != "":
|
|
ss.State, ss.Reason, ss.PID = StateSkipped, rec.Skipped, 0
|
|
case isCurrent(rec):
|
|
ss.State = StateRunning
|
|
if rec.Health != "" && probe {
|
|
healthy := healthy(ctx, rec.Health)
|
|
ss.Healthy = &healthy
|
|
if !healthy {
|
|
ss.LogTail = logTail(rec.Log, logTailLines)
|
|
if !rec.Optional {
|
|
ready = false
|
|
}
|
|
}
|
|
}
|
|
default:
|
|
ss.State = StateExited
|
|
ss.LogTail = logTail(rec.Log, logTailLines)
|
|
if !rec.Optional {
|
|
ready = false
|
|
}
|
|
}
|
|
out.Services = append(out.Services, ss)
|
|
}
|
|
if out.Services == nil {
|
|
out.Services = []ServiceStatus{}
|
|
}
|
|
settle := st.SettleTime
|
|
if settle == 0 {
|
|
settle = Options{}.WithDefaults().SettleTime
|
|
}
|
|
out.healthyNow = ready && settled(st.Services, settle)
|
|
// A stack is ready only after a reconcile finished: an interrupted
|
|
// start may have saved just part of the profile.
|
|
out.Ready = out.healthyNow && (st.Phase == PhaseReady || st.Phase == PhaseDegraded)
|
|
return out
|
|
}
|
|
|
|
// settled reports whether every required service without a health check
|
|
// has been alive for at least settle.
|
|
func settled(recs []ServiceRecord, settle time.Duration) bool {
|
|
for _, rec := range recs {
|
|
if rec.Skipped != "" || rec.Optional || rec.Health != "" {
|
|
continue
|
|
}
|
|
if time.Since(rec.StartedAt) < settle {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// URLs lists the stack's entry points by port name.
|
|
func URLs(ports map[string]int) map[string]string {
|
|
urls := map[string]string{}
|
|
for name, port := range ports {
|
|
switch name {
|
|
case "web", "api", "mcp", "model":
|
|
urls[name] = fmt.Sprintf("http://localhost:%d", port)
|
|
}
|
|
}
|
|
return urls
|
|
}
|
|
|
|
// logTailLines is how many log lines a failing service reports.
|
|
const logTailLines = 30
|
|
|
|
const maxLogTailBytes = 64 * 1024
|
|
|
|
func readLogTail(path string) ([]byte, error) {
|
|
file, err := os.Open(path)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = file.Close() }()
|
|
info, err := file.Stat()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if info.Size() > maxLogTailBytes {
|
|
if _, err := file.Seek(info.Size()-maxLogTailBytes, io.SeekStart); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return io.ReadAll(io.LimitReader(file, maxLogTailBytes))
|
|
}
|
|
|
|
// logTail returns the last n lines of a log file, or "" when unreadable.
|
|
func logTail(path string, n int) string {
|
|
if path == "" || n <= 0 {
|
|
return ""
|
|
}
|
|
data, err := readLogTail(path)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
lines := strings.Split(strings.TrimRight(string(data), "\n"), "\n")
|
|
if len(lines) > n {
|
|
lines = lines[len(lines)-n:]
|
|
}
|
|
return strings.Join(lines, "\n")
|
|
}
|
|
|
|
// healthy reports whether url answers 200 within 3 seconds and before ctx
|
|
// ends, whichever comes first.
|
|
func healthy(ctx context.Context, url string) bool {
|
|
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
|
|
defer cancel()
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
resp, err := http.DefaultClient.Do(req)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
_ = resp.Body.Close()
|
|
return resp.StatusCode == http.StatusOK
|
|
}
|
|
|
|
// waitReady polls the services until every required one is running and
|
|
// healthy and has been alive for settle, or a required one exited, or the
|
|
// timeout passes.
|
|
func (s *Stack) waitReady(ctx context.Context, st *State, timeout, settle time.Duration) (*Status, error) {
|
|
deadline := time.Now().Add(timeout)
|
|
// Health probes never run past the deadline.
|
|
probeCtx, cancel := context.WithDeadline(ctx, deadline)
|
|
defer cancel()
|
|
for {
|
|
var conflicts []string
|
|
for i := range st.Services {
|
|
rec := &st.Services[i]
|
|
rememberGroup(rec)
|
|
if failedBind(*rec) {
|
|
conflicts = append(conflicts, rec.Slug)
|
|
}
|
|
}
|
|
if err := s.WriteState(st); err != nil {
|
|
return nil, err
|
|
}
|
|
status := s.statusCtx(probeCtx, st, true)
|
|
if len(conflicts) > 0 {
|
|
return status, &bindFailure{services: conflicts}
|
|
}
|
|
var exited []string
|
|
for _, ss := range status.Services {
|
|
if ss.State == StateExited && !ss.Optional {
|
|
exited = append(exited, ss.Name)
|
|
}
|
|
}
|
|
if len(exited) > 0 {
|
|
err := s.fail(st, fmt.Errorf("services exited during startup: %s", strings.Join(exited, ", ")))
|
|
return s.status(st, true), err
|
|
}
|
|
if status.healthyNow {
|
|
st.Phase = PhaseReady
|
|
if err := s.WriteState(st); err != nil {
|
|
return status, err
|
|
}
|
|
// Return the snapshot that passed, not a fresh probe that could
|
|
// disagree with the nil error.
|
|
status.Phase = PhaseReady
|
|
status.Ready = true
|
|
return status, nil
|
|
}
|
|
if time.Now().After(deadline) {
|
|
st.Phase = PhaseDegraded
|
|
err := fmt.Errorf("stack not ready after %s", timeout)
|
|
st.Errors = append(st.Errors, err.Error())
|
|
if werr := s.WriteState(st); werr != nil {
|
|
return status, werr
|
|
}
|
|
// Report the probe that timed out; do not probe again.
|
|
status.Phase = PhaseDegraded
|
|
status.Errors = st.Errors
|
|
return status, err
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return status, ctx.Err()
|
|
case <-time.After(time.Second):
|
|
}
|
|
}
|
|
}
|