Repository: istio/istio
File: pkg/kube/multicluster/secretcontroller.go (Go)

Code before:
	namespaces := kclient.New[*corev1.Namespace](kubeclientset)
	controller.DiscoveryNamespacesFilter = filter.NewDiscoveryNamespacesFilter(namespaces, meshWatcher.Mesh().GetDiscoverySelectors())
	controller.queue = controllers.NewQueue("multicluster secret",
		controllers.WithMaxAttempts(maxRetries),
		controllers.WithReconciler(controller.processItem))

	_, _ = secretsInformer.AddEventHandler(controllers.ObjectHandler(controller.queue.AddObject))
	return controller
}

func (c *Controller) AddHandler(h ClusterHandler) {
	c.handlers = append(c.handlers, h)
}

// Run starts the controller until it receives a message over stopCh
func (c *Controller) Run(stopCh <-chan struct{}) error {
	// run handlers for the config cluster; do not store this *Cluster in the ClusterStore or give it a SyncTimeout
	// this is done outside the goroutine, we should block other Run/startFuncs until this is registered
	configCluster := &Cluster{Client: c.configClusterClient, ID: c.configClusterID}
	if err := c.handleAdd(configCluster, stopCh); err != nil {
		return fmt.Errorf("failed initializing primary cluster %s: %v", c.configClusterID, err)
	}
	go func() {
		t0 := time.Now()
		log.Info("Starting multicluster remote secrets controller")

		go c.informer.Run(stopCh)

		if !kube.WaitForCacheSync("multicluster remote secrets", stopCh, c.informer.HasSynced) {
			return
		}
		log.Infof("multicluster remote secrets controller cache synced in %v", time.Since(t0))
		c.queue.Run(stopCh)
	}()
	return nil
}

func (c *Controller) HasSynced() bool {
	if !c.queue.HasSynced() {
		log.Debug("secret controller did not sync secrets presented at startup")
		// we haven't finished processing the secrets that were present at startup
		return false
	}
	c.cs.RLock()
	defer c.cs.RUnlock()
	for _, clusterMap := range c.cs.remoteClusters {
		for _, cluster := range clusterMap {
			if !cluster.HasSynced() {
				log.Debugf("remote cluster %s registered informers have not been synced up yet", cluster.ID)
				return false
			}

Review comment (bug):
So we need this because we don't enable the handlers for it until initDiscoveryHandlers in kube/controller...?

That seems a bit odd. Why don't we just add the handlers here and wait for it to sync? 

I guess its maybe wrong since the queue would be different.

Overall it seems the design of DiscoveryFilter may be causing some issues. I would think we should have the DiscoveryFilter work on its own (and we run it fully here), and allow other controllers to register callbacks to it?

On a related note - do other users like NamespaceController react to changes in DiscoveryFilter properly? I suspect there is some issues there..

Code after:
	}

	namespaces := kclient.New[*corev1.Namespace](kubeclientset)
	controller.namespaces = namespaces
	controller.DiscoveryNamespacesFilter = filter.NewDiscoveryNamespacesFilter(namespaces, meshWatcher.Mesh().GetDiscoverySelectors())
	controller.queue = controllers.NewQueue("multicluster secret",
		controllers.WithMaxAttempts(maxRetries),
		controllers.WithReconciler(controller.processItem))

	_, _ = secretsInformer.AddEventHandler(controllers.ObjectHandler(controller.queue.AddObject))
	return controller
}

func (c *Controller) AddHandler(h ClusterHandler) {
	c.handlers = append(c.handlers, h)
}

// Run starts the controller until it receives a message over stopCh
func (c *Controller) Run(stopCh <-chan struct{}) error {
	// wait for namespace informer synced
	if !kube.WaitForCacheSync("namespace", stopCh, c.namespaces.HasSynced) {
		return fmt.Errorf("failed to sync namespaces")
	}
	// run discovery namespace filter initiation before kube registry run
	// otherwise istio may do some useless xds push for partial resources before namespace filter has been been completely initiated.
	c.DiscoveryNamespacesFilter.SyncNamespaces()
	// run handlers for the config cluster; do not store this *Cluster in the ClusterStore or give it a SyncTimeout
	// this is done outside the goroutine, we should block other Run/startFuncs until this is registered
	configCluster := &Cluster{Client: c.configClusterClient, ID: c.configClusterID}
	if err := c.handleAdd(configCluster, stopCh); err != nil {
		return fmt.Errorf("failed initializing primary cluster %s: %v", c.configClusterID, err)
	}
	go func() {
		t0 := time.Now()
		log.Info("Starting multicluster remote secrets controller")

		go c.informer.Run(stopCh)

		if !kube.WaitForCacheSync("multicluster remote secrets", stopCh, c.informer.HasSynced) {
			return
		}
		log.Infof("multicluster remote secrets controller cache synced in %v", time.Since(t0))
		c.queue.Run(stopCh)
	}()
	return nil
}

func (c *Controller) HasSynced() bool {
	if !c.queue.HasSynced() {
		log.Debug("secret controller did not sync secrets presented at startup")
		// we haven't finished processing the secrets that were present at startup