Kubevirt VirtController 初始化流程分析

 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

func (vca *VirtControllerApp) Run() {
	logger := log.Log

	promCertManager := bootstrap.NewFileCertificateManager(vca.promCertFilePath, vca.promKeyFilePath)
	go promCertManager.Start()
	promTLSConfig := kvtls.SetupPromTLS(promCertManager, vca.clusterConfig)

	go func() {
		httpLogger := logger.With("service", "http")
		_ = httpLogger.Level(log.INFO).Log("action", "listening", "interface", vca.BindAddress, "port", vca.Port)
		http.Handle("/metrics", promhttp.Handler())
		server := http.Server{
			Addr:      vca.Address(),
			Handler:   http.DefaultServeMux,
			TLSConfig: promTLSConfig,
		}
		if err := server.ListenAndServeTLS("", ""); err != nil {
			golog.Fatal(err)
		}
	}()

	if err := vca.setupLeaderElector(); err != nil {
		golog.Fatal(err)
	}

	readyGauge.Set(1)
	vca.leaderElector.Run(vca.ctx)
	readyGauge.Set(0)
	panic("unreachable")
}

 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

func (vca *VirtControllerApp) onStartedLeading() func(ctx context.Context) {
	return func(ctx context.Context) {
		stop := ctx.Done()
		vca.informerFactory.Start(stop)

		golog.Printf("STARTING controllers with following threads : "+
			"node %d, vmi %d, replicaset %d, vm %d, migration %d, evacuation %d, disruptionBudget %d",
			vca.nodeControllerThreads, vca.vmiControllerThreads, vca.rsControllerThreads,
			vca.vmControllerThreads, vca.migrationControllerThreads, vca.evacuationControllerThreads,
			vca.disruptionBudgetControllerThreads)

		vmiprom.SetupVMICollector(vca.vmiInformer, vca.clusterConfig)
		vmprom.SetupVMCollector(vca.vmInformer)
		perfscale.RegisterPerfScaleMetrics(vca.vmiInformer)
		if vca.migrationInformer == nil {
			vca.migrationInformer = vca.informerFactory.VirtualMachineInstanceMigration()
		}
		golog.Printf("\nvca.migrationInformer :%v\n", vca.migrationInformer)
		migration.RegisterMigrationMetrics(vca.migrationInformer)
		migrationstats.SetupMigrationsCollector(vca.migrationInformer)

		go vca.evacuationController.Run(vca.evacuationControllerThreads, stop)
		go vca.disruptionBudgetController.Run(vca.disruptionBudgetControllerThreads, stop)
		go vca.nodeController.Run(vca.nodeControllerThreads, stop)
		go vca.vmiController.Run(vca.vmiControllerThreads, stop)
		go vca.rsController.Run(vca.rsControllerThreads, stop)
		go vca.poolController.Run(vca.poolControllerThreads, stop)
		go vca.vmController.Run(vca.vmControllerThreads, stop)
		go vca.migrationController.Run(vca.migrationControllerThreads, stop)
		go func() {
			if err := vca.snapshotController.Run(vca.snapshotControllerThreads, stop); err != nil {
				log.Log.Warningf("error running the snapshot controller: %v", err)
			}
		}()
		go func() {
			if err := vca.restoreController.Run(vca.restoreControllerThreads, stop); err != nil {
				log.Log.Warningf("error running the restore controller: %v", err)
			}
		}()
		go func() {
			if err := vca.exportController.Run(vca.exportControllerThreads, stop); err != nil {
				log.Log.Warningf("error running the export controller: %v", err)
			}
		}()
		go vca.workloadUpdateController.Run(stop)
		go vca.nodeTopologyUpdater.Run(vca.nodeTopologyUpdatePeriod, stop)
		go func() {
			if err := vca.vmCloneController.Run(vca.cloneControllerThreads, stop); err != nil {
				log.Log.Warningf("error running the clone controller: %v", err)
			}
		}()

		cache.WaitForCacheSync(stop, vca.persistentVolumeClaimInformer.HasSynced)
		close(vca.readyChan)
		leaderGauge.Set(1)
	}
}