From 1d86c45ba4e4bbcf48d5222dfcb628c617e9b210 Mon Sep 17 00:00:00 2001 From: GnomeZworc Date: Thu, 23 Apr 2026 23:11:10 +0200 Subject: [PATCH] f-21: dispatch: add a prepare step to dispatch Signed-off-by: GnomeZworc --- internal/api/agent/subnet.go | 13 +++++-- internal/api/agent/subnets.go | 41 +++++++++++++++----- internal/api/agent/vpc.go | 11 +++++- internal/api/agent/vpcs.go | 11 +++++- internal/dispatcher/agent/dispatcher.go | 5 +++ internal/dispatcher/agent/subnet_commands.go | 24 +++++++++++- internal/dispatcher/agent/vpc_commands.go | 18 ++++++++- 7 files changed, 102 insertions(+), 21 deletions(-) diff --git a/internal/api/agent/subnet.go b/internal/api/agent/subnet.go index 61cbd6f..556b7c3 100644 --- a/internal/api/agent/subnet.go +++ b/internal/api/agent/subnet.go @@ -61,8 +61,15 @@ func (s *Server) getSubnet(w http.ResponseWriter, _ *http.Request, name string) json.NewEncoder(w).Encode(sub) } -func (s *Server) deleteSubnet(w http.ResponseWriter, r *http.Request, name string) { - s.dispatcher.Dispatch(dispatcher.DeleteSubnetCommand{Name: name}) +func (s *Server) deleteSubnet(w http.ResponseWriter, _ *http.Request, name string) { + cmd := dispatcher.DeleteSubnetCommand{Name: name} + if err := s.dispatcher.Prepare(cmd); err != nil { + w.WriteHeader(http.StatusNotFound) + json.NewEncoder(w).Encode(ErrorResponse{Error: err.Error()}) + return + } + s.dispatcher.Dispatch(cmd) + state, _ := kv.GetFromDB(s.db, "subnet/"+name+"/state") w.WriteHeader(http.StatusAccepted) - json.NewEncoder(w).Encode(Subnet{Name: name, State: "deleting"}) + json.NewEncoder(w).Encode(Subnet{Name: name, State: state}) } diff --git a/internal/api/agent/subnets.go b/internal/api/agent/subnets.go index 18fb81f..9e99a45 100644 --- a/internal/api/agent/subnets.go +++ b/internal/api/agent/subnets.go @@ -74,21 +74,42 @@ func (s *Server) postSubnet(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(ErrorResponse{Error: "name, vpc, iface_type, gateway_ip and cidr are required"}) return } - s.dispatcher.Dispatch(dispatcher.CreateSubnetCommand{ + cmd := dispatcher.CreateSubnetCommand{ Name: req.Name, VPC: req.VPC, VxlanID: req.VxlanID, IfaceType: req.IfaceType, GatewayIP: req.GatewayIP, CIDR: req.CIDR, - }) + } + if err := s.dispatcher.Prepare(cmd); err != nil { + w.WriteHeader(http.StatusConflict) + json.NewEncoder(w).Encode(ErrorResponse{Error: err.Error()}) + return + } + s.dispatcher.Dispatch(cmd) + entries, _ := kv.ListByPrefix(s.db, "subnet/"+req.Name+"/") + sub := Subnet{Name: req.Name} + for key, value := range entries { + parts := strings.Split(key, "/") + if len(parts) != 3 { + continue + } + switch parts[2] { + case "state": + sub.State = value + case "vpc": + sub.VPC = value + case "vxlan_id": + sub.VxlanID, _ = strconv.Atoi(value) + case "local_iface": + sub.LocalIface = value + case "gateway_ip": + sub.GatewayIP = value + case "cidr": + sub.CIDR = value + } + } w.WriteHeader(http.StatusAccepted) - json.NewEncoder(w).Encode(Subnet{ - Name: req.Name, - State: "creating", - VPC: req.VPC, - VxlanID: req.VxlanID, - GatewayIP: req.GatewayIP, - CIDR: req.CIDR, - }) + json.NewEncoder(w).Encode(sub) } diff --git a/internal/api/agent/vpc.go b/internal/api/agent/vpc.go index fd1bfdb..be0724e 100644 --- a/internal/api/agent/vpc.go +++ b/internal/api/agent/vpc.go @@ -40,7 +40,14 @@ func (s *Server) getVpc(w http.ResponseWriter, _ *http.Request, name string) { } func (s *Server) deleteVpc(w http.ResponseWriter, _ *http.Request, name string) { - s.dispatcher.Dispatch(dispatcher.DeleteVPCCommand{Name: name}) + cmd := dispatcher.DeleteVPCCommand{Name: name} + if err := s.dispatcher.Prepare(cmd); err != nil { + w.WriteHeader(http.StatusNotFound) + json.NewEncoder(w).Encode(ErrorResponse{Error: err.Error()}) + return + } + s.dispatcher.Dispatch(cmd) + state, _ := kv.GetFromDB(s.db, "vpc/"+name+"/state") w.WriteHeader(http.StatusAccepted) - json.NewEncoder(w).Encode(VPC{Name: name, State: "deleting"}) + json.NewEncoder(w).Encode(VPC{Name: name, State: state}) } diff --git a/internal/api/agent/vpcs.go b/internal/api/agent/vpcs.go index 735259c..3087456 100644 --- a/internal/api/agent/vpcs.go +++ b/internal/api/agent/vpcs.go @@ -62,7 +62,14 @@ func (s *Server) postVpc(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(ErrorResponse{Error: "name is required"}) return } - s.dispatcher.Dispatch(dispatcher.CreateVPCCommand{Name: req.Name}) + cmd := dispatcher.CreateVPCCommand{Name: req.Name} + if err := s.dispatcher.Prepare(cmd); err != nil { + w.WriteHeader(http.StatusConflict) + json.NewEncoder(w).Encode(ErrorResponse{Error: err.Error()}) + return + } + s.dispatcher.Dispatch(cmd) + state, _ := kv.GetFromDB(s.db, "vpc/"+req.Name+"/state") w.WriteHeader(http.StatusAccepted) - json.NewEncoder(w).Encode(VPC{Name: req.Name, State: "creating"}) + json.NewEncoder(w).Encode(VPC{Name: req.Name, State: state}) } diff --git a/internal/dispatcher/agent/dispatcher.go b/internal/dispatcher/agent/dispatcher.go index 8e94827..6b1ca67 100644 --- a/internal/dispatcher/agent/dispatcher.go +++ b/internal/dispatcher/agent/dispatcher.go @@ -9,6 +9,7 @@ import ( ) type Command interface { + Prepare(db *badger.DB, cfg *configuration.Config) error Execute(db *badger.DB, cfg *configuration.Config) error } @@ -22,6 +23,10 @@ func New(queue *worker.Queue, db *badger.DB, cfg *configuration.Config) *Dispatc return &Dispatcher{queue: queue, db: db, cfg: cfg} } +func (d *Dispatcher) Prepare(cmd Command) error { + return cmd.Prepare(d.db, d.cfg) +} + func (d *Dispatcher) Dispatch(cmd Command) { d.queue.Submit(func() { if err := cmd.Execute(d.db, d.cfg); err != nil { diff --git a/internal/dispatcher/agent/subnet_commands.go b/internal/dispatcher/agent/subnet_commands.go index 0d471c3..c86cc77 100644 --- a/internal/dispatcher/agent/subnet_commands.go +++ b/internal/dispatcher/agent/subnet_commands.go @@ -20,7 +20,17 @@ type CreateSubnetCommand struct { CIDR string } -func (c CreateSubnetCommand) Execute(db *badger.DB, cfg *configuration.Config) error { +func (c CreateSubnetCommand) Prepare(db *badger.DB, cfg *configuration.Config) error { + if _, err := kv.GetFromDB(db, "subnet/"+c.Name+"/state"); err == nil { + return fmt.Errorf("subnet %q already exists", c.Name) + } + vpcState, err := kv.GetFromDB(db, "vpc/"+c.VPC+"/state") + if err != nil { + return fmt.Errorf("vpc %q not found", c.VPC) + } + if vpcState == "deleting" || vpcState == "deleted" { + return fmt.Errorf("vpc %q is %s", c.VPC, vpcState) + } localIface, ok := cfg.Interfaces[c.IfaceType] if !ok { localIface = cfg.DefaultInterface @@ -31,6 +41,10 @@ func (c CreateSubnetCommand) Execute(db *badger.DB, cfg *configuration.Config) e kv.AddInDB(db, "subnet/"+c.Name+"/local_iface", localIface) kv.AddInDB(db, "subnet/"+c.Name+"/gateway_ip", c.GatewayIP) kv.AddInDB(db, "subnet/"+c.Name+"/cidr", c.CIDR) + return nil +} + +func (c CreateSubnetCommand) Execute(db *badger.DB, _ *configuration.Config) error { return subnet.CreateSubnet(db, c.Name) } @@ -38,8 +52,14 @@ type DeleteSubnetCommand struct { Name string } +func (c DeleteSubnetCommand) Prepare(db *badger.DB, _ *configuration.Config) error { + if _, err := kv.GetFromDB(db, "subnet/"+c.Name+"/state"); err != nil { + return fmt.Errorf("subnet %q not found", c.Name) + } + return kv.AddInDB(db, "subnet/"+c.Name+"/state", "deleting") +} + func (c DeleteSubnetCommand) Execute(db *badger.DB, _ *configuration.Config) error { - kv.AddInDB(db, "subnet/"+c.Name+"/state", "deleting") if err := subnet.DeleteSubnet(db, c.Name); err != nil { fmt.Println(err) os.Exit(1) diff --git a/internal/dispatcher/agent/vpc_commands.go b/internal/dispatcher/agent/vpc_commands.go index b2687fc..a1313e1 100644 --- a/internal/dispatcher/agent/vpc_commands.go +++ b/internal/dispatcher/agent/vpc_commands.go @@ -1,6 +1,8 @@ package dispatcher import ( + "fmt" + configuration "git.g3e.fr/syonad/two/internal/config/agent" "git.g3e.fr/syonad/two/internal/vpc" "git.g3e.fr/syonad/two/pkg/db/kv" @@ -11,8 +13,14 @@ type CreateVPCCommand struct { Name string } +func (c CreateVPCCommand) Prepare(db *badger.DB, _ *configuration.Config) error { + if _, err := kv.GetFromDB(db, "vpc/"+c.Name+"/state"); err == nil { + return fmt.Errorf("vpc %q already exists", c.Name) + } + return kv.AddInDB(db, "vpc/"+c.Name+"/state", "creating") +} + func (c CreateVPCCommand) Execute(db *badger.DB, _ *configuration.Config) error { - kv.AddInDB(db, "vpc/"+c.Name+"/state", "creating") return vpc.CreateVPC(db, c.Name) } @@ -20,8 +28,14 @@ type DeleteVPCCommand struct { Name string } +func (c DeleteVPCCommand) Prepare(db *badger.DB, _ *configuration.Config) error { + if _, err := kv.GetFromDB(db, "vpc/"+c.Name+"/state"); err != nil { + return fmt.Errorf("vpc %q not found", c.Name) + } + return kv.AddInDB(db, "vpc/"+c.Name+"/state", "deleting") +} + func (c DeleteVPCCommand) Execute(db *badger.DB, _ *configuration.Config) error { - kv.AddInDB(db, "vpc/"+c.Name+"/state", "deleting") if err := vpc.DeleteVPC(db, c.Name); err != nil { return err }