diff --git a/docs/configs/README.md b/docs/configs/README.md index ba8d58c..6bb9b35 100644 --- a/docs/configs/README.md +++ b/docs/configs/README.md @@ -178,7 +178,7 @@ Supported fields: | `memory_mb` | no | Memory in MiB | | `kernel_args` | no | Kernel boot args | | `network` | no | When true, networking is configured | -| `port_forwards` | no | Host-to-guest DNAT mappings; required for remote Traefik routing. Reachable from external traffic and from co-located guests addressing the host IP (hairpin) | +| `port_forwards` | no | Host-to-guest DNAT mappings; required for remote Traefik routing. Reachable from external traffic and from co-located guests addressing the host IP (hairpin). Each `host_port` is a node-exclusive claim — see [Host-port claims](#host-port-claims) | | `health_check` | no | `type` supports `http` or `tcp` | | `env` | no | Env vars injected via kernel args; values with whitespace are encoded | | `links` | no | Same-node service links (`env` gets resolved URL) | @@ -197,6 +197,26 @@ non-empty tenant `volumes` list replaces the inherited list. See [Persistent Volumes](../persistent-volumes.md) for lifecycle and safety semantics. +#### Host-port claims + +A `host_port` is a node-scoped exclusive resource: the agent installs one DNAT +rule per claim, and two colocated services claiming the same port would produce +rules with identical match criteria and different guest destinations, so only +one of them would ever receive traffic. + +- Repeating a `host_port` inside one service is always invalid and is rejected + by `configcheck` and by the enricher. +- Repeating a `host_port` across services is valid only while placement keeps + them on different nodes. `configcheck` reports it as a `repeated_host_port` + warning because it cannot know the placement. +- The scheduler treats a service's claims atomically: it places the service + only on a node where every claim is free, and otherwise leaves it `pending` + with reason `host_port_conflict` (or `duplicate_host_port_claims` for a + service that repeats its own port). +- The agent rejects a rendered node config whose services conflict on a host + port before changing any networking or service state, reporting + `host_port_conflict` and leaving the revision unapplied. + ### 2.3 `tenants/*` (optional) Tenant files support two modes: diff --git a/docs/deployment-visibility.md b/docs/deployment-visibility.md index e10013e..6700db7 100644 --- a/docs/deployment-visibility.md +++ b/docs/deployment-visibility.md @@ -43,7 +43,10 @@ of the service's volumes. Missing data fails closed: - expired node leases become `stale`; -- unplaced desired services are `pending`; +- unplaced desired services are `pending`, carrying the scheduler's reason code + (for example `insufficient_compute_capacity`, `volume_capacity_unavailable`, + or `host_port_conflict`, which names the contested port and the service + already holding it); - placed services with missing, stale, or unsupported agent status are `unknown`; - VM state and health remain separate, so a service can be `running` and diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 4d6317e..fafedfc 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -405,6 +405,18 @@ func (a *Agent) tick(ctx context.Context) { a.setStatusServices(*merged, "") } + // Reject a node config whose services conflict on a host port before any + // networking or service state changes. The control plane keeps colocated + // claims unique, so this is defense in depth for a stale, hand-written, or + // older-controller config: installing its DNAT rules would silently deliver + // host-port traffic to whichever guest matched first. + if err := config.ValidateNodePortClaims(*merged); err != nil { + a.logger.Error("host-port claim preflight failed; not advancing revision", "error", err) + a.failAgentStatus("NetworkReady", "host_port_conflict", err.Error()) + a.syncRegistry(ctx, nodeCap, capacity.NodeCapacity{}) + return + } + // Preflight routing metadata before any reconciliation so an invalid // direct node config or a missing ingress_domain fails the revision early // instead of being recorded as applied. diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 71334f8..79408a0 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -878,6 +878,45 @@ func TestReadNodeCapacity_UsesLastKnownOnError(t *testing.T) { } } +func TestTick_HostPortConflict_RejectsConfigBeforeReconcile(t *testing.T) { + nodeYAML := []byte("node: web\nservices:\n" + + "- name: tenant-1-elasticsearch\n image: /img/es\n kernel: /kern\n vcpus: 1\n memory_mb: 256\n network: {}\n port_forwards:\n - host_port: 9200\n vm_port: 9200\n" + + "- name: tenant-2-elasticsearch\n image: /img/es\n kernel: /kern\n vcpus: 1\n memory_mb: 256\n network: {}\n port_forwards:\n - host_port: 9200\n vm_port: 9200\n") + s := &fakeStore{data: map[string][]byte{"web": nodeYAML}, revision: "rev-1"} + + a := New(testAgentConfig(t), s, testLogger()) + a.tick(context.Background()) + + if instances := a.vmManager.List(); len(instances) != 0 { + t.Errorf("expected no VMs to be started for a conflicting config, got %d", len(instances)) + } + status := a.agentStatusSnapshot() + if status.Phase != "failed" || status.ReasonCode != "host_port_conflict" { + t.Fatalf("unexpected status: %#v", status) + } + if !strings.Contains(status.Message, "9200") { + t.Errorf("status message %q does not identify the conflicting port", status.Message) + } + // The revision must not be recorded as applied, so the next poll retries. + if a.lastRevision != "" { + t.Errorf("expected revision not to advance, got %q", a.lastRevision) + } +} + +func TestTick_DistinctHostPortsReconcileNormally(t *testing.T) { + nodeYAML := []byte("node: web\nservices:\n" + + "- name: tenant-1-elasticsearch\n image: /img/es\n kernel: /kern\n vcpus: 1\n memory_mb: 256\n network: {}\n port_forwards:\n - host_port: 9200\n vm_port: 9200\n" + + "- name: tenant-2-elasticsearch\n image: /img/es\n kernel: /kern\n vcpus: 1\n memory_mb: 256\n network: {}\n port_forwards:\n - host_port: 9201\n vm_port: 9200\n") + s := &fakeStore{data: map[string][]byte{"web": nodeYAML}, revision: "rev-1"} + + a := New(testAgentConfig(t), s, testLogger()) + a.tick(context.Background()) + + if status := a.agentStatusSnapshot(); status.ReasonCode == "host_port_conflict" { + t.Fatalf("distinct host ports must not be rejected: %#v", status) + } +} + // runningVMManager reports a service as already running, which is what steady // state looks like: port forwards are only asserted for services that have a // guest to forward to. diff --git a/internal/config/portclaim.go b/internal/config/portclaim.go new file mode 100644 index 0000000..ec021e9 --- /dev/null +++ b/internal/config/portclaim.go @@ -0,0 +1,133 @@ +package config + +import ( + "fmt" + "sort" + "strings" +) + +// ProtocolTCP is the effective protocol of every port forward today. The agent +// installs DNAT rules with "-p tcp", so a claim is currently identified by that +// protocol plus the host port. Protocol is carried explicitly in PortClaim so +// adding UDP forwards later widens the key without changing its callers. +const ProtocolTCP = "tcp" + +// PortClaim is a node-exclusive host endpoint requested by a service. Two +// services on the same node may not hold the same claim: their DNAT rules would +// match identical traffic and only the first installed rule would win. +type PortClaim struct { + Protocol string + HostPort int +} + +func (c PortClaim) String() string { + return fmt.Sprintf("%s/%d", c.Protocol, c.HostPort) +} + +// PortClaims returns the host-port claims a service makes, in declaration +// order. Entries without a positive host port are not forwarded by the agent +// and therefore claim nothing. +func (s ServiceConfig) PortClaims() []PortClaim { + var claims []PortClaim + for _, pf := range s.PortForwards { + if pf.HostPort <= 0 { + continue + } + claims = append(claims, PortClaim{Protocol: ProtocolTCP, HostPort: pf.HostPort}) + } + return claims +} + +// DuplicatePortClaims returns the claims a single service requests more than +// once, sorted by host port. Such a service is never schedulable anywhere: the +// duplicate DNAT rules conflict with each other on any node. +func (s ServiceConfig) DuplicatePortClaims() []PortClaim { + seen := make(map[PortClaim]int, len(s.PortForwards)) + for _, claim := range s.PortClaims() { + seen[claim]++ + } + var dupes []PortClaim + for claim, count := range seen { + if count > 1 { + dupes = append(dupes, claim) + } + } + sortClaims(dupes) + return dupes +} + +// PortClaimConflict reports a claim requested by more than one service in the +// same node scope. Services are sorted so the message is stable. +type PortClaimConflict struct { + Claim PortClaim + Services []string +} + +func (c PortClaimConflict) String() string { + return fmt.Sprintf("host port %d (%s) is claimed by %s", + c.Claim.HostPort, c.Claim.Protocol, strings.Join(c.Services, ", ")) +} + +// ConflictingPortClaims returns claims held by more than one of the given +// services. It is the cross-service half of the invariant only; use +// DuplicatePortClaims for claims a single service repeats. +func ConflictingPortClaims(services []ServiceConfig) []PortClaimConflict { + holders := make(map[PortClaim][]string) + for _, svc := range services { + claimed := make(map[PortClaim]bool) + for _, claim := range svc.PortClaims() { + if claimed[claim] { + continue + } + claimed[claim] = true + holders[claim] = append(holders[claim], svc.Name) + } + } + + var conflicts []PortClaimConflict + for claim, names := range holders { + if len(names) < 2 { + continue + } + sorted := append([]string(nil), names...) + sort.Strings(sorted) + conflicts = append(conflicts, PortClaimConflict{Claim: claim, Services: sorted}) + } + sort.Slice(conflicts, func(i, j int) bool { + return claimLess(conflicts[i].Claim, conflicts[j].Claim) + }) + return conflicts +} + +// ValidateNodePortClaims rejects a rendered node config whose services cannot +// coexist on one node. It is the agent's admission boundary: a stale, +// hand-written, or older-controller config must be refused before any +// networking or service state changes, because the resulting DNAT rules would +// silently deliver traffic to the wrong guest. +func ValidateNodePortClaims(nc NodeConfig) error { + var problems []string + for _, svc := range nc.Services { + for _, dupe := range svc.DuplicatePortClaims() { + problems = append(problems, fmt.Sprintf( + "service %s claims host port %d (%s) more than once", svc.Name, dupe.HostPort, dupe.Protocol)) + } + } + for _, conflict := range ConflictingPortClaims(nc.Services) { + problems = append(problems, conflict.String()) + } + if len(problems) == 0 { + return nil + } + return fmt.Errorf("conflicting host-port claims on node %s: %s", nc.Node, strings.Join(problems, "; ")) +} + +func sortClaims(claims []PortClaim) { + sort.Slice(claims, func(i, j int) bool { return claimLess(claims[i], claims[j]) }) +} + +func claimLess(a, b PortClaim) bool { + if a.HostPort != b.HostPort { + return a.HostPort < b.HostPort + } + return a.Protocol < b.Protocol +} diff --git a/internal/config/portclaim_test.go b/internal/config/portclaim_test.go new file mode 100644 index 0000000..76fdd70 --- /dev/null +++ b/internal/config/portclaim_test.go @@ -0,0 +1,101 @@ +package config + +import ( + "strings" + "testing" +) + +func TestPortClaims_SkipsEntriesWithoutHostPort(t *testing.T) { + svc := ServiceConfig{Name: "a", PortForwards: []PortForward{ + {HostPort: 9200, VMPort: 9200}, + {VMPort: 9300}, + {HostPort: 0, VMPort: 5601}, + }} + + claims := svc.PortClaims() + if len(claims) != 1 || claims[0] != (PortClaim{Protocol: ProtocolTCP, HostPort: 9200}) { + t.Fatalf("unexpected claims: %#v", claims) + } +} + +func TestDuplicatePortClaims(t *testing.T) { + svc := ServiceConfig{Name: "a", PortForwards: []PortForward{ + {HostPort: 9200, VMPort: 9200}, + {HostPort: 9200, VMPort: 9201}, + {HostPort: 5601, VMPort: 5601}, + }} + + dupes := svc.DuplicatePortClaims() + if len(dupes) != 1 || dupes[0].HostPort != 9200 { + t.Fatalf("expected 9200 reported once, got %#v", dupes) + } +} + +func TestConflictingPortClaims_ReportsHoldersSorted(t *testing.T) { + services := []ServiceConfig{ + {Name: "tenant-2-elasticsearch", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}}}, + {Name: "tenant-1-elasticsearch", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}}}, + {Name: "tenant-1-kibana", PortForwards: []PortForward{{HostPort: 5601, VMPort: 5601}}}, + } + + conflicts := ConflictingPortClaims(services) + if len(conflicts) != 1 { + t.Fatalf("expected one conflict, got %#v", conflicts) + } + if conflicts[0].Claim.HostPort != 9200 { + t.Errorf("expected conflict on 9200, got %d", conflicts[0].Claim.HostPort) + } + if got := strings.Join(conflicts[0].Services, ","); got != "tenant-1-elasticsearch,tenant-2-elasticsearch" { + t.Errorf("unexpected holders: %s", got) + } +} + +// A service repeating one claim conflicts with itself, not with its peers. +func TestConflictingPortClaims_IgnoresRepeatsWithinOneService(t *testing.T) { + services := []ServiceConfig{ + {Name: "a", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}, {HostPort: 9200, VMPort: 9201}}}, + } + + if conflicts := ConflictingPortClaims(services); len(conflicts) != 0 { + t.Fatalf("expected no cross-service conflict, got %#v", conflicts) + } +} + +func TestValidateNodePortClaims_RejectsCollocatedDuplicate(t *testing.T) { + nc := NodeConfig{Node: "i-001", Services: []ServiceConfig{ + {Name: "tenant-1-elasticsearch", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}}}, + {Name: "tenant-2-elasticsearch", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}}}, + }} + + err := ValidateNodePortClaims(nc) + if err == nil { + t.Fatal("expected conflicting claims to be rejected") + } + for _, want := range []string{"i-001", "9200", "tenant-1-elasticsearch", "tenant-2-elasticsearch"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("error %q does not identify %q", err, want) + } + } +} + +func TestValidateNodePortClaims_RejectsSelfDuplicate(t *testing.T) { + nc := NodeConfig{Node: "i-001", Services: []ServiceConfig{ + {Name: "a", PortForwards: []PortForward{{HostPort: 8080, VMPort: 80}, {HostPort: 8080, VMPort: 81}}}, + }} + + if err := ValidateNodePortClaims(nc); err == nil { + t.Fatal("expected a service claiming one host port twice to be rejected") + } +} + +func TestValidateNodePortClaims_AllowsDistinctClaims(t *testing.T) { + nc := NodeConfig{Node: "i-001", Services: []ServiceConfig{ + {Name: "a", PortForwards: []PortForward{{HostPort: 9200, VMPort: 9200}, {HostPort: 9300, VMPort: 9300}}}, + {Name: "b", PortForwards: []PortForward{{HostPort: 9201, VMPort: 9200}}}, + {Name: "c"}, + }} + + if err := ValidateNodePortClaims(nc); err != nil { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/internal/enricher/validation.go b/internal/enricher/validation.go index c2a49fc..d6a2857 100644 --- a/internal/enricher/validation.go +++ b/internal/enricher/validation.go @@ -4,6 +4,7 @@ import ( "fmt" "path/filepath" "regexp" + "sort" "strings" "github.com/artemnikitin/firework/internal/config" @@ -43,6 +44,11 @@ const ( // port_forwards host port, so it cannot participate in remote multi-node // routing (remote nodes proxy through the host port). WarnRemoteRoutingNoHostPort = "remote_routing_no_host_port" + // WarnRepeatedHostPort: several services of the same node type request the + // same host port. This is legitimate only while placement keeps them on + // different nodes; the scheduler leaves the extra services pending with + // host_port_conflict when it cannot. + WarnRepeatedHostPort = "repeated_host_port" ) // Warn represents a non-fatal issue found during validation. @@ -95,6 +101,7 @@ func ValidateInput(input *InputConfig) error { } validateRouting(ve, s, input.Defaults, subSeen, hostSeen) + validatePortForwards(ve, s) validateVolumes(ve, s) } @@ -104,6 +111,17 @@ func ValidateInput(input *InputConfig) error { return nil } +// validatePortForwards rejects a service that claims the same host port twice. +// Such a service is never schedulable: its own DNAT rules would conflict on any +// node. Repeats across services are left to placement, which keeps colocated +// claims unique, so they cannot be judged statically here. +func validatePortForwards(ve *ValidationError, s ServiceSpec) { + svc := config.ServiceConfig{Name: s.Name, PortForwards: s.PortForwards} + for _, dupe := range svc.DuplicatePortClaims() { + ve.addf("service %s: duplicate port_forwards host port %d", s.Name, dupe.HostPort) + } +} + var volumeNamePattern = regexp.MustCompile(`^[a-z0-9](?:[-a-z0-9]{0,61}[a-z0-9])?$`) func validateVolumeDefaults(ve *ValidationError, defs VolumeDefaults) { @@ -281,6 +299,9 @@ func ValidateOutput(nc config.NodeConfig) error { ve.addf("service %s volume %s: size_bytes must be positive", svc.Name, volume.Name) } } + for _, dupe := range svc.DuplicatePortClaims() { + ve.addf("service %s: duplicate port_forwards host port %d", svc.Name, dupe.HostPort) + } } if ve.hasErrors() { @@ -293,6 +314,8 @@ func ValidateOutput(nc config.NodeConfig) error { func CheckWarnings(input *InputConfig) []Warn { var warns []Warn + warns = append(warns, repeatedHostPortWarnings(input.Services)...) + for _, svc := range input.Services { if svc.HealthCheck != nil && !svc.Network { warns = append(warns, Warn{ @@ -315,3 +338,33 @@ func CheckWarnings(input *InputConfig) []Warn { return warns } + +// repeatedHostPortWarnings reports host ports requested by several services of +// the same node type. Only placement can decide whether that is safe, so this +// stays a warning: it is valid while the scheduler separates the services, and +// becomes a pending host_port_conflict when it cannot. +func repeatedHostPortWarnings(specs []ServiceSpec) []Warn { + byNodeType := make(map[string][]config.ServiceConfig) + for _, spec := range specs { + byNodeType[spec.NodeType] = append(byNodeType[spec.NodeType], + config.ServiceConfig{Name: spec.Name, PortForwards: spec.PortForwards}) + } + + nodeTypes := make([]string, 0, len(byNodeType)) + for nodeType := range byNodeType { + nodeTypes = append(nodeTypes, nodeType) + } + sort.Strings(nodeTypes) + + var warns []Warn + for _, nodeType := range nodeTypes { + for _, conflict := range config.ConflictingPortClaims(byNodeType[nodeType]) { + warns = append(warns, Warn{ + Code: WarnRepeatedHostPort, + Message: fmt.Sprintf("node type %s: %s; placement must keep them on different nodes", + nodeType, conflict), + }) + } + } + return warns +} diff --git a/internal/enricher/validation_test.go b/internal/enricher/validation_test.go index 477933e..0f44311 100644 --- a/internal/enricher/validation_test.go +++ b/internal/enricher/validation_test.go @@ -305,3 +305,92 @@ func TestCheckWarnings_NoWarnings(t *testing.T) { t.Errorf("expected no warnings, got: %v", warns) } } + +func hasWarn(warns []Warn, code string) bool { + for _, w := range warns { + if w.Code == code { + return true + } + } + return false +} + +func TestValidateInput_RejectsDuplicateHostPortWithinService(t *testing.T) { + input := &InputConfig{Services: []ServiceSpec{{ + Name: "web", + Image: "/img/web.ext4", + NodeType: "compute", + PortForwards: []config.PortForward{ + {HostPort: 8080, VMPort: 80}, + {HostPort: 8080, VMPort: 81}, + }, + }}} + + err := ValidateInput(input) + if err == nil { + t.Fatal("expected a service claiming one host port twice to be rejected") + } + if !strings.Contains(err.Error(), "8080") { + t.Errorf("error %q does not identify the port", err) + } +} + +// Repeated host ports across services are decided by placement, not statically: +// they are valid whenever the scheduler puts the services on different nodes. +func TestValidateInput_AllowsRepeatedHostPortAcrossServicesWithWarning(t *testing.T) { + input := &InputConfig{Services: []ServiceSpec{ + {Name: "tenant-1-es", Image: "/img/es.ext4", NodeType: "compute", + PortForwards: []config.PortForward{{HostPort: 9200, VMPort: 9200}}}, + {Name: "tenant-2-es", Image: "/img/es.ext4", NodeType: "compute", + PortForwards: []config.PortForward{{HostPort: 9200, VMPort: 9200}}}, + }} + + if err := ValidateInput(input); err != nil { + t.Fatalf("expected valid, got: %v", err) + } + warns := CheckWarnings(input) + if !hasWarn(warns, WarnRepeatedHostPort) { + t.Fatalf("expected a repeated-host-port warning, got: %v", warns) + } + for _, w := range warns { + if w.Code != WarnRepeatedHostPort { + continue + } + for _, want := range []string{"compute", "9200", "tenant-1-es", "tenant-2-es"} { + if !strings.Contains(w.Message, want) { + t.Errorf("warning %q does not mention %q", w.Message, want) + } + } + } +} + +func TestCheckWarnings_NoRepeatedHostPortAcrossNodeTypes(t *testing.T) { + input := &InputConfig{Services: []ServiceSpec{ + {Name: "a", Image: "/img/a.ext4", NodeType: "compute", + PortForwards: []config.PortForward{{HostPort: 9200, VMPort: 9200}}}, + {Name: "b", Image: "/img/b.ext4", NodeType: "storage", + PortForwards: []config.PortForward{{HostPort: 9200, VMPort: 9200}}}, + }} + + if warns := CheckWarnings(input); hasWarn(warns, WarnRepeatedHostPort) { + t.Fatalf("services of different node types never share a node: %v", warns) + } +} + +func TestValidateOutput_RejectsDuplicateHostPortWithinService(t *testing.T) { + nc := config.NodeConfig{Node: "compute", Services: []config.ServiceConfig{{ + Name: "web", + Image: "/images/web.ext4", + Kernel: "/kernels/vmlinux", + VCPUs: 2, + MemoryMB: 1024, + PortForwards: []config.PortForward{ + {HostPort: 8080, VMPort: 80}, + {HostPort: 8080, VMPort: 81}, + }, + }}} + + if err := ValidateOutput(nc); err == nil { + t.Fatal("expected duplicate host port in rendered output to be rejected") + } +} diff --git a/internal/network/setup_test.go b/internal/network/setup_test.go index e6a37f2..56679ee 100644 --- a/internal/network/setup_test.go +++ b/internal/network/setup_test.go @@ -110,3 +110,25 @@ func TestScopedPortForwardSpec(t *testing.T) { } } } + +// Removing one service's forward must not remove another service's rule for the +// same host port: every rule spec used for teardown carries the guest +// destination, so the match is service-scoped even during a same-port handoff. +func TestPortForwardSpecsAreScopedToGuestDestination(t *testing.T) { + t.Parallel() + + specs := map[string][]string{ + "scoped": scopedPortForwardSpec("enp3s0", "10.0.100.91", 9200, "172.16.0.6", 9200), + "hairpin": hairpinPortForwardSpec("br0", "10.0.100.91", 9200, "172.16.0.6", 9200), + "legacy": legacyPortForwardSpec(9200, "172.16.0.6", 9200), + } + for name, spec := range specs { + got := strings.Join(spec, " ") + if !strings.Contains(got, "--to-destination 172.16.0.6:9200") { + t.Errorf("%s spec is not destination-scoped: %s", name, got) + } + if strings.Contains(got, "--to-destination 172.16.0.7:9200") { + t.Errorf("%s spec matches a peer guest: %s", name, got) + } + } +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index aa3e052..9039370 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -50,6 +50,9 @@ type Pending struct { // existingAssignment maps service name → instance ID from the previous run. // The scheduler preserves existing assignments when possible. // +// It does not enforce host-port claims; use ScheduleWithStorage for placement +// that keeps colocated services from conflicting on a host port. +// // Returns a map of instance ID → services assigned to that node. func Schedule( services []config.ServiceConfig, @@ -193,8 +196,15 @@ func BuildNodeConfigs(assignment map[string][]config.ServiceConfig) []config.Nod } // ScheduleWithStorage preserves the legacy CPU/memory behavior while adding -// retained-volume constraints and per-service pending results. It is kept -// separate from Schedule so existing direct callers retain error semantics. +// retained-volume constraints, host-port claims, and per-service pending +// results. It is kept separate from Schedule so existing direct callers retain +// error semantics. +// +// Host ports are node-scoped resources: two colocated services claiming the +// same port produce DNAT rules with identical match criteria and different +// guest destinations, so traffic silently reaches only one of them. A service +// is therefore placed only on a node where all of its claims are free, and its +// claims are taken atomically. func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing map[string]string, reservations StorageReservations) (map[string][]config.ServiceConfig, []Pending) { result := make(map[string][]config.ServiceConfig, len(nodes)) usedVCPU := make(map[string]int, len(nodes)) @@ -202,10 +212,12 @@ func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing usedLocal := make(map[string]int64, len(nodes)) usedShared := make(map[string]int64) groups := make(map[string]map[string]bool, len(nodes)) + claimedPorts := make(map[string]map[config.PortClaim]string, len(nodes)) nodeByID := make(map[string]Node, len(nodes)) for _, node := range nodes { result[node.InstanceID] = nil groups[node.InstanceID] = make(map[string]bool) + claimedPorts[node.InstanceID] = make(map[config.PortClaim]string) nodeByID[node.InstanceID] = node } @@ -219,6 +231,14 @@ func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing var pending []Pending for _, service := range ordered { + if dupes := service.DuplicatePortClaims(); len(dupes) > 0 { + pending = append(pending, Pending{ + Service: service.Name, + ReasonCode: "duplicate_host_port_claims", + Message: fmt.Sprintf("service claims host port %d more than once", dupes[0].HostPort), + }) + continue + } boundNode, split := localBinding(service) if split { pending = append(pending, Pending{Service: service.Name, ReasonCode: "local_volume_binding_conflict", Message: "local volumes are retained on different nodes"}) @@ -258,8 +278,10 @@ func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing return candidates[i].InstanceID < candidates[j].InstanceID }) + claims := service.PortClaims() chosen := "" chosenService := service + portConflict := "" for _, node := range candidates { if boundNode != "" && node.InstanceID != boundNode { continue @@ -267,6 +289,14 @@ func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing if usedVCPU[node.InstanceID]+service.VCPUs > node.CapacityVCPUs || usedMem[node.InstanceID]+service.MemoryMB > node.CapacityMemMB { continue } + // All claims must fit on the same node, and the check runs before + // fitStorage so a port-rejected node commits no storage usage. + if conflict, blocked := firstPortConflict(node.InstanceID, claimedPorts[node.InstanceID], claims); blocked { + if portConflict == "" { + portConflict = conflict + } + continue + } candidateService, localDelta, sharedDelta, ok := fitStorage(service, node, reservations, usedLocal, usedShared) if !ok { continue @@ -286,12 +316,19 @@ func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing reason = "volume_capacity_unavailable" message = "no active node satisfies volume binding and capacity" } + if portConflict != "" { + reason = "host_port_conflict" + message = portConflict + } pending = append(pending, Pending{Service: service.Name, ReasonCode: reason, Message: message}) continue } result[chosen] = append(result[chosen], chosenService) usedVCPU[chosen] += service.VCPUs usedMem[chosen] += service.MemoryMB + for _, claim := range claims { + claimedPorts[chosen][claim] = service.Name + } if service.AntiAffinityGroup != "" { groups[chosen][service.AntiAffinityGroup] = true } @@ -300,6 +337,20 @@ func ScheduleWithStorage(services []config.ServiceConfig, nodes []Node, existing return result, pending } +// firstPortConflict reports the first claim already held on a node, together +// with a message naming the port, the holding service, and the node. Only the +// conflicting claim is named so the reason stays actionable without exposing +// unrelated configuration. +func firstPortConflict(node string, held map[config.PortClaim]string, claims []config.PortClaim) (string, bool) { + for _, claim := range claims { + if holder, taken := held[claim]; taken { + return fmt.Sprintf("host port %d (%s) is already claimed by service %s on node %s", + claim.HostPort, claim.Protocol, holder, node), true + } + } + return "", false +} + func localBinding(service config.ServiceConfig) (string, bool) { bound := "" for _, volume := range service.Volumes { diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 6c67ea8..ec41d9f 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -1,6 +1,7 @@ package scheduler import ( + "strings" "testing" "github.com/artemnikitin/firework/internal/config" @@ -278,3 +279,140 @@ func TestScheduleWithStorageKeepsSharedPendingUntilSafetyGate(t *testing.T) { t.Fatalf("unexpected pending result: %#v", pending) } } + +func withPorts(name string, vcpus, memMB int, hostPorts ...int) config.ServiceConfig { + service := svc(name, vcpus, memMB) + for _, hostPort := range hostPorts { + service.PortForwards = append(service.PortForwards, config.PortForward{HostPort: hostPort, VMPort: hostPort}) + } + return service +} + +func nodeOf(t *testing.T, assignment map[string][]config.ServiceConfig, service string) string { + t.Helper() + for instanceID, services := range assignment { + for _, placed := range services { + if placed.Name == service { + return instanceID + } + } + } + return "" +} + +func TestScheduleWithStorageSeparatesServicesSharingHostPort(t *testing.T) { + // i-001 is large enough that plain bin-packing would colocate both + // services there, so only the host-port claim separates them. + services := []config.ServiceConfig{ + withPorts("tenant-1-elasticsearch", 2, 512, 9200), + withPorts("tenant-2-elasticsearch", 2, 512, 9200), + } + nodes := []Node{node("i-001", 32, 16384), node("i-002", 4, 2048)} + + result, pending := ScheduleWithStorage(services, nodes, nil, StorageReservations{}) + if len(pending) != 0 { + t.Fatalf("unexpected pending services: %#v", pending) + } + if len(result["i-001"]) != 1 || len(result["i-002"]) != 1 { + t.Fatalf("expected the two claims of host port 9200 to be split across nodes, got %#v", result) + } +} + +func TestScheduleWithStorageKeepsRepeatedHostPortsOnDifferentNodes(t *testing.T) { + // The same host port on separate nodes is legitimate and must stay placed. + services := []config.ServiceConfig{ + withPorts("a", 2, 512, 9200), + withPorts("b", 2, 512, 9200), + } + nodes := []Node{node("i-001", 8, 4096), node("i-002", 8, 4096)} + existing := map[string]string{"a": "i-001", "b": "i-002"} + + result, pending := ScheduleWithStorage(services, nodes, existing, StorageReservations{}) + if len(pending) != 0 { + t.Fatalf("unexpected pending services: %#v", pending) + } + if nodeOf(t, result, "a") != "i-001" || nodeOf(t, result, "b") != "i-002" { + t.Fatalf("expected conflict-free existing placement to be preserved, got %#v", result) + } +} + +func TestScheduleWithStorageLeavesConflictingServicePending(t *testing.T) { + services := []config.ServiceConfig{ + withPorts("tenant-1-elasticsearch", 2, 512, 9200), + withPorts("tenant-2-elasticsearch", 2, 512, 9200), + } + nodes := []Node{node("i-001", 8, 4096)} + + result, pending := ScheduleWithStorage(services, nodes, nil, StorageReservations{}) + if len(result["i-001"]) != 1 { + t.Fatalf("expected exactly one service placed, got %#v", result) + } + if len(pending) != 1 || pending[0].ReasonCode != "host_port_conflict" { + t.Fatalf("unexpected pending result: %#v", pending) + } + placed := result["i-001"][0].Name + if !strings.Contains(pending[0].Message, "9200") || !strings.Contains(pending[0].Message, placed) { + t.Errorf("pending message %q does not identify the port and the holding service", pending[0].Message) + } +} + +func TestScheduleWithStorageRelocatesExistingPlacementOnNewConflict(t *testing.T) { + // Both services were colocated while they claimed different ports; the + // desired state now gives them the same one, so the existing placement is + // no longer conflict-free and must be re-evaluated. + services := []config.ServiceConfig{ + withPorts("a", 2, 512, 9200), + withPorts("b", 2, 512, 9200), + } + nodes := []Node{node("i-001", 8, 4096), node("i-002", 8, 4096)} + existing := map[string]string{"a": "i-001", "b": "i-001"} + + result, pending := ScheduleWithStorage(services, nodes, existing, StorageReservations{}) + if len(pending) != 0 { + t.Fatalf("unexpected pending services: %#v", pending) + } + if nodeOf(t, result, "a") == nodeOf(t, result, "b") { + t.Fatalf("expected the conflicting service to be relocated, got %#v", result) + } +} + +func TestScheduleWithStorageTreatsMultipleClaimsAtomically(t *testing.T) { + // i-001 is the preferred candidate for every service (most free vCPU), and + // "keeper" holds 9300 there. "mover" needs both 9200 and 9300, so it must + // take neither on i-001 and move as a unit; "later" then proves the + // rejected node kept no partial 9200 claim. + services := []config.ServiceConfig{ + withPorts("keeper", 4, 1024, 9300), + withPorts("mover", 2, 512, 9200, 9300), + withPorts("later", 1, 256, 9200), + } + nodes := []Node{node("i-001", 16, 8192), node("i-002", 8, 4096)} + existing := map[string]string{"keeper": "i-001"} + + result, pending := ScheduleWithStorage(services, nodes, existing, StorageReservations{}) + if len(pending) != 0 { + t.Fatalf("unexpected pending services: %#v", pending) + } + if nodeOf(t, result, "keeper") != "i-001" { + t.Fatalf("expected keeper to stay on i-001, got %#v", result) + } + if got := nodeOf(t, result, "mover"); got != "i-002" { + t.Fatalf("expected mover to move as a unit to i-002, got %q (%#v)", got, result) + } + if got := nodeOf(t, result, "later"); got != "i-001" { + t.Fatalf("expected 9200 to remain free on i-001, got %q (%#v)", got, result) + } +} + +func TestScheduleWithStorageRejectsSelfConflictingService(t *testing.T) { + service := withPorts("broken", 2, 512, 8080, 8080) + nodes := []Node{node("i-001", 8, 4096), node("i-002", 8, 4096)} + + result, pending := ScheduleWithStorage([]config.ServiceConfig{service}, nodes, nil, StorageReservations{}) + if len(pending) != 1 || pending[0].ReasonCode != "duplicate_host_port_claims" { + t.Fatalf("unexpected pending result: %#v", pending) + } + if len(result["i-001"]) != 0 || len(result["i-002"]) != 0 { + t.Fatalf("expected no placement for a self-conflicting service, got %#v", result) + } +}