diff --git a/agent b/agent new file mode 100755 index 0000000..e6b9fa4 Binary files /dev/null and b/agent differ diff --git a/api/openapi.yaml b/api/openapi.yaml new file mode 100644 index 0000000..be328d0 --- /dev/null +++ b/api/openapi.yaml @@ -0,0 +1,296 @@ +openapi: "3.1.0" +info: + title: Two API + version: "0.1.0" + description: REST API for managing VPCs and Subnets in the Two orchestrator. + +servers: + - url: http://localhost:8080 + description: Local development server + +paths: + + # ── VPC ──────────────────────────────────────────────────────────────────── + + /vpcs: + get: + summary: List all VPCs + operationId: listVPCs + responses: + "200": + description: List of VPCs + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/VPC" + "500": + $ref: "#/components/responses/InternalError" + + post: + summary: Create a VPC + operationId: createVPC + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/VPCCreateRequest" + responses: + "202": + description: VPC creation accepted + content: + application/json: + schema: + $ref: "#/components/schemas/VPC" + "409": + description: VPC already exists + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + "500": + $ref: "#/components/responses/InternalError" + + /vpcs/{name}: + parameters: + - $ref: "#/components/parameters/ResourceName" + + get: + summary: Get VPC status and info + operationId: getVPC + responses: + "200": + description: VPC found + content: + application/json: + schema: + $ref: "#/components/schemas/VPC" + "404": + $ref: "#/components/responses/NotFound" + "500": + $ref: "#/components/responses/InternalError" + + delete: + summary: Delete a VPC + operationId: deleteVPC + responses: + "202": + description: VPC deletion accepted + content: + application/json: + schema: + $ref: "#/components/schemas/VPC" + "404": + $ref: "#/components/responses/NotFound" + "409": + description: VPC not in a deletable state + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + "500": + $ref: "#/components/responses/InternalError" + + # ── Subnet ───────────────────────────────────────────────────────────────── + + /subnets: + get: + summary: List all subnets + operationId: listSubnets + responses: + "200": + description: List of subnets + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/Subnet" + "500": + $ref: "#/components/responses/InternalError" + + post: + summary: Create a subnet + operationId: createSubnet + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/SubnetCreateRequest" + responses: + "202": + description: Subnet creation accepted + content: + application/json: + schema: + $ref: "#/components/schemas/Subnet" + "409": + description: Subnet already exists + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + "422": + description: Parent VPC does not exist or is not ready + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + "500": + $ref: "#/components/responses/InternalError" + + /subnets/{name}: + parameters: + - $ref: "#/components/parameters/ResourceName" + + get: + summary: Get subnet status and info + operationId: getSubnet + responses: + "200": + description: Subnet found + content: + application/json: + schema: + $ref: "#/components/schemas/Subnet" + "404": + $ref: "#/components/responses/NotFound" + "500": + $ref: "#/components/responses/InternalError" + + delete: + summary: Delete a subnet + operationId: deleteSubnet + responses: + "202": + description: Subnet deletion accepted + content: + application/json: + schema: + $ref: "#/components/schemas/Subnet" + "404": + $ref: "#/components/responses/NotFound" + "409": + description: Subnet not in a deletable state + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + "500": + $ref: "#/components/responses/InternalError" + +# ── Components ────────────────────────────────────────────────────────────── + +components: + + parameters: + ResourceName: + name: name + in: path + required: true + schema: + type: string + description: Resource name + + schemas: + + VPCCreateRequest: + type: object + required: [name] + properties: + name: + type: string + description: Unique name for the VPC + example: vpc1 + + VPC: + type: object + properties: + name: + type: string + example: vpc1 + state: + type: string + enum: [creating, created, deleting, deleted] + example: created + + SubnetCreateRequest: + type: object + required: [name, vpc, vxlan_id, local_ip, gateway_ip, cidr] + properties: + name: + type: string + description: Unique name for the subnet + example: sn-00001 + vpc: + type: string + description: Parent VPC name + example: vpc1 + vxlan_id: + type: integer + description: VXLAN VNI identifier + example: 100 + local_ip: + type: string + format: ipv4 + description: Local VTEP IP address + example: "10.0.0.5" + gateway_ip: + type: string + format: ipv4 + description: Gateway IP for the subnet + example: "10.10.10.1" + cidr: + type: string + description: Subnet CIDR block + example: "10.10.10.0/24" + + Subnet: + type: object + properties: + name: + type: string + example: sn-00001 + state: + type: string + enum: [creating, created, deleting, deleted] + example: created + vpc: + type: string + example: vpc1 + vxlan_id: + type: integer + example: 100 + local_ip: + type: string + example: "10.0.0.5" + gateway_ip: + type: string + example: "10.10.10.1" + cidr: + type: string + example: "10.10.10.0/24" + + Error: + type: object + properties: + error: + type: string + example: "resource not found" + + responses: + NotFound: + description: Resource not found + content: + application/json: + schema: + $ref: "#/components/schemas/Error" + InternalError: + description: Internal server error + content: + application/json: + schema: + $ref: "#/components/schemas/Error" diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 0f82e93..56011bf 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -1,17 +1,42 @@ package main import ( + "flag" "fmt" - "os" -) + "log" -var ( - bin_name = os.Args[0] + agentapi "git.g3e.fr/syonad/two/internal/api/agent" + configuration "git.g3e.fr/syonad/two/internal/config/agent" + 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" ) func main() { + confFile := flag.String("config", "/etc/two/agent.yml", "config file path") + flag.Parse() - fmt.Printf("%s: Start process\n", bin_name) + cfg, err := configuration.LoadConfig(*confFile) + if err != nil { + log.Fatalf("failed to load config: %v", err) + } - os.Exit(0) + db := kv.InitDB(kv.Config{Path: cfg.Database.Path}, true) + defer db.Close() + + q := worker.New(cfg.Worker.BufferSize) + q.Start(cfg.Worker.Count) + + registry := prometheus.NewRegistry() + registry.MustRegister(agentmetrics.NewAgentCollector(db)) + + 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/go.mod b/go.mod index 7430e72..a4b0b38 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.24.0 toolchain go1.24.11 require ( + github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/coreos/go-systemd/v22 v22.6.0 // indirect github.com/dgraph-io/badger/v4 v4.8.0 // indirect @@ -17,7 +18,12 @@ require ( github.com/godbus/dbus/v5 v5.1.0 // indirect github.com/google/flatbuffers v25.2.10+incompatible // indirect github.com/klauspost/compress v1.18.0 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect + github.com/prometheus/client_golang v1.23.2 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.66.1 // indirect + github.com/prometheus/procfs v0.16.1 // indirect github.com/sagikazarmark/locafero v0.11.0 // indirect github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect github.com/spf13/afero v1.15.0 // indirect @@ -31,9 +37,10 @@ require ( go.opentelemetry.io/otel v1.37.0 // indirect go.opentelemetry.io/otel/metric v1.37.0 // indirect go.opentelemetry.io/otel/trace v1.37.0 // indirect + go.yaml.in/yaml/v2 v2.4.2 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect - golang.org/x/net v0.41.0 // indirect + golang.org/x/net v0.43.0 // indirect golang.org/x/sys v0.39.0 // indirect golang.org/x/text v0.28.0 // indirect - google.golang.org/protobuf v1.36.6 // indirect + google.golang.org/protobuf v1.36.8 // indirect ) diff --git a/go.sum b/go.sum index 73d16f3..402452d 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/coreos/go-systemd/v22 v22.6.0 h1:aGVa/v8B7hpb0TKl0MWoAavPDmHvobFe5R5zn0bCJWo= @@ -23,8 +25,18 @@ github.com/google/flatbuffers v25.2.10+incompatible h1:F3vclr7C3HpB1k9mxCGRMXq6F github.com/google/flatbuffers v25.2.10+incompatible/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8= github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= +github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= +github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= +github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= +github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg= +github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc= github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik= github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw= @@ -51,10 +63,14 @@ go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/Wgbsd go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E= go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4= go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0= +go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= +go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/net v0.41.0 h1:vBTly1HeNPEn3wtREYfy4GZ/NECgw2Cnl+nK6Nz3uvw= golang.org/x/net v0.41.0/go.mod h1:B/K4NNqkfmg07DQYrbwvSluqCJOOXwUjeb/5lOisjbA= +golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE= +golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg= golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.34.0 h1:H5Y5sJ2L2JRdyv7ROF1he/lPdvFsd0mJHFw2ThKHxLA= @@ -65,4 +81,6 @@ golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng= golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU= google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY= google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY= +google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= +google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/internal/api/agent/server.go b/internal/api/agent/server.go new file mode 100644 index 0000000..aebe2c3 --- /dev/null +++ b/internal/api/agent/server.go @@ -0,0 +1,26 @@ +package agentapi + +import ( + "log" + "net/http" + + "git.g3e.fr/syonad/two/pkg/worker" +) + +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, mux)) +} diff --git a/internal/api/agent/subnet.go b/internal/api/agent/subnet.go new file mode 100644 index 0000000..52b27d0 --- /dev/null +++ b/internal/api/agent/subnet.go @@ -0,0 +1,30 @@ +package agentapi + +import ( + "encoding/json" + "net/http" + "strings" +) + +func (s *Server) SubnetByNameHandler(w http.ResponseWriter, r *http.Request) { + name := strings.TrimPrefix(r.URL.Path, "/subnets/") + if name == "" { + http.NotFound(w, r) + 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}) + 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 new file mode 100644 index 0000000..b3d587b --- /dev/null +++ b/internal/api/agent/subnets.go @@ -0,0 +1,24 @@ +package agentapi + +import ( + "encoding/json" + "net/http" +) + +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 new file mode 100644 index 0000000..73cecd9 --- /dev/null +++ b/internal/api/agent/vpc.go @@ -0,0 +1,30 @@ +package agentapi + +import ( + "encoding/json" + "net/http" + "strings" +) + +func (s *Server) VpcByNameHandler(w http.ResponseWriter, r *http.Request) { + name := strings.TrimPrefix(r.URL.Path, "/vpcs/") + if name == "" { + http.NotFound(w, r) + 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}) + 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 new file mode 100644 index 0000000..57386b6 --- /dev/null +++ b/internal/api/agent/vpcs.go @@ -0,0 +1,24 @@ +package agentapi + +import ( + "encoding/json" + "net/http" +) + +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 c9537bf..24ad0f5 100644 --- a/internal/config/agent/struct.go +++ b/internal/config/agent/struct.go @@ -8,6 +8,18 @@ type Config struct { Database struct { Path string `mapstructure:"path"` } `mapstructure:"database"` + Api struct { + Address string `mapstructure:"address"` + Port int `mapstructure:"port"` + } `mapstructure:"api"` + Prometheus 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) { @@ -16,6 +28,12 @@ func LoadConfig(path string) (*Config, error) { v.SetConfigType("yaml") v.SetDefault("database.path", "/var/lib/two/data/") + v.SetDefault("api.address", "") + 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/internal/prometheus/agent/collector.go b/internal/prometheus/agent/collector.go new file mode 100644 index 0000000..04712ad --- /dev/null +++ b/internal/prometheus/agent/collector.go @@ -0,0 +1,67 @@ +package agentmetrics + +import ( + "strings" + + "git.g3e.fr/syonad/two/pkg/db/kv" + "github.com/dgraph-io/badger/v4" + "github.com/prometheus/client_golang/prometheus" +) + +var allStates = []string{"creating", "created", "deleting", "deleted"} + +// AgentCollector implements prometheus.Collector and exposes agent metrics +// by querying the BadgerDB on each scrape. +type AgentCollector struct { + db *badger.DB + vpcsTotal *prometheus.Desc + subnetsTotal *prometheus.Desc +} + +func NewAgentCollector(db *badger.DB) *AgentCollector { + return &AgentCollector{ + db: db, + vpcsTotal: prometheus.NewDesc( + "syonad_vpcs_total", + "Number of VPCs by state.", + []string{"state"}, nil, + ), + subnetsTotal: prometheus.NewDesc( + "syonad_subnets_total", + "Number of subnets by state.", + []string{"state"}, nil, + ), + } +} + +func (c *AgentCollector) Describe(ch chan<- *prometheus.Desc) { + ch <- c.vpcsTotal + ch <- c.subnetsTotal +} + +func (c *AgentCollector) Collect(ch chan<- prometheus.Metric) { + c.collectStates(ch, "vpc/", c.vpcsTotal) + c.collectStates(ch, "subnet/", c.subnetsTotal) +} + +// collectStates counts resources under the given DB prefix by their state value +// and emits one gauge per state label. +func (c *AgentCollector) collectStates(ch chan<- prometheus.Metric, prefix string, desc *prometheus.Desc) { + counts := make(map[string]float64, len(allStates)) + for _, s := range allStates { + counts[s] = 0 + } + + items, err := kv.ListByPrefix(c.db, prefix) + if err == nil { + for key, val := range items { + if strings.HasSuffix(key, "/state") { + counts[val]++ + } + } + } + + for _, state := range allStates { + ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, counts[state], state) + } +} diff --git a/pkg/db/kv/listByPrefix.go b/pkg/db/kv/listByPrefix.go new file mode 100644 index 0000000..e6c095e --- /dev/null +++ b/pkg/db/kv/listByPrefix.go @@ -0,0 +1,33 @@ +package kv + +import ( + "github.com/dgraph-io/badger/v4" +) + +// ListByPrefix returns all key-value pairs whose key starts with prefix. +func ListByPrefix(db *badger.DB, prefix string) (map[string]string, error) { + result := make(map[string]string) + p := []byte(prefix) + + err := db.View(func(txn *badger.Txn) error { + opts := badger.DefaultIteratorOptions + opts.PrefetchSize = 10 + + it := txn.NewIterator(opts) + defer it.Close() + + for it.Seek(p); it.ValidForPrefix(p); it.Next() { + item := it.Item() + key := string(item.Key()) + + val, err := item.ValueCopy(nil) + if err != nil { + return err + } + result[key] = string(val) + } + return nil + }) + + return result, err +} diff --git a/pkg/prometheus/server.go b/pkg/prometheus/server.go new file mode 100644 index 0000000..9f8e557 --- /dev/null +++ b/pkg/prometheus/server.go @@ -0,0 +1,20 @@ +package promserver + +import ( + "log" + "net/http" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +// Start launches the Prometheus metrics HTTP server on the given address. +// The provided registry is used to expose metrics at /metrics. +func Start(address string, registry *prometheus.Registry) { + mux := http.NewServeMux() + mux.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{ + EnableOpenMetrics: true, + })) + log.Printf("Prometheus server listening on %s", address) + log.Fatal(http.ListenAndServe(address, mux)) +} 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) + } +}