From 88f4e135d003b37ef8d4ba7fc55285e3c76892fc Mon Sep 17 00:00:00 2001 From: Thenujan Date: Fri, 7 Aug 2026 10:28:09 +0530 Subject: [PATCH 01/10] Add Redis client utility and corresponding tests --- sdk/core/go.mod | 12 ++ sdk/core/go.sum | 26 ++++ sdk/core/utils/redisclient/redisclient.go | 93 ++++++++++++ .../utils/redisclient/redisclient_test.go | 136 ++++++++++++++++++ 4 files changed, 267 insertions(+) create mode 100644 sdk/core/go.sum create mode 100644 sdk/core/utils/redisclient/redisclient.go create mode 100644 sdk/core/utils/redisclient/redisclient_test.go diff --git a/sdk/core/go.mod b/sdk/core/go.mod index fa1ad4c5b9..b43acbabf8 100644 --- a/sdk/core/go.mod +++ b/sdk/core/go.mod @@ -1,3 +1,15 @@ module github.com/wso2/api-platform/sdk/core go 1.26.2 + +require ( + github.com/alicebob/miniredis/v2 v2.38.0 + github.com/redis/go-redis/v9 v9.22.0 +) + +require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/yuin/gopher-lua v1.1.1 // indirect + go.uber.org/atomic v1.11.0 // indirect + golang.org/x/sys v0.30.0 // indirect +) diff --git a/sdk/core/go.sum b/sdk/core/go.sum new file mode 100644 index 0000000000..e9ba59b600 --- /dev/null +++ b/sdk/core/go.sum @@ -0,0 +1,26 @@ +github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= +github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0= +github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4= +github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= +golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= diff --git a/sdk/core/utils/redisclient/redisclient.go b/sdk/core/utils/redisclient/redisclient.go new file mode 100644 index 0000000000..52fec8a28d --- /dev/null +++ b/sdk/core/utils/redisclient/redisclient.go @@ -0,0 +1,93 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * + * 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 redisclient shares one process-wide *redis.Client (one connection +// pool) per distinct connection configuration, across every Redis-using +// policy that imports it - see GetOrCreateRedisClient. +package redisclient + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "sync" + "time" + + "github.com/redis/go-redis/v9" +) + +// redisConnKey identifies a distinct Redis connection configuration. Two policy +// instances with identical connection settings share one *redis.Client (one pool). +type redisConnKey struct { + addr string + username string + passwordHash string // sha256 hex; keeps the secret out of the in-process map key + db int + dialTimeout time.Duration + readTimeout time.Duration + writeTimeout time.Duration + poolSize int +} + +// redisClients is the process-wide registry of shared Redis clients. Without it, +// GetPolicy creates a new *redis.Client (a whole connection pool) per policy instance +// and per config reload, leaking pools and exploding Redis connections at scale. +var redisClients = struct { + mu sync.Mutex + m map[redisConnKey]*redis.Client +}{m: make(map[redisConnKey]*redis.Client)} + +func hashRedisPassword(p string) string { + if p == "" { + return "" + } + sum := sha256.Sum256([]byte(p)) + return hex.EncodeToString(sum[:]) +} + +// GetOrCreateRedisClient returns the process-wide shared client for these connection +// settings, creating (and pinging once) it on first use. created reports whether this +// call created the client; pingErr is non-nil only when created and the initial ping +// failed. The client is registered and returned even on ping failure (go-redis +// reconnects lazily). Clients are never closed — they live for the process lifetime. +func GetOrCreateRedisClient(opts *redis.Options, pingTimeout time.Duration) (client *redis.Client, created bool, pingErr error) { + key := redisConnKey{ + addr: opts.Addr, + username: opts.Username, + passwordHash: hashRedisPassword(opts.Password), + db: opts.DB, + dialTimeout: opts.DialTimeout, + readTimeout: opts.ReadTimeout, + writeTimeout: opts.WriteTimeout, + poolSize: opts.PoolSize, + } + + redisClients.mu.Lock() + defer redisClients.mu.Unlock() + + if c, ok := redisClients.m[key]; ok { + return c, false, nil + } + + c := redis.NewClient(opts) + redisClients.m[key] = c + + ctx, cancel := context.WithTimeout(context.Background(), pingTimeout) + defer cancel() + pingErr = c.Ping(ctx).Err() + return c, true, pingErr +} diff --git a/sdk/core/utils/redisclient/redisclient_test.go b/sdk/core/utils/redisclient/redisclient_test.go new file mode 100644 index 0000000000..c2766b25d1 --- /dev/null +++ b/sdk/core/utils/redisclient/redisclient_test.go @@ -0,0 +1,136 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com). + * + * WSO2 LLC. licenses this file to you 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 redisclient + +import ( + "context" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" +) + +func TestGetOrCreateClient_SharesClientForIdenticalConfig(t *testing.T) { + mr := miniredis.RunT(t) + opts := &redis.Options{Addr: mr.Addr(), DB: 0} + + c1, created1, err1 := GetOrCreateRedisClient(opts, time.Second) + if !created1 || err1 != nil { + t.Fatalf("first call: created=%v err=%v (want true,nil)", created1, err1) + } + + c2, created2, err2 := GetOrCreateRedisClient(opts, time.Second) + if created2 || err2 != nil { + t.Fatalf("second call: created=%v err=%v (want false,nil)", created2, err2) + } + if c1 != c2 { + t.Error("expected identical connection settings to share one *redis.Client") + } +} + +func TestGetOrCreateClient_DistinctClientForDifferentConfig(t *testing.T) { + mr := miniredis.RunT(t) + + c1, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), DB: 0}, time.Second) + c2, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), DB: 1}, time.Second) + + if c1 == c2 { + t.Error("expected different DB selection to produce a distinct *redis.Client") + } +} + +func TestGetOrCreateClient_DifferentPasswordProducesDistinctClient(t *testing.T) { + mr := miniredis.RunT(t) + + c1, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), Password: "one"}, time.Second) + c2, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), Password: "two"}, time.Second) + c3, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr()}, time.Second) // no password at all + + if c1 == c2 { + t.Error("expected different passwords to produce distinct clients") + } + if c1 == c3 || c2 == c3 { + t.Error("expected an absent password not to collide with a present one") + } +} + +func TestGetOrCreateClient_SharedAcrossSimulatedPolicies(t *testing.T) { + mr := miniredis.RunT(t) + opts := func() *redis.Options { return &redis.Options{Addr: mr.Addr(), DB: 0} } + + // Two "different policies" (distinct call sites, distinct *redis.Options + // values) with identical connection settings must still land on the + // same underlying client - this is the whole point of centralizing the + // registry here instead of each policy keeping its own. + fromPolicyA, _, _ := GetOrCreateRedisClient(opts(), time.Second) + fromPolicyB, _, _ := GetOrCreateRedisClient(opts(), time.Second) + + if fromPolicyA != fromPolicyB { + t.Fatal("expected two distinct callers with identical config to share one client") + } + + ctx := context.Background() + if err := fromPolicyA.Set(ctx, "shared-key", "value", 0).Err(); err != nil { + t.Fatalf("unexpected error writing via the shared client: %v", err) + } + got, err := fromPolicyB.Get(ctx, "shared-key").Result() + if err != nil { + t.Fatalf("unexpected error reading via the shared client: %v", err) + } + if got != "value" { + t.Errorf("got %q, want %q", got, "value") + } +} + +// TestGetOrCreateClient_ReuseSkipsPing locks in that only CREATION pings - +// a reused client is assumed healthy (go-redis reconnects lazily) and must +// never be re-pinged, or a client that legitimately reused a pool would +// spuriously start reporting errors the moment Redis blips after creation. +func TestGetOrCreateClient_ReuseSkipsPing(t *testing.T) { + mr := miniredis.RunT(t) + addr := mr.Addr() // capture before mr.Close() below + opts := &redis.Options{Addr: addr, DB: 0} + + c1, created1, err1 := GetOrCreateRedisClient(opts, time.Second) + if !created1 || err1 != nil { + t.Fatalf("first call: created=%v err=%v (want true,nil)", created1, err1) + } + + mr.Close() + c2, created2, err2 := GetOrCreateRedisClient(opts, time.Second) + if created2 || err2 != nil || c2 != c1 { + t.Fatalf("reuse after Redis went down should skip the ping: created=%v err=%v same=%v", created2, err2, c2 == c1) + } +} + +func TestHashPassword(t *testing.T) { + if hashRedisPassword("") != "" { + t.Error("expected an empty password to hash to empty, not sha256(\"\")") + } + if hashRedisPassword("secret") == "secret" { + t.Error("expected the password to actually be hashed, not passed through") + } + if hashRedisPassword("secret") != hashRedisPassword("secret") { + t.Error("expected hashing to be deterministic") + } + if hashRedisPassword("secret") == hashRedisPassword("different") { + t.Error("expected different passwords to hash differently") + } +} From d419ca091a40d3255b5f1d484f49450941351329 Mon Sep 17 00:00:00 2001 From: Thenujan Date: Fri, 7 Aug 2026 11:31:24 +0530 Subject: [PATCH 02/10] Enhance Redis client to support distinct protocols and improve connection handling with TLS and credentials provider --- sdk/core/utils/redisclient/redisclient.go | 28 +++- .../utils/redisclient/redisclient_test.go | 125 ++++++++++++++++-- 2 files changed, 140 insertions(+), 13 deletions(-) diff --git a/sdk/core/utils/redisclient/redisclient.go b/sdk/core/utils/redisclient/redisclient.go index 52fec8a28d..7d3d90e7cc 100644 --- a/sdk/core/utils/redisclient/redisclient.go +++ b/sdk/core/utils/redisclient/redisclient.go @@ -32,11 +32,15 @@ import ( // redisConnKey identifies a distinct Redis connection configuration. Two policy // instances with identical connection settings share one *redis.Client (one pool). +// +// Excludes TLSConfig and any credentials-provider option - see +// GetOrCreateRedisClient's bypass for those. type redisConnKey struct { addr string username string passwordHash string // sha256 hex; keeps the secret out of the in-process map key db int + protocol int dialTimeout time.Duration readTimeout time.Duration writeTimeout time.Duration @@ -65,26 +69,44 @@ func hashRedisPassword(p string) string { // failed. The client is registered and returned even on ping failure (go-redis // reconnects lazily). Clients are never closed — they live for the process lifetime. func GetOrCreateRedisClient(opts *redis.Options, pingTimeout time.Duration) (client *redis.Client, created bool, pingErr error) { + // TLSConfig and credentials-provider hooks can't be fingerprinted + // safely: a *tls.Config's pointer says nothing about its content, and + // Go func values aren't comparable at all. Bypass the registry rather + // than risk silently reusing a client built for a different config. + if opts.TLSConfig != nil || opts.CredentialsProvider != nil || opts.CredentialsProviderContext != nil || opts.StreamingCredentialsProvider != nil { + c := redis.NewClient(opts) + ctx, cancel := context.WithTimeout(context.Background(), pingTimeout) + defer cancel() + pingErr = c.Ping(ctx).Err() + return c, true, pingErr + } + key := redisConnKey{ addr: opts.Addr, username: opts.Username, passwordHash: hashRedisPassword(opts.Password), db: opts.DB, + protocol: opts.Protocol, dialTimeout: opts.DialTimeout, readTimeout: opts.ReadTimeout, writeTimeout: opts.WriteTimeout, poolSize: opts.PoolSize, } + // Lock guards only the map lookup/insert, never the ping below - mu is + // process-wide, so holding it during a slow/down connection's ping + // would stall every other caller's get-or-create too. A concurrent + // caller for the same key may see the just-inserted client before this + // ping finishes - fine, since a reused client is already "assumed + // healthy" regardless of timing, never gated on this call's pingErr. redisClients.mu.Lock() - defer redisClients.mu.Unlock() - if c, ok := redisClients.m[key]; ok { + redisClients.mu.Unlock() return c, false, nil } - c := redis.NewClient(opts) redisClients.m[key] = c + redisClients.mu.Unlock() ctx, cancel := context.WithTimeout(context.Background(), pingTimeout) defer cancel() diff --git a/sdk/core/utils/redisclient/redisclient_test.go b/sdk/core/utils/redisclient/redisclient_test.go index c2766b25d1..ba0126cf36 100644 --- a/sdk/core/utils/redisclient/redisclient_test.go +++ b/sdk/core/utils/redisclient/redisclient_test.go @@ -20,6 +20,8 @@ package redisclient import ( "context" + "crypto/tls" + "net" "testing" "time" @@ -75,10 +77,8 @@ func TestGetOrCreateClient_SharedAcrossSimulatedPolicies(t *testing.T) { mr := miniredis.RunT(t) opts := func() *redis.Options { return &redis.Options{Addr: mr.Addr(), DB: 0} } - // Two "different policies" (distinct call sites, distinct *redis.Options - // values) with identical connection settings must still land on the - // same underlying client - this is the whole point of centralizing the - // registry here instead of each policy keeping its own. + // Two distinct call sites with identical settings must share one + // client - the whole point of centralizing the registry. fromPolicyA, _, _ := GetOrCreateRedisClient(opts(), time.Second) fromPolicyB, _, _ := GetOrCreateRedisClient(opts(), time.Second) @@ -99,10 +99,8 @@ func TestGetOrCreateClient_SharedAcrossSimulatedPolicies(t *testing.T) { } } -// TestGetOrCreateClient_ReuseSkipsPing locks in that only CREATION pings - -// a reused client is assumed healthy (go-redis reconnects lazily) and must -// never be re-pinged, or a client that legitimately reused a pool would -// spuriously start reporting errors the moment Redis blips after creation. +// TestGetOrCreateClient_ReuseSkipsPing locks in that only creation pings - +// a reused client is assumed healthy and must never be re-pinged. func TestGetOrCreateClient_ReuseSkipsPing(t *testing.T) { mr := miniredis.RunT(t) addr := mr.Addr() // capture before mr.Close() below @@ -120,6 +118,112 @@ func TestGetOrCreateClient_ReuseSkipsPing(t *testing.T) { } } +func TestGetOrCreateClient_DifferentProtocolProducesDistinctClient(t *testing.T) { + mr := miniredis.RunT(t) + + c1, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), Protocol: 2}, time.Second) + c2, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), Protocol: 3}, time.Second) + c3, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), Protocol: 2}, time.Second) + + if c1 == c2 { + t.Error("expected different RESP protocol versions to produce distinct clients") + } + if c1 != c3 { + t.Error("expected the same protocol version to reuse the existing client") + } +} + +// TestGetOrCreateClient_TLSConfigBypassesRegistry locks in that a TLSConfig +// always gets a fresh, unshared client, even with otherwise-identical +// options - neither it nor a credentials-provider func can be fingerprinted +// safely, so sharing would risk a silent cross-config mixup. +func TestGetOrCreateClient_TLSConfigBypassesRegistry(t *testing.T) { + mr := miniredis.RunT(t) + + optsA := &redis.Options{Addr: mr.Addr(), TLSConfig: &tls.Config{}} //nolint:gosec // test-only, no real handshake asserted + optsB := &redis.Options{Addr: mr.Addr(), TLSConfig: &tls.Config{}} //nolint:gosec + + c1, created1, _ := GetOrCreateRedisClient(optsA, time.Second) + c2, created2, _ := GetOrCreateRedisClient(optsB, time.Second) + + if !created1 || !created2 { + t.Fatalf("expected every TLSConfig-bearing call to report created=true (never reused), got %v and %v", created1, created2) + } + if c1 == c2 { + t.Error("expected two TLSConfig-bearing calls to never share a client, even with identical-looking options") + } +} + +func TestGetOrCreateClient_CredentialsProviderBypassesRegistry(t *testing.T) { + mr := miniredis.RunT(t) + provider := func() (string, string) { return "", "" } + + c1, created1, err1 := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), CredentialsProvider: provider}, time.Second) + c2, created2, err2 := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), CredentialsProvider: provider}, time.Second) + + if !created1 || err1 != nil { + t.Fatalf("first call: created=%v err=%v (want true,nil)", created1, err1) + } + if !created2 || err2 != nil { + t.Fatalf("second call: created=%v err=%v (want true,nil - bypassed, not reused)", created2, err2) + } + if c1 == c2 { + t.Error("expected two CredentialsProvider-bearing calls to never share a client") + } +} + +// TestGetOrCreateClient_DoesNotHoldLockDuringPing proves the registry lock +// guards only the map lookup/insert, never c.Ping - mu is process-wide, so +// holding it during a slow/unreachable Redis's ping would stall every other +// caller too, even for an unrelated, healthy endpoint. +func TestGetOrCreateClient_DoesNotHoldLockDuringPing(t *testing.T) { + // Accepts but never responds, so Ping against it blocks until the + // deadline - a reliable window to prove a concurrent, unrelated key + // isn't blocked by it. + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("failed to start hanging listener: %v", err) + } + defer ln.Close() + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + _ = conn // held open, never responded to + } + }() + + done := make(chan struct{}) + go func() { + defer close(done) + // ReadTimeout set explicitly - the dial succeeds, it's the + // read-for-a-reply that hangs, and go-redis's default (5s) would + // otherwise bound that wait regardless of pingTimeout. + GetOrCreateRedisClient(&redis.Options{ + Addr: ln.Addr().String(), + DB: 0, + ReadTimeout: time.Second, + WriteTimeout: time.Second, + }, time.Second) + }() + + // Generous margin for the slow call to insert+unlock and enter Ping. + time.Sleep(100 * time.Millisecond) + + mr := miniredis.RunT(t) + fastStart := time.Now() + if _, _, err := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), DB: 1}, 500*time.Millisecond); err != nil { + t.Fatalf("unexpected error on the fast, unrelated key: %v", err) + } + if elapsed := time.Since(fastStart); elapsed > 300*time.Millisecond { + t.Errorf("expected the unrelated key's get-or-create to complete quickly (the registry lock must not be held during the other call's ping), took %s", elapsed) + } + + <-done // let the slow goroutine finish before the test exits +} + func TestHashPassword(t *testing.T) { if hashRedisPassword("") != "" { t.Error("expected an empty password to hash to empty, not sha256(\"\")") @@ -127,8 +231,9 @@ func TestHashPassword(t *testing.T) { if hashRedisPassword("secret") == "secret" { t.Error("expected the password to actually be hashed, not passed through") } - if hashRedisPassword("secret") != hashRedisPassword("secret") { - t.Error("expected hashing to be deterministic") + const wantSecretSHA256 = "2bb80d537b1da3e38bd30361aa855686bde0eacd7162fef6a25fe97bf527a25b" + if got := hashRedisPassword("secret"); got != wantSecretSHA256 { + t.Errorf("hashRedisPassword(%q) = %q, want %q", "secret", got, wantSecretSHA256) } if hashRedisPassword("secret") == hashRedisPassword("different") { t.Error("expected different passwords to hash differently") From 961f77a8eff529d80241dbfaad554a15fbf64c6f Mon Sep 17 00:00:00 2001 From: Thenujan Date: Fri, 7 Aug 2026 14:06:15 +0530 Subject: [PATCH 03/10] Improve test for Redis client connection handling during ping to ensure accurate timeout signaling --- .../utils/redisclient/redisclient_test.go | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/sdk/core/utils/redisclient/redisclient_test.go b/sdk/core/utils/redisclient/redisclient_test.go index ba0126cf36..6b82487c3f 100644 --- a/sdk/core/utils/redisclient/redisclient_test.go +++ b/sdk/core/utils/redisclient/redisclient_test.go @@ -22,6 +22,7 @@ import ( "context" "crypto/tls" "net" + "sync" "testing" "time" @@ -184,13 +185,20 @@ func TestGetOrCreateClient_DoesNotHoldLockDuringPing(t *testing.T) { if err != nil { t.Fatalf("failed to start hanging listener: %v", err) } - defer ln.Close() + defer func() { _ = ln.Close() }() + + // Signaled once the slow client's connection is actually accepted - + // proof it has dialed and is now blocked reading the Ping reply, rather + // than guessing via a fixed sleep how long that takes to happen. + accepted := make(chan struct{}) + var acceptedOnce sync.Once go func() { for { conn, err := ln.Accept() if err != nil { return } + acceptedOnce.Do(func() { close(accepted) }) _ = conn // held open, never responded to } }() @@ -201,7 +209,7 @@ func TestGetOrCreateClient_DoesNotHoldLockDuringPing(t *testing.T) { // ReadTimeout set explicitly - the dial succeeds, it's the // read-for-a-reply that hangs, and go-redis's default (5s) would // otherwise bound that wait regardless of pingTimeout. - GetOrCreateRedisClient(&redis.Options{ + _, _, _ = GetOrCreateRedisClient(&redis.Options{ Addr: ln.Addr().String(), DB: 0, ReadTimeout: time.Second, @@ -209,8 +217,11 @@ func TestGetOrCreateClient_DoesNotHoldLockDuringPing(t *testing.T) { }, time.Second) }() - // Generous margin for the slow call to insert+unlock and enter Ping. - time.Sleep(100 * time.Millisecond) + select { + case <-accepted: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the slow client's connection to be accepted") + } mr := miniredis.RunT(t) fastStart := time.Now() From 7f08d9bf24a7e495eda868110ee02af684af3346 Mon Sep 17 00:00:00 2001 From: Thenujan Date: Fri, 7 Aug 2026 14:10:29 +0530 Subject: [PATCH 04/10] Update license header --- sdk/core/utils/redisclient/redisclient.go | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/sdk/core/utils/redisclient/redisclient.go b/sdk/core/utils/redisclient/redisclient.go index 7d3d90e7cc..a7e91724b7 100644 --- a/sdk/core/utils/redisclient/redisclient.go +++ b/sdk/core/utils/redisclient/redisclient.go @@ -1,18 +1,19 @@ /* - * Copyright (c) 2026, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com). * - * 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 + * WSO2 LLC. licenses this file to you 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. + * 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 redisclient shares one process-wide *redis.Client (one connection From e29f6ace4b966fa186c2425586070fbd723f68d6 Mon Sep 17 00:00:00 2001 From: Thenujan Date: Tue, 11 Aug 2026 23:11:30 +0530 Subject: [PATCH 05/10] Implement shared Redis client initialization and configuration handling --- .../policy-engine/cmd/policy-engine/main.go | 14 + gateway/gateway-runtime/policy-engine/go.mod | 4 + gateway/gateway-runtime/policy-engine/go.sum | 18 +- sdk/core/utils/redisclient/redisclient.go | 256 +++++++++++++++++- .../utils/redisclient/redisclient_test.go | 256 ++++++++++++++++++ sdk/core/utils/redisclient/testing.go | 43 +++ 6 files changed, 580 insertions(+), 11 deletions(-) create mode 100644 sdk/core/utils/redisclient/testing.go diff --git a/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go b/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go index 9978f14ae8..3a82b80460 100644 --- a/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go +++ b/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go @@ -47,6 +47,7 @@ import ( "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/tracing" "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/utils" "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/xdsclient" + "github.com/wso2/api-platform/sdk/core/utils/redisclient" ) // Version information (set via ldflags during build) @@ -164,6 +165,19 @@ func main() { } slog.InfoContext(ctx, "Config set in registry for ${config} CEL resolution") + // Initialize the gateway-level shared Redis client (top-level "redis" config + // section - gateway infrastructure, not nested under policy_configurations, + // since it's not scoped to policies even though policy-engine is its current + // consumer). Must run before any policy chain is built (below) - a policy + // instance that calls redisclient.Shared()/Resolve() during construction + // assumes this has already run. A missing "redis" section is not an error + // here - it's only surfaced lazily, the first time some policy actually + // needs it. + if err := redisclient.InitFromConfig(cfg.PolicyEngine.RawConfig); err != nil { + slog.ErrorContext(ctx, "Failed to initialize shared redis client", "error", err) + os.Exit(1) + } + // Initialize CEL evaluator celEvaluator, err := cel.NewCELEvaluator() if err != nil { diff --git a/gateway/gateway-runtime/policy-engine/go.mod b/gateway/gateway-runtime/policy-engine/go.mod index e4914e8ec3..551e25171b 100644 --- a/gateway/gateway-runtime/policy-engine/go.mod +++ b/gateway/gateway-runtime/policy-engine/go.mod @@ -51,10 +51,12 @@ require ( github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.66.1 // indirect github.com/prometheus/procfs v0.19.2 // indirect + github.com/redis/go-redis/v9 v9.22.0 // indirect github.com/stoewer/go-strcase v1.3.1 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect + go.uber.org/atomic v1.11.0 // indirect go.yaml.in/yaml/v2 v2.4.3 // indirect golang.org/x/exp v0.0.0-20260112195511-716be5621a96 // indirect golang.org/x/net v0.56.0 // indirect @@ -65,3 +67,5 @@ require ( ) replace github.com/wso2/api-platform/common => ../../../common + +replace github.com/wso2/api-platform/sdk/core => ../../../sdk/core diff --git a/gateway/gateway-runtime/policy-engine/go.sum b/gateway/gateway-runtime/policy-engine/go.sum index 2939f20ddd..3e2a2de00c 100644 --- a/gateway/gateway-runtime/policy-engine/go.sum +++ b/gateway/gateway-runtime/policy-engine/go.sum @@ -1,11 +1,17 @@ cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4= cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= +github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= +github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ= github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ= github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= @@ -41,6 +47,8 @@ github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF2 github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs= github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/knadh/koanf/maps v0.1.2 h1:RBfmAW5CnZT+PJ1CVc1QSJKf4Xu9kxfQgYVQSu8hpbo= github.com/knadh/koanf/maps v0.1.2/go.mod h1:npD/QZY3V6ghQDdcQzl1W4ICNVTkohC8E73eI2xW4yI= github.com/knadh/koanf/parsers/toml/v2 v2.2.0 h1:2nV7tHYJ5OZy2BynQ4mOJ6k5bDqbbCzRERLUKBytz3A= @@ -80,6 +88,8 @@ github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9Z github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= github.com/prometheus/procfs v0.19.2 h1:zUMhqEW66Ex7OXIiDkll3tl9a1ZdilUOd/F6ZXw4Vws= github.com/prometheus/procfs v0.19.2/go.mod h1:M0aotyiemPhBCM0z5w87kL22CxfcH05ZpYlu+b4J7mw= +github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0= +github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/stoewer/go-strcase v1.3.1 h1:iS0MdW+kVTxgMoE1LAZyMiYJFKlOzLooE4MxjirtkAs= @@ -92,10 +102,12 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -github.com/wso2/api-platform/sdk/core v0.3.3 h1:5bBapq9tWQf/kXZfYNnofJUTwZLyZ0PZpe7skbeZd/I= -github.com/wso2/api-platform/sdk/core v0.3.3/go.mod h1:TgjpOk3QBPc7xEQC+NctWcNq5dSPBkn7wSKh+hULTj8= github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= @@ -114,6 +126,8 @@ go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/ go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0= diff --git a/sdk/core/utils/redisclient/redisclient.go b/sdk/core/utils/redisclient/redisclient.go index a7e91724b7..a5dc1eb9e7 100644 --- a/sdk/core/utils/redisclient/redisclient.go +++ b/sdk/core/utils/redisclient/redisclient.go @@ -17,14 +17,21 @@ */ // Package redisclient shares one process-wide *redis.Client (one connection -// pool) per distinct connection configuration, across every Redis-using -// policy that imports it - see GetOrCreateRedisClient. +// pool) per distinct connection configuration, across every caller that +// imports it - see GetOrCreateRedisClient. It also exposes a single +// gateway-wide default client (Shared, backed by the operator's top-level +// "redis" config section - gateway infrastructure, not something scoped to +// policies) that a policy falls back to when it has no Redis config of its +// own - see Resolve. package redisclient import ( "context" "crypto/sha256" "encoding/hex" + "fmt" + "strconv" + "strings" "sync" "time" @@ -75,11 +82,7 @@ func GetOrCreateRedisClient(opts *redis.Options, pingTimeout time.Duration) (cli // Go func values aren't comparable at all. Bypass the registry rather // than risk silently reusing a client built for a different config. if opts.TLSConfig != nil || opts.CredentialsProvider != nil || opts.CredentialsProviderContext != nil || opts.StreamingCredentialsProvider != nil { - c := redis.NewClient(opts) - ctx, cancel := context.WithTimeout(context.Background(), pingTimeout) - defer cancel() - pingErr = c.Ping(ctx).Err() - return c, true, pingErr + return newAndPingClient(opts, pingTimeout) } key := redisConnKey{ @@ -109,8 +112,243 @@ func GetOrCreateRedisClient(opts *redis.Options, pingTimeout time.Duration) (cli redisClients.m[key] = c redisClients.mu.Unlock() + pingErr = pingClient(c, pingTimeout) + return c, true, pingErr +} + +// defaultSharedPingTimeout bounds the one-time creation ping for the shared +// client. Not configurable: it only affects how long InitFromConfig blocks +// during gateway-runtime startup, never a per-request path. +const defaultSharedPingTimeout = 5 * time.Second + +// shared holds the process-wide gateway-level default client. inited +// distinguishes "InitFromConfig ran and found no redis section" (client nil, +// inited true - Shared reports a config-gap error) from "InitFromConfig was +// never called at all" (a gateway-runtime wiring bug - Shared reports that +// distinctly, since it means something programming-level is missing, not an +// operator config gap). +var shared struct { + mu sync.Mutex + client *redis.Client + inited bool +} + +// InitFromConfig resolves the operator-level top-level "redis" section from +// raw (e.g. cfg.PolicyEngine.RawConfig - raw's other top-level sections like +// "analytics"/"router"/"policy_configurations" are ignored here) and creates +// the process-wide shared client. This is gateway-wide infrastructure, not +// something scoped to policies - deliberately NOT nested under +// "policy_configurations" (that namespace is policy-engine's own ${config...} +// CEL-resolution mechanism for per-policy system parameters; a shared +// resource other gateway components could reach doesn't belong inside it). +// Must be called exactly once, at gateway-runtime startup, before any policy +// factory runs - see Shared and Resolve. A missing "redis" key is not an +// error: most gateways may have zero Redis-consuming policies configured, +// and that absence only matters lazily, the first time some policy actually +// calls Shared. A connection/ping failure is likewise not fatal here - the +// client is still created and stored (go-redis reconnects lazily), matching +// GetOrCreateRedisClient's own create-time philosophy. +func InitFromConfig(raw map[string]interface{}) error { + shared.mu.Lock() + defer shared.mu.Unlock() + if shared.inited { + return fmt.Errorf("redisclient: InitFromConfig called more than once") + } + shared.inited = true + + opts, err := resolveOptionsFromConfig(raw) + if err != nil { + return fmt.Errorf("redisclient: invalid \"redis\" config: %w", err) + } + if opts == nil { + return nil + } + + c, _, _ := newAndPingClient(opts, defaultSharedPingTimeout) + shared.client = c + return nil +} + +// Shared returns the process-wide gateway-level default client, backed by +// the top-level "redis" config section. It errors if InitFromConfig was +// never called (a gateway-runtime wiring bug, not a normal runtime +// condition) or if no "redis" section was configured at all - callers must +// treat the latter as a real configuration gap rather than assuming a +// shared Redis is always available. +func Shared() (*redis.Client, error) { + shared.mu.Lock() + defer shared.mu.Unlock() + if !shared.inited { + return nil, fmt.Errorf("redisclient: Shared() called before InitFromConfig") + } + if shared.client == nil { + return nil, fmt.Errorf(`redisclient: no shared redis configured ("redis" section)`) + } + return shared.client, nil +} + +// Resolve returns the client a policy instance should use: opts's own +// connection settings when the policy resolved one from its own config +// section, otherwise the gateway-level Shared client. +// +// opts must be nil - never a zero-value *redis.Options - when the policy's +// own section was absent. A struct pre-filled with the policy's schema +// defaults would always look "configured," and this fallback would never +// trigger; the presence check belongs to the caller's own config-extraction +// code, on whichever field has no default (e.g. host). +func Resolve(opts *redis.Options, pingTimeout time.Duration) (*redis.Client, error) { + if opts == nil { + return Shared() + } + client, _, _ := GetOrCreateRedisClient(opts, pingTimeout) + return client, nil +} + +// newAndPingClient creates a client and pings it once. created is always +// true - only present so this matches GetOrCreateRedisClient's own return +// shape at its call sites. +func newAndPingClient(opts *redis.Options, pingTimeout time.Duration) (client *redis.Client, created bool, pingErr error) { + c := redis.NewClient(opts) + return c, true, pingClient(c, pingTimeout) +} + +// pingClient pings an already-constructed client once, bounded by +// pingTimeout. Split out from newAndPingClient because GetOrCreateRedisClient's +// main path must insert the client into the registry BEFORE pinging (so a +// concurrent caller for the same key sees it immediately), not create-then-ping +// as one atomic step. +func pingClient(c *redis.Client, pingTimeout time.Duration) error { ctx, cancel := context.WithTimeout(context.Background(), pingTimeout) defer cancel() - pingErr = c.Ping(ctx).Err() - return c, true, pingErr + return c.Ping(ctx).Err() +} + +// resolveOptionsFromConfig extracts *redis.Options from raw["redis"] - a +// top-level section, sibling to "router"/"analytics"/etc in the gateway's +// complete config tree, not nested under "policy_configurations" (this is +// gateway-wide infrastructure, not a per-policy setting) - using the +// operator-facing defaults (host "localhost", port 6379, db 0, poolSize 0 +// (go-redis default), connectionTimeout 5s, readTimeout/writeTimeout 3s). +// Returns (nil, nil) - not an error - when raw has no "redis" key at all. +func resolveOptionsFromConfig(raw map[string]interface{}) (*redis.Options, error) { + section, ok := raw["redis"] + if !ok || section == nil { + return nil, nil + } + m, ok := section.(map[string]interface{}) + if !ok { + return nil, fmt.Errorf(`"redis" must be a table, got %T`, section) + } + + connectionTimeout, err := durationParam(m, "connection_timeout", 5*time.Second) + if err != nil { + return nil, fmt.Errorf("connection_timeout: %w", err) + } + readTimeout, err := durationParam(m, "read_timeout", 3*time.Second) + if err != nil { + return nil, fmt.Errorf("read_timeout: %w", err) + } + writeTimeout, err := durationParam(m, "write_timeout", 3*time.Second) + if err != nil { + return nil, fmt.Errorf("write_timeout: %w", err) + } + port, err := intParam(m, "port", 6379) + if err != nil { + return nil, fmt.Errorf("port: %w", err) + } + db, err := intParam(m, "db", 0) + if err != nil { + return nil, fmt.Errorf("db: %w", err) + } + poolSize, err := intParam(m, "pool_size", 0) + if err != nil { + return nil, fmt.Errorf("pool_size: %w", err) + } + + host, err := stringParam(m, "host", "localhost") + if err != nil { + return nil, fmt.Errorf("host: %w", err) + } + username, err := stringParam(m, "username", "") + if err != nil { + return nil, fmt.Errorf("username: %w", err) + } + password, err := stringParam(m, "password", "") + if err != nil { + return nil, fmt.Errorf("password: %w", err) + } + + return &redis.Options{ + Addr: fmt.Sprintf("%s:%d", host, port), + Username: username, + Password: password, + DB: db, + DialTimeout: connectionTimeout, + ReadTimeout: readTimeout, + WriteTimeout: writeTimeout, + PoolSize: poolSize, + }, nil +} + +// stringParam/intParam/durationParam read key from m, applying def when the +// key is absent or nil. They error on a present-but-wrong-shaped value +// rather than silently falling back to def - a typo'd config value should +// surface at startup, not resolve to a default the operator never asked for. +func stringParam(m map[string]interface{}, key, def string) (string, error) { + v, ok := m[key] + if !ok || v == nil { + return def, nil + } + s, ok := v.(string) + if !ok { + return "", fmt.Errorf("expected a string, got %T", v) + } + return s, nil +} + +func intParam(m map[string]interface{}, key string, def int) (int, error) { + v, ok := m[key] + if !ok || v == nil { + return def, nil + } + switch n := v.(type) { + case int: + return n, nil + case int64: + return int(n), nil + case float64: + return int(n), nil + case string: + // A TOML value written as {{ env "VAR" "default" }} must be a quoted + // string literal (TOML has no unquoted template syntax) - gateway-runtime's + // config interpolation resolves the token in place but never changes the + // field's type, so a numeric config value arrives here as a numeric + // string, not an int. Reject anything that isn't actually numeric. + parsed, err := strconv.Atoi(strings.TrimSpace(n)) + if err != nil { + return 0, fmt.Errorf("expected an integer, got non-numeric string %q", n) + } + return parsed, nil + default: + return 0, fmt.Errorf("expected an integer, got %T", v) + } +} + +func durationParam(m map[string]interface{}, key string, def time.Duration) (time.Duration, error) { + v, ok := m[key] + if !ok || v == nil { + return def, nil + } + switch d := v.(type) { + case string: + parsed, err := time.ParseDuration(d) + if err != nil { + return 0, fmt.Errorf("invalid duration %q: %w", d, err) + } + return parsed, nil + case time.Duration: + return d, nil + default: + return 0, fmt.Errorf("expected a duration string, got %T", v) + } } diff --git a/sdk/core/utils/redisclient/redisclient_test.go b/sdk/core/utils/redisclient/redisclient_test.go index 6b82487c3f..7c9c23470b 100644 --- a/sdk/core/utils/redisclient/redisclient_test.go +++ b/sdk/core/utils/redisclient/redisclient_test.go @@ -22,6 +22,7 @@ import ( "context" "crypto/tls" "net" + "strconv" "sync" "testing" "time" @@ -235,6 +236,261 @@ func TestGetOrCreateClient_DoesNotHoldLockDuringPing(t *testing.T) { <-done // let the slow goroutine finish before the test exits } +// resetSharedForTest clears the package-level shared client state before a +// test runs (so InitFromConfig can be called again despite its once-only +// guard) and restores whatever was there before once the test ends. White-box +// access is fine here - this file is part of the package. +func resetSharedForTest(t *testing.T) { + t.Helper() + shared.mu.Lock() + prevClient, prevInited := shared.client, shared.inited + shared.client, shared.inited = nil, false + shared.mu.Unlock() + t.Cleanup(func() { + shared.mu.Lock() + shared.client, shared.inited = prevClient, prevInited + shared.mu.Unlock() + }) +} + +func TestResolveOptionsFromConfig_NoRedisSectionReturnsNil(t *testing.T) { + opts, err := resolveOptionsFromConfig(map[string]interface{}{}) + if err != nil || opts != nil { + t.Fatalf("got opts=%v err=%v, want nil,nil when \"redis\" is absent entirely", opts, err) + } +} + +// TestResolveOptionsFromConfig_IgnoresSiblingSections proves resolveOptionsFromConfig +// looks at the top-level "redis" key only - other top-level sections (including +// policy_configurations, which is a separate, policy-engine-internal namespace +// this package deliberately does NOT nest under) have no bearing on it. +func TestResolveOptionsFromConfig_IgnoresSiblingSections(t *testing.T) { + raw := map[string]interface{}{ + "router": map[string]interface{}{"gateway_host": "*"}, + "policy_configurations": map[string]interface{}{"oauth2_generator_v1": map[string]interface{}{"redis": map[string]interface{}{"key_prefix": "x:"}}}, + } + opts, err := resolveOptionsFromConfig(raw) + if err != nil || opts != nil { + t.Fatalf("got opts=%v err=%v, want nil,nil when top-level \"redis\" is absent (unrelated sibling sections present)", opts, err) + } +} + +func TestResolveOptionsFromConfig_AppliesDefaults(t *testing.T) { + opts, err := resolveOptionsFromConfig(map[string]interface{}{"redis": map[string]interface{}{}}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + want := &redis.Options{ + Addr: "localhost:6379", + DialTimeout: 5 * time.Second, + ReadTimeout: 3 * time.Second, + WriteTimeout: 3 * time.Second, + } + if opts.Addr != want.Addr || opts.DialTimeout != want.DialTimeout || + opts.ReadTimeout != want.ReadTimeout || opts.WriteTimeout != want.WriteTimeout || + opts.Username != "" || opts.Password != "" || opts.DB != 0 || opts.PoolSize != 0 { + t.Errorf("got %+v, want defaults %+v (username/password/db/poolSize zero-valued)", opts, want) + } +} + +// TestResolveOptionsFromConfig_ParsesConfiguredValues covers both the string +// (typical koanf/TOML decode) and numeric (int64/float64 - decoder-dependent) +// shapes a value might arrive in. +func TestResolveOptionsFromConfig_ParsesConfiguredValues(t *testing.T) { + raw := map[string]interface{}{ + "redis": map[string]interface{}{ + "host": "redis.example.com", + "port": int64(6380), + "username": "app", + "password": "secret", + "db": float64(2), + "connection_timeout": "10s", + "read_timeout": "7s", + "write_timeout": "7s", + "pool_size": 20, + }, + } + opts, err := resolveOptionsFromConfig(raw) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if opts.Addr != "redis.example.com:6380" || opts.Username != "app" || opts.Password != "secret" || + opts.DB != 2 || opts.DialTimeout != 10*time.Second || opts.ReadTimeout != 7*time.Second || + opts.WriteTimeout != 7*time.Second || opts.PoolSize != 20 { + t.Errorf("got %+v, did not match configured values", opts) + } +} + +// TestResolveOptionsFromConfig_ParsesNumericStringPort locks in the shape +// gateway-runtime's own config interpolation actually produces: a TOML value +// written as {{ env "VAR" "6379" }} must be a quoted string literal (TOML has +// no unquoted template syntax), and interpolation resolves the token in place +// without ever changing the field's type - so a "numeric" config.toml value +// arrives here as a numeric string, not an int, even though int/int64/float64 +// are also accepted (e.g. from a JSON-sourced config path). +func TestResolveOptionsFromConfig_ParsesNumericStringPort(t *testing.T) { + raw := map[string]interface{}{ + "redis": map[string]interface{}{ + "host": "redis.example.com", + "port": "6380", + "db": "2", + }, + } + opts, err := resolveOptionsFromConfig(raw) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if opts.Addr != "redis.example.com:6380" || opts.DB != 2 { + t.Errorf("got %+v, want port 6380 and db 2 parsed from numeric strings", opts) + } +} + +func TestResolveOptionsFromConfig_RejectsWrongShapedValue(t *testing.T) { + _, err := resolveOptionsFromConfig(map[string]interface{}{ + "redis": map[string]interface{}{"port": "not-a-number"}, + }) + if err == nil { + t.Error("expected an error for a non-numeric port, so a config typo surfaces at startup instead of silently defaulting") + } +} + +func TestResolveOptionsFromConfig_RejectsNonTableSection(t *testing.T) { + _, err := resolveOptionsFromConfig(map[string]interface{}{"redis": "not-a-table"}) + if err == nil { + t.Error("expected an error when \"redis\" isn't a table") + } +} + +func TestInitFromConfig_NoRedisSectionLeavesSharedUnconfigured(t *testing.T) { + resetSharedForTest(t) + + if err := InitFromConfig(map[string]interface{}{}); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if _, err := Shared(); err == nil { + t.Error("expected Shared() to report a config-gap error when no redis section was ever configured") + } +} + +func TestInitFromConfig_CalledTwiceErrors(t *testing.T) { + resetSharedForTest(t) + + if err := InitFromConfig(map[string]interface{}{}); err != nil { + t.Fatalf("first call: unexpected error: %v", err) + } + if err := InitFromConfig(map[string]interface{}{}); err == nil { + t.Error("expected a second InitFromConfig call to error - it must run exactly once") + } +} + +func TestSharedBeforeInitFromConfigErrors(t *testing.T) { + resetSharedForTest(t) + + if _, err := Shared(); err == nil { + t.Error("expected Shared() to error when InitFromConfig was never called") + } +} + +// TestSharedReturnsIdenticalPointer is the actual "single instance" contract: +// not merely "these two configs happen to compare equal" (GetOrCreateRedisClient's +// dedup guarantee) but "there is exactly one gateway-level client, full stop." +func TestSharedReturnsIdenticalPointer(t *testing.T) { + resetSharedForTest(t) + mr := miniredis.RunT(t) + + port, err := strconv.Atoi(mr.Port()) + if err != nil { + t.Fatalf("failed to parse miniredis port: %v", err) + } + raw := map[string]interface{}{"redis": map[string]interface{}{"host": mr.Host(), "port": port}} + if err := InitFromConfig(raw); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + c1, err1 := Shared() + c2, err2 := Shared() + if err1 != nil || err2 != nil { + t.Fatalf("unexpected errors: %v, %v", err1, err2) + } + if c1 != c2 { + t.Error("expected every Shared() call to return the identical *redis.Client pointer") + } +} + +func TestResolve_NilOptsFallsBackToShared(t *testing.T) { + resetSharedForTest(t) + mr := miniredis.RunT(t) + sharedClient := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + SetSharedForTesting(t, sharedClient) + + got, err := Resolve(nil, time.Second) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != sharedClient { + t.Error("expected Resolve(nil, ...) to return the gateway-level Shared client") + } +} + +func TestResolve_NonNilOptsBypassesShared(t *testing.T) { + resetSharedForTest(t) + sharedMR := miniredis.RunT(t) + SetSharedForTesting(t, redis.NewClient(&redis.Options{Addr: sharedMR.Addr()})) + + overrideMR := miniredis.RunT(t) + got, err := Resolve(&redis.Options{Addr: overrideMR.Addr()}, time.Second) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got.Options().Addr != overrideMR.Addr() { + t.Errorf("expected a policy-supplied override to take precedence over the shared client, got client for %q", got.Options().Addr) + } +} + +// TestResolve_NonNilOptsStillDedupes proves Resolve's override branch keeps +// GetOrCreateRedisClient's existing sharing behavior - two policies that both +// explicitly override to the same config still get one pool between them, +// not one pool each. +func TestResolve_NonNilOptsStillDedupes(t *testing.T) { + resetSharedForTest(t) + mr := miniredis.RunT(t) + + c1, err1 := Resolve(&redis.Options{Addr: mr.Addr()}, time.Second) + c2, err2 := Resolve(&redis.Options{Addr: mr.Addr()}, time.Second) + if err1 != nil || err2 != nil { + t.Fatalf("unexpected errors: %v, %v", err1, err2) + } + if c1 != c2 { + t.Error("expected two identical explicit overrides to still share one client") + } +} + +// TestSetSharedForTesting_RestoresPreviousStateAfterTest proves the override +// is scoped to one (sub)test: a subtest's t.Cleanup runs when that subtest +// returns, before the parent continues, so the parent sees the pre-override +// "unconfigured" state again immediately afterward. +func TestSetSharedForTesting_RestoresPreviousStateAfterTest(t *testing.T) { + resetSharedForTest(t) + if err := InitFromConfig(map[string]interface{}{}); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if _, err := Shared(); err == nil { + t.Fatal("expected Shared() to error before any override is set") + } + + t.Run("override active inside subtest", func(t *testing.T) { + mr := miniredis.RunT(t) + SetSharedForTesting(t, redis.NewClient(&redis.Options{Addr: mr.Addr()})) + if _, err := Shared(); err != nil { + t.Fatalf("unexpected error while override was active: %v", err) + } + }) + + if _, err := Shared(); err == nil { + t.Error("expected the override to be reverted once the subtest returned") + } +} + func TestHashPassword(t *testing.T) { if hashRedisPassword("") != "" { t.Error("expected an empty password to hash to empty, not sha256(\"\")") diff --git a/sdk/core/utils/redisclient/testing.go b/sdk/core/utils/redisclient/testing.go new file mode 100644 index 0000000000..c574da54e0 --- /dev/null +++ b/sdk/core/utils/redisclient/testing.go @@ -0,0 +1,43 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com). + * + * WSO2 LLC. licenses this file to you 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 redisclient + +import ( + "testing" + + "github.com/redis/go-redis/v9" +) + +// SetSharedForTesting overrides the process-wide Shared client for the +// duration of t, restoring the previous state automatically via t.Cleanup. +// Test-only - never call this from production code. Since Shared is a +// single global, tests using this helper must not run in parallel with each +// other (no t.Parallel) or they will race on the same override. +func SetSharedForTesting(t testing.TB, client *redis.Client) { + t.Helper() + shared.mu.Lock() + prevClient, prevInited := shared.client, shared.inited + shared.client, shared.inited = client, true + shared.mu.Unlock() + t.Cleanup(func() { + shared.mu.Lock() + shared.client, shared.inited = prevClient, prevInited + shared.mu.Unlock() + }) +} From d469b1cd8e2ffe82c68bc904e43592e392b647dc Mon Sep 17 00:00:00 2001 From: Thenujan Date: Wed, 12 Aug 2026 09:07:30 +0530 Subject: [PATCH 06/10] Enhance Redis client timeout handling and validation in configuration parsing --- sdk/core/utils/redisclient/redisclient.go | 34 +++++++++--- .../utils/redisclient/redisclient_test.go | 54 +++++++++++++++++++ 2 files changed, 82 insertions(+), 6 deletions(-) diff --git a/sdk/core/utils/redisclient/redisclient.go b/sdk/core/utils/redisclient/redisclient.go index a5dc1eb9e7..7d2938b760 100644 --- a/sdk/core/utils/redisclient/redisclient.go +++ b/sdk/core/utils/redisclient/redisclient.go @@ -30,6 +30,8 @@ import ( "crypto/sha256" "encoding/hex" "fmt" + "math" + "net" "strconv" "strings" "sync" @@ -116,10 +118,21 @@ func GetOrCreateRedisClient(opts *redis.Options, pingTimeout time.Duration) (cli return c, true, pingErr } -// defaultSharedPingTimeout bounds the one-time creation ping for the shared -// client. Not configurable: it only affects how long InitFromConfig blocks -// during gateway-runtime startup, never a per-request path. -const defaultSharedPingTimeout = 5 * time.Second +// pingTimeoutMargin is added on top of a client's own configured dial/read/ +// write timeouts to derive the one-time creation ping's timeout (see +// pingTimeoutFor) - enough for the ping's own command round-trip on top of +// whatever the connection attempt itself is allowed to take. +const pingTimeoutMargin = 2 * time.Second + +// pingTimeoutFor derives the creation-ping timeout from opts's own configured +// timeouts, so the ping's context stays alive at least as long as the +// connection attempt permitted by DialTimeout (plus the read/write round-trip +// and a safety margin) - a fixed constant shorter than an operator's +// configured DialTimeout would cut the ping's context before a legitimately +// slow-but-successful connection attempt could complete. +func pingTimeoutFor(opts *redis.Options) time.Duration { + return opts.DialTimeout + opts.ReadTimeout + opts.WriteTimeout + pingTimeoutMargin +} // shared holds the process-wide gateway-level default client. inited // distinguishes "InitFromConfig ran and found no redis section" (client nil, @@ -164,7 +177,7 @@ func InitFromConfig(raw map[string]interface{}) error { return nil } - c, _, _ := newAndPingClient(opts, defaultSharedPingTimeout) + c, _, _ := newAndPingClient(opts, pingTimeoutFor(opts)) shared.client = c return nil } @@ -279,7 +292,7 @@ func resolveOptionsFromConfig(raw map[string]interface{}) (*redis.Options, error } return &redis.Options{ - Addr: fmt.Sprintf("%s:%d", host, port), + Addr: net.JoinHostPort(host, strconv.Itoa(port)), Username: username, Password: password, DB: db, @@ -317,6 +330,15 @@ func intParam(m map[string]interface{}, key string, def int) (int, error) { case int64: return int(n), nil case float64: + if math.IsNaN(n) || math.IsInf(n, 0) { + return 0, fmt.Errorf("expected an integer, got %v", n) + } + if n != math.Trunc(n) { + return 0, fmt.Errorf("expected an integer, got non-integer value %v", n) + } + if n < float64(math.MinInt) || n > float64(math.MaxInt) { + return 0, fmt.Errorf("value %v out of range for int", n) + } return int(n), nil case string: // A TOML value written as {{ env "VAR" "default" }} must be a quoted diff --git a/sdk/core/utils/redisclient/redisclient_test.go b/sdk/core/utils/redisclient/redisclient_test.go index 7c9c23470b..613d4829ac 100644 --- a/sdk/core/utils/redisclient/redisclient_test.go +++ b/sdk/core/utils/redisclient/redisclient_test.go @@ -21,6 +21,7 @@ package redisclient import ( "context" "crypto/tls" + "math" "net" "strconv" "sync" @@ -345,6 +346,23 @@ func TestResolveOptionsFromConfig_ParsesNumericStringPort(t *testing.T) { } } +// TestResolveOptionsFromConfig_BracketsIPv6Host proves Addr is built via +// net.JoinHostPort - a plain fmt.Sprintf("%s:%d", host, port) would produce +// "::1:6380", which is ambiguous/invalid, instead of the required +// "[::1]:6380". +func TestResolveOptionsFromConfig_BracketsIPv6Host(t *testing.T) { + raw := map[string]interface{}{ + "redis": map[string]interface{}{"host": "::1", "port": 6380}, + } + opts, err := resolveOptionsFromConfig(raw) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if want := "[::1]:6380"; opts.Addr != want { + t.Errorf("got Addr %q, want %q", opts.Addr, want) + } +} + func TestResolveOptionsFromConfig_RejectsWrongShapedValue(t *testing.T) { _, err := resolveOptionsFromConfig(map[string]interface{}{ "redis": map[string]interface{}{"port": "not-a-number"}, @@ -354,6 +372,42 @@ func TestResolveOptionsFromConfig_RejectsWrongShapedValue(t *testing.T) { } } +// TestIntParam_RejectsInvalidFloat64 locks in that intParam validates a +// float64 before converting it - NaN/Inf/fractional/out-of-range values must +// error rather than silently truncating or converting a NaN/Inf into +// undefined behavior. +func TestIntParam_RejectsInvalidFloat64(t *testing.T) { + cases := []struct { + name string + v float64 + }{ + {"NaN", math.NaN()}, + {"+Inf", math.Inf(1)}, + {"-Inf", math.Inf(-1)}, + {"fractional", 1.5}, + {"aboveMaxInt", 1e19}, + {"belowMinInt", -1e19}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + _, err := intParam(map[string]interface{}{"port": c.v}, "port", 6379) + if err == nil { + t.Errorf("expected an error for float64 value %v, got nil", c.v) + } + }) + } +} + +func TestIntParam_AcceptsIntegralFloat64(t *testing.T) { + got, err := intParam(map[string]interface{}{"port": float64(6380)}, "port", 6379) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != 6380 { + t.Errorf("got %d, want 6380", got) + } +} + func TestResolveOptionsFromConfig_RejectsNonTableSection(t *testing.T) { _, err := resolveOptionsFromConfig(map[string]interface{}{"redis": "not-a-table"}) if err == nil { From 514be8d0ccddc43420e627c398d1cd2d3afd6953 Mon Sep 17 00:00:00 2001 From: Thenujan Date: Wed, 19 Aug 2026 21:06:43 +0530 Subject: [PATCH 07/10] Add mapstructure integration for Redis configuration handling --- sdk/core/go.mod | 1 + sdk/core/go.sum | 2 + sdk/core/utils/redisclient/redisclient.go | 180 +++++++----------- .../utils/redisclient/redisclient_test.go | 23 +-- 4 files changed, 85 insertions(+), 121 deletions(-) diff --git a/sdk/core/go.mod b/sdk/core/go.mod index b43acbabf8..6bf7ce1c35 100644 --- a/sdk/core/go.mod +++ b/sdk/core/go.mod @@ -4,6 +4,7 @@ go 1.26.2 require ( github.com/alicebob/miniredis/v2 v2.38.0 + github.com/go-viper/mapstructure/v2 v2.5.0 github.com/redis/go-redis/v9 v9.22.0 ) diff --git a/sdk/core/go.sum b/sdk/core/go.sum index e9ba59b600..9013c6e309 100644 --- a/sdk/core/go.sum +++ b/sdk/core/go.sum @@ -8,6 +8,8 @@ github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UF github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro= +github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= diff --git a/sdk/core/utils/redisclient/redisclient.go b/sdk/core/utils/redisclient/redisclient.go index 7d2938b760..283228b617 100644 --- a/sdk/core/utils/redisclient/redisclient.go +++ b/sdk/core/utils/redisclient/redisclient.go @@ -32,11 +32,12 @@ import ( "fmt" "math" "net" + "reflect" "strconv" - "strings" "sync" "time" + "github.com/go-viper/mapstructure/v2" "github.com/redis/go-redis/v9" ) @@ -236,13 +237,34 @@ func pingClient(c *redis.Client, pingTimeout time.Duration) error { return c.Ping(ctx).Err() } +// redisSectionFields is the decode target for the "redis" table, pre-filled +// with operator-facing defaults (host "localhost", port 6379, db 0, poolSize +// 0 (go-redis default), connectionTimeout 5s, readTimeout/writeTimeout 3s) +// before mapstructure.Decode overwrites only the keys actually present in +// the config - the same weakly-typed decode (string durations, a numeric +// string from a {{ env }} token, int/int64/float64 from a JSON-sourced +// config path) every other config section in the repo goes through via +// koanf/mapstructure; see gateway-runtime/policy-engine's +// internal/config.Load. sdk/core otherwise has no koanf dependency, so this +// package decodes the raw map directly with mapstructure rather than pulling +// in koanf's file/merge machinery just for this one section. +type redisSectionFields struct { + Host string `mapstructure:"host"` + Port int `mapstructure:"port"` + Username string `mapstructure:"username"` + Password string `mapstructure:"password"` + DB int `mapstructure:"db"` + PoolSize int `mapstructure:"pool_size"` + ConnectionTimeout time.Duration `mapstructure:"connection_timeout"` + ReadTimeout time.Duration `mapstructure:"read_timeout"` + WriteTimeout time.Duration `mapstructure:"write_timeout"` +} + // resolveOptionsFromConfig extracts *redis.Options from raw["redis"] - a // top-level section, sibling to "router"/"analytics"/etc in the gateway's // complete config tree, not nested under "policy_configurations" (this is -// gateway-wide infrastructure, not a per-policy setting) - using the -// operator-facing defaults (host "localhost", port 6379, db 0, poolSize 0 -// (go-redis default), connectionTimeout 5s, readTimeout/writeTimeout 3s). -// Returns (nil, nil) - not an error - when raw has no "redis" key at all. +// gateway-wide infrastructure, not a per-policy setting). Returns (nil, nil) +// - not an error - when raw has no "redis" key at all. func resolveOptionsFromConfig(raw map[string]interface{}) (*redis.Options, error) { section, ok := raw["redis"] if !ok || section == nil { @@ -253,124 +275,62 @@ func resolveOptionsFromConfig(raw map[string]interface{}) (*redis.Options, error return nil, fmt.Errorf(`"redis" must be a table, got %T`, section) } - connectionTimeout, err := durationParam(m, "connection_timeout", 5*time.Second) - if err != nil { - return nil, fmt.Errorf("connection_timeout: %w", err) - } - readTimeout, err := durationParam(m, "read_timeout", 3*time.Second) - if err != nil { - return nil, fmt.Errorf("read_timeout: %w", err) - } - writeTimeout, err := durationParam(m, "write_timeout", 3*time.Second) - if err != nil { - return nil, fmt.Errorf("write_timeout: %w", err) - } - port, err := intParam(m, "port", 6379) - if err != nil { - return nil, fmt.Errorf("port: %w", err) - } - db, err := intParam(m, "db", 0) - if err != nil { - return nil, fmt.Errorf("db: %w", err) - } - poolSize, err := intParam(m, "pool_size", 0) - if err != nil { - return nil, fmt.Errorf("pool_size: %w", err) - } - - host, err := stringParam(m, "host", "localhost") - if err != nil { - return nil, fmt.Errorf("host: %w", err) + fields := redisSectionFields{ + Host: "localhost", + Port: 6379, + ConnectionTimeout: 5 * time.Second, + ReadTimeout: 3 * time.Second, + WriteTimeout: 3 * time.Second, } - username, err := stringParam(m, "username", "") + decoder, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{ + WeaklyTypedInput: true, + DecodeHook: mapstructure.ComposeDecodeHookFunc( + rejectNonIntegralFloatHookFunc, + mapstructure.StringToTimeDurationHookFunc(), + ), + Result: &fields, + }) if err != nil { - return nil, fmt.Errorf("username: %w", err) + return nil, fmt.Errorf("building redis config decoder: %w", err) } - password, err := stringParam(m, "password", "") - if err != nil { - return nil, fmt.Errorf("password: %w", err) + if err := decoder.Decode(m); err != nil { + return nil, err } return &redis.Options{ - Addr: net.JoinHostPort(host, strconv.Itoa(port)), - Username: username, - Password: password, - DB: db, - DialTimeout: connectionTimeout, - ReadTimeout: readTimeout, - WriteTimeout: writeTimeout, - PoolSize: poolSize, + Addr: net.JoinHostPort(fields.Host, strconv.Itoa(fields.Port)), + Username: fields.Username, + Password: fields.Password, + DB: fields.DB, + DialTimeout: fields.ConnectionTimeout, + ReadTimeout: fields.ReadTimeout, + WriteTimeout: fields.WriteTimeout, + PoolSize: fields.PoolSize, }, nil } -// stringParam/intParam/durationParam read key from m, applying def when the -// key is absent or nil. They error on a present-but-wrong-shaped value -// rather than silently falling back to def - a typo'd config value should -// surface at startup, not resolve to a default the operator never asked for. -func stringParam(m map[string]interface{}, key, def string) (string, error) { - v, ok := m[key] - if !ok || v == nil { - return def, nil +// rejectNonIntegralFloatHookFunc runs before mapstructure's own weakly-typed +// float->int conversion (int64(dataVal.Float()), reached for e.g. a +// JSON-sourced "port"/"db"/"pool_size" value decoded as float64) - that +// conversion is undefined/lossy for NaN, +-Inf, a fractional value, or a +// magnitude outside the int range, so reject those here instead of letting +// them silently truncate into a bogus port/db/pool size. +func rejectNonIntegralFloatHookFunc(_, to reflect.Kind, data interface{}) (interface{}, error) { + if to != reflect.Int { + return data, nil } - s, ok := v.(string) + f, ok := data.(float64) if !ok { - return "", fmt.Errorf("expected a string, got %T", v) + return data, nil } - return s, nil -} - -func intParam(m map[string]interface{}, key string, def int) (int, error) { - v, ok := m[key] - if !ok || v == nil { - return def, nil - } - switch n := v.(type) { - case int: - return n, nil - case int64: - return int(n), nil - case float64: - if math.IsNaN(n) || math.IsInf(n, 0) { - return 0, fmt.Errorf("expected an integer, got %v", n) - } - if n != math.Trunc(n) { - return 0, fmt.Errorf("expected an integer, got non-integer value %v", n) - } - if n < float64(math.MinInt) || n > float64(math.MaxInt) { - return 0, fmt.Errorf("value %v out of range for int", n) - } - return int(n), nil - case string: - // A TOML value written as {{ env "VAR" "default" }} must be a quoted - // string literal (TOML has no unquoted template syntax) - gateway-runtime's - // config interpolation resolves the token in place but never changes the - // field's type, so a numeric config value arrives here as a numeric - // string, not an int. Reject anything that isn't actually numeric. - parsed, err := strconv.Atoi(strings.TrimSpace(n)) - if err != nil { - return 0, fmt.Errorf("expected an integer, got non-numeric string %q", n) - } - return parsed, nil - default: - return 0, fmt.Errorf("expected an integer, got %T", v) + if math.IsNaN(f) || math.IsInf(f, 0) { + return nil, fmt.Errorf("expected an integer, got %v", f) } -} - -func durationParam(m map[string]interface{}, key string, def time.Duration) (time.Duration, error) { - v, ok := m[key] - if !ok || v == nil { - return def, nil + if f != math.Trunc(f) { + return nil, fmt.Errorf("expected an integer, got non-integer value %v", f) } - switch d := v.(type) { - case string: - parsed, err := time.ParseDuration(d) - if err != nil { - return 0, fmt.Errorf("invalid duration %q: %w", d, err) - } - return parsed, nil - case time.Duration: - return d, nil - default: - return 0, fmt.Errorf("expected a duration string, got %T", v) + if f < float64(math.MinInt) || f > float64(math.MaxInt) { + return nil, fmt.Errorf("value %v out of range for int", f) } + return data, nil } diff --git a/sdk/core/utils/redisclient/redisclient_test.go b/sdk/core/utils/redisclient/redisclient_test.go index 613d4829ac..097208b1b1 100644 --- a/sdk/core/utils/redisclient/redisclient_test.go +++ b/sdk/core/utils/redisclient/redisclient_test.go @@ -372,11 +372,12 @@ func TestResolveOptionsFromConfig_RejectsWrongShapedValue(t *testing.T) { } } -// TestIntParam_RejectsInvalidFloat64 locks in that intParam validates a -// float64 before converting it - NaN/Inf/fractional/out-of-range values must -// error rather than silently truncating or converting a NaN/Inf into -// undefined behavior. -func TestIntParam_RejectsInvalidFloat64(t *testing.T) { +// TestResolveOptionsFromConfig_RejectsInvalidFloat64Port locks in that a +// float64 "port" (e.g. from a JSON-sourced config path, decoded as float64 +// rather than int) is validated before conversion - +// NaN/Inf/fractional/out-of-range values must error rather than silently +// truncating or converting a NaN/Inf into undefined behavior. +func TestResolveOptionsFromConfig_RejectsInvalidFloat64Port(t *testing.T) { cases := []struct { name string v float64 @@ -390,21 +391,21 @@ func TestIntParam_RejectsInvalidFloat64(t *testing.T) { } for _, c := range cases { t.Run(c.name, func(t *testing.T) { - _, err := intParam(map[string]interface{}{"port": c.v}, "port", 6379) + _, err := resolveOptionsFromConfig(map[string]interface{}{"redis": map[string]interface{}{"port": c.v}}) if err == nil { - t.Errorf("expected an error for float64 value %v, got nil", c.v) + t.Errorf("expected an error for float64 port %v, got nil", c.v) } }) } } -func TestIntParam_AcceptsIntegralFloat64(t *testing.T) { - got, err := intParam(map[string]interface{}{"port": float64(6380)}, "port", 6379) +func TestResolveOptionsFromConfig_AcceptsIntegralFloat64Port(t *testing.T) { + opts, err := resolveOptionsFromConfig(map[string]interface{}{"redis": map[string]interface{}{"port": float64(6380)}}) if err != nil { t.Fatalf("unexpected error: %v", err) } - if got != 6380 { - t.Errorf("got %d, want 6380", got) + if want := "localhost:6380"; opts.Addr != want { + t.Errorf("got Addr %q, want %q", opts.Addr, want) } } From 2e4bf6cb6e6fa050e8fa1375215d92fafafe21cc Mon Sep 17 00:00:00 2001 From: Thenujan Date: Thu, 20 Aug 2026 12:43:15 +0530 Subject: [PATCH 08/10] Refactor Redis client functions and add parameter extraction for connection options --- .../policy-engine/cmd/policy-engine/main.go | 9 +- sdk/core/utils/redisclient/redisclient.go | 242 +++++++++++------- .../utils/redisclient/redisclient_test.go | 162 +++++++++--- 3 files changed, 267 insertions(+), 146 deletions(-) diff --git a/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go b/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go index 3a82b80460..ce9263f79a 100644 --- a/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go +++ b/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go @@ -165,14 +165,7 @@ func main() { } slog.InfoContext(ctx, "Config set in registry for ${config} CEL resolution") - // Initialize the gateway-level shared Redis client (top-level "redis" config - // section - gateway infrastructure, not nested under policy_configurations, - // since it's not scoped to policies even though policy-engine is its current - // consumer). Must run before any policy chain is built (below) - a policy - // instance that calls redisclient.Shared()/Resolve() during construction - // assumes this has already run. A missing "redis" section is not an error - // here - it's only surfaced lazily, the first time some policy actually - // needs it. + // Initialize the gateway-level shared Redis client (top-level "redis" config section) if err := redisclient.InitFromConfig(cfg.PolicyEngine.RawConfig); err != nil { slog.ErrorContext(ctx, "Failed to initialize shared redis client", "error", err) os.Exit(1) diff --git a/sdk/core/utils/redisclient/redisclient.go b/sdk/core/utils/redisclient/redisclient.go index 283228b617..c25c0d89c6 100644 --- a/sdk/core/utils/redisclient/redisclient.go +++ b/sdk/core/utils/redisclient/redisclient.go @@ -16,13 +16,11 @@ * under the License. */ -// Package redisclient shares one process-wide *redis.Client (one connection -// pool) per distinct connection configuration, across every caller that -// imports it - see GetOrCreateRedisClient. It also exposes a single -// gateway-wide default client (Shared, backed by the operator's top-level -// "redis" config section - gateway infrastructure, not something scoped to -// policies) that a policy falls back to when it has no Redis config of its -// own - see Resolve. +// Package redisclient shares one process-wide *redis.Client per distinct +// connection config across every caller - see GetOrCreate. It also exposes a +// gateway-wide default client (Shared, backed by the top-level "redis" +// config section) that a policy without its own Redis config falls back to +// - see Resolve. package redisclient import ( @@ -34,6 +32,7 @@ import ( "net" "reflect" "strconv" + "strings" "sync" "time" @@ -41,11 +40,9 @@ import ( "github.com/redis/go-redis/v9" ) -// redisConnKey identifies a distinct Redis connection configuration. Two policy -// instances with identical connection settings share one *redis.Client (one pool). -// -// Excludes TLSConfig and any credentials-provider option - see -// GetOrCreateRedisClient's bypass for those. +// redisConnKey identifies a distinct Redis connection config; identical +// settings share one *redis.Client. Excludes TLSConfig/credentials-provider +// options - see GetOrCreate's bypass for those. type redisConnKey struct { addr string username string @@ -58,9 +55,8 @@ type redisConnKey struct { poolSize int } -// redisClients is the process-wide registry of shared Redis clients. Without it, -// GetPolicy creates a new *redis.Client (a whole connection pool) per policy instance -// and per config reload, leaking pools and exploding Redis connections at scale. +// redisClients is the process-wide registry of shared Redis clients - without +// it, every policy instance/reload would open its own connection pool. var redisClients = struct { mu sync.Mutex m map[redisConnKey]*redis.Client @@ -74,16 +70,16 @@ func hashRedisPassword(p string) string { return hex.EncodeToString(sum[:]) } -// GetOrCreateRedisClient returns the process-wide shared client for these connection -// settings, creating (and pinging once) it on first use. created reports whether this -// call created the client; pingErr is non-nil only when created and the initial ping -// failed. The client is registered and returned even on ping failure (go-redis -// reconnects lazily). Clients are never closed — they live for the process lifetime. -func GetOrCreateRedisClient(opts *redis.Options, pingTimeout time.Duration) (client *redis.Client, created bool, pingErr error) { - // TLSConfig and credentials-provider hooks can't be fingerprinted - // safely: a *tls.Config's pointer says nothing about its content, and - // Go func values aren't comparable at all. Bypass the registry rather - // than risk silently reusing a client built for a different config. +// GetOrCreate returns the process-wide shared client for these connection +// settings, creating (and pinging once) it on first use. created reports +// whether this call created the client; pingErr is non-nil only then. The +// client is registered even on ping failure (go-redis reconnects lazily) and +// is never closed - it lives for the process lifetime. +func GetOrCreate(opts *redis.Options, pingTimeout time.Duration) (client *redis.Client, created bool, pingErr error) { + // TLSConfig/credentials-provider can't be fingerprinted safely (a + // *tls.Config pointer says nothing about content; func values aren't + // comparable) - bypass the registry rather than risk reusing a client + // built for a different config. if opts.TLSConfig != nil || opts.CredentialsProvider != nil || opts.CredentialsProviderContext != nil || opts.StreamingCredentialsProvider != nil { return newAndPingClient(opts, pingTimeout) } @@ -100,12 +96,8 @@ func GetOrCreateRedisClient(opts *redis.Options, pingTimeout time.Duration) (cli poolSize: opts.PoolSize, } - // Lock guards only the map lookup/insert, never the ping below - mu is - // process-wide, so holding it during a slow/down connection's ping - // would stall every other caller's get-or-create too. A concurrent - // caller for the same key may see the just-inserted client before this - // ping finishes - fine, since a reused client is already "assumed - // healthy" regardless of timing, never gated on this call's pingErr. + // Lock guards only the map lookup/insert, never the ping below - holding + // it during a slow ping would stall every other caller's get-or-create. redisClients.mu.Lock() if c, ok := redisClients.m[key]; ok { redisClients.mu.Unlock() @@ -119,49 +111,35 @@ func GetOrCreateRedisClient(opts *redis.Options, pingTimeout time.Duration) (cli return c, true, pingErr } -// pingTimeoutMargin is added on top of a client's own configured dial/read/ -// write timeouts to derive the one-time creation ping's timeout (see -// pingTimeoutFor) - enough for the ping's own command round-trip on top of -// whatever the connection attempt itself is allowed to take. +// pingTimeoutMargin is added on top of a client's own dial/read/write +// timeouts to derive the one-time creation ping's timeout - room for the +// ping's own round-trip on top of the connection attempt itself. const pingTimeoutMargin = 2 * time.Second -// pingTimeoutFor derives the creation-ping timeout from opts's own configured -// timeouts, so the ping's context stays alive at least as long as the -// connection attempt permitted by DialTimeout (plus the read/write round-trip -// and a safety margin) - a fixed constant shorter than an operator's -// configured DialTimeout would cut the ping's context before a legitimately -// slow-but-successful connection attempt could complete. +// pingTimeoutFor derives the creation-ping timeout from opts's own timeouts, +// so a fixed constant can't cut the ping short before a legitimately slow +// connection attempt completes. func pingTimeoutFor(opts *redis.Options) time.Duration { return opts.DialTimeout + opts.ReadTimeout + opts.WriteTimeout + pingTimeoutMargin } // shared holds the process-wide gateway-level default client. inited -// distinguishes "InitFromConfig ran and found no redis section" (client nil, -// inited true - Shared reports a config-gap error) from "InitFromConfig was -// never called at all" (a gateway-runtime wiring bug - Shared reports that -// distinctly, since it means something programming-level is missing, not an -// operator config gap). +// distinguishes "InitFromConfig ran, no redis section" (client nil, inited +// true) from "InitFromConfig never called" (a wiring bug) - Shared reports +// each distinctly. var shared struct { mu sync.Mutex client *redis.Client inited bool } -// InitFromConfig resolves the operator-level top-level "redis" section from -// raw (e.g. cfg.PolicyEngine.RawConfig - raw's other top-level sections like -// "analytics"/"router"/"policy_configurations" are ignored here) and creates -// the process-wide shared client. This is gateway-wide infrastructure, not -// something scoped to policies - deliberately NOT nested under -// "policy_configurations" (that namespace is policy-engine's own ${config...} -// CEL-resolution mechanism for per-policy system parameters; a shared -// resource other gateway components could reach doesn't belong inside it). -// Must be called exactly once, at gateway-runtime startup, before any policy -// factory runs - see Shared and Resolve. A missing "redis" key is not an -// error: most gateways may have zero Redis-consuming policies configured, -// and that absence only matters lazily, the first time some policy actually -// calls Shared. A connection/ping failure is likewise not fatal here - the -// client is still created and stored (go-redis reconnects lazily), matching -// GetOrCreateRedisClient's own create-time philosophy. +// InitFromConfig resolves the top-level "redis" section from raw (e.g. +// cfg.PolicyEngine.RawConfig) and creates the process-wide shared client - +// gateway-wide infrastructure, deliberately not nested under +// "policy_configurations" (policy-engine's per-policy ${config...} namespace). +// Must be called exactly once, at startup, before any policy factory runs - +// see Shared/Resolve. A missing "redis" key or a failed ping is not fatal +// here; both only matter lazily, the first time a policy calls Shared. func InitFromConfig(raw map[string]interface{}) error { shared.mu.Lock() defer shared.mu.Unlock() @@ -184,11 +162,9 @@ func InitFromConfig(raw map[string]interface{}) error { } // Shared returns the process-wide gateway-level default client, backed by -// the top-level "redis" config section. It errors if InitFromConfig was -// never called (a gateway-runtime wiring bug, not a normal runtime -// condition) or if no "redis" section was configured at all - callers must -// treat the latter as a real configuration gap rather than assuming a -// shared Redis is always available. +// the top-level "redis" config section. Errors if InitFromConfig was never +// called, or if no "redis" section was configured - callers must treat the +// latter as a real config gap, not assume a shared Redis always exists. func Shared() (*redis.Client, error) { shared.mu.Lock() defer shared.mu.Unlock() @@ -202,24 +178,20 @@ func Shared() (*redis.Client, error) { } // Resolve returns the client a policy instance should use: opts's own -// connection settings when the policy resolved one from its own config -// section, otherwise the gateway-level Shared client. -// -// opts must be nil - never a zero-value *redis.Options - when the policy's -// own section was absent. A struct pre-filled with the policy's schema -// defaults would always look "configured," and this fallback would never -// trigger; the presence check belongs to the caller's own config-extraction -// code, on whichever field has no default (e.g. host). +// settings if the policy configured its own connection, otherwise the +// gateway-level Shared client. opts must be nil - never a schema-defaulted +// zero-value *redis.Options - when the policy's own section was absent, or +// this fallback never triggers; that presence check is the caller's own. func Resolve(opts *redis.Options, pingTimeout time.Duration) (*redis.Client, error) { if opts == nil { return Shared() } - client, _, _ := GetOrCreateRedisClient(opts, pingTimeout) + client, _, _ := GetOrCreate(opts, pingTimeout) return client, nil } // newAndPingClient creates a client and pings it once. created is always -// true - only present so this matches GetOrCreateRedisClient's own return +// true - only present so this matches GetOrCreate's own return // shape at its call sites. func newAndPingClient(opts *redis.Options, pingTimeout time.Duration) (client *redis.Client, created bool, pingErr error) { c := redis.NewClient(opts) @@ -227,10 +199,8 @@ func newAndPingClient(opts *redis.Options, pingTimeout time.Duration) (client *r } // pingClient pings an already-constructed client once, bounded by -// pingTimeout. Split out from newAndPingClient because GetOrCreateRedisClient's -// main path must insert the client into the registry BEFORE pinging (so a -// concurrent caller for the same key sees it immediately), not create-then-ping -// as one atomic step. +// pingTimeout. Split out since GetOrCreate's main path must insert the +// client into the registry BEFORE pinging, not create-then-ping atomically. func pingClient(c *redis.Client, pingTimeout time.Duration) error { ctx, cancel := context.WithTimeout(context.Background(), pingTimeout) defer cancel() @@ -238,16 +208,10 @@ func pingClient(c *redis.Client, pingTimeout time.Duration) error { } // redisSectionFields is the decode target for the "redis" table, pre-filled -// with operator-facing defaults (host "localhost", port 6379, db 0, poolSize -// 0 (go-redis default), connectionTimeout 5s, readTimeout/writeTimeout 3s) -// before mapstructure.Decode overwrites only the keys actually present in -// the config - the same weakly-typed decode (string durations, a numeric -// string from a {{ env }} token, int/int64/float64 from a JSON-sourced -// config path) every other config section in the repo goes through via -// koanf/mapstructure; see gateway-runtime/policy-engine's -// internal/config.Load. sdk/core otherwise has no koanf dependency, so this -// package decodes the raw map directly with mapstructure rather than pulling -// in koanf's file/merge machinery just for this one section. +// with defaults before mapstructure.Decode overwrites only the keys present +// in the config - the same weakly-typed decode every other config section +// goes through via koanf/mapstructure (see policy-engine's +// internal/config.Load), just without pulling koanf itself into sdk/core. type redisSectionFields struct { Host string `mapstructure:"host"` Port int `mapstructure:"port"` @@ -261,10 +225,9 @@ type redisSectionFields struct { } // resolveOptionsFromConfig extracts *redis.Options from raw["redis"] - a -// top-level section, sibling to "router"/"analytics"/etc in the gateway's -// complete config tree, not nested under "policy_configurations" (this is +// top-level section, not nested under "policy_configurations" (this is // gateway-wide infrastructure, not a per-policy setting). Returns (nil, nil) -// - not an error - when raw has no "redis" key at all. +// when raw has no "redis" key at all. func resolveOptionsFromConfig(raw map[string]interface{}) (*redis.Options, error) { section, ok := raw["redis"] if !ok || section == nil { @@ -310,11 +273,9 @@ func resolveOptionsFromConfig(raw map[string]interface{}) (*redis.Options, error } // rejectNonIntegralFloatHookFunc runs before mapstructure's own weakly-typed -// float->int conversion (int64(dataVal.Float()), reached for e.g. a -// JSON-sourced "port"/"db"/"pool_size" value decoded as float64) - that -// conversion is undefined/lossy for NaN, +-Inf, a fractional value, or a -// magnitude outside the int range, so reject those here instead of letting -// them silently truncate into a bogus port/db/pool size. +// float->int conversion, which is undefined/lossy for NaN, +-Inf, a +// fractional value, or an out-of-range magnitude - reject those instead of +// letting them silently truncate into a bogus port/db/pool size. func rejectNonIntegralFloatHookFunc(_, to reflect.Kind, data interface{}) (interface{}, error) { if to != reflect.Int { return data, nil @@ -334,3 +295,88 @@ func rejectNonIntegralFloatHookFunc(_, to reflect.Kind, data interface{}) (inter } return data, nil } + +// ExtractOverrideFromParams reads systemParameters.redis.* from a policy's +// own params into a *redis.Options for that connection. Returns nil when +// redis.host is absent - unlike other fields, host has no default, or the +// gateway-wide fallback (Resolve/Shared) would never trigger. A malformed +// field silently falls back to its default rather than erroring, unlike +// resolveOptionsFromConfig's stricter decode of operator config. Dotted-key +// lookups tolerate both a flattened key (params["redis.host"]) and a nested +// map (params["redis"]["host"]). +func ExtractOverrideFromParams(params map[string]interface{}) *redis.Options { + host := paramString(params, "redis.host", "") + if host == "" { + return nil + } + return &redis.Options{ + Addr: net.JoinHostPort(host, strconv.Itoa(paramInt(params, "redis.port", 6379))), + Username: paramString(params, "redis.username", ""), + Password: paramString(params, "redis.password", ""), + DB: paramInt(params, "redis.db", 0), + DialTimeout: paramDuration(params, "redis.connectionTimeout", 5*time.Second), + ReadTimeout: paramDuration(params, "redis.readTimeout", 3*time.Second), + WriteTimeout: paramDuration(params, "redis.writeTimeout", 3*time.Second), + PoolSize: paramInt(params, "redis.poolSize", 0), + } +} + +// paramLookup resolves a dotted key ("redis.host") against params, tolerating +// either a flattened key (params["redis.host"]) or nested maps +// (params["redis"]["host"]). +func paramLookup(params map[string]interface{}, dottedKey string) (interface{}, bool) { + if v, ok := params[dottedKey]; ok { + return v, true + } + var cur interface{} = params + for _, part := range strings.Split(dottedKey, ".") { + m, ok := cur.(map[string]interface{}) + if !ok { + return nil, false + } + v, ok := m[part] + if !ok { + return nil, false + } + cur = v + } + return cur, true +} + +func paramString(params map[string]interface{}, dottedKey, def string) string { + if v, ok := paramLookup(params, dottedKey); ok { + if s, ok := v.(string); ok && strings.TrimSpace(s) != "" { + return strings.TrimSpace(s) + } + } + return def +} + +func paramInt(params map[string]interface{}, dottedKey string, def int) int { + if v, ok := paramLookup(params, dottedKey); ok { + switch n := v.(type) { + case int: + return n + case int64: + return int(n) + case float64: + return int(n) + case string: + if parsed, err := strconv.Atoi(strings.TrimSpace(n)); err == nil { + return parsed + } + } + } + return def +} + +func paramDuration(params map[string]interface{}, dottedKey string, def time.Duration) time.Duration { + if v, ok := paramLookup(params, dottedKey); ok { + if s, ok := v.(string); ok { + if d, err := time.ParseDuration(strings.TrimSpace(s)); err == nil { + return d + } + } + } + return def +} diff --git a/sdk/core/utils/redisclient/redisclient_test.go b/sdk/core/utils/redisclient/redisclient_test.go index 097208b1b1..0f912fec69 100644 --- a/sdk/core/utils/redisclient/redisclient_test.go +++ b/sdk/core/utils/redisclient/redisclient_test.go @@ -36,12 +36,12 @@ func TestGetOrCreateClient_SharesClientForIdenticalConfig(t *testing.T) { mr := miniredis.RunT(t) opts := &redis.Options{Addr: mr.Addr(), DB: 0} - c1, created1, err1 := GetOrCreateRedisClient(opts, time.Second) + c1, created1, err1 := GetOrCreate(opts, time.Second) if !created1 || err1 != nil { t.Fatalf("first call: created=%v err=%v (want true,nil)", created1, err1) } - c2, created2, err2 := GetOrCreateRedisClient(opts, time.Second) + c2, created2, err2 := GetOrCreate(opts, time.Second) if created2 || err2 != nil { t.Fatalf("second call: created=%v err=%v (want false,nil)", created2, err2) } @@ -53,8 +53,8 @@ func TestGetOrCreateClient_SharesClientForIdenticalConfig(t *testing.T) { func TestGetOrCreateClient_DistinctClientForDifferentConfig(t *testing.T) { mr := miniredis.RunT(t) - c1, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), DB: 0}, time.Second) - c2, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), DB: 1}, time.Second) + c1, _, _ := GetOrCreate(&redis.Options{Addr: mr.Addr(), DB: 0}, time.Second) + c2, _, _ := GetOrCreate(&redis.Options{Addr: mr.Addr(), DB: 1}, time.Second) if c1 == c2 { t.Error("expected different DB selection to produce a distinct *redis.Client") @@ -64,9 +64,9 @@ func TestGetOrCreateClient_DistinctClientForDifferentConfig(t *testing.T) { func TestGetOrCreateClient_DifferentPasswordProducesDistinctClient(t *testing.T) { mr := miniredis.RunT(t) - c1, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), Password: "one"}, time.Second) - c2, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), Password: "two"}, time.Second) - c3, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr()}, time.Second) // no password at all + c1, _, _ := GetOrCreate(&redis.Options{Addr: mr.Addr(), Password: "one"}, time.Second) + c2, _, _ := GetOrCreate(&redis.Options{Addr: mr.Addr(), Password: "two"}, time.Second) + c3, _, _ := GetOrCreate(&redis.Options{Addr: mr.Addr()}, time.Second) // no password at all if c1 == c2 { t.Error("expected different passwords to produce distinct clients") @@ -82,8 +82,8 @@ func TestGetOrCreateClient_SharedAcrossSimulatedPolicies(t *testing.T) { // Two distinct call sites with identical settings must share one // client - the whole point of centralizing the registry. - fromPolicyA, _, _ := GetOrCreateRedisClient(opts(), time.Second) - fromPolicyB, _, _ := GetOrCreateRedisClient(opts(), time.Second) + fromPolicyA, _, _ := GetOrCreate(opts(), time.Second) + fromPolicyB, _, _ := GetOrCreate(opts(), time.Second) if fromPolicyA != fromPolicyB { t.Fatal("expected two distinct callers with identical config to share one client") @@ -109,13 +109,13 @@ func TestGetOrCreateClient_ReuseSkipsPing(t *testing.T) { addr := mr.Addr() // capture before mr.Close() below opts := &redis.Options{Addr: addr, DB: 0} - c1, created1, err1 := GetOrCreateRedisClient(opts, time.Second) + c1, created1, err1 := GetOrCreate(opts, time.Second) if !created1 || err1 != nil { t.Fatalf("first call: created=%v err=%v (want true,nil)", created1, err1) } mr.Close() - c2, created2, err2 := GetOrCreateRedisClient(opts, time.Second) + c2, created2, err2 := GetOrCreate(opts, time.Second) if created2 || err2 != nil || c2 != c1 { t.Fatalf("reuse after Redis went down should skip the ping: created=%v err=%v same=%v", created2, err2, c2 == c1) } @@ -124,9 +124,9 @@ func TestGetOrCreateClient_ReuseSkipsPing(t *testing.T) { func TestGetOrCreateClient_DifferentProtocolProducesDistinctClient(t *testing.T) { mr := miniredis.RunT(t) - c1, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), Protocol: 2}, time.Second) - c2, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), Protocol: 3}, time.Second) - c3, _, _ := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), Protocol: 2}, time.Second) + c1, _, _ := GetOrCreate(&redis.Options{Addr: mr.Addr(), Protocol: 2}, time.Second) + c2, _, _ := GetOrCreate(&redis.Options{Addr: mr.Addr(), Protocol: 3}, time.Second) + c3, _, _ := GetOrCreate(&redis.Options{Addr: mr.Addr(), Protocol: 2}, time.Second) if c1 == c2 { t.Error("expected different RESP protocol versions to produce distinct clients") @@ -146,8 +146,8 @@ func TestGetOrCreateClient_TLSConfigBypassesRegistry(t *testing.T) { optsA := &redis.Options{Addr: mr.Addr(), TLSConfig: &tls.Config{}} //nolint:gosec // test-only, no real handshake asserted optsB := &redis.Options{Addr: mr.Addr(), TLSConfig: &tls.Config{}} //nolint:gosec - c1, created1, _ := GetOrCreateRedisClient(optsA, time.Second) - c2, created2, _ := GetOrCreateRedisClient(optsB, time.Second) + c1, created1, _ := GetOrCreate(optsA, time.Second) + c2, created2, _ := GetOrCreate(optsB, time.Second) if !created1 || !created2 { t.Fatalf("expected every TLSConfig-bearing call to report created=true (never reused), got %v and %v", created1, created2) @@ -161,8 +161,8 @@ func TestGetOrCreateClient_CredentialsProviderBypassesRegistry(t *testing.T) { mr := miniredis.RunT(t) provider := func() (string, string) { return "", "" } - c1, created1, err1 := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), CredentialsProvider: provider}, time.Second) - c2, created2, err2 := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), CredentialsProvider: provider}, time.Second) + c1, created1, err1 := GetOrCreate(&redis.Options{Addr: mr.Addr(), CredentialsProvider: provider}, time.Second) + c2, created2, err2 := GetOrCreate(&redis.Options{Addr: mr.Addr(), CredentialsProvider: provider}, time.Second) if !created1 || err1 != nil { t.Fatalf("first call: created=%v err=%v (want true,nil)", created1, err1) @@ -211,7 +211,7 @@ func TestGetOrCreateClient_DoesNotHoldLockDuringPing(t *testing.T) { // ReadTimeout set explicitly - the dial succeeds, it's the // read-for-a-reply that hangs, and go-redis's default (5s) would // otherwise bound that wait regardless of pingTimeout. - _, _, _ = GetOrCreateRedisClient(&redis.Options{ + _, _, _ = GetOrCreate(&redis.Options{ Addr: ln.Addr().String(), DB: 0, ReadTimeout: time.Second, @@ -227,7 +227,7 @@ func TestGetOrCreateClient_DoesNotHoldLockDuringPing(t *testing.T) { mr := miniredis.RunT(t) fastStart := time.Now() - if _, _, err := GetOrCreateRedisClient(&redis.Options{Addr: mr.Addr(), DB: 1}, 500*time.Millisecond); err != nil { + if _, _, err := GetOrCreate(&redis.Options{Addr: mr.Addr(), DB: 1}, 500*time.Millisecond); err != nil { t.Fatalf("unexpected error on the fast, unrelated key: %v", err) } if elapsed := time.Since(fastStart); elapsed > 300*time.Millisecond { @@ -237,10 +237,8 @@ func TestGetOrCreateClient_DoesNotHoldLockDuringPing(t *testing.T) { <-done // let the slow goroutine finish before the test exits } -// resetSharedForTest clears the package-level shared client state before a -// test runs (so InitFromConfig can be called again despite its once-only -// guard) and restores whatever was there before once the test ends. White-box -// access is fine here - this file is part of the package. +// resetSharedForTest clears shared so InitFromConfig can run again despite +// its once-only guard, restoring the prior state once the test ends. func resetSharedForTest(t *testing.T) { t.Helper() shared.mu.Lock() @@ -261,10 +259,9 @@ func TestResolveOptionsFromConfig_NoRedisSectionReturnsNil(t *testing.T) { } } -// TestResolveOptionsFromConfig_IgnoresSiblingSections proves resolveOptionsFromConfig -// looks at the top-level "redis" key only - other top-level sections (including -// policy_configurations, which is a separate, policy-engine-internal namespace -// this package deliberately does NOT nest under) have no bearing on it. +// TestResolveOptionsFromConfig_IgnoresSiblingSections proves it looks at the +// top-level "redis" key only - other sections, including policy_configurations, +// have no bearing on it. func TestResolveOptionsFromConfig_IgnoresSiblingSections(t *testing.T) { raw := map[string]interface{}{ "router": map[string]interface{}{"gateway_host": "*"}, @@ -323,12 +320,9 @@ func TestResolveOptionsFromConfig_ParsesConfiguredValues(t *testing.T) { } // TestResolveOptionsFromConfig_ParsesNumericStringPort locks in the shape -// gateway-runtime's own config interpolation actually produces: a TOML value -// written as {{ env "VAR" "6379" }} must be a quoted string literal (TOML has -// no unquoted template syntax), and interpolation resolves the token in place -// without ever changing the field's type - so a "numeric" config.toml value -// arrives here as a numeric string, not an int, even though int/int64/float64 -// are also accepted (e.g. from a JSON-sourced config path). +// config interpolation actually produces: a TOML {{ env "VAR" "6379" }} +// token resolves in place without changing the field's type, so a "numeric" +// value arrives as a numeric string, not an int. func TestResolveOptionsFromConfig_ParsesNumericStringPort(t *testing.T) { raw := map[string]interface{}{ "redis": map[string]interface{}{ @@ -373,10 +367,8 @@ func TestResolveOptionsFromConfig_RejectsWrongShapedValue(t *testing.T) { } // TestResolveOptionsFromConfig_RejectsInvalidFloat64Port locks in that a -// float64 "port" (e.g. from a JSON-sourced config path, decoded as float64 -// rather than int) is validated before conversion - -// NaN/Inf/fractional/out-of-range values must error rather than silently -// truncating or converting a NaN/Inf into undefined behavior. +// float64 "port" is validated before conversion - NaN/Inf/fractional/ +// out-of-range values must error, not silently truncate. func TestResolveOptionsFromConfig_RejectsInvalidFloat64Port(t *testing.T) { cases := []struct { name string @@ -447,7 +439,7 @@ func TestSharedBeforeInitFromConfigErrors(t *testing.T) { } // TestSharedReturnsIdenticalPointer is the actual "single instance" contract: -// not merely "these two configs happen to compare equal" (GetOrCreateRedisClient's +// not merely "these two configs happen to compare equal" (GetOrCreate's // dedup guarantee) but "there is exactly one gateway-level client, full stop." func TestSharedReturnsIdenticalPointer(t *testing.T) { resetSharedForTest(t) @@ -503,7 +495,7 @@ func TestResolve_NonNilOptsBypassesShared(t *testing.T) { } // TestResolve_NonNilOptsStillDedupes proves Resolve's override branch keeps -// GetOrCreateRedisClient's existing sharing behavior - two policies that both +// GetOrCreate's existing sharing behavior - two policies that both // explicitly override to the same config still get one pool between them, // not one pool each. func TestResolve_NonNilOptsStillDedupes(t *testing.T) { @@ -561,3 +553,93 @@ func TestHashPassword(t *testing.T) { t.Error("expected different passwords to hash differently") } } + +func TestExtractOverrideFromParams_NoHostReturnsNil(t *testing.T) { + if got := ExtractOverrideFromParams(map[string]interface{}{}); got != nil { + t.Fatalf("got %+v, want nil when redis.host is absent", got) + } + if got := ExtractOverrideFromParams(map[string]interface{}{"redis": map[string]interface{}{"port": 6380}}); got != nil { + t.Fatalf("got %+v, want nil when redis.host is absent even with sibling redis.* fields set", got) + } +} + +func TestExtractOverrideFromParams_NestedMapShape(t *testing.T) { + params := map[string]interface{}{ + "redis": map[string]interface{}{ + "host": "redis.example.com", + "port": 6380, + "username": "app", + "password": "secret", + "db": 2, + "poolSize": 10, + "connectionTimeout": "10s", + "readTimeout": "7s", + "writeTimeout": "7s", + }, + } + opts := ExtractOverrideFromParams(params) + if opts == nil { + t.Fatal("expected a non-nil override") + } + if opts.Addr != "redis.example.com:6380" || opts.Username != "app" || opts.Password != "secret" || + opts.DB != 2 || opts.PoolSize != 10 || opts.DialTimeout != 10*time.Second || + opts.ReadTimeout != 7*time.Second || opts.WriteTimeout != 7*time.Second { + t.Errorf("got %+v, did not match configured values", opts) + } +} + +func TestExtractOverrideFromParams_FlattenedKeyShape(t *testing.T) { + params := map[string]interface{}{ + "redis.host": "redis.example.com", + "redis.port": "6380", + "redis.db": "2", + } + opts := ExtractOverrideFromParams(params) + if opts == nil { + t.Fatal("expected a non-nil override") + } + if opts.Addr != "redis.example.com:6380" || opts.DB != 2 { + t.Errorf("got %+v, want port 6380 and db 2 parsed from a flattened-key, numeric-string shape", opts) + } +} + +func TestExtractOverrideFromParams_AppliesDefaults(t *testing.T) { + opts := ExtractOverrideFromParams(map[string]interface{}{"redis": map[string]interface{}{"host": "redis.example.com"}}) + if opts == nil { + t.Fatal("expected a non-nil override") + } + if opts.Addr != "redis.example.com:6379" || opts.DB != 0 || opts.PoolSize != 0 || + opts.DialTimeout != 5*time.Second || opts.ReadTimeout != 3*time.Second || opts.WriteTimeout != 3*time.Second { + t.Errorf("got %+v, want defaults applied for every field but host", opts) + } +} + +func TestExtractOverrideFromParams_BracketsIPv6Host(t *testing.T) { + opts := ExtractOverrideFromParams(map[string]interface{}{"redis": map[string]interface{}{"host": "::1", "port": 6380}}) + if opts == nil { + t.Fatal("expected a non-nil override") + } + if want := "[::1]:6380"; opts.Addr != want { + t.Errorf("got Addr %q, want %q", opts.Addr, want) + } +} + +func TestExtractOverrideFromParams_UnparseableValueFallsBackToDefaultSilently(t *testing.T) { + params := map[string]interface{}{ + "redis": map[string]interface{}{ + "host": "redis.example.com", + "port": "not-a-number", + "connectionTimeout": "not-a-duration", + }, + } + opts := ExtractOverrideFromParams(params) + if opts == nil { + t.Fatal("expected a non-nil override") + } + if opts.Addr != "redis.example.com:6379" { + t.Errorf("got Addr %q, want the default port 6379 silently applied for an unparseable port", opts.Addr) + } + if opts.DialTimeout != 5*time.Second { + t.Errorf("got DialTimeout %v, want the default 5s silently applied for an unparseable duration", opts.DialTimeout) + } +} From b63bff1348add715962d0c3a961ea780e2fed880 Mon Sep 17 00:00:00 2001 From: Thenujan Date: Thu, 20 Aug 2026 13:34:58 +0530 Subject: [PATCH 09/10] Add AI suggestions --- sdk/core/utils/redisclient/redisclient.go | 41 ++++++- .../utils/redisclient/redisclient_test.go | 113 +++++++++++++++++- 2 files changed, 152 insertions(+), 2 deletions(-) diff --git a/sdk/core/utils/redisclient/redisclient.go b/sdk/core/utils/redisclient/redisclient.go index c25c0d89c6..3e9c3cf94d 100644 --- a/sdk/core/utils/redisclient/redisclient.go +++ b/sdk/core/utils/redisclient/redisclient.go @@ -259,6 +259,9 @@ func resolveOptionsFromConfig(raw map[string]interface{}) (*redis.Options, error if err := decoder.Decode(m); err != nil { return nil, err } + if err := validateRedisSectionFields(fields); err != nil { + return nil, err + } return &redis.Options{ Addr: net.JoinHostPort(fields.Host, strconv.Itoa(fields.Port)), @@ -272,6 +275,39 @@ func resolveOptionsFromConfig(raw map[string]interface{}) (*redis.Options, error }, nil } +// minRedisTimeout is the smallest connection/read/write timeout accepted. +// go-redis treats a negative ReadTimeout/WriteTimeout (-1/-2) as "disable +// timeout enforcement entirely" rather than an error, and time.Duration's +// underlying kind is int64 (not caught by rejectNonIntegralFloatHookFunc's +// int-only guard), so a bare numeric config value like connection_timeout = 5 +// would otherwise decode as 5 nanoseconds. Rejecting anything below 1ms +// catches both cases with one check. +const minRedisTimeout = time.Millisecond + +// validateRedisSectionFields rejects decoded values that would otherwise +// silently produce an unsafe or unusable *redis.Options. +func validateRedisSectionFields(f redisSectionFields) error { + if f.Port < 1 || f.Port > 65535 { + return fmt.Errorf("port must be between 1 and 65535, got %d", f.Port) + } + if f.DB < 0 { + return fmt.Errorf("db must not be negative, got %d", f.DB) + } + if f.PoolSize < 0 { + return fmt.Errorf("pool_size must not be negative, got %d", f.PoolSize) + } + if f.ConnectionTimeout < minRedisTimeout { + return fmt.Errorf("connection_timeout must be at least %s, got %s", minRedisTimeout, f.ConnectionTimeout) + } + if f.ReadTimeout < minRedisTimeout { + return fmt.Errorf("read_timeout must be at least %s, got %s", minRedisTimeout, f.ReadTimeout) + } + if f.WriteTimeout < minRedisTimeout { + return fmt.Errorf("write_timeout must be at least %s, got %s", minRedisTimeout, f.WriteTimeout) + } + return nil +} + // rejectNonIntegralFloatHookFunc runs before mapstructure's own weakly-typed // float->int conversion, which is undefined/lossy for NaN, +-Inf, a // fractional value, or an out-of-range magnitude - reject those instead of @@ -360,7 +396,10 @@ func paramInt(params map[string]interface{}, dottedKey string, def int) int { case int64: return int(n) case float64: - return int(n) + if !math.IsNaN(n) && !math.IsInf(n, 0) && n == math.Trunc(n) && + n >= float64(math.MinInt) && n <= float64(math.MaxInt) { + return int(n) + } case string: if parsed, err := strconv.Atoi(strings.TrimSpace(n)); err == nil { return parsed diff --git a/sdk/core/utils/redisclient/redisclient_test.go b/sdk/core/utils/redisclient/redisclient_test.go index 0f912fec69..ad5a9a7096 100644 --- a/sdk/core/utils/redisclient/redisclient_test.go +++ b/sdk/core/utils/redisclient/redisclient_test.go @@ -33,6 +33,7 @@ import ( ) func TestGetOrCreateClient_SharesClientForIdenticalConfig(t *testing.T) { + resetRegistryForTest(t) mr := miniredis.RunT(t) opts := &redis.Options{Addr: mr.Addr(), DB: 0} @@ -51,6 +52,7 @@ func TestGetOrCreateClient_SharesClientForIdenticalConfig(t *testing.T) { } func TestGetOrCreateClient_DistinctClientForDifferentConfig(t *testing.T) { + resetRegistryForTest(t) mr := miniredis.RunT(t) c1, _, _ := GetOrCreate(&redis.Options{Addr: mr.Addr(), DB: 0}, time.Second) @@ -62,6 +64,7 @@ func TestGetOrCreateClient_DistinctClientForDifferentConfig(t *testing.T) { } func TestGetOrCreateClient_DifferentPasswordProducesDistinctClient(t *testing.T) { + resetRegistryForTest(t) mr := miniredis.RunT(t) c1, _, _ := GetOrCreate(&redis.Options{Addr: mr.Addr(), Password: "one"}, time.Second) @@ -77,6 +80,7 @@ func TestGetOrCreateClient_DifferentPasswordProducesDistinctClient(t *testing.T) } func TestGetOrCreateClient_SharedAcrossSimulatedPolicies(t *testing.T) { + resetRegistryForTest(t) mr := miniredis.RunT(t) opts := func() *redis.Options { return &redis.Options{Addr: mr.Addr(), DB: 0} } @@ -105,6 +109,7 @@ func TestGetOrCreateClient_SharedAcrossSimulatedPolicies(t *testing.T) { // TestGetOrCreateClient_ReuseSkipsPing locks in that only creation pings - // a reused client is assumed healthy and must never be re-pinged. func TestGetOrCreateClient_ReuseSkipsPing(t *testing.T) { + resetRegistryForTest(t) mr := miniredis.RunT(t) addr := mr.Addr() // capture before mr.Close() below opts := &redis.Options{Addr: addr, DB: 0} @@ -122,6 +127,7 @@ func TestGetOrCreateClient_ReuseSkipsPing(t *testing.T) { } func TestGetOrCreateClient_DifferentProtocolProducesDistinctClient(t *testing.T) { + resetRegistryForTest(t) mr := miniredis.RunT(t) c1, _, _ := GetOrCreate(&redis.Options{Addr: mr.Addr(), Protocol: 2}, time.Second) @@ -141,6 +147,7 @@ func TestGetOrCreateClient_DifferentProtocolProducesDistinctClient(t *testing.T) // options - neither it nor a credentials-provider func can be fingerprinted // safely, so sharing would risk a silent cross-config mixup. func TestGetOrCreateClient_TLSConfigBypassesRegistry(t *testing.T) { + resetRegistryForTest(t) mr := miniredis.RunT(t) optsA := &redis.Options{Addr: mr.Addr(), TLSConfig: &tls.Config{}} //nolint:gosec // test-only, no real handshake asserted @@ -158,6 +165,7 @@ func TestGetOrCreateClient_TLSConfigBypassesRegistry(t *testing.T) { } func TestGetOrCreateClient_CredentialsProviderBypassesRegistry(t *testing.T) { + resetRegistryForTest(t) mr := miniredis.RunT(t) provider := func() (string, string) { return "", "" } @@ -180,10 +188,12 @@ func TestGetOrCreateClient_CredentialsProviderBypassesRegistry(t *testing.T) { // holding it during a slow/unreachable Redis's ping would stall every other // caller too, even for an unrelated, healthy endpoint. func TestGetOrCreateClient_DoesNotHoldLockDuringPing(t *testing.T) { + resetRegistryForTest(t) // Accepts but never responds, so Ping against it blocks until the // deadline - a reliable window to prove a concurrent, unrelated key // isn't blocked by it. - ln, err := net.Listen("tcp", "127.0.0.1:0") + var lc net.ListenConfig + ln, err := lc.Listen(context.Background(), "tcp", "127.0.0.1:0") if err != nil { t.Fatalf("failed to start hanging listener: %v", err) } @@ -237,6 +247,24 @@ func TestGetOrCreateClient_DoesNotHoldLockDuringPing(t *testing.T) { <-done // let the slow goroutine finish before the test exits } +// resetRegistryForTest clears the process-wide redisClients registry so a +// closed miniredis instance's address (freed and possibly reused by a later +// miniredis.RunT in the same test binary) can never resolve to another +// test's stale, already-pinged client - restoring the prior state once the +// test ends. +func resetRegistryForTest(t *testing.T) { + t.Helper() + redisClients.mu.Lock() + prev := redisClients.m + redisClients.m = make(map[redisConnKey]*redis.Client) + redisClients.mu.Unlock() + t.Cleanup(func() { + redisClients.mu.Lock() + redisClients.m = prev + redisClients.mu.Unlock() + }) +} + // resetSharedForTest clears shared so InitFromConfig can run again despite // its once-only guard, restoring the prior state once the test ends. func resetSharedForTest(t *testing.T) { @@ -408,6 +436,59 @@ func TestResolveOptionsFromConfig_RejectsNonTableSection(t *testing.T) { } } +// TestResolveOptionsFromConfig_RejectsOutOfRangePort locks in that a port +// outside 1-65535 - a plain literal int, never routed through +// rejectNonIntegralFloatHookFunc since no type coercion is needed - errors +// instead of silently producing an unusable *redis.Options. +func TestResolveOptionsFromConfig_RejectsOutOfRangePort(t *testing.T) { + for _, port := range []int{0, -1, 65536} { + t.Run("", func(t *testing.T) { + _, err := resolveOptionsFromConfig(map[string]interface{}{"redis": map[string]interface{}{"host": "x", "port": port}}) + if err == nil { + t.Errorf("expected an error for out-of-range port %d", port) + } + }) + } +} + +// TestResolveOptionsFromConfig_RejectsNegativeDBOrPoolSize locks in the same +// gap as the port check above, for db/pool_size. +func TestResolveOptionsFromConfig_RejectsNegativeDBOrPoolSize(t *testing.T) { + if _, err := resolveOptionsFromConfig(map[string]interface{}{"redis": map[string]interface{}{"host": "x", "db": -1}}); err == nil { + t.Error("expected an error for a negative db") + } + if _, err := resolveOptionsFromConfig(map[string]interface{}{"redis": map[string]interface{}{"host": "x", "pool_size": -1}}); err == nil { + t.Error("expected an error for a negative pool_size") + } +} + +// TestResolveOptionsFromConfig_RejectsSubMillisecondTimeout locks in that a +// timeout below 1ms - including a negative value, which go-redis would +// otherwise silently treat as "disable timeout enforcement entirely" - errors +// rather than producing an unsafe client. Also catches a bare numeric +// duration value (e.g. connection_timeout = 5), which mapstructure decodes +// as nanoseconds since time.Duration's underlying kind is int64, not caught +// by rejectNonIntegralFloatHookFunc's int-only guard. +func TestResolveOptionsFromConfig_RejectsSubMillisecondTimeout(t *testing.T) { + cases := []struct { + name string + key string + val interface{} + }{ + {"negative connection_timeout", "connection_timeout", "-1s"}, + {"negative read_timeout", "read_timeout", "-2s"}, + {"bare numeric write_timeout decodes as nanoseconds", "write_timeout", 5}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + _, err := resolveOptionsFromConfig(map[string]interface{}{"redis": map[string]interface{}{"host": "x", c.key: c.val}}) + if err == nil { + t.Errorf("expected an error for %s=%v", c.key, c.val) + } + }) + } +} + func TestInitFromConfig_NoRedisSectionLeavesSharedUnconfigured(t *testing.T) { resetSharedForTest(t) @@ -481,6 +562,7 @@ func TestResolve_NilOptsFallsBackToShared(t *testing.T) { func TestResolve_NonNilOptsBypassesShared(t *testing.T) { resetSharedForTest(t) + resetRegistryForTest(t) sharedMR := miniredis.RunT(t) SetSharedForTesting(t, redis.NewClient(&redis.Options{Addr: sharedMR.Addr()})) @@ -500,6 +582,7 @@ func TestResolve_NonNilOptsBypassesShared(t *testing.T) { // not one pool each. func TestResolve_NonNilOptsStillDedupes(t *testing.T) { resetSharedForTest(t) + resetRegistryForTest(t) mr := miniredis.RunT(t) c1, err1 := Resolve(&redis.Options{Addr: mr.Addr()}, time.Second) @@ -643,3 +726,31 @@ func TestExtractOverrideFromParams_UnparseableValueFallsBackToDefaultSilently(t t.Errorf("got DialTimeout %v, want the default 5s silently applied for an unparseable duration", opts.DialTimeout) } } + +// TestExtractOverrideFromParams_RejectsMalformedFloat64Port locks in that +// paramInt validates a float64 before converting it - a NaN/Inf/fractional/ +// out-of-range port (e.g. from a JSON-sourced params map, decoded as +// float64) falls back to the default rather than int(n)'s +// implementation-defined result. +func TestExtractOverrideFromParams_RejectsMalformedFloat64Port(t *testing.T) { + cases := []struct { + name string + port float64 + }{ + {"NaN", math.NaN()}, + {"+Inf", math.Inf(1)}, + {"-Inf", math.Inf(-1)}, + {"fractional", 6380.5}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + opts := ExtractOverrideFromParams(map[string]interface{}{"redis": map[string]interface{}{"host": "x", "port": c.port}}) + if opts == nil { + t.Fatal("expected a non-nil override") + } + if opts.Addr != "x:6379" { + t.Errorf("got Addr %q, want the default port 6379 silently applied for malformed float64 port %v", opts.Addr, c.port) + } + }) + } +} From 1ceb8697ea78c05aac0a8102106d479355276f65 Mon Sep 17 00:00:00 2001 From: Thenujan Date: Thu, 20 Aug 2026 14:08:54 +0530 Subject: [PATCH 10/10] Validate Redis host field and add test for blank host handling --- sdk/core/utils/redisclient/redisclient.go | 3 +++ sdk/core/utils/redisclient/redisclient_test.go | 14 ++++++++++++++ 2 files changed, 17 insertions(+) diff --git a/sdk/core/utils/redisclient/redisclient.go b/sdk/core/utils/redisclient/redisclient.go index 3e9c3cf94d..77076f0882 100644 --- a/sdk/core/utils/redisclient/redisclient.go +++ b/sdk/core/utils/redisclient/redisclient.go @@ -287,6 +287,9 @@ const minRedisTimeout = time.Millisecond // validateRedisSectionFields rejects decoded values that would otherwise // silently produce an unsafe or unusable *redis.Options. func validateRedisSectionFields(f redisSectionFields) error { + if strings.TrimSpace(f.Host) == "" { + return fmt.Errorf("host must not be blank") + } if f.Port < 1 || f.Port > 65535 { return fmt.Errorf("port must be between 1 and 65535, got %d", f.Port) } diff --git a/sdk/core/utils/redisclient/redisclient_test.go b/sdk/core/utils/redisclient/redisclient_test.go index ad5a9a7096..1f215b28aa 100644 --- a/sdk/core/utils/redisclient/redisclient_test.go +++ b/sdk/core/utils/redisclient/redisclient_test.go @@ -462,6 +462,20 @@ func TestResolveOptionsFromConfig_RejectsNegativeDBOrPoolSize(t *testing.T) { } } +// TestResolveOptionsFromConfig_RejectsBlankHost locks in that an empty or +// whitespace-only host - which would otherwise silently produce a +// ""-only *redis.Options via net.JoinHostPort - errors instead. +func TestResolveOptionsFromConfig_RejectsBlankHost(t *testing.T) { + for _, host := range []string{"", " "} { + t.Run("", func(t *testing.T) { + _, err := resolveOptionsFromConfig(map[string]interface{}{"redis": map[string]interface{}{"host": host}}) + if err == nil { + t.Errorf("expected an error for blank host %q", host) + } + }) + } +} + // TestResolveOptionsFromConfig_RejectsSubMillisecondTimeout locks in that a // timeout below 1ms - including a negative value, which go-redis would // otherwise silently treat as "disable timeout enforcement entirely" - errors