diff --git a/internal/config/agent/struct.go b/internal/config/agent/struct.go index 1c9fc9f..8818541 100644 --- a/internal/config/agent/struct.go +++ b/internal/config/agent/struct.go @@ -20,6 +20,10 @@ type Config struct { Count int `mapstructure:"count"` BufferSize int `mapstructure:"buffer_size"` } `mapstructure:"worker"` + Dispatcher struct { + TimeoutSeconds int `mapstructure:"timeout_seconds"` + PollSeconds int `mapstructure:"poll_seconds"` + } `mapstructure:"dispatcher"` DefaultInterface string `mapstructure:"default_interface"` Interfaces map[string]string `mapstructure:"interfaces"` } @@ -36,6 +40,8 @@ func LoadConfig(path string) (*Config, error) { v.SetDefault("prometheus.port", 9090) v.SetDefault("worker.count", 4) v.SetDefault("worker.buffer_size", 100) + v.SetDefault("dispatcher.timeout_seconds", 300) + v.SetDefault("dispatcher.poll_seconds", 2) v.SetDefault("default_interface", "br-000000") v.ReadInConfig() diff --git a/internal/dispatcher/agent/subnet_commands.go b/internal/dispatcher/agent/subnet_commands.go index c86cc77..7bc214d 100644 --- a/internal/dispatcher/agent/subnet_commands.go +++ b/internal/dispatcher/agent/subnet_commands.go @@ -4,6 +4,7 @@ import ( "fmt" "os" "strconv" + "time" configuration "git.g3e.fr/syonad/two/internal/config/agent" "git.g3e.fr/syonad/two/internal/subnet" @@ -44,7 +45,22 @@ func (c CreateSubnetCommand) Prepare(db *badger.DB, cfg *configuration.Config) e return nil } -func (c CreateSubnetCommand) Execute(db *badger.DB, _ *configuration.Config) error { +func (c CreateSubnetCommand) Execute(db *badger.DB, cfg *configuration.Config) error { + timeout := time.After(time.Duration(cfg.Dispatcher.TimeoutSeconds) * time.Second) + for { + state, err := kv.GetFromDB(db, "vpc/"+c.VPC+"/state") + if err != nil { + return fmt.Errorf("vpc %q not found while waiting", c.VPC) + } + if state == "created" { + break + } + select { + case <-timeout: + return fmt.Errorf("timed out waiting for vpc %q to be created", c.VPC) + case <-time.After(time.Duration(cfg.Dispatcher.PollSeconds) * time.Second): + } + } return subnet.CreateSubnet(db, c.Name) } diff --git a/internal/dispatcher/agent/vpc_commands.go b/internal/dispatcher/agent/vpc_commands.go index a1313e1..c03dd77 100644 --- a/internal/dispatcher/agent/vpc_commands.go +++ b/internal/dispatcher/agent/vpc_commands.go @@ -2,6 +2,8 @@ package dispatcher import ( "fmt" + "strings" + "time" configuration "git.g3e.fr/syonad/two/internal/config/agent" "git.g3e.fr/syonad/two/internal/vpc" @@ -32,10 +34,50 @@ 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) } + entries, err := kv.ListByPrefix(db, "subnet/") + if err != nil { + return fmt.Errorf("failed to list subnets: %w", err) + } + for key, value := range entries { + if !strings.HasSuffix(key, "/vpc") || value != c.Name { + continue + } + subnetName := strings.Split(key, "/")[1] + state, err := kv.GetFromDB(db, "subnet/"+subnetName+"/state") + if err != nil || (state != "deleting" && state != "deleted") { + return fmt.Errorf("subnet %q must be deleted before deleting vpc %q", subnetName, c.Name) + } + } return kv.AddInDB(db, "vpc/"+c.Name+"/state", "deleting") } -func (c DeleteVPCCommand) Execute(db *badger.DB, _ *configuration.Config) error { +func (c DeleteVPCCommand) Execute(db *badger.DB, cfg *configuration.Config) error { + timeout := time.After(time.Duration(cfg.Dispatcher.TimeoutSeconds) * time.Second) + for { + entries, err := kv.ListByPrefix(db, "subnet/") + if err != nil { + return fmt.Errorf("failed to list subnets: %w", err) + } + pending := false + for key, value := range entries { + if strings.HasSuffix(key, "/vpc") && value == c.Name { + subnetName := strings.Split(key, "/")[1] + state, _ := kv.GetFromDB(db, "subnet/"+subnetName+"/state") + if state == "deleting" { + pending = true + break + } + } + } + if !pending { + break + } + select { + case <-timeout: + return fmt.Errorf("timed out waiting for subnets of vpc %q to be deleted", c.Name) + case <-time.After(time.Duration(cfg.Dispatcher.PollSeconds) * time.Second): + } + } if err := vpc.DeleteVPC(db, c.Name); err != nil { return err }