Wasm 插件原理

vm_config:
  vm_id: "foo"
  runtime: "envoy.wasm.runtime.v8"
  configuration:
    "@type": type.googleapis.com/google.protobuf.StringValue
    value: '{"my-vm-env": "dev"}'
  code:
    local:
      filename: "example.wasm"
configuration:
  "@type": type.googleapis.com/google.protobuf.StringValue
  value: '{"my-plugin-config": "bar"}'

http_filters:
  - name: envoy.filters.http.wasm
    typed_config:
      "@type": type.googleapis.com/udpa.type.v1.TypedStruct
      type_url: type.googleapis.com/envoy.extensions.filters.http.wasm.v3.Wasm
      value:
        config:
          configuration:
            "@type": type.googleapis.com/google.protobuf.StringValue
            value: |
              {
                "header": "x-wasm-header",
                "value": "demo-wasm"
              }
          vm_config:
            runtime: "envoy.wasm.runtime.v8"
            code:
              local:
                filename: "./examples/http_headers/main.wasm"
  - name: envoy.filters.http.router
    typed_config:
      "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router

filter_chains:
- filters:
    - name: envoy.filters.network.wasm
      typed_config:
        "@type": type.googleapis.com/envoy.extensions.filters.network.wasm.v3.Wasm
        config:
          vm_config: { ... }
          # ... plugin config follows
    - name: envoy.tcp_proxy

bootstrap_extensions:
- name: envoy.bootstrap.wasm
  typed_config:
    "@type": type.googleapis.com/envoy.extensions.wasm.v3.WasmService
    singleton: true
    config:
      vm_config: { ... }
      # ... plugin config follows

static_resources:
  listeners:
    - name: http-header-operation
      address:
        socket_address:
          address: 0.0.0.0
          port_value: 18000
      filter_chains:
        - filters:
            - name: envoy.http_connection_manager
              typed_config:
                # ....
                http_filters:
                  - name: envoy.filters.http.wasm
                    typed_config:
                      "@type": type.googleapis.com/envoy.extensions.filters.http.wasm.v3.Wasm
                      config:
                        configuration:
                          "@type": type.googleapis.com/google.protobuf.StringValue
                          value: "http-header-operation"
                        vm_config:
                          vm_id: "my-vm-id"
                          runtime: "envoy.wasm.runtime.v8"
                          configuration:
                            "@type": type.googleapis.com/google.protobuf.StringValue
                            value: "my-vm-configuration"
                          code:
                            local:
                              filename: "all-in-one.wasm"
                  - name: envoy.filters.http.router

    - name: http-body-operation
      address:
        socket_address:
          address: 0.0.0.0
          port_value: 18001
      filter_chains:
        - filters:
            - name: envoy.http_connection_manager
              typed_config:
                # ....
                http_filters:
                  - name: envoy.filters.http.wasm
                    typed_config:
                      "@type": type.googleapis.com/envoy.extensions.filters.http.wasm.v3.Wasm
                      config:
                        configuration:
                          "@type": type.googleapis.com/google.protobuf.StringValue
                          value: "http-body-operation"
                        vm_config:
                          vm_id: "my-vm-id"
                          runtime: "envoy.wasm.runtime.v8"
                          configuration:
                            "@type": type.googleapis.com/google.protobuf.StringValue
                            value: "my-vm-configuration"
                          code:
                            local:
                              filename: "all-in-one.wasm"
                  - name: envoy.filters.http.router

    - name: tcp-total-data-size-counter
      address:
        socket_address:
          address: 0.0.0.0
          port_value: 18002
      filter_chains:
        - filters:
            - name: envoy.filters.network.wasm
              typed_config:
                "@type": type.googleapis.com/envoy.extensions.filters.network.wasm.v3.Wasm
                config:
                  configuration:
                    "@type": type.googleapis.com/google.protobuf.StringValue
                    value: "tcp-total-data-size-counter"
                    vm_config:
                      vm_id: "my-vm-id"
                      runtime: "envoy.wasm.runtime.v8"
                      configuration:
                        "@type": type.googleapis.com/google.protobuf.StringValue
                        value: "my-vm-configuration"
                      code:
                        local:
                          filename: "all-in-one.wasm"
            - name: envoy.tcp_proxy
              typed_config: # ...

                    Wasm Virtual Machine
                      (.vm_config.code)
┌────────────────────────────────────────────────────────────────┐
│  Your program (.vm_config.code)                TcpContext      │
│          │                                  ╱ (Tcp stream)     │
│          │ 1: 1                            ╱                   │
│          │         1: N                   ╱ 1: N               │
│      VMContext  ──────────  PluginContext                      │
│                                (Plugin)   ╲ 1: N               │
│                                            ╲                   │
│                                             ╲  HttpContext     │
│                                               (Http stream)    │
└────────────────────────────────────────────────────────────────┘

type VMContext interface {
  // OnVMStart 在 VM 创建和调用 main 函数后被调用。
  // 在此调用期间，可以通过 GetVMConfiguration 获取在 vm_config.configuration 设置的配置。
  // 这主要用于执行 Wasm VM 范围内的初始化。
  OnVMStart(vmConfigurationSize int) OnVMStartStatus

  // NewPluginContext 用于为每个插件配置创建 PluginContext。
  NewPluginContext(contextID uint32) PluginContext
}

type PluginContext interface {
  // OnPluginStart 在所有插件上下文上调用（如果在这是 VM 上下文，则在 OnVmStart 之后）。
  // 在此调用期间，可以通过 GetPluginConfiguration 获取 envoy.yaml 中 config.configuration 设置的配置。
  OnPluginStart(pluginConfigurationSize int) OnPluginStartStatus

  // 以下函数用于在流上创建上下文，
  // *必须* 实现它们中的任一个，对应于扩展点。例如，如果您配置此插件上下文在 Http 过滤器上运行，那么必须实现 NewHttpContext。
  // 对 Tcp 过滤器也是如此。
  //
  // NewTcpContext 用于为每个 Tcp 流创建 TcpContext。
  NewTcpContext(contextID uint32) TcpContext
  // NewHttpContext 用于为每个 Http 流创建 HttpContext。
  NewHttpContext(contextID uint32) HttpContext
}

func main() {
  proxywasm.SetVMContext(&myVMContext{})
}

type myVMContext struct { .... }

var _ types.VMContext = &myVMContext{}

// Implementations follow...

// GetSharedData 用于检索给定 "key" 的值。
// 返回的 "cas" 应用于 SetSharedData 以实现该键的线程安全更新。
func GetSharedData(key string) (value []byte, cas uint32, err error)

// SetSharedData 用于在共享数据存储中设置键值对。
// 共享数据存储按主机中的 "vm_config.vm_id" 定义。
//
// 当给定的 CAS 值与当前值不匹配时，将返回 ErrorStatusCasMismatch。
// 这表明其他 Wasm VM 已经成功设置相同键的值，并且该键的当前 CAS 已递增。
// 建议在遇到此错误时实现重试逻辑。
//
// 将 cas 设置为 0 将永远不会返回 ErrorStatusCasMismatch 并且总是成功的，
// 但这并不是线程安全的，即可能在您调用此函数时另一个 VM 已经设置了该值，
// 看到的值与存储时的值已经不同。
func SetSharedData(key string, value []byte, cas uint32) error

// DequeueSharedQueue 从给定 queueID 的共享队列中出队数据。
// 要获取目标队列的 queue id，请先使用 "ResolveSharedQueue"。
func DequeueSharedQueue(queueID uint32) ([]byte, error)

// RegisterSharedQueue 在此插件上下文中注册共享队列。
// "注册" 意味着每当该 queueID 上有新数据入队时，将对此插件上下文调用 OnQueueReady。
// 仅适用于 types.PluginContext。返回的 queueID 可用于 Enqueue/DequeueSharedQueue。
// 请注意 "name" 必须在所有共享相同 "vm_id" 的 Wasm VMs 中是唯一的。使用 "vm_id" 来分隔共享队列的命名空间。
//
// 只有在调用 RegisterSharedQueue 之后，ResolveSharedQueue("此 vm_id", "名称") 才能成功
// 通过其他 VMs 检索 queueID。
func RegisterSharedQueue(name string) (queueID uint32, err error)

// EnqueueSharedQueue 将数据入队到给定 queueID 的共享队列。
// 要获取目标队列的 queue id，请先使用 "ResolveSharedQueue"。
func EnqueueSharedQueue(queueID uint32, data []byte) error

// ResolveSharedQueue 获取给定 vmID 和队列名称的 queueID。
// 返回的 queueID 可用于 Enqueue/DequeueSharedQueue。
func ResolveSharedQueue(vmID, queueName string) (queueID uint32, err error)

package main

import (
  "github.com/higress-group/proxy-wasm-go-sdk/proxywasm"
  "github.com/higress-group/proxy-wasm-go-sdk/proxywasm/types"
)

// 插件入口
func main() {
  proxywasm.SetVMContext(&vmContext{})
}

// VM 上下文
type vmContext struct {
  // Embed the default VM context here,
  types.DefaultVMContext
  // 这里添加 VM 配置
}

// VM 启动回调
func (*vmContext) OnVMStart(vmConfigurationSize int) types.OnVMStartStatus {
  proxywasm.LogInfof("OnVMStart()")
  // 获取 VM 配置
  _, err := proxywasm.GetVMConfiguration()
  if err != nil {
    proxywasm.LogCriticalf("error reading vm configuration: %v", err)
  }
  // 这里解析 VM 配置
  return types.OnVMStartStatusOK
}

// 生成插件上下文
func (*vmContext) NewPluginContext(contextID uint32) types.PluginContext {
  proxywasm.LogInfof("NewPluginContex()")
  return &pluginContext{}
}

// 插件上下文
type pluginContext struct {
  // Embed the default plugin context here,
  types.DefaultPluginContext
  // 这里添加插件配置
}

// Http 上下文
type httpContext struct {
  // Embed the default root http context here,
  // so that we don't need to reimplement all the methods.
  types.DefaultHttpContext
  // 这里添加http 上下文属性
  requestBodySize       int
  responseBodySize      int
}

// 生成 Http 上下文
func (ctx *pluginContext) NewHttpContext(contextID uint32) types.HttpContext {
  proxywasm.LogInfof("NewHttpContext()")
  return &httpContext{}
}

// 插件启动回调，
func (ctx *pluginContext) OnPluginStart(pluginConfigurationSize int) types.OnPluginStartStatus {
  proxywasm.LogInfof("OnPluginStart()")
  // 获取插件配置
  _, err := proxywasm.GetPluginConfiguration()
  if err != nil {
    proxywasm.LogCriticalf("error reading plugin configuration: %v", err)
  }
  // 这里解析插件配置

  return types.OnPluginStartStatusOK
}

// http 请求头回调
func (ctx *httpContext) OnHttpRequestHeaders(numHeaders int, endOfStream bool) types.Action {
  proxywasm.LogInfof("OnHttpRequestHeaders()")
  // 这里处理请求头回调
  return types.ActionContinue
}

// http 请求体回调，注意这里流式处理
func (ctx *httpContext) OnHttpRequestBody(bodySize int, endOfStream bool) types.Action {
  proxywasm.LogInfof("OnHttpRequestBody()")
  ctx.requestBodySize += bodySize
  if !endOfStream {
    // Wait until we see the entire body to replace.
    return types.ActionPause
  }
  _, err := proxywasm.GetHttpRequestBody(0, ctx.requestBodySize)
  if err != nil {
    proxywasm.LogErrorf("failed to get request body: %v", err)
    return types.ActionContinue
  }

  return types.ActionContinue
}

// http 响应头回调
func (ctx *httpContext) OnHttpResponseHeaders(numHeaders int, endOfStream bool) types.Action {
  proxywasm.LogInfof("OnHttpResponseHeaders()")
  // 这里响应头回调
  return types.ActionContinue
}

// http 响应体回调， 注意这里流式处理
func (ctx *httpContext) OnHttpResponseBody(bodySize int, endOfStream bool) types.Action {
  proxywasm.LogInfof("OnHttpResponseBody()")
  ctx.responseBodySize += bodySize
  // 判断是否响应体结束
  if !endOfStream {
    // Wait until we see the entire body to replace.
    return types.ActionPause
  }
  _, err := proxywasm.GetHttpResponseBody(0, ctx.responseBodySize)
  if err != nil {
    proxywasm.LogErrorf("failed to get response body: %v", err)
    return types.ActionContinue
  }
  return types.ActionContinue
}