diff --git a/.gitignore b/.gitignore index 19a3bf5..c94f010 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,10 @@ *.exe *.out +# Local Firecracker assets: kernel image, converted rootfs cache, and per-VM +# work dirs the agent creates at runtime (some root-owned). Not source. +/fc-assets/ + # Environment .env diff --git a/Makefile b/Makefile index 3536a23..3d397c2 100644 --- a/Makefile +++ b/Makefile @@ -1,10 +1,11 @@ -.PHONY: run run-agent build build-agent test test-e2e test-kvm test-bpf tidy fmt bpf-generate +.PHONY: run run-agent build build-agent test test-e2e test-kvm test-bpf tidy fmt bpf-generate proto-generate run: go run ./cmd/server -# Run a host agent (P3). Defaults target a local control plane; override via env, -# e.g. MODE=agent CONTROL_PLANE_URL=... ADVERTISE_ADDR=... PORT=9000. +# Run a host agent (P3). The agent dials the control plane's gRPC AgentLink and +# holds a stream open; override the target via env, e.g. +# MODE=agent CONTROL_PLANE_GRPC_ADDR=localhost:8090 AGENT_RUNTIME=fake. run-agent: MODE=agent go run ./cmd/agent @@ -49,6 +50,19 @@ bpf-generate: bpftool btf dump file /sys/kernel/btf/vmlinux format c > $(BPF_DIR)/vmlinux.h go generate ./$(BPF_DIR)/... +# Regenerate the gRPC AgentLink stubs from proto/agentlink/agentlink.proto. This +# is a MAINTAINER step; the outputs (internal/agentlink/pb/*.pb.go) are committed +# so `go build`/CI need only the Go toolchain. Run after editing the .proto, then +# `git add` the regenerated files. Requires protoc plus the Go plugins: +# go install google.golang.org/protobuf/cmd/protoc-gen-go@latest +# go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest +proto-generate: + @command -v protoc >/dev/null || { echo "need protoc (protobuf-compiler)"; exit 1; } + protoc \ + --go_out=. --go_opt=module=github.com/aarani/craftling-go \ + --go-grpc_out=. --go-grpc_opt=module=github.com/aarani/craftling-go \ + proto/agentlink/agentlink.proto + tidy: go mod tidy diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 91b6426..785e114 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -1,6 +1,8 @@ -// Command agent is the host-side worker (P3). It exposes a VM API the control -// plane calls to provision/start/stop/deprovision local VMs, and it registers + -// heartbeats with the control plane so the scheduler can place servers on it. +// Command agent is the host-side worker (P3). It dials the control plane and +// holds a persistent gRPC stream open over which the control plane pushes VM +// lifecycle commands (provision/start/stop/deprovision); the agent runs them +// against its local Runtime and answers on the same stream. It has no inbound +// API — the open stream both delivers commands and proves the host's liveness. // // It ships with the in-memory FakeRuntime; a real Firecracker driver (P4) slots // in behind the same Runtime interface without changing this wiring. @@ -8,10 +10,8 @@ package main import ( "context" - "errors" "fmt" "log" - "net/http" "os" "os/signal" "path/filepath" @@ -27,15 +27,6 @@ import ( "go.uber.org/zap" ) -const ( - // heartbeatInterval is how often the agent proves liveness to the control - // plane. It must be comfortably below the control plane's host TTL (30s). - heartbeatInterval = 10 * time.Second - // registerRetryInterval is how long to wait between registration attempts - // while the control plane is unreachable. - registerRetryInterval = 5 * time.Second -) - func main() { cfg := config.Load() @@ -45,57 +36,28 @@ func main() { } defer func() { _ = zlog.Sync() }() - advertiseAddr := cfg.Agent.AdvertiseAddr - if advertiseAddr == "" { - // Best-effort default so a local single-host run works out of the box. - advertiseAddr = "localhost:" + cfg.Port - } - - // The runtime that actually runs VMs, fronted by the agent HTTP API. + // The runtime that actually runs VMs, driven by commands off the link. rt, err := newRuntime(cfg, zlog) if err != nil { zlog.Fatal("init runtime", zap.Error(err)) } - srv := &http.Server{ - Addr: ":" + cfg.Port, - Handler: agent.NewRouter(rt, zlog), - ReadTimeout: 10 * time.Second, - WriteTimeout: 10 * time.Second, - IdleTimeout: 60 * time.Second, - } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() - go func() { - zlog.Info("agent listening", - zap.String("port", cfg.Port), zap.String("advertise_addr", advertiseAddr)) - if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { - zlog.Fatal("agent listen failed", zap.Error(err)) - } - }() - - // Register with the control plane and keep the host alive via heartbeats. - cp := agent.NewCPClient(cfg.Agent.ControlPlaneURL, &http.Client{Timeout: 10 * time.Second}) - go runRegistration(ctx, zlog, cp, agent.RegisterRequest{ + // Hold a persistent connection to the control plane for the agent's lifetime. + // RunLink blocks until ctx is cancelled, reconnecting on its own if the stream + // drops, so this is the agent's main loop. + zlog.Info("connecting to control plane", zap.String("addr", cfg.Agent.ControlPlaneGRPCAddr)) + agent.RunLink(ctx, cfg.Agent.ControlPlaneGRPCAddr, rt, agent.LinkInfo{ ID: cfg.Agent.ID, Hostname: cfg.Agent.Hostname, - Address: advertiseAddr, Zone: cfg.Agent.Zone, + AgentVersion: cfg.Agent.Version, CPUsTotal: cfg.Agent.CPUsTotal, MemoryMBTotal: cfg.Agent.MemoryMBTotal, - AgentVersion: cfg.Agent.Version, - }) + }, zlog) - <-ctx.Done() - stop() - zlog.Info("shutting down agent...") - - shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - if err := srv.Shutdown(shutdownCtx); err != nil { - zlog.Fatal("forced shutdown", zap.Error(err)) - } zlog.Info("agent exited") } @@ -161,57 +123,3 @@ func imageCacheDir(fc config.FirecrackerConfig) string { } return filepath.Join(workDir, "images") } - -// runRegistration registers the host then heartbeats on an interval until ctx is -// cancelled. A heartbeat that the control plane rejects with "not found" (it was -// restarted and forgot us) triggers a re-register, restoring the same identity. -func runRegistration(ctx context.Context, log *zap.Logger, cp *agent.CPClient, req agent.RegisterRequest) { - id := register(ctx, log, cp, req) - if id == "" { - return // ctx cancelled before we registered - } - // Re-register under the assigned id so identity stays stable across restarts. - req.ID = id - - ticker := time.NewTicker(heartbeatInterval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - found, err := cp.Heartbeat(ctx, id) - if err != nil { - log.Warn("heartbeat failed", zap.Error(err)) - continue - } - if !found { - log.Warn("control plane forgot host; re-registering", zap.String("id", id)) - if newID := register(ctx, log, cp, req); newID != "" { - id = newID - req.ID = newID - } - } - } - } -} - -// register retries registration until it succeeds or ctx is cancelled, -// returning the assigned host id (empty on cancellation). -func register(ctx context.Context, log *zap.Logger, cp *agent.CPClient, req agent.RegisterRequest) string { - for { - id, err := cp.Register(ctx, req) - if err == nil { - log.Info("registered with control plane", zap.String("id", id)) - return id - } - log.Warn("register failed; retrying", zap.Error(err)) - - select { - case <-ctx.Done(): - return "" - case <-time.After(registerRetryInterval): - } - } -} diff --git a/cmd/server/main.go b/cmd/server/main.go index b6b31a0..9f3e168 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -4,12 +4,14 @@ import ( "context" "errors" "log" + "net" "net/http" "os/signal" "syscall" "time" - "github.com/aarani/craftling-go/internal/agent" + "github.com/aarani/craftling-go/internal/agentlink" + pb "github.com/aarani/craftling-go/internal/agentlink/pb" "github.com/aarani/craftling-go/internal/config" "github.com/aarani/craftling-go/internal/db" "github.com/aarani/craftling-go/internal/handler" @@ -22,6 +24,7 @@ import ( "github.com/aarani/craftling-go/internal/seed" "github.com/aarani/craftling-go/internal/worldstore" "go.uber.org/zap" + "google.golang.org/grpc" ) const ( @@ -34,8 +37,6 @@ const ( // hostHeartbeatTTL is how long a host may go without heartbeating before it // is marked down. hostHeartbeatTTL = 30 * time.Second - // agentCallTimeout bounds each control-plane→agent VM API call. - agentCallTimeout = 10 * time.Second // worldGCInterval is how often the durable world store is swept for // snapshots belonging to no live server (P5b). worldGCInterval = time.Hour @@ -73,8 +74,10 @@ func main() { dbCancel() // The fleet inventory lives in process memory (P1). It is shared between the - // HTTP handlers (register/heartbeat) and the host reaper. + // agent link hub (which registers/heartbeats hosts as their streams come and + // go) and the host reaper. hostRepo := repository.NewHostRepository() + gameServerRepo := repository.NewGameServerRepository(pool) router := handler.NewRouter(cfg, zlog, pool, hostRepo) @@ -86,6 +89,18 @@ func main() { IdleTimeout: 60 * time.Second, } + // The hub is the control plane's end of the persistent agent connection: + // agents dial the gRPC listener and hold a stream open, and the hub pushes VM + // commands down it. It registers hosts (reconstructing committed capacity + // from the durable server records) and tracks liveness off the stream. + hub := agentlink.NewHub(hostRepo, gameServerRepo, zlog) + grpcSrv := grpc.NewServer() + pb.RegisterAgentLinkServer(grpcSrv, hub) + grpcLis, err := net.Listen("tcp", ":"+cfg.GRPCPort) + if err != nil { + zlog.Fatal("listen for agent gRPC", zap.Error(err)) + } + // ctx is cancelled on the first interrupt/terminate signal, which both // stops the reaper and triggers graceful shutdown. ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) @@ -103,8 +118,8 @@ func main() { // then drives the VM by calling the assigned host's agent (the control plane // never touches KVM itself). sched := scheduler.New(hostRepo) - prov := provisioner.NewRemote(hostRepo, agent.NewClient(&http.Client{Timeout: agentCallTimeout})) - rec := reconciler.New(repository.NewGameServerRepository(pool), prov, sched, zlog) + prov := provisioner.NewRemote(hub) + rec := reconciler.New(gameServerRepo, prov, sched, zlog) go rec.Run(ctx, reconcileInterval) // If a durable world store is configured, periodically GC snapshots that no @@ -116,9 +131,17 @@ func main() { if err != nil { zlog.Warn("world store unavailable; world GC disabled", zap.Error(err)) } else if worldStore != nil { - go reaper.Worlds(ctx, zlog, worldStore, repository.NewGameServerRepository(pool), worldGCInterval) + go reaper.Worlds(ctx, zlog, worldStore, gameServerRepo, worldGCInterval) } + // Serve the agent gRPC link alongside the HTTP API. + go func() { + zlog.Info("agent gRPC listening", zap.String("port", cfg.GRPCPort)) + if err := grpcSrv.Serve(grpcLis); err != nil && !errors.Is(err, grpc.ErrServerStopped) { + zlog.Fatal("agent gRPC serve failed", zap.Error(err)) + } + }() + // Start the server in a goroutine so it doesn't block graceful shutdown handling. go func() { zlog.Info("server listening", zap.String("port", cfg.Port), zap.String("env", cfg.Env)) @@ -131,6 +154,8 @@ func main() { stop() // restore default signal handling so a second signal force-quits zlog.Info("shutting down server...") + grpcSrv.GracefulStop() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() if err := srv.Shutdown(shutdownCtx); err != nil { diff --git a/docker-compose.yml b/docker-compose.yml index 1bdc8c0..953d85c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -38,6 +38,10 @@ services: ADMIN_PASSWORD: adminpassword ports: - "8082:8080" + # The gRPC AgentLink listener (default :8090) is reachable to agents over the + # compose network; no host publish is needed. + expose: + - "8090" depends_on: db: condition: service_healthy @@ -66,11 +70,11 @@ services: dockerfile: Dockerfile.agent environment: MODE: agent - PORT: "9000" APP_ENV: production - CONTROL_PLANE_URL: http://server:8080 - # Address the control plane calls back (reachable on the compose network). - ADVERTISE_ADDR: agent:9000 + # The agent dials this and holds a stream open; the control plane pushes VM + # commands down it. No inbound address is advertised — the control plane + # never dials the agent. + CONTROL_PLANE_GRPC_ADDR: server:8090 # Player-facing connect host VMs report (override for real connectivity). ADVERTISE_HOST: 127.0.0.1 AGENT_RUNTIME: firecracker @@ -90,9 +94,8 @@ services: privileged: true volumes: - ./fc-assets:/var/lib/craftling - ports: - # Exposed for debugging the agent API directly. - - "9000:9000" + # No published ports: the agent serves no inbound API. It reaches the control + # plane's gRPC link at server:8090 over the compose network. depends_on: - server restart: unless-stopped diff --git a/go.mod b/go.mod index 3a20cef..ed933bf 100644 --- a/go.mod +++ b/go.mod @@ -22,6 +22,8 @@ require ( go.uber.org/zap v1.28.0 golang.org/x/crypto v0.52.0 golang.org/x/sys v0.46.0 + google.golang.org/grpc v1.81.1 + google.golang.org/protobuf v1.36.11 ) require ( @@ -134,7 +136,7 @@ require ( golang.org/x/net v0.55.0 // indirect golang.org/x/sync v0.20.0 // indirect golang.org/x/text v0.37.0 // indirect - google.golang.org/protobuf v1.36.11 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529 // indirect gopkg.in/ini.v1 v1.67.2 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index ba115e5..acad684 100644 --- a/go.sum +++ b/go.sum @@ -133,6 +133,8 @@ github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM= github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA= github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= @@ -329,6 +331,12 @@ golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529 h1:XF8+t6QQiS0o9ArVan/HW8Q7cycNPGsJf6GA2nXxYAg= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.81.1 h1:VnnIIZ88UzOOKLukQi+ImGz8O1Wdp8nAGGnvOfEIWQQ= +google.golang.org/grpc v1.81.1/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index b64848c..e8c2991 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -2,11 +2,11 @@ package agent import ( "context" + "encoding/json" "errors" - "net/http/httptest" "testing" - "go.uber.org/zap" + pb "github.com/aarani/craftling-go/internal/agentlink/pb" ) // TestFakeRuntimeLifecycle exercises the in-memory runtime directly through its @@ -63,74 +63,53 @@ func TestFakeRuntimeIdempotency(t *testing.T) { } } -// TestAgentServerClientRoundTrip drives the runtime through the HTTP API the -// control plane uses, verifying the wire contract end-to-end. -func TestAgentServerClientRoundTrip(t *testing.T) { +// TestExecOpDispatch verifies the link's command dispatch: each op reaches the +// runtime, results are JSON-encoded the way the hub decodes them, and an unknown +// op surfaces an error rather than panicking. This is the agent half of the +// control-plane → agent command contract. +func TestExecOpDispatch(t *testing.T) { ctx := context.Background() - srv := httptest.NewServer(NewRouter(NewFakeRuntime("10.0.0.9"), zap.NewNop())) - defer srv.Close() + rt := NewFakeRuntime("10.0.0.9") - client := NewClient(nil) - base := srv.URL - - vm, err := client.Provision(ctx, base, VMSpec{ServerID: "s2", Version: "1.20.4", CPUs: 1, MemoryMB: 1024}) - if err != nil { - t.Fatalf("provision: %v", err) + // Provision returns a running VM payload, no error. + specJSON, _ := json.Marshal(VMSpec{ServerID: "s2", Version: "1.20.4", CPUs: 1, MemoryMB: 1024}) + payload, errStr := execOp(ctx, rt, &pb.Command{Op: OpProvision, Payload: specJSON}) + if errStr != "" { + t.Fatalf("provision op error = %q, want none", errStr) } - if vm == nil || vm.ID == "" || vm.State != StateRunning { - t.Fatalf("provisioned vm = %+v, want running with id", vm) + var vm VM + if err := json.Unmarshal(payload, &vm); err != nil { + t.Fatalf("decode provision result: %v", err) } - if vm.Host != "10.0.0.9" || vm.Port != defaultMinecraftPort { - t.Errorf("connect = %s:%d, want 10.0.0.9:%d", vm.Host, vm.Port, defaultMinecraftPort) + if vm.ID == "" || vm.State != StateRunning { + t.Fatalf("provisioned vm = %+v, want running with id", vm) } - if got := statusOf(t, client, base, vm.ID); got != StateRunning { - t.Errorf("after provision state = %q, want running", got) - } - if err := client.Stop(ctx, base, vm.ID); err != nil { - t.Fatalf("stop: %v", err) - } - if got := statusOf(t, client, base, vm.ID); got != StateStopped { - t.Errorf("after stop state = %q, want stopped", got) - } - if _, err := client.Start(ctx, base, vm.ID); err != nil { - t.Fatalf("start: %v", err) - } - if got := statusOf(t, client, base, vm.ID); got != StateRunning { - t.Errorf("after start state = %q, want running", got) + ref, _ := json.Marshal(VMRef{VMID: vm.ID}) + + // Stop then Status reflects the stopped state across the seam. + if _, errStr := execOp(ctx, rt, &pb.Command{Op: OpStop, Payload: ref}); errStr != "" { + t.Fatalf("stop op error = %q, want none", errStr) } - if err := client.Deprovision(ctx, base, vm.ID); err != nil { - t.Fatalf("deprovision: %v", err) + statusPayload, errStr := execOp(ctx, rt, &pb.Command{Op: OpStatus, Payload: ref}) + if errStr != "" { + t.Fatalf("status op error = %q, want none", errStr) } - if got := statusOf(t, client, base, vm.ID); got != StateMissing { - t.Errorf("after deprovision state = %q, want missing", got) + var stopped VM + _ = json.Unmarshal(statusPayload, &stopped) + if stopped.State != StateStopped { + t.Errorf("status after stop = %q, want stopped", stopped.State) } - // Starting a VM the agent does not know is an error over the wire. - if _, err := client.Start(ctx, base, "vm-ghost"); err == nil { - t.Error("start unknown vm: expected error, got nil") + // Starting a VM the runtime does not know surfaces an error string. + ghost, _ := json.Marshal(VMRef{VMID: "vm-ghost"}) + if _, errStr := execOp(ctx, rt, &pb.Command{Op: OpStart, Payload: ghost}); errStr == "" { + t.Error("start unknown vm: expected error string, got none") } -} - -// TestAgentSnapshotRoundTrip exercises the on-demand snapshot endpoint over the -// real HTTP seam: a known VM succeeds (the fake runtime no-ops), an unknown one -// surfaces a not-found error to the caller. -func TestAgentSnapshotRoundTrip(t *testing.T) { - ctx := context.Background() - srv := httptest.NewServer(NewRouter(NewFakeRuntime("10.0.0.9"), zap.NewNop())) - defer srv.Close() - client := NewClient(nil) - base := srv.URL - vm, err := client.Provision(ctx, base, VMSpec{ServerID: "s3", CPUs: 1, MemoryMB: 1024}) - if err != nil { - t.Fatalf("provision: %v", err) - } - if err := client.Snapshot(ctx, base, vm.ID); err != nil { - t.Fatalf("snapshot known vm: %v", err) - } - if err := client.Snapshot(ctx, base, "vm-ghost"); err == nil { - t.Error("snapshot unknown vm: expected error, got nil") + // An unrecognized op is reported, not fatal. + if _, errStr := execOp(ctx, rt, &pb.Command{Op: "bogus"}); errStr == "" { + t.Error("unknown op: expected error string, got none") } } @@ -144,12 +123,3 @@ func assertState(t *testing.T, rt Runtime, vmID, want string) { t.Fatalf("state = %q, want %q", vm.State, want) } } - -func statusOf(t *testing.T, c *Client, base, vmID string) string { - t.Helper() - vm, err := c.Status(context.Background(), base, vmID) - if err != nil { - t.Fatalf("status: %v", err) - } - return vm.State -} diff --git a/internal/agent/client.go b/internal/agent/client.go deleted file mode 100644 index a4325c1..0000000 --- a/internal/agent/client.go +++ /dev/null @@ -1,109 +0,0 @@ -package agent - -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "io" - "net/http" - "strings" -) - -// Client calls an agent's VM API. One Client is shared across all agents; the -// target agent's base URL is passed per call (resolved from the host inventory), -// since the control plane talks to many hosts. -type Client struct { - http *http.Client -} - -// NewClient constructs a Client over the given HTTP client (supply one with a -// sensible timeout). A nil httpClient falls back to http.DefaultClient. -func NewClient(httpClient *http.Client) *Client { - if httpClient == nil { - httpClient = http.DefaultClient - } - return &Client{http: httpClient} -} - -// BaseURL normalizes a host address (e.g. "10.0.0.1:9000") into an agent base -// URL, defaulting to http:// when no scheme is present. -func BaseURL(address string) string { - if strings.HasPrefix(address, "http://") || strings.HasPrefix(address, "https://") { - return strings.TrimRight(address, "/") - } - return "http://" + strings.TrimRight(address, "/") -} - -// Provision asks the agent to create and boot a VM for the spec. -func (c *Client) Provision(ctx context.Context, baseURL string, spec VMSpec) (*VM, error) { - return c.doVM(ctx, http.MethodPost, baseURL+"/vms", spec) -} - -// Start asks the agent to boot an existing VM. -func (c *Client) Start(ctx context.Context, baseURL, vmID string) (*VM, error) { - return c.doVM(ctx, http.MethodPost, baseURL+"/vms/"+vmID+"/start", nil) -} - -// Stop asks the agent to halt a VM. -func (c *Client) Stop(ctx context.Context, baseURL, vmID string) error { - _, err := c.doVM(ctx, http.MethodPost, baseURL+"/vms/"+vmID+"/stop", nil) - return err -} - -// Snapshot asks the agent to take an on-demand world snapshot of a running VM. -func (c *Client) Snapshot(ctx context.Context, baseURL, vmID string) error { - _, err := c.doVM(ctx, http.MethodPost, baseURL+"/vms/"+vmID+"/snapshot", nil) - return err -} - -// Deprovision asks the agent to destroy a VM. -func (c *Client) Deprovision(ctx context.Context, baseURL, vmID string) error { - _, err := c.doVM(ctx, http.MethodDelete, baseURL+"/vms/"+vmID, nil) - return err -} - -// Status fetches a VM's observed state. -func (c *Client) Status(ctx context.Context, baseURL, vmID string) (*VM, error) { - return c.doVM(ctx, http.MethodGet, baseURL+"/vms/"+vmID, nil) -} - -// doVM performs an agent request and decodes a VM body when one is returned. -// Endpoints that reply with a plain {"status":"ok"} simply yield a nil VM and a -// nil error. -func (c *Client) doVM(ctx context.Context, method, url string, body any) (*VM, error) { - var reader io.Reader - if body != nil { - b, err := json.Marshal(body) - if err != nil { - return nil, fmt.Errorf("marshal request: %w", err) - } - reader = bytes.NewReader(b) - } - - req, err := http.NewRequestWithContext(ctx, method, url, reader) - if err != nil { - return nil, fmt.Errorf("build request: %w", err) - } - if body != nil { - req.Header.Set("Content-Type", "application/json") - } - - resp, err := c.http.Do(req) - if err != nil { - return nil, fmt.Errorf("call agent: %w", err) - } - defer resp.Body.Close() - - data, _ := io.ReadAll(resp.Body) - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return nil, fmt.Errorf("agent %s %s: status %d: %s", method, url, resp.StatusCode, strings.TrimSpace(string(data))) - } - - // Lifecycle replies carry a VM; control replies ({"status":"ok"}) do not. - var vm VM - if err := json.Unmarshal(data, &vm); err != nil || vm.ID == "" { - return nil, nil - } - return &vm, nil -} diff --git a/internal/agent/controlplane.go b/internal/agent/controlplane.go deleted file mode 100644 index a58cae6..0000000 --- a/internal/agent/controlplane.go +++ /dev/null @@ -1,101 +0,0 @@ -package agent - -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "io" - "net/http" - "strings" -) - -// CPClient is the agent's view of the control plane: it registers the host and -// keeps it alive via heartbeats (the P1 agent endpoints). This is the "push -// status up" half of the control-plane-authoritative model; the control plane -// pushes desired state back down via the agent VM API. -type CPClient struct { - http *http.Client - baseURL string -} - -// NewCPClient constructs a control-plane client for the given base URL -// (e.g. "http://control-plane:8080"). -func NewCPClient(baseURL string, httpClient *http.Client) *CPClient { - if httpClient == nil { - httpClient = http.DefaultClient - } - return &CPClient{http: httpClient, baseURL: strings.TrimRight(baseURL, "/")} -} - -// RegisterRequest is the host registration payload. ID is the agent's own stable -// id; supplying it keeps the host's identity stable across a control-plane -// restart (P1 agent-owned ids). -type RegisterRequest struct { - ID string `json:"id,omitempty"` - Hostname string `json:"hostname"` - Address string `json:"address"` - Zone string `json:"zone,omitempty"` - CPUsTotal int `json:"cpus_total"` - MemoryMBTotal int `json:"memory_mb_total"` - AgentVersion string `json:"agent_version,omitempty"` -} - -// Register registers (or re-registers) this host and returns its assigned id. -func (c *CPClient) Register(ctx context.Context, req RegisterRequest) (string, error) { - b, err := json.Marshal(req) - if err != nil { - return "", fmt.Errorf("marshal register: %w", err) - } - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, - c.baseURL+"/api/v1/agent/hosts/register", bytes.NewReader(b)) - if err != nil { - return "", fmt.Errorf("build register request: %w", err) - } - httpReq.Header.Set("Content-Type", "application/json") - - resp, err := c.http.Do(httpReq) - if err != nil { - return "", fmt.Errorf("register: %w", err) - } - defer resp.Body.Close() - - data, _ := io.ReadAll(resp.Body) - if resp.StatusCode != http.StatusCreated { - return "", fmt.Errorf("register: status %d: %s", resp.StatusCode, strings.TrimSpace(string(data))) - } - var host struct { - ID string `json:"id"` - } - if err := json.Unmarshal(data, &host); err != nil { - return "", fmt.Errorf("decode register response: %w", err) - } - return host.ID, nil -} - -// Heartbeat reports liveness for the host. It returns found=false when the -// control plane returns 404 (it has forgotten this host), signalling the agent -// to re-register. -func (c *CPClient) Heartbeat(ctx context.Context, id string) (found bool, err error) { - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, - c.baseURL+"/api/v1/agent/hosts/"+id+"/heartbeat", nil) - if err != nil { - return false, fmt.Errorf("build heartbeat request: %w", err) - } - - resp, err := c.http.Do(httpReq) - if err != nil { - return false, fmt.Errorf("heartbeat: %w", err) - } - defer resp.Body.Close() - _, _ = io.Copy(io.Discard, resp.Body) - - switch resp.StatusCode { - case http.StatusOK: - return true, nil - case http.StatusNotFound: - return false, nil - default: - return false, fmt.Errorf("heartbeat: status %d", resp.StatusCode) - } -} diff --git a/internal/agent/firecracker/config.go b/internal/agent/firecracker/config.go index 1b5518b..80df5b4 100644 --- a/internal/agent/firecracker/config.go +++ b/internal/agent/firecracker/config.go @@ -14,6 +14,7 @@ package firecracker import ( + "context" "errors" "fmt" "net" @@ -24,13 +25,22 @@ import ( "strings" "time" - "github.com/aarani/craftling-go/internal/image" "github.com/aarani/craftling-go/internal/runspec" "github.com/aarani/craftling-go/internal/storage" v1 "github.com/google/go-containerregistry/pkg/v1" "go.uber.org/zap" ) +// ImageEnsurer resolves an OCI ref (pinned to a platform) to a +// content-addressed read-only squashfs rootfs path and the RunSpec distilled +// from the image's OCI config, building the rootfs on first use and reusing the +// cached artifact thereafter. *image.Store is the production implementation; the +// seam exists so the non-KVM tests can substitute a fake and assert what context +// the (long, idempotent) build runs under. +type ImageEnsurer interface { + Ensure(ctx context.Context, ref string, platform *v1.Platform) (string, runspec.RunSpec, error) +} + // Config configures the Firecracker Runtime. Paths point at host artifacts // (kernel, the squashfs image cache) provided out of band on the agent host. type Config struct { @@ -42,7 +52,17 @@ type Config struct { // rootfs files and resolves their RunSpec. Provision pulls the spec's image // through it, attaches the resulting squashfs as a read-only /dev/vda, and // publishes the RunSpec into MMDS for the in-VM init. Required. - ImageStore *image.Store + // *image.Store is the production implementation; the interface seam lets the + // non-KVM tests substitute a fake. + ImageStore ImageEnsurer + // ImagePullTimeout caps a single image build (pull + flatten + squashfs). + // The build is deliberately decoupled from the per-command context — it is + // content-addressed, idempotent, and shared across every VM booting the ref, + // so it must not be aborted because a control-plane command's context was + // cancelled (a stream reconnect, a caller deadline). Anchored to the + // runtime's lifetime instead, this is the only deadline bounding a stuck + // pull. Default DefaultImagePullTimeout. + ImagePullTimeout time.Duration // ImageRef is the OCI image reference the driver converts for a server. A // "{version}" placeholder is substituted with the spec's version, so one // template (e.g. "myrepo/minecraft:{version}") covers every version. A ref @@ -152,6 +172,12 @@ const ( DefaultRCONPort = 25575 ) +// DefaultImagePullTimeout bounds a single image build (pull + flatten + +// squashfs) when ImagePullTimeout is unset. Loose on purpose: a cold pull of a +// large game-server image over a slow link can run into minutes, and the build +// is shared, so over-tight is worse than over-loose. +const DefaultImagePullTimeout = 10 * time.Minute + // DefaultBootArgs is a minimal serial-console boot line that mounts the // read-only squashfs rootfs off the first virtio block device and hands PID 1 // to the injected init agent. Writable state rides the world disk (/dev/vdb) @@ -188,6 +214,9 @@ func (c *Config) validate() error { if c.ImageStore == nil { return errors.New("firecracker: ImageStore is required") } + if c.ImagePullTimeout <= 0 { + c.ImagePullTimeout = DefaultImagePullTimeout + } if c.ImageRef == "" && c.DefaultImageRef == "" { return errors.New("firecracker: ImageRef or DefaultImageRef is required") } diff --git a/internal/agent/firecracker/runtime.go b/internal/agent/firecracker/runtime.go index 3c25a64..e4f0e0d 100644 --- a/internal/agent/firecracker/runtime.go +++ b/internal/agent/firecracker/runtime.go @@ -38,6 +38,15 @@ type Runtime struct { done chan struct{} sweepWG sync.WaitGroup + // baseCtx is the runtime's lifetime context, cancelled by Close. Long, + // idempotent work that must outlive the control-plane command that triggered + // it — notably the shared image build — derives from baseCtx instead of the + // per-command context, so a dropped/reconnected agent stream doesn't abort a + // half-finished build. Cancelling baseCtx at shutdown still tears those + // builds down. + baseCtx context.Context + baseCancel context.CancelFunc + mu sync.Mutex vms map[string]*machine } @@ -56,6 +65,7 @@ func New(cfg Config) (*Runtime, error) { } r := &Runtime{cfg: cfg, vms: make(map[string]*machine), done: make(chan struct{})} + r.baseCtx, r.baseCancel = context.WithCancel(context.Background()) if cfg.persistEnabled() { r.store = cfg.WorldStore } @@ -89,6 +99,7 @@ func New(cfg Config) (*Runtime, error) { // Close stops the periodic snapshot sweep and tears down the NAT dataplane // (detaching all eBPF programs). It is safe to call when neither was enabled. func (r *Runtime) Close() { + r.baseCancel() close(r.done) r.sweepWG.Wait() if r.dp != nil { @@ -157,7 +168,15 @@ func (r *Runtime) Provision(ctx context.Context, spec agent.VMSpec) (*agent.VM, if err != nil { return nil, err } - rootfs, baseSpec, err := r.cfg.ImageStore.Ensure(ctx, ref, hostPlatform()) + // Build the rootfs under the runtime's lifetime, not the command ctx: the + // build is content-addressed, idempotent, and shared across every VM booting + // this ref, so it must not die because this command's context was cancelled + // (a stream reconnect, a caller deadline). A cancelled provision then still + // leaves a populated cache for the next attempt instead of restarting from + // zero. ImagePullTimeout is the only deadline on a stuck pull. + imgCtx, cancelImg := context.WithTimeout(r.baseCtx, r.cfg.ImagePullTimeout) + defer cancelImg() + rootfs, baseSpec, err := r.cfg.ImageStore.Ensure(imgCtx, ref, hostPlatform()) if err != nil { return nil, fmt.Errorf("firecracker: resolve image %q: %w", ref, err) } diff --git a/internal/agent/firecracker/runtime_test.go b/internal/agent/firecracker/runtime_test.go index 3a154e6..046e73a 100644 --- a/internal/agent/firecracker/runtime_test.go +++ b/internal/agent/firecracker/runtime_test.go @@ -6,13 +6,37 @@ import ( "os" "path/filepath" "reflect" + "sync" "testing" "github.com/aarani/craftling-go/internal/agent" "github.com/aarani/craftling-go/internal/image" "github.com/aarani/craftling-go/internal/runspec" + v1 "github.com/google/go-containerregistry/pkg/v1" ) +// fakeEnsurer records the context its Ensure was invoked with, so a test can +// assert the image build does not inherit the (cancellable) command context. +// It returns errStop to abort Provision right after the build seam, before any +// real microVM work. +type fakeEnsurer struct { + errStop error + + mu sync.Mutex + called bool + ctxErr error + hasDeadline bool +} + +func (f *fakeEnsurer) Ensure(ctx context.Context, _ string, _ *v1.Platform) (string, runspec.RunSpec, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.called = true + f.ctxErr = ctx.Err() + _, f.hasDeadline = ctx.Deadline() + return "", runspec.RunSpec{}, f.errStop +} + // testImageStore returns an image.Store over a throwaway cache dir. The // non-KVM unit tests never actually convert an image (they stop before any // Ensure call), so the store only needs to satisfy Config.validate. @@ -166,6 +190,49 @@ func TestProvisionRejectsInvalidSpec(t *testing.T) { } } +// TestProvisionImageBuildOutlivesCommandContext is the regression guard for the +// "context canceled" provision failures: the shared, idempotent image build must +// run under the runtime's lifetime, not the per-command context, so a cancelled +// command (e.g. a dropped/reconnected agent stream) cannot abort a build +// mid-stream. The fake ensurer records the context it was handed. +func TestProvisionImageBuildOutlivesCommandContext(t *testing.T) { + dir := t.TempDir() + kernel := filepath.Join(dir, "vmlinux") + if err := os.WriteFile(kernel, []byte("kernel"), 0o600); err != nil { + t.Fatalf("write kernel: %v", err) + } + fake := &fakeEnsurer{errStop: errors.New("stop after build seam")} + rt, err := New(Config{ + KernelPath: kernel, + ImageStore: fake, + ImageRef: "example.invalid/mc:{version}", + WorkDir: filepath.Join(dir, "work"), + AdvertiseHost: "10.0.0.5", + }) + if err != nil { + t.Fatalf("New: %v", err) + } + defer rt.Close() + + // An already-cancelled command context — exactly what a dropped agent stream + // hands Provision mid-flight. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + if _, err := rt.Provision(ctx, agent.VMSpec{Version: "1.20.4", CPUs: 2, MemoryMB: 1024}); err == nil { + t.Fatal("Provision: expected the fake ensurer's error") + } + if !fake.called { + t.Fatal("Ensure never ran: Provision aborted on the command ctx before the build") + } + if fake.ctxErr != nil { + t.Errorf("image build saw ctx.Err() = %v; want nil — it must not inherit the cancelled command context", fake.ctxErr) + } + if !fake.hasDeadline { + t.Error("image build ctx had no deadline; want ImagePullTimeout applied") + } +} + // TestProvisionUnresolvableImage checks Provision fails fast — before any // network pull — when the image reference can't be resolved. newTestRuntime's // ImageRef is templated with no DefaultImageRef, so a version-less spec has no diff --git a/internal/agent/link.go b/internal/agent/link.go new file mode 100644 index 0000000..fe00df3 --- /dev/null +++ b/internal/agent/link.go @@ -0,0 +1,216 @@ +package agent + +import ( + "context" + "encoding/json" + "sync" + "time" + + pb "github.com/aarani/craftling-go/internal/agentlink/pb" + "go.uber.org/zap" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// Command op names. They are the contract between the control-plane hub (which +// emits Commands) and the agent link (which dispatches them to the Runtime). +// They live here, in the shared agent package the hub already imports, so both +// ends reference the same constants. +const ( + OpProvision = "provision" + OpStart = "start" + OpStop = "stop" + OpSnapshot = "snapshot" + OpDeprovision = "deprovision" + OpStatus = "status" +) + +// VMRef is the JSON payload for the ops that act on an existing VM by id +// (everything except provision, which carries a full VMSpec). The hub marshals +// it; the agent link decodes it. +type VMRef struct { + VMID string `json:"vm_id"` +} + +// LinkInfo is what the agent announces about itself when it opens the stream. +// It mirrors the old HTTP registration minus the advertise address: the control +// plane no longer dials the agent, so there is nothing to advertise. +type LinkInfo struct { + ID string + Hostname string + Zone string + AgentVersion string + CPUsTotal int + MemoryMBTotal int +} + +const ( + // linkHeartbeatInterval is how often the agent sends a heartbeat over the + // open stream. It must stay comfortably below the control plane's host TTL. + linkHeartbeatInterval = 10 * time.Second + // linkReconnectInterval is how long to wait before redialing after the + // stream drops or the control plane is unreachable. + linkReconnectInterval = 5 * time.Second +) + +// RunLink keeps a persistent control-plane connection open for the agent's +// lifetime: it dials the control plane's gRPC AgentLink service, registers, +// then serves Commands the control plane pushes down the stream until ctx is +// cancelled. A dropped stream is retried with a fixed backoff, so a control +// plane restart heals on its own. +func RunLink(ctx context.Context, cpAddr string, rt Runtime, info LinkInfo, log *zap.Logger) { + for { + if ctx.Err() != nil { + return + } + if err := connectOnce(ctx, cpAddr, rt, info, log); err != nil && ctx.Err() == nil { + log.Warn("control-plane link dropped; reconnecting", zap.Error(err)) + } + select { + case <-ctx.Done(): + return + case <-time.After(linkReconnectInterval): + } + } +} + +// connectOnce dials the control plane, opens the stream, registers, and serves +// commands until the stream ends (returning the terminating error, if any). +func connectOnce(ctx context.Context, cpAddr string, rt Runtime, info LinkInfo, log *zap.Logger) error { + cc, err := grpc.NewClient(cpAddr, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return err + } + defer func() { _ = cc.Close() }() + + stream, err := pb.NewAgentLinkClient(cc).Connect(ctx) + if err != nil { + return err + } + + // A single mutex serializes Send across the heartbeat ticker and the + // per-command result goroutines (gRPC forbids concurrent Send on a stream). + send := &sender{stream: stream} + + if err := send.message(&pb.AgentMessage{Body: &pb.AgentMessage_Register{Register: &pb.Register{ + Id: info.ID, + Hostname: info.Hostname, + Zone: info.Zone, + CpusTotal: int32(info.CPUsTotal), + MemoryMbTotal: int32(info.MemoryMBTotal), + AgentVersion: info.AgentVersion, + }}}); err != nil { + return err + } + log.Info("connected to control plane", zap.String("addr", cpAddr)) + + // streamCtx is cancelled when this stream ends, stopping the heartbeat loop. + streamCtx, cancel := context.WithCancel(ctx) + defer cancel() + go heartbeatLoop(streamCtx, send, log) + + for { + msg, err := stream.Recv() + if err != nil { + return err + } + cmd := msg.GetCommand() + if cmd == nil { + continue + } + // Handle each command on its own goroutine so a slow op (e.g. a VM boot) + // doesn't stall the receive loop or other in-flight commands. + go func(cmd *pb.Command) { + payload, errStr := execOp(streamCtx, rt, cmd) + if err := send.message(&pb.AgentMessage{Body: &pb.AgentMessage_Result{Result: &pb.Result{ + Id: cmd.Id, + Payload: payload, + Error: errStr, + }}}); err != nil { + log.Warn("send command result", zap.String("op", cmd.Op), zap.Error(err)) + } + }(cmd) + } +} + +// heartbeatLoop sends a heartbeat on an interval until the stream ends. +func heartbeatLoop(ctx context.Context, send *sender, log *zap.Logger) { + ticker := time.NewTicker(linkHeartbeatInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := send.message(&pb.AgentMessage{Body: &pb.AgentMessage_Heartbeat{Heartbeat: &pb.Heartbeat{}}}); err != nil { + log.Warn("send heartbeat", zap.Error(err)) + return + } + } + } +} + +// execOp dispatches a command to the runtime and returns the JSON-encoded VM +// result (nil when the op returns no VM) and an error string ("" on success). +func execOp(ctx context.Context, rt Runtime, cmd *pb.Command) (payload []byte, errStr string) { + switch cmd.Op { + case OpProvision: + var spec VMSpec + if err := json.Unmarshal(cmd.Payload, &spec); err != nil { + return nil, "decode spec: " + err.Error() + } + vm, err := rt.Provision(ctx, spec) + return marshalVM(vm), errString(err) + case OpStart: + vm, err := rt.Start(ctx, vmRef(cmd)) + return marshalVM(vm), errString(err) + case OpStop: + return nil, errString(rt.Stop(ctx, vmRef(cmd))) + case OpSnapshot: + return nil, errString(rt.Snapshot(ctx, vmRef(cmd))) + case OpDeprovision: + return nil, errString(rt.Deprovision(ctx, vmRef(cmd))) + case OpStatus: + vm, err := rt.Status(ctx, vmRef(cmd)) + return marshalVM(vm), errString(err) + default: + return nil, "unknown op " + cmd.Op + } +} + +// vmRef extracts the target VM id from a command payload (best-effort; an +// undecodable payload yields an empty id, which the runtime treats as missing). +func vmRef(cmd *pb.Command) string { + var ref VMRef + _ = json.Unmarshal(cmd.Payload, &ref) + return ref.VMID +} + +// marshalVM JSON-encodes a VM, or returns nil for a nil VM (ops with no result). +func marshalVM(vm *VM) []byte { + if vm == nil { + return nil + } + b, _ := json.Marshal(vm) + return b +} + +// errString renders an error for the wire: "" for success, else its message. +func errString(err error) string { + if err == nil { + return "" + } + return err.Error() +} + +// sender serializes Send calls on a client stream. +type sender struct { + mu sync.Mutex + stream grpc.BidiStreamingClient[pb.AgentMessage, pb.ControlMessage] +} + +func (s *sender) message(m *pb.AgentMessage) error { + s.mu.Lock() + defer s.mu.Unlock() + return s.stream.Send(m) +} diff --git a/internal/agent/server.go b/internal/agent/server.go deleted file mode 100644 index c50235d..0000000 --- a/internal/agent/server.go +++ /dev/null @@ -1,125 +0,0 @@ -package agent - -import ( - "errors" - "net/http" - - "github.com/aarani/craftling-go/internal/logger" - "github.com/aarani/craftling-go/internal/middleware" - "github.com/gin-gonic/gin" - "go.uber.org/zap" -) - -// Server exposes a Runtime over HTTP so the control plane can drive local VMs. -// It is the host-side half of the agent split: the reconciler's RemoteProvisioner -// calls these endpoints instead of touching compute in-process. -// -// These routes are unauthenticated for now; per-host auth / mTLS is hardened in -// P10, alongside the control plane's matching agent-auth seam. -type Server struct { - rt Runtime -} - -// NewServer constructs an agent Server over the given runtime. -func NewServer(rt Runtime) *Server { return &Server{rt: rt} } - -// NewRouter builds the agent's Gin engine: shared middleware plus the VM -// lifecycle routes the control plane calls. -func NewRouter(rt Runtime, log *zap.Logger) *gin.Engine { - s := NewServer(rt) - - r := gin.New() - r.Use(gin.Recovery()) - r.Use(middleware.RequestID()) - r.Use(middleware.RequestLogger(log)) - - r.GET("/healthz", func(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"status": "ok"}) }) - - vms := r.Group("/vms") - { - vms.POST("", s.Provision) - vms.POST("/:id/start", s.Start) - vms.POST("/:id/stop", s.Stop) - vms.POST("/:id/snapshot", s.Snapshot) - vms.DELETE("/:id", s.Deprovision) - vms.GET("/:id", s.Status) - } - return r -} - -// Provision creates and boots a VM for the requested spec. -func (s *Server) Provision(c *gin.Context) { - var spec VMSpec - if err := c.ShouldBindJSON(&spec); err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) - return - } - vm, err := s.rt.Provision(c.Request.Context(), spec) - if err != nil { - logger.FromContext(c).Error("provision vm", zap.Error(err)) - c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"}) - return - } - c.JSON(http.StatusCreated, vm) -} - -// Start boots an existing stopped VM. -func (s *Server) Start(c *gin.Context) { - vm, err := s.rt.Start(c.Request.Context(), c.Param("id")) - if errors.Is(err, ErrVMNotFound) { - c.JSON(http.StatusNotFound, gin.H{"error": "vm not found"}) - return - } - if err != nil { - logger.FromContext(c).Error("start vm", zap.Error(err)) - c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"}) - return - } - c.JSON(http.StatusOK, vm) -} - -// Stop halts a VM without destroying it (idempotent). -func (s *Server) Stop(c *gin.Context) { - if err := s.rt.Stop(c.Request.Context(), c.Param("id")); err != nil { - logger.FromContext(c).Error("stop vm", zap.Error(err)) - c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"}) - return - } - c.JSON(http.StatusOK, gin.H{"status": "ok"}) -} - -// Snapshot takes an on-demand world snapshot of a running VM (P5c). -func (s *Server) Snapshot(c *gin.Context) { - err := s.rt.Snapshot(c.Request.Context(), c.Param("id")) - if errors.Is(err, ErrVMNotFound) { - c.JSON(http.StatusNotFound, gin.H{"error": "vm not found"}) - return - } - if err != nil { - logger.FromContext(c).Error("snapshot vm", zap.Error(err)) - c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"}) - return - } - c.JSON(http.StatusOK, gin.H{"status": "ok"}) -} - -// Deprovision destroys a VM (idempotent). -func (s *Server) Deprovision(c *gin.Context) { - if err := s.rt.Deprovision(c.Request.Context(), c.Param("id")); err != nil { - logger.FromContext(c).Error("deprovision vm", zap.Error(err)) - c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"}) - return - } - c.JSON(http.StatusOK, gin.H{"status": "ok"}) -} - -// Status reports a VM's observed state (StateMissing for an unknown id). -func (s *Server) Status(c *gin.Context) { - vm, err := s.rt.Status(c.Request.Context(), c.Param("id")) - if err != nil { - logger.FromContext(c).Error("vm status", zap.Error(err)) - c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"}) - return - } - c.JSON(http.StatusOK, vm) -} diff --git a/internal/agentlink/hub.go b/internal/agentlink/hub.go new file mode 100644 index 0000000..3b72b84 --- /dev/null +++ b/internal/agentlink/hub.go @@ -0,0 +1,278 @@ +// Package agentlink is the control-plane side of the agent control channel. The +// agent dials in and holds one long-lived bidirectional gRPC stream open; the +// Hub registers the host, tracks the live connection, and pushes VM lifecycle +// commands down the stream on the provisioner's behalf. This inverts the older +// model where the control plane dialed each agent's HTTP API: agents now need no +// inbound reachability, and the open stream is itself the host's liveness signal. +package agentlink + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + + "github.com/aarani/craftling-go/internal/agent" + pb "github.com/aarani/craftling-go/internal/agentlink/pb" + "github.com/aarani/craftling-go/internal/model" + "github.com/google/uuid" + "go.uber.org/zap" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// ErrHostNotConnected means the target host has no live agent stream, so a +// command cannot be delivered. The provisioner surfaces it like any other +// transport failure. +var ErrHostNotConnected = errors.New("host has no live agent connection") + +// HostInventory is the slice of the fleet inventory the hub drives: it (re)adds +// a host on stream open, refreshes liveness on heartbeats, and marks it down on +// disconnect. *repository.HostRepository satisfies it. +type HostInventory interface { + RegisterReserved(ctx context.Context, h *model.Host, reservedCPUs, reservedMemMB int) (*model.Host, error) + Heartbeat(ctx context.Context, id string) error + MarkDown(ctx context.Context, id string) error +} + +// CapacityReconstructor reconstructs the capacity already committed to a host id +// from the durable record, so a host re-registering after a control-plane +// restart comes back with its real allocatable. *repository.GameServerRepository +// satisfies it (this is the same seam the old HTTP register handler used). +type CapacityReconstructor interface { + UsedCapacity(ctx context.Context, hostID string) (cpus, memoryMB int, err error) +} + +// Hub is the gRPC AgentLink server plus an in-memory registry of live agent +// connections keyed by host id. Like the host inventory it fronts, the registry +// is per-process: it assumes a single control-plane instance, the same +// assumption repository.HostRepository already makes. +type Hub struct { + pb.UnimplementedAgentLinkServer + + hosts HostInventory + cap CapacityReconstructor + log *zap.Logger + + mu sync.RWMutex + conns map[string]*conn // by host id +} + +// NewHub constructs a Hub over the fleet inventory and the capacity source. +func NewHub(hosts HostInventory, cap CapacityReconstructor, log *zap.Logger) *Hub { + return &Hub{ + hosts: hosts, + cap: cap, + log: log, + conns: make(map[string]*conn), + } +} + +// conn is one live agent stream and the commands awaiting their replies. +type conn struct { + stream grpc.BidiStreamingServer[pb.AgentMessage, pb.ControlMessage] + + sendMu sync.Mutex // serializes Send (gRPC forbids concurrent Send) + + mu sync.Mutex + waiters map[string]chan *pb.Result // by command id +} + +func (c *conn) send(m *pb.ControlMessage) error { + c.sendMu.Lock() + defer c.sendMu.Unlock() + return c.stream.Send(m) +} + +func (c *conn) addWaiter(id string, ch chan *pb.Result) { + c.mu.Lock() + defer c.mu.Unlock() + c.waiters[id] = ch +} + +func (c *conn) removeWaiter(id string) { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.waiters, id) +} + +// resolve hands a result to its waiter, if one is still listening. +func (c *conn) resolve(res *pb.Result) { + c.mu.Lock() + ch, ok := c.waiters[res.Id] + c.mu.Unlock() + if !ok { + return // caller timed out and gave up + } + select { + case ch <- res: + default: // buffered chan; a duplicate result is dropped + } +} + +// Connect handles one agent's lifetime. The agent sends a Register frame first; +// the hub adds it to the inventory and registry, then services Results and +// Heartbeats until the stream ends, at which point the host is marked down. +func (h *Hub) Connect(stream grpc.BidiStreamingServer[pb.AgentMessage, pb.ControlMessage]) error { + ctx := stream.Context() + + first, err := stream.Recv() + if err != nil { + return err + } + reg := first.GetRegister() + if reg == nil { + return status.Error(codes.InvalidArgument, "first message must be register") + } + + // Reconstruct any capacity already committed to this host (only meaningful + // when the agent supplies its stable id), mirroring the old register path. + usedCPUs, usedMemMB, err := h.cap.UsedCapacity(ctx, reg.Id) + if err != nil { + h.log.Error("reconstruct host capacity", zap.Error(err)) + return status.Error(codes.Internal, "reconstruct host capacity") + } + + host, err := h.hosts.RegisterReserved(ctx, &model.Host{ + ID: reg.Id, + Hostname: reg.Hostname, + Zone: reg.Zone, + CPUsTotal: int(reg.CpusTotal), + MemoryMBTotal: int(reg.MemoryMbTotal), + AgentVersion: reg.AgentVersion, + }, usedCPUs, usedMemMB) + if err != nil { + h.log.Error("register host", zap.Error(err)) + return status.Error(codes.Internal, "register host") + } + hostID := host.ID + + c := &conn{stream: stream, waiters: make(map[string]chan *pb.Result)} + h.add(hostID, c) + h.log.Info("agent connected", zap.String("host_id", hostID), zap.String("hostname", reg.Hostname)) + + defer func() { + h.remove(hostID, c) + // MarkDown on a fresh context: stream.Context() is already cancelled. + if err := h.hosts.MarkDown(context.Background(), hostID); err != nil { + h.log.Warn("mark host down", zap.String("host_id", hostID), zap.Error(err)) + } + h.log.Info("agent disconnected", zap.String("host_id", hostID)) + }() + + for { + msg, err := stream.Recv() + if err != nil { + return err // io.EOF on clean close, or a transport error + } + switch { + case msg.GetResult() != nil: + c.resolve(msg.GetResult()) + case msg.GetHeartbeat() != nil: + if err := h.hosts.Heartbeat(ctx, hostID); err != nil { + h.log.Warn("host heartbeat", zap.String("host_id", hostID), zap.Error(err)) + } + } + } +} + +func (h *Hub) add(hostID string, c *conn) { + h.mu.Lock() + defer h.mu.Unlock() + h.conns[hostID] = c +} + +// remove drops c from the registry only if it is still the current connection +// for hostID, so a reconnect that raced ahead is not clobbered. +func (h *Hub) remove(hostID string, c *conn) { + h.mu.Lock() + defer h.mu.Unlock() + if h.conns[hostID] == c { + delete(h.conns, hostID) + } +} + +func (h *Hub) get(hostID string) *conn { + h.mu.RLock() + defer h.mu.RUnlock() + return h.conns[hostID] +} + +// Provision asks the host's agent to create and boot a VM for the spec. +func (h *Hub) Provision(ctx context.Context, hostID string, spec agent.VMSpec) (*agent.VM, error) { + return h.call(ctx, hostID, agent.OpProvision, spec) +} + +// Start asks the host's agent to boot an existing VM. +func (h *Hub) Start(ctx context.Context, hostID, vmID string) (*agent.VM, error) { + return h.call(ctx, hostID, agent.OpStart, agent.VMRef{VMID: vmID}) +} + +// Stop asks the host's agent to halt a VM without destroying it. +func (h *Hub) Stop(ctx context.Context, hostID, vmID string) error { + _, err := h.call(ctx, hostID, agent.OpStop, agent.VMRef{VMID: vmID}) + return err +} + +// Snapshot asks the host's agent to take an on-demand world snapshot. +func (h *Hub) Snapshot(ctx context.Context, hostID, vmID string) error { + _, err := h.call(ctx, hostID, agent.OpSnapshot, agent.VMRef{VMID: vmID}) + return err +} + +// Deprovision asks the host's agent to destroy a VM. +func (h *Hub) Deprovision(ctx context.Context, hostID, vmID string) error { + _, err := h.call(ctx, hostID, agent.OpDeprovision, agent.VMRef{VMID: vmID}) + return err +} + +// Status fetches a VM's observed state from the host's agent. +func (h *Hub) Status(ctx context.Context, hostID, vmID string) (*agent.VM, error) { + return h.call(ctx, hostID, agent.OpStatus, agent.VMRef{VMID: vmID}) +} + +// call sends one command down the host's stream and blocks for the correlated +// reply (or until ctx is done). It returns the decoded VM when the op yields +// one, a nil VM for ops that don't, or an error from the agent or transport. +func (h *Hub) call(ctx context.Context, hostID, op string, reqPayload any) (*agent.VM, error) { + c := h.get(hostID) + if c == nil { + return nil, ErrHostNotConnected + } + + payload, err := json.Marshal(reqPayload) + if err != nil { + return nil, fmt.Errorf("marshal %s payload: %w", op, err) + } + + id := uuid.NewString() + ch := make(chan *pb.Result, 1) + c.addWaiter(id, ch) + defer c.removeWaiter(id) + + if err := c.send(&pb.ControlMessage{Command: &pb.Command{Id: id, Op: op, Payload: payload}}); err != nil { + return nil, fmt.Errorf("send %s command: %w", op, err) + } + + select { + case <-ctx.Done(): + return nil, ctx.Err() + case res := <-ch: + if res.Error != "" { + return nil, fmt.Errorf("agent %s: %s", op, res.Error) + } + if len(res.Payload) == 0 { + return nil, nil + } + var vm agent.VM + if err := json.Unmarshal(res.Payload, &vm); err != nil { + return nil, fmt.Errorf("decode %s result: %w", op, err) + } + if vm.ID == "" { + return nil, nil + } + return &vm, nil + } +} diff --git a/internal/agentlink/hub_test.go b/internal/agentlink/hub_test.go new file mode 100644 index 0000000..510fe89 --- /dev/null +++ b/internal/agentlink/hub_test.go @@ -0,0 +1,152 @@ +package agentlink + +import ( + "context" + "encoding/json" + "net" + "sync" + "testing" + "time" + + "github.com/aarani/craftling-go/internal/agent" + pb "github.com/aarani/craftling-go/internal/agentlink/pb" + "github.com/aarani/craftling-go/internal/model" + "go.uber.org/zap" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/test/bufconn" +) + +// fakeInventory records the inventory transitions the hub drives so a test can +// assert that a stream's lifecycle (register -> heartbeat -> disconnect) is +// reflected in the fleet. +type fakeInventory struct { + registered chan string // host id, on RegisterReserved + heartbeat chan string // host id, on Heartbeat + down chan string // host id, on MarkDown +} + +func newFakeInventory() *fakeInventory { + return &fakeInventory{ + registered: make(chan string, 1), + heartbeat: make(chan string, 1), + down: make(chan string, 1), + } +} + +func (f *fakeInventory) RegisterReserved(_ context.Context, h *model.Host, _, _ int) (*model.Host, error) { + f.registered <- h.ID + return &model.Host{ID: h.ID, Hostname: h.Hostname, Status: model.HostReady}, nil +} +func (f *fakeInventory) Heartbeat(_ context.Context, id string) error { + select { + case f.heartbeat <- id: + default: + } + return nil +} +func (f *fakeInventory) MarkDown(_ context.Context, id string) error { + f.down <- id + return nil +} + +// zeroCapacity is a CapacityReconstructor that reports no committed capacity. +type zeroCapacity struct{} + +func (zeroCapacity) UsedCapacity(context.Context, string) (int, int, error) { return 0, 0, nil } + +// TestHubRoundTrip exercises the full control channel against a real (in-memory) +// gRPC stream: an agent registers, the hub pushes a Provision command and gets +// the VM back, a heartbeat refreshes liveness, and dropping the stream marks the +// host down. +func TestHubRoundTrip(t *testing.T) { + inv := newFakeInventory() + hub := NewHub(inv, zeroCapacity{}, zap.NewNop()) + + lis := bufconn.Listen(1 << 20) + srv := grpc.NewServer() + pb.RegisterAgentLinkServer(srv, hub) + go func() { _ = srv.Serve(lis) }() + defer srv.Stop() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + cc, err := grpc.NewClient("passthrough:///bufnet", + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }), + grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer func() { _ = cc.Close() }() + + stream, err := pb.NewAgentLinkClient(cc).Connect(ctx) + if err != nil { + t.Fatalf("connect: %v", err) + } + + var sendMu sync.Mutex + send := func(m *pb.AgentMessage) error { + sendMu.Lock() + defer sendMu.Unlock() + return stream.Send(m) + } + + // Register, then wait until the hub has wired the connection in. + if err := send(&pb.AgentMessage{Body: &pb.AgentMessage_Register{Register: &pb.Register{ + Id: "h1", Hostname: "host-1", CpusTotal: 4, MemoryMbTotal: 4096, + }}}); err != nil { + t.Fatalf("send register: %v", err) + } + if got := <-inv.registered; got != "h1" { + t.Fatalf("registered host = %q, want h1", got) + } + + // Agent responder: answer the one Provision command with a running VM. + go func() { + for { + msg, err := stream.Recv() + if err != nil { + return + } + cmd := msg.GetCommand() + if cmd == nil { + continue + } + vm, _ := json.Marshal(agent.VM{ID: "vm-1", ServerID: "s1", Host: "10.0.0.5", Port: 25565, State: agent.StateRunning}) + _ = send(&pb.AgentMessage{Body: &pb.AgentMessage_Result{Result: &pb.Result{Id: cmd.Id, Payload: vm}}}) + } + }() + + vm, err := hub.Provision(ctx, "h1", agent.VMSpec{ServerID: "s1", CPUs: 1, MemoryMB: 1024}) + if err != nil { + t.Fatalf("provision: %v", err) + } + if vm == nil || vm.ID != "vm-1" || vm.State != agent.StateRunning || vm.Host != "10.0.0.5" { + t.Fatalf("provisioned vm = %+v, want vm-1 running on 10.0.0.5", vm) + } + + // A command to a host with no connection is reported, not blocked. + if _, err := hub.Provision(ctx, "unknown", agent.VMSpec{}); err != ErrHostNotConnected { + t.Errorf("provision unknown host = %v, want ErrHostNotConnected", err) + } + + // Heartbeat refreshes liveness. + if err := send(&pb.AgentMessage{Body: &pb.AgentMessage_Heartbeat{Heartbeat: &pb.Heartbeat{}}}); err != nil { + t.Fatalf("send heartbeat: %v", err) + } + if got := <-inv.heartbeat; got != "h1" { + t.Fatalf("heartbeat host = %q, want h1", got) + } + + // Dropping the stream marks the host down. + _ = cc.Close() + select { + case got := <-inv.down: + if got != "h1" { + t.Fatalf("marked down host = %q, want h1", got) + } + case <-ctx.Done(): + t.Fatal("host was not marked down after disconnect") + } +} diff --git a/internal/agentlink/pb/agentlink.pb.go b/internal/agentlink/pb/agentlink.pb.go new file mode 100644 index 0000000..d8ff4df --- /dev/null +++ b/internal/agentlink/pb/agentlink.pb.go @@ -0,0 +1,519 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v3.21.12 +// source: proto/agentlink/agentlink.proto + +// Package agentlink is the control-plane <-> agent control channel. The agent +// dials the control plane and holds one long-lived bidirectional stream open; +// the control plane pushes commands down it and the agent answers. This inverts +// the older model where the control plane dialed each agent's HTTP API, so +// agents need no inbound reachability. + +package agentlinkpb + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// AgentMessage is anything the agent sends up the stream. +type AgentMessage struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Types that are valid to be assigned to Body: + // + // *AgentMessage_Register + // *AgentMessage_Result + // *AgentMessage_Heartbeat + Body isAgentMessage_Body `protobuf_oneof:"body"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentMessage) Reset() { + *x = AgentMessage{} + mi := &file_proto_agentlink_agentlink_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentMessage) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentMessage) ProtoMessage() {} + +func (x *AgentMessage) ProtoReflect() protoreflect.Message { + mi := &file_proto_agentlink_agentlink_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AgentMessage.ProtoReflect.Descriptor instead. +func (*AgentMessage) Descriptor() ([]byte, []int) { + return file_proto_agentlink_agentlink_proto_rawDescGZIP(), []int{0} +} + +func (x *AgentMessage) GetBody() isAgentMessage_Body { + if x != nil { + return x.Body + } + return nil +} + +func (x *AgentMessage) GetRegister() *Register { + if x != nil { + if x, ok := x.Body.(*AgentMessage_Register); ok { + return x.Register + } + } + return nil +} + +func (x *AgentMessage) GetResult() *Result { + if x != nil { + if x, ok := x.Body.(*AgentMessage_Result); ok { + return x.Result + } + } + return nil +} + +func (x *AgentMessage) GetHeartbeat() *Heartbeat { + if x != nil { + if x, ok := x.Body.(*AgentMessage_Heartbeat); ok { + return x.Heartbeat + } + } + return nil +} + +type isAgentMessage_Body interface { + isAgentMessage_Body() +} + +type AgentMessage_Register struct { + Register *Register `protobuf:"bytes,1,opt,name=register,proto3,oneof"` // first frame only +} + +type AgentMessage_Result struct { + Result *Result `protobuf:"bytes,2,opt,name=result,proto3,oneof"` // answer to a Command, correlated by id +} + +type AgentMessage_Heartbeat struct { + Heartbeat *Heartbeat `protobuf:"bytes,3,opt,name=heartbeat,proto3,oneof"` // liveness, on a ticker +} + +func (*AgentMessage_Register) isAgentMessage_Body() {} + +func (*AgentMessage_Result) isAgentMessage_Body() {} + +func (*AgentMessage_Heartbeat) isAgentMessage_Body() {} + +// ControlMessage is anything the control plane pushes down the stream. +type ControlMessage struct { + state protoimpl.MessageState `protogen:"open.v1"` + Command *Command `protobuf:"bytes,1,opt,name=command,proto3" json:"command,omitempty"` // a VM lifecycle command to execute locally + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ControlMessage) Reset() { + *x = ControlMessage{} + mi := &file_proto_agentlink_agentlink_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ControlMessage) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ControlMessage) ProtoMessage() {} + +func (x *ControlMessage) ProtoReflect() protoreflect.Message { + mi := &file_proto_agentlink_agentlink_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ControlMessage.ProtoReflect.Descriptor instead. +func (*ControlMessage) Descriptor() ([]byte, []int) { + return file_proto_agentlink_agentlink_proto_rawDescGZIP(), []int{1} +} + +func (x *ControlMessage) GetCommand() *Command { + if x != nil { + return x.Command + } + return nil +} + +// Register identifies the host on stream open. It mirrors the fields the old +// HTTP register carried, minus address: the control plane no longer dials the +// agent, so there is nothing to advertise. +type Register struct { + state protoimpl.MessageState `protogen:"open.v1"` + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` // agent-owned stable id (keeps identity across restarts) + Hostname string `protobuf:"bytes,2,opt,name=hostname,proto3" json:"hostname,omitempty"` + Zone string `protobuf:"bytes,3,opt,name=zone,proto3" json:"zone,omitempty"` + CpusTotal int32 `protobuf:"varint,4,opt,name=cpus_total,json=cpusTotal,proto3" json:"cpus_total,omitempty"` + MemoryMbTotal int32 `protobuf:"varint,5,opt,name=memory_mb_total,json=memoryMbTotal,proto3" json:"memory_mb_total,omitempty"` + AgentVersion string `protobuf:"bytes,6,opt,name=agent_version,json=agentVersion,proto3" json:"agent_version,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Register) Reset() { + *x = Register{} + mi := &file_proto_agentlink_agentlink_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Register) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Register) ProtoMessage() {} + +func (x *Register) ProtoReflect() protoreflect.Message { + mi := &file_proto_agentlink_agentlink_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Register.ProtoReflect.Descriptor instead. +func (*Register) Descriptor() ([]byte, []int) { + return file_proto_agentlink_agentlink_proto_rawDescGZIP(), []int{2} +} + +func (x *Register) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *Register) GetHostname() string { + if x != nil { + return x.Hostname + } + return "" +} + +func (x *Register) GetZone() string { + if x != nil { + return x.Zone + } + return "" +} + +func (x *Register) GetCpusTotal() int32 { + if x != nil { + return x.CpusTotal + } + return 0 +} + +func (x *Register) GetMemoryMbTotal() int32 { + if x != nil { + return x.MemoryMbTotal + } + return 0 +} + +func (x *Register) GetAgentVersion() string { + if x != nil { + return x.AgentVersion + } + return "" +} + +// Command is one VM lifecycle request. payload is the JSON of the existing Go +// type for the op (agent.VMSpec for provision; {"vm_id":...} otherwise), so the +// command schema stays single-sourced in Go rather than duplicated here. +type Command struct { + state protoimpl.MessageState `protogen:"open.v1"` + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` // correlation id, echoed in the matching Result + Op string `protobuf:"bytes,2,opt,name=op,proto3" json:"op,omitempty"` // provision|start|stop|snapshot|deprovision|status + Payload []byte `protobuf:"bytes,3,opt,name=payload,proto3" json:"payload,omitempty"` // JSON request body for the op + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Command) Reset() { + *x = Command{} + mi := &file_proto_agentlink_agentlink_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Command) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Command) ProtoMessage() {} + +func (x *Command) ProtoReflect() protoreflect.Message { + mi := &file_proto_agentlink_agentlink_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Command.ProtoReflect.Descriptor instead. +func (*Command) Descriptor() ([]byte, []int) { + return file_proto_agentlink_agentlink_proto_rawDescGZIP(), []int{3} +} + +func (x *Command) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *Command) GetOp() string { + if x != nil { + return x.Op + } + return "" +} + +func (x *Command) GetPayload() []byte { + if x != nil { + return x.Payload + } + return nil +} + +// Result answers a Command. payload is the JSON of agent.VM when the op returns +// one (provision/start/status); error is non-empty when the op failed. +type Result struct { + state protoimpl.MessageState `protogen:"open.v1"` + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` // matches the Command id + Payload []byte `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"` // JSON agent.VM, or empty + Error string `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"` // non-empty on failure + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Result) Reset() { + *x = Result{} + mi := &file_proto_agentlink_agentlink_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Result) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Result) ProtoMessage() {} + +func (x *Result) ProtoReflect() protoreflect.Message { + mi := &file_proto_agentlink_agentlink_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Result.ProtoReflect.Descriptor instead. +func (*Result) Descriptor() ([]byte, []int) { + return file_proto_agentlink_agentlink_proto_rawDescGZIP(), []int{4} +} + +func (x *Result) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *Result) GetPayload() []byte { + if x != nil { + return x.Payload + } + return nil +} + +func (x *Result) GetError() string { + if x != nil { + return x.Error + } + return "" +} + +// Heartbeat proves liveness over the same stream. The stream itself is the +// primary liveness signal; this keeps the control plane's heartbeat-TTL reaper +// working as a backstop. +type Heartbeat struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Heartbeat) Reset() { + *x = Heartbeat{} + mi := &file_proto_agentlink_agentlink_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Heartbeat) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Heartbeat) ProtoMessage() {} + +func (x *Heartbeat) ProtoReflect() protoreflect.Message { + mi := &file_proto_agentlink_agentlink_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Heartbeat.ProtoReflect.Descriptor instead. +func (*Heartbeat) Descriptor() ([]byte, []int) { + return file_proto_agentlink_agentlink_proto_rawDescGZIP(), []int{5} +} + +var File_proto_agentlink_agentlink_proto protoreflect.FileDescriptor + +const file_proto_agentlink_agentlink_proto_rawDesc = "" + + "\n" + + "\x1fproto/agentlink/agentlink.proto\x12\tagentlink\"\xac\x01\n" + + "\fAgentMessage\x121\n" + + "\bregister\x18\x01 \x01(\v2\x13.agentlink.RegisterH\x00R\bregister\x12+\n" + + "\x06result\x18\x02 \x01(\v2\x11.agentlink.ResultH\x00R\x06result\x124\n" + + "\theartbeat\x18\x03 \x01(\v2\x14.agentlink.HeartbeatH\x00R\theartbeatB\x06\n" + + "\x04body\">\n" + + "\x0eControlMessage\x12,\n" + + "\acommand\x18\x01 \x01(\v2\x12.agentlink.CommandR\acommand\"\xb6\x01\n" + + "\bRegister\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id\x12\x1a\n" + + "\bhostname\x18\x02 \x01(\tR\bhostname\x12\x12\n" + + "\x04zone\x18\x03 \x01(\tR\x04zone\x12\x1d\n" + + "\n" + + "cpus_total\x18\x04 \x01(\x05R\tcpusTotal\x12&\n" + + "\x0fmemory_mb_total\x18\x05 \x01(\x05R\rmemoryMbTotal\x12#\n" + + "\ragent_version\x18\x06 \x01(\tR\fagentVersion\"C\n" + + "\aCommand\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id\x12\x0e\n" + + "\x02op\x18\x02 \x01(\tR\x02op\x12\x18\n" + + "\apayload\x18\x03 \x01(\fR\apayload\"H\n" + + "\x06Result\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id\x12\x18\n" + + "\apayload\x18\x02 \x01(\fR\apayload\x12\x14\n" + + "\x05error\x18\x03 \x01(\tR\x05error\"\v\n" + + "\tHeartbeat2N\n" + + "\tAgentLink\x12A\n" + + "\aConnect\x12\x17.agentlink.AgentMessage\x1a\x19.agentlink.ControlMessage(\x010\x01BBZ@github.com/aarani/craftling-go/internal/agentlink/pb;agentlinkpbb\x06proto3" + +var ( + file_proto_agentlink_agentlink_proto_rawDescOnce sync.Once + file_proto_agentlink_agentlink_proto_rawDescData []byte +) + +func file_proto_agentlink_agentlink_proto_rawDescGZIP() []byte { + file_proto_agentlink_agentlink_proto_rawDescOnce.Do(func() { + file_proto_agentlink_agentlink_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_proto_agentlink_agentlink_proto_rawDesc), len(file_proto_agentlink_agentlink_proto_rawDesc))) + }) + return file_proto_agentlink_agentlink_proto_rawDescData +} + +var file_proto_agentlink_agentlink_proto_msgTypes = make([]protoimpl.MessageInfo, 6) +var file_proto_agentlink_agentlink_proto_goTypes = []any{ + (*AgentMessage)(nil), // 0: agentlink.AgentMessage + (*ControlMessage)(nil), // 1: agentlink.ControlMessage + (*Register)(nil), // 2: agentlink.Register + (*Command)(nil), // 3: agentlink.Command + (*Result)(nil), // 4: agentlink.Result + (*Heartbeat)(nil), // 5: agentlink.Heartbeat +} +var file_proto_agentlink_agentlink_proto_depIdxs = []int32{ + 2, // 0: agentlink.AgentMessage.register:type_name -> agentlink.Register + 4, // 1: agentlink.AgentMessage.result:type_name -> agentlink.Result + 5, // 2: agentlink.AgentMessage.heartbeat:type_name -> agentlink.Heartbeat + 3, // 3: agentlink.ControlMessage.command:type_name -> agentlink.Command + 0, // 4: agentlink.AgentLink.Connect:input_type -> agentlink.AgentMessage + 1, // 5: agentlink.AgentLink.Connect:output_type -> agentlink.ControlMessage + 5, // [5:6] is the sub-list for method output_type + 4, // [4:5] is the sub-list for method input_type + 4, // [4:4] is the sub-list for extension type_name + 4, // [4:4] is the sub-list for extension extendee + 0, // [0:4] is the sub-list for field type_name +} + +func init() { file_proto_agentlink_agentlink_proto_init() } +func file_proto_agentlink_agentlink_proto_init() { + if File_proto_agentlink_agentlink_proto != nil { + return + } + file_proto_agentlink_agentlink_proto_msgTypes[0].OneofWrappers = []any{ + (*AgentMessage_Register)(nil), + (*AgentMessage_Result)(nil), + (*AgentMessage_Heartbeat)(nil), + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_agentlink_agentlink_proto_rawDesc), len(file_proto_agentlink_agentlink_proto_rawDesc)), + NumEnums: 0, + NumMessages: 6, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_proto_agentlink_agentlink_proto_goTypes, + DependencyIndexes: file_proto_agentlink_agentlink_proto_depIdxs, + MessageInfos: file_proto_agentlink_agentlink_proto_msgTypes, + }.Build() + File_proto_agentlink_agentlink_proto = out.File + file_proto_agentlink_agentlink_proto_goTypes = nil + file_proto_agentlink_agentlink_proto_depIdxs = nil +} diff --git a/internal/agentlink/pb/agentlink_grpc.pb.go b/internal/agentlink/pb/agentlink_grpc.pb.go new file mode 100644 index 0000000..5f9d072 --- /dev/null +++ b/internal/agentlink/pb/agentlink_grpc.pb.go @@ -0,0 +1,131 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.6.2 +// - protoc v3.21.12 +// source: proto/agentlink/agentlink.proto + +// Package agentlink is the control-plane <-> agent control channel. The agent +// dials the control plane and holds one long-lived bidirectional stream open; +// the control plane pushes commands down it and the agent answers. This inverts +// the older model where the control plane dialed each agent's HTTP API, so +// agents need no inbound reachability. + +package agentlinkpb + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + AgentLink_Connect_FullMethodName = "/agentlink.AgentLink/Connect" +) + +// AgentLinkClient is the client API for AgentLink service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +// +// AgentLink is served by the control plane and dialed by every agent. +type AgentLinkClient interface { + // Connect is opened once by the agent and kept open for its lifetime. The + // agent sends a Register frame first, then Results (answers to Commands) and + // periodic Heartbeats; the control plane streams Commands back down. + Connect(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[AgentMessage, ControlMessage], error) +} + +type agentLinkClient struct { + cc grpc.ClientConnInterface +} + +func NewAgentLinkClient(cc grpc.ClientConnInterface) AgentLinkClient { + return &agentLinkClient{cc} +} + +func (c *agentLinkClient) Connect(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[AgentMessage, ControlMessage], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &AgentLink_ServiceDesc.Streams[0], AgentLink_Connect_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[AgentMessage, ControlMessage]{ClientStream: stream} + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type AgentLink_ConnectClient = grpc.BidiStreamingClient[AgentMessage, ControlMessage] + +// AgentLinkServer is the server API for AgentLink service. +// All implementations must embed UnimplementedAgentLinkServer +// for forward compatibility. +// +// AgentLink is served by the control plane and dialed by every agent. +type AgentLinkServer interface { + // Connect is opened once by the agent and kept open for its lifetime. The + // agent sends a Register frame first, then Results (answers to Commands) and + // periodic Heartbeats; the control plane streams Commands back down. + Connect(grpc.BidiStreamingServer[AgentMessage, ControlMessage]) error + mustEmbedUnimplementedAgentLinkServer() +} + +// UnimplementedAgentLinkServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedAgentLinkServer struct{} + +func (UnimplementedAgentLinkServer) Connect(grpc.BidiStreamingServer[AgentMessage, ControlMessage]) error { + return status.Error(codes.Unimplemented, "method Connect not implemented") +} +func (UnimplementedAgentLinkServer) mustEmbedUnimplementedAgentLinkServer() {} +func (UnimplementedAgentLinkServer) testEmbeddedByValue() {} + +// UnsafeAgentLinkServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to AgentLinkServer will +// result in compilation errors. +type UnsafeAgentLinkServer interface { + mustEmbedUnimplementedAgentLinkServer() +} + +func RegisterAgentLinkServer(s grpc.ServiceRegistrar, srv AgentLinkServer) { + // If the following call panics, it indicates UnimplementedAgentLinkServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } + s.RegisterService(&AgentLink_ServiceDesc, srv) +} + +func _AgentLink_Connect_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(AgentLinkServer).Connect(&grpc.GenericServerStream[AgentMessage, ControlMessage]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type AgentLink_ConnectServer = grpc.BidiStreamingServer[AgentMessage, ControlMessage] + +// AgentLink_ServiceDesc is the grpc.ServiceDesc for AgentLink service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var AgentLink_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "agentlink.AgentLink", + HandlerType: (*AgentLinkServer)(nil), + Methods: []grpc.MethodDesc{}, + Streams: []grpc.StreamDesc{ + { + StreamName: "Connect", + Handler: _AgentLink_Connect_Handler, + ServerStreams: true, + ClientStreams: true, + }, + }, + Metadata: "proto/agentlink/agentlink.proto", +} diff --git a/internal/config/config.go b/internal/config/config.go index 2240d21..666bff0 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -25,6 +25,11 @@ type Config struct { AccessTTL time.Duration RefreshTTL time.Duration + // GRPCPort is the control plane's gRPC AgentLink listener (ModeServer). It + // is separate from the HTTP API on Port: agents dial it and hold a stream + // open for the control plane to push VM commands down. + GRPCPort string + // TemplateIndexURL is the registry/marketplace index the control plane fetches // the list of game-server templates from. TemplateIndexURL string @@ -33,8 +38,9 @@ type Config struct { AdminEmail string AdminPassword string - // Agent configuration (ModeAgent only). The host worker registers with the - // control plane and exposes its VM API for the control plane to call back. + // Agent configuration (ModeAgent only). The host worker dials the control + // plane and holds a stream open over which the control plane pushes VM + // commands; the agent never exposes an inbound API. Agent AgentConfig } @@ -47,8 +53,9 @@ const ( // AgentConfig holds the host-worker settings used when Mode == ModeAgent. type AgentConfig struct { - // ControlPlaneURL is where the agent registers and heartbeats. - ControlPlaneURL string + // ControlPlaneGRPCAddr is the control plane's gRPC AgentLink address + // (host:port) the agent dials and keeps a stream open to. + ControlPlaneGRPCAddr string // Runtime selects the VM backend: "fake" (default) or "firecracker". Runtime string // Firecracker holds the real-microVM driver settings (Runtime == "firecracker"). @@ -57,9 +64,6 @@ type AgentConfig struct { ID string // Hostname identifies the host in the fleet view. Hostname string - // AdvertiseAddr is the agent's own API address the control plane calls back - // (host:port reachable from the control plane). - AdvertiseAddr string // AdvertiseHost is the player-facing connect address VMs report. AdvertiseHost string // Zone is an optional placement/locality label. @@ -148,6 +152,7 @@ func Load() *Config { JWTSecret: getEnv("JWT_SECRET", "dev-secret-change-me"), AccessTTL: getDurationEnv("ACCESS_TTL", 15*time.Minute), RefreshTTL: getDurationEnv("REFRESH_TTL", 30*24*time.Hour), + GRPCPort: getEnv("GRPC_PORT", "8090"), TemplateIndexURL: getEnv("TEMPLATE_INDEX_URL", "https://registry.craftling.io/manifest.json"), @@ -155,8 +160,8 @@ func Load() *Config { AdminPassword: getEnv("ADMIN_PASSWORD", ""), Agent: AgentConfig{ - ControlPlaneURL: getEnv("CONTROL_PLANE_URL", "http://localhost:8080"), - Runtime: getEnv("AGENT_RUNTIME", RuntimeFake), + ControlPlaneGRPCAddr: getEnv("CONTROL_PLANE_GRPC_ADDR", "localhost:8090"), + Runtime: getEnv("AGENT_RUNTIME", RuntimeFake), Firecracker: FirecrackerConfig{ BinaryPath: getEnv("FC_BINARY", ""), KernelPath: getEnv("FC_KERNEL", ""), @@ -186,7 +191,6 @@ func Load() *Config { }, ID: getEnv("AGENT_ID", ""), Hostname: getEnv("AGENT_HOSTNAME", defaultHostname()), - AdvertiseAddr: getEnv("ADVERTISE_ADDR", ""), AdvertiseHost: getEnv("ADVERTISE_HOST", "127.0.0.1"), Zone: getEnv("ZONE", ""), Version: getEnv("AGENT_VERSION", "0.1.0"), diff --git a/internal/handler/agent.go b/internal/handler/agent.go deleted file mode 100644 index 5975d05..0000000 --- a/internal/handler/agent.go +++ /dev/null @@ -1,92 +0,0 @@ -package handler - -import ( - "errors" - "net/http" - - "github.com/aarani/craftling-go/internal/logger" - "github.com/aarani/craftling-go/internal/model" - "github.com/aarani/craftling-go/internal/repository" - "github.com/gin-gonic/gin" - "go.uber.org/zap" -) - -// AgentHandler serves the agent-facing host endpoints: a host agent registers -// itself and then heartbeats to prove liveness. -type AgentHandler struct { - hosts *repository.HostRepository - servers *repository.GameServerRepository -} - -// NewAgentHandler constructs an AgentHandler. -func NewAgentHandler(hosts *repository.HostRepository, servers *repository.GameServerRepository) *AgentHandler { - return &AgentHandler{hosts: hosts, servers: servers} -} - -type registerHostRequest struct { - // ID is the agent's own stable identity. Optional, but supplying it lets a - // host keep the same id across a control-plane restart (see HostRepository). - ID string `json:"id" binding:"omitempty,uuid"` - Hostname string `json:"hostname" binding:"required,min=1,max=253"` - Address string `json:"address" binding:"required"` - Zone string `json:"zone" binding:"omitempty,max=64"` - CPUsTotal int `json:"cpus_total" binding:"required,min=1"` - MemoryMBTotal int `json:"memory_mb_total" binding:"required,min=1"` - AgentVersion string `json:"agent_version" binding:"omitempty,max=64"` -} - -// Register adds (or re-registers) the calling host to the fleet inventory and -// returns the stored record, including its assigned id. -func (h *AgentHandler) Register(c *gin.Context) { - var req registerHostRequest - if err := c.ShouldBindJSON(&req); err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) - return - } - - ctx := c.Request.Context() - - // Reconstruct any capacity already committed to this host from the durable - // record, so a host re-registering after a control-plane restart comes back - // with its real allocatable rather than a clean slate. Only meaningful when - // the agent supplies its stable id (otherwise there is nothing to match). - usedCPUs, usedMemMB, err := h.servers.UsedCapacity(ctx, req.ID) - if err != nil { - logger.FromContext(c).Error("reconstruct host capacity", zap.Error(err)) - c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"}) - return - } - - host, err := h.hosts.RegisterReserved(ctx, &model.Host{ - ID: req.ID, - Hostname: req.Hostname, - Address: req.Address, - Zone: req.Zone, - CPUsTotal: req.CPUsTotal, - MemoryMBTotal: req.MemoryMBTotal, - AgentVersion: req.AgentVersion, - }, usedCPUs, usedMemMB) - if err != nil { - logger.FromContext(c).Error("register host", zap.Error(err)) - c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"}) - return - } - c.JSON(http.StatusCreated, host) -} - -// Heartbeat refreshes the liveness timestamp for the host named in the path. A -// host the control plane has never seen (or has forgotten) gets a 404 so the -// agent knows to re-register. -func (h *AgentHandler) Heartbeat(c *gin.Context) { - err := h.hosts.Heartbeat(c.Request.Context(), c.Param("id")) - if errors.Is(err, repository.ErrNotFound) { - c.JSON(http.StatusNotFound, gin.H{"error": "host not found"}) - return - } - if err != nil { - logger.FromContext(c).Error("host heartbeat", zap.Error(err)) - c.JSON(http.StatusInternalServerError, gin.H{"error": "internal error"}) - return - } - c.JSON(http.StatusOK, gin.H{"status": "ok"}) -} diff --git a/internal/handler/router.go b/internal/handler/router.go index 368c667..fa1662d 100644 --- a/internal/handler/router.go +++ b/internal/handler/router.go @@ -36,7 +36,6 @@ func NewRouter(cfg *config.Config, log *zap.Logger, pool *pgxpool.Pool, hostRepo // The scheduler is stateless over the shared in-memory host inventory, so the // handler builds its own; the reconciler builds another over the same store. serverHandler := NewServerHandler(gameServerRepo, scheduler.New(hostRepo), registryClient) - agentHandler := NewAgentHandler(hostRepo, gameServerRepo) templateHandler := NewTemplateHandler(registryClient) r := gin.New() @@ -90,14 +89,9 @@ func NewRouter(cfg *config.Config, log *zap.Logger, pool *pgxpool.Pool, hostRepo admin.GET("/hosts", adminHandler.ListHosts) } - // Agent-facing routes. Hosts register and heartbeat here. Auth is a - // placeholder until P10 (per-host tokens / mTLS). - agent := api.Group("/agent") - agent.Use(middleware.AgentAuth()) - { - agent.POST("/hosts/register", agentHandler.Register) - agent.POST("/hosts/:id/heartbeat", agentHandler.Heartbeat) - } + // Hosts no longer register/heartbeat over HTTP: each agent holds a + // persistent gRPC stream to the control plane (see internal/agentlink), + // which both delivers commands and serves as the host's liveness signal. } return r diff --git a/internal/middleware/agent.go b/internal/middleware/agent.go deleted file mode 100644 index dfd2388..0000000 --- a/internal/middleware/agent.go +++ /dev/null @@ -1,14 +0,0 @@ -package middleware - -import "github.com/gin-gonic/gin" - -// AgentAuth guards the agent-facing endpoints (/api/v1/agent/*). -// -// PLACEHOLDER: agents are not yet authenticated. P10 replaces this with -// per-host tokens or mTLS with rotation. It exists now as the seam those -// credentials will plug into, so routing and handlers don't move later. -func AgentAuth() gin.HandlerFunc { - return func(c *gin.Context) { - c.Next() - } -} diff --git a/internal/provisioner/remote.go b/internal/provisioner/remote.go index 66166f3..82f35fb 100644 --- a/internal/provisioner/remote.go +++ b/internal/provisioner/remote.go @@ -3,7 +3,6 @@ package provisioner import ( "context" "errors" - "fmt" "github.com/aarani/craftling-go/internal/agent" "github.com/aarani/craftling-go/internal/model" @@ -16,29 +15,36 @@ import ( // indicates a logic error rather than a transient condition. var ErrUnplaced = errors.New("server has no host assigned") -// HostResolver looks up a host by id to find the agent's address. The in-memory -// repository.HostRepository satisfies it. -type HostResolver interface { - GetByID(ctx context.Context, id string) (*model.Host, error) +// Commander delivers VM lifecycle commands to a host's agent over the agent's +// live control-plane connection, keyed by host id. It replaces the old outbound +// HTTP client: the control plane no longer dials agents, it pushes commands down +// the stream each agent holds open. *agentlink.Hub satisfies it. +type Commander interface { + Provision(ctx context.Context, hostID string, spec agent.VMSpec) (*agent.VM, error) + Start(ctx context.Context, hostID, vmID string) (*agent.VM, error) + Stop(ctx context.Context, hostID, vmID string) error + Snapshot(ctx context.Context, hostID, vmID string) error + Deprovision(ctx context.Context, hostID, vmID string) error + Status(ctx context.Context, hostID, vmID string) (*agent.VM, error) } -// RemoteProvisioner implements Provisioner by calling the agent on the host the -// scheduler assigned. The reconciler's calls keep the same shape as with Fake — -// they just become a network hop to the host that actually runs the VM, honoring -// the invariant that the control plane never touches KVM itself. +// RemoteProvisioner implements Provisioner by sending commands to the agent on +// the host the scheduler assigned. The reconciler's calls keep the same shape as +// with Fake — they become a message down the host's open stream rather than an +// in-process call, honoring the invariant that the control plane never touches +// KVM itself. type RemoteProvisioner struct { - hosts HostResolver - client *agent.Client + cmd Commander } -// NewRemote constructs a RemoteProvisioner over a host resolver and agent client. -func NewRemote(hosts HostResolver, client *agent.Client) *RemoteProvisioner { - return &RemoteProvisioner{hosts: hosts, client: client} +// NewRemote constructs a RemoteProvisioner over a command channel (the hub). +func NewRemote(cmd Commander) *RemoteProvisioner { + return &RemoteProvisioner{cmd: cmd} } // Provision asks the assigned host's agent to create and boot a VM. func (p *RemoteProvisioner) Provision(ctx context.Context, s *model.GameServer) (*Instance, error) { - base, err := p.baseURL(ctx, s) + hostID, err := assignedHost(s) if err != nil { return nil, err } @@ -59,7 +65,7 @@ func (p *RemoteProvisioner) Provision(ctx context.Context, s *model.GameServer) if len(s.Env) > 0 { spec.RunSpec = &runspec.RunSpec{Env: registry.SortedEnv(s.Env)} } - vm, err := p.client.Provision(ctx, base, spec) + vm, err := p.cmd.Provision(ctx, hostID, spec) if err != nil { return nil, err } @@ -72,11 +78,11 @@ func (p *RemoteProvisioner) Start(ctx context.Context, s *model.GameServer) (*In if s.VMID == nil || *s.VMID == "" { return p.Provision(ctx, s) } - base, err := p.baseURL(ctx, s) + hostID, err := assignedHost(s) if err != nil { return nil, err } - vm, err := p.client.Start(ctx, base, *s.VMID) + vm, err := p.cmd.Start(ctx, hostID, *s.VMID) if err != nil { return nil, err } @@ -88,11 +94,11 @@ func (p *RemoteProvisioner) Stop(ctx context.Context, s *model.GameServer) error if s.VMID == nil || *s.VMID == "" { return nil } - base, err := p.baseURL(ctx, s) + hostID, err := assignedHost(s) if err != nil { return err } - return p.client.Stop(ctx, base, *s.VMID) + return p.cmd.Stop(ctx, hostID, *s.VMID) } // Deprovision tears down the VM on its host (idempotent). A server that was @@ -101,11 +107,7 @@ func (p *RemoteProvisioner) Deprovision(ctx context.Context, s *model.GameServer if s.HostID == nil || *s.HostID == "" || s.VMID == nil || *s.VMID == "" { return nil } - base, err := p.baseURL(ctx, s) - if err != nil { - return err - } - return p.client.Deprovision(ctx, base, *s.VMID) + return p.cmd.Deprovision(ctx, *s.HostID, *s.VMID) } // Status reports the VM's observed state as seen by its host's agent. @@ -113,11 +115,11 @@ func (p *RemoteProvisioner) Status(ctx context.Context, s *model.GameServer) (St if s.VMID == nil || *s.VMID == "" { return StateMissing, nil } - base, err := p.baseURL(ctx, s) + hostID, err := assignedHost(s) if err != nil { return "", err } - vm, err := p.client.Status(ctx, base, *s.VMID) + vm, err := p.cmd.Status(ctx, hostID, *s.VMID) if err != nil { return "", err } @@ -130,23 +132,19 @@ func (p *RemoteProvisioner) Snapshot(ctx context.Context, s *model.GameServer) e if s.VMID == nil || *s.VMID == "" { return nil } - base, err := p.baseURL(ctx, s) + hostID, err := assignedHost(s) if err != nil { return err } - return p.client.Snapshot(ctx, base, *s.VMID) + return p.cmd.Snapshot(ctx, hostID, *s.VMID) } -// baseURL resolves the agent base URL for the server's assigned host. -func (p *RemoteProvisioner) baseURL(ctx context.Context, s *model.GameServer) (string, error) { +// assignedHost returns the server's assigned host id, or ErrUnplaced. +func assignedHost(s *model.GameServer) (string, error) { if s.HostID == nil || *s.HostID == "" { return "", ErrUnplaced } - h, err := p.hosts.GetByID(ctx, *s.HostID) - if err != nil { - return "", fmt.Errorf("resolve host %s: %w", *s.HostID, err) - } - return agent.BaseURL(h.Address), nil + return *s.HostID, nil } // instanceOf maps an agent VM to a provisioner Instance. diff --git a/internal/provisioner/remote_test.go b/internal/provisioner/remote_test.go index d33fa4d..2d8c97b 100644 --- a/internal/provisioner/remote_test.go +++ b/internal/provisioner/remote_test.go @@ -3,32 +3,79 @@ package provisioner import ( "context" "errors" - "net/http/httptest" "testing" "github.com/aarani/craftling-go/internal/agent" "github.com/aarani/craftling-go/internal/model" - "go.uber.org/zap" ) -// stubResolver resolves every host id to a fixed agent address. -type stubResolver struct{ addr string } +// fakeCommander routes commands to an in-process Runtime, ignoring the host id — +// it stands in for the hub so the provisioner can be driven without a real gRPC +// stream. It is the seam the control plane pushes commands through. +type fakeCommander struct{ rt agent.Runtime } -func (s stubResolver) GetByID(_ context.Context, id string) (*model.Host, error) { - return &model.Host{ID: id, Address: s.addr}, nil +func (c fakeCommander) Provision(ctx context.Context, _ string, spec agent.VMSpec) (*agent.VM, error) { + return c.rt.Provision(ctx, spec) +} +func (c fakeCommander) Start(ctx context.Context, _, vmID string) (*agent.VM, error) { + return c.rt.Start(ctx, vmID) +} +func (c fakeCommander) Stop(ctx context.Context, _, vmID string) error { + return c.rt.Stop(ctx, vmID) +} +func (c fakeCommander) Snapshot(ctx context.Context, _, vmID string) error { + return c.rt.Snapshot(ctx, vmID) +} +func (c fakeCommander) Deprovision(ctx context.Context, _, vmID string) error { + return c.rt.Deprovision(ctx, vmID) +} +func (c fakeCommander) Status(ctx context.Context, _, vmID string) (*agent.VM, error) { + return c.rt.Status(ctx, vmID) +} + +// errCommander fails loudly on every call, so a test can assert a code path is a +// no-op that never reaches the command channel. +type errCommander struct{ t *testing.T } + +func (c errCommander) Provision(context.Context, string, agent.VMSpec) (*agent.VM, error) { + c.t.Helper() + c.t.Fatal("Provision called, want no-op") + return nil, errors.New("unreachable") +} +func (c errCommander) Start(context.Context, string, string) (*agent.VM, error) { + c.t.Helper() + c.t.Fatal("Start called, want no-op") + return nil, errors.New("unreachable") +} +func (c errCommander) Stop(context.Context, string, string) error { + c.t.Helper() + c.t.Fatal("Stop called, want no-op") + return nil +} +func (c errCommander) Snapshot(context.Context, string, string) error { + c.t.Helper() + c.t.Fatal("Snapshot called, want no-op") + return nil +} +func (c errCommander) Deprovision(context.Context, string, string) error { + c.t.Helper() + c.t.Fatal("Deprovision called, want no-op") + return nil +} +func (c errCommander) Status(context.Context, string, string) (*agent.VM, error) { + c.t.Helper() + c.t.Fatal("Status called, want no-op") + return nil, errors.New("unreachable") } func ptr(s string) *string { return &s } // TestRemoteProvisionerLifecycle drives a game server through provision → stop → -// start → deprovision against a real in-process agent, asserting the observed -// state reported back across the seam at each step. +// start → deprovision against an in-process runtime behind the command channel, +// asserting the observed state reported back at each step. func TestRemoteProvisionerLifecycle(t *testing.T) { ctx := context.Background() - srv := httptest.NewServer(agent.NewRouter(agent.NewFakeRuntime("10.0.0.20"), zap.NewNop())) - defer srv.Close() - - p := NewRemote(stubResolver{addr: srv.URL}, agent.NewClient(nil)) + p := NewRemote(fakeCommander{rt: agent.NewFakeRuntime("10.0.0.20")}) s := &model.GameServer{ ID: "srv-1", HostID: ptr("host-1"), @@ -67,15 +114,15 @@ func TestRemoteProvisionerLifecycle(t *testing.T) { // TestRemoteProvisionerUnplaced verifies provisioning without a host assignment // is a logic error, while teardown of an unplaced/unprovisioned server is a -// harmless no-op. +// harmless no-op that never reaches the command channel. func TestRemoteProvisionerUnplaced(t *testing.T) { ctx := context.Background() - p := NewRemote(stubResolver{addr: "http://127.0.0.1:1"}, agent.NewClient(nil)) + p := NewRemote(errCommander{t: t}) if _, err := p.Provision(ctx, &model.GameServer{ID: "x"}); !errors.Is(err, ErrUnplaced) { t.Errorf("provision unplaced = %v, want ErrUnplaced", err) } - // No host and no VM: nothing to tear down, and we must not dial anyone. + // No host and no VM: nothing to tear down, and we must not send a command. if err := p.Deprovision(ctx, &model.GameServer{ID: "x"}); err != nil { t.Errorf("deprovision unplaced = %v, want nil", err) } @@ -88,10 +135,7 @@ func TestRemoteProvisionerUnplaced(t *testing.T) { // back to provisioning a fresh one. func TestRemoteProvisionerStartProvisions(t *testing.T) { ctx := context.Background() - srv := httptest.NewServer(agent.NewRouter(agent.NewFakeRuntime("10.0.0.21"), zap.NewNop())) - defer srv.Close() - - p := NewRemote(stubResolver{addr: srv.URL}, agent.NewClient(nil)) + p := NewRemote(fakeCommander{rt: agent.NewFakeRuntime("10.0.0.21")}) s := &model.GameServer{ID: "srv-2", HostID: ptr("host-2"), Version: "1.20.4", CPUs: 1, MemoryMB: 1024} inst, err := p.Start(ctx, s) @@ -104,14 +148,10 @@ func TestRemoteProvisionerStartProvisions(t *testing.T) { } // TestRemoteProvisionerSnapshot verifies a snapshot of a provisioned server is -// forwarded to its host's agent, and that a server with no VM is a no-op (no -// dial). +// forwarded to its host's agent, and that a server with no VM is a no-op. func TestRemoteProvisionerSnapshot(t *testing.T) { ctx := context.Background() - srv := httptest.NewServer(agent.NewRouter(agent.NewFakeRuntime("10.0.0.22"), zap.NewNop())) - defer srv.Close() - - p := NewRemote(stubResolver{addr: srv.URL}, agent.NewClient(nil)) + p := NewRemote(fakeCommander{rt: agent.NewFakeRuntime("10.0.0.22")}) s := &model.GameServer{ID: "srv-3", HostID: ptr("host-3"), Version: "1.20.4", CPUs: 1, MemoryMB: 1024} inst, err := p.Provision(ctx, s) @@ -123,9 +163,8 @@ func TestRemoteProvisionerSnapshot(t *testing.T) { t.Fatalf("snapshot: %v", err) } - // No VM: nothing to snapshot, and we must not dial anyone (the resolver - // points at an unroutable address, so a dial would error). - dead := NewRemote(stubResolver{addr: "http://127.0.0.1:1"}, agent.NewClient(nil)) + // No VM: nothing to snapshot, and we must not send a command. + dead := NewRemote(errCommander{t: t}) if err := dead.Snapshot(ctx, &model.GameServer{ID: "x", HostID: ptr("h")}); err != nil { t.Errorf("snapshot with no vm = %v, want nil", err) } @@ -149,10 +188,7 @@ func (r *recordingRuntime) Provision(ctx context.Context, spec agent.VMSpec) (*a func TestRemoteProvisionerDeliversTemplate(t *testing.T) { ctx := context.Background() rt := &recordingRuntime{FakeRuntime: agent.NewFakeRuntime("10.0.0.30")} - srv := httptest.NewServer(agent.NewRouter(rt, zap.NewNop())) - defer srv.Close() - - p := NewRemote(stubResolver{addr: srv.URL}, agent.NewClient(nil)) + p := NewRemote(fakeCommander{rt: rt}) imageRef := "itzg/minecraft-server:java21" tmpl := &model.GameServer{ diff --git a/internal/reconciler/reconciler.go b/internal/reconciler/reconciler.go index 5beb056..0c4d16f 100644 --- a/internal/reconciler/reconciler.go +++ b/internal/reconciler/reconciler.go @@ -43,11 +43,15 @@ func (r *Reconciler) Run(ctx context.Context, interval time.Duration) { } // ReconcileOnce processes one batch of servers needing reconciliation. +// +// It runs under the reconciler's lifetime context, with no per-batch deadline: +// a single step can legitimately take minutes (a cold image build pulls and +// flattens a multi-hundred-MB image), and a short request-scoped timeout here +// would abort that work and flip the server to an error status while the agent +// is still making progress. Provisioning is bounded by the agent-side image +// pull timeout and by process shutdown (ctx cancellation), not by this loop. func (r *Reconciler) ReconcileOnce(ctx context.Context) { - opCtx, cancel := context.WithTimeout(ctx, 30*time.Second) - defer cancel() - - servers, err := r.servers.ListReconcilable(opCtx) + servers, err := r.servers.ListReconcilable(ctx) if err != nil { r.log.Error("list reconcilable servers", zap.Error(err)) return @@ -55,9 +59,9 @@ func (r *Reconciler) ReconcileOnce(ctx context.Context) { for i := range servers { s := &servers[i] - if err := r.reconcile(opCtx, s); err != nil { + if err := r.reconcile(ctx, s); err != nil { r.log.Error("reconcile server", zap.String("id", s.ID), zap.Error(err)) - _ = r.servers.MarkStatus(opCtx, s.ID, model.StatusError, err.Error()) + _ = r.servers.MarkStatus(ctx, s.ID, model.StatusError, err.Error()) } } } diff --git a/internal/repository/host.go b/internal/repository/host.go index 7ff3842..11d830c 100644 --- a/internal/repository/host.go +++ b/internal/repository/host.go @@ -203,6 +203,24 @@ func (r *HostRepository) Release(_ context.Context, id string, cpus, memMB int) return nil } +// MarkDown marks a single host down, the immediate counterpart to MarkStale: +// the hub calls it the moment an agent's stream drops, so a disconnected host +// stops being scheduled without waiting for its heartbeat TTL to lapse. An +// unknown host is a no-op (the fleet lives in memory; a control-plane restart +// can legitimately forget a host that later reconnects). +func (r *HostRepository) MarkDown(_ context.Context, id string) error { + r.mu.Lock() + defer r.mu.Unlock() + + h, ok := r.hosts[id] + if !ok || h.Status == model.HostDown { + return nil + } + h.Status = model.HostDown + h.UpdatedAt = now() + return nil +} + // MarkStale marks every host whose last heartbeat predates cutoff as down, and // returns how many transitioned. Already-down hosts are left untouched. func (r *HostRepository) MarkStale(_ context.Context, cutoff time.Time) (int, error) { diff --git a/internal/repository/host_test.go b/internal/repository/host_test.go index 2c15140..74eb3d3 100644 --- a/internal/repository/host_test.go +++ b/internal/repository/host_test.go @@ -24,6 +24,31 @@ func TestRegisterReservedNewHost(t *testing.T) { } } +// TestMarkDown verifies a host is marked down on demand (the hub's +// disconnect path), and that marking an unknown host is a harmless no-op. +func TestMarkDown(t *testing.T) { + repo := NewHostRepository() + if _, err := repo.RegisterReserved(context.Background(), newHost("a", 4, 4096), 0, 0); err != nil { + t.Fatalf("register: %v", err) + } + + if err := repo.MarkDown(context.Background(), "a"); err != nil { + t.Fatalf("mark down: %v", err) + } + h, err := repo.GetByID(context.Background(), "a") + if err != nil { + t.Fatalf("get: %v", err) + } + if h.Status != model.HostDown { + t.Errorf("status = %q, want %q", h.Status, model.HostDown) + } + + // Unknown host: no error (a control-plane restart can forget a host). + if err := repo.MarkDown(context.Background(), "ghost"); err != nil { + t.Errorf("mark down unknown = %v, want nil", err) + } +} + // TestRegisterReservedClampsNegative guards against a reconstructed reservation // exceeding the host's reported total (allocatable floors at zero). func TestRegisterReservedClampsNegative(t *testing.T) { diff --git a/proto/agentlink/agentlink.proto b/proto/agentlink/agentlink.proto new file mode 100644 index 0000000..79638d1 --- /dev/null +++ b/proto/agentlink/agentlink.proto @@ -0,0 +1,66 @@ +syntax = "proto3"; + +// Package agentlink is the control-plane <-> agent control channel. The agent +// dials the control plane and holds one long-lived bidirectional stream open; +// the control plane pushes commands down it and the agent answers. This inverts +// the older model where the control plane dialed each agent's HTTP API, so +// agents need no inbound reachability. +package agentlink; + +option go_package = "github.com/aarani/craftling-go/internal/agentlink/pb;agentlinkpb"; + +// AgentLink is served by the control plane and dialed by every agent. +service AgentLink { + // Connect is opened once by the agent and kept open for its lifetime. The + // agent sends a Register frame first, then Results (answers to Commands) and + // periodic Heartbeats; the control plane streams Commands back down. + rpc Connect(stream AgentMessage) returns (stream ControlMessage); +} + +// AgentMessage is anything the agent sends up the stream. +message AgentMessage { + oneof body { + Register register = 1; // first frame only + Result result = 2; // answer to a Command, correlated by id + Heartbeat heartbeat = 3; // liveness, on a ticker + } +} + +// ControlMessage is anything the control plane pushes down the stream. +message ControlMessage { + Command command = 1; // a VM lifecycle command to execute locally +} + +// Register identifies the host on stream open. It mirrors the fields the old +// HTTP register carried, minus address: the control plane no longer dials the +// agent, so there is nothing to advertise. +message Register { + string id = 1; // agent-owned stable id (keeps identity across restarts) + string hostname = 2; + string zone = 3; + int32 cpus_total = 4; + int32 memory_mb_total = 5; + string agent_version = 6; +} + +// Command is one VM lifecycle request. payload is the JSON of the existing Go +// type for the op (agent.VMSpec for provision; {"vm_id":...} otherwise), so the +// command schema stays single-sourced in Go rather than duplicated here. +message Command { + string id = 1; // correlation id, echoed in the matching Result + string op = 2; // provision|start|stop|snapshot|deprovision|status + bytes payload = 3; // JSON request body for the op +} + +// Result answers a Command. payload is the JSON of agent.VM when the op returns +// one (provision/start/status); error is non-empty when the op failed. +message Result { + string id = 1; // matches the Command id + bytes payload = 2; // JSON agent.VM, or empty + string error = 3; // non-empty on failure +} + +// Heartbeat proves liveness over the same stream. The stream itself is the +// primary liveness signal; this keeps the control plane's heartbeat-TTL reaper +// working as a backstop. +message Heartbeat {} diff --git a/test/e2e/agent_test.go b/test/e2e/agent_test.go index 47fad0a..3541d84 100644 --- a/test/e2e/agent_test.go +++ b/test/e2e/agent_test.go @@ -3,14 +3,16 @@ package e2e import ( - "encoding/json" + "context" "net/http" "testing" + + "github.com/aarani/craftling-go/internal/agent" ) // TestAgentSeam verifies the control plane drives the VM on the host agent -// across the network seam (P3): a created server's VM actually exists and runs -// on the in-process agent, and deleting the server tears that VM down. +// across the gRPC stream seam (P3): a created server's VM actually exists and +// runs on the in-process agent, and deleting the server tears that VM down. func TestAgentSeam(t *testing.T) { user := registerUser(t, "seam-user@example.com", "hunter2pass") tok := user.AccessToken @@ -25,11 +27,11 @@ func TestAgentSeam(t *testing.T) { // The agent must report this VM as running, and tagged with the server id. vm := agentVM(t, vmID) - if vm["state"] != "running" { - t.Errorf("agent vm state = %v, want running", vm["state"]) + if vm.State != agent.StateRunning { + t.Errorf("agent vm state = %v, want running", vm.State) } - if vm["server_id"] != id { - t.Errorf("agent vm server_id = %v, want %s", vm["server_id"], id) + if vm.ServerID != id { + t.Errorf("agent vm server_id = %v, want %s", vm.ServerID, id) } // Deleting the server deprovisions the VM on the agent. @@ -39,25 +41,19 @@ func TestAgentSeam(t *testing.T) { } waitForGone(t, tok, id) - if vm := agentVM(t, vmID); vm["state"] != "missing" { - t.Errorf("after delete, agent vm state = %v, want missing", vm["state"]) + if vm := agentVM(t, vmID); vm.State != agent.StateMissing { + t.Errorf("after delete, agent vm state = %v, want missing", vm.State) } } -// agentVM fetches a VM's record directly from the in-process agent API. -func agentVM(t *testing.T, vmID string) map[string]any { +// agentVM fetches a VM's record directly from the placement host's in-process +// runtime — the same FakeRuntime the control plane drives over the stream — so +// the test can confirm the command actually landed on the agent. +func agentVM(t *testing.T, vmID string) *agent.VM { t.Helper() - resp, err := http.Get(agentBaseURL + "/vms/" + vmID) + vm, err := fakeRT.Status(context.Background(), vmID) if err != nil { - t.Fatalf("get agent vm: %v", err) - } - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - t.Fatalf("agent vm status = %d", resp.StatusCode) - } - var vm map[string]any - if err := json.NewDecoder(resp.Body).Decode(&vm); err != nil { - t.Fatalf("decode agent vm: %v", err) + t.Fatalf("agent vm status: %v", err) } return vm } diff --git a/test/e2e/hosts_test.go b/test/e2e/hosts_test.go index e2f5224..202809d 100644 --- a/test/e2e/hosts_test.go +++ b/test/e2e/hosts_test.go @@ -4,38 +4,18 @@ package e2e import ( "encoding/json" - "net/http" "testing" "time" -) -// registerHost registers a host via the agent endpoint and returns its record. -func registerHost(t *testing.T, hostname string) map[string]any { - t.Helper() - resp, body := doJSON(t, http.MethodPost, "/api/v1/agent/hosts/register", "", map[string]any{ - "hostname": hostname, - "address": "10.0.0.1:9000", - "zone": "zone-a", - "cpus_total": 8, - "memory_mb_total": 16384, - "agent_version": "0.1.0", - }) - if resp.StatusCode != http.StatusCreated { - t.Fatalf("register host status = %d, body = %s", resp.StatusCode, body) - } - var h map[string]any - if err := json.Unmarshal(body, &h); err != nil { - t.Fatalf("decode host: %v (body=%s)", err, body) - } - return h -} + "github.com/aarani/craftling-go/internal/agent" +) // adminHostByID returns the fleet host with the given id from the admin view, // or nil if absent. func adminHostByID(t *testing.T, adminToken, id string) map[string]any { t.Helper() resp, body := get(t, "/api/v1/admin/hosts", adminToken) - if resp.StatusCode != http.StatusOK { + if resp.StatusCode != 200 { t.Fatalf("list hosts status = %d, body = %s", resp.StatusCode, body) } var out struct { @@ -65,144 +45,81 @@ func waitForHostStatus(t *testing.T, adminToken, id, want string) { t.Fatalf("host %s did not reach status %q within timeout", id, want) } -// TestHostFleetLifecycle exercises register -> heartbeat -> stale -> down -> -// recover through the agent endpoints and the admin fleet view. +// TestHostFleetLifecycle exercises a host's liveness over the gRPC stream: +// connecting registers it ready, dropping the stream marks it down, and +// reconnecting brings it back — the stream is the host's liveness signal now +// that the HTTP register/heartbeat endpoints are gone. func TestHostFleetLifecycle(t *testing.T) { admin := makeAdmin(t, "host-fleet-admin@example.com", "hunter2pass") - host := registerHost(t, "host-lifecycle") - id, _ := host["id"].(string) - if id == "" { - t.Fatalf("no id in register response: %v", host) - } - if host["status"] != "ready" { - t.Errorf("status = %v, want ready", host["status"]) + const id = "44444444-4444-4444-4444-444444444444" + info := agent.LinkInfo{ + ID: id, Hostname: "host-lifecycle", Zone: "zone-a", + CPUsTotal: 8, MemoryMBTotal: 16384, AgentVersion: "0.1.0", } - // Allocatable capacity is initialised to total on registration. - if host["cpus_allocatable"] != host["cpus_total"] { - t.Errorf("cpus_allocatable = %v, want %v", host["cpus_allocatable"], host["cpus_total"]) + + // Connecting the agent registers the host as ready. + stop := startAgent(t, info) + waitForHostStatus(t, admin.AccessToken, id, "ready") + + h := adminHostByID(t, admin.AccessToken, id) + if h == nil { + t.Fatalf("host not present in admin fleet view after connect") } - if host["memory_mb_allocatable"] != host["memory_mb_total"] { - t.Errorf("memory_mb_allocatable = %v, want %v", host["memory_mb_allocatable"], host["memory_mb_total"]) + // Allocatable capacity is initialised to total on a fresh registration. + if h["cpus_allocatable"] != h["cpus_total"] { + t.Errorf("cpus_allocatable = %v, want %v", h["cpus_allocatable"], h["cpus_total"]) } - - // A heartbeat keeps the host alive. - resp, body := doJSON(t, http.MethodPost, "/api/v1/agent/hosts/"+id+"/heartbeat", "", nil) - if resp.StatusCode != http.StatusOK { - t.Fatalf("heartbeat status = %d, body = %s", resp.StatusCode, body) + if h["memory_mb_allocatable"] != h["memory_mb_total"] { + t.Errorf("memory_mb_allocatable = %v, want %v", h["memory_mb_allocatable"], h["memory_mb_total"]) } - // Stop heartbeating: the reaper marks the host down once its TTL lapses. + // Dropping the stream marks the host down. + stop() waitForHostStatus(t, admin.AccessToken, id, "down") - // A fresh heartbeat brings a downed host back to ready. - resp, _ = doJSON(t, http.MethodPost, "/api/v1/agent/hosts/"+id+"/heartbeat", "", nil) - if resp.StatusCode != http.StatusOK { - t.Fatalf("recovery heartbeat status = %d", resp.StatusCode) - } - if h := adminHostByID(t, admin.AccessToken, id); h == nil || h["status"] != "ready" { - t.Fatalf("host did not recover to ready: %v", h) - } -} - -// TestHostReRegisterKeepsID verifies that re-registering the same hostname -// updates the existing record in place rather than creating a duplicate. -func TestHostReRegisterKeepsID(t *testing.T) { - first := registerHost(t, "host-stable") - second := registerHost(t, "host-stable") - if first["id"] != second["id"] { - t.Errorf("re-register changed id: %v -> %v", first["id"], second["id"]) - } + // Reconnecting brings a downed host back to ready. + stop2 := startAgent(t, info) + defer stop2() + waitForHostStatus(t, admin.AccessToken, id, "ready") } -// TestRegisterWithAgentSuppliedID verifies that an agent-owned id is honored and -// is the authoritative key on re-registration — the basis for identity surviving -// a control-plane restart. A second register under the same id updates the -// existing record in place rather than minting a new one. +// TestRegisterWithAgentSuppliedID verifies the agent-owned id is the +// authoritative key on the stream: a reconnect under the same id updates the +// existing fleet record in place (here, with changed capacity) rather than +// minting a new one — the basis for host identity surviving a reconnect or a +// control-plane restart. func TestRegisterWithAgentSuppliedID(t *testing.T) { - const agentID = "11111111-1111-1111-1111-111111111111" - - resp, body := doJSON(t, http.MethodPost, "/api/v1/agent/hosts/register", "", map[string]any{ - "id": agentID, - "hostname": "host-owned-id", - "address": "10.0.0.5:9000", - "cpus_total": 4, - "memory_mb_total": 8192, - }) - if resp.StatusCode != http.StatusCreated { - t.Fatalf("register status = %d, body = %s", resp.StatusCode, body) - } - var first map[string]any - if err := json.Unmarshal(body, &first); err != nil { - t.Fatalf("decode: %v", err) - } - if first["id"] != agentID { - t.Fatalf("id = %v, want agent-supplied %s", first["id"], agentID) - } + admin := makeAdmin(t, "agent-id-admin@example.com", "hunter2pass") - // Re-register under the same id with changed capacity: same record, updated. - resp, body = doJSON(t, http.MethodPost, "/api/v1/agent/hosts/register", "", map[string]any{ - "id": agentID, - "hostname": "host-owned-id", - "address": "10.0.0.5:9000", - "cpus_total": 16, - "memory_mb_total": 32768, + const id = "11111111-1111-1111-1111-111111111111" + + stop := startAgent(t, agent.LinkInfo{ + ID: id, Hostname: "host-owned-id", CPUsTotal: 4, MemoryMBTotal: 8192, }) - if resp.StatusCode != http.StatusCreated { - t.Fatalf("re-register status = %d, body = %s", resp.StatusCode, body) - } - var second map[string]any - if err := json.Unmarshal(body, &second); err != nil { - t.Fatalf("decode: %v", err) - } - if second["id"] != agentID { - t.Errorf("re-register changed id: %v", second["id"]) - } - if second["cpus_total"].(float64) != 16 { - t.Errorf("cpus_total = %v, want 16 (updated in place)", second["cpus_total"]) + waitForHostStatus(t, admin.AccessToken, id, "ready") + if h := adminHostByID(t, admin.AccessToken, id); h == nil || h["id"] != id { + t.Fatalf("host not registered under agent-supplied id %s: %v", id, h) } -} -// TestRegisterRejectsBadID verifies a malformed agent id is rejected. -func TestRegisterRejectsBadID(t *testing.T) { - resp, body := doJSON(t, http.MethodPost, "/api/v1/agent/hosts/register", "", map[string]any{ - "id": "not-a-uuid", - "hostname": "host-bad-id", - "address": "10.0.0.6:9000", - "cpus_total": 4, - "memory_mb_total": 8192, + // Disconnect, then reconnect under the same id with changed capacity. + stop() + waitForHostStatus(t, admin.AccessToken, id, "down") + + stop2 := startAgent(t, agent.LinkInfo{ + ID: id, Hostname: "host-owned-id", CPUsTotal: 16, MemoryMBTotal: 32768, }) - if resp.StatusCode != http.StatusBadRequest { - t.Fatalf("status = %d, body = %s", resp.StatusCode, body) - } -} + defer stop2() + waitForHostStatus(t, admin.AccessToken, id, "ready") -// TestHeartbeatUnknownHost verifies an unknown host id is rejected with 404 so -// the agent knows to re-register. -func TestHeartbeatUnknownHost(t *testing.T) { - resp, _ := doJSON(t, http.MethodPost, "/api/v1/agent/hosts/00000000-0000-0000-0000-000000000000/heartbeat", "", nil) - if resp.StatusCode != http.StatusNotFound { - t.Fatalf("status = %d, want 404", resp.StatusCode) + h := adminHostByID(t, admin.AccessToken, id) + if h == nil { + t.Fatalf("host vanished after reconnect") } -} - -// TestRegisterHostValidation covers request validation on the register endpoint. -func TestRegisterHostValidation(t *testing.T) { - cases := []struct { - name string - body map[string]any - }{ - {"missing hostname", map[string]any{"address": "10.0.0.1:9000", "cpus_total": 8, "memory_mb_total": 1024}}, - {"missing address", map[string]any{"hostname": "h", "cpus_total": 8, "memory_mb_total": 1024}}, - {"zero cpus", map[string]any{"hostname": "h", "address": "10.0.0.1:9000", "cpus_total": 0, "memory_mb_total": 1024}}, - {"zero memory", map[string]any{"hostname": "h", "address": "10.0.0.1:9000", "cpus_total": 8, "memory_mb_total": 0}}, + if h["cpus_total"].(float64) != 16 { + t.Errorf("cpus_total = %v, want 16 (updated in place on reconnect)", h["cpus_total"]) } - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - resp, body := doJSON(t, http.MethodPost, "/api/v1/agent/hosts/register", "", tc.body) - if resp.StatusCode != http.StatusBadRequest { - t.Fatalf("status = %d, body = %s", resp.StatusCode, body) - } - }) + if h["memory_mb_total"].(float64) != 32768 { + t.Errorf("memory_mb_total = %v, want 32768 (updated in place on reconnect)", h["memory_mb_total"]) } } diff --git a/test/e2e/main_test.go b/test/e2e/main_test.go index 8348df1..0de9001 100644 --- a/test/e2e/main_test.go +++ b/test/e2e/main_test.go @@ -1,7 +1,13 @@ //go:build e2e -// Package e2e contains end-to-end tests that exercise the real HTTP server -// against a real PostgreSQL instance started via testcontainers. +// Package e2e contains end-to-end tests that exercise the real HTTP server and +// the real agent control plane against a live PostgreSQL instance started via +// testcontainers. +// +// The control plane and a host agent are wired exactly as in production (P3): +// the agent dials the hub's gRPC listener and holds a persistent stream open, +// over which the hub pushes VM lifecycle commands. There is no inbound agent +// API — the open stream both delivers commands and proves the host's liveness. // // Run with: go test -tags e2e ./test/e2e/... (requires Docker). package e2e @@ -9,18 +15,20 @@ package e2e import ( "context" "fmt" + "net" "net/http/httptest" "os" "testing" "time" "github.com/aarani/craftling-go/internal/agent" + "github.com/aarani/craftling-go/internal/agentlink" + pb "github.com/aarani/craftling-go/internal/agentlink/pb" "github.com/aarani/craftling-go/internal/config" "github.com/aarani/craftling-go/internal/db" "github.com/aarani/craftling-go/internal/handler" "github.com/aarani/craftling-go/internal/model" "github.com/aarani/craftling-go/internal/provisioner" - "github.com/aarani/craftling-go/internal/reaper" "github.com/aarani/craftling-go/internal/reconciler" "github.com/aarani/craftling-go/internal/repository" "github.com/aarani/craftling-go/internal/scheduler" @@ -29,35 +37,34 @@ import ( "github.com/testcontainers/testcontainers-go/modules/postgres" "github.com/testcontainers/testcontainers-go/wait" "go.uber.org/zap" + "google.golang.org/grpc" ) // Shared test fixtures, set up in TestMain. var ( - baseURL string // address of the test HTTP server - pool *pgxpool.Pool // direct DB access for integration assertions + baseURL string // address of the test HTTP server + pool *pgxpool.Pool // direct DB access for integration assertions + grpcAddr string // address of the agent-link gRPC hub + hostRepo *repository.HostRepository // shared fleet inventory ) -// Host-reaper timing for tests: short so the stale->down transition is quick. -const ( - hostHeartbeatTTL = 200 * time.Millisecond - hostReapInterval = 25 * time.Millisecond -) - -// Capacity of the always-on placement host registered in TestMain. It is sized -// large in cpu so every test's server can be placed (the scheduler needs a ready -// host), but its memory total is deliberately below the maximum allowed server -// spec so a create request can still exceed it and exercise the oversize path. +// Capacity of the always-on placement host the agent registers in TestMain. It +// is sized large in cpu so every test's server can be placed (the scheduler +// needs a ready host), but its memory total is deliberately below the maximum +// allowed server spec so a create request can still exceed it and exercise the +// oversize path. const ( placementHostCPUs = 64 placementHostMemoryMB = 32768 ) -// placementHostID is the id of the kept-alive host the reconciler schedules onto -// in e2e; agentBaseURL is that host's in-process agent API. Set in TestMain. -var ( - placementHostID string - agentBaseURL string -) +// placementHostID is the stable id the always-on placement agent registers +// under; fakeRT is that agent's in-process runtime, queried directly by the +// agent-seam test to confirm a VM really landed on the agent. Both are set in +// TestMain. +const placementHostID = "22222222-2222-2222-2222-222222222222" + +var fakeRT *agent.FakeRuntime func TestMain(m *testing.M) { ctx := context.Background() @@ -99,66 +106,95 @@ func TestMain(m *testing.M) { AccessTTL: time.Hour, RefreshTTL: time.Hour, } - hostRepo := repository.NewHostRepository() + hostRepo = repository.NewHostRepository() + gameServerRepo := repository.NewGameServerRepository(pool) + srv := httptest.NewServer(handler.NewRouter(cfg, zap.NewNop(), pool, hostRepo)) baseURL = srv.URL - // Run an in-process host agent (FakeRuntime) so the control plane drives VMs - // across the real network seam (P3): the reconciler's RemoteProvisioner calls - // this agent's HTTP API to provision/start/stop/deprovision. - agentSrv := httptest.NewServer(agent.NewRouter(agent.NewFakeRuntime("127.0.0.1"), zap.NewNop())) - agentBaseURL = agentSrv.URL + // The hub is the control plane's end of the persistent agent connection: + // agents dial this gRPC listener and hold a stream open, and the hub registers + // hosts and pushes VM commands down the stream. This is the real P3 seam, + // replacing the old outbound HTTP client the control plane used to dial. + hub := agentlink.NewHub(hostRepo, gameServerRepo, zap.NewNop()) + grpcSrv := grpc.NewServer() + pb.RegisterAgentLinkServer(grpcSrv, hub) + grpcLis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + fmt.Fprintf(os.Stderr, "listen for agent gRPC: %v\n", err) + os.Exit(1) + } + grpcAddr = grpcLis.Addr().String() + go func() { _ = grpcSrv.Serve(grpcLis) }() - // Run the reconciler with a fast tick so lifecycle tests converge quickly. + // Run the reconciler with a fast tick so lifecycle tests converge quickly. The + // remote provisioner drives VMs by pushing commands down the host's stream via + // the hub (the control plane never touches a runtime directly). recCtx, recCancel := context.WithCancel(ctx) sched := scheduler.New(hostRepo) - prov := provisioner.NewRemote(hostRepo, agent.NewClient(nil)) - rec := reconciler.New(repository.NewGameServerRepository(pool), prov, sched, zap.NewNop()) + prov := provisioner.NewRemote(hub) + rec := reconciler.New(gameServerRepo, prov, sched, zap.NewNop()) go rec.Run(recCtx, 100*time.Millisecond) - // Run the host reaper with short timing so the stale->down test converges. - go reaper.Hosts(recCtx, zap.NewNop(), hostRepo, hostReapInterval, hostHeartbeatTTL) - - // Register an always-on host whose Address points at the in-process agent, so - // the scheduler can place servers and the RemoteProvisioner can reach them. - // Keep it alive against the reaper's short TTL by heartbeating (not - // re-registering, which would reset its allocatable capacity). - placed, err := hostRepo.Register(ctx, &model.Host{ + // Bring up the always-on placement host as an in-process agent dialing the hub + // over the real gRPC stream: a FakeRuntime that registers the host (so the + // scheduler can place onto it) and answers VM commands. Its FakeRuntime is the + // same instance the agent-seam test inspects directly. RunLink holds the stream + // open for the suite's lifetime and reconnects on its own; recCancel stops it. + fakeRT = agent.NewFakeRuntime("127.0.0.1") + go agent.RunLink(recCtx, grpcAddr, fakeRT, agent.LinkInfo{ + ID: placementHostID, Hostname: "placement-host", - Address: agentBaseURL, Zone: "zone-a", + AgentVersion: "test", CPUsTotal: placementHostCPUs, MemoryMBTotal: placementHostMemoryMB, - AgentVersion: "test", - }) - if err != nil { - fmt.Fprintf(os.Stderr, "register placement host: %v\n", err) + }, zap.NewNop()) + + // Wait for the placement host's stream to register before running tests, so the + // scheduler always has a ready host to place onto. + if err := waitHostReady(ctx, placementHostID, 10*time.Second); err != nil { + fmt.Fprintf(os.Stderr, "placement host did not register: %v\n", err) os.Exit(1) } - placementHostID = placed.ID - go keepHostAlive(recCtx, hostRepo, placementHostID) code := m.Run() recCancel() + grpcSrv.Stop() srv.Close() - agentSrv.Close() pool.Close() _ = pg.Terminate(ctx) os.Exit(code) } -// keepHostAlive heartbeats a host well within the reaper TTL so it stays ready -// for the whole suite, without re-registering (which would reset its capacity). -func keepHostAlive(ctx context.Context, repo *repository.HostRepository, id string) { - ticker := time.NewTicker(hostHeartbeatTTL / 4) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - _ = repo.Heartbeat(ctx, id) +// startAgent dials the hub as an in-process agent with the given identity and a +// throwaway FakeRuntime, returning a stop func that disconnects it (closing the +// stream, which the hub observes as the host going down). RunLink reconnects on +// its own until stop() cancels its context. +func startAgent(t *testing.T, info agent.LinkInfo) (stop func()) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + agent.RunLink(ctx, grpcAddr, agent.NewFakeRuntime("127.0.0.1"), info, zap.NewNop()) + close(done) + }() + return func() { + cancel() + <-done + } +} + +// waitHostReady polls the fleet inventory until host id is registered and ready, +// or the timeout elapses. +func waitHostReady(ctx context.Context, id string, timeout time.Duration) error { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if h, err := hostRepo.GetByID(ctx, id); err == nil && h.Status == model.HostReady { + return nil } + time.Sleep(20 * time.Millisecond) } + return fmt.Errorf("host %s not ready within %s", id, timeout) }