|
| 1 | +package clusternode |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "fmt" |
| 6 | + "net/http" |
| 7 | + "time" |
| 8 | + |
| 9 | + corev1 "k8s.io/api/core/v1" |
| 10 | + "k8s.io/apimachinery/pkg/labels" |
| 11 | + "k8s.io/apimachinery/pkg/util/runtime" |
| 12 | + "k8s.io/apimachinery/pkg/util/wait" |
| 13 | + coreinformers "k8s.io/client-go/informers/core/v1" |
| 14 | + clientset "k8s.io/client-go/kubernetes" |
| 15 | + corelisters "k8s.io/client-go/listers/core/v1" |
| 16 | + "k8s.io/client-go/tools/cache" |
| 17 | + "k8s.io/client-go/util/workqueue" |
| 18 | + cloudprovider "k8s.io/cloud-provider" |
| 19 | + cloudproviderapp "k8s.io/cloud-provider/app" |
| 20 | + cloudcontrollerconfig "k8s.io/cloud-provider/app/config" |
| 21 | + genericcontrollermanager "k8s.io/controller-manager/app" |
| 22 | + "k8s.io/klog" |
| 23 | + |
| 24 | + "github.com/yunify/qingcloud-cloud-controller-manager/pkg/qingcloud" |
| 25 | +) |
| 26 | + |
| 27 | +const ( |
| 28 | + clusterNodeRunWorkerPeriod = 1 * time.Second |
| 29 | + clusterNodeWorkers = 1 |
| 30 | +) |
| 31 | + |
| 32 | +type ClusterNodeController struct { |
| 33 | + cloud cloudprovider.Interface |
| 34 | + |
| 35 | + // svc |
| 36 | + serviceLister corelisters.ServiceLister |
| 37 | + serviceListerSynced cache.InformerSynced |
| 38 | + |
| 39 | + // clusternode |
| 40 | + nodeLister corelisters.NodeLister |
| 41 | + nodeListerSynced cache.InformerSynced |
| 42 | + nodeQueue workqueue.RateLimitingInterface |
| 43 | +} |
| 44 | + |
| 45 | +func StartClusterNodeControllerWrapper(completedConfig *cloudcontrollerconfig.CompletedConfig, cloud cloudprovider.Interface) cloudproviderapp.InitFunc { |
| 46 | + return func(ctx genericcontrollermanager.ControllerContext) (http.Handler, bool, error) { |
| 47 | + return startClusterNodeController(completedConfig, cloud, ctx.Stop) |
| 48 | + } |
| 49 | +} |
| 50 | + |
| 51 | +func startClusterNodeController(ctx *cloudcontrollerconfig.CompletedConfig, cloud cloudprovider.Interface, stopCh <-chan struct{}) (http.Handler, bool, error) { |
| 52 | + // Start the endpoint controller |
| 53 | + clusterNodeController, err := New( |
| 54 | + cloud, |
| 55 | + ctx.ClientBuilder.ClientOrDie("clusternode-controller"), |
| 56 | + ctx.SharedInformers.Core().V1().Services(), |
| 57 | + ctx.SharedInformers.Core().V1().Nodes(), |
| 58 | + ) |
| 59 | + if err != nil { |
| 60 | + // This error shouldn't fail. It lives like this as a legacy. |
| 61 | + klog.Errorf("Failed to start endpoint controller: %v", err) |
| 62 | + return nil, false, nil |
| 63 | + } |
| 64 | + |
| 65 | + go clusterNodeController.Run(stopCh, clusterNodeWorkers) |
| 66 | + |
| 67 | + return nil, true, nil |
| 68 | +} |
| 69 | + |
| 70 | +func New( |
| 71 | + cloud cloudprovider.Interface, |
| 72 | + kubeClient clientset.Interface, |
| 73 | + serviceInformer coreinformers.ServiceInformer, |
| 74 | + nodeInformer coreinformers.NodeInformer, |
| 75 | +) (*ClusterNodeController, error) { |
| 76 | + |
| 77 | + cnc := &ClusterNodeController{ |
| 78 | + cloud: cloud, |
| 79 | + serviceLister: serviceInformer.Lister(), |
| 80 | + serviceListerSynced: serviceInformer.Informer().HasSynced, |
| 81 | + nodeLister: nodeInformer.Lister(), |
| 82 | + nodeListerSynced: nodeInformer.Informer().HasSynced, |
| 83 | + nodeQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "cluster-node"), |
| 84 | + } |
| 85 | + |
| 86 | + nodeInformer.Informer().AddEventHandler( |
| 87 | + cache.ResourceEventHandlerFuncs{ |
| 88 | + UpdateFunc: func(old, cur interface{}) { |
| 89 | + oldNode, ok1 := old.(*corev1.Node) |
| 90 | + curNode, ok2 := cur.(*corev1.Node) |
| 91 | + if ok1 && ok2 && cnc.needsUpdate(oldNode, curNode) { |
| 92 | + cnc.enqueueNode(cur) |
| 93 | + } |
| 94 | + }, |
| 95 | + }, |
| 96 | + ) |
| 97 | + |
| 98 | + return cnc, nil |
| 99 | +} |
| 100 | + |
| 101 | +//check if node label changed |
| 102 | +func (cnc *ClusterNodeController) needsUpdate(old, new *corev1.Node) bool { |
| 103 | + |
| 104 | + if len(old.Labels) != len(new.Labels) { |
| 105 | + return true |
| 106 | + } |
| 107 | + |
| 108 | + for newLabelKey, newLabelValue := range new.Labels { |
| 109 | + oldLabelValuk, ok := old.Labels[newLabelKey] |
| 110 | + if !ok || newLabelValue != oldLabelValuk { |
| 111 | + return true |
| 112 | + } |
| 113 | + } |
| 114 | + |
| 115 | + return false |
| 116 | +} |
| 117 | + |
| 118 | +func (cnc *ClusterNodeController) enqueueNode(obj interface{}) { |
| 119 | + key, err := cache.MetaNamespaceKeyFunc(obj) |
| 120 | + if err != nil { |
| 121 | + runtime.HandleError(fmt.Errorf("couldn't get key for object %#v: %v", obj, err)) |
| 122 | + return |
| 123 | + } |
| 124 | + cnc.nodeQueue.Add(key) |
| 125 | +} |
| 126 | + |
| 127 | +func (cnc *ClusterNodeController) Run(stopCh <-chan struct{}, workers int) { |
| 128 | + defer runtime.HandleCrash() |
| 129 | + defer cnc.nodeQueue.ShutDown() |
| 130 | + |
| 131 | + klog.Info("Starting cluster node controller") |
| 132 | + defer klog.Info("Shutting down cluster node controller") |
| 133 | + |
| 134 | + if !cache.WaitForCacheSync(stopCh, cnc.serviceListerSynced, cnc.nodeListerSynced) { |
| 135 | + return |
| 136 | + } |
| 137 | + |
| 138 | + for i := 0; i < workers; i++ { |
| 139 | + go wait.Until(cnc.worker, clusterNodeRunWorkerPeriod, stopCh) |
| 140 | + } |
| 141 | + |
| 142 | + <-stopCh |
| 143 | +} |
| 144 | + |
| 145 | +func (cnc *ClusterNodeController) worker() { |
| 146 | + for cnc.processNextWorkItem() { |
| 147 | + } |
| 148 | +} |
| 149 | + |
| 150 | +func (cnc *ClusterNodeController) processNextWorkItem() bool { |
| 151 | + key, quit := cnc.nodeQueue.Get() |
| 152 | + if quit { |
| 153 | + return false |
| 154 | + } |
| 155 | + defer cnc.nodeQueue.Done(key) |
| 156 | + |
| 157 | + err := cnc.handleNodesUpdate(key.(string)) |
| 158 | + if err == nil { |
| 159 | + cnc.nodeQueue.Forget(key) |
| 160 | + return true |
| 161 | + } |
| 162 | + |
| 163 | + runtime.HandleError(fmt.Errorf("error processing cluster node %v (will retry): %v", key, err)) |
| 164 | + cnc.nodeQueue.AddRateLimited(key) |
| 165 | + |
| 166 | + return true |
| 167 | +} |
| 168 | + |
| 169 | +// handleNodesUpdate handle service backend according to node lables |
| 170 | +func (cnc *ClusterNodeController) handleNodesUpdate(key string) error { |
| 171 | + startTime := time.Now() |
| 172 | + defer func() { |
| 173 | + klog.V(4).Infof("Finished handleNodesUpdate %q (%v)", key, time.Since(startTime)) |
| 174 | + }() |
| 175 | + |
| 176 | + // 1. get node list |
| 177 | + var nodes []*corev1.Node |
| 178 | + nodeList, err := cnc.nodeLister.List(labels.NewSelector()) |
| 179 | + if err != nil { |
| 180 | + return fmt.Errorf("get node list error: %v", err) |
| 181 | + } |
| 182 | + for i, _ := range nodeList { |
| 183 | + nodes = append(nodes, nodeList[i]) |
| 184 | + } |
| 185 | + |
| 186 | + // 2. list all service |
| 187 | + svcs, err := cnc.serviceLister.List(labels.NewSelector()) |
| 188 | + if err != nil { |
| 189 | + return fmt.Errorf("list service error: %v", err) |
| 190 | + } |
| 191 | + |
| 192 | + // 3. filter service which externalTrafficPolicy=cluster and has annotation service.beta.kubernetes.io/qingcloud-lb-backend-label |
| 193 | + for _, svc := range svcs { |
| 194 | + _, ok := svc.Annotations[qingcloud.ServiceAnnotationBackendLabel] |
| 195 | + if ok && svc.Spec.Type == corev1.ServiceTypeLoadBalancer && |
| 196 | + svc.Spec.ExternalTrafficPolicy == corev1.ServiceExternalTrafficPolicyTypeCluster { |
| 197 | + klog.Infof("service %s serviceType = %s, externalTrafficPolicy = %s, also has backend label annotation , going to update loadbalancer", svc.Name, svc.Spec.Type, svc.Spec.ExternalTrafficPolicy) |
| 198 | + |
| 199 | + // 4. update lb |
| 200 | + lbInterface, _ := cnc.cloud.LoadBalancer() |
| 201 | + err = lbInterface.UpdateLoadBalancer(context.TODO(), "", svc, nodes) |
| 202 | + if err != nil { |
| 203 | + return fmt.Errorf("update loadbalancer for service %s/%s error: %v", svc.Namespace, svc.Name, err) |
| 204 | + } |
| 205 | + klog.Infof("update loadbalancer for service %s/%s success", svc.Namespace, svc.Name) |
| 206 | + } |
| 207 | + } |
| 208 | + |
| 209 | + return nil |
| 210 | +} |
0 commit comments