diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml new file mode 100644 index 0000000..aae0902 --- /dev/null +++ b/.github/workflows/e2e.yml @@ -0,0 +1,61 @@ +name: E2E + +# A real single-node cluster on a GitHub-hosted runner (nested KVM), built from +# this commit and created through the local action. +on: + schedule: + - cron: "17 3 * * *" + workflow_dispatch: + +permissions: + contents: read + +jobs: + cluster: + runs-on: ubuntu-latest + timeout-minutes: 75 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 + with: + go-version-file: go.mod + - run: make build + + - id: proxbase + uses: ./ + with: + binary: bin/proxbase + name: e2e + nodes: "1" + + - name: Health check + run: proxbase status e2e --check + + - name: API call with the exported token + env: + ENDPOINT: ${{ steps.proxbase.outputs.endpoint }} + CA: ${{ steps.proxbase.outputs.ca-cert }} + TOKEN: ${{ steps.proxbase.outputs.api-token }} + ENV_FILE: ${{ steps.proxbase.outputs.env-file }} + run: | + curl -fsS --cacert "$CA" -H "Authorization: PVEAPIToken=$TOKEN" "${ENDPOINT}api2/json/version" | jq . + # the env file works too + # shellcheck source=/dev/null + . "$ENV_FILE" + curl -fsS --cacert "$PROXBASE_CA_CERT" -H "Authorization: PVEAPIToken=$PROXMOX_VE_API_TOKEN" "${PROXMOX_VE_ENDPOINT}api2/json/nodes" | jq -e '.data | length == 1' + + - name: MCP server (status, exec, env, stop, snapshot, restore, start) + env: + PROXBASE_BIN: ${{ github.workspace }}/bin/proxbase + PROXBASE_E2E_CLUSTER: e2e + run: go test -tags e2e -run TestE2E -timeout 30m -v ./internal/mcpserver + + - name: Logs + if: failure() + run: tail -n 100 ~/.local/share/proxbase/clusters/e2e/logs/*.log || true + + - name: Destroy cluster + if: always() + run: proxbase destroy e2e --yes || true diff --git a/.github/workflows/pages.yml b/.github/workflows/pages.yml index f615aa2..dfb35bd 100644 --- a/.github/workflows/pages.yml +++ b/.github/workflows/pages.yml @@ -1,10 +1,11 @@ name: Pages -# Publishes the JSON schema at https://proxbase.virtbase.com/schema/. +# Publishes the JSON schema at https://proxbase.virtbase.com/schema/ and the MCP +# registry entry at https://proxbase.virtbase.com/server.json. on: push: branches: [main] - paths: [schema/**, .github/workflows/pages.yml] + paths: [schema/**, server.json, .github/workflows/pages.yml] workflow_dispatch: permissions: {} @@ -30,6 +31,7 @@ jobs: run: | mkdir -p _site cp -r schema _site/schema + cp server.json _site/server.json echo proxbase.virtbase.com > _site/CNAME printf 'Proxbase\n' > _site/index.html - uses: actions/configure-pages@45bfe0192ca1faeb007ade9deae92b16b8254a0d # v6.0.0 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 34a0a59..f276330 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -70,3 +70,21 @@ jobs: - uses: actions/attest-build-provenance@4d101475d8b20a2381f78447822ac1eab6504dd8 # v4.2.2 with: subject-checksums: dist/checksums.txt + + # Moves the major tag (v0, v1, ...) used by `uses: virtbase/proxbase@v0`. + action-tag: + needs: [release-please, goreleaser] + if: ${{ !cancelled() && needs.goreleaser.result == 'success' }} + runs-on: ubuntu-latest + permissions: + contents: write + env: + RELEASE_TAG: ${{ inputs.tag || needs.release-please.outputs.tag }} + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ env.RELEASE_TAG }} + - run: | + major="${RELEASE_TAG%%.*}" + git tag -f "$major" "$RELEASE_TAG^{commit}" + git push -f origin "refs/tags/$major" diff --git a/.goreleaser.yaml b/.goreleaser.yaml index f7e877f..4cac399 100644 --- a/.goreleaser.yaml +++ b/.goreleaser.yaml @@ -79,6 +79,7 @@ dockers_v2: org.opencontainers.image.version: "{{ .Version }}" org.opencontainers.image.revision: "{{ .FullCommit }}" org.opencontainers.image.licenses: Apache-2.0 + io.modelcontextprotocol.server.name: io.github.virtbase/proxbase # MCP registry ownership, see server.json docker_signs: - cmd: cosign diff --git a/README.md b/README.md index 410b0c4..4c91ddd 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,10 @@ pve3 running online 10.10.10.13 https://127.0.0.1:18003 127.0.0.1:18103 - **Fault injection:** power loss, hung nodes, pulled cables, network partitions, latency and loss, to test HA, fencing and Ceph recovery. - **Ready for automation:** API token and CA export, `proxbase env` for shells, JSON - and the bpg/proxmox Terraform provider, `-o json` for scripts. + and the bpg/proxmox Terraform provider, `-o json` and `--progress json` for scripts, + and a [GitHub Action](docs/ci.md) for CI. +- **For AI agents:** `proxbase mcp` is an [MCP server](docs/mcp.md): agents create + clusters, run commands on nodes, snapshot and inject faults through tools. - **Fast:** `--golden` clones nodes from a cached base image: a 3-node cluster in about 1.5 minutes, with a unique identity per node. - **Declarative:** one YAML file with a JSON schema; every flag overrides a field. @@ -77,6 +80,18 @@ proxbase snapshot restore lab base && proxbase start lab proxbase env lab --format terraform # provider block for bpg/proxmox proxbase fault partition lab pve3 # isolate a node, then: fault clear proxbase status lab --check # exit code for CI and health checks +claude mcp add proxbase -- proxbase mcp # let an AI agent drive proxbase +``` + +In GitHub Actions: + +```yaml +- id: pve + uses: virtbase/proxbase@v0 # KVM, QEMU, cache, create; outputs endpoint, api-token, env-file +- env: {PVE_ENV: "${{ steps.pve.outputs.env-file }}"} + run: . "$PVE_ENV" && go test ./... -tags integration +- if: always() + run: proxbase destroy ci --yes ``` A cluster file looks like this (everything has a default): @@ -103,13 +118,14 @@ storage: | [Configuration](docs/configuration.md) | Every field, default and limit | | [Networking](docs/networking.md) · [Storage](docs/storage.md) · [Lifecycle](docs/lifecycle.md) | How it works and what to expect | | [Fault injection](docs/fault-injection.md) · [Golden images](docs/golden-images.md) | Breaking the cluster on purpose, faster creates | +| [CI and automation](docs/ci.md) · [MCP server](docs/mcp.md) | GitHub Action, JSON progress, AI agents | | [Docker](docs/docker.md) · [CLI reference](docs/cli.md) · [Troubleshooting](docs/troubleshooting.md) | | | [Architecture](docs/architecture.md) | Packages, create flow, state on disk | | [Examples](examples/) | Ready-to-use cluster files for common setups | ## Status and limits -Proxbase is young; the file format may change before v0.1.0. It is tested with +Proxbase is young; the file format may change before v1.0. It is tested with Proxmox VE 9.2 on Debian 13 (QEMU 10.0, AMD). Clusters need a Linux host with KVM: macOS builds are experimental and Docker Desktop has no `/dev/kvm`. Nodes are reachable from the host through forwarded ports on `127.0.0.1`; host-reachable bridge diff --git a/action.yml b/action.yml new file mode 100644 index 0000000..f0f6670 --- /dev/null +++ b/action.yml @@ -0,0 +1,185 @@ +name: Proxbase +description: Create a Proxmox VE cluster in QEMU/KVM on the runner for integration tests +author: virtbase +branding: + icon: server + color: orange + +inputs: + version: + description: proxbase release to install, e.g. v0.2.0, or "latest" + default: latest + binary: + description: Path to a proxbase binary to use instead of downloading a release + default: "" + name: + description: Cluster name + default: ci + config: + description: Cluster file (YAML); the inputs below override it + default: "" + nodes: + description: Number of nodes + default: "1" + memory: + description: Memory per node, e.g. 4G + default: "" + data-disks: + description: Data disks per node, e.g. 2x32G or none + default: "" + storage: + description: Storage on the data disks (zfs, ceph or none) + default: "" + pve-version: + description: Proxmox VE ISO version, e.g. 9.2 + default: "" + golden: + description: Clone nodes from a cached base image (faster with cache) + default: "true" + cache: + description: Cache the ISO and base image with actions/cache + default: "true" + verify-signature: + description: Also verify the release signature with cosign (cosign must be installed) + default: "false" + github-token: + description: Token for the GitHub API (release lookup) + default: ${{ github.token }} + +outputs: + cluster: + description: Cluster name + value: ${{ inputs.name }} + endpoint: + description: API endpoint, e.g. https://127.0.0.1:18001/ + value: ${{ steps.env.outputs.endpoint }} + api-token: + description: API token (user@realm!id=secret) + value: ${{ steps.env.outputs.api-token }} + ca-cert: + description: Path of the cluster CA certificate + value: ${{ steps.env.outputs.ca-cert }} + ssh-key: + description: Path of the root SSH private key + value: ${{ steps.env.outputs.ssh-key }} + env-file: + description: Shell file exporting PROXMOX_VE_* and PROXBASE_* variables (source it) + value: ${{ steps.env.outputs.env-file }} + +runs: + using: composite + steps: + - name: Enable KVM + shell: bash + run: | + echo 'KERNEL=="kvm", GROUP="kvm", MODE="0666", OPTIONS+="static_node=kvm"' | sudo tee /etc/udev/rules.d/99-kvm.rules >/dev/null + sudo udevadm control --reload-rules + sudo udevadm trigger --name-match=kvm + test -w /dev/kvm || { echo "::error::/dev/kvm is not available on this runner"; exit 1; } + + - name: Install QEMU + shell: bash + run: | + if ! command -v qemu-system-x86_64 >/dev/null || ! command -v qemu-img >/dev/null; then + sudo apt-get update -qq + sudo apt-get install -y -qq --no-install-recommends qemu-system-x86 qemu-utils >/dev/null + fi + + - name: Install proxbase + id: install + shell: bash + env: + VERSION: ${{ inputs.version }} + BINARY: ${{ inputs.binary }} + VERIFY: ${{ inputs.verify-signature }} + GH_TOKEN: ${{ inputs.github-token }} + run: | + set -euo pipefail + bindir="$RUNNER_TEMP/proxbase-bin" + mkdir -p "$bindir" + if [ -n "$BINARY" ]; then + install -m 0755 "$BINARY" "$bindir/proxbase" + else + if [ "$VERSION" = latest ]; then + VERSION=$(curl -fsSL -H "Authorization: Bearer $GH_TOKEN" https://api.github.com/repos/virtbase/proxbase/releases/latest | jq -r .tag_name) + fi + arch=$(uname -m); case "$arch" in x86_64) arch=amd64 ;; aarch64) arch=arm64 ;; esac + base="https://github.com/virtbase/proxbase/releases/download/$VERSION" + archive="proxbase_${VERSION#v}_linux_${arch}.tar.gz" + cd "$RUNNER_TEMP" + curl -fsSLO "$base/$archive" + curl -fsSLO "$base/checksums.txt" + sha256sum --ignore-missing -c checksums.txt + if [ "$VERIFY" = true ]; then + curl -fsSLO "$base/checksums.txt.sigstore.json" + cosign verify-blob --bundle checksums.txt.sigstore.json \ + --certificate-identity-regexp '^https://github.com/virtbase/proxbase/\.github/workflows/release\.yml@' \ + --certificate-oidc-issuer https://token.actions.githubusercontent.com checksums.txt + fi + tar -xzf "$archive" -C "$bindir" proxbase + fi + echo "$bindir" >> "$GITHUB_PATH" + "$bindir/proxbase" version + echo "version=$("$bindir/proxbase" version | cut -d' ' -f2)" >> "$GITHUB_OUTPUT" + + - name: Restore ISO and base image cache + if: inputs.cache == 'true' + uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0 + with: + path: ~/.cache/proxbase + key: proxbase-${{ runner.os }}-${{ runner.arch }}-${{ steps.install.outputs.version }}-${{ inputs.pve-version }} + restore-keys: | + proxbase-${{ runner.os }}-${{ runner.arch }}- + + - name: Create cluster + shell: bash + env: + NAME: ${{ inputs.name }} + CONFIG: ${{ inputs.config }} + NODES: ${{ inputs.nodes }} + MEMORY: ${{ inputs.memory }} + DATA_DISKS: ${{ inputs.data-disks }} + STORAGE: ${{ inputs.storage }} + PVE_VERSION: ${{ inputs.pve-version }} + GOLDEN: ${{ inputs.golden }} + run: | + set -euo pipefail + args=(create "$NAME") + [ -n "$CONFIG" ] && args+=(-f "$CONFIG") + [ -n "$NODES" ] && args+=(--nodes "$NODES") + [ -n "$MEMORY" ] && args+=(--memory "$MEMORY") + [ -n "$DATA_DISKS" ] && args+=(--data-disks "$DATA_DISKS") + [ -n "$STORAGE" ] && args+=(--storage "$STORAGE") + [ -n "$PVE_VERSION" ] && args+=(--pve-version "$PVE_VERSION") + [ "$GOLDEN" = true ] && args+=(--golden) + if ! proxbase "${args[@]}"; then + logs="${XDG_DATA_HOME:-$HOME/.local/share}/proxbase/clusters/$NAME/logs" + for f in "$logs"/*.log; do + [ -f "$f" ] || continue + echo "::group::$(basename "$f")" + tail -n 100 "$f" + echo "::endgroup::" + done + exit 1 + fi + + - name: Export credentials + id: env + shell: bash + env: + NAME: ${{ inputs.name }} + run: | + set -euo pipefail + dir="$RUNNER_TEMP/proxbase-$NAME" + proxbase env "$NAME" --export "$dir" + proxbase env "$NAME" --format json > "$dir/env.json" + echo "::add-mask::$(jq -r .tokenSecret "$dir/env.json")" + echo "::add-mask::$(jq -r .rootPassword "$dir/env.json")" + { + echo "endpoint=$(jq -r .endpoint "$dir/env.json")" + echo "api-token=$(jq -r .apiToken "$dir/env.json")" + echo "ca-cert=$dir/pve-root-ca.pem" + echo "ssh-key=$dir/id_ed25519" + echo "env-file=$dir/env.sh" + } >> "$GITHUB_OUTPUT" + rm "$dir/env.json" diff --git a/docs/README.md b/docs/README.md index 2782424..8e18713 100644 --- a/docs/README.md +++ b/docs/README.md @@ -11,6 +11,8 @@ | [Golden images](golden-images.md) | Faster creates from a cached base image | | [Fault injection](fault-injection.md) | Power loss, hangs, pulled cables, partitions, latency and loss | | [Docker](docker.md) | Running clusters in a container with Docker Compose | +| [CI and automation](ci.md) | GitHub Action, other CI systems, JSON progress | +| [MCP server](mcp.md) | Letting AI agents create, use and break clusters | | [CLI reference](cli.md) | All commands and flags | | [Troubleshooting](troubleshooting.md) | Logs, common errors and how to resume | | [Architecture](architecture.md) | How Proxbase is built, with diagrams | diff --git a/docs/architecture.md b/docs/architecture.md index 729d2d2..9f1d2cc 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -12,6 +12,11 @@ flowchart TD cli["cli
cobra commands"] --> cluster cli --> config cli --> network + cli --> mcpserver["mcpserver
MCP tools"] + mcpserver --> cluster + mcpserver --> job["job
detached creates"] + cluster --> progress["progress
events, text/JSON sinks"] + job --> progress cluster["cluster
orchestration"] --> vm cluster --> network cluster --> storage @@ -51,7 +56,10 @@ flowchart TD | `remote` | SSH to nodes through the forwarded ports | | `retry` | Retry with timeout, interval and permanent errors | | `golden` | Cached base images and the personalization of cloned nodes | -| `cluster` | Orchestration: create, lifecycle, nodes, snapshots, faults, status | +| `cluster` | Orchestration: create, lifecycle, nodes, snapshots, faults, status, exec | +| `progress` | Progress events and sinks: text, JSON lines, MCP notifications | +| `job` | Runs a create detached from its caller, with a JSON-lines progress log | +| `mcpserver` | MCP server (`proxbase mcp`): tools, the schema resource, progress forwarding | | `cli` | Commands, flags and output | Extension points are small interfaces: a new hypervisor implements `vm.Runtime` (and @@ -89,6 +97,33 @@ sequenceDiagram Each step checks first and skips what is done, so a failed create resumes when run again. Progress (`installed` per node, phase, errors) is kept in `state.json`. +## Progress and the MCP server + +Long operations report through a `progress.Sink` (a function taking an `Event`). +The CLI passes a text or JSON-lines sink for stderr (`--progress`); the MCP server +passes one that also forwards events as MCP progress notifications. + +```mermaid +sequenceDiagram + participant Agent as MCP client + participant MCP as proxbase mcp + participant Job as proxbase create --progress json (detached) + participant Log as job log (jobs/NAME.jsonl) + Agent->>MCP: tools/call cluster_create (progressToken) + MCP->>Job: start in its own session + loop every second until done or timeoutSeconds + Job->>Log: one event per line + MCP->>Log: read new lines + MCP-->>Agent: notifications/progress + end + MCP-->>Agent: result (status, or running=true) + Agent->>MCP: cluster_wait (if still running) +``` + +The create runs as a separate process so it survives client timeouts and restarts +of the MCP server; everything else (`node_exec`, `cluster_start`, snapshots, faults) +runs in the server process, guarded by the same cluster lock as the CLI. + ## Nodes - QEMU/KVM with q35, SeaBIOS and fixed PCI addresses for every device, identical @@ -118,6 +153,7 @@ again. Progress (`installed` per node, phase, errors) is kept in `state.json`. ├── disks/ -root.qcow2, -dataN.qcow2 ├── run/ pidfiles, QMP/console sockets, switch and NIC sockets └── logs/ install, console, QEMU and switch logs +~/.local/share/proxbase/jobs/ .jsonl, .pid, .yaml of creates started over MCP ~/.cache/proxbase/iso/ ISOs, SHA256SUMS, extracted installer files ~/.cache/proxbase/images/ golden base images (create --golden) ``` diff --git a/docs/ci.md b/docs/ci.md new file mode 100644 index 0000000..12431ca --- /dev/null +++ b/docs/ci.md @@ -0,0 +1,84 @@ +# CI and automation + +GitHub-hosted Linux runners expose `/dev/kvm`, so they can run a real cluster. A +single node is ready in 2-3 minutes with a cached base image. + +## GitHub Action + +```yaml +jobs: + integration: + runs-on: ubuntu-latest + timeout-minutes: 45 + steps: + - uses: actions/checkout@v7 + - id: pve + uses: virtbase/proxbase@v0 # or pin a commit SHA + with: + nodes: "1" + - env: + PVE_ENV: ${{ steps.pve.outputs.env-file }} + run: | + . "$PVE_ENV" + go test ./... -tags integration + - if: always() + run: proxbase destroy ci --yes +``` + +The action enables KVM for the runner user, installs QEMU, downloads the release +binary and checks it against `checksums.txt` (and the cosign signature with +`verify-signature: "true"`), restores the ISO and base image from `actions/cache`, +and creates the cluster with `--golden`. On failure it prints the node logs. + +| Input | Default | | +| --- | --- | --- | +| `version` | `latest` | Release tag, e.g. `v0.2.0` | +| `binary` | | Use this binary instead of downloading one | +| `name` | `ci` | Cluster name | +| `config` | | Cluster file; the inputs below override it | +| `nodes`, `memory`, `data-disks`, `storage`, `pve-version` | `1`, file defaults | As the `create` flags | +| `golden` | `true` | Clone from a cached base image | +| `cache` | `true` | Cache `~/.cache/proxbase` | +| `verify-signature` | `false` | Verify with cosign (install it first) | + +| Output | | +| --- | --- | +| `endpoint` | `https://127.0.0.1:18001/` | +| `api-token` | `proxbase@pve!api=…` (masked in logs) | +| `ca-cert`, `ssh-key` | Paths of the cluster CA and the root key | +| `env-file` | Shell file with `PROXMOX_VE_*` and `PROXBASE_*`; source it | + +Composite actions have no post step, so destroy the cluster with an `if: always()` +step. A complete workflow is in [examples/ci/github-actions.yml](../examples/ci/github-actions.yml); +the repository's own [E2E workflow](../.github/workflows/e2e.yml) uses the action +from the checkout with a freshly built binary. + +## Other CI systems + +Any Linux runner with KVM works: install QEMU and the binary, then + +```bash +proxbase create ci --nodes 1 --golden --progress json # JSON lines on stderr +proxbase status ci --check # exit code for health checks +eval "$(proxbase env ci)" +proxbase destroy ci --yes +``` + +Cache `~/.cache/proxbase` between runs to skip the ISO download and the base image +build. + +## Machine-readable output + +- `-o json` on `create`, `status` and `list` prints JSON on stdout. +- `--progress json` (any command) writes progress and messages to stderr as one + JSON object per line: + + ```json + {"time":"…","type":"step","cluster":"ci","node":"pve1","message":"pve1 installed in 1m51s","elapsed":111.5} + {"time":"…","type":"error","error":"pve1: SSH not reachable: …","hint":"Re-run `proxbase create ci` to resume, or `proxbase destroy ci` to start over","log":"…/clusters/ci/logs"} + ``` + + `type` is `step`, `log`, `done` (last line on success) or `error` (last line on + failure, with `hint` and `log` where known). +- The exit code is 0 on success and 1 on any error. +- For AI agents, [`proxbase mcp`](mcp.md) offers the same operations as MCP tools. diff --git a/docs/cli.md b/docs/cli.md index 2c0dcb9..f2f83c4 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -3,6 +3,10 @@ Commands take the cluster name as first argument. Without it they use `default`, or the only existing cluster. `proxbase --help` shows the same information. +Every command accepts `--progress text|json`: with `json`, progress and messages on +stderr are JSON lines, ending with a `done` or `error` event (see +[CI and automation](ci.md#machine-readable-output)). Results (`-o json`) stay on stdout. + ## Clusters | Command | Description | @@ -66,6 +70,7 @@ See [fault-injection.md](fault-injection.md). | `ssh [cluster] [node] [-- command…]` | SSH as root (default: first node) | | `console [cluster] ` | Serial console, Ctrl-] to detach | | `env [cluster]` | API endpoint and credentials: `--format shell\|json\|terraform`, `--export ` | +| `mcp` | MCP server on stdio for AI agents ([mcp.md](mcp.md)) | ## Host and files @@ -84,6 +89,6 @@ See [fault-injection.md](fault-injection.md). | Variable | Effect | | --- | --- | -| `XDG_DATA_HOME` | Cluster state in `$XDG_DATA_HOME/proxbase/clusters` (default `~/.local/share`) | +| `XDG_DATA_HOME` | Cluster state in `$XDG_DATA_HOME/proxbase/clusters`, MCP create jobs in `…/proxbase/jobs` (default `~/.local/share`) | | `XDG_CACHE_HOME` | ISO cache in `$XDG_CACHE_HOME/proxbase/iso` (default `~/.cache`) | | `PROXBASE_BIND_ADDRESS` | Default for `access.bindAddress` of new clusters | diff --git a/docs/docker.md b/docs/docker.md index 246a063..c588191 100644 --- a/docs/docker.md +++ b/docs/docker.md @@ -49,6 +49,9 @@ Limits: - A 3-node ZFS cluster needs about 12 GiB RAM and 15 GiB in the volume; Ceph about 18 GiB RAM. - The image is built for linux/amd64. +The image also runs the MCP server for AI agents (`… ghcr.io/virtbase/proxbase mcp`); +see [mcp.md](mcp.md#setup) for the client configuration. + Building the image yourself: `docker build -t proxbase .` (source build) or the release image `ghcr.io/virtbase/proxbase`. On networks where the Debian mirror is unreliable over IPv6, pass `--build-arg APT_OPTS="-o Acquire::ForceIPv4=true"`. diff --git a/docs/mcp.md b/docs/mcp.md new file mode 100644 index 0000000..6997b5d --- /dev/null +++ b/docs/mcp.md @@ -0,0 +1,133 @@ +# MCP server for AI agents + +`proxbase mcp` is a [Model Context Protocol](https://modelcontextprotocol.io) server +built into the binary. An agent client (Claude Code, Cursor, VS Code, …) starts it +as a subprocess and talks to it over stdin/stdout. It needs no hosting: it runs on +the machine with KVM where the clusters run. Clusters it creates are ordinary +proxbase clusters, so the CLI sees them too and vice versa. + +## Setup + +Claude Code: + +```bash +claude mcp add proxbase -- proxbase mcp +``` + +Cursor (`.cursor/mcp.json`) and other clients with the common format: + +```json +{ + "mcpServers": { + "proxbase": { "command": "proxbase", "args": ["mcp"] } + } +} +``` + +VS Code (`.vscode/mcp.json`): + +```json +{ + "servers": { + "proxbase": { "type": "stdio", "command": "proxbase", "args": ["mcp"] } + } +} +``` + +With the container image instead of a local binary (host network so the agent and +your tools reach the forwarded ports; `--group-add` is the GID of `/dev/kvm`): + +```json +{ + "mcpServers": { + "proxbase": { + "command": "docker", + "args": ["run", "-i", "--rm", "--device", "/dev/kvm", "--group-add", "993", + "--network", "host", "-v", "proxbase:/data", + "ghcr.io/virtbase/proxbase:latest", "mcp"] + } + } +} +``` + +Clusters created in the container live in the `proxbase` volume; use +`docker run … proxbase list` to see them from the CLI. + +## Tools + +| Tool | Does | +| --- | --- | +| `cluster_list` | All clusters with phase and running nodes | +| `cluster_status` | Phase, quorum, nodes, Ceph health, faults; `healthy` and `problem` | +| `cluster_create` | Create (or resume) a cluster from parameters or a cluster file; reports progress | +| `cluster_wait` | Keep waiting for a create that is still running | +| `cluster_start`, `cluster_stop` | Boot or shut down all nodes | +| `cluster_destroy` | Delete a cluster; `confirm` must repeat the cluster name | +| `node_exec` | Run a command as root on a node; returns `stdout`, `stderr`, `exitCode` | +| `cluster_env` | API endpoint, token, CA certificate, SSH key and ports; root password only with `includeRootPassword` | +| `snapshot_save`, `snapshot_restore`, `snapshot_delete`, `snapshot_list` | Snapshots of the stopped cluster | +| `fault_apply`, `fault_clear` | kill, freeze, thaw, link-down/up, partition, degrade | +| `config_validate` | Check a cluster file; returns errors, warnings and the file with defaults | + +The resource `https://proxbase.virtbase.com/schema/v1alpha1/cluster.json` is the JSON +schema of cluster files. Tool annotations mark read-only and destructive tools, so +clients can ask before `cluster_destroy`, `snapshot_restore` or `fault_apply`. + +## Long-running creates + +A create takes 2-3 minutes with a cached [golden image](golden-images.md) (the +default for `cluster_create`) and 10-20 minutes the first time, when it downloads the +ISO and builds the base image. Many clients time out tool calls earlier, so: + +- `cluster_create` starts the create as a background job on the host + (`proxbase create --progress json`), follows it, and sends each step as an MCP + progress notification if the client passed a progress token. +- It returns when the cluster is ready or failed, or after `timeoutSeconds` + (default 600) with `running: true`; then `cluster_wait` continues. +- The job survives the MCP server: a restarted client can call `cluster_wait` or + `cluster_status` (which shows the latest steps while it runs). +- Cancelling `cluster_create` stops the create (SIGTERM); calling it again resumes. + Cancelling `cluster_wait` only stops waiting. + +The job log is `~/.local/share/proxbase/jobs/.jsonl` (same events as +`--progress json`); node install logs stay in the cluster's `logs/` directory. + +## Example session (abridged) + +```text +agent → cluster_create {"name": "lab", "nodes": 3, "storage": "ceph"} + ← progress: [ 12s] installing Proxmox VE 9.2-1 on pve1, pve2, pve3 … + ← {"running": false, "status": {"phase": "ready", "quorate": true, …}} +agent → node_exec {"cluster": "lab", "node": "pve2", "command": "ceph -s"} + ← {"stdout": " cluster:\n id: …\n health: HEALTH_OK …", "exitCode": 0} +agent → fault_apply {"cluster": "lab", "kind": "partition", "groups": [["pve3"]]} +agent → cluster_status {"cluster": "lab"} + ← {"healthy": false, "problem": "pve3 not running or offline", …} +agent → fault_clear {"cluster": "lab"} +agent → cluster_destroy {"cluster": "lab", "confirm": "lab"} +``` + +## Security + +The server runs with the rights of the user who starts it and can create VMs, +run commands as root inside them and delete clusters. It only talks to its client +over stdio and opens no network port. `cluster_env` returns the API token (full +privileges on the lab cluster) and, on request, the root password; treat the +conversation accordingly. Progress and errors are logged to stderr. + +## MCP registry + +[`server.json`](../server.json) describes the server for the +[MCP registry](https://registry.modelcontextprotocol.io) as the container image +(`io.github.virtbase/proxbase`); the image carries the matching +`io.modelcontextprotocol.server.name` label and Pages publishes the file at +`https://proxbase.virtbase.com/server.json`. Publishing is manual after a release: + +```bash +# set "version" and the image tag in server.json to the release, then: +mcp-publisher login github # as a member of the virtbase organization +mcp-publisher publish +``` + +`mcp-publisher` is in the +[registry releases](https://github.com/modelcontextprotocol/registry/releases). diff --git a/docs/releasing.md b/docs/releasing.md index 09f07b0..8de89cb 100644 --- a/docs/releasing.md +++ b/docs/releasing.md @@ -11,7 +11,10 @@ Releases are automatic; nobody tags by hand. release. The same `Release` workflow then runs GoReleaser, which uploads archives, deb/rpm/apk packages, SBOMs, `checksums.txt` with its cosign bundle, pushes the amd64 image to GHCR (signed), updates the Homebrew tap and attests build - provenance. + provenance. Afterwards the major tag (`v0`) is moved to the release for + `uses: virtbase/proxbase@v0`. +4. Optional, manual: update `version` and the image tag in `server.json` and publish + it to the MCP registry (see [mcp.md](mcp.md#mcp-registry)). Local dry run (no signing, no publishing): @@ -24,4 +27,5 @@ The image is built for linux/amd64 only; arm64 binaries and packages are release Dependency updates come from Renovate (weekly, grouped, actions pinned to commit SHAs). Security checks: CodeQL, govulncheck, gitleaks, Trivy (repository and image) and -OpenSSF Scorecard run on pushes, pull requests and weekly. +OpenSSF Scorecard run on pushes, pull requests and weekly. The `E2E` workflow creates +a real cluster through the action nightly and on demand. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index bc3c7f1..865e61f 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -14,7 +14,8 @@ Everything of a cluster lives in `~/.local/share/proxbase/clusters//`: | `secrets/` | Root password, SSH key, API token, CA (mode 0600) | `proxbase status ` shows the last error of a failed create; re-running -`proxbase create ` resumes it. +`proxbase create ` resumes it. A create started through the +[MCP server](mcp.md) logs its progress to `~/.local/share/proxbase/jobs/.jsonl`. ## Common problems @@ -28,7 +29,8 @@ Everything of a cluster lives in `~/.local/share/proxbase/clusters//`: | `Hash Sum mismatch` during apt (nodes or image build) | A broken mirror or path (seen over IPv6). Retry; for image builds pass `--build-arg APT_OPTS="-o Acquire::ForceIPv4=true"` | | `ceph not ready after 10m0s` | The error includes `ceph -s` and `ceph health detail`; usually too little memory per node | | Nested guests are slow or fail to start | Nested virtualization is off on the host (`proxbase doctor`) | -| `cluster "x" is busy` | Another proxbase command holds the cluster lock | +| `cluster "x" is busy` | Another proxbase command holds the cluster lock, e.g. a create started over MCP (`cluster_status` shows it) | +| `a job for cluster x is already running` | An MCP create of that cluster is still running; use `cluster_wait`, or `cluster_destroy` to stop it | ## Starting over diff --git a/examples/README.md b/examples/README.md index 52c3841..5bdf44c 100644 --- a/examples/README.md +++ b/examples/README.md @@ -8,7 +8,7 @@ | [networks.yaml](networks.yaml) | Separate management, cluster, storage, replication and guest (VLAN) networks | | [terraform/](terraform) | Test Terraform/OpenTofu code with the bpg/proxmox provider | | [ansible/](ansible) | Dynamic inventory and a playbook against the nodes | -| [ci/github-actions.yml](ci/github-actions.yml) | Integration tests against a real cluster in GitHub Actions | +| [ci/github-actions.yml](ci/github-actions.yml) | Integration tests against a real cluster in GitHub Actions, with the proxbase action | | [compose/](compose) | Run a cluster with Docker Compose | All cluster files are checked by `go test ./internal/config`. Run one with diff --git a/examples/ci/github-actions.yml b/examples/ci/github-actions.yml index ca6b57c..b3efba7 100644 --- a/examples/ci/github-actions.yml +++ b/examples/ci/github-actions.yml @@ -1,5 +1,6 @@ # Example workflow: run tests against a real single-node Proxmox VE cluster on a # GitHub-hosted Linux runner (they expose /dev/kvm). Copy to .github/workflows/. +# Pin the action to a release tag or, better, its commit SHA. name: Proxmox integration tests on: [pull_request] @@ -10,29 +11,30 @@ permissions: jobs: integration: runs-on: ubuntu-latest - timeout-minutes: 30 + timeout-minutes: 45 steps: - uses: actions/checkout@v7 - - name: Enable KVM for the runner user - run: | - echo 'KERNEL=="kvm", GROUP="kvm", MODE="0666", OPTIONS+="static_node=kvm"' | sudo tee /etc/udev/rules.d/99-kvm.rules - sudo udevadm control --reload-rules && sudo udevadm trigger --name-match=kvm - - name: Install QEMU and proxbase - run: | - sudo apt-get update && sudo apt-get install -y --no-install-recommends qemu-system-x86 qemu-utils - go install github.com/virtbase/proxbase/cmd/proxbase@latest - echo "$(go env GOPATH)/bin" >> "$GITHUB_PATH" - - name: Create cluster - run: proxbase create ci --nodes 1 --memory 4G --data-disks none + + # Installs QEMU and proxbase, enables KVM, caches the ISO and base image, + # and creates the cluster. Outputs: endpoint, api-token, ca-cert, ssh-key, env-file. + - id: pve + uses: virtbase/proxbase@v0 + with: + name: ci + nodes: "1" + # config: test/cluster.yaml # or a cluster file + - name: Run tests + env: + PVE_ENV: ${{ steps.pve.outputs.env-file }} run: | - eval "$(proxbase env ci)" + . "$PVE_ENV" # PROXMOX_VE_ENDPOINT, PROXMOX_VE_API_TOKEN, PROXBASE_CA_CERT, ... curl -sf --cacert "$PROXBASE_CA_CERT" -H "Authorization: PVEAPIToken=$PROXMOX_VE_API_TOKEN" \ "${PROXMOX_VE_ENDPOINT}api2/json/version" # your tests here, e.g. go test ./... -tags integration - - name: Logs on failure - if: failure() - run: tail -n 50 ~/.local/share/proxbase/clusters/ci/logs/*.log + proxbase ssh ci -- pvesh get /cluster/resources --output-format json + + # Composite actions have no post step: destroy the cluster explicitly. - name: Destroy cluster if: always() run: proxbase destroy ci --yes diff --git a/go.mod b/go.mod index d0dc954..b074353 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.27 require ( github.com/invopop/jsonschema v0.14.0 github.com/kdomanski/iso9660 v0.4.0 + github.com/modelcontextprotocol/go-sdk v1.8.0 github.com/spf13/cobra v1.10.2 golang.org/x/crypto v0.57.0 golang.org/x/sync v0.23.0 @@ -15,9 +16,15 @@ require ( require ( github.com/bahlo/generic-list-go v0.2.0 // indirect github.com/buger/jsonparser v1.1.2 // indirect + github.com/google/jsonschema-go v0.4.3 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/pb33f/ordered-map/v2 v2.3.1 // indirect + github.com/segmentio/asm v1.1.3 // indirect + github.com/segmentio/encoding v0.5.4 // indirect github.com/spf13/pflag v1.0.9 // indirect + github.com/yosida95/uritemplate/v3 v3.0.2 // indirect go.yaml.in/yaml/v4 v4.0.0-rc.2 // indirect + golang.org/x/oauth2 v0.35.0 // indirect golang.org/x/sys v0.48.0 // indirect + golang.org/x/time v0.15.0 // indirect ) diff --git a/go.sum b/go.sum index a0f997b..09686fc 100644 --- a/go.sum +++ b/go.sum @@ -5,34 +5,54 @@ github.com/buger/jsonparser v1.1.2/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx2 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/jsonschema-go v0.4.3 h1:/DBOLZTfDow7pe2GmaJNhltueGTtDKICi8V8p+DQPd0= +github.com/google/jsonschema-go v0.4.3/go.mod h1:r5quNTdLOYEz95Ru18zA0ydNbBuYoo9tgaYcxEYhJVE= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/invopop/jsonschema v0.14.0 h1:MHQqLhvpNUZfw+hM3AZDYK7jxO8FZoQeQM77g8iyZjg= github.com/invopop/jsonschema v0.14.0/go.mod h1:ygm6C2EaVNMBDPpaPlnOA2pFAxBnxGjFlMZABxm9n2I= github.com/kdomanski/iso9660 v0.4.0 h1:BPKKdcINz3m0MdjIMwS0wx1nofsOjxOq8TOr45WGHFg= github.com/kdomanski/iso9660 v0.4.0/go.mod h1:OxUSupHsO9ceI8lBLPJKWBTphLemjrCQY8LPXM7qSzU= +github.com/modelcontextprotocol/go-sdk v1.8.0 h1:KIvahhYqwtbeniWVPs3TcXEA7b8jEtwfBpOTAI+Urx4= +github.com/modelcontextprotocol/go-sdk v1.8.0/go.mod h1:dL7u98E/zjJTGzEq+j30jQ8K2k1mb6LeAH4inEcSGts= github.com/pb33f/ordered-map/v2 v2.3.1 h1:5319HDO0aw4DA4gzi+zv4FXU9UlSs3xGZ40wcP1nBjY= github.com/pb33f/ordered-map/v2 v2.3.1/go.mod h1:qxFQgd0PkVUtOMCkTapqotNgzRhMPL7VvaHKbd1HnmQ= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc= +github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg= +github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfvNt0= +github.com/segmentio/encoding v0.5.4/go.mod h1:HS1ZKa3kSN32ZHVZ7ZLPLXWvOVIiZtyJnO1gPH1sKt0= github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4= github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY= github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= +github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= go.yaml.in/yaml/v4 v4.0.0-rc.2 h1:/FrI8D64VSr4HtGIlUtlFMGsm7H7pWTbj6vOLVZcA6s= go.yaml.in/yaml/v4 v4.0.0-rc.2/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0= golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/oauth2 v0.35.0 h1:Mv2mzuHuZuY2+bkyWXIHMfhNdJAdwW3FuWeCPYN5GVQ= +golang.org/x/oauth2 v0.35.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE= golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= +golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/internal/cli/create.go b/internal/cli/create.go index 52ddc38..7547850 100644 --- a/internal/cli/create.go +++ b/internal/cli/create.go @@ -3,9 +3,6 @@ package cli import ( "fmt" "os" - "regexp" - "strconv" - "strings" "github.com/spf13/cobra" @@ -41,12 +38,9 @@ func createCmd() *cobra.Command { _, err = os.Stdout.Write(b) return err } - c, err := cluster.Create(cmd.Context(), cfg, logf) + c, err := cluster.Create(cmd.Context(), cfg, sink) if err != nil { - if c != nil { - return fmt.Errorf("%w\n\nRe-run `proxbase create %s` to resume, or `proxbase destroy %s` to start over", err, cfg.Name, cfg.Name) - } - return err + return createError(c, err) } logf("created in %s", c.St.Duration) return printStatus(c.Status(cmd.Context()), o.output) @@ -58,6 +52,16 @@ func createCmd() *cobra.Command { return cmd } +// createError tells how to go on after a failed create of an initialized cluster. +func createError(c *cluster.Cluster, err error) error { + if c == nil { + return err + } + name := c.Cfg.Name + return &hintError{err: err, log: c.Dir.Log(""), + hint: fmt.Sprintf("Re-run `proxbase create %s` to resume, or `proxbase destroy %s` to start over", name, name)} +} + // addFlags registers the flags that override fields of the cluster file. func (o *createOpts) addFlags(cmd *cobra.Command) { f := cmd.Flags() @@ -73,8 +77,6 @@ func (o *createOpts) addFlags(cmd *cobra.Command) { f.StringVar(&o.bind, "bind-address", "", "address for the UI/SSH forwards (default $PROXBASE_BIND_ADDRESS or 127.0.0.1)") } -var multiDiskRe = regexp.MustCompile(`^([0-9]+)x([0-9]+[MGT])$`) - func (o *createOpts) resolve(cmd *cobra.Command, args []string) (*config.Cluster, error) { cfg := &config.Cluster{} if o.file != "" { @@ -83,86 +85,36 @@ func (o *createOpts) resolve(cmd *cobra.Command, args []string) (*config.Cluster return nil, fmt.Errorf("%s: %w", o.file, err) } } + f := cmd.Flags() + ov := config.Overrides{Golden: o.golden} if len(args) > 0 { - cfg.Name = args[0] + ov.Name = args[0] + } + set := func(flag string, dst **string, v *string) { + if f.Changed(flag) { + *dst = v + } } - f := cmd.Flags() - d := &cfg.Nodes.Defaults if f.Changed("nodes") { - cfg.Nodes.Count = o.nodes + ov.Nodes = &o.nodes } if f.Changed("cpus") { - d.CPUs = o.cpus - } - if f.Changed("memory") { - d.Memory = o.memory + ov.CPUs = &o.cpus } - if f.Changed("disk") { - d.RootDisk.Size = o.disk + set("memory", &ov.Memory, &o.memory) + set("disk", &ov.Disk, &o.disk) + set("data-disks", &ov.DataDisks, &o.dataDisks) + set("storage", &ov.Storage, &o.storage) + set("pve-version", &ov.Version, &o.version) + set("bind-address", &ov.BindAddress, &o.bind) + if env := os.Getenv("PROXBASE_BIND_ADDRESS"); ov.BindAddress == nil && cfg.Access.BindAddress == "" && env != "" { + ov.BindAddress = &env } - if f.Changed("pve-version") { - cfg.Proxmox.Version = o.version - } - if o.golden { - cfg.Proxmox.Golden = true - } - if f.Changed("bind-address") { - cfg.Access.BindAddress = o.bind - } else if env := os.Getenv("PROXBASE_BIND_ADDRESS"); cfg.Access.BindAddress == "" && env != "" { - cfg.Access.BindAddress = env - } - if f.Changed("data-disks") { - disks, err := parseDisks(o.dataDisks) - if err != nil { - return nil, err - } - d.DataDisks = disks - cfg.Storage.ZFS = nil // re-derive the default pool from the new disks - } - if f.Changed("storage") { - switch o.storage { - case "zfs": - if len(d.DataDisks) == 0 && d.DataDisks != nil { - return nil, fmt.Errorf("--storage zfs needs data disks") - } - cfg.Storage.Ceph = nil - case "ceph": - cfg.Storage.ZFS = []config.ZFSPool{} - if cfg.Storage.Ceph == nil { - cfg.Storage.Ceph = &config.Ceph{} - } - cfg.Storage.Ceph.Enabled = true - case "none": - cfg.Storage.ZFS = []config.ZFSPool{} - cfg.Storage.Ceph = nil - default: - return nil, fmt.Errorf("--storage must be zfs, ceph or none") - } - } - cfg.SetDefaults() - if err := cfg.Validate(); err != nil { - return nil, fmt.Errorf("invalid cluster configuration:\n%w", err) + if err := ov.Apply(cfg); err != nil { + return nil, err } for _, w := range cfg.Warnings() { logf("warning: %s", w) } return cfg, nil } - -func parseDisks(s string) ([]config.Disk, error) { - disks := []config.Disk{} - if s == "none" || s == "0" || s == "" { - return disks, nil - } - if m := multiDiskRe.FindStringSubmatch(s); m != nil { - n, _ := strconv.Atoi(m[1]) - for i := 0; i < n; i++ { - disks = append(disks, config.Disk{Size: m[2]}) - } - return disks, nil - } - for _, size := range strings.Split(s, ",") { - disks = append(disks, config.Disk{Size: strings.TrimSpace(size)}) - } - return disks, nil -} diff --git a/internal/cli/env.go b/internal/cli/env.go index acb3d10..157e79d 100644 --- a/internal/cli/env.go +++ b/internal/cli/env.go @@ -4,7 +4,6 @@ import ( "fmt" "os" "path/filepath" - "strings" "github.com/spf13/cobra" @@ -22,12 +21,11 @@ func envCmd() *cobra.Command { if err != nil { return err } - token, err := c.Token() + env, err := c.Env() if err != nil { - return fmt.Errorf("cluster %s has no API token yet (create not finished or access.apiToken: false)", c.Cfg.Name) + return err } - first := c.St.Nodes[0] - endpoint := fmt.Sprintf("https://127.0.0.1:%d/", first.UIPort) + endpoint, token := env.Endpoint, env.APIToken shell := func(key, ca, prefix string) string { return prefix + fmt.Sprintf("export PROXMOX_VE_ENDPOINT=%q\n", endpoint) + fmt.Sprintf("export PROXMOX_VE_API_TOKEN='%s'\n", token) + @@ -47,28 +45,8 @@ func envCmd() *cobra.Command { fmt.Print(shell(fmt.Sprintf("%q", c.KeyPath()), fmt.Sprintf("%q", c.CAPath()), "")) return nil case "json": - type node struct { - Name string `json:"name"` - API string `json:"api"` - SSHHost string `json:"sshHost"` - SSHPort int `json:"sshPort"` - } - out := struct { - Endpoint string `json:"endpoint"` - APIToken string `json:"apiToken"` - TokenID string `json:"tokenId"` - TokenSecret string `json:"tokenSecret"` - CACert string `json:"caCert"` - SSHUser string `json:"sshUser"` - SSHKey string `json:"sshKey"` - RootPassword string `json:"rootPassword"` - Nodes []node `json:"nodes"` - }{Endpoint: endpoint, APIToken: token, CACert: c.CAPath(), SSHUser: "root", SSHKey: c.KeyPath(), RootPassword: c.Password()} - out.TokenID, out.TokenSecret, _ = strings.Cut(token, "=") - for _, n := range c.St.Nodes { - out.Nodes = append(out.Nodes, node{Name: n.Name, API: fmt.Sprintf("https://127.0.0.1:%d/", n.UIPort), SSHHost: "127.0.0.1", SSHPort: n.SSHPort}) - } - return printJSON(out) + env.RootPassword = c.Password() + return printJSON(env) case "terraform": fmt.Printf(`provider "proxmox" { endpoint = %q diff --git a/internal/cli/lifecycle.go b/internal/cli/lifecycle.go index 2c269c5..c93df4f 100644 --- a/internal/cli/lifecycle.go +++ b/internal/cli/lifecycle.go @@ -12,6 +12,7 @@ import ( "golang.org/x/term" "github.com/virtbase/proxbase/internal/cluster" + "github.com/virtbase/proxbase/internal/job" "github.com/virtbase/proxbase/internal/state" ) @@ -72,7 +73,7 @@ func listCmd() *cobra.Command { } rows := []row{} for _, name := range names { - c, err := cluster.Open(name, logf) + c, err := cluster.Open(name, sink) if err != nil { rows = append(rows, row{Name: name, Phase: "broken: " + err.Error()}) continue @@ -101,25 +102,6 @@ func listCmd() *cobra.Command { return cmd } -// healthy explains why a cluster is not fully healthy ("" if it is). -func healthy(s *cluster.Status) string { - switch { - case s.Phase != "ready": - return "phase " + s.Phase - case s.Quorate == nil || !*s.Quorate: - return "not quorate" - } - for _, n := range s.Nodes { - if !n.Running || n.Online == nil || !*n.Online { - return n.Name + " not running or offline" - } - } - if s.Ceph != "" && s.Ceph != "HEALTH_OK" { - return "ceph " + s.Ceph - } - return "" -} - func statusCmd() *cobra.Command { var output string var check bool @@ -134,7 +116,7 @@ func statusCmd() *cobra.Command { } s := c.Status(cmd.Context()) if check { - if why := healthy(s); why != "" { + if why := s.Problem(); why != "" { return fmt.Errorf("cluster %s is not healthy: %s", s.Name, why) } fmt.Printf("cluster %s is healthy\n", s.Name) @@ -202,9 +184,12 @@ func destroyCmd() *cobra.Command { return fmt.Errorf("aborted") } } - if err := cluster.Destroy(name, logf); err != nil { + if err := cluster.Destroy(name, sink); err != nil { return err } + if j, err := job.New(name); err == nil { + j.Remove() // of a create started over MCP + } logf("cluster %s destroyed", name) return nil }, diff --git a/internal/cli/lifecycle_test.go b/internal/cli/lifecycle_test.go deleted file mode 100644 index 77076d7..0000000 --- a/internal/cli/lifecycle_test.go +++ /dev/null @@ -1,31 +0,0 @@ -package cli - -import ( - "testing" - - "github.com/virtbase/proxbase/internal/cluster" -) - -func TestHealthy(t *testing.T) { - yes, no := true, false - ok := func() *cluster.Status { - return &cluster.Status{Phase: "ready", Quorate: &yes, Ceph: "HEALTH_OK", - Nodes: []cluster.NodeStatus{{Name: "pve1", Running: true, Online: &yes}}} - } - if why := healthy(ok()); why != "" { - t.Fatalf("healthy cluster reported %q", why) - } - cases := map[string]func(*cluster.Status){ - "phase creating": func(s *cluster.Status) { s.Phase = "creating" }, - "not quorate": func(s *cluster.Status) { s.Quorate = &no }, - "pve1 not running or offline": func(s *cluster.Status) { s.Nodes[0].Online = nil }, - "ceph HEALTH_WARN": func(s *cluster.Status) { s.Ceph = "HEALTH_WARN" }, - } - for want, mutate := range cases { - s := ok() - mutate(s) - if got := healthy(s); got != want { - t.Errorf("got %q, want %q", got, want) - } - } -} diff --git a/internal/cli/mcp.go b/internal/cli/mcp.go new file mode 100644 index 0000000..34bfdc2 --- /dev/null +++ b/internal/cli/mcp.go @@ -0,0 +1,25 @@ +package cli + +import ( + "github.com/spf13/cobra" + + "github.com/virtbase/proxbase/internal/mcpserver" +) + +func mcpCmd() *cobra.Command { + return &cobra.Command{ + Use: "mcp", + Short: "Run a Model Context Protocol server on stdio for AI agents", + Long: `mcp serves the proxbase tools (create, status, exec, snapshot, fault, destroy, ...) +to an MCP client such as Claude Code, Cursor or VS Code over stdin/stdout. +The client starts it; register it e.g. with: + + claude mcp add proxbase -- proxbase mcp + +Progress and logs go to stderr. See docs/mcp.md.`, + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + return mcpserver.Run(cmd.Context(), version, sink) + }, + } +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 9642031..a5cef71 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -4,41 +4,78 @@ package cli import ( "context" "encoding/json" + "errors" "fmt" "os" "os/signal" "syscall" + "time" "github.com/spf13/cobra" "github.com/virtbase/proxbase/internal/cluster" "github.com/virtbase/proxbase/internal/network" + "github.com/virtbase/proxbase/internal/progress" "github.com/virtbase/proxbase/internal/state" ) // Set via -ldflags "-X github.com/virtbase/proxbase/internal/cli.version=...". var version = "dev" +// sink receives progress and messages; --progress selects text or JSON lines. +var sink = progress.Text(os.Stderr) + +// Execute runs the command line and returns the process exit code. func Execute() int { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() + var format string root := &cobra.Command{ Use: "proxbase", Short: "Proxmox VE clusters in QEMU/KVM, for labs and CI", SilenceUsage: true, SilenceErrors: true, + PersistentPreRunE: func(*cobra.Command, []string) error { + switch format { + case "text": + case "json": + sink = progress.JSON(os.Stderr) + default: + return fmt.Errorf("--progress must be text or json") + } + return nil + }, } + root.PersistentFlags().StringVar(&format, "progress", "text", "progress and messages on stderr: text or json (one JSON object per line)") root.AddCommand(createCmd(), listCmd(), statusCmd(), startCmd(), stopCmd(), destroyCmd(), upCmd(), nodeCmd(), snapshotCmd(), faultCmd(), imageCmd(), sshCmd(), consoleCmd(), - envCmd(), doctorCmd(), configCmd(), versionCmd(), switchCmd()) - if err := root.ExecuteContext(ctx); err != nil { - fmt.Fprintln(os.Stderr, "Error:", err) + envCmd(), doctorCmd(), configCmd(), versionCmd(), switchCmd(), mcpCmd()) + t0 := time.Now() + err := root.ExecuteContext(ctx) + e := progress.Event{Time: time.Now(), Type: progress.TypeDone, Elapsed: time.Since(t0).Seconds()} + if err != nil { + e.Type, e.Error = progress.TypeError, err.Error() + if h, ok := errors.AsType[*hintError](err); ok { + e.Hint, e.Log = h.hint, h.log + } + } + sink(e) + if err != nil { return 1 } return 0 } -func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } +// hintError adds what to do next and where details are logged to an error. +type hintError struct { + err error + hint, log string +} + +func (e *hintError) Error() string { return e.err.Error() } +func (e *hintError) Unwrap() error { return e.err } + +func logf(format string, a ...any) { sink.Logf(format, a...) } // clusterName picks the argument, else "default", else the only existing cluster. func clusterName(args []string) string { @@ -52,7 +89,7 @@ func clusterName(args []string) string { } func openCluster(args []string) (*cluster.Cluster, error) { - return cluster.Open(clusterName(args), logf) + return cluster.Open(clusterName(args), sink) } func printJSON(v any) error { diff --git a/internal/cli/up.go b/internal/cli/up.go index f8e0ba6..7c001f9 100644 --- a/internal/cli/up.go +++ b/internal/cli/up.go @@ -27,7 +27,7 @@ cleanly and exits. An existing cluster keeps its stored configuration.`, return err } ctx := cmd.Context() - c, runErr := run(ctx, cfg.Name, func() (*cluster.Cluster, error) { return cluster.Create(ctx, cfg, logf) }) + c, runErr := run(ctx, cfg.Name, func() (*cluster.Cluster, error) { return cluster.Create(ctx, cfg, sink) }) if c == nil { return runErr } @@ -54,7 +54,7 @@ cleanly and exits. An existing cluster keeps its stored configuration.`, // run starts an existing ready cluster or creates/resumes one. func run(ctx context.Context, name string, create func() (*cluster.Cluster, error)) (*cluster.Cluster, error) { if state.ForCluster(name).Exists() { - c, err := cluster.Open(name, logf) + c, err := cluster.Open(name, sink) if err != nil { return nil, err } diff --git a/internal/cluster/cluster.go b/internal/cluster/cluster.go index f9f9887..153c949 100644 --- a/internal/cluster/cluster.go +++ b/internal/cluster/cluster.go @@ -19,6 +19,7 @@ import ( "github.com/virtbase/proxbase/internal/answer" "github.com/virtbase/proxbase/internal/config" "github.com/virtbase/proxbase/internal/network" + "github.com/virtbase/proxbase/internal/progress" "github.com/virtbase/proxbase/internal/qemu" "github.com/virtbase/proxbase/internal/state" "github.com/virtbase/proxbase/internal/storage" @@ -37,15 +38,12 @@ const ( TokenUser = "proxbase@pve" ) -// Logf reports progress. -type Logf func(format string, a ...any) - // Cluster is one cluster with its configuration, state and secrets. type Cluster struct { - Dir state.Dir - Cfg *config.Cluster - St *state.State - Log Logf + Dir state.Dir + Cfg *config.Cluster + St *state.State + Progress progress.Sink start time.Time signer ssh.Signer @@ -54,9 +52,12 @@ type Cluster struct { net network.Switch } -func newCluster(d state.Dir, cfg *config.Cluster, st *state.State, logf Logf) (*Cluster, error) { +func newCluster(d state.Dir, cfg *config.Cluster, st *state.State, p progress.Sink) (*Cluster, error) { + if p == nil { + p = progress.Discard + } c := &Cluster{ - Dir: d, Cfg: cfg, St: st, Log: logf, + Dir: d, Cfg: cfg, St: st, Progress: p, start: time.Now(), runtime: qemu.Runtime{}, net: network.Switch{Dir: d, Cluster: cfg.Name}, @@ -65,7 +66,10 @@ func newCluster(d state.Dir, cfg *config.Cluster, st *state.State, logf Logf) (* } // Open loads an existing cluster. -func Open(name string, logf Logf) (*Cluster, error) { +func Open(name string, p progress.Sink) (*Cluster, error) { + if err := config.CheckName(name); err != nil { + return nil, err + } d := state.ForCluster(name) st, err := d.Load() if err != nil { @@ -75,11 +79,11 @@ func Open(name string, logf Logf) (*Cluster, error) { if err != nil { return nil, fmt.Errorf("load %s: %w", d.Config(), err) } - return newCluster(d, cfg, st, logf) + return newCluster(d, cfg, st, p) } // initialize creates the directory, secrets and state for a new cluster. -func initialize(cfg *config.Cluster, logf Logf) (*Cluster, error) { +func initialize(cfg *config.Cluster, p progress.Sink) (*Cluster, error) { d := state.ForCluster(cfg.Name) if err := d.Init(); err != nil { return nil, err @@ -101,7 +105,7 @@ func initialize(cfg *config.Cluster, logf Logf) (*Cluster, error) { if err := d.Save(st); err != nil { return nil, err } - return newCluster(d, cfg, st, logf) + return newCluster(d, cfg, st, p) } // writeKeyPair generates the ed25519 key nodes trust for root logins. @@ -164,8 +168,17 @@ func (c *Cluster) ConsoleSocket(name string) (string, error) { return "", fmt.Errorf("cluster %s has no node %q", c.Cfg.Name, name) } +// step reports progress; a message starting with a node name is attributed to it. func (c *Cluster) step(format string, a ...any) { - c.Log("[%5.0fs] %s", time.Since(c.start).Seconds(), fmt.Sprintf(format, a...)) + msg := fmt.Sprintf(format, a...) + e := progress.Event{Time: time.Now(), Type: progress.TypeStep, Cluster: c.Cfg.Name, Message: msg, Elapsed: time.Since(c.start).Seconds()} + for _, n := range c.St.Nodes { + if rest, ok := strings.CutPrefix(msg, n.Name); ok && (rest == "" || rest[0] == ':' || rest[0] == ' ') { + e.Node = n.Name + break + } + } + c.Progress(e) } func (c *Cluster) save() error { return c.Dir.Save(c.St) } diff --git a/internal/cluster/create.go b/internal/cluster/create.go index 1f3a589..b1e6e12 100644 --- a/internal/cluster/create.go +++ b/internal/cluster/create.go @@ -13,28 +13,29 @@ import ( "github.com/virtbase/proxbase/internal/config" "github.com/virtbase/proxbase/internal/image" "github.com/virtbase/proxbase/internal/network" + "github.com/virtbase/proxbase/internal/progress" "github.com/virtbase/proxbase/internal/qemu" "github.com/virtbase/proxbase/internal/state" ) // Create builds a cluster, or resumes an unfinished create of the same name. -func Create(ctx context.Context, cfg *config.Cluster, logf Logf) (*Cluster, error) { +func Create(ctx context.Context, cfg *config.Cluster, p progress.Sink) (*Cluster, error) { d := state.ForCluster(cfg.Name) var c *Cluster var err error if d.Exists() { - if c, err = Open(cfg.Name, logf); err != nil { + if c, err = Open(cfg.Name, p); err != nil { return nil, err } if c.St.Phase == state.PhaseReady { return nil, fmt.Errorf("cluster %q already exists", cfg.Name) } - logf("resuming unfinished create of %q (using its stored %s)", cfg.Name, d.Config()) + c.Progress.Logf("resuming unfinished create of %q (using its stored %s)", cfg.Name, d.Config()) } else { if err := preflight(cfg, cfg.NodeList()); err != nil { return nil, err } - if c, err = initialize(cfg, logf); err != nil { + if c, err = initialize(cfg, p); err != nil { return nil, err } } diff --git a/internal/cluster/env.go b/internal/cluster/env.go new file mode 100644 index 0000000..2bcb2c0 --- /dev/null +++ b/internal/cluster/env.go @@ -0,0 +1,44 @@ +package cluster + +import ( + "fmt" + "strings" +) + +// NodeEndpoint is how clients reach one node. +type NodeEndpoint struct { + Name string `json:"name"` + API string `json:"api"` + SSHHost string `json:"sshHost"` + SSHPort int `json:"sshPort"` +} + +// Env is what API and SSH clients need to use the cluster. +type Env struct { + Endpoint string `json:"endpoint"` + APIToken string `json:"apiToken"` + TokenID string `json:"tokenId"` + TokenSecret string `json:"tokenSecret"` + CACert string `json:"caCert"` + SSHUser string `json:"sshUser"` + SSHKey string `json:"sshKey"` + RootPassword string `json:"rootPassword,omitempty"` + Nodes []NodeEndpoint `json:"nodes"` +} + +// Env returns endpoints and credentials, without the root password. +func (c *Cluster) Env() (*Env, error) { + token, err := c.Token() + if err != nil || token == "" { + return nil, fmt.Errorf("cluster %s has no API token yet (create not finished or access.apiToken: false)", c.Cfg.Name) + } + e := &Env{ + Endpoint: fmt.Sprintf("https://127.0.0.1:%d/", c.St.Nodes[0].UIPort), + APIToken: token, CACert: c.CAPath(), SSHUser: "root", SSHKey: c.KeyPath(), + } + e.TokenID, e.TokenSecret, _ = strings.Cut(token, "=") + for _, n := range c.St.Nodes { + e.Nodes = append(e.Nodes, NodeEndpoint{Name: n.Name, API: fmt.Sprintf("https://127.0.0.1:%d/", n.UIPort), SSHHost: "127.0.0.1", SSHPort: n.SSHPort}) + } + return e, nil +} diff --git a/internal/cluster/fault.go b/internal/cluster/fault.go index 30c8056..9958e42 100644 --- a/internal/cluster/fault.go +++ b/internal/cluster/fault.go @@ -220,8 +220,8 @@ func describe(f state.Fault) string { return f.Kind + " " + f.Node } -// faultList describes all active faults. -func (c *Cluster) faultList() []string { +// Faults describes all active faults. +func (c *Cluster) Faults() []string { var out []string for _, f := range c.St.Faults { out = append(out, describe(f)) diff --git a/internal/cluster/lifecycle.go b/internal/cluster/lifecycle.go index 8bd30d8..ee6b9f0 100644 --- a/internal/cluster/lifecycle.go +++ b/internal/cluster/lifecycle.go @@ -13,6 +13,8 @@ import ( "golang.org/x/sync/errgroup" + "github.com/virtbase/proxbase/internal/config" + "github.com/virtbase/proxbase/internal/progress" "github.com/virtbase/proxbase/internal/retry" "github.com/virtbase/proxbase/internal/state" ) @@ -117,7 +119,10 @@ func (c *Cluster) stop(ctx context.Context, timeout time.Duration) error { } // Destroy kills everything belonging to the cluster and removes its directory. -func Destroy(name string, logf Logf) error { +func Destroy(name string, p progress.Sink) error { + if err := config.CheckName(name); err != nil { + return err + } d := state.ForCluster(name) if _, err := os.Stat(string(d)); err != nil { return fmt.Errorf("cluster %q does not exist", name) @@ -127,7 +132,7 @@ func Destroy(name string, logf Logf) error { return err } defer unlock() - if c, err := Open(name, logf); err == nil { + if c, err := Open(name, p); err == nil { _ = c.stop(context.Background(), 0) } killStrays(string(d)) diff --git a/internal/cluster/node.go b/internal/cluster/node.go index de70709..3855327 100644 --- a/internal/cluster/node.go +++ b/internal/cluster/node.go @@ -134,6 +134,20 @@ func (c *Cluster) api(ctx context.Context, name string) (*pve.Client, error) { return cl, nil } +// Exec runs a command as root on a running node. +func (c *Cluster) Exec(ctx context.Context, name, command string) (remote.ExecResult, error) { + ns := c.St.Node(name) + if ns == nil { + return remote.ExecResult{}, fmt.Errorf("cluster %s has no node %q", c.Cfg.Name, name) + } + s, err := c.ssh(ctx, name) + if err != nil { + return remote.ExecResult{}, fmt.Errorf("%s: SSH: %w", name, err) + } + defer s.Close() + return s.Exec(ctx, command) +} + // tokenClient returns an API client using the token, verified against the cluster // CA, or nil if the cluster has no token yet. func (c *Cluster) tokenClient(port int) *pve.Client { diff --git a/internal/cluster/status.go b/internal/cluster/status.go index 00870ed..76311f3 100644 --- a/internal/cluster/status.go +++ b/internal/cluster/status.go @@ -32,10 +32,30 @@ type Status struct { Nodes []NodeStatus `json:"nodes"` } +// Problem explains why a cluster is not fully healthy ("" if it is): healthy +// means ready, quorate, every node running and online, and Ceph HEALTH_OK. +func (s *Status) Problem() string { + switch { + case s.Phase != "ready": + return "phase " + s.Phase + case s.Quorate == nil || !*s.Quorate: + return "not quorate" + } + for _, n := range s.Nodes { + if !n.Running || n.Online == nil || !*n.Online { + return n.Name + " not running or offline" + } + } + if s.Ceph != "" && s.Ceph != "HEALTH_OK" { + return "ceph " + s.Ceph + } + return "" +} + // Status reports processes and, if reachable, membership and storage health via // the API token. func (c *Cluster) Status(ctx context.Context) *Status { - s := &Status{Name: c.Cfg.Name, Phase: string(c.St.Phase), Error: c.St.Error, ISO: c.St.ISO, Switch: c.net.Running(), Duration: c.St.Duration, Faults: c.faultList()} + s := &Status{Name: c.Cfg.Name, Phase: string(c.St.Phase), Error: c.St.Error, ISO: c.St.ISO, Switch: c.net.Running(), Duration: c.St.Duration, Faults: c.Faults()} var api *pve.Client for _, n := range c.Cfg.NodeList() { ns := c.St.Node(n.Name) diff --git a/internal/cluster/status_test.go b/internal/cluster/status_test.go new file mode 100644 index 0000000..61fa6a6 --- /dev/null +++ b/internal/cluster/status_test.go @@ -0,0 +1,27 @@ +package cluster + +import "testing" + +func TestStatusProblem(t *testing.T) { + yes, no := true, false + ok := func() *Status { + return &Status{Phase: "ready", Quorate: &yes, Ceph: "HEALTH_OK", + Nodes: []NodeStatus{{Name: "pve1", Running: true, Online: &yes}}} + } + if why := ok().Problem(); why != "" { + t.Fatalf("healthy cluster reported %q", why) + } + cases := map[string]func(*Status){ + "phase creating": func(s *Status) { s.Phase = "creating" }, + "not quorate": func(s *Status) { s.Quorate = &no }, + "pve1 not running or offline": func(s *Status) { s.Nodes[0].Online = nil }, + "ceph HEALTH_WARN": func(s *Status) { s.Ceph = "HEALTH_WARN" }, + } + for want, mutate := range cases { + s := ok() + mutate(s) + if got := s.Problem(); got != want { + t.Errorf("got %q, want %q", got, want) + } + } +} diff --git a/internal/config/overrides.go b/internal/config/overrides.go new file mode 100644 index 0000000..f901737 --- /dev/null +++ b/internal/config/overrides.go @@ -0,0 +1,101 @@ +package config + +import ( + "fmt" + "regexp" + "strconv" + "strings" +) + +// Overrides change a cluster file from command line flags or tool parameters. +// Nil fields keep the file's value. +type Overrides struct { + Name string + Nodes, CPUs *int + Memory *string + Disk *string + DataDisks *string // "2x32G", "32G,64G" or "none" + Storage *string // zfs, ceph or none + Version *string + BindAddress *string + Golden bool +} + +// Apply sets the overrides on c, then defaults, and validates the result. +func (o Overrides) Apply(c *Cluster) error { + if o.Name != "" { + c.Name = o.Name + } + d := &c.Nodes.Defaults + if o.Nodes != nil { + c.Nodes.Count = *o.Nodes + } + if o.CPUs != nil { + d.CPUs = *o.CPUs + } + if o.Memory != nil { + d.Memory = *o.Memory + } + if o.Disk != nil { + d.RootDisk.Size = *o.Disk + } + if o.Version != nil { + c.Proxmox.Version = *o.Version + } + if o.Golden { + c.Proxmox.Golden = true + } + if o.BindAddress != nil { + c.Access.BindAddress = *o.BindAddress + } + if o.DataDisks != nil { + d.DataDisks = ParseDisks(*o.DataDisks) + c.Storage.ZFS = nil // re-derive the default pool from the new disks + } + if o.Storage != nil { + switch *o.Storage { + case "zfs": + if len(d.DataDisks) == 0 && d.DataDisks != nil { + return fmt.Errorf("storage zfs needs data disks") + } + c.Storage.Ceph = nil + case "ceph": + c.Storage.ZFS = []ZFSPool{} + if c.Storage.Ceph == nil { + c.Storage.Ceph = &Ceph{} + } + c.Storage.Ceph.Enabled = true + case "none": + c.Storage.ZFS = []ZFSPool{} + c.Storage.Ceph = nil + default: + return fmt.Errorf("storage must be zfs, ceph or none") + } + } + c.SetDefaults() + if err := c.Validate(); err != nil { + return fmt.Errorf("invalid cluster configuration:\n%w", err) + } + return nil +} + +var multiDiskRe = regexp.MustCompile(`^([0-9]+)x([0-9]+[MGT])$`) + +// ParseDisks parses "2x32G", "32G,64G" or "none" (validation checks the sizes). +func ParseDisks(s string) []Disk { + disks := []Disk{} + if s == "none" || s == "0" || s == "" { + return disks + } + if m := multiDiskRe.FindStringSubmatch(s); m != nil { + n, _ := strconv.Atoi(m[1]) + for range min(n, 9) { // more than 8 fails validation + disks = append(disks, Disk{Size: m[2]}) + } + return disks + } + for _, size := range strings.Split(s, ",") { + disks = append(disks, Disk{Size: strings.TrimSpace(size)}) + } + return disks +} diff --git a/internal/config/validate.go b/internal/config/validate.go index 8477840..37fd06a 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -19,6 +19,15 @@ var ( minDisks = map[string]int{"single": 1, "mirror": 2, "raid10": 4, "raidz": 3, "raidz2": 4, "raidz3": 5} ) +// CheckName validates a cluster name. Names become path elements of the state +// directory, so check names from outside (arguments, tool calls) before use. +func CheckName(name string) error { + if !nameRe.MatchString(name) { + return fmt.Errorf("invalid cluster name %q: must match %s", name, nameRe) + } + return nil +} + // Validate checks a defaulted config and returns all problems at once. func (c *Cluster) Validate() error { var errs []error diff --git a/internal/job/job.go b/internal/job/job.go new file mode 100644 index 0000000..021769d --- /dev/null +++ b/internal/job/job.go @@ -0,0 +1,153 @@ +// Package job runs a proxbase command detached from the caller (for example a +// create started by an MCP client), with its progress as JSON lines in a file. +// A job survives the caller; one job per cluster can run at a time. +package job + +import ( + "bufio" + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "syscall" + + "github.com/virtbase/proxbase/internal/config" + "github.com/virtbase/proxbase/internal/progress" + "github.com/virtbase/proxbase/internal/state" +) + +// Job is the detached command of one cluster. +type Job struct{ cluster string } + +// New returns the job slot of a cluster. +func New(cluster string) (Job, error) { + if err := config.CheckName(cluster); err != nil { + return Job{}, err + } + return Job{cluster: cluster}, nil +} + +// Cluster is the name of the job's cluster. +func (j Job) Cluster() string { return j.cluster } + +func dir() string { return filepath.Join(state.DataHome(), "jobs") } + +// Log is the file with the job's progress events. +func (j Job) Log() string { return j.file(".jsonl") } + +func (j Job) pidFile() string { return j.file(".pid") } + +func (j Job) file(ext string) string { return filepath.Join(dir(), j.cluster+ext) } + +// WriteInput stores the cluster file for the job and returns its path. +func (j Job) WriteInput(data []byte) (string, error) { + if err := os.MkdirAll(dir(), 0o755); err != nil { + return "", err + } + path := j.file(".yaml") + return path, state.WriteFileAtomic(path, data, 0o644) +} + +// Start runs ` args... --progress json` in its own session. +// exited is closed when the job ends, if the caller lives that long. +func (j Job) Start(args ...string) (exited <-chan struct{}, err error) { + if pid := j.PID(); pid != 0 { + return nil, fmt.Errorf("a job for cluster %s is already running (pid %d)", j.cluster, pid) + } + exe, err := os.Executable() + if err != nil { + return nil, err + } + if err := os.MkdirAll(dir(), 0o755); err != nil { + return nil, err + } + log, err := os.Create(j.Log()) + if err != nil { + return nil, err + } + cmd := exec.Command(exe, append(args, "--progress", "json")...) + cmd.Stderr = log // stdout (the final status table) is dropped; the log has a done/error event + cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true} + if err := cmd.Start(); err != nil { + log.Close() + return nil, err + } + if err := state.WriteFileAtomic(j.pidFile(), []byte(strconv.Itoa(cmd.Process.Pid)), 0o644); err != nil { + _ = cmd.Process.Kill() + } + done := make(chan struct{}) + go func() { + _ = cmd.Wait() // reap, so PID does not see a zombie + log.Close() + close(done) + }() + return done, err +} + +// PID returns the process ID of the running job, or 0. +func (j Job) PID() int { + b, err := os.ReadFile(j.pidFile()) + if err != nil { + return 0 + } + pid, err := strconv.Atoi(strings.TrimSpace(string(b))) + if err != nil || pid <= 0 || syscall.Kill(pid, 0) != nil { + return 0 + } + // Guard against PID reuse where /proc exists. + if cmdline, err := os.ReadFile(fmt.Sprintf("/proc/%d/cmdline", pid)); err == nil && !bytes.Contains(cmdline, []byte("--progress\x00json")) { + return 0 + } + return pid +} + +// Signal sends sig to the running job, if any. +func (j Job) Signal(sig syscall.Signal) error { + if pid := j.PID(); pid != 0 { + return syscall.Kill(pid, sig) + } + return nil +} + +// Remove deletes the job's files. +func (j Job) Remove() { + for _, p := range []string{j.Log(), j.pidFile(), j.file(".yaml")} { + _ = os.Remove(p) + } +} + +// Events returns the events logged after byte offset from, and the new offset. +// A partially written last line is left for the next call. +func (j Job) Events(from int64) ([]progress.Event, int64, error) { + f, err := os.Open(j.Log()) + if errors.Is(err, os.ErrNotExist) { + return nil, from, nil + } + if err != nil { + return nil, from, err + } + defer f.Close() + if _, err := f.Seek(from, io.SeekStart); err != nil { + return nil, from, err + } + var out []progress.Event + r := bufio.NewReader(f) + for { + line, err := r.ReadBytes('\n') + if err != nil { + return out, from, nil + } + from += int64(len(line)) + var e progress.Event + if json.Unmarshal(line, &e) != nil { + e = progress.Event{Type: progress.TypeLog, Message: strings.TrimSpace(string(line))} + } + out = append(out, e) + } +} diff --git a/internal/job/job_test.go b/internal/job/job_test.go new file mode 100644 index 0000000..3f73293 --- /dev/null +++ b/internal/job/job_test.go @@ -0,0 +1,66 @@ +package job + +import ( + "fmt" + "os" + "testing" + "time" + + "github.com/virtbase/proxbase/internal/progress" +) + +// The test binary is the job: it logs a step, a non-JSON line and success. +func TestMain(m *testing.M) { + if os.Getenv("PROXBASE_FAKE_JOB") == "1" { + fmt.Fprintln(os.Stderr, `{"type":"step","message":"working","elapsed":1}`) + fmt.Fprint(os.Stderr, "not json\n") + time.Sleep(200 * time.Millisecond) + fmt.Fprintln(os.Stderr, `{"type":"done"}`) + os.Exit(0) + } + os.Exit(m.Run()) +} + +func TestJob(t *testing.T) { + t.Setenv("XDG_DATA_HOME", t.TempDir()) + t.Setenv("PROXBASE_FAKE_JOB", "1") + if _, err := New("../lab"); err == nil { + t.Error("New accepted a path as cluster name") + } + j, err := New("lab") + if err != nil { + t.Fatal(err) + } + exited, err := j.Start("create") + if err != nil { + t.Fatal(err) + } + if j.PID() == 0 { + t.Error("PID: job not running") + } + if _, err := j.Start("create"); err == nil { + t.Error("second job for the same cluster started") + } + select { + case <-exited: + case <-time.After(10 * time.Second): + t.Fatal("job did not exit") + } + if pid := j.PID(); pid != 0 { + t.Errorf("PID %d after exit", pid) + } + events, off, err := j.Events(0) + if err != nil || len(events) != 3 { + t.Fatalf("events %v, %v", events, err) + } + if events[0].Message != "working" || events[1].Type != progress.TypeLog || events[1].Message != "not json" || events[2].Type != progress.TypeDone { + t.Errorf("events %+v", events) + } + if more, _, _ := j.Events(off); len(more) != 0 { + t.Errorf("events after the end offset: %v", more) + } + j.Remove() + if _, err := os.Stat(j.Log()); !os.IsNotExist(err) { + t.Errorf("log not removed: %v", err) + } +} diff --git a/internal/mcpserver/create.go b/internal/mcpserver/create.go new file mode 100644 index 0000000..f918ad8 --- /dev/null +++ b/internal/mcpserver/create.go @@ -0,0 +1,256 @@ +package mcpserver + +import ( + "context" + "errors" + "fmt" + "os" + "strings" + "syscall" + "time" + + "github.com/modelcontextprotocol/go-sdk/mcp" + + "github.com/virtbase/proxbase/internal/cluster" + "github.com/virtbase/proxbase/internal/config" + "github.com/virtbase/proxbase/internal/job" + "github.com/virtbase/proxbase/internal/progress" + "github.com/virtbase/proxbase/internal/state" +) + +const createDescription = `Create a Proxmox VE cluster (or resume a failed create of the same name). +Without config, it creates 1 node with 4 vCPUs, 4G memory, a 32G ZFS root disk and one 32G data disk +holding the ZFS pool "tank". Set nodes: 3 for a real cluster, storage: ceph for Ceph (3+ nodes). +The create runs in the background on this host and survives this connection; the tool reports +progress and returns when the cluster is ready, or with running=true after timeoutSeconds +(then call cluster_wait). Cancelling the call cancels the create (re-run to resume). +Needs free memory for all nodes; the first create downloads a ~1.5 GB ISO.` + +type createIn struct { + Name string `json:"name" jsonschema:"cluster name: lowercase letters, digits and dashes, starting with a letter, at most 15 characters"` + Config string `json:"config,omitempty" jsonschema:"cluster file (YAML, apiVersion proxbase.virtbase.com/v1alpha1, kind Cluster); the other parameters override it"` + Nodes int `json:"nodes,omitempty" jsonschema:"number of nodes (default 1)"` + CPUs int `json:"cpus,omitempty" jsonschema:"vCPUs per node (default 4)"` + Memory string `json:"memory,omitempty" jsonschema:"memory per node, e.g. 4G (default 4G, 6G with Ceph)"` + RootDisk string `json:"rootDisk,omitempty" jsonschema:"root disk size, e.g. 32G"` + DataDisks string `json:"dataDisks,omitempty" jsonschema:"data disks per node: 2x32G, 32G,64G or none"` + Storage string `json:"storage,omitempty" jsonschema:"storage on the data disks: zfs, ceph or none"` + PVEVersion string `json:"pveVersion,omitempty" jsonschema:"Proxmox VE ISO version, e.g. 9.2 or 9.2-1 (default: newest supported)"` + Golden *bool `json:"golden,omitempty" jsonschema:"clone nodes from a cached installed base image instead of installing each (default true; the first create builds the image)"` + TimeoutSeconds int `json:"timeoutSeconds,omitempty" jsonschema:"return after this many seconds while the create keeps running (default 600, max 3600)"` +} + +type waitIn struct { + Cluster string `json:"cluster" jsonschema:"cluster name"` + TimeoutSeconds int `json:"timeoutSeconds,omitempty" jsonschema:"return after this many seconds if the create is still running (default 600, max 3600)"` +} + +// jobOut is the state of a create in the background. +type jobOut struct { + Cluster string `json:"cluster"` + Running bool `json:"running" jsonschema:"the create is still running; call cluster_wait"` + Error string `json:"error,omitempty"` + Hint string `json:"hint,omitempty"` + Log string `json:"log" jsonschema:"progress log (JSON lines); node install logs are in the cluster's logs directory"` + Steps []string `json:"steps" jsonschema:"latest progress messages"` + Status *cluster.Status `json:"status,omitempty"` +} + +// stepsShown is how many progress messages results carry. +const stepsShown = 10 + +// resolve builds the cluster file from the tool parameters. +func (in createIn) resolve() (*config.Cluster, error) { + cfg := &config.Cluster{} + if in.Config != "" { + var err error + if cfg, err = config.Decode(strings.NewReader(in.Config)); err != nil { + return nil, fmt.Errorf("config: %w", err) + } + } + golden := in.Golden == nil || *in.Golden + cfg.Proxmox.Golden = cfg.Proxmox.Golden && golden + ov := config.Overrides{Name: in.Name, Golden: golden} + if in.Nodes != 0 { + ov.Nodes = &in.Nodes + } + if in.CPUs != 0 { + ov.CPUs = &in.CPUs + } + for _, f := range []struct { + dst **string + v *string + }{{&ov.Memory, &in.Memory}, {&ov.Disk, &in.RootDisk}, {&ov.DataDisks, &in.DataDisks}, {&ov.Storage, &in.Storage}, {&ov.Version, &in.PVEVersion}} { + if *f.v != "" { + *f.dst = f.v + } + } + return cfg, ov.Apply(cfg) +} + +func (h *handlers) create(ctx context.Context, req *mcp.CallToolRequest, in createIn) (*mcp.CallToolResult, jobOut, error) { + cfg, err := in.resolve() + if err != nil { + return nil, jobOut{}, err + } + if d := state.ForCluster(cfg.Name); d.Exists() { + if st, err := d.Load(); err == nil && st.Phase == state.PhaseReady { + return nil, jobOut{}, fmt.Errorf("cluster %q already exists and is ready; use it, or cluster_destroy it first", cfg.Name) + } + } + b, err := cfg.YAML() + if err != nil { + return nil, jobOut{}, err + } + j, err := job.New(cfg.Name) + if err != nil { + return nil, jobOut{}, err + } + file, err := j.WriteInput(b) + if err != nil { + return nil, jobOut{}, err + } + exited, err := j.Start("create", "-f", file) + if err != nil { + return nil, jobOut{}, err + } + out, err := h.follow(ctx, req, j, exited, seconds(in.TimeoutSeconds, 600, 3600)) + if ctx.Err() != nil { + // The create saves its state and exits; calling cluster_create again resumes it. + if serr := j.Signal(syscall.SIGTERM); serr != nil { + h.log.Logf("cancel create of %s: %v", cfg.Name, serr) + } + } + return nil, out, err +} + +func (h *handlers) wait(ctx context.Context, req *mcp.CallToolRequest, in waitIn) (*mcp.CallToolResult, jobOut, error) { + j, err := job.New(in.Cluster) + if err != nil { + return nil, jobOut{}, err + } + if _, err := os.Stat(j.Log()); err != nil { + return nil, jobOut{}, fmt.Errorf("no create was started for cluster %q from MCP; see cluster_status", in.Cluster) + } + out, err := h.follow(ctx, req, j, nil, seconds(in.TimeoutSeconds, 600, 3600)) + return nil, out, err +} + +// follow forwards the job's progress until it ends or timeout passes. exited, +// if not nil, is closed when the job (a child of this process) ends. +func (h *handlers) follow(ctx context.Context, req *mcp.CallToolRequest, j job.Job, exited <-chan struct{}, timeout time.Duration) (jobOut, error) { + p := h.sink(ctx, req) + deadline := time.NewTimer(timeout) + defer deadline.Stop() + tick := time.NewTicker(time.Second) + defer tick.Stop() + var offset int64 + var last progress.Event + forward := func() error { + events, next, err := j.Events(offset) + if err != nil { + return fmt.Errorf("read progress of %s: %w", j.Cluster(), err) + } + offset = next + for _, e := range events { + p(e) + last = e + } + return nil + } + for exited != nil || j.PID() != 0 { + if err := forward(); err != nil { + return jobOut{}, err + } + select { + case <-exited: + exited = nil + case <-tick.C: + case <-ctx.Done(): + return jobOut{}, ctx.Err() + case <-deadline.C: + out, err := summarize(j) + out.Running = true + return out, err + } + } + if err := forward(); err != nil { + return jobOut{}, err + } + out, err := summarize(j) + if err != nil { + return out, err + } + switch last.Type { + case progress.TypeError: + out.Error, out.Hint = last.Error, last.Hint + case progress.TypeDone: + default: + out.Error = "the create ended without a result; see the log" + } + if c, err := cluster.Open(j.Cluster(), nil); err == nil { + out.Status = c.Status(ctx) + } else if out.Error == "" { + return out, err + } + if out.Error != "" { + return out, errors.New(out.Error + toolHint(out.Hint)) + } + return out, nil +} + +// toolHint rewrites a CLI hint in terms of the tools. +func toolHint(hint string) string { + if hint == "" { + return "" + } + r := strings.NewReplacer("`proxbase create ", "`cluster_create name=", "`proxbase destroy ", "`cluster_destroy cluster=") + return "\n" + r.Replace(hint) +} + +// summarize returns the latest progress messages of a job. +func summarize(j job.Job) (jobOut, error) { + out := jobOut{Cluster: j.Cluster(), Log: j.Log(), Running: j.PID() != 0, Steps: []string{}} + events, _, err := j.Events(0) + if err != nil { + return out, fmt.Errorf("read progress of %s: %w", j.Cluster(), err) + } + for _, e := range events { + switch { + case e.Type == progress.TypeStep: + out.Steps = append(out.Steps, fmt.Sprintf("[%4.0fs] %s", e.Elapsed, e.Message)) + case e.Type == progress.TypeLog && e.Message != "": + out.Steps = append(out.Steps, e.Message) + } + } + out.Steps = out.Steps[max(0, len(out.Steps)-stepsShown):] + return out, nil +} + +// stopJob ends a running create: SIGTERM, then SIGKILL after 30 seconds. +func stopJob(ctx context.Context, j job.Job) error { + if j.PID() == 0 { + return nil + } + if err := j.Signal(syscall.SIGTERM); err != nil { + return fmt.Errorf("stop create of %s: %w", j.Cluster(), err) + } + deadline := time.Now().Add(30 * time.Second) + tick := time.NewTicker(500 * time.Millisecond) + defer tick.Stop() + killed := false + for j.PID() != 0 { + if !killed && time.Now().After(deadline) { + if err := j.Signal(syscall.SIGKILL); err != nil { + return fmt.Errorf("kill create of %s: %w", j.Cluster(), err) + } + killed = true + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-tick.C: + } + } + return nil +} diff --git a/internal/mcpserver/e2e_test.go b/internal/mcpserver/e2e_test.go new file mode 100644 index 0000000..cccc497 --- /dev/null +++ b/internal/mcpserver/e2e_test.go @@ -0,0 +1,115 @@ +//go:build e2e + +package mcpserver + +import ( + "context" + "os" + "os/exec" + "strings" + "testing" + "time" + + "github.com/modelcontextprotocol/go-sdk/mcp" + + "github.com/virtbase/proxbase/internal/state" +) + +// TestE2E drives a real `proxbase mcp` over stdio against a real 1-node cluster: +// +// go build -o proxbase ./cmd/proxbase +// PROXBASE_BIN=$PWD/proxbase go test -tags e2e -run TestE2E -timeout 60m -v ./internal/mcpserver +// +// It needs KVM, QEMU and about 5 GB of free memory, and destroys the cluster at +// the end. If PROXBASE_E2E_CLUSTER names an existing cluster, it uses that one +// and keeps it. +func TestE2E(t *testing.T) { + bin := os.Getenv("PROXBASE_BIN") + if bin == "" { + t.Skip("PROXBASE_BIN not set") + } + name := os.Getenv("PROXBASE_E2E_CLUSTER") + if name == "" { + name = "mcp-e2e" + } + ctx, cancel := context.WithTimeout(context.Background(), 55*time.Minute) + defer cancel() + p := &progressLog{} + cmd := exec.Command(bin, "mcp") //nolint:gosec // the binary under test, set by whoever runs the test + cmd.Stderr = os.Stderr + cs, err := mcp.NewClient(&mcp.Implementation{Name: "e2e"}, &mcp.ClientOptions{ProgressNotificationHandler: p.add}). + Connect(ctx, &mcp.CommandTransport{Command: cmd}, nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { cs.Close() }) // runs after the destroy registered below + t.Logf("server %s %s", cs.InitializeResult().ServerInfo.Name, cs.InitializeResult().ServerInfo.Version) + + if state.ForCluster(name).Exists() { + t.Logf("using existing cluster %s", name) + } else { + t.Cleanup(func() { + res, _ := call(t, cs, "cluster_destroy", map[string]any{"cluster": name, "confirm": name}) + t.Logf("cluster_destroy: %s", text(res)) + }) + res, out := call(t, cs, "cluster_create", map[string]any{"name": name, "timeoutSeconds": 3000}) + if res.IsError { + t.Fatalf("cluster_create: %s", text(res)) + } + for out["running"] == true { + res, out = call(t, cs, "cluster_wait", map[string]any{"cluster": name, "timeoutSeconds": 600}) + if res.IsError { + t.Fatalf("cluster_wait: %s", text(res)) + } + } + p.mu.Lock() + t.Logf("%d progress notifications, last: %q", len(p.msgs), p.msgs[max(0, len(p.msgs)-3):]) + p.mu.Unlock() + } + + res, out := call(t, cs, "cluster_status", map[string]any{"cluster": name}) + if out["healthy"] != true { + t.Fatalf("cluster_status: %s", text(res)) + } + res, out = call(t, cs, "node_exec", map[string]any{"cluster": name, "command": "pveversion"}) + if res.IsError || out["exitCode"] != float64(0) || !strings.Contains(out["stdout"].(string), "pve-manager/") { + t.Fatalf("node_exec pveversion: %s", text(res)) + } + t.Logf("pveversion: %s", strings.TrimSpace(out["stdout"].(string))) + _, out = call(t, cs, "node_exec", map[string]any{"cluster": name, "command": "echo err >&2; exit 3"}) + if out["exitCode"] != float64(3) || out["stderr"] != "err\n" { + t.Errorf("node_exec exit code: %v", out) + } + res, out = call(t, cs, "cluster_env", map[string]any{"cluster": name}) + env, _ := out["env"].(map[string]any) + if res.IsError || env["rootPassword"] != nil || !strings.HasPrefix(out["caCertPem"].(string), "-----BEGIN CERTIFICATE-----") { + t.Errorf("cluster_env: %s", text(res)) + } + + for _, step := range []struct { + tool string + args map[string]any + }{ + {"cluster_stop", map[string]any{"cluster": name}}, + {"snapshot_save", map[string]any{"cluster": name, "name": "clean"}}, + {"snapshot_restore", map[string]any{"cluster": name, "name": "clean"}}, + {"cluster_start", map[string]any{"cluster": name}}, + } { + t0 := time.Now() + res, _ := call(t, cs, step.tool, step.args) + if res.IsError { + t.Fatalf("%s: %s", step.tool, text(res)) + } + t.Logf("%s: %s (%s)", step.tool, firstLine(text(res)), time.Since(t0).Round(time.Second)) + } + res, out = call(t, cs, "snapshot_list", map[string]any{"cluster": name}) + if snaps, _ := out["snapshots"].([]any); len(snaps) != 1 { + t.Errorf("snapshot_list: %s", text(res)) + } + res, out = call(t, cs, "cluster_status", map[string]any{"cluster": name}) + if out["healthy"] != true { + t.Errorf("cluster_status after restore: %s", text(res)) + } +} + +func firstLine(s string) string { return strings.SplitN(s, "\n", 2)[0] } diff --git a/internal/mcpserver/server.go b/internal/mcpserver/server.go new file mode 100644 index 0000000..f9b9049 --- /dev/null +++ b/internal/mcpserver/server.go @@ -0,0 +1,122 @@ +// Package mcpserver exposes proxbase to AI agents as a Model Context Protocol +// server. Clients start it as `proxbase mcp` and talk to it over stdio. +package mcpserver + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/modelcontextprotocol/go-sdk/mcp" + + "github.com/virtbase/proxbase/internal/cluster" + "github.com/virtbase/proxbase/internal/config" + "github.com/virtbase/proxbase/internal/progress" + "github.com/virtbase/proxbase/internal/state" +) + +// SchemaURI is the resource with the JSON schema of cluster files. +const SchemaURI = "https://proxbase.virtbase.com/schema/v1alpha1/cluster.json" + +const instructions = `proxbase creates Proxmox VE clusters out of QEMU/KVM VMs on this host, for tests and labs. +Every node is reachable on 127.0.0.1 through forwarded ports (web UI/API and SSH). + +Typical flow: +1. cluster_create: blocks and reports progress. The first create downloads the ISO and builds a + base image (10-20 minutes); later creates clone it (a few minutes). If it returns with + running=true, call cluster_wait until running=false. +2. cluster_status: healthy=true means ready, quorate, all nodes online, Ceph HEALTH_OK. +3. node_exec runs shell commands as root on a node (e.g. pvesh, qm, pct, ceph). + cluster_env returns the API endpoint, token and CA certificate for API clients. +4. To reset quickly between tests: cluster_stop, snapshot_save once, then snapshot_restore + cluster_start. +5. fault_apply/fault_clear inject failures (power loss, hangs, cable pulls, partitions, latency). +6. cluster_destroy deletes everything; it needs confirm set to the cluster name. + +Clusters persist across sessions; call cluster_list first to reuse one. +The cluster file format is the resource ` + SchemaURI + `.` + +// New returns the server. log receives the progress of all operations as well +// (e.g. for the client's server log on stderr). +func New(version string, log progress.Sink) *mcp.Server { + if log == nil { + log = progress.Discard + } + s := mcp.NewServer(&mcp.Implementation{Name: "proxbase", Title: "Proxbase", Version: version}, + &mcp.ServerOptions{Instructions: instructions}) + h := &handlers{log: log} + h.register(s) + s.AddResource(&mcp.Resource{ + URI: SchemaURI, + Name: "cluster-schema", + Title: "Cluster file JSON schema", + Description: "JSON schema of proxbase cluster files (apiVersion proxbase.virtbase.com/v1alpha1, kind Cluster)", + MIMEType: "application/schema+json", + }, func(context.Context, *mcp.ReadResourceRequest) (*mcp.ReadResourceResult, error) { + b, err := config.Schema() + if err != nil { + return nil, err + } + return &mcp.ReadResourceResult{Contents: []*mcp.ResourceContents{{URI: SchemaURI, MIMEType: "application/schema+json", Text: string(b)}}}, nil + }) + return s +} + +// Run serves one client on stdin/stdout until it disconnects. +func Run(ctx context.Context, version string, log progress.Sink) error { + return New(version, log).Run(ctx, &mcp.StdioTransport{}) +} + +type handlers struct{ log progress.Sink } + +// sink forwards progress to the client (if it asked for it) and the server log. +func (h *handlers) sink(ctx context.Context, req *mcp.CallToolRequest) progress.Sink { + token := req.Params.GetProgressToken() + if token == nil || req.Session == nil { + return h.log + } + var mu sync.Mutex + n := 0 + return progress.Tee(h.log, func(e progress.Event) { + msg := e.Message + switch { + case e.Type == progress.TypeStep: + msg = fmt.Sprintf("[%4.0fs] %s", e.Elapsed, e.Message) + case e.Type == progress.TypeError: + msg = "error: " + e.Error + case msg == "": + return + } + // Held while sending: progress values must arrive in increasing order. + mu.Lock() + defer mu.Unlock() + n++ + if err := req.Session.NotifyProgress(ctx, &mcp.ProgressNotificationParams{ProgressToken: token, Progress: float64(n), Message: msg}); err != nil && ctx.Err() == nil { + h.log.Logf("progress notification: %v", err) + } + }) +} + +// clusterName picks the given name, else "default", else the only cluster. +func clusterName(name string) string { + if name != "" { + return name + } + if names, _ := state.List(); len(names) == 1 && !state.ForCluster("default").Exists() { + return names[0] + } + return "default" +} + +func (h *handlers) open(name string, p progress.Sink) (*cluster.Cluster, error) { + return cluster.Open(clusterName(name), p) +} + +func seconds(v, def, max int) time.Duration { + if v <= 0 { + v = def + } + return time.Duration(min(v, max)) * time.Second +} + +func ptr[T any](v T) *T { return &v } diff --git a/internal/mcpserver/server_test.go b/internal/mcpserver/server_test.go new file mode 100644 index 0000000..e4b1d12 --- /dev/null +++ b/internal/mcpserver/server_test.go @@ -0,0 +1,197 @@ +package mcpserver + +import ( + "context" + "encoding/json" + "fmt" + "os" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// The test binary stands in for proxbase when a tool starts a job: it prints +// the progress events of a create that fails. +func TestMain(m *testing.M) { + if os.Getenv("PROXBASE_FAKE_JOB") == "1" { + fmt.Fprintln(os.Stderr, `{"type":"step","cluster":"lab","message":"cluster lab: 1 node(s)","elapsed":0.1}`) + fmt.Fprintln(os.Stderr, `{"type":"step","cluster":"lab","node":"pve1","message":"pve1 installed in 1s","elapsed":1.2}`) + fmt.Fprintln(os.Stderr, `{"type":"error","error":"boom","hint":"Re-run `+"`proxbase create lab`"+` to resume"}`) + os.Exit(1) + } + os.Exit(m.Run()) +} + +type progressLog struct { + mu sync.Mutex + msgs []string +} + +func (p *progressLog) add(_ context.Context, req *mcp.ProgressNotificationClientRequest) { + p.mu.Lock() + defer p.mu.Unlock() + p.msgs = append(p.msgs, req.Params.Message) +} + +func connect(t *testing.T) (*mcp.ClientSession, *progressLog) { + t.Helper() + t.Setenv("XDG_DATA_HOME", t.TempDir()) + t.Setenv("XDG_CACHE_HOME", t.TempDir()) + ctx := context.Background() + st, ct := mcp.NewInMemoryTransports() + if _, err := New("test", nil).Connect(ctx, st, nil); err != nil { + t.Fatal(err) + } + p := &progressLog{} + cs, err := mcp.NewClient(&mcp.Implementation{Name: "test"}, &mcp.ClientOptions{ProgressNotificationHandler: p.add}).Connect(ctx, ct, nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { cs.Close() }) + return cs, p +} + +func call(t *testing.T, cs *mcp.ClientSession, name string, args map[string]any) (*mcp.CallToolResult, map[string]any) { + t.Helper() + params := &mcp.CallToolParams{Name: name, Arguments: args} + params.SetProgressToken("t1") + res, err := cs.CallTool(context.Background(), params) + if err != nil { + t.Fatalf("%s: %v", name, err) + } + var out map[string]any + if res.StructuredContent != nil { + b, _ := json.Marshal(res.StructuredContent) + _ = json.Unmarshal(b, &out) + } + return res, out +} + +func text(res *mcp.CallToolResult) string { + var b strings.Builder + for _, c := range res.Content { + if t, ok := c.(*mcp.TextContent); ok { + b.WriteString(t.Text) + } + } + return b.String() +} + +func TestTools(t *testing.T) { + cs, _ := connect(t) + var names []string + for tool, err := range cs.Tools(context.Background(), nil) { + if err != nil { + t.Fatal(err) + } + if tool.Description == "" { + t.Errorf("%s has no description", tool.Name) + } + names = append(names, tool.Name) + } + for _, want := range []string{"cluster_list", "cluster_status", "cluster_create", "cluster_wait", "cluster_start", "cluster_stop", + "cluster_destroy", "node_exec", "cluster_env", "snapshot_save", "snapshot_restore", "snapshot_list", "fault_apply", "fault_clear", "config_validate"} { + if !slices.Contains(names, want) { + t.Errorf("tool %s missing (have %v)", want, names) + } + } +} + +func TestSchemaResource(t *testing.T) { + cs, _ := connect(t) + res, err := cs.ReadResource(context.Background(), &mcp.ReadResourceParams{URI: SchemaURI}) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(res.Contents[0].Text, `"$id": "`+SchemaURI+`"`) { + t.Errorf("schema resource does not carry its $id:\n%.300s", res.Contents[0].Text) + } +} + +func TestValidate(t *testing.T) { + cs, _ := connect(t) + _, out := call(t, cs, "config_validate", map[string]any{"config": "name: lab\nnodes:\n count: 3\n"}) + if out["valid"] != true || !strings.Contains(out["resolved"].(string), "count: 3") { + t.Errorf("valid file: %v", out) + } + _, out = call(t, cs, "config_validate", map[string]any{"config": "name: Lab_1\n"}) + if out["valid"] != false || len(out["errors"].([]any)) == 0 { + t.Errorf("invalid name accepted: %v", out) + } + _, out = call(t, cs, "config_validate", map[string]any{"config": "name: lab\nnodez: 3\n"}) + if out["valid"] != false { + t.Errorf("unknown field accepted: %v", out) + } +} + +func TestEmptyHost(t *testing.T) { + cs, _ := connect(t) + _, out := call(t, cs, "cluster_list", nil) + if c := out["clusters"].([]any); len(c) != 0 { + t.Errorf("clusters: %v", c) + } + res, _ := call(t, cs, "cluster_status", map[string]any{"cluster": "lab"}) + if !res.IsError || !strings.Contains(text(res), "does not exist") { + t.Errorf("status of a missing cluster: %s", text(res)) + } + res, _ = call(t, cs, "cluster_destroy", map[string]any{"cluster": "lab", "confirm": "other"}) + if !res.IsError || !strings.Contains(text(res), "confirm") { + t.Errorf("destroy without confirmation: %s", text(res)) + } + res, _ = call(t, cs, "cluster_destroy", map[string]any{"cluster": "lab", "confirm": "lab"}) + if !res.IsError || !strings.Contains(text(res), "does not exist") { + t.Errorf("destroy of a missing cluster: %s", text(res)) + } + const path = "../../tmp" + for tool, args := range map[string]map[string]any{ + "cluster_destroy": {"cluster": path, "confirm": path}, + "cluster_status": {"cluster": path}, + "cluster_wait": {"cluster": path}, + "node_exec": {"cluster": path, "command": "true"}, + } { + res, _ = call(t, cs, tool, args) + if !res.IsError || !strings.Contains(text(res), "invalid cluster name") { + t.Errorf("%s accepted a path as cluster name: %s", tool, text(res)) + } + } + res, _ = call(t, cs, "cluster_create", map[string]any{"name": "lab", "storage": "nfs"}) + if !res.IsError || !strings.Contains(text(res), "storage must be") { + t.Errorf("create with bad storage: %s", text(res)) + } + res, _ = call(t, cs, "fault_apply", map[string]any{"cluster": "lab", "kind": "kill"}) + if !res.IsError { + t.Errorf("fault on a missing cluster succeeded") + } +} + +func TestCreateReportsJobProgress(t *testing.T) { + cs, p := connect(t) + t.Setenv("PROXBASE_FAKE_JOB", "1") + t0 := time.Now() + res, _ := call(t, cs, "cluster_create", map[string]any{"name": "lab", "timeoutSeconds": 30}) + if time.Since(t0) > 20*time.Second { + t.Errorf("create did not return when the job ended") + } + if !res.IsError || !strings.Contains(text(res), "boom") || !strings.Contains(text(res), "cluster_create name=lab") { + t.Errorf("create result: %s", text(res)) + } + p.mu.Lock() + defer p.mu.Unlock() + if !slices.Contains(p.msgs, "[ 1s] pve1 installed in 1s") || !slices.Contains(p.msgs, "error: boom") { + t.Errorf("progress notifications: %q", p.msgs) + } + + // The job has ended: waiting again reports its result at once. + res, _ = call(t, cs, "cluster_wait", map[string]any{"cluster": "lab", "timeoutSeconds": 5}) + if !res.IsError || !strings.Contains(text(res), "boom") { + t.Errorf("wait for an ended job: %s", text(res)) + } + res, _ = call(t, cs, "cluster_wait", map[string]any{"cluster": "other"}) + if !res.IsError || !strings.Contains(text(res), "no create") { + t.Errorf("wait without job: %s", text(res)) + } +} diff --git a/internal/mcpserver/tools.go b/internal/mcpserver/tools.go new file mode 100644 index 0000000..b689694 --- /dev/null +++ b/internal/mcpserver/tools.go @@ -0,0 +1,389 @@ +package mcpserver + +import ( + "context" + "errors" + "fmt" + "os" + "slices" + "strings" + "time" + + "github.com/modelcontextprotocol/go-sdk/mcp" + + "github.com/virtbase/proxbase/internal/cluster" + "github.com/virtbase/proxbase/internal/config" + "github.com/virtbase/proxbase/internal/job" + "github.com/virtbase/proxbase/internal/state" +) + +type clusterIn struct { + Cluster string `json:"cluster,omitempty" jsonschema:"cluster name; default: the only cluster, else \"default\""` +} + +type listOut struct { + Clusters []clusterSummary `json:"clusters"` +} + +type clusterSummary struct { + Name string `json:"name"` + Phase string `json:"phase"` + Nodes int `json:"nodes"` + Running int `json:"running"` + Job bool `json:"createRunning,omitempty"` +} + +type statusOut struct { + Healthy bool `json:"healthy"` + Problem string `json:"problem,omitempty"` + Status *cluster.Status `json:"status"` + Job *jobOut `json:"create,omitempty" jsonschema:"progress of a create still running in the background"` +} + +type messageOut struct { + Message string `json:"message"` +} + +type stopIn struct { + Cluster string `json:"cluster,omitempty" jsonschema:"cluster name; default: the only cluster"` + TimeoutSeconds int `json:"timeoutSeconds,omitempty" jsonschema:"time for a clean shutdown before powering off hard (default 180; nodes shut their guests down first)"` +} + +type destroyIn struct { + Cluster string `json:"cluster" jsonschema:"cluster name"` + Confirm string `json:"confirm" jsonschema:"must equal the cluster name; destroying deletes all disks and cannot be undone"` +} + +type execIn struct { + Cluster string `json:"cluster,omitempty" jsonschema:"cluster name; default: the only cluster"` + Node string `json:"node,omitempty" jsonschema:"node name, e.g. pve1 (default: the first node)"` + Command string `json:"command" jsonschema:"shell command line, run as root through the login shell"` + TimeoutSeconds int `json:"timeoutSeconds,omitempty" jsonschema:"kill the command after this many seconds (default 120, max 3600)"` +} + +type execOut struct { + Stdout string `json:"stdout"` + Stderr string `json:"stderr"` + ExitCode int `json:"exitCode" jsonschema:"exit status; -1 if the command was killed"` + Truncated bool `json:"truncated,omitempty" jsonschema:"stdout or stderr was cut to the last 64 KiB"` +} + +type envIn struct { + Cluster string `json:"cluster,omitempty" jsonschema:"cluster name; default: the only cluster"` + IncludeRootPassword bool `json:"includeRootPassword,omitempty" jsonschema:"also return the root password of the nodes (only if needed, e.g. for root@pam API logins)"` +} + +type envOut struct { + Env *cluster.Env `json:"env"` + CACertPEM string `json:"caCertPem" jsonschema:"the cluster CA; API certificates are signed by it"` +} + +type snapshotIn struct { + Cluster string `json:"cluster,omitempty" jsonschema:"cluster name; default: the only cluster"` + Name string `json:"name" jsonschema:"snapshot name"` +} + +type snapshotsOut struct { + Snapshots []state.Snapshot `json:"snapshots"` +} + +type faultIn struct { + Cluster string `json:"cluster,omitempty" jsonschema:"cluster name; default: the only cluster"` + Kind string `json:"kind" jsonschema:"one of: kill (power the node off hard; cluster_start boots it), freeze (stop its vCPUs), thaw, link-down, link-up (the node's cable on a network), partition (split a network into groups), degrade (latency and frame loss on a network)"` + Node string `json:"node,omitempty" jsonschema:"node for kill, freeze, thaw, link-down, link-up"` + Network string `json:"network,omitempty" jsonschema:"network for link-*, partition and degrade (default: corosync link0)"` + Groups [][]string `json:"groups,omitempty" jsonschema:"partition: lists of node names; unlisted nodes form one more group"` + DelayMs int `json:"delayMs,omitempty" jsonschema:"degrade: delay per frame and direction in milliseconds"` + LossPercent float64 `json:"lossPercent,omitempty" jsonschema:"degrade: percentage of frames dropped per direction"` +} + +type faultsOut struct { + Faults []string `json:"faults" jsonschema:"active faults"` + Cleared []string `json:"cleared,omitempty"` +} + +type validateIn struct { + Config string `json:"config" jsonschema:"cluster file (YAML)"` +} + +type validateOut struct { + Valid bool `json:"valid"` + Errors []string `json:"errors,omitempty"` + Warnings []string `json:"warnings,omitempty"` + Resolved string `json:"resolved,omitempty" jsonschema:"the cluster file with all defaults, as create would use it"` +} + +const maxOutput = 64 << 10 + +func (h *handlers) register(s *mcp.Server) { + readOnly := &mcp.ToolAnnotations{ReadOnlyHint: true, OpenWorldHint: ptr(false)} + local := func(destructive, idempotent bool) *mcp.ToolAnnotations { + return &mcp.ToolAnnotations{DestructiveHint: &destructive, IdempotentHint: idempotent, OpenWorldHint: ptr(false)} + } + + mcp.AddTool(s, &mcp.Tool{Name: "cluster_list", Description: "List all proxbase clusters on this host with phase and running nodes.", Annotations: readOnly}, h.list) + mcp.AddTool(s, &mcp.Tool{Name: "cluster_status", Description: "Show a cluster: phase, quorum, nodes (running/online, IP, web UI URL, SSH port), Ceph health, active faults, and the progress of a create running in the background. healthy=true means ready, quorate, all nodes online and Ceph HEALTH_OK.", Annotations: readOnly}, h.status) + mcp.AddTool(s, &mcp.Tool{Name: "cluster_create", Description: createDescription, Annotations: local(false, true)}, h.create) + mcp.AddTool(s, &mcp.Tool{Name: "cluster_wait", Description: "Wait for a create running in the background (after cluster_create returned running=true), reporting its progress. Returns running=false with the status when it finished, or running=true again after timeoutSeconds.", Annotations: readOnly}, h.wait) + mcp.AddTool(s, &mcp.Tool{Name: "cluster_start", Description: "Boot a stopped cluster and wait for quorum, the API and storage. Also boots nodes killed by fault_apply kill.", Annotations: local(false, true)}, h.start) + mcp.AddTool(s, &mcp.Tool{Name: "cluster_stop", Description: "Shut all nodes of a cluster down (cleanly, then hard after the timeout). Disks are kept; cluster_start boots it again. Clears all faults.", Annotations: local(false, true)}, h.stop) + mcp.AddTool(s, &mcp.Tool{Name: "cluster_destroy", Description: "Delete a cluster: kill its VMs and remove all its disks, snapshots and credentials. Also stops a create running in the background. Set confirm to the cluster name.", Annotations: local(true, true)}, h.destroy) + mcp.AddTool(s, &mcp.Tool{Name: "node_exec", Description: "Run a shell command as root on a running node over SSH and return stdout, stderr and the exit code. Use it for pvesh, qm, pct, pvecm, ceph and the like. Non-interactive; output is capped at 64 KiB per stream.", Annotations: local(true, false)}, h.exec) + mcp.AddTool(s, &mcp.Tool{Name: "cluster_env", Description: "Return what API and SSH clients need: API endpoint, API token (proxbase@pve!api, full privileges), the cluster CA certificate (PEM and path), SSH user, key path and ports per node. Root password only if includeRootPassword is set.", Annotations: readOnly}, h.env) + mcp.AddTool(s, &mcp.Tool{Name: "snapshot_save", Description: "Snapshot all disks of a stopped cluster (call cluster_stop first). Restoring is much faster than creating again.", Annotations: local(false, false)}, h.snapshotOp((*cluster.Cluster).SnapshotSave, "saved")) + mcp.AddTool(s, &mcp.Tool{Name: "snapshot_restore", Description: "Reset all disks of a stopped cluster to a snapshot (call cluster_stop first, cluster_start afterwards). Changes since the snapshot are lost.", Annotations: local(true, true)}, h.snapshotOp((*cluster.Cluster).SnapshotRestore, "restored")) + mcp.AddTool(s, &mcp.Tool{Name: "snapshot_delete", Description: "Delete a snapshot of a cluster.", Annotations: local(true, true)}, h.snapshotOp((*cluster.Cluster).SnapshotDelete, "deleted")) + mcp.AddTool(s, &mcp.Tool{Name: "snapshot_list", Description: "List the snapshots of a cluster.", Annotations: readOnly}, h.snapshots) + mcp.AddTool(s, &mcp.Tool{Name: "fault_apply", Description: "Inject a failure into a running cluster to test HA, fencing, Ceph or applications: kill, freeze, thaw, link-down, link-up, partition or degrade. fault_clear undoes them.", Annotations: local(true, false)}, h.fault) + mcp.AddTool(s, &mcp.Tool{Name: "fault_clear", Description: "Undo all injected faults (thaw, plug cables in, heal partitions and degradations). Killed nodes need cluster_start.", Annotations: local(false, true)}, h.clearFaults) + mcp.AddTool(s, &mcp.Tool{Name: "config_validate", Description: "Check a cluster file (YAML) without creating anything; returns errors, warnings and the file with all defaults. The schema is the resource " + SchemaURI + ".", Annotations: readOnly}, h.validate) +} + +func (h *handlers) list(ctx context.Context, _ *mcp.CallToolRequest, _ struct{}) (*mcp.CallToolResult, listOut, error) { + names, err := state.List() + if err != nil { + return nil, listOut{}, err + } + out := listOut{Clusters: []clusterSummary{}} + for _, name := range names { + c, err := cluster.Open(name, nil) + if err != nil { + out.Clusters = append(out.Clusters, clusterSummary{Name: name, Phase: "broken: " + err.Error()}) + continue + } + s := c.Status(ctx) + r := clusterSummary{Name: name, Phase: s.Phase, Nodes: len(s.Nodes)} + if j, err := job.New(name); err == nil { + r.Job = j.PID() != 0 + } + for _, n := range s.Nodes { + if n.Running { + r.Running++ + } + } + out.Clusters = append(out.Clusters, r) + } + return nil, out, nil +} + +func (h *handlers) status(ctx context.Context, _ *mcp.CallToolRequest, in clusterIn) (*mcp.CallToolResult, statusOut, error) { + name := clusterName(in.Cluster) + j, err := job.New(name) + if err != nil { + return nil, statusOut{}, err + } + var out statusOut + if j.PID() != 0 { + js, err := summarize(j) + if err != nil { + return nil, out, err + } + out.Job = &js + } + c, err := cluster.Open(name, nil) + if err != nil { + if out.Job != nil { // create started, cluster not initialized yet + return nil, out, nil + } + return nil, out, err + } + out.Status = c.Status(ctx) + out.Problem = out.Status.Problem() + out.Healthy = out.Problem == "" + return nil, out, nil +} + +func (h *handlers) start(ctx context.Context, req *mcp.CallToolRequest, in clusterIn) (*mcp.CallToolResult, statusOut, error) { + c, err := h.open(in.Cluster, h.sink(ctx, req)) + if err != nil { + return nil, statusOut{}, err + } + if err := c.Start(ctx); err != nil { + return nil, statusOut{}, err + } + s := c.Status(ctx) + return nil, statusOut{Status: s, Problem: s.Problem(), Healthy: s.Problem() == ""}, nil +} + +func (h *handlers) stop(ctx context.Context, req *mcp.CallToolRequest, in stopIn) (*mcp.CallToolResult, messageOut, error) { + c, err := h.open(in.Cluster, h.sink(ctx, req)) + if err != nil { + return nil, messageOut{}, err + } + if err := c.Stop(ctx, seconds(in.TimeoutSeconds, 180, 1800)); err != nil { + return nil, messageOut{}, err + } + return nil, messageOut{Message: "cluster " + c.Cfg.Name + " stopped"}, nil +} + +func (h *handlers) destroy(ctx context.Context, req *mcp.CallToolRequest, in destroyIn) (*mcp.CallToolResult, messageOut, error) { + if in.Cluster == "" || in.Confirm != in.Cluster { + return nil, messageOut{}, errors.New("set confirm to the cluster name to destroy it") + } + j, err := job.New(in.Cluster) + if err != nil { + return nil, messageOut{}, err + } + if err := stopJob(ctx, j); err != nil { + return nil, messageOut{}, err + } + if _, err := os.Stat(string(state.ForCluster(in.Cluster))); err == nil { + if err := cluster.Destroy(in.Cluster, h.sink(ctx, req)); err != nil { + return nil, messageOut{}, err + } + } else if _, err := os.Stat(j.Log()); err != nil { + return nil, messageOut{}, fmt.Errorf("cluster %q does not exist", in.Cluster) + } + j.Remove() + return nil, messageOut{Message: "cluster " + in.Cluster + " destroyed"}, nil +} + +func (h *handlers) exec(ctx context.Context, _ *mcp.CallToolRequest, in execIn) (*mcp.CallToolResult, execOut, error) { + c, err := h.open(in.Cluster, nil) + if err != nil { + return nil, execOut{}, err + } + node := in.Node + if node == "" { + node = c.St.Nodes[0].Name + } + timeout := seconds(in.TimeoutSeconds, 120, 3600) + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + r, err := c.Exec(ctx, node, in.Command) + if errors.Is(err, context.DeadlineExceeded) { + err = fmt.Errorf("%s: command killed after %s", node, timeout) + } + if err != nil { + return nil, execOut{}, err + } + out := execOut{ExitCode: r.ExitCode} + out.Stdout, out.Truncated = tail(r.Stdout) + var cut bool + out.Stderr, cut = tail(r.Stderr) + out.Truncated = out.Truncated || cut + return nil, out, nil +} + +// tail keeps the last maxOutput bytes. +func tail(s string) (string, bool) { + if len(s) <= maxOutput { + return s, false + } + return s[len(s)-maxOutput:], true +} + +func (h *handlers) env(_ context.Context, _ *mcp.CallToolRequest, in envIn) (*mcp.CallToolResult, envOut, error) { + c, err := h.open(in.Cluster, nil) + if err != nil { + return nil, envOut{}, err + } + e, err := c.Env() + if err != nil { + return nil, envOut{}, err + } + if in.IncludeRootPassword { + e.RootPassword = c.Password() + } + ca, err := os.ReadFile(c.CAPath()) + if err != nil { + return nil, envOut{}, err + } + return nil, envOut{Env: e, CACertPEM: string(ca)}, nil +} + +func (h *handlers) snapshotOp(f func(*cluster.Cluster, context.Context, string) error, done string) mcp.ToolHandlerFor[snapshotIn, messageOut] { + return func(ctx context.Context, req *mcp.CallToolRequest, in snapshotIn) (*mcp.CallToolResult, messageOut, error) { + c, err := h.open(in.Cluster, h.sink(ctx, req)) + if err != nil { + return nil, messageOut{}, err + } + t0 := time.Now() + if err := f(c, ctx, in.Name); err != nil { + return nil, messageOut{}, err + } + return nil, messageOut{Message: fmt.Sprintf("snapshot %s %s in %s", in.Name, done, time.Since(t0).Round(time.Millisecond))}, nil + } +} + +func (h *handlers) snapshots(_ context.Context, _ *mcp.CallToolRequest, in clusterIn) (*mcp.CallToolResult, snapshotsOut, error) { + c, err := h.open(in.Cluster, nil) + if err != nil { + return nil, snapshotsOut{}, err + } + return nil, snapshotsOut{Snapshots: append([]state.Snapshot{}, c.St.Snapshots...)}, nil +} + +func (h *handlers) fault(ctx context.Context, req *mcp.CallToolRequest, in faultIn) (*mcp.CallToolResult, faultsOut, error) { + c, err := h.open(in.Cluster, h.sink(ctx, req)) + if err != nil { + return nil, faultsOut{}, err + } + needNode := slices.Contains([]string{"kill", "freeze", "thaw", "link-down", "link-up"}, in.Kind) + if needNode && in.Node == "" { + return nil, faultsOut{}, fmt.Errorf("fault %s needs node", in.Kind) + } + switch in.Kind { + case "kill": + err = c.KillNode(in.Node) + case "freeze": + err = c.FreezeNode(ctx, in.Node) + case "thaw": + err = c.ThawNode(ctx, in.Node) + case "link-down", "link-up": + err = c.SetLink(ctx, in.Node, in.Network, in.Kind == "link-up") + case "partition": + if len(in.Groups) == 0 { + return nil, faultsOut{}, errors.New("partition needs groups") + } + err = c.Partition(in.Network, in.Groups) + case "degrade": + err = c.Degrade(in.Network, time.Duration(in.DelayMs)*time.Millisecond, in.LossPercent/100) + default: + return nil, faultsOut{}, fmt.Errorf("unknown fault kind %q", in.Kind) + } + if err != nil { + return nil, faultsOut{}, err + } + return nil, faultsOut{Faults: nonNil(c.Faults())}, nil +} + +func (h *handlers) clearFaults(ctx context.Context, req *mcp.CallToolRequest, in clusterIn) (*mcp.CallToolResult, faultsOut, error) { + c, err := h.open(in.Cluster, h.sink(ctx, req)) + if err != nil { + return nil, faultsOut{}, err + } + cleared, err := c.ClearFaults(ctx) + if err != nil { + return nil, faultsOut{}, err + } + return nil, faultsOut{Faults: nonNil(c.Faults()), Cleared: cleared}, nil +} + +func (h *handlers) validate(_ context.Context, _ *mcp.CallToolRequest, in validateIn) (*mcp.CallToolResult, validateOut, error) { + cfg, err := config.Decode(strings.NewReader(in.Config)) + if err == nil { + err = config.Overrides{}.Apply(cfg) + } + if err != nil { + var errs []string + for _, line := range strings.Split(err.Error(), "\n") { + if line = strings.TrimSpace(line); line != "" && !strings.HasPrefix(line, "invalid cluster configuration") { + errs = append(errs, line) + } + } + return nil, validateOut{Errors: errs}, nil + } + b, err := cfg.YAML() + if err != nil { + return nil, validateOut{}, err + } + return nil, validateOut{Valid: true, Warnings: cfg.Warnings(), Resolved: string(b)}, nil +} + +func nonNil(s []string) []string { + if s == nil { + return []string{} + } + return s +} diff --git a/internal/progress/progress.go b/internal/progress/progress.go new file mode 100644 index 0000000..214342a --- /dev/null +++ b/internal/progress/progress.go @@ -0,0 +1,87 @@ +// Package progress carries progress events of long operations (create, start, +// node add, ...) to a sink: human-readable text, JSON lines or an MCP client. +package progress + +import ( + "encoding/json" + "fmt" + "io" + "sync" + "time" +) + +// Event types. +const ( + TypeStep = "step" // a step of an operation finished or started + TypeLog = "log" // any other message + TypeDone = "done" // the command succeeded + TypeError = "error" // the command failed +) + +// Event is one progress report. Elapsed is seconds since the operation started. +type Event struct { + Time time.Time `json:"time"` + Type string `json:"type"` + Cluster string `json:"cluster,omitempty"` + Node string `json:"node,omitempty"` + Message string `json:"message,omitempty"` + Elapsed float64 `json:"elapsed,omitempty"` + Error string `json:"error,omitempty"` + Hint string `json:"hint,omitempty"` + Log string `json:"log,omitempty"` // file or directory with details +} + +// Sink receives events. It must be safe for concurrent use. +type Sink func(Event) + +// Logf sends a log event. +func (s Sink) Logf(format string, a ...any) { + if s != nil { + s(Event{Time: time.Now(), Type: TypeLog, Message: fmt.Sprintf(format, a...)}) + } +} + +// Discard drops all events. +func Discard(Event) {} + +// Text writes events as lines for humans, steps prefixed with the elapsed time. +func Text(w io.Writer) Sink { + var mu sync.Mutex + return func(e Event) { + mu.Lock() + defer mu.Unlock() + switch e.Type { + case TypeStep: + fmt.Fprintf(w, "[%5.0fs] %s\n", e.Elapsed, e.Message) + case TypeError: + fmt.Fprintln(w, "Error:", e.Error) + if e.Hint != "" { + fmt.Fprintf(w, "\n%s\n", e.Hint) + } + default: + if e.Message != "" { + fmt.Fprintln(w, e.Message) + } + } + } +} + +// JSON writes one JSON object per event and line. +func JSON(w io.Writer) Sink { + var mu sync.Mutex + enc := json.NewEncoder(w) + return func(e Event) { + mu.Lock() + defer mu.Unlock() + _ = enc.Encode(e) + } +} + +// Tee sends every event to all sinks. +func Tee(sinks ...Sink) Sink { + return func(e Event) { + for _, s := range sinks { + s(e) + } + } +} diff --git a/internal/progress/progress_test.go b/internal/progress/progress_test.go new file mode 100644 index 0000000..cabfdf3 --- /dev/null +++ b/internal/progress/progress_test.go @@ -0,0 +1,39 @@ +package progress + +import ( + "bytes" + "encoding/json" + "strings" + "testing" +) + +func TestText(t *testing.T) { + var b bytes.Buffer + s := Text(&b) + s(Event{Type: TypeStep, Message: "pve1 installed", Elapsed: 42.4}) + s.Logf("warning: %s", "low memory") + s(Event{Type: TypeDone}) + s(Event{Type: TypeError, Error: "boom", Hint: "re-run"}) + want := "[ 42s] pve1 installed\nwarning: low memory\nError: boom\n\nre-run\n" + if b.String() != want { + t.Errorf("got %q, want %q", b.String(), want) + } +} + +func TestJSON(t *testing.T) { + var b bytes.Buffer + s := JSON(&b) + s(Event{Type: TypeStep, Cluster: "lab", Node: "pve1", Message: "pve1 installed", Elapsed: 1.5}) + s(Event{Type: TypeError, Error: "boom", Hint: "re-run", Log: "/x/logs"}) + lines := strings.Split(strings.TrimSpace(b.String()), "\n") + if len(lines) != 2 { + t.Fatalf("want 2 lines, got %q", b.String()) + } + var e Event + if err := json.Unmarshal([]byte(lines[1]), &e); err != nil || e.Error != "boom" || e.Log != "/x/logs" { + t.Errorf("error event %q: %v", lines[1], err) + } + if !strings.Contains(lines[0], `"node":"pve1"`) || strings.Contains(lines[0], `"error"`) { + t.Errorf("step event %s", lines[0]) + } +} diff --git a/internal/remote/ssh.go b/internal/remote/ssh.go index a65f5f2..594032e 100644 --- a/internal/remote/ssh.go +++ b/internal/remote/ssh.go @@ -5,6 +5,7 @@ import ( "bytes" "context" "encoding/base64" + "errors" "fmt" "net" "strings" @@ -58,6 +59,46 @@ func (s *SSH) Close() error { return s.client.Close() } +// ExecResult is the outcome of a command; ExitCode is -1 if it did not exit normally. +type ExecResult struct { + Stdout string `json:"stdout"` + Stderr string `json:"stderr"` + ExitCode int `json:"exitCode"` +} + +// Exec runs a command line through the login shell. A non-zero exit is not an +// error; a cancelled ctx kills the command. +func (s *SSH) Exec(ctx context.Context, command string) (ExecResult, error) { + sess, err := s.client.NewSession() + if err != nil { + return ExecResult{}, fmt.Errorf("ssh session: %w", err) + } + defer sess.Close() + var stdout, stderr bytes.Buffer + sess.Stdout, sess.Stderr = &stdout, &stderr + done := make(chan error, 1) + go func() { done <- sess.Run(command) }() + select { + case err = <-done: + case <-ctx.Done(): + // Best effort: closing the session ends Run even if the signal is not supported. + _ = sess.Signal(ssh.SIGKILL) + _ = sess.Close() + <-done + return ExecResult{Stdout: stdout.String(), Stderr: stderr.String(), ExitCode: -1}, ctx.Err() + } + r := ExecResult{Stdout: stdout.String(), Stderr: stderr.String()} + if err == nil { + return r, nil + } + if exit, ok := errors.AsType[*ssh.ExitError](err); ok { + r.ExitCode = exit.ExitStatus() + return r, nil + } + r.ExitCode = -1 + return r, fmt.Errorf("run command: %w", err) +} + // Run executes a shell script (passed on stdin to bash) and returns stdout. func (s *SSH) Run(script string) (string, error) { sess, err := s.client.NewSession() diff --git a/server.json b/server.json new file mode 100644 index 0000000..254c486 --- /dev/null +++ b/server.json @@ -0,0 +1,62 @@ +{ + "$schema": "https://static.modelcontextprotocol.io/schemas/2025-12-11/server.schema.json", + "name": "io.github.virtbase/proxbase", + "title": "Proxbase", + "description": "Create, test and break real Proxmox VE clusters in QEMU/KVM on the local host.", + "repository": { + "url": "https://github.com/virtbase/proxbase", + "source": "github" + }, + "websiteUrl": "https://github.com/virtbase/proxbase/blob/main/docs/mcp.md", + "version": "0.2.0", + "packages": [ + { + "registryType": "oci", + "identifier": "ghcr.io/virtbase/proxbase:0.2.0", + "transport": { + "type": "stdio" + }, + "runtimeArguments": [ + { + "type": "named", + "name": "--device", + "value": "/dev/kvm", + "description": "KVM for the node VMs", + "isRequired": true + }, + { + "type": "named", + "name": "--group-add", + "value": "{kvm_gid}", + "description": "Group that may use /dev/kvm", + "isRequired": true, + "variables": { + "kvm_gid": { + "description": "GID of /dev/kvm on the host (stat -c %g /dev/kvm)", + "isRequired": true + } + } + }, + { + "type": "named", + "name": "--network", + "value": "host", + "description": "Reach the nodes' web UI, API and SSH ports from the host" + }, + { + "type": "named", + "name": "--volume", + "value": "proxbase:/data", + "description": "Keep clusters, ISOs and base images between runs", + "isRequired": true + } + ], + "packageArguments": [ + { + "type": "positional", + "value": "mcp" + } + ] + } + ] +}