feat(ingress): observe upstream network readiness

components/ingress/main.go

@@ -30,6 +30,7 @@ import (
 
 	"github.com/alibaba/opensandbox/ingress/pkg/flag"
 	"github.com/alibaba/opensandbox/ingress/pkg/proxy"
+	"github.com/alibaba/opensandbox/ingress/pkg/proxy/connectivity"
 	"github.com/alibaba/opensandbox/ingress/pkg/renewintent"
 	"github.com/alibaba/opensandbox/ingress/pkg/routescope"
 	"github.com/alibaba/opensandbox/ingress/pkg/sandbox"
@@ -124,11 +125,32 @@ func main() {
 		})
 	}
 
-	// Create reverse proxy with sandbox provider
-	reverseProxy := proxy.NewProxy(ctx, sandboxProvider, proxy.Mode(flag.Mode), renewPublisher, secure, scopeVerifier)
-	mux := http.NewServeMux()
-	mux.Handle("/", reverseProxy)
-	mux.HandleFunc("/status.ok", proxy.Healthz)
+	connectObserver, networkReadiness, err := newNetworkReadiness(connectivity.TrackerConfig{
+		Window:                   flag.NetworkReadinessShadowWindow,
+		MaxDistinctTargets:       flag.NetworkReadinessShadowMaxTargets,
+		MinAttempts:              flag.NetworkReadinessShadowMinAttempts,
+		MinDistinctTargets:       flag.NetworkReadinessShadowMinTargets,
+		MinDistinctSignalTargets: flag.NetworkReadinessShadowMinSignalTargets,
+		DegradedFailureRatio:     flag.NetworkReadinessShadowDegradedFailureRatio,
+	})
+	proxyOptions := make([]proxy.Option, 0, 1)
+	if err != nil {
+		log.Printf("network readiness shadow assessment disabled (invalid configuration): %v", err)
+	} else {
+		proxyOptions = append(proxyOptions, proxy.WithConnectObserver(connectObserver))
+	}
+
+	// Create reverse proxy with sandbox provider.
+	reverseProxy := proxy.NewProxy(
+		ctx,
+		sandboxProvider,
+		proxy.Mode(flag.Mode),
+		renewPublisher,
+		secure,
+		scopeVerifier,
+		proxyOptions...,
+	)
+	mux := newIngressMux(reverseProxy, networkReadiness)
 
 	if err := http.ListenAndServe(fmt.Sprintf(":%v", flag.Port), mux); err != nil {
 		log.Panicf("Error starting http server: %v", err)
@@ -137,6 +159,43 @@ func main() {
 	panic("unreachable")
 }
 
+func newNetworkReadiness(config connectivity.TrackerConfig) (connectivity.Observer, http.Handler, error) {
+	tracker, err := connectivity.NewTracker(config)
+	if err != nil {
+		telemetry.SetConnectivitySnapshotProvider(nil)
+		return nil, http.NotFoundHandler(), err
+	}
+
+	observer := connectivity.ObserverFunc(func(observation connectivity.Observation) {
+		tracker.Observe(observation)
+		telemetry.RecordUpstreamConnect(
+			string(observation.Result),
+			observation.Protocol,
+			float64(observation.Duration)/float64(time.Millisecond),
+		)
+	})
+	telemetry.SetConnectivitySnapshotProvider(func() telemetry.ConnectivitySnapshot {
+		snapshot := tracker.Snapshot(time.Now())
+		return telemetry.ConnectivitySnapshot{
+			Attempts:              int64(snapshot.Attempts),
+			SignalFailures:        int64(snapshot.SignalFailures),
+			DistinctTargets:       int64(snapshot.DistinctTargets),
+			DistinctSignalTargets: int64(snapshot.DistinctSignalTargets),
+			Qualified:             snapshot.Qualified,
+			Degraded:              snapshot.Degraded,
+		}
+	})
+	return observer, connectivity.NewReadinessHandler(tracker), nil
+}
+
+func newIngressMux(reverseProxy, networkReadiness http.Handler) *http.ServeMux {
+	mux := http.NewServeMux()
+	mux.Handle("/", reverseProxy)
+	mux.Handle("/status.ok/network-readiness", networkReadiness)
+	mux.HandleFunc("/status.ok", proxy.Healthz)
+	return mux
+}
+
 func withLogger(ctx context.Context, logLevel string) context.Context {
 	logger := slogger.MustNew(slogger.Config{Level: logLevel}).Named("opensandbox.ingress")
 	return proxy.WithLogger(ctx, logger)

components/ingress/main_test.go

@@ -0,0 +1,66 @@
+// 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 main
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/alibaba/opensandbox/ingress/pkg/proxy/connectivity"
+)
+
+func TestIngressMuxReservesOnlyExactHealthPaths(t *testing.T) {
+	dataPlane := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte("proxied:" + r.URL.Path))
+	})
+	networkReadiness := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = w.Write([]byte("network-readiness"))
+	})
+	mux := newIngressMux(dataPlane, networkReadiness)
+
+	tests := []struct {
+		path string
+		want string
+	}{
+		{path: "/readyz", want: "proxied:/readyz"},
+		{path: "/livez", want: "proxied:/livez"},
+		{path: "/status.ok/network-readiness", want: "network-readiness"},
+		{path: "/status.ok/network-readiness/", want: "proxied:/status.ok/network-readiness/"},
+		{path: "/status.ok", want: "OK"},
+	}
+	for _, test := range tests {
+		t.Run(test.path, func(t *testing.T) {
+			recorder := httptest.NewRecorder()
+			mux.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, test.path, nil))
+			if recorder.Code != http.StatusOK || recorder.Body.String() != test.want {
+				t.Fatalf("response = %d %q, want 200 %q", recorder.Code, recorder.Body.String(), test.want)
+			}
+		})
+	}
+}
+
+func TestInvalidNetworkReadinessConfigDisablesObserver(t *testing.T) {
+	observer, handler, err := newNetworkReadiness(connectivity.TrackerConfig{})
+	if err == nil || observer != nil {
+		t.Fatalf("newNetworkReadiness() = (%v, _, %v), want nil observer and error", observer, err)
+	}
+
+	recorder := httptest.NewRecorder()
+	handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/status.ok/network-readiness", nil))
+	if recorder.Code != http.StatusNotFound {
+		t.Fatalf("disabled handler status = %d, want 404", recorder.Code)
+	}
+}

components/ingress/pkg/flag/flags.go

@@ -14,6 +14,8 @@
 
 package flag
 
+import "time"
+
 var (
 	// LogLevel controls the router log verbosity.
 	LogLevel string
@@ -38,4 +40,11 @@ var (
 	FastPathEndpoint          string
 	FastPathAccessMode        string
 	FastPathWaitTimeoutMillis int
+
+	NetworkReadinessShadowWindow               time.Duration
+	NetworkReadinessShadowMaxTargets           int
+	NetworkReadinessShadowMinAttempts          uint64
+	NetworkReadinessShadowMinTargets           int
+	NetworkReadinessShadowMinSignalTargets     int
+	NetworkReadinessShadowDegradedFailureRatio float64
 )

components/ingress/pkg/flag/parser.go

@@ -16,6 +16,7 @@ package flag
 
 import (
 	"flag"
+	"time"
 )
 
 var (
@@ -40,5 +41,12 @@ func InitFlags() {
 	flag.StringVar(&FastPathAccessMode, "fastpath-access-mode", "direct-fastlet-proxy", "FastPath fleets data-plane mode: central-proxy or direct-fastlet-proxy")
 	flag.IntVar(&FastPathWaitTimeoutMillis, "fastpath-wait-timeout-millis", 2000, "Bounded FastPath readiness wait for one ingress request")
 
+	flag.DurationVar(&NetworkReadinessShadowWindow, "network-readiness-shadow-window", time.Minute, "Shadow connectivity assessment window")
+	flag.IntVar(&NetworkReadinessShadowMaxTargets, "network-readiness-shadow-max-targets", 1024, "Maximum distinct upstream targets retained per shadow window")
+	flag.Uint64Var(&NetworkReadinessShadowMinAttempts, "network-readiness-shadow-min-attempts", 20, "Minimum connection attempts required for a shadow assessment")
+	flag.IntVar(&NetworkReadinessShadowMinTargets, "network-readiness-shadow-min-targets", 5, "Minimum distinct upstream targets required for a shadow assessment")
+	flag.IntVar(&NetworkReadinessShadowMinSignalTargets, "network-readiness-shadow-min-signal-targets", 2, "Minimum distinct upstream targets with timeout or unreachable results required for a degraded shadow assessment")
+	flag.Float64Var(&NetworkReadinessShadowDegradedFailureRatio, "network-readiness-shadow-failure-ratio", 0.2, "Timeout or unreachable ratio reported as degraded in shadow mode")
+
 	flag.Parse()
 }

components/ingress/pkg/proxy/connectivity/dialer.go

@@ -0,0 +1,60 @@
+// 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 connectivity
+
+import (
+	"context"
+	"net"
+	"strings"
+	"time"
+)
+
+// DialContextFunc matches net.Dialer.DialContext.
+type DialContextFunc func(context.Context, string, string) (net.Conn, error)
+
+// WrapDialContext observes exactly one result for every TCP invocation of next.
+func WrapDialContext(next DialContextFunc, observer Observer, protocol string) DialContextFunc {
+	if next == nil {
+		return nil
+	}
+	return wrapDialContext(next, observer, protocol, time.Now)
+}
+
+func wrapDialContext(next DialContextFunc, observer Observer, protocol string, now func() time.Time) DialContextFunc {
+	return func(ctx context.Context, network, address string) (net.Conn, error) {
+		startedAt := now()
+		conn, err := next(ctx, network, address)
+		finishedAt := now()
+		if observer == nil || !strings.HasPrefix(network, "tcp") {
+			return conn, err
+		}
+
+		result := ClassifyConnectError(err)
+		// A request can be canceled while DialContext is returning a more
+		// specific network error. Preserve that error and use cancellation only
+		// when the dial result itself cannot be classified.
+		if err != nil && result == ResultOther && ctx.Err() == context.Canceled {
+			result = ResultCanceled
+		}
+		observer.Observe(Observation{
+			At:       finishedAt,
+			Protocol: protocol,
+			Target:   address,
+			Result:   result,
+			Duration: finishedAt.Sub(startedAt),
+		})
+		return conn, err
+	}
+}

components/ingress/pkg/proxy/connectivity/dialer_test.go

@@ -0,0 +1,79 @@
+// 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 connectivity
+
+import (
+	"context"
+	"errors"
+	"net"
+	"syscall"
+	"testing"
+	"time"
+)
+
+func TestWrapDialContextObservesTCPAttempt(t *testing.T) {
+	var got Observation
+	times := []time.Time{
+		time.Date(2026, 8, 26, 10, 0, 0, 0, time.UTC),
+		time.Date(2026, 8, 26, 10, 0, 0, 25_000_000, time.UTC),
+	}
+	now := func() time.Time {
+		value := times[0]
+		times = times[1:]
+		return value
+	}
+	next := func(context.Context, string, string) (net.Conn, error) { return nil, syscall.ETIMEDOUT }
+	observer := ObserverFunc(func(observation Observation) { got = observation })
+
+	_, err := wrapDialContext(next, observer, "http", now)(context.Background(), "tcp", "Sandbox.EXAMPLE.:28888")
+	if !errors.Is(err, syscall.ETIMEDOUT) {
+		t.Fatalf("wrapped dial error = %v, want timeout", err)
+	}
+	if got.Result != ResultTimeout || got.Protocol != "http" || got.Target != "Sandbox.EXAMPLE.:28888" || got.Duration != 25*time.Millisecond {
+		t.Fatalf("unexpected observation: %+v", got)
+	}
+}
+
+func TestWrapDialContextPreservesNetworkErrorWhenContextIsCanceled(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	var got Observation
+	observer := ObserverFunc(func(observation Observation) { got = observation })
+	next := func(context.Context, string, string) (net.Conn, error) { return nil, syscall.ENETUNREACH }
+
+	_, _ = WrapDialContext(next, observer, "websocket")(ctx, "tcp", "10.0.0.1:80")
+	if got.Result != ResultUnreachable {
+		t.Fatalf("result = %q, want %q", got.Result, ResultUnreachable)
+	}
+}
+
+func TestWrapDialContextUsesCancellationAsFallback(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	var got Observation
+	observer := ObserverFunc(func(observation Observation) { got = observation })
+	next := func(context.Context, string, string) (net.Conn, error) { return nil, errors.New("dial interrupted") }
+
+	_, _ = WrapDialContext(next, observer, "http")(ctx, "tcp", "10.0.0.1:80")
+	if got.Result != ResultCanceled {
+		t.Fatalf("result = %q, want %q", got.Result, ResultCanceled)
+	}
+}
+
+func TestWrapDialContextAllowsNilBaseDialer(t *testing.T) {
+	if wrapped := WrapDialContext(nil, ObserverFunc(func(Observation) {}), "http"); wrapped != nil {
+		t.Fatalf("WrapDialContext(nil, ...) = %v, want nil", wrapped)
+	}
+}

components/ingress/pkg/proxy/connectivity/handler.go

@@ -0,0 +1,45 @@
+// 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 connectivity
+
+import (
+	"net/http"
+	"time"
+)
+
+// Snapshotter supplies the current fixed-window connectivity assessment.
+type Snapshotter interface {
+	Snapshot(time.Time) Snapshot
+}
+
+// NewReadinessHandler creates a shadow-only readiness endpoint. It reports an
+// observed degradation in the body but intentionally never removes traffic.
+func NewReadinessHandler(tracker Snapshotter) http.Handler {
+	return newReadinessHandler(tracker, time.Now)
+}
+
+func newReadinessHandler(tracker Snapshotter, now func() time.Time) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		snapshot := tracker.Snapshot(now())
+		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+		w.Header().Set("Cache-Control", "no-store")
+		w.WriteHeader(http.StatusOK)
+		if snapshot.Degraded {
+			_, _ = w.Write([]byte("DEGRADED"))
+			return
+		}
+		_, _ = w.Write([]byte("OK"))
+	})
+}