From 3d5bac66b443a1b71b34e67ad58d00d520cf8f45 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Sat, 13 Jun 2026 14:23:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(provision):=20=E5=BC=B9=E6=80=A7=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E5=9F=BA=E5=BB=BA=20Terraform=20+=20=E4=B8=80?= =?UTF-8?q?=E9=94=AE=E6=9B=B4=E6=8D=A2=20(tsk=5F6u0FxmbC7Yeq)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit IaC 面 (infra/) 与控制面 (server/internal/provision) 双产出，落地 doc/04 §4 「节点是牲口」弹性拓扑与 make-before-break 一键更换。 server/internal/provision: - CloudAdapter 适配层 + Registry（首发 vultr 消耗品池 / hetzner 精品池各一）； 厂商凭证仅从 PROVISION__* env 注入，不入库不入 git。 - ProvisionService：CreateNode(幂等键重放不重复开机)、DestroyNode(幂等)、 RotateIP(换 IP 不换机 + version bump)、ListProviders。 - Replace 一键更换：先建后拆，新机 up 先于旧机 draining（容量不下降）， replacement_uuid 幂等键 + replacements 表分步记录，崩溃可续跑不重复。 - RotatePool：池内滚动轮换，并发度 1–2。 - cmd/nodectl CLI：create/destroy/rotate-ip/replace/rotate-pool/providers。 - 单测（mock 厂商 API + 内存 Store）：幂等重放、make-before-break 时序断言、 开机失败/探活超时→destroyed+失败计数+告警钩子、崩溃续跑、RotatePool。 infra/: - terraform/：探针机 + 控制面基线模块化（probe / control-plane）+ README， 低频基线进 state，节点不进 Terraform。 - cloud-init/node.yaml.tmpl：节点引导模板（注入一次性 bootstrap token，task #5）。 - identity-isolation.md：身份隔离登记表（doc/06 §2 红线，无任何凭证）。 migrations/000008：nodes 增 provider_instance_id/elastic_ip_id、node_events 增 ip_rotated、provision_idempotency / replacements 表（附加式，不动现网）。 红线：仅面向新厂商池，绝不纳管现网生产 EC2(deploy/ marzban)。 Co-Authored-By: Claude Opus 4.8 --- .gitignore | 8 + infra/README.md | 44 ++ infra/cloud-init/node.yaml.tmpl | 64 +++ infra/identity-isolation.md | 58 +++ infra/terraform/README.md | 67 +++ infra/terraform/main.tf | 42 ++ .../terraform/modules/control-plane/README.md | 30 ++ infra/terraform/modules/control-plane/main.tf | 53 ++ .../modules/control-plane/outputs.tf | 9 + .../modules/control-plane/variables.tf | 29 ++ infra/terraform/modules/probe/README.md | 38 ++ infra/terraform/modules/probe/main.tf | 57 +++ infra/terraform/modules/probe/outputs.tf | 9 + infra/terraform/modules/probe/variables.tf | 33 ++ infra/terraform/outputs.tf | 9 + infra/terraform/terraform.tfvars.example | 18 + infra/terraform/variables.tf | 53 ++ infra/terraform/versions.tf | 22 + server/cmd/nodectl/main.go | 267 ++++++++++ server/internal/provision/adapter.go | 80 +++ server/internal/provision/cloudinit.go | 66 +++ server/internal/provision/cloudinit_test.go | 84 ++++ server/internal/provision/deps.go | 97 ++++ server/internal/provision/doc.go | 31 ++ server/internal/provision/fakes_test.go | 462 ++++++++++++++++++ server/internal/provision/mysqlstore.go | 366 ++++++++++++++ server/internal/provision/providers/doc.go | 16 + .../internal/provision/providers/hetzner.go | 154 ++++++ .../internal/provision/providers/registry.go | 121 +++++ .../provision/providers/registry_test.go | 76 +++ server/internal/provision/providers/vultr.go | 110 +++++ server/internal/provision/replace.go | 312 ++++++++++++ server/internal/provision/replace_test.go | 213 ++++++++ server/internal/provision/service.go | 384 +++++++++++++++ server/internal/provision/service_test.go | 155 ++++++ server/internal/provision/store.go | 81 +++ server/internal/provision/types.go | 121 +++++ server/migrations/000008_provision.down.sql | 8 + server/migrations/000008_provision.up.sql | 37 ++ 39 files changed, 3884 insertions(+) create mode 100644 infra/README.md create mode 100644 infra/cloud-init/node.yaml.tmpl create mode 100644 infra/identity-isolation.md create mode 100644 infra/terraform/README.md create mode 100644 infra/terraform/main.tf create mode 100644 infra/terraform/modules/control-plane/README.md create mode 100644 infra/terraform/modules/control-plane/main.tf create mode 100644 infra/terraform/modules/control-plane/outputs.tf create mode 100644 infra/terraform/modules/control-plane/variables.tf create mode 100644 infra/terraform/modules/probe/README.md create mode 100644 infra/terraform/modules/probe/main.tf create mode 100644 infra/terraform/modules/probe/outputs.tf create mode 100644 infra/terraform/modules/probe/variables.tf create mode 100644 infra/terraform/outputs.tf create mode 100644 infra/terraform/terraform.tfvars.example create mode 100644 infra/terraform/variables.tf create mode 100644 infra/terraform/versions.tf create mode 100644 server/cmd/nodectl/main.go create mode 100644 server/internal/provision/adapter.go create mode 100644 server/internal/provision/cloudinit.go create mode 100644 server/internal/provision/cloudinit_test.go create mode 100644 server/internal/provision/deps.go create mode 100644 server/internal/provision/doc.go create mode 100644 server/internal/provision/fakes_test.go create mode 100644 server/internal/provision/mysqlstore.go create mode 100644 server/internal/provision/providers/doc.go create mode 100644 server/internal/provision/providers/hetzner.go create mode 100644 server/internal/provision/providers/registry.go create mode 100644 server/internal/provision/providers/registry_test.go create mode 100644 server/internal/provision/providers/vultr.go create mode 100644 server/internal/provision/replace.go create mode 100644 server/internal/provision/replace_test.go create mode 100644 server/internal/provision/service.go create mode 100644 server/internal/provision/service_test.go create mode 100644 server/internal/provision/store.go create mode 100644 server/internal/provision/types.go create mode 100644 server/migrations/000008_provision.down.sql create mode 100644 server/migrations/000008_provision.up.sql diff --git a/.gitignore b/.gitignore index d0c339e..4b5fa3c 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,14 @@ deploy/backup/backup.env *.age-private* age-keypair.txt +# Terraform 状态与本地变量（含厂商/管理凭证）—— 绝不入库 +infra/**/.terraform/ +infra/**/.terraform.lock.hcl +*.tfstate +*.tfstate.* +infra/**/terraform.tfvars +*.auto.tfvars + # 杂项 .DS_Store *.log diff --git a/infra/README.md b/infra/README.md new file mode 100644 index 0000000..9c1ad93 --- /dev/null +++ b/infra/README.md @@ -0,0 +1,44 @@ +# `infra/` — elastic-node infrastructure + +This directory is the **IaC half** of the elastic-node base (task #14). The +**control-plane half** — vendor adapters and the make-before-break orchestration +— lives in [`../server/internal/provision`](../server/internal/provision). + +``` +infra/ + terraform/ baseline resources: probe machines + control-plane env + cloud-init/ + node.yaml.tmpl data-plane node bootstrap (rendered by provision svc) + identity-isolation.md per-asset identity isolation register (doc/06 §2) +``` + +## Division of labour (doc/04 §4) + +- **Terraform** (`terraform/`) — low-frequency baseline only (probes, control + plane). These are in Terraform state. +- **Provision service** (`server/internal/provision`) — high-frequency, + minute-scale, idempotent node open/退机 via vendor APIs. Nodes are cattle and + are **never** in Terraform state. + +## `cloud-init/node.yaml.tmpl` + +The entire persistent footprint of a data-plane node: sing-box + agent + a +one-time bootstrap token (doc/04 §2). It is a Go `text/template` rendered by the +provision service's `CloudInitRenderer`, injecting: + +`NodeUUID`, `BootstrapToken` (task #5), `Region`, `Role`, `Tier`, +`ControlPlaneURL`. + +The rendered output carries the one-time enrollment token and **must not be +logged**. The agent self-registers over mTLS (task #6); the control plane then +drives the node `provisioning → probing → up`. + +## Red lines (doc/06 §2) + +- Everything here targets the **new vendor pools**. It must **never** manage the + production EC2 host (the `deploy/` marzban machine). +- Credentials are never committed: Terraform tokens via `TF_VAR_*` / CI secrets, + vendor API keys via `PROVISION__*` env consumed by the provision + service. The `providers` table stores only `name/api_kind/regions/pool/enabled`. +- Each vendor/domain/account is an independent identity with zero cross-linkage; + see `identity-isolation.md`. diff --git a/infra/cloud-init/node.yaml.tmpl b/infra/cloud-init/node.yaml.tmpl new file mode 100644 index 0000000..b11bfa6 --- /dev/null +++ b/infra/cloud-init/node.yaml.tmpl @@ -0,0 +1,64 @@ +## template: jinja-free Go text/template — rendered by +## server/internal/provision (CloudInitRenderer). Fields: +## {{.NodeUUID}} {{.BootstrapToken}} {{.Region}} {{.Role}} {{.Tier}} {{.ControlPlaneURL}} +## +## "Nodes are cattle, not pets" (doc/04 §2): this is the ENTIRE persistent +## footprint of a data-plane node — sing-box + agent binaries + a one-time +## bootstrap token. Zero user DB, zero logs, zero persistent state. +## +## SECURITY: the rendered output carries a one-time bootstrap token. It MUST NOT +## be logged. The token is consumed exactly once during mTLS enrollment (task #5) +## and is useless afterwards. +#cloud-config + +write_files: + # Node identity + enrollment parameters consumed by the agent on first boot. + - path: /etc/pangolin/bootstrap.env + permissions: "0600" + owner: root:root + content: | + PANGOLIN_NODE_UUID={{.NodeUUID}} + PANGOLIN_BOOTSTRAP_TOKEN={{.BootstrapToken}} + PANGOLIN_CONTROL_PLANE_URL={{.ControlPlaneURL}} + PANGOLIN_REGION={{.Region}} + PANGOLIN_ROLE={{.Role}} + PANGOLIN_TIER={{.Tier}} + + # systemd unit: agent enrolls (mTLS) then pulls runtime config from control plane. + - path: /etc/systemd/system/pangolin-agent.service + permissions: "0644" + owner: root:root + content: | + [Unit] + Description=Pangolin node agent (self-register + config pull) + After=network-online.target + Wants=network-online.target + + [Service] + Type=simple + EnvironmentFile=/etc/pangolin/bootstrap.env + ExecStart=/usr/local/bin/pangolin-agent run + # No-log node: agent does not persist connection logs. + Restart=always + RestartSec=5 + + [Install] + WantedBy=multi-user.target + +runcmd: + # 1. Harden: key-only SSH, no root password (doc/06 §3). + - sed -i 's/^#\?PermitRootLogin.*/PermitRootLogin prohibit-password/' /etc/ssh/sshd_config + - sed -i 's/^#\?PasswordAuthentication.*/PasswordAuthentication no/' /etc/ssh/sshd_config + - systemctl restart sshd || true + # 2. Install the agent binary (pinned, checksum-verified by the install script, + # delivered out-of-band by task #6). Placeholder URL resolved at build time. + - install -d -m 0755 /etc/pangolin + - /usr/local/bin/pangolin-agent-install || true + # 3. Enroll + start. The agent reads bootstrap.env, performs mTLS enrollment, + # and the control plane flips the node provisioning → probing → up. + - systemctl daemon-reload + - systemctl enable --now pangolin-agent.service + # 4. Shred the one-time token from disk after enrollment (defence in depth). + - bash -c 'sleep 60; shred -u /etc/pangolin/bootstrap.env 2>/dev/null || rm -f /etc/pangolin/bootstrap.env' + +# No swap file written to disk, no persistent data dirs — a destroyed node leaks nothing. diff --git a/infra/identity-isolation.md b/infra/identity-isolation.md new file mode 100644 index 0000000..0d3ec3c --- /dev/null +++ b/infra/identity-isolation.md @@ -0,0 +1,58 @@ +# Identity isolation register (doc/06 §2 红线) + +> **RED LINE (any phase, never crossed):** servers, domains, CDN, object +> storage, payment, email and phone numbers MUST NOT share any linkable identity. +> Each vendor uses an **independent account + independent email + crypto +> payment**. This file is the authoritative register of which isolated identity +> backs which asset. **It contains NO secrets** — only the isolation mapping. +> Credentials live in independent secret stores (env / files), never here, never +> in the database, never in git. + +## How to use this register + +1. Onboarding a new vendor/asset → add a row before provisioning anything. +2. Each row gets a distinct `identity-id` (an opaque internal label, e.g. + `id-a7`). Never reuse an identity across rows. +3. Record the asset, pool, payment rail, and the **secret location** (where the + credential is injected from), not the credential itself. + +## Data-plane vendor pools (doc/04 §5.2) + +Consumed by `server/internal/provision`. The `providers` table stores only +`name/api_kind/regions/pool/enabled`; the matching credential is injected from +the secret location below as `PROVISION__*`. + +| identity-id | asset | pool | api_kind | payment | secret location (env) | notes | +|-------------|-------|------|----------|---------|-----------------------|-------| +| id-c1 | _vendor A_ | consumable | `vultr` | USDT/crypto | `PROVISION_VULTR_API_KEY` | small/cheap entry pool; ≥3 vendors target | +| id-c2 | _vendor B_ | consumable | _tbd_ | crypto | `PROVISION__API_KEY` | second consumable vendor | +| id-c3 | _vendor C_ | consumable | _tbd_ | crypto | `PROVISION__API_KEY` | third consumable vendor | +| id-p1 | _vendor D_ | premium | `hetzner` | crypto | `PROVISION_HETZNER_API_TOKEN` | stable exit/pro entry; native IPs | + +> Consumable pool and premium pool MUST use **different vendor accounts** so a +> mass-burn of the consumable pool never touches the premium pool (doc/04 §4.2). + +## Management / baseline (Terraform) + +Managed by `infra/terraform`. Independent from all data-plane vendor identities. + +| identity-id | asset | purpose | payment | secret location | +|-------------|-------|---------|---------|-----------------| +| id-m1 | mgmt/probe cloud | probe machines + control-plane host | crypto | `TF_VAR_hcloud_token` (CI secret) | +| id-m2 | control-plane DB/Redis | user DB, audit_log | — | server deploy secrets | + +## Adjacent assets (registered elsewhere, listed for completeness) + +| identity-id | asset | owner doc | +|-------------|-------|-----------| +| id-d1 | domain registrar (WHOIS privacy) | doc/05 | +| id-d2 | CDN account | doc/05 | +| id-pay1 | payment / card store (USDT-TRC20) | doc/02 §4 | +| id-tg1 | ops TG bot (anonymous) | doc/04 §5.3 | + +## Audit checklist (run before each onboarding) + +- [ ] New identity-id is unique; no email/account reused from another row. +- [ ] Payment is crypto / non-KYC; not bound to a real-name account. +- [ ] Credential stored only in its secret location; absent from git & DB. +- [ ] `ci/scan-redline.sh` and a repo/DB secret scan pass. diff --git a/infra/terraform/README.md b/infra/terraform/README.md new file mode 100644 index 0000000..835490c --- /dev/null +++ b/infra/terraform/README.md @@ -0,0 +1,67 @@ +# `infra/terraform` — baseline infrastructure as code + +Terraform here manages **only low-frequency baseline resources**: + +- **probe machines** — overseas reference dial-test points (doc/04 §1, §4.2). +- **control-plane environment** — the API + MySQL + Redis host (doc/04 §5.2), + off by default because it changes rarely. + +## What is deliberately NOT here + +**Data-plane nodes are not in Terraform.** They are disposable cattle, created +and destroyed minute-by-minute through vendor APIs by the provision service +(`server/internal/provision`, doc/04 §4). Putting them in Terraform state would +fight the "nodes are cattle, not pets" model and serialise every open/退机 +behind a state lock. Division of labour: + +| concern | owner | frequency | in TF state | +|---------|-------|-----------|-------------| +| probe machines | Terraform (this dir) | low | ✅ | +| control-plane env | Terraform (this dir) | one-off | ✅ | +| data-plane nodes (entry/relay/exit) | provision service + vendor API | minutes | ❌ | + +## Identity isolation (RED LINE, doc/06 §2) + +Every vendor/account/domain used here MUST be an **independent identity** — its +own account, email, and crypto payment, with **zero cross-linkage** to other +assets. This Terraform code and the data-plane vendor accounts use **separate** +credentials: + +- `var.hcloud_token` here is the **management/probe** account token — *not* a + data-plane vendor token. +- Data-plane vendor credentials live only in `PROVISION__*` env secrets + consumed by the provision service, never in Terraform, never in the DB. + +The full isolation register (which account/email/payment maps to which asset) is +maintained in [`../identity-isolation.md`](../identity-isolation.md). + +## Usage + +```bash +cd infra/terraform +cp terraform.tfvars.example terraform.tfvars # gitignored; fill in +export TF_VAR_hcloud_token=... # inject secret, never commit + +terraform init +terraform fmt -check +terraform validate +terraform plan +terraform apply +``` + +CI/management operations run on a **dedicated channel** (doc/06 §2), not from an +operator's everyday machine. + +## Layout + +``` +terraform/ + versions.tf provider + backend pins + variables.tf root inputs + main.tf wires probe[] + control-plane modules + outputs.tf probe IPs, control-plane IP + terraform.tfvars.example + modules/ + probe/ one overseas reference probe point + control-plane/ API+MySQL+Redis host (prevent_destroy) +``` diff --git a/infra/terraform/main.tf b/infra/terraform/main.tf new file mode 100644 index 0000000..aec5a40 --- /dev/null +++ b/infra/terraform/main.tf @@ -0,0 +1,42 @@ +provider "hcloud" { + token = var.hcloud_token +} + +# SSH keys for the management identity (probe + control-plane access). +resource "hcloud_ssh_key" "mgmt" { + for_each = { for idx, key in var.ssh_public_keys : idx => key } + name = "pangolin-mgmt-${each.key}" + public_key = each.value +} + +locals { + ssh_key_ids = [for k in hcloud_ssh_key.mgmt : k.id] +} + +# Overseas reference probe points (low frequency → Terraform-managed). +# 境内 probe points are serverless / third-party dialing services with strict +# identity isolation and are intentionally OUT of Terraform scope (registered in +# ../identity-isolation.md). +module "probe" { + source = "./modules/probe" + for_each = var.probe_overseas + + name = each.key + location = each.value.location + server_type = each.value.server_type + image = var.probe_image + ssh_key_ids = local.ssh_key_ids +} + +# Control-plane baseline env (API + MySQL + Redis), provisioned once and rarely +# changed. Disabled by default; flip control_plane_enabled to manage it here. +module "control_plane" { + source = "./modules/control-plane" + count = var.control_plane_enabled ? 1 : 0 + + server_type = var.control_plane_server_type + location = var.control_plane_location + image = var.probe_image + ssh_key_ids = local.ssh_key_ids + allowlist_cidrs = var.admin_allowlist_cidrs +} diff --git a/infra/terraform/modules/control-plane/README.md b/infra/terraform/modules/control-plane/README.md new file mode 100644 index 0000000..bc4d5d2 --- /dev/null +++ b/infra/terraform/modules/control-plane/README.md @@ -0,0 +1,30 @@ +# Module: `control-plane` + +Provisions the **control-plane baseline host** (API + MySQL + Redis), doc/04 §5.2. + +Key properties: + +- Deployed on a **stable cloud unrelated to the data-plane node vendors**. Nodes + can all be destroyed without affecting the control plane, and vice-versa. The + control plane carries **no proxy traffic**. +- `prevent_destroy = true` — it holds the user DB; it is never recreated + implicitly. Application bring-up (containers, migrations) is done by the server + deploy pipeline, not Terraform. +- Admin/SSH surface is **never on the open internet** (doc/06 §2): inbound is + locked to `allowlist_cidrs`; the public API sits behind a CDN/proxy terminated + elsewhere. + +Disabled by default in the root module (`control_plane_enabled = false`) because +it changes rarely. + +## Inputs + +| name | description | +|------|-------------| +| `server_type` / `location` / `image` | host sizing | +| `ssh_key_ids` | authorised SSH keys | +| `allowlist_cidrs` | admin/SSH inbound allowlist (tighten in prod) | + +## Outputs + +`ipv4_address`, `server_id`. diff --git a/infra/terraform/modules/control-plane/main.tf b/infra/terraform/modules/control-plane/main.tf new file mode 100644 index 0000000..d2ff17d --- /dev/null +++ b/infra/terraform/modules/control-plane/main.tf @@ -0,0 +1,53 @@ +terraform { + required_providers { + hcloud = { + source = "hetznercloud/hcloud" + version = "~> 1.45" + } + } +} + +# Control-plane baseline host (API + MySQL + Redis). Deployed on a stable cloud +# COMPLETELY UNRELATED to the data-plane node vendors (doc/04 §5.2): nodes can +# all die without touching the control plane, and vice-versa. Carries NO proxy +# traffic. +# +# Provisioned once and rarely changed — hence Terraform-managed. Application +# bring-up (containers, migrations) is handled separately by the server deploy +# pipeline, not here. +resource "hcloud_server" "control_plane" { + name = "pangolin-control-plane" + server_type = var.server_type + image = var.image + location = var.location + ssh_keys = var.ssh_key_ids + + labels = { + role = "control-plane" + project = "pangolin" + } + + lifecycle { + # Never recreate the control plane implicitly — it holds the user DB. + prevent_destroy = true + } +} + +# Admin/SSH surface is never on the open internet (doc/06 §2). Inbound is locked +# to the management allowlist; the public API is expected to sit behind a CDN / +# reverse proxy terminated elsewhere. +resource "hcloud_firewall" "control_plane" { + name = "pangolin-control-plane" + + rule { + direction = "in" + protocol = "tcp" + port = "22" + source_ips = length(var.allowlist_cidrs) > 0 ? var.allowlist_cidrs : ["0.0.0.0/0", "::/0"] + } +} + +resource "hcloud_firewall_attachment" "control_plane" { + firewall_id = hcloud_firewall.control_plane.id + server_ids = [hcloud_server.control_plane.id] +} diff --git a/infra/terraform/modules/control-plane/outputs.tf b/infra/terraform/modules/control-plane/outputs.tf new file mode 100644 index 0000000..4f1dd3b --- /dev/null +++ b/infra/terraform/modules/control-plane/outputs.tf @@ -0,0 +1,9 @@ +output "ipv4_address" { + description = "Public IPv4 of the control-plane host." + value = hcloud_server.control_plane.ipv4_address +} + +output "server_id" { + description = "Vendor server ID." + value = hcloud_server.control_plane.id +} diff --git a/infra/terraform/modules/control-plane/variables.tf b/infra/terraform/modules/control-plane/variables.tf new file mode 100644 index 0000000..32b42f3 --- /dev/null +++ b/infra/terraform/modules/control-plane/variables.tf @@ -0,0 +1,29 @@ +variable "server_type" { + description = "Vendor server type/size for the control-plane host." + type = string + default = "cpx21" +} + +variable "location" { + description = "Vendor location/datacenter." + type = string + default = "hel1" +} + +variable "image" { + description = "OS image." + type = string + default = "debian-12" +} + +variable "ssh_key_ids" { + description = "Vendor SSH key IDs authorised on the host." + type = list(string) + default = [] +} + +variable "allowlist_cidrs" { + description = "CIDRs allowed to reach SSH/admin. Empty = open (MUST be tightened in prod; doc/06 §2)." + type = list(string) + default = [] +} diff --git a/infra/terraform/modules/probe/README.md b/infra/terraform/modules/probe/README.md new file mode 100644 index 0000000..7379389 --- /dev/null +++ b/infra/terraform/modules/probe/README.md @@ -0,0 +1,38 @@ +# Module: `probe` + +Provisions **one overseas reference probe point** (doc/04 §1, §4.2). + +Probes are low-frequency baseline resources, so they live in Terraform state — +unlike data-plane nodes, which are disposable cattle managed by the provision +service via vendor APIs (never in Terraform). + +## What it creates + +- `hcloud_server` running the probe agent (installed out-of-band; the probe + holds **no** data-plane secrets and **no** user data). +- A firewall locking inbound to SSH (optionally CIDR-restricted). + +## Role in the system + +Overseas probes cross-check the in-China probe fleet so the scheduler can tell +**blocked** (in-China fails, overseas OK) from **machine down** (both fail), +avoiding false positives before triggering an automatic replacement (doc/04 §4.2). + +> In-China probe points use serverless / third-party dialing services under a +> separate isolated identity and are **out of Terraform scope** — registered in +> [`../../identity-isolation.md`](../../identity-isolation.md). + +## Inputs + +| name | description | +|------|-------------| +| `name` | short id (e.g. `eu`, `us`) | +| `location` | vendor location | +| `server_type` | vendor size | +| `image` | OS image | +| `ssh_key_ids` | authorised SSH key IDs | +| `ssh_allowlist_cidrs` | inbound SSH allowlist | + +## Outputs + +`ipv4_address`, `server_id`. diff --git a/infra/terraform/modules/probe/main.tf b/infra/terraform/modules/probe/main.tf new file mode 100644 index 0000000..9599441 --- /dev/null +++ b/infra/terraform/modules/probe/main.tf @@ -0,0 +1,57 @@ +terraform { + required_providers { + hcloud = { + source = "hetznercloud/hcloud" + version = "~> 1.45" + } + } +} + +# One overseas reference probe point (doc/04 §4.2). It runs the probe script and +# reports TCP/TLS/real-handshake reachability of data-plane nodes back to the +# control plane, used to distinguish "blocked" from "machine down". +resource "hcloud_server" "probe" { + name = "pangolin-probe-${var.name}" + server_type = var.server_type + image = var.image + location = var.location + ssh_keys = var.ssh_key_ids + + labels = { + role = "probe" + project = "pangolin" + scope = "overseas-reference" + } + + # Probe bootstrap: pin + install the probe agent out-of-band. Kept minimal; + # probes hold NO data-plane secrets and NO user data. + user_data = <<-EOT + #cloud-config + package_update: true + runcmd: + - sed -i 's/^#\?PermitRootLogin.*/PermitRootLogin prohibit-password/' /etc/ssh/sshd_config + - sed -i 's/^#\?PasswordAuthentication.*/PasswordAuthentication no/' /etc/ssh/sshd_config + - systemctl restart sshd || true + EOT + + lifecycle { + create_before_destroy = true + } +} + +# Firewall: probes only need outbound; lock inbound to SSH from management. +resource "hcloud_firewall" "probe" { + name = "pangolin-probe-${var.name}" + + rule { + direction = "in" + protocol = "tcp" + port = "22" + source_ips = length(var.ssh_allowlist_cidrs) > 0 ? var.ssh_allowlist_cidrs : ["0.0.0.0/0", "::/0"] + } +} + +resource "hcloud_firewall_attachment" "probe" { + firewall_id = hcloud_firewall.probe.id + server_ids = [hcloud_server.probe.id] +} diff --git a/infra/terraform/modules/probe/outputs.tf b/infra/terraform/modules/probe/outputs.tf new file mode 100644 index 0000000..08c9308 --- /dev/null +++ b/infra/terraform/modules/probe/outputs.tf @@ -0,0 +1,9 @@ +output "ipv4_address" { + description = "Public IPv4 of the probe machine." + value = hcloud_server.probe.ipv4_address +} + +output "server_id" { + description = "Vendor server ID." + value = hcloud_server.probe.id +} diff --git a/infra/terraform/modules/probe/variables.tf b/infra/terraform/modules/probe/variables.tf new file mode 100644 index 0000000..f617c51 --- /dev/null +++ b/infra/terraform/modules/probe/variables.tf @@ -0,0 +1,33 @@ +variable "name" { + description = "Short probe identifier (e.g. eu, us)." + type = string +} + +variable "location" { + description = "Vendor location/datacenter for the probe." + type = string +} + +variable "server_type" { + description = "Vendor server type/size." + type = string + default = "cpx11" +} + +variable "image" { + description = "OS image." + type = string + default = "debian-12" +} + +variable "ssh_key_ids" { + description = "Vendor SSH key IDs authorised on the probe." + type = list(string) + default = [] +} + +variable "ssh_allowlist_cidrs" { + description = "CIDRs allowed to SSH the probe. Empty = open (tighten in prod)." + type = list(string) + default = [] +} diff --git a/infra/terraform/outputs.tf b/infra/terraform/outputs.tf new file mode 100644 index 0000000..68528cc --- /dev/null +++ b/infra/terraform/outputs.tf @@ -0,0 +1,9 @@ +output "probe_ips" { + description = "Public IPv4 of each overseas probe machine, keyed by probe name." + value = { for k, m in module.probe : k => m.ipv4_address } +} + +output "control_plane_ip" { + description = "Public IPv4 of the control-plane host, if managed here." + value = var.control_plane_enabled ? module.control_plane[0].ipv4_address : null +} diff --git a/infra/terraform/terraform.tfvars.example b/infra/terraform/terraform.tfvars.example new file mode 100644 index 0000000..84966b1 --- /dev/null +++ b/infra/terraform/terraform.tfvars.example @@ -0,0 +1,18 @@ +# Copy to terraform.tfvars (gitignored) and fill in. NEVER commit real values. +# hcloud_token is injected via TF_VAR_hcloud_token in CI instead of this file. + +# hcloud_token = "REDACTED — inject via TF_VAR_hcloud_token" + +ssh_public_keys = [ + # "ssh-ed25519 AAAA... pangolin-mgmt" +] + +probe_overseas = { + eu = { location = "hel1", server_type = "cpx11" } + us = { location = "ash", server_type = "cpx11" } +} + +control_plane_enabled = false +admin_allowlist_cidrs = [ + # "203.0.113.10/32" # management bastion only — admin never on public internet +] diff --git a/infra/terraform/variables.tf b/infra/terraform/variables.tf new file mode 100644 index 0000000..fe91989 --- /dev/null +++ b/infra/terraform/variables.tf @@ -0,0 +1,53 @@ +variable "hcloud_token" { + description = "Hetzner Cloud API token for the PROBE/management account ONLY (independent identity; never the data-plane vendor accounts). Inject via TF_VAR_hcloud_token / CI secret, never commit." + type = string + sensitive = true +} + +variable "ssh_public_keys" { + description = "SSH public keys authorised on probe machines (management identity)." + type = list(string) + default = [] +} + +variable "probe_overseas" { + description = "Overseas reference probe points (doc/04 §4.2). Each emits a probe machine." + type = map(object({ + location = string # vendor location, e.g. "hel1", "ash" + server_type = string # e.g. "cpx11" + })) + default = { + eu = { location = "hel1", server_type = "cpx11" } + us = { location = "ash", server_type = "cpx11" } + } +} + +variable "probe_image" { + description = "OS image for probe machines." + type = string + default = "debian-12" +} + +variable "control_plane_enabled" { + description = "Whether to manage the control-plane baseline env in this state. Off by default — the control plane is provisioned once and rarely changes." + type = bool + default = false +} + +variable "control_plane_server_type" { + description = "Server type for the control-plane host (API + MySQL + Redis), if managed here." + type = string + default = "cpx21" +} + +variable "control_plane_location" { + description = "Location for the control-plane host." + type = string + default = "hel1" +} + +variable "admin_allowlist_cidrs" { + description = "CIDRs allowed to reach the admin/SSH surface of the control plane (doc/06 §2: admin never on public internet)." + type = list(string) + default = [] +} diff --git a/infra/terraform/versions.tf b/infra/terraform/versions.tf new file mode 100644 index 0000000..de9449c --- /dev/null +++ b/infra/terraform/versions.tf @@ -0,0 +1,22 @@ +terraform { + required_version = ">= 1.5.0" + + required_providers { + # Probe machines (overseas reference points, doc/04 §1) run on a small, + # privacy-friendly cloud. Hetzner Cloud is used as the reference example; + # swap the provider per the identity-isolation register. + hcloud = { + source = "hetznercloud/hcloud" + version = "~> 1.45" + } + } + + # State holds ONLY low-frequency baseline resources (probes, control-plane + # env). Disposable data-plane nodes are NEVER in state — they are cattle, + # managed by the provision service via vendor APIs (doc/04 §4). + # + # Configure a remote, access-restricted backend out-of-band (e.g. an S3- + # compatible bucket on the management account). Left as local here so the + # module stays runnable for `terraform validate`. + # backend "s3" { ... } +} diff --git a/server/cmd/nodectl/main.go b/server/cmd/nodectl/main.go new file mode 100644 index 0000000..dc1f0c4 --- /dev/null +++ b/server/cmd/nodectl/main.go @@ -0,0 +1,267 @@ +// Command nodectl is the operator CLI for the elastic-node control plane +// (task #14). It drives the same provision.Service that the admin panel (#8) +// calls behind its buttons, so CLI and panel share one idempotent code path. +// +// Subcommands: +// +// nodectl create -provider=ID -region=hkg -tier=free|pro [-plan=...] [-key=...] +// nodectl destroy -node=ID +// nodectl rotate-ip -node=ID +// nodectl replace -node=ID [-replacement=UUID] +// nodectl rotate-pool -pool=consumable|premium [-concurrency=1|2] +// nodectl providers [-pool=consumable|premium] +// +// Configuration (env, never flags — credentials must not land in shell history): +// +// DB_DSN MySQL DSN (required) +// PROVISION_CONTROL_PLANE_URL agent enrollment URL injected into cloud-init +// PROVISION_CLOUD_INIT_TMPL path to infra/cloud-init/node.yaml.tmpl +// PROVISION__* per-vendor credentials (see providers/) +// +// nodectl performs NO operation against the production EC2 host: it only knows +// the providers/nodes registered in the new vendor pools (doc/06 §2 red line). +package main + +import ( + "context" + "flag" + "fmt" + "os" + "strings" + "text/tabwriter" + "time" + + "github.com/wangjia/pangolin/server/internal/db" + "github.com/wangjia/pangolin/server/internal/provision" + "github.com/wangjia/pangolin/server/internal/provision/providers" +) + +func main() { + if len(os.Args) < 2 { + usage() + os.Exit(2) + } + cmd := os.Args[1] + args := os.Args[2:] + + ctx := context.Background() + if err := run(ctx, cmd, args); err != nil { + fmt.Fprintln(os.Stderr, "nodectl: "+err.Error()) + os.Exit(1) + } +} + +func usage() { + fmt.Fprintln(os.Stderr, strings.TrimSpace(` +nodectl — elastic-node control plane CLI + + nodectl create -provider=ID -region=hkg -tier=free|pro [-plan=...] + nodectl destroy -node=ID + nodectl rotate-ip -node=ID + nodectl replace -node=ID [-replacement=UUID] + nodectl rotate-pool -pool=consumable|premium [-concurrency=1] + nodectl providers [-pool=consumable|premium] + +Config via env: DB_DSN, PROVISION_CONTROL_PLANE_URL, PROVISION_CLOUD_INIT_TMPL, + PROVISION__* credentials.`)) +} + +func buildService(ctx context.Context) (*provision.Service, func(), error) { + dsn := os.Getenv("DB_DSN") + if dsn == "" { + return nil, nil, fmt.Errorf("DB_DSN is required") + } + conn, err := db.Open(dsn) + if err != nil { + return nil, nil, err + } + cfg := provision.Config{ + Store: provision.NewMySQLStore(conn), + Adapters: providers.NewRegistry(), + ControlPlaneURL: os.Getenv("PROVISION_CONTROL_PLANE_URL"), + } + if tmpl := os.Getenv("PROVISION_CLOUD_INIT_TMPL"); tmpl != "" { + r, err := provision.NewTemplateRenderer(tmpl) + if err != nil { + conn.Close() + return nil, nil, err + } + cfg.Renderer = r + } + svc, err := provision.NewService(cfg) + if err != nil { + conn.Close() + return nil, nil, err + } + return svc, func() { conn.Close() }, nil +} + +func run(ctx context.Context, cmd string, args []string) error { + switch cmd { + case "create": + return cmdCreate(ctx, args) + case "destroy": + return cmdDestroy(ctx, args) + case "rotate-ip": + return cmdRotateIP(ctx, args) + case "replace": + return cmdReplace(ctx, args) + case "rotate-pool": + return cmdRotatePool(ctx, args) + case "providers": + return cmdProviders(ctx, args) + case "-h", "--help", "help": + usage() + return nil + default: + usage() + return fmt.Errorf("unknown subcommand %q", cmd) + } +} + +func cmdCreate(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("create", flag.ExitOnError) + provider := fs.Int64("provider", 0, "provider ID (required)") + region := fs.String("region", "", "vendor region ID (required)") + tier := fs.String("tier", "free", "free|pro") + plan := fs.String("plan", "", "vendor instance size") + role := fs.String("role", "entry", "entry|relay|exit") + pbk := fs.String("reality-pbk", "", "REALITY public key") + sni := fs.String("reality-sni", "www.cloudflare.com", "REALITY masquerade SNI") + hy2 := fs.Int("hy2-port", 443, "Hysteria2 port") + key := fs.String("key", "", "idempotency key (default: generated time-based)") + _ = fs.Parse(args) + if *provider == 0 || *region == "" { + return fmt.Errorf("create: -provider and -region are required") + } + idem := *key + if idem == "" { + idem = fmt.Sprintf("cli-create-%d", time.Now().UnixNano()) + } + svc, closeFn, err := buildService(ctx) + if err != nil { + return err + } + defer closeFn() + + n, err := svc.CreateNode(ctx, provision.NodeSpec{ + Region: *region, + Role: provision.Role(*role), + Tier: provision.Tier(*tier), + ProviderID: *provider, + Plan: *plan, + RealityPBK: *pbk, + RealitySNI: *sni, + HY2Port: *hy2, + }, idem) + if err != nil { + return err + } + fmt.Printf("created node id=%d uuid=%s status=%s endpoint=%s\n", n.ID, n.UUID, n.Status, n.Endpoint) + return nil +} + +func cmdDestroy(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("destroy", flag.ExitOnError) + node := fs.Int64("node", 0, "node ID (required)") + _ = fs.Parse(args) + if *node == 0 { + return fmt.Errorf("destroy: -node is required") + } + svc, closeFn, err := buildService(ctx) + if err != nil { + return err + } + defer closeFn() + if err := svc.DestroyNode(ctx, *node); err != nil { + return err + } + fmt.Printf("destroyed node id=%d\n", *node) + return nil +} + +func cmdRotateIP(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("rotate-ip", flag.ExitOnError) + node := fs.Int64("node", 0, "node ID (required)") + _ = fs.Parse(args) + if *node == 0 { + return fmt.Errorf("rotate-ip: -node is required") + } + svc, closeFn, err := buildService(ctx) + if err != nil { + return err + } + defer closeFn() + n, err := svc.RotateIP(ctx, *node) + if err != nil { + return err + } + fmt.Printf("rotated IP node id=%d new endpoint=%s\n", n.ID, n.Endpoint) + return nil +} + +func cmdReplace(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("replace", flag.ExitOnError) + node := fs.Int64("node", 0, "node ID (required)") + repl := fs.String("replacement", "", "replacement UUID (to resume a crashed replace)") + _ = fs.Parse(args) + if *node == 0 { + return fmt.Errorf("replace: -node is required") + } + svc, closeFn, err := buildService(ctx) + if err != nil { + return err + } + defer closeFn() + res, err := svc.Replace(ctx, *node, *repl) + if err != nil { + return err + } + fmt.Printf("replaced node id=%d → new id=%d (replacement=%s)\n", res.OldNodeID, res.NewNodeID, res.ReplacementUUID) + return nil +} + +func cmdRotatePool(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("rotate-pool", flag.ExitOnError) + pool := fs.String("pool", "", "consumable|premium (required)") + conc := fs.Int("concurrency", 1, "1 or 2") + _ = fs.Parse(args) + if *pool == "" { + return fmt.Errorf("rotate-pool: -pool is required") + } + svc, closeFn, err := buildService(ctx) + if err != nil { + return err + } + defer closeFn() + results, err := svc.RotatePool(ctx, provision.Pool(*pool), *conc) + if err != nil { + return fmt.Errorf("rotate-pool (%d succeeded before error): %w", len(results), err) + } + fmt.Printf("rotated %d node(s) in pool %s\n", len(results), *pool) + for _, r := range results { + fmt.Printf(" node %d → %d\n", r.OldNodeID, r.NewNodeID) + } + return nil +} + +func cmdProviders(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("providers", flag.ExitOnError) + pool := fs.String("pool", "", "consumable|premium (default: all)") + _ = fs.Parse(args) + svc, closeFn, err := buildService(ctx) + if err != nil { + return err + } + defer closeFn() + ps, err := svc.ListProviders(ctx, provision.Pool(*pool)) + if err != nil { + return err + } + w := tabwriter.NewWriter(os.Stdout, 0, 4, 2, ' ', 0) + fmt.Fprintln(w, "ID\tNAME\tAPI_KIND\tPOOL\tREGIONS") + for _, p := range ps { + fmt.Fprintf(w, "%d\t%s\t%s\t%s\t%s\n", p.ID, p.Name, p.APIKind, p.Pool, strings.Join(p.Regions, ",")) + } + return w.Flush() +} diff --git a/server/internal/provision/adapter.go b/server/internal/provision/adapter.go new file mode 100644 index 0000000..477530f --- /dev/null +++ b/server/internal/provision/adapter.go @@ -0,0 +1,80 @@ +package provision + +import ( + "context" + "errors" +) + +// Region is a vendor region/datacenter descriptor. +type Region struct { + // ID is the vendor-native region identifier (e.g. "hkg", "nbg1"). + ID string + // Country is the ISO-3166 alpha-2 country code (e.g. "HK", "JP"). + Country string + // City is a human-readable city name (optional). + City string +} + +// Instance is the vendor-side view of a freshly created machine. +type Instance struct { + // ID is the vendor-native instance identifier (used to destroy / rotate). + ID string + // IP is the public IPv4 address. + IP string + // Region echoes the region the instance was created in. + Region string +} + +// CreateInput carries everything an adapter needs to boot one node. +type CreateInput struct { + // Region is the vendor-native region ID. + Region string + // Plan is the vendor-native instance size ID. + Plan string + // Label is a vendor-side label/hostname (we pass the node UUID). + Label string + // UserData is the rendered cloud-init document (base64-encoded by the + // adapter if the vendor requires it). It carries the one-time bootstrap + // token, so it MUST NOT be logged. + UserData string + // SSHKeyIDs are optional vendor-registered SSH key IDs. + SSHKeyIDs []string + // Tags are optional vendor-side tags. + Tags []string +} + +// CloudAdapter is the unified interface every vendor adapter implements +// (doc/04 §4: "Terraform + 厂商 API 适配层"). Implementations live in +// providers/ and read their credentials only from independent secrets. +type CloudAdapter interface { + // Kind returns the adapter key matching providers.api_kind. + Kind() string + // CreateInstance boots a node and returns once the vendor has assigned an + // instance ID and IP (not necessarily once the OS is up). + CreateInstance(ctx context.Context, in CreateInput) (*Instance, error) + // DestroyInstance tears down the instance and releases its primary IP. + // It must be idempotent: destroying an already-gone instance is a no-op. + DestroyInstance(ctx context.Context, instanceID string) error + // AttachIP allocates a fresh elastic IP, binds it to the instance, and + // returns the new public IP. Only meaningful when SupportsElasticIP. + AttachIP(ctx context.Context, instanceID string) (newIP string, err error) + // ListRegions enumerates the vendor's regions. + ListRegions(ctx context.Context) ([]Region, error) + // SupportsElasticIP reports whether RotateIP (change IP, keep the box) is + // available for this vendor. + SupportsElasticIP() bool +} + +// AdapterFactory resolves the CloudAdapter for a given provider row. +// providers/Registry implements it; tests inject a fake. +type AdapterFactory interface { + For(p *Provider) (CloudAdapter, error) +} + +// ErrElasticIPUnsupported is returned by RotateIP when the provider's adapter +// does not support elastic IPs. +var ErrElasticIPUnsupported = errors.New("provision: provider does not support elastic IP") + +// ErrNoCredentials signals an adapter could not find its credentials in the +// configured secrets (env/file). +var ErrNoCredentials = errors.New("provision: vendor credentials not configured") diff --git a/server/internal/provision/cloudinit.go b/server/internal/provision/cloudinit.go new file mode 100644 index 0000000..1f0d49c --- /dev/null +++ b/server/internal/provision/cloudinit.go @@ -0,0 +1,66 @@ +package provision + +import ( + "crypto/rand" + "fmt" + "os" + "text/template" +) + +// newUUID returns a RFC-4122 v4 UUID string. Self-contained (crypto/rand) so +// the package does not depend on the not-yet-implemented idgen package. +func newUUID() (string, error) { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + return "", fmt.Errorf("provision: uuid entropy: %w", err) + } + b[6] = (b[6] & 0x0f) | 0x40 // version 4 + b[8] = (b[8] & 0x3f) | 0x80 // variant 10 + return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]), nil +} + +// TemplateRenderer renders the node cloud-init document from a Go text/template +// file (infra/cloud-init/node.yaml.tmpl). It satisfies CloudInitRenderer. +// +// The rendered output carries the one-time bootstrap token and MUST NOT be +// logged. text/template (not html/template) is used because the output is YAML. +type TemplateRenderer struct { + tmpl *template.Template +} + +// NewTemplateRenderer parses the cloud-init template file at path. +func NewTemplateRenderer(path string) (*TemplateRenderer, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("provision: read cloud-init template: %w", err) + } + return NewTemplateRendererFromString(string(raw)) +} + +// NewTemplateRendererFromString parses an in-memory template (used in tests). +func NewTemplateRendererFromString(body string) (*TemplateRenderer, error) { + t, err := template.New("node-cloud-init").Option("missingkey=error").Parse(body) + if err != nil { + return nil, fmt.Errorf("provision: parse cloud-init template: %w", err) + } + return &TemplateRenderer{tmpl: t}, nil +} + +// Render executes the template against data. +func (r *TemplateRenderer) Render(data CloudInitData) (string, error) { + var sb stringsBuilder + if err := r.tmpl.Execute(&sb, data); err != nil { + return "", fmt.Errorf("provision: render cloud-init: %w", err) + } + return sb.String(), nil +} + +// stringsBuilder is a tiny alias to avoid importing strings just for Builder +// alongside the strings import elsewhere; kept local for clarity. +type stringsBuilder struct{ buf []byte } + +func (s *stringsBuilder) Write(p []byte) (int, error) { + s.buf = append(s.buf, p...) + return len(p), nil +} +func (s *stringsBuilder) String() string { return string(s.buf) } diff --git a/server/internal/provision/cloudinit_test.go b/server/internal/provision/cloudinit_test.go new file mode 100644 index 0000000..b9dba81 --- /dev/null +++ b/server/internal/provision/cloudinit_test.go @@ -0,0 +1,84 @@ +package provision + +import ( + "os" + "strings" + "testing" +) + +// TestRealCloudInitTemplate ensures the shipped infra/cloud-init/node.yaml.tmpl +// stays a valid Go template and renders the bootstrap token + node identity. +func TestRealCloudInitTemplate(t *testing.T) { + const path = "../../../infra/cloud-init/node.yaml.tmpl" + if _, err := os.Stat(path); err != nil { + t.Skipf("template not found at %s: %v", path, err) + } + r, err := NewTemplateRenderer(path) + if err != nil { + t.Fatalf("parse real template: %v", err) + } + out, err := r.Render(CloudInitData{ + NodeUUID: "11111111-2222-4333-8444-555555555555", + BootstrapToken: "deadbeef", + Region: "hkg", + Role: RoleEntry, + Tier: TierFree, + ControlPlaneURL: "https://cp.example", + }) + if err != nil { + t.Fatalf("render real template: %v", err) + } + for _, want := range []string{ + "PANGOLIN_NODE_UUID=11111111-2222-4333-8444-555555555555", + "PANGOLIN_BOOTSTRAP_TOKEN=deadbeef", + "PANGOLIN_CONTROL_PLANE_URL=https://cp.example", + "#cloud-config", + } { + if !strings.Contains(out, want) { + t.Errorf("rendered cloud-init missing %q", want) + } + } +} + +func TestNewUUID_Format(t *testing.T) { + u, err := newUUID() + if err != nil { + t.Fatalf("newUUID: %v", err) + } + if len(u) != 36 || strings.Count(u, "-") != 4 { + t.Errorf("uuid format wrong: %q", u) + } + if u[14] != '4' { + t.Errorf("not a v4 uuid: %q", u) + } + u2, _ := newUUID() + if u == u2 { + t.Error("uuid not unique") + } +} + +func TestTemplateRenderer(t *testing.T) { + r, err := NewTemplateRendererFromString("id={{.NodeUUID}}\ntoken={{.BootstrapToken}}\ncp={{.ControlPlaneURL}}") + if err != nil { + t.Fatalf("parse: %v", err) + } + out, err := r.Render(CloudInitData{NodeUUID: "abc", BootstrapToken: "tok", ControlPlaneURL: "https://cp"}) + if err != nil { + t.Fatalf("render: %v", err) + } + for _, want := range []string{"id=abc", "token=tok", "cp=https://cp"} { + if !strings.Contains(out, want) { + t.Errorf("rendered output missing %q:\n%s", want, out) + } + } +} + +func TestTemplateRenderer_MissingKeyErrors(t *testing.T) { + r, err := NewTemplateRendererFromString("{{.NoSuchField}}") + if err != nil { + t.Fatalf("parse: %v", err) + } + if _, err := r.Render(CloudInitData{}); err == nil { + t.Error("expected error for missing template key") + } +} diff --git a/server/internal/provision/deps.go b/server/internal/provision/deps.go new file mode 100644 index 0000000..08724f9 --- /dev/null +++ b/server/internal/provision/deps.go @@ -0,0 +1,97 @@ +package provision + +import ( + "context" + "time" +) + +// Clock abstracts time for testability (Replace drains on a timer). +type Clock interface { + Now() time.Time + // Sleep blocks for d or until ctx is done, returning ctx.Err() on cancel. + Sleep(ctx context.Context, d time.Duration) error +} + +// realClock is the production Clock. +type realClock struct{} + +func (realClock) Now() time.Time { return time.Now().UTC() } + +func (realClock) Sleep(ctx context.Context, d time.Duration) error { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + return nil + } +} + +// Prober verifies a node is reachable before it joins the directory. +// +// The full version (three-way probe cross-check, doc/04 §4.2) depends on the +// probe fleet (task #15). Until then a simplified Prober is injected: +// "heartbeat present + an overseas reachability check". Replace calls WaitReady +// after a node is created; on success the node is promoted to up. +type Prober interface { + // WaitReady blocks until the node passes simplified probing, or returns an + // error on timeout / ctx cancellation / probe failure. + WaitReady(ctx context.Context, node *Node) error +} + +// AlertHook receives operational alerts (doc/04 §5.3 — TG bot in production). +// Replace / CreateNode call Fire on boot failure, probe timeout, and capacity +// protection breaches. +type AlertHook interface { + Fire(ctx context.Context, alert Alert) +} + +// AlertKind classifies an alert. +type AlertKind string + +const ( + AlertBootFailed AlertKind = "boot_failed" + AlertProbeTimeout AlertKind = "probe_timeout" + AlertReplaceFail AlertKind = "replace_failed" + AlertCapacity AlertKind = "capacity_low" +) + +// Alert is a single operational alert payload. +type Alert struct { + Kind AlertKind + NodeUUID string + Pool Pool + Message string + // FailCount is the running failure counter for capacity-protection alerts. + FailCount int +} + +// noopAlert discards alerts; used when no hook is configured. +type noopAlert struct{} + +func (noopAlert) Fire(context.Context, Alert) {} + +// BootstrapIssuer issues one-time enrollment tokens (task #5). +// *mtls.BootstrapTokenManager satisfies this interface. +type BootstrapIssuer interface { + IssueToken(ctx context.Context, nodeUUID string) (string, error) +} + +// CloudInitRenderer renders the node cloud-init document from a template, +// injecting the node UUID and bootstrap token (template lives at +// infra/cloud-init/node.yaml.tmpl, task #6). +type CloudInitRenderer interface { + Render(data CloudInitData) (string, error) +} + +// CloudInitData is the template context for the node cloud-init document. +type CloudInitData struct { + NodeUUID string + BootstrapToken string + Region string + Role Role + Tier Tier + // ControlPlaneURL is the mTLS enrollment endpoint the agent registers to. + ControlPlaneURL string +} diff --git a/server/internal/provision/doc.go b/server/internal/provision/doc.go new file mode 100644 index 0000000..6cf16a9 --- /dev/null +++ b/server/internal/provision/doc.go @@ -0,0 +1,31 @@ +// Package provision owns the elastic-node control plane: it turns the +// "nodes are cattle, not pets" principle (doc/04 §2) into running code. +// +// Responsibility split (doc/04 §4): +// +// - Terraform (infra/terraform) manages low-frequency baseline resources: +// probe machines and the control-plane environment. Those live in +// Terraform state. +// - This package drives high-frequency, minute-scale node lifecycle through +// vendor APIs. Disposable nodes are NOT in Terraform state. +// +// The package exposes a ProvisionService with idempotent operations: +// +// - CreateNode — insert nodes(status=provisioning) → vendor API boot → +// render cloud-init (injecting a one-time bootstrap token, task #5) → +// return. The agent self-registers (task #6) and flips the node to up. +// - DestroyNode — vendor destroy + IP release + nodes→destroyed. +// - RotateIP — swap the elastic IP without re-creating the machine +// (change IP, keep the box), then bump the directory version. +// - Replace — make-before-break one-click replacement: bring a fresh +// node up BEFORE draining/destroying the old one, so capacity never dips. +// - RotatePool — rolling Replace across a whole pool, concurrency 1–2. +// +// Vendor adapters live in providers/ behind the CloudAdapter interface. +// Vendor credentials are injected ONLY from independent secrets (env/file) — +// never stored in the database and never committed to git. The providers table +// holds just name/api_kind/regions/pool/enabled. +// +// Hard red line (doc/06 §2): every operation here targets the new vendor pools. +// It MUST NOT ever manage the production EC2 host (deploy/ marzban machine). +package provision diff --git a/server/internal/provision/fakes_test.go b/server/internal/provision/fakes_test.go new file mode 100644 index 0000000..25175a5 --- /dev/null +++ b/server/internal/provision/fakes_test.go @@ -0,0 +1,462 @@ +package provision + +import ( + "context" + "fmt" + "sync" + "time" +) + +// --------------------------------------------------------------------------- +// fakeStore — an in-memory Store that records the ordered sequence of status +// transitions so tests can assert make-before-break timing. +// --------------------------------------------------------------------------- + +type statusTransition struct { + Seq int + NodeID int64 + Status Status +} + +type fakeStore struct { + mu sync.Mutex + + nodes map[int64]*Node + byUUID map[string]int64 + providers map[int64]*Provider + idem map[string]string + replaces map[string]*Replacement + events []struct { + NodeID int64 + Event Event + Detail string + } + audits []string + transitions []statusTransition + + nextNodeID int64 + version int64 + seq int +} + +func newFakeStore() *fakeStore { + return &fakeStore{ + nodes: map[int64]*Node{}, + byUUID: map[string]int64{}, + providers: map[int64]*Provider{}, + idem: map[string]string{}, + replaces: map[string]*Replacement{}, + nextNodeID: 0, + version: 1, + } +} + +func (f *fakeStore) addProvider(p *Provider) { + f.mu.Lock() + defer f.mu.Unlock() + f.providers[p.ID] = p +} + +func (f *fakeStore) InsertNode(_ context.Context, n *Node) (int64, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.nextNodeID++ + id := f.nextNodeID + cp := *n + cp.ID = id + if cp.Status == "" { + cp.Status = StatusProvisioning + } + cp.CreatedAt = time.Now() + f.nodes[id] = &cp + f.byUUID[cp.UUID] = id + f.seq++ + f.transitions = append(f.transitions, statusTransition{f.seq, id, cp.Status}) + return id, nil +} + +func (f *fakeStore) GetNode(_ context.Context, id int64) (*Node, error) { + f.mu.Lock() + defer f.mu.Unlock() + n, ok := f.nodes[id] + if !ok { + return nil, nil + } + cp := *n + return &cp, nil +} + +func (f *fakeStore) GetNodeByUUID(_ context.Context, uuid string) (*Node, error) { + f.mu.Lock() + id, ok := f.byUUID[uuid] + f.mu.Unlock() + if !ok { + return nil, nil + } + return f.GetNode(context.Background(), id) +} + +func (f *fakeStore) UpdateNodeStatus(_ context.Context, id int64, status Status) error { + f.mu.Lock() + defer f.mu.Unlock() + n, ok := f.nodes[id] + if !ok { + return fmt.Errorf("fakeStore: node %d not found", id) + } + n.Status = status + f.seq++ + f.transitions = append(f.transitions, statusTransition{f.seq, id, status}) + return nil +} + +func (f *fakeStore) UpdateNodeEndpoint(_ context.Context, id int64, endpoint string) error { + f.mu.Lock() + defer f.mu.Unlock() + if n, ok := f.nodes[id]; ok { + n.Endpoint = endpoint + } + return nil +} + +func (f *fakeStore) SetNodeInstance(_ context.Context, id int64, instanceID, endpoint string) error { + f.mu.Lock() + defer f.mu.Unlock() + if n, ok := f.nodes[id]; ok { + n.ProviderInstanceID = instanceID + n.Endpoint = endpoint + } + return nil +} + +func (f *fakeStore) SetNodeWeight(_ context.Context, id int64, weight int) error { + f.mu.Lock() + defer f.mu.Unlock() + if n, ok := f.nodes[id]; ok { + n.Weight = weight + } + return nil +} + +func (f *fakeStore) ListNodesByPool(_ context.Context, pool Pool, status Status) ([]*Node, error) { + f.mu.Lock() + defer f.mu.Unlock() + var out []*Node + for _, n := range f.nodes { + p := f.providers[n.ProviderID] + if p == nil || p.Pool != pool { + continue + } + if status != "" && n.Status != status { + continue + } + cp := *n + out = append(out, &cp) + } + return out, nil +} + +func (f *fakeStore) ListProviders(_ context.Context, pool Pool) ([]*Provider, error) { + f.mu.Lock() + defer f.mu.Unlock() + var out []*Provider + for _, p := range f.providers { + if !p.Enabled { + continue + } + if pool != "" && p.Pool != pool { + continue + } + cp := *p + out = append(out, &cp) + } + return out, nil +} + +func (f *fakeStore) GetProvider(_ context.Context, id int64) (*Provider, error) { + f.mu.Lock() + defer f.mu.Unlock() + p, ok := f.providers[id] + if !ok { + return nil, nil + } + cp := *p + return &cp, nil +} + +func (f *fakeStore) WriteNodeEvent(_ context.Context, nodeID int64, event Event, detail string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.events = append(f.events, struct { + NodeID int64 + Event Event + Detail string + }{nodeID, event, detail}) + return nil +} + +func (f *fakeStore) WriteAuditLog(_ context.Context, actor, action, target, meta string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.audits = append(f.audits, fmt.Sprintf("%s|%s|%s", actor, action, target)) + return nil +} + +func (f *fakeStore) BumpDirectoryVersion(_ context.Context) (int64, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.version++ + return f.version, nil +} + +func (f *fakeStore) LookupIdempotency(_ context.Context, key string) (string, bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + v, ok := f.idem[key] + return v, ok, nil +} + +func (f *fakeStore) SaveIdempotency(_ context.Context, key, nodeUUID string) error { + f.mu.Lock() + defer f.mu.Unlock() + if _, ok := f.idem[key]; !ok { + f.idem[key] = nodeUUID + } + return nil +} + +func (f *fakeStore) CreateReplacement(_ context.Context, r *Replacement) error { + f.mu.Lock() + defer f.mu.Unlock() + cp := *r + f.replaces[r.UUID] = &cp + return nil +} + +func (f *fakeStore) GetReplacement(_ context.Context, uuid string) (*Replacement, error) { + f.mu.Lock() + defer f.mu.Unlock() + r, ok := f.replaces[uuid] + if !ok { + return nil, nil + } + cp := *r + return &cp, nil +} + +func (f *fakeStore) UpdateReplacement(_ context.Context, r *Replacement) error { + f.mu.Lock() + defer f.mu.Unlock() + cp := *r + f.replaces[r.UUID] = &cp + return nil +} + +// helpers for assertions + +func (f *fakeStore) countEvents(e Event) int { + f.mu.Lock() + defer f.mu.Unlock() + n := 0 + for _, ev := range f.events { + if ev.Event == e { + n++ + } + } + return n +} + +// firstSeqForStatus returns the seq of the first transition of nodeID into status, +// or -1 if it never happened. +func (f *fakeStore) firstSeqForStatus(nodeID int64, status Status) int { + f.mu.Lock() + defer f.mu.Unlock() + for _, t := range f.transitions { + if t.NodeID == nodeID && t.Status == status { + return t.Seq + } + } + return -1 +} + +// --------------------------------------------------------------------------- +// fakeAdapter / fakeFactory +// --------------------------------------------------------------------------- + +type fakeAdapter struct { + mu sync.Mutex + createCalls int + destroyCalls int + createErr error + elastic bool + nextIP int + created []string // instance IDs created + destroyed []string +} + +func newFakeAdapter() *fakeAdapter { return &fakeAdapter{elastic: true} } + +func (a *fakeAdapter) Kind() string { return "fake" } +func (a *fakeAdapter) SupportsElasticIP() bool { return a.elastic } + +func (a *fakeAdapter) CreateInstance(_ context.Context, in CreateInput) (*Instance, error) { + a.mu.Lock() + defer a.mu.Unlock() + a.createCalls++ + if a.createErr != nil { + return nil, a.createErr + } + a.nextIP++ + id := fmt.Sprintf("inst-%d", a.createCalls) + a.created = append(a.created, id) + return &Instance{ID: id, IP: fmt.Sprintf("203.0.113.%d", a.nextIP), Region: in.Region}, nil +} + +func (a *fakeAdapter) DestroyInstance(_ context.Context, instanceID string) error { + a.mu.Lock() + defer a.mu.Unlock() + a.destroyCalls++ + a.destroyed = append(a.destroyed, instanceID) + return nil +} + +func (a *fakeAdapter) AttachIP(_ context.Context, _ string) (string, error) { + a.mu.Lock() + defer a.mu.Unlock() + a.nextIP++ + return fmt.Sprintf("198.51.100.%d", a.nextIP), nil +} + +func (a *fakeAdapter) ListRegions(context.Context) ([]Region, error) { + return []Region{{ID: "hkg", Country: "HK"}}, nil +} + +func (a *fakeAdapter) createCount() int { + a.mu.Lock() + defer a.mu.Unlock() + return a.createCalls +} + +func (a *fakeAdapter) destroyCount() int { + a.mu.Lock() + defer a.mu.Unlock() + return a.destroyCalls +} + +type fakeFactory struct{ a CloudAdapter } + +func (f fakeFactory) For(*Provider) (CloudAdapter, error) { return f.a, nil } + +// --------------------------------------------------------------------------- +// fakeProber / fakeClock / fakeAlert / fakeTokens +// --------------------------------------------------------------------------- + +type fakeProber struct { + err error + calls int +} + +func (p *fakeProber) WaitReady(ctx context.Context, _ *Node) error { + p.calls++ + if p.err != nil { + return p.err + } + return nil +} + +// fakeClock records sleeps without actually blocking. +type fakeClock struct { + now time.Time + slept []time.Duration +} + +func (c *fakeClock) Now() time.Time { return c.now } +func (c *fakeClock) Sleep(_ context.Context, d time.Duration) error { + c.slept = append(c.slept, d) + c.now = c.now.Add(d) + return nil +} + +type fakeAlert struct { + mu sync.Mutex + alerts []Alert +} + +func (a *fakeAlert) Fire(_ context.Context, al Alert) { + a.mu.Lock() + defer a.mu.Unlock() + a.alerts = append(a.alerts, al) +} + +func (a *fakeAlert) count(kind AlertKind) int { + a.mu.Lock() + defer a.mu.Unlock() + n := 0 + for _, al := range a.alerts { + if al.Kind == kind { + n++ + } + } + return n +} + +type fakeTokens struct{ calls int } + +func (t *fakeTokens) IssueToken(context.Context, string) (string, error) { + t.calls++ + return fmt.Sprintf("token-%d", t.calls), nil +} + +// --------------------------------------------------------------------------- +// test harness builder +// --------------------------------------------------------------------------- + +type harness struct { + svc *Service + store *fakeStore + adapter *fakeAdapter + prober *fakeProber + clock *fakeClock + alert *fakeAlert + tokens *fakeTokens +} + +func newHarness(t interface{ Fatalf(string, ...any) }) *harness { + store := newFakeStore() + // One consumable + one premium provider. + store.addProvider(&Provider{ID: 1, Name: "v", APIKind: "fake", Pool: PoolConsumable, Enabled: true}) + store.addProvider(&Provider{ID: 2, Name: "h", APIKind: "fake", Pool: PoolPremium, Enabled: true}) + + adapter := newFakeAdapter() + prober := &fakeProber{} + clock := &fakeClock{now: time.Unix(1700000000, 0).UTC()} + alert := &fakeAlert{} + tokens := &fakeTokens{} + renderer, err := NewTemplateRendererFromString("uuid={{.NodeUUID}} token={{.BootstrapToken}}") + if err != nil { + t.Fatalf("renderer: %v", err) + } + svc, err := NewService(Config{ + Store: store, + Adapters: fakeFactory{adapter}, + Tokens: tokens, + Renderer: renderer, + Prober: prober, + Alert: alert, + Clock: clock, + ControlPlaneURL: "https://cp.example", + DrainTimeout: 30 * time.Minute, + ProbeTimeout: time.Minute, + }) + if err != nil { + t.Fatalf("NewService: %v", err) + } + return &harness{svc, store, adapter, prober, clock, alert, tokens} +} + +func proSpec() NodeSpec { + return NodeSpec{Region: "hkg", Tier: TierPro, ProviderID: 2, Plan: "cpx11", RealityPBK: "pbk", RealitySNI: "www.example.com", HY2Port: 443, NameZH: "香港", NameEn: "HK"} +} + +func freeSpec() NodeSpec { + return NodeSpec{Region: "hkg", Tier: TierFree, ProviderID: 1, Plan: "vc2-1c-1gb", RealityPBK: "pbk", RealitySNI: "www.example.com", HY2Port: 443, NameZH: "香港", NameEn: "HK"} +} diff --git a/server/internal/provision/mysqlstore.go b/server/internal/provision/mysqlstore.go new file mode 100644 index 0000000..a088086 --- /dev/null +++ b/server/internal/provision/mysqlstore.go @@ -0,0 +1,366 @@ +package provision + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "strings" +) + +// MySQLStore is the production Store backed by the shared *sql.DB pool. +type MySQLStore struct { + db *sql.DB +} + +// NewMySQLStore wraps a MySQL connection pool. +func NewMySQLStore(db *sql.DB) *MySQLStore { return &MySQLStore{db: db} } + +var _ Store = (*MySQLStore)(nil) + +func marshalJSONList(v []string) string { + if v == nil { + v = []string{} + } + b, _ := json.Marshal(v) + return string(b) +} + +func unmarshalJSONList(s sql.NullString) []string { + if !s.Valid || s.String == "" { + return nil + } + var out []string + _ = json.Unmarshal([]byte(s.String), &out) + return out +} + +// --- nodes --- + +func (s *MySQLStore) InsertNode(ctx context.Context, n *Node) (int64, error) { + endpoint := n.Endpoint + if endpoint == "" { + endpoint = pendingEndpoint + } + status := n.Status + if status == "" { + status = StatusProvisioning + } + weight := n.Weight + if weight == 0 { + weight = 100 + } + res, err := s.db.ExecContext(ctx, + `INSERT INTO nodes + (uuid, region, name_zh, name_en, role, tier, endpoint, hy2_port, + reality_pbk, reality_sni, provider_id, tags, status, weight, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, UTC_TIMESTAMP(6))`, + n.UUID, n.Region, n.NameZH, n.NameEn, string(n.Role), string(n.Tier), + endpoint, nullInt(n.HY2Port), n.RealityPBK, n.RealitySNI, n.ProviderID, + marshalJSONList(n.Tags), string(status), weight) + if err != nil { + return 0, fmt.Errorf("store.InsertNode: %w", err) + } + id, err := res.LastInsertId() + if err != nil { + return 0, fmt.Errorf("store.InsertNode last id: %w", err) + } + return id, nil +} + +const nodeColumns = `id, uuid, region, name_zh, name_en, role, tier, endpoint, + hy2_port, reality_pbk, reality_sni, provider_id, provider_instance_id, + elastic_ip_id, tags, status, weight, created_at` + +func scanNode(row interface{ Scan(...any) error }) (*Node, error) { + var ( + n Node + hy2 sql.NullInt64 + instanceID sql.NullString + elasticID sql.NullString + tags sql.NullString + ) + if err := row.Scan( + &n.ID, &n.UUID, &n.Region, &n.NameZH, &n.NameEn, &n.Role, &n.Tier, + &n.Endpoint, &hy2, &n.RealityPBK, &n.RealitySNI, &n.ProviderID, + &instanceID, &elasticID, &tags, &n.Status, &n.Weight, &n.CreatedAt, + ); err != nil { + return nil, err + } + n.HY2Port = int(hy2.Int64) + n.ProviderInstanceID = instanceID.String + n.ElasticIPID = elasticID.String + n.Tags = unmarshalJSONList(tags) + return &n, nil +} + +func (s *MySQLStore) GetNode(ctx context.Context, id int64) (*Node, error) { + row := s.db.QueryRowContext(ctx, `SELECT `+nodeColumns+` FROM nodes WHERE id=?`, id) + n, err := scanNode(row) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("store.GetNode: %w", err) + } + return n, nil +} + +func (s *MySQLStore) GetNodeByUUID(ctx context.Context, uuid string) (*Node, error) { + row := s.db.QueryRowContext(ctx, `SELECT `+nodeColumns+` FROM nodes WHERE uuid=?`, uuid) + n, err := scanNode(row) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("store.GetNodeByUUID: %w", err) + } + return n, nil +} + +func (s *MySQLStore) UpdateNodeStatus(ctx context.Context, id int64, status Status) error { + _, err := s.db.ExecContext(ctx, `UPDATE nodes SET status=? WHERE id=?`, string(status), id) + if err != nil { + return fmt.Errorf("store.UpdateNodeStatus: %w", err) + } + return nil +} + +func (s *MySQLStore) UpdateNodeEndpoint(ctx context.Context, id int64, endpoint string) error { + _, err := s.db.ExecContext(ctx, `UPDATE nodes SET endpoint=? WHERE id=?`, endpoint, id) + if err != nil { + return fmt.Errorf("store.UpdateNodeEndpoint: %w", err) + } + return nil +} + +func (s *MySQLStore) SetNodeInstance(ctx context.Context, id int64, instanceID, endpoint string) error { + _, err := s.db.ExecContext(ctx, + `UPDATE nodes SET provider_instance_id=?, endpoint=? WHERE id=?`, + instanceID, endpoint, id) + if err != nil { + return fmt.Errorf("store.SetNodeInstance: %w", err) + } + return nil +} + +func (s *MySQLStore) SetNodeWeight(ctx context.Context, id int64, weight int) error { + _, err := s.db.ExecContext(ctx, `UPDATE nodes SET weight=? WHERE id=?`, weight, id) + if err != nil { + return fmt.Errorf("store.SetNodeWeight: %w", err) + } + return nil +} + +func (s *MySQLStore) ListNodesByPool(ctx context.Context, pool Pool, status Status) ([]*Node, error) { + q := `SELECT ` + qualify(nodeColumns, "n") + ` + FROM nodes n JOIN providers p ON p.id = n.provider_id + WHERE p.pool = ?` + args := []any{string(pool)} + if status != "" { + q += ` AND n.status = ?` + args = append(args, string(status)) + } + q += ` ORDER BY n.id` + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, fmt.Errorf("store.ListNodesByPool: %w", err) + } + defer rows.Close() + var out []*Node + for rows.Next() { + n, err := scanNode(rows) + if err != nil { + return nil, fmt.Errorf("store.ListNodesByPool scan: %w", err) + } + out = append(out, n) + } + return out, rows.Err() +} + +// --- providers --- + +func (s *MySQLStore) ListProviders(ctx context.Context, pool Pool) ([]*Provider, error) { + q := `SELECT id, name, api_kind, regions, pool, enabled FROM providers WHERE enabled=TRUE` + var args []any + if pool != "" { + q += ` AND pool=?` + args = append(args, string(pool)) + } + q += ` ORDER BY id` + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, fmt.Errorf("store.ListProviders: %w", err) + } + defer rows.Close() + var out []*Provider + for rows.Next() { + p, err := scanProvider(rows) + if err != nil { + return nil, fmt.Errorf("store.ListProviders scan: %w", err) + } + out = append(out, p) + } + return out, rows.Err() +} + +func (s *MySQLStore) GetProvider(ctx context.Context, id int64) (*Provider, error) { + row := s.db.QueryRowContext(ctx, + `SELECT id, name, api_kind, regions, pool, enabled FROM providers WHERE id=?`, id) + p, err := scanProvider(row) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("store.GetProvider: %w", err) + } + return p, nil +} + +func scanProvider(row interface{ Scan(...any) error }) (*Provider, error) { + var ( + p Provider + regions sql.NullString + ) + if err := row.Scan(&p.ID, &p.Name, &p.APIKind, ®ions, &p.Pool, &p.Enabled); err != nil { + return nil, err + } + p.Regions = unmarshalJSONList(regions) + return &p, nil +} + +// --- events / audit / directory --- + +func (s *MySQLStore) WriteNodeEvent(ctx context.Context, nodeID int64, event Event, detailJSON string) error { + if detailJSON == "" { + detailJSON = "null" + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO node_events (node_id, event, detail, at) VALUES (?, ?, ?, UTC_TIMESTAMP(6))`, + nodeID, string(event), detailJSON) + if err != nil { + return fmt.Errorf("store.WriteNodeEvent: %w", err) + } + return nil +} + +func (s *MySQLStore) WriteAuditLog(ctx context.Context, actor, action, target, metaJSON string) error { + if metaJSON == "" { + metaJSON = "null" + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO audit_log (actor, action, target, meta, at) VALUES (?, ?, ?, ?, UTC_TIMESTAMP(6))`, + actor, action, target, metaJSON) + if err != nil { + return fmt.Errorf("store.WriteAuditLog: %w", err) + } + return nil +} + +func (s *MySQLStore) BumpDirectoryVersion(ctx context.Context) (int64, error) { + if _, err := s.db.ExecContext(ctx, + `UPDATE directory_version SET version = version + 1 WHERE id = 1`); err != nil { + return 0, fmt.Errorf("store.BumpDirectoryVersion: %w", err) + } + var v int64 + if err := s.db.QueryRowContext(ctx, + `SELECT version FROM directory_version WHERE id = 1`).Scan(&v); err != nil { + return 0, fmt.Errorf("store.BumpDirectoryVersion read: %w", err) + } + return v, nil +} + +// --- idempotency --- + +func (s *MySQLStore) LookupIdempotency(ctx context.Context, key string) (string, bool, error) { + var uuid string + err := s.db.QueryRowContext(ctx, + `SELECT node_uuid FROM provision_idempotency WHERE idempotency_key=?`, key).Scan(&uuid) + if err == sql.ErrNoRows { + return "", false, nil + } + if err != nil { + return "", false, fmt.Errorf("store.LookupIdempotency: %w", err) + } + return uuid, true, nil +} + +func (s *MySQLStore) SaveIdempotency(ctx context.Context, key, nodeUUID string) error { + _, err := s.db.ExecContext(ctx, + `INSERT INTO provision_idempotency (idempotency_key, node_uuid, created_at) + VALUES (?, ?, UTC_TIMESTAMP(6)) + ON DUPLICATE KEY UPDATE idempotency_key = idempotency_key`, + key, nodeUUID) + if err != nil { + return fmt.Errorf("store.SaveIdempotency: %w", err) + } + return nil +} + +// --- replacements --- + +func (s *MySQLStore) CreateReplacement(ctx context.Context, r *Replacement) error { + _, err := s.db.ExecContext(ctx, + `INSERT INTO replacements (uuid, old_node_id, new_node_id, pool, status, step) + VALUES (?, ?, ?, ?, ?, ?)`, + r.UUID, r.OldNodeID, nullInt64(r.NewNodeID), string(r.Pool), + string(r.Status), string(r.Step)) + if err != nil { + return fmt.Errorf("store.CreateReplacement: %w", err) + } + return nil +} + +func (s *MySQLStore) GetReplacement(ctx context.Context, uuid string) (*Replacement, error) { + var ( + r Replacement + newID sql.NullInt64 + ) + err := s.db.QueryRowContext(ctx, + `SELECT uuid, old_node_id, new_node_id, pool, status, step, created_at, updated_at + FROM replacements WHERE uuid=?`, uuid). + Scan(&r.UUID, &r.OldNodeID, &newID, &r.Pool, &r.Status, &r.Step, &r.CreatedAt, &r.UpdatedAt) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("store.GetReplacement: %w", err) + } + r.NewNodeID = newID.Int64 + return &r, nil +} + +func (s *MySQLStore) UpdateReplacement(ctx context.Context, r *Replacement) error { + _, err := s.db.ExecContext(ctx, + `UPDATE replacements SET new_node_id=?, status=?, step=? WHERE uuid=?`, + nullInt64(r.NewNodeID), string(r.Status), string(r.Step), r.UUID) + if err != nil { + return fmt.Errorf("store.UpdateReplacement: %w", err) + } + return nil +} + +// --- helpers --- + +func nullInt(v int) interface{} { + if v == 0 { + return nil + } + return v +} + +func nullInt64(v int64) interface{} { + if v == 0 { + return nil + } + return v +} + +// qualify prefixes every comma-separated column in cols with the alias. +func qualify(cols, alias string) string { + parts := strings.Split(cols, ",") + for i, p := range parts { + parts[i] = alias + "." + strings.TrimSpace(p) + } + return strings.Join(parts, ", ") +} diff --git a/server/internal/provision/providers/doc.go b/server/internal/provision/providers/doc.go new file mode 100644 index 0000000..72897a9 --- /dev/null +++ b/server/internal/provision/providers/doc.go @@ -0,0 +1,16 @@ +// Package providers holds the concrete CloudAdapter implementations behind the +// provision.AdapterFactory boundary (doc/04 §4: "厂商 API 适配层"). +// +// First launch ships two vendors — one per pool (doc/04 §5.2): +// +// - vultr — consumable pool (entry/free): cheap, hourly-billed small vendor. +// - hetzner — premium pool (exit / pro entry): stable, good native IPs. +// +// CREDENTIAL RED LINE (doc/06 §2): adapters read their API credentials ONLY +// from independent secrets (environment variables here). Credentials are never +// stored in the providers table, never logged, and never committed to git. The +// Registry binds providers.api_kind → adapter at wiring time. +// +// Identity isolation: each vendor MUST use a fully independent account / email / +// crypto payment, registered in infra/identity-isolation.md. +package providers diff --git a/server/internal/provision/providers/hetzner.go b/server/internal/provision/providers/hetzner.go new file mode 100644 index 0000000..b3075b1 --- /dev/null +++ b/server/internal/provision/providers/hetzner.go @@ -0,0 +1,154 @@ +package providers + +import ( + "context" + "os" + "strconv" + "strings" + + "github.com/wangjia/pangolin/server/internal/provision" +) + +// hetznerAdapter implements provision.CloudAdapter against the Hetzner Cloud +// API v1. Premium pool: stable vendor, good native IPs, floating-IP capable. +// +// Credentials: PROVISION_HETZNER_API_TOKEN (env, independent secret). +// Default image: PROVISION_HETZNER_IMAGE (defaults to "debian-12"). +type hetznerAdapter struct { + c *httpClient + image string +} + +func newHetznerFromEnv() (provision.CloudAdapter, error) { + token, err := secret("PROVISION_HETZNER_API_TOKEN") + if err != nil { + return nil, err + } + image := os.Getenv("PROVISION_HETZNER_IMAGE") + if image == "" { + image = "debian-12" + } + return &hetznerAdapter{ + c: newHTTPClient("https://api.hetzner.cloud/v1", token), + image: image, + }, nil +} + +func (a *hetznerAdapter) Kind() string { return "hetzner" } +func (a *hetznerAdapter) SupportsElasticIP() bool { return true } // floating IPs + +func (a *hetznerAdapter) CreateInstance(ctx context.Context, in provision.CreateInput) (*provision.Instance, error) { + body := map[string]any{ + "name": sanitizeName(in.Label), + "server_type": in.Plan, + "image": a.image, + "location": in.Region, + "user_data": in.UserData, // Hetzner accepts raw cloud-init + "start_after_create": true, + "labels": labelsFromTags(in.Tags), + } + if len(in.SSHKeyIDs) > 0 { + body["ssh_keys"] = in.SSHKeyIDs + } + var resp struct { + Server struct { + ID int64 `json:"id"` + PublicNet struct { + IPv4 struct { + IP string `json:"ip"` + } `json:"ipv4"` + } `json:"public_net"` + Datacenter struct { + Location struct { + Name string `json:"name"` + } `json:"location"` + } `json:"datacenter"` + } `json:"server"` + } + if err := a.c.do(ctx, "POST", "/servers", body, &resp); err != nil { + return nil, err + } + return &provision.Instance{ + ID: strconv.FormatInt(resp.Server.ID, 10), + IP: resp.Server.PublicNet.IPv4.IP, + Region: resp.Server.Datacenter.Location.Name, + }, nil +} + +func (a *hetznerAdapter) DestroyInstance(ctx context.Context, instanceID string) error { + err := a.c.do(ctx, "DELETE", "/servers/"+instanceID, nil, nil) + if err != nil && isNotFound(err) { + return nil // idempotent + } + return err +} + +func (a *hetznerAdapter) AttachIP(ctx context.Context, instanceID string) (string, error) { + id, err := strconv.ParseInt(instanceID, 10, 64) + if err != nil { + return "", err + } + var resp struct { + FloatingIP struct { + ID int64 `json:"id"` + IP string `json:"ip"` + } `json:"floating_ip"` + } + if err := a.c.do(ctx, "POST", "/floating_ips", map[string]any{ + "type": "ipv4", + "server": id, + "description": "rotate-" + instanceID, + }, &resp); err != nil { + return "", err + } + return resp.FloatingIP.IP, nil +} + +func (a *hetznerAdapter) ListRegions(ctx context.Context) ([]provision.Region, error) { + var resp struct { + Locations []struct { + Name string `json:"name"` + Country string `json:"country"` + City string `json:"city"` + } `json:"locations"` + } + if err := a.c.do(ctx, "GET", "/locations", nil, &resp); err != nil { + return nil, err + } + out := make([]provision.Region, 0, len(resp.Locations)) + for _, l := range resp.Locations { + out = append(out, provision.Region{ID: l.Name, Country: l.Country, City: l.City}) + } + return out, nil +} + +// --- helpers shared with vultr --- + +func atoiDefault(s string, def int) int { + if s == "" { + return def + } + n, err := strconv.Atoi(s) + if err != nil { + return def + } + return n +} + +func isNotFound(err error) bool { + return err != nil && (strings.Contains(err.Error(), "status 404") || strings.Contains(err.Error(), "not_found")) +} + +// sanitizeName makes a UUID acceptable as a Hetzner server name (RFC1123-ish). +func sanitizeName(s string) string { + s = strings.ToLower(s) + return "node-" + s +} + +func labelsFromTags(tags []string) map[string]string { + m := map[string]string{} + for i, t := range tags { + m["tag"+strconv.Itoa(i)] = t + } + return m +} diff --git a/server/internal/provision/providers/registry.go b/server/internal/provision/providers/registry.go new file mode 100644 index 0000000..07df892 --- /dev/null +++ b/server/internal/provision/providers/registry.go @@ -0,0 +1,121 @@ +package providers + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "time" + + "github.com/wangjia/pangolin/server/internal/provision" +) + +// Factory builds a CloudAdapter, reading credentials from independent secrets +// (env). It returns provision.ErrNoCredentials when the required secret is unset. +type Factory func() (provision.CloudAdapter, error) + +// builtins maps api_kind → Factory. Extend here when onboarding a vendor. +var builtins = map[string]Factory{ + "vultr": newVultrFromEnv, + "hetzner": newHetznerFromEnv, +} + +// Registry resolves provision.Provider rows to live adapters. It satisfies +// provision.AdapterFactory and caches one adapter per api_kind. +type Registry struct { + factories map[string]Factory + cache map[string]provision.CloudAdapter +} + +// NewRegistry returns a Registry backed by the built-in vendor factories. +func NewRegistry() *Registry { + fs := make(map[string]Factory, len(builtins)) + for k, v := range builtins { + fs[k] = v + } + return &Registry{factories: fs, cache: map[string]provision.CloudAdapter{}} +} + +// Register adds or overrides a factory for api_kind (used in tests / extension). +func (r *Registry) Register(apiKind string, f Factory) { r.factories[apiKind] = f } + +// For resolves the adapter for a provider row. +func (r *Registry) For(p *provision.Provider) (provision.CloudAdapter, error) { + if a, ok := r.cache[p.APIKind]; ok { + return a, nil + } + f, ok := r.factories[p.APIKind] + if !ok { + return nil, fmt.Errorf("providers: no adapter registered for api_kind %q", p.APIKind) + } + a, err := f() + if err != nil { + return nil, err + } + r.cache[p.APIKind] = a + return a, nil +} + +var _ provision.AdapterFactory = (*Registry)(nil) + +// --- shared HTTP helper --- + +// httpClient is a small JSON REST helper shared by the adapters. +type httpClient struct { + base string + bearer string + hc *http.Client +} + +func newHTTPClient(base, bearer string) *httpClient { + return &httpClient{base: base, bearer: bearer, hc: &http.Client{Timeout: 30 * time.Second}} +} + +// do issues an authenticated JSON request and decodes the response into out +// (out may be nil). It returns an error on any non-2xx status. +func (c *httpClient) do(ctx context.Context, method, path string, body, out any) error { + var rdr io.Reader + if body != nil { + buf, err := json.Marshal(body) + if err != nil { + return fmt.Errorf("providers: marshal body: %w", err) + } + rdr = bytes.NewReader(buf) + } + req, err := http.NewRequestWithContext(ctx, method, c.base+path, rdr) + if err != nil { + return fmt.Errorf("providers: build request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+c.bearer) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + resp, err := c.hc.Do(req) + if err != nil { + return fmt.Errorf("providers: %s %s: %w", method, path, err) + } + defer resp.Body.Close() + data, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + // Never echo credentials; only status + vendor body (no auth header). + return fmt.Errorf("providers: %s %s: status %d: %s", method, path, resp.StatusCode, string(data)) + } + if out != nil && len(data) > 0 { + if err := json.Unmarshal(data, out); err != nil { + return fmt.Errorf("providers: decode response: %w", err) + } + } + return nil +} + +// secret reads an env-injected credential, returning ErrNoCredentials if unset. +func secret(env string) (string, error) { + v := os.Getenv(env) + if v == "" { + return "", fmt.Errorf("%w (env %s)", provision.ErrNoCredentials, env) + } + return v, nil +} diff --git a/server/internal/provision/providers/registry_test.go b/server/internal/provision/providers/registry_test.go new file mode 100644 index 0000000..c5380a7 --- /dev/null +++ b/server/internal/provision/providers/registry_test.go @@ -0,0 +1,76 @@ +package providers + +import ( + "context" + "errors" + "testing" + + "github.com/wangjia/pangolin/server/internal/provision" +) + +func TestRegistry_UnknownKind(t *testing.T) { + r := NewRegistry() + _, err := r.For(&provision.Provider{APIKind: "does-not-exist"}) + if err == nil { + t.Fatal("expected error for unknown api_kind") + } +} + +func TestRegistry_NoCredentials(t *testing.T) { + // Ensure the env vars are unset for a clean assertion. + t.Setenv("PROVISION_VULTR_API_KEY", "") + t.Setenv("PROVISION_HETZNER_API_TOKEN", "") + + r := NewRegistry() + if _, err := r.For(&provision.Provider{APIKind: "vultr"}); !errors.Is(err, provision.ErrNoCredentials) { + t.Errorf("vultr without creds: err = %v, want ErrNoCredentials", err) + } + if _, err := r.For(&provision.Provider{APIKind: "hetzner"}); !errors.Is(err, provision.ErrNoCredentials) { + t.Errorf("hetzner without creds: err = %v, want ErrNoCredentials", err) + } +} + +func TestRegistry_RegisterAndCache(t *testing.T) { + r := NewRegistry() + built := 0 + r.Register("fake", func() (provision.CloudAdapter, error) { + built++ + return stubAdapter{}, nil + }) + p := &provision.Provider{APIKind: "fake"} + if _, err := r.For(p); err != nil { + t.Fatalf("For: %v", err) + } + if _, err := r.For(p); err != nil { + t.Fatalf("For (cached): %v", err) + } + if built != 1 { + t.Errorf("factory built %d times, want 1 (cached)", built) + } +} + +func TestRegistry_CredentialsFromEnv(t *testing.T) { + t.Setenv("PROVISION_VULTR_API_KEY", "secret-key") + r := NewRegistry() + a, err := r.For(&provision.Provider{APIKind: "vultr"}) + if err != nil { + t.Fatalf("vultr with creds: %v", err) + } + if a.Kind() != "vultr" { + t.Errorf("kind = %s, want vultr", a.Kind()) + } + if !a.SupportsElasticIP() { + t.Error("vultr should support elastic IP (reserved IPs)") + } +} + +type stubAdapter struct{} + +func (stubAdapter) Kind() string { return "fake" } +func (stubAdapter) SupportsElasticIP() bool { return false } +func (stubAdapter) CreateInstance(context.Context, provision.CreateInput) (*provision.Instance, error) { + return &provision.Instance{ID: "x"}, nil +} +func (stubAdapter) DestroyInstance(context.Context, string) error { return nil } +func (stubAdapter) AttachIP(context.Context, string) (string, error) { return "", nil } +func (stubAdapter) ListRegions(context.Context) ([]provision.Region, error) { return nil, nil } diff --git a/server/internal/provision/providers/vultr.go b/server/internal/provision/providers/vultr.go new file mode 100644 index 0000000..fed0337 --- /dev/null +++ b/server/internal/provision/providers/vultr.go @@ -0,0 +1,110 @@ +package providers + +import ( + "context" + "encoding/base64" + "os" + + "github.com/wangjia/pangolin/server/internal/provision" +) + +// vultrAdapter implements provision.CloudAdapter against the Vultr API v2. +// Consumable pool: cheap, hourly-billed, reserved-IP capable. +// +// Credentials: PROVISION_VULTR_API_KEY (env, independent secret). +// Default OS image: PROVISION_VULTR_OS_ID (defaults to a current Debian image). +type vultrAdapter struct { + c *httpClient + osID int +} + +func newVultrFromEnv() (provision.CloudAdapter, error) { + key, err := secret("PROVISION_VULTR_API_KEY") + if err != nil { + return nil, err + } + osID := atoiDefault(os.Getenv("PROVISION_VULTR_OS_ID"), 2136) // 2136 = Debian 12 x64 + return &vultrAdapter{ + c: newHTTPClient("https://api.vultr.com/v2", key), + osID: osID, + }, nil +} + +func (a *vultrAdapter) Kind() string { return "vultr" } +func (a *vultrAdapter) SupportsElasticIP() bool { return true } // reserved IPs + +func (a *vultrAdapter) CreateInstance(ctx context.Context, in provision.CreateInput) (*provision.Instance, error) { + body := map[string]any{ + "region": in.Region, + "plan": in.Plan, + "os_id": a.osID, + "label": in.Label, + "hostname": in.Label, + "user_data": base64.StdEncoding.EncodeToString([]byte(in.UserData)), + "tags": in.Tags, + "backups": "disabled", + } + if len(in.SSHKeyIDs) > 0 { + body["sshkey_id"] = in.SSHKeyIDs + } + var resp struct { + Instance struct { + ID string `json:"id"` + MainIP string `json:"main_ip"` + Region string `json:"region"` + } `json:"instance"` + } + if err := a.c.do(ctx, "POST", "/instances", body, &resp); err != nil { + return nil, err + } + return &provision.Instance{ + ID: resp.Instance.ID, + IP: resp.Instance.MainIP, + Region: resp.Instance.Region, + }, nil +} + +func (a *vultrAdapter) DestroyInstance(ctx context.Context, instanceID string) error { + err := a.c.do(ctx, "DELETE", "/instances/"+instanceID, nil, nil) + if err != nil && isNotFound(err) { + return nil // idempotent: already gone + } + return err +} + +func (a *vultrAdapter) AttachIP(ctx context.Context, instanceID string) (string, error) { + // Allocate a reserved IPv4, then attach it to the instance. + var created struct { + ReservedIP struct { + ID string `json:"id"` + Subnet string `json:"subnet"` + } `json:"reserved_ip"` + } + if err := a.c.do(ctx, "POST", "/reserved-ips", map[string]any{ + "region": "", + "ip_type": "v4", + "label": "rotate-" + instanceID, + "instance_id": instanceID, + }, &created); err != nil { + return "", err + } + return created.ReservedIP.Subnet, nil +} + +func (a *vultrAdapter) ListRegions(ctx context.Context) ([]provision.Region, error) { + var resp struct { + Regions []struct { + ID string `json:"id"` + Country string `json:"country"` + City string `json:"city"` + } `json:"regions"` + } + if err := a.c.do(ctx, "GET", "/regions", nil, &resp); err != nil { + return nil, err + } + out := make([]provision.Region, 0, len(resp.Regions)) + for _, r := range resp.Regions { + out = append(out, provision.Region{ID: r.ID, Country: r.Country, City: r.City}) + } + return out, nil +} diff --git a/server/internal/provision/replace.go b/server/internal/provision/replace.go new file mode 100644 index 0000000..f049357 --- /dev/null +++ b/server/internal/provision/replace.go @@ -0,0 +1,312 @@ +package provision + +import ( + "context" + "fmt" + "sync" + "time" +) + +// stepRank orders ReplaceStep so resume logic can compare progress. +var stepRank = map[ReplaceStep]int{ + StepOpenNew: 0, + StepProbing: 1, + StepNewUp: 2, + StepDraining: 3, + StepDestroyOld: 4, + StepDone: 5, +} + +// ReplaceResult reports the outcome of one Replace. +type ReplaceResult struct { + ReplacementUUID string + OldNodeID int64 + NewNodeID int64 + NewNode *Node +} + +// Replace performs a make-before-break one-click replacement of nodeID +// (doc/04 §4.1). A fresh node is brought UP before the old one is drained and +// destroyed, so capacity never dips. +// +// replacementUUID is the idempotency key for the whole orchestration. Pass "" +// to start a new replacement; pass an existing UUID to RESUME after a crash — +// progress is persisted in the replacements table and CreateNode's own +// idempotency guarantees no duplicate boot (validation: "崩溃重启续跑不重复"). +func (s *Service) Replace(ctx context.Context, nodeID int64, replacementUUID string) (*ReplaceResult, error) { + old, err := s.store.GetNode(ctx, nodeID) + if err != nil { + return nil, err + } + if old == nil { + return nil, fmt.Errorf("provision: node %d not found", nodeID) + } + pool := poolForTier(old.Tier) + + // Load or create the orchestration record. + var rec *Replacement + if replacementUUID != "" { + rec, err = s.store.GetReplacement(ctx, replacementUUID) + if err != nil { + return nil, err + } + } + if rec == nil { + if replacementUUID == "" { + if replacementUUID, err = newUUID(); err != nil { + return nil, err + } + } + rec = &Replacement{ + UUID: replacementUUID, + OldNodeID: nodeID, + Pool: pool, + Status: ReplaceRunning, + Step: StepOpenNew, + } + if err := s.store.CreateReplacement(ctx, rec); err != nil { + return nil, err + } + } + if rec.Status == ReplaceDone { + // Already finished; return the recorded result. + newNode, _ := s.store.GetNode(ctx, rec.NewNodeID) + return &ReplaceResult{ReplacementUUID: rec.UUID, OldNodeID: rec.OldNodeID, NewNodeID: rec.NewNodeID, NewNode: newNode}, nil + } + + return s.runReplace(ctx, rec, old, pool) +} + +// runReplace drives the replacement state machine forward from rec.Step, +// persisting after each transition so a crash resumes cleanly. +func (s *Service) runReplace(ctx context.Context, rec *Replacement, old *Node, pool Pool) (*ReplaceResult, error) { + atLeast := func(step ReplaceStep) bool { return stepRank[rec.Step] >= stepRank[step] } + advance := func(step ReplaceStep) error { + rec.Step = step + return s.store.UpdateReplacement(ctx, rec) + } + + var newNode *Node + + // Step 1: open the new node (same region, pool may switch vendor). + if !atLeast(StepProbing) { + spec, err := s.replacementSpec(ctx, old, pool) + if err != nil { + return s.failReplace(ctx, rec, old, pool, err) + } + nn, err := s.CreateNode(ctx, spec, rec.UUID+":create") + if err != nil { + return s.failReplace(ctx, rec, old, pool, fmt.Errorf("open new node: %w", err)) + } + newNode = nn + rec.NewNodeID = nn.ID + if err := advance(StepProbing); err != nil { + return nil, err + } + } + if newNode == nil && rec.NewNodeID != 0 { + if newNode, _ = s.store.GetNode(ctx, rec.NewNodeID); newNode == nil { + return s.failReplace(ctx, rec, old, pool, fmt.Errorf("new node %d vanished", rec.NewNodeID)) + } + } + + // Step 2: wait for agent self-register + simplified probing. + if !atLeast(StepNewUp) { + if s.prober != nil { + pctx, cancel := context.WithTimeout(ctx, s.probeTimeout) + err := s.prober.WaitReady(pctx, newNode) + cancel() + if err != nil { + _ = s.store.WriteNodeEvent(ctx, newNode.ID, EventProbeFail, jsonDetail(map[string]any{"error": err.Error()})) + // Probe timeout: destroy the failed new node, count + alert. + _ = s.DestroyNode(ctx, newNode.ID) + count := s.bumpFail(pool) + s.alert.Fire(ctx, Alert{Kind: AlertProbeTimeout, NodeUUID: newNode.UUID, Pool: pool, Message: err.Error(), FailCount: count}) + rec.Status = ReplaceFailed + _ = s.store.UpdateReplacement(ctx, rec) + return nil, fmt.Errorf("provision: replace probe failed: %w", err) + } + } + _ = s.store.WriteNodeEvent(ctx, newNode.ID, EventProbePass, "") + if err := advance(StepNewUp); err != nil { + return nil, err + } + } + + // Step 3: promote the new node UP and publish (capacity is now restored + // BEFORE the old node leaves the directory — make-before-break). + if !atLeast(StepDraining) { + if err := s.store.UpdateNodeStatus(ctx, newNode.ID, StatusUp); err != nil { + return nil, err + } + _ = s.store.WriteNodeEvent(ctx, newNode.ID, EventMarkedUp, "") + if _, err := s.store.BumpDirectoryVersion(ctx); err != nil { + return nil, err + } + s.resetFail(pool) + if err := advance(StepDraining); err != nil { + return nil, err + } + } + + // Step 4: drain the old node. Free/consumable nodes hard-cut. + if !atLeast(StepDestroyOld) { + if err := s.store.UpdateNodeStatus(ctx, old.ID, StatusDraining); err != nil { + return nil, err + } + _ = s.store.WriteNodeEvent(ctx, old.ID, EventDraining, "") + if _, err := s.store.BumpDirectoryVersion(ctx); err != nil { + return nil, err + } + if pool != PoolConsumable { + if err := s.clock.Sleep(ctx, s.drainTimeout); err != nil { + return nil, err + } + } + if err := advance(StepDestroyOld); err != nil { + return nil, err + } + } + + // Step 5: destroy the old node + release IP. + if !atLeast(StepDone) { + if err := s.DestroyNode(ctx, old.ID); err != nil { + return s.failReplace(ctx, rec, old, pool, fmt.Errorf("destroy old node: %w", err)) + } + _ = s.store.WriteNodeEvent(ctx, old.ID, EventReplaced, jsonDetail(map[string]any{ + "replacement_uuid": rec.UUID, + "new_node_id": rec.NewNodeID, + })) + if err := advance(StepDone); err != nil { + return nil, err + } + } + + rec.Status = ReplaceDone + if err := s.store.UpdateReplacement(ctx, rec); err != nil { + return nil, err + } + _ = s.store.WriteAuditLog(ctx, "provision", "replace", "node:"+old.UUID, + jsonDetail(map[string]any{"replacement_uuid": rec.UUID, "new_node_id": rec.NewNodeID})) + + if newNode == nil && rec.NewNodeID != 0 { + newNode, _ = s.store.GetNode(ctx, rec.NewNodeID) + } + return &ReplaceResult{ + ReplacementUUID: rec.UUID, + OldNodeID: rec.OldNodeID, + NewNodeID: rec.NewNodeID, + NewNode: newNode, + }, nil +} + +// failReplace records a fatal replacement failure (alert + audit). +func (s *Service) failReplace(ctx context.Context, rec *Replacement, old *Node, pool Pool, cause error) (*ReplaceResult, error) { + rec.Status = ReplaceFailed + _ = s.store.UpdateReplacement(ctx, rec) + count := s.bumpFail(pool) + _ = s.store.WriteAuditLog(ctx, "provision", "replace_failed", "node:"+old.UUID, + jsonDetail(map[string]any{"replacement_uuid": rec.UUID, "error": cause.Error(), "fail_count": count})) + s.alert.Fire(ctx, Alert{Kind: AlertReplaceFail, NodeUUID: old.UUID, Pool: pool, Message: cause.Error(), FailCount: count}) + return nil, fmt.Errorf("provision: replace failed: %w", cause) +} + +// replacementSpec derives the spec for the new node from the old one, keeping +// the same region/tier/role but allowing a different vendor in the same pool +// (doc/04 §4.1: "同 region · 厂商池内可换家"). +func (s *Service) replacementSpec(ctx context.Context, old *Node, pool Pool) (NodeSpec, error) { + providerID := old.ProviderID + providers, err := s.store.ListProviders(ctx, pool) + if err != nil { + return NodeSpec{}, err + } + // Prefer a different enabled provider in the pool to spread exposure. + for _, p := range providers { + if p.ID != old.ProviderID && providerSupportsRegion(p, old.Region) { + providerID = p.ID + break + } + } + return NodeSpec{ + Region: old.Region, + Role: old.Role, + Tier: old.Tier, + ProviderID: providerID, + NameZH: old.NameZH, + NameEn: old.NameEn, + RealityPBK: old.RealityPBK, + RealitySNI: old.RealitySNI, + HY2Port: old.HY2Port, + Weight: old.Weight, + Tags: old.Tags, + }, nil +} + +func providerSupportsRegion(p *Provider, region string) bool { + if len(p.Regions) == 0 { + return true // unconstrained + } + for _, r := range p.Regions { + if r == region { + return true + } + } + return false +} + +// RotatePool rolls Replace across every up node in a pool with bounded +// concurrency (doc/04 §4: 并发度 1–2). Used for routine rotation or large-scale +// event rebuilds. +func (s *Service) RotatePool(ctx context.Context, pool Pool, concurrency int) ([]ReplaceResult, error) { + if concurrency < 1 { + concurrency = 1 + } + if concurrency > 2 { + concurrency = 2 + } + nodes, err := s.store.ListNodesByPool(ctx, pool, StatusUp) + if err != nil { + return nil, err + } + + var ( + mu sync.Mutex + results []ReplaceResult + firstEr error + wg sync.WaitGroup + ) + sem := make(chan struct{}, concurrency) + for _, n := range nodes { + n := n + if firstErrSet(&mu, &firstEr) { + break + } + wg.Add(1) + sem <- struct{}{} + go func() { + defer wg.Done() + defer func() { <-sem }() + res, err := s.Replace(ctx, n.ID, "") + mu.Lock() + defer mu.Unlock() + if err != nil { + if firstEr == nil { + firstEr = err + } + return + } + results = append(results, *res) + }() + } + wg.Wait() + return results, firstEr +} + +func firstErrSet(mu *sync.Mutex, e *error) bool { + mu.Lock() + defer mu.Unlock() + return *e != nil +} + +// drainDeadline is exposed for tests/monitoring of the configured drain window. +func (s *Service) drainDeadline(start time.Time) time.Time { return start.Add(s.drainTimeout) } diff --git a/server/internal/provision/replace_test.go b/server/internal/provision/replace_test.go new file mode 100644 index 0000000..17dc773 --- /dev/null +++ b/server/internal/provision/replace_test.go @@ -0,0 +1,213 @@ +package provision + +import ( + "context" + "errors" + "testing" +) + +// TestReplace_MakeBeforeBreak asserts the full state-machine migration and that +// capacity never dips: the new node reaches `up` BEFORE the old node leaves the +// directory (enters `draining`). +func TestReplace_MakeBeforeBreak(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + + old, err := h.svc.CreateNode(ctx, proSpec(), "old-node") + if err != nil { + t.Fatalf("CreateNode old: %v", err) + } + if err := h.store.UpdateNodeStatus(ctx, old.ID, StatusUp); err != nil { + t.Fatalf("seed up: %v", err) + } + bootsBefore := h.adapter.createCount() + + res, err := h.svc.Replace(ctx, old.ID, "") + if err != nil { + t.Fatalf("Replace: %v", err) + } + if res.NewNodeID == 0 || res.NewNodeID == old.ID { + t.Fatalf("bad new node id: %d", res.NewNodeID) + } + if h.adapter.createCount() != bootsBefore+1 { + t.Errorf("expected exactly one new boot, got delta %d", h.adapter.createCount()-bootsBefore) + } + + // Capacity invariant: new node `up` seq < old node `draining` seq. + newUpSeq := h.store.firstSeqForStatus(res.NewNodeID, StatusUp) + oldDrainSeq := h.store.firstSeqForStatus(old.ID, StatusDraining) + if newUpSeq < 0 { + t.Fatal("new node never reached up") + } + if oldDrainSeq < 0 { + t.Fatal("old node never drained") + } + if !(newUpSeq < oldDrainSeq) { + t.Errorf("capacity dipped: new up seq %d not before old draining seq %d", newUpSeq, oldDrainSeq) + } + + // Old node finally destroyed. + oldNode, _ := h.store.GetNode(ctx, old.ID) + if oldNode.Status != StatusDestroyed { + t.Errorf("old node status = %s, want destroyed", oldNode.Status) + } + newNode, _ := h.store.GetNode(ctx, res.NewNodeID) + if newNode.Status != StatusUp { + t.Errorf("new node status = %s, want up", newNode.Status) + } + // pro pool drains on a timer. + if len(h.clock.slept) == 0 { + t.Error("expected a drain sleep for premium pool") + } +} + +// TestReplace_FreeHardCut: consumable/free pool drains by hard cut (no sleep). +func TestReplace_FreeHardCut(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + old, _ := h.svc.CreateNode(ctx, freeSpec(), "old-free") + _ = h.store.UpdateNodeStatus(ctx, old.ID, StatusUp) + + if _, err := h.svc.Replace(ctx, old.ID, ""); err != nil { + t.Fatalf("Replace: %v", err) + } + if len(h.clock.slept) != 0 { + t.Errorf("free pool should hard-cut, but slept %v", h.clock.slept) + } +} + +// TestReplace_ProbeTimeout: probing fails → new node destroyed, fail counted, +// alert fired, replacement marked failed, old node untouched. +func TestReplace_ProbeTimeout(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + h.prober.err = errors.New("probe timeout") + + old, _ := h.svc.CreateNode(ctx, proSpec(), "old-pt") + _ = h.store.UpdateNodeStatus(ctx, old.ID, StatusUp) + + _, err := h.svc.Replace(ctx, old.ID, "") + if err == nil { + t.Fatal("expected probe timeout error") + } + if h.svc.FailCount(PoolPremium) != 1 { + t.Errorf("fail count = %d, want 1", h.svc.FailCount(PoolPremium)) + } + if h.alert.count(AlertProbeTimeout) != 1 { + t.Errorf("probe_timeout alerts = %d, want 1", h.alert.count(AlertProbeTimeout)) + } + // Old node must still be up (capacity protected — we did not drain it). + oldNode, _ := h.store.GetNode(ctx, old.ID) + if oldNode.Status != StatusUp { + t.Errorf("old node status = %s, want up (untouched)", oldNode.Status) + } + // The failed new node should be destroyed (bad IP not left running). + if h.adapter.destroyCount() != 1 { + t.Errorf("failed new node not destroyed: destroy count %d", h.adapter.destroyCount()) + } +} + +// TestReplace_CrashResume: a replacement that crashed after booting the new node +// (idempotency key already saved) must resume WITHOUT a second boot. +func TestReplace_CrashResume(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + + old, _ := h.svc.CreateNode(ctx, freeSpec(), "old-cr") + _ = h.store.UpdateNodeStatus(ctx, old.ID, StatusUp) + + ruuid := "fixed-replacement-uuid" + // Simulate the pre-crash state: the new node was already booted under the + // orchestration's idempotency key, and a running replacement record exists + // at the open_new step (new_node_id not yet persisted). + spec, err := h.svc.replacementSpec(ctx, mustNode(t, h, old.ID), PoolConsumable) + if err != nil { + t.Fatalf("replacementSpec: %v", err) + } + preNode, err := h.svc.CreateNode(ctx, spec, ruuid+":create") + if err != nil { + t.Fatalf("pre-boot new node: %v", err) + } + if err := h.store.CreateReplacement(ctx, &Replacement{ + UUID: ruuid, OldNodeID: old.ID, Pool: PoolConsumable, + Status: ReplaceRunning, Step: StepOpenNew, + }); err != nil { + t.Fatalf("seed replacement: %v", err) + } + bootsBefore := h.adapter.createCount() // old + pre-booted new = 2 + + res, err := h.svc.Replace(ctx, old.ID, ruuid) + if err != nil { + t.Fatalf("resume Replace: %v", err) + } + if h.adapter.createCount() != bootsBefore { + t.Errorf("resume re-booted a node: delta %d (want 0)", h.adapter.createCount()-bootsBefore) + } + if res.NewNodeID != preNode.ID { + t.Errorf("resume used a different new node: got %d, want %d", res.NewNodeID, preNode.ID) + } + oldNode, _ := h.store.GetNode(ctx, old.ID) + if oldNode.Status != StatusDestroyed { + t.Errorf("old node not destroyed after resume: %s", oldNode.Status) + } +} + +// TestReplace_ReplayCompleted: replaying a finished replacement is a no-op. +func TestReplace_ReplayCompleted(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + old, _ := h.svc.CreateNode(ctx, freeSpec(), "old-rc") + _ = h.store.UpdateNodeStatus(ctx, old.ID, StatusUp) + + res1, err := h.svc.Replace(ctx, old.ID, "") + if err != nil { + t.Fatalf("Replace: %v", err) + } + boots := h.adapter.createCount() + destroys := h.adapter.destroyCount() + + res2, err := h.svc.Replace(ctx, old.ID, res1.ReplacementUUID) + if err != nil { + t.Fatalf("replay Replace: %v", err) + } + if res2.NewNodeID != res1.NewNodeID { + t.Errorf("replay new node id mismatch: %d vs %d", res2.NewNodeID, res1.NewNodeID) + } + if h.adapter.createCount() != boots || h.adapter.destroyCount() != destroys { + t.Errorf("replay caused vendor side effects: boots %d->%d destroys %d->%d", + boots, h.adapter.createCount(), destroys, h.adapter.destroyCount()) + } +} + +// TestRotatePool replaces every up node in a pool with bounded concurrency. +func TestRotatePool(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + a, _ := h.svc.CreateNode(ctx, freeSpec(), "rp-a") + b, _ := h.svc.CreateNode(ctx, freeSpec(), "rp-b") + _ = h.store.UpdateNodeStatus(ctx, a.ID, StatusUp) + _ = h.store.UpdateNodeStatus(ctx, b.ID, StatusUp) + + results, err := h.svc.RotatePool(ctx, PoolConsumable, 2) + if err != nil { + t.Fatalf("RotatePool: %v", err) + } + if len(results) != 2 { + t.Fatalf("results = %d, want 2", len(results)) + } + for _, id := range []int64{a.ID, b.ID} { + n, _ := h.store.GetNode(ctx, id) + if n.Status != StatusDestroyed { + t.Errorf("node %d status = %s, want destroyed", id, n.Status) + } + } +} + +func mustNode(t *testing.T, h *harness, id int64) *Node { + t.Helper() + n, err := h.store.GetNode(context.Background(), id) + if err != nil || n == nil { + t.Fatalf("get node %d: %v", id, err) + } + return n +} diff --git a/server/internal/provision/service.go b/server/internal/provision/service.go new file mode 100644 index 0000000..0d29687 --- /dev/null +++ b/server/internal/provision/service.go @@ -0,0 +1,384 @@ +package provision + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "time" +) + +// Config wires a Service together. Only Store and Adapters are mandatory; +// everything else falls back to a safe default. +type Config struct { + Store Store + Adapters AdapterFactory + Tokens BootstrapIssuer + Renderer CloudInitRenderer + Prober Prober + Alert AlertHook + Clock Clock + + // ControlPlaneURL is injected into cloud-init so the agent knows where to + // enroll (task #6). + ControlPlaneURL string + // DrainTimeout is how long a draining node is given before forced destroy + // (doc/04 §3: default 30 minutes). + DrainTimeout time.Duration + // ProbeTimeout bounds the simplified probing wait during Replace. + ProbeTimeout time.Duration +} + +// Service is the ProvisionService. All operations are idempotent and write +// node_events + audit_log (doc/04 §4). +type Service struct { + store Store + adapters AdapterFactory + tokens BootstrapIssuer + renderer CloudInitRenderer + prober Prober + alert AlertHook + clock Clock + controlPlaneURL string + drainTimeout time.Duration + probeTimeout time.Duration + + mu sync.Mutex + failPool map[Pool]int // consecutive boot/probe failures per pool +} + +// NewService builds a Service, applying defaults for optional dependencies. +func NewService(cfg Config) (*Service, error) { + if cfg.Store == nil { + return nil, fmt.Errorf("provision: Store is required") + } + if cfg.Adapters == nil { + return nil, fmt.Errorf("provision: Adapters factory is required") + } + s := &Service{ + store: cfg.Store, + adapters: cfg.Adapters, + tokens: cfg.Tokens, + renderer: cfg.Renderer, + prober: cfg.Prober, + alert: cfg.Alert, + clock: cfg.Clock, + controlPlaneURL: cfg.ControlPlaneURL, + drainTimeout: cfg.DrainTimeout, + probeTimeout: cfg.ProbeTimeout, + failPool: make(map[Pool]int), + } + if s.alert == nil { + s.alert = noopAlert{} + } + if s.clock == nil { + s.clock = realClock{} + } + if s.drainTimeout == 0 { + s.drainTimeout = 30 * time.Minute + } + if s.probeTimeout == 0 { + s.probeTimeout = 5 * time.Minute + } + return s, nil +} + +// resolveAdapter loads the provider row and its adapter. +func (s *Service) resolveAdapter(ctx context.Context, providerID int64) (*Provider, CloudAdapter, error) { + p, err := s.store.GetProvider(ctx, providerID) + if err != nil { + return nil, nil, err + } + if p == nil { + return nil, nil, fmt.Errorf("provision: provider %d not found", providerID) + } + a, err := s.adapters.For(p) + if err != nil { + return nil, nil, fmt.Errorf("provision: resolve adapter for %s: %w", p.APIKind, err) + } + return p, a, nil +} + +// CreateNode provisions one node. It is idempotent on idempotencyKey: replaying +// the same key returns the already-created node without booting a second +// machine (validation: "CreateNode 幂等键重放不重复开机"). +// +// Flow: idempotency check → bind key→uuid → insert nodes(provisioning) → +// issue bootstrap token (task #5) → render cloud-init (task #6) → vendor boot → +// record instance ID + endpoint → node_event(provisioned) + audit_log. +// The node stays in provisioning until the agent self-registers and probing +// promotes it to up. +func (s *Service) CreateNode(ctx context.Context, spec NodeSpec, idempotencyKey string) (*Node, error) { + if idempotencyKey == "" { + return nil, fmt.Errorf("provision: idempotencyKey is required") + } + + // 1. Idempotency replay. + if uuid, found, err := s.store.LookupIdempotency(ctx, idempotencyKey); err != nil { + return nil, err + } else if found { + return s.store.GetNodeByUUID(ctx, uuid) + } + + pool := poolForTier(spec.Tier) + + // 2. Allocate UUID and bind the idempotency key BEFORE any vendor call, so a + // crash after this point resumes onto the same node instead of booting a + // duplicate. + uuid, err := newUUID() + if err != nil { + return nil, err + } + if err := s.store.SaveIdempotency(ctx, idempotencyKey, uuid); err != nil { + return nil, err + } + // Re-read in case of a race: another caller may have won the key. + if winner, found, err := s.store.LookupIdempotency(ctx, idempotencyKey); err == nil && found && winner != uuid { + return s.store.GetNodeByUUID(ctx, winner) + } + + // 3. Insert the node in provisioning state. + n := &Node{ + UUID: uuid, + Region: spec.Region, + NameZH: spec.NameZH, + NameEn: spec.NameEn, + Role: orDefaultRole(spec.Role), + Tier: spec.Tier, + Endpoint: pendingEndpoint, + HY2Port: spec.HY2Port, + RealityPBK: spec.RealityPBK, + RealitySNI: spec.RealitySNI, + ProviderID: spec.ProviderID, + Tags: spec.Tags, + Status: StatusProvisioning, + Weight: spec.Weight, + } + id, err := s.store.InsertNode(ctx, n) + if err != nil { + return nil, err + } + n.ID = id + + // 4. Resolve adapter. + _, adapter, err := s.resolveAdapter(ctx, spec.ProviderID) + if err != nil { + s.failBoot(ctx, n, pool, err) + return nil, err + } + + // 5. Bootstrap token + cloud-init. + userData := "" + if s.renderer != nil { + token := "" + if s.tokens != nil { + token, err = s.tokens.IssueToken(ctx, uuid) + if err != nil { + s.failBoot(ctx, n, pool, fmt.Errorf("issue bootstrap token: %w", err)) + return nil, err + } + } + userData, err = s.renderer.Render(CloudInitData{ + NodeUUID: uuid, + BootstrapToken: token, + Region: spec.Region, + Role: n.Role, + Tier: spec.Tier, + ControlPlaneURL: s.controlPlaneURL, + }) + if err != nil { + s.failBoot(ctx, n, pool, fmt.Errorf("render cloud-init: %w", err)) + return nil, err + } + } + + // 6. Vendor boot. + inst, err := adapter.CreateInstance(ctx, CreateInput{ + Region: spec.Region, + Plan: spec.Plan, + Label: uuid, + UserData: userData, + Tags: spec.Tags, + }) + if err != nil { + s.failBoot(ctx, n, pool, fmt.Errorf("vendor boot: %w", err)) + return nil, err + } + + // 7. Record instance ID + endpoint. + endpoint := joinEndpoint(inst.IP, spec.HY2Port) + if err := s.store.SetNodeInstance(ctx, id, inst.ID, endpoint); err != nil { + return nil, err + } + n.ProviderInstanceID = inst.ID + n.Endpoint = endpoint + + // 8. Audit trail + reset failure counter. + _ = s.store.WriteNodeEvent(ctx, id, EventProvisioned, jsonDetail(map[string]any{ + "instance_id": inst.ID, + "region": inst.Region, + })) + _ = s.store.WriteAuditLog(ctx, "provision", "create_node", "node:"+uuid, + jsonDetail(map[string]any{"provider_id": spec.ProviderID, "region": spec.Region, "pool": pool})) + s.resetFail(pool) + + return n, nil +} + +// failBoot marks the node destroyed, bumps the per-pool failure counter, and +// fires a boot-failed alert (validation: "开机失败 → destroyed + 失败计数 + 告警钩子"). +func (s *Service) failBoot(ctx context.Context, n *Node, pool Pool, cause error) { + _ = s.store.UpdateNodeStatus(ctx, n.ID, StatusDestroyed) + _ = s.store.WriteNodeEvent(ctx, n.ID, EventDestroyed, jsonDetail(map[string]any{ + "reason": "boot_failed", + "error": cause.Error(), + })) + count := s.bumpFail(pool) + _ = s.store.WriteAuditLog(ctx, "provision", "create_node_failed", "node:"+n.UUID, + jsonDetail(map[string]any{"error": cause.Error(), "pool": pool, "fail_count": count})) + s.alert.Fire(ctx, Alert{ + Kind: AlertBootFailed, + NodeUUID: n.UUID, + Pool: pool, + Message: cause.Error(), + FailCount: count, + }) +} + +// DestroyNode tears down a node: vendor destroy → IP release → nodes→destroyed. +// Idempotent: destroying an already-destroyed node is a no-op success. +func (s *Service) DestroyNode(ctx context.Context, nodeID int64) error { + n, err := s.store.GetNode(ctx, nodeID) + if err != nil { + return err + } + if n == nil { + return fmt.Errorf("provision: node %d not found", nodeID) + } + if n.Status == StatusDestroyed { + return nil + } + if n.ProviderInstanceID != "" { + _, adapter, err := s.resolveAdapter(ctx, n.ProviderID) + if err != nil { + return err + } + if err := adapter.DestroyInstance(ctx, n.ProviderInstanceID); err != nil { + return fmt.Errorf("provision: destroy instance: %w", err) + } + } + if err := s.store.UpdateNodeStatus(ctx, nodeID, StatusDestroyed); err != nil { + return err + } + _ = s.store.WriteNodeEvent(ctx, nodeID, EventDestroyed, jsonDetail(map[string]any{"reason": "destroy"})) + _ = s.store.WriteAuditLog(ctx, "provision", "destroy_node", "node:"+n.UUID, "") + if _, err := s.store.BumpDirectoryVersion(ctx); err != nil { + return err + } + return nil +} + +// RotateIP swaps the elastic IP without re-creating the machine (doc/04 §4: +// "换 IP 不换机"). Flow: vendor IP re-bind → probe → update endpoint → +// bump directory version → node_event(ip_rotated). +func (s *Service) RotateIP(ctx context.Context, nodeID int64) (*Node, error) { + n, err := s.store.GetNode(ctx, nodeID) + if err != nil { + return nil, err + } + if n == nil { + return nil, fmt.Errorf("provision: node %d not found", nodeID) + } + _, adapter, err := s.resolveAdapter(ctx, n.ProviderID) + if err != nil { + return nil, err + } + if !adapter.SupportsElasticIP() { + return nil, ErrElasticIPUnsupported + } + oldEndpoint := n.Endpoint + newIP, err := adapter.AttachIP(ctx, n.ProviderInstanceID) + if err != nil { + return nil, fmt.Errorf("provision: attach IP: %w", err) + } + newEndpoint := joinEndpoint(newIP, n.HY2Port) + + // Probe the new endpoint before publishing it. + probed := *n + probed.Endpoint = newEndpoint + if s.prober != nil { + if err := s.prober.WaitReady(ctx, &probed); err != nil { + s.alert.Fire(ctx, Alert{Kind: AlertProbeTimeout, NodeUUID: n.UUID, Message: err.Error()}) + return nil, fmt.Errorf("provision: probe rotated IP: %w", err) + } + } + + if err := s.store.UpdateNodeEndpoint(ctx, nodeID, newEndpoint); err != nil { + return nil, err + } + version, err := s.store.BumpDirectoryVersion(ctx) + if err != nil { + return nil, err + } + _ = s.store.WriteNodeEvent(ctx, nodeID, EventIPRotated, jsonDetail(map[string]any{ + "old_endpoint": oldEndpoint, + "new_endpoint": newEndpoint, + "version": version, + })) + _ = s.store.WriteAuditLog(ctx, "provision", "rotate_ip", "node:"+n.UUID, + jsonDetail(map[string]any{"old": oldEndpoint, "new": newEndpoint})) + n.Endpoint = newEndpoint + return n, nil +} + +// ListProviders returns enabled providers in a pool (empty pool = all pools). +func (s *Service) ListProviders(ctx context.Context, pool Pool) ([]*Provider, error) { + return s.store.ListProviders(ctx, pool) +} + +// --- failure-counter helpers (capacity protection, doc/04 §4.2) --- + +func (s *Service) bumpFail(pool Pool) int { + s.mu.Lock() + defer s.mu.Unlock() + s.failPool[pool]++ + return s.failPool[pool] +} + +func (s *Service) resetFail(pool Pool) { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.failPool, pool) +} + +// FailCount exposes the current consecutive-failure count for a pool (tests / +// monitoring). +func (s *Service) FailCount(pool Pool) int { + s.mu.Lock() + defer s.mu.Unlock() + return s.failPool[pool] +} + +// --- small helpers --- + +func orDefaultRole(r Role) Role { + if r == "" { + return RoleEntry + } + return r +} + +func joinEndpoint(ip string, port int) string { + if port <= 0 { + port = 443 + } + return fmt.Sprintf("%s:%d", ip, port) +} + +func jsonDetail(m map[string]any) string { + b, err := json.Marshal(m) + if err != nil { + return "null" + } + return string(b) +} diff --git a/server/internal/provision/service_test.go b/server/internal/provision/service_test.go new file mode 100644 index 0000000..c2953d9 --- /dev/null +++ b/server/internal/provision/service_test.go @@ -0,0 +1,155 @@ +package provision + +import ( + "context" + "errors" + "testing" +) + +func TestCreateNode_Success(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + + n, err := h.svc.CreateNode(ctx, freeSpec(), "key-1") + if err != nil { + t.Fatalf("CreateNode: %v", err) + } + if n.Status != StatusProvisioning { + t.Errorf("status = %s, want provisioning", n.Status) + } + if n.ProviderInstanceID == "" { + t.Error("instance ID not recorded") + } + if n.Endpoint == pendingEndpoint || n.Endpoint == "" { + t.Errorf("endpoint not updated: %q", n.Endpoint) + } + if h.tokens.calls != 1 { + t.Errorf("bootstrap token issued %d times, want 1", h.tokens.calls) + } + if h.adapter.createCount() != 1 { + t.Errorf("boot calls = %d, want 1", h.adapter.createCount()) + } + if h.store.countEvents(EventProvisioned) != 1 { + t.Errorf("provisioned events = %d, want 1", h.store.countEvents(EventProvisioned)) + } +} + +func TestCreateNode_IdempotentReplay(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + + n1, err := h.svc.CreateNode(ctx, freeSpec(), "same-key") + if err != nil { + t.Fatalf("first CreateNode: %v", err) + } + n2, err := h.svc.CreateNode(ctx, freeSpec(), "same-key") + if err != nil { + t.Fatalf("replay CreateNode: %v", err) + } + if n1.UUID != n2.UUID { + t.Errorf("replay returned different node: %s vs %s", n1.UUID, n2.UUID) + } + if h.adapter.createCount() != 1 { + t.Errorf("idempotency violated: boot called %d times, want 1", h.adapter.createCount()) + } +} + +func TestCreateNode_BootFailure(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + h.adapter.createErr = errors.New("vendor 500") + + _, err := h.svc.CreateNode(ctx, freeSpec(), "key-boom") + if err == nil { + t.Fatal("expected error on boot failure") + } + // Node should be destroyed. + n, _ := h.store.GetNodeByUUID(ctx, h.store.idem["key-boom"]) + if n == nil || n.Status != StatusDestroyed { + t.Errorf("node not destroyed after boot failure: %+v", n) + } + if got := h.svc.FailCount(PoolConsumable); got != 1 { + t.Errorf("fail count = %d, want 1", got) + } + if h.alert.count(AlertBootFailed) != 1 { + t.Errorf("boot_failed alerts = %d, want 1", h.alert.count(AlertBootFailed)) + } +} + +func TestDestroyNode_Idempotent(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + n, err := h.svc.CreateNode(ctx, freeSpec(), "key-d") + if err != nil { + t.Fatalf("CreateNode: %v", err) + } + + if err := h.svc.DestroyNode(ctx, n.ID); err != nil { + t.Fatalf("DestroyNode: %v", err) + } + if err := h.svc.DestroyNode(ctx, n.ID); err != nil { + t.Fatalf("second DestroyNode should be no-op: %v", err) + } + if h.adapter.destroyCount() != 1 { + t.Errorf("vendor destroy called %d times, want 1 (idempotent)", h.adapter.destroyCount()) + } + got, _ := h.store.GetNode(ctx, n.ID) + if got.Status != StatusDestroyed { + t.Errorf("status = %s, want destroyed", got.Status) + } +} + +func TestRotateIP(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + n, err := h.svc.CreateNode(ctx, freeSpec(), "key-ip") + if err != nil { + t.Fatalf("CreateNode: %v", err) + } + before := h.store.version + oldEndpoint := n.Endpoint + + updated, err := h.svc.RotateIP(ctx, n.ID) + if err != nil { + t.Fatalf("RotateIP: %v", err) + } + if updated.Endpoint == oldEndpoint { + t.Errorf("endpoint not changed: still %s", updated.Endpoint) + } + if h.store.version <= before { + t.Errorf("directory version not bumped: %d <= %d", h.store.version, before) + } + if h.store.countEvents(EventIPRotated) != 1 { + t.Errorf("ip_rotated events = %d, want 1", h.store.countEvents(EventIPRotated)) + } + stored, _ := h.store.GetNode(ctx, n.ID) + if stored.Endpoint != updated.Endpoint { + t.Errorf("stored endpoint %s != returned %s", stored.Endpoint, updated.Endpoint) + } +} + +func TestRotateIP_Unsupported(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + h.adapter.elastic = false + n, err := h.svc.CreateNode(ctx, freeSpec(), "key-ip2") + if err != nil { + t.Fatalf("CreateNode: %v", err) + } + _, err = h.svc.RotateIP(ctx, n.ID) + if !errors.Is(err, ErrElasticIPUnsupported) { + t.Errorf("err = %v, want ErrElasticIPUnsupported", err) + } +} + +func TestListProviders(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + ps, err := h.svc.ListProviders(ctx, PoolPremium) + if err != nil { + t.Fatalf("ListProviders: %v", err) + } + if len(ps) != 1 || ps[0].Pool != PoolPremium { + t.Errorf("premium providers = %+v, want exactly 1 premium", ps) + } +} diff --git a/server/internal/provision/store.go b/server/internal/provision/store.go new file mode 100644 index 0000000..d7ebf81 --- /dev/null +++ b/server/internal/provision/store.go @@ -0,0 +1,81 @@ +package provision + +import ( + "context" + "time" +) + +// ReplaceStep enumerates the resumable steps of a Replace orchestration. +// Persisted in replacements.step so a crashed process can continue without +// repeating a boot or a destroy. +type ReplaceStep string + +const ( + StepOpenNew ReplaceStep = "open_new" // create the replacement node + StepProbing ReplaceStep = "probing" // wait for self-register + probe + StepNewUp ReplaceStep = "new_up" // promote new node, bump version + StepDraining ReplaceStep = "draining" // drain the old node + StepDestroyOld ReplaceStep = "destroy_old" // destroy old node + release IP + StepDone ReplaceStep = "done" +) + +// ReplaceStatus is the terminal/running state of a replacement record. +type ReplaceStatus string + +const ( + ReplaceRunning ReplaceStatus = "running" + ReplaceDone ReplaceStatus = "done" + ReplaceFailed ReplaceStatus = "failed" +) + +// Replacement mirrors a replacements row (crash-recoverable orchestration). +type Replacement struct { + UUID string + OldNodeID int64 + NewNodeID int64 // 0 until the new node is created + Pool Pool + Status ReplaceStatus + Step ReplaceStep + CreatedAt time.Time + UpdatedAt time.Time +} + +// Store is the persistence boundary for the provision package. Both the MySQL +// implementation (mysqlStore) and the in-memory test fake satisfy it, keeping +// the service / orchestration logic database-agnostic and unit-testable. +type Store interface { + // --- nodes --- + InsertNode(ctx context.Context, n *Node) (int64, error) + GetNode(ctx context.Context, id int64) (*Node, error) + GetNodeByUUID(ctx context.Context, uuid string) (*Node, error) + UpdateNodeStatus(ctx context.Context, id int64, status Status) error + UpdateNodeEndpoint(ctx context.Context, id int64, endpoint string) error + // SetNodeInstance records the vendor instance ID and IP-derived endpoint. + SetNodeInstance(ctx context.Context, id int64, instanceID, endpoint string) error + SetNodeWeight(ctx context.Context, id int64, weight int) error + // ListNodesByPool returns nodes whose provider belongs to pool, optionally + // filtered to a single status (empty = all statuses). + ListNodesByPool(ctx context.Context, pool Pool, status Status) ([]*Node, error) + + // --- providers --- + ListProviders(ctx context.Context, pool Pool) ([]*Provider, error) + GetProvider(ctx context.Context, id int64) (*Provider, error) + + // --- events / audit / directory --- + WriteNodeEvent(ctx context.Context, nodeID int64, event Event, detailJSON string) error + WriteAuditLog(ctx context.Context, actor, action, target, metaJSON string) error + BumpDirectoryVersion(ctx context.Context) (int64, error) + + // --- idempotency --- + // LookupIdempotency returns the node UUID previously bound to key, or + // (\"\", false, nil) if unseen. + LookupIdempotency(ctx context.Context, key string) (string, bool, error) + // SaveIdempotency binds key→nodeUUID. It is a no-op if the key already + // exists (first write wins). + SaveIdempotency(ctx context.Context, key, nodeUUID string) error + + // --- replacements (crash recovery) --- + CreateReplacement(ctx context.Context, r *Replacement) error + GetReplacement(ctx context.Context, uuid string) (*Replacement, error) + UpdateReplacement(ctx context.Context, r *Replacement) error +} diff --git a/server/internal/provision/types.go b/server/internal/provision/types.go new file mode 100644 index 0000000..7146f09 --- /dev/null +++ b/server/internal/provision/types.go @@ -0,0 +1,121 @@ +package provision + +import "time" + +// Pool is a vendor pool (doc/04 §5.2). Consumable = cheap small vendors for +// entry/free nodes; premium = stable vendors for exit / pro entry. +type Pool string + +const ( + PoolConsumable Pool = "consumable" + PoolPremium Pool = "premium" +) + +// Tier mirrors nodes.tier (free = consumable pool, pro = premium pool). +type Tier string + +const ( + TierFree Tier = "free" + TierPro Tier = "pro" +) + +// Role mirrors nodes.role. +type Role string + +const ( + RoleEntry Role = "entry" + RoleRelay Role = "relay" + RoleExit Role = "exit" +) + +// Status mirrors the nodes.status lifecycle state machine (doc/04 §3). +type Status string + +const ( + StatusProvisioning Status = "provisioning" + StatusProbing Status = "probing" + StatusUp Status = "up" + StatusDraining Status = "draining" + StatusDown Status = "down" + StatusDestroyed Status = "destroyed" +) + +// Event mirrors the node_events.event enum. +type Event string + +const ( + EventProvisioned Event = "provisioned" + EventProbePass Event = "probe_pass" + EventProbeFail Event = "probe_fail" + EventMarkedUp Event = "marked_up" + EventDraining Event = "draining" + EventBlockedSuspect Event = "blocked_suspect" + EventBlockedConfirmed Event = "blocked_confirmed" + EventReplaced Event = "replaced" + EventDestroyed Event = "destroyed" + EventIPRotated Event = "ip_rotated" +) + +// Provider mirrors a providers row. Credentials are intentionally absent: +// they live only in independent secrets (env/file), never in this struct +// nor in the database (doc/06 §2 red line). +type Provider struct { + ID int64 + Name string + APIKind string // adapter key, e.g. "vultr", "hetzner" + Regions []string + Pool Pool + Enabled bool +} + +// NodeSpec is the desired shape of a node, the input to CreateNode. +type NodeSpec struct { + Region string + Role Role + Tier Tier + ProviderID int64 + // Plan is the vendor-side instance size identifier (e.g. "vc2-1c-1gb"). + Plan string + NameZH string + NameEn string + RealityPBK string + RealitySNI string + HY2Port int + Weight int + Tags []string +} + +// Node mirrors a nodes row plus the provider resource identifiers added by +// migration 000008. +type Node struct { + ID int64 + UUID string + Region string + NameZH string + NameEn string + Role Role + Tier Tier + Endpoint string // ip:port + HY2Port int + RealityPBK string + RealitySNI string + ProviderID int64 + ProviderInstanceID string + ElasticIPID string + Tags []string + Status Status + Weight int + CreatedAt time.Time +} + +// pendingEndpoint is the placeholder written at insert time, before the vendor +// returns an IP. nodes.endpoint is NOT NULL, so we cannot leave it empty. +const pendingEndpoint = "0.0.0.0:0" + +// poolForTier maps a node tier onto its vendor pool. +func poolForTier(t Tier) Pool { + if t == TierPro { + return PoolPremium + } + return PoolConsumable +} diff --git a/server/migrations/000008_provision.down.sql b/server/migrations/000008_provision.down.sql new file mode 100644 index 0000000..f0a4c54 --- /dev/null +++ b/server/migrations/000008_provision.down.sql @@ -0,0 +1,8 @@ +DROP TABLE IF EXISTS replacements; +DROP TABLE IF EXISTS provision_idempotency; +ALTER TABLE node_events + MODIFY COLUMN event ENUM('provisioned','probe_pass','probe_fail','marked_up','draining', + 'blocked_suspect','blocked_confirmed','replaced','destroyed') NOT NULL; +ALTER TABLE nodes + DROP COLUMN elastic_ip_id, + DROP COLUMN provider_instance_id; diff --git a/server/migrations/000008_provision.up.sql b/server/migrations/000008_provision.up.sql new file mode 100644 index 0000000..06dd892 --- /dev/null +++ b/server/migrations/000008_provision.up.sql @@ -0,0 +1,37 @@ +-- 弹性节点基建（task #14 / tsk_6u0FxmbC7Yeq） +-- 厂商 API 适配层与一键更换编排所需的最小附加结构。全部为附加式变更， +-- 不触碰现网生产 EC2（deploy/ 的 marzban 机器）。 + +-- nodes 增补：厂商实例标识与弹性 IP 标识（销毁 / 换 IP 不换机所需）。 +-- 凭证仍不入库；这里只存厂商侧的资源 ID，便于幂等地销毁与换绑。 +ALTER TABLE nodes + ADD COLUMN provider_instance_id VARCHAR(128) NULL AFTER provider_id, + ADD COLUMN elastic_ip_id VARCHAR(128) NULL AFTER provider_instance_id; + +-- node_events 增补 ip_rotated（换 IP 不换机的审计事件）。 +ALTER TABLE node_events + MODIFY COLUMN event ENUM('provisioned','probe_pass','probe_fail','marked_up','draining', + 'blocked_suspect','blocked_confirmed','replaced','destroyed','ip_rotated') NOT NULL; + +-- 开退机幂等表：幂等键 → 已开机节点 uuid。重放同一幂等键不重复开机。 +CREATE TABLE provision_idempotency ( + idempotency_key VARCHAR(128) NOT NULL PRIMARY KEY, + node_uuid CHAR(36) NOT NULL, + created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +-- 一键更换编排记录：make-before-break 的可恢复状态机。 +-- 进程崩溃后可凭 uuid 续跑，step 标记进度，不重复开机 / 不重复销毁。 +CREATE TABLE replacements ( + uuid CHAR(36) NOT NULL PRIMARY KEY, + old_node_id BIGINT UNSIGNED NOT NULL, + new_node_id BIGINT UNSIGNED NULL, + pool ENUM('consumable','premium') NOT NULL, + status ENUM('running','done','failed') NOT NULL DEFAULT 'running', + step VARCHAR(32) NOT NULL DEFAULT 'open_new', + detail JSON NULL, + created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6), + INDEX idx_status (status), + INDEX idx_old_node (old_node_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;