From d20a232996007dfe6ab425abc0a39a3ae9a0889b Mon Sep 17 00:00:00 2001 From: fatedier Date: Thu, 10 Sep 2026 01:28:26 +0800 Subject: [PATCH] test: strengthen health check coverage (#5528) --- Release.md | 1 + client/health/health.go | 82 +++-- client/health/health_test.go | 556 ++++++++++++++++++++++++++++++++-- test/e2e/v1/features/group.go | 271 ++++++++++++++--- 4 files changed, 824 insertions(+), 86 deletions(-) diff --git a/Release.md b/Release.md index 5199f328..68d7929d 100644 --- a/Release.md +++ b/Release.md @@ -1,3 +1,4 @@ ## Fixes * Fixed VirtualNet route lifecycle issues during reconnect and shutdown, including stale route cleanup, shutdown races, and reconnect backoff overflow. +* Fixed health check failure counts not resetting after a successful check, ensuring `healthCheck.maxFailed` applies to consecutive failures. diff --git a/client/health/health.go b/client/health/health.go index 8cb1122a..cedbba37 100644 --- a/client/health/health.go +++ b/client/health/health.go @@ -30,6 +30,11 @@ import ( var ErrHealthCheckType = errors.New("error health check type") +func newHealthTimer(interval time.Duration) (<-chan time.Time, func()) { + timer := time.NewTimer(interval) + return timer.C, func() { timer.Stop() } +} + type Monitor struct { checkType string interval time.Duration @@ -49,6 +54,9 @@ type Monitor struct { ctx context.Context cancel context.CancelFunc + doneCh chan struct{} + + timerFactory func(time.Duration) (<-chan time.Time, func()) } func NewMonitor(ctx context.Context, cfg v1.HealthCheckConfig, addr string, @@ -91,6 +99,8 @@ func NewMonitor(ctx context.Context, cfg v1.HealthCheckConfig, addr string, statusFailedFn: statusFailedFn, ctx: newctx, cancel: cancel, + doneCh: make(chan struct{}), + timerFactory: newHealthTimer, } } @@ -102,40 +112,66 @@ func (monitor *Monitor) Stop() { monitor.cancel() } +// Done is closed when the worker launched by Start has exited. +func (monitor *Monitor) Done() <-chan struct{} { + return monitor.doneCh +} + func (monitor *Monitor) checkWorker() { - xl := xlog.FromContextSafe(monitor.ctx) + defer close(monitor.doneCh) + for { + if monitor.ctx.Err() != nil { + return + } + doCtx, cancel := context.WithDeadline(monitor.ctx, time.Now().Add(monitor.timeout)) err := monitor.doCheck(doCtx) + cancel() // check if this monitor has been closed - select { - case <-monitor.ctx.Done(): - cancel() + if monitor.ctx.Err() != nil { return - default: - cancel() } + monitor.handleCheckResult(err) - if err == nil { - xl.Tracef("do one health check success") - monitor.failedTimes = 0 - if !monitor.statusOK && monitor.statusNormalFn != nil { - xl.Infof("health check status change to success") - monitor.statusOK = true - monitor.statusNormalFn() - } - } else { - xl.Warnf("do one health check failed: %v", err) - monitor.failedTimes++ - if monitor.statusOK && int(monitor.failedTimes) >= monitor.maxFailedTimes && monitor.statusFailedFn != nil { - xl.Warnf("health check status change to failed") - monitor.statusOK = false - monitor.statusFailedFn() - } + if !monitor.waitForNextCheck() { + return } + } +} - time.Sleep(monitor.interval) +func (monitor *Monitor) handleCheckResult(err error) { + xl := xlog.FromContextSafe(monitor.ctx) + if err == nil { + xl.Tracef("do one health check success") + monitor.failedTimes = 0 + if !monitor.statusOK && monitor.statusNormalFn != nil { + xl.Infof("health check status change to success") + monitor.statusOK = true + monitor.statusNormalFn() + } + return + } + + xl.Warnf("do one health check failed: %v", err) + monitor.failedTimes++ + if monitor.statusOK && int(monitor.failedTimes) >= monitor.maxFailedTimes && monitor.statusFailedFn != nil { + xl.Warnf("health check status change to failed") + monitor.statusOK = false + monitor.statusFailedFn() + } +} + +func (monitor *Monitor) waitForNextCheck() bool { + timerC, stopTimer := monitor.timerFactory(monitor.interval) + defer stopTimer() + + select { + case <-monitor.ctx.Done(): + return false + case <-timerC: + return monitor.ctx.Err() == nil } } diff --git a/client/health/health_test.go b/client/health/health_test.go index f43edd04..80705384 100644 --- a/client/health/health_test.go +++ b/client/health/health_test.go @@ -16,9 +16,12 @@ package health import ( "context" + "errors" + "net" "net/http" "net/http/httptest" "strings" + "sync" "sync/atomic" "testing" "time" @@ -28,38 +31,545 @@ import ( v1 "github.com/fatedier/frp/pkg/config/v1" ) -func TestMonitorResetsFailedTimesAfterSuccess(t *testing.T) { - var checkCount atomic.Int32 - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - count := checkCount.Add(1) - if count == 1 || count == 2 || count == 4 { - w.WriteHeader(http.StatusServiceUnavailable) +type tcpHealthBackend struct { + listener net.Listener + accepted chan struct{} + done chan struct{} +} + +func newTCPHealthBackend(t *testing.T, addr string, accepted chan struct{}) *tcpHealthBackend { + listener, err := net.Listen("tcp", addr) + require.NoError(t, err) + + backend := &tcpHealthBackend{ + listener: listener, + accepted: accepted, + done: make(chan struct{}), + } + go func() { + defer close(backend.done) + for { + conn, err := backend.listener.Accept() + if err != nil { + return + } + _ = conn.Close() + select { + case backend.accepted <- struct{}{}: + default: + } + } + }() + return backend +} + +func (backend *tcpHealthBackend) Close() { + _ = backend.listener.Close() + <-backend.done +} + +func TestMonitorConsecutiveFailureWindows(t *testing.T) { + tests := []struct { + name string + setup func(*testing.T, func(), func()) (*Monitor, func(bool)) + }{ + { + name: "HTTP", + setup: func(t *testing.T, normalFn, failedFn func()) (*Monitor, func(bool)) { + var healthy atomic.Bool + healthy.Store(true) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if !healthy.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + } + })) + t.Cleanup(server.Close) + + monitor := NewMonitor( + context.Background(), + v1.HealthCheckConfig{ + Type: "http", + Path: "/health", + TimeoutSeconds: 1, + IntervalSeconds: 1, + MaxFailed: 3, + }, + strings.TrimPrefix(server.URL, "http://"), + normalFn, + failedFn, + ) + return monitor, healthy.Store + }, + }, + { + name: "TCP", + setup: func(t *testing.T, normalFn, failedFn func()) (*Monitor, func(bool)) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = listener.Close() }) + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + _ = conn.Close() + } + }() + + monitor := NewMonitor( + context.Background(), + v1.HealthCheckConfig{ + Type: "tcp", + TimeoutSeconds: 1, + IntervalSeconds: 1, + MaxFailed: 3, + }, + listener.Addr().String(), + normalFn, + failedFn, + ) + healthyAddr := monitor.addr + return monitor, func(healthy bool) { + if healthy { + monitor.addr = healthyAddr + } else { + monitor.addr = "127.0.0.1:0" + } + } + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var events []string + monitor, setHealthy := test.setup( + t, + func() { events = append(events, "normal") }, + func() { events = append(events, "failed") }, + ) + t.Cleanup(monitor.Stop) + + runCheck := func(healthy bool) { + t.Helper() + setHealthy(healthy) + ctx, cancel := context.WithTimeout(monitor.ctx, time.Second) + err := monitor.doCheck(ctx) + cancel() + if healthy { + require.NoError(t, err) + } else { + require.Error(t, err) + } + monitor.handleCheckResult(err) + } + + runCheck(true) + require.True(t, monitor.statusOK) + require.Zero(t, monitor.failedTimes) + require.Equal(t, []string{"normal"}, events) + + runCheck(false) + runCheck(false) + require.True(t, monitor.statusOK) + require.Equal(t, uint64(2), monitor.failedTimes) + require.Equal(t, []string{"normal"}, events) + + runCheck(true) + require.True(t, monitor.statusOK) + require.Zero(t, monitor.failedTimes) + require.Equal(t, []string{"normal"}, events) + + runCheck(false) + require.True(t, monitor.statusOK) + require.Equal(t, uint64(1), monitor.failedTimes) + require.Equal(t, []string{"normal"}, events) + + runCheck(false) + require.True(t, monitor.statusOK) + require.Equal(t, uint64(2), monitor.failedTimes) + + runCheck(false) + require.False(t, monitor.statusOK) + require.Equal(t, uint64(3), monitor.failedTimes) + require.Equal(t, []string{"normal", "failed"}, events) + + runCheck(true) + require.True(t, monitor.statusOK) + require.Zero(t, monitor.failedTimes) + require.Equal(t, []string{"normal", "failed", "normal"}, events) + }) + } +} + +type workerStatus struct { + failedTimes uint64 + statusOK bool +} + +type workerInterval struct { + status workerStatus + timer chan time.Time +} + +func observeWorkerIntervals(monitor *Monitor) <-chan workerInterval { + intervals := make(chan workerInterval, 1) + monitor.timerFactory = func(time.Duration) (<-chan time.Time, func()) { + timer := make(chan time.Time, 1) + // The worker takes this snapshot after handleCheckResult. Sending it + // publishes the state to the test; no test goroutine reads live fields. + interval := workerInterval{ + status: workerStatus{failedTimes: monitor.failedTimes, statusOK: monitor.statusOK}, + timer: timer, + } + select { + case intervals <- interval: + case <-monitor.ctx.Done(): + } + return timer, func() {} + } + return intervals +} + +func awaitWorkerInterval(t *testing.T, intervals <-chan workerInterval, want workerStatus) chan time.Time { + t.Helper() + select { + case interval := <-intervals: + require.Equal(t, want, interval.status) + return interval.timer + case <-time.After(time.Second): + t.Fatal("health worker did not reach the interval barrier") + return nil + } +} + +func stopMonitorWorker(t *testing.T, monitor *Monitor) { + t.Helper() + monitor.Stop() + select { + case <-monitor.Done(): + case <-time.After(time.Second): + t.Error("health worker did not exit after Stop") + } +} + +func TestMonitorWorkerProcessesResults(t *testing.T) { + requestReady := make(chan struct{}, 1) + responses := make(chan int) + requestCanceled := make(chan struct{}) + var cancelOnce sync.Once + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case requestReady <- struct{}{}: + case <-r.Context().Done(): return } - w.WriteHeader(http.StatusOK) + select { + case code := <-responses: + w.WriteHeader(code) + case <-r.Context().Done(): + cancelOnce.Do(func() { close(requestCanceled) }) + } })) - defer server.Close() + t.Cleanup(server.Close) - var failedCount atomic.Int32 + events := make(chan string, 3) monitor := NewMonitor( context.Background(), v1.HealthCheckConfig{ - Type: "http", - Path: "/health", - TimeoutSeconds: 1, - IntervalSeconds: 1, - MaxFailed: 3, + Type: "http", + Path: "/health", + TimeoutSeconds: 5, + MaxFailed: 3, }, strings.TrimPrefix(server.URL, "http://"), - func() {}, - func() { failedCount.Add(1) }, + func() { events <- "normal" }, + func() { events <- "failed" }, ) - monitor.interval = 10 * time.Millisecond - monitor.Start() - defer monitor.Stop() + intervals := observeWorkerIntervals(monitor) + t.Cleanup(func() { stopMonitorWorker(t, monitor) }) - require.Eventually(t, func() bool { - return checkCount.Load() >= 5 - }, time.Second, 10*time.Millisecond) - require.Equal(t, int32(0), failedCount.Load()) + awaitRequest := func() { + t.Helper() + select { + case <-requestReady: + case <-time.After(time.Second): + t.Fatal("health check request did not start") + } + } + awaitEvent := func(want string) { + t.Helper() + select { + case got := <-events: + require.Equal(t, want, got) + case <-time.After(time.Second): + t.Fatalf("health check callback %q was not called", want) + } + } + checkResult := func(code int, want workerStatus, event string) { + t.Helper() + awaitRequest() + select { + case responses <- code: + case <-time.After(time.Second): + t.Fatal("health check handler did not accept the response") + } + timer := awaitWorkerInterval(t, intervals, want) + if event != "" { + awaitEvent(event) + } + // The worker has finished this result and cannot emit another + // callback until the test releases the next check. + select { + case got := <-events: + t.Fatalf("unexpected health check callback %q", got) + default: + } + timer <- time.Now() + } + + monitor.Start() + checkResult(http.StatusOK, workerStatus{statusOK: true}, "normal") + checkResult(http.StatusServiceUnavailable, workerStatus{failedTimes: 1, statusOK: true}, "") + checkResult(http.StatusServiceUnavailable, workerStatus{failedTimes: 2, statusOK: true}, "") + checkResult(http.StatusOK, workerStatus{statusOK: true}, "") + checkResult(http.StatusServiceUnavailable, workerStatus{failedTimes: 1, statusOK: true}, "") + checkResult(http.StatusServiceUnavailable, workerStatus{failedTimes: 2, statusOK: true}, "") + checkResult(http.StatusServiceUnavailable, workerStatus{failedTimes: 3}, "failed") + checkResult(http.StatusOK, workerStatus{statusOK: true}, "normal") + + awaitRequest() + stopMonitorWorker(t, monitor) + select { + case <-requestCanceled: + case <-time.After(time.Second): + t.Fatal("health check request context was not canceled") + } +} + +func TestMonitorTCPWorkerProcessesResults(t *testing.T) { + initialAccepted := make(chan struct{}, 1) + initialBackend := newTCPHealthBackend(t, "127.0.0.1:0", initialAccepted) + addr := initialBackend.listener.Addr().String() + recoveryAccepted := make(chan struct{}, 1) + + normalCallbacks := make(chan workerStatus, 2) + failedCallbacks := make(chan workerStatus, 2) + var monitor *Monitor + monitor = NewMonitor( + context.Background(), + v1.HealthCheckConfig{ + Type: "tcp", + TimeoutSeconds: 1, + MaxFailed: 3, + }, + addr, + func() { + normalCallbacks <- workerStatus{failedTimes: monitor.failedTimes, statusOK: monitor.statusOK} + }, + func() { + failedCallbacks <- workerStatus{failedTimes: monitor.failedTimes, statusOK: monitor.statusOK} + }, + ) + intervals := observeWorkerIntervals(monitor) + recoveryBackend := (*tcpHealthBackend)(nil) + t.Cleanup(func() { + stopMonitorWorker(t, monitor) + initialBackend.Close() + if recoveryBackend != nil { + recoveryBackend.Close() + } + }) + + awaitStatus := func(ch <-chan workerStatus, want workerStatus, message string) { + t.Helper() + select { + case got := <-ch: + require.Equal(t, want, got) + case <-time.After(time.Second): + t.Fatal(message) + } + } + + monitor.Start() + select { + case <-initialAccepted: + case <-time.After(time.Second): + t.Fatal("TCP health check did not reach the initial backend") + } + awaitStatus(normalCallbacks, workerStatus{failedTimes: 0, statusOK: true}, "TCP worker did not report the initial success") + + initialTimer := awaitWorkerInterval(t, intervals, workerStatus{statusOK: true}) + initialBackend.Close() + initialTimer <- time.Now() + + firstFailureTimer := awaitWorkerInterval(t, intervals, workerStatus{failedTimes: 1, statusOK: true}) + firstFailureTimer <- time.Now() + + secondFailureTimer := awaitWorkerInterval(t, intervals, workerStatus{failedTimes: 2, statusOK: true}) + secondFailureTimer <- time.Now() + + awaitStatus(failedCallbacks, workerStatus{failedTimes: 3, statusOK: false}, "TCP worker did not report the third failed health check") + thirdFailureTimer := awaitWorkerInterval(t, intervals, workerStatus{failedTimes: 3}) + + recoveryBackend = newTCPHealthBackend(t, addr, recoveryAccepted) + thirdFailureTimer <- time.Now() + select { + case <-recoveryAccepted: + case <-time.After(time.Second): + t.Fatal("TCP health check did not reach the recovery backend") + } + awaitStatus(normalCallbacks, workerStatus{failedTimes: 0, statusOK: true}, "TCP worker did not report recovery") + + recoveryTimer := awaitWorkerInterval(t, intervals, workerStatus{statusOK: true}) + recoveryBackend.Close() + recoveryTimer <- time.Now() + + firstRecoveryFailureTimer := awaitWorkerInterval(t, intervals, workerStatus{failedTimes: 1, statusOK: true}) + firstRecoveryFailureTimer <- time.Now() + + secondRecoveryFailureTimer := awaitWorkerInterval(t, intervals, workerStatus{failedTimes: 2, statusOK: true}) + secondRecoveryFailureTimer <- time.Now() + + awaitStatus(failedCallbacks, workerStatus{failedTimes: 3, statusOK: false}, "TCP worker did not report the third failed health check after recovery") + stopMonitorWorker(t, monitor) +} + +func TestMonitorMaxFailedOne(t *testing.T) { + var events []string + monitor := NewMonitor( + context.Background(), + v1.HealthCheckConfig{Type: "tcp", MaxFailed: 1}, + "", + func() { events = append(events, "normal") }, + func() { events = append(events, "failed") }, + ) + t.Cleanup(monitor.Stop) + + checkErr := errors.New("health check failed") + monitor.handleCheckResult(nil) + monitor.handleCheckResult(checkErr) + require.Equal(t, []string{"normal", "failed"}, events) + require.False(t, monitor.statusOK) + require.Equal(t, uint64(1), monitor.failedTimes) + + monitor.handleCheckResult(checkErr) + require.Equal(t, []string{"normal", "failed"}, events) + + monitor.handleCheckResult(nil) + monitor.handleCheckResult(checkErr) + require.Equal(t, []string{"normal", "failed", "normal", "failed"}, events) +} + +func TestMonitorStopCancelsWork(t *testing.T) { + t.Run("in-flight check", func(t *testing.T) { + requestStarted := make(chan struct{}) + requestCanceled := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + close(requestStarted) + <-r.Context().Done() + close(requestCanceled) + })) + t.Cleanup(server.Close) + + monitor := NewMonitor( + context.Background(), + v1.HealthCheckConfig{Type: "http", Path: "/health"}, + strings.TrimPrefix(server.URL, "http://"), + func() {}, + func() {}, + ) + t.Cleanup(func() { stopMonitorWorker(t, monitor) }) + monitor.Start() + + select { + case <-requestStarted: + case <-time.After(time.Second): + t.Fatal("health check request did not start") + } + stopMonitorWorker(t, monitor) + select { + case <-requestCanceled: + case <-time.After(time.Second): + t.Fatal("health check request context was not canceled") + } + }) + + t.Run("interval wait", func(t *testing.T) { + monitor := NewMonitor(context.Background(), v1.HealthCheckConfig{Type: "tcp"}, "", nil, nil) + intervals := observeWorkerIntervals(monitor) + t.Cleanup(func() { stopMonitorWorker(t, monitor) }) + monitor.Start() + + // A real worker has completed its first check and installed a timer + // that will never fire. Stop must release the wait and exit the loop. + awaitWorkerInterval(t, intervals, workerStatus{}) + stopMonitorWorker(t, monitor) + }) +} + +type gatedErrContext struct { + context.Context + errCalled chan struct{} + releaseErr chan struct{} +} + +func (ctx *gatedErrContext) Err() error { + select { + case ctx.errCalled <- struct{}{}: + default: + } + <-ctx.releaseErr + return ctx.Context.Err() +} + +func TestMonitorTimerCancellationAfterTimerFires(t *testing.T) { + baseCtx, cancel := context.WithCancel(context.Background()) + ctx := &gatedErrContext{ + Context: baseCtx, + errCalled: make(chan struct{}, 1), + releaseErr: make(chan struct{}), + } + monitor := NewMonitor( + context.Background(), + v1.HealthCheckConfig{Type: "tcp"}, + "", + nil, + nil, + ) + monitor.ctx = ctx + monitor.cancel = cancel + + timerReady := make(chan time.Time, 1) + timerReady <- time.Now() + var timerStopped atomic.Bool + monitor.timerFactory = func(time.Duration) (<-chan time.Time, func()) { + return timerReady, func() { timerStopped.Store(true) } + } + + waitResult := make(chan bool, 1) + go func() { + waitResult <- monitor.waitForNextCheck() + }() + + // ctx.Done is not ready when select runs, so receiving timerReady is the + // only possible branch. Err signals after that receive and blocks until the + // test cancels the context, deterministically exercising the cancellation + // re-check in the timer branch. + select { + case <-ctx.errCalled: + case <-time.After(time.Second): + t.Fatal("timer branch did not re-check the monitor context") + } + cancel() + close(ctx.releaseErr) + + select { + case shouldContinue := <-waitResult: + require.False(t, shouldContinue) + case <-time.After(time.Second): + t.Fatal("timer wait did not observe context cancellation") + } + require.True(t, timerStopped.Load()) } diff --git a/test/e2e/v1/features/group.go b/test/e2e/v1/features/group.go index 85b938a9..87270e7a 100644 --- a/test/e2e/v1/features/group.go +++ b/test/e2e/v1/features/group.go @@ -1,14 +1,20 @@ package features import ( + "context" "crypto/tls" + "encoding/json" "fmt" + "net/http" + "net/url" "strconv" "sync" "time" "github.com/onsi/ginkgo/v2" + "github.com/onsi/gomega" + "github.com/fatedier/frp/pkg/sdk/client" "github.com/fatedier/frp/pkg/transport" "github.com/fatedier/frp/test/e2e/framework" "github.com/fatedier/frp/test/e2e/framework/consts" @@ -17,6 +23,80 @@ import ( "github.com/fatedier/frp/test/e2e/pkg/request" ) +func waitForProxyStatus(proxyClient *client.Client, proxyName, want string) error { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + ticker := time.NewTicker(25 * time.Millisecond) + defer ticker.Stop() + + var lastStatus string + var lastErr error + for { + status, err := proxyClient.GetProxyStatus(ctx, proxyName) + if err == nil { + lastErr = nil + lastStatus = status.Status + if status.Status == want { + return nil + } + } else { + lastErr = err + } + + select { + case <-ctx.Done(): + if lastErr != nil { + return fmt.Errorf("timeout waiting for proxy %q status %q: last error: %w", proxyName, want, lastErr) + } + return fmt.Errorf("timeout waiting for proxy %q status %q: last status %q", proxyName, want, lastStatus) + case <-ticker.C: + } + } +} + +func waitForSignal(signal <-chan struct{}) error { + timer := time.NewTimer(5 * time.Second) + defer timer.Stop() + + select { + case <-signal: + return nil + case <-timer.C: + return fmt.Errorf("timeout waiting for backend health-check signal") + } +} + +// Keep this longer than client/proxy.statusCheckInterval (3s). The wrapper's +// health notification is deliberately non-blocking, so the E2E assertion must +// also cover the fallback poll plus scheduling margin while recovery is gated. +const proxyFallbackObservationWindow = 4 * time.Second + +func waitForServerProxyStatus(port int, proxyName, want string, timeout time.Duration) error { + return waitForLifecycleCondition(timeout, func() error { + body, err := getLifecycleEndpoint(port, "/api/proxies/"+url.PathEscape(proxyName)) + if err != nil { + return err + } + var status struct { + Status string `json:"status"` + } + if err := json.Unmarshal([]byte(body), &status); err != nil { + return err + } + if status.Status != want { + return fmt.Errorf("frps proxy %q status %q, want %q", proxyName, status.Status, want) + } + return nil + }) +} + +type httpHealthStage struct { + healthy bool + signal chan<- struct{} + release <-chan struct{} +} + var _ = ginkgo.Describe("[Feature: Group]", func() { f := framework.NewDefaultFramework() @@ -252,7 +332,10 @@ var _ = ginkgo.Describe("[Feature: Group]", func() { ginkgo.Describe("Health Check", func() { ginkgo.It("TCP", func() { - serverConf := consts.DefaultServerConfig + dashboardPort := f.AllocPort() + serverConf := consts.DefaultServerConfig + fmt.Sprintf(` + webServer.port = %d + `, dashboardPort) clientConf := consts.DefaultClientConfig fooPort := f.AllocPort() @@ -260,11 +343,17 @@ var _ = ginkgo.Describe("[Feature: Group]", func() { f.RunServer("", fooServer) barPort := f.AllocPort() - barServer := streamserver.New(streamserver.TCP, streamserver.WithBindPort(barPort), streamserver.WithRespContent([]byte("bar"))) + newBarServer := func() *streamserver.Server { + return streamserver.New(streamserver.TCP, streamserver.WithBindPort(barPort), streamserver.WithRespContent([]byte("bar"))) + } + barServer := newBarServer() f.RunServer("", barServer) remotePort := f.AllocPort() + adminPort := f.AllocPort() clientConf += fmt.Sprintf(` + webServer.port = %d + [[proxies]] name = "foo" type = "tcp" @@ -284,11 +373,16 @@ var _ = ginkgo.Describe("[Feature: Group]", func() { loadBalancer.groupKey = "123" healthCheck.type = "tcp" healthCheck.intervalSeconds = 1 - `, fooPort, remotePort, barPort, remotePort) + healthCheck.maxFailed = 3 + `, adminPort, fooPort, remotePort, barPort, remotePort) - _, clientProcesses := f.RunProcesses(serverConf, []string{clientConf}) + f.RunProcesses(serverConf, []string{clientConf}) + proxyClient := f.APIClientForFrpc(adminPort) + framework.ExpectNoError(waitForProxyStatus(proxyClient, "foo", "running")) + framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "running")) - // check foo and bar is ok + // Both requests traverse frps, the load-balancing group, frpc, and the + // corresponding local backend. results := []string{} for range 10 { framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool { @@ -298,33 +392,38 @@ var _ = ginkgo.Describe("[Feature: Group]", func() { } framework.ExpectContainElements(results, []string{"foo", "bar"}) - // close bar server, check foo is ok - failedCount := clientProcesses[0].CountOutput("[bar] health check failed") - barServer.Close() - framework.ExpectNoError(clientProcesses[0].WaitForOutput("[bar] health check failed", failedCount+1, 5*time.Second)) - for range 10 { - framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure() - } + // frps removes the group listener before deleting the proxy from + // its manager. Its offline status is the removal barrier; frpc's + // local check-failed status alone does not acknowledge that work. + for range 2 { + framework.ExpectNoError(barServer.Close()) + framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "check failed")) + framework.ExpectNoError(waitForServerProxyStatus(dashboardPort, "bar", "offline", 5*time.Second)) + for range 10 { + framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure() + } - // resume bar server, check foo and bar is ok - successCount := clientProcesses[0].CountOutput("[bar] health check success") - f.RunServer("", barServer) - framework.ExpectNoError(clientProcesses[0].WaitForOutput("[bar] health check success", successCount+1, 5*time.Second)) - results = []string{} - for range 10 { - framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool { - results = append(results, string(resp.Content)) - return true - }) + barServer = newBarServer() + f.RunServer("", barServer) + framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "running")) + results = []string{} + for range 10 { + framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool { + results = append(results, string(resp.Content)) + return true + }) + } + framework.ExpectContainElements(results, []string{"foo", "bar"}) } - framework.ExpectContainElements(results, []string{"foo", "bar"}) }) ginkgo.It("HTTP", func() { vhostPort := f.AllocPort() + dashboardPort := f.AllocPort() serverConf := consts.DefaultServerConfig + fmt.Sprintf(` vhostHTTPPort = %d - `, vhostPort) + webServer.port = %d + `, vhostPort, dashboardPort) clientConf := consts.DefaultClientConfig fooPort := f.AllocPort() @@ -332,10 +431,41 @@ var _ = ginkgo.Describe("[Feature: Group]", func() { f.RunServer("", fooServer) barPort := f.AllocPort() - barServer := newHTTPServer(barPort, "bar") + var stageMu sync.RWMutex + stage := &httpHealthStage{healthy: true, signal: make(chan struct{}, 1)} + barServer := httpserver.New( + httpserver.WithBindPort(barPort), + httpserver.WithHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + stageMu.RLock() + currentStage := stage + stageMu.RUnlock() + + if r.URL.Path == "/healthz" { + if !currentStage.healthy { + w.WriteHeader(http.StatusServiceUnavailable) + } + select { + case currentStage.signal <- struct{}{}: + default: + } + if currentStage.release != nil { + select { + case <-currentStage.release: + case <-r.Context().Done(): + return + } + } + return + } + _, _ = w.Write([]byte("bar")) + })), + ) f.RunServer("", barServer) + adminPort := f.AllocPort() clientConf += fmt.Sprintf(` + webServer.port = %d + [[proxies]] name = "foo" type = "http" @@ -356,12 +486,17 @@ var _ = ginkgo.Describe("[Feature: Group]", func() { loadBalancer.groupKey = "123" healthCheck.type = "http" healthCheck.intervalSeconds = 1 + healthCheck.maxFailed = 3 + healthCheck.timeoutSeconds = 10 healthCheck.path = "/healthz" - `, fooPort, barPort) + `, adminPort, fooPort, barPort) - _, clientProcesses := f.RunProcesses(serverConf, []string{clientConf}) + f.RunProcesses(serverConf, []string{clientConf}) + proxyClient := f.APIClientForFrpc(adminPort) + framework.ExpectNoError(waitForProxyStatus(proxyClient, "foo", "running")) + framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "running")) - // send first HTTP request + // Ordinary requests traverse the real HTTP proxy path. var contents []string framework.NewRequestExpect(f).Port(vhostPort). RequestModify(func(r *request.Request) { @@ -371,8 +506,6 @@ var _ = ginkgo.Describe("[Feature: Group]", func() { contents = append(contents, string(resp.Content)) return true }) - - // send second HTTP request, should be forwarded to another service framework.NewRequestExpect(f).Port(vhostPort). RequestModify(func(r *request.Request) { r.HTTP().HTTPHost("example.com") @@ -381,25 +514,83 @@ var _ = ginkgo.Describe("[Feature: Group]", func() { contents = append(contents, string(resp.Content)) return true }) - framework.ExpectContainElements(contents, []string{"foo", "bar"}) - // check foo and bar is ok + // Every failed health response is gated by the test. This gives each + // stage a request-level barrier and avoids cumulative process logs. + runFailureRecovery := func() { + failureSignals := make(chan struct{}, 1) + release := make(chan struct{}, 1) + stageMu.Lock() + stage = &httpHealthStage{healthy: false, signal: failureSignals, release: release} + stageMu.Unlock() + + framework.ExpectNoError(waitForSignal(failureSignals)) + release <- struct{}{} + framework.ExpectNoError(waitForSignal(failureSignals)) + + recoverySignals := make(chan struct{}, 1) + recoveryRelease := make(chan struct{}) + stageMu.Lock() + stage = &httpHealthStage{healthy: true, signal: recoverySignals, release: recoveryRelease} + stageMu.Unlock() + release <- struct{}{} + + // The next request can only start after the worker consumed the + // second failed response. Hold this success response until the + // assertion finishes, so premature removal cannot be hidden by + // recovery. Consistently also gives the proxy worker time to act + // on an erroneous failure notification without retrying it away. + framework.ExpectNoError(waitForSignal(recoverySignals)) + gomega.Consistently(func() string { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + status, err := proxyClient.GetProxyStatus(ctx, "bar") + if err != nil { + return err.Error() + } + return status.Status + }, proxyFallbackObservationWindow, 25*time.Millisecond).Should(gomega.Equal("running")) + close(recoveryRelease) + } + runFailureRecovery() results := doFooBarHTTPRequest(vhostPort, "example.com") framework.ExpectContainElements(results, []string{"foo", "bar"}) - // close bar server, check foo is ok - failedCount := clientProcesses[0].CountOutput("[bar] health check failed") - barServer.Close() - framework.ExpectNoError(clientProcesses[0].WaitForOutput("[bar] health check failed", failedCount+1, 5*time.Second)) + // Repeat the two-failure window to verify that recovery cleared the + // counter before another failure sequence began. + runFailureRecovery() + results = doFooBarHTTPRequest(vhostPort, "example.com") + framework.ExpectContainElements(results, []string{"foo", "bar"}) + + // Three failed responses reach MaxFailed and remove bar from the + // group; a healthy response then re-registers it. + failureSignals := make(chan struct{}, 1) + release := make(chan struct{}, 1) + stageMu.Lock() + stage = &httpHealthStage{healthy: false, signal: failureSignals, release: release} + stageMu.Unlock() + for range 3 { + framework.ExpectNoError(waitForSignal(failureSignals)) + release <- struct{}{} + } + framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "check failed")) + // HTTPProxy.Close unregisters the group route before frps reports + // offline. Only then assert that fresh data-plane requests use foo. + framework.ExpectNoError(waitForServerProxyStatus(dashboardPort, "bar", "offline", 5*time.Second)) results = doFooBarHTTPRequest(vhostPort, "example.com") framework.ExpectContainElements(results, []string{"foo"}) framework.ExpectNotContainElements(results, []string{"bar"}) - // resume bar server, check foo and bar is ok - successCount := clientProcesses[0].CountOutput("[bar] health check success") - f.RunServer("", barServer) - framework.ExpectNoError(clientProcesses[0].WaitForOutput("[bar] health check success", successCount+1, 5*time.Second)) + recoverySignals := make(chan struct{}, 1) + stageMu.Lock() + stage = &httpHealthStage{healthy: true, signal: recoverySignals} + stageMu.Unlock() + // Release any additional failed request already waiting while the + // removal and data-plane assertions were in progress. + close(release) + framework.ExpectNoError(waitForSignal(recoverySignals)) + framework.ExpectNoError(waitForProxyStatus(proxyClient, "bar", "running")) results = doFooBarHTTPRequest(vhostPort, "example.com") framework.ExpectContainElements(results, []string{"foo", "bar"}) })