Refactor: extract Valkey Del command and ratelimit ring client injection

net/valkey.go

@@ -277,6 +277,11 @@ func (vr *valkeyRing) Expire(ctx context.Context, key string, expire time.Durati
 	return shard.Do(ctx, shard.B().Expire().Key(key).Seconds(int64(expire.Seconds())).Build())
 }
 
+func (vr *valkeyRing) Del(ctx context.Context, key string) valkey.ValkeyResult {
+	shard := vr.shardForKey(key)
+	return shard.Do(ctx, shard.B().Del().Key(key).Build())
+}
+
 func (vr *valkeyRing) Get(ctx context.Context, key string) valkey.ValkeyResult {
 	shard := vr.shardForKey(key)
 	return shard.Do(ctx, shard.B().Get().Key(key).Build())
@@ -520,6 +525,11 @@ func (vrc *ValkeyRingClient) Expire(ctx context.Context, key string, d time.Dura
 	return res.ToInt64()
 }
 
+func (vrc *ValkeyRingClient) Del(ctx context.Context, key string) (int64, error) {
+	res := vrc.ring.Del(ctx, key)
+	return res.ToInt64()
+}
+
 func (vrc *ValkeyRingClient) Get(ctx context.Context, key string) (string, error) {
 	res := vrc.ring.Get(ctx, key)
 	return res.ToString()
@@ -532,15 +542,15 @@ func (vrc *ValkeyRingClient) Set(ctx context.Context, key, val string) (string,
 
 func (vrc *ValkeyRingClient) SetWithExpire(ctx context.Context, key string, value string, expire time.Duration) error {
 	results := vrc.ring.SetWithExpire(ctx, key, value, expire)
+	if len(results) == 0 {
+		return fmt.Errorf("failed to SetWithExpire, no result")
+	}
 	for _, res := range results {
 		if err := res.Error(); err != nil {
 			return err
 		}
 	}
-	if len(results) == 0 {
-		return fmt.Errorf("failed to SetWithExpire, no result")
-	}
-	return results[len(results)-1].Error()
+	return nil
 }
 
 func (vrc *ValkeyRingClient) ZAdd(ctx context.Context, key, val string, score float64) (int64, error) {

ratelimit/redis.go

@@ -28,11 +28,11 @@ type clusterLimitRedis struct {
 }
 
 const (
-	redisMetricsPrefix                    = "swarm.redis."
-	redisAllowMetricsFormat               = redisMetricsPrefix + "query.allow.%s"
-	redisRetryAfterMetricsFormat          = redisMetricsPrefix + "query.retryafter.%s"
-	redisAllowMetricsFormatWithGroup      = redisMetricsPrefix + "query.allow.%s.%s"
-	redisRetryAfterMetricsFormatWithGroup = redisMetricsPrefix + "query.retryafter.%s.%s"
+	RedisMetricsPrefix                    = "swarm.redis."
+	redisAllowMetricsFormat               = RedisMetricsPrefix + "query.allow.%s"
+	redisRetryAfterMetricsFormat          = RedisMetricsPrefix + "query.retryafter.%s"
+	redisAllowMetricsFormatWithGroup      = RedisMetricsPrefix + "query.allow.%s.%s"
+	redisRetryAfterMetricsFormatWithGroup = RedisMetricsPrefix + "query.retryafter.%s.%s"
 
 	redisAllowSpanName       = "redis_allow"
 	redisOldestScoreSpanName = "redis_oldest_score"
@@ -111,7 +111,7 @@ func (c *clusterLimitRedis) commonTags() opentracing.Tags {
 //
 // Uses provided context for creating an OpenTracing span.
 func (c *clusterLimitRedis) Allow(ctx context.Context, clearText string) bool {
-	c.metrics.IncCounter(redisMetricsPrefix + "total")
+	c.metrics.IncCounter(RedisMetricsPrefix + "total")
 	now := time.Now()
 
 	var span opentracing.Span
@@ -135,9 +135,9 @@ func (c *clusterLimitRedis) Allow(ctx context.Context, clearText string) bool {
 	c.measureQuery(redisAllowMetricsFormat, redisAllowMetricsFormatWithGroup, &failed, now)
 
 	if allow {
-		c.metrics.IncCounter(redisMetricsPrefix + "allows")
+		c.metrics.IncCounter(RedisMetricsPrefix + "allows")
 	} else {
-		c.metrics.IncCounter(redisMetricsPrefix + "forbids")
+		c.metrics.IncCounter(RedisMetricsPrefix + "forbids")
 	}
 	return allow
 }

ratelimit/registry.go

@@ -84,17 +84,38 @@ func NewSwimSwarmRegistry(swarm Swarmer, settings ...Settings) *Registry {
 // the optional provided default settings.
 func NewRedisSwarmRegistry(ro *net.RedisOptions, settings ...Settings) *Registry {
 	if ro != nil && ro.MetricsPrefix == "" {
-		ro.MetricsPrefix = redisMetricsPrefix
+		ro.MetricsPrefix = RedisMetricsPrefix
 	}
 
+	return NewRatelimitRegistryRedis(net.NewRedisRingClient(ro), settings...)
+}
+
+// NewValkeySwarmRegistry initializes a registry with Valkey shards
+// and the optional provided default settings.
+func NewValkeySwarmRegistry(vo *net.ValkeyOptions, settings ...Settings) (*Registry, error) {
+	if vo != nil && vo.MetricsPrefix == "" {
+		vo.MetricsPrefix = ValkeyMetricsPrefix
+	}
+
+	ring, err := net.NewValkeyRingClient(vo)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewRatelimitRegistryValkey(ring, settings...), nil
+}
+
+// NewRatelimitRegistryRedis creates a registry for the given redis ring client.
+func NewRatelimitRegistryRedis(ring *net.RedisRingClient, settings ...Settings) *Registry {
 	r := &Registry{
 		once:      sync.Once{},
 		global:    getSwarmRegistryDefaultSettings(),
 		lookup:    make(map[Settings]*Ratelimit),
-		redisRing: net.NewRedisRingClient(ro),
+		redisRing: ring,
 	}
-	if ro != nil {
-		r.redisRing.StartMetricsCollection()
+
+	if ring != nil {
+		ring.StartMetricsCollection()
 	}
 
 	if len(settings) > 0 {
@@ -104,18 +125,8 @@ func NewRedisSwarmRegistry(ro *net.RedisOptions, settings ...Settings) *Registry
 	return r
 }
 
-// NewValkeySwarmRegistry initializes a registry with Valkey shards
-// and the optional provided default settings.
-func NewValkeySwarmRegistry(vo *net.ValkeyOptions, settings ...Settings) (*Registry, error) {
-	if vo != nil && vo.MetricsPrefix == "" {
-		vo.MetricsPrefix = valkeyMetricsPrefix
-	}
-
-	ring, err := net.NewValkeyRingClient(vo)
-	if err != nil {
-		return nil, err
-	}
-
+// NewRatelimitRegistryValkey creates a registry for the given valkey ring client.
+func NewRatelimitRegistryValkey(ring *net.ValkeyRingClient, settings ...Settings) *Registry {
 	r := &Registry{
 		once:       sync.Once{},
 		global:     getSwarmRegistryDefaultSettings(),
@@ -127,7 +138,7 @@ func NewValkeySwarmRegistry(vo *net.ValkeyOptions, settings ...Settings) (*Regis
 		r.global = settings[0]
 	}
 
-	return r, nil
+	return r
 }
 
 // Close teardown Registry and dependent resources

ratelimit/valkey.go

@@ -28,11 +28,11 @@ type clusterLimitValkey struct {
 }
 
 const (
-	valkeyMetricsPrefix                    = "swarm.valkey."
-	valkeyAllowMetricsFormat               = valkeyMetricsPrefix + "query.allow.%s"
-	valkeyRetryAfterMetricsFormat          = valkeyMetricsPrefix + "query.retryafter.%s"
-	valkeyAllowMetricsFormatWithGroup      = valkeyMetricsPrefix + "query.allow.%s.%s"
-	valkeyRetryAfterMetricsFormatWithGroup = valkeyMetricsPrefix + "query.retryafter.%s.%s"
+	ValkeyMetricsPrefix                    = "swarm.valkey."
+	valkeyAllowMetricsFormat               = ValkeyMetricsPrefix + "query.allow.%s"
+	valkeyRetryAfterMetricsFormat          = ValkeyMetricsPrefix + "query.retryafter.%s"
+	valkeyAllowMetricsFormatWithGroup      = ValkeyMetricsPrefix + "query.allow.%s.%s"
+	valkeyRetryAfterMetricsFormatWithGroup = ValkeyMetricsPrefix + "query.retryafter.%s.%s"
 
 	valkeyAllowSpanName       = "valkey_allow"
 	valkeyOldestScoreSpanName = "valkey_oldest_score"
@@ -111,7 +111,7 @@ func (c *clusterLimitValkey) commonTags() opentracing.Tags {
 //
 // Uses provided context for creating an OpenTracing span.
 func (c *clusterLimitValkey) Allow(ctx context.Context, clearText string) bool {
-	c.metrics.IncCounter(valkeyMetricsPrefix + "total")
+	c.metrics.IncCounter(ValkeyMetricsPrefix + "total")
 	now := time.Now()
 
 	var span opentracing.Span
@@ -135,9 +135,9 @@ func (c *clusterLimitValkey) Allow(ctx context.Context, clearText string) bool {
 	c.measureQuery(valkeyAllowMetricsFormat, valkeyAllowMetricsFormatWithGroup, &failed, now)
 
 	if allow {
-		c.metrics.IncCounter(valkeyMetricsPrefix + "allows")
+		c.metrics.IncCounter(ValkeyMetricsPrefix + "allows")
 	} else {
-		c.metrics.IncCounter(valkeyMetricsPrefix + "forbids") // TODO(sszuecs) forbids or better deny?
+		c.metrics.IncCounter(ValkeyMetricsPrefix + "forbids") // TODO(sszuecs) forbids or better deny?
 	}
 	return allow
 }

skipper.go

@@ -2181,11 +2181,41 @@ func run(o Options, sig chan os.Signal, idleConnsCH chan struct{}) error {
 		}
 	}
 
-	var ratelimitRegistry *ratelimit.Registry
-	var failClosedRatelimitPostProcessor *ratelimitfilters.FailClosedPostProcessor
+	var valkeyRing *skpnet.ValkeyRingClient
+	if valkeyOptions != nil {
+		if valkeyOptions.MetricsPrefix == "" {
+			valkeyOptions.MetricsPrefix = ratelimit.ValkeyMetricsPrefix
+		}
+		valkeyRing, err = skpnet.NewValkeyRingClient(valkeyOptions)
+		if err != nil {
+			return err
+		}
+		defer valkeyRing.Close()
+	}
+
+	var (
+		ratelimitRegistry                *ratelimit.Registry
+		failClosedRatelimitPostProcessor *ratelimitfilters.FailClosedPostProcessor
+	)
 	if o.EnableRatelimiters || len(o.RatelimitSettings) > 0 {
 		log.Infof("enabled ratelimiters %v: %v", o.EnableRatelimiters, o.RatelimitSettings)
-		ratelimitRegistry = ratelimit.NewSwarmRegistry(swarmer, redisOptions, valkeyOptions, o.RatelimitSettings...)
+
+		switch {
+		case valkeyRing != nil:
+			ratelimitRegistry = ratelimit.NewRatelimitRegistryValkey(valkeyRing, o.RatelimitSettings...)
+
+		case redisOptions != nil:
+			if redisOptions.MetricsPrefix == "" {
+				redisOptions.MetricsPrefix = ratelimit.RedisMetricsPrefix
+			}
+			redisRing := skpnet.NewRedisRingClient(redisOptions)
+			ratelimitRegistry = ratelimit.NewRatelimitRegistryRedis(redisRing, o.RatelimitSettings...)
+			defer redisRing.Close()
+
+		default:
+			// swim based Swarmer (deprecated)
+			ratelimitRegistry = ratelimit.NewSwarmRegistry(swarmer, redisOptions, valkeyOptions, o.RatelimitSettings...)
+		}
 		defer ratelimitRegistry.Close()
 
 		if hook := o.SwarmRegistry; hook != nil {