mirror of
https://github.com/hanzoai/ingress.git
synced 2026-10-30 14:58:23 +00:00
zeekayandHanzo Dev
8c0c819f02
hanzokv/go v9.22.0 renames its Go package from `redis` to `kv`; the import
path is unchanged. Update the call sites and rename the middleware's own
client vocabulary to match:
redis_limiter.go -> kv_limiter.go
redisLimiter -> kvLimiter
newRedisLimiter -> newKVLimiter
redisPrefix -> kvPrefix (key prefix value "rate:" is unchanged)
Rediser -> KVClient
mockRedisClient -> mockKVClient
Three things deliberately keep the redis name because they are not ours
to rename:
- `redis.call(...)` inside AllowTokenBucketRaw, and the matching
state.SetGlobal("redis", mod) in the test's Lua VM. That is the KV
server's scripting API — a wire contract, not vocabulary.
- dynamic.Redis / config.Redis. That field is CRD schema: it has
generated deepcopy and applyconfiguration code and it is what a
Middleware CR spells today. Renaming it breaks every deployed CR and
needs a codegen pass, so it belongs in its own change.
- pkg/provider/kv/redis rides github.com/kvtools/redis, a foreign
client, and never touched hanzokv/go.
Verified with go vet ./pkg/middlewares/ratelimiter/... (clean) and
CGO_ENABLED=0 go test ./pkg/middlewares/ratelimiter/... (ok, 121s). A
full `go build ./...` cannot run on darwin/arm64 here: luxfi/accel's cgo
fails with 'stdlib.h' file not found, and linking the test binary fails
with ld: library 'resolv' not found. Both are local toolchain gaps that
predate this change.
Co-authored-by: Hanzo Dev <[email protected]>
119 lines
3.0 KiB
Go
119 lines
3.0 KiB
Go
package ratelimiter
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/hanzokv/go/v9"
|
|
"github.com/rs/zerolog"
|
|
ptypes "github.com/hanzoai/ingress-parser/types"
|
|
"github.com/hanzoai/ingress/pkg/config/dynamic"
|
|
"golang.org/x/time/rate"
|
|
)
|
|
|
|
const kvPrefix = "rate:"
|
|
|
|
type kvLimiter struct {
|
|
rate rate.Limit // reqs/s
|
|
burst int64
|
|
maxDelay time.Duration
|
|
period ptypes.Duration
|
|
logger *zerolog.Logger
|
|
ttl int
|
|
client KVClient
|
|
}
|
|
|
|
func newKVLimiter(ctx context.Context, rate rate.Limit, burst int64, maxDelay time.Duration, ttl int, config dynamic.RateLimit, logger *zerolog.Logger) (limiter, error) {
|
|
options := &kv.UniversalOptions{
|
|
Addrs: config.Redis.Endpoints,
|
|
Username: config.Redis.Username,
|
|
Password: config.Redis.Password,
|
|
DB: config.Redis.DB,
|
|
PoolSize: config.Redis.PoolSize,
|
|
MinIdleConns: config.Redis.MinIdleConns,
|
|
MaxActiveConns: config.Redis.MaxActiveConns,
|
|
}
|
|
|
|
if config.Redis.DialTimeout != nil && *config.Redis.DialTimeout > 0 {
|
|
options.DialTimeout = time.Duration(*config.Redis.DialTimeout)
|
|
}
|
|
|
|
if config.Redis.ReadTimeout != nil {
|
|
if *config.Redis.ReadTimeout > 0 {
|
|
options.ReadTimeout = time.Duration(*config.Redis.ReadTimeout)
|
|
} else {
|
|
options.ReadTimeout = -1
|
|
}
|
|
}
|
|
|
|
if config.Redis.WriteTimeout != nil {
|
|
if *config.Redis.ReadTimeout > 0 {
|
|
options.WriteTimeout = time.Duration(*config.Redis.WriteTimeout)
|
|
} else {
|
|
options.WriteTimeout = -1
|
|
}
|
|
}
|
|
|
|
if config.Redis.TLS != nil {
|
|
var err error
|
|
options.TLSConfig, err = config.Redis.TLS.CreateTLSConfig(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("creating TLS config: %w", err)
|
|
}
|
|
}
|
|
|
|
return &kvLimiter{
|
|
rate: rate,
|
|
burst: burst,
|
|
period: config.Period,
|
|
maxDelay: maxDelay,
|
|
logger: logger,
|
|
ttl: ttl,
|
|
client: kv.NewUniversalClient(options),
|
|
}, nil
|
|
}
|
|
|
|
func (r *kvLimiter) Allow(ctx context.Context, source string) (*time.Duration, error) {
|
|
ok, delay, err := r.evaluateScript(ctx, source)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("evaluating script: %w", err)
|
|
}
|
|
if !ok {
|
|
return nil, nil
|
|
}
|
|
return delay, nil
|
|
}
|
|
|
|
func (r *kvLimiter) evaluateScript(ctx context.Context, key string) (bool, *time.Duration, error) {
|
|
if r.rate == rate.Inf {
|
|
return true, nil, nil
|
|
}
|
|
|
|
params := []any{
|
|
float64(r.rate / 1000000),
|
|
r.burst,
|
|
r.ttl,
|
|
time.Now().UnixMicro(),
|
|
r.maxDelay.Microseconds(),
|
|
}
|
|
v, err := AllowTokenBucketScript.Run(ctx, r.client, []string{kvPrefix + key}, params...).Result()
|
|
if err != nil {
|
|
return false, nil, fmt.Errorf("running script: %w", err)
|
|
}
|
|
|
|
values := v.([]any)
|
|
ok, err := strconv.ParseBool(values[0].(string))
|
|
if err != nil {
|
|
return false, nil, fmt.Errorf("parsing ok value from kv rate lua script: %w", err)
|
|
}
|
|
delay, err := strconv.ParseFloat(values[1].(string), 64)
|
|
if err != nil {
|
|
return false, nil, fmt.Errorf("parsing delay value from kv rate lua script: %w", err)
|
|
}
|
|
|
|
microDelay := time.Duration(delay * float64(time.Microsecond))
|
|
return ok, µDelay, nil
|
|
}
|