From dbf32d390fea0a1b31b213acb3ec001e864dc14b Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Wed, 16 Sep 2026 19:56:55 +0000 Subject: [PATCH 1/3] Improve production peer health-check efficiency and live reconfiguration Spread initial and updated probe cadences by directed peer identity. Apply resolved profiles in place without losing health state, counters, or flap history; disabled associations block fallback. Default to 15s probes and document the nominal 45s failure-detection tradeoff while retaining explicit overrides. Preserve the newer routing rollback, localCIDRs, and gateway-pool peering protocol precedence. Adapt newer routing/protocol test callback signatures without changing their assertions. Regenerate the five consuming CRDs using controller-gen v0.21.0; only default documentation changes result. No simulator, e2e, or status transport/collection changes. Validated: scoped make fmt and make lint; go test for internal/net/healthcheck, cmd/unbounded-net-node, cmd/kubectl-unbounded/app/net, api/net/v1alpha1, api/machina/v1alpha3; go test -race for healthcheck, node, and net CLI; go build for affected production packages and kubectl-unbounded; repeat API generation is idempotent. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- api/net/v1alpha1/types.go | 2 + cmd/kubectl-unbounded/app/net/create.go | 4 +- cmd/kubectl-unbounded/app/net/create_test.go | 31 ++ cmd/unbounded-net-node/cni_reconcile_test.go | 3 +- .../gateway_pool_peering_protocol_test.go | 3 +- cmd/unbounded-net-node/main_update_test.go | 15 +- cmd/unbounded-net-node/peer_healthcheck.go | 76 +++- .../peer_healthcheck_test.go | 247 +++++++++++ .../reconciliation_helpers.go | 10 +- .../reconciliation_helpers_test.go | 2 + .../site_routing_reconcile_test.go | 5 +- .../site_watch_reconcile.go | 17 +- cmd/unbounded-net-node/tunnel_config.go | 13 +- cmd/unbounded-net-node/wireguard_config.go | 10 +- .../machina/crd/unbounded-cloud.io_sites.yaml | 2 + ...nbounded-cloud.io_gatewaypoolpeerings.yaml | 2 + .../net.unbounded-cloud.io_gatewaypools.yaml | 2 + ...d-cloud.io_sitegatewaypoolassignments.yaml | 2 + .../net.unbounded-cloud.io_sitepeerings.yaml | 2 + docs/net/architecture.md | 5 +- docs/net/configuration.md | 39 +- internal/net/healthcheck/manager.go | 65 ++- internal/net/healthcheck/probe_phase.go | 32 ++ internal/net/healthcheck/probe_phase_test.go | 327 +++++++++++++++ .../net/healthcheck/reconfiguration_test.go | 390 ++++++++++++++++++ internal/net/healthcheck/session.go | 195 +++++++-- internal/net/healthcheck/types.go | 30 +- 27 files changed, 1422 insertions(+), 109 deletions(-) create mode 100644 cmd/unbounded-net-node/peer_healthcheck_test.go create mode 100644 internal/net/healthcheck/probe_phase.go create mode 100644 internal/net/healthcheck/probe_phase_test.go create mode 100644 internal/net/healthcheck/reconfiguration_test.go diff --git a/api/net/v1alpha1/types.go b/api/net/v1alpha1/types.go index 04d0d731f..8882224cb 100644 --- a/api/net/v1alpha1/types.go +++ b/api/net/v1alpha1/types.go @@ -92,12 +92,14 @@ type HealthCheckSettings struct { // ReceiveInterval is the minimum interval between received health check packets. // Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + // Defaults to 15s when omitted from the selected health check scope. // +kubebuilder:validation:XIntOrString // +optional ReceiveInterval *intstr.IntOrString `json:"receiveInterval,omitempty"` // TransmitInterval is the minimum interval between transmitted health check packets. // Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + // Defaults to 15s when omitted from the selected health check scope. // +kubebuilder:validation:XIntOrString // +optional TransmitInterval *intstr.IntOrString `json:"transmitInterval,omitempty"` diff --git a/cmd/kubectl-unbounded/app/net/create.go b/cmd/kubectl-unbounded/app/net/create.go index 8a188080d..c2dd74fb9 100644 --- a/cmd/kubectl-unbounded/app/net/create.go +++ b/cmd/kubectl-unbounded/app/net/create.go @@ -36,8 +36,8 @@ type healthCheckFlags struct { func (b *healthCheckFlags) addToFlags(cmd *cobra.Command) { cmd.Flags().BoolVar(&b.enabled, "health-check-enabled", false, "Enable UDP health probes over tunnels") cmd.Flags().Int32Var(&b.detectMultiplier, "health-check-detect-multiplier", 0, "Number of missed probes before marking a peer down") - cmd.Flags().StringVar(&b.receiveInterval, "health-check-receive-interval", "", "Min interval between received probes before declaring down, e.g. 300ms") - cmd.Flags().StringVar(&b.transmitInterval, "health-check-transmit-interval", "", "Interval between transmitted health probes, e.g. 300ms") + cmd.Flags().StringVar(&b.receiveInterval, "health-check-receive-interval", "", "Min interval between received probes before declaring down, e.g. 300ms (node default: 15s)") + cmd.Flags().StringVar(&b.transmitInterval, "health-check-transmit-interval", "", "Interval between transmitted health probes, e.g. 300ms (node default: 15s)") cmd.Flags().Int32Var(&b.tunnelMTU, "tunnel-mtu", 0, "MTU for tunnel interfaces in this scope") cmd.Flags().StringVar(&b.tunnelProtocol, "tunnel-protocol", "", "Tunnel encapsulation protocol (WireGuard, GENEVE, or Auto)") _ = cmd.RegisterFlagCompletionFunc("tunnel-protocol", func(*cobra.Command, []string, string) ([]string, cobra.ShellCompDirective) { //nolint:errcheck diff --git a/cmd/kubectl-unbounded/app/net/create_test.go b/cmd/kubectl-unbounded/app/net/create_test.go index d70c48be0..563252c1c 100644 --- a/cmd/kubectl-unbounded/app/net/create_test.go +++ b/cmd/kubectl-unbounded/app/net/create_test.go @@ -7,8 +7,39 @@ import ( "bytes" "strings" "testing" + + "github.com/spf13/cobra" ) +func TestHealthCheckFlagsPreserveRuntimeDefaults(t *testing.T) { + cmd := &cobra.Command{} + flags := &healthCheckFlags{} + flags.addToFlags(cmd) + flags.selectedFrom(cmd) + + if flags.toObject() != nil { + t.Fatal("omitted health flags must preserve the node's runtime defaults") + } + + for _, name := range []string{"health-check-transmit-interval", "health-check-receive-interval"} { + flag := cmd.Flags().Lookup(name) + if flag.DefValue != "" || !strings.Contains(flag.Usage, "15s") { + t.Fatalf("flag %s must document the inherited 15s default without serializing it", name) + } + } + + if err := cmd.Flags().Set("health-check-transmit-interval", "60s"); err != nil { + t.Fatal(err) + } + + flags.selectedFrom(cmd) + + got := flags.toObject() + if len(got) != 1 || got["transmitInterval"] != "60s" { + t.Fatalf("explicit interval or partial settings changed: %v", got) + } +} + func TestCreateSiteUsesSharedSiteAPI(t *testing.T) { t.Parallel() diff --git a/cmd/unbounded-net-node/cni_reconcile_test.go b/cmd/unbounded-net-node/cni_reconcile_test.go index eb2ee2b21..54fe326f1 100644 --- a/cmd/unbounded-net-node/cni_reconcile_test.go +++ b/cmd/unbounded-net-node/cni_reconcile_test.go @@ -14,6 +14,7 @@ import ( "k8s.io/client-go/kubernetes/fake" unboundedv1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3" + "github.com/Azure/unbounded/internal/net/healthcheck" unboundednetnetlink "github.com/Azure/unbounded/internal/net/netlink" ) @@ -57,7 +58,7 @@ func TestCNIReconciliationDisablesAndRecoversWithoutMTUChange(t *testing.T) { }) ensureCNIBridgeMTUFunc = func(string, int, *unboundednetnetlink.NetlinkCache, bool) error { return nil } - configureWireGuardFunc = func(_ context.Context, _ *config, _ string, _ []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState) error { + configureWireGuardFunc = func(_ context.Context, _ *config, _ string, _ []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState, _ map[string]healthcheck.HealthCheckSettings) error { return nil } diff --git a/cmd/unbounded-net-node/gateway_pool_peering_protocol_test.go b/cmd/unbounded-net-node/gateway_pool_peering_protocol_test.go index a3111d9b9..bea4442bf 100644 --- a/cmd/unbounded-net-node/gateway_pool_peering_protocol_test.go +++ b/cmd/unbounded-net-node/gateway_pool_peering_protocol_test.go @@ -15,6 +15,7 @@ import ( unboundedv1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3" unboundednetv1alpha1 "github.com/Azure/unbounded/api/net/v1alpha1" + "github.com/Azure/unbounded/internal/net/healthcheck" unboundednetnetlink "github.com/Azure/unbounded/internal/net/netlink" ) @@ -180,7 +181,7 @@ func newPoolPeeringProtocolFixture(t *testing.T, gateway bool) *poolPeeringProto } original := configureWireGuardFunc - configureWireGuardFunc = func(_ context.Context, _ *config, _ string, mesh []meshPeerInfo, gateways []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState) error { + configureWireGuardFunc = func(_ context.Context, _ *config, _ string, mesh []meshPeerInfo, gateways []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState, _ map[string]healthcheck.HealthCheckSettings) error { f.configureCalls++ f.gateways = append([]gatewayPeerInfo(nil), gateways...) diff --git a/cmd/unbounded-net-node/main_update_test.go b/cmd/unbounded-net-node/main_update_test.go index fc88bf5ac..5bdb9c792 100644 --- a/cmd/unbounded-net-node/main_update_test.go +++ b/cmd/unbounded-net-node/main_update_test.go @@ -17,6 +17,7 @@ import ( unboundedv1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3" unboundednetv1alpha1 "github.com/Azure/unbounded/api/net/v1alpha1" + "github.com/Azure/unbounded/internal/net/healthcheck" unboundednetnetlink "github.com/Azure/unbounded/internal/net/netlink" ) @@ -140,7 +141,7 @@ func TestUpdateWireGuardFromSlices_SitePodCIDRPoolChanges(t *testing.T) { ) origConfigure := configureWireGuardFunc - configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, gatewayPeers []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, state *wireGuardState) error { + configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, gatewayPeers []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, state *wireGuardState, _ map[string]healthcheck.HealthCheckSettings) error { configureCalls++ gotPools = append([]string(nil), state.sitePodCIDRPools...) @@ -334,7 +335,7 @@ func TestUpdateWireGuardFromSlices_GatewayMeshPeersUseOnlyDirectConnectedSites(t var gotPeers []meshPeerInfo origConfigure := configureWireGuardFunc - configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState) error { + configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState, _ map[string]healthcheck.HealthCheckSettings) error { gotPeers = append([]meshPeerInfo(nil), peers...) return nil } @@ -439,7 +440,7 @@ func TestUpdateWireGuardFromSlices_ExternalGatewayIncludesAssignedNonDirectSites var gotPeers []meshPeerInfo origConfigure := configureWireGuardFunc - configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState) error { + configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState, _ map[string]healthcheck.HealthCheckSettings) error { gotPeers = append([]meshPeerInfo(nil), peers...) return nil } @@ -524,7 +525,7 @@ func TestUpdateWireGuardFromSlices_NonGatewayMeshPeersUseOnlyPeeredSites(t *test var gotPeers []meshPeerInfo origConfigure := configureWireGuardFunc - configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState) error { + configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState, _ map[string]healthcheck.HealthCheckSettings) error { gotPeers = append([]meshPeerInfo(nil), peers...) return nil } @@ -610,7 +611,7 @@ func TestUpdateWireGuardFromSlices_ManageCniPluginFalseSkipsPodCIDRRoutesForSame var gotGatewayPeers []gatewayPeerInfo origConfigure := configureWireGuardFunc - configureWireGuardFunc = func(_ context.Context, _ *config, _ string, _ []meshPeerInfo, gatewayPeers []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState) error { + configureWireGuardFunc = func(_ context.Context, _ *config, _ string, _ []meshPeerInfo, gatewayPeers []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState, _ map[string]healthcheck.HealthCheckSettings) error { gotGatewayPeers = append([]gatewayPeerInfo(nil), gatewayPeers...) return nil } @@ -697,7 +698,7 @@ func TestUpdateWireGuardFromSlices_ManageCniPluginFalseSkipsPodCIDRRoutesForSame var gotPeers []meshPeerInfo origConfigure := configureWireGuardFunc - configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState) error { + configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState, _ map[string]healthcheck.HealthCheckSettings) error { gotPeers = append([]meshPeerInfo(nil), peers...) return nil } @@ -791,7 +792,7 @@ func TestUpdateWireGuardFromSlices_ManageCniPluginFalseKeepsRemotePeeredMeshPeer var gotPeers []meshPeerInfo origConfigure := configureWireGuardFunc - configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState) error { + configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, _ []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState, _ map[string]healthcheck.HealthCheckSettings) error { gotPeers = append([]meshPeerInfo(nil), peers...) return nil } diff --git a/cmd/unbounded-net-node/peer_healthcheck.go b/cmd/unbounded-net-node/peer_healthcheck.go index f12110122..43e6c3b0d 100644 --- a/cmd/unbounded-net-node/peer_healthcheck.go +++ b/cmd/unbounded-net-node/peer_healthcheck.go @@ -4,13 +4,38 @@ package main import ( + "errors" + "fmt" "net" + "time" "k8s.io/klog/v2" "github.com/Azure/unbounded/internal/net/healthcheck" ) +// A disabled association must block fallback to less-specific enabled profiles. +const disabledHealthCheckProfile = "disabled" + +var errRegisterHealthChecks = errors.New("health check registration failed") + +func resolvedHealthCheckSettings(name string, profiles map[string]healthcheck.HealthCheckSettings, maxBackoff time.Duration) (healthcheck.HealthCheckSettings, bool, error) { + if name == "" || name == disabledHealthCheckProfile { + return healthcheck.HealthCheckSettings{}, false, nil + } + + settings, ok := profiles[name] + if !ok { + return healthcheck.HealthCheckSettings{}, false, fmt.Errorf("health check profile %q is missing from the current reconciliation", name) + } + + if maxBackoff > 0 { + settings.MaxBackoff = maxBackoff + } + + return settings, true, nil +} + // registerPeersWithHealthCheck registers mesh and gateway peers with the // healthcheck manager, resolving HC profiles for each peer. It sets // state.meshPeerHealthCheckEnabled and state.gatewayPeerHealthCheckEnabled @@ -24,7 +49,8 @@ import ( // when no pool/assignment-level profile is found for a gateway peer. This is // used by GENEVE which has no WireGuard handshake as a liveness signal. // -// Returns the set of peer names that were registered (desiredHCPeers). +// Returns desired peer names and registration errors. On error, names are retained +// so a failed configuration does not remove an existing healthy session. func registerPeersWithHealthCheck( meshPeers []meshPeerInfo, gatewayPeers []gatewayPeerInfo, @@ -35,15 +61,18 @@ func registerPeersWithHealthCheck( assignmentSiteHCProfileNames map[string]string, assignmentPoolHCProfileNames map[string]string, poolHCProfileNames map[string]string, + profiles map[string]healthcheck.HealthCheckSettings, state *wireGuardState, peerIfaceNameFn func(gatewayPeerInfo) string, useSiteFallbackForGateway bool, -) map[string]bool { +) (map[string]bool, error) { desiredHCPeers := make(map[string]bool) if state.healthCheckManager == nil { - return desiredHCPeers + return desiredHCPeers, nil } + var registrationErrors []error + // Mesh peers. for _, peer := range meshPeers { overlayIP := getHealthIPFromPodCIDRs(peer.PodCIDRs) @@ -53,7 +82,16 @@ func registerPeersWithHealthCheck( hcProfileName := resolveMeshPeerHealthCheckProfileName(isGatewayNode, peer, mySiteName, siteHCProfileNames, peeringHCProfileNames, assignmentSiteHCProfileNames) - if hcProfileName == "" { + + settings, enabled, err := resolvedHealthCheckSettings(hcProfileName, profiles, state.healthFlapMaxBackoff) + if err != nil { + desiredHCPeers[peer.Name] = true + registrationErrors = append(registrationErrors, fmt.Errorf("mesh peer %s: %w", peer.Name, err)) + + continue + } + + if !enabled { continue } @@ -64,13 +102,8 @@ func registerPeersWithHealthCheck( state.mu.Unlock() } - settings := healthcheck.DefaultSettings() - if state.healthFlapMaxBackoff > 0 { - settings.MaxBackoff = state.healthFlapMaxBackoff - } - if err := state.healthCheckManager.AddPeer(peer.Name, net.ParseIP(overlayIP), settings); err != nil { - klog.V(2).Infof("Healthcheck: failed to register mesh peer %s at %s: %v", peer.Name, overlayIP, err) + registrationErrors = append(registrationErrors, fmt.Errorf("register mesh peer %s at %s: %w", peer.Name, overlayIP, err)) } else { klog.V(4).Infof("Healthcheck: registered mesh peer %s at %s", peer.Name, overlayIP) } @@ -94,7 +127,15 @@ func registerPeersWithHealthCheck( hcProfileName = siteHCProfileNames[mySiteName] } - if hcProfileName == "" { + settings, enabled, err := resolvedHealthCheckSettings(hcProfileName, profiles, state.healthFlapMaxBackoff) + if err != nil { + desiredHCPeers[gwPeer.Name] = true + registrationErrors = append(registrationErrors, fmt.Errorf("gateway peer %s: %w", gwPeer.Name, err)) + + continue + } + + if !enabled { continue } @@ -104,19 +145,18 @@ func registerPeersWithHealthCheck( state.gatewayPeerHealthCheckEnabled[ifName] = true state.mu.Unlock() - settings := healthcheck.DefaultSettings() - if state.healthFlapMaxBackoff > 0 { - settings.MaxBackoff = state.healthFlapMaxBackoff - } - if err := state.healthCheckManager.AddPeer(gwPeer.Name, net.ParseIP(overlayIP), settings); err != nil { - klog.V(2).Infof("Healthcheck: failed to register gateway peer %s at %s: %v", gwPeer.Name, overlayIP, err) + registrationErrors = append(registrationErrors, fmt.Errorf("register gateway peer %s at %s: %w", gwPeer.Name, overlayIP, err)) } else { klog.V(4).Infof("Healthcheck: registered gateway peer %s at %s (iface %s)", gwPeer.Name, overlayIP, ifName) } } - return desiredHCPeers + if len(registrationErrors) > 0 { + return desiredHCPeers, fmt.Errorf("%w: %w", errRegisterHealthChecks, errors.Join(registrationErrors...)) + } + + return desiredHCPeers, nil } // peerIfaceNameWireGuard maps a gateway peer to its WireGuard interface name diff --git a/cmd/unbounded-net-node/peer_healthcheck_test.go b/cmd/unbounded-net-node/peer_healthcheck_test.go new file mode 100644 index 000000000..9ced587c6 --- /dev/null +++ b/cmd/unbounded-net-node/peer_healthcheck_test.go @@ -0,0 +1,247 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "context" + "errors" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/client-go/kubernetes/fake" + + unboundedv1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3" + unboundednetv1alpha1 "github.com/Azure/unbounded/api/net/v1alpha1" + "github.com/Azure/unbounded/internal/net/healthcheck" + unboundednetnetlink "github.com/Azure/unbounded/internal/net/netlink" +) + +func TestRegisterHealthProfilesAllTunnelModes(t *testing.T) { + for _, protocol := range []string{"WireGuard", "GENEVE", "VXLAN", "IPIP", "None"} { + for _, tc := range []struct { + name string + gatewayPeer, gatewayNode bool + override, site, peering, assignmentSite, assignmentLocal, assignmentRemote, pool, want string + }{ + {name: "mesh default", site: "site", want: "site"}, + {name: "mesh peering", site: "site", peering: "peering", want: "peering"}, + {name: "mesh assignment", site: "site", peering: "peering", assignmentSite: "assignment", want: "assignment"}, + {name: "mesh gateway role", gatewayNode: true, site: "site", peering: "peering", assignmentSite: "assignment", want: "assignment"}, + {name: "mesh explicit pool", override: "pool", assignmentSite: "assignment", want: "pool"}, + {name: "mesh explicit pool peering", override: "pool-peering", assignmentSite: "assignment", want: "pool-peering"}, + {name: "mesh disabled assignment", site: "site", assignmentSite: disabledHealthCheckProfile}, + {name: "mesh disabled override", override: disabledHealthCheckProfile, site: "site"}, + {name: "gateway assignment", gatewayPeer: true, assignmentLocal: "assignment", assignmentRemote: "remote", pool: "pool", want: "assignment"}, + {name: "gateway pool", gatewayPeer: true, gatewayNode: true, assignmentRemote: "remote", pool: "pool", want: "pool"}, + {name: "gateway remote assignment fallback", gatewayPeer: true, gatewayNode: true, assignmentRemote: "remote", want: "remote"}, + {name: "gateway explicit", gatewayPeer: true, override: "pool-peering", assignmentLocal: "assignment", want: "pool-peering"}, + {name: "gateway disabled assignment", gatewayPeer: true, site: "site", assignmentLocal: disabledHealthCheckProfile}, + {name: "gateway disabled pool", gatewayPeer: true, gatewayNode: true, site: "site", pool: disabledHealthCheckProfile, assignmentRemote: "remote"}, + {name: "gateway disabled override", gatewayPeer: true, override: disabledHealthCheckProfile, site: "site", assignmentLocal: "assignment"}, + } { + t.Run(protocol+"/"+tc.name, func(t *testing.T) { + manager, err := healthcheck.NewManager("local", 0, nil) + if err != nil { + t.Fatal(err) + } + defer manager.Stop() + + profiles := make(map[string]healthcheck.HealthCheckSettings) + + for i, name := range []string{"site", "peering", "assignment", "remote", "pool", "pool-peering"} { + _, profile := healthCheckProfileFromSettings(&unboundednetv1alpha1.HealthCheckSettings{ + TransmitInterval: ptrIntOrString(intstr.FromInt(1000 + i)), + }, name) + profiles[name] = profile + } + + state := &wireGuardState{ + healthCheckManager: manager, healthFlapMaxBackoff: 300 * time.Second, + healthCheckProfiles: map[string]healthcheck.HealthCheckSettings{"site": {TransmitInterval: time.Millisecond}}, + meshPeerHealthCheckEnabled: make(map[string]bool), gatewayPeerHealthCheckEnabled: make(map[string]bool), + } + + var ( + mesh []meshPeerInfo + gateways []gatewayPeerInfo + ) + if tc.gatewayPeer { + gateways = []gatewayPeerInfo{{Name: "peer", SiteName: "remote", PoolName: "pool", PodCIDRs: []string{"10.244.1.0/24"}, HealthCheckProfileName: tc.override, TunnelProtocol: protocol}} + } else { + mesh = []meshPeerInfo{{Name: "peer", SiteName: "remote", WireGuardPublicKey: "pub", PodCIDRs: []string{"10.244.1.0/24"}, HealthCheckProfileName: tc.override, TunnelProtocol: protocol}} + } + + desired, err := registerPeersWithHealthCheck(mesh, gateways, "local", tc.gatewayNode, + map[string]string{"local": tc.site, "remote": tc.site}, map[string]string{"remote": tc.peering}, + map[string]string{"remote": tc.assignmentSite}, map[string]string{"local|pool": tc.assignmentLocal, "remote|pool": tc.assignmentRemote}, + map[string]string{"pool": tc.pool}, profiles, state, func(gatewayPeerInfo) string { return "iface" }, protocol != "WireGuard") + if err != nil { + t.Fatal(err) + } + + if tc.want == "" { + if desired["peer"] || len(manager.GetAllPeerStatuses()) != 0 || state.meshPeerHealthCheckEnabled["pub"] || state.gatewayPeerHealthCheckEnabled["iface"] { + t.Fatal("disabled profile fell through to enabled lower-precedence profile") + } + + return + } + + want := profiles[tc.want] + want.MaxBackoff = 300 * time.Second + + got, err := manager.GetPeerSettings("peer") + if err != nil || got != want || !desired["peer"] { + t.Fatalf("got %+v, %v; want %+v", got, err, want) + } + + if got.ReceiveInterval != 15*time.Second { + t.Fatal("partial profile lost 15s receive default") + } + }) + } + } +} + +func TestRegisterHealthProfileFallbackAndMissing(t *testing.T) { + for _, siteFallback := range []bool{false, true} { + manager, err := healthcheck.NewManager("local", 0, nil) + if err != nil { + t.Fatal(err) + } + + state := &wireGuardState{healthCheckManager: manager, meshPeerHealthCheckEnabled: make(map[string]bool), gatewayPeerHealthCheckEnabled: make(map[string]bool)} + gateways := []gatewayPeerInfo{{Name: "peer", PoolName: "pool", PodCIDRs: []string{"10.244.1.0/24"}}} + + desired, err := registerPeersWithHealthCheck(nil, gateways, "local", false, map[string]string{"local": "site"}, nil, nil, nil, nil, + map[string]healthcheck.HealthCheckSettings{"site": healthcheck.DefaultSettings()}, state, func(gatewayPeerInfo) string { return "iface" }, siteFallback) + if err != nil || desired["peer"] != siteFallback { + t.Fatalf("site fallback %t: desired=%v err=%v", siteFallback, desired, err) + } + + gateways[0].HealthCheckProfileName = "missing" + + desired, err = registerPeersWithHealthCheck(nil, gateways, "local", false, nil, nil, nil, nil, nil, nil, state, func(gatewayPeerInfo) string { return "iface" }, siteFallback) + if !errors.Is(err, errRegisterHealthChecks) || !desired["peer"] { + t.Fatal("missing fresh profile must fail without deleting existing session") + } + + manager.Stop() + } +} + +func TestHealthProfilePartialUpdateResetsUnspecifiedValues(t *testing.T) { + manager, err := healthcheck.NewManager("local", 0, nil) + if err != nil { + t.Fatal(err) + } + defer manager.Stop() + + state := &wireGuardState{healthCheckManager: manager, healthFlapMaxBackoff: 240 * time.Second, meshPeerHealthCheckEnabled: make(map[string]bool)} + peer := meshPeerInfo{Name: "peer", SiteName: "local", WireGuardPublicKey: "pub", PodCIDRs: []string{"10.244.1.0/24"}} + profile := healthcheck.DefaultSettings() + profile.TransmitInterval, profile.ReceiveInterval = time.Second, time.Second + profiles := map[string]healthcheck.HealthCheckSettings{"site": profile} + register := func() { + t.Helper() + + if _, err := registerPeersWithHealthCheck([]meshPeerInfo{peer}, nil, "local", false, map[string]string{"local": "site"}, nil, nil, nil, nil, profiles, state, nil, false); err != nil { + t.Fatal(err) + } + } + register() + + _, profiles["site"] = healthCheckProfileFromSettings(&unboundednetv1alpha1.HealthCheckSettings{ + TransmitInterval: ptrIntOrString(intstr.FromString("60s")), + }, "site") + + register() + + got, err := manager.GetPeerSettings("peer") + if err != nil || got.TransmitInterval != 60*time.Second || got.ReceiveInterval != 15*time.Second || got.MaxBackoff != 240*time.Second { + t.Fatalf("partial update reused stale values: %+v %v", got, err) + } +} + +func TestDisabledAssignmentBlocksLowerPrecedence(t *testing.T) { + assignment := unboundednetv1alpha1.SiteGatewayPoolAssignment{ + ObjectMeta: metav1.ObjectMeta{Name: "assignment"}, + Spec: unboundednetv1alpha1.SiteGatewayPoolAssignmentSpec{ + Sites: []string{"local", "remote"}, GatewayPools: []string{"pool"}, + HealthCheckSettings: &unboundednetv1alpha1.HealthCheckSettings{Enabled: ptrBool(false)}, + }, + } + profiles := make(map[string]healthcheck.HealthCheckSettings) + pools, sites := make(map[string]string), make(map[string]string) + mergeAssignmentHealthCheckState(assignment, "local", nil, profiles, nil, pools, make(map[string]string), sites, make(map[string]string)) + + if len(profiles) != 0 || pools["local|pool"] != disabledHealthCheckProfile || sites["remote"] != disabledHealthCheckProfile { + t.Fatalf("disabled association was discarded: pools=%v sites=%v", pools, sites) + } +} + +func TestReconciliationPassesFreshHealthProfilesBeforeStateCommit(t *testing.T) { + manager, err := healthcheck.NewManager("node-self", 0, nil) + if err != nil { + t.Fatal(err) + } + defer manager.Stop() + + site := &unboundedv1alpha3.Site{ + ObjectMeta: metav1.ObjectMeta{Name: "site"}, + Spec: unboundedv1alpha3.SiteSpec{HealthCheckSettings: &unboundednetv1alpha1.HealthCheckSettings{ + TransmitInterval: ptrIntOrString(intstr.FromString("60s")), + }}, + } + siteInformer := newInformerWithObjects(toUnstructured(t, site)) + slices := newInformerWithObjects(toUnstructured(t, &unboundednetv1alpha1.SiteNodeSlice{ + ObjectMeta: metav1.ObjectMeta{Name: "slice"}, SiteName: "site", + Nodes: []unboundednetv1alpha1.NodeInfo{{Name: "peer", WireGuardPublicKey: "pub-peer", InternalIPs: []string{"10.0.0.2"}, PodCIDRs: []string{"10.244.1.0/24"}}}, + })) + stale := healthcheck.DefaultSettings() + stale.TransmitInterval = time.Second + state := &wireGuardState{ + clientset: fake.NewClientset(&corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node-self"}}), + nodeName: "node-self", healthCheckManager: manager, + healthCheckProfiles: map[string]healthcheck.HealthCheckSettings{"s-site": stale}, + } + + original := configureWireGuardFunc + defer func() { configureWireGuardFunc = original }() + + called := false + configureWireGuardFunc = func(_ context.Context, _ *config, _ string, peers []meshPeerInfo, gateways []gatewayPeerInfo, siteName string, _, _, _ map[string]bool, + siteNames, peeringNames, assignmentSites, assignmentPools, poolNames map[string]string, _, _, _, _, _ map[string]int, + _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, s *wireGuardState, profiles map[string]healthcheck.HealthCheckSettings, + ) error { + called = true + + if s.healthCheckProfiles["s-site"] != stale { + t.Fatal("test must observe uncommitted state") + } + + if profiles["s-site"].TransmitInterval != 60*time.Second { + t.Fatal("fresh profile was not passed") + } + + _, err := registerPeersWithHealthCheck(peers, gateways, siteName, false, siteNames, peeringNames, assignmentSites, assignmentPools, poolNames, profiles, s, func(gatewayPeerInfo) string { return "" }, false) + + return err + } + empty := newInformerWithObjects() + + err = updateWireGuardFromSlices(context.Background(), nil, siteInformer, slices, empty, empty, empty, empty, empty, + &config{NodeName: "node-self", WireGuardPort: 51820}, "site", "private", "pub-self", true, state) + if err != nil || !called { + t.Fatalf("reconciliation: called=%t err=%v", called, err) + } + + got, err := manager.GetPeerSettings("peer") + if err != nil || got.TransmitInterval != 60*time.Second || got.ReceiveInterval != 15*time.Second { + t.Fatalf("registration used stale state: %+v %v", got, err) + } +} diff --git a/cmd/unbounded-net-node/reconciliation_helpers.go b/cmd/unbounded-net-node/reconciliation_helpers.go index 0d9f5381f..091872623 100644 --- a/cmd/unbounded-net-node/reconciliation_helpers.go +++ b/cmd/unbounded-net-node/reconciliation_helpers.go @@ -104,9 +104,7 @@ func healthCheckProfileNameForGatewayPoolPeering(name string) string { } func healthCheckProfilesEqual(a, b healthcheck.HealthCheckSettings) bool { - return a.DetectMultiplier == b.DetectMultiplier && - a.ReceiveInterval == b.ReceiveInterval && - a.TransmitInterval == b.TransmitInterval + return a == b } // mergeAssignmentHealthCheckState merges SiteGatewayPoolAssignment health check settings into active maps. @@ -122,7 +120,7 @@ func mergeAssignmentHealthCheckState( assignmentSiteHealthCheckProfileNames map[string]string, assignmentSiteHealthCheckSourceAssignment map[string]string, ) { - assignmentHealthCheckProfileName := "" + assignmentHealthCheckProfileName := disabledHealthCheckProfile assignmentScope := healthCheckLogScope(siteGatewayPoolAssignmentGVR, assignment.Name) if enabled, profile := healthCheckProfileFromSettings(assignment.Spec.HealthCheckSettings, assignmentScope); enabled { @@ -137,10 +135,6 @@ func mergeAssignmentHealthCheckState( } } - if assignmentHealthCheckProfileName == "" { - return - } - for _, poolName := range assignment.Spec.GatewayPools { poolName = strings.TrimSpace(poolName) if poolName == "" { diff --git a/cmd/unbounded-net-node/reconciliation_helpers_test.go b/cmd/unbounded-net-node/reconciliation_helpers_test.go index 0bab1020b..617a69d84 100644 --- a/cmd/unbounded-net-node/reconciliation_helpers_test.go +++ b/cmd/unbounded-net-node/reconciliation_helpers_test.go @@ -72,6 +72,7 @@ func TestHealthCheckProfileSettingsHelpers(t *testing.T) { } want := healthcheck.HealthCheckSettings{ + MaxBackoff: 120 * time.Second, DetectMultiplier: 5, ReceiveInterval: 150 * time.Millisecond, TransmitInterval: 275 * time.Millisecond, @@ -99,6 +100,7 @@ func TestHealthCheckProfileSettingsHelpers(t *testing.T) { } siteWant := healthcheck.HealthCheckSettings{ + MaxBackoff: 120 * time.Second, DetectMultiplier: 7, ReceiveInterval: 200 * time.Millisecond, TransmitInterval: 400 * time.Millisecond, diff --git a/cmd/unbounded-net-node/site_routing_reconcile_test.go b/cmd/unbounded-net-node/site_routing_reconcile_test.go index 0acbe6696..12a4b17c4 100644 --- a/cmd/unbounded-net-node/site_routing_reconcile_test.go +++ b/cmd/unbounded-net-node/site_routing_reconcile_test.go @@ -15,6 +15,7 @@ import ( unboundedv1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3" unboundednetv1alpha1 "github.com/Azure/unbounded/api/net/v1alpha1" + "github.com/Azure/unbounded/internal/net/healthcheck" unboundednetnetlink "github.com/Azure/unbounded/internal/net/netlink" ) @@ -49,7 +50,7 @@ func TestUpdateWireGuardFromSlices_LocalGatewayExclusions(t *testing.T) { var gotGatewayPeers []gatewayPeerInfo original := configureWireGuardFunc - configureWireGuardFunc = func(_ context.Context, _ *config, _ string, _ []meshPeerInfo, gatewayPeers []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState) error { + configureWireGuardFunc = func(_ context.Context, _ *config, _ string, _ []meshPeerInfo, gatewayPeers []gatewayPeerInfo, _ string, _, _, _ map[string]bool, _, _, _, _, _ map[string]string, _, _, _, _, _ map[string]int, _ []unboundednetnetlink.DesiredRoute, _ map[string]bool, _ *wireGuardState, _ map[string]healthcheck.HealthCheckSettings) error { calls++ gotGatewayPeers = gatewayPeers @@ -208,7 +209,7 @@ func TestUpdateWireGuardFromSlices_GatewayRoutingChanges(t *testing.T) { var configureErr error original := configureWireGuardFunc - configureWireGuardFunc = func(context.Context, *config, string, []meshPeerInfo, []gatewayPeerInfo, string, map[string]bool, map[string]bool, map[string]bool, map[string]string, map[string]string, map[string]string, map[string]string, map[string]string, map[string]int, map[string]int, map[string]int, map[string]int, map[string]int, []unboundednetnetlink.DesiredRoute, map[string]bool, *wireGuardState) error { + configureWireGuardFunc = func(context.Context, *config, string, []meshPeerInfo, []gatewayPeerInfo, string, map[string]bool, map[string]bool, map[string]bool, map[string]string, map[string]string, map[string]string, map[string]string, map[string]string, map[string]int, map[string]int, map[string]int, map[string]int, map[string]int, []unboundednetnetlink.DesiredRoute, map[string]bool, *wireGuardState, map[string]healthcheck.HealthCheckSettings) error { calls++ return configureErr } diff --git a/cmd/unbounded-net-node/site_watch_reconcile.go b/cmd/unbounded-net-node/site_watch_reconcile.go index 6a62a0aa5..8abc5ff5c 100644 --- a/cmd/unbounded-net-node/site_watch_reconcile.go +++ b/cmd/unbounded-net-node/site_watch_reconcile.go @@ -989,6 +989,7 @@ func updateWireGuardFromSlices(ctx context.Context, dynamicClient dynamic.Interf enabled, profile := healthCheckProfileFromSettings(site.Spec.HealthCheckSettings, siteScope) if !enabled { + siteHealthCheckProfileNames[siteName] = disabledHealthCheckProfile continue } @@ -1109,7 +1110,7 @@ func updateWireGuardFromSlices(ctx context.Context, dynamicClient dynamic.Interf // If our site is in this peering, add all other sites if mySiteInPeering { remoteSites := make([]string, 0, len(peering.Spec.Sites)) - peeringHealthCheckProfileName := "" + peeringHealthCheckProfileName := disabledHealthCheckProfile peeringScope := healthCheckLogScope(sitePeeringGVR, peering.Name) if enabled, profile := healthCheckProfileFromSettings(peering.Spec.HealthCheckSettings, peeringScope); enabled { @@ -1287,6 +1288,8 @@ func updateWireGuardFromSlices(ctx context.Context, dynamicClient dynamic.Interf poolHealthCheckProfileNames[pool.Name] = profileName healthCheckProfileSources[profileName] = poolScope } + } else { + poolHealthCheckProfileNames[pool.Name] = disabledHealthCheckProfile } // Collect pool-level tunnelMTU override. if v := tunnelMTUFromSpec(pool.Spec.TunnelMTU); v > 0 { @@ -1407,7 +1410,7 @@ func updateWireGuardFromSlices(ctx context.Context, dynamicClient dynamic.Interf continue } - peeringHealthCheckProfileName := "" + peeringHealthCheckProfileName := disabledHealthCheckProfile peeringScope := healthCheckLogScope(gatewayPoolPeeringGVR, peering.Name) if enabled, profile := healthCheckProfileFromSettings(peering.Spec.HealthCheckSettings, peeringScope); enabled { @@ -2100,7 +2103,7 @@ func updateWireGuardFromSlices(ctx context.Context, dynamicClient dynamic.Interf assignmentSiteHealthCheckProfileNames, assignmentPoolHealthCheckProfileNames, poolHealthCheckProfileNames, siteTunnelMTUs, peeringSiteTunnelMTUs, assignmentSiteTunnelMTUs, - assignmentPoolTunnelMTUs, poolTunnelMTUs, state) + assignmentPoolTunnelMTUs, poolTunnelMTUs, state, healthCheckProfiles) if sharedTunnelErr != nil { klog.Warningf("Tunnel configuration failed (WireGuard will still be configured): %v", sharedTunnelErr) } @@ -2133,11 +2136,15 @@ func updateWireGuardFromSlices(ctx context.Context, dynamicClient dynamic.Interf // Configure WireGuard with WG peers, merging tunnel routes into // the unified route manager's SyncRoutes call. - if err := configureWireGuardFunc(ctx, cfg, privKey, wgMeshPeers, wgGatewayPeers, mySiteName, peeredSites, networkPeeredSites, gatewayNodePubKeys, siteHealthCheckProfileNames, peeringSiteHealthCheckProfileNames, assignmentSiteHealthCheckProfileNames, assignmentPoolHealthCheckProfileNames, poolHealthCheckProfileNames, siteTunnelMTUs, peeringSiteTunnelMTUs, assignmentSiteTunnelMTUs, assignmentPoolTunnelMTUs, poolTunnelMTUs, tunnelRoutes, tunnelHCPeers, state); err != nil { + if err := configureWireGuardFunc(ctx, cfg, privKey, wgMeshPeers, wgGatewayPeers, mySiteName, peeredSites, networkPeeredSites, gatewayNodePubKeys, siteHealthCheckProfileNames, peeringSiteHealthCheckProfileNames, assignmentSiteHealthCheckProfileNames, assignmentPoolHealthCheckProfileNames, poolHealthCheckProfileNames, siteTunnelMTUs, peeringSiteTunnelMTUs, assignmentSiteTunnelMTUs, assignmentPoolTunnelMTUs, poolTunnelMTUs, tunnelRoutes, tunnelHCPeers, state, healthCheckProfiles); err != nil { return err } - if sharedTunnelErr != nil && fabricMTUIncreased { + if sharedTunnelErr != nil && (fabricMTUIncreased || errors.Is(sharedTunnelErr, errRegisterHealthChecks)) { + if errors.Is(sharedTunnelErr, errRegisterHealthChecks) { + return sharedTunnelErr + } + return fmt.Errorf("cannot raise fabric MTU while tunnel reconciliation is incomplete: %w", sharedTunnelErr) } diff --git a/cmd/unbounded-net-node/tunnel_config.go b/cmd/unbounded-net-node/tunnel_config.go index c390ec924..8e0551479 100644 --- a/cmd/unbounded-net-node/tunnel_config.go +++ b/cmd/unbounded-net-node/tunnel_config.go @@ -16,6 +16,7 @@ import ( unboundednetv1alpha1 "github.com/Azure/unbounded/api/net/v1alpha1" ebpfpkg "github.com/Azure/unbounded/internal/net/ebpf" + "github.com/Azure/unbounded/internal/net/healthcheck" unboundednetnetlink "github.com/Azure/unbounded/internal/net/netlink" ) @@ -204,6 +205,7 @@ func configureTunnelPeers( assignmentPoolTunnelMTUs map[string]int, poolTunnelMTUs map[string]int, state *wireGuardState, + profiles map[string]healthcheck.HealthCheckSettings, ) ([]unboundednetnetlink.DesiredRoute, map[string]bool, error) { // Do NOT early-return when both peer lists are empty. Even when // there are no tunnel-protocol peers (e.g. WG-only gateway @@ -518,12 +520,19 @@ func configureTunnelPeers( // TC egress BPF intercepts and redirects to geneve0. var routes []unboundednetnetlink.DesiredRoute - hcPeers := registerPeersWithHealthCheck(meshPeers, gatewayPeers, mySiteName, false, + state.mu.Lock() + isGatewayNode := state.isGatewayNode + state.mu.Unlock() + + hcPeers, healthErr := registerPeersWithHealthCheck(meshPeers, gatewayPeers, mySiteName, isGatewayNode, siteHealthCheckProfileNames, peeringSiteHealthCheckProfileNames, assignmentSiteHealthCheckProfileNames, assignmentPoolHealthCheckProfileNames, - poolHealthCheckProfileNames, state, + poolHealthCheckProfileNames, profiles, state, func(gw gatewayPeerInfo) string { return peerIfaceName(cfg, gw) }, true) + if healthErr != nil { + return routes, hcPeers, fmt.Errorf("register shared-tunnel health checks: %w", healthErr) + } klog.V(2).Infof("eBPF tunnel: configured %d mesh + %d gateway peers, %d BPF entries, %d supernet routes on %s", len(meshPeers), len(gatewayPeers), len(bpfEntries), len(routes), ifName) diff --git a/cmd/unbounded-net-node/wireguard_config.go b/cmd/unbounded-net-node/wireguard_config.go index bf9dbcde6..4c3e486b5 100644 --- a/cmd/unbounded-net-node/wireguard_config.go +++ b/cmd/unbounded-net-node/wireguard_config.go @@ -12,6 +12,7 @@ import ( "k8s.io/klog/v2" unboundednetv1alpha1 "github.com/Azure/unbounded/api/net/v1alpha1" + "github.com/Azure/unbounded/internal/net/healthcheck" unboundednetnetlink "github.com/Azure/unbounded/internal/net/netlink" "github.com/Azure/unbounded/internal/net/routeplan" ) @@ -20,7 +21,7 @@ import ( // - wg: Main mesh interface for all mesh peers (intra-site, remote, same-pool gateways) // - wg: Separate interfaces for each gateway peer (for ECMP routing) // Endpoint and routing decisions are driven by peer.SiteName and peeredSites membership. -func configureWireGuard(ctx context.Context, cfg *config, privKey string, peers []meshPeerInfo, gatewayPeers []gatewayPeerInfo, mySiteName string, peeredSites, networkPeeredSites, gatewayNodePubKeys map[string]bool, siteHealthCheckProfileNames, peeringSiteHealthCheckProfileNames, assignmentSiteHealthCheckProfileNames, assignmentPoolHealthCheckProfileNames, poolHealthCheckProfileNames map[string]string, siteTunnelMTUs, peeringSiteTunnelMTUs, assignmentSiteTunnelMTUs, assignmentPoolTunnelMTUs, poolTunnelMTUs map[string]int, additionalRoutes []unboundednetnetlink.DesiredRoute, geneveHCPeers map[string]bool, state *wireGuardState) error { +func configureWireGuard(ctx context.Context, cfg *config, privKey string, peers []meshPeerInfo, gatewayPeers []gatewayPeerInfo, mySiteName string, peeredSites, networkPeeredSites, gatewayNodePubKeys map[string]bool, siteHealthCheckProfileNames, peeringSiteHealthCheckProfileNames, assignmentSiteHealthCheckProfileNames, assignmentPoolHealthCheckProfileNames, poolHealthCheckProfileNames map[string]string, siteTunnelMTUs, peeringSiteTunnelMTUs, assignmentSiteTunnelMTUs, assignmentPoolTunnelMTUs, poolTunnelMTUs map[string]int, additionalRoutes []unboundednetnetlink.DesiredRoute, geneveHCPeers map[string]bool, state *wireGuardState, profiles map[string]healthcheck.HealthCheckSettings) error { nodePodCIDRs := state.nodePodCIDRs isGatewayNode := state.isGatewayNode myGatewayPort := state.myGatewayPort @@ -619,12 +620,15 @@ func configureWireGuard(ctx context.Context, cfg *config, privKey string, peers len(peers), len(gatewayPeers), len(allDesiredRoutes)) // === Register healthcheck peers via shared HC registration === - wgHCPeers := registerPeersWithHealthCheck(peers, gatewayPeers, mySiteName, isGatewayNode, + wgHCPeers, err := registerPeersWithHealthCheck(peers, gatewayPeers, mySiteName, isGatewayNode, siteHealthCheckProfileNames, peeringSiteHealthCheckProfileNames, assignmentSiteHealthCheckProfileNames, assignmentPoolHealthCheckProfileNames, - poolHealthCheckProfileNames, state, + poolHealthCheckProfileNames, profiles, state, func(gw gatewayPeerInfo) string { return peerIfaceNameWireGuard(cfg, gw) }, false) + if err != nil { + return fmt.Errorf("register WireGuard health checks: %w", err) + } // Remove peers that are no longer desired (preserve GENEVE HC peers) if state.healthCheckManager != nil { diff --git a/deploy/machina/crd/unbounded-cloud.io_sites.yaml b/deploy/machina/crd/unbounded-cloud.io_sites.yaml index dcb2d2897..4161a7b45 100644 --- a/deploy/machina/crd/unbounded-cloud.io_sites.yaml +++ b/deploy/machina/crd/unbounded-cloud.io_sites.yaml @@ -174,6 +174,7 @@ spec: description: |- ReceiveInterval is the minimum interval between received health check packets. Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + Defaults to 15s when omitted from the selected health check scope. x-kubernetes-int-or-string: true transmitInterval: anyOf: @@ -182,6 +183,7 @@ spec: description: |- TransmitInterval is the minimum interval between transmitted health check packets. Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + Defaults to 15s when omitted from the selected health check scope. x-kubernetes-int-or-string: true type: object localCidrs: diff --git a/deploy/net/crd/net.unbounded-cloud.io_gatewaypoolpeerings.yaml b/deploy/net/crd/net.unbounded-cloud.io_gatewaypoolpeerings.yaml index cdfcdf1dd..20c215143 100644 --- a/deploy/net/crd/net.unbounded-cloud.io_gatewaypoolpeerings.yaml +++ b/deploy/net/crd/net.unbounded-cloud.io_gatewaypoolpeerings.yaml @@ -82,6 +82,7 @@ spec: description: |- ReceiveInterval is the minimum interval between received health check packets. Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + Defaults to 15s when omitted from the selected health check scope. x-kubernetes-int-or-string: true transmitInterval: anyOf: @@ -90,6 +91,7 @@ spec: description: |- TransmitInterval is the minimum interval between transmitted health check packets. Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + Defaults to 15s when omitted from the selected health check scope. x-kubernetes-int-or-string: true type: object tunnelMTU: diff --git a/deploy/net/crd/net.unbounded-cloud.io_gatewaypools.yaml b/deploy/net/crd/net.unbounded-cloud.io_gatewaypools.yaml index 5bac17b5a..b5eaf4baf 100644 --- a/deploy/net/crd/net.unbounded-cloud.io_gatewaypools.yaml +++ b/deploy/net/crd/net.unbounded-cloud.io_gatewaypools.yaml @@ -76,6 +76,7 @@ spec: description: |- ReceiveInterval is the minimum interval between received health check packets. Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + Defaults to 15s when omitted from the selected health check scope. x-kubernetes-int-or-string: true transmitInterval: anyOf: @@ -84,6 +85,7 @@ spec: description: |- TransmitInterval is the minimum interval between transmitted health check packets. Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + Defaults to 15s when omitted from the selected health check scope. x-kubernetes-int-or-string: true type: object nodeSelector: diff --git a/deploy/net/crd/net.unbounded-cloud.io_sitegatewaypoolassignments.yaml b/deploy/net/crd/net.unbounded-cloud.io_sitegatewaypoolassignments.yaml index e84653908..240b1d1ae 100644 --- a/deploy/net/crd/net.unbounded-cloud.io_sitegatewaypoolassignments.yaml +++ b/deploy/net/crd/net.unbounded-cloud.io_sitegatewaypoolassignments.yaml @@ -82,6 +82,7 @@ spec: description: |- ReceiveInterval is the minimum interval between received health check packets. Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + Defaults to 15s when omitted from the selected health check scope. x-kubernetes-int-or-string: true transmitInterval: anyOf: @@ -90,6 +91,7 @@ spec: description: |- TransmitInterval is the minimum interval between transmitted health check packets. Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + Defaults to 15s when omitted from the selected health check scope. x-kubernetes-int-or-string: true type: object sites: diff --git a/deploy/net/crd/net.unbounded-cloud.io_sitepeerings.yaml b/deploy/net/crd/net.unbounded-cloud.io_sitepeerings.yaml index 10796e16d..272bbcbff 100644 --- a/deploy/net/crd/net.unbounded-cloud.io_sitepeerings.yaml +++ b/deploy/net/crd/net.unbounded-cloud.io_sitepeerings.yaml @@ -84,6 +84,7 @@ spec: description: |- ReceiveInterval is the minimum interval between received health check packets. Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + Defaults to 15s when omitted from the selected health check scope. x-kubernetes-int-or-string: true transmitInterval: anyOf: @@ -92,6 +93,7 @@ spec: description: |- TransmitInterval is the minimum interval between transmitted health check packets. Accepts either a duration string (e.g. "300ms") or an integer interpreted as milliseconds. + Defaults to 15s when omitted from the selected health check scope. x-kubernetes-int-or-string: true type: object meshNodes: diff --git a/docs/net/architecture.md b/docs/net/architecture.md index e8a9a33bb..6a90368ee 100644 --- a/docs/net/architecture.md +++ b/docs/net/architecture.md @@ -474,7 +474,7 @@ sequenceDiagram participant Routes as Netlink Route Table Agent->>HC: Start health check sessions for peers - loop Every transmitInterval (default 1s) + loop Every transmitInterval (default 15s) HC->>Peer: UDP probe (over WireGuard tunnel) alt Healthy Peer-->>HC: UDP response @@ -488,8 +488,9 @@ sequenceDiagram ``` **Key Design Decisions:** -- Probes are sent and received over WireGuard tunnels at configurable intervals (default 1s) +- Probes are sent and received over supported tunnel types at configurable intervals (default 15s) - Failure detection uses `detectMultiplier * max(transmitInterval, receiveInterval)` to determine when a peer is down +- The 15s defaults reduce probe traffic versus the previous 1s defaults, trading a nominal 3s detection timeout for 45s; explicit intervals retain their requested cadence - On failure, route metrics are increased to deprioritize unhealthy paths rather than removing routes entirely - On recovery, route metrics are restored to their base values diff --git a/docs/net/configuration.md b/docs/net/configuration.md index 1569b0785..1cc49ffe2 100644 --- a/docs/net/configuration.md +++ b/docs/net/configuration.md @@ -356,24 +356,47 @@ The **security-wins rule** ensures that if any scope in the hierarchy explicitly ### Health Check (UDP Probe over Tunnel) -The health check protocol provides sub-second failure detection for overlay peers using a custom UDP probe protocol (similar to SBFD) running over all tunnel types. Sessions are automatically created for all routes with nexthops (supernet/RoutedCidrs routes, podCIDR routes, and internal IP routes). Bootstrap routes (/32 and /128 host routes for peer nexthops) do not use health checks to avoid a chicken-and-egg dependency. +The health check protocol monitors overlay peers using a custom UDP probe protocol (similar to SBFD) running over all tunnel types. +The node registers per-peer sessions using the peer's overlay health IP when the resolved health-check profile is enabled. +Transmit and receive intervals default to **15s**, with detect multiplier **3** and maximum flap backoff **120s**. +The detection timeout is `detectMultiplier * max(transmitInterval, receiveInterval)`, or **45s** with defaults. +Compared with the previous 1s default, this sends one-fifteenth as many probes per peer, trading a nominal 3s detection timeout for 45s to reduce steady-state traffic and CPU. +Existing explicit intervals are unchanged; set both intervals to `1s` to retain the previous cadence and nominal timeout. +Timeouts are checked every half-timeout (at least 100ms), so an unresponsive established session can take up to another check interval to be marked down. +Shorter explicit intervals can provide faster detection at the cost of additional probe traffic and CPU. **Health Check Behavior:** -- Health check sessions are managed automatically for all routed traffic +- Health check sessions are managed automatically for peers with enabled profiles - Session status is displayed on the `/status` endpoint - Health checks replace the legacy gateway health checking mechanism -- route metric adjustment on health check failure provides faster and more reliable failover -- No additional configuration flags are needed -- health checks are always active for routed traffic +- Profiles are enabled by default; `healthCheckSettings.enabled: false` disables the selected association without falling back to a less-specific enabled profile **Health Check Settings Precedence (CRDs):** - `Site.spec.healthCheckSettings` applies to node-to-node routes within the same site. - `SitePeering.spec.healthCheckSettings` applies to node-to-node routes between sites in that peering. -- `GatewayPool.spec.healthCheckSettings` applies to routes from nodes to peers in that gateway pool. +- `GatewayPool.spec.healthCheckSettings` governs same-pool gateway peers. - `GatewayPoolPeering.spec.healthCheckSettings` applies to routes between gateway pools in that peering. -For gateway-pool routes, precedence is: -1. `SiteGatewayPoolAssignment.spec.healthCheckSettings` -2. `GatewayPool.spec.healthCheckSettings` -3. `Site.spec.healthCheckSettings` +For mesh peers, an explicit gateway-pool or pool-peering profile wins, followed by a `SiteGatewayPoolAssignment`, a `SitePeering`, and then the peer's `Site`. +For node-to-gateway peers, the node's site/pool assignment governs; gateway nodes use their explicit peer profile or the peer's pool profile, with a remote-site/pool assignment as fallback. +Shared-tunnel gateway peers may fall back to the local site's profile when no governing association exists; WireGuard gateway peers do not use that site fallback. +An explicitly disabled governing profile blocks every fallback. + +The selected scope is merged with fresh defaults, not the last applied profile or values from lower-priority scopes. +For example, specifying only `transmitInterval: 60s` uses a 15s receive interval and detect multiplier 3. +Both duration strings and integer milliseconds are accepted; explicit settings and examples retain their requested intervals. +The node's global maximum flap-backoff setting overrides the selected profile's backoff. + +Reconciliation passes the freshly resolved settings to every supported transport before committing its cached profile maps. +Settings-only changes for an existing peer and overlay IP update the live session in place, preserving health state, uptime, RTT, packet counters, and flap history. +Probe and detection timers wake promptly rather than waiting for the previous interval to elapse. +Each directed `(local node, peer)` identity has a deterministic transmit phase strictly greater than zero and no larger than the configured transmit interval. +Initial probes and transmit-interval changes use this phase to spread fleet-wide starts and updates; subsequent probes use the exact configured interval, without per-probe jitter or extra retries. +Receive-only, backoff-only, and unchanged settings do not reset the transmit phase. +When an established healthy session's detection timeout is shortened, one bounded transition window lets the first probe under the new cadence receive a reply before applying the shorter timeout to old-cadence data. +The window is at most one new transmit interval plus one new nominal detection timeout; a fresh reply immediately restores ordinary detection. +The nominal timeout remains `detectMultiplier * max(transmitInterval, receiveInterval)`, and the transition does not fabricate a reply or reset counters. +Changing a peer's overlay IP still replaces and cancels the old session; newly created sessions begin down until sufficient replies arrive. If multiple peerings define conflicting health check settings for the same target site or gateway pool, the controller processes peerings in deterministic name order and keeps the first profile, diff --git a/internal/net/healthcheck/manager.go b/internal/net/healthcheck/manager.go index f511d9d9a..5bc291f58 100644 --- a/internal/net/healthcheck/manager.go +++ b/internal/net/healthcheck/manager.go @@ -26,6 +26,7 @@ type Manager struct { conn net.PacketConn mu sync.RWMutex + peerMu sync.Mutex sessions map[string]*session ctx context.Context @@ -90,15 +91,26 @@ func (m *Manager) Start(ctx context.Context) error { // Stop gracefully shuts down the listener and all sessions. func (m *Manager) Stop() { + m.peerMu.Lock() + defer m.peerMu.Unlock() + if m.cancel != nil { m.cancel() } - m.mu.Lock() + m.mu.RLock() + + sessions := make([]*session, 0, len(m.sessions)) for _, s := range m.sessions { + sessions = append(sessions, s) + } + + m.mu.RUnlock() + + for _, s := range sessions { s.stop() } - m.mu.Unlock() + m.listener.Stop() klog.Info("healthcheck manager stopped") } @@ -106,19 +118,30 @@ func (m *Manager) Stop() { // AddPeer registers a new peer for health checking. If the manager is // already running, the session starts immediately. func (m *Manager) AddPeer(peerHostname string, overlayIP net.IP, settings HealthCheckSettings) error { - m.mu.Lock() - defer m.mu.Unlock() + if err := settings.validate(); err != nil { + return err + } + m.peerMu.Lock() + defer m.peerMu.Unlock() + + m.mu.Lock() if existing, exists := m.sessions[peerHostname]; exists { - // Peer already registered -- update settings/IP if changed, otherwise no-op - if existing.overlayIP.Equal(overlayIP) && existing.settings == settings { + if existing.overlayIP.Equal(overlayIP) { + existing.updateSettings(settings) + m.mu.Unlock() + return nil } - // Settings or IP changed -- stop old session and replace + + delete(m.sessions, peerHostname) + m.mu.Unlock() + // Callbacks may query the manager while stop waits for their completion. existing.stop() klog.V(4).Infof("healthcheck: updating peer %s (%s -> %s)", peerHostname, existing.overlayIP, overlayIP) - delete(m.sessions, peerHostname) + m.mu.Lock() } + defer m.mu.Unlock() s := newSession(sessionConfig{ peerHostname: peerHostname, @@ -145,6 +168,9 @@ func (m *Manager) AddPeer(peerHostname string, overlayIP net.IP, settings Health // RemovePeer stops and removes a peer session. func (m *Manager) RemovePeer(peerHostname string) error { + m.peerMu.Lock() + defer m.peerMu.Unlock() + m.mu.Lock() s, exists := m.sessions[peerHostname] @@ -165,9 +191,14 @@ func (m *Manager) RemovePeer(peerHostname string) error { // UpdatePeerSettings modifies the health check parameters for an existing peer. func (m *Manager) UpdatePeerSettings(peerHostname string, settings HealthCheckSettings) error { + if err := settings.validate(); err != nil { + return err + } + m.mu.RLock() + defer m.mu.RUnlock() + s, exists := m.sessions[peerHostname] - m.mu.RUnlock() if !exists { return fmt.Errorf("peer %q not found", peerHostname) @@ -178,6 +209,22 @@ func (m *Manager) UpdatePeerSettings(peerHostname string, settings HealthCheckSe return nil } +// GetPeerSettings returns a copy of the currently applied session settings. +func (m *Manager) GetPeerSettings(peerHostname string) (HealthCheckSettings, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + s, exists := m.sessions[peerHostname] + if !exists { + return HealthCheckSettings{}, fmt.Errorf("peer %q not found", peerHostname) + } + + s.mu.Lock() + defer s.mu.Unlock() + + return s.settings, nil +} + // GetPeerStatus returns the current health status for a single peer. func (m *Manager) GetPeerStatus(peerHostname string) (*PeerStatus, error) { m.mu.RLock() diff --git a/internal/net/healthcheck/probe_phase.go b/internal/net/healthcheck/probe_phase.go new file mode 100644 index 000000000..6f047bc9e --- /dev/null +++ b/internal/net/healthcheck/probe_phase.go @@ -0,0 +1,32 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package healthcheck + +import ( + "crypto/sha256" + "encoding/binary" + "math/bits" + "time" +) + +// Directional identities spread both a node's outgoing probes and the probes +// arriving at one peer from different nodes. No process-global RNG is needed. +func peerProbePhaseSeed(local, remote string) uint64 { + key := make([]byte, 0, 16+len(local)+len(remote)) + key = binary.BigEndian.AppendUint64(key, uint64(len(local))) + key = append(key, local...) + key = binary.BigEndian.AppendUint64(key, uint64(len(remote))) + key = append(key, remote...) + hash := sha256.Sum256(key) + + return binary.BigEndian.Uint64(hash[:8]) +} + +// probePhase scales a stable fraction into [1ns, interval], without overflow or +// floating-point rounding. Only the first probe is offset; steady ticks retain +// the configured interval. Manager validation guarantees interval is positive. +func (s *session) probePhase(interval time.Duration) time.Duration { + high, _ := bits.Mul64(s.probePhaseSeed, uint64(interval)) + return time.Duration(high + 1) +} diff --git a/internal/net/healthcheck/probe_phase_test.go b/internal/net/healthcheck/probe_phase_test.go new file mode 100644 index 000000000..fa3e7a248 --- /dev/null +++ b/internal/net/healthcheck/probe_phase_test.go @@ -0,0 +1,327 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package healthcheck + +import ( + "context" + "fmt" + "net" + "sync" + "testing" + "testing/synctest" + "time" + + "google.golang.org/protobuf/proto" + + pb "github.com/Azure/unbounded/internal/net/healthcheck/proto" +) + +var intervalEndProbePhaseSeed = ^uint64(0) + +func TestProbePhaseBoundsAndIdentitySpread(t *testing.T) { + for _, interval := range []time.Duration{1, 2, 15 * time.Second, 60 * time.Second, time.Duration(1<<63 - 1)} { + for _, seed := range []uint64{0, 1, 1 << 63, ^uint64(0)} { + s := newSession(sessionConfig{probePhaseSeed: &seed}) + + phase := s.probePhase(interval) + if phase <= 0 || phase > interval { + t.Fatalf("seed=%d interval=%v phase=%v", seed, interval, phase) + } + + if seed == 0 && phase != time.Nanosecond { + t.Fatal("minimum phase must be positive") + } + + if seed == ^uint64(0) && phase != interval { + t.Fatal("maximum injected fraction must reach the interval boundary") + } + } + } + + if peerProbePhaseSeed("ab", "c") == peerProbePhaseSeed("a", "bc") || + peerProbePhaseSeed("node-a", "node-b") == peerProbePhaseSeed("node-b", "node-a") { + t.Fatal("phase identities are ambiguous or not directional") + } + + for _, incoming := range []bool{false, true} { + buckets := make([]int, 60) + + for i := range 2000 { + local, remote := "node-fixed", fmt.Sprintf("node-%d", i) + if incoming { + local, remote = remote, local + } + + s := newSession(sessionConfig{localHostname: local, peerHostname: remote}) + phase := s.probePhase(time.Minute) + + buckets[int((phase-1)/time.Second)]++ + if phase != newSession(sessionConfig{localHostname: local, peerHostname: remote}).probePhase(time.Minute) { + t.Fatal("identity-derived phase is not deterministic") + } + } + + for second, count := range buckets { + if count == 0 || count > 80 { + t.Fatalf("incoming=%t second=%d count=%d: phases concentrate peers", incoming, second, count) + } + } + + minimum, maximum := 2000, 0 + for _, count := range buckets { + minimum, maximum = min(minimum, count), max(maximum, count) + } + + t.Logf("incoming=%t: 2000 peers span all 60 one-second buckets; min=%d max=%d", incoming, minimum, maximum) + } +} + +type phaseRecordingConn struct { + discardProbeConn + mu sync.Mutex + writes map[string][]time.Time +} + +func (c *phaseRecordingConn) WriteTo(data []byte, _ net.Addr) (int, error) { + var packet pb.HealthCheckPacket + if err := proto.Unmarshal(data, &packet); err != nil { + return 0, err + } + + c.mu.Lock() + defer c.mu.Unlock() + + key := packet.SourceHostname + "|" + packet.DestinationHostname + c.writes[key] = append(c.writes[key], time.Now()) + + return len(data), nil +} + +func TestProbePhasesSpreadFleetUpdatesAndPreserveRate(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + conn := &phaseRecordingConn{writes: make(map[string][]time.Time)} + settings := DefaultSettings() + settings.TransmitInterval, settings.ReceiveInterval = time.Minute, time.Minute + start := time.Now() + + var sessions []*session + + for i := range 128 { + s := newSession(sessionConfig{localHostname: "local", peerHostname: fmt.Sprintf("peer-%d", i), settings: settings, conn: conn}) + s.start(ctx) + sessions = append(sessions, s) + } + + defer func() { + cancel() + + for _, s := range sessions { + s.stop() + } + }() + + synctest.Wait() + time.Sleep(2 * time.Minute) + synctest.Wait() + + buckets := make(map[int]int) + + for _, writes := range conn.writes { + if len(writes) != 2 || writes[1].Sub(writes[0]) != time.Minute { + t.Fatalf("steady cadence changed: %v", writes) + } + + buckets[int(writes[0].Sub(start)/time.Second)]++ + } + + if len(conn.writes) != len(sessions) || len(buckets) < 30 { + t.Fatalf("initial phases concentrated: peers=%d buckets=%d", len(conn.writes), len(buckets)) + } + + for _, count := range buckets { + if count > 10 { + t.Fatalf("initial one-second burst contains %d/128 peers", count) + } + } + + t.Logf("128 initial sessions: %d one-second buckets, exactly one probe per 60s interval", len(buckets)) + + changedAt := time.Now() + + for _, s := range sessions { + for _, interval := range []time.Duration{15 * time.Second, time.Minute, 15 * time.Second} { + settings.TransmitInterval, settings.ReceiveInterval = interval, interval + s.updateSettings(settings) + } + } + + synctest.Wait() + time.Sleep(30 * time.Second) + synctest.Wait() + + buckets = make(map[int]int) + + for _, writes := range conn.writes { + if len(writes) != 4 || writes[3].Sub(writes[2]) != 15*time.Second { + t.Fatalf("coalesced update changed steady cadence: %v", writes) + } + + phase := writes[2].Sub(changedAt) + if phase <= 0 || phase > 15*time.Second { + t.Fatalf("updated phase out of bounds: %v", phase) + } + + buckets[int(phase/time.Second)]++ + } + + if len(buckets) < 12 { + t.Fatalf("reconfiguration phases concentrated in %d seconds", len(buckets)) + } + + t.Logf("128 reconfigured sessions: %d one-second buckets, exactly one probe per 15s interval", len(buckets)) + cancel() + + for _, s := range sessions { + s.stop() + } + + time.Sleep(time.Minute) + + for _, writes := range conn.writes { + if len(writes) != 4 { + t.Fatal("stopped sessions leaked probes") + } + } + }) +} + +func TestProbePhaseCancellationAndNonTransmitUpdates(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + settings := DefaultSettings() + settings.TransmitInterval = time.Minute + s := newSession(sessionConfig{settings: settings, conn: &discardProbeConn{}, probePhaseSeed: &intervalEndProbePhaseSeed}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + s.start(ctx) + defer s.stop() + + synctest.Wait() + time.Sleep(10 * time.Second) + + settings.ReceiveInterval = 2 * time.Minute + s.updateSettings(settings) + settings.MaxBackoff = 240 * time.Second + s.updateSettings(settings) + s.updateSettings(settings) + synctest.Wait() + time.Sleep(50 * time.Second) + synctest.Wait() + + if s.status().PacketsSent != 1 { + t.Fatal("receive/backoff/unchanged settings disturbed the original transmit phase") + } + + settings.TransmitInterval = time.Hour + s.updateSettings(settings) + synctest.Wait() + cancel() + s.stop() + time.Sleep(2 * time.Hour) + + if s.status().PacketsSent != 1 { + t.Fatal("canceled initial phase sent probes") + } + }) +} + +func TestShortenedIntervalsAllowFirstProbeNominalTimeout(t *testing.T) { + for _, reply := range []bool{false, true} { + t.Run(fmt.Sprintf("reply-%t", reply), func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + settings := DefaultSettings() + settings.TransmitInterval, settings.ReceiveInterval = time.Minute, time.Minute + s := newSession(sessionConfig{settings: settings, conn: &discardProbeConn{}, probePhaseSeed: &intervalEndProbePhaseSeed}) + s.state, s.stateSince = StateUp, time.Now().Add(-time.Hour) + s.lastReceived = time.Now().Add(-50 * time.Second) + oldReply := s.lastReceived + s.packetsReceived = 10 + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + s.start(ctx) + defer s.stop() + + synctest.Wait() + + changedAt := time.Now() + + s.updateSettings(DefaultSettings()) + synctest.Wait() + + if s.detectTimeout() != 45*time.Second || s.detectGraceUntil.Sub(changedAt) != time.Minute { + t.Fatal("nominal timeout or bounded transition window changed") + } + + time.Sleep(23 * time.Second) + synctest.Wait() + + if s.status().State != StateUp || s.packetsReceived != 10 || !s.lastReceived.Equal(oldReply) { + t.Fatal("shortening applied the new timeout to an old-cadence reply or fabricated liveness") + } + + if reply { + s.receiveReply(&pb.HealthCheckPacket{TimestampNs: time.Now().UnixNano()}) + synctest.Wait() + + if !s.detectGraceUntil.IsZero() { + t.Fatal("fresh reply did not restore ordinary detection") + } + + time.Sleep(23 * time.Second) + synctest.Wait() + + if s.status().State != StateUp { + t.Fatal("healthy peer flapped during interval shortening") + } + + time.Sleep(45 * time.Second) + } else { + time.Sleep(45 * time.Second) + } + + synctest.Wait() + + if s.status().State != StateDown { + t.Fatal("transition protection silently disabled failure detection") + } + }) + }) + } +} + +func TestTimeoutTransitionRechecksFreshReply(t *testing.T) { + settings := DefaultSettings() + s := newSession(sessionConfig{settings: settings}) + s.state = StateUp + s.lastReceived = time.Now().Add(-time.Minute) + s.mu.Lock() + expired := s.replyTimedOut(time.Now()) + s.mu.Unlock() + + if !expired { + t.Fatal("test requires expired old snapshot") + } + + s.receiveReply(&pb.HealthCheckPacket{TimestampNs: time.Now().UnixNano()}) + + if s.setStateIf(StateDown, func() bool { return s.state == StateUp && s.replyTimedOut(time.Now()) }) { + t.Fatal("stale timeout snapshot overrode a fresh reply") + } +} diff --git a/internal/net/healthcheck/reconfiguration_test.go b/internal/net/healthcheck/reconfiguration_test.go new file mode 100644 index 000000000..e081a643a --- /dev/null +++ b/internal/net/healthcheck/reconfiguration_test.go @@ -0,0 +1,390 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package healthcheck + +import ( + "context" + "io" + "net" + "reflect" + "sync" + "testing" + "testing/synctest" + "time" + + pb "github.com/Azure/unbounded/internal/net/healthcheck/proto" +) + +type discardProbeConn struct{} + +func (*discardProbeConn) ReadFrom([]byte) (int, net.Addr, error) { return 0, nil, io.EOF } +func (*discardProbeConn) WriteTo(b []byte, _ net.Addr) (int, error) { return len(b), nil } +func (*discardProbeConn) Close() error { return nil } +func (*discardProbeConn) LocalAddr() net.Addr { return &net.UDPAddr{} } +func (*discardProbeConn) SetDeadline(time.Time) error { return nil } +func (*discardProbeConn) SetReadDeadline(time.Time) error { return nil } +func (*discardProbeConn) SetWriteDeadline(time.Time) error { return nil } + +func TestDefaultHealthCheckIntervals(t *testing.T) { + settings := DefaultSettings() + if settings.TransmitInterval != 15*time.Second || settings.ReceiveInterval != 15*time.Second || + settings.DetectMultiplier != 3 || settings.MaxBackoff != 120*time.Second { + t.Fatalf("unexpected defaults: %+v", settings) + } + + s := newSession(sessionConfig{settings: settings}) + if s.detectTimeout() != 45*time.Second { + t.Fatal("unexpected default detection timeout") + } +} + +func TestManagerLiveSettingsPreserveHealthySession(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + m, err := NewManager("local", 0, nil) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + m.ctx, m.conn = ctx, &discardProbeConn{} + defer m.Stop() + + ip := net.ParseIP("10.0.0.1") + if err := m.AddPeer("peer", ip, DefaultSettings()); err != nil { + t.Fatal(err) + } + + synctest.Wait() + + s := m.sessions["peer"] + + time.Sleep(15 * time.Second) + synctest.Wait() + + for range 3 { + s.receiveReply(&pb.HealthCheckPacket{TimestampNs: time.Now().Add(-17 * time.Millisecond).UnixNano()}) + } + + synctest.Wait() + + before := s.status() + if before.State != StateUp || before.PacketsSent != 1 || before.PacketsReceived != 3 || before.LastRTT != 17*time.Millisecond { + t.Fatalf("session must start healthy with measurements: %+v", before) + } + + for _, interval := range []time.Duration{60 * time.Second, 15 * time.Second, time.Second} { + settings := DefaultSettings() + + settings.TransmitInterval, settings.ReceiveInterval = interval, interval + if err := m.AddPeer("peer", ip, settings); err != nil { + t.Fatal(err) + } + + synctest.Wait() + + if m.sessions["peer"] != s { + t.Fatal("settings-only AddPeer replaced session") + } + + got, err := m.GetPeerSettings("peer") + if err != nil || got != settings { + t.Fatalf("applied settings: %+v, %v", got, err) + } + + after := s.status() + + after.RequiredReplies = before.RequiredReplies + if !reflect.DeepEqual(before, after) { + t.Fatalf("health history reset: before=%+v after=%+v", before, after) + } + + settings.MaxBackoff = 300 * time.Second + if err := m.UpdatePeerSettings("peer", settings); err != nil { + t.Fatal(err) + } + + synctest.Wait() + + after = s.status() + + after.RequiredReplies = before.RequiredReplies + if !reflect.DeepEqual(before, after) { + t.Fatal("UpdatePeerSettings reset health history") + } + } + }) +} + +func TestSessionSettingsWakeProbeTimer(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + settings := DefaultSettings() + settings.TransmitInterval, settings.ReceiveInterval = time.Hour, time.Hour + s := newSession(sessionConfig{settings: settings, conn: &discardProbeConn{}, probePhaseSeed: &intervalEndProbePhaseSeed}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + s.start(ctx) + defer s.stop() + + synctest.Wait() + + settings.TransmitInterval = 20 * time.Millisecond + s.updateSettings(settings) + synctest.Wait() + time.Sleep(21 * time.Millisecond) + synctest.Wait() + + if s.status().PacketsSent != 1 { + t.Fatal("shortened interval waited for old one-hour timer") + } + + settings.TransmitInterval = time.Hour + s.updateSettings(settings) + synctest.Wait() + time.Sleep(time.Second) + synctest.Wait() + + if s.status().PacketsSent != 1 { + t.Fatal("lengthened interval allowed old fast probes") + } + + for _, interval := range []time.Duration{10 * time.Millisecond, time.Hour, 20 * time.Millisecond} { + settings.TransmitInterval = interval + s.updateSettings(settings) + } + + synctest.Wait() + time.Sleep(21 * time.Millisecond) + synctest.Wait() + + if s.status().PacketsSent != 2 { + t.Fatal("coalesced updates did not use latest interval") + } + + cancel() + s.stop() + count := s.status().PacketsSent + s.updateSettings(DefaultSettings()) + time.Sleep(time.Minute) + + if s.status().PacketsSent != count { + t.Fatal("stopped session resumed probes") + } + }) +} + +func TestSessionSettingsWakeDetectionTimer(t *testing.T) { + for _, increase := range []bool{false, true} { + t.Run(map[bool]string{false: "shorten", true: "lengthen"}[increase], func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + settings := DefaultSettings() + + settings.TransmitInterval, settings.ReceiveInterval = time.Hour, time.Hour + if increase { + settings.TransmitInterval, settings.ReceiveInterval = 20*time.Millisecond, 20*time.Millisecond + } + + s := newSession(sessionConfig{settings: settings, conn: &discardProbeConn{}, probePhaseSeed: &intervalEndProbePhaseSeed}) + s.state, s.lastReceived = StateUp, time.Now() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + s.start(ctx) + defer s.stop() + + synctest.Wait() + + settings.TransmitInterval, settings.ReceiveInterval = 20*time.Millisecond, 20*time.Millisecond + if increase { + settings.TransmitInterval, settings.ReceiveInterval = time.Hour, time.Hour + } + + s.updateSettings(settings) + synctest.Wait() + time.Sleep(201 * time.Millisecond) + synctest.Wait() + + want := StateDown + if increase { + want = StateUp + } + + if got := s.status().State; got != want { + t.Fatalf("state=%v want %v after timer update", got, want) + } + }) + }) + } +} + +func TestSessionIdenticalSettingsKeepProbeDeadline(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + settings := DefaultSettings() + settings.TransmitInterval = 50 * time.Millisecond + s := newSession(sessionConfig{settings: settings, conn: &discardProbeConn{}}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + s.start(ctx) + defer s.stop() + + synctest.Wait() + time.Sleep(20 * time.Millisecond) + s.updateSettings(settings) + synctest.Wait() + time.Sleep(31 * time.Millisecond) + synctest.Wait() + + if s.status().PacketsSent != 1 { + t.Fatal("identical settings reset probe deadline") + } + }) +} + +func TestManagerIPReplacementJoinsSessionOutsideLookupLock(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + entered, release := make(chan struct{}), make(chan struct{}) + + var ( + m *Manager + err error + ) + + m, err = NewManager("local", 0, func(string, SessionState, SessionState) { + close(entered) + <-release + m.GetAllPeerStatuses() + }) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + m.ctx, m.conn = ctx, &discardProbeConn{} + defer m.Stop() + + if err := m.AddPeer("peer", net.ParseIP("10.0.0.1"), DefaultSettings()); err != nil { + t.Fatal(err) + } + + old := m.sessions["peer"] + old.setState(StateUp) + <-entered + + done := make(chan error, 1) + go func() { done <- m.AddPeer("peer", net.ParseIP("10.0.0.2"), DefaultSettings()) }() + + synctest.Wait() + close(release) + + if err := <-done; err != nil { + t.Fatal(err) + } + + if m.sessions["peer"] == old || m.sessions["peer"].status().State != StateDown { + t.Fatal("IP replacement retained old session") + } + + synctest.Wait() + + before := old.status().PacketsSent + + time.Sleep(16 * time.Second) + synctest.Wait() + + if old.status().PacketsSent != before { + t.Fatal("old IP session still sends probes") + } + }) +} + +func TestManagerConcurrentSettingsAndStatus(t *testing.T) { + m, err := NewManager("local", 0, nil) + if err != nil { + t.Fatal(err) + } + + ip := net.ParseIP("10.0.0.1") + if err := m.AddPeer("peer", ip, DefaultSettings()); err != nil { + t.Fatal(err) + } + + var wg sync.WaitGroup + for worker := range 4 { + wg.Go(func() { + for i := range 100 { + settings := DefaultSettings() + settings.TransmitInterval += time.Duration(i) * time.Millisecond + + var err error + + switch worker { + case 0: + err = m.AddPeer("peer", ip, settings) + case 1: + err = m.UpdatePeerSettings("peer", settings) + case 2: + _, err = m.GetPeerSettings("peer") + case 3: + _, err = m.GetPeerStatus("peer") + } + + if err != nil { + t.Errorf("concurrent operation: %v", err) + return + } + } + }) + } + + wg.Wait() +} + +func TestManagerRejectsInvalidSettings(t *testing.T) { + for name, change := range map[string]func(*HealthCheckSettings){ + "tx": func(s *HealthCheckSettings) { s.TransmitInterval = 0 }, + "rx": func(s *HealthCheckSettings) { s.ReceiveInterval = -1 }, + "multiplier": func(s *HealthCheckSettings) { s.DetectMultiplier = 0 }, + "large multiplier": func(s *HealthCheckSettings) { s.DetectMultiplier = 256 }, + "backoff": func(s *HealthCheckSettings) { s.MaxBackoff = -1 }, + "overflow": func(s *HealthCheckSettings) { s.ReceiveInterval = time.Duration(1<<63 - 1) }, + } { + t.Run(name, func(t *testing.T) { + m, err := NewManager("local", 0, nil) + if err != nil { + t.Fatal(err) + } + + settings := DefaultSettings() + change(&settings) + + if err := m.AddPeer("peer", net.ParseIP("10.0.0.1"), settings); err == nil { + t.Fatal("invalid settings accepted") + } + + if len(m.sessions) != 0 { + t.Fatal("invalid AddPeer mutated sessions") + } + + if err := m.AddPeer("peer", net.ParseIP("10.0.0.1"), DefaultSettings()); err != nil { + t.Fatal(err) + } + + if err := m.UpdatePeerSettings("peer", settings); err == nil { + t.Fatal("invalid update accepted") + } + + if got, _ := m.GetPeerSettings("peer"); got != DefaultSettings() { + t.Fatal("invalid update changed settings") + } + }) + } +} diff --git a/internal/net/healthcheck/session.go b/internal/net/healthcheck/session.go index 49b30e26a..3c5a6403b 100644 --- a/internal/net/healthcheck/session.go +++ b/internal/net/healthcheck/session.go @@ -31,16 +31,19 @@ const flapWindowDuration = 5 * time.Minute // It sends periodic probes and monitors for replies to determine // if the peer is up. type session struct { - peerHostname string - overlayIP net.IP - port int - localHost string + peerHostname string + overlayIP net.IP + port int + localHost string + probePhaseSeed uint64 mu sync.Mutex settings HealthCheckSettings + probeRevision uint64 state SessionState stateSince time.Time lastReceived time.Time + detectGraceUntil time.Time lastRTT time.Duration packetsSent uint64 packetsReceived uint64 @@ -49,11 +52,13 @@ type session struct { seqNum atomic.Uint64 onStateChange StateChangeFunc - conn net.PacketConn - cancel context.CancelFunc - callbackCh chan stateEvent - wg sync.WaitGroup - started bool + conn net.PacketConn + cancel context.CancelFunc + callbackCh chan stateEvent + probeSettingsCh chan struct{} + detectSettingsCh chan struct{} + wg sync.WaitGroup + started bool } // sessionConfig holds the parameters needed to create a new session. @@ -65,23 +70,33 @@ type sessionConfig struct { settings HealthCheckSettings onChange StateChangeFunc conn net.PacketConn + // Tests can inject the phase fraction, not an unchecked timer duration. + probePhaseSeed *uint64 } // newSession creates a new health check session for a remote peer. func newSession(cfg sessionConfig) *session { now := time.Now() + seed := peerProbePhaseSeed(cfg.localHostname, cfg.peerHostname) + if cfg.probePhaseSeed != nil { + seed = *cfg.probePhaseSeed + } + return &session{ - peerHostname: cfg.peerHostname, - overlayIP: cfg.overlayIP, - port: cfg.port, - localHost: cfg.localHostname, - settings: cfg.settings, - state: StateDown, - stateSince: now, - onStateChange: cfg.onChange, - conn: cfg.conn, - callbackCh: make(chan stateEvent, 8), + peerHostname: cfg.peerHostname, + overlayIP: cfg.overlayIP, + port: cfg.port, + localHost: cfg.localHostname, + probePhaseSeed: seed, + settings: cfg.settings, + state: StateDown, + stateSince: now, + onStateChange: cfg.onChange, + conn: cfg.conn, + callbackCh: make(chan stateEvent, 8), + probeSettingsCh: make(chan struct{}, 1), + detectSettingsCh: make(chan struct{}, 1), } } @@ -118,6 +133,7 @@ func (s *session) receiveReply(pkt *pb.HealthCheckPacket) { s.mu.Lock() s.lastReceived = now + s.detectGraceUntil = time.Time{} s.lastRTT = rtt s.packetsReceived++ @@ -158,9 +174,48 @@ func (s *session) status() *PeerStatus { // updateSettings applies new health check settings. func (s *session) updateSettings(settings HealthCheckSettings) { s.mu.Lock() - defer s.mu.Unlock() + if s.settings == settings { + s.mu.Unlock() + return + } + + oldTimeout := s.detectTimeout() + probeChanged := s.settings.TransmitInterval != settings.TransmitInterval s.settings = settings + if probeChanged { + s.probeRevision++ + } + + newTimeout := s.detectTimeout() + if s.state == StateUp && newTimeout < oldTimeout { + // A reply from the old cadence may already exceed the new timeout. + // Allow the first newly scheduled probe one nominal reply timeout. + // Receive-only changes retain the existing phase, at most one TX away. + firstProbe := settings.TransmitInterval + if probeChanged { + firstProbe = s.probePhase(settings.TransmitInterval) + } + + s.detectGraceUntil = time.Now().Add(firstProbe).Add(newTimeout) + } + s.mu.Unlock() + + // Coalesced wakeups read the latest settings. Receive/backoff changes do not + // disturb a running transmit phase. + if probeChanged { + select { + case s.probeSettingsCh <- struct{}{}: + default: + } + } + + if newTimeout != oldTimeout { + select { + case s.detectSettingsCh <- struct{}{}: + default: + } + } } func (s *session) probeLoop(ctx context.Context) { @@ -168,26 +223,68 @@ func (s *session) probeLoop(ctx context.Context) { s.mu.Lock() interval := s.settings.TransmitInterval + revision := s.probeRevision s.mu.Unlock() - ticker := time.NewTicker(interval) - defer ticker.Stop() + phaseTimer := time.NewTimer(s.probePhase(interval)) + defer phaseTimer.Stop() + + var ( + ticker *time.Ticker + ticks <-chan time.Time + ) + + defer func() { + if ticker != nil { + ticker.Stop() + } + }() + + resetPhase := func() bool { + s.mu.Lock() + newInterval := s.settings.TransmitInterval + newRevision := s.probeRevision + s.mu.Unlock() + + if newRevision == revision { + return false + } + + interval = newInterval + revision = newRevision + + if ticker != nil { + ticker.Stop() + ticker = nil + ticks = nil + } + + phaseTimer.Reset(s.probePhase(interval)) + + return true + } for { select { case <-ctx.Done(): return - case <-ticker.C: - s.sendProbe() - // Check if interval changed. - s.mu.Lock() - newInterval := s.settings.TransmitInterval - s.mu.Unlock() + case <-s.probeSettingsCh: + resetPhase() + case <-phaseTimer.C: + if resetPhase() { + continue + } + + ticker = time.NewTicker(interval) + ticks = ticker.C - if newInterval != interval { - interval = newInterval - ticker.Reset(interval) + s.sendProbe() + case <-ticks: + if resetPhase() { + continue } + + s.sendProbe() } } } @@ -211,15 +308,22 @@ func (s *session) detectLoop(ctx context.Context) { select { case <-ctx.Done(): return + case <-s.detectSettingsCh: + s.mu.Lock() + timeout = s.detectTimeout() + s.mu.Unlock() + + checkInterval = max(timeout/2, 100*time.Millisecond) + ticker.Reset(checkInterval) case <-ticker.C: s.mu.Lock() state := s.state - lastRecv := s.lastReceived timeout = s.detectTimeout() + expired := s.replyTimedOut(time.Now()) // Reset consecutive replies counter when we detect a timeout, // whether currently Up (transitioning to Down) or already Down // (stale counter from a partial reply burst). - if !lastRecv.IsZero() && time.Since(lastRecv) > timeout { + if expired { s.consecutiveReplies = 0 } s.mu.Unlock() @@ -228,9 +332,13 @@ func (s *session) detectLoop(ctx context.Context) { continue } - if state == StateUp && !lastRecv.IsZero() && time.Since(lastRecv) > timeout { - metricPacketsTimeout.Inc() - s.setState(StateDown) + if state == StateUp && expired { + // A reply or settings update may have arrived after the snapshot. + if s.setStateIf(StateDown, func() bool { + return s.state == StateUp && s.replyTimedOut(time.Now()) + }) { + metricPacketsTimeout.Inc() + } } // Update check interval if settings changed. @@ -247,6 +355,11 @@ func (s *session) detectLoop(ctx context.Context) { } } +// replyTimedOut requires s.mu and leaves the configured timeout unchanged. +func (s *session) replyTimedOut(now time.Time) bool { + return !s.lastReceived.IsZero() && now.Sub(s.lastReceived) > s.detectTimeout() && !now.Before(s.detectGraceUntil) +} + func (s *session) detectTimeout() time.Duration { tx := s.settings.TransmitInterval rx := s.settings.ReceiveInterval @@ -293,12 +406,16 @@ func (s *session) sendProbe() { } func (s *session) setState(newState SessionState) { + s.setStateIf(newState, nil) +} + +func (s *session) setStateIf(newState SessionState, ready func() bool) bool { s.mu.Lock() oldState := s.state - if oldState == newState { + if oldState == newState || (ready != nil && !ready()) { s.mu.Unlock() - return + return false } s.state = newState @@ -322,6 +439,8 @@ func (s *session) setState(newState SessionState) { klog.V(2).Infof("healthcheck: callback channel full for peer %s, dropping %s -> %s", s.peerHostname, oldState, newState) } + + return true } // trimFlapTimestamps removes flap timestamps older than flapWindowDuration. diff --git a/internal/net/healthcheck/types.go b/internal/net/healthcheck/types.go index 47d8c348d..4ebd30a6a 100644 --- a/internal/net/healthcheck/types.go +++ b/internal/net/healthcheck/types.go @@ -3,7 +3,10 @@ package healthcheck -import "time" +import ( + "fmt" + "time" +) // SessionState represents the health state of a peer session. type SessionState int @@ -42,13 +45,34 @@ type HealthCheckSettings struct { // DefaultSettings returns the default health check settings. func DefaultSettings() HealthCheckSettings { return HealthCheckSettings{ - TransmitInterval: 1000 * time.Millisecond, - ReceiveInterval: 1000 * time.Millisecond, + TransmitInterval: 15 * time.Second, + ReceiveInterval: 15 * time.Second, DetectMultiplier: 3, MaxBackoff: 120 * time.Second, } } +func (s HealthCheckSettings) validate() error { + if s.TransmitInterval <= 0 || s.ReceiveInterval <= 0 { + return fmt.Errorf("health check intervals must be positive") + } + + if s.DetectMultiplier < 1 || s.DetectMultiplier > 255 { + return fmt.Errorf("health check detect multiplier must be between 1 and 255") + } + + const maxDuration = time.Duration(1<<63 - 1) + if max(s.TransmitInterval, s.ReceiveInterval) > maxDuration/time.Duration(s.DetectMultiplier) { + return fmt.Errorf("health check detection timeout overflows time.Duration") + } + + if s.MaxBackoff < 0 { + return fmt.Errorf("health check maximum backoff must not be negative") + } + + return nil +} + // PeerStatus contains the current health status of a peer. type PeerStatus struct { State SessionState From 3e71c03bd664d7b306f119c37b2d3c22573b9a70 Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Fri, 18 Sep 2026 20:42:08 +0000 Subject: [PATCH 2/3] fix(net): address health-check review feedback Rebound active timeout-transition grace when cadence settings change and align public reference defaults and precedence documentation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../reference/networking/architecture.md | 7 ++-- .../reference/networking/custom-resources.md | 21 ++++++---- docs/net/custom-resources.md | 19 +++++---- internal/net/healthcheck/probe_phase_test.go | 41 +++++++++++++++++++ internal/net/healthcheck/session.go | 12 +++++- 5 files changed, 78 insertions(+), 22 deletions(-) diff --git a/docs/content/reference/networking/architecture.md b/docs/content/reference/networking/architecture.md index 5d60d5bea..e8adb75eb 100644 --- a/docs/content/reference/networking/architecture.md +++ b/docs/content/reference/networking/architecture.md @@ -249,9 +249,10 @@ and upserts desired entries, keeping IPv4 and IPv6 tries synchronized. ### Gateway Health Checks Node agents send UDP probes over WireGuard tunnels at configurable intervals -(default 1s). Failure detection uses `detectMultiplier * max(transmitInterval, -receiveInterval)`. On failure, route metrics are increased to deprioritize -unhealthy paths. On recovery, metrics are restored. +(default 15s). Failure detection uses `detectMultiplier * max(transmitInterval, +receiveInterval)`, which is 45s with the default multiplier of 3. On failure, +route metrics are increased to deprioritize unhealthy paths. On recovery, +metrics are restored. ### Status Aggregation diff --git a/docs/content/reference/networking/custom-resources.md b/docs/content/reference/networking/custom-resources.md index 4430bbfe6..3c3ca5aef 100644 --- a/docs/content/reference/networking/custom-resources.md +++ b/docs/content/reference/networking/custom-resources.md @@ -253,8 +253,8 @@ spec: healthCheckSettings: enabled: true detectMultiplier: 3 - receiveInterval: 300ms - transmitInterval: 300ms + receiveInterval: 15s + transmitInterval: 15s tunnelProtocol: Auto ``` @@ -384,15 +384,20 @@ GatewayPool, SiteGatewayPoolAssignment, and GatewayPoolPeering: |-------|------|---------|-------------| | `enabled` | `*bool` | `true` | Enable health checks for routes in this scope. | | `detectMultiplier` | `*int32` | 3 | Number of consecutive failures before marking unhealthy. | -| `receiveInterval` | `string` | `300ms` | Expected interval between received probes. | -| `transmitInterval` | `string` | `300ms` | Interval between sent probes. | +| `receiveInterval` | `string` | `15s` | Expected interval between received probes. | +| `transmitInterval` | `string` | `15s` | Interval between sent probes. | ### Precedence -- **Same-site routes**: `Site.spec.healthCheckSettings` -- **Peered-site routes**: `SitePeering.spec.healthCheckSettings` -- **Gateway pool routes**: `SiteGatewayPoolAssignment.spec.healthCheckSettings` - → `GatewayPool.spec.healthCheckSettings` → `Site.spec.healthCheckSettings` +- `Site.spec.healthCheckSettings` applies to node-to-node routes within the same site. +- `SitePeering.spec.healthCheckSettings` applies to node-to-node routes between sites in that peering. +- `GatewayPool.spec.healthCheckSettings` governs same-pool gateway peers. +- `GatewayPoolPeering.spec.healthCheckSettings` applies to routes between gateway pools in that peering. + +For mesh peers, an explicit gateway-pool or pool-peering profile wins, followed by a `SiteGatewayPoolAssignment`, a `SitePeering`, and then the peer's `Site`. +For node-to-gateway peers, the node's site/pool assignment governs; gateway nodes use their explicit peer profile or the peer's pool profile, with a remote-site/pool assignment as fallback. +Shared-tunnel gateway peers may fall back to the local site's profile when no governing association exists; WireGuard gateway peers do not use that site fallback. +An explicitly disabled governing profile blocks every fallback. If multiple peerings define conflicting settings, peerings are processed in deterministic name order and the first profile is kept. diff --git a/docs/net/custom-resources.md b/docs/net/custom-resources.md index 8c50de44f..9dbbe9c69 100644 --- a/docs/net/custom-resources.md +++ b/docs/net/custom-resources.md @@ -520,8 +520,8 @@ spec: healthCheckSettings: enabled: true detectMultiplier: 3 - receiveInterval: 300ms - transmitInterval: 300ms + receiveInterval: 15s + transmitInterval: 15s # Optional: Tunnel encapsulation type (WireGuard, IPIP, GENEVE, VXLAN, None, or Auto; default: Auto) tunnelProtocol: Auto @@ -566,14 +566,15 @@ When sites are peered via SitePeering: The node agent resolves health check settings per route scope: -- `Site.spec.healthCheckSettings`: node-to-node routes within the same site. -- `SitePeering.spec.healthCheckSettings`: node-to-node routes between sites in that peering. -- `GatewayPool.spec.healthCheckSettings`: routes from nodes to peers in that gateway pool. +- `Site.spec.healthCheckSettings` applies to node-to-node routes within the same site. +- `SitePeering.spec.healthCheckSettings` applies to node-to-node routes between sites in that peering. +- `GatewayPool.spec.healthCheckSettings` governs same-pool gateway peers. +- `GatewayPoolPeering.spec.healthCheckSettings` applies to routes between gateway pools in that peering. -For routes to gateway pool peers, precedence is: - -1. `SiteGatewayPoolAssignment.spec.healthCheckSettings` (site-to-pool relationships) -2. `GatewayPool.spec.healthCheckSettings` (gateway-to-gateway relationships) +For mesh peers, an explicit gateway-pool or pool-peering profile wins, followed by a `SiteGatewayPoolAssignment`, a `SitePeering`, and then the peer's `Site`. +For node-to-gateway peers, the node's site/pool assignment governs; gateway nodes use their explicit peer profile or the peer's pool profile, with a remote-site/pool assignment as fallback. +Shared-tunnel gateway peers may fall back to the local site's profile when no governing association exists; WireGuard gateway peers do not use that site fallback. +An explicitly disabled governing profile blocks every fallback. If multiple peerings define conflicting health check settings for the same site or gateway pool, peerings are processed in deterministic name order, and the first profile is kept. diff --git a/internal/net/healthcheck/probe_phase_test.go b/internal/net/healthcheck/probe_phase_test.go index fa3e7a248..f197c0ee1 100644 --- a/internal/net/healthcheck/probe_phase_test.go +++ b/internal/net/healthcheck/probe_phase_test.go @@ -306,6 +306,47 @@ func TestShortenedIntervalsAllowFirstProbeNominalTimeout(t *testing.T) { } } +func TestSuccessiveSettingsUpdatesReboundTimeoutTransition(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + settings := DefaultSettings() + settings.TransmitInterval, settings.ReceiveInterval = time.Minute, time.Minute + s := newSession(sessionConfig{settings: settings, probePhaseSeed: &intervalEndProbePhaseSeed}) + s.state = StateUp + s.lastReceived = time.Now() + + settings.DetectMultiplier = 1 + s.updateSettings(settings) + + firstUpdate := time.Now() + if got := s.detectGraceUntil.Sub(firstUpdate); got != 2*time.Minute { + t.Fatalf("initial transition window=%v want %v", got, 2*time.Minute) + } + + settings.TransmitInterval = time.Second + s.updateSettings(settings) + + secondUpdate := time.Now() + if got := s.detectGraceUntil.Sub(secondUpdate); got != 61*time.Second { + t.Fatalf("rebounded transition window=%v want %v", got, 61*time.Second) + } + + settings.ReceiveInterval = 2 * time.Minute + s.updateSettings(settings) + + thirdUpdate := time.Now() + if got := s.detectGraceUntil.Sub(thirdUpdate); got != 121*time.Second { + t.Fatalf("rebounded increased-timeout window=%v want %v", got, 121*time.Second) + } + + settings.MaxBackoff = 5 * time.Minute + s.updateSettings(settings) + + if got := s.detectGraceUntil.Sub(thirdUpdate); got != 121*time.Second { + t.Fatalf("backoff-only update changed transition window to %v", got) + } + }) +} + func TestTimeoutTransitionRechecksFreshReply(t *testing.T) { settings := DefaultSettings() s := newSession(sessionConfig{settings: settings}) diff --git a/internal/net/healthcheck/session.go b/internal/net/healthcheck/session.go index 3c5a6403b..9e9468afe 100644 --- a/internal/net/healthcheck/session.go +++ b/internal/net/healthcheck/session.go @@ -181,6 +181,9 @@ func (s *session) updateSettings(settings HealthCheckSettings) { oldTimeout := s.detectTimeout() probeChanged := s.settings.TransmitInterval != settings.TransmitInterval + cadenceChanged := probeChanged || + s.settings.ReceiveInterval != settings.ReceiveInterval || + s.settings.DetectMultiplier != settings.DetectMultiplier s.settings = settings if probeChanged { @@ -188,7 +191,10 @@ func (s *session) updateSettings(settings HealthCheckSettings) { } newTimeout := s.detectTimeout() - if s.state == StateUp && newTimeout < oldTimeout { + now := time.Now() + + graceActive := now.Before(s.detectGraceUntil) + if s.state == StateUp && cadenceChanged && (newTimeout < oldTimeout || graceActive) { // A reply from the old cadence may already exceed the new timeout. // Allow the first newly scheduled probe one nominal reply timeout. // Receive-only changes retain the existing phase, at most one TX away. @@ -197,7 +203,9 @@ func (s *session) updateSettings(settings HealthCheckSettings) { firstProbe = s.probePhase(settings.TransmitInterval) } - s.detectGraceUntil = time.Now().Add(firstProbe).Add(newTimeout) + s.detectGraceUntil = now.Add(firstProbe).Add(newTimeout) + } else if cadenceChanged { + s.detectGraceUntil = time.Time{} } s.mu.Unlock() From d680f23978b1efb9ed7b392736f1e7f351030676 Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Fri, 18 Sep 2026 21:07:20 +0000 Subject: [PATCH 3/3] docs(net): clarify health-check CLI defaults Document the inherited detect multiplier and explain how receive and transmit intervals determine the down timeout. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 07b1482a-4420-4169-8102-a2d2026849e4 --- cmd/kubectl-unbounded/app/net/create.go | 6 +++--- cmd/kubectl-unbounded/app/net/create_test.go | 10 ++++++++++ 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/cmd/kubectl-unbounded/app/net/create.go b/cmd/kubectl-unbounded/app/net/create.go index c2dd74fb9..e9c990715 100644 --- a/cmd/kubectl-unbounded/app/net/create.go +++ b/cmd/kubectl-unbounded/app/net/create.go @@ -35,8 +35,8 @@ type healthCheckFlags struct { // addToFlags registers health check flags with a command. func (b *healthCheckFlags) addToFlags(cmd *cobra.Command) { cmd.Flags().BoolVar(&b.enabled, "health-check-enabled", false, "Enable UDP health probes over tunnels") - cmd.Flags().Int32Var(&b.detectMultiplier, "health-check-detect-multiplier", 0, "Number of missed probes before marking a peer down") - cmd.Flags().StringVar(&b.receiveInterval, "health-check-receive-interval", "", "Min interval between received probes before declaring down, e.g. 300ms (node default: 15s)") + cmd.Flags().Int32Var(&b.detectMultiplier, "health-check-detect-multiplier", 0, "Number of missed probes before marking a peer down (node default: 3)") + cmd.Flags().StringVar(&b.receiveInterval, "health-check-receive-interval", "", "Expected interval between received probes; down timeout is detect multiplier * max(receive, transmit), e.g. 300ms (node default: 15s)") cmd.Flags().StringVar(&b.transmitInterval, "health-check-transmit-interval", "", "Interval between transmitted health probes, e.g. 300ms (node default: 15s)") cmd.Flags().Int32Var(&b.tunnelMTU, "tunnel-mtu", 0, "MTU for tunnel interfaces in this scope") cmd.Flags().StringVar(&b.tunnelProtocol, "tunnel-protocol", "", "Tunnel encapsulation protocol (WireGuard, GENEVE, or Auto)") @@ -44,7 +44,7 @@ func (b *healthCheckFlags) addToFlags(cmd *cobra.Command) { return []string{"WireGuard", "GENEVE", "Auto"}, cobra.ShellCompDirectiveNoFileComp }) _ = cmd.RegisterFlagCompletionFunc("health-check-receive-interval", func(*cobra.Command, []string, string) ([]string, cobra.ShellCompDirective) { //nolint:errcheck - return cobra.AppendActiveHelp(nil, "Minimum interval between received health probes before declaring down. Duration, e.g. 300ms or 1s"), cobra.ShellCompDirectiveNoFileComp + return cobra.AppendActiveHelp(nil, "Expected interval between received health probes. Down timeout is detect multiplier * max(receive, transmit). Duration, e.g. 300ms or 1s"), cobra.ShellCompDirectiveNoFileComp }) _ = cmd.RegisterFlagCompletionFunc("health-check-transmit-interval", func(*cobra.Command, []string, string) ([]string, cobra.ShellCompDirective) { //nolint:errcheck return cobra.AppendActiveHelp(nil, "Interval between transmitted health probes. Duration, e.g. 300ms or 1s"), cobra.ShellCompDirectiveNoFileComp diff --git a/cmd/kubectl-unbounded/app/net/create_test.go b/cmd/kubectl-unbounded/app/net/create_test.go index 563252c1c..f09197c92 100644 --- a/cmd/kubectl-unbounded/app/net/create_test.go +++ b/cmd/kubectl-unbounded/app/net/create_test.go @@ -28,6 +28,16 @@ func TestHealthCheckFlagsPreserveRuntimeDefaults(t *testing.T) { } } + multiplier := cmd.Flags().Lookup("health-check-detect-multiplier") + if multiplier.DefValue != "0" || !strings.Contains(multiplier.Usage, "node default: 3") { + t.Fatal("detect multiplier must document the inherited default without serializing it") + } + + receive := cmd.Flags().Lookup("health-check-receive-interval") + if !strings.Contains(receive.Usage, "detect multiplier * max(receive, transmit)") { + t.Fatal("receive interval must document how it contributes to the down timeout") + } + if err := cmd.Flags().Set("health-check-transmit-interval", "60s"); err != nil { t.Fatal(err) }