Higress 插件 Go SDK 与处理流程

type CommonVmCtx[PluginConfig any] struct {
    // proxy-wasm-go-sdk DefaultVMContext 默认实现
    types.DefaultVMContext
    // 插件名称
    pluginName                  string
    // 插件日志工具
    log                         Log
    hasCustomConfig             bool
    // 插件配置解析函数
    parseConfig                 ParseConfigFunc[PluginConfig]
    // 插件路由、域名、服务级别配置解析函数
    parseRuleConfig             ParseRuleConfigFunc[PluginConfig]
    // 以下是自定义插件回调钩子函数
    onHttpRequestHeaders        onHttpHeadersFunc[PluginConfig]
    onHttpRequestBody           onHttpBodyFunc[PluginConfig]
    onHttpStreamingRequestBody  onHttpStreamingBodyFunc[PluginConfig]
    onHttpResponseHeaders       onHttpHeadersFunc[PluginConfig]
    onHttpResponseBody          onHttpBodyFunc[PluginConfig]
    onHttpStreamingResponseBody onHttpStreamingBodyFunc[PluginConfig]
    onHttpStreamDone            onHttpStreamDoneFunc[PluginConfig]
}

type CommonPluginCtx[PluginConfig any] struct {
    // proxy-wasm-go-sdk DefaultPluginContext 默认实现
    types.DefaultPluginContext
    // 解析后保存路由、域名、服务级别配置和全局插件配置
    matcher.RuleMatcher[PluginConfig]
    // 引用 CommonVmCtx
    vm          *CommonVmCtx[PluginConfig]
    // tickFunc 数组
    onTickFuncs []TickFuncEntry
}

type CommonHttpCtx[PluginConfig any] struct {
  // proxy-wasm-go-sdk DefaultHttpContext 默认实现
  types.DefaultHttpContext
  // 引用 CommonPluginCtx
  plugin                *CommonPluginCtx[PluginConfig]
  // 当前 Http 上下文下匹配插件配置，可能是路由、域名、服务级别配置或者全局配置
  config                *PluginConfig
  // 是否处理请求体
  needRequestBody       bool
  // 是否处理响应体
  needResponseBody      bool
  // 是否处理流式请求体
  streamingRequestBody  bool
  // 是否处理流式响应体
  streamingResponseBody bool
  // 非流式处理缓存请求体大小
  requestBodySize       int
  // 非流式处理缓存响应体大小
  responseBodySize      int
  // Http 上下文 ID
  contextID             uint32
  // 自定义插件设置自定义插件上下文
  userContext           map[string]interface{}
}

func main() {
  wrapper.SetCtx(
    // 插件名称
    "hello-world",
    // 设置自定义函数解析插件配置，这个方法适合插件全局配置和路由、域名、服务级别配置内容规则是一样
    wrapper.ParseConfigBy(parseConfig),
    // 设置自定义函数解析插件全局配置和路由、域名、服务级别配置，这个方法适合插件全局配置和路由、域名、服务级别配置内容规则不一样
    wrapper.ParseOverrideConfigBy(parseConfig, parseRuleConfig)
    // 设置自定义函数处理请求头
    wrapper.ProcessRequestHeadersBy(onHttpRequestHeaders),
    // 设置自定义函数处理请求体
    wrapper.ProcessRequestBodyBy(onHttpRequestBody),
    // 设置自定义函数处理响应头
    wrapper.ProcessResponseHeadersBy(onHttpResponseHeaders),
    // 设置自定义函数处理响应体
    wrapper.ProcessResponseBodyBy(onHttpResponseBody),
    // 设置自定义函数处理流式请求体
    wrapper.ProcessStreamingRequestBodyBy(onHttpStreamingRequestBody),
    // 设置自定义函数处理流式响应体
    wrapper.ProcessStreamingResponseBodyBy(onHttpStreamingResponseBody),
    // 设置自定义函数处理流式请求完成
    wrappper.ProcessStreamDoneBy(onHttpStreamDone)
  )
}

func SetCtx[PluginConfig any](pluginName string, setFuncs ...SetPluginFunc[PluginConfig]) {
  // 调用 proxywasm.SetVMContext 设置 VMContext
  proxywasm.SetVMContext(NewCommonVmCtx(pluginName, setFuncs...))
}

func NewCommonVmCtx[PluginConfig any](pluginName string, setFuncs ...SetPluginFunc[PluginConfig]) *CommonVmCtx[PluginConfig] {
  ctx := &CommonVmCtx[PluginConfig]{
    pluginName:      pluginName,
    log:             Log{pluginName},
    hasCustomConfig: true,
  }
  // CommonVmCtx 里设置自定义插件回调钩子函数
  for _, set := range setFuncs {
    set(ctx)
  }
  ...
  return ctx

func (ctx *CommonVmCtx[PluginConfig]) NewPluginContext(uint32) types.PluginContext {
  return &CommonPluginCtx[PluginConfig]{
    vm: ctx,
  }
}

func (ctx *CommonPluginCtx[PluginConfig]) OnPluginStart(int) types.OnPluginStartStatus {
  // 调用 proxywasm.GetPluginConfiguration 获取插件配置
  data, err := proxywasm.GetPluginConfiguration()
  globalOnTickFuncs = nil
  ...
  var jsonData gjson.Result
  // 插件配置转成 json
  jsonData = gjson.ParseBytes(data)

  // 设置 parseOverrideConfig
  var parseOverrideConfig func(gjson.Result, PluginConfig, *PluginConfig) error
  if ctx.vm.parseRuleConfig != nil {
    parseOverrideConfig = func(js gjson.Result, global PluginConfig, cfg *PluginConfig) error {
      // 解析插件路由、域名、服务级别插件配置
      return ctx.vm.parseRuleConfig(js, global, cfg, ctx.vm.log)
    }
  }
  ...
  // 解析插件配置
  err = ctx.ParseRuleConfig(jsonData,
    func(js gjson.Result, cfg *PluginConfig) error {
      // 解析插件全局或者当 parseRuleConfig 没有设置时候同时解析路由、域名、服务级别插件配置
      return ctx.vm.parseConfig(js, cfg, ctx.vm.log)
    },
    parseOverrideConfig,
  )
  ...
  if globalOnTickFuncs != nil {
    ctx.onTickFuncs = globalOnTickFuncs
    ...
  }
  return types.OnPluginStartStatusOK
}

apiVersion: extensions.higress.io/v1alpha1
kind: WasmPlugin
metadata:
  name: cpp-basic-auth
  namespace: higress-system
spec:
  defaultConfig:
    consumers:
      - credential: admin:123456
        name: consumer1
      - credential: guest:abc
        name: consumer2
    global_auth: false
  defaultConfigDisable: false
  matchRules:
    - config:
        allow:
          - consumer1
      configDisable: false
      ingress:
        - higress-conformance-infra/wasmplugin-cpp-basic-auth
  url: file:///opt/plugins/wasm-cpp/extensions/basic-auth/plugin.wasm

func main() {
  wrapper.SetCtx(
    "basic-auth",
    // 要同时设置 parseConfig 和 parseRuleConfig 回调钩子函数
    // ParseOverrideConfigBy 函数的第一个参数接收 parseConfig 回调钩子函数，第二个参数接收 parseRuleConfig 回调钩子函数
    wrapper.ParseOverrideConfigBy(parseGlobalConfig, parseOverrideRuleConfig),
    wrapper.ProcessRequestHeadersBy(onHttpRequestHeaders),
  )
}

// 自定义插件配置
type BasicAuthConfig struct {
  // 插件全局配置内容
  globalAuth *bool `yaml:"global_auth"`
  consumers []Consumer `yaml:"consumers"`
  ...
  // 插件路由、域名、服务级别配置内容
  allow []string `yaml:"allow"`
}

type Consumer struct {
  name string `yaml:"name"`
  credential string `yaml:"credential"`
}

// 解析插件全局配置回调钩子函数
func parseGlobalConfig(json gjson.Result, global *BasicAuthConfig, log wrapper.Log) error {
  log.Debug("global config")
  // 解析插件全局配置
  ...
  consumers := json.Get("consumers")
  ...
  globalAuth := json.Get("global_auth")
  ...
  return nil
}

// 解析插件路由、域名、服务级别配置回调钩子函数
func parseOverrideRuleConfig(json gjson.Result, global BasicAuthConfig, config *BasicAuthConfig, log wrapper.Log) error {
  log.Debug("domain/route config")
  // 这里要注意要用全局配置内容复制到路由、域名、服务级别配置中，这样后续在 HttpContext 中可以获取当前 Http 请求下插件配置包括全局配置和路由、域名、服务级别配置
  *config = global
  // 解析插件路由、域名、服务级别配置
  allow := json.Get("allow")
  ...
  for _, item := range allow.Array() {
    config.allow = append(config.allow, item.String())
  }
  ...
  return nil
}

func (ctx *CommonPluginCtx[PluginConfig]) NewHttpContext(contextID uint32) types.HttpContext {
  httpCtx := &CommonHttpCtx[PluginConfig]{
    plugin:      ctx,
    contextID:   contextID,
    userContext: map[string]interface{}{},
  }
  // 根据插件配置判断设置是否需要处理请求和响应的 body
  if ctx.vm.onHttpRequestBody != nil || ctx.vm.onHttpStreamingRequestBody != nil {
    httpCtx.needRequestBody = true
  }
  if ctx.vm.onHttpResponseBody != nil || ctx.vm.onHttpStreamingResponseBody != nil {
    httpCtx.needResponseBody = true
  }
  if ctx.vm.onHttpStreamingRequestBody != nil {
    httpCtx.streamingRequestBody = true
  }
  if ctx.vm.onHttpStreamingResponseBody != nil {
    httpCtx.streamingResponseBody = true
  }

  return httpCtx
}

func (ctx *CommonHttpCtx[PluginConfig]) OnHttpRequestHeaders(numHeaders int, endOfStream bool) types.Action {
  // 获取当前 HTTP 请求生效插件配置
  config, err := ctx.plugin.GetMatchConfig()
  ...
  // 设置插件配置到 HttpContext
  ctx.config = config
  // 如果请求 content-type 是 octet-stream/grpc 或者定义 content-encoding，则不处理请求 body
  if IsBinaryRequestBody() {
    ctx.needRequestBody = false
  }
  ...
  // 调用自定义插件 onHttpRequestHeaders 回调钩子函数
  return ctx.plugin.vm.onHttpRequestHeaders(ctx, *config, ctx.plugin.vm.log)
}

func (ctx *CommonHttpCtx[PluginConfig]) OnHttpRequestBody(bodySize int, endOfStream bool) types.Action {
  ...
  // 如果不需要处理请求 body，则直接返回，继续后续处理
  if !ctx.needRequestBody {
    return types.ActionContinue
  }
  // 先判断是否要需要进行流式处理，如果需要则调用自定义插件 onHttpStreamingRequestBody 回调钩子函数
  if ctx.plugin.vm.onHttpStreamingRequestBody != nil && ctx.streamingRequestBody {
    chunk, _ := proxywasm.GetHttpRequestBody(0, bodySize)
    // 调用自定义插件 onHttpStreamingRequestBody 回调钩子函数
    modifiedChunk := ctx.plugin.vm.onHttpStreamingRequestBody(ctx, *ctx.config, chunk, endOfStream, ctx.plugin.vm.log)
    err := proxywasm.ReplaceHttpRequestBody(modifiedChunk)
    ...
    return types.ActionContinue
  }
  // 再判断是否要需要进行非流式处理，需要缓存请求 body，等读取整个请求 body 后调用自定义插件 onHttpRequestBody 回调钩子函数
  if ctx.plugin.vm.onHttpRequestBody != nil {
    ctx.requestBodySize += bodySize
    if !endOfStream {
      return types.ActionPause
    }
    body, err := proxywasm.GetHttpRequestBody(0, ctx.requestBodySize)
    ...
    // 调用自定义插件 onHttpRequestBody 回调钩子函数
    return ctx.plugin.vm.onHttpRequestBody(ctx, *ctx.config, body, ctx.plugin.vm.log)
  }
  return types.ActionContinue
}

func (ctx *CommonHttpCtx[PluginConfig]) OnHttpResponseHeaders(numHeaders int, endOfStream bool) types.Action {
  ...
  // To avoid unexpected operations, plugins do not read the binary content body
  if IsBinaryResponseBody() {
    ctx.needResponseBody = false
  }
  ...
  return ctx.plugin.vm.onHttpResponseHeaders(ctx, *ctx.config, ctx.plugin.vm.log)
}

func (ctx *CommonHttpCtx[PluginConfig]) OnHttpResponseBody(bodySize int, endOfStream bool) types.Action {
    ...
  // 如果不需要处理响应 body，则直接返回，继续后续处理
  if !ctx.needResponseBody {
    return types.ActionContinue
  }
  // 先判断是否要需要进行流式处理，如果需要则调用自定义插件 onHttpStreamingResponseBod 回调钩子函数
  if ctx.plugin.vm.onHttpStreamingResponseBody != nil && ctx.streamingResponseBody {
    chunk, _ := proxywasm.GetHttpResponseBody(0, bodySize)
    // 调用自定义插件 onHttpStreamingResponseBody 回调钩子函数
    modifiedChunk := ctx.plugin.vm.onHttpStreamingResponseBody(ctx, *ctx.config, chunk, endOfStream, ctx.plugin.vm.log)
    ...
    return types.ActionContinue
  }
  // 再判断是否要需要进行非流式处理，需要缓存响应 body，等读取整个响应 body 后调用自定义插件 onHttpResponseBody 回调钩子函数
  if ctx.plugin.vm.onHttpResponseBody != nil {
    ctx.responseBodySize += bodySize
    if !endOfStream {
      return types.ActionPause
    }
    body, err := proxywasm.GetHttpResponseBody(0, ctx.responseBodySize)
    ...
    // 调用自定义插件 onHttpResponseBody 回调钩子函数
    return ctx.plugin.vm.onHttpResponseBody(ctx, *ctx.config, body, ctx.plugin.vm.log)
  }
  return types.ActionContinue
}

func (ctx *CommonHttpCtx[PluginConfig]) OnHttpStreamDone() {
  ...
  ctx.plugin.vm.onHttpStreamDone(ctx, *ctx.config, ctx.plugin.vm.log)
}

// 设置自定义上下文，这个上下文可以在自定义插件所有回调钩子函数中可以获取
func (ctx *CommonHttpCtx[PluginConfig]) SetContext(key string, value interface{}) {
  ctx.userContext[key] = value
}
// 获取自定义上下文，这个上下文可以在自定义插件所有回调钩子函数中可以获取
func (ctx *CommonHttpCtx[PluginConfig]) GetContext(key string) interface{} {
  return ctx.userContext[key]
}
// 获取 bool 类型自定义上下文
func (ctx *CommonHttpCtx[PluginConfig]) GetBoolContext(key string, defaultValue bool) bool {
  if b, ok := ctx.userContext[key].(bool); ok {
    return b
  }
  return defaultValue
}
// 获取 string 类型自定义上下文
func (ctx *CommonHttpCtx[PluginConfig]) GetStringContext(key, defaultValue string) string {
  if s, ok := ctx.userContext[key].(string); ok {
    return s
  }
  return defaultValue
}
// 获取请求 scheme
func (ctx *CommonHttpCtx[PluginConfig]) Scheme() string {
  proxywasm.SetEffectiveContext(ctx.contextID)
  return GetRequestScheme()
}
// 获取请求 host
func (ctx *CommonHttpCtx[PluginConfig]) Host() string {
  proxywasm.SetEffectiveContext(ctx.contextID)
  return GetRequestHost()
}
// 获取请求 path
func (ctx *CommonHttpCtx[PluginConfig]) Path() string {
  proxywasm.SetEffectiveContext(ctx.contextID)
  return GetRequestPath()
}
// 获取请求 method
func (ctx *CommonHttpCtx[PluginConfig]) Method() string {
  proxywasm.SetEffectiveContext(ctx.contextID)
  return GetRequestMethod()
}

// 调用这个方法可以禁止读取请求 body 和处理。
// 比如在 onHttpRequestHeaders 回调钩子函数中根据某些条件时调用，可以跳过后续的请求 body 读取和处理
func (ctx *CommonHttpCtx[PluginConfig]) DontReadRequestBody() {
  ctx.needRequestBody = false
}

// 调用这个方法禁止读取响应 body，
// 比如在 onHttpResponseHeaders 回调钩子函数中根据某些条件时调用，可以跳过后续的响应 body 读取和处理
func (ctx *CommonHttpCtx[PluginConfig]) DontReadResponseBody() {
  ctx.needResponseBody = false
}
// 调用这个方法禁止请求流式处理
// 比如在 onHttpRequestHeaders 回调钩子函数中根据某些条件时调用，跳过后续的 onHttpStreamingRequestBody 流式处理, 转成 onHttpRequestBody 处理
func (ctx *CommonHttpCtx[PluginConfig]) BufferRequestBody() {
  ctx.streamingRequestBody = false
}

// 调用这个方法禁止响应流式处理
// 比如在 onHttpResponseHeaders 回调钩子函数中根据某些条件时调用，跳过后续的 onHttpStreamingResponseBody 流式处理, 转成 onHttpResponseBody 处理
func (ctx *CommonHttpCtx[PluginConfig]) BufferResponseBody() {
  ctx.streamingResponseBody = false
}

// 调用这个方法可以禁止重新计算路由，Envoy 默认在 Http 头发生变更时会重新计算路由，调用这个方法，可以禁止重新计算路由。
// 比如在 onHttpRequestHeaders 回调钩子函数中根据某些条件时调用这个方法，可以禁止重新计算路由。
// 关于 DisableReroute 使用场景可以参考 Higress 官方提供 [ai-proxy 插件](https://github.com/alibaba/higress/blob/main/plugins/wasm-go/extensions/ai-proxy/main.go#L76)
func (ctx *CommonHttpCtx[PluginConfig]) DisableReroute() {
  _ = proxywasm.SetProperty([]string{"clear_route_cache"}, []byte("off"))
}
// 设置请求 body buffer limit
// 关于 DisableReroute 使用场景可以参考 Higress 官方提供 [ai-proxy 插件](https://github.com/alibaba/higress/blob/main/plugins/wasm-go/extensions/ai-proxy/main.go#L81)
func (ctx *CommonHttpCtx[PluginConfig]) SetRequestBodyBufferLimit(size uint32) {
  ctx.plugin.vm.log.Infof("SetRequestBodyBufferLimit: %d", size)
  _ = proxywasm.SetProperty([]string{"set_decoder_buffer_limit"}, []byte(strconv.Itoa(int(size))))
}
// 设置响应 body buffer limit
func (ctx *CommonHttpCtx[PluginConfig]) SetResponseBodyBufferLimit(size uint32) {
  ctx.plugin.vm.log.Infof("SetResponseBodyBufferLimit: %d", size)
  _ = proxywasm.SetProperty([]string{"set_encoder_buffer_limit"}, []byte(strconv.Itoa(int(size))))
}