经常有人问:“浏览器输入url后发生了什么”,这个问题看似简单,但是却能全面的考察一个人对系统的了解程度。如果把这个问题引申到k8s领域,就可以问出“K8sClient提交Yaml后发生了什么”这样类似的问题。同样的,要回答这个问题,需要我们对k8s的设计有一个比较完整的了解。本文就试图回答这个问题,带着大家体验一下一份Yaml的K8S之旅。
k8s可以说是松耦合设计的一个典型,如下图所示,各个组件都和ApiServer进行通信,只有ApiServer可以写Etcd,这样做的的好处有许多:各个组件解耦了,可以独立发展;各组件也可以分布在不同的机器上,避免单机繁忙,甚至对某些关键组件可以多实例部署,增强性能和可用性;由于数据库Etcd维护了集群的核心元数据和状态,由ApiServer统一验证鉴权更合理;等等。
了解了k8s的整体设计后,下面我们以Service这个应用最关心的资源的Yaml文件提交后的效果进行分析,其它资源也是大同小异。
系统的整体处理流程图大致如下,首先在k8s启动后,各个组件包括CoreDNS、各个Controller都会连接到ApiServer(list/watch),在client如kubectl提交yaml后,API server会把相关资源存储到Ectd中并通知各个组件,各个组件然后各自进行自己的相关操作,最后产生了一个可以对外提供服务的service。
了解了整体流程后,我们来对流程中涉及的各个组件进行细致的分析。
ApiServer 会启动一个httpsserver,并把相关端点注册到具体的storage,其中以“api”开头的属于legacy,其注册的部分常见端点有:
restStorageMap := map[string]rest.Storage{ "pods": podStorage.Pod, "services": serviceRest, "endpoints": endpointsStorage, "nodes": nodeStorage.Node, ...... }
首先我们来看看通用的storage
type Store struct { // NewFunc returns a new instance of the type this registry returns for a // GET of a single object, e.g.: // // curl GET /apis/group/version/namespaces/my-ns/myresource/name-of-object NewFunc func() runtime.Object // NewListFunc returns a new list of the type this registry; it is the // type returned when the resource is listed, e.g.: // // curl GET /apis/group/version/namespaces/my-ns/myresource NewListFunc func() runtime.Object } // 资源创建方法 func (e *Store) Create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) { // 校验资源合法性 if createValidation != nil { if err := createValidation(ctx, obj.DeepCopyObject()); err != nil { return nil, err } } name, err := e.ObjectNameFunc(obj) key, err := e.KeyFunc(ctx, name) qualifiedResource := e.qualifiedResourceFromContext(ctx) ttl, err := e.calculateTTL(obj, 0, false) out := e.NewFunc() // 最终调用的要么是 dryrun,要么是etcd3 // https://github.com/kubernetes/kubernetes/blob/7f7378eddfe7a817c47fc75c220a729f4b78b913/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go#L144 if err := e.Storage.Create(ctx, key, obj, out, ttl, dryrun.IsDryRun(options.DryRun)); err != nil { err = storeerr.InterpretCreateError(err, qualifiedResource, name) err = rest.CheckGeneratedNameError(e.CreateStrategy, err, obj) // 资源已存在则可以原谅 if !apierrors.IsAlreadyExists(err) { return nil, err } // 创建后没法获得则不能原谅 if errGet := e.Storage.Get(ctx, key, "", out, false); errGet != nil { return nil, err } } // 切面 if e.AfterCreate != nil { if err := e.AfterCreate(out); err != nil { return nil, err } } return out, nil }
从restStorageMap可见处理逻辑是Service和Endpoints对象都要被存入etcd,其中Service还有一些特殊的逻辑(如分配IP,健康检查等),而Endpoint没啥额外的逻辑,直接使用通用的storage即可。
// service 创建逻辑 func (rs *REST) Create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) { service := obj.(*api.Service) // 切面 if err := rest.BeforeCreate(registry.Strategy, ctx, obj); err != nil { return nil, err } // 是否需要释放IP,类似于事务,如果分配出错的话,把ip还给资源池 releaseServiceIP := false defer func() { if releaseServiceIP { if helper.IsServiceIPSet(service) { allocator := rs.getAllocatorByClusterIP(service) allocator.Release(net.ParseIP(service.Spec.ClusterIP)) } } }() var err error if !dryrun.IsDryRun(options.DryRun) { // 对于不是ExternalName类型的service才分配IP if service.Spec.Type != api.ServiceTypeExternalName { // 这个 分配器 实际上基于etcd allocator := rs.getAllocatorBySpec(service) if releaseServiceIP, err = initClusterIP(service, allocator); err != nil { return nil, err } } } // 由 分配器 分配端口 nodePortOp := portallocator.StartOperation(rs.serviceNodePorts, dryrun.IsDryRun(options.DryRun)) // 同样需要判断是否需要回收 defer nodePortOp.Finish() // 对于 NodePort和LoadBalance类型的service都要分配端口 if service.Spec.Type == api.ServiceTypeNodePort || service.Spec.Type == api.ServiceTypeLoadBalancer { if err := initNodePorts(service, nodePortOp); err != nil { return nil, err } } // 对于需要健康检查的service分配专门的端口 // 至于loadbalance类型且ExternalTrafficPolicy为Local的才需要分配 if apiservice.NeedsHealthCheck(service) { if err := allocateHealthCheckNodePort(service, nodePortOp); err != nil { return nil, errors.NewInternalError(err) } } // 实际创建 out, err := rs.services.Create(ctx, service, createValidation, options) if err != nil { err = rest.CheckGeneratedNameError(registry.Strategy, err, service) } } // 基于 etcd 的 ip 分配器 serviceClusterIPAllocator, err := ipallocator.NewAllocatorCIDRRange(&serviceClusterIPRange, func(max int, rangeSpec string) (allocator.Interface, error) { mem := allocator.NewAllocationMap(max, rangeSpec) // TODO etcdallocator package to return a storage interface via the storageFactory etcd, err := serviceallocator.NewEtcd(mem, "/ranges/serviceips", api.Resource("serviceipallocations"), serviceStorageConfig) if err != nil { return nil, err } serviceClusterIPRegistry = etcd return etcd, nil }) if err != nil { return LegacyRESTStorage{}, genericapiserver.APIGroupInfo{}, fmt.Errorf("cannot create cluster IP allocator: %v", err) } restStorage.ServiceClusterIPAllocator = serviceClusterIPRegistry
对于Pod对象来说,除了存储外,还要将pod绑定到特定的机器上去:
func (r *BindingREST) setPodHostAndAnnotations(ctx context.Context, podID, oldMachine, machine string, annotations map[string]string, dryRun bool) (finalPod *api.Pod, err error) { podKey, err := r.store.KeyFunc(ctx, podID) if err != nil { return nil, err } err = r.store.Storage.GuaranteedUpdate(ctx, podKey, &api.Pod{}, false, nil, storage.SimpleUpdate(func(obj runtime.Object) (runtime.Object, error) { pod, ok := obj.(*api.Pod) // 分配到机器 pod.Spec.NodeName = machine // 设置注解 if pod.Annotations == nil { pod.Annotations = make(map[string]string) } for k, v := range annotations { pod.Annotations[k] = v } // condition,表明已被调度 podutil.UpdatePodCondition(&pod.Status, &api.PodCondition{ Type: api.PodScheduled, Status: api.ConditionTrue, }) finalPod = pod return pod, nil }), dryRun) return finalPod, err }
EndpointController监听Service和Pod的变化事件,并注册回调函数,通过Informer实现。同时利用Informer缓存最新的endpoint到本地,但是并不注册回调事件,因为endpoint基本上是最底层的概念,不需要额外的处理逻辑。
// 监听service serviceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ // 增删改 AddFunc: e.onServiceUpdate, UpdateFunc: func(old, cur interface{}) { e.onServiceUpdate(cur) }, DeleteFunc: e.onServiceDelete, }) ... // 监听pod podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: e.addPod, UpdateFunc: e.updatePod, DeleteFunc: e.deletePod, }) // 利用Informer可以获取最新的endpoint状况 e.endpointsLister = endpointsInformer.Lister() e.endpointsSynced = endpointsInformer.Informer().HasSynced
收到相关资源增加事件后,把需要处理的service加入队列
func (e *EndpointController) onServiceUpdate(obj interface{}) { // 获得service key 可能是 name 或则 namespace/name key, err := controller.KeyFunc(obj) if err != nil { utilruntime.HandleError(fmt.Errorf("Couldn't get key for object %+v: %v", obj, err)) return } // 更新service的selector _ = e.serviceSelectorCache.Update(key, obj.(*v1.Service).Spec.Selector) // 将service加入待处理队列 e.queue.Add(key) } func (e *EndpointController) addPod(obj interface{}) { pod := obj.(*v1.Pod) // 获得该pod相关的service,这些service的selector包含这个pod services, err := e.serviceSelectorCache.GetPodServiceMemberships(e.serviceLister, pod) if err != nil { utilruntime.HandleError(fmt.Errorf("Unable to get pod %s/%s's service memberships: %v", pod.Namespace, pod.Name, err)) return } // 将该pod相关的service加入待处理队列 for key := range services { e.queue.AddAfter(key, e.endpointUpdatesBatchPeriod) } }
另一方面,当 EndpointController Run 起来过后,其实是循环处理队列中的service,处理内容包括修改Service本身和其对应的Endpoints
func (e *EndpointController) Run(workers int, stopCh <-chan struct{}) { ... // 可以启动多个 goroutine ,来处理endpoint变化 for i := 0; i < workers; i++ { go wait.Until(e.worker, e.workerLoopPeriod, stopCh) } go func() { defer utilruntime.HandleCrash() // 处理无主(没有对应service)的endpoint,类似垃圾回收, // 当然这个方法只是遍历service的key并加入队列,实际处理由syncService完成 e.checkLeftoverEndpoints() }() } // 具体处理方法 func (e *EndpointController) syncService(key string) error { // 获得service service, err := e.serviceLister.Services(namespace).Get(name) if err != nil { // 不是没找到,返回错误 if !errors.IsNotFound(err) { return err } // 没有这个service,删掉相应的endpoint。这两者由key关联 err = e.client.CoreV1().Endpoints(namespace).Delete(context.TODO(), name, metav1.DeleteOptions{}) return nil } if service.Spec.Selector == nil { // 没有selector的service,其endpoint只能是手动创建的,与本Controller无关,直接返回 // https://kubernetes.io/docs/concepts/services-networking/service/#services-without-selectors return nil } // 获得相应pod pods, err := e.podLister.Pods(service.Namespace).List(labels.Set(service.Spec.Selector).AsSelectorPreValidated()) // 遍历这些pod,把合适的pod的ip加入该service的endpoints集合 for _, pod := range pods { // 返回这个pod的端点地址,需要处理v4 v6两类情况 ep, err := podToEndpointAddressForService(service, pod) // headless service 可以不指定端口. if len(service.Spec.Ports) == 0 { if service.Spec.ClusterIP == api.ClusterIPNone { subsets, totalReadyEps, totalNotReadyEps = addEndpointSubset(subsets, pod, epa, nil, tolerateUnreadyEndpoints) } } else { // 针对每个port映射,生成端点地址 for i := range service.Spec.Ports { servicePort := &service.Spec.Ports[i] portNum, err := podutil.FindPort(pod, servicePort) epp := endpointPortFromServicePort(servicePort, portNum) subsets, readyEps, notReadyEps = addEndpointSubset(subsets, pod, epa, epp, tolerateUnreadyEndpoints) } } } // 检测service是否真的有变化 // 首先获得最新的端点状况 currentEndpoints, err := e.endpointsLister.Endpoints(service.Namespace).Get(service.Name) if err != nil { // 不存在该endpoint就创建 if errors.IsNotFound(err) { currentEndpoints = &v1.Endpoints{ ObjectMeta: metav1.ObjectMeta{ Name: service.Name, Labels: service.Labels, }, } } else { return err } } createEndpoints := len(currentEndpoints.ResourceVersion) == 0 // 如果不是新创建的endpoint,则比较是否相同,相同则说明不需要修改 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 // 调用go client ,让APIservier 创建/更新 endpoint对象 if createEndpoints { // No previous endpoints, create them _, err = e.client.CoreV1().Endpoints(service.Namespace).Create(context.TODO(), newEndpoints, metav1.CreateOptions{}) } else { // Pre-existing _, err = e.client.CoreV1().Endpoints(service.Namespace).Update(context.TODO(), newEndpoints, metav1.UpdateOptions{}) } return nil }
ServiceController监听Service和Node的变化事件,原理与EndpointController一致,都是利用Informer,Informer的事件回调方法主要也是把需要处理的service加入队列,以及处理node。
serviceInformer.Informer().AddEventHandlerWithResyncPeriod( cache.ResourceEventHandlerFuncs{ AddFunc: func(cur interface{}) { svc, ok := cur.(*v1.Service) // 将需要分配负载均衡器或者清理的service加入待处理队列 if ok && (wantsLoadBalancer(svc) || needsCleanup(svc)) { s.enqueueService(cur) } }, UpdateFunc: func(old, cur interface{}) { oldSvc, ok1 := old.(*v1.Service) curSvc, ok2 := cur.(*v1.Service) if ok1 && ok2 && (s.needsUpdate(oldSvc, curSvc) || needsCleanup(curSvc)) { s.enqueueService(cur) } }, }, serviceSyncPeriod, ) nodeInformer.Informer().AddEventHandlerWithResyncPeriod( cache.ResourceEventHandlerFuncs{ AddFunc: func(cur interface{}) { s.nodeSyncLoop() }, UpdateFunc: func(old, cur interface{}) { s.nodeSyncLoop() }, DeleteFunc: func(old interface{}) { s.nodeSyncLoop() },
其中处理node的方法nodeSyncLoop,主要工作是对比最新节点和原有节点,若有变化则更新对应的service。
func (s *Controller) nodeSyncLoop() { // 最新且ready的所有节点 // 要所有节点是因为loadbalancer可能需要挂载到所有节点 // 取决于具体策略externalTrafficPolicy,不同云厂商实现大同小异 // https://aws.amazon.com/cn/blogs/opensource/network-load-balancer-support-in-kubernetes-1-9/ // https://help.aliyun.com/document_detail/86531.html#title-cn3-euk-ij6 newHosts, err := listWithPredicate(s.nodeLister, getNodeConditionPredicate()) // 节点未变化,本来不需要变动,但是可以在这里处理上次处理失败的service if nodeSlicesEqualForLB(newHosts, s.knownHosts) { s.servicesToUpdate = s.updateLoadBalancerHosts(s.servicesToUpdate, newHosts) return } // 获得所有service s.servicesToUpdate = s.cache.allServices() // 处理service,保留本次处理失败的service留给下次处理 s.servicesToUpdate = s.updateLoadBalancerHosts(s.servicesToUpdate, newHosts) // 更新本地service s.knownHosts = newHosts } // 处理service,保留本次处理失败的service func (s *Controller) updateLoadBalancerHosts(services []*v1.Service, hosts []*v1.Node) (servicesToRetry []*v1.Service) { for _, service := range services { func() { if err := s.lockedUpdateLoadBalancerHosts(service, hosts); err != nil { servicesToRetry = append(servicesToRetry, service) } }() } return servicesToRetry } func (s *Controller) lockedUpdateLoadBalancerHosts(service *v1.Service, hosts []*v1.Node) error { // 只处理 loadbalance 类型的service if !wantsLoadBalancer(service) { return nil } // 由云厂商实现loadBalancer的分配,比如 aws aliyun等 err := s.balancer.UpdateLoadBalancer(context.TODO(), s.clusterName, service, hosts) if err == nil { return nil } if _, exists, err := s.balancer.GetLoadBalancer(context.TODO(), s.clusterName, service); err != nil { runtime.HandleError(fmt.Errorf("failed to check if load balancer exists for service %s/%s: %v", service.Namespace, service.Name, err)) } else if !exists { return nil } s.eventRecorder.Eventf(service, v1.EventTypeWarning, "UpdateLoadBalancerFailed", "Error updating load balancer with new hosts %v: %v", nodeNames(hosts), err) return err }
另一方面,当 ServiceController Run 起来过后,其实是循环处理队列中的service和node,主要完成的工作是LoadBalancer类型的service与后端node的映射关系的维护。
func (s *Controller) Run(stopCh <-chan struct{}, workers int) { // 启动多个协程来处理service for i := 0; i < workers; i++ { go wait.Until(s.worker, time.Second, stopCh) } // 处理节点,也就是说不仅有事件触发,也有主动循环来处理节点变化 go wait.Until(s.nodeSyncLoop, nodeSyncPeriod, stopCh) } // 具体处理service的方法 func (s *Controller) syncService(key string) error { // 由key获得命名空间和service name namespace, name, err := cache.SplitMetaNamespaceKey(key) // 最新的service service, err := s.serviceLister.Services(namespace).Get(name) switch { case errors.IsNotFound(err): // 没找到,说明该删除这个service了 err = s.processServiceDeletion(key) case err != nil: runtime.HandleError(fmt.Errorf("Unable to retrieve service %v from store: %v", key, err)) default: // 创建或者更新service err = s.processServiceCreateOrUpdate(service, key) } return err }
CoreDNS的 kubernetes 插件配置好并启动后,以 service 的形式(名字就叫 kube-dns 兼容之前的dns插件名称)运行在k8s集群中,DNSController 监听 namespace、service和pod(可选)、endpoint(可选)的变化,并通过 Informer 机制缓存在本地。
func (k *Kubernetes) InitKubeCache(ctx context.Context) (err error) { // 获取配置 config, err := k.getClientConfig() // 依据配置获得client kubeClient, err := kubernetes.NewForConfig(config) if err != nil { return fmt.Errorf("failed to create kubernetes notification controller: %q", err) } k.opts.initPodCache = k.podMode == podModeVerified // controller中监听各资源 k.APIConn = newdnsController(ctx, kubeClient, k.opts) return err } // 在 controller 中监听各个资源的变化,并存储在本地 dns.svcLister, dns.svcController = object.NewIndexerInformer( &cache.ListWatch{ ListFunc: serviceListFunc(ctx, dns.client, api.NamespaceAll, dns.selector), WatchFunc: serviceWatchFunc(ctx, dns.client, api.NamespaceAll, dns.selector), }, &api.Service{}, ) dns.nsLister, dns.nsController = cache.NewInformer( &cache.ListWatch{ ListFunc: namespaceListFunc(ctx, dns.client, dns.namespaceSelector), WatchFunc: namespaceWatchFunc(ctx, dns.client, dns.namespaceSelector), }, &api.Namespace{}, ) pod和endpoint可选 ...
CoreDNS处理域名查询通过每一个插件的ServeDNS方法完成,在 kubernetes插件 中如下:
// ServeDNS implements the plugin.Handler interface. func (k Kubernetes) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (int, error) { state := request.Request{W: w, Req: r} // 处理多种请求类型 switch state.QType() { case dns.TypeAXFR, dns.TypeIXFR: k.Transfer(ctx, state) case dns.TypeA: records, err = plugin.A(ctx, &k, zone, state, nil, plugin.Options{}) case dns.TypeAAAA: records, err = plugin.AAAA(ctx, &k, zone, state, nil, plugin.Options{}) case dns.TypeTXT: records, err = plugin.TXT(ctx, &k, zone, state, nil, plugin.Options{}) case dns.TypeCNAME: records, err = plugin.CNAME(ctx, &k, zone, state, plugin.Options{}) case dns.TypePTR: records, err = plugin.PTR(ctx, &k, zone, state, plugin.Options{}) .... default: // Do a fake A lookup, so we can distinguish between NODATA and NXDOMAIN fake := state.NewWithQuestion(state.QName(), dns.TypeA) fake.Zone = state.Zone _, err = plugin.A(ctx, &k, zone, fake, nil, plugin.Options{}) } return dns.RcodeSuccess, nil }
该方法处理具体请求时,是通过informer查找存在本地的service或者pod的endpoints信息完成域名和ip的映射。
func (k *Kubernetes) Services(ctx context.Context, state request.Request, exact bool, opt plugin.Options) (svcs []msg.Service, err error) { // 特殊dns请求类型直接可以返回 switch state.QType() { case dns.TypeTXT: return []msg.Service{svc}, nil case dns.TypeNS: return svcs, nil } if isDefaultNS(state.Name(), state.Zone) { return svcs, nil } // 其余类型需要查询k8s返回记录 s, e := k.Records(ctx, state, false) internal := []msg.Service{} for _, svc := range s { if t, _ := svc.HostType(); t != dns.TypeCNAME { internal = append(internal, svc) } } return internal, e } // records 方法解析请求,做些验证,然后查找k8s中的相应记录 func (k *Kubernetes) Records(ctx context.Context, state request.Request, exact bool) ([]msg.Service, error) { r, e := parseRequest(state.Name(), state.Zone) ... if r.podOrSvc == Pod { // 处理pod请求 pods, err := k.findPods(r, state.Zone) return pods, err } // 处理service请求 services, err := k.findServices(r, state.Zone) return services, err } // 这个方法处理pod的dns请求 func (k *Kubernetes) findPods(r recordRequest, zone string) (pods []msg.Service, err error) { // 特性关闭,返回空 if k.podMode == podModeDisabled { return nil, errNoItems } podname := r.service // pod的name直接可以解析ip,区分ipv4 ipv6 if strings.Count(podname, "-") == 3 && !strings.Contains(podname, "--") { ip = strings.Replace(podname, "-", ".", -1) } else { ip = strings.Replace(podname, "-", ":", -1) } if k.podMode == podModeInsecure { // 不需检查模式,直接返回记录 return []msg.Service{{Key: strings.Join([]string{zonePath, Pod, namespace, podname}, "/"), Host: ip, TTL: k.ttl}}, err } // 需要检查的模式,只返回存在的pod的记录 for _, p := range k.APIConn.PodIndex(ip) { // check for matching ip and namespace if ip == p.PodIP && match(namespace, p.Namespace) { s := msg.Service{Key: strings.Join([]string{zonePath, Pod, namespace, podname}, "/"), Host: ip, TTL: k.ttl} pods = append(pods, s) } } return pods, err } // 这个方法处理各个类型的service请求,从本地cache中读取相应记录 func (k *Kubernetes) findServices(r recordRequest, zone string) (services []msg.Service, err error) { // 确定是否是模糊查询 if wildcard(r.service) || wildcard(r.namespace) { // 返回所有状态正确的service serviceList = k.APIConn.ServiceList() endpointsListFunc = func() []*object.Endpoints { return k.APIConn.EndpointsList() } } else { // 只返回 name + "." + namespace 类型的service idx := object.ServiceKey(r.service, r.namespace) serviceList = k.APIConn.SvcIndex(idx) endpointsListFunc = func() []*object.Endpoints { return k.APIConn.EpIndex(idx) } } zonePath := msg.Path(zone, coredns) for _, svc := range serviceList { // service name和 namespace都要匹配才行 if !(match(r.namespace, svc.Namespace) && match(r.service, svc.Name)) { continue } // 如果是模糊查询,需要namespace被暴露才行 if wildcard(r.namespace) && !k.namespaceExposed(svc.Namespace) { continue } // 处理endpoint或者headless service,这两类请求都需遍历endpoint if svc.ClusterIP == api.ClusterIPNone || r.endpoint != "" { for _, ep := range endpointsList { if ep.Name != svc.Name || ep.Namespace != svc.Namespace { continue } // 遍历endpoint的每一个ip和port for _, eps := range ep.Subsets { for _, addr := range eps.Addresses { for _, p := range eps.Ports { s := msg.Service{Host: addr.IP, Port: int(p.Port), TTL: k.ttl} s.Key = strings.Join([]string{zonePath, Svc, svc.Namespace, svc.Name, endpointHostname(addr, k.endpointNameMode)}, "/") err = nil services = append(services, s) } } } } continue } // 处理 External service if svc.Type == api.ServiceTypeExternalName { // 如 cluster.local/svc/namespace/service s := msg.Service{Key: strings.Join([]string{zonePath, Svc, svc.Namespace, svc.Name}, "/"), Host: svc.ExternalName, TTL: k.ttl} if t, _ := s.HostType(); t == dns.TypeCNAME { // 只有 cname 记录 s.Key = strings.Join([]string{zonePath, Svc, svc.Namespace, svc.Name}, "/") services = append(services, s) err = nil } continue } // 处理 ClusterIP service for _, p := range svc.Ports { if !(match(r.port, p.Name) && match(r.protocol, string(p.Protocol))) { continue } s := msg.Service{Host: svc.ClusterIP, Port: int(p.Port), TTL: k.ttl} s.Key = strings.Join([]string{zonePath, Svc, svc.Namespace, svc.Name}, "/") services = append(services, s) } } return services, err }
KubeProxy的主要工作是监听Service和Endpoints等的变化,并把路由规则(如何根据service的域名或者ip获得后端真实pod ip)刷新到节点上。
这样,每个pod在访问service时,就会向CoreDNS要到对应的service ip 或者直接是每个backend pod的ip(如 headless service),对于前者由本地路由规则将service ip的流量引导至真正的pod ip。我们的这份yaml也终于成了一个可以对外提供服务的service。
KubeProxy这部分由于经过了多个版本的迭代,目前包括三种类型,限于篇幅本文不展开,且待下回分解。