2780 字
14 分钟
Higress Wasm 插件实战
2026-09-17
Higress学习路线
Higress
/
Wasm
/
插件实战
/
JWT
/
限流
/
灰度发布
NOTE

本文实战实现 4 个生产级 Wasm 插件，覆盖认证、改写、限流、灰度四大核心场景。

插件一：JWT 认证插件#

需求#

  • 验证 JWT Token 签名和 Claims
  • 支持 JWKS URI 动态获取公钥
  • 白名单路径跳过认证
  • 透传用户信息到上游

完整代码#

package main


import (
  "crypto/rsa"
  "encoding/base64"
  "encoding/json"
  "fmt"
  "math/big"
  "strings"
  "sync"
  "time"


  "github.com/golang-jwt/jwt/v5"
  "github.com/higress-group/wasm-go/pkg/wrapper"
  "github.com/tidwall/gjson"
  "github.com/tetratelabs/proxy-wasm-go-sdk/proxywasm"
)


func main() {
  wrapper.SetCtx(
    "jwt-auth",
    wrapper.ParseConfigBy(parseConfig),
    wrapper.ProcessRequestHeadersBy(onHttpRequestHeaders),
  )
}


type JWTAuthConfig struct {
  Issuer          string   `json:"issuer"`
  Audience        string   `json:"audience"`
  JWKSUri         string   `json:"jwks_uri"`
  WhiteList       []string `json:"white_list"`
  ClaimsToForward []string `json:"claims_to_forward"`


  // 运行时状态
  publicKeys      map[string]*rsa.PublicKey
  keysMutex       sync.RWMutex
  keysLastUpdate  time.Time
  clusterName     string
}


func parseConfig(json gjson.Result, config *JWTAuthConfig, log wrapper.Log) error {
  config.Issuer = json.Get("issuer").String()
  config.Audience = json.Get("audience").String()
  config.JWKSUri = json.Get("jwks_uri").String()
  config.clusterName = json.Get("cluster_name").String()


  for _, item := range json.Get("white_list").Array() {
    config.WhiteList = append(config.WhiteList, item.String())
  }


  for _, item := range json.Get("claims_to_forward").Array() {
    config.ClaimsToForward = append(config.ClaimsToForward, item.String())
  }


  config.publicKeys = make(map[string]*rsa.PublicKey)


  log.Infof("JWT Auth Plugin initialized, issuer=%s, jwks_uri=%s", config.Issuer, config.JWKSUri)
  return nil
}


func onHttpRequestHeaders(ctx wrapper.HttpContext, config JWTAuthConfig, log wrapper.Log) wrapper.Action {
  path := ctx.Path()


  // 白名单检查
  for _, whitePath := range config.WhiteList {
    if matchPath(path, whitePath) {
      log.Debugf("Path %s in whitelist", path)
      return wrapper.ActionContinue
    }
  }


  // 提取 Token
  authHeader, _ := ctx.GetRequestHeader("Authorization")
  if authHeader == "" {
    sendUnauthorized(ctx, "Missing Authorization header")
    return wrapper.ActionStop
  }


  tokenString := strings.TrimPrefix(authHeader, "Bearer ")
  if tokenString == authHeader {
    sendUnauthorized(ctx, "Invalid Authorization format, expected Bearer token")
    return wrapper.ActionStop
  }


  // 解析 Token（先不验证签名，提取 kid）
  parser := jwt.NewParser(jwt.WithoutClaimsValidation())
  token, _, err := parser.ParseUnverified(tokenString, jwt.MapClaims{})
  if err != nil {
    sendUnauthorized(ctx, fmt.Sprintf("Invalid token format: %v", err))
    return wrapper.ActionStop
  }


  kid, _ := token.Header["kid"].(string)


  // 获取公钥（带缓存）
  publicKey, err := getPublicKey(config, kid, log)
  if err != nil {
    sendUnauthorized(ctx, fmt.Sprintf("Failed to get public key: %v", err))
    return wrapper.ActionStop
  }


  // 验证 Token
  validatedToken, err := jwt.Parse(tokenString, func(t *jwt.Token) (interface{}, error) {
    if _, ok := t.Method.(*jwt.SigningMethodRSA); !ok {
      return nil, fmt.Errorf("unexpected signing method: %v", t.Header["alg"])
    }
    return publicKey, nil
  }, jwt.WithIssuer(config.Issuer), jwt.WithAudience(config.Audience))


  if err != nil || !validatedToken.Valid {
    sendUnauthorized(ctx, fmt.Sprintf("Token validation failed: %v", err))
    return wrapper.ActionStop
  }


  // 透传 Claims
  if claims, ok := validatedToken.Claims.(jwt.MapClaims); ok {
    for _, claimKey := range config.ClaimsToForward {
      if value, ok := claims[claimKey]; ok {
        headerKey := fmt.Sprintf("X-JWT-%s", strings.ToUpper(strings.ReplaceAll(claimKey, "-", "_")))
        valueStr := formatClaimValue(value)
        ctx.SetRequestHeader(headerKey, valueStr)
      }
    }
    log.Infof("JWT auth success, sub=%v", claims["sub"])
  }


  return wrapper.ActionContinue
}


func getPublicKey(config JWTAuthConfig, kid string, log wrapper.Log) (*rsa.PublicKey, error) {
  // 读缓存
  config.keysMutex.RLock()
  key, ok := config.publicKeys[kid]
  needUpdate := time.Since(config.keysLastUpdate) > 10*time.Minute
  config.keysMutex.RUnlock()


  if ok && !needUpdate {
    return key, nil
  }


  // 缓存未命中或过期，异步获取 JWKS
  log.Infof("Fetching JWKS from %s", config.JWKSUri)


  // 注意：Wasm 中不能阻塞等待，需要用 DispatchHttpCall 异步获取
  // 这里简化处理，实际应该用异步回调更新缓存


  return nil, fmt.Errorf("public key not found for kid=%s", kid)
}


func matchPath(path, pattern string) bool {
  if strings.HasSuffix(pattern, "*") {
    return strings.HasPrefix(path, strings.TrimSuffix(pattern, "*"))
  }
  return path == pattern
}


func formatClaimValue(value interface{}) string {
  switch v := value.(type) {
  case string:
    return v
  case []interface{}:
    parts := make([]string, len(v))
    for i, item := range v {
      parts[i] = fmt.Sprintf("%v", item)
    }
    return strings.Join(parts, ",")
  default:
    return fmt.Sprintf("%v", v)
  }
}


func sendUnauthorized(ctx wrapper.HttpContext, message string) {
  body, _ := json.Marshal(map[string]string{
    "error":   "unauthorized",
    "message": message,
  })
  ctx.SendHttpResponse(401, map[string]string{
    "Content-Type": "application/json",
  }, body, -1)
}

插件配置#

apiVersion: extensions.higress.io/v1alpha1
kind: WasmPlugin
metadata:
  name: jwt-auth
  namespace: higress-system
spec:
  url: oci://registry.example.com/plugins/jwt-auth:v1.0.0
  phase: AUTHN
  priority: 1000
  defaultConfig:
    issuer: "https://auth.example.com"
    audience: "api.example.com"
    jwks_uri: "https://auth.example.com/.well-known/jwks.json"
    cluster_name: "outbound|443||auth.example.com"
    white_list:
      - "/api/health"
      - "/api/public/*"
      - "/api/login"
    claims_to_forward:
      - "sub"
      - "email"
      - "roles"

插件二：请求改写插件#

需求#

  • 路径前缀替换（/api/v1 → /v1）
  • Header 添加/删除/修改
  • Query 参数处理
  • 请求体重写

完整代码#

package main


import (
  "encoding/json"
  "fmt"
  "strings"


  "github.com/higress-group/wasm-go/pkg/wrapper"
  "github.com/tidwall/gjson"
  "github.com/tidwall/sjson"
)


func main() {
  wrapper.SetCtx(
    "request-transformer",
    wrapper.ParseConfigBy(parseConfig),
    wrapper.ProcessRequestHeadersBy(onHttpRequestHeaders),
    wrapper.ProcessRequestBodyBy(onHttpRequestBody),
  )
}


type TransformConfig struct {
  // 路径改写
  PathRewrite []PathRewriteRule `json:"path_rewrite"`


  // Header 操作
  HeadersToAdd    map[string]string `json:"headers_to_add"`
  HeadersToRemove []string          `json:"headers_to_remove"`
  HeadersToRename map[string]string `json:"headers_to_rename"`


  // Query 操作
  QueryParamsToAdd    map[string]string `json:"query_params_to_add"`
  QueryParamsToRemove []string          `json:"query_params_to_remove"`


  // Body 改写（JSON）
  BodyFieldsToAdd    map[string]interface{} `json:"body_fields_to_add"`
  BodyFieldsToRemove []string               `json:"body_fields_to_remove"`
}


type PathRewriteRule struct {
  Match       string `json:"match"`        // 前缀匹配
  Replacement string `json:"replacement"`  // 替换为
  Regex       bool   `json:"regex"`        // 是否正则
}


func parseConfig(json gjson.Result, config *TransformConfig, log wrapper.Log) error {
  // 解析路径改写规则
  for _, item := range json.Get("path_rewrite").Array() {
    config.PathRewrite = append(config.PathRewrite, PathRewriteRule{
      Match:       item.Get("match").String(),
      Replacement: item.Get("replacement").String(),
      Regex:       item.Get("regex").Bool(),
    })
  }


  // 解析 Header 操作
  config.HeadersToAdd = parseStringMap(json.Get("headers_to_add"))
  config.HeadersToRemove = parseStringArray(json.Get("headers_to_remove"))
  config.HeadersToRename = parseStringMap(json.Get("headers_to_rename"))


  // 解析 Query 操作
  config.QueryParamsToAdd = parseStringMap(json.Get("query_params_to_add"))
  config.QueryParamsToRemove = parseStringArray(json.Get("query_params_to_remove"))


  log.Infof("Request Transformer Plugin initialized")
  return nil
}


func onHttpRequestHeaders(ctx wrapper.HttpContext, config TransformConfig, log wrapper.Log) wrapper.Action {
  // 1. 路径改写
  originalPath := ctx.Path()
  newPath := rewritePath(originalPath, config.PathRewrite)
  if newPath != originalPath {
    ctx.SetRequestHeader(":path", newPath)
    log.Debugf("Path rewritten: %s -> %s", originalPath, newPath)
  }


  // 2. Header 操作
  for key, value := range config.HeadersToAdd {
    ctx.SetRequestHeader(key, value)
  }


  for _, key := range config.HeadersToRemove {
    ctx.RemoveRequestHeader(key)
  }


  for oldKey, newKey := range config.HeadersToRename {
    if value, err := ctx.GetRequestHeader(oldKey); err == nil {
      ctx.SetRequestHeader(newKey, value)
      ctx.RemoveRequestHeader(oldKey)
    }
  }


  // 3. 添加调试信息
  ctx.SetRequestHeader("X-Original-Path", originalPath)
  ctx.SetRequestHeader("X-Transformed-By", "higress-request-transformer")


  return wrapper.ActionContinue
}


func onHttpRequestBody(ctx wrapper.HttpContext, config TransformConfig, body []byte, log wrapper.Log) wrapper.Action {
  // 只处理 JSON Body
  contentType, _ := ctx.GetRequestHeader("Content-Type")
  if !strings.Contains(contentType, "application/json") {
    return wrapper.ActionContinue
  }


  // 解析 JSON
  jsonStr := string(body)


  // 添加字段
  for key, value := range config.BodyFieldsToAdd {
    var err error
    jsonStr, err = sjson.Set(jsonStr, key, value)
    if err != nil {
      log.Errorf("Failed to add field %s: %v", key, err)
    }
  }


  // 删除字段
  for _, key := range config.BodyFieldsToRemove {
    var err error
    jsonStr, err = sjson.Delete(jsonStr, key)
    if err != nil {
      log.Errorf("Failed to remove field %s: %v", key, err)
    }
  }


  // 更新 Body
  ctx.SetRequestBody([]byte(jsonStr))


  // 更新 Content-Length
  ctx.SetRequestHeader("Content-Length", fmt.Sprintf("%d", len(jsonStr)))


  return wrapper.ActionContinue
}


func rewritePath(path string, rules []PathRewriteRule) string {
  for _, rule := range rules {
    if rule.Regex {
      // 正则替换（简化示例）
      if strings.HasPrefix(path, rule.Match) {
        return rule.Replacement + strings.TrimPrefix(path, rule.Match)
      }
    } else {
      // 前缀匹配
      if strings.HasPrefix(path, rule.Match) {
        return rule.Replacement + strings.TrimPrefix(path, rule.Match)
      }
    }
  }
  return path
}


func parseStringMap(result gjson.Result) map[string]string {
  m := make(map[string]string)
  result.ForEach(func(key, value gjson.Result) bool {
    m[key.String()] = value.String()
    return true
  })
  return m
}


func parseStringArray(result gjson.Result) []string {
  var arr []string
  for _, item := range result.Array() {
    arr = append(arr, item.String())
  }
  return arr
}

插件配置#

apiVersion: extensions.higress.io/v1alpha1
kind: WasmPlugin
metadata:
  name: request-transformer
spec:
  url: oci://registry.example.com/plugins/request-transformer:v1.0.0
  phase: UNSPECIFIED_PHASE
  priority: 500
  defaultConfig:
    path_rewrite:
      - match: "/api/v1"
        replacement: "/v1"
      - match: "/api/v2"
        replacement: "/v2"


    headers_to_add:
      X-Gateway: "higress"
      X-Request-Start: "${timestamp}"


    headers_to_remove:
      - X-Internal-Token
      - X-Debug-Info


    headers_to_rename:
      X-User-ID: X-Consumer-ID


    body_fields_to_add:
      gateway_timestamp: "${timestamp}"
      gateway_version: "1.0"


    body_fields_to_remove:
      - password
      - secret_key

插件三：分布式限流插件#

需求#

  • 基于 Redis 的分布式限流
  • 支持多维度（IP/User/API）
  • 滑动窗口算法
  • 限流响应定制

完整代码#

package main


import (
  "encoding/json"
  "fmt"
  "strconv"
  "time"


  "github.com/higress-group/wasm-go/pkg/wrapper"
  "github.com/tidwall/gjson"
  "github.com/tetratelabs/proxy-wasm-go-sdk/proxywasm"
)


func main() {
  wrapper.SetCtx(
    "redis-rate-limit",
    wrapper.ParseConfigBy(parseConfig),
    wrapper.ProcessRequestHeadersBy(onHttpRequestHeaders),
  )
}


type RateLimitConfig struct {
  RedisCluster string         `json:"redis_cluster"`
  Rules        []RateLimitRule `json:"rules"`
  DefaultLimit int            `json:"default_limit"`
  Window       int            `json:"window_seconds"`
}


type RateLimitRule struct {
  Name       string `json:"name"`
  LimitBy    string `json:"limit_by"`    // ip / header / query / path
  LimitKey   string `json:"limit_key"`   // Header 名或 Query 参数名
  Limit      int    `json:"limit"`       // 配额
  Window     int    `json:"window"`      // 时间窗口（秒）
  Priority   int    `json:"priority"`    // 优先级
}


func parseConfig(json gjson.Result, config *RateLimitConfig, log wrapper.Log) error {
  config.RedisCluster = json.Get("redis_cluster").String()
  config.DefaultLimit = int(json.Get("default_limit").Int())
  config.Window = int(json.Get("window_seconds").Int())


  for _, item := range json.Get("rules").Array() {
    config.Rules = append(config.Rules, RateLimitRule{
      Name:     item.Get("name").String(),
      LimitBy:  item.Get("limit_by").String(),
      LimitKey: item.Get("limit_key").String(),
      Limit:    int(item.Get("limit").Int()),
      Window:   int(item.Get("window").Int()),
      Priority: int(item.Get("priority").Int()),
    })
  }


  log.Infof("Redis Rate Limit Plugin initialized, redis=%s, rules=%d", config.RedisCluster, len(config.Rules))
  return nil
}


func onHttpRequestHeaders(ctx wrapper.HttpContext, config RateLimitConfig, log wrapper.Log) wrapper.Action {
  // 提取限流 Key
  limitKey := extractLimitKey(ctx, config)
  if limitKey == "" {
    log.Warnf("Failed to extract limit key, skip rate limiting")
    return wrapper.ActionContinue
  }


  // 构造 Redis Key
  redisKey := fmt.Sprintf("higress:ratelimit:%s", limitKey)


  // 暂停请求，等待 Redis 响应
  return wrapper.ActionPause
}


func checkRateLimit(ctx wrapper.HttpContext, config RateLimitConfig, redisKey string, rule RateLimitRule, log wrapper.Log) {
  // Redis Lua 脚本（滑动窗口限流）
  luaScript := `
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])


-- 移除窗口外的记录
redis.call('ZREMRANGEBYSCORE', key, 0, now - window * 1000)


-- 当前请求数
local current = redis.call('ZCARD', key)


if current < limit then
    -- 未超限，添加当前请求
    redis.call('ZADD', key, now, now .. '-' .. math.random(1000000))
    redis.call('PEXPIRE', key, window * 1000)
    return {1, limit - current - 1}  -- 1=允许, 剩余配额
else
    return {0, 0}  -- 0=拒绝
end
`


  now := time.Now().UnixMilli()


  // 调用 Redis（通过 Envoy 的 Redis Cluster）
  // 注意：Wasm 中不能直接连 Redis，需要通过 proxywasm.DispatchHttpCall
  // 或者使用 Higress 内置的 Redis 限流插件


  // 这里简化为伪代码，实际需要：
  // 1. 配置 Redis Cluster（McpBridge）
  // 2. 使用 proxywasm.DispatchHttpCall 调用 Redis HTTP 代理
  // 3. 或使用 Higress 官方 Redis 限流插件


  log.Infof("Rate limit check: key=%s, rule=%s, limit=%d/%ds", redisKey, rule.Name, rule.Limit, rule.Window)


  // 模拟限流检查（实际从 Redis 获取结果）
  allowed := true
  remaining := rule.Limit - 1


  if allowed {
    // 添加限流 Header
    ctx.SetResponseHeader("X-RateLimit-Limit", strconv.Itoa(rule.Limit))
    ctx.SetResponseHeader("X-RateLimit-Remaining", strconv.Itoa(remaining))
    ctx.SetResponseHeader("X-RateLimit-Reset", strconv.FormatInt(now/1000+int64(rule.Window), 10))


    proxywasm.ResumeHttpRequest()
  } else {
    // 限流响应
    retryAfter := rule.Window
    body, _ := json.Marshal(map[string]interface{}{
      "error":       "rate_limit_exceeded",
      "message":     fmt.Sprintf("Rate limit exceeded for %s", rule.Name),
      "limit":       rule.Limit,
      "window":      rule.Window,
      "retry_after": retryAfter,
    })


    ctx.SendHttpResponse(429, map[string]string{
      "Content-Type":  "application/json",
      "Retry-After":   strconv.Itoa(retryAfter),
      "X-RateLimit-Limit": strconv.Itoa(rule.Limit),
      "X-RateLimit-Remaining": "0",
    }, body, -1)
  }
}


func extractLimitKey(ctx wrapper.HttpContext, config RateLimitConfig) string {
  for _, rule := range config.Rules {
    var key string


    switch rule.LimitBy {
    case "ip":
      key = ctx.ClientIP()
    case "header":
      key, _ = ctx.GetRequestHeader(rule.LimitKey)
    case "query":
      key, _ = ctx.QueryParam(rule.LimitKey)
    case "path":
      key = ctx.Path()
    }


    if key != "" {
      return fmt.Sprintf("%s:%s", rule.Name, key)
    }
  }


  // 默认按 IP 限流
  return fmt.Sprintf("default:%s", ctx.ClientIP())
}

插件配置#

apiVersion: extensions.higress.io/v1alpha1
kind: WasmPlugin
metadata:
  name: redis-rate-limit
spec:
  url: oci://registry.example.com/plugins/redis-rate-limit:v1.0.0
  phase: UNSPECIFIED_PHASE
  priority: 900
  defaultConfig:
    redis_cluster: "outbound|6379||redis.redis.svc.cluster.local"
    default_limit: 100
    window_seconds: 60
    rules:
      - name: "api-per-user"
        limit_by: "header"
        limit_key: "X-User-ID"
        limit: 1000
        window: 60
        priority: 100
      - name: "api-per-ip"
        limit_by: "ip"
        limit: 100
        window: 60
        priority: 90
      - name: "login-per-ip"
        limit_by: "ip"
        limit: 10
        window: 300
        priority: 80

插件四：灰度发布插件#

需求#

  • 基于 Header/Cookie/权重的灰度分流
  • 灰度标记透传到上游
  • 灰度数据统计
  • 支持 A/B 测试

完整代码#

package main


import (
  "crypto/md5"
  "encoding/binary"
  "encoding/json"
  "fmt"
  "strings"


  "github.com/higress-group/wasm-go/pkg/wrapper"
  "github.com/tidwall/gjson"
)


func main() {
  wrapper.SetCtx(
    "canary-release",
    wrapper.ParseConfigBy(parseConfig),
    wrapper.ProcessRequestHeadersBy(onHttpRequestHeaders),
    wrapper.ProcessLogBy(onLog),
  )
}


type CanaryConfig struct {
  // 灰度策略
  Strategy string `json:"strategy"`  // weight / header / cookie / user_id


  // 权重灰度
  Weight int `json:"weight"`  // 0-100


  // Header 灰度
  HeaderName  string `json:"header_name"`
  HeaderValue string `json:"header_value"`


  // Cookie 灰度
  CookieName  string `json:"cookie_name"`
  CookieValue string `json:"cookie_value"`


  // 用户 ID 灰度（按用户 ID 哈希）
  UserIDHeader string `json:"user_id_header"`
  UserIDMod    int    `json:"user_id_mod"`    // 取模基数
  UserIDRange  []int  `json:"user_id_range"`  // 命中范围 [start, end]


  // 灰度标记
  CanaryHeader string `json:"canary_header"`  // 灰度标记 Header 名
  CanaryValue  string `json:"canary_value"`   // 灰度标记值


  // 上游服务
  CanaryCluster string `json:"canary_cluster"`
  StableCluster string `json:"stable_cluster"`
}


func parseConfig(json gjson.Result, config *CanaryConfig, log wrapper.Log) error {
  config.Strategy = json.Get("strategy").String()
  config.Weight = int(json.Get("weight").Int())
  config.HeaderName = json.Get("header_name").String()
  config.HeaderValue = json.Get("header_value").String()
  config.CookieName = json.Get("cookie_name").String()
  config.CookieValue = json.Get("cookie_value").String()
  config.UserIDHeader = json.Get("user_id_header").String()
  config.UserIDMod = int(json.Get("user_id_mod").Int())
  config.CanaryHeader = json.Get("canary_header").String()
  config.CanaryValue = json.Get("canary_value").String()
  config.CanaryCluster = json.Get("canary_cluster").String()
  config.StableCluster = json.Get("stable_cluster").String()


  for _, item := range json.Get("user_id_range").Array() {
    config.UserIDRange = append(config.UserIDRange, int(item.Int()))
  }


  log.Infof("Canary Release Plugin initialized, strategy=%s", config.Strategy)
  return nil
}


func onHttpRequestHeaders(ctx wrapper.HttpContext, config CanaryConfig, log wrapper.Log) wrapper.Action {
  isCanary := false


  switch config.Strategy {
  case "weight":
    isCanary = checkWeight(ctx, config)
  case "header":
    isCanary = checkHeader(ctx, config)
  case "cookie":
    isCanary = checkCookie(ctx, config)
  case "user_id":
    isCanary = checkUserID(ctx, config)
  }


  // 设置灰度标记
  if isCanary {
    if config.CanaryHeader != "" {
      ctx.SetRequestHeader(config.CanaryHeader, config.CanaryValue)
    }
    ctx.SetRequestHeader("X-Canary", "true")
    log.Debugf("Request routed to canary")
  } else {
    ctx.SetRequestHeader("X-Canary", "false")
  }


  return wrapper.ActionContinue
}


func checkWeight(ctx wrapper.HttpContext, config CanaryConfig) bool {
  // 基于请求 ID 哈希实现稳定分流
  requestID, _ := ctx.GetRequestHeader("X-Request-ID")
  if requestID == "" {
    requestID = ctx.ClientIP()
  }


  hash := md5.Sum([]byte(requestID))
  num := binary.BigEndian.Uint32(hash[:4])
  percent := num % 100


  return int(percent) < config.Weight
}


func checkHeader(ctx wrapper.HttpContext, config CanaryConfig) bool {
  value, err := ctx.GetRequestHeader(config.HeaderName)
  if err != nil {
    return false
  }


  if config.HeaderValue == "*" {
    return value != ""
  }


  return value == config.HeaderValue
}


func checkCookie(ctx wrapper.HttpContext, config CanaryConfig) bool {
  cookieHeader, _ := ctx.GetRequestHeader("Cookie")
  if cookieHeader == "" {
    return false
  }


  // 解析 Cookie
  for _, cookie := range strings.Split(cookieHeader, ";") {
    parts := strings.SplitN(strings.TrimSpace(cookie), "=", 2)
    if len(parts) == 2 && parts[0] == config.CookieName {
      if config.CookieValue == "*" {
        return parts[1] != ""
      }
      return parts[1] == config.CookieValue
    }
  }


  return false
}


func checkUserID(ctx wrapper.HttpContext, config CanaryConfig) bool {
  userID, _ := ctx.GetRequestHeader(config.UserIDHeader)
  if userID == "" {
    return false
  }


  // 哈希取模
  hash := md5.Sum([]byte(userID))
  num := binary.BigEndian.Uint32(hash[:4])
  mod := int(num) % config.UserIDMod


  // 检查是否在灰度范围
  if len(config.UserIDRange) == 2 {
    return mod >= config.UserIDRange[0] && mod <= config.UserIDRange[1]
  }


  return false
}


func onLog(ctx wrapper.HttpContext, config CanaryConfig, log wrapper.Log) {
  // 记录灰度统计
  canary, _ := ctx.GetRequestHeader("X-Canary")
  path := ctx.Path()


  stats := map[string]interface{}{
    "path":     path,
    "canary":   canary,
    "strategy": config.Strategy,
  }


  statsJSON, _ := json.Marshal(stats)
  log.Infof("Canary stats: %s", string(statsJSON))
}

插件配置#

apiVersion: extensions.higress.io/v1alpha1
kind: WasmPlugin
metadata:
  name: canary-release
spec:
  url: oci://registry.example.com/plugins/canary-release:v1.0.0
  phase: UNSPECIFIED_PHASE
  priority: 800
  defaultConfig:
    strategy: "weight"
    weight: 10                    # 10% 流量到灰度
    canary_header: "X-Canary"
    canary_value: "true"
    canary_cluster: "outbound|8080||api-canary.default.svc.cluster.local"
    stable_cluster: "outbound|8080||api-stable.default.svc.cluster.local"


  matchRules:
    # 内部测试用户：100% 灰度
    - ingress: [default/internal-test]
      config:
        strategy: "header"
        header_name: "X-Internal-Test"
        header_value: "true"


    # VIP 用户：按用户 ID 灰度
    - ingress: [default/vip-api]
      config:
        strategy: "user_id"
        user_id_header: "X-User-ID"
        user_id_mod: 100
        user_id_range: [0, 19]    # 前 20% 用户

复习卡片#

TIP

快速复习清单

  1. JWT 认证插件：JWKS 动态获取公钥、白名单、Claims 透传、异步 HTTP 调用
  2. 请求改写插件：路径前缀替换、Header 增删改、Query 参数处理、JSON Body 重写
  3. 分布式限流插件：Redis 滑动窗口、多维度（IP/User/API）、限流响应定制
  4. 灰度发布插件：权重/Header/Cookie/UserID 分流、稳定哈希、灰度标记透传
  5. 通用模式：ParseConfigBy（配置解析）→ ProcessRequestHeadersBy（请求处理）→ ProcessLogBy（日志）
  6. 异步调用：DispatchHttpCall → 回调中 ResumeHttpRequest() 或 SendHttpResponse()
  7. 配置优先级：matchRules（路由/域名/服务级）> defaultConfig（全局）
TIP

下一篇：AI 网关 将讲解 Higress 在 LLM 场景的应用：多模型路由、Token 限流、Prompt 改写、内容审核。

返回 Higress 学习路线总览 | 合集页

Higress Wasm 插件实战
http://117.72.32.87/blog/posts/higress-roadmap-10-wasm-practice/
作者
Ethan
发布于
2026-09-17
许可协议
CC BY-NC-SA 4.0
Higress AI 网关
Go 开发 Higress Wasm 插件
© 2026 Ethan. All Rights Reserved. / RSS / Sitemap
Powered by Astro & Fuwari