diff --git a/README.md b/README.md index d556f6e..824c29d 100644 --- a/README.md +++ b/README.md @@ -750,6 +750,11 @@ Long-running `./jeannie up` steps keep compact output by default, but the active step line refreshes with elapsed time and the latest log line. Set `JEANNIE_PROGRESS_INTERVAL_SECONDS=5` for faster heartbeats, or `JEANNIE_VERBOSE=true` to stream full command output. +Independent host services run through the Go DAG runner during `up`: Gitea +deploy/bootstrap, RPi services, Pimox workers, OpenWrt, and OCI edge host checks +can progress concurrently. Set `JEANNIE_UP_PARALLELISM=1` to force sequential +execution while troubleshooting, or raise it above the default `4` if the lab +can absorb more concurrent work. Run a full cluster rebuild from the Debian server with: diff --git a/README.md.tmpl b/README.md.tmpl index b4c7034..0868164 100644 --- a/README.md.tmpl +++ b/README.md.tmpl @@ -750,6 +750,11 @@ Long-running `./{{ main_script }} up` steps keep compact output by default, but the active step line refreshes with elapsed time and the latest log line. Set `JEANNIE_PROGRESS_INTERVAL_SECONDS=5` for faster heartbeats, or `JEANNIE_VERBOSE=true` to stream full command output. +Independent host services run through the Go DAG runner during `up`: Gitea +deploy/bootstrap, RPi services, Pimox workers, OpenWrt, and OCI edge host checks +can progress concurrently. Set `JEANNIE_UP_PARALLELISM=1` to force sequential +execution while troubleshooting, or raise it above the default `4` if the lab +can absorb more concurrent work. Run a full cluster rebuild from the Debian server with: diff --git a/cmd/jeannie-core/main.go b/cmd/jeannie-core/main.go index d826cb3..8dd2853 100644 --- a/cmd/jeannie-core/main.go +++ b/cmd/jeannie-core/main.go @@ -1,13 +1,18 @@ package main import ( + "bufio" + "context" "encoding/json" "flag" "fmt" "io" "os" + "os/exec" "sort" "strings" + "sync" + "time" ) type reportRow struct { @@ -33,6 +38,11 @@ func main() { fmt.Fprintln(os.Stderr, err) os.Exit(1) } + case "run-dag": + if err := runDAG(os.Args[2:], os.Stdout, os.Stderr); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } case "help", "-h", "--help": usage(os.Stdout) default: @@ -42,7 +52,184 @@ func main() { } func usage(out io.Writer) { - fmt.Fprintln(out, "Usage: jeannie-core report-render [--title TITLE] [--details] [--only failures|warnings|problems|all] [--json]") + fmt.Fprintln(out, "Usage: jeannie-core {report-render|run-dag}") +} + +type dagSpec struct { + Tasks []dagTask `json:"tasks"` + Parallelism int `json:"parallelism"` +} + +type dagTask struct { + ID string `json:"id"` + Title string `json:"title"` + Command []string `json:"command"` + Needs []string `json:"needs"` +} + +type dagTaskResult struct { + id string + exitCode int + err error +} + +func runDAG(args []string, out io.Writer, errOut io.Writer) error { + flags := flag.NewFlagSet("run-dag", flag.ContinueOnError) + flags.SetOutput(io.Discard) + specPath := flags.String("spec", "", "") + if err := flags.Parse(args); err != nil { + return err + } + if *specPath == "" { + return fmt.Errorf("run-dag requires --spec") + } + + specData, err := os.ReadFile(*specPath) + if err != nil { + return err + } + var spec dagSpec + if err := json.Unmarshal(specData, &spec); err != nil { + return err + } + return executeDAG(context.Background(), spec, out, errOut) +} + +func executeDAG(ctx context.Context, spec dagSpec, out io.Writer, errOut io.Writer) error { + if spec.Parallelism <= 0 { + spec.Parallelism = 4 + } + + taskByID := map[string]dagTask{} + dependents := map[string][]string{} + remainingNeeds := map[string]int{} + for _, task := range spec.Tasks { + if task.ID == "" { + return fmt.Errorf("dag task has empty id") + } + if _, exists := taskByID[task.ID]; exists { + return fmt.Errorf("duplicate dag task id %q", task.ID) + } + if len(task.Command) == 0 { + return fmt.Errorf("dag task %q has empty command", task.ID) + } + taskByID[task.ID] = task + remainingNeeds[task.ID] = len(task.Needs) + for _, need := range task.Needs { + dependents[need] = append(dependents[need], task.ID) + } + } + for _, task := range spec.Tasks { + for _, need := range task.Needs { + if _, exists := taskByID[need]; !exists { + return fmt.Errorf("dag task %q depends on unknown task %q", task.ID, need) + } + } + } + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + ready := make(chan string, len(spec.Tasks)) + results := make(chan dagTaskResult, len(spec.Tasks)) + sem := make(chan struct{}, spec.Parallelism) + var writersMu sync.Mutex + var wg sync.WaitGroup + started := map[string]bool{} + completed := map[string]bool{} + + for id, count := range remainingNeeds { + if count == 0 { + ready <- id + } + } + + failed := false + for len(completed) < len(spec.Tasks) { + select { + case id := <-ready: + if started[id] || failed { + continue + } + started[id] = true + task := taskByID[id] + wg.Add(1) + go func() { + defer wg.Done() + sem <- struct{}{} + defer func() { <-sem }() + results <- runDAGTask(ctx, task, out, errOut, &writersMu) + }() + case result := <-results: + completed[result.id] = true + if result.err != nil || result.exitCode != 0 { + failed = true + cancel() + wg.Wait() + if result.err != nil { + return result.err + } + return fmt.Errorf("dag task %q failed with exit code %d", result.id, result.exitCode) + } + for _, dependent := range dependents[result.id] { + remainingNeeds[dependent]-- + if remainingNeeds[dependent] == 0 { + ready <- dependent + } + } + } + } + wg.Wait() + return nil +} + +func runDAGTask(ctx context.Context, task dagTask, out io.Writer, errOut io.Writer, writersMu *sync.Mutex) dagTaskResult { + title := task.Title + if title == "" { + title = task.ID + } + started := time.Now() + writeDAGLine(out, writersMu, task.ID, "started "+title) + + cmd := exec.CommandContext(ctx, task.Command[0], task.Command[1:]...) + cmd.Env = os.Environ() + stdout, err := cmd.StdoutPipe() + if err != nil { + return dagTaskResult{id: task.ID, exitCode: 1, err: err} + } + cmd.Stderr = cmd.Stdout + if err := cmd.Start(); err != nil { + return dagTaskResult{id: task.ID, exitCode: 1, err: err} + } + + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for scanner.Scan() { + writeDAGLine(out, writersMu, task.ID, scanner.Text()) + } + if scanErr := scanner.Err(); scanErr != nil { + writeDAGLine(errOut, writersMu, task.ID, "output read error: "+scanErr.Error()) + } + + err = cmd.Wait() + elapsed := time.Since(started).Round(time.Second) + exitCode := 0 + if err != nil { + exitCode = 1 + if exitErr, ok := err.(*exec.ExitError); ok { + exitCode = exitErr.ExitCode() + } + writeDAGLine(out, writersMu, task.ID, fmt.Sprintf("failed after %s", elapsed)) + return dagTaskResult{id: task.ID, exitCode: exitCode, err: nil} + } + writeDAGLine(out, writersMu, task.ID, fmt.Sprintf("completed in %s", elapsed)) + return dagTaskResult{id: task.ID, exitCode: 0, err: nil} +} + +func writeDAGLine(out io.Writer, writersMu *sync.Mutex, id string, line string) { + writersMu.Lock() + defer writersMu.Unlock() + fmt.Fprintf(out, "[%s] %s\n", id, line) } func runReportRender(args []string, in io.Reader, out io.Writer) error { diff --git a/cmd/jeannie-core/main_test.go b/cmd/jeannie-core/main_test.go index d0f9de3..5958c74 100644 --- a/cmd/jeannie-core/main_test.go +++ b/cmd/jeannie-core/main_test.go @@ -2,6 +2,9 @@ package main import ( "bytes" + "context" + "os" + "path/filepath" "strings" "testing" ) @@ -42,3 +45,46 @@ func TestReportRenderHidesOKGraphAndQuery(t *testing.T) { t.Fatalf("ok row leaked graph/query links:\n%s", got) } } + +func TestExecuteDAGRespectsDependencies(t *testing.T) { + dir := t.TempDir() + first := filepath.Join(dir, "first") + second := filepath.Join(dir, "second") + spec := dagSpec{ + Parallelism: 2, + Tasks: []dagTask{ + { + ID: "first", + Command: []string{"sh", "-c", "printf first > \"$1\"", "sh", first}, + }, + { + ID: "second", + Needs: []string{"first"}, + Command: []string{"sh", "-c", "test -f \"$1\" && printf second > \"$2\"", "sh", first, second}, + }, + }, + } + + var output bytes.Buffer + if err := executeDAG(context.Background(), spec, &output, &output); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(second); err != nil { + t.Fatalf("dependent task did not run after dependency: %v\n%s", err, output.String()) + } +} + +func TestExecuteDAGReturnsFailedTask(t *testing.T) { + spec := dagSpec{ + Parallelism: 2, + Tasks: []dagTask{ + {ID: "bad", Command: []string{"sh", "-c", "exit 7"}}, + }, + } + + var output bytes.Buffer + err := executeDAG(context.Background(), spec, &output, &output) + if err == nil || !strings.Contains(err.Error(), `dag task "bad" failed`) { + t.Fatalf("expected failed task error, got %v\n%s", err, output.String()) + } +} diff --git a/docs/jeannie.1.md b/docs/jeannie.1.md index cabfff5..b1134ad 100644 --- a/docs/jeannie.1.md +++ b/docs/jeannie.1.md @@ -535,6 +535,10 @@ falls back to the first Lynis warning, then the first suggestion. `LAB_AUTO_APPROVE=false` : Disable automatic OpenTofu approval for apply paths that support confirmation. +`JEANNIE_UP_PARALLELISM` +: Maximum number of independent `up` branches the Go DAG runner may execute at +once. Defaults to `4`; set to `1` for sequential troubleshooting. + `LAB_PIMOX_WORKER_COUNT` : Desired Pimox worker count when not supplied by generated cluster topology state. diff --git a/jeannie b/jeannie index a1e25cd..d55e0d2 100755 --- a/jeannie +++ b/jeannie @@ -4546,22 +4546,111 @@ ensure_cluster_worker_var_file() { fi } +run_up_parallel_stage() { + local spec_file + local parallelism="${JEANNIE_UP_PARALLELISM:-4}" + local status + + if ! [[ "${parallelism}" =~ ^[0-9]+$ ]] || ((parallelism < 1)); then + echo "JEANNIE_UP_PARALLELISM must be a positive integer." >&2 + return 1 + fi + + spec_file="$(mktemp)" + python3 - "${spec_file}" "${REPO_ROOT}/jeannie" "${parallelism}" <<'PY' +import json +import sys + +spec_file, jeannie, parallelism = sys.argv[1:4] +spec = { + "parallelism": int(parallelism), + "tasks": [ + { + "id": "gitea-deploy", + "title": "Gitea deploy", + "command": [jeannie, "__up-task", "gitea-deploy"], + }, + { + "id": "gitea-bootstrap", + "title": "Gitea repo bootstrap", + "needs": ["gitea-deploy"], + "command": [jeannie, "__up-task", "gitea-bootstrap"], + }, + { + "id": "rpi-services", + "title": "RPi services", + "command": [jeannie, "__up-task", "rpi-services"], + }, + { + "id": "pimox-workers", + "title": "Pimox provisioning and workers", + "command": [jeannie, "__up-task", "pimox-workers"], + }, + { + "id": "openwrt", + "title": "OpenWrt VM", + "command": [jeannie, "__up-task", "openwrt"], + }, + { + "id": "edge-host", + "title": "OCI edge host check", + "command": [jeannie, "__up-task", "edge-host"], + }, + ], +} +with open(spec_file, "w", encoding="utf-8") as handle: + json.dump(spec, handle, indent=2) + handle.write("\n") +PY + set +e + "${REPO_ROOT}/scripts/jeannie-core" run-dag --spec "${spec_file}" + status=$? + set -e + rm -f "${spec_file}" + return "${status}" +} + +up_task() { + require_debian_server "__up-task" + + case "${1:-}" in + gitea-deploy) + deploy_gitea + ;; + gitea-bootstrap) + bootstrap_gitea_repo + ;; + rpi-services) + deploy_rpi_services + ;; + pimox-workers) + run_pimox_pipeline + ;; + openwrt) + run_openwrt_pipeline + ;; + edge-host) + check_edge_ssh + ;; + *) + echo "Unknown __up-task '${1:-}'." >&2 + return 1 + ;; + esac +} + up() { require_debian_server "up" echo "Deploying the homelab infrastructure..." jeannie_log_start "up" - jeannie_step_plan 16 + jeannie_step_plan 12 run_step "Early preflight" homelab_preflight early run_step "Go toolchain" ensure_go_toolchain - run_step "Gitea deploy" deploy_gitea - run_step "Gitea repo bootstrap" bootstrap_gitea_repo - run_step "RPi services" deploy_rpi_services + run_step "Independent host services" run_up_parallel_stage run_step "Full preflight" homelab_preflight full run_step "Pre-apply doctor" doctor_preapply - run_step "Pimox provisioning and workers" run_pimox_pipeline - run_step "OpenWrt VM" run_openwrt_pipeline run_step "Worker var file" ensure_cluster_worker_var_file run_step "Start existing cluster if stopped" ensure_existing_cluster_started_for_up run_step "Cluster OpenTofu apply" run_tofu_stack "bootstrap/cluster" @@ -7016,6 +7105,9 @@ if [[ "${JEANNIE_LIBRARY_MODE:-false}" == "true" ]]; then fi case "${1:-}" in + __up-task) + up_task "${2:-}" + ;; "" | help | -h | --help) print_usage ;; diff --git a/scripts/jeannie-core b/scripts/jeannie-core new file mode 100755 index 0000000..3f0ae13 --- /dev/null +++ b/scripts/jeannie-core @@ -0,0 +1,27 @@ +#!/usr/bin/env bash +set -euo pipefail + +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +CORE_BIN="${JEANNIE_CORE_BIN:-${REPO_ROOT}/.lab/bin/jeannie-core}" + +core_sources_newer() { + local source + + while IFS= read -r source; do + if [[ "${source}" -nt "${CORE_BIN}" ]]; then + return 0 + fi + done < <(find "${REPO_ROOT}/cmd/jeannie-core" -name '*.go' -type f) + return 1 +} + +if [[ ! -x "${CORE_BIN}" || "${REPO_ROOT}/go.mod" -nt "${CORE_BIN}" ]] || core_sources_newer; then + if ! command -v go >/dev/null 2>&1; then + echo "go is required to build ${CORE_BIN}; run ./jeannie up or install golang-go." >&2 + exit 1 + fi + mkdir -p "$(dirname "${CORE_BIN}")" + GOCACHE="${GOCACHE:-${REPO_ROOT}/.lab/go-build-cache}" go build -o "${CORE_BIN}" "${REPO_ROOT}/cmd/jeannie-core" +fi + +exec "${CORE_BIN}" "$@" diff --git a/scripts/report-render b/scripts/report-render index de9d167..f959248 100755 --- a/scripts/report-render +++ b/scripts/report-render @@ -2,19 +2,6 @@ set -euo pipefail REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -CORE_BIN="${JEANNIE_CORE_BIN:-${REPO_ROOT}/.lab/bin/jeannie-core}" - -core_sources_newer() { - local source - - while IFS= read -r source; do - if [[ "${source}" -nt "${CORE_BIN}" ]]; then - return 0 - fi - done < <(find "${REPO_ROOT}/cmd/jeannie-core" -name '*.go' -type f) - return 1 -} - usage() { cat <<'EOF' Usage: report-render [--title TITLE] [--details] [--only failures|warnings|problems|all] [--json] @@ -35,13 +22,4 @@ if (($# > 0)); then esac fi -if [[ ! -x "${CORE_BIN}" || "${REPO_ROOT}/go.mod" -nt "${CORE_BIN}" ]] || core_sources_newer; then - if ! command -v go >/dev/null 2>&1; then - echo "go is required to build ${CORE_BIN}; run ./jeannie up or install golang-go." >&2 - exit 1 - fi - mkdir -p "$(dirname "${CORE_BIN}")" - GOCACHE="${GOCACHE:-${REPO_ROOT}/.lab/go-build-cache}" go build -o "${CORE_BIN}" "${REPO_ROOT}/cmd/jeannie-core" -fi - -exec "${CORE_BIN}" report-render "$@" +exec "${REPO_ROOT}/scripts/jeannie-core" report-render "$@"