feat: cluster key rate limit (#1002)

2026-02-23 12:10:53 +08:00 · 2024-06-12 14:51:46 +08:00
parent d53c713561
commit 12cc44b324
6 changed files with 662 additions and 0 deletions
--- a/plugins/wasm-go/extensions/cluster-key-rate-limit/README.md
+++ b/plugins/wasm-go/extensions/cluster-key-rate-limit/README.md
@@ -0,0 +1,133 @@
+# 功能说明
+
+`key-cluster-rate-limit`插件实现了基于特定键值实现集群限流，键值来源可以是 URL 参数、HTTP 请求头、客户端 IP 地址
+
+
+
+# 配置说明
+
+| 配置项                  | 类型   | 必填 | 默认值 | 说明 |
+| ----------------------- | ------ | ---- | ------ | ---- |
+| rule_name               | string | 是 | - | 限流规则名称，根据限流规则名称和限流的客户端IP段来拼装redis key |
+| limit_by_header         | string          | 否，`limit_by_header`,`limit_by_param`,`limit_by_per_ip` 中选填一项 | -                 | 配置获取限流键值的来源 http 请求头名称                       |
+| limit_by_param          | string          | 否，`limit_by_header`,`limit_by_param`,`limit_by_per_ip` 中选填一项 | -                 | 配置获取限流键值的来源 URL 参数名称                          |
+| limit_by_per_ip         | string          | 否，`limit_by_header`,`limit_by_param`,`limit_by_per_ip` 中选填一项 | -                 | 配置获取限流键值的来源 IP 参数名称，从请求头获取，以`from-header-对应的header名`，示例：`from-header-x-forwarded-for`，直接获取对端socket ip，配置为`from-remote-addr` |
+| limit_keys              | array of object | 是 | - | 配置匹配键值后的限流次数 |
+| show_limit_quota_header | bool | 否 | false | 响应头中是否显示`X-RateLimit-Limit`（限制的总请求数）和`X-RateLimit-Remaining`（剩余还可以发送的请求数） |
+| rejected_code           | int | 否 | 429 | 请求被限流时，返回的HTTP状态码 |
+| rejected_msg            | string | 否 | Too many requests | 请求被限流时，返回的响应体 |
+| redis                   | object          | 是                                                           | -                 | redis相关配置                                                |
+
+`limit_keys`中每一项的配置字段说明
+
+| 配置项           | 类型   | 必填                                                         | 默认值 | 说明               |
+| ---------------- | ------ | ------------------------------------------------------------ | ------ | ------------------ |
+| key              | string | 是                                                           | -      | 匹配的键值         |
+| query_per_second | int    | 否，`query_per_second`,`query_per_minute`,`query_per_hour`,`query_per_day` 中选填一项 | -      | 允许每秒请求次数   |
+| query_per_minute | int    | 否，`query_per_second`,`query_per_minute`,`query_per_hour`,`query_per_day` 中选填一项 | -      | 允许每分钟请求次数 |
+| query_per_hour   | int    | 否，`query_per_second`,`query_per_minute`,`query_per_hour`,`query_per_day` 中选填一项 | -      | 允许每小时请求次数 |
+| query_per_day    | int    | 否，`query_per_second`,`query_per_minute`,`query_per_hour`,`query_per_day` 中选填一项 | -      | 允许每天请求次数   |
+
+`redis`中每一项的配置字段说明
+
+| 配置项       | 类型   | 必填 | 默认值                                                     | 说明                        |
+| ------------ | ------ | ---- | ---------------------------------------------------------- | --------------------------- |
+| service_name | string | 必填 | -                                                          | 输入redis服务的注册名称     |
+| service_port | int    | 否   | 服务类型为固定地址（static service）默认值为80，其他为6379 | 输入redis服务的服务端口     |
+| username     | string | 否   | -                                                          | redis用户名                 |
+| password     | string | 否   | -                                                          | redis密码                   |
+| timeout      | int    | 否   | 1000                                                       | redis连接超时时间，单位毫秒 |
+
+
+
+# 配置示例
+
+## 识别请求参数 apikey，进行区别限流
+```yaml
+rule_name: limit_by_param_apikey
+limit_by_param: apikey
+limit_keys:
+- key: 9a342114-ba8a-11ec-b1bf-00163e1250b5
+  query_per_second: 10
+- key: a6a6d7f2-ba8a-11ec-bec2-00163e1250b5
+  query_per_minute: 100
+redis:
+  service_name: redis.static
+show_limit_quota_header: true
+```
+
+## 识别请求头 x-ca-key，进行区别限流
+```yaml
+rule_name: limit_by_param_x-ca-key
+limit_by_header: x-ca-key
+limit_keys:
+- key: 102234
+  query_per_second: 10
+- key: 308239
+  query_per_hour: 10
+redis:
+  service_name: redis.static
+show_limit_quota_header: true  
+```
+
+## 根据请求头 x-forwarded-for 获取对端IP，进行区别限流
+
+```yaml
+rule_name: limit_by_per_ip_from-header-x-forwarded-for
+limit_by_per_ip: from-header-x-forwarded-for
+limit_keys:
+	# 精确ip
+- key: 1.1.1.1
+  query_per_day: 10
+  # ip段，符合这个ip段的ip，每个ip 100qps
+- key: 1.1.1.0/24
+  query_per_day: 100
+  # 兜底用，即默认每个ip 1000qps
+- key: 0.0.0.0/0
+  query_per_day: 1000
+redis:
+  service_name: redis.static
+show_limit_quota_header: true  
+```
+
+## 对特定路由或域名开启
+
+```yaml
+# 使用_rules_字段进行细粒度规则配置
+_rules_:
+# 规则一：按路由名称匹配生效
+- _match_route_:
+  - route-a
+  - route-b
+  rule_name: limit_rule1
+  limit_by_per_ip: from-header-x-forwarded-for
+  limit_keys:
+    # 精确ip
+  - key: 1.1.1.1
+    query_per_day: 10
+    # ip段，符合这个ip段的ip，每个ip 100qps
+  - key: 1.1.1.0/24
+    query_per_day: 100
+    # 兜底用，即默认每个ip 1000qps
+  - key: 0.0.0.0/0
+    query_per_day: 1000
+  redis:
+  	service_name: redis.static
+# 规则二：按域名匹配生效
+- _match_domain_:
+  - "*.example.com"
+  - test.com
+  rule_name: limit_rule2
+  limit_by_param: apikey
+  limit_keys:
+  - key: 9a342114-ba8a-11ec-b1bf-00163e1250b5
+    query_per_second: 10
+  - key: a6a6d7f2-ba8a-11ec-bec2-00163e1250b5
+    query_per_minute: 100
+  redis:
+  	service_name: redis.static
+  show_limit_quota_header: true 
+```
+此例 `_match_route_` 中指定的 `route-a` 和 `route-b` 即在创建网关路由时填写的路由名称，当匹配到这两个路由时，将使用此段配置；
+此例 `_match_domain_` 中指定的 `*.example.com` 和 `test.com` 用于匹配请求的域名，当发现域名匹配时，将使用此段配置；
+配置的匹配生效顺序，将按照 `_rules_` 下规则的排列顺序，匹配第一个规则后生效对应配置，后续规则将被忽略
--- a/plugins/wasm-go/extensions/cluster-key-rate-limit/config.go
+++ b/plugins/wasm-go/extensions/cluster-key-rate-limit/config.go
@@ -0,0 +1,230 @@
+package main
+
+import (
+	"errors"
+	"github.com/alibaba/higress/plugins/wasm-go/pkg/wrapper"
+	"github.com/tidwall/gjson"
+	"github.com/zmap/go-iptree/iptree"
+	"strings"
+)
+
+// 限流规则类型
+type limitRuleType string
+
+const (
+	limitByHeaderType limitRuleType = "limitByHeader"
+	limitByParamType  limitRuleType = "limitByParam"
+	limitByPerIpType  limitRuleType = "limitByPerIp"
+
+	RemoteAddrSourceType = "remote-addr"
+	HeaderSourceType     = "header"
+
+	DefaultRejectedCode uint32 = 429
+	DefaultRejectedMsg  string = "Too many requests"
+
+	Second           int64 = 1
+	SecondsPerMinute       = 60 * Second
+	SecondsPerHour         = 60 * SecondsPerMinute
+	SecondsPerDay          = 24 * SecondsPerHour
+)
+
+type ClusterKeyRateLimitConfig struct {
+	ruleName             string        // 限流规则名称
+	limitType            limitRuleType // 限流类型
+	limitByHeader        string        // 根据http请求头限流
+	limitByParam         string        // 根据url参数限流
+	limitByPerIp         LimitByPerIp  // 根据对端ip限流
+	limitItems           []LimitItem   // 限流配置 key为限流的ip地址或者ip段
+	showLimitQuotaHeader bool          // 响应头中是否显示X-RateLimit-Limit和X-RateLimit-Remaining
+	rejectedCode         uint32        // 当请求超过阈值被拒绝时,返回的HTTP状态码
+	rejectedMsg          string        // 当请求超过阈值被拒绝时,返回的响应体
+	redisClient          wrapper.RedisClient
+}
+
+type LimitByPerIp struct {
+	sourceType string // ip来源类型
+	headerName string // 根据该请求头获取客户端ip
+}
+
+type LimitItem struct {
+	key        string         // 限流key
+	ipNet      *iptree.IPTree // 限流key转换的ip地址或者ip段
+	count      int64          // 指定时间窗口内的总请求数量阈值
+	timeWindow int64          // 时间窗口大小
+}
+
+func initRedisClusterClient(json gjson.Result, config *ClusterKeyRateLimitConfig) error {
+	redisConfig := json.Get("redis")
+	if !redisConfig.Exists() {
+		return errors.New("missing redis in config")
+	}
+	serviceName := redisConfig.Get("service_name").String()
+	if serviceName == "" {
+		return errors.New("redis service name must not be empty")
+	}
+	servicePort := int(redisConfig.Get("service_port").Int())
+	if servicePort == 0 {
+		if strings.HasSuffix(serviceName, ".static") {
+			// use default logic port which is 80 for static service
+			servicePort = 80
+		} else {
+			servicePort = 6379
+		}
+	}
+	username := redisConfig.Get("username").String()
+	password := redisConfig.Get("password").String()
+	timeout := int(redisConfig.Get("timeout").Int())
+	if timeout == 0 {
+		timeout = 1000
+	}
+	config.redisClient = wrapper.NewRedisClusterClient(wrapper.FQDNCluster{
+		FQDN: serviceName,
+		Port: int64(servicePort),
+	})
+	return config.redisClient.Init(username, password, int64(timeout))
+}
+
+func parseClusterKeyRateLimitConfig(json gjson.Result, config *ClusterKeyRateLimitConfig, log wrapper.Log) error {
+	ruleName := json.Get("rule_name")
+	if !ruleName.Exists() {
+		return errors.New("missing rule_name in config")
+	}
+	config.ruleName = ruleName.String()
+
+	// 根据配置区分限流类型
+	var limitType limitRuleType
+	limitByHeader := json.Get("limit_by_header")
+	if limitByHeader.Exists() && limitByHeader.String() != "" {
+		config.limitByHeader = limitByHeader.String()
+		limitType = limitByHeaderType
+	}
+
+	limitByParam := json.Get("limit_by_param")
+	if limitByParam.Exists() && limitByParam.String() != "" {
+		config.limitByParam = limitByParam.String()
+		limitType = limitByParamType
+	}
+
+	limitByPerIpResult := json.Get("limit_by_per_ip")
+	if limitByPerIpResult.Exists() && limitByPerIpResult.String() != "" {
+		limitByPerIp := limitByPerIpResult.String()
+		if strings.HasPrefix(limitByPerIp, "from-header-") {
+			headerName := limitByPerIp[len("from-header-"):]
+			if headerName == "" {
+				return errors.New("limit_by_per_ip parse error: empty after 'from-header-'")
+			}
+			config.limitByPerIp = LimitByPerIp{
+				sourceType: HeaderSourceType,
+				headerName: headerName,
+			}
+		} else if limitByPerIp == "from-remote-addr" {
+			config.limitByPerIp = LimitByPerIp{
+				sourceType: RemoteAddrSourceType,
+				headerName: "",
+			}
+		} else {
+			return errors.New("the 'limit_by_per_ip' restriction must start with 'from-header-' or be exactly 'from-remote-addr'")
+		}
+		limitType = limitByPerIpType
+	}
+	if limitType == "" {
+		return errors.New("only one of 'limit_by_header' and 'limit_by_param' and 'limit_by_per_ip' can be set")
+	}
+	config.limitType = limitType
+
+	// 初始化LimitItem
+	err := initLimitItems(json, config, log)
+	if err != nil {
+		return err
+	}
+
+	showLimitQuotaHeader := json.Get("show_limit_quota_header")
+	if showLimitQuotaHeader.Exists() {
+		config.showLimitQuotaHeader = showLimitQuotaHeader.Bool()
+	}
+
+	rejectedCode := json.Get("rejected_code")
+	if rejectedCode.Exists() {
+		config.rejectedCode = uint32(rejectedCode.Uint())
+	} else {
+		config.rejectedCode = DefaultRejectedCode
+	}
+	rejectedMsg := json.Get("rejected_msg")
+	if rejectedCode.Exists() {
+		config.rejectedMsg = rejectedMsg.String()
+	} else {
+		config.rejectedMsg = DefaultRejectedMsg
+	}
+	return nil
+}
+
+func initLimitItems(json gjson.Result, config *ClusterKeyRateLimitConfig, log wrapper.Log) error {
+	limitKeys := json.Get("limit_keys")
+	if !limitKeys.Exists() {
+		return errors.New("missing limit_keys in config")
+	}
+	if len(limitKeys.Array()) == 0 {
+		return errors.New("config limit_keys cannot be empty")
+	}
+	var limitItems []LimitItem
+	for _, item := range limitKeys.Array() {
+		key := item.Get("key")
+		if !key.Exists() || key.String() == "" {
+			return errors.New("limit_keys key is required")
+		}
+		var ipNet *iptree.IPTree
+		if config.limitType == limitByPerIpType {
+			var err error
+			ipNet, err = parseIPNet(key.String())
+			if err != nil {
+				log.Errorf("parseIPNet error: %v", err)
+				return err
+			}
+		} else {
+			ipNet = nil
+		}
+
+		qps := item.Get("query_per_second")
+		if qps.Exists() && qps.Int() > 0 {
+			limitItems = append(limitItems, LimitItem{
+				key:        key.String(),
+				ipNet:      ipNet,
+				count:      qps.Int(),
+				timeWindow: Second,
+			})
+			continue
+		}
+		qpm := item.Get("query_per_minute")
+		if qpm.Exists() && qpm.Int() > 0 {
+			limitItems = append(limitItems, LimitItem{
+				key:        key.String(),
+				ipNet:      ipNet,
+				count:      qpm.Int(),
+				timeWindow: SecondsPerMinute,
+			})
+			continue
+		}
+		qph := item.Get("query_per_hour")
+		if qph.Exists() && qph.Int() > 0 {
+			limitItems = append(limitItems, LimitItem{
+				key:        key.String(),
+				ipNet:      ipNet,
+				count:      qph.Int(),
+				timeWindow: SecondsPerHour,
+			})
+			continue
+		}
+		qpd := item.Get("query_per_day")
+		if qpd.Exists() && qpd.Int() > 0 {
+			limitItems = append(limitItems, LimitItem{
+				key:        key.String(),
+				ipNet:      ipNet,
+				count:      qpd.Int(),
+				timeWindow: SecondsPerDay,
+			})
+			continue
+		}
+	}
+	config.limitItems = limitItems
+	return nil
+}
--- a/plugins/wasm-go/extensions/cluster-key-rate-limit/go.mod
+++ b/plugins/wasm-go/extensions/cluster-key-rate-limit/go.mod
@@ -0,0 +1,22 @@
+module github.com/alibaba/higress/plugins/wasm-go/extensions/key-cluster-rate-limit
+
+go 1.19
+
+replace github.com/alibaba/higress/plugins/wasm-go => ../..
+
+require (
+	github.com/alibaba/higress/plugins/wasm-go v0.0.0
+	github.com/higress-group/proxy-wasm-go-sdk v0.0.0-20240327114451-d6b7174a84fc
+	github.com/tidwall/gjson v1.14.3
+	github.com/tidwall/resp v0.1.1
+	github.com/zmap/go-iptree v0.0.0-20210731043055-d4e632617837
+)
+
+require (
+	github.com/asergeyev/nradix v0.0.0-20170505151046-3872ab85bb56 // indirect
+	github.com/google/uuid v1.3.0 // indirect
+	github.com/higress-group/nottinygc v0.0.0-20231101025119-e93c4c2f8520 // indirect
+	github.com/magefile/mage v1.14.0 // indirect
+	github.com/tidwall/match v1.1.1 // indirect
+	github.com/tidwall/pretty v1.2.0 // indirect
+)
--- a/plugins/wasm-go/extensions/cluster-key-rate-limit/go.sum
+++ b/plugins/wasm-go/extensions/cluster-key-rate-limit/go.sum
@@ -0,0 +1,24 @@
+github.com/asergeyev/nradix v0.0.0-20170505151046-3872ab85bb56 h1:Wi5Tgn8K+jDcBYL+dIMS1+qXYH2r7tpRAyBgqrWfQtw=
+github.com/asergeyev/nradix v0.0.0-20170505151046-3872ab85bb56/go.mod h1:8BhOLuqtSuT5NZtZMwfvEibi09RO3u79uqfHZzfDTR4=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/higress-group/nottinygc v0.0.0-20231101025119-e93c4c2f8520 h1:IHDghbGQ2DTIXHBHxWfqCYQW1fKjyJ/I7W1pMyUDeEA=
+github.com/higress-group/nottinygc v0.0.0-20231101025119-e93c4c2f8520/go.mod h1:Nz8ORLaFiLWotg6GeKlJMhv8cci8mM43uEnLA5t8iew=
+github.com/higress-group/proxy-wasm-go-sdk v0.0.0-20240327114451-d6b7174a84fc h1:t2AT8zb6N/59Y78lyRWedVoVWHNRSCBh0oWCC+bluTQ=
+github.com/higress-group/proxy-wasm-go-sdk v0.0.0-20240327114451-d6b7174a84fc/go.mod h1:hNFjhrLUIq+kJ9bOcs8QtiplSQ61GZXtd2xHKx4BYRo=
+github.com/magefile/mage v1.14.0 h1:6QDX3g6z1YvJ4olPhT1wksUcSa/V0a1B+pJb73fBjyo=
+github.com/magefile/mage v1.14.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
+github.com/tidwall/gjson v1.14.3 h1:9jvXn7olKEHU1S9vwoMGliaT8jq1vJ7IH/n9zD9Dnlw=
+github.com/tidwall/gjson v1.14.3/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
+github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
+github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
+github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
+github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
+github.com/tidwall/resp v0.1.1 h1:Ly20wkhqKTmDUPlyM1S7pWo5kk0tDu8OoC/vFArXmwE=
+github.com/tidwall/resp v0.1.1/go.mod h1:3/FrruOBAxPTPtundW0VXgmsQ4ZBA0Aw714lVYgwFa0=
+github.com/zmap/go-iptree v0.0.0-20210731043055-d4e632617837 h1:DjHnADS2r2zynZ3WkCFAQ+PNYngMSNceRROi0pO6c3M=
+github.com/zmap/go-iptree v0.0.0-20210731043055-d4e632617837/go.mod h1:9vp0bxqozzQwcjBwenEXfKVq8+mYbwHkQ1NF9Ap0DMw=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
--- a/plugins/wasm-go/extensions/cluster-key-rate-limit/main.go
+++ b/plugins/wasm-go/extensions/cluster-key-rate-limit/main.go
@@ -0,0 +1,208 @@
+// Copyright (c) 2024 Alibaba Group Holding Ltd.
+//
+// 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 main
+
+import (
+	"fmt"
+	"github.com/alibaba/higress/plugins/wasm-go/pkg/wrapper"
+	"github.com/higress-group/proxy-wasm-go-sdk/proxywasm"
+	"github.com/higress-group/proxy-wasm-go-sdk/proxywasm/types"
+	"github.com/tidwall/gjson"
+	"github.com/tidwall/resp"
+	"net"
+	"net/url"
+	"strconv"
+	"strings"
+)
+
+func main() {
+	wrapper.SetCtx(
+		"cluster-key-rate-limit",
+		wrapper.ParseConfigBy(parseConfig),
+		wrapper.ProcessRequestHeadersBy(onHttpRequestHeaders),
+		wrapper.ProcessResponseHeadersBy(onHttpResponseHeaders),
+	)
+}
+
+const (
+	ClusterRateLimitFormat string = "higress-cluster-key-rate-limit:%s:%s"
+	FixedWindowScript      string = `
+    	local ttl = redis.call('ttl', KEYS[1])
+    	if ttl < 0 then
+        	redis.call('set', KEYS[1], ARGV[1] - 1, 'EX', ARGV[2])
+        	return {ARGV[1], ARGV[1] - 1, ARGV[2]}
+    	end
+    	return {ARGV[1], redis.call('incrby', KEYS[1], -1), ttl}
+	`
+)
+
+const (
+	LimitContextKey string = "LimitContext" // 限流上下文信息
+
+	RateLimitLimitHeader     string = "X-RateLimit-Limit"     // 限制的总请求数
+	RateLimitRemainingHeader string = "X-RateLimit-Remaining" // 剩余还可以发送的请求数
+	RateLimitResetHeader     string = "X-RateLimit-Reset"     // 限流重置时间（触发限流时返回）
+)
+
+type LimitContext struct {
+	count     int
+	remaining int
+	reset     int
+}
+
+func parseConfig(json gjson.Result, config *ClusterKeyRateLimitConfig, log wrapper.Log) error {
+	err := initRedisClusterClient(json, config)
+	if err != nil {
+		return err
+	}
+	err = parseClusterKeyRateLimitConfig(json, config, log)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func onHttpRequestHeaders(ctx wrapper.HttpContext, config ClusterKeyRateLimitConfig, log wrapper.Log) types.Action {
+	// 判断是否命中限流规则
+	key, limitItem := hitRateLimitRule(ctx, config, log)
+	if limitItem == nil {
+		return types.ActionContinue
+	}
+
+	// 构建redis限流key和参数
+	limitKey := fmt.Sprintf(ClusterRateLimitFormat, config.ruleName, key)
+	keys := []interface{}{limitKey}
+	args := []interface{}{limitItem.count, limitItem.timeWindow}
+	// 执行限流逻辑
+	err := config.redisClient.Eval(FixedWindowScript, 1, keys, args, func(response resp.Value) {
+		defer func() {
+			_ = proxywasm.ResumeHttpRequest()
+		}()
+		resultArray := response.Array()
+		if len(resultArray) != 3 {
+			log.Errorf("redis response parse error, response: %v", response)
+			return
+		}
+		context := LimitContext{
+			count:     resultArray[0].Integer(),
+			remaining: resultArray[1].Integer(),
+			reset:     resultArray[2].Integer(),
+		}
+		if context.remaining < 0 {
+			// 触发限流
+			rejected(config, context)
+		} else {
+			ctx.SetContext(LimitContextKey, context)
+		}
+	})
+	if err != nil {
+		log.Errorf("redis call failed: %v", err)
+		return types.ActionContinue
+	}
+	return types.ActionPause
+}
+
+func onHttpResponseHeaders(ctx wrapper.HttpContext, config ClusterKeyRateLimitConfig, log wrapper.Log) types.Action {
+	limitContext, ok := ctx.GetContext(LimitContextKey).(LimitContext)
+	if !ok {
+		return types.ActionContinue
+	}
+	if config.showLimitQuotaHeader {
+		_ = proxywasm.ReplaceHttpResponseHeader(RateLimitLimitHeader, strconv.Itoa(limitContext.count))
+		_ = proxywasm.ReplaceHttpResponseHeader(RateLimitRemainingHeader, strconv.Itoa(limitContext.remaining))
+	}
+	return types.ActionContinue
+}
+
+func hitRateLimitRule(ctx wrapper.HttpContext, config ClusterKeyRateLimitConfig, log wrapper.Log) (string, *LimitItem) {
+	switch config.limitType {
+	case limitByHeaderType:
+		headerVal, err := proxywasm.GetHttpRequestHeader(config.limitByHeader)
+		if err != nil {
+			log.Debugf("failed to get request header %s: %v", config.limitByHeader, err)
+			return "", nil
+		}
+		return headerVal, findMatchingItem(config.limitItems, headerVal)
+	case limitByParamType:
+		parse, _ := url.Parse(ctx.Path())
+		query, _ := url.ParseQuery(parse.RawQuery)
+		val, ok := query[config.limitByParam]
+		if !ok {
+			log.Debugf("request param %s is empty", config.limitByParam)
+			return "", nil
+		} else {
+			return val[0], findMatchingItem(config.limitItems, val[0])
+		}
+	case limitByPerIpType:
+		realIp, err := getDownStreamIp(config)
+		if err != nil {
+			log.Warnf("failed to get down stream ip: %v", err)
+			return "", nil
+		}
+		for _, item := range config.limitItems {
+			if _, found, _ := item.ipNet.Get(realIp); !found {
+				continue
+			}
+			return realIp.String(), &item
+		}
+	}
+	return "", nil
+}
+
+func findMatchingItem(items []LimitItem, key string) *LimitItem {
+	for _, item := range items {
+		if item.key == key {
+			return &item
+		}
+	}
+	return nil
+}
+
+func getDownStreamIp(config ClusterKeyRateLimitConfig) (net.IP, error) {
+	var (
+		realIpStr string
+		err       error
+	)
+	if config.limitByPerIp.sourceType == HeaderSourceType {
+		realIpStr, err = proxywasm.GetHttpRequestHeader(config.limitByPerIp.headerName)
+		if err == nil {
+			realIpStr = strings.Split(strings.Trim(realIpStr, " "), ",")[0]
+		}
+	} else {
+		var bs []byte
+		bs, err = proxywasm.GetProperty([]string{"source", "address"})
+		realIpStr = string(bs)
+	}
+	if err != nil {
+		return nil, err
+	}
+	ip := parseIP(realIpStr)
+	realIP := net.ParseIP(ip)
+	if realIP == nil {
+		return nil, fmt.Errorf("invalid ip[%s]", ip)
+	}
+	return realIP, nil
+}
+
+func rejected(config ClusterKeyRateLimitConfig, context LimitContext) {
+	headers := make(map[string][]string)
+	headers[RateLimitResetHeader] = []string{strconv.Itoa(context.reset)}
+	if config.showLimitQuotaHeader {
+		headers[RateLimitLimitHeader] = []string{strconv.Itoa(context.count)}
+		headers[RateLimitRemainingHeader] = []string{strconv.Itoa(0)}
+	}
+	_ = proxywasm.SendHttpResponse(
+		config.rejectedCode, reconvertHeaders(headers), []byte(config.rejectedMsg), -1)
+}
--- a/plugins/wasm-go/extensions/cluster-key-rate-limit/utils.go
+++ b/plugins/wasm-go/extensions/cluster-key-rate-limit/utils.go
@@ -0,0 +1,45 @@
+package main
+
+import (
+	"fmt"
+	"github.com/zmap/go-iptree/iptree"
+	"sort"
+	"strings"
+)
+
+// parseIPNet 解析Ip段配置
+func parseIPNet(key string) (*iptree.IPTree, error) {
+	tree := iptree.New()
+	err := tree.AddByString(key, 0)
+	if err != nil {
+		return nil, fmt.Errorf("invalid IP[%s]", key)
+	}
+	return tree, nil
+}
+
+// parseIP 解析IP
+func parseIP(source string) string {
+	if strings.Contains(source, ".") {
+		// parse ipv4
+		return strings.Split(source, ":")[0]
+	}
+	// parse ipv6
+	if strings.Contains(source, "]") {
+		return strings.Split(source, "]")[0][1:]
+	}
+	return source
+}
+
+// reconvertHeaders headers: map[string][]string -> [][2]string
+func reconvertHeaders(hs map[string][]string) [][2]string {
+	var ret [][2]string
+	for k, vs := range hs {
+		for _, v := range vs {
+			ret = append(ret, [2]string{k, v})
+		}
+	}
+	sort.SliceStable(ret, func(i, j int) bool {
+		return ret[i][0] < ret[j][0]
+	})
+	return ret
+}