在 kubernetes，可以从集群外部和内部两种方式访问 kubernetes API，在集群外直接访问 apiserver 提供的 API，在集群内即 pod 中可以通过访问 service 为 kubernetes 的 ClusterIP。kubernetes 集群在初始化完成后就会创建一个 kubernetes service，该 service 是 kube-apiserver 创建并进行维护的，如下所示：

  1. $ kubectl get service
  2. NAME         TYPE        CLUSTER-IP   EXTERNAL-IP   PORT(S)   AGE
  3. kubernetes   ClusterIP   10.96.0.1    <none>        443/TCP   4d22h
  5. $ kubectl get endpoints kubernetes
  6. NAME         ENDPOINTS             AGE
  7. kubernetes   192.168.99.113:6443   4d22h

内置的 kubernetes service 无法删除，其 ClusterIP 为通过 --service-cluster-ip-range 参数指定的 ip 段中的首个 ip，kubernetes endpoints 中的 ip 以及 port 可以通过 --advertise-address 和 --secure-port 启动参数来指定。

kubernetes service 是由 kube-apiserver 中的 bootstrap controller 进行控制的，其主要以下几个功能：

  • 创建 kubernetes service；
  • 创建 default、kube-system 和 kube-public 命名空间，如果启用了 NodeLease 特性还会创建 kube-node-lease 命名空间；
  • 提供基于 Service ClusterIP 的修复及检查功能；
  • 提供基于 Service NodePort 的修复及检查功能；

kubernetes service 默认使用 ClusterIP 对外暴露服务，若要使用 nodePort 的方式可在 kube-apiserver 启动时通过 --kubernetes-service-node-port 参数指定对应的端口。

bootstrap controller 源码分析

kubernetes 版本：v1.16

bootstrap controller 的初始化以及启动是在 CreateKubeAPIServer 调用链的 InstallLegacyAPI 方法中完成的，bootstrap controller 的启停是由 apiserver 的 PostStartHook 和 ShutdownHook 进行控制的。

k8s.io/kubernetes/pkg/master/master.go:406

  1. func (m *Master) InstallLegacyAPI(......) error {
  2.     legacyRESTStorage, apiGroupInfo, err := legacyRESTStorageProvider.NewLegacyRESTStorage(restOptionsGetter)
  3.     if err != nil {
  4.         return fmt.Errorf("Error building core storage: %v", err)
  5.     }
  7.     // 初始化 bootstrap-controller
  8.     controllerName := "bootstrap-controller"
  9.     coreClient := corev1client.NewForConfigOrDie(c.GenericConfig.LoopbackClientConfig)
  10.     bootstrapController := c.NewBootstrapController(......)
  11.     m.GenericAPIServer.AddPostStartHookOrDie(controllerName, bootstrapController.PostStartHook)
  12.     m.GenericAPIServer.AddPreShutdownHookOrDie(controllerName, bootstrapController.PreShutdownHook)
  14.     if err := m.GenericAPIServer.InstallLegacyAPIGroup(genericapiserver.DefaultLegacyAPIPrefix, &apiGroupInfo); err != nil {
  15.         return fmt.Errorf("Error in registering group versions: %v", err)
  16.     }
  17.     return nil
  18. }

postStartHooks 会在 kube-apiserver 的启动方法 prepared.Run 中调用 RunPostStartHooks 启动所有 Hook。

NewBootstrapController

bootstrap controller 在初始化时需要设定多个参数，主要有 PublicIP、ServiceCIDR、PublicServicePort 等。PublicIP 是通过命令行参数 --advertise-address 指定的，如果没有指定，系统会自动选出一个 global IP。PublicServicePort 通过 --secure-port 启动参数来指定（默认为 6443），ServiceCIDR 通过 --service-cluster-ip-range 参数指定（默认为 10.0.0.0/24）。

k8s.io/kubernetes/pkg/master/controller.go:89

  1. func (c *completedConfig) NewBootstrapController(......) *Controller {
  2.     // 1、获取 PublicServicePort
  3.     _, publicServicePort, err := c.GenericConfig.SecureServing.HostPort()
  4.     if err != nil {
  5.         klog.Fatalf("failed to get listener address: %v", err)
  6.     }
  8.     // 2、指定需要创建的 kube-system 和 kube-public
  9.     systemNamespaces := []string{metav1.NamespaceSystem, metav1.NamespacePublic}
  10.     if utilfeature.DefaultFeatureGate.Enabled(features.NodeLease) {
  11.         systemNamespaces = append(systemNamespaces, corev1.NamespaceNodeLease)
  12.     }
  14.     return &Controller{
  15.         ......
  16.         // ServiceClusterIPRegistry 是在 CreateKubeAPIServer 初始化 RESTStorage 时初始化的，是一个 etcd 实例
  17.         ServiceClusterIPRegistry:          legacyRESTStorage.ServiceClusterIPAllocator,
  18.         ServiceClusterIPRange:             c.ExtraConfig.ServiceIPRange,
  19.         SecondaryServiceClusterIPRegistry: legacyRESTStorage.SecondaryServiceClusterIPAllocator,
  21.         // SecondaryServiceClusterIPRange 需要在启用 IPv6DualStack 后才能使用
  22.         SecondaryServiceClusterIPRange:    c.ExtraConfig.SecondaryServiceIPRange,
  24.         ServiceClusterIPInterval: 3 * time.Minute,
  26.         ServiceNodePortRegistry: legacyRESTStorage.ServiceNodePortAllocator,
  27.         ServiceNodePortRange:    c.ExtraConfig.ServiceNodePortRange,
  28.         ServiceNodePortInterval: 3 * time.Minute,
  30.         // API Server 绑定的IP，这个IP会作为kubernetes service的Endpoint的IP
  31.         PublicIP: c.GenericConfig.PublicAddress,
  32.         // 取 clusterIP range 中的第一个 IP
  33.         ServiceIP:                 c.ExtraConfig.APIServerServiceIP,
  34.         // 默认为 6443
  35.         ServicePort:               c.ExtraConfig.APIServerServicePort,
  36.         ExtraServicePorts:         c.ExtraConfig.ExtraServicePorts,
  37.         ExtraEndpointPorts:        c.ExtraConfig.ExtraEndpointPorts,
  38.         // 这里为 6443
  39.         PublicServicePort:         publicServicePort,
  41.         // 缺省是基于 ClusterIP 启动模式，这里为0
  42.         KubernetesServiceNodePort: c.ExtraConfig.KubernetesServiceNodePort,
  43.     }
  44. }

自动选出 global IP 的代码如下所示：

k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/net/interface.go:323

  1. func ChooseHostInterface() (net.IP, error) {
  2.     var nw networkInterfacer = networkInterface{}
  3.     if _, err := os.Stat(ipv4RouteFile); os.IsNotExist(err) {
  4.         return chooseIPFromHostInterfaces(nw)
  5.     }
  6.     routes, err := getAllDefaultRoutes()
  7.     if err != nil {
  8.         return nil, err
  9.     }
  10.     return chooseHostInterfaceFromRoute(routes, nw)
  11. }

bootstrapController.Start

上文已经提到了 bootstrap controller 主要的四个功能：修复 ClusterIP、修复 NodePort、更新 kubernetes service、创建系统所需要的名字空间（default、kube-system、kube-public）。bootstrap controller 在启动后首先会完成一次 ClusterIP、NodePort 和 Kubernets 服务的处理，然后异步循环运行上面的4个工作。以下是其 start 方法：

k8s.io/kubernetes/pkg/master/controller.go:146

  1. func (c *Controller) Start() {
  2.     if c.runner != nil {
  3.         return
  4.     }
  6.     // 1、首次启动时首先从 kubernetes endpoints 中移除自身的配置，
  7.     // 此时 kube-apiserver 可能处于非 ready 状态
  8.     endpointPorts := createEndpointPortSpec(c.PublicServicePort, "https", c.ExtraEndpointPorts)
  9.     if err := c.EndpointReconciler.RemoveEndpoints(kubernetesServiceName, c.PublicIP, endpointPorts); err != nil {
  10.         klog.Errorf("Unable to remove old endpoints from kubernetes service: %v", err)
  11.     }
  13.     // 2、初始化 repairClusterIPs 和 repairNodePorts 对象
  14.     repairClusterIPs := servicecontroller.NewRepair(......)
  15.     repairNodePorts := portallocatorcontroller.NewRepair(......)
  17.     // 3、首先运行一次 epairClusterIPs 和 repairNodePorts，即进行初始化
  18.     if err := repairClusterIPs.RunOnce(); err != nil {
  19.         klog.Fatalf("Unable to perform initial IP allocation check: %v", err)
  20.     }
  21.     if err := repairNodePorts.RunOnce(); err != nil {
  22.         klog.Fatalf("Unable to perform initial service nodePort check: %v", err)
  23.     }
  24.     // 4、定期执行 bootstrap controller 主要的四个功能
  25.     c.runner = async.NewRunner(c.RunKubernetesNamespaces, c.RunKubernetesService, repairClusterIPs.RunUntil, repairNodePorts.RunUntil)
  26.     c.runner.Start()
  27. }

c.RunKubernetesNamespaces

c.RunKubernetesNamespaces 主要功能是创建 kube-system 和 kube-public 命名空间，如果启用了 NodeLease 特性功能还会创建 kube-node-lease 命名空间，之后每隔一分钟检查一次。

k8s.io/kubernetes/pkg/master/controller.go:199

  1. func (c *Controller) RunKubernetesNamespaces(ch chan struct{}) {
  2.     wait.Until(func() {
  3.         for _, ns := range c.SystemNamespaces {
  4.             if err := createNamespaceIfNeeded(c.NamespaceClient, ns); err != nil {
  5.                 runtime.HandleError(fmt.Errorf("unable to create required kubernetes system namespace %s: %v", ns, err))
  6.             }
  7.         }
  8.     }, c.SystemNamespacesInterval, ch)
  9. }

c.RunKubernetesService

c.RunKubernetesService 主要是检查 kubernetes service 是否处于正常状态，并定期执行同步操作。首先调用 /healthz 接口检查 apiserver 当前是否处于 ready 状态，若处于 ready 状态然后调用 c.UpdateKubernetesService 服务更新 kubernetes service 状态。

k8s.io/kubernetes/pkg/master/controller.go:210

  1. func (c *Controller) RunKubernetesService(ch chan struct{}) {
  2.     wait.PollImmediateUntil(100*time.Millisecond, func() (bool, error) {
  3.         var code int
  4.         c.healthClient.Get().AbsPath("/healthz").Do().StatusCode(&code)
  5.         return code == http.StatusOK, nil
  6.     }, ch)
  8.     wait.NonSlidingUntil(func() {
  9.         if err := c.UpdateKubernetesService(false); err != nil {
  10.             runtime.HandleError(fmt.Errorf("unable to sync kubernetes service: %v", err))
  11.         }
  12.     }, c.EndpointInterval, ch)
  13. }
c.UpdateKubernetesService

c.UpdateKubernetesService 的主要逻辑为：

  • 1、调用 createNamespaceIfNeeded 创建 default namespace；
  • 2、调用 c.CreateOrUpdateMasterServiceIfNeeded 为 master 创建 kubernetes service；
  • 3、调用 c.EndpointReconciler.ReconcileEndpoints 更新 master 的 endpoint；

k8s.io/kubernetes/pkg/master/controller.go:230

  1. func (c *Controller) UpdateKubernetesService(reconcile bool) error {
  2.     if err := createNamespaceIfNeeded(c.NamespaceClient, metav1.NamespaceDefault); err != nil {
  3.         return err
  4.     }
  6.     servicePorts, serviceType := createPortAndServiceSpec(c.ServicePort, c.PublicServicePort, c.KubernetesServiceNodePort, "https", c.ExtraServicePorts)
  7.     if err := c.CreateOrUpdateMasterServiceIfNeeded(kubernetesServiceName, c.ServiceIP, servicePorts, serviceType, reconcile); err != nil {
  8.             return err
  9.     }
  10.     endpointPorts := createEndpointPortSpec(c.PublicServicePort, "https", c.ExtraEndpointPorts)
  11.     if err := c.EndpointReconciler.ReconcileEndpoints(kubernetesServiceName, c.PublicIP, endpointPorts, reconcile); err != nil {
  12.         return err
  13.     }
  14.     return nil
  15. }
c.EndpointReconciler.ReconcileEndpoints

EndpointReconciler 的具体实现由 EndpointReconcilerType 决定，EndpointReconcilerType 是 --endpoint-reconciler-type 参数指定的，可选的参数有 master-count, lease, none，每种类型对应不同的 EndpointReconciler 实例，在 v1.16 中默认为 lease，此处仅分析 lease 对应的 EndpointReconciler 的实现。

一个集群中可能会有多个 apiserver 实例，因此需要统一管理 apiserver service 的 endpoints，c.EndpointReconciler.ReconcileEndpoints 就是用来管理 apiserver endpoints 的。一个集群中 apiserver 的所有实例会在 etcd 中的对应目录下创建 key，并定期更新这个 key 来上报自己的心跳信息，ReconcileEndpoints 会从 etcd 中获取 apiserver 的实例信息并更新 endpoint。

k8s.io/kubernetes/pkg/master/reconcilers/lease.go:144

  1. func (r *leaseEndpointReconciler) ReconcileEndpoints(......) error {
  2.     r.reconcilingLock.Lock()
  3.     defer r.reconcilingLock.Unlock()
  5.     if r.stopReconcilingCalled {
  6.         return nil
  7.     }
  9.     // 更新 lease 信息
  10.     if err := r.masterLeases.UpdateLease(ip.String()); err != nil {
  11.         return err
  12.     }
  14.     return r.doReconcile(serviceName, endpointPorts, reconcilePorts)
  15. }
  16. func (r *leaseEndpointReconciler) doReconcile(......) error {
  17.     // 1、获取 master 的 endpoint
  18.     e, err := r.epAdapter.Get(corev1.NamespaceDefault, serviceName, metav1.GetOptions{})
  19.     shouldCreate := false
  20.     if err != nil {
  21.         if !errors.IsNotFound(err) {
  22.             return err
  23.         }
  25.         shouldCreate = true
  26.         e = &corev1.Endpoints{
  27.             ObjectMeta: metav1.ObjectMeta{
  28.                 Name:      serviceName,
  29.                 Namespace: corev1.NamespaceDefault,
  30.             },
  31.         }
  32.     }
  34.     // 2、从 etcd 中获取所有的 master
  35.     masterIPs, err := r.masterLeases.ListLeases()
  36.     if err != nil {
  37.         return err
  38.     }
  40.     if len(masterIPs) == 0 {
  41.         return fmt.Errorf("no master IPs were listed in storage, refusing to erase all endpoints for the kubernetes service")
  42.     }
  44.     // 3、检查 endpoint 中 master 信息，如果与 etcd 中的不一致则进行更新
  45.     formatCorrect, ipCorrect, portsCorrect := checkEndpointSubsetFormatWithLease(e, masterIPs, endpointPorts, reconcilePorts)
  46.     if formatCorrect && ipCorrect && portsCorrect {
  47.         return nil
  48.     }
  50.     if !formatCorrect {
  51.         e.Subsets = []corev1.EndpointSubset{{
  52.             Addresses: []corev1.EndpointAddress{},
  53.             Ports:     endpointPorts,
  54.         }}
  55.     }
  56.     if !formatCorrect || !ipCorrect {
  57.         e.Subsets[0].Addresses = make([]corev1.EndpointAddress, len(masterIPs))
  58.         for ind, ip := range masterIPs {
  59.             e.Subsets[0].Addresses[ind] = corev1.EndpointAddress{IP: ip}
  60.         }
  62.         e.Subsets = endpointsv1.RepackSubsets(e.Subsets)
  63.     }
  65.     if !portsCorrect {
  66.         e.Subsets[0].Ports = endpointPorts
  67.     }
  69.     if shouldCreate {
  70.         if _, err = r.epAdapter.Create(corev1.NamespaceDefault, e); errors.IsAlreadyExists(err) {
  71.             err = nil
  72.         }
  73.     } else {
  74.         _, err = r.epAdapter.Update(corev1.NamespaceDefault, e)
  75.     }
  76.     return err
  77. }

repairClusterIPs.RunUntil

repairClusterIP 主要解决的问题有：

  • 保证集群中所有的 ClusterIP 都是唯一分配的；
  • 保证分配的 ClusterIP 不会超出指定范围；
  • 确保已经分配给 service 但是因为 crash 等其他原因没有正确创建 ClusterIP；
  • 自动将旧版本的 Kubernetes services 迁移到 ipallocator 原子性模型；

repairClusterIPs.RunUntil 其实是调用 repairClusterIPs.runOnce 来处理的，其代码中的主要逻辑如下所示：

k8s.io/kubernetes/pkg/registry/core/service/ipallocator/controller/repair.go:134

  1. func (c *Repair) runOnce() error {
  2.     ......
  4.     // 1、首先从 etcd 中获取已经使用 ClusterIP 的快照
  5.     err = wait.PollImmediate(time.Second, 10*time.Second, func() (bool, error) {
  6.         var err error
  7.         snapshot, err = c.alloc.Get()
  8.         if err != nil {
  9.             return false, err
  10.         }
  12.         if c.shouldWorkOnSecondary() {
  13.             secondarySnapshot, err = c.secondaryAlloc.Get()
  14.             if err != nil {
  15.                 return false, err
  16.             }
  17.         }
  18.         return true, nil
  19.     })
  20.     if err != nil {
  21.         return fmt.Errorf("unable to refresh the service IP block: %v", err)
  22.     }
  23.     // 2、判断 snapshot 是否已经初始化
  24.     if snapshot.Range == "" {
  25.         snapshot.Range = c.network.String()
  26.     }
  28.     if c.shouldWorkOnSecondary() && secondarySnapshot.Range == "" {
  29.         secondarySnapshot.Range = c.secondaryNetwork.String()
  30.     }
  32.     stored, err = ipallocator.NewFromSnapshot(snapshot)
  33.     if c.shouldWorkOnSecondary() {
  34.         secondaryStored, secondaryErr = ipallocator.NewFromSnapshot(secondarySnapshot)
  35.     }
  37.     if err != nil || secondaryErr != nil {
  38.         return fmt.Errorf("unable to rebuild allocator from snapshots: %v", err)
  39.     }
  40.     // 3、获取 service list
  41.     list, err := c.serviceClient.Services(metav1.NamespaceAll).List(metav1.ListOptions{})
  42.     if err != nil {
  43.         return fmt.Errorf("unable to refresh the service IP block: %v", err)
  44.     }
  46.     // 4、将 CIDR 转换为对应的 IP range 格式
  47.     var rebuilt, secondaryRebuilt *ipallocator.Range
  48.     rebuilt, err = ipallocator.NewCIDRRange(c.network)
  50.     ......
  52.     // 5、检查每个 Service 的 ClusterIP，保证其处于正常状态
  53.     for _, svc := range list.Items {
  54.         if !helper.IsServiceIPSet(&svc) {
  55.             continue
  56.         }
  57.         ip := net.ParseIP(svc.Spec.ClusterIP)
  58.         ......
  60.         actualAlloc := c.selectAllocForIP(ip, rebuilt, secondaryRebuilt)
  61.         switch err := actualAlloc.Allocate(ip); err {
  62.         // 6、检查 ip 是否泄漏
  63.         case nil:
  64.             actualStored := c.selectAllocForIP(ip, stored, secondaryStored)
  65.             if actualStored.Has(ip) {
  66.                 actualStored.Release(ip)
  67.             } else {
  68.                 ......
  69.             }
  70.             delete(c.leaks, ip.String())
  71.         // 7、ip 重复分配
  72.         case ipallocator.ErrAllocated:
  73.             ......
  74.         // 8、ip 超出范围
  75.         case err.(*ipallocator.ErrNotInRange):
  76.             ......
  77.         // 9、ip 已经分配完
  78.         case ipallocator.ErrFull:
  79.             ......
  80.         default:
  81.             ......
  82.         }
  83.     }
  84.     // 10、对比是否有泄漏 ip
  85.     c.checkLeaked(stored, rebuilt)
  86.     if c.shouldWorkOnSecondary() {
  87.         c.checkLeaked(secondaryStored, secondaryRebuilt)
  88.     }
  90.     // 11、更新快照
  91.     err = c.saveSnapShot(rebuilt, c.alloc, snapshot)
  92.     if err != nil {
  93.         return err
  94.     }
  96.     if c.shouldWorkOnSecondary() {
  97.         err := c.saveSnapShot(secondaryRebuilt, c.secondaryAlloc, secondarySnapshot)
  98.         if err != nil {
  99.             return nil
  100.         }
  101.     }
  102.     return nil
  103. }

repairNodePorts.RunUnti

repairNodePorts 主要是用来纠正 service 中 nodePort 的信息，保证所有的 ports 都基于 cluster 创建的，当没有与 cluster 同步时会触发告警，其最终是调用 repairNodePorts.runOnce 进行处理的，主要逻辑与 ClusterIP 的处理逻辑类似。

k8s.io/kubernetes/pkg/registry/core/service/portallocator/controller/repair.go:84

  1. func (c *Repair) runOnce() error {
  2.     // 1、首先从 etcd 中获取已使用 nodeport 的快照
  3.     err := wait.PollImmediate(time.Second, 10*time.Second, func() (bool, error) {
  4.         var err error
  5.         snapshot, err = c.alloc.Get()
  6.         return err == nil, err
  7.     })
  8.     if err != nil {
  9.         return fmt.Errorf("unable to refresh the port allocations: %v", err)
  10.     }
  11.     // 2、检查 snapshot 是否初始化
  12.     if snapshot.Range == "" {
  13.         snapshot.Range = c.portRange.String()
  14.     }
  15.     // 3、获取已分配 nodePort 信息
  16.     stored, err := portallocator.NewFromSnapshot(snapshot)
  17.     if err != nil {
  18.         return fmt.Errorf("unable to rebuild allocator from snapshot: %v", err)
  19.     }
  20.     // 4、获取 service list
  21.     list, err := c.serviceClient.Services(metav1.NamespaceAll).List(metav1.ListOptions{})
  22.     if err != nil {
  23.         return fmt.Errorf("unable to refresh the port block: %v", err)
  24.     }
  26.     rebuilt, err := portallocator.NewPortAllocator(c.portRange)
  27.     if err != nil {
  28.         return fmt.Errorf("unable to create port allocator: %v", err)
  29.     }
  31.     // 5、检查每个 Service ClusterIP 的 port，保证其处于正常状态
  32.     for i := range list.Items {
  33.         svc := &list.Items[i]
  34.         ports := collectServiceNodePorts(svc)
  35.         if len(ports) == 0 {
  36.             continue
  37.         }
  38.         for _, port := range ports {
  39.             switch err := rebuilt.Allocate(port); err {
  40.             // 6、检查 port 是否泄漏
  41.             case nil:
  42.                 if stored.Has(port) {
  43.                     stored.Release(port)
  44.                 } else {
  45.                     ......
  46.                 }
  47.                 delete(c.leaks, port)
  48.             // 7、port 重复分配
  49.             case portallocator.ErrAllocated:
  50.                 ......
  51.             // 8、port 超出分配范围
  52.             case err.(*portallocator.ErrNotInRange):
  53.                 ......
  54.             // 9、port 已经分配完
  55.             case portallocator.ErrFull:
  56.                 ......
  57.             default:
  58.                 ......
  59.             }
  60.         }
  61.     }
  62.     // 10、检查 port 是否泄漏
  63.     stored.ForEach(func(port int) {
  64.         count, found := c.leaks[port]
  65.         switch {
  66.         case !found:
  67.             ......
  68.             count = numRepairsBeforeLeakCleanup - 1
  69.             fallthrough
  70.         case count > 0:
  71.             c.leaks[port] = count - 1
  72.             if err := rebuilt.Allocate(port); err != nil {
  73.                 runtime.HandleError(fmt.Errorf("the node port %d may have leaked, but can not be allocated: %v", port, err))
  74.             }
  75.         default:
  76.             ......
  77.         }
  78.     })
  80.     // 11、更新 snapshot
  81.     if err := rebuilt.Snapshot(snapshot); err != nil {
  82.         return fmt.Errorf("unable to snapshot the updated port allocations: %v", err)
  83.     }
  84.     ......
  85.     return nil
  86. }

以上就是 bootstrap controller 的主要实现。

总结

本文主要分析了 kube-apiserver 中 apiserver service 的实现，apiserver service 是通过 bootstrap controller 控制的，bootstrap controller 会保证 apiserver service 以及其 endpoint 处于正常状态，需要注意的是，apiserver service 的 endpoint 根据启动时指定的参数分为三种控制方式，本文仅分析了 lease 的实现方式，如果使用 master-count 方式，需要将每个 master 实例的 port、apiserver-count 等配置参数改为一致。