diff --git a/api/openapi.yaml b/api/agent.yaml similarity index 100% rename from api/openapi.yaml rename to api/agent.yaml diff --git a/cmd/agent/main.go b/cmd/agent/main.go index ce3df4e..5ebf09f 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -6,10 +6,11 @@ import ( "log" agentapi "git.g3e.fr/syonad/two/internal/api/agent" - agentmetrics "git.g3e.fr/syonad/two/internal/prometheus/agent" configuration "git.g3e.fr/syonad/two/internal/config/agent" - promserver "git.g3e.fr/syonad/two/pkg/prometheus" + agentmetrics "git.g3e.fr/syonad/two/internal/prometheus/agent" "git.g3e.fr/syonad/two/pkg/db/kv" + promserver "git.g3e.fr/syonad/two/pkg/prometheus" + "git.g3e.fr/syonad/two/pkg/worker" "github.com/prometheus/client_golang/prometheus" ) @@ -22,16 +23,19 @@ func main() { log.Fatalf("failed to load config: %v", err) } - db := kv.InitDB(kv.Config{Path: cfg.Database.Path}, true) + db := kv.InitDB(kv.Config{Path: cfg.Database.Path}, false) defer db.Close() - apiAddr := fmt.Sprintf("%s:%d", cfg.Api.Address, cfg.Api.Port) - promAddr := fmt.Sprintf("%s:%d", cfg.Prometheus.Address, cfg.Prometheus.Port) + q := worker.New(cfg.Worker.BufferSize) + q.Start(cfg.Worker.Count) registry := prometheus.NewRegistry() registry.MustRegister(agentmetrics.NewAgentCollector(db)) - go agentapi.Start(apiAddr) + apiAddr := fmt.Sprintf("%s:%d", cfg.Api.Address, cfg.Api.Port) + promAddr := fmt.Sprintf("%s:%d", cfg.Prometheus.Address, cfg.Prometheus.Port) + + go agentapi.New(q).Start(apiAddr) go promserver.Start(promAddr, registry) select {} diff --git a/internal/api/agent/models.go b/internal/api/agent/models.go new file mode 100644 index 0000000..bd09c76 --- /dev/null +++ b/internal/api/agent/models.go @@ -0,0 +1,33 @@ +package agentapi + +type VPCCreateRequest struct { + Name string `json:"name"` +} + +type VPC struct { + Name string `json:"name"` + State string `json:"state"` +} + +type SubnetCreateRequest struct { + Name string `json:"name"` + VPC string `json:"vpc"` + VxlanID int `json:"vxlan_id"` + LocalIP string `json:"local_ip"` + GatewayIP string `json:"gateway_ip"` + CIDR string `json:"cidr"` +} + +type Subnet struct { + Name string `json:"name"` + State string `json:"state"` + VPC string `json:"vpc"` + VxlanID int `json:"vxlan_id"` + LocalIP string `json:"local_ip"` + GatewayIP string `json:"gateway_ip"` + CIDR string `json:"cidr"` +} + +type ErrorResponse struct { + Error string `json:"error"` +} diff --git a/internal/api/agent/server.go b/internal/api/agent/server.go index 05653a4..75833c9 100644 --- a/internal/api/agent/server.go +++ b/internal/api/agent/server.go @@ -3,14 +3,31 @@ package agentapi import ( "log" "net/http" + + "git.g3e.fr/syonad/two/pkg/worker" ) -func Start(address string) { - mux := http.NewServeMux() - mux.HandleFunc("/vpcs", VpcsHandler) - mux.HandleFunc("/vpcs/", VpcByNameHandler) - mux.HandleFunc("/subnets", SubnetsHandler) - mux.HandleFunc("/subnets/", SubnetByNameHandler) - log.Printf("API server listening on %s", address) - log.Fatal(http.ListenAndServe(address, mux)) +type Server struct { + queue *worker.Queue +} + +func New(queue *worker.Queue) *Server { + return &Server{queue: queue} +} + +func (s *Server) Start(address string) { + mux := http.NewServeMux() + mux.HandleFunc("/vpcs", s.VpcsHandler) + mux.HandleFunc("/vpcs/", s.VpcByNameHandler) + mux.HandleFunc("/subnets", s.SubnetsHandler) + mux.HandleFunc("/subnets/", s.SubnetByNameHandler) + log.Printf("API server listening on %s", address) + log.Fatal(http.ListenAndServe(address, logMiddleware(mux))) +} + +func logMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + log.Printf("%s %s %s", r.RemoteAddr, r.Method, r.URL.Path) + next.ServeHTTP(w, r) + }) } diff --git a/internal/api/agent/subnet.go b/internal/api/agent/subnet.go index e46e4f4..289667f 100644 --- a/internal/api/agent/subnet.go +++ b/internal/api/agent/subnet.go @@ -6,20 +6,36 @@ import ( "strings" ) -func SubnetByNameHandler(w http.ResponseWriter, r *http.Request) { +func (s *Server) SubnetByNameHandler(w http.ResponseWriter, r *http.Request) { name := strings.TrimPrefix(r.URL.Path, "/subnets/") if name == "" { - http.NotFound(w, r) + 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: - w.WriteHeader(http.StatusOK) - json.NewEncoder(w).Encode(map[string]string{"name": name}) + s.getSubnet(w, r, name) case http.MethodDelete: - w.WriteHeader(http.StatusAccepted) + s.deleteSubnet(w, r, name) default: - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed) } } + +func (s *Server) getSubnet(w http.ResponseWriter, r *http.Request, name string) { + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(Subnet{Name: name, State: "created"}) +} + +func (s *Server) deleteSubnet(w http.ResponseWriter, r *http.Request, name string) { + s.queue.Submit(func() { + destroySubnet(name) + }) + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(Subnet{Name: name, State: "deleting"}) +} + +func destroySubnet(name string) {} diff --git a/internal/api/agent/subnets.go b/internal/api/agent/subnets.go index 53dce2e..5e0ebdc 100644 --- a/internal/api/agent/subnets.go +++ b/internal/api/agent/subnets.go @@ -5,15 +5,48 @@ import ( "net/http" ) -func SubnetsHandler(w http.ResponseWriter, r *http.Request) { +func (s *Server) SubnetsHandler(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch r.Method { case http.MethodGet: - w.WriteHeader(http.StatusOK) - json.NewEncoder(w).Encode([]interface{}{}) + s.listSubnets(w, r) case http.MethodPost: - w.WriteHeader(http.StatusAccepted) + s.postSubnet(w, r) default: - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed) } } + +func (s *Server) listSubnets(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode([]Subnet{}) +} + +func (s *Server) postSubnet(w http.ResponseWriter, r *http.Request) { + var req SubnetCreateRequest + 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.VPC == "" || req.LocalIP == "" || req.GatewayIP == "" || req.CIDR == "" { + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(ErrorResponse{Error: "name, vpc, local_ip, gateway_ip and cidr are required"}) + return + } + s.queue.Submit(func() { + createSubnet(req) + }) + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(Subnet{ + Name: req.Name, + State: "creating", + VPC: req.VPC, + VxlanID: req.VxlanID, + LocalIP: req.LocalIP, + GatewayIP: req.GatewayIP, + CIDR: req.CIDR, + }) +} + +func createSubnet(req SubnetCreateRequest) {} diff --git a/internal/api/agent/vpc.go b/internal/api/agent/vpc.go index aecfaa1..f1d34c1 100644 --- a/internal/api/agent/vpc.go +++ b/internal/api/agent/vpc.go @@ -6,20 +6,36 @@ import ( "strings" ) -func VpcByNameHandler(w http.ResponseWriter, r *http.Request) { +func (s *Server) VpcByNameHandler(w http.ResponseWriter, r *http.Request) { name := strings.TrimPrefix(r.URL.Path, "/vpcs/") if name == "" { - http.NotFound(w, r) + 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: - w.WriteHeader(http.StatusOK) - json.NewEncoder(w).Encode(map[string]string{"name": name}) + s.getVpc(w, r, name) case http.MethodDelete: - w.WriteHeader(http.StatusAccepted) + s.deleteVpc(w, r, name) default: - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed) } } + +func (s *Server) getVpc(w http.ResponseWriter, r *http.Request, name string) { + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(VPC{Name: name, State: "created"}) +} + +func (s *Server) deleteVpc(w http.ResponseWriter, r *http.Request, name string) { + s.queue.Submit(func() { + destroyVpc(name) + }) + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(VPC{Name: name, State: "deleting"}) +} + +func destroyVpc(name string) {} diff --git a/internal/api/agent/vpcs.go b/internal/api/agent/vpcs.go index c4e9b27..aba4e7a 100644 --- a/internal/api/agent/vpcs.go +++ b/internal/api/agent/vpcs.go @@ -5,15 +5,40 @@ import ( "net/http" ) -func VpcsHandler(w http.ResponseWriter, r *http.Request) { +func (s *Server) VpcsHandler(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch r.Method { case http.MethodGet: - w.WriteHeader(http.StatusOK) - json.NewEncoder(w).Encode([]interface{}{}) + s.listVpcs(w, r) case http.MethodPost: - w.WriteHeader(http.StatusAccepted) + s.postVpc(w, r) default: - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed) } } + +func (s *Server) listVpcs(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode([]VPC{}) +} + +func (s *Server) postVpc(w http.ResponseWriter, r *http.Request) { + var req VPCCreateRequest + 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 == "" { + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(ErrorResponse{Error: "name is required"}) + return + } + s.queue.Submit(func() { + createVpc(req.Name) + }) + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(VPC{Name: req.Name, State: "creating"}) +} + +func createVpc(name string) {} diff --git a/internal/config/agent/struct.go b/internal/config/agent/struct.go index 0cca127..24ad0f5 100644 --- a/internal/config/agent/struct.go +++ b/internal/config/agent/struct.go @@ -16,6 +16,10 @@ type Config struct { Address string `mapstructure:"address"` Port int `mapstructure:"port"` } `mapstructure:"prometheus"` + Worker struct { + Count int `mapstructure:"count"` + BufferSize int `mapstructure:"buffer_size"` + } `mapstructure:"worker"` } func LoadConfig(path string) (*Config, error) { @@ -28,6 +32,8 @@ func LoadConfig(path string) (*Config, error) { v.SetDefault("api.port", 8080) v.SetDefault("prometheus.address", "") v.SetDefault("prometheus.port", 9090) + v.SetDefault("worker.count", 4) + v.SetDefault("worker.buffer_size", 100) v.ReadInConfig() diff --git a/pkg/worker/queue.go b/pkg/worker/queue.go new file mode 100644 index 0000000..109c726 --- /dev/null +++ b/pkg/worker/queue.go @@ -0,0 +1,33 @@ +package worker + +import "log" + +// Task is a function to be executed asynchronously by a worker. +type Task func() + +// Queue is a FIFO channel-backed task queue consumed by worker goroutines. +type Queue struct { + tasks chan Task +} + +// New creates a Queue with the given channel buffer size. +func New(bufferSize int) *Queue { + return &Queue{tasks: make(chan Task, bufferSize)} +} + +// Submit enqueues a task. Blocks if the queue is full. +func (q *Queue) Submit(t Task) { + q.tasks <- t +} + +// Start launches n worker goroutines that consume and execute tasks. +func (q *Queue) Start(n int) { + log.Printf("worker: starting %d workers", n) + for i := range n { + go func(id int) { + for task := range q.tasks { + task() + } + }(i) + } +}