Multicluster

Multicluster

func (s *Server) initServiceControllers(args *PilotArgs) error {
    serviceControllers := s.ServiceController()

    s.serviceEntryStore = serviceentry.NewServiceDiscovery(s.configController, s.environment.IstioConfigStore, s.XDSServer)
    serviceControllers.AddRegistry(s.serviceEntryStore)

    registered := make(map[serviceregistry.ProviderID]bool)
    for _, r := range args.RegistryOptions.Registries {
        serviceRegistry := serviceregistry.ProviderID(r)
        if _, exists := registered[serviceRegistry]; exists {
            log.Warnf("%s registry specified multiple times.", r)
            continue
        }
        registered[serviceRegistry] = true
        log.Infof("Adding %s registry adapter", serviceRegistry)
        switch serviceRegistry {
        case serviceregistry.Kubernetes:
            if err := s.initKubeRegistry(args); err != nil {
                return err
            }
        case serviceregistry.Mock:
            s.initMockRegistry()
        default:
            return fmt.Errorf("service registry %s is not supported", r)
        }
    }

    // Defer running of the service controllers.
    s.addStartFunc(func(stop <-chan struct{}) error {
        go serviceControllers.Run(stop)
        return nil
    })

    return nil
}

func (s *Server) initKubeRegistry(args *PilotArgs) (err error) {
    args.RegistryOptions.KubeOptions.ClusterID = s.clusterID
    args.RegistryOptions.KubeOptions.Metrics = s.environment
    args.RegistryOptions.KubeOptions.XDSUpdater = s.XDSServer
    args.RegistryOptions.KubeOptions.NetworksWatcher = s.environment.NetworksWatcher
    args.RegistryOptions.KubeOptions.SystemNamespace = args.Namespace

    caBundlePath := s.caBundlePath
    if hasCustomTLSCerts(args.ServerOptions.TLSOptions) {
        caBundlePath = args.ServerOptions.TLSOptions.CaCertFile
    }

    mc := kubecontroller.NewMulticluster(args.PodName,
        s.kubeClient,
        args.RegistryOptions.ClusterRegistriesNamespace,
        args.RegistryOptions.KubeOptions,
        s.ServiceController(),
        s.serviceEntryStore,
        caBundlePath,
        args.Revision,
        s.fetchCARoot,
        s.environment)

    // initialize the "main" cluster registry before starting controllers for remote clusters
    if err := mc.AddMemberCluster(s.kubeClient, args.RegistryOptions.KubeOptions.ClusterID); err != nil {
        log.Errorf("failed initializing registry for %s: %v", args.RegistryOptions.KubeOptions.ClusterID, err)
        return err
    }

    // start remote cluster controllers
    s.addStartFunc(func(stop <-chan struct{}) error {
        mc.InitSecretController(stop)
        return nil
    })

    s.multicluster = mc
    return
}

func (m *Multicluster) InitSecretController(stop <-chan struct{}) {
    m.secretController = secretcontroller.StartSecretController(
        m.client, m.AddMemberCluster, m.UpdateMemberCluster, m.DeleteMemberCluster,
        m.secretNamespace, m.syncInterval, stop)
}

// AddMemberCluster is passed to the secret controller as a callback to be called
// when a remote cluster is added.  This function needs to set up all the handlers
// to watch for resources being added, deleted or changed on remote clusters.
func (m *Multicluster) AddMemberCluster(client kubelib.Client, clusterID string) error {
    // stopCh to stop controller created here when cluster removed.
    stopCh := make(chan struct{})
    m.m.Lock()
    options := m.opts
    options.ClusterID = clusterID

    log.Infof("Initializing Kubernetes service registry %q", options.ClusterID)
    kubeRegistry := NewController(client, options)
    m.serviceController.AddRegistry(kubeRegistry)
    m.remoteKubeControllers[clusterID] = &kubeController{
        Controller: kubeRegistry,
        stopCh:     stopCh,
    }
    localCluster := m.opts.ClusterID == clusterID

    m.m.Unlock()

    // Only need to add service handler for kubernetes registry as `initRegistryEventHandlers`,
    // because when endpoints update `XDSUpdater.EDSUpdate` has already been called.
    kubeRegistry.AppendServiceHandler(func(svc *model.Service, ev model.Event) { m.updateHandler(svc) })

    // TODO move instance cache out of registries
    if m.serviceEntryStore != nil && features.EnableServiceEntrySelectPods {
        // Add an instance handler in the kubernetes registry to notify service entry store about pod events
        kubeRegistry.AppendWorkloadHandler(m.serviceEntryStore.WorkloadInstanceHandler)
    }

    if localCluster {
        // TODO implement deduping in aggregate registry to allow multiple k8s registries to handle WorkloadEntry
        if m.serviceEntryStore != nil && features.EnableK8SServiceSelectWorkloadEntries {
            // Add an instance handler in the service entry store to notify kubernetes about workload entry events
            m.serviceEntryStore.AppendWorkloadHandler(kubeRegistry.WorkloadInstanceHandler)
        }
    }

    // TODO only create namespace controller and cert patch for remote clusters (no way to tell currently)
    if m.serviceController.Running() {
        go kubeRegistry.Run(stopCh)
    }
    if m.fetchCaRoot != nil && m.fetchCaRoot() != nil && (features.ExternalIstioD || features.CentralIstioD || localCluster) {
        log.Infof("joining leader-election for %s in %s", leaderelection.NamespaceController, options.SystemNamespace)
        go leaderelection.
            NewLeaderElection(options.SystemNamespace, m.serverID, leaderelection.NamespaceController, client.Kube()).
            AddRunFunction(func(leaderStop <-chan struct{}) {
                log.Infof("starting namespace controller for cluster %s", clusterID)
                nc := NewNamespaceController(m.fetchCaRoot, client)
                // Start informers again. This fixes the case where informers for namespace do not start,
                // as we create them only after acquiring the leader lock
                // Note: stop here should be the overall pilot stop, NOT the leader election stop. We are
                // basically lazy loading the informer, if we stop it when we lose the lock we will never
                // recreate it again.
                client.RunAndWait(stopCh)
                nc.Run(leaderStop)
            }).Run(stopCh)
    }

    // Patch cert if a webhook config name is provided.
    // This requires RBAC permissions - a low-priv Istiod should not attempt to patch but rely on
    // operator or CI/CD
    webhookConfigName := strings.ReplaceAll(validationWebhookConfigNameTemplate, validationWebhookConfigNameTemplateVar, m.secretNamespace)
    if features.InjectionWebhookConfigName.Get() != "" && m.caBundlePath != "" && !localCluster && (features.ExternalIstioD || features.CentralIstioD) {
        // TODO remove the patch loop init from initSidecarInjector (does this need leader elect? how well does it work with multi-primary?)
        log.Infof("initializing webhook cert patch for cluster %s", clusterID)
        patcher, err := webhooks.NewWebhookCertPatcher(client.Kube(), m.revision, webhookName, m.caBundlePath)
        if err != nil {
            log.Errorf("could not initialize webhook cert patcher")
        } else {
            patcher.Run(stopCh)
        }
        validationWebhookController := webhooks.CreateValidationWebhookController(client, webhookConfigName,
            m.secretNamespace, m.caBundlePath, true)
        if validationWebhookController != nil {
            go validationWebhookController.Start(stopCh)
        }
    }

    client.RunAndWait(stopCh)
    return nil
}

func (m *Multicluster) UpdateMemberCluster(clients kubelib.Client, clusterID string) error {
    if err := m.DeleteMemberCluster(clusterID); err != nil {
        return err
    }
    return m.AddMemberCluster(clients, clusterID)
}

// DeleteMemberCluster is passed to the secret controller as a callback to be called
// when a remote cluster is deleted.  Also must clear the cache so remote resources
// are removed.
func (m *Multicluster) DeleteMemberCluster(clusterID string) error {

    m.m.Lock()
    defer m.m.Unlock()
    m.serviceController.DeleteRegistry(clusterID)
    kc, ok := m.remoteKubeControllers[clusterID]
    if !ok {
        log.Infof("cluster %s does not exist, maybe caused by invalid kubeconfig", clusterID)
        return nil
    }
    if err := kc.Cleanup(); err != nil {
        log.Warnf("failed cleaning up services in %s: %v", clusterID, err)
    }
    close(m.remoteKubeControllers[clusterID].stopCh)
    delete(m.remoteKubeControllers, clusterID)
    if m.XDSUpdater != nil {
        m.XDSUpdater.ConfigUpdate(&model.PushRequest{Full: true})
    }

    return nil
}