gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-16 01:22:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(network): move NATS VPC subscribers to network/subscribers

Browse Source 
Phase 4 Slice C (mulga-siv-125.6). Promotes the NATS event handlers out of
services/vpcd into a new spinifex/network/subscribers package. Subscriber
holds the topology/policy/external managers directly — the lazy *Once
adapter shim (topology_manager.go) is deleted along with vpcd's local
event-struct copies.

launchService now constructs topology.LiveManager eagerly with a DNS
callback derived from external pools (preserving the unscoped-pool lookup
old TopologyHandler.lmDNSServer did), builds every manager up front, then
hands them to subscribers.New + Subscribe. reconcile.New receives
topology.Manager directly instead of through the adapter.

Bridge-mode constants and waitForFlowsHV move into vpcd.go alongside the
launcher; they were left behind as topology.go's last responsibilities.

Handler-layer tests (topology_test.go, security_test.go, integration_test.go)
are dropped: substance is covered by network/topology, network/policy,
network/external manager tests. subscribers/subscribers_test.go adds
focused coverage: missing-dep rejection, topic registration smoke, and
bad-JSON envelope per topic.

services/vpcd is now down to the launcher + the ovn/mock compat aliases
(removed in Slice D when the launcher promotes to cmd/vpcd/).
This commit is contained in:
joshsiv-mulga
2026-05-25 14:44:22 +10:00
parent b4d784bc65
commit c3e2d57209
14 changed files with 896 additions and 4974 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spinifex/handlers/ec2/vpc/eni.go
+2 -2
View File
@@ -588,8 +588,8 @@ func generateENIMac(eniId string) string {
}
// portEventPayload is the wire shape for vpc.create-port / vpc.delete-port.
// Mirrors vpcd.PortEvent — duplicated here to avoid a vpcd → handlers import
// cycle.
// Mirrors network/subscribers.PortEvent — duplicated here to avoid a
// subscribers → handlers import cycle.
type portEventPayload struct {
NetworkInterfaceId string `json:"network_interface_id"`
SubnetId string `json:"subnet_id"`
spinifex/network/subscribers/events.go
+95
View File
@@ -0,0 +1,95 @@
package subscribers
// NATS topic names for VPC lifecycle events published by daemon.
const (
TopicVPCCreate = "vpc.create"
TopicVPCDelete = "vpc.delete"
TopicSubnetCreate = "vpc.create-subnet"
TopicSubnetDelete = "vpc.delete-subnet"
TopicCreatePort = "vpc.create-port"
TopicDeletePort = "vpc.delete-port"
TopicUpdatePortSGs = "vpc.update-port-sgs"
TopicIGWAttach = "vpc.igw-attach"
TopicIGWDetach = "vpc.igw-detach"
TopicAddNAT = "vpc.add-nat"
TopicDeleteNAT = "vpc.delete-nat"
TopicAddNATGateway = "vpc.add-nat-gateway"
TopicDeleteNATGateway = "vpc.delete-nat-gateway"
TopicCreateSG = "vpc.create-sg"
TopicDeleteSG = "vpc.delete-sg"
TopicUpdateSG = "vpc.update-sg"
)
// VPCEvent is published on vpc.create / vpc.delete.
type VPCEvent struct {
VpcId string `json:"vpc_id"`
CidrBlock string `json:"cidr_block"`
VNI int64 `json:"vni"`
}
// SubnetEvent is published on vpc.create-subnet / vpc.delete-subnet.
type SubnetEvent struct {
SubnetId string `json:"subnet_id"`
VpcId string `json:"vpc_id"`
CidrBlock string `json:"cidr_block"`
}
// PortEvent is published on vpc.create-port / vpc.delete-port.
//
// SecurityGroupIds carries the SG membership the port should join at create
// time so vpcd can wire OVN port-group membership atomically with the LSP
// create. Empty on delete-port (handleDeletePort discovers current
// memberships from the libovsdb cache).
type PortEvent struct {
NetworkInterfaceId string `json:"network_interface_id"`
SubnetId string `json:"subnet_id"`
VpcId string `json:"vpc_id"`
PrivateIpAddress string `json:"private_ip_address"`
MacAddress string `json:"mac_address"`
SecurityGroupIds []string `json:"security_group_ids,omitempty"`
}
// UpdatePortSGsEvent is published on vpc.update-port-sgs after
// ModifyNetworkInterfaceAttribute changes an ENI's SG membership. The payload
// is declarative — vpcd reads its libovsdb cache to discover current
// memberships and computes the diff against SecurityGroupIds.
type UpdatePortSGsEvent struct {
NetworkInterfaceId string `json:"network_interface_id"`
PrivateIpAddress string `json:"private_ip_address"`
SecurityGroupIds []string `json:"security_group_ids"`
}
// NATEvent is published on vpc.add-nat / vpc.delete-nat for 1:1 public IP NAT.
type NATEvent struct {
VpcId string `json:"vpc_id"`
ExternalIP string `json:"external_ip"`
LogicalIP string `json:"logical_ip"`
PortName string `json:"port_name"` // logical port for distributed NAT
MAC string `json:"mac"` // external MAC for distributed NAT
}
// NATGatewayEvent is published on vpc.add-nat-gateway / vpc.delete-nat-gateway.
type NATGatewayEvent struct {
VpcId string `json:"vpc_id"`
NatGatewayId string `json:"nat_gateway_id"`
PublicIp string `json:"public_ip"`
SubnetCidr string `json:"subnet_cidr"` // private subnet CIDR for SNAT rule
}
// SGRule mirrors the on-wire payload from handlers/ec2/vpc.SGRule. Kept
// local so subscribers do not import the handlers package.
type SGRule struct {
IpProtocol string `json:"ip_protocol"`
FromPort int64 `json:"from_port"`
ToPort int64 `json:"to_port"`
CidrIp string `json:"cidr_ip,omitempty"`
SourceSG string `json:"source_sg,omitempty"`
}
// SGEvent mirrors handlers/ec2/vpc.SGEvent.
type SGEvent struct {
GroupId string `json:"group_id"`
VpcId string `json:"vpc_id"`
IngressRules []SGRule `json:"ingress_rules,omitempty"`
EgressRules []SGRule `json:"egress_rules,omitempty"`
}
spinifex/network/subscribers/respond.go
+34
View File
@@ -0,0 +1,34 @@
package subscribers
import (
"encoding/json"
"log/slog"
"github.com/nats-io/nats.go"
)
type respondResponse struct {
Success bool `json:"success"`
Error string `json:"error,omitempty"`
}
// respond sends a JSON success/error envelope to a NATS request. Fire-and-
// forget when no reply is set on msg.
func respond(msg *nats.Msg, err error) {
if msg.Reply == "" {
return
}
resp := respondResponse{Success: true}
if err != nil {
resp.Success = false
resp.Error = err.Error()
}
data, marshalErr := json.Marshal(resp)
if marshalErr != nil {
slog.Error("subscribers: failed to marshal NATS response", "err", marshalErr)
data = []byte(`{"success":false,"error":"internal marshal failure"}`)
}
if err := msg.Respond(data); err != nil {
slog.Error("subscribers: failed to respond to NATS request", "err", err)
}
}
spinifex/network/subscribers/security.go
+104
View File
@@ -0,0 +1,104 @@
package subscribers
import (
"context"
"encoding/json"
"log/slog"
"github.com/mulgadc/spinifex/spinifex/network/policy"
"github.com/nats-io/nats.go"
)
func (e SGEvent) toSpec() policy.SGSpec {
return policy.SGSpec{
GroupID: e.GroupId,
VPCID: e.VpcId,
IngressRules: toPolicyRules(e.IngressRules),
EgressRules: toPolicyRules(e.EgressRules),
}
}
func toPolicyRules(in []SGRule) []policy.Rule {
out := make([]policy.Rule, len(in))
for i, r := range in {
out[i] = policy.Rule{
IPProtocol: r.IpProtocol,
FromPort: r.FromPort,
ToPort: r.ToPort,
CIDR: r.CidrIp,
SourceSG: r.SourceSG,
}
}
return out
}
// handleCreateSG ensures the OVN port group exists then applies the ACL set
// for a new security group. Two-step (topology + policy) because port-group
// lifecycle is L2 and ACL programming is L3 — see plan §8.1.
func (s *Subscriber) handleCreateSG(msg *nats.Msg) {
var evt SGEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.create-sg event", "err", err)
respond(msg, err)
return
}
ctx := context.Background()
if err := s.topology.EnsureSGPortGroup(ctx, evt.GroupId); err != nil {
slog.Error("subscribers: EnsureSGPortGroup failed", "group_id", evt.GroupId, "err", err)
respond(msg, err)
return
}
if err := s.sg.EnsureSG(ctx, evt.toSpec()); err != nil {
slog.Error("subscribers: EnsureSG failed", "group_id", evt.GroupId, "err", err)
respond(msg, err)
return
}
slog.Info("subscribers: created security group",
"group_id", evt.GroupId,
"vpc_id", evt.VpcId,
"ingress_rules", len(evt.IngressRules),
"egress_rules", len(evt.EgressRules),
)
respond(msg, nil)
}
// handleDeleteSG removes the SG's port group (and its ACLs in one OVN
// transaction). Idempotent on already-absent.
func (s *Subscriber) handleDeleteSG(msg *nats.Msg) {
var evt SGEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.delete-sg event", "err", err)
respond(msg, err)
return
}
if err := s.topology.DeleteSGPortGroup(context.Background(), evt.GroupId); err != nil {
slog.Error("subscribers: DeleteSGPortGroup failed", "group_id", evt.GroupId, "err", err)
respond(msg, err)
return
}
slog.Info("subscribers: deleted security group", "group_id", evt.GroupId, "vpc_id", evt.VpcId)
respond(msg, nil)
}
// handleUpdateSG replaces the SG's ACL set with the rules from the event.
// Port-group lifecycle is unaffected — UpdateSG only touches ACLs.
func (s *Subscriber) handleUpdateSG(msg *nats.Msg) {
var evt SGEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.update-sg event", "err", err)
respond(msg, err)
return
}
if err := s.sg.UpdateSG(context.Background(), evt.toSpec()); err != nil {
slog.Error("subscribers: UpdateSG failed", "group_id", evt.GroupId, "err", err)
respond(msg, err)
return
}
slog.Info("subscribers: updated security group",
"group_id", evt.GroupId,
"vpc_id", evt.VpcId,
"ingress_rules", len(evt.IngressRules),
"egress_rules", len(evt.EgressRules),
)
respond(msg, nil)
}
spinifex/network/subscribers/subscribers.go
+100
View File
@@ -0,0 +1,100 @@
// Package subscribers translates VPC lifecycle NATS events into calls
// against the network/{topology,policy,external} managers. Each handler is
// a thin JSON-decode + manager-call adapter; ACL semantics, OVN object
// lifecycle, and convergence live in the managers themselves. Constructed
// eagerly with every dependency injected — no lazy *Once builders.
package subscribers
import (
"errors"
"fmt"
"log/slog"
"github.com/mulgadc/spinifex/spinifex/network/external"
"github.com/mulgadc/spinifex/spinifex/network/policy"
"github.com/mulgadc/spinifex/spinifex/network/topology"
"github.com/nats-io/nats.go"
)
// QueueGroup is the NATS queue group used for every VPC lifecycle topic.
// Membership is exclusive within the cluster: only one vpcd processes any
// given event, since all vpcds share the same OVN NB DB.
const QueueGroup = "vpcd-workers"
// Subscriber wires VPC lifecycle NATS topics to the network managers.
type Subscriber struct {
topology topology.Manager
sg policy.SecurityGroupManager
nat policy.NATManager
igw external.IGWManager
}
// Config is the construction-time bag. Every field is required.
type Config struct {
Topology topology.Manager
SG policy.SecurityGroupManager
NAT policy.NATManager
IGW external.IGWManager
}
// New constructs a Subscriber, returning an error when any manager is nil.
func New(cfg Config) (*Subscriber, error) {
switch {
case cfg.Topology == nil:
return nil, errors.New("subscribers: Topology manager required")
case cfg.SG == nil:
return nil, errors.New("subscribers: SecurityGroupManager required")
case cfg.NAT == nil:
return nil, errors.New("subscribers: NATManager required")
case cfg.IGW == nil:
return nil, errors.New("subscribers: IGWManager required")
}
return &Subscriber{
topology: cfg.Topology,
sg: cfg.SG,
nat: cfg.NAT,
igw: cfg.IGW,
}, nil
}
// Subscribe registers NATS queue subscriptions for every VPC lifecycle
// topic. Cleanup on partial failure: previously-registered subscriptions
// are unsubscribed before the error is returned.
func (s *Subscriber) Subscribe(nc *nats.Conn) ([]*nats.Subscription, error) {
type sub struct {
topic string
handler nats.MsgHandler
}
subs := []sub{
{TopicVPCCreate, s.handleVPCCreate},
{TopicVPCDelete, s.handleVPCDelete},
{TopicSubnetCreate, s.handleSubnetCreate},
{TopicSubnetDelete, s.handleSubnetDelete},
{TopicCreatePort, s.handleCreatePort},
{TopicDeletePort, s.handleDeletePort},
{TopicUpdatePortSGs, s.handleUpdatePortSGs},
{TopicIGWAttach, s.handleIGWAttach},
{TopicIGWDetach, s.handleIGWDetach},
{TopicAddNAT, s.handleAddNAT},
{TopicDeleteNAT, s.handleDeleteNAT},
{TopicAddNATGateway, s.handleAddNATGateway},
{TopicDeleteNATGateway, s.handleDeleteNATGateway},
{TopicCreateSG, s.handleCreateSG},
{TopicDeleteSG, s.handleDeleteSG},
{TopicUpdateSG, s.handleUpdateSG},
}
var result []*nats.Subscription
for _, item := range subs {
natsSub, err := nc.QueueSubscribe(item.topic, QueueGroup, item.handler)
if err != nil {
for _, r := range result {
_ = r.Unsubscribe()
}
return nil, fmt.Errorf("subscribe %s: %w", item.topic, err)
}
result = append(result, natsSub)
slog.Info("Subscribed to VPC topic", "topic", item.topic)
}
return result, nil
}
spinifex/network/subscribers/subscribers_test.go
+144
View File
@@ -0,0 +1,144 @@
package subscribers
import (
"context"
"encoding/json"
"testing"
"time"
"github.com/mulgadc/spinifex/spinifex/network/external"
"github.com/mulgadc/spinifex/spinifex/network/ovn/mock"
"github.com/mulgadc/spinifex/spinifex/network/policy"
"github.com/mulgadc/spinifex/spinifex/network/topology"
"github.com/mulgadc/spinifex/spinifex/testutil"
)
const requestTimeout = 2 * time.Second
// newTestSubscriber wires every manager against one MockOVNClient so the
// subscriber acts as a real NATS → OVN bridge in tests. The mock OVN client
// satisfies every manager constructor's nil-check at zero cost.
func newTestSubscriber(t *testing.T) (*Subscriber, *mock.Client) {
t.Helper()
m := mock.New()
_ = m.Connect(context.Background())
topo := topology.NewLiveManager(m)
sg := policy.NewSecurityGroupManager(m)
nat, err := policy.NewNATManager(m, policy.NATModeDistributed)
if err != nil {
t.Fatalf("NewNATManager: %v", err)
}
routes := policy.NewRouteManager(m)
igw, err := external.NewIGWManager(external.IGWManagerConfig{
OVN: m,
Routes: routes,
NAT: nat,
Allocator: external.NewStaticRangeAllocator(m),
Chassis: []string{"hv1"},
NATMode: policy.NATModeDistributed,
})
if err != nil {
t.Fatalf("NewIGWManager: %v", err)
}
s, err := New(Config{Topology: topo, SG: sg, NAT: nat, IGW: igw})
if err != nil {
t.Fatalf("New: %v", err)
}
return s, m
}
func TestNew_RejectsMissingDeps(t *testing.T) {
cases := []struct {
name string
cfg Config
}{
{"no topology", Config{SG: policy.NewSecurityGroupManager(mock.New())}},
{"no sg", Config{Topology: topology.NewLiveManager(mock.New())}},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
if _, err := New(c.cfg); err == nil {
t.Fatal("expected error, got nil")
}
})
}
}
func TestSubscribe_RegistersAllTopics(t *testing.T) {
_, nc := testutil.StartTestNATS(t)
sub, _ := newTestSubscriber(t)
subs, err := sub.Subscribe(nc)
if err != nil {
t.Fatalf("Subscribe: %v", err)
}
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
wantTopics := []string{
TopicVPCCreate, TopicVPCDelete,
TopicSubnetCreate, TopicSubnetDelete,
TopicCreatePort, TopicDeletePort, TopicUpdatePortSGs,
TopicIGWAttach, TopicIGWDetach,
TopicAddNAT, TopicDeleteNAT,
TopicAddNATGateway, TopicDeleteNATGateway,
TopicCreateSG, TopicDeleteSG, TopicUpdateSG,
}
if len(subs) != len(wantTopics) {
t.Fatalf("subscription count: got %d, want %d", len(subs), len(wantTopics))
}
got := map[string]bool{}
for _, s := range subs {
got[s.Subject] = true
}
for _, topic := range wantTopics {
if !got[topic] {
t.Errorf("missing subscription for topic %q", topic)
}
}
}
// TestBadJSON_AllRequestReplies verifies every request/reply topic returns a
// structured error envelope (not a NATS timeout) when given malformed JSON.
// Fire-and-forget topics (NAT gateway) just have to not panic.
func TestBadJSON_AllRequestReplies(t *testing.T) {
_, nc := testutil.StartTestNATS(t)
sub, _ := newTestSubscriber(t)
subs, err := sub.Subscribe(nc)
if err != nil {
t.Fatalf("Subscribe: %v", err)
}
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
topics := []string{
TopicVPCCreate, TopicVPCDelete,
TopicSubnetCreate, TopicSubnetDelete,
TopicCreatePort, TopicDeletePort, TopicUpdatePortSGs,
TopicIGWAttach, TopicIGWDetach,
TopicAddNAT, TopicDeleteNAT,
TopicCreateSG, TopicDeleteSG, TopicUpdateSG,
}
for _, topic := range topics {
t.Run(topic, func(t *testing.T) {
resp, err := nc.Request(topic, []byte("not json"), requestTimeout)
if err != nil {
t.Fatalf("request %s: %v", topic, err)
}
var env respondResponse
if err := json.Unmarshal(resp.Data, &env); err != nil {
t.Fatalf("unmarshal envelope: %v", err)
}
if env.Success {
t.Errorf("topic %s: expected success=false on bad JSON, got success=true", topic)
}
})
}
}
spinifex/network/subscribers/topology.go
+311
View File
@@ -0,0 +1,311 @@
package subscribers
import (
"context"
"encoding/json"
"log/slog"
"net"
"net/netip"
"github.com/mulgadc/spinifex/spinifex/network/external"
"github.com/mulgadc/spinifex/spinifex/network/policy"
"github.com/mulgadc/spinifex/spinifex/network/topology"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/nats-io/nats.go"
)
// --- VPC (LogicalRouter) ---
func (s *Subscriber) handleVPCCreate(msg *nats.Msg) {
var evt VPCEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.create event", "err", err)
respond(msg, err)
return
}
spec := topology.VPCSpec{VPCID: evt.VpcId, VNI: evt.VNI}
if evt.CidrBlock != "" {
cidr, err := netip.ParsePrefix(evt.CidrBlock)
if err != nil {
slog.Error("subscribers: invalid CIDR in vpc.create event", "cidr", evt.CidrBlock, "err", err)
respond(msg, err)
return
}
spec.CIDR = cidr
}
if err := s.topology.EnsureVPC(context.Background(), spec); err != nil {
slog.Error("subscribers: EnsureVPC failed", "vpc_id", evt.VpcId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
func (s *Subscriber) handleVPCDelete(msg *nats.Msg) {
var evt VPCEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.delete event", "err", err)
respond(msg, err)
return
}
if err := s.topology.DeleteVPC(context.Background(), evt.VpcId); err != nil {
slog.Error("subscribers: DeleteVPC failed", "vpc_id", evt.VpcId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
// --- Subnet (LogicalSwitch + RouterPort + DHCP) ---
func (s *Subscriber) handleSubnetCreate(msg *nats.Msg) {
var evt SubnetEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.create-subnet event", "err", err)
respond(msg, err)
return
}
cidr, err := netip.ParsePrefix(evt.CidrBlock)
if err != nil {
slog.Error("subscribers: invalid CIDR in vpc.create-subnet event", "cidr", evt.CidrBlock, "err", err)
respond(msg, err)
return
}
ctx := context.Background()
// vpc.create and vpc.create-subnet are independent NATS events with no
// inter-event ordering guarantee. Tenant bootstrap publishes both
// within <1ms, so EnsureSubnet can race the parent VPC's
// CreateLogicalRouter and fail with "logical router not found".
// EnsureVPC is idempotent on the router name (mulga-siv-133).
if err := s.topology.EnsureVPC(ctx, topology.VPCSpec{VPCID: evt.VpcId}); err != nil {
slog.Error("subscribers: EnsureVPC (subnet pre-create) failed", "vpc_id", evt.VpcId, "err", err)
respond(msg, err)
return
}
if err := s.topology.EnsureSubnet(ctx, topology.SubnetSpec{
SubnetID: evt.SubnetId,
VPCID: evt.VpcId,
CIDR: cidr,
}); err != nil {
slog.Error("subscribers: EnsureSubnet failed", "subnet_id", evt.SubnetId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
func (s *Subscriber) handleSubnetDelete(msg *nats.Msg) {
var evt SubnetEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.delete-subnet event", "err", err)
respond(msg, err)
return
}
spec := topology.SubnetSpec{SubnetID: evt.SubnetId, VPCID: evt.VpcId}
if evt.CidrBlock != "" {
if cidr, perr := netip.ParsePrefix(evt.CidrBlock); perr == nil {
spec.CIDR = cidr
}
}
if err := s.topology.DeleteSubnet(context.Background(), spec); err != nil {
slog.Error("subscribers: DeleteSubnet failed", "subnet_id", evt.SubnetId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
// --- Port (LogicalSwitchPort for VM/ENI) ---
func (s *Subscriber) handleCreatePort(msg *nats.Msg) {
var evt PortEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.create-port event", "err", err)
respond(msg, err)
return
}
ip, err := netip.ParseAddr(evt.PrivateIpAddress)
if err != nil {
slog.Error("subscribers: invalid private IP in vpc.create-port event", "ip", evt.PrivateIpAddress, "err", err)
respond(msg, err)
return
}
mac, err := net.ParseMAC(evt.MacAddress)
if err != nil {
slog.Error("subscribers: invalid MAC in vpc.create-port event", "mac", evt.MacAddress, "err", err)
respond(msg, err)
return
}
if err := s.topology.EnsurePort(context.Background(), topology.PortSpec{
PortID: evt.NetworkInterfaceId,
SubnetID: evt.SubnetId,
VPCID: evt.VpcId,
PrivateIP: ip,
MAC: mac,
SGIDs: evt.SecurityGroupIds,
}); err != nil {
slog.Error("subscribers: EnsurePort failed", "eni_id", evt.NetworkInterfaceId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
func (s *Subscriber) handleDeletePort(msg *nats.Msg) {
var evt PortEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.delete-port event", "err", err)
respond(msg, err)
return
}
if err := s.topology.DeletePort(context.Background(), topology.PortSpec{
PortID: evt.NetworkInterfaceId,
SubnetID: evt.SubnetId,
VPCID: evt.VpcId,
}); err != nil {
slog.Error("subscribers: DeletePort failed", "eni_id", evt.NetworkInterfaceId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
// handleUpdatePortSGs reconciles the port group membership for an ENI's LSP
// against the desired SG list in the event. The payload is declarative — the
// manager computes the add/remove diff from current OVN state.
func (s *Subscriber) handleUpdatePortSGs(msg *nats.Msg) {
var evt UpdatePortSGsEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.update-port-sgs event", "err", err)
respond(msg, err)
return
}
if err := s.topology.SetPortSecurityGroups(context.Background(), evt.NetworkInterfaceId, evt.SecurityGroupIds); err != nil {
slog.Error("subscribers: SetPortSecurityGroups failed",
"eni_id", evt.NetworkInterfaceId, "sgs", evt.SecurityGroupIds, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
// --- Internet Gateway (external connectivity + NAT) ---
func (s *Subscriber) handleIGWAttach(msg *nats.Msg) {
var evt types.IGWEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.igw-attach event", "err", err)
respond(msg, err)
return
}
if err := s.igw.AttachIGW(context.Background(), external.IGWSpec{
VPCID: evt.VpcId,
InternetGatewayID: evt.InternetGatewayId,
}); err != nil {
slog.Error("subscribers: AttachIGW failed",
"vpc_id", evt.VpcId, "igw_id", evt.InternetGatewayId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
func (s *Subscriber) handleIGWDetach(msg *nats.Msg) {
var evt types.IGWEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.igw-detach event", "err", err)
respond(msg, err)
return
}
if err := s.igw.DetachIGW(context.Background(), evt.VpcId); err != nil {
slog.Error("subscribers: DetachIGW failed", "vpc_id", evt.VpcId, "err", err)
respond(msg, err)
return
}
slog.Info("subscribers: detached internet gateway from VPC",
"igw_id", evt.InternetGatewayId, "vpc_id", evt.VpcId)
respond(msg, nil)
}
// --- NAT (dnat_and_snat for public IPs) ---
func (s *Subscriber) handleAddNAT(msg *nats.Msg) {
var evt NATEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.add-nat event", "err", err)
respond(msg, err)
return
}
if err := s.nat.AddEIP(context.Background(), policy.EIPSpec{
VPCID: evt.VpcId,
ExternalIP: evt.ExternalIP,
LogicalIP: evt.LogicalIP,
PortName: evt.PortName,
MAC: evt.MAC,
}); err != nil {
slog.Error("subscribers: AddEIP failed",
"vpc_id", evt.VpcId, "external_ip", evt.ExternalIP,
"logical_ip", evt.LogicalIP, "err", err)
respond(msg, err)
return
}
slog.Info("subscribers: added dnat_and_snat rule",
"vpc_id", evt.VpcId, "external_ip", evt.ExternalIP,
"logical_ip", evt.LogicalIP, "port", evt.PortName)
respond(msg, nil)
}
func (s *Subscriber) handleDeleteNAT(msg *nats.Msg) {
var evt NATEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.delete-nat event", "err", err)
respond(msg, err)
return
}
if err := s.nat.DeleteEIP(context.Background(), evt.VpcId, evt.LogicalIP); err != nil {
slog.Error("subscribers: DeleteEIP failed",
"vpc_id", evt.VpcId, "logical_ip", evt.LogicalIP, "err", err)
respond(msg, err)
return
}
slog.Info("subscribers: deleted dnat_and_snat rule",
"vpc_id", evt.VpcId, "logical_ip", evt.LogicalIP)
respond(msg, nil)
}
func (s *Subscriber) handleAddNATGateway(msg *nats.Msg) {
var evt NATGatewayEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.add-nat-gateway event", "err", err)
return
}
if err := s.nat.AddNATGateway(context.Background(), policy.NATGWSpec{
VPCID: evt.VpcId,
NATGatewayID: evt.NatGatewayId,
PublicIP: evt.PublicIp,
SubnetCIDR: evt.SubnetCidr,
}); err != nil {
slog.Error("subscribers: AddNATGateway failed",
"vpc_id", evt.VpcId, "natgw_id", evt.NatGatewayId,
"public_ip", evt.PublicIp, "subnet_cidr", evt.SubnetCidr, "err", err)
return
}
slog.Info("subscribers: NAT Gateway SNAT rule added",
"vpc_id", evt.VpcId, "natgw_id", evt.NatGatewayId,
"public_ip", evt.PublicIp, "subnet_cidr", evt.SubnetCidr)
}
func (s *Subscriber) handleDeleteNATGateway(msg *nats.Msg) {
var evt NATGatewayEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("subscribers: failed to unmarshal vpc.delete-nat-gateway event", "err", err)
return
}
if err := s.nat.DeleteNATGateway(context.Background(), evt.VpcId, evt.SubnetCidr); err != nil {
slog.Warn("subscribers: DeleteNATGateway failed",
"vpc_id", evt.VpcId, "subnet_cidr", evt.SubnetCidr, "err", err)
return
}
slog.Info("subscribers: NAT Gateway SNAT rule removed",
"vpc_id", evt.VpcId, "natgw_id", evt.NatGatewayId, "subnet_cidr", evt.SubnetCidr)
}
spinifex/services/vpcd/integration_test.go
-603
View File
@@ -1,603 +0,0 @@
package vpcd
import (
"context"
"encoding/json"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/testutil"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
)
const testIGWAccountID = "123456789012"
// startTestJetStreamNATS starts an embedded NATS server with JetStream for integration tests.
func startTestJetStreamNATS(t *testing.T) (*server.Server, *nats.Conn) {
t.Helper()
ns, nc, _ := testutil.StartTestJetStream(t)
return ns, nc
}
// TestIntegration_VPCLifecycle tests the full VPC lifecycle:
// VPC create → subnet create → ENI create → IGW attach → IGW detach → cleanup.
// This validates the NATS event flow between the daemon and vpcd topology handler.
func TestIntegration_VPCLifecycle(t *testing.T) {
_, nc := startTestJetStreamNATS(t)
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
// Set up topology handler subscriptions (simulates vpcd)
topo := NewTopologyHandler(mock)
subs, err := topo.Subscribe(nc)
if err != nil {
t.Fatalf("subscribe topology handler: %v", err)
}
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
// === Phase 1: Create VPC ===
vpcEvt := VPCEvent{VpcId: "vpc-integ1", CidrBlock: "10.0.0.0/16", VNI: 1000}
vpcData, _ := json.Marshal(vpcEvt)
resp, err := nc.Request(TopicVPCCreate, vpcData, 5*time.Second)
if err != nil {
t.Fatalf("vpc.create: %v", err)
}
assertSuccess(t, resp, "create VPC")
// Verify OVN: logical router exists
router, err := mock.GetLogicalRouter(ctx, "vpc-vpc-integ1")
if err != nil {
t.Fatalf("expected VPC router: %v", err)
}
if router.ExternalIDs["spinifex:vpc_id"] != "vpc-integ1" {
t.Errorf("router vpc_id = %s, want vpc-integ1", router.ExternalIDs["spinifex:vpc_id"])
}
// === Phase 2: Create Subnet ===
subEvt := SubnetEvent{SubnetId: "subnet-integ1", VpcId: "vpc-integ1", CidrBlock: "10.0.1.0/24"}
subData, _ := json.Marshal(subEvt)
resp, err = nc.Request(TopicSubnetCreate, subData, 5*time.Second)
if err != nil {
t.Fatalf("vpc.create-subnet: %v", err)
}
assertSuccess(t, resp, "create subnet")
// Verify OVN: logical switch, router port, DHCP
ls, err := mock.GetLogicalSwitch(ctx, "subnet-subnet-integ1")
if err != nil {
t.Fatalf("expected logical switch: %v", err)
}
if ls.ExternalIDs["spinifex:subnet_id"] != "subnet-integ1" {
t.Errorf("switch subnet_id = %s, want subnet-integ1", ls.ExternalIDs["spinifex:subnet_id"])
}
dhcp, err := mock.FindDHCPOptionsByCIDR(ctx, "10.0.1.0/24")
if err != nil {
t.Fatalf("expected DHCP options: %v", err)
}
if dhcp.Options["router"] != "10.0.1.1" {
t.Errorf("DHCP router = %s, want 10.0.1.1", dhcp.Options["router"])
}
// Verify router has 1 port (subnet port)
router, _ = mock.GetLogicalRouter(ctx, "vpc-vpc-integ1")
if len(router.Ports) != 1 {
t.Errorf("router ports = %d, want 1", len(router.Ports))
}
// === Phase 3: Create ENI port ===
portEvt := PortEvent{
NetworkInterfaceId: "eni-integ1",
SubnetId: "subnet-integ1",
VpcId: "vpc-integ1",
PrivateIpAddress: "10.0.1.10",
MacAddress: "02:00:00:aa:bb:01",
}
portData, _ := json.Marshal(portEvt)
resp, err = nc.Request(TopicCreatePort, portData, 5*time.Second)
if err != nil {
t.Fatalf("vpc.create-port: %v", err)
}
assertSuccess(t, resp, "create ENI port")
// Verify OVN: logical switch port with correct addresses and DHCP
lsp, err := mock.GetLogicalSwitchPort(ctx, "port-eni-integ1")
if err != nil {
t.Fatalf("expected logical switch port: %v", err)
}
if lsp.Addresses[0] != "02:00:00:aa:bb:01 10.0.1.10" {
t.Errorf("port addresses = %v, want [02:00:00:aa:bb:01 10.0.1.10]", lsp.Addresses)
}
if lsp.DHCPv4Options == nil || *lsp.DHCPv4Options != dhcp.UUID {
t.Error("port should have DHCPv4Options set to subnet DHCP UUID")
}
// Verify switch has 2 ports (router port + ENI port)
ls, _ = mock.GetLogicalSwitch(ctx, "subnet-subnet-integ1")
if len(ls.Ports) != 2 {
t.Errorf("switch ports = %d, want 2 (router + ENI)", len(ls.Ports))
}
// === Phase 4: Create and Attach Internet Gateway ===
// Create VPC KV bucket and register test VPC so IGW ownership checks pass
js, err := nc.JetStream()
if err != nil {
t.Fatalf("JetStream: %v", err)
}
vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
if err != nil {
t.Fatalf("create VPC KV: %v", err)
}
if _, err := vpcKV.Put(utils.AccountKey(testIGWAccountID, "vpc-integ1"), []byte(`{"vpc_id":"vpc-integ1","state":"available"}`)); err != nil {
t.Fatalf("register test VPC: %v", err)
}
igwSvc, err := handlers_ec2_igw.NewIGWServiceImplWithNATS(nil, nc)
if err != nil {
t.Fatalf("create IGW service: %v", err)
}
igwOut, err := igwSvc.CreateInternetGateway(&ec2.CreateInternetGatewayInput{
TagSpecifications: []*ec2.TagSpecification{
{
ResourceType: aws.String("internet-gateway"),
Tags: []*ec2.Tag{
{Key: aws.String("Name"), Value: aws.String("integ-igw")},
},
},
},
}, testIGWAccountID)
if err != nil {
t.Fatalf("CreateInternetGateway: %v", err)
}
igwID := *igwOut.InternetGateway.InternetGatewayId
// Attach IGW to VPC
_, err = igwSvc.AttachInternetGateway(&ec2.AttachInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
VpcId: aws.String("vpc-integ1"),
}, testIGWAccountID)
if err != nil {
t.Fatalf("AttachInternetGateway: %v", err)
}
// Wait briefly for NATS event to be processed
time.Sleep(100 * time.Millisecond)
// Verify OVN: external switch, SNAT, default route
_, err = mock.GetLogicalSwitch(ctx, "ext-vpc-integ1")
if err != nil {
t.Fatalf("expected external switch after IGW attach: %v", err)
}
router, _ = mock.GetLogicalRouter(ctx, "vpc-vpc-integ1")
// Router should have 2 ports (subnet + gateway)
if len(router.Ports) != 2 {
t.Errorf("router ports = %d, want 2 (subnet + gateway)", len(router.Ports))
}
// No blanket SNAT — only per-VM dnat_and_snat rules provide NAT (AWS parity)
if len(router.NAT) != 0 {
t.Errorf("router NAT rules = %d, want 0 (no blanket SNAT)", len(router.NAT))
}
// Default route should exist
if len(router.StaticRoutes) != 1 {
t.Errorf("router static routes = %d, want 1", len(router.StaticRoutes))
}
// Verify IGW state via Describe
descOut, err := igwSvc.DescribeInternetGateways(&ec2.DescribeInternetGatewaysInput{
InternetGatewayIds: []*string{aws.String(igwID)},
}, testIGWAccountID)
if err != nil {
t.Fatalf("DescribeInternetGateways: %v", err)
}
if len(descOut.InternetGateways) != 1 {
t.Fatalf("expected 1 IGW, got %d", len(descOut.InternetGateways))
}
if len(descOut.InternetGateways[0].Attachments) != 1 {
t.Fatal("expected IGW to have 1 attachment")
}
if *descOut.InternetGateways[0].Attachments[0].VpcId != "vpc-integ1" {
t.Errorf("IGW attachment VpcId = %s, want vpc-integ1", *descOut.InternetGateways[0].Attachments[0].VpcId)
}
// === Phase 5: Verify Full Topology Summary ===
switches, _ := mock.ListLogicalSwitches(ctx)
if len(switches) != 2 {
t.Errorf("total switches = %d, want 2 (subnet + external)", len(switches))
}
routers, _ := mock.ListLogicalRouters(ctx)
if len(routers) != 1 {
t.Errorf("total routers = %d, want 1", len(routers))
}
dhcpList, _ := mock.ListDHCPOptions(ctx)
if len(dhcpList) != 1 {
t.Errorf("total DHCP options = %d, want 1", len(dhcpList))
}
// === Phase 6: Detach IGW ===
_, err = igwSvc.DetachInternetGateway(&ec2.DetachInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
VpcId: aws.String("vpc-integ1"),
}, testIGWAccountID)
if err != nil {
t.Fatalf("DetachInternetGateway: %v", err)
}
time.Sleep(100 * time.Millisecond)
// External switch should be cleaned up
_, err = mock.GetLogicalSwitch(ctx, "ext-vpc-integ1")
if err == nil {
t.Error("expected external switch to be deleted after IGW detach")
}
// Router should have 1 port (subnet only), no NAT, no routes
router, _ = mock.GetLogicalRouter(ctx, "vpc-vpc-integ1")
if len(router.Ports) != 1 {
t.Errorf("router ports after detach = %d, want 1", len(router.Ports))
}
if len(router.NAT) != 0 {
t.Errorf("router NAT after detach = %d, want 0", len(router.NAT))
}
if len(router.StaticRoutes) != 0 {
t.Errorf("router routes after detach = %d, want 0", len(router.StaticRoutes))
}
// === Phase 7: Delete IGW ===
_, err = igwSvc.DeleteInternetGateway(&ec2.DeleteInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
}, testIGWAccountID)
if err != nil {
t.Fatalf("DeleteInternetGateway: %v", err)
}
// === Phase 8: Clean up — delete ENI, subnet, VPC ===
resp, _ = nc.Request(TopicDeletePort, portData, 5*time.Second)
assertSuccess(t, resp, "delete ENI port")
resp, _ = nc.Request(TopicSubnetDelete, subData, 5*time.Second)
assertSuccess(t, resp, "delete subnet")
resp, _ = nc.Request(TopicVPCDelete, vpcData, 5*time.Second)
assertSuccess(t, resp, "delete VPC")
// === Phase 9: Verify Complete Cleanup ===
switches, _ = mock.ListLogicalSwitches(ctx)
if len(switches) != 0 {
t.Errorf("expected 0 switches after cleanup, got %d", len(switches))
}
routers, _ = mock.ListLogicalRouters(ctx)
if len(routers) != 0 {
t.Errorf("expected 0 routers after cleanup, got %d", len(routers))
}
dhcpList, _ = mock.ListDHCPOptions(ctx)
if len(dhcpList) != 0 {
t.Errorf("expected 0 DHCP options after cleanup, got %d", len(dhcpList))
}
descOut, err = igwSvc.DescribeInternetGateways(&ec2.DescribeInternetGatewaysInput{}, testIGWAccountID)
if err != nil {
t.Fatalf("DescribeInternetGateways: %v", err)
}
if len(descOut.InternetGateways) != 0 {
t.Errorf("expected 0 IGWs after cleanup, got %d", len(descOut.InternetGateways))
}
}
// TestIntegration_MultiSubnetWithIGW tests a VPC with multiple subnets and IGW.
func TestIntegration_MultiSubnetWithIGW(t *testing.T) {
_, nc := startTestJetStreamNATS(t)
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock)
subs, err := topo.Subscribe(nc)
if err != nil {
t.Fatalf("subscribe: %v", err)
}
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
// Create VPC
vpcEvt := VPCEvent{VpcId: "vpc-multi", CidrBlock: "10.0.0.0/16", VNI: 2000}
vpcData, _ := json.Marshal(vpcEvt)
resp, _ := nc.Request(TopicVPCCreate, vpcData, 5*time.Second)
assertSuccess(t, resp, "create VPC")
// Create 3 subnets in the same VPC
subnets := []SubnetEvent{
{SubnetId: "subnet-a", VpcId: "vpc-multi", CidrBlock: "10.0.1.0/24"},
{SubnetId: "subnet-b", VpcId: "vpc-multi", CidrBlock: "10.0.2.0/24"},
{SubnetId: "subnet-c", VpcId: "vpc-multi", CidrBlock: "10.0.3.0/24"},
}
for _, sub := range subnets {
data, _ := json.Marshal(sub)
resp, _ = nc.Request(TopicSubnetCreate, data, 5*time.Second)
assertSuccess(t, resp, "create subnet "+sub.SubnetId)
}
// Create ENI in each subnet
ports := []PortEvent{
{NetworkInterfaceId: "eni-a1", SubnetId: "subnet-a", VpcId: "vpc-multi", PrivateIpAddress: "10.0.1.10", MacAddress: "02:00:00:01:01:01"},
{NetworkInterfaceId: "eni-b1", SubnetId: "subnet-b", VpcId: "vpc-multi", PrivateIpAddress: "10.0.2.10", MacAddress: "02:00:00:02:02:02"},
{NetworkInterfaceId: "eni-c1", SubnetId: "subnet-c", VpcId: "vpc-multi", PrivateIpAddress: "10.0.3.10", MacAddress: "02:00:00:03:03:03"},
}
for _, port := range ports {
data, _ := json.Marshal(port)
resp, _ = nc.Request(TopicCreatePort, data, 5*time.Second)
assertSuccess(t, resp, "create port "+port.NetworkInterfaceId)
}
// Verify: 1 router, 3 switches, 3 DHCP options
routerList, _ := mock.ListLogicalRouters(ctx)
if len(routerList) != 1 {
t.Errorf("routers = %d, want 1", len(routerList))
}
switchList, _ := mock.ListLogicalSwitches(ctx)
if len(switchList) != 3 {
t.Errorf("switches = %d, want 3", len(switchList))
}
dhcpList, _ := mock.ListDHCPOptions(ctx)
if len(dhcpList) != 3 {
t.Errorf("DHCP options = %d, want 3", len(dhcpList))
}
// Router should have 3 ports (one per subnet)
router, _ := mock.GetLogicalRouter(ctx, "vpc-vpc-multi")
if len(router.Ports) != 3 {
t.Errorf("router ports = %d, want 3", len(router.Ports))
}
// Create VPC KV bucket and register test VPC so IGW ownership checks pass
js, err := nc.JetStream()
if err != nil {
t.Fatalf("JetStream: %v", err)
}
vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
if err != nil {
t.Fatalf("create VPC KV: %v", err)
}
if _, err := vpcKV.Put(utils.AccountKey(testIGWAccountID, "vpc-multi"), []byte(`{"vpc_id":"vpc-multi","state":"available"}`)); err != nil {
t.Fatalf("register test VPC: %v", err)
}
// Attach IGW
igwSvc, _ := handlers_ec2_igw.NewIGWServiceImplWithNATS(nil, nc)
igwOut, _ := igwSvc.CreateInternetGateway(&ec2.CreateInternetGatewayInput{}, testIGWAccountID)
igwID := *igwOut.InternetGateway.InternetGatewayId
_, err = igwSvc.AttachInternetGateway(&ec2.AttachInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
VpcId: aws.String("vpc-multi"),
}, testIGWAccountID)
if err != nil {
t.Fatalf("AttachInternetGateway: %v", err)
}
time.Sleep(100 * time.Millisecond)
// Now: 4 switches (3 subnets + 1 external), router has 4 ports + NAT + route
switchList, _ = mock.ListLogicalSwitches(ctx)
if len(switchList) != 4 {
t.Errorf("switches after IGW = %d, want 4", len(switchList))
}
router, _ = mock.GetLogicalRouter(ctx, "vpc-vpc-multi")
if len(router.Ports) != 4 {
t.Errorf("router ports after IGW = %d, want 4", len(router.Ports))
}
if len(router.NAT) != 0 {
t.Errorf("router NAT = %d, want 0 (no blanket SNAT)", len(router.NAT))
}
if len(router.StaticRoutes) != 1 {
t.Errorf("router routes = %d, want 1", len(router.StaticRoutes))
}
// === Cleanup: detach IGW, delete everything ===
_, _ = igwSvc.DetachInternetGateway(&ec2.DetachInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
VpcId: aws.String("vpc-multi"),
}, testIGWAccountID)
time.Sleep(100 * time.Millisecond)
_, _ = igwSvc.DeleteInternetGateway(&ec2.DeleteInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
}, testIGWAccountID)
// Delete ports
for _, port := range ports {
data, _ := json.Marshal(port)
resp, _ = nc.Request(TopicDeletePort, data, 5*time.Second)
assertSuccess(t, resp, "delete port "+port.NetworkInterfaceId)
}
// Delete subnets
for _, sub := range subnets {
data, _ := json.Marshal(sub)
resp, _ = nc.Request(TopicSubnetDelete, data, 5*time.Second)
assertSuccess(t, resp, "delete subnet "+sub.SubnetId)
}
// Delete VPC
resp, _ = nc.Request(TopicVPCDelete, vpcData, 5*time.Second)
assertSuccess(t, resp, "delete VPC")
// Verify complete cleanup
routerList, _ = mock.ListLogicalRouters(ctx)
if len(routerList) != 0 {
t.Errorf("routers after cleanup = %d, want 0", len(routerList))
}
switchList, _ = mock.ListLogicalSwitches(ctx)
if len(switchList) != 0 {
t.Errorf("switches after cleanup = %d, want 0", len(switchList))
}
}
// TestIntegration_IGWErrorPaths tests IGW error handling across the integration boundary.
func TestIntegration_IGWErrorPaths(t *testing.T) {
_, nc := startTestJetStreamNATS(t)
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
topo := NewTopologyHandler(mock)
subs, err := topo.Subscribe(nc)
if err != nil {
t.Fatalf("subscribe: %v", err)
}
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
// Create VPC KV bucket and register test VPCs so IGW ownership checks pass
js, err := nc.JetStream()
if err != nil {
t.Fatalf("JetStream: %v", err)
}
vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
if err != nil {
t.Fatalf("create VPC KV: %v", err)
}
for _, vpcID := range []string{"vpc-err1", "vpc-err2"} {
if _, err := vpcKV.Put(utils.AccountKey(testIGWAccountID, vpcID), []byte(`{"vpc_id":"`+vpcID+`","state":"available"}`)); err != nil {
t.Fatalf("register test VPC: %v", err)
}
}
igwSvc, err := handlers_ec2_igw.NewIGWServiceImplWithNATS(nil, nc)
if err != nil {
t.Fatalf("create IGW service: %v", err)
}
// Create and attach IGW
igwOut, _ := igwSvc.CreateInternetGateway(&ec2.CreateInternetGatewayInput{}, testIGWAccountID)
igwID := *igwOut.InternetGateway.InternetGatewayId
// Can't delete while attached
_, _ = igwSvc.AttachInternetGateway(&ec2.AttachInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
VpcId: aws.String("vpc-err1"),
}, testIGWAccountID)
_, err = igwSvc.DeleteInternetGateway(&ec2.DeleteInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
}, testIGWAccountID)
if err == nil {
t.Error("expected DependencyViolation error")
}
// Can't attach twice
_, err = igwSvc.AttachInternetGateway(&ec2.AttachInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
VpcId: aws.String("vpc-err2"),
}, testIGWAccountID)
if err == nil {
t.Error("expected ResourceAlreadyAssociated error")
}
// Can't detach from wrong VPC
_, err = igwSvc.DetachInternetGateway(&ec2.DetachInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
VpcId: aws.String("vpc-wrong"),
}, testIGWAccountID)
if err == nil {
t.Error("expected GatewayNotAttached error")
}
// Operations on nonexistent IGW
_, err = igwSvc.AttachInternetGateway(&ec2.AttachInternetGatewayInput{
InternetGatewayId: aws.String("igw-doesnotexist"),
VpcId: aws.String("vpc-err1"),
}, testIGWAccountID)
if err == nil {
t.Error("expected NotFound error for attach")
}
_, err = igwSvc.DetachInternetGateway(&ec2.DetachInternetGatewayInput{
InternetGatewayId: aws.String("igw-doesnotexist"),
VpcId: aws.String("vpc-err1"),
}, testIGWAccountID)
if err == nil {
t.Error("expected NotFound error for detach")
}
// Cleanup
_, _ = igwSvc.DetachInternetGateway(&ec2.DetachInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
VpcId: aws.String("vpc-err1"),
}, testIGWAccountID)
_, _ = igwSvc.DeleteInternetGateway(&ec2.DeleteInternetGatewayInput{
InternetGatewayId: aws.String(igwID),
}, testIGWAccountID)
}
// TestIntegration_SubnetBeforeVPC asserts that vpc.create-subnet does not
// fail when it arrives before the matching vpc.create-vpc (mulga-siv-133).
// The subnet handler must idempotently EnsureVPC so the parent logical
// router is in place by the time CreateLogicalRouterPort runs.
func TestIntegration_SubnetBeforeVPC(t *testing.T) {
_, nc := startTestJetStreamNATS(t)
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock)
subs, err := topo.Subscribe(nc)
if err != nil {
t.Fatalf("subscribe: %v", err)
}
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
// Subnet event arrives first — no prior vpc.create-vpc.
subEvt := SubnetEvent{SubnetId: "subnet-race", VpcId: "vpc-race", CidrBlock: "10.0.1.0/24"}
subData, _ := json.Marshal(subEvt)
resp, err := nc.Request(TopicSubnetCreate, subData, 5*time.Second)
if err != nil {
t.Fatalf("vpc.create-subnet (race): %v", err)
}
assertSuccess(t, resp, "subnet-before-vpc create")
// Router was created defensively.
router, err := mock.GetLogicalRouter(ctx, "vpc-vpc-race")
if err != nil {
t.Fatalf("expected VPC router after subnet pre-create: %v", err)
}
// Subnet topology landed.
if _, err := mock.GetLogicalSwitch(ctx, "subnet-subnet-race"); err != nil {
t.Fatalf("expected subnet switch: %v", err)
}
if len(router.Ports) != 1 {
t.Errorf("router ports = %d, want 1", len(router.Ports))
}
// Now publish the (delayed) vpc.create-vpc with full metadata. EnsureVPC
// must short-circuit on the existing LR — no duplicate, no error.
vpcEvt := VPCEvent{VpcId: "vpc-race", CidrBlock: "10.0.0.0/16", VNI: 4242}
vpcData, _ := json.Marshal(vpcEvt)
resp, err = nc.Request(TopicVPCCreate, vpcData, 5*time.Second)
if err != nil {
t.Fatalf("vpc.create (late): %v", err)
}
assertSuccess(t, resp, "late VPC create")
routerList, _ := mock.ListLogicalRouters(ctx)
if len(routerList) != 1 {
t.Errorf("routers = %d, want 1 (duplicate after race-loser create)", len(routerList))
}
}
spinifex/services/vpcd/security.go
-143
View File
@@ -1,143 +0,0 @@
package vpcd
// Phase 2.6 (mulga-siv-129) thin NATS wrappers for SG lifecycle. Each handler
// deserialises the event, translates to the policy/topology layer's spec
// types, and calls the manager. ACL semantics, OVN port-group lifecycle and
// the full all-or-nothing convergence model live in network/topology and
// network/policy; this file owns NATS plumbing only.
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"github.com/mulgadc/spinifex/spinifex/network/policy"
"github.com/nats-io/nats.go"
)
// sgRule mirrors the on-wire payload from handlers/ec2/vpc.SGRule. Kept local
// so vpcd does not import the handlers package.
type sgRule struct {
IpProtocol string `json:"ip_protocol"`
FromPort int64 `json:"from_port"`
ToPort int64 `json:"to_port"`
CidrIp string `json:"cidr_ip,omitempty"`
SourceSG string `json:"source_sg,omitempty"`
}
// sgEvent mirrors handlers/ec2/vpc.SGEvent.
type sgEvent struct {
GroupId string `json:"group_id"`
VpcId string `json:"vpc_id"`
IngressRules []sgRule `json:"ingress_rules,omitempty"`
EgressRules []sgRule `json:"egress_rules,omitempty"`
}
func (e sgEvent) toSpec() policy.SGSpec {
return policy.SGSpec{
GroupID: e.GroupId,
VPCID: e.VpcId,
IngressRules: toPolicyRules(e.IngressRules),
EgressRules: toPolicyRules(e.EgressRules),
}
}
func toPolicyRules(in []sgRule) []policy.Rule {
out := make([]policy.Rule, len(in))
for i, r := range in {
out[i] = policy.Rule{
IPProtocol: r.IpProtocol,
FromPort: r.FromPort,
ToPort: r.ToPort,
CIDR: r.CidrIp,
SourceSG: r.SourceSG,
}
}
return out
}
// handleCreateSG ensures the OVN port group exists then applies the ACL set
// for a new security group. Two-step (topology + policy) because port-group
// lifecycle is L2 and ACL programming is L3 — see plan §8.1.
func (h *TopologyHandler) handleCreateSG(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt sgEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.create-sg event", "err", err)
respond(msg, err)
return
}
ctx := context.Background()
if err := h.EnsureSGPortGroup(ctx, evt.GroupId); err != nil {
slog.Error("vpcd: EnsureSGPortGroup failed", "group_id", evt.GroupId, "err", err)
respond(msg, err)
return
}
if err := h.securityGroupManager().EnsureSG(ctx, evt.toSpec()); err != nil {
slog.Error("vpcd: EnsureSG failed", "group_id", evt.GroupId, "err", err)
respond(msg, err)
return
}
slog.Info("vpcd: created security group",
"group_id", evt.GroupId,
"vpc_id", evt.VpcId,
"ingress_rules", len(evt.IngressRules),
"egress_rules", len(evt.EgressRules),
)
respond(msg, nil)
}
// handleDeleteSG removes the SG's port group (and its ACLs in one OVN
// transaction). Idempotent on already-absent.
func (h *TopologyHandler) handleDeleteSG(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt sgEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.delete-sg event", "err", err)
respond(msg, err)
return
}
ctx := context.Background()
if err := h.DeleteSGPortGroup(ctx, evt.GroupId); err != nil {
slog.Error("vpcd: DeleteSGPortGroup failed", "group_id", evt.GroupId, "err", err)
respond(msg, err)
return
}
slog.Info("vpcd: deleted security group", "group_id", evt.GroupId, "vpc_id", evt.VpcId)
respond(msg, nil)
}
// handleUpdateSG replaces the SG's ACL set with the rules from the event.
// Port-group lifecycle is unaffected — UpdateSG only touches ACLs.
func (h *TopologyHandler) handleUpdateSG(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt sgEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.update-sg event", "err", err)
respond(msg, err)
return
}
ctx := context.Background()
if err := h.securityGroupManager().UpdateSG(ctx, evt.toSpec()); err != nil {
slog.Error("vpcd: UpdateSG failed", "group_id", evt.GroupId, "err", err)
respond(msg, err)
return
}
slog.Info("vpcd: updated security group",
"group_id", evt.GroupId,
"vpc_id", evt.VpcId,
"ingress_rules", len(evt.IngressRules),
"egress_rules", len(evt.EgressRules),
)
respond(msg, nil)
}
spinifex/services/vpcd/security_test.go
-261
View File
@@ -1,261 +0,0 @@
package vpcd
// Phase 2.6 (mulga-siv-129) tests for the SG NATS wrappers in security.go.
// ACL emission semantics live in network/policy/sg_test.go +
// network/policy/acl_test.go; tests here cover the wrapper layer:
// event-shape errors, OVN-client wiring, and end-to-end NATS → OVN happy
// paths through the new managers.
import (
"context"
"encoding/json"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func sgTopoFixture(t *testing.T) (*TopologyHandler, *MockOVNClient) {
t.Helper()
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
return NewTopologyHandler(mock), mock
}
func TestSG_CreateSG_HappyPath(t *testing.T) {
_, nc := startTestNATS(t)
topo, mock := sgTopoFixture(t)
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
evt := sgEvent{GroupId: "sg-create1", VpcId: "vpc-test"}
data, _ := json.Marshal(evt)
resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
require.NoError(t, err)
assertSuccess(t, resp, "create SG")
mock.Mu.Lock()
_, pgExists := mock.PortGroups["sg_create1"]
aclCount := len(mock.ACLs)
mock.Mu.Unlock()
assert.True(t, pgExists, "port group sg_create1 must exist after create-sg")
assert.Equal(t, 4, aclCount, "infra ACL set (2 deny + 2 DHCP) must be attached")
}
func TestSG_CreateSG_Idempotent(t *testing.T) {
_, nc := startTestNATS(t)
topo, mock := sgTopoFixture(t)
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
data, _ := json.Marshal(sgEvent{GroupId: "sg-idem", VpcId: "vpc-test"})
for range 2 {
resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
require.NoError(t, err)
assertSuccess(t, resp, "create SG retry")
}
mock.Mu.Lock()
pgCount := len(mock.PortGroups)
mock.Mu.Unlock()
assert.Equal(t, 1, pgCount, "duplicate create-sg must not create a second port group")
}
func TestSG_DeleteSG_HappyPath(t *testing.T) {
_, nc := startTestNATS(t)
topo, mock := sgTopoFixture(t)
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
data, _ := json.Marshal(sgEvent{GroupId: "sg-del1", VpcId: "vpc-test"})
resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
require.NoError(t, err)
assertSuccess(t, resp, "create SG before delete")
resp, err = nc.Request(TopicDeleteSG, data, 5_000_000_000)
require.NoError(t, err)
assertSuccess(t, resp, "delete SG")
mock.Mu.Lock()
_, pgExists := mock.PortGroups["sg_del1"]
mock.Mu.Unlock()
assert.False(t, pgExists, "port group must be gone after delete-sg")
}
// Delete of a never-created SG must succeed: the new wrapper delegates to
// topology.Manager.DeleteSGPortGroup which is idempotent on absent port
// groups (the reconciler retries delete events that race with vm-cleanup).
func TestSG_DeleteSG_IdempotentOnAbsent(t *testing.T) {
_, nc := startTestNATS(t)
topo, _ := sgTopoFixture(t)
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
data, _ := json.Marshal(sgEvent{GroupId: "sg-ghost", VpcId: "vpc-test"})
resp, err := nc.Request(TopicDeleteSG, data, 5_000_000_000)
require.NoError(t, err)
assertSuccess(t, resp, "delete absent SG must be a no-op")
}
func TestSG_UpdateSG_ReplacesACLSet(t *testing.T) {
_, nc := startTestNATS(t)
topo, mock := sgTopoFixture(t)
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
createData, _ := json.Marshal(sgEvent{GroupId: "sg-upd", VpcId: "vpc-test"})
resp, err := nc.Request(TopicCreateSG, createData, 5_000_000_000)
require.NoError(t, err)
assertSuccess(t, resp, "create SG before update")
updateData, _ := json.Marshal(sgEvent{
GroupId: "sg-upd",
VpcId: "vpc-test",
IngressRules: []sgRule{
{IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "10.0.0.0/8"},
},
})
resp, err = nc.Request(TopicUpdateSG, updateData, 5_000_000_000)
require.NoError(t, err)
assertSuccess(t, resp, "update SG")
mock.Mu.Lock()
pg := mock.PortGroups["sg_upd"]
aclCount := 0
hasTenantRule := false
if pg != nil {
aclCount = len(pg.ACLs)
for _, uuid := range pg.ACLs {
if acl := mock.ACLs[uuid]; acl != nil && acl.Priority == 1000 {
hasTenantRule = true
}
}
}
mock.Mu.Unlock()
assert.Equal(t, 5, aclCount, "update must leave 4 infra + 1 tenant ACL on the port group")
assert.True(t, hasTenantRule, "tenant priority-1000 ACL must be present after update")
}
func TestSG_HandleCreateSG_OVNNotConnected(t *testing.T) {
_, nc := startTestNATS(t)
topo := NewTopologyHandler(nil)
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
data, _ := json.Marshal(sgEvent{GroupId: "sg-noovn"})
resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000)
require.NoError(t, err)
assertFailure(t, resp, "create SG must fail when ovn client is nil")
}
func TestSG_HandleDeleteSG_OVNNotConnected(t *testing.T) {
_, nc := startTestNATS(t)
topo := NewTopologyHandler(nil)
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
data, _ := json.Marshal(sgEvent{GroupId: "sg-noovn"})
resp, err := nc.Request(TopicDeleteSG, data, 5_000_000_000)
require.NoError(t, err)
assertFailure(t, resp, "delete SG must fail when ovn client is nil")
}
func TestSG_HandleUpdateSG_OVNNotConnected(t *testing.T) {
_, nc := startTestNATS(t)
topo := NewTopologyHandler(nil)
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
data, _ := json.Marshal(sgEvent{GroupId: "sg-noovn"})
resp, err := nc.Request(TopicUpdateSG, data, 5_000_000_000)
require.NoError(t, err)
assertFailure(t, resp, "update SG must fail when ovn client is nil")
}
func TestSG_HandleCreateSG_BadJSON(t *testing.T) {
_, nc := startTestNATS(t)
topo, _ := sgTopoFixture(t)
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
resp, err := nc.Request(TopicCreateSG, []byte("not json"), 5_000_000_000)
require.NoError(t, err)
assertFailure(t, resp, "create SG with malformed payload must fail")
}
func TestSG_HandleDeleteSG_BadJSON(t *testing.T) {
_, nc := startTestNATS(t)
topo, _ := sgTopoFixture(t)
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
resp, err := nc.Request(TopicDeleteSG, []byte("not json"), 5_000_000_000)
require.NoError(t, err)
assertFailure(t, resp, "delete SG with malformed payload must fail")
}
func TestSG_HandleUpdateSG_BadJSON(t *testing.T) {
_, nc := startTestNATS(t)
topo, _ := sgTopoFixture(t)
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
resp, err := nc.Request(TopicUpdateSG, []byte("not json"), 5_000_000_000)
require.NoError(t, err)
assertFailure(t, resp, "update SG with malformed payload must fail")
}
spinifex/services/vpcd/topology.go
-719
View File
@@ -1,719 +0,0 @@
package vpcd
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"net"
"net/netip"
"strings"
"sync"
"time"
"github.com/mulgadc/spinifex/spinifex/network/external"
"github.com/mulgadc/spinifex/spinifex/network/policy"
"github.com/mulgadc/spinifex/spinifex/network/topology"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/nats-io/nats.go"
)
// waitForFlowsHV shells out to `ovn-nbctl --wait=hv sync`, which bumps
// NB_Global.nb_cfg and blocks until every connected chassis has acknowledged
// the new sequence number — i.e. ovn-northd has compiled NB -> SB and
// ovn-controller has installed the resulting flows. Used after IGW attach
// so newly-launched VMs aren't unreachable while their gateway chassis is
// still catching up (mulga-siv-105).
//
// Bounded by ovn-nbctl --timeout=30 (seconds). On overrun we log a WARN
// and return nil — the caller continues. In practice flows converge within
// seconds; a 30s overrun means something OVN-side is wedged enough that
// failing VPC create wouldn't improve things.
//
// Declared as a var so tests can stub it.
var waitForFlowsHV = func() error {
start := time.Now()
cmd := sudoCommand("ovn-nbctl",
"--no-leader-only",
"--timeout=30",
"--wait=hv",
"sync",
)
if out, err := cmd.CombinedOutput(); err != nil {
slog.Warn("vpcd: OVN flows-ready barrier overran; continuing without confirmation",
"elapsed", time.Since(start),
"err", err,
"output", strings.TrimSpace(string(out)),
)
return nil
}
slog.Debug("vpcd: OVN flows-ready barrier complete", "elapsed", time.Since(start))
return nil
}
// NATS topics for VPC lifecycle events published by the daemon.
const (
TopicVPCCreate = "vpc.create"
TopicVPCDelete = "vpc.delete"
TopicSubnetCreate = "vpc.create-subnet"
TopicSubnetDelete = "vpc.delete-subnet"
TopicCreatePort = "vpc.create-port"
TopicDeletePort = "vpc.delete-port"
TopicUpdatePortSGs = "vpc.update-port-sgs"
TopicPortStatus = "vpc.port-status"
TopicIGWAttach = "vpc.igw-attach"
TopicIGWDetach = "vpc.igw-detach"
TopicAddNAT = "vpc.add-nat"
TopicDeleteNAT = "vpc.delete-nat"
TopicAddNATGateway = "vpc.add-nat-gateway"
TopicDeleteNATGateway = "vpc.delete-nat-gateway"
TopicCreateSG = "vpc.create-sg"
TopicDeleteSG = "vpc.delete-sg"
TopicUpdateSG = "vpc.update-sg"
)
// VPCEvent is published on vpc.create after a VPC is persisted.
type VPCEvent struct {
VpcId string `json:"vpc_id"`
CidrBlock string `json:"cidr_block"`
VNI int64 `json:"vni"`
}
// SubnetEvent is published on vpc.create-subnet / vpc.delete-subnet.
type SubnetEvent struct {
SubnetId string `json:"subnet_id"`
VpcId string `json:"vpc_id"`
CidrBlock string `json:"cidr_block"`
}
// PortEvent is published on vpc.create-port / vpc.delete-port.
//
// SecurityGroupIds carries the SG membership the port should join at create
// time so vpcd can wire OVN port-group membership atomically with the LSP
// create. Empty on delete-port (handleDeletePort discovers current
// memberships from the libovsdb cache).
type PortEvent struct {
NetworkInterfaceId string `json:"network_interface_id"`
SubnetId string `json:"subnet_id"`
VpcId string `json:"vpc_id"`
PrivateIpAddress string `json:"private_ip_address"`
MacAddress string `json:"mac_address"`
SecurityGroupIds []string `json:"security_group_ids,omitempty"`
}
// UpdatePortSGsEvent is published on vpc.update-port-sgs after
// ModifyNetworkInterfaceAttribute changes an ENI's SG membership. The payload
// is declarative — vpcd reads its libovsdb cache to discover current
// memberships and computes the diff against SecurityGroupIds.
type UpdatePortSGsEvent struct {
NetworkInterfaceId string `json:"network_interface_id"`
PrivateIpAddress string `json:"private_ip_address"`
SecurityGroupIds []string `json:"security_group_ids"`
}
// NATEvent is published on vpc.add-nat / vpc.delete-nat for 1:1 public IP NAT.
type NATEvent struct {
VpcId string `json:"vpc_id"`
ExternalIP string `json:"external_ip"`
LogicalIP string `json:"logical_ip"`
PortName string `json:"port_name"` // logical port for distributed NAT
MAC string `json:"mac"` // external MAC for distributed NAT
}
// NATGatewayEvent is published on vpc.add-nat-gateway / vpc.delete-nat-gateway.
type NATGatewayEvent struct {
VpcId string `json:"vpc_id"`
NatGatewayId string `json:"nat_gateway_id"`
PublicIp string `json:"public_ip"`
SubnetCidr string `json:"subnet_cidr"` // private subnet CIDR for SNAT rule
}
// Bridge mode constants for external connectivity.
const (
// BridgeModeDirect adds the WAN NIC directly to br-external as an OVS port.
// Enables distributed NAT. Only safe when the WAN NIC is NOT the SSH/
// management NIC.
BridgeModeDirect = "direct"
// BridgeModeVeth uses a veth pair to link a Linux bridge (br-wan) to an
// OVS bridge (br-ext). Requires centralized NAT because the Linux bridge
// intermediary breaks distributed NAT hairpin routing.
BridgeModeVeth = "veth"
// OvnExternalBridge is the OVS bridge that ovn-bridge-mappings targets
// for the "external" localnet. Owned by setup-ovn.sh's ovn-bridge-mappings
// setup and independent of the WAN bridge (which is Linux-side in veth
// mode). Named as a constant so the vpcd sanity check refers to the
// contract, not a hardcoded string (D18).
OvnExternalBridge = "br-ext"
)
// TopologyHandler translates VPC lifecycle NATS events into OVN NB DB operations.
type TopologyHandler struct {
ovn OVNClient
externalMode string
externalPools []ExternalPoolConfig
chassisNames []string // OVN chassis names for gateway HA scheduling
natMode policy.NATMode // distributed (direct uplink) or centralized (veth uplink)
// lm + lmOnce cache the topology.Manager produced from h.ovn so each
// adapter call doesn't re-construct (Phase 2.6, mulga-siv-129).
lmOnce sync.Once
lm topology.Manager
// sgm + sgmOnce cache the policy.SecurityGroupManager backed by h.ovn so
// SG subscribers reuse one manager instance across events.
sgmOnce sync.Once
sgm policy.SecurityGroupManager
// natm + natmOnce cache the policy.NATManager backed by h.ovn. NAT mode is
// the value injected via WithNATMode; the FlowsBarrier closure shells out
// to ovn-nbctl --wait=hv sync via waitForFlowsHV.
natmOnce sync.Once
natm policy.NATManager
natmErr error
// igwm + igwmOnce cache the external.IGWManager. All IGW attach/detach
// paths route through it.
igwmOnce sync.Once
igwm external.IGWManager
igwmErr error
}
// NewTopologyHandler creates a new TopologyHandler with optional external network config.
// NAT mode defaults to Distributed (direct uplink) when WithNATMode is omitted;
// matches the pre-Slice-B behaviour where unset bridgeMode also implied
// distributed via useCentralizedNAT().
func NewTopologyHandler(ovn OVNClient, opts ...TopologyOption) *TopologyHandler {
h := &TopologyHandler{ovn: ovn, natMode: policy.NATModeDistributed}
for _, opt := range opts {
opt(h)
}
return h
}
// TopologyOption configures a TopologyHandler.
type TopologyOption func(*TopologyHandler)
// WithExternalNetwork configures external connectivity for public subnets.
func WithExternalNetwork(mode string, pools []ExternalPoolConfig) TopologyOption {
return func(h *TopologyHandler) {
h.externalMode = mode
h.externalPools = pools
}
}
// WithChassisNames sets the OVN chassis names for gateway HA scheduling.
func WithChassisNames(names []string) TopologyOption {
return func(h *TopologyHandler) {
h.chassisNames = names
}
}
// WithNATMode sets the NAT mode used by lazily-constructed managers. Derived
// upstream from the uplink/bridge mode (direct→distributed, veth→centralized).
func WithNATMode(mode policy.NATMode) TopologyOption {
return func(h *TopologyHandler) {
h.natMode = mode
}
}
// findExternalPool returns the first pool matching the given region/AZ,
// using the fallback order: AZ-scoped → region-scoped → unscoped.
func (h *TopologyHandler) findExternalPool(region, az string) *ExternalPoolConfig {
// 1. AZ-scoped match
for i := range h.externalPools {
p := &h.externalPools[i]
if p.AZ != "" && p.AZ == az && p.Region == region {
return p
}
}
// 2. Region-scoped (no AZ)
for i := range h.externalPools {
p := &h.externalPools[i]
if p.AZ == "" && p.Region != "" && p.Region == region {
return p
}
}
// 3. Unscoped (no region, no AZ — global pool)
for i := range h.externalPools {
p := &h.externalPools[i]
if p.Region == "" && p.AZ == "" {
return p
}
}
return nil
}
// Subscribe registers NATS subscriptions for VPC lifecycle topics.
// Global topology events (VPC/subnet/IGW) use a queue group so exactly one vpcd
// instance handles each event — all instances connect to the same OVN NB DB, so
// any one can process these. Per-node port events use regular subscriptions so
// all instances see them (future: route to the specific node).
func (h *TopologyHandler) Subscribe(nc *nats.Conn) ([]*nats.Subscription, error) {
type sub struct {
topic string
handler nats.MsgHandler
queue bool // true = use queue group (one handler), false = fan-out (all handlers)
}
subs := []sub{
{TopicVPCCreate, h.handleVPCCreate, true},
{TopicVPCDelete, h.handleVPCDelete, true},
{TopicSubnetCreate, h.handleSubnetCreate, true},
{TopicSubnetDelete, h.handleSubnetDelete, true},
{TopicCreatePort, h.handleCreatePort, true},
{TopicDeletePort, h.handleDeletePort, true},
{TopicUpdatePortSGs, h.handleUpdatePortSGs, true},
{TopicIGWAttach, h.handleIGWAttach, true},
{TopicIGWDetach, h.handleIGWDetach, true},
{TopicAddNAT, h.handleAddNAT, true},
{TopicDeleteNAT, h.handleDeleteNAT, true},
{TopicAddNATGateway, h.handleAddNATGateway, true},
{TopicDeleteNATGateway, h.handleDeleteNATGateway, true},
{TopicCreateSG, h.handleCreateSG, true},
{TopicDeleteSG, h.handleDeleteSG, true},
{TopicUpdateSG, h.handleUpdateSG, true},
}
var result []*nats.Subscription
for _, s := range subs {
var natsSub *nats.Subscription
var err error
if s.queue {
natsSub, err = nc.QueueSubscribe(s.topic, "vpcd-workers", s.handler)
} else {
natsSub, err = nc.Subscribe(s.topic, s.handler)
}
if err != nil {
for _, r := range result {
_ = r.Unsubscribe()
}
return nil, fmt.Errorf("subscribe %s: %w", s.topic, err)
}
result = append(result, natsSub)
slog.Info("Subscribed to VPC topic", "topic", s.topic)
}
return result, nil
}
// --- VPC (LogicalRouter) ---
func (h *TopologyHandler) handleVPCCreate(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt VPCEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.create event", "err", err)
respond(msg, err)
return
}
spec := topology.VPCSpec{VPCID: evt.VpcId, VNI: evt.VNI}
if evt.CidrBlock != "" {
cidr, err := netip.ParsePrefix(evt.CidrBlock)
if err != nil {
slog.Error("vpcd: invalid CIDR in vpc.create event", "cidr", evt.CidrBlock, "err", err)
respond(msg, err)
return
}
spec.CIDR = cidr
}
if err := h.EnsureVPC(context.Background(), spec); err != nil {
slog.Error("vpcd: EnsureVPC failed", "vpc_id", evt.VpcId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
func (h *TopologyHandler) handleVPCDelete(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt VPCEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.delete event", "err", err)
respond(msg, err)
return
}
if err := h.DeleteVPC(context.Background(), evt.VpcId); err != nil {
slog.Error("vpcd: DeleteVPC failed", "vpc_id", evt.VpcId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
// --- Subnet (LogicalSwitch + RouterPort + DHCP) ---
func (h *TopologyHandler) handleSubnetCreate(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt SubnetEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.create-subnet event", "err", err)
respond(msg, err)
return
}
cidr, err := netip.ParsePrefix(evt.CidrBlock)
if err != nil {
slog.Error("vpcd: invalid CIDR in vpc.create-subnet event", "cidr", evt.CidrBlock, "err", err)
respond(msg, err)
return
}
ctx := context.Background()
// vpc.create-vpc and vpc.create-subnet are independent NATS events with no
// inter-event ordering guarantee. Tenant bootstrap publishes both within
// <1ms, so EnsureSubnet can race the parent VPC's CreateLogicalRouter and
// fail with "logical router not found". EnsureVPC is idempotent on the
// router name (mulga-siv-133).
if err := h.EnsureVPC(ctx, topology.VPCSpec{VPCID: evt.VpcId}); err != nil {
slog.Error("vpcd: EnsureVPC (subnet pre-create) failed", "vpc_id", evt.VpcId, "err", err)
respond(msg, err)
return
}
if err := h.EnsureSubnet(ctx, topology.SubnetSpec{
SubnetID: evt.SubnetId,
VPCID: evt.VpcId,
CIDR: cidr,
}); err != nil {
slog.Error("vpcd: EnsureSubnet failed", "subnet_id", evt.SubnetId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
func (h *TopologyHandler) handleSubnetDelete(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt SubnetEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.delete-subnet event", "err", err)
respond(msg, err)
return
}
spec := topology.SubnetSpec{SubnetID: evt.SubnetId, VPCID: evt.VpcId}
if evt.CidrBlock != "" {
if cidr, perr := netip.ParsePrefix(evt.CidrBlock); perr == nil {
spec.CIDR = cidr
}
}
if err := h.DeleteSubnet(context.Background(), spec); err != nil {
slog.Error("vpcd: DeleteSubnet failed", "subnet_id", evt.SubnetId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
// --- Port (LogicalSwitchPort for VM/ENI) ---
func (h *TopologyHandler) handleCreatePort(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt PortEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.create-port event", "err", err)
respond(msg, err)
return
}
ip, err := netip.ParseAddr(evt.PrivateIpAddress)
if err != nil {
slog.Error("vpcd: invalid private IP in vpc.create-port event", "ip", evt.PrivateIpAddress, "err", err)
respond(msg, err)
return
}
mac, err := net.ParseMAC(evt.MacAddress)
if err != nil {
slog.Error("vpcd: invalid MAC in vpc.create-port event", "mac", evt.MacAddress, "err", err)
respond(msg, err)
return
}
if err := h.EnsurePort(context.Background(), topology.PortSpec{
PortID: evt.NetworkInterfaceId,
SubnetID: evt.SubnetId,
VPCID: evt.VpcId,
PrivateIP: ip,
MAC: mac,
SGIDs: evt.SecurityGroupIds,
}); err != nil {
slog.Error("vpcd: EnsurePort failed", "eni_id", evt.NetworkInterfaceId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
func (h *TopologyHandler) handleDeletePort(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt PortEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.delete-port event", "err", err)
respond(msg, err)
return
}
if err := h.DeletePort(context.Background(), topology.PortSpec{
PortID: evt.NetworkInterfaceId,
SubnetID: evt.SubnetId,
VPCID: evt.VpcId,
}); err != nil {
slog.Error("vpcd: DeletePort failed", "eni_id", evt.NetworkInterfaceId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
// handleUpdatePortSGs reconciles the port group membership for an ENI's LSP
// against the desired SG list in the event. The payload is declarative — the
// manager computes the add/remove diff from current OVN state.
func (h *TopologyHandler) handleUpdatePortSGs(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt UpdatePortSGsEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.update-port-sgs event", "err", err)
respond(msg, err)
return
}
if err := h.SetPortSecurityGroups(context.Background(), evt.NetworkInterfaceId, evt.SecurityGroupIds); err != nil {
slog.Error("vpcd: SetPortSecurityGroups failed",
"eni_id", evt.NetworkInterfaceId, "sgs", evt.SecurityGroupIds, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
// --- Internet Gateway (external connectivity + NAT) ---
// handleIGWAttach decodes the event and delegates to
// external.IGWManager.AttachIGW.
func (h *TopologyHandler) handleIGWAttach(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt types.IGWEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.igw-attach event", "err", err)
respond(msg, err)
return
}
ctx := context.Background()
igw, err := h.igwManager()
if err != nil {
slog.Error("vpcd: IGW manager init failed", "err", err)
respond(msg, err)
return
}
if err := igw.AttachIGW(ctx, external.IGWSpec{
VPCID: evt.VpcId,
InternetGatewayID: evt.InternetGatewayId,
}); err != nil {
slog.Error("vpcd: AttachIGW failed",
"vpc_id", evt.VpcId, "igw_id", evt.InternetGatewayId, "err", err)
respond(msg, err)
return
}
respond(msg, nil)
}
// handleIGWDetach decodes the event and delegates the OVN teardown to
// external.IGWManager.DetachIGW. Static gateway LRP IPs live in OVN
// external_ids and are cleared when the LRP itself is deleted.
func (h *TopologyHandler) handleIGWDetach(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt types.IGWEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.igw-detach event", "err", err)
respond(msg, err)
return
}
ctx := context.Background()
igw, err := h.igwManager()
if err != nil {
slog.Error("vpcd: IGW manager init failed", "err", err)
respond(msg, err)
return
}
if err := igw.DetachIGW(ctx, evt.VpcId); err != nil {
slog.Error("vpcd: DetachIGW failed", "vpc_id", evt.VpcId, "err", err)
respond(msg, err)
return
}
slog.Info("vpcd: detached internet gateway from VPC",
"igw_id", evt.InternetGatewayId, "vpc_id", evt.VpcId)
respond(msg, nil)
}
// --- NAT (dnat_and_snat for public IPs) ---
// handleAddNAT decodes the EIP event and delegates to
// policy.NATManager.AddEIP. The manager owns the idempotency-skip,
// stale-rule cleanup, distributed-NAT mode selection, and post-write
// flow-install barrier.
func (h *TopologyHandler) handleAddNAT(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt NATEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.add-nat event", "err", err)
respond(msg, err)
return
}
nm, err := h.natManager()
if err != nil {
slog.Error("vpcd: NAT manager init failed", "err", err)
respond(msg, err)
return
}
if err := nm.AddEIP(context.Background(), policy.EIPSpec{
VPCID: evt.VpcId,
ExternalIP: evt.ExternalIP,
LogicalIP: evt.LogicalIP,
PortName: evt.PortName,
MAC: evt.MAC,
}); err != nil {
slog.Error("vpcd: AddEIP failed",
"vpc_id", evt.VpcId, "external_ip", evt.ExternalIP,
"logical_ip", evt.LogicalIP, "err", err)
respond(msg, err)
return
}
slog.Info("vpcd: added dnat_and_snat rule",
"vpc_id", evt.VpcId, "external_ip", evt.ExternalIP,
"logical_ip", evt.LogicalIP, "port", evt.PortName)
respond(msg, nil)
}
// handleDeleteNAT decodes the event and delegates to
// policy.NATManager.DeleteEIP. Idempotency on already-absent rules is
// handled by the manager.
func (h *TopologyHandler) handleDeleteNAT(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt NATEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.delete-nat event", "err", err)
respond(msg, err)
return
}
nm, err := h.natManager()
if err != nil {
slog.Error("vpcd: NAT manager init failed", "err", err)
respond(msg, err)
return
}
if err := nm.DeleteEIP(context.Background(), evt.VpcId, evt.LogicalIP); err != nil {
slog.Error("vpcd: DeleteEIP failed",
"vpc_id", evt.VpcId, "logical_ip", evt.LogicalIP, "err", err)
respond(msg, err)
return
}
slog.Info("vpcd: deleted dnat_and_snat rule",
"vpc_id", evt.VpcId, "logical_ip", evt.LogicalIP)
respond(msg, nil)
}
// --- Helpers ---
// respond sends a simple JSON response to a NATS request.
func respond(msg *nats.Msg, err error) {
if msg.Reply == "" {
return // fire-and-forget, no reply expected
}
type response struct {
Success bool `json:"success"`
Error string `json:"error,omitempty"`
}
resp := response{Success: true}
if err != nil {
resp.Success = false
resp.Error = err.Error()
}
data, marshalErr := json.Marshal(resp)
if marshalErr != nil {
slog.Error("vpcd: failed to marshal NATS response", "err", marshalErr)
data = []byte(`{"success":false,"error":"internal marshal failure"}`)
}
if err := msg.Respond(data); err != nil {
slog.Error("vpcd: failed to respond to NATS request", "err", err)
}
}
// handleAddNATGateway is a thin wrapper that decodes the NATS event and
// delegates the SNAT installation to policy.NATManager.AddNATGateway.
func (h *TopologyHandler) handleAddNATGateway(msg *nats.Msg) {
var evt NATGatewayEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.add-nat-gateway event", "err", err)
return
}
nm, err := h.natManager()
if err != nil {
slog.Error("vpcd: NAT manager init failed", "err", err)
return
}
if err := nm.AddNATGateway(context.Background(), policy.NATGWSpec{
VPCID: evt.VpcId,
NATGatewayID: evt.NatGatewayId,
PublicIP: evt.PublicIp,
SubnetCIDR: evt.SubnetCidr,
}); err != nil {
slog.Error("vpcd: AddNATGateway failed",
"vpc_id", evt.VpcId, "natgw_id", evt.NatGatewayId,
"public_ip", evt.PublicIp, "subnet_cidr", evt.SubnetCidr, "err", err)
return
}
slog.Info("vpcd: NAT Gateway SNAT rule added",
"vpc_id", evt.VpcId, "natgw_id", evt.NatGatewayId,
"public_ip", evt.PublicIp, "subnet_cidr", evt.SubnetCidr)
}
// handleDeleteNATGateway delegates SNAT teardown to policy.NATManager.
func (h *TopologyHandler) handleDeleteNATGateway(msg *nats.Msg) {
var evt NATGatewayEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.delete-nat-gateway event", "err", err)
return
}
nm, err := h.natManager()
if err != nil {
slog.Error("vpcd: NAT manager init failed", "err", err)
return
}
if err := nm.DeleteNATGateway(context.Background(), evt.VpcId, evt.SubnetCidr); err != nil {
slog.Warn("vpcd: DeleteNATGateway failed",
"vpc_id", evt.VpcId, "subnet_cidr", evt.SubnetCidr, "err", err)
return
}
slog.Info("vpcd: NAT Gateway SNAT rule removed",
"vpc_id", evt.VpcId, "natgw_id", evt.NatGatewayId, "subnet_cidr", evt.SubnetCidr)
}
spinifex/services/vpcd/topology_manager.go
-157
View File
@@ -1,157 +0,0 @@
package vpcd
// Phase 2.6 (mulga-siv-129) adapter: TopologyHandler still implements
// topology.Manager so existing NATS subscribers and tests stay green, but the
// concrete logic now lives in network/topology.NewLiveManager. The shim here
// constructs the live manager lazily on first use, passing TopologyHandler's
// per-VPC external-pool resolver as the DHCP dns_server callback.
//
// Once all NATS subscribers route through the live manager directly (the
// remainder of mulga-siv-129), this adapter — and TopologyHandler's
// topology.Manager conformance — can be deleted along with topology.go.
import (
"context"
"strings"
"github.com/mulgadc/spinifex/spinifex/network/external"
"github.com/mulgadc/spinifex/spinifex/network/policy"
"github.com/mulgadc/spinifex/spinifex/network/topology"
)
var _ topology.Manager = (*TopologyHandler)(nil)
func (h *TopologyHandler) topologyManager() topology.Manager {
h.lmOnce.Do(func() {
h.lm = topology.NewLiveManager(h.ovn, topology.WithDNSServer(h.lmDNSServer))
})
return h.lm
}
// lmDNSServer is the callback handed to topology.NewLiveManager so per-subnet
// DHCPOptions inherit the external pool's configured DNS list. Mirrors the
// legacy dnsServer() method without depending on TopologyHandler.findExternalPool's
// region/AZ keys (Phase 2.6 leaves multi-AZ routing to Phase 3).
func (h *TopologyHandler) lmDNSServer() string {
pool := h.findExternalPool("", "")
if pool != nil && len(pool.DNSServers) > 0 {
return "{" + strings.Join(pool.DNSServers, ", ") + "}"
}
return "{8.8.8.8, 1.1.1.1}"
}
// EnsureVPC creates the OVN logical router for the VPC, idempotently.
func (h *TopologyHandler) EnsureVPC(ctx context.Context, spec topology.VPCSpec) error {
return h.topologyManager().EnsureVPC(ctx, spec)
}
// DeleteVPC removes the OVN logical router for the VPC and cascades through
// any subnets that belong to it.
func (h *TopologyHandler) DeleteVPC(ctx context.Context, vpcID string) error {
return h.topologyManager().DeleteVPC(ctx, vpcID)
}
// EnsureSubnet creates the OVN logical switch, subnet→VPC router port pair,
// and DHCP options for the subnet.
func (h *TopologyHandler) EnsureSubnet(ctx context.Context, spec topology.SubnetSpec) error {
return h.topologyManager().EnsureSubnet(ctx, spec)
}
// DeleteSubnet tears down the subnet's logical switch, router port, switch
// router port, and DHCP options.
func (h *TopologyHandler) DeleteSubnet(ctx context.Context, spec topology.SubnetSpec) error {
return h.topologyManager().DeleteSubnet(ctx, spec)
}
// EnsurePort creates the ENI's LogicalSwitchPort and joins it to its initial
// SG port groups in one OVSDB transaction.
func (h *TopologyHandler) EnsurePort(ctx context.Context, spec topology.PortSpec) error {
return h.topologyManager().EnsurePort(ctx, spec)
}
// DeletePort clears the ENI's port-group memberships and removes the LSP.
func (h *TopologyHandler) DeletePort(ctx context.Context, spec topology.PortSpec) error {
return h.topologyManager().DeletePort(ctx, spec)
}
// SetPortSecurityGroups reconciles the port's port-group memberships against
// the declared list.
func (h *TopologyHandler) SetPortSecurityGroups(ctx context.Context, portID string, sgIDs []string) error {
return h.topologyManager().SetPortSecurityGroups(ctx, portID, sgIDs)
}
// EnsureSGPortGroup creates the OVN port group for a security group.
func (h *TopologyHandler) EnsureSGPortGroup(ctx context.Context, groupID string) error {
return h.topologyManager().EnsureSGPortGroup(ctx, groupID)
}
// DeleteSGPortGroup removes the OVN port group for a security group.
func (h *TopologyHandler) DeleteSGPortGroup(ctx context.Context, groupID string) error {
return h.topologyManager().DeleteSGPortGroup(ctx, groupID)
}
// DeleteSGPortGroupByName removes the OVN port group given its raw name.
func (h *TopologyHandler) DeleteSGPortGroupByName(ctx context.Context, pgName string) error {
return h.topologyManager().DeleteSGPortGroupByName(ctx, pgName)
}
// securityGroupManager returns the lazily-constructed policy SG manager
// backed by the same OVN client as the live topology manager. Subscribers
// use it to apply ACL sets against the SG's port group.
func (h *TopologyHandler) securityGroupManager() policy.SecurityGroupManager {
h.sgmOnce.Do(func() {
h.sgm = policy.NewSecurityGroupManager(h.ovn)
})
return h.sgm
}
// natManager returns the lazily-constructed policy.NATManager. NAT mode is
// the value injected via WithNATMode; the FlowsBarrier is bound to
// waitForFlowsHV so the EIP subscriber blocks on hypervisor flow-install
// before responding.
func (h *TopologyHandler) natManager() (policy.NATManager, error) {
h.natmOnce.Do(func() {
h.natm, h.natmErr = policy.NewNATManager(h.ovn, h.natMode, policy.WithFlowsBarrier(waitForFlowsHV))
})
return h.natm, h.natmErr
}
// igwManager returns the lazily-constructed external.IGWManager. All
// IGW attach/detach paths route through it.
func (h *TopologyHandler) igwManager() (external.IGWManager, error) {
h.igwmOnce.Do(func() {
nm, err := h.natManager()
if err != nil {
h.igwmErr = err
return
}
var poolCfg *external.ExternalPoolConfig
if p := h.findExternalPool("", ""); p != nil {
shared := external.ExternalPoolConfig{
Name: p.Name,
RangeStart: p.RangeStart,
RangeEnd: p.RangeEnd,
Gateway: p.Gateway,
GatewayIP: p.GatewayIP,
PrefixLen: p.PrefixLen,
DNSServers: p.DNSServers,
Region: p.Region,
AZ: p.AZ,
GwLrpRangeStart: p.GwLrpRangeStart,
GwLrpRangeEnd: p.GwLrpRangeEnd,
}
poolCfg = &shared
}
h.igwm, h.igwmErr = external.NewIGWManager(external.IGWManagerConfig{
OVN: h.ovn,
Routes: policy.NewRouteManager(h.ovn),
NAT: nm,
Pool: poolCfg,
Allocator: external.NewStaticRangeAllocator(h.ovn),
Chassis: h.chassisNames,
NATMode: h.natMode,
FlowsBarrier: waitForFlowsHV,
})
})
return h.igwm, h.igwmErr
}
spinifex/services/vpcd/topology_test.go
-3052
View File
File diff suppressed because it is too large Load Diff
spinifex/services/vpcd/vpcd.go
+106 -37
View File
@@ -19,9 +19,57 @@ import (
"github.com/mulgadc/spinifex/spinifex/network/host"
"github.com/mulgadc/spinifex/spinifex/network/policy"
"github.com/mulgadc/spinifex/spinifex/network/reconcile"
"github.com/mulgadc/spinifex/spinifex/network/subscribers"
"github.com/mulgadc/spinifex/spinifex/network/topology"
"github.com/mulgadc/spinifex/spinifex/utils"
)
// Bridge mode constants for external connectivity. Bridge mode selects how
// the WAN NIC reaches the OVS integration bridge:
// - Direct: WAN NIC is added directly to br-external as an OVS port.
// Enables distributed NAT. Only safe when the WAN NIC is NOT the SSH/
// management NIC.
// - Veth: a veth pair links a Linux bridge (br-wan) to the OVS bridge
// (br-ext). Requires centralized NAT — the Linux bridge intermediary
// breaks distributed NAT hairpin routing.
const (
BridgeModeDirect = "direct"
BridgeModeVeth = "veth"
// OvnExternalBridge is the OVS bridge that ovn-bridge-mappings targets
// for the "external" localnet. Owned by setup-ovn.sh and independent of
// the WAN bridge (which is Linux-side in veth mode).
OvnExternalBridge = "br-ext"
)
// waitForFlowsHV shells out to `ovn-nbctl --wait=hv sync`, bumping
// NB_Global.nb_cfg and blocking until every connected chassis has acknowledged
// the new sequence — i.e. ovn-northd has compiled NB→SB and ovn-controller
// has installed the resulting flows. Used as the NAT manager's flows barrier
// so newly-launched VMs aren't unreachable while their gateway chassis is
// still catching up (mulga-siv-105). Bounded by ovn-nbctl --timeout=30; on
// overrun we log a Warn and return nil so the caller continues.
//
// Declared as a var so tests can stub it.
var waitForFlowsHV = func() error {
start := time.Now()
cmd := sudoCommand("ovn-nbctl",
"--no-leader-only",
"--timeout=30",
"--wait=hv",
"sync",
)
if out, err := cmd.CombinedOutput(); err != nil {
slog.Warn("vpcd: OVN flows-ready barrier overran; continuing without confirmation",
"elapsed", time.Since(start),
"err", err,
"output", strings.TrimSpace(string(out)),
)
return nil
}
slog.Debug("vpcd: OVN flows-ready barrier complete", "elapsed", time.Since(start))
return nil
}
// sudoCommand wraps exec.Command with sudo when running as non-root.
// OVS/OVN commands require elevated privileges; when running as root
// (Docker, production) no wrapper is needed.
@@ -416,12 +464,7 @@ func launchService(cfg *Config) error {
return err
}
// Reconcile OVN topology from bootstrap config before subscribing.
// This ensures the default VPC topology exists even if admin init ran
// before services were started (first-install case).
var topoOpts []TopologyOption
if cfg.ExternalMode != "" {
topoOpts = append(topoOpts, WithExternalNetwork(cfg.ExternalMode, cfg.ExternalPools))
slog.Info("External network enabled", "mode", cfg.ExternalMode, "pools", len(cfg.ExternalPools))
}
// Discover chassis from OVN SBDB. The OVS-managed system-id (persisted at
@@ -436,46 +479,24 @@ func launchService(cfg *Config) error {
if len(chassisNames) == 0 {
return fmt.Errorf("vpcd: no OVN chassis registered in SBDB — is ovn-controller running and connected?")
}
topoOpts = append(topoOpts, WithChassisNames(chassisNames))
slog.Info("vpcd: gateway chassis discovered", "chassis", chassisNames)
uplinkMode := host.UplinkModePhysical
if bridgeMode == BridgeModeVeth {
uplinkMode = host.UplinkModeVeth
}
natMode := policy.NATModeFromUplinkMode(uplinkMode)
topoOpts = append(topoOpts, WithNATMode(natMode))
topo := NewTopologyHandler(liveClient, topoOpts...)
// Elect a single vpcd to run startup reconcile. Without this, N vpcds in a
// multi-node cluster all hit Get-then-Create on Logical_Router with no
// atomicity, producing duplicate rows that ovn-nbctl rejects with
// "Multiple logical routers named '...'. Use a UUID." (mulga-siv-29).
// Runtime VPC events still fan out via the vpcd-workers queue group, so
// non-leaders remain functional after Subscribe below.
holder, _ := os.Hostname()
releaseLeader, isLeader := reconcile.AcquireLeader(nc, holder)
subs, err := topo.Subscribe(nc)
if err != nil {
slog.Error("Failed to subscribe to VPC topics", "err", err)
return err
// Construct the network manager stack. The reconciler is the single
// intent-driven pass that runs once at startup (leader-gated) and then
// on a 5-minute drift ticker; together they replace the five separate
// passes the old vpcd ran on boot (mulga-siv-125, Phase 2).
var topoOpts []topology.Option
if dns := pickDNSServer(cfg.ExternalPools); dns != "" {
topoOpts = append(topoOpts, topology.WithDNSServer(func() string { return dns }))
}
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
topoMgr := topology.NewLiveManager(liveClient, topoOpts...)
js, err := nc.JetStream()
if err != nil {
slog.Error("Failed to get JetStream context", "err", err)
return err
}
// Construct the network/reconcile manager stack. The reconciler is the
// single intent-driven pass that runs once at startup (leader-gated) and
// then on a 5-minute drift ticker; together they replace the five
// separate passes the old vpcd ran on boot (mulga-siv-125, Phase 2).
sgMgr := policy.NewSecurityGroupManager(liveClient)
natMgr, err := policy.NewNATManager(liveClient, natMode, policy.WithFlowsBarrier(waitForFlowsHV))
if err != nil {
@@ -502,13 +523,48 @@ func launchService(cfg *Config) error {
return fmt.Errorf("construct IGW manager: %w", err)
}
// Elect a single vpcd to run startup reconcile. Without this, N vpcds in
// a multi-node cluster all hit Get-then-Create on Logical_Router with no
// atomicity, producing duplicate rows that ovn-nbctl rejects with
// "Multiple logical routers named '...'. Use a UUID." (mulga-siv-29).
// Runtime VPC events still fan out via the vpcd-workers queue group, so
// non-leaders remain functional after Subscribe below.
holder, _ := os.Hostname()
releaseLeader, isLeader := reconcile.AcquireLeader(nc, holder)
subscriber, err := subscribers.New(subscribers.Config{
Topology: topoMgr,
SG: sgMgr,
NAT: natMgr,
IGW: igwMgr,
})
if err != nil {
return fmt.Errorf("construct subscriber: %w", err)
}
subs, err := subscriber.Subscribe(nc)
if err != nil {
slog.Error("Failed to subscribe to VPC topics", "err", err)
return err
}
defer func() {
for _, s := range subs {
_ = s.Unsubscribe()
}
}()
js, err := nc.JetStream()
if err != nil {
slog.Error("Failed to get JetStream context", "err", err)
return err
}
rec, err := reconcile.New(reconcile.Config{
OVN: liveClient,
SG: sgMgr,
NAT: natMgr,
Routes: routeMgr,
IGW: igwMgr,
Topology: topo,
Topology: topoMgr,
LocalAZ: cfg.AZ,
NodeHostname: holder,
Chassis: chassisNames,
@@ -558,6 +614,19 @@ func launchService(cfg *Config) error {
return nil
}
// pickDNSServer returns the OVN dhcp_options dns_server string ("{a, b}")
// from the first unscoped (no region, no AZ) pool with DNS servers set.
// Empty string falls back to topology.NewLiveManager's default
// ("{8.8.8.8, 1.1.1.1}").
func pickDNSServer(pools []ExternalPoolConfig) string {
for _, p := range pools {
if p.Region == "" && p.AZ == "" && len(p.DNSServers) > 0 {
return "{" + strings.Join(p.DNSServers, ", ") + "}"
}
}
return ""
}
// externalPoolConfigToShared translates the vpcd-local ExternalPoolConfig
// into the network/external package's shared shape. The two will merge
// once topology.go is split (bead mulga-siv-125.3.4 — see external.go's