mirror of
https://github.com/lukaszraczylo/graphql-monitoring-proxy.git
synced 2026-10-12 05:32:36 +00:00
lukaszraczylo
d410fb3c7e
PingInterval/PongTimeout were configured but unused, so a half-open peer (TCP open but silent) left both proxy goroutines and the handler blocked in ReadMessage forever. Wire both directions: read deadline refreshed by real traffic or pong, plus a sidecar ping goroutine (WriteControl is safe to call concurrently with WriteMessage). Add regression test TestWebSocketProxy_ReapsSilentBackend_Keepalive proving a silent backend is reaped via the read deadline instead of hanging.
398 lines
11 KiB
Go
398 lines
11 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
gorillaws "github.com/gorilla/websocket"
|
|
libpack_logger "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
|
|
libpack_monitoring "github.com/lukaszraczylo/graphql-monitoring-proxy/monitoring"
|
|
"github.com/stretchr/testify/assert"
|
|
)
|
|
|
|
func TestNewWebSocketProxy(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
backendURL string
|
|
config WebSocketConfig
|
|
}{
|
|
{
|
|
name: "default config",
|
|
backendURL: "http://localhost:8080",
|
|
config: WebSocketConfig{
|
|
Enabled: true,
|
|
PingInterval: 30 * time.Second,
|
|
PongTimeout: 60 * time.Second,
|
|
MaxMessageSize: 512 * 1024,
|
|
},
|
|
},
|
|
{
|
|
name: "custom config",
|
|
backendURL: "https://graphql.example.com",
|
|
config: WebSocketConfig{
|
|
Enabled: true,
|
|
PingInterval: 10 * time.Second,
|
|
PongTimeout: 20 * time.Second,
|
|
MaxMessageSize: 1024 * 1024,
|
|
},
|
|
},
|
|
{
|
|
name: "disabled config",
|
|
backendURL: "http://localhost:8080",
|
|
config: WebSocketConfig{
|
|
Enabled: false,
|
|
},
|
|
},
|
|
{
|
|
name: "zero values use defaults",
|
|
backendURL: "http://localhost:8080",
|
|
config: WebSocketConfig{
|
|
Enabled: true,
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
logger := libpack_logger.New()
|
|
monitoring := libpack_monitoring.NewMonitoring(&libpack_monitoring.InitConfig{})
|
|
|
|
wsp := NewWebSocketProxy(tt.backendURL, tt.config, logger, monitoring)
|
|
|
|
assert.NotNil(t, wsp)
|
|
assert.Equal(t, tt.backendURL, wsp.backendURL)
|
|
assert.Equal(t, tt.config.Enabled, wsp.enabled)
|
|
|
|
// Check defaults were applied
|
|
if tt.config.PingInterval == 0 {
|
|
assert.Equal(t, 30*time.Second, wsp.pingInterval)
|
|
} else {
|
|
assert.Equal(t, tt.config.PingInterval, wsp.pingInterval)
|
|
}
|
|
|
|
if tt.config.PongTimeout == 0 {
|
|
assert.Equal(t, 60*time.Second, wsp.pongTimeout)
|
|
} else {
|
|
assert.Equal(t, tt.config.PongTimeout, wsp.pongTimeout)
|
|
}
|
|
|
|
if tt.config.MaxMessageSize == 0 {
|
|
assert.Equal(t, int64(512*1024), wsp.maxMessageSize)
|
|
} else {
|
|
assert.Equal(t, tt.config.MaxMessageSize, wsp.maxMessageSize)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestWebSocketProxy_GetStats(t *testing.T) {
|
|
config := WebSocketConfig{
|
|
Enabled: true,
|
|
PingInterval: 30 * time.Second,
|
|
PongTimeout: 60 * time.Second,
|
|
MaxMessageSize: 512 * 1024,
|
|
}
|
|
|
|
wsp := NewWebSocketProxy("http://localhost:8080", config, libpack_logger.New(), nil)
|
|
|
|
// Simulate some activity
|
|
wsp.activeConnections.Store(5)
|
|
wsp.totalConnections.Store(100)
|
|
wsp.messagesSent.Store(1000)
|
|
wsp.messagesReceived.Store(2000)
|
|
wsp.errors.Store(10)
|
|
|
|
stats := wsp.GetStats()
|
|
|
|
assert.Equal(t, true, stats["enabled"])
|
|
assert.Equal(t, int64(5), stats["active_connections"])
|
|
assert.Equal(t, int64(100), stats["total_connections"])
|
|
assert.Equal(t, int64(1000), stats["messages_sent"])
|
|
assert.Equal(t, int64(2000), stats["messages_received"])
|
|
assert.Equal(t, int64(10), stats["errors"])
|
|
assert.Equal(t, "30s", stats["ping_interval"])
|
|
assert.Equal(t, "1m0s", stats["pong_timeout"])
|
|
assert.Equal(t, int64(512*1024), stats["max_message_size"])
|
|
}
|
|
|
|
func TestWebSocketProxy_GetStats_Disabled(t *testing.T) {
|
|
config := WebSocketConfig{
|
|
Enabled: false,
|
|
}
|
|
|
|
wsp := NewWebSocketProxy("http://localhost:8080", config, libpack_logger.New(), nil)
|
|
|
|
stats := wsp.GetStats()
|
|
|
|
assert.Equal(t, false, stats["enabled"])
|
|
assert.Equal(t, int64(0), stats["active_connections"])
|
|
assert.Equal(t, int64(0), stats["total_connections"])
|
|
}
|
|
|
|
func TestWebSocketProxy_DialBackend_URLConversion(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
backendURL string
|
|
expectedURL string
|
|
}{
|
|
{
|
|
name: "http to ws",
|
|
backendURL: "http://localhost:8080",
|
|
expectedURL: "ws://localhost:8080",
|
|
},
|
|
{
|
|
name: "https to wss",
|
|
backendURL: "https://localhost:8080",
|
|
expectedURL: "wss://localhost:8080",
|
|
},
|
|
{
|
|
name: "http with path",
|
|
backendURL: "http://localhost:8080/graphql",
|
|
expectedURL: "ws://localhost:8080/graphql",
|
|
},
|
|
{
|
|
name: "https with path",
|
|
backendURL: "https://example.com/v1/graphql",
|
|
expectedURL: "wss://example.com/v1/graphql",
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
config := WebSocketConfig{Enabled: true}
|
|
wsp := NewWebSocketProxy(tt.backendURL, config, libpack_logger.New(), nil)
|
|
|
|
assert.Equal(t, tt.backendURL, wsp.backendURL)
|
|
|
|
// We can't fully test dialBackend without a real WebSocket server,
|
|
// but we can verify the URL conversion logic
|
|
ctx := context.Background()
|
|
headers := http.Header{}
|
|
_, err := wsp.dialBackend(ctx, headers)
|
|
|
|
// We expect an error since there's no server, but we verify the conversion happened
|
|
assert.Error(t, err) // Should fail to connect to non-existent server
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestWebSocketProxy_ActiveConnectionTracking(t *testing.T) {
|
|
config := WebSocketConfig{
|
|
Enabled: true,
|
|
MaxMessageSize: 512 * 1024,
|
|
}
|
|
|
|
wsp := NewWebSocketProxy("http://localhost:8080", config, libpack_logger.New(), nil)
|
|
|
|
// Simulate connection lifecycle
|
|
wsp.activeConnections.Add(1)
|
|
wsp.totalConnections.Add(1)
|
|
assert.Equal(t, int64(1), wsp.activeConnections.Load())
|
|
assert.Equal(t, int64(1), wsp.totalConnections.Load())
|
|
|
|
// Simulate more connections
|
|
wsp.activeConnections.Add(1)
|
|
wsp.totalConnections.Add(1)
|
|
assert.Equal(t, int64(2), wsp.activeConnections.Load())
|
|
assert.Equal(t, int64(2), wsp.totalConnections.Load())
|
|
|
|
// Simulate disconnect
|
|
wsp.activeConnections.Add(-1)
|
|
assert.Equal(t, int64(1), wsp.activeConnections.Load())
|
|
assert.Equal(t, int64(2), wsp.totalConnections.Load()) // Total stays the same
|
|
|
|
// Simulate another disconnect
|
|
wsp.activeConnections.Add(-1)
|
|
assert.Equal(t, int64(0), wsp.activeConnections.Load())
|
|
assert.Equal(t, int64(2), wsp.totalConnections.Load())
|
|
}
|
|
|
|
func TestWebSocketProxy_MessageTracking(t *testing.T) {
|
|
config := WebSocketConfig{
|
|
Enabled: true,
|
|
}
|
|
|
|
wsp := NewWebSocketProxy("http://localhost:8080", config, libpack_logger.New(), nil)
|
|
|
|
// Simulate messages
|
|
wsp.messagesSent.Add(10)
|
|
wsp.messagesReceived.Add(20)
|
|
wsp.errors.Add(2)
|
|
|
|
assert.Equal(t, int64(10), wsp.messagesSent.Load())
|
|
assert.Equal(t, int64(20), wsp.messagesReceived.Load())
|
|
assert.Equal(t, int64(2), wsp.errors.Load())
|
|
|
|
stats := wsp.GetStats()
|
|
assert.Equal(t, int64(10), stats["messages_sent"])
|
|
assert.Equal(t, int64(20), stats["messages_received"])
|
|
assert.Equal(t, int64(2), stats["errors"])
|
|
}
|
|
|
|
func TestWebSocketProxy_ConcurrentStats(t *testing.T) {
|
|
config := WebSocketConfig{
|
|
Enabled: true,
|
|
}
|
|
|
|
wsp := NewWebSocketProxy("http://localhost:8080", config, libpack_logger.New(), nil)
|
|
|
|
// Concurrent updates
|
|
done := make(chan bool)
|
|
goroutines := 100
|
|
|
|
for i := 0; i < goroutines; i++ {
|
|
go func() {
|
|
wsp.messagesSent.Add(1)
|
|
wsp.messagesReceived.Add(1)
|
|
wsp.errors.Add(1)
|
|
done <- true
|
|
}()
|
|
}
|
|
|
|
// Wait for all goroutines
|
|
for i := 0; i < goroutines; i++ {
|
|
<-done
|
|
}
|
|
|
|
assert.Equal(t, int64(goroutines), wsp.messagesSent.Load())
|
|
assert.Equal(t, int64(goroutines), wsp.messagesReceived.Load())
|
|
assert.Equal(t, int64(goroutines), wsp.errors.Load())
|
|
}
|
|
|
|
func TestWebSocketProxy_GlobalInstance(t *testing.T) {
|
|
config := WebSocketConfig{
|
|
Enabled: true,
|
|
PingInterval: 30 * time.Second,
|
|
MaxMessageSize: 512 * 1024,
|
|
}
|
|
|
|
wsp := InitializeWebSocketProxy("http://localhost:8080", config, libpack_logger.New(), nil)
|
|
assert.NotNil(t, wsp)
|
|
|
|
// Should return the same instance
|
|
wsp2 := GetWebSocketProxy()
|
|
assert.Equal(t, wsp, wsp2)
|
|
}
|
|
|
|
func TestWebSocketProxy_ConfigValidation(t *testing.T) {
|
|
t.Run("ping interval defaults", func(t *testing.T) {
|
|
config := WebSocketConfig{
|
|
Enabled: true,
|
|
PingInterval: 0, // Should use default
|
|
}
|
|
|
|
wsp := NewWebSocketProxy("http://localhost:8080", config, libpack_logger.New(), nil)
|
|
assert.Equal(t, 30*time.Second, wsp.pingInterval)
|
|
})
|
|
|
|
t.Run("pong timeout defaults", func(t *testing.T) {
|
|
config := WebSocketConfig{
|
|
Enabled: true,
|
|
PongTimeout: 0, // Should use default
|
|
}
|
|
|
|
wsp := NewWebSocketProxy("http://localhost:8080", config, libpack_logger.New(), nil)
|
|
assert.Equal(t, 60*time.Second, wsp.pongTimeout)
|
|
})
|
|
|
|
t.Run("max message size defaults", func(t *testing.T) {
|
|
config := WebSocketConfig{
|
|
Enabled: true,
|
|
MaxMessageSize: 0, // Should use default
|
|
}
|
|
|
|
wsp := NewWebSocketProxy("http://localhost:8080", config, libpack_logger.New(), nil)
|
|
assert.Equal(t, int64(512*1024), wsp.maxMessageSize)
|
|
})
|
|
}
|
|
|
|
func TestWebSocketProxy_ReapsSilentBackend_Keepalive(t *testing.T) {
|
|
config := WebSocketConfig{
|
|
Enabled: true,
|
|
PingInterval: 30 * time.Second,
|
|
PongTimeout: 150 * time.Millisecond,
|
|
}
|
|
wsp := NewWebSocketProxy("ws://placeholder", config, libpack_logger.New(), nil)
|
|
|
|
// Real backend that completes the WebSocket upgrade then stays silent (TCP
|
|
// open, no frames) - simulates a half-open peer whose connection the
|
|
// keepalive read deadline must reap instead of hanging forever.
|
|
serverRelease := make(chan struct{})
|
|
upgrader := gorillaws.Upgrader{}
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
conn, err := upgrader.Upgrade(w, r, nil)
|
|
if err != nil {
|
|
return
|
|
}
|
|
<-serverRelease
|
|
_ = conn.Close()
|
|
}))
|
|
defer srv.Close()
|
|
defer close(serverRelease)
|
|
|
|
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http")
|
|
backend, _, err := gorillaws.DefaultDialer.Dial(wsURL, nil)
|
|
if err != nil {
|
|
t.Fatalf("dial backend: %v", err)
|
|
}
|
|
defer backend.Close()
|
|
|
|
// Mirror handleConnection's keepalive: refresh the read deadline on the
|
|
// configured pong timeout so an idle backend is reaped.
|
|
backend.SetReadDeadline(time.Now().Add(config.PongTimeout))
|
|
|
|
done := make(chan struct{})
|
|
go func() {
|
|
// client conn is nil and safe here: the silent backend sends no
|
|
// message, so proxyBackendToClient returns on the read deadline
|
|
// before any client.WriteMessage call.
|
|
wsp.proxyBackendToClient(context.Background(), backend, nil, "test-conn")
|
|
close(done)
|
|
}()
|
|
|
|
select {
|
|
case <-done:
|
|
// Reaped via the keepalive read deadline rather than blocking forever.
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("proxyBackendToClient hung; keepalive read deadline did not reap silent backend")
|
|
}
|
|
}
|
|
|
|
func TestWebSocketProxy_StatsStructure(t *testing.T) {
|
|
config := WebSocketConfig{
|
|
Enabled: true,
|
|
PingInterval: 15 * time.Second,
|
|
PongTimeout: 30 * time.Second,
|
|
MaxMessageSize: 1024 * 1024,
|
|
}
|
|
|
|
wsp := NewWebSocketProxy("http://localhost:8080", config, libpack_logger.New(), nil)
|
|
|
|
stats := wsp.GetStats()
|
|
|
|
// Verify all expected fields are present
|
|
_, hasEnabled := stats["enabled"]
|
|
_, hasActive := stats["active_connections"]
|
|
_, hasTotal := stats["total_connections"]
|
|
_, hasSent := stats["messages_sent"]
|
|
_, hasReceived := stats["messages_received"]
|
|
_, hasErrors := stats["errors"]
|
|
_, hasPing := stats["ping_interval"]
|
|
_, hasPong := stats["pong_timeout"]
|
|
_, hasSize := stats["max_message_size"]
|
|
|
|
assert.True(t, hasEnabled)
|
|
assert.True(t, hasActive)
|
|
assert.True(t, hasTotal)
|
|
assert.True(t, hasSent)
|
|
assert.True(t, hasReceived)
|
|
assert.True(t, hasErrors)
|
|
assert.True(t, hasPing)
|
|
assert.True(t, hasPong)
|
|
assert.True(t, hasSize)
|
|
}
|