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..ce9263f79a 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,12 @@ 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) + 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/go.mod b/sdk/core/go.mod index fa1ad4c5b9..6bf7ce1c35 100644 --- a/sdk/core/go.mod +++ b/sdk/core/go.mod @@ -1,3 +1,16 @@ module github.com/wso2/api-platform/sdk/core 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 +) + +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..9013c6e309 --- /dev/null +++ b/sdk/core/go.sum @@ -0,0 +1,28 @@ +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/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= +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..77076f0882 --- /dev/null +++ b/sdk/core/utils/redisclient/redisclient.go @@ -0,0 +1,424 @@ +/* + * 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 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 ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "math" + "net" + "reflect" + "strconv" + "strings" + "sync" + "time" + + "github.com/go-viper/mapstructure/v2" + "github.com/redis/go-redis/v9" +) + +// 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 + 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 + poolSize int +} + +// 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 +}{m: make(map[redisConnKey]*redis.Client)} + +func hashRedisPassword(p string) string { + if p == "" { + return "" + } + sum := sha256.Sum256([]byte(p)) + return hex.EncodeToString(sum[:]) +} + +// 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) + } + + 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 - 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() + return c, false, nil + } + c := redis.NewClient(opts) + redisClients.m[key] = c + redisClients.mu.Unlock() + + pingErr = pingClient(c, pingTimeout) + return c, true, pingErr +} + +// 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 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, 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 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() + 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, pingTimeoutFor(opts)) + shared.client = c + return nil +} + +// Shared returns the process-wide gateway-level default client, backed by +// 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() + 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 +// 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, _, _ := GetOrCreate(opts, pingTimeout) + return client, nil +} + +// newAndPingClient creates a client and pings it once. created is always +// 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) + return c, true, pingClient(c, pingTimeout) +} + +// pingClient pings an already-constructed client once, bounded by +// 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() + return c.Ping(ctx).Err() +} + +// redisSectionFields is the decode target for the "redis" table, pre-filled +// 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"` + 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, not nested under "policy_configurations" (this is +// gateway-wide infrastructure, not a per-policy setting). Returns (nil, nil) +// 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) + } + + fields := redisSectionFields{ + Host: "localhost", + Port: 6379, + ConnectionTimeout: 5 * time.Second, + ReadTimeout: 3 * time.Second, + WriteTimeout: 3 * time.Second, + } + decoder, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{ + WeaklyTypedInput: true, + DecodeHook: mapstructure.ComposeDecodeHookFunc( + rejectNonIntegralFloatHookFunc, + mapstructure.StringToTimeDurationHookFunc(), + ), + Result: &fields, + }) + if err != nil { + return nil, fmt.Errorf("building redis config decoder: %w", err) + } + 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)), + Username: fields.Username, + Password: fields.Password, + DB: fields.DB, + DialTimeout: fields.ConnectionTimeout, + ReadTimeout: fields.ReadTimeout, + WriteTimeout: fields.WriteTimeout, + PoolSize: fields.PoolSize, + }, 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 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) + } + 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 +// 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 + } + f, ok := data.(float64) + if !ok { + return data, nil + } + if math.IsNaN(f) || math.IsInf(f, 0) { + return nil, fmt.Errorf("expected an integer, got %v", f) + } + if f != math.Trunc(f) { + return nil, fmt.Errorf("expected an integer, got non-integer value %v", f) + } + if f < float64(math.MinInt) || f > float64(math.MaxInt) { + return nil, fmt.Errorf("value %v out of range for int", f) + } + 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: + 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 + } + } + } + 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 new file mode 100644 index 0000000000..1f215b28aa --- /dev/null +++ b/sdk/core/utils/redisclient/redisclient_test.go @@ -0,0 +1,770 @@ +/* + * 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" + "crypto/tls" + "math" + "net" + "strconv" + "sync" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" +) + +func TestGetOrCreateClient_SharesClientForIdenticalConfig(t *testing.T) { + resetRegistryForTest(t) + mr := miniredis.RunT(t) + opts := &redis.Options{Addr: mr.Addr(), DB: 0} + + 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 := GetOrCreate(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) { + resetRegistryForTest(t) + mr := miniredis.RunT(t) + + 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") + } +} + +func TestGetOrCreateClient_DifferentPasswordProducesDistinctClient(t *testing.T) { + resetRegistryForTest(t) + mr := miniredis.RunT(t) + + 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") + } + if c1 == c3 || c2 == c3 { + t.Error("expected an absent password not to collide with a present one") + } +} + +func TestGetOrCreateClient_SharedAcrossSimulatedPolicies(t *testing.T) { + resetRegistryForTest(t) + mr := miniredis.RunT(t) + opts := func() *redis.Options { return &redis.Options{Addr: mr.Addr(), DB: 0} } + + // Two distinct call sites with identical settings must share one + // client - the whole point of centralizing the registry. + 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") + } + + 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 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} + + 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 := 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) + } +} + +func TestGetOrCreateClient_DifferentProtocolProducesDistinctClient(t *testing.T) { + resetRegistryForTest(t) + mr := miniredis.RunT(t) + + 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") + } + 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) { + resetRegistryForTest(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, _ := 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) + } + 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) { + resetRegistryForTest(t) + mr := miniredis.RunT(t) + provider := func() (string, string) { return "", "" } + + 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) + } + 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) { + 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. + 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) + } + 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 + } + }() + + 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. + _, _, _ = GetOrCreate(&redis.Options{ + Addr: ln.Addr().String(), + DB: 0, + ReadTimeout: time.Second, + WriteTimeout: time.Second, + }, time.Second) + }() + + 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() + 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 { + 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 +} + +// 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) { + 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 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": "*"}, + "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 +// 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{}{ + "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) + } +} + +// 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"}, + }) + if err == nil { + t.Error("expected an error for a non-numeric port, so a config typo surfaces at startup instead of silently defaulting") + } +} + +// TestResolveOptionsFromConfig_RejectsInvalidFloat64Port locks in that a +// 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 + 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 := resolveOptionsFromConfig(map[string]interface{}{"redis": map[string]interface{}{"port": c.v}}) + if err == nil { + t.Errorf("expected an error for float64 port %v, got nil", c.v) + } + }) + } +} + +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 want := "localhost:6380"; opts.Addr != want { + t.Errorf("got Addr %q, want %q", opts.Addr, want) + } +} + +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") + } +} + +// 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_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 +// 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) + + 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" (GetOrCreate'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) + resetRegistryForTest(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 +// 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) { + resetSharedForTest(t) + resetRegistryForTest(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(\"\")") + } + if hashRedisPassword("secret") == "secret" { + t.Error("expected the password to actually be hashed, not passed through") + } + 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") + } +} + +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) + } +} + +// 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) + } + }) + } +} 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() + }) +}