diff --git a/api/agent.yaml b/api/agent.yaml index da20853..00d4fd8 100644 --- a/api/agent.yaml +++ b/api/agent.yaml @@ -99,6 +99,95 @@ paths: "500": $ref: "#/components/responses/InternalError" + # ── VM ───────────────────────────────────────────────────────────────────── + + /vms: + get: + summary: List all VMs + operationId: listVMs + responses: + "200": + description: List of VMs + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/VM" + "500": + $ref: "#/components/responses/InternalError" + + post: + summary: Start a VM + operationId: startVM + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/VMCreateRequest" + responses: + "202": + description: VM start accepted + content: + application/json: + schema: + $ref: "#/components/schemas/VM" + "400": + description: Missing required field or invalid request body + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + "409": + description: VM already exists + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + "422": + description: Subnet not found or not in created state + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + "500": + $ref: "#/components/responses/InternalError" + + /vms/{name}: + parameters: + - $ref: "#/components/parameters/ResourceName" + + get: + summary: Get VM status and info + operationId: getVM + responses: + "200": + description: VM found + content: + application/json: + schema: + $ref: "#/components/schemas/VM" + "404": + $ref: "#/components/responses/NotFound" + "500": + $ref: "#/components/responses/InternalError" + + delete: + summary: Stop a VM + operationId: stopVM + responses: + "202": + description: VM stop accepted + content: + application/json: + schema: + $ref: "#/components/schemas/VM" + "404": + $ref: "#/components/responses/NotFound" + "500": + $ref: "#/components/responses/InternalError" + # ── Subnet ───────────────────────────────────────────────────────────────── /subnets: @@ -281,6 +370,97 @@ components: type: string example: "10.10.10.0/24" + VMCreateRequest: + type: object + required: [name, metadata_port, interfaces, storage] + properties: + name: + type: string + example: vm-00001 + metadata_port: + type: string + example: "80" + memory: + type: integer + description: Memory in MB (default 512) + example: 1024 + cpus: + type: integer + description: Number of vCPUs (default 1) + example: 2 + password: + type: string + sshkey: + type: string + example: "ssh-ed25519 AAAA..." + interfaces: + type: array + minItems: 1 + items: + $ref: "#/components/schemas/VMInterface" + storage: + type: array + minItems: 1 + items: + $ref: "#/components/schemas/VMStorage" + + VMInterface: + type: object + required: [subnet, ip, primary] + properties: + subnet: + type: string + example: sn-00001 + ip: + type: string + format: ipv4 + example: "10.0.0.5" + primary: + type: boolean + example: true + + VMStorage: + type: object + required: [path, dev] + properties: + path: + type: string + description: Path to the disk image on the host + example: /var/lib/two/volumes/abc.qcow2 + dev: + type: string + description: Device name inside the VM + pattern: '^[sv]d[a-z]$' + example: vda + + VM: + type: object + properties: + name: + type: string + example: vm-00001 + state: + type: string + enum: [starting, started, stopping, stopped] + example: started + metadata_port: + type: string + example: "80" + memory: + type: integer + example: 1024 + cpus: + type: integer + example: 2 + interfaces: + type: array + items: + $ref: "#/components/schemas/VMInterface" + storage: + type: array + items: + $ref: "#/components/schemas/VMStorage" + Error: type: object properties: diff --git a/internal/api/agent/models.go b/internal/api/agent/models.go index e826082..f13407b 100644 --- a/internal/api/agent/models.go +++ b/internal/api/agent/models.go @@ -28,6 +28,38 @@ type Subnet struct { CIDR string `json:"cidr"` } +type VMInterface struct { + Subnet string `json:"subnet"` + IP string `json:"ip"` + Primary bool `json:"primary"` +} + +type VMStorage struct { + Path string `json:"path"` + Dev string `json:"dev"` +} + +type VMCreateRequest struct { + Name string `json:"name"` + MetadataPort string `json:"metadata_port"` + Memory int `json:"memory"` + CPUs int `json:"cpus"` + Password string `json:"password"` + SSHKey string `json:"sshkey"` + Interfaces []VMInterface `json:"interfaces"` + Storage []VMStorage `json:"storage"` +} + +type VM struct { + Name string `json:"name"` + State string `json:"state"` + MetadataPort string `json:"metadata_port"` + Memory int `json:"memory"` + CPUs int `json:"cpus"` + Interfaces []VMInterface `json:"interfaces"` + Storage []VMStorage `json:"storage"` +} + type ErrorResponse struct { Error string `json:"error"` } diff --git a/internal/api/agent/server.go b/internal/api/agent/server.go index 8e3c4e3..4a0fba4 100644 --- a/internal/api/agent/server.go +++ b/internal/api/agent/server.go @@ -27,6 +27,8 @@ func (s *Server) Start(address string) { mux.HandleFunc("/vpcs/", s.VpcByNameHandler) mux.HandleFunc("/subnets", s.SubnetsHandler) mux.HandleFunc("/subnets/", s.SubnetByNameHandler) + mux.HandleFunc("/vms", s.VmsHandler) + mux.HandleFunc("/vms/", s.VmByNameHandler) s.logger.Info("API server listening", "address", address) if err := http.ListenAndServe(address, s.logMiddleware(mux)); err != nil { s.logger.Error("API server stopped", "error", err) diff --git a/internal/api/agent/vm.go b/internal/api/agent/vm.go new file mode 100644 index 0000000..a150d16 --- /dev/null +++ b/internal/api/agent/vm.go @@ -0,0 +1,94 @@ +package agentapi + +import ( + "encoding/json" + "net/http" + "strconv" + "strings" + + dispatcher "git.g3e.fr/syonad/two/internal/dispatcher/agent" + "git.g3e.fr/syonad/two/pkg/db/kv" +) + +func (s *Server) VmByNameHandler(w http.ResponseWriter, r *http.Request) { + name := strings.TrimPrefix(r.URL.Path, "/vms/") + if name == "" { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + json.NewEncoder(w).Encode(ErrorResponse{Error: "resource not found"}) + return + } + w.Header().Set("Content-Type", "application/json") + switch r.Method { + case http.MethodGet: + s.getVM(w, r, name) + case http.MethodDelete: + s.stopVM(w, r, name) + default: + w.WriteHeader(http.StatusMethodNotAllowed) + json.NewEncoder(w).Encode(ErrorResponse{Error: "method not allowed"}) + } +} + +func (s *Server) getVM(w http.ResponseWriter, _ *http.Request, name string) { + entries, err := kv.ListByPrefix(s.db, "vm/"+name+"/") + if err != nil || len(entries) == 0 { + w.WriteHeader(http.StatusNotFound) + json.NewEncoder(w).Encode(ErrorResponse{Error: "vm not found"}) + return + } + vm, err := vmFromDB(name, entries) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + json.NewEncoder(w).Encode(ErrorResponse{Error: "failed to read vm"}) + return + } + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(vm) +} + +func (s *Server) stopVM(w http.ResponseWriter, _ *http.Request, name string) { + cmd := dispatcher.StopVMCommand{Name: name} + if err := s.dispatcher.Prepare(cmd); err != nil { + if _, dbErr := kv.GetFromDB(s.db, "vm/"+name+"/state"); dbErr != nil { + w.WriteHeader(http.StatusNotFound) + } else { + w.WriteHeader(http.StatusConflict) + } + json.NewEncoder(w).Encode(ErrorResponse{Error: err.Error()}) + return + } + s.dispatcher.Dispatch(cmd) + + entries, _ := kv.ListByPrefix(s.db, "vm/"+name+"/") + vm, err := vmFromDB(name, entries) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + json.NewEncoder(w).Encode(ErrorResponse{Error: "failed to read vm state"}) + return + } + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(vm) +} + +func vmFromDB(name string, entries map[string]string) (VM, error) { + prefix := "vm/" + name + "/" + vm := VM{Name: name} + + vm.State = entries[prefix+"state"] + vm.MetadataPort = entries[prefix+"metadata_port"] + vm.Memory, _ = strconv.Atoi(entries[prefix+"memory"]) + vm.CPUs, _ = strconv.Atoi(entries[prefix+"cpus"]) + + subnet := entries[prefix+"subnet"] + ip := entries[prefix+"ip"] + if subnet != "" || ip != "" { + vm.Interfaces = []VMInterface{{Subnet: subnet, IP: ip, Primary: true}} + } + + if path := entries[prefix+"volume_path"]; path != "" { + vm.Storage = []VMStorage{{Path: path}} + } + + return vm, nil +} diff --git a/internal/api/agent/vms.go b/internal/api/agent/vms.go new file mode 100644 index 0000000..841c429 --- /dev/null +++ b/internal/api/agent/vms.go @@ -0,0 +1,112 @@ +package agentapi + +import ( + "encoding/json" + "net/http" + "strings" + + dispatcher "git.g3e.fr/syonad/two/internal/dispatcher/agent" + "git.g3e.fr/syonad/two/pkg/db/kv" +) + +func (s *Server) VmsHandler(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch r.Method { + case http.MethodGet: + s.listVMs(w, r) + case http.MethodPost: + s.startVM(w, r) + default: + w.WriteHeader(http.StatusMethodNotAllowed) + json.NewEncoder(w).Encode(ErrorResponse{Error: "method not allowed"}) + } +} + +func (s *Server) listVMs(w http.ResponseWriter, _ *http.Request) { + entries, err := kv.ListByPrefix(s.db, "vm/") + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + json.NewEncoder(w).Encode(ErrorResponse{Error: "failed to list vms"}) + return + } + + names := map[string]struct{}{} + for key := range entries { + parts := strings.Split(key, "/") + if len(parts) >= 2 { + names[parts[1]] = struct{}{} + } + } + + result := make([]VM, 0, len(names)) + for name := range names { + vm, err := vmFromDB(name, entries) + if err != nil { + continue + } + result = append(result, vm) + } + + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(result) +} + +func (s *Server) startVM(w http.ResponseWriter, r *http.Request) { + var req VMCreateRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(ErrorResponse{Error: "invalid request body"}) + return + } + if req.Name == "" || req.MetadataPort == "" || len(req.Interfaces) == 0 || len(req.Storage) == 0 { + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(ErrorResponse{Error: "name, metadata_port, interfaces and storage are required"}) + return + } + + var primary *VMInterface + for i := range req.Interfaces { + if req.Interfaces[i].Primary { + primary = &req.Interfaces[i] + break + } + } + if primary == nil { + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(ErrorResponse{Error: "one interface must be primary"}) + return + } + + cmd := dispatcher.StartVMCommand{ + Name: req.Name, + Subnet: primary.Subnet, + IP: primary.IP, + MetadataPort: req.MetadataPort, + VolumePath: req.Storage[0].Path, + Memory: req.Memory, + CPUs: req.CPUs, + Password: req.Password, + SSHKey: req.SSHKey, + } + + if err := s.dispatcher.Prepare(cmd); err != nil { + if _, dbErr := kv.GetFromDB(s.db, "vm/"+req.Name+"/state"); dbErr == nil { + w.WriteHeader(http.StatusConflict) + } else { + w.WriteHeader(http.StatusUnprocessableEntity) + } + json.NewEncoder(w).Encode(ErrorResponse{Error: err.Error()}) + return + } + s.dispatcher.Dispatch(cmd) + + entries, _ := kv.ListByPrefix(s.db, "vm/"+req.Name+"/") + vm, err := vmFromDB(req.Name, entries) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + json.NewEncoder(w).Encode(ErrorResponse{Error: "failed to read vm state"}) + return + } + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(vm) +} diff --git a/internal/dispatcher/agent/vm_commands.go b/internal/dispatcher/agent/vm_commands.go new file mode 100644 index 0000000..c0245ca --- /dev/null +++ b/internal/dispatcher/agent/vm_commands.go @@ -0,0 +1,95 @@ +package dispatcher + +import ( + "fmt" + "strconv" + "time" + + configuration "git.g3e.fr/syonad/two/internal/config/agent" + "git.g3e.fr/syonad/two/internal/vm" + "git.g3e.fr/syonad/two/pkg/db/kv" + "github.com/dgraph-io/badger/v4" +) + +type StartVMCommand struct { + Name string + Subnet string + IP string + MetadataPort string + VolumePath string + Memory int + CPUs int + Password string + SSHKey string +} + +func (c StartVMCommand) Prepare(db *badger.DB, _ *configuration.Config) error { + if _, err := kv.GetFromDB(db, "vm/"+c.Name+"/state"); err == nil { + return fmt.Errorf("vm %q already exists", c.Name) + } + subnetState, err := kv.GetFromDB(db, "subnet/"+c.Subnet+"/state") + if err != nil { + return fmt.Errorf("subnet %q not found", c.Subnet) + } + if subnetState == "deleting" || subnetState == "deleted" { + return fmt.Errorf("subnet %q is %s", c.Subnet, subnetState) + } + kv.AddInDB(db, "vm/"+c.Name+"/state", "starting") + kv.AddInDB(db, "vm/"+c.Name+"/subnet", c.Subnet) + kv.AddInDB(db, "vm/"+c.Name+"/ip", c.IP) + kv.AddInDB(db, "vm/"+c.Name+"/metadata_port", c.MetadataPort) + kv.AddInDB(db, "vm/"+c.Name+"/volume_path", c.VolumePath) + kv.AddInDB(db, "vm/"+c.Name+"/memory", strconv.Itoa(c.Memory)) + kv.AddInDB(db, "vm/"+c.Name+"/cpus", strconv.Itoa(c.CPUs)) + if c.Password != "" { + kv.AddInDB(db, "vm/"+c.Name+"/password", c.Password) + } + if c.SSHKey != "" { + kv.AddInDB(db, "vm/"+c.Name+"/sshkey", c.SSHKey) + } + return nil +} + +func (c StartVMCommand) Execute(db *badger.DB, cfg *configuration.Config) error { + timeout := time.After(time.Duration(cfg.Dispatcher.TimeoutSeconds) * time.Second) + for { + state, err := kv.GetFromDB(db, "subnet/"+c.Subnet+"/state") + if err != nil { + return fmt.Errorf("subnet %q not found while waiting", c.Subnet) + } + if state == "created" { + break + } + select { + case <-timeout: + return fmt.Errorf("timed out waiting for subnet %q to be created", c.Subnet) + case <-time.After(time.Duration(cfg.Dispatcher.PollSeconds) * time.Second): + } + } + return vm.StartVM(db, c.Name) +} + +type StopVMCommand struct { + Name string +} + +func (c StopVMCommand) Prepare(db *badger.DB, _ *configuration.Config) error { + if _, err := kv.GetFromDB(db, "vm/"+c.Name+"/state"); err != nil { + return fmt.Errorf("vm %q not found", c.Name) + } + return kv.AddInDB(db, "vm/"+c.Name+"/state", "stopping") +} + +func (c StopVMCommand) Execute(db *badger.DB, cfg *configuration.Config) error { + if err := vm.StopVM(db, c.Name, cfg); err != nil { + return err + } + state, err := kv.GetFromDB(db, "vm/"+c.Name+"/state") + if err != nil { + return err + } + if state == "stopped" { + kv.DeleteInDB(db, "vm/"+c.Name) + } + return nil +} diff --git a/internal/qemu/start_linux.go b/internal/qemu/start_linux.go index cc9c283..6119278 100644 --- a/internal/qemu/start_linux.go +++ b/internal/qemu/start_linux.go @@ -5,13 +5,10 @@ package qemu import ( "fmt" "os/exec" - - "git.g3e.fr/syonad/two/internal/netns" ) type Config struct { Name string - VpcName string TapID int Mac string VolumePath string @@ -30,24 +27,22 @@ func Start(cfg Config) error { cpus = 1 } - return netns.Call(cfg.VpcName, func() error { - cmd := exec.Command("qemu-system-x86_64", - "-enable-kvm", - "-cpu", "host", - "-m", fmt.Sprintf("%d", memory), - "-smp", fmt.Sprintf("%d", cpus), - "-serial", fmt.Sprintf("unix:/tmp/%s.sock,server,nowait", cfg.Name), - "-monitor", fmt.Sprintf("unix:/tmp/%s.mon-sock,server,nowait", cfg.Name), - "-qmp", fmt.Sprintf("unix:/tmp/%s.qmp-sock,server,nowait", cfg.Name), - "-display", "none", - "-drive", fmt.Sprintf("file=%s,if=virtio", cfg.VolumePath), - "-netdev", fmt.Sprintf("tap,id=net0,ifname=tap%d,script=no,downscript=no", cfg.TapID), - "-device", fmt.Sprintf("virtio-net-pci,netdev=net0,mac=%s", cfg.Mac), - "-daemonize", - ) - if err := cmd.Run(); err != nil { - return fmt.Errorf("qemu-system-x86_64: %w", err) - } - return nil - }) + cmd := exec.Command("qemu-system-x86_64", + "-enable-kvm", + "-cpu", "host", + "-m", fmt.Sprintf("%d", memory), + "-smp", fmt.Sprintf("%d", cpus), + "-serial", fmt.Sprintf("unix:/tmp/%s.sock,server,nowait", cfg.Name), + "-monitor", fmt.Sprintf("unix:/tmp/%s.mon-sock,server,nowait", cfg.Name), + "-qmp", fmt.Sprintf("unix:/tmp/%s.qmp-sock,server,nowait", cfg.Name), + "-display", "none", + "-drive", fmt.Sprintf("file=%s,if=virtio", cfg.VolumePath), + "-netdev", fmt.Sprintf("tap,id=net0,ifname=tap%d,script=no,downscript=no", cfg.TapID), + "-device", fmt.Sprintf("virtio-net-pci,netdev=net0,mac=%s", cfg.Mac), + "-daemonize", + ) + if err := cmd.Run(); err != nil { + return fmt.Errorf("qemu-system-x86_64: %w", err) + } + return nil } diff --git a/internal/qemu/start_other.go b/internal/qemu/start_other.go index 9fa57a1..782a7ee 100644 --- a/internal/qemu/start_other.go +++ b/internal/qemu/start_other.go @@ -5,10 +5,10 @@ package qemu import "errors" type Config struct { - Name, VpcName, Mac, VolumePath string - TapID, Memory, CPUs int + Name, Mac, VolumePath string + TapID, Memory, CPUs int } -func Start(cfg Config) error { +func Start(_ Config) error { return errors.New("vm: not supported on this platform") } diff --git a/internal/vm/create.go b/internal/vm/create.go index 6f732c3..4737e62 100644 --- a/internal/vm/create.go +++ b/internal/vm/create.go @@ -48,14 +48,15 @@ func StartVM(db *badger.DB, name string) error { return fmt.Errorf("start metadata: %w", err) } - if err := qemu.Start(qemu.Config{ - Name: name, - VpcName: d.vpcName, - TapID: d.tapID, - Mac: d.mac, - VolumePath: d.volumePath, - Memory: d.memory, - CPUs: d.cpus, + if err := netns.Call(d.vpcName, func() error { + return qemu.Start(qemu.Config{ + Name: name, + TapID: d.tapID, + Mac: d.mac, + VolumePath: d.volumePath, + Memory: d.memory, + CPUs: d.cpus, + }) }); err != nil { return fmt.Errorf("start qemu: %w", err) } diff --git a/internal/vm/data.go b/internal/vm/data.go index dfceb49..0563fc0 100644 --- a/internal/vm/data.go +++ b/internal/vm/data.go @@ -91,13 +91,19 @@ func loadVM(db *badger.DB, name string) (vmData, error) { if err != nil { return d, fmt.Errorf("get memory: %w", err) } - d.memory, _ = strconv.Atoi(memoryStr) + d.memory, err = strconv.Atoi(memoryStr) + if err != nil { + return d, fmt.Errorf("parse memory: %w", err) + } cpusStr, err := kv.GetFromDB(db, "vm/"+name+"/cpus") if err != nil { return d, fmt.Errorf("get cpus: %w", err) } - d.cpus, _ = strconv.Atoi(cpusStr) + d.cpus, err = strconv.Atoi(cpusStr) + if err != nil { + return d, fmt.Errorf("parse cpus: %w", err) + } d.password, _ = kv.GetFromDB(db, "vm/"+name+"/password") d.sshkey, _ = kv.GetFromDB(db, "vm/"+name+"/sshkey")