From ae80d7be5c09c167f1da77b27ecd23a46ae69f1f Mon Sep 17 00:00:00 2001 From: Afshin Arani Date: Sat, 20 Jun 2026 18:23:43 +0200 Subject: [PATCH 1/2] Revert "chore: gitignore local fc-assets (kernel, rootfs cache, VM work dirs)" This reverts commit fb8fbbac55653f03ffcf7278c79b3a45acf707f4. --- .gitignore | 4 ---- 1 file changed, 4 deletions(-) diff --git a/.gitignore b/.gitignore index c94f010..19a3bf5 100644 --- a/.gitignore +++ b/.gitignore @@ -3,10 +3,6 @@ *.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 From 3b23bce83c547db062e13e04126f6bc4d40a2b32 Mon Sep 17 00:00:00 2001 From: Afshin Arani Date: Sat, 20 Jun 2026 18:23:43 +0200 Subject: [PATCH 2/2] =?UTF-8?q?Revert=20"feat:=20invert=20control=20plane?= =?UTF-8?q?=20=E2=86=94=20agent=20to=20an=20agent-initiated=20gRPC=20strea?= =?UTF-8?q?m"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This reverts commit b351afe52998c9462f3792023690af2c7652d4af. --- Makefile | 20 +- cmd/agent/main.go | 118 ++++- cmd/server/main.go | 39 +- docker-compose.yml | 17 +- go.mod | 4 +- go.sum | 8 - internal/agent/agent_test.go | 104 +++-- internal/agent/client.go | 109 +++++ internal/agent/controlplane.go | 101 ++++ internal/agent/link.go | 216 --------- internal/agent/server.go | 125 +++++ internal/agentlink/hub.go | 278 ----------- internal/agentlink/hub_test.go | 152 ------ internal/agentlink/pb/agentlink.pb.go | 519 --------------------- internal/agentlink/pb/agentlink_grpc.pb.go | 131 ------ internal/config/config.go | 24 +- internal/handler/agent.go | 92 ++++ internal/handler/router.go | 12 +- internal/middleware/agent.go | 14 + internal/provisioner/remote.go | 70 +-- internal/provisioner/remote_test.go | 100 ++-- internal/repository/host.go | 18 - internal/repository/host_test.go | 25 - proto/agentlink/agentlink.proto | 66 --- 24 files changed, 718 insertions(+), 1644 deletions(-) create mode 100644 internal/agent/client.go create mode 100644 internal/agent/controlplane.go delete mode 100644 internal/agent/link.go create mode 100644 internal/agent/server.go delete mode 100644 internal/agentlink/hub.go delete mode 100644 internal/agentlink/hub_test.go delete mode 100644 internal/agentlink/pb/agentlink.pb.go delete mode 100644 internal/agentlink/pb/agentlink_grpc.pb.go create mode 100644 internal/handler/agent.go create mode 100644 internal/middleware/agent.go delete mode 100644 proto/agentlink/agentlink.proto diff --git a/Makefile b/Makefile index 3d397c2..3536a23 100644 --- a/Makefile +++ b/Makefile @@ -1,11 +1,10 @@ -.PHONY: run run-agent build build-agent test test-e2e test-kvm test-bpf tidy fmt bpf-generate proto-generate +.PHONY: run run-agent build build-agent test test-e2e test-kvm test-bpf tidy fmt bpf-generate run: go run ./cmd/server -# 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 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-agent: MODE=agent go run ./cmd/agent @@ -50,19 +49,6 @@ 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 785e114..91b6426 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -1,8 +1,6 @@ -// 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. +// 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. // // It ships with the in-memory FakeRuntime; a real Firecracker driver (P4) slots // in behind the same Runtime interface without changing this wiring. @@ -10,8 +8,10 @@ package main import ( "context" + "errors" "fmt" "log" + "net/http" "os" "os/signal" "path/filepath" @@ -27,6 +27,15 @@ 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() @@ -36,28 +45,57 @@ func main() { } defer func() { _ = zlog.Sync() }() - // The runtime that actually runs VMs, driven by commands off the link. + 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. 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() - // 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{ + 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{ 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, - }, zlog) + AgentVersion: cfg.Agent.Version, + }) + <-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") } @@ -123,3 +161,57 @@ 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 9f3e168..b6b31a0 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -4,14 +4,12 @@ import ( "context" "errors" "log" - "net" "net/http" "os/signal" "syscall" "time" - "github.com/aarani/craftling-go/internal/agentlink" - pb "github.com/aarani/craftling-go/internal/agentlink/pb" + "github.com/aarani/craftling-go/internal/agent" "github.com/aarani/craftling-go/internal/config" "github.com/aarani/craftling-go/internal/db" "github.com/aarani/craftling-go/internal/handler" @@ -24,7 +22,6 @@ import ( "github.com/aarani/craftling-go/internal/seed" "github.com/aarani/craftling-go/internal/worldstore" "go.uber.org/zap" - "google.golang.org/grpc" ) const ( @@ -37,6 +34,8 @@ 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 @@ -74,10 +73,8 @@ func main() { dbCancel() // The fleet inventory lives in process memory (P1). It is shared between the - // agent link hub (which registers/heartbeats hosts as their streams come and - // go) and the host reaper. + // HTTP handlers (register/heartbeat) and the host reaper. hostRepo := repository.NewHostRepository() - gameServerRepo := repository.NewGameServerRepository(pool) router := handler.NewRouter(cfg, zlog, pool, hostRepo) @@ -89,18 +86,6 @@ 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) @@ -118,8 +103,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(hub) - rec := reconciler.New(gameServerRepo, prov, sched, zlog) + prov := provisioner.NewRemote(hostRepo, agent.NewClient(&http.Client{Timeout: agentCallTimeout})) + rec := reconciler.New(repository.NewGameServerRepository(pool), prov, sched, zlog) go rec.Run(ctx, reconcileInterval) // If a durable world store is configured, periodically GC snapshots that no @@ -131,17 +116,9 @@ 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, gameServerRepo, worldGCInterval) + go reaper.Worlds(ctx, zlog, worldStore, repository.NewGameServerRepository(pool), 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)) @@ -154,8 +131,6 @@ 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 953d85c..1bdc8c0 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -38,10 +38,6 @@ 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 @@ -70,11 +66,11 @@ services: dockerfile: Dockerfile.agent environment: MODE: agent + PORT: "9000" APP_ENV: production - # 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 + CONTROL_PLANE_URL: http://server:8080 + # Address the control plane calls back (reachable on the compose network). + ADVERTISE_ADDR: agent:9000 # Player-facing connect host VMs report (override for real connectivity). ADVERTISE_HOST: 127.0.0.1 AGENT_RUNTIME: firecracker @@ -94,8 +90,9 @@ services: privileged: true volumes: - ./fc-assets:/var/lib/craftling - # No published ports: the agent serves no inbound API. It reaches the control - # plane's gRPC link at server:8090 over the compose network. + ports: + # Exposed for debugging the agent API directly. + - "9000:9000" depends_on: - server restart: unless-stopped diff --git a/go.mod b/go.mod index ed933bf..3a20cef 100644 --- a/go.mod +++ b/go.mod @@ -22,8 +22,6 @@ 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 ( @@ -136,7 +134,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/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529 // indirect + google.golang.org/protobuf v1.36.11 // 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 acad684..ba115e5 100644 --- a/go.sum +++ b/go.sum @@ -133,8 +133,6 @@ 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= @@ -331,12 +329,6 @@ 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 e8c2991..b64848c 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" - pb "github.com/aarani/craftling-go/internal/agentlink/pb" + "go.uber.org/zap" ) // TestFakeRuntimeLifecycle exercises the in-memory runtime directly through its @@ -63,53 +63,74 @@ func TestFakeRuntimeIdempotency(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) { +// TestAgentServerClientRoundTrip drives the runtime through the HTTP API the +// control plane uses, verifying the wire contract end-to-end. +func TestAgentServerClientRoundTrip(t *testing.T) { ctx := context.Background() - rt := NewFakeRuntime("10.0.0.9") + srv := httptest.NewServer(NewRouter(NewFakeRuntime("10.0.0.9"), zap.NewNop())) + defer srv.Close() - // 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) - } - var vm VM - if err := json.Unmarshal(payload, &vm); err != nil { - t.Fatalf("decode provision result: %v", err) + 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) } - if vm.ID == "" || vm.State != StateRunning { + if vm == nil || vm.ID == "" || vm.State != StateRunning { t.Fatalf("provisioned vm = %+v, want running with id", vm) } + 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) + } - 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 got := statusOf(t, client, base, vm.ID); got != StateRunning { + t.Errorf("after provision state = %q, want running", got) } - statusPayload, errStr := execOp(ctx, rt, &pb.Command{Op: OpStatus, Payload: ref}) - if errStr != "" { - t.Fatalf("status op error = %q, want none", errStr) + 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) + } + if err := client.Deprovision(ctx, base, vm.ID); err != nil { + t.Fatalf("deprovision: %v", err) } - var stopped VM - _ = json.Unmarshal(statusPayload, &stopped) - if stopped.State != StateStopped { - t.Errorf("status after stop = %q, want stopped", stopped.State) + if got := statusOf(t, client, base, vm.ID); got != StateMissing { + t.Errorf("after deprovision state = %q, want missing", got) } - // 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") + // 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") } +} + +// 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 - // 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") + 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") } } @@ -123,3 +144,12 @@ 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 new file mode 100644 index 0000000..a4325c1 --- /dev/null +++ b/internal/agent/client.go @@ -0,0 +1,109 @@ +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 new file mode 100644 index 0000000..a58cae6 --- /dev/null +++ b/internal/agent/controlplane.go @@ -0,0 +1,101 @@ +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/link.go b/internal/agent/link.go deleted file mode 100644 index fe00df3..0000000 --- a/internal/agent/link.go +++ /dev/null @@ -1,216 +0,0 @@ -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 new file mode 100644 index 0000000..c50235d --- /dev/null +++ b/internal/agent/server.go @@ -0,0 +1,125 @@ +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 deleted file mode 100644 index 3b72b84..0000000 --- a/internal/agentlink/hub.go +++ /dev/null @@ -1,278 +0,0 @@ -// 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 deleted file mode 100644 index 510fe89..0000000 --- a/internal/agentlink/hub_test.go +++ /dev/null @@ -1,152 +0,0 @@ -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 deleted file mode 100644 index d8ff4df..0000000 --- a/internal/agentlink/pb/agentlink.pb.go +++ /dev/null @@ -1,519 +0,0 @@ -// 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 deleted file mode 100644 index 5f9d072..0000000 --- a/internal/agentlink/pb/agentlink_grpc.pb.go +++ /dev/null @@ -1,131 +0,0 @@ -// 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 666bff0..2240d21 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -25,11 +25,6 @@ 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 @@ -38,9 +33,8 @@ type Config struct { AdminEmail string AdminPassword string - // 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 configuration (ModeAgent only). The host worker registers with the + // control plane and exposes its VM API for the control plane to call back. Agent AgentConfig } @@ -53,9 +47,8 @@ const ( // AgentConfig holds the host-worker settings used when Mode == ModeAgent. type AgentConfig struct { - // ControlPlaneGRPCAddr is the control plane's gRPC AgentLink address - // (host:port) the agent dials and keeps a stream open to. - ControlPlaneGRPCAddr string + // ControlPlaneURL is where the agent registers and heartbeats. + ControlPlaneURL string // Runtime selects the VM backend: "fake" (default) or "firecracker". Runtime string // Firecracker holds the real-microVM driver settings (Runtime == "firecracker"). @@ -64,6 +57,9 @@ 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. @@ -152,7 +148,6 @@ 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"), @@ -160,8 +155,8 @@ func Load() *Config { AdminPassword: getEnv("ADMIN_PASSWORD", ""), Agent: AgentConfig{ - ControlPlaneGRPCAddr: getEnv("CONTROL_PLANE_GRPC_ADDR", "localhost:8090"), - Runtime: getEnv("AGENT_RUNTIME", RuntimeFake), + ControlPlaneURL: getEnv("CONTROL_PLANE_URL", "http://localhost:8080"), + Runtime: getEnv("AGENT_RUNTIME", RuntimeFake), Firecracker: FirecrackerConfig{ BinaryPath: getEnv("FC_BINARY", ""), KernelPath: getEnv("FC_KERNEL", ""), @@ -191,6 +186,7 @@ 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 new file mode 100644 index 0000000..5975d05 --- /dev/null +++ b/internal/handler/agent.go @@ -0,0 +1,92 @@ +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 fa1662d..368c667 100644 --- a/internal/handler/router.go +++ b/internal/handler/router.go @@ -36,6 +36,7 @@ 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() @@ -89,9 +90,14 @@ func NewRouter(cfg *config.Config, log *zap.Logger, pool *pgxpool.Pool, hostRepo admin.GET("/hosts", adminHandler.ListHosts) } - // 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. + // 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) + } } return r diff --git a/internal/middleware/agent.go b/internal/middleware/agent.go new file mode 100644 index 0000000..dfd2388 --- /dev/null +++ b/internal/middleware/agent.go @@ -0,0 +1,14 @@ +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 82f35fb..66166f3 100644 --- a/internal/provisioner/remote.go +++ b/internal/provisioner/remote.go @@ -3,6 +3,7 @@ package provisioner import ( "context" "errors" + "fmt" "github.com/aarani/craftling-go/internal/agent" "github.com/aarani/craftling-go/internal/model" @@ -15,36 +16,29 @@ import ( // indicates a logic error rather than a transient condition. var ErrUnplaced = errors.New("server has no host assigned") -// 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) +// 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) } -// 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. +// 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. type RemoteProvisioner struct { - cmd Commander + hosts HostResolver + client *agent.Client } -// NewRemote constructs a RemoteProvisioner over a command channel (the hub). -func NewRemote(cmd Commander) *RemoteProvisioner { - return &RemoteProvisioner{cmd: cmd} +// 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} } // 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) { - hostID, err := assignedHost(s) + base, err := p.baseURL(ctx, s) if err != nil { return nil, err } @@ -65,7 +59,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.cmd.Provision(ctx, hostID, spec) + vm, err := p.client.Provision(ctx, base, spec) if err != nil { return nil, err } @@ -78,11 +72,11 @@ func (p *RemoteProvisioner) Start(ctx context.Context, s *model.GameServer) (*In if s.VMID == nil || *s.VMID == "" { return p.Provision(ctx, s) } - hostID, err := assignedHost(s) + base, err := p.baseURL(ctx, s) if err != nil { return nil, err } - vm, err := p.cmd.Start(ctx, hostID, *s.VMID) + vm, err := p.client.Start(ctx, base, *s.VMID) if err != nil { return nil, err } @@ -94,11 +88,11 @@ func (p *RemoteProvisioner) Stop(ctx context.Context, s *model.GameServer) error if s.VMID == nil || *s.VMID == "" { return nil } - hostID, err := assignedHost(s) + base, err := p.baseURL(ctx, s) if err != nil { return err } - return p.cmd.Stop(ctx, hostID, *s.VMID) + return p.client.Stop(ctx, base, *s.VMID) } // Deprovision tears down the VM on its host (idempotent). A server that was @@ -107,7 +101,11 @@ func (p *RemoteProvisioner) Deprovision(ctx context.Context, s *model.GameServer if s.HostID == nil || *s.HostID == "" || s.VMID == nil || *s.VMID == "" { return nil } - return p.cmd.Deprovision(ctx, *s.HostID, *s.VMID) + base, err := p.baseURL(ctx, s) + if err != nil { + return err + } + return p.client.Deprovision(ctx, base, *s.VMID) } // Status reports the VM's observed state as seen by its host's agent. @@ -115,11 +113,11 @@ func (p *RemoteProvisioner) Status(ctx context.Context, s *model.GameServer) (St if s.VMID == nil || *s.VMID == "" { return StateMissing, nil } - hostID, err := assignedHost(s) + base, err := p.baseURL(ctx, s) if err != nil { return "", err } - vm, err := p.cmd.Status(ctx, hostID, *s.VMID) + vm, err := p.client.Status(ctx, base, *s.VMID) if err != nil { return "", err } @@ -132,19 +130,23 @@ func (p *RemoteProvisioner) Snapshot(ctx context.Context, s *model.GameServer) e if s.VMID == nil || *s.VMID == "" { return nil } - hostID, err := assignedHost(s) + base, err := p.baseURL(ctx, s) if err != nil { return err } - return p.cmd.Snapshot(ctx, hostID, *s.VMID) + return p.client.Snapshot(ctx, base, *s.VMID) } -// assignedHost returns the server's assigned host id, or ErrUnplaced. -func assignedHost(s *model.GameServer) (string, error) { +// baseURL resolves the agent base URL for the server's assigned host. +func (p *RemoteProvisioner) baseURL(ctx context.Context, s *model.GameServer) (string, error) { if s.HostID == nil || *s.HostID == "" { return "", ErrUnplaced } - return *s.HostID, nil + 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 } // instanceOf maps an agent VM to a provisioner Instance. diff --git a/internal/provisioner/remote_test.go b/internal/provisioner/remote_test.go index 2d8c97b..d33fa4d 100644 --- a/internal/provisioner/remote_test.go +++ b/internal/provisioner/remote_test.go @@ -3,79 +3,32 @@ 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" ) -// 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 } +// stubResolver resolves every host id to a fixed agent address. +type stubResolver struct{ addr string } -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 (s stubResolver) GetByID(_ context.Context, id string) (*model.Host, error) { + return &model.Host{ID: id, Address: s.addr}, nil } func ptr(s string) *string { return &s } // TestRemoteProvisionerLifecycle drives a game server through provision → stop → -// start → deprovision against an in-process runtime behind the command channel, -// asserting the observed state reported back at each step. +// start → deprovision against a real in-process agent, asserting the observed +// state reported back across the seam at each step. func TestRemoteProvisionerLifecycle(t *testing.T) { ctx := context.Background() - p := NewRemote(fakeCommander{rt: agent.NewFakeRuntime("10.0.0.20")}) + 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)) s := &model.GameServer{ ID: "srv-1", HostID: ptr("host-1"), @@ -114,15 +67,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 that never reaches the command channel. +// harmless no-op. func TestRemoteProvisionerUnplaced(t *testing.T) { ctx := context.Background() - p := NewRemote(errCommander{t: t}) + p := NewRemote(stubResolver{addr: "http://127.0.0.1:1"}, agent.NewClient(nil)) 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 send a command. + // No host and no VM: nothing to tear down, and we must not dial anyone. if err := p.Deprovision(ctx, &model.GameServer{ID: "x"}); err != nil { t.Errorf("deprovision unplaced = %v, want nil", err) } @@ -135,7 +88,10 @@ func TestRemoteProvisionerUnplaced(t *testing.T) { // back to provisioning a fresh one. func TestRemoteProvisionerStartProvisions(t *testing.T) { ctx := context.Background() - p := NewRemote(fakeCommander{rt: agent.NewFakeRuntime("10.0.0.21")}) + 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)) s := &model.GameServer{ID: "srv-2", HostID: ptr("host-2"), Version: "1.20.4", CPUs: 1, MemoryMB: 1024} inst, err := p.Start(ctx, s) @@ -148,10 +104,14 @@ 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. +// forwarded to its host's agent, and that a server with no VM is a no-op (no +// dial). func TestRemoteProvisionerSnapshot(t *testing.T) { ctx := context.Background() - p := NewRemote(fakeCommander{rt: agent.NewFakeRuntime("10.0.0.22")}) + 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)) s := &model.GameServer{ID: "srv-3", HostID: ptr("host-3"), Version: "1.20.4", CPUs: 1, MemoryMB: 1024} inst, err := p.Provision(ctx, s) @@ -163,8 +123,9 @@ func TestRemoteProvisionerSnapshot(t *testing.T) { t.Fatalf("snapshot: %v", err) } - // No VM: nothing to snapshot, and we must not send a command. - dead := NewRemote(errCommander{t: t}) + // 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)) if err := dead.Snapshot(ctx, &model.GameServer{ID: "x", HostID: ptr("h")}); err != nil { t.Errorf("snapshot with no vm = %v, want nil", err) } @@ -188,7 +149,10 @@ 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")} - p := NewRemote(fakeCommander{rt: rt}) + srv := httptest.NewServer(agent.NewRouter(rt, zap.NewNop())) + defer srv.Close() + + p := NewRemote(stubResolver{addr: srv.URL}, agent.NewClient(nil)) imageRef := "itzg/minecraft-server:java21" tmpl := &model.GameServer{ diff --git a/internal/repository/host.go b/internal/repository/host.go index 11d830c..7ff3842 100644 --- a/internal/repository/host.go +++ b/internal/repository/host.go @@ -203,24 +203,6 @@ 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 74eb3d3..2c15140 100644 --- a/internal/repository/host_test.go +++ b/internal/repository/host_test.go @@ -24,31 +24,6 @@ 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 deleted file mode 100644 index 79638d1..0000000 --- a/proto/agentlink/agentlink.proto +++ /dev/null @@ -1,66 +0,0 @@ -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 {}