您好,登录后才能下订单哦!
如何进行main方法与Leader选举分析,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。
主要对main方法的主要逻辑进行分析,以及分析下组件的EventHandler,看该组件list/watch哪些对象,对象事件来了怎么处理,以及claimQueue与volumeQueue的对象来源。
main方法主要逻辑:
(1)解析启动参数;
(2)根据配置建立clientset;
(3)建立grpcclient;
(4)进行grpc探测(探测cephcsi-rbd服务是否准备好),直至探测成功;
(5)通过grpc获取driver名称与能力;
(6)根据clientset建立informers;
(7)构建provisionController对象;
(8)定义run方法(包括了provisionController.Run);
(9)根据--enable-leader-election
组件启动参数配置决定是否开启Leader 选举,当不开启时,直接运行run方法,开启时调用le.Run()。
func main() { var config *rest.Config var err error flag.Var(utilflag.NewMapStringBool(&featureGates), "feature-gates", "A set of key=value pairs that describe feature gates for alpha/experimental features. "+ "Options are:\n"+strings.Join(utilfeature.DefaultFeatureGate.KnownFeatures(), "\n")) klog.InitFlags(nil) flag.CommandLine.AddGoFlagSet(goflag.CommandLine) flag.Set("logtostderr", "true") flag.Parse() if err := utilfeature.DefaultMutableFeatureGate.SetFromMap(featureGates); err != nil { klog.Fatal(err) } if *showVersion { fmt.Println(os.Args[0], version) os.Exit(0) } klog.Infof("Version: %s", version) // get the KUBECONFIG from env if specified (useful for local/debug cluster) kubeconfigEnv := os.Getenv("KUBECONFIG") if kubeconfigEnv != "" { klog.Infof("Found KUBECONFIG environment variable set, using that..") kubeconfig = &kubeconfigEnv } if *master != "" || *kubeconfig != "" { klog.Infof("Either master or kubeconfig specified. building kube config from that..") config, err = clientcmd.BuildConfigFromFlags(*master, *kubeconfig) } else { klog.Infof("Building kube configs for running in cluster...") config, err = rest.InClusterConfig() } if err != nil { klog.Fatalf("Failed to create config: %v", err) } clientset, err := kubernetes.NewForConfig(config) if err != nil { klog.Fatalf("Failed to create client: %v", err) } // snapclientset.NewForConfig creates a new Clientset for VolumesnapshotV1beta1Client snapClient, err := snapclientset.NewForConfig(config) if err != nil { klog.Fatalf("Failed to create snapshot client: %v", err) } // The controller needs to know what the server version is because out-of-tree // provisioners aren't officially supported until 1.5 serverVersion, err := clientset.Discovery().ServerVersion() if err != nil { klog.Fatalf("Error getting server version: %v", err) } metricsManager := metrics.NewCSIMetricsManager("" /* driverName */) grpcClient, err := ctrl.Connect(*csiEndpoint, metricsManager) if err != nil { klog.Error(err.Error()) os.Exit(1) } // 循环探测,直至CSI driver即cephcsi-rbd服务准备好 err = ctrl.Probe(grpcClient, *operationTimeout) if err != nil { klog.Error(err.Error()) os.Exit(1) } // 从ceph-csi组件中获取driver name provisionerName, err := ctrl.GetDriverName(grpcClient, *operationTimeout) if err != nil { klog.Fatalf("Error getting CSI driver name: %s", err) } klog.V(2).Infof("Detected CSI driver %s", provisionerName) metricsManager.SetDriverName(provisionerName) metricsManager.StartMetricsEndpoint(*metricsAddress, *metricsPath) // 从ceph-csi组件中获取driver能力 pluginCapabilities, controllerCapabilities, err := ctrl.GetDriverCapabilities(grpcClient, *operationTimeout) if err != nil { klog.Fatalf("Error getting CSI driver capabilities: %s", err) } // Generate a unique ID for this provisioner timeStamp := time.Now().UnixNano() / int64(time.Millisecond) identity := strconv.FormatInt(timeStamp, 10) + "-" + strconv.Itoa(rand.Intn(10000)) + "-" + provisionerName // 开始构建infomer factory := informers.NewSharedInformerFactory(clientset, ctrl.ResyncPeriodOfCsiNodeInformer) // ------------------------------- // Listers // Create informer to prevent hit the API server for all resource request scLister := factory.Storage().V1().StorageClasses().Lister() claimLister := factory.Core().V1().PersistentVolumeClaims().Lister() var csiNodeLister storagelistersv1beta1.CSINodeLister var nodeLister v1.NodeLister if ctrl.SupportsTopology(pluginCapabilities) { csiNodeLister = factory.Storage().V1beta1().CSINodes().Lister() nodeLister = factory.Core().V1().Nodes().Lister() } // ------------------------------- // PersistentVolumeClaims informer rateLimiter := workqueue.NewItemExponentialFailureRateLimiter(*retryIntervalStart, *retryIntervalMax) claimQueue := workqueue.NewNamedRateLimitingQueue(rateLimiter, "claims") claimInformer := factory.Core().V1().PersistentVolumeClaims().Informer() // Setup options provisionerOptions := []func(*controller.ProvisionController) error{ controller.LeaderElection(false), // Always disable leader election in provisioner lib. Leader election should be done here in the CSI provisioner level instead. controller.FailedProvisionThreshold(0), controller.FailedDeleteThreshold(0), controller.RateLimiter(rateLimiter), controller.Threadiness(int(*workerThreads)), controller.CreateProvisionedPVLimiter(workqueue.DefaultControllerRateLimiter()), controller.ClaimsInformer(claimInformer), } translator := csitrans.New() supportsMigrationFromInTreePluginName := "" if translator.IsMigratedCSIDriverByName(provisionerName) { supportsMigrationFromInTreePluginName, err = translator.GetInTreeNameFromCSIName(provisionerName) if err != nil { klog.Fatalf("Failed to get InTree plugin name for migrated CSI plugin %s: %v", provisionerName, err) } klog.V(2).Infof("Supports migration from in-tree plugin: %s", supportsMigrationFromInTreePluginName) provisionerOptions = append(provisionerOptions, controller.AdditionalProvisionerNames([]string{supportsMigrationFromInTreePluginName})) } // Create the provisioner: it implements the Provisioner interface expected by // the controller csiProvisioner := ctrl.NewCSIProvisioner( clientset, *operationTimeout, identity, *volumeNamePrefix, *volumeNameUUIDLength, grpcClient, snapClient, provisionerName, pluginCapabilities, controllerCapabilities, supportsMigrationFromInTreePluginName, *strictTopology, translator, scLister, csiNodeLister, nodeLister, claimLister, *extraCreateMetadata, ) provisionController = controller.NewProvisionController( clientset, provisionerName, csiProvisioner, serverVersion.GitVersion, provisionerOptions..., ) csiClaimController := ctrl.NewCloningProtectionController( clientset, claimLister, claimInformer, claimQueue, ) // 主循环函数 run := func(context.Context) { stopCh := context.Background().Done() factory.Start(stopCh) cacheSyncResult := factory.WaitForCacheSync(stopCh) for _, v := range cacheSyncResult { if !v { klog.Fatalf("Failed to sync Informers!") } } // 跑两个controller,后面主要分析provisionController go csiClaimController.Run(int(*finalizerThreads), stopCh) provisionController.Run(wait.NeverStop) } // Leader 选举相关 if !*enableLeaderElection { run(context.TODO()) } else { // this lock name pattern is also copied from sigs.k8s.io/sig-storage-lib-external-provisioner/v5/controller // to preserve backwards compatibility lockName := strings.Replace(provisionerName, "/", "-", -1) // 使用endpoints或leases资源对象来选leader var le leaderElection if *leaderElectionType == "endpoints" { klog.Warning("The 'endpoints' leader election type is deprecated and will be removed in a future release. Use '--leader-election-type=leases' instead.") le = leaderelection.NewLeaderElectionWithEndpoints(clientset, lockName, run) } else if *leaderElectionType == "leases" { le = leaderelection.NewLeaderElection(clientset, lockName, run) } else { klog.Error("--leader-election-type must be either 'endpoints' or 'leases'") os.Exit(1) } if *leaderElectionNamespace != "" { le.WithNamespace(*leaderElectionNamespace) } // 处理Leader 选举逻辑的方法 if err := le.Run(); err != nil { klog.Fatalf("failed to initialize leader election: %v", err) } } }
主要看到EventHandler,定义了该组件list/watch的对象,对象事件来了怎么处理,以及claimQueue与volumeQueue的对象来源。
可以看到,claimQueue的来源是pvc对象的新增、更新事件(对claimQueue的处理已在external-provisioner源码分析(1)-主体处理逻辑分析中讲过,忘了的话可以回头看下)。
... // PersistentVolumeClaims claimHandler := cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { controller.enqueueClaim(obj) }, UpdateFunc: func(oldObj, newObj interface{}) { controller.enqueueClaim(newObj) }, DeleteFunc: func(obj interface{}) { // NOOP. The claim is either in claimsInProgress and in the queue, so it will be processed as usual // or it's not in claimsInProgress and then we don't care }, } if controller.claimInformer != nil { controller.claimInformer.AddEventHandlerWithResyncPeriod(claimHandler, controller.resyncPeriod) } else { controller.claimInformer = informer.Core().V1().PersistentVolumeClaims().Informer() controller.claimInformer.AddEventHandler(claimHandler) } ...
// enqueueClaim takes an obj and converts it into UID that is then put onto claim work queue. func (ctrl *ProvisionController) enqueueClaim(obj interface{}) { uid, err := getObjectUID(obj) if err != nil { utilruntime.HandleError(err) return } if ctrl.claimQueue.NumRequeues(uid) == 0 { ctrl.claimQueue.Add(uid) } }
可以看到,volumeQueue的来源是pv对象的新增、更新事件(对volumeQueue的处理已在external-provisioner源码分析(1)-主体处理逻辑分析中讲过,忘了的话可以回头看下)。
... // PersistentVolumes volumeHandler := cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { controller.enqueueVolume(obj) }, UpdateFunc: func(oldObj, newObj interface{}) { controller.enqueueVolume(newObj) }, DeleteFunc: func(obj interface{}) { controller.forgetVolume(obj) }, } if controller.volumeInformer != nil { controller.volumeInformer.AddEventHandlerWithResyncPeriod(volumeHandler, controller.resyncPeriod) } else { controller.volumeInformer = informer.Core().V1().PersistentVolumes().Informer() controller.volumeInformer.AddEventHandler(volumeHandler) } ...
// enqueueVolume takes an obj and converts it into a namespace/name string which // is then put onto the given work queue. func (ctrl *ProvisionController) enqueueVolume(obj interface{}) { var key string var err error if key, err = cache.DeletionHandlingMetaNamespaceKeyFunc(obj); err != nil { utilruntime.HandleError(err) return } // Re-Adding is harmless but try to add it to the queue only if it is not // already there, because if it is already there we *must* be retrying it if ctrl.volumeQueue.NumRequeues(key) == 0 { ctrl.volumeQueue.Add(key) } }
// forgetVolume Forgets an obj from the given work queue, telling the queue to // stop tracking its retries because e.g. the obj was deleted func (ctrl *ProvisionController) forgetVolume(obj interface{}) { var key string var err error if key, err = cache.DeletionHandlingMetaNamespaceKeyFunc(obj); err != nil { utilruntime.HandleError(err) return } ctrl.volumeQueue.Forget(key) ctrl.volumeQueue.Done(key) }
在 Golang 中,k8s client-go 这个package 针对 Leader 相关功能进行了封装,支持3种锁资源,endpoint,configmap,lease,方便使用。
Leader 选举基本原理其实就是利用通过Kubernetes中 configmap , endpoints 或者 lease 资源实现一个分布式锁,抢(acqure)到锁的节点成为leader,并且定期更新(renew)。其他进程也在不断的尝试进行抢占,抢占不到则继续等待下次循环。当leader节点挂掉之后,租约到期,其他节点就成为新的leader。
抢到锁其实就是成功把该进程的相关信息(如进程唯一标识)写入configmap、endpoints 或者 lease 资源对象中;而定期更新其实就是定期更新该资源的锁更新时间,以延续租期。
多个进程同时获取锁(更新锁资源)时,由apiserver来保证锁资源update的原子操作,通过对比resourceVersion版本号(resourceVersion的取值最终来源于etcd的modifiedindex),保证只有一个进程能修改成功,也即只有一个进程能成功获取到锁。
锁示例如下:
apiVersion: coordination.k8s.io/v1 kind: Lease metadata: creationTimestamp: "2020-08-21T11:56:46Z" name: rbd-csi-ceph-com namespace: default resourceVersion: "69642798" selfLink: /apis/coordination.k8s.io/v1/namespaces/default/leases/rbd-csi-ceph-com uid: c9a7ea00-c000-4c5c-b90f-d0e7c85240ca spec: acquireTime: "2020-08-21T11:56:46.907075Z" holderIdentity: cld-dnode3-1091-i-nease-net leaseDurationSeconds: 15 leaseTransitions: 0 renewTime: "2020-09-07T02:38:24.587170Z"
其中holderIdentity
记录了获取到锁的进程信息,renewTime
记录了锁更新时间。
从main方法代码中可以看出,在external-provisioner组件中,仅支持endpoint与lease两种锁资源,且endpoints锁会在后续被弃用,所以建议使用lease锁。
external-provisioner组件的高可用选主逻辑与k8s中的kube-controller-manager、kube-scheduler等组件的高可用选主逻辑类似。
概要过程:
(1)组件启动时,定期循环的去获取lease锁,获取成功则成为leader且返回,否则一直阻塞;
(2)获取lease锁成功,则竞选leader成功,然后运行external-provisioner组件的主体处理逻辑;
(3)竞选leader成功后,继续定期循环续约,以保证leader的长久性。
下面进行代码的分析。
当--enable-leader-election
组件启动参数为true时,运行该方法,主要逻辑为:
(1)定义leaderConfig结构体;
(2)调用leaderelection.RunOrDie做进一步的选主逻辑处理。
func (l *leaderElection) Run() error { if l.identity == "" { id, err := defaultLeaderElectionIdentity() if err != nil { return fmt.Errorf("error getting the default leader identity: %v", err) } l.identity = id } if l.namespace == "" { l.namespace = inClusterNamespace() } broadcaster := record.NewBroadcaster() broadcaster.StartRecordingToSink(&corev1.EventSinkImpl{Interface: l.clientset.CoreV1().Events(l.namespace)}) eventRecorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: fmt.Sprintf("%s/%s", l.lockName, string(l.identity))}) rlConfig := resourcelock.ResourceLockConfig{ Identity: sanitizeName(l.identity), EventRecorder: eventRecorder, } lock, err := resourcelock.New(l.resourceLock, l.namespace, sanitizeName(l.lockName), l.clientset.CoreV1(), l.clientset.CoordinationV1(), rlConfig) if err != nil { return err } leaderConfig := leaderelection.LeaderElectionConfig{ Lock: lock, LeaseDuration: l.leaseDuration, RenewDeadline: l.renewDeadline, RetryPeriod: l.retryPeriod, Callbacks: leaderelection.LeaderCallbacks{ OnStartedLeading: func(ctx context.Context) { klog.V(2).Info("became leader, starting") l.runFunc(ctx) }, OnStoppedLeading: func() { klog.Fatal("stopped leading") }, OnNewLeader: func(identity string) { klog.V(3).Infof("new leader detected, current leader: %s", identity) }, }, } leaderelection.RunOrDie(context.TODO(), leaderConfig) return nil // should never reach here }
主要逻辑:
(1)调用le.acquire()方法来尝试竞选为leader(acquire方法会定期循环的去获取lease锁,获取成功则成为leader且返回,否则一直阻塞);
(2)竞选leader成功,运行run方法;
(3)调用le.renew()续约方法,定期循环续约。
// RunOrDie starts a client with the provided config or panics if the config // fails to validate. func RunOrDie(ctx context.Context, lec LeaderElectionConfig) { le, err := NewLeaderElector(lec) if err != nil { panic(err) } if lec.WatchDog != nil { lec.WatchDog.SetLeaderElection(le) } le.Run(ctx) } // Run starts the leader election loop func (le *LeaderElector) Run(ctx context.Context) { defer func() { runtime.HandleCrash() le.config.Callbacks.OnStoppedLeading() }() if !le.acquire(ctx) { return // ctx signalled done } ctx, cancel := context.WithCancel(ctx) defer cancel() go le.config.Callbacks.OnStartedLeading(ctx) le.renew(ctx) } // acquire会不断循环的去获取lease锁,获取成功则成为leader且返回 // acquire loops calling tryAcquireOrRenew and returns true immediately when tryAcquireOrRenew succeeds. // Returns false if ctx signals done. func (le *LeaderElector) acquire(ctx context.Context) bool { ctx, cancel := context.WithCancel(ctx) defer cancel() succeeded := false desc := le.config.Lock.Describe() klog.Infof("attempting to acquire leader lease %v...", desc) wait.JitterUntil(func() { succeeded = le.tryAcquireOrRenew() le.maybeReportTransition() if !succeeded { klog.V(4).Infof("failed to acquire lease %v", desc) return } le.config.Lock.RecordEvent("became leader") le.metrics.leaderOn(le.config.Name) klog.Infof("successfully acquired lease %v", desc) cancel() }, le.config.RetryPeriod, JitterFactor, true, ctx.Done()) return succeeded } // 续约方法,不断循环续约 // renew loops calling tryAcquireOrRenew and returns immediately when tryAcquireOrRenew fails or ctx signals done. func (le *LeaderElector) renew(ctx context.Context) { ctx, cancel := context.WithCancel(ctx) defer cancel() wait.Until(func() { timeoutCtx, timeoutCancel := context.WithTimeout(ctx, le.config.RenewDeadline) defer timeoutCancel() err := wait.PollImmediateUntil(le.config.RetryPeriod, func() (bool, error) { done := make(chan bool, 1) go func() { defer close(done) done <- le.tryAcquireOrRenew() }() select { case <-timeoutCtx.Done(): return false, fmt.Errorf("failed to tryAcquireOrRenew %s", timeoutCtx.Err()) case result := <-done: return result, nil } }, timeoutCtx.Done()) le.maybeReportTransition() desc := le.config.Lock.Describe() if err == nil { klog.V(5).Infof("successfully renewed lease %v", desc) return } le.config.Lock.RecordEvent("stopped leading") le.metrics.leaderOff(le.config.Name) klog.Infof("failed to renew lease %v: %v", desc, err) cancel() }, le.config.RetryPeriod, ctx.Done()) // if we hold the lease, give it up if le.config.ReleaseOnCancel { le.release() } }
看完上述内容,你们掌握如何进行main方法与Leader选举分析的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注亿速云行业资讯频道,感谢各位的阅读!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。