feat(egress): upstream health probes, active list, configurable probe name (opensandbox-group#655)

* feat(egress): upstream health probes, active list, configurable probe name

* feat(egress): upstream health probes, active list, configurable probe name

Author: Pangjiping

.github/workflows/egress-test.yaml.yml

@@ -61,6 +61,12 @@ jobs:
           chmod +x tests/smoke-dynamic-ip.sh
           ./tests/smoke-dynamic-ip.sh
 
+      - name: Run dns upstream probe test
+        working-directory: components/egress
+        run: |
+          chmod +x tests/smoke-dns-upstream-probe.sh
+          ./tests/smoke-dns-upstream-probe.sh
+
   bench:
     runs-on: ubuntu-latest
     steps:

components/egress/main.go

@@ -75,7 +75,7 @@ func main() {
 	if err != nil {
 		log.Fatalf("failed to init dns proxy: %v", err)
 	}
-	if err := proxy.Start(); err != nil {
+	if err := proxy.Start(ctx); err != nil {
 		log.Fatalf("failed to start dns proxy: %v", err)
 	}
 	log.Infof("dns proxy started on 127.0.0.1:15353")

components/egress/pkg/constants/configuration.go

@@ -33,10 +33,10 @@ const (
 	EnvNameserverExempt = "OPENSANDBOX_EGRESS_NAMESERVER_EXEMPT"
 
 	// EnvDNSUpstream comma-separated upstream resolvers; each address must be a literal IPv4/IPv6 (optional :port). Hostnames are rejected (DNS recursion via REDIRECT).
-	EnvDNSUpstream = "OPENSANDBOX_EGRESS_DNS_UPSTREAM"
-
-	// EnvDNSUpstreamTimeout is the per-upstream DNS forward timeout in whole seconds (default 5). Invalid/empty uses default.
-	EnvDNSUpstreamTimeout = "OPENSANDBOX_EGRESS_DNS_UPSTREAM_TIMEOUT"
+	EnvDNSUpstream                 = "OPENSANDBOX_EGRESS_DNS_UPSTREAM"
+	EnvDNSUpstreamTimeout          = "OPENSANDBOX_EGRESS_DNS_UPSTREAM_TIMEOUT"
+	EnvDNSUpstreamProbe            = "OPENSANDBOX_EGRESS_DNS_UPSTREAM_PROBE"
+	EnvDNSUpstreamProbeIntervalSec = "OPENSANDBOX_EGRESS_DNS_UPSTREAM_PROBE_INTERVAL_SEC"
 )
 
 const (
@@ -45,8 +45,7 @@ const (
 )
 
 const (
-	DefaultEgressServerAddr = ":18080"
-	// ResolvNameserverCap is the max number of nameserver lines read from /etc/resolv.conf for nft allow-list merge and auto upstream chain (not configurable).
+	DefaultEgressServerAddr      = ":18080"
 	ResolvNameserverCap          = 10
 	DefaultMaxEgressRules        = 4096
 	DefaultDNSUpstreamTimeoutSec = 5

components/egress/pkg/dnsproxy/proxy.go

@@ -15,6 +15,7 @@
 package dnsproxy
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"net/netip"
@@ -44,7 +45,12 @@ type Proxy struct {
 	alwaysDeny              []policy.EgressRule
 	alwaysAllow             []policy.EgressRule
 	listenAddr              string
-	upstreams               []string // ordered resolver chain; try next on forward failure
+	upstreams               []string // ordered resolver chain from discovery (immutable after New)
+	upstreamMu              sync.RWMutex
+	activeUpstreams         []string // healthy subset; same order as upstreams; used for forwarding
+	upstreamProbeName       string   // wire name for probe (FQDN or "." for root)
+	upstreamProbeQType      uint16   // dns.TypeA or dns.TypeNS etc.
+	upstreamProbeInterval   time.Duration
 	upstreamExchangeTimeout time.Duration
 	servers                 []*dns.Server
 	shutdownOnce            sync.Once
@@ -70,9 +76,14 @@ func New(p *policy.NetworkPolicy, listenAddr string, alwaysDeny, alwaysAllow []p
 	if err != nil {
 		return nil, err
 	}
+	probeName, probeQType := upstreamProbeFromEnv()
 	proxy := &Proxy{
 		listenAddr:              listenAddr,
 		upstreams:               upstreams,
+		activeUpstreams:         append([]string(nil), upstreams...),
+		upstreamProbeName:       probeName,
+		upstreamProbeQType:      probeQType,
+		upstreamProbeInterval:   upstreamProbeIntervalFromEnv(),
 		upstreamExchangeTimeout: upstreamExchangeTimeoutFromEnv(),
 		userPolicy:              ensurePolicyDefaults(p),
 		alwaysDeny:              append([]policy.EgressRule(nil), alwaysDeny...),
@@ -101,7 +112,7 @@ func upstreamExchangeTimeoutFromEnv() time.Duration {
 	return time.Duration(n) * time.Second
 }
 
-func (p *Proxy) Start() error {
+func (p *Proxy) Start(ctx context.Context) error {
 	handler := dns.HandlerFunc(p.serveDNS)
 
 	udpServer := &dns.Server{Addr: p.listenAddr, Net: "udp", Handler: handler}
@@ -118,13 +129,18 @@ func (p *Proxy) Start() error {
 		}()
 	}
 
+	timer := time.NewTimer(200 * time.Millisecond)
+	defer timer.Stop()
 	select {
 	case err := <-errCh:
 		return fmt.Errorf("dns proxy failed: %w", err)
-	case <-time.After(200 * time.Millisecond):
-		// small grace window; running fine
-		return nil
+	case <-timer.C:
+		// listeners bound; start upstream probes only after DNS servers are up
 	}
+
+	go p.runUpstreamProbes(ctx)
+
+	return nil
 }
 
 // Shutdown stops UDP/TCP DNS listeners. Safe to call more than once.
@@ -193,8 +209,9 @@ func (p *Proxy) maybeNotifyResolved(domain string, resp *dns.Msg) {
 }
 
 func (p *Proxy) forward(r *dns.Msg) (*dns.Msg, error) {
+	list := p.forwardUpstreams()
 	var lastErr error
-	for i, upstream := range p.upstreams {
+	for i, upstream := range list {
 		c := &dns.Client{
 			Timeout: p.upstreamExchangeTimeout,
 			Dialer:  p.dialerForUpstream(upstream),
@@ -209,7 +226,7 @@ func (p *Proxy) forward(r *dns.Msg) (*dns.Msg, error) {
 			lastErr = fmt.Errorf("nil response from %s", upstream)
 			continue
 		}
-		if tryNext, reason := p.shouldFailoverAfterResponse(resp, i); tryNext {
+		if tryNext, reason := p.shouldFailoverAfterResponse(resp, i, len(list)); tryNext {
 			lastErr = fmt.Errorf("%s from %s", reason, upstream)
 			log.Warnf("[dns] upstream %s: %s; trying next", upstream, reason)
 			continue
@@ -225,15 +242,15 @@ func (p *Proxy) forward(r *dns.Msg) (*dns.Msg, error) {
 // shouldFailoverAfterResponse returns whether to try the next upstream.
 // NXDOMAIN is not retried. NOERROR with an empty Answer (NODATA) is retried when another upstream exists:
 // a broken or non-recursive first hop may return empty NOERROR while a public resolver succeeds.
-func (p *Proxy) shouldFailoverAfterResponse(resp *dns.Msg, upstreamIdx int) (tryNext bool, reason string) {
+func (p *Proxy) shouldFailoverAfterResponse(resp *dns.Msg, upstreamIdx, upstreamCount int) (tryNext bool, reason string) {
 	if resp == nil {
 		return true, "nil response"
 	}
 	switch resp.Rcode {
 	case dns.RcodeNameError:
 		return false, ""
 	case dns.RcodeSuccess:
-		if len(resp.Answer) == 0 && upstreamIdx < len(p.upstreams)-1 {
+		if len(resp.Answer) == 0 && upstreamIdx < upstreamCount-1 {
 			return true, "empty NOERROR"
 		}
 		return false, ""
@@ -246,12 +263,13 @@ func (p *Proxy) shouldFailoverAfterResponse(resp *dns.Msg, upstreamIdx int) (try
 	}
 }
 
-// UpstreamHost returns the host part of the first upstream resolver, empty on parse error.
+// UpstreamHost returns the host part of the first upstream resolver used for forwarding, empty on parse error.
 func (p *Proxy) UpstreamHost() string {
-	if len(p.upstreams) == 0 {
+	list := p.forwardUpstreams()
+	if len(list) == 0 {
 		return ""
 	}
-	host, _, err := net.SplitHostPort(p.upstreams[0])
+	host, _, err := net.SplitHostPort(list[0])
 	if err != nil {
 		return ""
 	}
@@ -262,15 +280,17 @@ func (p *Proxy) UpstreamHost() string {
 // Passing nil reverts to the default deny-all policy.
 func (p *Proxy) UpdatePolicy(newPolicy *policy.NetworkPolicy) {
 	p.policyMu.Lock()
+	defer p.policyMu.Unlock()
+
 	p.userPolicy = ensurePolicyDefaults(newPolicy)
 	p.refreshEffectivePolicy()
-	p.policyMu.Unlock()
 }
 
 // CurrentPolicy returns the user policy (POST/PATCH/GET), not the always-deny/allow overlay.
 func (p *Proxy) CurrentPolicy() *policy.NetworkPolicy {
 	p.policyMu.RLock()
 	defer p.policyMu.RUnlock()
+
 	return p.userPolicy
 }
 

components/egress/pkg/dnsproxy/upstream.go

@@ -0,0 +1,149 @@
+// Copyright 2026 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dnsproxy
+
+import (
+	"context"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/miekg/dns"
+
+	"github.com/alibaba/opensandbox/egress/pkg/constants"
+	"github.com/alibaba/opensandbox/egress/pkg/log"
+)
+
+const defaultUpstreamProbeInterval = 30 * time.Second
+
+func upstreamProbeIntervalFromEnv() time.Duration {
+	s := strings.TrimSpace(os.Getenv(constants.EnvDNSUpstreamProbeIntervalSec))
+	if s == "" {
+		return defaultUpstreamProbeInterval
+	}
+	n, err := strconv.Atoi(s)
+	if err != nil || n < 1 {
+		return defaultUpstreamProbeInterval
+	}
+	if n > 3600 {
+		n = 3600
+	}
+	return time.Duration(n) * time.Second
+}
+
+// upstreamProbeFromEnv returns the DNS question used for upstream liveness checks.
+// Default is root IN NS (primers/recursors answer without resolving a public TLD).
+// Set OPENSANDBOX_EGRESS_DNS_UPSTREAM_PROBE to an FQDN that your resolvers always
+// answer (e.g. split-horizon internal name) when the default is inappropriate.
+func upstreamProbeFromEnv() (name string, qtype uint16) {
+	raw := strings.TrimSpace(os.Getenv(constants.EnvDNSUpstreamProbe))
+	if raw == "" || raw == "." {
+		return ".", dns.TypeNS
+	}
+	return dns.Fqdn(raw), dns.TypeA
+}
+
+func (p *Proxy) runUpstreamProbes(ctx context.Context) {
+	p.probeUpstreams()
+
+	t := time.NewTicker(p.upstreamProbeInterval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			p.probeUpstreams()
+		}
+	}
+}
+
+// forwardUpstreams returns the ordered list used for DNS forwarding: last healthy probe
+// results when non-empty; otherwise the configured chain (e.g. before first probe).
+func (p *Proxy) forwardUpstreams() []string {
+	p.upstreamMu.RLock()
+	active := p.activeUpstreams
+	p.upstreamMu.RUnlock()
+	if len(active) > 0 {
+		return active
+	}
+	return p.upstreams
+}
+
+// probeUpstreams checks each configured resolver in parallel and refreshes activeUpstreams.
+// If every probe fails, the full upstream list is kept so forwarding still attempts all resolvers.
+func (p *Proxy) probeUpstreams() {
+	all := p.upstreams
+	if len(all) == 0 {
+		return
+	}
+
+	timeout := probeExchangeTimeout(p.upstreamExchangeTimeout)
+	healthy := make([]bool, len(all))
+	var wg sync.WaitGroup
+	for i := range all {
+		wg.Add(1)
+		go func(idx int, addr string) {
+			defer wg.Done()
+			healthy[idx] = p.probeOneUpstream(addr, timeout)
+		}(i, all[i])
+	}
+	wg.Wait()
+
+	var active []string
+	for i := range all {
+		if healthy[i] {
+			active = append(active, all[i])
+		}
+	}
+	if len(active) == 0 {
+		log.Warnf("[dns] all upstream probes failed; using full upstream list for forwarding")
+		active = append([]string(nil), all...)
+	}
+
+	p.upstreamMu.Lock()
+	p.activeUpstreams = active
+	p.upstreamMu.Unlock()
+}
+
+func probeExchangeTimeout(upstreamTimeout time.Duration) time.Duration {
+	const maxProbe = 2 * time.Second
+	if upstreamTimeout <= 0 {
+		return maxProbe
+	}
+	if upstreamTimeout > maxProbe {
+		return maxProbe
+	}
+	return upstreamTimeout
+}
+
+func (p *Proxy) probeOneUpstream(addr string, timeout time.Duration) bool {
+	m := new(dns.Msg)
+	m.SetQuestion(p.upstreamProbeName, p.upstreamProbeQType)
+	m.RecursionDesired = true
+
+	c := &dns.Client{
+		Timeout: timeout,
+		Dialer:  p.dialerForUpstream(addr),
+	}
+	resp, _, err := c.Exchange(m, addr)
+	if err != nil {
+		log.Errorf("[dns] upstream probe %s failed: %v", addr, err)
+		return false
+	}
+	return resp != nil
+}