syonad/two
2
0
Fork 0
Code Issues 6 Pull requests 1 Projects Releases 9 Packages Wiki Activity Actions

Merge branch 'feature-21'

Browse source
This commit is contained in:
GnomeZworc 2026-04-26 16:16:48 +02:00
commit 877cfc7260
Signed by: nicolas.boufideline
GPG key ID: 4406BBBF8845D632
43 changed files with 2480 additions and 302 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.forgejo/workflows/prerelease.yml
View file

@ -37,9 +37,6 @@ jobs:
- metadata - metadata
- metacli - metacli
- agent - agent
- vpc
- dhcp
- subnet
uses: ./.forgejo/workflows/build.yml uses: ./.forgejo/workflows/build.yml
with: with:
tag: ${{ needs.set-release-target.outputs.release_cible }} tag: ${{ needs.set-release-target.outputs.release_cible }}

3
.gitignore vendored
View file

@ -15,6 +15,9 @@
# Output of the go coverage tool, specifically when used with LiteIDE # Output of the go coverage tool, specifically when used with LiteIDE
*.out *.out
# Output of the go coverage tool, specifically when used with LiteIDE
.claude
# Dependency directories (remove the comment below to include it) # Dependency directories (remove the comment below to include it)
# vendor/ # vendor/

303
api/agent.yaml Normal file
View file

@ -0,0 +1,303 @@
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"
"400":
description: Missing required field or invalid request body
content:
application/json:
schema:
$ref: "#/components/schemas/Error"
"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"
"400":
description: Missing required field or unknown iface_type
content:
application/json:
schema:
$ref: "#/components/schemas/Error"
"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"
"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, must follow the format vp-[id]
pattern: '^vp-.+'
example: vp-00001
VPC:
type: object
properties:
name:
type: string
example: vp-00001
state:
type: string
enum: [creating, created, deleting, deleted]
example: created
SubnetCreateRequest:
type: object
required: [name, vpc, vxlan_id, 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
iface_type:
type: string
description: Interface type key defined in the agent config (e.g. vms, internet, admin). Falls back to default_interface if omitted or unknown.
example: vms
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_iface:
type: string
description: Resolved interface name
example: br-000000
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"

51
cmd/agent/main.go
View file

@ -1,17 +1,56 @@
package main package main
import ( import (
"flag"
"fmt" "fmt"
"os" "log/slog"
)
var ( agentapi "git.g3e.fr/syonad/two/internal/api/agent"
bin_name = os.Args[0] configuration "git.g3e.fr/syonad/two/internal/config/agent"
dispatcher "git.g3e.fr/syonad/two/internal/dispatcher/agent"
agentmetrics "git.g3e.fr/syonad/two/internal/prometheus/agent"
"git.g3e.fr/syonad/two/pkg/db/kv"
"git.g3e.fr/syonad/two/pkg/logger"
promserver "git.g3e.fr/syonad/two/pkg/prometheus"
"git.g3e.fr/syonad/two/pkg/worker"
"github.com/prometheus/client_golang/prometheus"
) )
func main() { 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 {
slog.Error("failed to load config", "error", err)
return
}
os.Exit(0) log := logger.New(cfg.Logger.Level, cfg.Logger.Debug)
db := kv.InitDB(kv.Config{Path: cfg.Database.Path}, false)
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)
log.Info("starting agent",
"api", apiAddr,
"prometheus", promAddr,
"workers", cfg.Worker.Count,
"log_level", cfg.Logger.Level,
"debug", cfg.Logger.Debug,
)
d := dispatcher.New(q, db, cfg, log.With(slog.String("component", "dispatcher")))
go agentapi.New(d, db, log.With(slog.String("component", "api"))).Start(apiAddr)
go promserver.Start(promAddr, registry)
select {}
} }

64
cmd/dhcp/main.go
View file

@ -1,64 +0,0 @@
package main
import (
"flag"
"fmt"
"net"
"os"
"git.g3e.fr/syonad/two/internal/dhcp"
"git.g3e.fr/syonad/two/pkg/systemd"
)
func main() {
subnet := flag.String("subnet", "", "Subnet CIDR (e.g. 10.10.10.0/24)")
name := flag.String("name", "", "Config name (e.g. vpc1_br-00002)")
gateway := flag.String("gateway", "", "Gateway IP (e.g. 10.10.10.1)")
confDir := flag.String("confdir", "/etc/dnsmasq.d", "dnsmasq config directory")
flag.Parse()
if *subnet == "" || *name == "" || *gateway == "" {
flag.Usage()
os.Exit(1)
}
_, network, err := net.ParseCIDR(*subnet)
if err != nil {
fmt.Fprintf(os.Stderr, "invalid subnet: %v\n", err)
os.Exit(1)
}
gw := net.ParseIP(*gateway)
if gw == nil {
fmt.Fprintf(os.Stderr, "invalid gateway IP: %q\n", *gateway)
os.Exit(1)
}
conf := dhcp.Config{
Network: network,
Gateway: gw,
Name: *name,
ConfDir: *confDir,
}
confPath, err := dhcp.GenerateConfig(conf)
if err != nil {
fmt.Fprintf(os.Stderr, "error generating config: %v\n", err)
os.Exit(1)
}
fmt.Printf("dnsmasq config written to %s\n", confPath)
svc, err := systemd.New()
if err != nil {
fmt.Fprintf(os.Stderr, "error connecting to systemd: %v\n", err)
os.Exit(1)
}
defer svc.Close()
unit := "dnsmasq@" + *name + ".service"
if err := svc.Start(unit); err != nil {
fmt.Fprintf(os.Stderr, "error starting %s: %v\n", unit, err)
os.Exit(1)
}
fmt.Printf("started %s\n", unit)
}

93
cmd/subnet/main.go
View file

@ -1,93 +0,0 @@
package main
import (
"flag"
"fmt"
"os"
configuration "git.g3e.fr/syonad/two/internal/config/agent"
"git.g3e.fr/syonad/two/internal/subnet"
"git.g3e.fr/syonad/two/pkg/db/kv"
"github.com/dgraph-io/badger/v4"
)
var (
name = flag.String("name", "", "Subnet name (ex: sn-00001)")
vpcName = flag.String("vpc", "", "VPC name")
vxlanID = flag.String("vxlan-id", "", "VXLAN ID")
localIP = flag.String("local-ip", "", "Local VTEP IP")
gatewayIP = flag.String("gateway-ip", "", "Gateway IP")
cidr = flag.String("cidr", "", "Subnet CIDR (ex: 10.10.10.0/24)")
action = flag.String("action", "", "Action à effectuer")
conf_file = flag.String("conf", "/etc/two/agent.yml", "Configuration file")
)
var DB *badger.DB
func main() {
flag.Parse()
conf, err := configuration.LoadConfig(*conf_file)
if err != nil {
fmt.Println(err)
os.Exit(1)
}
DB = kv.InitDB(kv.Config{
Path: conf.Database.Path,
}, false)
defer DB.Close()
switch *action {
case "create":
if *name == "" || *vpcName == "" || *vxlanID == "" || *localIP == "" || *gatewayIP == "" || *cidr == "" {
fmt.Println("create requires: -name -vpc -vxlan-id -local-ip -gateway-ip -cidr")
os.Exit(1)
}
kv.AddInDB(DB, "subnet/"+*name+"/state", "creating")
kv.AddInDB(DB, "subnet/"+*name+"/vpc", *vpcName)
kv.AddInDB(DB, "subnet/"+*name+"/vxlan_id", *vxlanID)
kv.AddInDB(DB, "subnet/"+*name+"/local_ip", *localIP)
kv.AddInDB(DB, "subnet/"+*name+"/gateway_ip", *gatewayIP)
kv.AddInDB(DB, "subnet/"+*name+"/cidr", *cidr)
if err := subnet.CreateSubnet(DB, *name); err != nil {
fmt.Println(err)
os.Exit(1)
}
case "delete":
if *name == "" {
fmt.Println("delete requires: -name")
os.Exit(1)
}
kv.AddInDB(DB, "subnet/"+*name+"/state", "deleting")
if err := subnet.DeleteSubnet(DB, *name); err != nil {
fmt.Println(err)
os.Exit(1)
}
if state, err := kv.GetFromDB(DB, "subnet/"+*name+"/state"); err != nil {
fmt.Println(err)
os.Exit(1)
} else if state == "deleted" {
kv.DeleteInDB(DB, "subnet/"+*name)
}
case "check":
if *name == "" {
fmt.Println("check requires: -name")
os.Exit(1)
}
if state, err := kv.GetFromDB(DB, "subnet/"+*name+"/state"); err != nil {
os.Exit(1)
} else if state != "created" {
os.Exit(1)
}
default:
fmt.Printf("Available commands:\n - create\n - delete\n - check\n")
os.Exit(1)
}
os.Exit(0)
}

65
cmd/vpc/main.go
View file

@ -1,65 +0,0 @@
package main
import (
"flag"
"fmt"
"os"
configuration "git.g3e.fr/syonad/two/internal/config/agent"
"git.g3e.fr/syonad/two/internal/vpc"
"git.g3e.fr/syonad/two/pkg/db/kv"
"github.com/dgraph-io/badger/v4"
)
var (
netns = flag.String("netns", "", "Network namespace à faire")
name = flag.String("name", "", "interface name")
action = flag.String("action", "", "Action a faire")
conf_file = flag.String("conf", "/etc/two/agent.yml", "configuration file")
)
var DB *badger.DB
func main() {
flag.Parse()
conf, err := configuration.LoadConfig(*conf_file)
if err != nil {
fmt.Println(err)
return
}
DB = kv.InitDB(kv.Config{
Path: conf.Database.Path,
}, false)
defer DB.Close()
switch *action {
case "create":
kv.AddInDB(DB, "vpc/"+*name+"/state", "creating")
if err := vpc.CreateVPC(DB, *name); err != nil {
fmt.Println(err)
}
case "delete":
kv.AddInDB(DB, "vpc/"+*name+"/state", "deleting")
if err := vpc.DeleteVPC(DB, *name); err != nil {
fmt.Println(err)
}
if state, err := kv.GetFromDB(DB, "vpc/"+*name+"/state"); err != nil {
fmt.Println(err)
os.Exit(1)
} else if state == "deleted" {
kv.DeleteInDB(DB, "vpc/"+*name)
}
case "check":
if state, err := kv.GetFromDB(DB, "vpc/"+*name+"/state"); err != nil {
os.Exit(1)
} else if state != "created" {
os.Exit(1)
}
default:
fmt.Printf("Available commande:\n - create\n - delete\n - check\n")
os.Exit(1)
}
os.Exit(0)
}

41
conf/agent/config.exemple.yml
View file

@ -1,2 +1,43 @@
# Path to the Badger key-value database directory
database: database:
path: "/var/lib/two/data/" path: "/var/lib/two/data/"
# REST API server
api:
address: "0.0.0.0"
port: 8080
# Prometheus metrics server
prometheus:
address: "0.0.0.0"
port: 9090
# Worker pool that executes dispatched commands
worker:
# Number of concurrent worker goroutines
count: 4
# Maximum number of commands queued before Dispatch blocks
buffer_size: 100
# Timing for commands that wait on resource state transitions
dispatcher:
# How long (in seconds) to wait before giving up
timeout_seconds: 300
# Interval (in seconds) between each state check
poll_seconds: 2
# Bridge interface used when the requested iface_type is not found in interfaces
default_interface: br-000000
# Map of logical interface types to physical bridge names on this host
interfaces:
vms: br-000000
internet: br-000000
admin: br-000000
# Logging configuration
logger:
# Log level: debug, info, warn, error (default: info)
level: info
# Force debug level regardless of level setting (default: false)
debug: false

11
go.mod
View file

@ -5,6 +5,7 @@ go 1.24.0
toolchain go1.24.11 toolchain go1.24.11
require ( require (
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/coreos/go-systemd/v22 v22.6.0 // indirect github.com/coreos/go-systemd/v22 v22.6.0 // indirect
github.com/dgraph-io/badger/v4 v4.8.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/godbus/dbus/v5 v5.1.0 // indirect
github.com/google/flatbuffers v25.2.10+incompatible // indirect github.com/google/flatbuffers v25.2.10+incompatible // indirect
github.com/klauspost/compress v1.18.0 // 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/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/sagikazarmark/locafero v0.11.0 // indirect
github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect
github.com/spf13/afero v1.15.0 // 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 v1.37.0 // indirect
go.opentelemetry.io/otel/metric v1.37.0 // indirect go.opentelemetry.io/otel/metric v1.37.0 // indirect
go.opentelemetry.io/otel/trace 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 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/sys v0.39.0 // indirect
golang.org/x/text v0.28.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
) )

18
go.sum
View file

@ -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 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= 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= 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/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 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= 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 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4=
github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= 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 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc=
github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik= github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik=
github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw= 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/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 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4=
go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0= 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 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= 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 h1:vBTly1HeNPEn3wtREYfy4GZ/NECgw2Cnl+nK6Nz3uvw=
golang.org/x/net v0.41.0/go.mod h1:B/K4NNqkfmg07DQYrbwvSluqCJOOXwUjeb/5lOisjbA= 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.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.10.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= 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= 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 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY= 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= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

27
internal/api/agent/helpers_test.go Normal file
View file

@ -0,0 +1,27 @@
package agentapi
import (
"io"
"log/slog"
"testing"
configuration "git.g3e.fr/syonad/two/internal/config/agent"
dispatcher "git.g3e.fr/syonad/two/internal/dispatcher/agent"
"git.g3e.fr/syonad/two/pkg/db/kv"
"git.g3e.fr/syonad/two/pkg/worker"
"github.com/dgraph-io/badger/v4"
)
// newTestServer builds a Server backed by an in-memory Badger DB.
// The worker queue is buffered but has no running goroutines: Dispatch enqueues
// without blocking and Execute never runs, so DB state reflects only Prepare writes.
func newTestServer(t *testing.T) (*Server, *badger.DB) {
t.Helper()
db := kv.InitDB(kv.Config{Path: t.TempDir()}, false)
t.Cleanup(func() { db.Close() })
q := worker.New(100)
cfg := &configuration.Config{DefaultInterface: "br-test"}
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
d := dispatcher.New(q, db, cfg, logger)
return New(d, db, logger), db
}

33
internal/api/agent/models.go Normal file
View file

@ -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"`
IfaceType string `json:"iface_type"`
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"`
LocalIface string `json:"local_iface"`
GatewayIP string `json:"gateway_ip"`
CIDR string `json:"cidr"`
}
type ErrorResponse struct {
Error string `json:"error"`
}

65
internal/api/agent/server.go Normal file
View file

@ -0,0 +1,65 @@
package agentapi
import (
"crypto/rand"
"encoding/hex"
"log/slog"
"net/http"
"time"
dispatcher "git.g3e.fr/syonad/two/internal/dispatcher/agent"
"github.com/dgraph-io/badger/v4"
)
type Server struct {
dispatcher *dispatcher.Dispatcher
db *badger.DB
logger *slog.Logger
}
func New(d *dispatcher.Dispatcher, db *badger.DB, logger *slog.Logger) *Server {
return &Server{dispatcher: d, db: db, logger: logger}
}
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)
s.logger.Info("API server listening", "address", address)
if err := http.ListenAndServe(address, s.logMiddleware(mux)); err != nil {
s.logger.Error("API server stopped", "error", err)
}
}
type statusWriter struct {
http.ResponseWriter
status int
}
func (sw *statusWriter) WriteHeader(code int) {
sw.status = code
sw.ResponseWriter.WriteHeader(code)
}
func (s *Server) logMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var b [4]byte
rand.Read(b[:])
reqID := hex.EncodeToString(b[:])
sw := &statusWriter{ResponseWriter: w, status: http.StatusOK}
start := time.Now()
next.ServeHTTP(sw, r)
s.logger.Info("request",
"request_id", reqID,
"method", r.Method,
"path", r.URL.Path,
"status", sw.status,
"duration_ms", time.Since(start).Milliseconds(),
"remote", r.RemoteAddr,
)
})
}

80
internal/api/agent/subnet.go Normal file
View file

@ -0,0 +1,80 @@
package agentapi
import (
"encoding/json"
"net/http"
"strconv"
"strings"
dispatcher "git.g3e.fr/syonad/two/internal/dispatcher/agent"
"git.g3e.fr/syonad/two/pkg/db/kv"
)
func (s *Server) SubnetByNameHandler(w http.ResponseWriter, r *http.Request) {
name := strings.TrimPrefix(r.URL.Path, "/subnets/")
if name == "" {
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:
s.getSubnet(w, r, name)
case http.MethodDelete:
s.deleteSubnet(w, r, name)
default:
http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed)
}
}
func (s *Server) getSubnet(w http.ResponseWriter, _ *http.Request, name string) {
entries, err := kv.ListByPrefix(s.db, "subnet/"+name+"/")
if err != nil || len(entries) == 0 {
w.WriteHeader(http.StatusNotFound)
json.NewEncoder(w).Encode(ErrorResponse{Error: "subnet not found"})
return
}
sub := Subnet{Name: 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.StatusOK)
json.NewEncoder(w).Encode(sub)
}
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, err := kv.GetFromDB(s.db, "subnet/"+name+"/state")
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(ErrorResponse{Error: "failed to read subnet state"})
return
}
w.WriteHeader(http.StatusAccepted)
json.NewEncoder(w).Encode(Subnet{Name: name, State: state})
}

252
internal/api/agent/subnet_test.go Normal file
View file

@ -0,0 +1,252 @@
package agentapi
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"git.g3e.fr/syonad/two/pkg/db/kv"
)
// --- SubnetsHandler ---
func TestListSubnets_Empty(t *testing.T) {
s, _ := newTestServer(t)
w := httptest.NewRecorder()
s.SubnetsHandler(w, httptest.NewRequest(http.MethodGet, "/subnets", nil))
if w.Code != http.StatusOK {
t.Fatalf("attendu 200, obtenu %d", w.Code)
}
var result []Subnet
json.NewDecoder(w.Body).Decode(&result)
if len(result) != 0 {
t.Errorf("attendu liste vide, obtenu %v", result)
}
}
func TestListSubnets_WithData(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "subnet/sn-1/state", "created")
kv.AddInDB(db, "subnet/sn-1/vpc", "vpc-1")
kv.AddInDB(db, "subnet/sn-2/state", "creating")
kv.AddInDB(db, "subnet/sn-2/vpc", "vpc-1")
w := httptest.NewRecorder()
s.SubnetsHandler(w, httptest.NewRequest(http.MethodGet, "/subnets", nil))
if w.Code != http.StatusOK {
t.Fatalf("attendu 200, obtenu %d", w.Code)
}
var result []Subnet
json.NewDecoder(w.Body).Decode(&result)
if len(result) != 2 {
t.Errorf("attendu 2 subnets, obtenu %d", len(result))
}
}
func TestListSubnets_InvalidMethod(t *testing.T) {
s, _ := newTestServer(t)
w := httptest.NewRecorder()
s.SubnetsHandler(w, httptest.NewRequest(http.MethodPut, "/subnets", nil))
if w.Code != http.StatusMethodNotAllowed {
t.Errorf("attendu 405, obtenu %d", w.Code)
}
}
func TestPostSubnet_Created(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "vpc/vpc-1/state", "created")
req := SubnetCreateRequest{
Name: "sn-new",
VPC: "vpc-1",
IfaceType: "vms",
GatewayIP: "10.0.0.1",
CIDR: "10.0.0.0/24",
}
body, _ := json.Marshal(req)
w := httptest.NewRecorder()
s.SubnetsHandler(w, httptest.NewRequest(http.MethodPost, "/subnets", bytes.NewReader(body)))
if w.Code != http.StatusAccepted {
t.Fatalf("attendu 202, obtenu %d: %s", w.Code, w.Body.String())
}
var result Subnet
json.NewDecoder(w.Body).Decode(&result)
if result.Name != "sn-new" {
t.Errorf("name attendu sn-new, obtenu %q", result.Name)
}
if result.State != "creating" {
t.Errorf("state attendu creating, obtenu %q", result.State)
}
}
func TestPostSubnet_MissingFields(t *testing.T) {
s, _ := newTestServer(t)
body, _ := json.Marshal(SubnetCreateRequest{Name: "sn-1"}) // vpc, gateway_ip, cidr manquants
w := httptest.NewRecorder()
s.SubnetsHandler(w, httptest.NewRequest(http.MethodPost, "/subnets", bytes.NewReader(body)))
if w.Code != http.StatusBadRequest {
t.Errorf("attendu 400, obtenu %d", w.Code)
}
}
func TestPostSubnet_IfaceTypeOptional(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "vpc/vpc-1/state", "created")
req := SubnetCreateRequest{
Name: "sn-opt",
VPC: "vpc-1",
GatewayIP: "10.0.0.1",
CIDR: "10.0.0.0/24",
// IfaceType omis — doit utiliser default_interface
}
body, _ := json.Marshal(req)
w := httptest.NewRecorder()
s.SubnetsHandler(w, httptest.NewRequest(http.MethodPost, "/subnets", bytes.NewReader(body)))
if w.Code != http.StatusAccepted {
t.Fatalf("attendu 202, obtenu %d: %s", w.Code, w.Body.String())
}
}
func TestPostSubnet_VPCNotFound(t *testing.T) {
s, _ := newTestServer(t)
req := SubnetCreateRequest{
Name: "sn-1",
VPC: "vpc-inexistant",
IfaceType: "vms",
GatewayIP: "10.0.0.1",
CIDR: "10.0.0.0/24",
}
body, _ := json.Marshal(req)
w := httptest.NewRecorder()
s.SubnetsHandler(w, httptest.NewRequest(http.MethodPost, "/subnets", bytes.NewReader(body)))
if w.Code != http.StatusUnprocessableEntity {
t.Errorf("attendu 422, obtenu %d", w.Code)
}
}
func TestPostSubnet_Duplicate(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "vpc/vpc-1/state", "created")
kv.AddInDB(db, "subnet/sn-exist/state", "created")
req := SubnetCreateRequest{
Name: "sn-exist",
VPC: "vpc-1",
IfaceType: "vms",
GatewayIP: "10.0.0.1",
CIDR: "10.0.0.0/24",
}
body, _ := json.Marshal(req)
w := httptest.NewRecorder()
s.SubnetsHandler(w, httptest.NewRequest(http.MethodPost, "/subnets", bytes.NewReader(body)))
if w.Code != http.StatusConflict {
t.Errorf("attendu 409, obtenu %d", w.Code)
}
}
func TestPostSubnet_VPCDeleting(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "vpc/vpc-dying/state", "deleting")
req := SubnetCreateRequest{
Name: "sn-1",
VPC: "vpc-dying",
IfaceType: "vms",
GatewayIP: "10.0.0.1",
CIDR: "10.0.0.0/24",
}
body, _ := json.Marshal(req)
w := httptest.NewRecorder()
s.SubnetsHandler(w, httptest.NewRequest(http.MethodPost, "/subnets", bytes.NewReader(body)))
if w.Code != http.StatusUnprocessableEntity {
t.Errorf("attendu 422, obtenu %d", w.Code)
}
}
func TestPostSubnet_InvalidBody(t *testing.T) {
s, _ := newTestServer(t)
w := httptest.NewRecorder()
s.SubnetsHandler(w, httptest.NewRequest(http.MethodPost, "/subnets", bytes.NewReader([]byte("not json"))))
if w.Code != http.StatusBadRequest {
t.Errorf("attendu 400, obtenu %d", w.Code)
}
}
// --- SubnetByNameHandler ---
func TestGetSubnet_Found(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "subnet/sn-1/state", "created")
kv.AddInDB(db, "subnet/sn-1/vpc", "vpc-1")
kv.AddInDB(db, "subnet/sn-1/cidr", "10.0.0.0/24")
kv.AddInDB(db, "subnet/sn-1/gateway_ip", "10.0.0.1")
req := httptest.NewRequest(http.MethodGet, "/subnets/sn-1", nil)
w := httptest.NewRecorder()
s.SubnetByNameHandler(w, req)
if w.Code != http.StatusOK {
t.Fatalf("attendu 200, obtenu %d", w.Code)
}
var result Subnet
json.NewDecoder(w.Body).Decode(&result)
if result.Name != "sn-1" || result.State != "created" {
t.Errorf("résultat inattendu : %+v", result)
}
if result.VPC != "vpc-1" {
t.Errorf("vpc attendu vpc-1, obtenu %q", result.VPC)
}
}
func TestGetSubnet_NotFound(t *testing.T) {
s, _ := newTestServer(t)
req := httptest.NewRequest(http.MethodGet, "/subnets/inexistant", nil)
w := httptest.NewRecorder()
s.SubnetByNameHandler(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("attendu 404, obtenu %d", w.Code)
}
}
func TestGetSubnet_EmptyName(t *testing.T) {
s, _ := newTestServer(t)
req := httptest.NewRequest(http.MethodGet, "/subnets/", nil)
w := httptest.NewRecorder()
s.SubnetByNameHandler(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("attendu 404, obtenu %d", w.Code)
}
}
func TestDeleteSubnet_Success(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "subnet/sn-del/state", "created")
req := httptest.NewRequest(http.MethodDelete, "/subnets/sn-del", nil)
w := httptest.NewRecorder()
s.SubnetByNameHandler(w, req)
if w.Code != http.StatusAccepted {
t.Fatalf("attendu 202, obtenu %d: %s", w.Code, w.Body.String())
}
var result Subnet
json.NewDecoder(w.Body).Decode(&result)
if result.State != "deleting" {
t.Errorf("state attendu deleting, obtenu %q", result.State)
}
}
func TestDeleteSubnet_NotFound(t *testing.T) {
s, _ := newTestServer(t)
req := httptest.NewRequest(http.MethodDelete, "/subnets/inexistant", nil)
w := httptest.NewRecorder()
s.SubnetByNameHandler(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("attendu 404, obtenu %d", w.Code)
}
}
func TestSubnetByName_InvalidMethod(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "subnet/sn-1/state", "created")
req := httptest.NewRequest(http.MethodPut, "/subnets/sn-1", nil)
w := httptest.NewRecorder()
s.SubnetByNameHandler(w, req)
if w.Code != http.StatusMethodNotAllowed {
t.Errorf("attendu 405, obtenu %d", w.Code)
}
}

124
internal/api/agent/subnets.go Normal file
View file

@ -0,0 +1,124 @@
package agentapi
import (
"encoding/json"
"net/http"
"strconv"
"strings"
dispatcher "git.g3e.fr/syonad/two/internal/dispatcher/agent"
"git.g3e.fr/syonad/two/pkg/db/kv"
)
func (s *Server) SubnetsHandler(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
switch r.Method {
case http.MethodGet:
s.listSubnets(w, r)
case http.MethodPost:
s.postSubnet(w, r)
default:
http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed)
}
}
func (s *Server) listSubnets(w http.ResponseWriter, _ *http.Request) {
entries, err := kv.ListByPrefix(s.db, "subnet/")
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(ErrorResponse{Error: "failed to list subnets"})
return
}
subnets := make(map[string]*Subnet)
for key, value := range entries {
parts := strings.Split(key, "/")
if len(parts) != 3 {
continue
}
name := parts[1]
if _, ok := subnets[name]; !ok {
subnets[name] = &Subnet{Name: name}
}
switch parts[2] {
case "state":
subnets[name].State = value
case "vpc":
subnets[name].VPC = value
case "vxlan_id":
subnets[name].VxlanID, _ = strconv.Atoi(value)
case "local_iface":
subnets[name].LocalIface = value
case "gateway_ip":
subnets[name].GatewayIP = value
case "cidr":
subnets[name].CIDR = value
}
}
result := make([]Subnet, 0, len(subnets))
for _, sub := range subnets {
result = append(result, *sub)
}
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(result)
}
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.GatewayIP == "" || req.CIDR == "" {
w.WriteHeader(http.StatusBadRequest)
json.NewEncoder(w).Encode(ErrorResponse{Error: "name, vpc, gateway_ip and cidr are required"})
return
}
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 {
if _, dbErr := kv.GetFromDB(s.db, "subnet/"+req.Name+"/state"); dbErr == nil {
w.WriteHeader(http.StatusConflict)
} else {
w.WriteHeader(http.StatusUnprocessableEntity)
}
json.NewEncoder(w).Encode(ErrorResponse{Error: err.Error()})
return
}
s.dispatcher.Dispatch(cmd)
entries, err := kv.ListByPrefix(s.db, "subnet/"+req.Name+"/")
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(ErrorResponse{Error: "failed to read subnet state"})
return
}
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(sub)
}

62
internal/api/agent/vpc.go Normal file
View file

@ -0,0 +1,62 @@
package agentapi
import (
"encoding/json"
"net/http"
"strings"
dispatcher "git.g3e.fr/syonad/two/internal/dispatcher/agent"
"git.g3e.fr/syonad/two/pkg/db/kv"
)
func (s *Server) VpcByNameHandler(w http.ResponseWriter, r *http.Request) {
name := strings.TrimPrefix(r.URL.Path, "/vpcs/")
if name == "" {
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:
s.getVpc(w, r, name)
case http.MethodDelete:
s.deleteVpc(w, r, name)
default:
http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed)
}
}
func (s *Server) getVpc(w http.ResponseWriter, _ *http.Request, name string) {
state, err := kv.GetFromDB(s.db, "vpc/"+name+"/state")
if err != nil {
w.WriteHeader(http.StatusNotFound)
json.NewEncoder(w).Encode(ErrorResponse{Error: "vpc not found"})
return
}
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(VPC{Name: name, State: state})
}
func (s *Server) deleteVpc(w http.ResponseWriter, _ *http.Request, name string) {
cmd := dispatcher.DeleteVPCCommand{Name: name}
if err := s.dispatcher.Prepare(cmd); err != nil {
if _, dbErr := kv.GetFromDB(s.db, "vpc/"+name+"/state"); dbErr != nil {
w.WriteHeader(http.StatusNotFound)
} else {
w.WriteHeader(http.StatusConflict)
}
json.NewEncoder(w).Encode(ErrorResponse{Error: err.Error()})
return
}
s.dispatcher.Dispatch(cmd)
state, err := kv.GetFromDB(s.db, "vpc/"+name+"/state")
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(ErrorResponse{Error: "failed to read vpc state"})
return
}
w.WriteHeader(http.StatusAccepted)
json.NewEncoder(w).Encode(VPC{Name: name, State: state})
}

188
internal/api/agent/vpc_test.go Normal file
View file

@ -0,0 +1,188 @@
package agentapi
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"git.g3e.fr/syonad/two/pkg/db/kv"
)
// --- VpcsHandler ---
func TestListVpcs_Empty(t *testing.T) {
s, _ := newTestServer(t)
w := httptest.NewRecorder()
s.VpcsHandler(w, httptest.NewRequest(http.MethodGet, "/vpcs", nil))
if w.Code != http.StatusOK {
t.Fatalf("attendu 200, obtenu %d", w.Code)
}
var result []VPC
json.NewDecoder(w.Body).Decode(&result)
if len(result) != 0 {
t.Errorf("attendu liste vide, obtenu %v", result)
}
}
func TestListVpcs_WithData(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "vpc/v1/state", "created")
kv.AddInDB(db, "vpc/v2/state", "creating")
w := httptest.NewRecorder()
s.VpcsHandler(w, httptest.NewRequest(http.MethodGet, "/vpcs", nil))
if w.Code != http.StatusOK {
t.Fatalf("attendu 200, obtenu %d", w.Code)
}
var result []VPC
json.NewDecoder(w.Body).Decode(&result)
if len(result) != 2 {
t.Errorf("attendu 2 VPCs, obtenu %d", len(result))
}
}
func TestListVpcs_InvalidMethod(t *testing.T) {
s, _ := newTestServer(t)
w := httptest.NewRecorder()
s.VpcsHandler(w, httptest.NewRequest(http.MethodPut, "/vpcs", nil))
if w.Code != http.StatusMethodNotAllowed {
t.Errorf("attendu 405, obtenu %d", w.Code)
}
}
func TestPostVpc_Created(t *testing.T) {
s, _ := newTestServer(t)
body, _ := json.Marshal(VPCCreateRequest{Name: "vpc-new"})
w := httptest.NewRecorder()
s.VpcsHandler(w, httptest.NewRequest(http.MethodPost, "/vpcs", bytes.NewReader(body)))
if w.Code != http.StatusAccepted {
t.Fatalf("attendu 202, obtenu %d: %s", w.Code, w.Body.String())
}
var result VPC
json.NewDecoder(w.Body).Decode(&result)
if result.Name != "vpc-new" {
t.Errorf("name attendu vpc-new, obtenu %q", result.Name)
}
if result.State != "creating" {
t.Errorf("state attendu creating, obtenu %q", result.State)
}
}
func TestPostVpc_MissingName(t *testing.T) {
s, _ := newTestServer(t)
body, _ := json.Marshal(VPCCreateRequest{})
w := httptest.NewRecorder()
s.VpcsHandler(w, httptest.NewRequest(http.MethodPost, "/vpcs", bytes.NewReader(body)))
if w.Code != http.StatusBadRequest {
t.Errorf("attendu 400, obtenu %d", w.Code)
}
}
func TestPostVpc_Duplicate(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "vpc/vpc-exist/state", "created")
body, _ := json.Marshal(VPCCreateRequest{Name: "vpc-exist"})
w := httptest.NewRecorder()
s.VpcsHandler(w, httptest.NewRequest(http.MethodPost, "/vpcs", bytes.NewReader(body)))
if w.Code != http.StatusConflict {
t.Errorf("attendu 409, obtenu %d", w.Code)
}
}
func TestPostVpc_InvalidBody(t *testing.T) {
s, _ := newTestServer(t)
w := httptest.NewRecorder()
s.VpcsHandler(w, httptest.NewRequest(http.MethodPost, "/vpcs", bytes.NewReader([]byte("not json"))))
if w.Code != http.StatusBadRequest {
t.Errorf("attendu 400, obtenu %d", w.Code)
}
}
// --- VpcByNameHandler ---
func TestGetVpc_Found(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "vpc/vpc-1/state", "created")
req := httptest.NewRequest(http.MethodGet, "/vpcs/vpc-1", nil)
w := httptest.NewRecorder()
s.VpcByNameHandler(w, req)
if w.Code != http.StatusOK {
t.Fatalf("attendu 200, obtenu %d", w.Code)
}
var result VPC
json.NewDecoder(w.Body).Decode(&result)
if result.Name != "vpc-1" || result.State != "created" {
t.Errorf("résultat inattendu : %+v", result)
}
}
func TestGetVpc_NotFound(t *testing.T) {
s, _ := newTestServer(t)
req := httptest.NewRequest(http.MethodGet, "/vpcs/inexistant", nil)
w := httptest.NewRecorder()
s.VpcByNameHandler(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("attendu 404, obtenu %d", w.Code)
}
}
func TestGetVpc_EmptyName(t *testing.T) {
s, _ := newTestServer(t)
req := httptest.NewRequest(http.MethodGet, "/vpcs/", nil)
w := httptest.NewRecorder()
s.VpcByNameHandler(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("attendu 404, obtenu %d", w.Code)
}
}
func TestDeleteVpc_Success(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "vpc/vpc-del/state", "created")
req := httptest.NewRequest(http.MethodDelete, "/vpcs/vpc-del", nil)
w := httptest.NewRecorder()
s.VpcByNameHandler(w, req)
if w.Code != http.StatusAccepted {
t.Fatalf("attendu 202, obtenu %d: %s", w.Code, w.Body.String())
}
var result VPC
json.NewDecoder(w.Body).Decode(&result)
if result.State != "deleting" {
t.Errorf("state attendu deleting, obtenu %q", result.State)
}
}
func TestDeleteVpc_NotFound(t *testing.T) {
s, _ := newTestServer(t)
req := httptest.NewRequest(http.MethodDelete, "/vpcs/inexistant", nil)
w := httptest.NewRecorder()
s.VpcByNameHandler(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("attendu 404, obtenu %d", w.Code)
}
}
func TestDeleteVpc_BlockedByActiveSubnet(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "vpc/vpc-busy/state", "created")
kv.AddInDB(db, "subnet/sn-1/state", "created")
kv.AddInDB(db, "subnet/sn-1/vpc", "vpc-busy")
req := httptest.NewRequest(http.MethodDelete, "/vpcs/vpc-busy", nil)
w := httptest.NewRecorder()
s.VpcByNameHandler(w, req)
if w.Code != http.StatusConflict {
t.Errorf("attendu 409, obtenu %d: %s", w.Code, w.Body.String())
}
}
func TestVpcByName_InvalidMethod(t *testing.T) {
s, db := newTestServer(t)
kv.AddInDB(db, "vpc/vpc-1/state", "created")
req := httptest.NewRequest(http.MethodPut, "/vpcs/vpc-1", nil)
w := httptest.NewRecorder()
s.VpcByNameHandler(w, req)
if w.Code != http.StatusMethodNotAllowed {
t.Errorf("attendu 405, obtenu %d", w.Code)
}
}

80
internal/api/agent/vpcs.go Normal file
View file

@ -0,0 +1,80 @@
package agentapi
import (
"encoding/json"
"net/http"
"strings"
dispatcher "git.g3e.fr/syonad/two/internal/dispatcher/agent"
"git.g3e.fr/syonad/two/pkg/db/kv"
)
func (s *Server) VpcsHandler(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
switch r.Method {
case http.MethodGet:
s.listVpcs(w, r)
case http.MethodPost:
s.postVpc(w, r)
default:
http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed)
}
}
func (s *Server) listVpcs(w http.ResponseWriter, _ *http.Request) {
entries, err := kv.ListByPrefix(s.db, "vpc/")
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(ErrorResponse{Error: "failed to list vpcs"})
return
}
vpcs := make(map[string]*VPC)
for key, value := range entries {
parts := strings.Split(key, "/")
if len(parts) != 3 {
continue
}
name := parts[1]
if _, ok := vpcs[name]; !ok {
vpcs[name] = &VPC{Name: name}
}
if parts[2] == "state" {
vpcs[name].State = value
}
}
result := make([]VPC, 0, len(vpcs))
for _, v := range vpcs {
result = append(result, *v)
}
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(result)
}
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
}
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, err := kv.GetFromDB(s.db, "vpc/"+req.Name+"/state")
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(ErrorResponse{Error: "failed to read vpc state"})
return
}
w.WriteHeader(http.StatusAccepted)
json.NewEncoder(w).Encode(VPC{Name: req.Name, State: state})
}

33
internal/config/agent/struct.go
View file

@ -8,6 +8,28 @@ type Config struct {
Database struct { Database struct {
Path string `mapstructure:"path"` Path string `mapstructure:"path"`
} `mapstructure:"database"` } `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"`
Dispatcher struct {
TimeoutSeconds int `mapstructure:"timeout_seconds"`
PollSeconds int `mapstructure:"poll_seconds"`
} `mapstructure:"dispatcher"`
Logger struct {
Level string `mapstructure:"level"`
Debug bool `mapstructure:"debug"`
} `mapstructure:"logger"`
DefaultInterface string `mapstructure:"default_interface"`
Interfaces map[string]string `mapstructure:"interfaces"`
} }
func LoadConfig(path string) (*Config, error) { func LoadConfig(path string) (*Config, error) {
@ -16,6 +38,17 @@ func LoadConfig(path string) (*Config, error) {
v.SetConfigType("yaml") v.SetConfigType("yaml")
v.SetDefault("database.path", "/var/lib/two/data/") 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.SetDefault("dispatcher.timeout_seconds", 300)
v.SetDefault("dispatcher.poll_seconds", 2)
v.SetDefault("default_interface", "br-000000")
v.SetDefault("logger.level", "info")
v.SetDefault("logger.debug", false)
v.ReadInConfig() v.ReadInConfig()

50
internal/dispatcher/agent/dispatcher.go Normal file
View file

@ -0,0 +1,50 @@
package dispatcher
import (
"fmt"
"log/slog"
"time"
configuration "git.g3e.fr/syonad/two/internal/config/agent"
"git.g3e.fr/syonad/two/pkg/worker"
"github.com/dgraph-io/badger/v4"
)
type Command interface {
Prepare(db *badger.DB, cfg *configuration.Config) error
Execute(db *badger.DB, cfg *configuration.Config) error
}
type Dispatcher struct {
queue *worker.Queue
db *badger.DB
cfg *configuration.Config
logger *slog.Logger
}
func New(queue *worker.Queue, db *badger.DB, cfg *configuration.Config, logger *slog.Logger) *Dispatcher {
return &Dispatcher{queue: queue, db: db, cfg: cfg, logger: logger}
}
func (d *Dispatcher) Prepare(cmd Command) error {
d.logger.Debug("prepare", "command", fmt.Sprintf("%T", cmd))
return cmd.Prepare(d.db, d.cfg)
}
func (d *Dispatcher) Dispatch(cmd Command) {
cmdType := fmt.Sprintf("%T", cmd)
d.logger.Debug("dispatch", "command", cmdType)
d.queue.Submit(func() {
start := time.Now()
err := cmd.Execute(d.db, d.cfg)
attrs := []any{
"command", cmdType,
"duration_ms", time.Since(start).Milliseconds(),
}
if err != nil {
d.logger.Error("command failed", append(attrs, "error", err)...)
} else {
d.logger.Info("command done", attrs...)
}
})
}

63
internal/dispatcher/agent/dispatcher_test.go Normal file
View file

@ -0,0 +1,63 @@
package dispatcher
import (
"errors"
"sync"
"testing"
configuration "git.g3e.fr/syonad/two/internal/config/agent"
"github.com/dgraph-io/badger/v4"
)
func TestDispatcher_Prepare_Success(t *testing.T) {
d, _ := newTestDispatcher(t)
cmd := mockCmd{
prepareFn: func(*badger.DB, *configuration.Config) error { return nil },
executeFn: func(*badger.DB, *configuration.Config) error { return nil },
}
if err := d.Prepare(cmd); err != nil {
t.Errorf("Prepare devrait retourner nil, obtenu : %v", err)
}
}
func TestDispatcher_Prepare_PropagatesError(t *testing.T) {
d, _ := newTestDispatcher(t)
want := errors.New("prepare failed")
cmd := mockCmd{
prepareFn: func(*badger.DB, *configuration.Config) error { return want },
executeFn: func(*badger.DB, *configuration.Config) error { return nil },
}
if err := d.Prepare(cmd); !errors.Is(err, want) {
t.Errorf("attendu %v, obtenu %v", want, err)
}
}
func TestDispatcher_Dispatch_ExecutesCommand(t *testing.T) {
d, _ := newTestDispatcher(t)
var wg sync.WaitGroup
wg.Add(1)
cmd := mockCmd{
prepareFn: func(*badger.DB, *configuration.Config) error { return nil },
executeFn: func(*badger.DB, *configuration.Config) error {
wg.Done()
return nil
},
}
d.Dispatch(cmd)
wg.Wait()
}
func TestDispatcher_Dispatch_ExecuteErrorLogged(t *testing.T) {
d, _ := newTestDispatcher(t)
var wg sync.WaitGroup
wg.Add(1)
cmd := mockCmd{
prepareFn: func(*badger.DB, *configuration.Config) error { return nil },
executeFn: func(*badger.DB, *configuration.Config) error {
defer wg.Done()
return errors.New("execute failed")
},
}
d.Dispatch(cmd)
wg.Wait() // Execute s'est terminé — l'erreur est loggée, pas propagée
}

38
internal/dispatcher/agent/helpers_test.go Normal file
View file

@ -0,0 +1,38 @@
package dispatcher
import (
"io"
"log/slog"
"testing"
configuration "git.g3e.fr/syonad/two/internal/config/agent"
"git.g3e.fr/syonad/two/pkg/db/kv"
"git.g3e.fr/syonad/two/pkg/worker"
"github.com/dgraph-io/badger/v4"
)
func newTestDispatcher(t *testing.T) (*Dispatcher, *badger.DB) {
t.Helper()
db := kv.InitDB(kv.Config{Path: t.TempDir()}, false)
t.Cleanup(func() { db.Close() })
q := worker.New(100)
q.Start(2)
cfg := &configuration.Config{DefaultInterface: "br-default"}
cfg.Interfaces = map[string]string{"vms": "br-vms"}
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
return New(q, db, cfg, logger), db
}
// mockCmd implémente Command sans aucune dépendance système.
type mockCmd struct {
prepareFn func(*badger.DB, *configuration.Config) error
executeFn func(*badger.DB, *configuration.Config) error
}
func (m mockCmd) Prepare(db *badger.DB, cfg *configuration.Config) error {
return m.prepareFn(db, cfg)
}
func (m mockCmd) Execute(db *badger.DB, cfg *configuration.Config) error {
return m.executeFn(db, cfg)
}

89
internal/dispatcher/agent/subnet_commands.go Normal file
View file

@ -0,0 +1,89 @@
package dispatcher
import (
"fmt"
"strconv"
"time"
configuration "git.g3e.fr/syonad/two/internal/config/agent"
"git.g3e.fr/syonad/two/internal/subnet"
"git.g3e.fr/syonad/two/pkg/db/kv"
"github.com/dgraph-io/badger/v4"
)
type CreateSubnetCommand struct {
Name string
VPC string
VxlanID int
IfaceType string
GatewayIP string
CIDR string
}
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
}
kv.AddInDB(db, "subnet/"+c.Name+"/state", "creating")
kv.AddInDB(db, "subnet/"+c.Name+"/vpc", c.VPC)
kv.AddInDB(db, "subnet/"+c.Name+"/vxlan_id", strconv.Itoa(c.VxlanID))
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, 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)
}
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 {
if err := subnet.DeleteSubnet(db, c.Name); err != nil {
return err
}
state, err := kv.GetFromDB(db, "subnet/"+c.Name+"/state")
if err != nil {
return err
}
if state == "deleted" {
kv.DeleteInDB(db, "subnet/"+c.Name)
}
return nil
}

135
internal/dispatcher/agent/subnet_commands_test.go Normal file
View file

@ -0,0 +1,135 @@
package dispatcher
import (
"testing"
configuration "git.g3e.fr/syonad/two/internal/config/agent"
"git.g3e.fr/syonad/two/pkg/db/kv"
)
func testCfg() *configuration.Config {
cfg := &configuration.Config{DefaultInterface: "br-default"}
cfg.Interfaces = map[string]string{"vms": "br-vms"}
return cfg
}
// --- CreateSubnetCommand.Prepare ---
func TestCreateSubnetCommand_Prepare_Success(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "vpc/vpc-1/state", "created")
cmd := CreateSubnetCommand{
Name: "sn-1", VPC: "vpc-1", VxlanID: 100,
IfaceType: "vms", GatewayIP: "10.0.0.1", CIDR: "10.0.0.0/24",
}
if err := cmd.Prepare(db, testCfg()); err != nil {
t.Fatalf("Prepare a échoué : %v", err)
}
state, _ := kv.GetFromDB(db, "subnet/sn-1/state")
if state != "creating" {
t.Errorf("state attendu creating, obtenu %q", state)
}
vpc, _ := kv.GetFromDB(db, "subnet/sn-1/vpc")
if vpc != "vpc-1" {
t.Errorf("vpc attendu vpc-1, obtenu %q", vpc)
}
}
func TestCreateSubnetCommand_Prepare_UsesIfaceTypeMapping(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "vpc/vpc-1/state", "created")
cmd := CreateSubnetCommand{
Name: "sn-1", VPC: "vpc-1", VxlanID: 100,
IfaceType: "vms", GatewayIP: "10.0.0.1", CIDR: "10.0.0.0/24",
}
cmd.Prepare(db, testCfg())
iface, _ := kv.GetFromDB(db, "subnet/sn-1/local_iface")
if iface != "br-vms" {
t.Errorf("local_iface attendu br-vms, obtenu %q", iface)
}
}
func TestCreateSubnetCommand_Prepare_UsesDefaultIfaceWhenTypeUnknown(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "vpc/vpc-1/state", "created")
cmd := CreateSubnetCommand{
Name: "sn-1", VPC: "vpc-1", VxlanID: 100,
IfaceType: "inconnu", GatewayIP: "10.0.0.1", CIDR: "10.0.0.0/24",
}
cmd.Prepare(db, testCfg())
iface, _ := kv.GetFromDB(db, "subnet/sn-1/local_iface")
if iface != "br-default" {
t.Errorf("local_iface attendu br-default, obtenu %q", iface)
}
}
func TestCreateSubnetCommand_Prepare_Duplicate(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "vpc/vpc-1/state", "created")
kv.AddInDB(db, "subnet/sn-exist/state", "created")
cmd := CreateSubnetCommand{
Name: "sn-exist", VPC: "vpc-1", VxlanID: 100,
IfaceType: "vms", GatewayIP: "10.0.0.1", CIDR: "10.0.0.0/24",
}
if err := cmd.Prepare(db, testCfg()); err == nil {
t.Error("Prepare devrait échouer sur un subnet déjà existant")
}
}
func TestCreateSubnetCommand_Prepare_VPCNotFound(t *testing.T) {
_, db := newTestDispatcher(t)
cmd := CreateSubnetCommand{
Name: "sn-1", VPC: "vpc-inexistant", VxlanID: 100,
IfaceType: "vms", GatewayIP: "10.0.0.1", CIDR: "10.0.0.0/24",
}
if err := cmd.Prepare(db, testCfg()); err == nil {
t.Error("Prepare devrait échouer si le VPC n'existe pas")
}
}
func TestCreateSubnetCommand_Prepare_VPCDeleting(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "vpc/vpc-dying/state", "deleting")
cmd := CreateSubnetCommand{
Name: "sn-1", VPC: "vpc-dying", VxlanID: 100,
IfaceType: "vms", GatewayIP: "10.0.0.1", CIDR: "10.0.0.0/24",
}
if err := cmd.Prepare(db, testCfg()); err == nil {
t.Error("Prepare devrait échouer si le VPC est en cours de suppression")
}
}
func TestCreateSubnetCommand_Prepare_VPCDeleted(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "vpc/vpc-gone/state", "deleted")
cmd := CreateSubnetCommand{
Name: "sn-1", VPC: "vpc-gone", VxlanID: 100,
IfaceType: "vms", GatewayIP: "10.0.0.1", CIDR: "10.0.0.0/24",
}
if err := cmd.Prepare(db, testCfg()); err == nil {
t.Error("Prepare devrait échouer si le VPC est supprimé")
}
}
// --- DeleteSubnetCommand.Prepare ---
func TestDeleteSubnetCommand_Prepare_Success(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "subnet/sn-del/state", "created")
cmd := DeleteSubnetCommand{Name: "sn-del"}
if err := cmd.Prepare(db, nil); err != nil {
t.Fatalf("Prepare a échoué : %v", err)
}
state, _ := kv.GetFromDB(db, "subnet/sn-del/state")
if state != "deleting" {
t.Errorf("state attendu deleting, obtenu %q", state)
}
}
func TestDeleteSubnetCommand_Prepare_NotFound(t *testing.T) {
_, db := newTestDispatcher(t)
cmd := DeleteSubnetCommand{Name: "sn-inexistant"}
if err := cmd.Prepare(db, nil); err == nil {
t.Error("Prepare devrait échouer si le subnet n'existe pas")
}
}

92
internal/dispatcher/agent/vpc_commands.go Normal file
View file

@ -0,0 +1,92 @@
package dispatcher
import (
"fmt"
"strings"
"time"
configuration "git.g3e.fr/syonad/two/internal/config/agent"
"git.g3e.fr/syonad/two/internal/vpc"
"git.g3e.fr/syonad/two/pkg/db/kv"
"github.com/dgraph-io/badger/v4"
)
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 {
return vpc.CreateVPC(db, c.Name)
}
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)
}
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, 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
}
state, err := kv.GetFromDB(db, "vpc/"+c.Name+"/state")
if err != nil {
return err
}
if state == "deleted" {
kv.DeleteInDB(db, "vpc/"+c.Name)
}
return nil
}

89
internal/dispatcher/agent/vpc_commands_test.go Normal file
View file

@ -0,0 +1,89 @@
package dispatcher
import (
"testing"
"git.g3e.fr/syonad/two/pkg/db/kv"
)
// --- CreateVPCCommand.Prepare ---
func TestCreateVPCCommand_Prepare_NewVPC(t *testing.T) {
_, db := newTestDispatcher(t)
cmd := CreateVPCCommand{Name: "vpc-1"}
if err := cmd.Prepare(db, nil); err != nil {
t.Fatalf("Prepare a échoué : %v", err)
}
state, err := kv.GetFromDB(db, "vpc/vpc-1/state")
if err != nil {
t.Fatalf("état non écrit en DB : %v", err)
}
if state != "creating" {
t.Errorf("state attendu creating, obtenu %q", state)
}
}
func TestCreateVPCCommand_Prepare_Duplicate(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "vpc/vpc-exist/state", "created")
cmd := CreateVPCCommand{Name: "vpc-exist"}
if err := cmd.Prepare(db, nil); err == nil {
t.Error("Prepare devrait échouer sur un VPC déjà existant")
}
}
// --- DeleteVPCCommand.Prepare ---
func TestDeleteVPCCommand_Prepare_Success(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "vpc/vpc-del/state", "created")
cmd := DeleteVPCCommand{Name: "vpc-del"}
if err := cmd.Prepare(db, nil); err != nil {
t.Fatalf("Prepare a échoué : %v", err)
}
state, _ := kv.GetFromDB(db, "vpc/vpc-del/state")
if state != "deleting" {
t.Errorf("state attendu deleting, obtenu %q", state)
}
}
func TestDeleteVPCCommand_Prepare_NotFound(t *testing.T) {
_, db := newTestDispatcher(t)
cmd := DeleteVPCCommand{Name: "vpc-inexistant"}
if err := cmd.Prepare(db, nil); err == nil {
t.Error("Prepare devrait échouer si le VPC n'existe pas")
}
}
func TestDeleteVPCCommand_Prepare_BlockedByActiveSubnet(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "vpc/vpc-busy/state", "created")
kv.AddInDB(db, "subnet/sn-1/state", "created")
kv.AddInDB(db, "subnet/sn-1/vpc", "vpc-busy")
cmd := DeleteVPCCommand{Name: "vpc-busy"}
if err := cmd.Prepare(db, nil); err == nil {
t.Error("Prepare devrait échouer si un subnet actif existe")
}
}
func TestDeleteVPCCommand_Prepare_AllowedWhenSubnetDeleted(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "vpc/vpc-ok/state", "created")
kv.AddInDB(db, "subnet/sn-1/state", "deleted")
kv.AddInDB(db, "subnet/sn-1/vpc", "vpc-ok")
cmd := DeleteVPCCommand{Name: "vpc-ok"}
if err := cmd.Prepare(db, nil); err != nil {
t.Fatalf("Prepare devrait réussir si le subnet est deleted : %v", err)
}
}
func TestDeleteVPCCommand_Prepare_AllowedWhenSubnetDeleting(t *testing.T) {
_, db := newTestDispatcher(t)
kv.AddInDB(db, "vpc/vpc-ok/state", "created")
kv.AddInDB(db, "subnet/sn-1/state", "deleting")
kv.AddInDB(db, "subnet/sn-1/vpc", "vpc-ok")
cmd := DeleteVPCCommand{Name: "vpc-ok"}
if err := cmd.Prepare(db, nil); err != nil {
t.Fatalf("Prepare devrait réussir si le subnet est deleting : %v", err)
}
}

58
internal/ebtables/ebtables.go Normal file
View file

@ -0,0 +1,58 @@
package ebtables
import (
"fmt"
"os/exec"
)
func addRule(args ...string) error {
return exec.Command("ebtables", append([]string{"-A"}, args...)...).Run()
}
func deleteRule(args ...string) error {
return exec.Command("ebtables", append([]string{"-D"}, args...)...).Run()
}
func DropARPToGateway(bridge, gatewayIP string) error {
if err := addRule("FORWARD",
"--out-interface", bridge,
"-p", "arp",
"--arp-op", "Request",
"--arp-ip-dst", gatewayIP,
"-j", "DROP"); err != nil {
return fmt.Errorf("ebtables arp rule: %w", err)
}
return nil
}
func DropDHCP(bridge string) error {
if err := addRule("FORWARD",
"--out-interface", bridge,
"-p", "IPv4",
"--ip-protocol", "udp",
"--ip-source-port", "67:68",
"--ip-destination-port", "67:68",
"-j", "DROP"); err != nil {
return fmt.Errorf("ebtables dhcp rule: %w", err)
}
return nil
}
func DeleteARPToGateway(bridge, gatewayIP string) error {
return deleteRule("FORWARD",
"--out-interface", bridge,
"-p", "arp",
"--arp-op", "Request",
"--arp-ip-dst", gatewayIP,
"-j", "DROP")
}
func DeleteDHCP(bridge string) error {
return deleteRule("FORWARD",
"--out-interface", bridge,
"-p", "IPv4",
"--ip-protocol", "udp",
"--ip-source-port", "67:68",
"--ip-destination-port", "67:68",
"-j", "DROP")
}

99
internal/metadata/metadata_test.go
View file

@ -1,6 +1,8 @@
package metadata package metadata
import ( import (
"net/http"
"net/http/httptest"
"strings" "strings"
"testing" "testing"
@ -181,6 +183,103 @@ func TestUnLoadNoCloudInDB_RemovesAllKeys(t *testing.T) {
} }
} }
// --- getIP ---
func TestGetIP_ValidHostPort(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "/", nil)
req.RemoteAddr = "10.0.0.1:4567"
if ip := getIP(req); ip != "10.0.0.1" {
t.Errorf("attendu 10.0.0.1, obtenu %q", ip)
}
}
func TestGetIP_IPv6(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "/", nil)
req.RemoteAddr = "[::1]:8080"
if ip := getIP(req); ip != "::1" {
t.Errorf("attendu ::1, obtenu %q", ip)
}
}
func TestGetIP_NoPort(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "/", nil)
req.RemoteAddr = "10.0.0.1"
if ip := getIP(req); ip != "10.0.0.1" {
t.Errorf("attendu RemoteAddr brut, obtenu %q", ip)
}
}
// --- rootHandler ---
func TestRootHandler_UserData(t *testing.T) {
data = NoCloudData{UserData: "userdata-content"}
w := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/user-data", nil)
rootHandler(w, req)
if w.Code != http.StatusOK {
t.Errorf("attendu 200, obtenu %d", w.Code)
}
if body := w.Body.String(); body != "userdata-content" {
t.Errorf("body inattendu : %q", body)
}
}
func TestRootHandler_MetaData(t *testing.T) {
data = NoCloudData{MetaData: "metadata-content"}
w := httptest.NewRecorder()
rootHandler(w, httptest.NewRequest(http.MethodGet, "/meta-data", nil))
if w.Code != http.StatusOK {
t.Errorf("attendu 200, obtenu %d", w.Code)
}
if body := w.Body.String(); body != "metadata-content" {
t.Errorf("body inattendu : %q", body)
}
}
func TestRootHandler_NetworkConfig(t *testing.T) {
data = NoCloudData{NetworkConfig: "network-content"}
w := httptest.NewRecorder()
rootHandler(w, httptest.NewRequest(http.MethodGet, "/network-config", nil))
if w.Code != http.StatusOK {
t.Errorf("attendu 200, obtenu %d", w.Code)
}
if body := w.Body.String(); body != "network-content" {
t.Errorf("body inattendu : %q", body)
}
}
func TestRootHandler_VendorData(t *testing.T) {
data = NoCloudData{VendorData: "vendor-content"}
w := httptest.NewRecorder()
rootHandler(w, httptest.NewRequest(http.MethodGet, "/vendor-data", nil))
if w.Code != http.StatusOK {
t.Errorf("attendu 200, obtenu %d", w.Code)
}
if body := w.Body.String(); body != "vendor-content" {
t.Errorf("body inattendu : %q", body)
}
}
func TestRootHandler_UnknownPath(t *testing.T) {
data = NoCloudData{}
w := httptest.NewRecorder()
rootHandler(w, httptest.NewRequest(http.MethodGet, "/unknown", nil))
if w.Code != http.StatusNotFound {
t.Errorf("attendu 404, obtenu %d", w.Code)
}
}
func TestRootHandler_ContentType(t *testing.T) {
data = NoCloudData{MetaData: "x"}
w := httptest.NewRecorder()
rootHandler(w, httptest.NewRequest(http.MethodGet, "/meta-data", nil))
if ct := w.Header().Get("Content-Type"); ct != "text/yaml" {
t.Errorf("Content-Type attendu text/yaml, obtenu %q", ct)
}
}
// --- UnLoadNoCloudInDB_DoesNotAffectOtherVMs ---
func TestUnLoadNoCloudInDB_DoesNotAffectOtherVMs(t *testing.T) { func TestUnLoadNoCloudInDB_DoesNotAffectOtherVMs(t *testing.T) {
db := kv.InitDB(kv.Config{Path: t.TempDir()}, false) db := kv.InitDB(kv.Config{Path: t.TempDir()}, false)
t.Cleanup(func() { db.Close() }) t.Cleanup(func() { db.Close() })

11
internal/netif/vxlan.go
View file

@ -1,19 +1,22 @@
package netif package netif
import ( import (
"net"
"github.com/vishvananda/netlink" "github.com/vishvananda/netlink"
) )
func CreateVxlan(name string, vxlanID int, localIP net.IP) error { func CreateVxlan(name string, vxlanID int, localIface string, mtu int) error {
link, err := netlink.LinkByName(localIface)
if err != nil {
return err
}
vxlan := &netlink.Vxlan{ vxlan := &netlink.Vxlan{
LinkAttrs: netlink.LinkAttrs{ LinkAttrs: netlink.LinkAttrs{
Name: name, Name: name,
MTU: mtu,
}, },
VxlanId: vxlanID, VxlanId: vxlanID,
Port: 4789, Port: 4789,
SrcAddr: localIP, VtepDevIndex: link.Attrs().Index,
Learning: false, Learning: false,
} }
return netlink.LinkAdd(vxlan) return netlink.LinkAdd(vxlan)

67
internal/prometheus/agent/collector.go Normal file
View file

@ -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)
}
}

34
internal/subnet/create.go
View file

@ -3,11 +3,11 @@ package subnet
import ( import (
"fmt" "fmt"
"net" "net"
"os/exec"
"strconv" "strconv"
"strings" "strings"
"git.g3e.fr/syonad/two/internal/dhcp" "git.g3e.fr/syonad/two/internal/dhcp"
"git.g3e.fr/syonad/two/internal/ebtables"
"git.g3e.fr/syonad/two/internal/netif" "git.g3e.fr/syonad/two/internal/netif"
"git.g3e.fr/syonad/two/internal/netns" "git.g3e.fr/syonad/two/internal/netns"
"git.g3e.fr/syonad/two/pkg/db/kv" "git.g3e.fr/syonad/two/pkg/db/kv"
@ -40,13 +40,9 @@ func CreateSubnet(db *badger.DB, subnetName string) error {
return fmt.Errorf("parse vxlan_id: %w", err) return fmt.Errorf("parse vxlan_id: %w", err)
} }
localIPStr, err := kv.GetFromDB(db, "subnet/"+subnetName+"/local_ip") localIface, err := kv.GetFromDB(db, "subnet/"+subnetName+"/local_iface")
if err != nil { if err != nil {
return fmt.Errorf("get local_ip: %w", err) return fmt.Errorf("get local_iface: %w", err)
}
localIP := net.ParseIP(localIPStr)
if localIP == nil {
return fmt.Errorf("invalid local_ip: %s", localIPStr)
} }
gatewayIPStr, err := kv.GetFromDB(db, "subnet/"+subnetName+"/gateway_ip") gatewayIPStr, err := kv.GetFromDB(db, "subnet/"+subnetName+"/gateway_ip")
@ -90,7 +86,7 @@ func CreateSubnet(db *badger.DB, subnetName string) error {
} }
// vxlan // vxlan
if err := netif.CreateVxlan(vxlanIface, vxlanID, localIP); err != nil { if err := netif.CreateVxlan(vxlanIface, vxlanID, localIface, 1500); err != nil {
return fmt.Errorf("create vxlan: %w", err) return fmt.Errorf("create vxlan: %w", err)
} }
@ -140,25 +136,11 @@ func CreateSubnet(db *badger.DB, subnetName string) error {
return fmt.Errorf("add route in netns: %w", err) return fmt.Errorf("add route in netns: %w", err)
} }
// ebtables : drop ARP Request vers la gateway sur ce bridge if err := ebtables.DropARPToGateway(bridge, gatewayIP.String()); err != nil {
if err := exec.Command("ebtables", "-A", "FORWARD", return err
"--out-interface", bridge,
"-p", "arp",
"--arp-op", "Request",
"--arp-ip-dst", gatewayIP.String(),
"-j", "DROP").Run(); err != nil {
return fmt.Errorf("ebtables arp rule: %w", err)
} }
if err := ebtables.DropDHCP(bridge); err != nil {
// ebtables : drop trafic DHCP sur ce bridge return err
if err := exec.Command("ebtables", "-A", "FORWARD",
"--out-interface", bridge,
"-p", "IPv4",
"--ip-protocol", "udp",
"--ip-source-port", "67:68",
"--ip-destination-port", "67:68",
"-j", "DROP").Run(); err != nil {
return fmt.Errorf("ebtables dhcp rule: %w", err)
} }
// génération de la config dnsmasq et démarrage du service // génération de la config dnsmasq et démarrage du service

26
internal/subnet/delete.go
View file

@ -3,9 +3,9 @@ package subnet
import ( import (
"fmt" "fmt"
"os" "os"
"os/exec"
"strings" "strings"
"git.g3e.fr/syonad/two/internal/ebtables"
"git.g3e.fr/syonad/two/internal/netif" "git.g3e.fr/syonad/two/internal/netif"
"git.g3e.fr/syonad/two/internal/netns" "git.g3e.fr/syonad/two/internal/netns"
"git.g3e.fr/syonad/two/pkg/db/kv" "git.g3e.fr/syonad/two/pkg/db/kv"
@ -33,6 +33,11 @@ func DeleteSubnet(db *badger.DB, subnetName string) error {
return fmt.Errorf("get vxlan_id: %w", err) return fmt.Errorf("get vxlan_id: %w", err)
} }
gatewayIP, err := kv.GetFromDB(db, "subnet/"+subnetName+"/gateway_ip")
if err != nil {
return fmt.Errorf("get gateway_ip: %w", err)
}
subnetID := strings.SplitN(subnetName, "-", 2)[1] subnetID := strings.SplitN(subnetName, "-", 2)[1]
bridge := "br-" + subnetID bridge := "br-" + subnetID
vxlanIface := "vxlan-" + vxlanIDStr vxlanIface := "vxlan-" + vxlanIDStr
@ -55,19 +60,12 @@ func DeleteSubnet(db *badger.DB, subnetName string) error {
} }
// suppression des règles ebtables // suppression des règles ebtables
exec.Command("ebtables", "-D", "FORWARD", if err := ebtables.DeleteARPToGateway(bridge, gatewayIP); err != nil {
"--out-interface", bridge, return fmt.Errorf("delete ebtables arp rule: %w", err)
"-p", "arp", }
"--arp-op", "Request", if err := ebtables.DeleteDHCP(bridge); err != nil {
"-j", "DROP").Run() return fmt.Errorf("delete ebtables dhcp rule: %w", err)
}
exec.Command("ebtables", "-D", "FORWARD",
"--out-interface", bridge,
"-p", "IPv4",
"--ip-protocol", "udp",
"--ip-source-port", "67:68",
"--ip-destination-port", "67:68",
"-j", "DROP").Run()
// suppression du bridge dans le netns VPC // suppression du bridge dans le netns VPC
if err := netns.Call(vpcName, func() error { if err := netns.Call(vpcName, func() error {

25
internal/vpc/create.go
View file

@ -1,6 +1,8 @@
package vpc package vpc
import ( import (
"strings"
"git.g3e.fr/syonad/two/internal/netif" "git.g3e.fr/syonad/two/internal/netif"
"git.g3e.fr/syonad/two/internal/netns" "git.g3e.fr/syonad/two/internal/netns"
"git.g3e.fr/syonad/two/pkg/db/kv" "git.g3e.fr/syonad/two/pkg/db/kv"
@ -9,49 +11,40 @@ import (
) )
func CreateVPC(db *badger.DB, name string) error { func CreateVPC(db *badger.DB, name string) error {
// missing
// search data in db
// change state in db
// create netns
if state, err := kv.GetFromDB(db, "vpc/"+name+"/state"); err != nil { if state, err := kv.GetFromDB(db, "vpc/"+name+"/state"); err != nil {
return err return err
} else if state == "creating" { } else if state == "creating" {
vpcID := strings.SplitN(name, "-", 2)[1]
if err := netns.Create(name); err != nil { if err := netns.Create(name); err != nil {
return err return err
} }
// create veth public for this netns if err := netif.CreateVethToNetns("vp-"+vpcID+"-e", "vp-"+vpcID+"-i", "/var/run/netns/"+name, 9000); err != nil {
if err := netif.CreateVethToNetns("vp-"+name+"-e", "vp-public-i", "/var/run/netns/"+name, 9000); err != nil {
return err return err
} }
// create public bridge in netns
if err := netns.Call(name, func() error { if err := netns.Call(name, func() error {
return netif.CreateBridge("br-public", 1500) return netif.CreateBridge("br-public", 1500)
}); err != nil { }); err != nil {
return err return err
} }
// set veth to ext public bridge if err := netif.BridgeSetMaster("vp-"+vpcID+"-e", "br-public"); err != nil {
if err := netif.BridgeSetMaster("vp-"+name+"-e", "br-public"); err != nil {
return err return err
} }
// set veth to int public bridge
if err := netns.Call(name, func() error { if err := netns.Call(name, func() error {
return netif.BridgeSetMaster("vp-public-i", "br-public") return netif.BridgeSetMaster("vp-"+vpcID+"-i", "br-public")
}); err != nil { }); err != nil {
return err return err
} }
// set set ext veth up if err := netif.LinkSetUp("vp-" + vpcID + "-e"); err != nil {
if err := netif.LinkSetUp("vp-" + name + "-e"); err != nil {
return err return err
} }
// set set int veth up
if err := netns.Call(name, func() error { if err := netns.Call(name, func() error {
return netif.LinkSetUp("vp-public-i") return netif.LinkSetUp("vp-" + vpcID + "-i")
}); err != nil { }); err != nil {
return err return err
} }

6
internal/vpc/delete.go
View file

@ -1,6 +1,8 @@
package vpc package vpc
import ( import (
"strings"
"git.g3e.fr/syonad/two/internal/netif" "git.g3e.fr/syonad/two/internal/netif"
"git.g3e.fr/syonad/two/internal/netns" "git.g3e.fr/syonad/two/internal/netns"
"git.g3e.fr/syonad/two/pkg/db/kv" "git.g3e.fr/syonad/two/pkg/db/kv"
@ -12,7 +14,9 @@ func DeleteVPC(db *badger.DB, name string) error {
if state, err := kv.GetFromDB(db, "vpc/"+name+"/state"); err != nil { if state, err := kv.GetFromDB(db, "vpc/"+name+"/state"); err != nil {
return err return err
} else if state == "deleting" { } else if state == "deleting" {
if err := netif.DeleteLink("vp-" + name + "-e"); err != nil { vpcID := strings.SplitN(name, "-", 2)[1]
if err := netif.DeleteLink("vp-" + vpcID + "-e"); err != nil {
return err return err
} }

4
pkg/db/kv/deleteInDB.go
View file

@ -1,8 +1,6 @@
package kv package kv
import ( import (
"log"
"github.com/dgraph-io/badger/v4" "github.com/dgraph-io/badger/v4"
) )
@ -35,7 +33,7 @@ func DeleteInDB(db *badger.DB, key string) error {
return nil return nil
}) })
if err != nil { if err != nil {
log.Fatal(err) return err
} }
return deleteKey(db, key) return deleteKey(db, key)

66
pkg/db/kv/kv_test.go
View file

@ -151,3 +151,69 @@ func TestDeleteInDB_MissingKey(t *testing.T) {
t.Logf("DeleteInDB clé inexistante retourne : %v (non bloquant)", err) t.Logf("DeleteInDB clé inexistante retourne : %v (non bloquant)", err)
} }
} }
// --- ListByPrefix ---
func TestListByPrefix_MatchingKeys(t *testing.T) {
db := newTestDB(t)
AddInDB(db, "subnet/sn1/state", "created")
AddInDB(db, "subnet/sn1/vpc", "vpc-1")
AddInDB(db, "subnet/sn1/cidr", "10.0.0.0/24")
entries, err := ListByPrefix(db, "subnet/sn1/")
if err != nil {
t.Fatalf("ListByPrefix a échoué : %v", err)
}
if len(entries) != 3 {
t.Fatalf("attendu 3 entrées, obtenu %d", len(entries))
}
if entries["subnet/sn1/state"] != "created" {
t.Errorf("valeur inattendue pour state : %q", entries["subnet/sn1/state"])
}
if entries["subnet/sn1/vpc"] != "vpc-1" {
t.Errorf("valeur inattendue pour vpc : %q", entries["subnet/sn1/vpc"])
}
}
func TestListByPrefix_NoMatch(t *testing.T) {
db := newTestDB(t)
AddInDB(db, "vpc/v1/state", "created")
entries, err := ListByPrefix(db, "subnet/")
if err != nil {
t.Fatalf("ListByPrefix a échoué : %v", err)
}
if len(entries) != 0 {
t.Errorf("attendu 0 entrées, obtenu %d", len(entries))
}
}
func TestListByPrefix_IsolatesPrefix(t *testing.T) {
db := newTestDB(t)
AddInDB(db, "subnet/sn1/state", "created")
AddInDB(db, "subnet/sn2/state", "creating")
AddInDB(db, "vpc/v1/state", "created")
entries, err := ListByPrefix(db, "subnet/sn1/")
if err != nil {
t.Fatalf("ListByPrefix a échoué : %v", err)
}
if len(entries) != 1 {
t.Errorf("attendu 1 entrée, obtenu %d : %v", len(entries), entries)
}
if _, ok := entries["subnet/sn1/state"]; !ok {
t.Error("subnet/sn1/state devrait être présent")
}
}
func TestListByPrefix_EmptyDB(t *testing.T) {
db := newTestDB(t)
entries, err := ListByPrefix(db, "subnet/")
if err != nil {
t.Fatalf("ListByPrefix a échoué : %v", err)
}
if len(entries) != 0 {
t.Errorf("attendu 0 entrées sur DB vide, obtenu %d", len(entries))
}
}

33
pkg/db/kv/listByPrefix.go Normal file
View file

@ -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
}

25
pkg/logger/logger.go Normal file
View file

@ -0,0 +1,25 @@
package logger
import (
"log/slog"
"os"
)
var Level = new(slog.LevelVar)
func New(level string, debug bool) *slog.Logger {
switch level {
case "debug":
Level.Set(slog.LevelDebug)
case "warn":
Level.Set(slog.LevelWarn)
case "error":
Level.Set(slog.LevelError)
default:
Level.Set(slog.LevelInfo)
}
if debug {
Level.Set(slog.LevelDebug)
}
return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: Level}))
}

20
pkg/prometheus/server.go Normal file
View file

@ -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))
}

33
pkg/worker/queue.go Normal file
View file

@ -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)
}
}

103
pkg/worker/queue_test.go Normal file
View file

@ -0,0 +1,103 @@
package worker
import (
"sync"
"sync/atomic"
"testing"
"time"
)
func TestNew_ReturnsQueue(t *testing.T) {
q := New(10)
if q == nil {
t.Fatal("New devrait retourner une queue non-nil")
}
}
func TestQueue_SingleTaskExecuted(t *testing.T) {
q := New(1)
q.Start(1)
var done atomic.Bool
var wg sync.WaitGroup
wg.Add(1)
q.Submit(func() {
done.Store(true)
wg.Done()
})
wg.Wait()
if !done.Load() {
t.Error("la tâche n'a pas été exécutée")
}
}
func TestQueue_AllTasksExecuted(t *testing.T) {
const n = 50
q := New(n)
q.Start(1)
var count atomic.Int32
var wg sync.WaitGroup
wg.Add(n)
for range n {
q.Submit(func() {
count.Add(1)
wg.Done()
})
}
wg.Wait()
if count.Load() != n {
t.Errorf("attendu %d exécutions, obtenu %d", n, count.Load())
}
}
func TestQueue_MultipleWorkers(t *testing.T) {
const n = 100
q := New(n)
q.Start(4)
var count atomic.Int32
var wg sync.WaitGroup
wg.Add(n)
for range n {
q.Submit(func() {
count.Add(1)
wg.Done()
})
}
wg.Wait()
if count.Load() != n {
t.Errorf("attendu %d exécutions, obtenu %d", n, count.Load())
}
}
func TestQueue_SubmitBlocksWhenFull(t *testing.T) {
q := New(1)
// Remplit le buffer sans worker
q.Submit(func() {})
submitted := make(chan struct{})
go func() {
q.Submit(func() {}) // doit bloquer jusqu'à ce qu'un worker consomme
close(submitted)
}()
select {
case <-submitted:
t.Error("Submit aurait dû bloquer sur une queue pleine")
case <-time.After(50 * time.Millisecond):
// comportement attendu : goroutine bloquée
}
// Démarre un worker pour débloquer
q.Start(1)
select {
case <-submitted:
// Submit a pu avancer
case <-time.After(time.Second):
t.Error("Submit aurait dû se débloquer après démarrage d'un worker")
}
}

10
systemd/agent.service Normal file
View file

@ -0,0 +1,10 @@
[Unit]
Description=Agent service
After=network.target
[Service]
Type=simple
ExecStart=/opt/two/bin/agent
[Install]
WantedBy=multi-user.target