Istio Pilot控制平面深度源码剖析：配置分发与服务发现的核心实现

 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

// pilot/pkg/bootstrap/server.go
type Server struct {
    // XDS服务器 - 负责配置分发的核心组件
    XDSServer *xds.DiscoveryServer
    
    // 集群标识符
    clusterID cluster.ID
    
    // 环境上下文 - 包含网格配置、服务发现等
    environment *model.Environment
    
    // Kubernetes客户端
    kubeClient kubelib.Client
    
    // 多集群控制器
    multiclusterController *multicluster.Controller
    
    // 配置存储控制器
    configController model.ConfigStoreController
    ConfigStores     []model.ConfigStoreController
    
    // 外部服务入口控制器  
    serviceEntryController *serviceentry.Controller
    
    // HTTP服务器 - 调试、监控和就绪检查
    httpServer  *http.Server
    httpAddr    string
    
    // HTTPS服务器 - webhook处理
    httpsServer *http.Server  
    httpsAddr   string
    
    // gRPC服务器 - XDS协议通信
    grpcServer        *grpc.Server
    grpcAddress       string
    secureGrpcServer  *grpc.Server
    secureGrpcAddress string
    
    // 证书管理相关
    CA                      *ca.IstioCA
    RA                      ra.RegistrationAuthority
    caServer                *caserver.Server
    workloadTrustBundle     *tb.TrustBundle
    istiodCertBundleWatcher *keycertbundle.Watcher
    
    // 文件监听器 - 监控配置文件变化
    fileWatcher filewatcher.FileWatcher
    
    // 就绪状态探针
    readinessProbes map[string]readinessProbe
    readinessFlags  *readinessFlags
    
    // 优雅关闭持续时间
    shutdownDuration time.Duration
    
    // 内部停止信号通道
    internalStop chan struct{}
    
    // webhook信息
    webhookInfo *webhookInfo
    
    // 状态管理器
    statusManager *status.Manager
    
    // 读写配置存储
    RWConfigStore model.ConfigStoreController
}

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

// pilot/cmd/pilot-discovery/main.go
func main() {
    // 启用Klog与Cobra的集成，统一日志管理
    log.EnableKlogWithCobra()
    
    // 创建根命令，包含所有子命令和配置选项
    rootCmd := app.NewRootCommand()
    
    // 执行命令并处理错误
    if err := rootCmd.Execute(); err != nil {
        log.Error(err)
        os.Exit(-1)
    }
}

 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

// pilot/cmd/pilot-discovery/app/cmd.go
func NewRootCommand() *cobra.Command {
    rootCmd := &cobra.Command{
        Use:          "pilot-discovery",
        Short:        "Istio Pilot.",
        Long:         "Istio Pilot provides mesh-wide traffic management, security and policy capabilities in the Istio Service Mesh.",
        SilenceUsage: true,
        FParseErrWhitelist: cobra.FParseErrWhitelist{
            // 允许未知标志，保持向后兼容性
            UnknownFlags: true,
        },
        PreRunE: func(c *cobra.Command, args []string) error {
            cmd.AddFlags(c)
            return nil
        },
    }

    // 创建discovery子命令 - 这是主要的服务发现命令
    discoveryCmd := newDiscoveryCommand()
    addFlags(discoveryCmd)
    rootCmd.AddCommand(discoveryCmd)
    
    // 添加版本信息命令
    rootCmd.AddCommand(version.CobraCommand())
    
    // 添加文档生成命令
    rootCmd.AddCommand(collateral.CobraCommand(rootCmd, collateral.Metadata{
        Title:   "Istio Pilot Discovery",
        Section: "pilot-discovery CLI",
        Manual:  "Istio Pilot Discovery",
    }))
    
    // 添加请求处理命令
    rootCmd.AddCommand(requestCmd)

    return rootCmd
}

 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

func newDiscoveryCommand() *cobra.Command {
    return &cobra.Command{
        Use:   "discovery",
        Short: "Start Istio proxy discovery service.",
        Args:  cobra.ExactArgs(0),
        PreRunE: func(c *cobra.Command, args []string) error {
            // 1. 配置日志系统
            if err := log.Configure(loggingOptions); err != nil {
                return err
            }
            
            // 2. 验证命令行参数
            if err := validateFlags(serverArgs); err != nil {
                return err
            }
            
            // 3. 完成参数处理和默认值设置
            if err := serverArgs.Complete(); err != nil {
                return err
            }
            return nil
        },
        RunE: func(c *cobra.Command, args []string) error {
            cmd.PrintFlags(c.Flags())

            // 创建全局停止信号通道
            stop := make(chan struct{})

            // 创建发现服务器实例
            discoveryServer, err := bootstrap.NewServer(serverArgs)
            if err != nil {
                return fmt.Errorf("failed to create discovery service: %v", err)
            }

            // 启动服务器
            if err := discoveryServer.Start(stop); err != nil {
                return fmt.Errorf("failed to start discovery service: %v", err)
            }

            // 等待退出信号
            cmd.WaitSignal(stop)
            
            // 等待优雅关闭完成
            discoveryServer.WaitUntilCompletion()
            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
 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
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

// pilot/pkg/bootstrap/server.go
func NewServer(args *PilotArgs, initFuncs ...func(*Server)) (*Server, error) {
    // 1. 创建环境对象，包含网格配置和域名后缀
    e := model.NewEnvironment()
    e.DomainSuffix = args.RegistryOptions.KubeOptions.DomainSuffix

    // 2. 创建聚合服务发现控制器
    ac := aggregate.NewController(aggregate.Options{
        MeshHolder:      e,
        ConfigClusterID: getClusterID(args),
    })
    e.ServiceDiscovery = ac

    // 3. 注册Prometheus监控导出器
    exporter, err := monitoring.RegisterPrometheusExporter(nil, nil)
    if err != nil {
        return nil, fmt.Errorf("could not set up prometheus exporter: %v", err)
    }
    
    // 4. 创建Server结构体实例并初始化核心字段
    s := &Server{
        clusterID:               getClusterID(args),
        environment:             e,
        fileWatcher:             filewatcher.NewWatcher(),
        httpMux:                 http.NewServeMux(),
        monitoringMux:           http.NewServeMux(),
        readinessProbes:         make(map[string]readinessProbe),
        readinessFlags:          &readinessFlags{},
        server:                  server.New(),
        shutdownDuration:        args.ShutdownDuration,
        internalStop:            make(chan struct{}),
        istiodCertBundleWatcher: keycertbundle.NewWatcher(),
        webhookInfo:             &webhookInfo{},
        metricsExporter:         exporter,
        krtDebugger:             args.KrtDebugger,
    }

    // 5. 应用自定义初始化函数
    for _, fn := range initFuncs {
        fn(s)
    }
    
    // 6. 创建XDS服务器 - 这是配置分发的核心
    s.XDSServer = xds.NewDiscoveryServer(e, args.RegistryOptions.KubeOptions.ClusterAliases, args.KrtDebugger)
    configGen := core.NewConfigGenerator(s.XDSServer.Cache)

    // 7. 启用gRPC性能监控
    grpcprom.EnableHandlingTimeHistogram()

    // 8. 初始化各个组件
    s.initReadinessProbes()
    s.initServers(args)
    
    if err := s.initIstiodAdminServer(args, s.webhookInfo.GetTemplates); err != nil {
        return nil, fmt.Errorf("error initializing debug server: %v", err)
    }
    
    if err := s.serveHTTP(); err != nil {
        return nil, fmt.Errorf("error serving http: %v", err)
    }

    // 9. 初始化Kubernetes客户端
    if err := s.initKubeClient(args); err != nil {
        return nil, fmt.Errorf("error initializing kube client: %v", err)
    }

    // 10. 初始化网格配置管理
    s.initMeshConfiguration(args, s.fileWatcher)
    
    // 11. 设置Kubernetes资源过滤器
    if s.kubeClient != nil {
        namespaces := kclient.New[*corev1.Namespace](s.kubeClient)
        filter := namespace.NewDiscoveryNamespacesFilter(namespaces, s.environment.Watcher, s.internalStop)
        s.kubeClient = kubelib.SetObjectFilter(s.kubeClient, filter)
    }

    // 12. 初始化网格网络配置
    s.initMeshNetworks(args, s.fileWatcher)
    s.initMeshHandlers(configGen.MeshConfigChanged)
    
    // 13. 初始化环境和网络管理器
    s.environment.Init()
    if err := s.environment.InitNetworksManager(s.XDSServer); err != nil {
        return nil, err
    }

    // 14. 多根网格支持（可选）
    if features.MultiRootMesh {
        s.workloadTrustBundle = tb.NewTrustBundle(nil, e.Watcher)
        e.TrustBundle = s.workloadTrustBundle
    }

    // 15. 证书管理选项配置
    caOpts := &caOptions{
        TrustDomain:      s.environment.Mesh().TrustDomain,
        Namespace:        args.Namespace,
        ExternalCAType:   ra.CaExternalType(externalCaType),
        CertSignerDomain: features.CertSignerDomain,
    }

    // 16. 创建CA证书（如果需要）
    if err := s.maybeCreateCA(caOpts); err != nil {
        return nil, err
    }

    // 17. 初始化各种控制器
    if err := s.initControllers(args); err != nil {
        return nil, err
    }

    // 18. 初始化XDS生成器
    InitGenerators(s.XDSServer, configGen, args.Namespace, s.clusterID, s.internalDebugMux)

    // 19. 初始化工作负载信任包
    if err := s.initWorkloadTrustBundle(args); err != nil {
        return nil, err
    }

    return s, 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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

// pilot/pkg/xds/discovery.go
type DiscoveryServer struct {
    // 环境模型 - 包含网格配置和服务发现
    Env *model.Environment

    // 生成器映射 - 支持自定义配置生成逻辑
    // Key为生成器类型，与客户端元数据匹配选择生成器
    Generators map[string]model.XdsResourceGenerator

    // 代理推送需求判断函数 - 决定是否需要推送配置
    ProxyNeedsPush func(proxy *model.Proxy, req *model.PushRequest) (*model.PushRequest, bool)

    // 并发推送限制信号量 - 控制XDS推送的并发数量
    concurrentPushLimit chan struct{}
    
    // 请求速率限制器 - 防止新XDS请求的雷群效应
    RequestRateLimit *rate.Limiter

    // 入站更新计数器 - 收到的配置更新数量
    InboundUpdates *atomic.Int64
    
    // 已提交更新计数器 - 已处理并存储在推送上下文中的配置更新数量  
    CommittedUpdates *atomic.Int64

    // 推送通道 - 用于去抖动的缓冲区
    pushChannel chan *model.PushRequest

    // 推送队列 - 去抖动后、真正XDS推送前的缓冲区
    pushQueue *PushQueue

    // 调试处理器列表
    debugHandlers map[string]string

    // 启动时间
    DiscoveryStartTime time.Time

    // 工作负载入口控制器
    WorkloadEntryController *autoregistration.Controller

    // 配置缓存
    Cache model.XdsCache

    // 去抖动选项
    DebounceOptions DebounceOptions
    
    // 服务器就绪状态
    serverReady atomic.Bool
    
    // 认证器列表
    Authenticators []security.Authenticator
}

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

// pilot/pkg/xds/discovery.go 
func (s *DiscoveryServer) Start(stopCh <-chan struct{}) {
    // 1. 启动工作负载入口控制器 - 管理动态工作负载注册
    go s.WorkloadEntryController.Run(stopCh)
    
    // 2. 启动更新处理协程 - 处理配置更新的去抖动逻辑
    go s.handleUpdates(stopCh)
    
    // 3. 启动指标刷新协程 - 定期更新推送状态指标  
    go s.periodicRefreshMetrics(stopCh)
    
    // 4. 启动推送处理协程 - 执行实际的配置推送
    go s.sendPushes(stopCh)
    
    // 5. 启动配置缓存 - 管理生成的配置缓存
    go s.Cache.Run(stopCh)
}

 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

// ConfigUpdate实现ConfigUpdater接口，用于请求配置推送
func (s *DiscoveryServer) ConfigUpdate(req *model.PushRequest) {
    // 1. 安全断言检查 - 确保Service类型不会设置在ConfigKey中
    if features.EnableUnsafeAssertions {
        if model.HasConfigsOfKind(req.ConfigsUpdated, kind.Service) {
            panic("assertion failed kind.Service can not be set in ConfigKey")
        }
    }
    
    // 2. 地址类型特殊处理 - 清除所有缓存因为Address类型是动态获取的
    if model.HasConfigsOfKind(req.ConfigsUpdated, kind.Address) {
        s.Cache.ClearAll()
    }
    
    // 3. 更新统计指标
    inboundConfigUpdates.Increment()
    s.InboundUpdates.Inc()
    
    // 4. 全量推送调试日志
    if req.Full && fullPushLog.DebugEnabled() {
        configs := slices.Sort(slices.Map(req.ConfigsUpdated.UnsortedList(), model.ConfigKey.String))
        reasons := maps.Keys(req.Reason)
        fullPushLog.Debugf("full push triggered configs=%v reasons=%v", configs, reasons)
    }
    
    // 5. 将请求发送到推送通道，触发去抖动处理
    s.pushChannel <- req
}

 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
70

// handleUpdates在独立线程中处理pushChannel的事件
// 确保在最后一个事件后至少经过minQuiet时间才处理
// 同时确保接收事件和处理事件之间最多间隔maxDelay时间
func (s *DiscoveryServer) handleUpdates(stopCh <-chan struct{}) {
    debounce(s.pushChannel, stopCh, s.DebounceOptions, s.Push, s.CommittedUpdates)
}

// 去抖动辅助函数实现
func debounce(ch chan *model.PushRequest, stopCh <-chan struct{}, opts DebounceOptions, pushFn func(req *model.PushRequest), updateSent *atomic.Int64) {
    var timeChan <-chan time.Time
    var startDebounce time.Time
    var lastConfigUpdateTime time.Time

    pushCounter := 0
    debouncedEvents := 0

    // 跟踪推送请求。如果更新被去抖动，它们会被合并
    var req *model.PushRequest

    free := true
    freeCh := make(chan struct{}, 1)

    push := func(req *model.PushRequest, debouncedEvents int, startDebounce time.Time) {
        pushFn(req)
        updateSent.Add(int64(debouncedEvents))
        debounceTime.Record(time.Since(startDebounce).Seconds())
        freeCh <- struct{}{}
    }

    pushWorker := func(req *model.PushRequest, debouncedEvents int, startDebounce time.Time) {
        go push(req, debouncedEvents, startDebounce)
    }

    for {
        select {
        case <-freeCh:
            free = true
            pushCounter++
        case r := <-ch:
            // 合并配置更新请求
            if req == nil {
                req = r
            } else {
                req = req.Merge(r)
            }
            lastConfigUpdateTime = time.Now()
            
            if debouncedEvents == 0 {
                startDebounce = lastConfigUpdateTime
                timeChan = time.After(opts.debounceAfter)
            }
            debouncedEvents++

        case <-timeChan:
            if free {
                free = false
                pushWorker(req, debouncedEvents, startDebounce)
                
                req = nil
                debouncedEvents = 0
                timeChan = nil
            } else {
                // 如果推送正在进行中，短时间后重试
                timeChan = time.After(opts.debounceAfter / 10)
            }
        case <-stopCh:
            return
        }
    }
}

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

// pilot/pkg/xds/ads.go （概念化示例）
func (s *DiscoveryServer) handleRequest(con *Connection, req *discovery.DiscoveryRequest) error {
    // 1. 记录客户端已知的version_info与response_nonce
    typeURL := req.TypeUrl
    nonce := req.ResponseNonce
    version := req.VersionInfo

    // 2. 处理NACK：当ErrorDetail非空且nonce匹配，标记上次推送失败
    if req.ErrorDetail != nil && nonce != "" {
        nackErrors.With(typeTag.Value(typeURL)).Increment()
        s.onNack(con, typeURL, nonce, req.ErrorDetail)
        return nil // NACK不触发立即重推，由后续变更或重试驱动
    }

    // 3. 处理ACK：记录客户端接受的最新版本
    if nonce != "" && req.ErrorDetail == nil {
        s.onAck(con, typeURL, version, nonce)
        return nil
    }

    // 4. 首次订阅/资源变更：生成并发送资源
    watched := con.Watched(typeURL)
    return s.pushXds(con, watched, &model.PushRequest{Full: false})
}

 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

// pilot/pkg/serviceregistry/aggregate/controller.go
type Controller struct {
    // 网格持有者 - 提供网格配置
    meshHolder mesh.Holder
    
    // 注册的服务发现实例列表
    registries []serviceregistry.Instance
    
    // 服务处理器列表
    serviceHandlers []model.ServiceHandler
    
    // 工作负载处理器列表  
    workloadHandlers []func(*model.WorkloadInstance, model.Event)
    
    // 网络网关处理器
    model.NetworkGatewaysHandler
    
    // 读写锁保护并发访问
    storeLock sync.RWMutex
}

// 添加服务发现注册表
func (c *Controller) AddRegistry(registry serviceregistry.Instance) {
    c.storeLock.Lock()
    defer c.storeLock.Unlock()
    
    c.registries = append(c.registries, registry)
    
    // 为新注册表注册已有的服务处理器
    for _, h := range c.serviceHandlers {
        registry.AppendServiceHandler(h)
    }
    
    // 为新注册表注册已有的工作负载处理器
    for _, h := range c.workloadHandlers {
        registry.AppendWorkloadHandler(h)
    }
}

// 获取所有服务的聚合视图
func (c *Controller) Services() []*model.Service {
    // 去重和合并逻辑
    services := map[host.Name]*model.Service{}
    
    c.storeLock.RLock()
    defer c.storeLock.RUnlock()
    
    for _, r := range c.registries {
        for _, s := range r.Services() {
            sp, f := services[s.Hostname]
            if !f {
                services[s.Hostname] = s
                continue
            }
            // 如果服务已存在，需要合并端口信息
            sp.Ports = model.MergeServicePorts(sp.Ports, s.Ports, sp.Hostname)
        }
    }
    
    out := make([]*model.Service, 0, len(services))
    for _, service := range services {
        out = append(out, service)
    }
    return out
}

 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
70
71
72
73

// pilot/pkg/serviceregistry/kube/controller/controller.go - Informer事件处理
func (c *Controller) setupInformers() error {
    // 1. Service资源Informer
    c.serviceInformer = cache.NewSharedIndexInformer(
        &cache.ListWatch{
            ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) {
                return c.client.CoreV1().Services(metav1.NamespaceAll).List(context.TODO(), opts)
            },
            WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
                return c.client.CoreV1().Services(metav1.NamespaceAll).Watch(context.TODO(), opts)
            },
        },
        &corev1.Service{},
        resyncPeriod,
        cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
    )
    
    // 注册Service事件处理器
    c.serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            c.onServiceEvent(obj.(*corev1.Service), model.EventAdd)
        },
        UpdateFunc: func(oldObj, newObj interface{}) {
            // 比较资源版本避免重复处理
            oldSvc := oldObj.(*corev1.Service)
            newSvc := newObj.(*corev1.Service)
            if oldSvc.ResourceVersion != newSvc.ResourceVersion {
                c.onServiceEvent(newSvc, model.EventUpdate)
            }
        },
        DeleteFunc: func(obj interface{}) {
            if svc, ok := obj.(*corev1.Service); ok {
                c.onServiceEvent(svc, model.EventDelete)
            }
        },
    })
    
    // 2. EndpointSlice资源Informer
    c.endpointSliceInformer = cache.NewSharedIndexInformer(
        &cache.ListWatch{
            ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) {
                return c.client.DiscoveryV1().EndpointSlices(metav1.NamespaceAll).List(context.TODO(), opts)
            },
            WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
                return c.client.DiscoveryV1().EndpointSlices(metav1.NamespaceAll).Watch(context.TODO(), opts)
            },
        },
        &discoveryv1.EndpointSlice{},
        resyncPeriod,
        cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
    )
    
    // 注册EndpointSlice事件处理器
    c.endpointSliceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            c.onEndpointSliceEvent(obj.(*discoveryv1.EndpointSlice), model.EventAdd)
        },
        UpdateFunc: func(oldObj, newObj interface{}) {
            oldES := oldObj.(*discoveryv1.EndpointSlice)
            newES := newObj.(*discoveryv1.EndpointSlice)
            if !reflect.DeepEqual(oldES.Endpoints, newES.Endpoints) {
                c.onEndpointSliceEvent(newES, model.EventUpdate)
            }
        },
        DeleteFunc: func(obj interface{}) {
            if es, ok := obj.(*discoveryv1.EndpointSlice); ok {
                c.onEndpointSliceEvent(es, model.EventDelete)
            }
        },
    })
    
    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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

// 基于业界实践的服务发现优化策略
type OptimizedServiceDiscovery struct {
    // 分层缓存机制
    l1Cache *sync.Map           // 内存一级缓存
    l2Cache *persistent.Cache   // 持久化二级缓存
    
    // 事件去重器
    eventDeduplicator *EventDeduplicator
    
    // 批量处理队列
    batchProcessor *BatchProcessor
    
    // 性能监控
    metrics *ServiceDiscoveryMetrics
}

// 事件去重逻辑
type EventDeduplicator struct {
    recentEvents *lru.Cache      // LRU缓存近期事件
    window      time.Duration    // 去重时间窗口
}

func (ed *EventDeduplicator) isDuplicate(event ServiceEvent) bool {
    key := fmt.Sprintf("%s-%s-%s", event.Type, event.Namespace, event.Name)
    
    if lastTime, exists := ed.recentEvents.Get(key); exists {
        if time.Since(lastTime.(time.Time)) < ed.window {
            return true  // 重复事件，跳过处理
        }
    }
    
    ed.recentEvents.Add(key, time.Now())
    return false
}

// 批量处理机制
func (c *OptimizedServiceDiscovery) processBatch(events []ServiceEvent) {
    // 1. 按命名空间分组
    eventsByNamespace := make(map[string][]ServiceEvent)
    for _, event := range events {
        eventsByNamespace[event.Namespace] = append(eventsByNamespace[event.Namespace], event)
    }
    
    // 2. 并行处理各命名空间的事件
    var wg sync.WaitGroup
    for ns, nsEvents := range eventsByNamespace {
        wg.Add(1)
        go func(namespace string, events []ServiceEvent) {
            defer wg.Done()
            c.processNamespaceEvents(namespace, events)
        }(ns, nsEvents)
    }
    
    wg.Wait()
    
    // 3. 触发全局配置更新
    c.triggerConfigUpdate()
}

 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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

// pilot/pkg/serviceregistry/kube/controller/controller.go
type Controller struct {
    // Kubernetes客户端
    client kubelib.Client
    
    // 域名后缀  
    domainSuffix string
    
    // 集群ID
    clusterID cluster.ID
    
    // XDS更新器
    xdsUpdater model.XDSUpdater
    
    // 网络ID
    networkID network.ID
    
    // 服务处理器列表
    serviceHandlers []model.ServiceHandler
    
    // 工作负载处理器列表
    workloadHandlers []func(*model.WorkloadInstance, model.Event)
    
    // Pod到工作负载的映射
    pods *krt.Collection[*v1.Pod]
    
    // 服务到模型服务的映射  
    services *krt.Collection[*v1.Service]
    
    // 端点切片集合
    endpointSlices *krt.Collection[*v1.EndpointSlice]
    
    // 网格配置监听器
    meshWatcher mesh.Watcher
    
    // 命名空间控制器
    namespaceController namespace.Controller
    
    // 停止信号
    stop <-chan struct{}
}

// 服务事件处理器
func (c *Controller) onServiceEvent(svc *v1.Service, event model.Event) error {
    log.Debugf("Handle service %s in namespace %s", svc.Name, svc.Namespace)
    
    // 转换Kubernetes Service到Istio模型
    svcConv := kube.ConvertService(*svc, c.domainSuffix, c.clusterID)
    
    switch event {
    case model.EventAdd:
        log.Infof("Service %s/%s added", svc.Namespace, svc.Name)
        
    case model.EventUpdate:
        log.Debugf("Service %s/%s updated", svc.Namespace, svc.Name)
        
    case model.EventDelete:
        log.Infof("Service %s/%s deleted", svc.Namespace, svc.Name)
    }
    
    // 通知所有服务处理器
    for _, handler := range c.serviceHandlers {
        handler(svcConv, nil, event)
    }
    
    return nil
}

// 端点切片事件处理器  
func (c *Controller) onEndpointSliceEvent(es *v1.EndpointSlice, event model.Event) error {
    serviceName := es.Labels[v1.LabelServiceName]
    if serviceName == "" {
        log.Warnf("EndpointSlice %s/%s missing service name label", es.Namespace, es.Name)
        return nil
    }
    
    log.Debugf("Handle endpoint slice %s in namespace %s for service %s", es.Name, es.Namespace, serviceName)
    
    // 转换端点切片到Istio端点
    endpoints := kube.ConvertEndpointSlice(es, c.domainSuffix, c.clusterID)
    
    // 通过XDS更新器推送端点变更
    if c.xdsUpdater != nil {
        shard := model.ShardKey{
            Cluster: c.clusterID,
            Provider: provider.Kubernetes,
        }
        
        c.xdsUpdater.EDSUpdate(shard, serviceName, es.Namespace, endpoints)
    }
    
    return nil
}