k8s中各种类型的资源都由各自的controller来管理，所有controller由kube-controller-manager组件来管理。各个控制器在cmd/kube-controller-manager/app/controllermanager.go中的Run函数中被启动。
func Run(c *config.CompletedConfig, stopCh <-chan struct{}) error {
//....
if err := StartControllers(controllerContext, saTokenControllerInitFunc, NewControllerInitializers(controllerContext.LoopMode), unsecuredMux); err != nil {
klog.Fatalf("error starting controllers: %v", err)
}
//....
}
其中的NewControllerInitializers()函数返回各个controller的启动函数。
controllers["endpoint"] = startEndpointController
再跳转至startEndpointController函数。注意NewEndpointController中传了三个informer过去。
func startEndpointController(ctx ControllerContext) (http.Handler, bool, error) {
go endpointcontroller.NewEndpointController(
ctx.InformerFactory.Core().V1().Pods(),
ctx.InformerFactory.Core().V1().Services(),
ctx.InformerFactory.Core().V1().Endpoints(),
ctx.ClientBuilder.ClientOrDie("endpoint-controller"),
).Run(int(ctx.ComponentConfig.EndpointController.ConcurrentEndpointSyncs), ctx.Stop)
return nil, true, nil
}
然后进入到的真正的EndpointController的Run函数。这里发现很多controller的Run都是差不多一样的，启动一定数量的worker，每个worker都是for循环，读取endpointController.queue队列，处理收到的事件。endpoint controller的处理事件的func 时syncService()
func (e *EndpointController) Run(workers int, stopCh <-chan struct{}) {
for i := 0; i < workers; i++ {
go wait.Until(e.worker, e.workerLoopPeriod, stopCh)
}
<-stopCh
}
func (e *EndpointController) worker() {
for e.processNextWorkItem() {
}
}
func (e *EndpointController) processNextWorkItem() bool {
eKey, quit := e.queue.Get()
if quit {
return false
}
defer e.queue.Done(eKey)
err := e.syncService(eKey.(string))
e.handleErr(err, eKey)
return true
}
现在进入最主要的处理逻辑syncService
func (e *EndpointController) syncService(key string) error {
//获取 namespace和name
namespace, name, err := cache.SplitMetaNamespaceKey(key)
//根据namespace 和name 获取service
service, err := e.serviceLister.Services(namespace).Get(name)
if err != nil {
//如果service notfound，那么对应的endpoints也要被删掉
err = e.client.CoreV1().Endpoints(namespace).Delete(name, nil)
if err != nil && !errors.IsNotFound(err) {
return err
}
return nil
}
//根据service 的 selector来获取pod列表
pods, err := e.podLister.Pods(service.Namespace).List(labels.Set(service.Spec.Selector).AsSelectorPreValidated())
subsets := []v1.EndpointSubset{}
var totalReadyEps int = 0
var totalNotReadyEps int = 0
for _, pod := range pods {
//如果pod ip没有，说明pod没有起来，不用把相关信息加到subset中
if len(pod.Status.PodIP) == 0 {
klog.V(5).Infof("Failed to find an IP for pod %s/%s", pod.Namespace, pod.Name)
continue
}
//过滤掉被删除的pod
if !tolerateUnreadyEndpoints && pod.DeletionTimestamp != nil {
klog.V(5).Infof("Pod is being deleted %s/%s", pod.Namespace, pod.Name)
continue
}
epa := *podToEndpointAddress(pod)
hostname := pod.Spec.Hostname
if len(hostname) > 0 && pod.Spec.Subdomain == service.Name && service.Namespace == pod.Namespace {
epa.Hostname = hostname
}
// Allow headless service not to have ports.
if len(service.Spec.Ports) == 0 {
if service.Spec.ClusterIP == api.ClusterIPNone {
subsets, totalReadyEps, totalNotReadyEps = addEndpointSubset(subsets, pod, epa, nil, tolerateUnreadyEndpoints)
// No need to repack subsets for headless service without ports.
}
} else {
for i := range service.Spec.Ports {
servicePort := &service.Spec.Ports[i]
portName := servicePort.Name
portProto := servicePort.Protocol
portNum, err := podutil.FindPort(pod, servicePort)
if err != nil {
klog.V(4).Infof("Failed to find port for service %s/%s: %v", service.Namespace, service.Name, err)
continue
}
var readyEps, notReadyEps int
epp := &v1.EndpointPort{Name: portName, Port: int32(portNum), Protocol: portProto}
//把当前pod的信息，追加至subsets
subsets, readyEps, notReadyEps = addEndpointSubset(subsets, pod, epa, epp, tolerateUnreadyEndpoints)
totalReadyEps = totalReadyEps + readyEps
totalNotReadyEps = totalNotReadyEps + notReadyEps
}
}
}
subsets = endpoints.RepackSubsets(subsets)
// See if there's actually an update here.
//接下来就是判断entpoints是否需要被更新，如果新生成的跟已经存在的一样，不用更新；如果当前集群中没有这个service的endpoints，需要create；如果已经存在但是跟新生成的endpoints信息不一致就更新endpoints
currentEndpoints, err := e.endpointsLister.Endpoints(service.Namespace).Get(service.Name)
if err != nil {
if errors.IsNotFound(err) {
currentEndpoints = &v1.Endpoints{
ObjectMeta: metav1.ObjectMeta{
Name: service.Name,
Labels: service.Labels,
},
}
} else {
return err
}
}
createEndpoints := len(currentEndpoints.ResourceVersion) == 0
if !createEndpoints &&
apiequality.Semantic.DeepEqual(currentEndpoints.Subsets, subsets) &&
apiequality.Semantic.DeepEqual(currentEndpoints.Labels, service.Labels) {
klog.V(5).Infof("endpoints are equal for %s/%s, skipping update", service.Namespace, service.Name)
return nil
}
newEndpoints := currentEndpoints.DeepCopy()
newEndpoints.Subsets = subsets
newEndpoints.Labels = service.Labels
if newEndpoints.Annotations == nil {
newEndpoints.Annotations = make(map[string]string)
}
klog.V(4).Infof("Update endpoints for %v/%v, ready: %d not ready: %d", service.Namespace, service.Name, totalReadyEps, totalNotReadyEps)
if createEndpoints {
// No previous endpoints, create them
_, err = e.client.CoreV1().Endpoints(service.Namespace).Create(newEndpoints)
} else {
// Pre-existing
_, err = e.client.CoreV1().Endpoints(service.Namespace).Update(newEndpoints)
}
if err != nil {
if createEndpoints && errors.IsForbidden(err) {
// A request is forbidden primarily for two reasons:
// 1. namespace is terminating, endpoint creation is not allowed by default.
// 2. policy is misconfigured, in which case no service would function anywhere.
// Given the frequency of 1, we log at a lower level.
klog.V(5).Infof("Forbidden from creating endpoints: %v", err)
}
return err
}
return nil
}
最后回过头看下e.queue中的key是哪里放进去的。搜了下queue，发现是由queue.Add方法向里面放数据，总的调用queue.Add的有四处。其中前面三个都在NewEndpointController时说明了，最后一个只会在controller刚起来的时候执行
func (e *EndpointController) addPod(obj interface{}) {
}
func (e *EndpointController) updatePod(old, cur interface{}) {
}
func (e *EndpointController) enqueueService(obj interface{}) {
}
func (e *EndpointController) checkLeftoverEndpoints() {
}
