从源码看istio如何下发wasmplugin

背景

记录istio中wasmplugin如何生效相关源码

简单总结

pilot-agent从istiod收到关于wasmplugin的xds响应
pilot-agent下载wasmplugin文件到本地缓存
pilot-agent转换wasmplugin配置为本地将转换后的响应发送给envoy
envoy加载wasmplugin

源码

pkg/istio-agent/xds_proxy.go中

处理xds请求
func (p *XdsProxy) handleUpstreamResponse(con *ProxyConnection) {
    ...
    处理从istiod收到的xds响应进行转换后发送给envoy
                    go p.rewriteAndForward(con, resp, func(resp *discovery.DiscoveryResponse) {
                        select {
                        case forwardEnvoyCh <- resp:
                        case <-con.stopChan:
                        }
                    })
    ...
}

转换wasmplugin配置并转发给envoy
func (p *XdsProxy) rewriteAndForward(con *ProxyConnection, resp *discovery.DiscoveryResponse, forward func(resp *discovery.DiscoveryResponse)) {
    ...
    转换wasmplugin配置
    if err := wasm.MaybeConvertWasmExtensionConfig(resp.Resources, p.wasmCache); err != nil {
        ...
        return
    }
    转发转换后的响应
    forward(resp)
    ...
}

pkg/wasm/convert.go中

转换wasmplugin配置
func MaybeConvertWasmExtensionConfig(resources []*anypb.Any, cache Cache) error {
    ...
    numResources := len(resources)
    ...
    并发转换wasmplugin配置
    for i := 0; i < numResources; i++ {
        go func(i int) {
            ...
            这里以httpwasmplugin配置为例
                newExtensionConfig, err := convertHTTPWasmConfigFromRemoteToLocal(extConfig, wasmHTTPConfig, cache)
            ...
                resources[i] = newExtensionConfig
            ...
        }
    }
    ...
    wg.Wait()
    ...
}

转换httpwasmplugin配置
func convertHTTPWasmConfigFromRemoteToLocal(ec *core.TypedExtensionConfig, wasmHTTPFilterConfig *httpwasm.Wasm, cache Cache) (*anypb.Any, error) {
    ...
    重写vmconfig中的remote地址为local地址
    err := rewriteVMConfig(ec.Name, wasmHTTPFilterConfig.Config.GetVmConfig(), &status, cache, wasmHTTPFilterConfig.Config.Name)
    if err != nil {
        return nil, err
    }
    ...
}

重写vmconfig中的remote地址为local地址
func rewriteVMConfig(resourceName string, vm *wasmextensions.VmConfig, status *string, cache Cache, configName string) error {
    ...
    获取wamsplugin文件路径
    f, err := cache.Get(httpURI.GetUri(), GetOptions{
        Checksum:        remote.Sha256,
        ResourceName:    resourceName,
        ResourceVersion: resourceVersion,
        RequestTimeout:  timeout,
        PullSecret:      pullSecret,
        PullPolicy:      pullPolicy,
    })
    ...
    重写vmconfig中的remote地址为local地址
    vm.Code = &core.AsyncDataSource{
        Specifier: &core.AsyncDataSource_Local{
            Local: &core.DataSource{
                Specifier: &core.DataSource_Filename{
                    Filename: f,
                },
            },
        },
    }
    ...
}

pkg/wasm/cache.go中

获取wamsplugin文件路径
func (c *LocalFileCache) Get(downloadURL string, opts GetOptions) (string, error) {
    ...
    从缓存中获取或下载文件
    entry, err := c.getOrFetch(key, opts)
    if err != nil {
        return "", err
    }
    ...
}

从缓存中获取或下载文件
func (c *LocalFileCache) getOrFetch(key cacheKey, opts GetOptions) (*cacheEntry, error) {
    ...
    检查本地缓存是否匹配
    ce, checksum := c.getEntry(key, shouldIgnoreResourceVersion(opts.PullPolicy, u))
    if ce != nil {
        缓存匹配直接返回
        return ce, nil
    }
    ...
    下载wasmplugin文件
        b, err = c.httpFetcher.Fetch(ctx, key.downloadURL, insecure)
    ...
    添加缓存
    return c.addEntry(key, b)
}