Istio Pilot 深度剖析：xDS 生成器与源码跳读导航

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

// 统一生成器接口：CDS/LDS/RDS/EDS 等实现按 TypeUrl 分派
type XdsResourceGenerator interface {
    Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (
        model.Resources, model.XdsLogDetails, error,
    )
}

// 客户端对某类资源的订阅视图（RDS/EDS 高频）
type WatchedResource struct {
    TypeUrl       string      // v3.ClusterType/v3.ListenerType/v3.RouteType/v3.EndpointType
    ResourceNames []string    // 订阅的路由名/集群名等
}

// 与单个 Envoy 会话的连接上下文
type Connection struct {
    proxy            *model.Proxy
    WatchedResources map[string]*model.WatchedResource
    ConID            string      // 便于日志/排错的连接标识
}

// 一次推送请求的元数据
type PushRequest struct {
    Full           bool
    Delta          model.Delta           // 增量订阅/退订详情
    ConfigsUpdated model.ConfigKeySet    // 本次受影响的配置集合
    Push           *model.PushContext    // 只读快照：生成器核心输入
    Reason         model.ReasonStats
}

// 推送快照（高频读取）：服务/路由/策略索引，已做预计算
type PushContext struct {
    Mesh                    *meshconfig.MeshConfig
    Env                     *model.Environment
    ServiceIndex            *ServiceIndex
    VirtualServiceIndex     *VirtualServiceIndex
    DestinationRuleIndex    *DestinationRuleIndex
    SidecarScopeByNamespace map[string]*SidecarScope
}

// 代理抽象：裁剪规模与版本兼容的重要依据
type Proxy struct {
    ID           string
    IPAddresses  []string
    IstioVersion *model.IstioVersion
    Metadata     *NodeMetadata
    SidecarScope *SidecarScope
}

// Sidecar 作用域 - 决定可见的服务/主机/监听
type SidecarScope struct {
    Namespace         string
    Services          []*model.Service
    EgressHosts       []host.Name
    InboundListeners  []*ListenerConfig
    OutboundListeners []*ListenerConfig
}

// EDS 端点构建器 - 按 locality/priority/健康度 组装
type EndpointBuilder struct {
    ClusterName string
    Proxy       *model.Proxy
    Push        *model.PushContext
}

// 生成日志标记：判断增量/过滤是否生效
type XdsLogDetails struct {
    Incremental    bool
    AdditionalInfo string
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

// pilot/pkg/xds/xdsgen.go（概念化精简示例）
// pushXds 将特定 TypeUrl 的资源推送给一个连接
func (s *DiscoveryServer) pushXds(con *Connection, w *model.WatchedResource, req *model.PushRequest) error {
    if w == nil { // 连接未订阅该类型，直接返回
        return nil
    }

    // 1) 选择生成器：按 TypeUrl + 连接元数据
    gen := s.findGenerator(w.TypeUrl, con)
    if gen == nil { // 未找到生成器，跳过
            return nil
    }

    // 2) 增量订阅过滤：仅保留新增订阅资源名
    if !req.Delta.IsEmpty() && !con.proxy.IsProxylessGrpc() {
        w = &model.WatchedResource{TypeUrl: w.TypeUrl, ResourceNames: req.Delta.Subscribed}
    }

    // 3) 生成资源：读取 Proxy/PushContext/订阅名
    resources, logdata, err := gen.Generate(con.proxy, w, req)
    if err != nil || resources == nil { // 生成失败或无资源
            return err
    }

    // 4) 构造响应：携带版本与 nonce，便于 ACK/NACK 追踪
    resp := &discovery.DiscoveryResponse{
        ControlPlane: ControlPlane(w.TypeUrl),
        TypeUrl:      w.TypeUrl,
        VersionInfo:  req.Push.PushVersion,
        Nonce:        nonce(req.Push.PushVersion),
        Resources:    xds.ResourcesToAny(resources),
    }

    // 5) 发送：gRPC 流写入，失败计数 + 记录原因
    if err := xds.Send(con, resp); err != nil {
        return err
    }
    _ = logdata // 可选：附加日志（增量/过滤信息）
        return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

// 以 CDS 为例：构建出站/入站集群（多数为 EDS 类型）
type CdsGenerator struct{}

func (g CdsGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
    cg := core.NewConfigGenerator(nil)
    clusters := cg.BuildClusters(proxy, req.Push) // 读取 SidecarScope/DR 等索引
    return model.ResourcesFromClusters(clusters), model.DefaultXdsLogDetails(), nil
}

// LDS：拼装入站/出站监听器与过滤器链（含 Authn/Authz/Telemetry）
type LdsGenerator struct{}

func (g LdsGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
    listeners := req.Push.Env.ConfigGenerator().BuildListeners(proxy, req.Push)
    return model.ResourcesFromListeners(listeners), model.DefaultXdsLogDetails(), nil
}

// RDS：按订阅的 route name 定向生成 RouteConfiguration
type RdsGenerator struct{}

func (g RdsGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
    routes := req.Push.Env.ConfigGenerator().BuildHTTPRoutes(proxy, req.Push, w.ResourceNames)
    return model.ResourcesFromRoutes(routes), model.DefaultXdsLogDetails(), nil
}

// EDS：仅对订阅的 cluster 生成 CLA；按 locality/priority 聚合
type EdsGenerator struct{}

func (g EdsGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
    out := make([]*endpoint.ClusterLoadAssignment, 0, len(w.ResourceNames))
    for _, cluster := range w.ResourceNames {
        cla := v1alpha3.NewEndpointBuilder(req.Push, proxy, cluster).BuildClusterLoadAssignment()
        if cla != nil { out = append(out, cla) }
    }
    return model.ResourcesFromCLAs(out), model.DefaultXdsLogDetails(), nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

// BuildClusters：将服务/子集/策略折叠为集群，对应 DR/TLS/Outlier/ConnPool
func (cg *ConfigGeneratorImpl) BuildClusters(node *model.Proxy, push *model.PushContext) []*cluster.Cluster {
    out := make([]*cluster.Cluster, 0, 256)
    out = append(out, cg.buildOutboundClusters(node, push)...)
    out = append(out, cg.buildInboundClusters(node, push)...)
    return out
}

// BuildListeners：生成入站/出站监听器，透明代理使用 original_dst 等策略
func (cg *ConfigGeneratorImpl) BuildListeners(node *model.Proxy, push *model.PushContext) []*listener.Listener {
    var out []*listener.Listener
    out = append(out, cg.buildSidecarInboundListeners(node, push)...)
    out = append(out, cg.buildSidecarOutboundListeners(node, push)...)
    return out
}

// BuildHTTPRoutes：将 VirtualService 规则汇编为 RouteConfiguration
func (cg *ConfigGeneratorImpl) BuildHTTPRoutes(node *model.Proxy, push *model.PushContext, routeNames []string) []*route.RouteConfiguration {
    var out []*route.RouteConfiguration
    for _, name := range routeNames {
        if rc := cg.buildSidecarOutboundHTTPRouteConfig(node, push, name); rc != nil {
            out = append(out, rc)
        }
    }
    return out
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 代理状态与断面
istioctl proxy-status

# 查看集群/监听/路由/端点（必要时 -o json）
istioctl proxy-config clusters  <pod> -n <ns>
istioctl proxy-config listeners <pod> -n <ns>
istioctl proxy-config routes    <pod> -n <ns>
istioctl proxy-config endpoints <pod> -n <ns>

# 证书/密钥
istioctl proxy-config secret    <pod> -n <ns>

# 生成/下发过程日志（Pilot）
kubectl -n istio-system logs deploy/istiod -f | rg -i "PUSH|NACK|ACK|xds|nonce"