From 8637c33bd8f7f7dda132a6706611734bfd1b3ae6 Mon Sep 17 00:00:00 2001 From: GnomeZworc Date: Tue, 14 Apr 2026 21:24:13 +0200 Subject: [PATCH 1/4] f-21: code: ajout d'un system de worker Signed-off-by: GnomeZworc --- cmd/agent/main.go | 14 +++++++++----- internal/api/agent/server.go | 20 +++++++++++++++----- internal/api/agent/subnet.go | 7 ++++++- internal/api/agent/subnets.go | 7 ++++++- internal/api/agent/vpc.go | 7 ++++++- internal/api/agent/vpcs.go | 7 ++++++- internal/config/agent/struct.go | 6 ++++++ pkg/worker/queue.go | 33 +++++++++++++++++++++++++++++++++ 8 files changed, 87 insertions(+), 14 deletions(-) create mode 100644 pkg/worker/queue.go diff --git a/cmd/agent/main.go b/cmd/agent/main.go index ce3df4e..56011bf 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" ) @@ -25,13 +26,16 @@ func main() { db := kv.InitDB(kv.Config{Path: cfg.Database.Path}, true) 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/server.go b/internal/api/agent/server.go index 05653a4..aebe2c3 100644 --- a/internal/api/agent/server.go +++ b/internal/api/agent/server.go @@ -3,14 +3,24 @@ package agentapi import ( "log" "net/http" + + "git.g3e.fr/syonad/two/pkg/worker" ) -func Start(address string) { +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", VpcsHandler) - mux.HandleFunc("/vpcs/", VpcByNameHandler) - mux.HandleFunc("/subnets", SubnetsHandler) - mux.HandleFunc("/subnets/", SubnetByNameHandler) + 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, mux)) } diff --git a/internal/api/agent/subnet.go b/internal/api/agent/subnet.go index e46e4f4..52b27d0 100644 --- a/internal/api/agent/subnet.go +++ b/internal/api/agent/subnet.go @@ -6,7 +6,7 @@ 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) @@ -18,8 +18,13 @@ func SubnetByNameHandler(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) json.NewEncoder(w).Encode(map[string]string{"name": name}) case http.MethodDelete: + s.queue.Submit(func() { + deleteSubnet(name) + }) w.WriteHeader(http.StatusAccepted) default: http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } } + +func deleteSubnet(name string) {} diff --git a/internal/api/agent/subnets.go b/internal/api/agent/subnets.go index 53dce2e..b3d587b 100644 --- a/internal/api/agent/subnets.go +++ b/internal/api/agent/subnets.go @@ -5,15 +5,20 @@ 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{}{}) case http.MethodPost: + s.queue.Submit(func() { + createSubnet() + }) w.WriteHeader(http.StatusAccepted) default: http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } } + +func createSubnet() {} diff --git a/internal/api/agent/vpc.go b/internal/api/agent/vpc.go index aecfaa1..73cecd9 100644 --- a/internal/api/agent/vpc.go +++ b/internal/api/agent/vpc.go @@ -6,7 +6,7 @@ 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) @@ -18,8 +18,13 @@ func VpcByNameHandler(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) json.NewEncoder(w).Encode(map[string]string{"name": name}) case http.MethodDelete: + s.queue.Submit(func() { + deleteVpc(name) + }) w.WriteHeader(http.StatusAccepted) default: http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } } + +func deleteVpc(name string) {} diff --git a/internal/api/agent/vpcs.go b/internal/api/agent/vpcs.go index c4e9b27..57386b6 100644 --- a/internal/api/agent/vpcs.go +++ b/internal/api/agent/vpcs.go @@ -5,15 +5,20 @@ 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{}{}) case http.MethodPost: + s.queue.Submit(func() { + createVpc() + }) w.WriteHeader(http.StatusAccepted) default: http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } } + +func createVpc() {} 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) + } +} From 24536ed89104198d795fb7167d4c8cc3183ee6d3 Mon Sep 17 00:00:00 2001 From: GnomeZworc Date: Thu, 16 Apr 2026 22:40:27 +0200 Subject: [PATCH 2/4] f-21: code: set db tu readwrite Signed-off-by: GnomeZworc --- cmd/agent/main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 56011bf..5ebf09f 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -23,7 +23,7 @@ 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() q := worker.New(cfg.Worker.BufferSize) From 75e02bd4f7f05692211b3083786c031fbe893c31 Mon Sep 17 00:00:00 2001 From: GnomeZworc Date: Thu, 16 Apr 2026 22:57:20 +0200 Subject: [PATCH 3/4] f-21: code: separate and create route fonction Signed-off-by: GnomeZworc --- internal/api/agent/models.go | 33 ++++++++++++++++++++++++++ internal/api/agent/server.go | 9 ++++++- internal/api/agent/subnet.go | 29 ++++++++++++++++------- internal/api/agent/subnets.go | 44 ++++++++++++++++++++++++++++------- internal/api/agent/vpc.go | 29 ++++++++++++++++------- internal/api/agent/vpcs.go | 36 +++++++++++++++++++++------- 6 files changed, 145 insertions(+), 35 deletions(-) create mode 100644 internal/api/agent/models.go 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 aebe2c3..75833c9 100644 --- a/internal/api/agent/server.go +++ b/internal/api/agent/server.go @@ -22,5 +22,12 @@ func (s *Server) Start(address string) { mux.HandleFunc("/subnets", s.SubnetsHandler) mux.HandleFunc("/subnets/", s.SubnetByNameHandler) log.Printf("API server listening on %s", address) - log.Fatal(http.ListenAndServe(address, mux)) + 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 52b27d0..289667f 100644 --- a/internal/api/agent/subnet.go +++ b/internal/api/agent/subnet.go @@ -9,22 +9,33 @@ import ( 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: - s.queue.Submit(func() { - deleteSubnet(name) - }) - 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 deleteSubnet(name string) {} +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 b3d587b..5e0ebdc 100644 --- a/internal/api/agent/subnets.go +++ b/internal/api/agent/subnets.go @@ -9,16 +9,44 @@ 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: - s.queue.Submit(func() { - createSubnet() - }) - 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 createSubnet() {} +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 73cecd9..f1d34c1 100644 --- a/internal/api/agent/vpc.go +++ b/internal/api/agent/vpc.go @@ -9,22 +9,33 @@ import ( 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: - s.queue.Submit(func() { - deleteVpc(name) - }) - 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 deleteVpc(name string) {} +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 57386b6..aba4e7a 100644 --- a/internal/api/agent/vpcs.go +++ b/internal/api/agent/vpcs.go @@ -9,16 +9,36 @@ 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: - s.queue.Submit(func() { - createVpc() - }) - 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 createVpc() {} +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) {} From b0c73b29522bdf6660ea1de898ee62fe9d4c7e25 Mon Sep 17 00:00:00 2001 From: GnomeZworc Date: Thu, 16 Apr 2026 22:57:53 +0200 Subject: [PATCH 4/4] f-21: doc: rename api doc Signed-off-by: GnomeZworc --- api/{openapi.yaml => agent.yaml} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename api/{openapi.yaml => agent.yaml} (100%) diff --git a/api/openapi.yaml b/api/agent.yaml similarity index 100% rename from api/openapi.yaml rename to api/agent.yaml