feat(dns,cni): cluster-domain derivation (241/242) + topology-aware peer latency (245)

pkg/monitors/network/cni.go

@@ -67,10 +67,17 @@ type ConnectivityConfig struct {
 	PingCount int
 	// PingTimeout is the timeout for each ping.
 	PingTimeout time.Duration
-	// WarningLatency is the latency threshold for warnings.
+	// WarningLatency is the latency threshold for warnings (same-zone peers).
 	WarningLatency time.Duration
-	// CriticalLatency is the latency threshold for critical conditions.
+	// CriticalLatency is the latency threshold for critical conditions (same-zone peers).
 	CriticalLatency time.Duration
+	// CrossZoneWarningLatency/CrossZoneCriticalLatency are the latency thresholds applied
+	// to peers in a DIFFERENT topology zone (e.g. cross-site nodes). When 0 (unset), the
+	// same-zone thresholds above are used for all peers — so topology awareness is inert
+	// until explicitly configured. Set these looser than the same-zone thresholds so normal
+	// inter-site WAN latency does not raise a false NetworkDegraded.
+	CrossZoneWarningLatency  time.Duration
+	CrossZoneCriticalLatency time.Duration
 	// FailureThreshold is consecutive failures before marking peer unreachable.
 	FailureThreshold int
 	// MinReachablePeers is the percentage of peers that must be reachable.
@@ -363,6 +370,20 @@ func parseCNIConfig(configMap map[string]interface{}) (*CNIMonitorConfig, error)
 			}
 			config.Connectivity.CriticalLatency = duration
 		}
+		if v, ok := connMap["crossZoneWarningLatency"]; ok {
+			duration, err := parseDuration(v)
+			if err != nil {
+				return nil, fmt.Errorf("invalid connectivity.crossZoneWarningLatency: %w", err)
+			}
+			config.Connectivity.CrossZoneWarningLatency = duration
+		}
+		if v, ok := connMap["crossZoneCriticalLatency"]; ok {
+			duration, err := parseDuration(v)
+			if err != nil {
+				return nil, fmt.Errorf("invalid connectivity.crossZoneCriticalLatency: %w", err)
+			}
+			config.Connectivity.CrossZoneCriticalLatency = duration
+		}
 		if failureThreshold, ok := connMap["failureThreshold"]; ok {
 			switch v := failureThreshold.(type) {
 			case int:
@@ -425,6 +446,17 @@ func ValidateCNIConfig(config types.MonitorConfig) error {
 }
 
 // checkCNI performs the CNI connectivity check.
+// warningLatencyFor returns the warning-latency threshold to apply to a peer. A peer
+// in a different topology zone uses CrossZoneWarningLatency when it is configured
+// (non-zero); otherwise the same-zone WarningLatency is used. This keeps topology
+// awareness inert until both zone labels exist AND a cross-zone threshold is set.
+func (m *CNIMonitor) warningLatencyFor(peer Peer) time.Duration {
+	if !peer.SameZone && m.config.Connectivity.CrossZoneWarningLatency > 0 {
+		return m.config.Connectivity.CrossZoneWarningLatency
+	}
+	return m.config.Connectivity.WarningLatency
+}
+
 func (m *CNIMonitor) checkCNI(ctx context.Context) (*types.Status, error) {
 	status := types.NewStatus(m.name)
 
@@ -490,10 +522,10 @@ func (m *CNIMonitor) checkCNI(ctx context.Context) (*types.Status, error) {
 			reachableCount++
 			totalLatency += peerStatus.AvgLatency
 
-			// Check for high latency (collect but don't emit individual events)
-			if peerStatus.AvgLatency > m.config.Connectivity.CriticalLatency {
-				highLatencyPeers = append(highLatencyPeers, fmt.Sprintf("%s (%.2fms)", result.peer.NodeName, float64(peerStatus.AvgLatency)/float64(time.Millisecond)))
-			} else if peerStatus.AvgLatency > m.config.Connectivity.WarningLatency {
+			// Check for high latency (collect but don't emit individual events).
+			// Cross-zone peers use the looser cross-zone warning threshold when configured,
+			// so normal inter-site WAN latency does not raise a false NetworkDegraded.
+			if peerStatus.AvgLatency > m.warningLatencyFor(result.peer) {
 				highLatencyPeers = append(highLatencyPeers, fmt.Sprintf("%s (%.2fms)", result.peer.NodeName, float64(peerStatus.AvgLatency)/float64(time.Millisecond)))
 			}
 		} else {

pkg/monitors/network/cni_topology_test.go

@@ -0,0 +1,57 @@
+package network
+
+import (
+	"testing"
+	"time"
+)
+
+func TestWarningLatencyFor(t *testing.T) {
+	mk := func(warn, crossWarn time.Duration) *CNIMonitor {
+		return &CNIMonitor{config: &CNIMonitorConfig{
+			Connectivity: ConnectivityConfig{
+				WarningLatency:          warn,
+				CrossZoneWarningLatency: crossWarn,
+			},
+		}}
+	}
+	sameZone := Peer{NodeName: "a", SameZone: true}
+	crossZone := Peer{NodeName: "b", SameZone: false}
+
+	// Cross-zone threshold configured: same-zone peer uses tight, cross-zone uses loose.
+	m := mk(200*time.Millisecond, 800*time.Millisecond)
+	if got := m.warningLatencyFor(sameZone); got != 200*time.Millisecond {
+		t.Errorf("same-zone threshold = %v, want 200ms", got)
+	}
+	if got := m.warningLatencyFor(crossZone); got != 800*time.Millisecond {
+		t.Errorf("cross-zone threshold = %v, want 800ms", got)
+	}
+
+	// Cross-zone threshold unset (0): topology awareness is inert — both use WarningLatency.
+	inert := mk(200*time.Millisecond, 0)
+	if got := inert.warningLatencyFor(crossZone); got != 200*time.Millisecond {
+		t.Errorf("inert cross-zone threshold = %v, want 200ms (same as warning)", got)
+	}
+	if got := inert.warningLatencyFor(sameZone); got != 200*time.Millisecond {
+		t.Errorf("inert same-zone threshold = %v, want 200ms", got)
+	}
+}
+
+func TestConnectivityCrossZoneLatencyParsing(t *testing.T) {
+	cfg, err := parseCNIConfig(map[string]interface{}{
+		"connectivity": map[string]interface{}{
+			"warningLatency":           "200ms",
+			"criticalLatency":          "500ms",
+			"crossZoneWarningLatency":  "1s",
+			"crossZoneCriticalLatency": "2s",
+		},
+	})
+	if err != nil {
+		t.Fatalf("parseCNIConfig: %v", err)
+	}
+	if cfg.Connectivity.CrossZoneWarningLatency != time.Second {
+		t.Errorf("crossZoneWarningLatency = %v, want 1s", cfg.Connectivity.CrossZoneWarningLatency)
+	}
+	if cfg.Connectivity.CrossZoneCriticalLatency != 2*time.Second {
+		t.Errorf("crossZoneCriticalLatency = %v, want 2s", cfg.Connectivity.CrossZoneCriticalLatency)
+	}
+}

pkg/monitors/network/dns.go

@@ -1232,9 +1232,23 @@
 //
 //nolint:gocyclo // config defaults are inherently branchy across many optional fields
 func (c *DNSMonitorConfig) applyDefaults() error {
-	// Default cluster domains - only apply if not explicitly set (nil vs empty slice)
+	// Default resolver path first — cluster-domain derivation reads it below.
+	if c.ResolverPath == "" {
+		c.ResolverPath = "/etc/resolv.conf"
+	}
+
+	// Default cluster domains - only apply if not explicitly set (nil vs empty slice).
+	// An explicit empty slice ([]) means "cluster DNS check disabled" and is left as-is;
+	// only a nil (unset) value gets a default.
+	//
+	// The default is DERIVED from the resolver's search domains rather than hardcoded to
+	// "kubernetes.default.svc.cluster.local". A hardcoded cluster.local produces a
+	// fleet-wide false ClusterDNSResolutionFailed on clusters that use a CUSTOM cluster
+	// domain (the target simply does not resolve). Derivation reads the real domain from
+	// /etc/resolv.conf `search` and builds the correct kubernetes.default.svc.<domain>,
+	// falling back to cluster.local only when derivation is not possible.
 	if c.ClusterDomains == nil {
-		c.ClusterDomains = []string{"kubernetes.default.svc.cluster.local"}
+		c.ClusterDomains = defaultClusterDomains(c.ResolverPath)
 	}
 
 	// Default external domains - only apply if not explicitly set (nil vs empty slice)
@@ -1247,11 +1261,6 @@
 		c.LatencyThreshold = 1 * time.Second
 	}
 
-	// Default nameserver check enabled
-	if c.ResolverPath == "" {
-		c.ResolverPath = "/etc/resolv.conf"
-	}
-
 	// Default failure count threshold
 	if c.FailureCountThreshold == 0 {
 		c.FailureCountThreshold = 3
@@ -2338,6 +2347,75 @@
 	return nameservers, nil
 }
 
+// defaultClusterDomains returns the default in-cluster DNS probe target(s). It derives
+// the cluster domain from the resolver's search domains and builds
+// "kubernetes.default.svc.<domain>", falling back to the well-known
+// "kubernetes.default.svc.cluster.local" when derivation is not possible. This prevents
+// the false ClusterDNSResolutionFailed that a hardcoded cluster.local causes on clusters
+// with a custom cluster domain.
+func defaultClusterDomains(resolverPath string) []string {
+	if domain, ok := deriveClusterDomainFromResolver(resolverPath); ok {
+		return []string{"kubernetes.default.svc." + domain}
+	}
+	return []string{"kubernetes.default.svc.cluster.local"}
+}
+
+// deriveClusterDomainFromResolver reads resolverPath and extracts the Kubernetes
+// cluster domain from its `search` line. Returns ("", false) if the file can't be read
+// or no cluster domain can be identified.
+func deriveClusterDomainFromResolver(resolverPath string) (string, bool) {
+	file, err := os.Open(resolverPath)
+	if err != nil {
+		return "", false
+	}
+	defer file.Close()
+
+	scanner := bufio.NewScanner(file)
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if line == "" || strings.HasPrefix(line, "#") {
+			continue
+		}
+		if strings.HasPrefix(line, "search") {
+			fields := strings.Fields(line)
+			if len(fields) >= 2 {
+				return deriveClusterDomain(fields[1:])
+			}
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return "", false
+	}
+	return "", false
+}
+
+// deriveClusterDomain extracts the cluster domain from a list of resolver search
+// domains. A Kubernetes pod's search list looks like:
+//
+//	<namespace>.svc.<clusterDomain>  svc.<clusterDomain>  <clusterDomain>
+//
+// so the cluster domain is the suffix after the "svc." label. It prefers the canonical
+// "svc.<clusterDomain>" entry, then any "<ns>.svc.<clusterDomain>" entry. Returns
+// ("", false) when no svc-scoped search domain is present (e.g. a non-Kubernetes
+// resolver), so the caller can fall back rather than probe a bogus target.
+func deriveClusterDomain(searchDomains []string) (string, bool) {
+	// Prefer the canonical middle entry: "svc.<clusterDomain>".
+	for _, d := range searchDomains {
+		d = strings.TrimSuffix(strings.TrimSpace(d), ".")
+		if strings.HasPrefix(d, "svc.") && len(d) > len("svc.") {
+			return d[len("svc."):], true
+		}
+	}
+	// Fall back to "<namespace>.svc.<clusterDomain>".
+	for _, d := range searchDomains {
+		d = strings.TrimSuffix(strings.TrimSpace(d), ".")
+		if _, domain, found := strings.Cut(d, ".svc."); found && domain != "" {
+			return domain, true
+		}
+	}
+	return "", false
+}
+
 // exclusiveCond describes one member of a mutually-exclusive condition group.
 type exclusiveCond struct{ Type, Reason, Message string }
 

pkg/monitors/network/dns_clusterdomain_test.go

@@ -0,0 +1,130 @@
+package network
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestDeriveClusterDomain(t *testing.T) {
+	tests := []struct {
+		name   string
+		search []string
+		want   string
+		wantOK bool
+	}{
+		{
+			name:   "standard cluster.local search list",
+			search: []string{"default.svc.cluster.local", "svc.cluster.local", "cluster.local"},
+			want:   "cluster.local",
+			wantOK: true,
+		},
+		{
+			name:   "custom cluster domain (the incident case)",
+			search: []string{"default.svc.k8s.example.com", "svc.k8s.example.com", "k8s.example.com"},
+			want:   "k8s.example.com",
+			wantOK: true,
+		},
+		{
+			name:   "only the ns-scoped entry present (no bare svc.)",
+			search: []string{"kube-system.svc.cluster.local"},
+			want:   "cluster.local",
+			wantOK: true,
+		},
+		{
+			name:   "trailing dots tolerated",
+			search: []string{"svc.cluster.local."},
+			want:   "cluster.local",
+			wantOK: true,
+		},
+		{
+			name:   "non-kubernetes resolver -> no derivation",
+			search: []string{"corp.example.com", "example.com"},
+			want:   "",
+			wantOK: false,
+		},
+		{
+			name:   "empty search",
+			search: nil,
+			want:   "",
+			wantOK: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, ok := deriveClusterDomain(tt.search)
+			if got != tt.want || ok != tt.wantOK {
+				t.Errorf("deriveClusterDomain(%v) = (%q, %v), want (%q, %v)", tt.search, got, ok, tt.want, tt.wantOK)
+			}
+		})
+	}
+}
+
+func writeResolv(t *testing.T, content string) string {
+	t.Helper()
+	p := filepath.Join(t.TempDir(), "resolv.conf")
+	if err := os.WriteFile(p, []byte(content), 0o644); err != nil {
+		t.Fatalf("write resolv.conf: %v", err)
+	}
+	return p
+}
+
+func TestDeriveClusterDomainFromResolver(t *testing.T) {
+	custom := writeResolv(t, "search default.svc.k8s.example.com svc.k8s.example.com k8s.example.com\nnameserver 10.43.0.10\noptions ndots:5\n")
+	if d, ok := deriveClusterDomainFromResolver(custom); !ok || d != "k8s.example.com" {
+		t.Errorf("custom domain: got (%q,%v), want (k8s.example.com,true)", d, ok)
+	}
+
+	// Missing file -> no derivation, no panic.
+	if d, ok := deriveClusterDomainFromResolver(filepath.Join(t.TempDir(), "nope")); ok || d != "" {
+		t.Errorf("missing file: got (%q,%v), want ('',false)", d, ok)
+	}
+
+	// No search line -> no derivation.
+	noSearch := writeResolv(t, "nameserver 1.1.1.1\n")
+	if d, ok := deriveClusterDomainFromResolver(noSearch); ok || d != "" {
+		t.Errorf("no search line: got (%q,%v), want ('',false)", d, ok)
+	}
+}
+
+func TestDefaultClusterDomains(t *testing.T) {
+	// Custom-domain cluster: derived target must use the real domain, NOT cluster.local.
+	custom := writeResolv(t, "search default.svc.mesh.internal svc.mesh.internal mesh.internal\nnameserver 10.96.0.10\n")
+	got := defaultClusterDomains(custom)
+	want := "kubernetes.default.svc.mesh.internal"
+	if len(got) != 1 || got[0] != want {
+		t.Errorf("defaultClusterDomains(custom) = %v, want [%q]", got, want)
+	}
+
+	// Non-derivable resolver: fall back to the well-known cluster.local target.
+	fallback := defaultClusterDomains(filepath.Join(t.TempDir(), "missing"))
+	if len(fallback) != 1 || fallback[0] != "kubernetes.default.svc.cluster.local" {
+		t.Errorf("defaultClusterDomains(fallback) = %v, want [kubernetes.default.svc.cluster.local]", fallback)
+	}
+}
+
+// TestApplyDefaultsDerivesClusterDomain verifies the config wiring: a nil ClusterDomains
+// gets the derived default, while an explicit empty slice (cluster DNS disabled) is left
+// untouched — the property that keeps this change from re-enabling the check on a fleet
+// that intentionally set clusterDomains: [].
+func TestApplyDefaultsDerivesClusterDomain(t *testing.T) {
+	resolv := writeResolv(t, "search default.svc.custom.zone svc.custom.zone custom.zone\nnameserver 10.96.0.10\n")
+
+	// nil -> derived
+	c := &DNSMonitorConfig{ResolverPath: resolv}
+	if err := c.applyDefaults(); err != nil {
+		t.Fatalf("applyDefaults: %v", err)
+	}
+	if len(c.ClusterDomains) != 1 || c.ClusterDomains[0] != "kubernetes.default.svc.custom.zone" {
+		t.Errorf("nil ClusterDomains derived = %v, want [kubernetes.default.svc.custom.zone]", c.ClusterDomains)
+	}
+
+	// explicit empty slice -> left disabled (NOT re-enabled)
+	disabled := &DNSMonitorConfig{ResolverPath: resolv, ClusterDomains: []string{}}
+	if err := disabled.applyDefaults(); err != nil {
+		t.Fatalf("applyDefaults: %v", err)
+	}
+	if len(disabled.ClusterDomains) != 0 {
+		t.Errorf("explicit empty ClusterDomains should stay empty, got %v", disabled.ClusterDomains)
+	}
+}