From 35a1c69935d678787a45635295744f8b84e6f73b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B2=90=E8=8F=93?= Date: Tue, 1 Sep 2026 12:13:19 +0800 Subject: [PATCH] feat(ingress): observe upstream network readiness --- components/ingress/main.go | 69 +++++- components/ingress/main_test.go | 66 ++++++ components/ingress/pkg/flag/flags.go | 9 + components/ingress/pkg/flag/parser.go | 8 + .../ingress/pkg/proxy/connectivity/dialer.go | 60 ++++++ .../pkg/proxy/connectivity/dialer_test.go | 79 +++++++ .../ingress/pkg/proxy/connectivity/handler.go | 45 ++++ .../pkg/proxy/connectivity/handler_test.go | 52 +++++ .../pkg/proxy/connectivity/observation.go | 40 ++++ .../ingress/pkg/proxy/connectivity/result.go | 74 +++++++ .../pkg/proxy/connectivity/result_test.go | 50 +++++ .../ingress/pkg/proxy/connectivity/tracker.go | 200 ++++++++++++++++++ .../pkg/proxy/connectivity/tracker_test.go | 179 ++++++++++++++++ .../ingress/pkg/proxy/connectivity_options.go | 68 ++++++ .../pkg/proxy/connectivity_options_test.go | 135 ++++++++++++ .../ingress/pkg/proxy/fleets_proxy_test.go | 17 +- components/ingress/pkg/proxy/http.go | 4 + components/ingress/pkg/proxy/proxy.go | 15 +- components/ingress/pkg/telemetry/metrics.go | 145 +++++++++++++ .../ingress/pkg/telemetry/metrics_test.go | 139 ++++++++++++ docs/components/ingress.md | 52 ++++- 21 files changed, 1497 insertions(+), 9 deletions(-) create mode 100644 components/ingress/main_test.go create mode 100644 components/ingress/pkg/proxy/connectivity/dialer.go create mode 100644 components/ingress/pkg/proxy/connectivity/dialer_test.go create mode 100644 components/ingress/pkg/proxy/connectivity/handler.go create mode 100644 components/ingress/pkg/proxy/connectivity/handler_test.go create mode 100644 components/ingress/pkg/proxy/connectivity/observation.go create mode 100644 components/ingress/pkg/proxy/connectivity/result.go create mode 100644 components/ingress/pkg/proxy/connectivity/result_test.go create mode 100644 components/ingress/pkg/proxy/connectivity/tracker.go create mode 100644 components/ingress/pkg/proxy/connectivity/tracker_test.go create mode 100644 components/ingress/pkg/proxy/connectivity_options.go create mode 100644 components/ingress/pkg/proxy/connectivity_options_test.go create mode 100644 components/ingress/pkg/telemetry/metrics_test.go diff --git a/components/ingress/main.go b/components/ingress/main.go index 3bb0dd8c1..62feb57aa 100644 --- a/components/ingress/main.go +++ b/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) diff --git a/components/ingress/main_test.go b/components/ingress/main_test.go new file mode 100644 index 000000000..322d74993 --- /dev/null +++ b/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) + } +} diff --git a/components/ingress/pkg/flag/flags.go b/components/ingress/pkg/flag/flags.go index f92cc7bde..f3caebe4c 100644 --- a/components/ingress/pkg/flag/flags.go +++ b/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 ) diff --git a/components/ingress/pkg/flag/parser.go b/components/ingress/pkg/flag/parser.go index 7bb3ff0cc..1f3eaa5a8 100644 --- a/components/ingress/pkg/flag/parser.go +++ b/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() } diff --git a/components/ingress/pkg/proxy/connectivity/dialer.go b/components/ingress/pkg/proxy/connectivity/dialer.go new file mode 100644 index 000000000..73af3cf4a --- /dev/null +++ b/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 + } +} diff --git a/components/ingress/pkg/proxy/connectivity/dialer_test.go b/components/ingress/pkg/proxy/connectivity/dialer_test.go new file mode 100644 index 000000000..2708e5d44 --- /dev/null +++ b/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) + } +} diff --git a/components/ingress/pkg/proxy/connectivity/handler.go b/components/ingress/pkg/proxy/connectivity/handler.go new file mode 100644 index 000000000..28a8dd0de --- /dev/null +++ b/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")) + }) +} diff --git a/components/ingress/pkg/proxy/connectivity/handler_test.go b/components/ingress/pkg/proxy/connectivity/handler_test.go new file mode 100644 index 000000000..54e0c911f --- /dev/null +++ b/components/ingress/pkg/proxy/connectivity/handler_test.go @@ -0,0 +1,52 @@ +// 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" + "net/http/httptest" + "testing" + "time" +) + +type staticSnapshotter struct { + snapshot Snapshot +} + +func (s staticSnapshotter) Snapshot(time.Time) Snapshot { return s.snapshot } + +func TestReadinessHandlerIsShadowOnly(t *testing.T) { + tests := []struct { + name string + degraded bool + body string + }{ + {name: "healthy", body: "OK"}, + {name: "degraded", degraded: true, body: "DEGRADED"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + recorder := httptest.NewRecorder() + handler := NewReadinessHandler(staticSnapshotter{snapshot: Snapshot{Degraded: test.degraded}}) + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/status.ok/network-readiness", nil)) + if recorder.Code != http.StatusOK || recorder.Body.String() != test.body { + t.Fatalf("response = %d %q, want 200 %q", recorder.Code, recorder.Body.String(), test.body) + } + if recorder.Header().Get("Content-Type") != "text/plain; charset=utf-8" || recorder.Header().Get("Cache-Control") != "no-store" { + t.Fatalf("unexpected response headers: %v", recorder.Header()) + } + }) + } +} diff --git a/components/ingress/pkg/proxy/connectivity/observation.go b/components/ingress/pkg/proxy/connectivity/observation.go new file mode 100644 index 000000000..19870c4dd --- /dev/null +++ b/components/ingress/pkg/proxy/connectivity/observation.go @@ -0,0 +1,40 @@ +// 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 "time" + +// Observation describes one TCP connection attempt as seen by the ingress. +type Observation struct { + At time.Time + Protocol string + // Target is the TCP dial address. Trackers normalize it to a network host. + Target string + Result Result + Duration time.Duration +} + +// Observer consumes TCP connection observations. +type Observer interface { + Observe(Observation) +} + +// ObserverFunc adapts a function to Observer. +type ObserverFunc func(Observation) + +// Observe implements Observer. +func (f ObserverFunc) Observe(observation Observation) { + f(observation) +} diff --git a/components/ingress/pkg/proxy/connectivity/result.go b/components/ingress/pkg/proxy/connectivity/result.go new file mode 100644 index 000000000..b62d77aef --- /dev/null +++ b/components/ingress/pkg/proxy/connectivity/result.go @@ -0,0 +1,74 @@ +// 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" +) + +// Result classifies the outcome of one TCP connection attempt. It deliberately +// excludes HTTP, TLS, and application protocol outcomes. +type Result string + +const ( + ResultSuccess Result = "success" + ResultTimeout Result = "timeout" + ResultUnreachable Result = "unreachable" + ResultRefused Result = "refused" + ResultDNS Result = "dns_error" + ResultCanceled Result = "canceled" + ResultOther Result = "other" +) + +// ClassifyConnectError classifies an error returned by a TCP DialContext call. +func ClassifyConnectError(err error) Result { + if err == nil { + return ResultSuccess + } + + switch { + case errors.Is(err, syscall.ETIMEDOUT): + return ResultTimeout + case errors.Is(err, syscall.EHOSTUNREACH), errors.Is(err, syscall.ENETUNREACH), errors.Is(err, syscall.ENETDOWN), errors.Is(err, syscall.EHOSTDOWN): + return ResultUnreachable + case errors.Is(err, syscall.ECONNREFUSED): + return ResultRefused + } + + var dnsErr *net.DNSError + if errors.As(err, &dnsErr) { + return ResultDNS + } + if errors.Is(err, context.DeadlineExceeded) { + return ResultTimeout + } + if errors.Is(err, context.Canceled) { + return ResultCanceled + } + + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return ResultTimeout + } + + return ResultOther +} + +func isDegradationSignal(result Result) bool { + return result == ResultTimeout || result == ResultUnreachable +} diff --git a/components/ingress/pkg/proxy/connectivity/result_test.go b/components/ingress/pkg/proxy/connectivity/result_test.go new file mode 100644 index 000000000..d7b80da96 --- /dev/null +++ b/components/ingress/pkg/proxy/connectivity/result_test.go @@ -0,0 +1,50 @@ +// 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" +) + +func TestClassifyConnectError(t *testing.T) { + tests := []struct { + name string + err error + want Result + }{ + {name: "success", want: ResultSuccess}, + {name: "timeout", err: syscall.ETIMEDOUT, want: ResultTimeout}, + {name: "host unreachable", err: syscall.EHOSTUNREACH, want: ResultUnreachable}, + {name: "network unreachable", err: syscall.ENETUNREACH, want: ResultUnreachable}, + {name: "network down", err: syscall.ENETDOWN, want: ResultUnreachable}, + {name: "host down", err: syscall.EHOSTDOWN, want: ResultUnreachable}, + {name: "refused", err: syscall.ECONNREFUSED, want: ResultRefused}, + {name: "DNS", err: &net.DNSError{Err: "no such host"}, want: ResultDNS}, + {name: "deadline", err: context.DeadlineExceeded, want: ResultTimeout}, + {name: "canceled", err: context.Canceled, want: ResultCanceled}, + {name: "other", err: errors.New("broken connection"), want: ResultOther}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := ClassifyConnectError(test.err); got != test.want { + t.Fatalf("ClassifyConnectError(%v) = %q, want %q", test.err, got, test.want) + } + }) + } +} diff --git a/components/ingress/pkg/proxy/connectivity/tracker.go b/components/ingress/pkg/proxy/connectivity/tracker.go new file mode 100644 index 000000000..28f3f0111 --- /dev/null +++ b/components/ingress/pkg/proxy/connectivity/tracker.go @@ -0,0 +1,200 @@ +// 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 ( + "errors" + "math" + "net" + "strings" + "sync" + "time" +) + +// TrackerConfig controls a fixed, wall-clock-aligned observation window. +// Thresholds only produce a shadow assessment. +type TrackerConfig struct { + Window time.Duration + MaxDistinctTargets int + MinAttempts uint64 + MinDistinctTargets int + MinDistinctSignalTargets int + DegradedFailureRatio float64 +} + +// Snapshot is a bounded aggregate for the most recent fully elapsed window. +type Snapshot struct { + WindowStart time.Time + Attempts uint64 + SignalFailures uint64 + DistinctTargets int + DistinctSignalTargets int + Qualified bool + Degraded bool +} + +// Tracker keeps bounded, concurrency-safe TCP connection aggregates. +type Tracker struct { + mu sync.RWMutex + + initialized bool + config TrackerConfig + current bucket + completed bucket +} + +type bucket struct { + start time.Time + attempts uint64 + signalFailures uint64 + targets map[string]struct{} + signalTargets map[string]struct{} +} + +// NewTracker validates config and constructs an empty tracker. +func NewTracker(config TrackerConfig) (*Tracker, error) { + switch { + case config.Window <= 0: + return nil, errors.New("connectivity tracker window must be positive") + case config.MaxDistinctTargets <= 0: + return nil, errors.New("connectivity tracker max distinct targets must be positive") + case config.MinAttempts == 0: + return nil, errors.New("connectivity tracker minimum attempts must be positive") + case config.MinDistinctTargets <= 0: + return nil, errors.New("connectivity tracker minimum distinct targets must be positive") + case config.MinDistinctTargets > config.MaxDistinctTargets: + return nil, errors.New("connectivity tracker minimum distinct targets exceeds cap") + case config.MinDistinctSignalTargets <= 0: + return nil, errors.New("connectivity tracker minimum distinct signal targets must be positive") + case config.MinDistinctSignalTargets > config.MaxDistinctTargets: + return nil, errors.New("connectivity tracker minimum distinct signal targets exceeds cap") + case math.IsNaN(config.DegradedFailureRatio) || config.DegradedFailureRatio <= 0 || config.DegradedFailureRatio > 1: + return nil, errors.New("connectivity tracker degraded failure ratio must be in (0,1]") + } + + return &Tracker{config: config}, nil +} + +// Observe records an observation in its fixed, wall-clock-aligned window. +func (t *Tracker) Observe(observation Observation) { + // Caller cancellation does not establish whether the network path was + // healthy or degraded, so it must not dilute the assessed failure ratio. + if observation.Result == ResultCanceled { + return + } + + windowStart := observation.At.Truncate(t.config.Window) + target := normalizeTarget(observation.Target) + + t.mu.Lock() + defer t.mu.Unlock() + + if !t.initialized { + t.initialized = true + t.current = newBucket(windowStart) + t.completed = newBucket(windowStart.Add(-t.config.Window)) + } else if windowStart.After(t.current.start) { + completedStart := windowStart.Add(-t.config.Window) + if t.current.start.Equal(completedStart) { + t.completed = t.current + } else { + t.completed = newBucket(completedStart) + } + t.current = newBucket(windowStart) + } + if windowStart.Before(t.current.start) { + if windowStart.Equal(t.completed.start) { + t.addObservation(&t.completed, target, observation.Result) + } + return + } + + t.addObservation(&t.current, target, observation.Result) +} + +// Snapshot returns the most recent fully elapsed fixed window without changing +// tracker state. +func (t *Tracker) Snapshot(now time.Time) Snapshot { + completedStart := now.Truncate(t.config.Window).Add(-t.config.Window) + + t.mu.RLock() + defer t.mu.RUnlock() + + if t.current.start.Equal(completedStart) { + return t.snapshot(t.current) + } + if t.completed.start.Equal(completedStart) { + return t.snapshot(t.completed) + } + return Snapshot{WindowStart: completedStart} +} + +func (t *Tracker) snapshot(source bucket) Snapshot { + snapshot := Snapshot{ + WindowStart: source.start, + Attempts: source.attempts, + SignalFailures: source.signalFailures, + DistinctTargets: len(source.targets), + DistinctSignalTargets: len(source.signalTargets), + } + snapshot.Qualified = snapshot.Attempts >= t.config.MinAttempts && + snapshot.DistinctTargets >= t.config.MinDistinctTargets + if snapshot.Qualified { + snapshot.Degraded = snapshot.DistinctSignalTargets >= t.config.MinDistinctSignalTargets && + float64(snapshot.SignalFailures)/float64(snapshot.Attempts) >= t.config.DegradedFailureRatio + } + return snapshot +} + +func (t *Tracker) addObservation(destination *bucket, target string, result Result) { + destination.attempts++ + if isDegradationSignal(result) { + destination.signalFailures++ + addTarget(destination.signalTargets, target, t.config.MaxDistinctTargets) + } + addTarget(destination.targets, target, t.config.MaxDistinctTargets) +} + +func newBucket(start time.Time) bucket { + return bucket{ + start: start, + targets: make(map[string]struct{}), + signalTargets: make(map[string]struct{}), + } +} + +func addTarget(targets map[string]struct{}, target string, limit int) { + if target != "" && len(targets) < limit { + targets[target] = struct{}{} + } +} + +func normalizeTarget(address string) string { + host, _, err := net.SplitHostPort(address) + if err == nil { + // Distinct targets represent network hosts. Multiple ports on one + // Sandbox Pod must not satisfy the cross-target qualification guard. + return canonicalHost(host) + } + return canonicalHost(strings.Trim(address, "[]")) +} + +func canonicalHost(host string) string { + host = strings.ToLower(strings.TrimSuffix(host, ".")) + if ip := net.ParseIP(host); ip != nil { + return ip.String() + } + return host +} diff --git a/components/ingress/pkg/proxy/connectivity/tracker_test.go b/components/ingress/pkg/proxy/connectivity/tracker_test.go new file mode 100644 index 000000000..e65b9f469 --- /dev/null +++ b/components/ingress/pkg/proxy/connectivity/tracker_test.go @@ -0,0 +1,179 @@ +// 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 ( + "fmt" + "math" + "sync" + "testing" + "time" +) + +func validTrackerConfig() TrackerConfig { + return TrackerConfig{ + Window: time.Minute, + MaxDistinctTargets: 8, + MinAttempts: 4, + MinDistinctTargets: 2, + MinDistinctSignalTargets: 2, + DegradedFailureRatio: 0.5, + } +} + +func TestNewTrackerRejectsInvalidConfig(t *testing.T) { + tests := []func(*TrackerConfig){ + func(config *TrackerConfig) { config.Window = 0 }, + func(config *TrackerConfig) { config.MaxDistinctTargets = 0 }, + func(config *TrackerConfig) { config.MinAttempts = 0 }, + func(config *TrackerConfig) { config.MinDistinctTargets = 9 }, + func(config *TrackerConfig) { config.MinDistinctSignalTargets = 9 }, + func(config *TrackerConfig) { config.DegradedFailureRatio = math.NaN() }, + } + for i, mutate := range tests { + config := validTrackerConfig() + mutate(&config) + if _, err := NewTracker(config); err == nil { + t.Fatalf("case %d: NewTracker() returned nil error", i) + } + } +} + +func TestTrackerUsesMostRecentCompletedWindow(t *testing.T) { + tracker, err := NewTracker(validTrackerConfig()) + if err != nil { + t.Fatal(err) + } + start := time.Date(2026, 8, 26, 10, 0, 0, 0, time.UTC) + observations := []Observation{ + {At: start, Target: "10.0.0.1:80", Result: ResultTimeout}, + {At: start, Target: "10.0.0.2:80", Result: ResultUnreachable}, + {At: start, Target: "10.0.0.1:443", Result: ResultSuccess}, + {At: start, Target: "10.0.0.2:443", Result: ResultRefused}, + } + for _, observation := range observations { + tracker.Observe(observation) + } + + if partial := tracker.Snapshot(start.Add(30 * time.Second)); partial.Attempts != 0 { + t.Fatalf("partial window was exposed: %+v", partial) + } + completed := tracker.Snapshot(start.Add(time.Minute)) + if !completed.Qualified || !completed.Degraded || completed.Attempts != 4 || completed.SignalFailures != 2 || completed.DistinctTargets != 2 || completed.DistinctSignalTargets != 2 { + t.Fatalf("unexpected completed snapshot: %+v", completed) + } +} + +func TestTrackerExcludesCanceledAttemptsFromAssessment(t *testing.T) { + tracker, err := NewTracker(validTrackerConfig()) + if err != nil { + t.Fatal(err) + } + start := time.Date(2026, 8, 26, 10, 0, 0, 0, time.UTC) + tracker.Observe(Observation{At: start, Target: "10.0.0.1:80", Result: ResultCanceled}) + + snapshot := tracker.Snapshot(start.Add(time.Minute)) + if snapshot.Attempts != 0 || snapshot.DistinctTargets != 0 { + t.Fatalf("canceled attempt changed assessment: %+v", snapshot) + } +} + +func TestTrackerAcceptsLateObservationForCompletedWindow(t *testing.T) { + tracker, err := NewTracker(validTrackerConfig()) + if err != nil { + t.Fatal(err) + } + start := time.Date(2026, 8, 26, 10, 0, 0, 0, time.UTC) + tracker.Observe(Observation{At: start.Add(time.Minute), Target: "10.0.0.2:80", Result: ResultSuccess}) + tracker.Observe(Observation{At: start.Add(time.Minute - time.Nanosecond), Target: "10.0.0.1:80", Result: ResultTimeout}) + + snapshot := tracker.Snapshot(start.Add(time.Minute)) + if snapshot.Attempts != 1 || snapshot.SignalFailures != 1 || snapshot.DistinctTargets != 1 { + t.Fatalf("late observation was not retained: %+v", snapshot) + } +} + +func TestTrackerAcceptsZeroObservationTime(t *testing.T) { + tracker, err := NewTracker(validTrackerConfig()) + if err != nil { + t.Fatal(err) + } + tracker.Observe(Observation{Target: "10.0.0.1:80", Result: ResultSuccess}) + tracker.Observe(Observation{Target: "10.0.0.2:80", Result: ResultSuccess}) + + if tracker.current.attempts != 2 { + t.Fatalf("attempts = %d, want 2", tracker.current.attempts) + } +} + +func TestTrackerDoesNotQualifyOneTarget(t *testing.T) { + tracker, err := NewTracker(validTrackerConfig()) + if err != nil { + t.Fatal(err) + } + start := time.Date(2026, 8, 26, 10, 0, 0, 0, time.UTC) + for range 4 { + tracker.Observe(Observation{At: start, Target: "10.0.0.1:80", Result: ResultTimeout}) + } + + snapshot := tracker.Snapshot(start.Add(time.Minute)) + if snapshot.Qualified || snapshot.Degraded || snapshot.DistinctTargets != 1 { + t.Fatalf("unexpected single-target snapshot: %+v", snapshot) + } +} + +func TestTrackerBoundsTargetsAndSupportsConcurrentObservation(t *testing.T) { + config := validTrackerConfig() + config.MaxDistinctTargets = 2 + config.MinDistinctTargets = 2 + config.MinDistinctSignalTargets = 2 + tracker, err := NewTracker(config) + if err != nil { + t.Fatal(err) + } + start := time.Date(2026, 8, 26, 10, 0, 0, 0, time.UTC) + + const workers = 8 + var waitGroup sync.WaitGroup + waitGroup.Add(workers) + for worker := 0; worker < workers; worker++ { + go func(target int) { + defer waitGroup.Done() + tracker.Observe(Observation{At: start, Target: fmt.Sprintf("10.0.0.%d", target+1), Result: ResultTimeout}) + }(worker) + } + waitGroup.Wait() + + snapshot := tracker.Snapshot(start.Add(time.Minute)) + if snapshot.Attempts != workers || snapshot.SignalFailures != workers || snapshot.DistinctTargets != 2 || snapshot.DistinctSignalTargets != 2 { + t.Fatalf("unexpected bounded snapshot: %+v", snapshot) + } +} + +func TestTrackerCanonicalizesEquivalentIPv6Targets(t *testing.T) { + tracker, err := NewTracker(validTrackerConfig()) + if err != nil { + t.Fatal(err) + } + start := time.Date(2026, 8, 26, 10, 0, 0, 0, time.UTC) + + tracker.Observe(Observation{At: start, Target: "[::1]:80", Result: ResultSuccess}) + tracker.Observe(Observation{At: start, Target: "[0:0:0:0:0:0:0:1]:443", Result: ResultTimeout}) + + snapshot := tracker.Snapshot(start.Add(time.Minute)) + if snapshot.DistinctTargets != 1 || snapshot.DistinctSignalTargets != 1 { + t.Fatalf("equivalent IPv6 targets were counted separately: %+v", snapshot) + } +} diff --git a/components/ingress/pkg/proxy/connectivity_options.go b/components/ingress/pkg/proxy/connectivity_options.go new file mode 100644 index 000000000..b1a72f3f6 --- /dev/null +++ b/components/ingress/pkg/proxy/connectivity_options.go @@ -0,0 +1,68 @@ +// 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 proxy + +import ( + "net" + "net/http" + + "github.com/gorilla/websocket" + + "github.com/alibaba/opensandbox/ingress/pkg/proxy/connectivity" +) + +// Option configures optional proxy behavior without changing existing callers. +type Option func(*proxyOptions) + +type proxyOptions struct { + connectObserver connectivity.Observer +} + +// WithConnectObserver observes HTTP and WebSocket TCP connection attempts. +func WithConnectObserver(observer connectivity.Observer) Option { + return func(options *proxyOptions) { + options.connectObserver = observer + } +} + +func newObservedHTTPTransport(observer connectivity.Observer) http.RoundTripper { + if observer == nil { + return nil + } + + baseTransport, ok := http.DefaultTransport.(*http.Transport) + if !ok { + return nil + } + transport := baseTransport.Clone() + baseDialContext := transport.DialContext + if baseDialContext == nil { + baseDialer := &net.Dialer{} + baseDialContext = baseDialer.DialContext + } + transport.DialContext = connectivity.WrapDialContext(baseDialContext, observer, "http") + return transport +} + +func newObservedWebSocketDialer(observer connectivity.Observer) *websocket.Dialer { + if observer == nil { + return nil + } + + dialer := *websocket.DefaultDialer + baseDialer := &net.Dialer{} + dialer.NetDialContext = connectivity.WrapDialContext(baseDialer.DialContext, observer, "websocket") + return &dialer +} diff --git a/components/ingress/pkg/proxy/connectivity_options_test.go b/components/ingress/pkg/proxy/connectivity_options_test.go new file mode 100644 index 000000000..f48d241ef --- /dev/null +++ b/components/ingress/pkg/proxy/connectivity_options_test.go @@ -0,0 +1,135 @@ +// 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 proxy + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/alibaba/opensandbox/ingress/pkg/proxy/connectivity" +) + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { + return f(request) +} + +func TestObservedHTTPTransportRecordsTCPConnect(t *testing.T) { + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNoContent) + })) + t.Cleanup(backend.Close) + + observations := make(chan connectivity.Observation, 2) + transport := newObservedHTTPTransport(connectivity.ObserverFunc(func(observation connectivity.Observation) { + observations <- observation + })) + t.Cleanup(transport.(*http.Transport).CloseIdleConnections) + client := &http.Client{Transport: transport} + for range 2 { + response, err := client.Get(backend.URL) + if err != nil { + t.Fatal(err) + } + _ = response.Body.Close() + } + + assertSuccessfulObservation(t, observations, "http") + if len(observations) != 0 { + t.Fatalf("keep-alive request opened %d additional TCP connections", len(observations)) + } +} + +func TestObservedWebSocketDialerRecordsTCPConnect(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + + accepted := make(chan struct{}) + go func() { + conn, acceptErr := listener.Accept() + if acceptErr == nil { + _ = conn.Close() + } + close(accepted) + }() + + observations := make(chan connectivity.Observation, 1) + dialer := newObservedWebSocketDialer(connectivity.ObserverFunc(func(observation connectivity.Observation) { + observations <- observation + })) + conn, err := dialer.NetDialContext(context.Background(), "tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + _ = conn.Close() + <-accepted + + assertSuccessfulObservation(t, observations, "websocket") +} + +func TestProxyConfiguresBothObservedDialers(t *testing.T) { + observer := connectivity.ObserverFunc(func(connectivity.Observation) {}) + proxy := NewProxy(context.Background(), nil, ModeHeader, nil, nil, nil, WithConnectObserver(observer)) + + if proxy.httpTransport == nil || proxy.websocketDialer == nil { + t.Fatalf("observed transports were not configured: %+v", proxy) + } +} + +func TestObservedHTTPTransportHandlesReplacedDefault(t *testing.T) { + previousTransport := http.DefaultTransport + http.DefaultTransport = roundTripFunc(func(*http.Request) (*http.Response, error) { return nil, nil }) + t.Cleanup(func() { http.DefaultTransport = previousTransport }) + + observer := connectivity.ObserverFunc(func(connectivity.Observation) {}) + if transport := newObservedHTTPTransport(observer); transport != nil { + t.Fatalf("transport = %T, want nil fallback for non-standard default transport", transport) + } +} + +func TestObservedHTTPTransportHandlesNilDefaultDialer(t *testing.T) { + previousTransport := http.DefaultTransport + http.DefaultTransport = &http.Transport{} + t.Cleanup(func() { http.DefaultTransport = previousTransport }) + + observer := connectivity.ObserverFunc(func(connectivity.Observation) {}) + transport, ok := newObservedHTTPTransport(observer).(*http.Transport) + if !ok { + t.Fatalf("observed transport = %T, want *http.Transport", transport) + } + if transport.DialContext == nil { + t.Fatal("observed transport has nil DialContext") + } +} + +func assertSuccessfulObservation(t *testing.T, observations <-chan connectivity.Observation, protocol string) { + t.Helper() + select { + case observation := <-observations: + if observation.Protocol != protocol || observation.Result != connectivity.ResultSuccess { + t.Fatalf("unexpected observation: %+v", observation) + } + case <-time.After(time.Second): + t.Fatalf("timed out waiting for %s observation", protocol) + } +} diff --git a/components/ingress/pkg/proxy/fleets_proxy_test.go b/components/ingress/pkg/proxy/fleets_proxy_test.go index 35274e914..c721f2979 100644 --- a/components/ingress/pkg/proxy/fleets_proxy_test.go +++ b/components/ingress/pkg/proxy/fleets_proxy_test.go @@ -24,6 +24,7 @@ import ( "strings" "testing" + "github.com/alibaba/opensandbox/ingress/pkg/proxy/connectivity" "github.com/alibaba/opensandbox/ingress/pkg/routescope" "github.com/alibaba/opensandbox/ingress/pkg/sandbox" slogger "github.com/alibaba/opensandbox/internal/logger" @@ -343,13 +344,27 @@ func TestFleetsProxyInvalidatesRouteOnUpstreamConnectionFailure(t *testing.T) { require.NoError(t, listener.Close()) provider := &fleetsProxyProvider{info: &sandbox.EndpointInfo{UpstreamURL: upstream}} - p := NewProxy(context.Background(), provider, ModeHeader, nil, nil, &routescope.Verifier{Keys: map[string][]byte{"k": []byte("shared-secret")}}) + observations := make(chan connectivity.Observation, 1) + p := NewProxy( + context.Background(), + provider, + ModeHeader, + nil, + nil, + &routescope.Verifier{Keys: map[string][]byte{"k": []byte("shared-secret")}}, + WithConnectObserver(connectivity.ObserverFunc(func(observation connectivity.Observation) { + observations <- observation + })), + ) request := httptest.NewRequest(http.MethodPost, "http://ingress/mutate", strings.NewReader("body")) request.Header.Set(SandboxIngress, fleetsScopeVector) response := httptest.NewRecorder() p.ServeHTTP(response, request) require.Equal(t, http.StatusBadGateway, response.Code) require.Equal(t, provider.target, provider.invalidated) + observation := <-observations + require.Equal(t, connectivity.ResultRefused, observation.Result) + require.Equal(t, "http", observation.Protocol) } func TestFleetsProxyWebSocketUsesBasePathAndUpstreamCredential(t *testing.T) { diff --git a/components/ingress/pkg/proxy/http.go b/components/ingress/pkg/proxy/http.go index 4bbeba815..df2f3f01e 100644 --- a/components/ingress/pkg/proxy/http.go +++ b/components/ingress/pkg/proxy/http.go @@ -26,6 +26,7 @@ import ( type HTTPProxy struct { responseObservers []func(*http.Response) errorObserver func(error) + transport http.RoundTripper } func NewHTTPProxy(observers ...func(*http.Response)) *HTTPProxy { @@ -42,6 +43,9 @@ func (hp *HTTPProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (hp *HTTPProxy) newReverseProxy(targetURL *url.URL) *httputil.ReverseProxy { proxy := httputil.NewSingleHostReverseProxy(targetURL) + if hp.transport != nil { + proxy.Transport = hp.transport + } proxy.Director = func(req *http.Request) { req.URL.Scheme = targetURL.Scheme req.URL.Host = targetURL.Host diff --git a/components/ingress/pkg/proxy/proxy.go b/components/ingress/pkg/proxy/proxy.go index 6f911507d..f994d0674 100644 --- a/components/ingress/pkg/proxy/proxy.go +++ b/components/ingress/pkg/proxy/proxy.go @@ -26,6 +26,7 @@ import ( "time" slogger "github.com/alibaba/opensandbox/internal/logger" + "github.com/gorilla/websocket" "github.com/alibaba/opensandbox/ingress/pkg/renewintent" "github.com/alibaba/opensandbox/ingress/pkg/routescope" @@ -43,15 +44,25 @@ type Proxy struct { secure *signature.Verifier scope *routescope.Verifier + + httpTransport http.RoundTripper + websocketDialer *websocket.Dialer } -func NewProxy(_ context.Context, sandboxProvider sandbox.Provider, mode Mode, renewIntentPublisher renewintent.Publisher, secure *signature.Verifier, scope *routescope.Verifier) *Proxy { +func NewProxy(_ context.Context, sandboxProvider sandbox.Provider, mode Mode, renewIntentPublisher renewintent.Publisher, secure *signature.Verifier, scope *routescope.Verifier, opts ...Option) *Proxy { + options := proxyOptions{} + for _, opt := range opts { + opt(&options) + } + return &Proxy{ sandboxProvider: sandboxProvider, mode: mode, renewIntentPublisher: renewIntentPublisher, secure: secure, scope: scope, + httpTransport: newObservedHTTPTransport(options.connectObserver), + websocketDialer: newObservedWebSocketDialer(options.connectObserver), } } @@ -172,6 +183,7 @@ func (p *Proxy) serve(w http.ResponseWriter, r *http.Request, target sandbox.End } websocketProxy := NewWebSocketProxy(r.URL, p.upstreamResponseObserver(target)) //nolint:bodyclose // Failed handshake bodies are closed by copyResponse. websocketProxy.errorObserver = p.upstreamErrorObserver(target) + websocketProxy.dialer = p.websocketDialer websocketProxy.ServeHTTP(w, r) } else { if r.URL.Scheme == "" { @@ -183,6 +195,7 @@ func (p *Proxy) serve(w http.ResponseWriter, r *http.Request, target sandbox.End } httpProxy := NewHTTPProxy(p.upstreamResponseObserver(target)) //nolint:bodyclose // httputil.ReverseProxy owns response bodies. httpProxy.errorObserver = p.upstreamErrorObserver(target) + httpProxy.transport = p.httpTransport httpProxy.ServeHTTP(w, r) } } diff --git a/components/ingress/pkg/telemetry/metrics.go b/components/ingress/pkg/telemetry/metrics.go index d5c0bd459..7b79b81bd 100644 --- a/components/ingress/pkg/telemetry/metrics.go +++ b/components/ingress/pkg/telemetry/metrics.go @@ -16,6 +16,7 @@ package telemetry import ( "context" + "sync" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" @@ -30,8 +31,24 @@ var ( routingResolutions metric.Int64Counter routingResolutionDuration metric.Float64Histogram + + upstreamConnectCount metric.Int64Counter + upstreamConnectDuration metric.Float64Histogram + + connectivityProviderMu sync.RWMutex + connectivityProvider func() ConnectivitySnapshot ) +// ConnectivitySnapshot contains bounded, low-cardinality shadow aggregates. +type ConnectivitySnapshot struct { + Attempts int64 + SignalFailures int64 + DistinctTargets int64 + DistinctSignalTargets int64 + Qualified bool + Degraded bool +} + func registerIngressMetrics() error { meter = otel.Meter("opensandbox/ingress") @@ -70,6 +87,23 @@ func registerIngressMetrics() error { return err } + upstreamConnectCount, err = meter.Int64Counter( + "ingress.upstream.connect.count", + metric.WithDescription("Ingress upstream TCP connection attempts by result"), + ) + if err != nil { + return err + } + + upstreamConnectDuration, err = meter.Float64Histogram( + "ingress.upstream.connect.duration", + metric.WithDescription("Ingress upstream TCP connection duration"), + metric.WithUnit("ms"), + ) + if err != nil { + return err + } + _, err = meter.Float64ObservableGauge( "ingress.system.cpu.usage", metric.WithDescription("System CPU utilization ratio 0-1"), @@ -104,9 +138,105 @@ func registerIngressMetrics() error { return nil }), ) + if err != nil { + return err + } + + return registerConnectivityMetrics() +} + +func registerConnectivityMetrics() error { + attempts, err := meter.Int64ObservableGauge( + "ingress.network.shadow.attempts", + metric.WithDescription("TCP connection attempts in the most recent complete shadow window"), + ) + if err != nil { + return err + } + signalFailures, err := meter.Int64ObservableGauge( + "ingress.network.shadow.signal_failures", + metric.WithDescription("Timeout and unreachable results in the most recent complete shadow window"), + ) + if err != nil { + return err + } + distinctTargets, err := meter.Int64ObservableGauge( + "ingress.network.shadow.distinct_targets", + metric.WithDescription("Bounded distinct upstream targets in the most recent complete shadow window"), + ) + if err != nil { + return err + } + distinctSignalTargets, err := meter.Int64ObservableGauge( + "ingress.network.shadow.distinct_signal_targets", + metric.WithDescription("Bounded distinct upstream targets with timeout or unreachable results in the most recent complete shadow window"), + ) + if err != nil { + return err + } + qualified, err := meter.Int64ObservableGauge( + "ingress.network.shadow.qualified", + metric.WithDescription("Whether the most recent complete shadow window has enough samples"), + ) + if err != nil { + return err + } + degraded, err := meter.Int64ObservableGauge( + "ingress.network.shadow.degraded", + metric.WithDescription("Whether the most recent complete qualified shadow window is degraded"), + ) + if err != nil { + return err + } + + _, err = meter.RegisterCallback( + func(_ context.Context, observer metric.Observer) error { + snapshot, ok := connectivitySnapshot() + if !ok { + return nil + } + observer.ObserveInt64(attempts, snapshot.Attempts) + observer.ObserveInt64(signalFailures, snapshot.SignalFailures) + observer.ObserveInt64(distinctTargets, snapshot.DistinctTargets) + observer.ObserveInt64(distinctSignalTargets, snapshot.DistinctSignalTargets) + observer.ObserveInt64(qualified, boolToInt64(snapshot.Qualified)) + observer.ObserveInt64(degraded, boolToInt64(snapshot.Degraded)) + return nil + }, + attempts, + signalFailures, + distinctTargets, + distinctSignalTargets, + qualified, + degraded, + ) return err } +// SetConnectivitySnapshotProvider installs the callback used by shadow gauges. +func SetConnectivitySnapshotProvider(provider func() ConnectivitySnapshot) { + connectivityProviderMu.Lock() + defer connectivityProviderMu.Unlock() + connectivityProvider = provider +} + +func connectivitySnapshot() (ConnectivitySnapshot, bool) { + connectivityProviderMu.RLock() + provider := connectivityProvider + connectivityProviderMu.RUnlock() + if provider == nil { + return ConnectivitySnapshot{}, false + } + return provider(), true +} + +func boolToInt64(value bool) int64 { + if value { + return 1 + } + return 0 +} + func RecordHTTPRequest(method string, statusCode int, proxyType string, durationMs float64) { if httpRequestCount == nil { return @@ -128,3 +258,18 @@ func RecordRouting(result string, durationMs float64) { routingResolutions.Add(context.Background(), 1, attrs) routingResolutionDuration.Record(context.Background(), durationMs, attrs) } + +// RecordUpstreamConnect records only low-cardinality connection attributes. +// The target address is intentionally excluded because Sandbox endpoints are +// high-cardinality and short-lived. +func RecordUpstreamConnect(result, proxyType string, durationMs float64) { + if upstreamConnectCount == nil || upstreamConnectDuration == nil { + return + } + attrs := metric.WithAttributes( + attribute.String("connect_result", result), + attribute.String("proxy_type", proxyType), + ) + upstreamConnectCount.Add(context.Background(), 1, attrs) + upstreamConnectDuration.Record(context.Background(), durationMs, attrs) +} diff --git a/components/ingress/pkg/telemetry/metrics_test.go b/components/ingress/pkg/telemetry/metrics_test.go new file mode 100644 index 000000000..4b34f8069 --- /dev/null +++ b/components/ingress/pkg/telemetry/metrics_test.go @@ -0,0 +1,139 @@ +// 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 telemetry + +import ( + "context" + "testing" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +func TestConnectivityMetricsUseLowCardinalityAttributes(t *testing.T) { + resetMetricState() + previousProvider := otel.GetMeterProvider() + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + otel.SetMeterProvider(provider) + t.Cleanup(func() { + SetConnectivitySnapshotProvider(nil) + _ = provider.Shutdown(context.Background()) + resetMetricState() + otel.SetMeterProvider(previousProvider) + }) + + if err := registerIngressMetrics(); err != nil { + t.Fatalf("registerIngressMetrics() error = %v", err) + } + RecordUpstreamConnect("timeout", "http", 125) + providerCalls := 0 + SetConnectivitySnapshotProvider(func() ConnectivitySnapshot { + providerCalls++ + return ConnectivitySnapshot{ + Attempts: 25, + SignalFailures: 5, + DistinctTargets: 7, + DistinctSignalTargets: 3, + Qualified: true, + Degraded: true, + } + }) + + metrics := collectMetrics(t, reader) + if providerCalls != 1 { + t.Fatalf("connectivity snapshot provider calls = %d, want 1", providerCalls) + } + assertInt64Gauge(t, metrics, "ingress.network.shadow.attempts", 25) + assertInt64Gauge(t, metrics, "ingress.network.shadow.signal_failures", 5) + assertInt64Gauge(t, metrics, "ingress.network.shadow.distinct_targets", 7) + assertInt64Gauge(t, metrics, "ingress.network.shadow.distinct_signal_targets", 3) + assertInt64Gauge(t, metrics, "ingress.network.shadow.qualified", 1) + assertInt64Gauge(t, metrics, "ingress.network.shadow.degraded", 1) + assertConnectMetrics(t, metrics) +} + +func collectMetrics(t *testing.T, reader *sdkmetric.ManualReader) map[string]metricdata.Aggregation { + t.Helper() + var resourceMetrics metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &resourceMetrics); err != nil { + t.Fatalf("Collect() error = %v", err) + } + + metrics := make(map[string]metricdata.Aggregation) + for _, scopeMetrics := range resourceMetrics.ScopeMetrics { + for _, collectedMetric := range scopeMetrics.Metrics { + metrics[collectedMetric.Name] = collectedMetric.Data + } + } + return metrics +} + +func assertInt64Gauge(t *testing.T, metrics map[string]metricdata.Aggregation, name string, want int64) { + t.Helper() + data, exists := metrics[name] + if !exists { + t.Fatalf("metric %q was not collected", name) + } + gauge, ok := data.(metricdata.Gauge[int64]) + if !ok || len(gauge.DataPoints) != 1 || gauge.DataPoints[0].Value != want { + t.Fatalf("metric %q data = %#v, want one data point with value %d", name, data, want) + } +} + +func assertConnectMetrics(t *testing.T, metrics map[string]metricdata.Aggregation) { + t.Helper() + count, ok := metrics["ingress.upstream.connect.count"].(metricdata.Sum[int64]) + if !ok || len(count.DataPoints) != 1 || count.DataPoints[0].Value != 1 { + t.Fatalf("unexpected connect count: %#v", metrics["ingress.upstream.connect.count"]) + } + duration, ok := metrics["ingress.upstream.connect.duration"].(metricdata.Histogram[float64]) + if !ok || len(duration.DataPoints) != 1 || duration.DataPoints[0].Count != 1 || duration.DataPoints[0].Sum != 125 { + t.Fatalf("unexpected connect duration: %#v", metrics["ingress.upstream.connect.duration"]) + } + assertConnectAttributes(t, count.DataPoints[0].Attributes) + assertConnectAttributes(t, duration.DataPoints[0].Attributes) +} + +func assertConnectAttributes(t *testing.T, attributes attribute.Set) { + t.Helper() + for key, want := range map[attribute.Key]string{ + "connect_result": "timeout", + "proxy_type": "http", + } { + value, exists := attributes.Value(key) + if !exists || value.AsString() != want { + t.Fatalf("attribute %q = %q, want %q", key, value.AsString(), want) + } + } + for _, forbidden := range []attribute.Key{"target", "target_ip", "sandbox_id"} { + if _, exists := attributes.Value(forbidden); exists { + t.Fatalf("metric contains forbidden high-cardinality attribute %q", forbidden) + } + } +} + +func resetMetricState() { + SetConnectivitySnapshotProvider(nil) + meter = nil + httpRequestCount = nil + httpRequestDuration = nil + routingResolutions = nil + routingResolutionDuration = nil + upstreamConnectCount = nil + upstreamConnectDuration = nil +} diff --git a/docs/components/ingress.md b/docs/components/ingress.md index 0d6b8b17b..992355904 100644 --- a/docs/components/ingress.md +++ b/docs/components/ingress.md @@ -12,7 +12,7 @@ description: HTTP/WebSocket reverse proxy that routes traffic to OpenSandbox ins - AgentSandbox: reads `status.serviceFQDN`. - Can serve fleets routes from the same ingress when `--fastpath-endpoint` is set. - Fleets routes lazily call FastPath v2 `ResolveEndpoint` when traffic arrives. -- Exposes `/status.ok` health check; prints build metadata (version, commit, time, Go/platform) at startup. +- Exposes `/status.ok` health check and a shadow-only network readiness assessment at `/status.ok/network-readiness`; prints build metadata (version, commit, time, Go/platform) at startup. ## Quick Start ```bash @@ -25,7 +25,55 @@ go run main.go \ --port 28888 \ --log-level info ``` -Endpoints: `/` (proxy), `/status.ok` (health). +Endpoints: `/` (proxy), `/status.ok` (health), `/status.ok/network-readiness` (shadow network assessment). + +## Network Readiness Observation + +Ingress observes the TCP connections that its HTTP transport and WebSocket +dialer open to upstream targets. Each connection is classified as `success`, +`timeout`, `unreachable`, `refused`, `dns_error`, `canceled`, or `other`. +Only timeouts and unreachable errors are treated as possible source-side +network degradation signals. Canceled connections remain visible in the +per-result connection metric but are excluded from the assessment denominator +because they do not establish whether the network path was healthy. + +The assessment uses the most recent complete fixed window. It requires enough +connection attempts, distinct upstream targets, and distinct failing targets +before reporting `DEGRADED`. The endpoint remains shadow-only: +`/status.ok/network-readiness` always returns HTTP 200 with a body of `OK` or +`DEGRADED`, and responses are never cacheable. This path is reserved by the +Ingress itself. The normal `/status.ok` liveness and readiness endpoint is +unchanged. + +| Flag | Default | Description | +|------|---------|-------------| +| `--network-readiness-shadow-window` | `1m` | Fixed aggregation window | +| `--network-readiness-shadow-max-targets` | `1024` | Maximum distinct targets retained per window | +| `--network-readiness-shadow-min-attempts` | `20` | Minimum connection attempts required to qualify a window | +| `--network-readiness-shadow-min-targets` | `5` | Minimum distinct targets required to qualify a window | +| `--network-readiness-shadow-min-signal-targets` | `2` | Minimum distinct targets with timeout or unreachable results | +| `--network-readiness-shadow-failure-ratio` | `0.2` | Failure ratio required to report `DEGRADED` | + +Invalid shadow settings disable connection observation and make the shadow +endpoint return HTTP 404; they do not stop the Ingress data plane. + +The following OpenTelemetry metrics are emitted when OTLP metrics are enabled: + +- `ingress.upstream.connect.count` and `ingress.upstream.connect.duration`, + labeled only by connection result and proxy type. +- `ingress.network.shadow.*` gauges for attempts, signal failures, distinct + targets, qualification, and the shadow decision. + +`attempts` counts physical TCP connections, not HTTP requests. Distinct targets +are network hosts; multiple ports on the same host intentionally count once. +HTTP keep-alive +can therefore make the sample count much lower than the request count. Target +addresses and Sandbox IDs are intentionally excluded from metric attributes. +Deployments that route through a fixed central proxy, or otherwise connect to +fewer than the configured minimum number of targets, may never qualify with +the default thresholds. If `HTTP_PROXY` or `HTTPS_PROXY` is configured, HTTP +observations describe the connection to that proxy rather than the final +Sandbox endpoint. ## Routing Modes