内置的 kubernetes service 无法删除，其 ClusterIP 为通过 参数指定的 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 的初始化以及启动是在 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. }

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

    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.     c.runner = async.NewRunner(c.RunKubernetesNamespaces, c.RunKubernetesService, repairClusterIPs.RunUntil, repairNodePorts.RunUntil)
    26. }

    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

    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

    一个集群中可能会有多个 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)
    65.     if !portsCorrect {
    66.         e.Subsets[0].Ports = endpointPorts
    67.     }
    69.         if _, err = r.epAdapter.Create(corev1.NamespaceDefault, e); errors.IsAlreadyExists(err) {
    70.             err = nil
    71.         }
    72.     } else {
    73.         _, err = r.epAdapter.Update(corev1.NamespaceDefault, e)
    74.     }
    75.     return err
    76. }

    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

    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 等配置参数改为一致。