feat(egress): count DNS resolution and nftables update failures

components/egress/docs/opentelemetry.md

@@ -11,9 +11,11 @@ This page lists the OpenTelemetry metrics currently implemented in egress.
 | Metric | Type | Unit | Meaning |
 |---|---|---|---|
 | `egress.dns.query.duration` | Histogram | `s` | Upstream DNS forward latency (recorded for allowed queries). |
+| `egress.dns.query.failed_total` | Counter | - | Queries the proxy could not resolve, by `reason`. |
 | `egress.policy.denied_total` | Counter | - | Number of DNS queries denied by policy. |
 | `egress.nftables.rules.count` | Observable Gauge | `{element}` | Approximate policy size after last successful static apply. |
 | `egress.nftables.updates.count` | Counter | - | Number of successful nftables updates (static apply + dynamic IP add). |
+| `egress.nftables.updates.failed_total` | Counter | - | nftables updates that failed, by `operation`. |
 | `egress.system.memory.usage_bytes` | Observable Gauge | `By` | System memory used bytes (Linux: gopsutil; non-Linux build: `0`). |
 | `egress.system.cpu.utilization` | Observable Gauge | `1` | CPU busy ratio in `[0,1]` (Linux: gopsutil; non-Linux build: `0`). |
 
@@ -42,6 +44,37 @@ per-exchange cap — has bigger problems than a percentile.
 Note both successful and failed lookups feed this histogram, so its tail mixes slow
 resolutions with exhausted retry chains.
 
+## Failure Signals
+
+`egress.dns.query.failed_total` and `egress.policy.denied_total` answer different
+questions, and confusing them inverts the diagnosis:
+
+- **denied** — the policy did its job. The workload asked for something it is not allowed
+  to reach. Expected traffic in a working system.
+- **failed** — the sidecar could not do its job. The workload asked for something allowed
+  and got `SERVFAIL`. Never expected.
+
+`reason` comes from a closed set, so the counter's cardinality is fixed and neither the
+queried name nor the error text is ever attached:
+
+| `reason` | Meaning |
+|---|---|
+| `no_upstreams` | No resolvers configured or discovered. |
+| `upstream_error` | Every resolver failed to answer (network error, timeout). |
+| `empty_response` | A resolver returned a nil message. |
+| `rcode` | The last resolver answered with a failover-worthy rcode, e.g. `SERVFAIL`. |
+
+`egress.nftables.updates.failed_total` covers the other silent failure. Its `operation`
+attribute is one of `static_apply`, `dynamic_add` or `remove`; `dynamic_add` is the one to
+alert on, because a failed add means the kernel never learned about IPs the policy allows,
+so the chain drops traffic that should pass — which looks exactly like a policy denial from
+inside the sandbox while `egress.policy.denied_total` stays flat.
+
+A `static_apply` failure happens during startup, where the sidecar logs and exits. Metrics
+leave through a periodic reader and `os.Exit` skips the deferred shutdown, so that path
+flushes telemetry explicitly before terminating — otherwise the one sample explaining why the
+sidecar died would never be exported.
+
 ## Shared Attributes
 
 All egress metrics may include shared attributes:

components/egress/nft.go

@@ -26,6 +26,7 @@ import (
 	"github.com/alibaba/opensandbox/egress/pkg/log"
 	"github.com/alibaba/opensandbox/egress/pkg/nftables"
 	"github.com/alibaba/opensandbox/egress/pkg/policy"
+	"github.com/alibaba/opensandbox/egress/pkg/telemetry"
 )
 
 // createNftManager is non-nil only when mode includes the nft token (e.g. dns+nft).
@@ -48,6 +49,14 @@ func setupNft(ctx context.Context, nftMgr nftApplier, initialPolicy *policy.Netw
 	merged := policy.MergeAlwaysOverlay(initialPolicy, alwaysDeny, alwaysAllow)
 	policyWithNS := merged.WithExtraAllowIPs(nameserverIPs)
 	if err := nftMgr.ApplyStatic(ctx, policyWithNS); err != nil {
+		// ApplyStatic recorded the failure, but Fatalf calls os.Exit and the periodic
+		// reader would never export it, nor would main's deferred shutdown run. Flush
+		// first, so the one sample explaining why the sidecar died actually leaves.
+		flushCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+		if flushErr := telemetry.ForceFlush(flushCtx); flushErr != nil {
+			log.Warnf("failed to flush telemetry before exit: %v", flushErr)
+		}
+		cancel()
 		log.Fatalf("nftables static apply failed: %v", err)
 	}
 	log.Infof("nftables static policy applied (table inet opensandbox); DNS-resolved IPs will be added to dynamic allow sets")

components/egress/pkg/dnsproxy/proxy.go

@@ -184,10 +184,11 @@ func (p *Proxy) serveDNS(w dns.ResponseWriter, r *dns.Msg) {
 	}
 
 	start := time.Now()
-	resp, err := p.forward(r)
+	resp, failure, err := p.forward(r)
 	elapsed := time.Since(start).Seconds()
 	if err != nil {
 		telemetry.RecordDNSForward(elapsed)
+		telemetry.RecordDNSQueryFailed(failure)
 		logOutboundDNS(host, nil, "", err.Error())
 		fail := new(dns.Msg)
 		fail.SetRcode(r, dns.RcodeServerFailure)
@@ -230,9 +231,13 @@ func (p *Proxy) maybeNotifyResolved(domain string, resp *dns.Msg) {
 	p.onResolved(domain, ips)
 }
 
-func (p *Proxy) forward(r *dns.Msg) (*dns.Msg, error) {
+// forward returns the response, or the bounded failure reason (a telemetry.DNSFailure*
+// constant) alongside the error. The reason is what the last attempted upstream failed
+// with: the loop keeps trying, so only the final outcome is reported.
+func (p *Proxy) forward(r *dns.Msg) (*dns.Msg, string, error) {
 	list := p.forwardUpstreams()
 	var lastErr error
+	lastFailure := telemetry.DNSFailureNoUpstreams
 	for _, upstream := range list {
 		const upstreamUDPSize = 4096
 		query := r.Copy()
@@ -247,24 +252,27 @@ func (p *Proxy) forward(r *dns.Msg) (*dns.Msg, error) {
 		resp, _, err := c.Exchange(query, upstream)
 		if err != nil {
 			lastErr = err
+			lastFailure = telemetry.DNSFailureUpstreamError
 			log.Warnf("[dns] upstream %s exchange error: %v", upstream, err)
 			continue
 		}
 		if resp == nil {
 			lastErr = fmt.Errorf("nil response from %s", upstream)
+			lastFailure = telemetry.DNSFailureEmptyResponse
 			continue
 		}
 		if tryNext, reason := p.shouldFailoverAfterResponse(resp); tryNext {
 			lastErr = fmt.Errorf("%s from %s", reason, upstream)
+			lastFailure = telemetry.DNSFailureRcode
 			log.Warnf("[dns] upstream %s: %s; trying next", upstream, reason)
 			continue
 		}
-		return resp, nil
+		return resp, "", nil
 	}
 	if lastErr != nil {
-		return nil, lastErr
+		return nil, lastFailure, lastErr
 	}
-	return nil, fmt.Errorf("no upstream resolvers configured")
+	return nil, telemetry.DNSFailureNoUpstreams, fmt.Errorf("no upstream resolvers configured")
 }
 
 // shouldFailoverAfterResponse: treat NXDOMAIN and NOERROR as final (no retry). Other rcodes may

components/egress/pkg/dnsproxy/proxy_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/alibaba/opensandbox/egress/pkg/constants"
 	"github.com/alibaba/opensandbox/egress/pkg/nftables"
 	"github.com/alibaba/opensandbox/egress/pkg/policy"
+	"github.com/alibaba/opensandbox/egress/pkg/telemetry"
 )
 
 func TestProxyUpdatePolicy(t *testing.T) {
@@ -119,12 +120,89 @@ func TestForwardAddsEDNS0BufferSize(t *testing.T) {
 	query := new(dns.Msg)
 	query.SetQuestion("example.com.", dns.TypeA)
 
-	resp, err := proxy.forward(query)
+	resp, failure, err := proxy.forward(query)
 	require.NoError(t, err)
+	require.Empty(t, failure, "a successful forward must not report a failure reason")
 	require.Len(t, resp.Answer, 1)
 	require.Equal(t, uint16(4096), <-seen)
 }
 
+// A failed lookup has to be classifiable: serveDNS turns the reason into the
+// egress.dns.query.failed_total attribute, which is the only signal an operator gets that
+// resolution is broken rather than merely denied by policy.
+func TestForwardClassifiesFailures(t *testing.T) {
+	t.Run("no upstreams configured", func(t *testing.T) {
+		proxy := &Proxy{upstreamExchangeTimeout: time.Second}
+		query := new(dns.Msg)
+		query.SetQuestion("example.com.", dns.TypeA)
+
+		resp, failure, err := proxy.forward(query)
+
+		require.Error(t, err)
+		require.Nil(t, resp)
+		require.Equal(t, telemetry.DNSFailureNoUpstreams, failure)
+	})
+
+	t.Run("every upstream unreachable", func(t *testing.T) {
+		// Exempt loopback so the dialer skips SO_MARK: without CAP_NET_ADMIN it fails with
+		// EPERM, which would classify as upstream_error for the wrong reason.
+		t.Setenv(constants.EnvNameserverExempt, "127.0.0.1")
+		resetNameserverExemptCache(t)
+
+		// Port 1 on loopback: nothing listens, so the exchange fails rather than timing out.
+		proxy := &Proxy{
+			upstreams:               []string{"127.0.0.1:1"},
+			activeUpstreams:         []string{"127.0.0.1:1"},
+			upstreamExchangeTimeout: 200 * time.Millisecond,
+		}
+		query := new(dns.Msg)
+		query.SetQuestion("example.com.", dns.TypeA)
+
+		resp, failure, err := proxy.forward(query)
+
+		require.Error(t, err)
+		require.Nil(t, resp)
+		require.Equal(t, telemetry.DNSFailureUpstreamError, failure)
+	})
+
+	t.Run("upstream answers with a failover rcode", func(t *testing.T) {
+		t.Setenv(constants.EnvNameserverExempt, "127.0.0.1")
+		resetNameserverExemptCache(t)
+
+		conn, err := net.ListenPacket("udp", "127.0.0.1:0")
+		require.NoError(t, err)
+		t.Cleanup(func() { _ = conn.Close() })
+
+		server := &dns.Server{
+			PacketConn: conn,
+			Handler: dns.HandlerFunc(func(w dns.ResponseWriter, r *dns.Msg) {
+				resp := new(dns.Msg)
+				resp.SetRcode(r, dns.RcodeServerFailure)
+				_ = w.WriteMsg(resp)
+			}),
+		}
+		started := make(chan struct{})
+		server.NotifyStartedFunc = func() { close(started) }
+		go func() { _ = server.ActivateAndServe() }()
+		t.Cleanup(func() { _ = server.Shutdown() })
+		<-started
+
+		proxy := &Proxy{
+			upstreams:               []string{conn.LocalAddr().String()},
+			activeUpstreams:         []string{conn.LocalAddr().String()},
+			upstreamExchangeTimeout: time.Second,
+		}
+		query := new(dns.Msg)
+		query.SetQuestion("example.com.", dns.TypeA)
+
+		resp, failure, err := proxy.forward(query)
+
+		require.Error(t, err, "SERVFAIL from the only upstream must exhaust the chain")
+		require.Nil(t, resp)
+		require.Equal(t, telemetry.DNSFailureRcode, failure)
+	})
+}
+
 func TestSetOnResolved(t *testing.T) {
 	proxy, err := New(policy.DefaultDenyPolicy(), "", nil, nil)
 	require.NoError(t, err)

components/egress/pkg/nftables/manager.go

@@ -113,6 +113,7 @@ func (m *Manager) ApplyStatic(ctx context.Context, p *policy.NetworkPolicy) erro
 				}
 			}
 		}
+		telemetry.RecordNftablesUpdateFailed(telemetry.NftOpStaticApply)
 		return err
 	}
 	m.tracker.clear()
@@ -135,11 +136,16 @@ func (m *Manager) AddResolvedIPs(ctx context.Context, ips []ResolvedIP) error {
 	}
 	log.Debugf("nftables: adding %d resolved IP(s) to dynamic allow sets with script statement %s", len(ips), script)
 	_, err := m.run(ctx, script)
-	if err == nil {
-		m.tracker.setDynamicIPs(ips)
-		telemetry.RecordNftablesUpdate()
+	if err != nil {
+		// The policy allows these destinations but the kernel does not know it yet, so
+		// the chain's final rule drops them. Indistinguishable from a policy denial
+		// inside the sandbox, hence its own counter.
+		telemetry.RecordNftablesUpdateFailed(telemetry.NftOpDynamicAdd)
+		return err
 	}
-	return err
+	m.tracker.setDynamicIPs(ips)
+	telemetry.RecordNftablesUpdate()
+	return nil
 }
 
 // StartConnectionRefresh keeps DNS-learned IPs authorized while a TCP
@@ -172,6 +178,7 @@ func (m *Manager) RemoveEnforcement(ctx context.Context) error {
 		if strings.Contains(msg, "no such file") || strings.Contains(msg, "does not exist") {
 			return nil
 		}
+		telemetry.RecordNftablesUpdateFailed(telemetry.NftOpRemove)
 		return err
 	}
 	m.tracker.clear()

components/egress/pkg/telemetry/metrics.go

@@ -32,13 +32,31 @@ import (
 var (
 	meter metric.Meter
 
-	dnsQueryDur  metric.Float64Histogram
-	policyDenied metric.Int64Counter
-	nftUpdates   metric.Int64Counter
+	dnsQueryDur     metric.Float64Histogram
+	dnsQueryFailed  metric.Int64Counter
+	policyDenied    metric.Int64Counter
+	nftUpdates      metric.Int64Counter
+	nftUpdateFailed metric.Int64Counter
 
 	lastNftRuleCount atomic.Int64
 )
 
+// Bounded reason values for RecordDNSQueryFailed. A closed set keeps the counter's
+// cardinality fixed: error strings and queried names must never reach an attribute.
+const (
+	DNSFailureNoUpstreams   = "no_upstreams"
+	DNSFailureUpstreamError = "upstream_error"
+	DNSFailureEmptyResponse = "empty_response"
+	DNSFailureRcode         = "rcode"
+)
+
+// Bounded operation values for RecordNftablesUpdateFailed.
+const (
+	NftOpStaticApply = "static_apply"
+	NftOpDynamicAdd  = "dynamic_add"
+	NftOpRemove      = "remove"
+)
+
 var egressSharedAttrs = sync.OnceValue(func() []attribute.KeyValue {
 	return inttelemetry.SharedAttrsFromEnv(inttelemetry.SharedAttrsEnvConfig{
 		SandboxIDEnv:  constants.EnvSandboxID,
@@ -51,6 +69,18 @@ var egressMetricOpt = sync.OnceValue(func() metric.MeasurementOption {
 	return metric.WithAttributes(egressSharedAttrs()...)
 })
 
+// egressMetricOptWith adds one attribute to the shared set. It copies rather than
+// appending to the slice returned by egressSharedAttrs: that slice is shared by every
+// caller and may have spare capacity, so append would write into the backing array and
+// let one call's attribute leak into another's.
+func egressMetricOptWith(kv attribute.KeyValue) metric.MeasurementOption {
+	shared := egressSharedAttrs()
+	attrs := make([]attribute.KeyValue, 0, len(shared)+1)
+	attrs = append(attrs, shared...)
+	attrs = append(attrs, kv)
+	return metric.WithAttributes(attrs...)
+}
+
 func EgressLogFields() []slogger.Field {
 	kvs := egressSharedAttrs()
 	out := make([]slogger.Field, 0, len(kvs))
@@ -96,6 +126,14 @@ func registerEgressMetrics() error {
 	if err != nil {
 		return err
 	}
+	dnsQueryFailed, err = meter.Int64Counter(
+		"egress.dns.query.failed_total",
+		metric.WithDescription("DNS queries the proxy could not resolve, by reason. "+
+			"Distinct from egress.policy.denied_total, which counts deliberate policy denials."),
+	)
+	if err != nil {
+		return err
+	}
 	policyDenied, err = meter.Int64Counter(
 		"egress.policy.denied_total",
 		metric.WithDescription("DNS policy denials"),
@@ -110,6 +148,14 @@ func registerEgressMetrics() error {
 	if err != nil {
 		return err
 	}
+	nftUpdateFailed, err = meter.Int64Counter(
+		"egress.nftables.updates.failed_total",
+		metric.WithDescription("nft updates that failed, by operation. A failed dynamic_add "+
+			"means an allowed destination is unreachable while the policy says otherwise."),
+	)
+	if err != nil {
+		return err
+	}
 
 	_, err = meter.Int64ObservableGauge(
 		"egress.nftables.rules.count",
@@ -149,6 +195,13 @@ func registerEgressMetrics() error {
 	return err
 }
 
+// ForceFlush exports pending metrics immediately. Callers that are about to terminate the
+// process must use it: metrics leave through a periodic reader, and the deferred shutdown in
+// main does not run past os.Exit.
+func ForceFlush(ctx context.Context) error {
+	return inttelemetry.ForceFlush(ctx)
+}
+
 func NftRuleCountFromPolicy(p *policy.NetworkPolicy) int64 {
 	if p == nil {
 		p = policy.DefaultDenyPolicy()
@@ -165,6 +218,15 @@ func RecordDNSForward(seconds float64) {
 	dnsQueryDur.Record(context.Background(), seconds, opt)
 }
 
+// RecordDNSQueryFailed counts a lookup the proxy could not answer. reason must be one of
+// the DNSFailure* constants.
+func RecordDNSQueryFailed(reason string) {
+	if dnsQueryFailed == nil {
+		return
+	}
+	dnsQueryFailed.Add(context.Background(), 1, egressMetricOptWith(attribute.String("reason", reason)))
+}
+
 func RecordDNSDenied() {
 	if policyDenied == nil {
 		return
@@ -182,3 +244,12 @@ func RecordNftablesUpdate() {
 	}
 	nftUpdates.Add(context.Background(), 1, egressMetricOpt())
 }
+
+// RecordNftablesUpdateFailed counts an update that did not reach the kernel. operation
+// must be one of the NftOp* constants.
+func RecordNftablesUpdateFailed(operation string) {
+	if nftUpdateFailed == nil {
+		return
+	}
+	nftUpdateFailed.Add(context.Background(), 1, egressMetricOptWith(attribute.String("operation", operation)))
+}