Kubernetes 源码分析 -- API Server 系统启动

API Server的服务

在进入代码分析之前，这里首先把API Server的服务理清，这样将会对后面的代码的分析有更好的理解，API Server中的服务架构图如下所示：

apiserver infrastructure.png

整个程序的基本过程为：

命令行参数解析，参数存储在ServerRunOptions
基于ServerRunOptions构建genericapiserver.Config
构建master.Config
构建apiextensionserver.Config，并创建API Extension Server
构建Master（API Server），把API Extension Server作为代理服务，这样它会融合API Extension Server的服务
构建Aggregator，把Master作为代理服务
启动http非安全服务
执行Aggregator的Run，启动Https安全服务

进程启动过程

kubernetes 1.10代码的启动都是采用了cobra框架，下面代码是为API Server创建一个cobra.Command。

func NewAPIServerCommand() *cobra.Command {
    s := options.NewServerRunOptions()
    cmd := &cobra.Command{
        Use: "kube-apiserver",
        Long: `The Kubernetes API server validates and configures data
for the api objects which include pods, services, replicationcontrollers, and
others. The API Server services REST operations and provides the frontend to the
cluster's shared state through which all other components interact.`,
        Run: func(cmd *cobra.Command, args []string) {
            verflag.PrintAndExitIfRequested()

            stopCh := server.SetupSignalHandler()
            if err := Run(s, stopCh); err != nil {
                fmt.Fprintf(os.Stderr, "%v\n", err)
                os.Exit(1)
            }
        },
    }
    s.AddFlags(cmd.Flags())

    return cmd
}

APIServer的启动参数存储在ServerRunOptions中，启动参数基于命令行参数进行解析。
cobra框架会自动分析命令行参数，命令行参数的解析是由s.AddFlag(cmd.Flags())中预先定义好的。在AddFlags方法中。

// AddFlags adds flags for a specific APIServer to the specified FlagSet
func (s *ServerRunOptions) AddFlags(fs *pflag.FlagSet) {
    // Add the generic flags.
    s.GenericServerRunOptions.AddUniversalFlags(fs)
    s.Etcd.AddFlags(fs)
    s.SecureServing.AddFlags(fs)
    s.InsecureServing.AddFlags(fs)
    s.InsecureServing.AddDeprecatedFlags(fs)
    s.Audit.AddFlags(fs)
    s.Features.AddFlags(fs)
    s.Authentication.AddFlags(fs)
    s.Authorization.AddFlags(fs)
    s.CloudProvider.AddFlags(fs)
    s.StorageSerialization.AddFlags(fs)
    s.APIEnablement.AddFlags(fs)
    s.Admission.AddFlags(fs)

       ......   
    fs.BoolVar(&s.EnableAggregatorRouting, "enable-aggregator-routing", s.EnableAggregatorRouting,
        "Turns on aggregator routing requests to endoints IP rather than cluster IP.")

}

我们先来看看都有哪些常用的命令行参数被传递给了APISrever对象，下面是运行Master节点的kube-apiserver进程的命令行信息：

kube-apiserver --admission-control=NamespaceLifecycle,LimitRanger,ServiceAccount,ResourceQuota --apiserver-count=3 --insecure-port=8080 --secure-port=6443 --insecure-bind-address=127.0.0.1 --tls-cert-file=/data/k8s/ssl/kubernetes.pem --tls-private-key-file=/data/k8s/ssl/kubernetes-key.pem --client-ca-file=/data/k8s/ssl/ca.pem --service-account-key-file=/data/k8s/ssl/kubernetes-key.pem --storage-backend=etcd3 --enable-swagger-ui --etcd-cafile=/data/k8s/ssl/ca.pem --etcd-certfile=/data/k8s/ssl/kubernetes.pem --etcd-keyfile=/data/k8s/ssl/kubernetes-key.pem --etcd-servers=https://221.228.106.151:2379,https://61.160.36.43:2379,https://58.215.170.46:2379 --kubelet-certificate-authority=/data/k8s/ssl/ca.pem --kubelet-client-certificate=/data/k8s/ssl/kubelet.pem --kubelet-client-key=/data/k8s/ssl/kubelet-key.pem --runtime-config=settings.k8s.io/v1alpha1=true --enable-bootstrap-token-auth=true --token-auth-file=/data/k8s/ssl/token.csv --allow-privileged=true --service-cluster-ip-range=10.254.0.0/16 --audit-policy-file=/data/k8s/ssl/audit-policy.yaml --audit-log-maxage=30 --audit-log-maxbackup=3 --audit-log-maxsize=100 --audit-log-path=/data/yy/log/kubeapiserver/audit.log --logtostderr=false --log-dir=/data/yy/log/kubeapiserver --v=2 --kubelet-preferred-address-types=InternalIP,ExternalIP,InternalDNS,ExternalDNS,Hostname

etcd相关的参数
storage-backend、etcd-prefix、etcd-cafile、etcd-certfile、etcd-keyfile、etcd-servers
BuildinAuthenticationOptions相关参数
enable-bootstrap-token-auth、token-auth-file、service-account-key-file（用于验证ServiceAccount tokens。如果没有指定，则使用tls-private-key-file）
AdmissionOptions相关参数
admission-control
InsecureServingOptions相关参数
insecure-port、insecure-bind-address
AuditOptions相关参数
audit-policy-file、audit-log-path、audit-log-maxsize、audit-log-maxbackup、audit-log-format
SecureServingOptions相关参数
bind-address、secure-port、tls-private-key-file、tls-cert-file、
ClientCertAuthenticationOptions相关参数
client-ca-file
公共参数
kubelet-preferred-address-types、
kubelet-certificate-authority、kubelet-client-key、kubelet-certificate-authority、

最终，我们会执行Command.Execute方法启动程序，它会首先解析参数，然后执行Run方法。

// Run runs the specified APIServer.  This should never exit.
func Run(runOptions *options.ServerRunOptions, stopCh <-chan struct{}) error {
    // To help debugging, immediately log version
    glog.Infof("Version: %+v", version.Get())

    server, err := CreateServerChain(runOptions, stopCh)
    if err != nil {
        return err
    }

    return server.PrepareRun().Run(stopCh)
}

构建服务器调用链

在上面的Run方法中，CreateServerChain负责构建服务器链，最终返回genericapiserver.GenericAPIServer实例，代码如下所示：

经过前面的架构分析，API Server主要工作都在CreateServerChain中完成的。

// CreateServerChain creates the apiservers connected via delegation.
func CreateServerChain(runOptions *options.ServerRunOptions, stopCh <-chan struct{}) (*genericapiserver.GenericAPIServer, error) {
       // 如果指定了SSHUser参数，CreateNodeDialer用于创建连接本机节点的相关传输方法和网络建立方法
       // 所以与本机是通过SSH隧道交互的。通常情况下，我们是没有设置SSHUser参数的。
    nodeTunneler, proxyTransport, err := CreateNodeDialer(runOptions)
    if err != nil {
        return nil, err
    }

    kubeAPIServerConfig, sharedInformers, versionedInformers, insecureServingOptions, serviceResolver, err := CreateKubeAPIServerConfig(runOptions, nodeTunneler, proxyTransport)
    if err != nil {
        return nil, err
    }

    // TPRs are enabled and not yet beta, since this these are the successor, they fall under the same enablement rule
    // If additional API servers are added, they should be gated.
    apiExtensionsConfig, err := createAPIExtensionsConfig(*kubeAPIServerConfig.GenericConfig, versionedInformers, runOptions)
    if err != nil {
        return nil, err
    }
    // API ExtensionsServer是代理用
    apiExtensionsServer, err := createAPIExtensionsServer(apiExtensionsConfig, genericapiserver.EmptyDelegate)
    if err != nil {
        return nil, err
    }

    kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer, sharedInformers, versionedInformers)
    if err != nil {
        return nil, err
    }

    // if we're starting up a hacked up version of this API server for a weird test case,
    // just start the API server as is because clients don't get built correctly when you do this
    if len(os.Getenv("KUBE_API_VERSIONS")) > 0 {
        if insecureServingOptions != nil {
            insecureHandlerChain := kubeserver.BuildInsecureHandlerChain(kubeAPIServer.GenericAPIServer.UnprotectedHandler(), kubeAPIServerConfig.GenericConfig)
            if err := kubeserver.NonBlockingRun(insecureServingOptions, insecureHandlerChain, kubeAPIServerConfig.GenericConfig.RequestTimeout, stopCh); err != nil {
                return nil, err
            }
        }

        return kubeAPIServer.GenericAPIServer, nil
    }

    // otherwise go down the normal path of standing the aggregator up in front of the API server
    // this wires up openapi
    kubeAPIServer.GenericAPIServer.PrepareRun()

    // This will wire up openapi for extension api server
    apiExtensionsServer.GenericAPIServer.PrepareRun()

    // aggregator comes last in the chain
    aggregatorConfig, err := createAggregatorConfig(*kubeAPIServerConfig.GenericConfig, runOptions, versionedInformers, serviceResolver, proxyTransport)
    if err != nil {
        return nil, err
    }
    aggregatorConfig.ExtraConfig.ProxyTransport = proxyTransport
    aggregatorConfig.ExtraConfig.ServiceResolver = serviceResolver
    aggregatorServer, err := createAggregatorServer(aggregatorConfig, kubeAPIServer.GenericAPIServer, apiExtensionsServer.Informers)
    if err != nil {
        // we don't need special handling for innerStopCh because the aggregator server doesn't create any go routines
        return nil, err
    }

    if insecureServingOptions != nil {
        insecureHandlerChain := kubeserver.BuildInsecureHandlerChain(aggregatorServer.GenericAPIServer.UnprotectedHandler(), kubeAPIServerConfig.GenericConfig)
        if err := kubeserver.NonBlockingRun(insecureServingOptions, insecureHandlerChain, kubeAPIServerConfig.GenericConfig.RequestTimeout, stopCh); err != nil {
            return nil, err
        }
    }

    return aggregatorServer.GenericAPIServer, nil
}

CreateServerChain完成了主要的流程，里面有几个步骤比较特殊，这里分别讲解

CreateKubeAPIServerConfig

CreateKubeAPIServerConfig负责创建Master Server的配置、以及相关的资源。

func BuildGenericConfig(s *options.ServerRunOptions, proxyTransport *http.Transport) (*genericapiserver.Config, informers.SharedInformerFactory, clientgoinformers.SharedInformerFactory, *kubeserver.InsecureServingInfo, aggregatorapiserver.ServiceResolver, error) {
    genericConfig := genericapiserver.NewConfig(legacyscheme.Codecs)        // 构建通用配置实例
        if err := s.GenericServerRunOptions.ApplyTo(genericConfig); err != nil {    // 运行参数->通用参数
        return nil, nil, nil, nil, nil, err
    }

    insecureServingOptions, err := s.InsecureServing.ApplyTo(genericConfig)     // 非安全服务属性 -> 通用参数
    if err != nil {
        return nil, nil, nil, nil, nil, err
    }
    if err := s.SecureServing.ApplyTo(genericConfig); err != nil {      // 安全服务属性 -> 通用参数
        return nil, nil, nil, nil, nil, err
    }
    if err := s.Authentication.ApplyTo(genericConfig); err != nil {     // 认证信息 -> 通用参数
        return nil, nil, nil, nil, nil, err
    }
    if err := s.Audit.ApplyTo(genericConfig); err != nil {              // 审计信息 -> 通用参数
        return nil, nil, nil, nil, nil, err
    }
    if err := s.Features.ApplyTo(genericConfig); err != nil {           // Features -> 通用参数（EnableProfiling、EnableContentionProfiling、EnableSwaggerUI）
        return nil, nil, nil, nil, nil, err
    }
    if err := s.APIEnablement.ApplyTo(genericConfig, master.DefaultAPIResourceConfigSource(), legacyscheme.Registry); err != nil {  // 启动API的资源信息，格式为  --runtime-config=api/all=false,api/v1=true
        return nil, nil, nil, nil, nil, err
    }

    // OpenAPI配置
    genericConfig.OpenAPIConfig = genericapiserver.DefaultOpenAPIConfig(generatedopenapi.GetOpenAPIDefinitions, legacyscheme.Scheme)
    genericConfig.OpenAPIConfig.PostProcessSpec = postProcessOpenAPISpecForBackwardCompatibility
    genericConfig.OpenAPIConfig.Info.Title = "Kubernetes"
    genericConfig.SwaggerConfig = genericapiserver.DefaultSwaggerConfig()
    genericConfig.EnableMetrics = true
    genericConfig.LongRunningFunc = filters.BasicLongRunningRequestCheck(
        sets.NewString("watch", "proxy"),
        sets.NewString("attach", "exec", "proxy", "log", "portforward"),
    )

    kubeVersion := version.Get()
    genericConfig.Version = &kubeVersion

    // 构建存储工厂实例，指定的参数中:
    // 1. s： 提供基本的Etcd的配置信息
    // 2. genericConfig.MergedResourceConfig：指定了Enable的API Resource
    storageFactory, err := BuildStorageFactory(s, genericConfig.MergedResourceConfig)
    if err != nil {
        return nil, nil, nil, nil, nil, err
    }
    // 把存储工厂实例封装存放到genericConfig.RESTOptionsGetter 变量（同时添加Etcd的健康检查方法）
    if err := s.Etcd.ApplyWithStorageFactoryTo(storageFactory, genericConfig); err != nil {
        return nil, nil, nil, nil, nil, err
    }

    // 使用protobufs用来内部交互，（外部的看来都是application/json）
    // 看来是基于Loopback的ip和端口来访问自己？？？？？
    // Use protobufs for self-communication.
    // Since not every generic apiserver has to support protobufs, we
    // cannot default to it in generic apiserver and need to explicitly
    // set it in kube-apiserver.
    genericConfig.LoopbackClientConfig.ContentConfig.ContentType = "application/vnd.kubernetes.protobuf"

    // 生成API 客户端
    client, err := internalclientset.NewForConfig(genericConfig.LoopbackClientConfig)
    if err != nil {
        kubeAPIVersions := os.Getenv("KUBE_API_VERSIONS")
        if len(kubeAPIVersions) == 0 {
            return nil, nil, nil, nil, nil, fmt.Errorf("failed to create clientset: %v", err)
        }

        // KUBE_API_VERSIONS is used in test-update-storage-objects.sh, disabling a number of API
        // groups. This leads to a nil client above and undefined behaviour further down.
        //
        // TODO: get rid of KUBE_API_VERSIONS or define sane behaviour if set
        glog.Errorf("Failed to create clientset with KUBE_API_VERSIONS=%q. KUBE_API_VERSIONS is only for testing. Things will break.", kubeAPIVersions)
    }

    kubeClientConfig := genericConfig.LoopbackClientConfig
    // 原来创建到自己的连接的目的在这里，实现共享通知功能，这个在出了API Server的其他模块都在大量使用。
    // 资源的变化，都能够通过该功能进行通知。从而可以依据它来做出必要的响应。
    sharedInformers := informers.NewSharedInformerFactory(client, 10*time.Minute)

    // 这里生成了另外一个Client
    // TODO：需要了解一下clientgoclientset 与 internelclientset的区别在哪里
    clientgoExternalClient, err := clientgoclientset.NewForConfig(kubeClientConfig)
    if err != nil {
        return nil, nil, nil, nil, nil, fmt.Errorf("failed to create real external clientset: %v", err)
    }
    versionedInformers := clientgoinformers.NewSharedInformerFactory(clientgoExternalClient, 10*time.Minute)

    var serviceResolver aggregatorapiserver.ServiceResolver // ServiceResolver让我们找到Srevice的URI
    if s.EnableAggregatorRouting {  // EnableAggregatorRouting似乎是让所有的Service，都基于Endpoint Ip来访问，而不基于Cluster Ip
        serviceResolver = aggregatorapiserver.NewEndpointServiceResolver(
            versionedInformers.Core().V1().Services().Lister(),
            versionedInformers.Core().V1().Endpoints().Lister(),
        )
    } else {
        serviceResolver = aggregatorapiserver.NewClusterIPServiceResolver(
            versionedInformers.Core().V1().Services().Lister(),
        )
    }

    // 创建认证实例
    // 注意：可能有多种认证，最终会合并成一个对象（合并方法：在union包中，func union.New(authRequestHandlers ...authenticator.Request) authenticator.Request）
    // 支持的认证：
    // 1. 请求Header认证
    // 2. Auth文件认证
    // 3. KeyStone认证，应该是KMS吧。
    // 4. CA认证认证（也就是常用的双向认证） 【 一般的使用方式 】
    // 5. Token文件认证
    // 6. ServiceAccount认证
    // 7. Bootstrap认证。。。。。          【一般使用，生成客户端证书】
    genericConfig.Authenticator, genericConfig.OpenAPIConfig.SecurityDefinitions, err = BuildAuthenticator(s, storageFactory, client, sharedInformers)
    if err != nil {
        return nil, nil, nil, nil, nil, fmt.Errorf("invalid authentication config: %v", err)
    }

    // 创建授权实例，这块用的不多，先不折腾
    genericConfig.Authorizer, genericConfig.RuleResolver, err = BuildAuthorizer(s, sharedInformers, versionedInformers)
    if err != nil {
        return nil, nil, nil, nil, nil, fmt.Errorf("invalid authorization config: %v", err)
    }
    if !sets.NewString(s.Authorization.Modes()...).Has(modes.ModeRBAC) {
        genericConfig.DisabledPostStartHooks.Insert(rbacrest.PostStartHookName)
    }

    webhookAuthResolverWrapper := func(delegate webhookconfig.AuthenticationInfoResolver) webhookconfig.AuthenticationInfoResolver {
        return &webhookconfig.AuthenticationInfoResolverDelegator{
            ClientConfigForFunc: func(server string) (*rest.Config, error) {
                if server == "kubernetes.default.svc" {
                    return genericConfig.LoopbackClientConfig, nil
                }
                return delegate.ClientConfigFor(server)
            },
            ClientConfigForServiceFunc: func(serviceName, serviceNamespace string) (*rest.Config, error) {
                if serviceName == "kubernetes" && serviceNamespace == "default" {
                    return genericConfig.LoopbackClientConfig, nil
                }
                ret, err := delegate.ClientConfigForService(serviceName, serviceNamespace)
                if err != nil {
                    return nil, err
                }
                if proxyTransport != nil && proxyTransport.Dial != nil {
                    ret.Dial = proxyTransport.Dial
                }
                return ret, err
            },
        }
    }
    // 准入插件初始化，CloudProvider???
    // 暂时没有研究，用的少，有必要时在研究。
    pluginInitializers, err := BuildAdmissionPluginInitializers(
        s,
        client,
        sharedInformers,
        serviceResolver,
        webhookAuthResolverWrapper,
    )
    if err != nil {
        return nil, nil, nil, nil, nil, fmt.Errorf("failed to create admission plugin initializer: %v", err)
    }

    err = s.Admission.ApplyTo(
        genericConfig,
        versionedInformers,
        kubeClientConfig,
        legacyscheme.Scheme,
        pluginInitializers...)
    if err != nil {
        return nil, nil, nil, nil, nil, fmt.Errorf("failed to initialize admission: %v", err)
    }

    return genericConfig, sharedInformers, versionedInformers, insecureServingOptions, serviceResolver, nil
}

BuildGenericConfig

构建Generic Config的过程，这块提供了通用配置，这块配置会被多个服务共同使用。

首先基于legacyscheme.Codecs创建一个缺省的GenericConfig实例，并把ServerRunOptions中的参数引用到GenrericConfig中，在这里构建了启动服务的基本要素：

1. genericConfig
1. sharedInformer：内部版本的共享数据缓存通知器
1. versionedInformers：基于clientgoclientset的共享缓存通知器
1. insecureServingOptions
1. servideResolve：服务（service资源）解析器，根据是否启用enable-aggrator-routing，来确定是基于cluster ip还是endpoint ip的解析方式

func BuildGenericConfig(s *options.ServerRunOptions, proxyTransport *http.Transport) (*genericapiserver.Config, informers.SharedInformerFactory, clientgoinformers.SharedInformerFactory, *kubeserver.InsecureServingInfo, aggregatorapiserver.ServiceResolver, error) {
    genericConfig := genericapiserver.NewConfig(legacyscheme.Codecs)        // 构建通用配置实例
        if err := s.GenericServerRunOptions.ApplyTo(genericConfig); err != nil {    // 运行参数->通用参数
        return nil, nil, nil, nil, nil, err
    }

    insecureServingOptions, err := s.InsecureServing.ApplyTo(genericConfig)     // 非安全服务属性 -> 通用参数
    if err != nil {
        return nil, nil, nil, nil, nil, err
    }
    if err := s.SecureServing.ApplyTo(genericConfig); err != nil {      // 安全服务属性 -> 通用参数
        return nil, nil, nil, nil, nil, err
    }
    if err := s.Authentication.ApplyTo(genericConfig); err != nil {     // 认证信息 -> 通用参数
        return nil, nil, nil, nil, nil, err
    }
    if err := s.Audit.ApplyTo(genericConfig); err != nil {              // 审计信息 -> 通用参数
        return nil, nil, nil, nil, nil, err
    }
    if err := s.Features.ApplyTo(genericConfig); err != nil {           // Features -> 通用参数（EnableProfiling、EnableContentionProfiling、EnableSwaggerUI）
        return nil, nil, nil, nil, nil, err
    }
    if err := s.APIEnablement.ApplyTo(genericConfig, master.DefaultAPIResourceConfigSource(), legacyscheme.Registry); err != nil {  // 启动API的资源信息，格式为  --runtime-config=api/all=false,api/v1=true
        return nil, nil, nil, nil, nil, err
    }

    // OpenAPI配置
    genericConfig.OpenAPIConfig = genericapiserver.DefaultOpenAPIConfig(generatedopenapi.GetOpenAPIDefinitions, legacyscheme.Scheme)
    genericConfig.OpenAPIConfig.PostProcessSpec = postProcessOpenAPISpecForBackwardCompatibility
    genericConfig.OpenAPIConfig.Info.Title = "Kubernetes"
    genericConfig.SwaggerConfig = genericapiserver.DefaultSwaggerConfig()
    genericConfig.EnableMetrics = true
    genericConfig.LongRunningFunc = filters.BasicLongRunningRequestCheck(
        sets.NewString("watch", "proxy"),
        sets.NewString("attach", "exec", "proxy", "log", "portforward"),
    )

    kubeVersion := version.Get()
    genericConfig.Version = &kubeVersion

    // 构建存储工厂实例，指定的参数中:
    // 1. s： 提供基本的Etcd的配置信息
    // 2. genericConfig.MergedResourceConfig：指定了Enable的API Resource
    storageFactory, err := BuildStorageFactory(s, genericConfig.MergedResourceConfig)
    if err != nil {
        return nil, nil, nil, nil, nil, err
    }
    // 把存储工厂实例封装存放到genericConfig.RESTOptionsGetter 变量（同时添加Etcd的健康检查方法）
    if err := s.Etcd.ApplyWithStorageFactoryTo(storageFactory, genericConfig); err != nil {
        return nil, nil, nil, nil, nil, err
    }

    // 使用protobufs用来内部交互，（外部的看来都是application/json）
    // 看来是基于Loopback的ip和端口来访问自己？？？？？
    // Use protobufs for self-communication.
    // Since not every generic apiserver has to support protobufs, we
    // cannot default to it in generic apiserver and need to explicitly
    // set it in kube-apiserver.
    genericConfig.LoopbackClientConfig.ContentConfig.ContentType = "application/vnd.kubernetes.protobuf"

    // 生成API 客户端
    client, err := internalclientset.NewForConfig(genericConfig.LoopbackClientConfig)
    if err != nil {
        kubeAPIVersions := os.Getenv("KUBE_API_VERSIONS")
        if len(kubeAPIVersions) == 0 {
            return nil, nil, nil, nil, nil, fmt.Errorf("failed to create clientset: %v", err)
        }

        // KUBE_API_VERSIONS is used in test-update-storage-objects.sh, disabling a number of API
        // groups. This leads to a nil client above and undefined behaviour further down.
        //
        // TODO: get rid of KUBE_API_VERSIONS or define sane behaviour if set
        glog.Errorf("Failed to create clientset with KUBE_API_VERSIONS=%q. KUBE_API_VERSIONS is only for testing. Things will break.", kubeAPIVersions)
    }

    kubeClientConfig := genericConfig.LoopbackClientConfig
    // 原来创建到自己的连接的目的在这里，实现共享通知功能，这个在出了API Server的其他模块都在大量使用。
    // 资源的变化，都能够通过该功能进行通知。从而可以依据它来做出必要的响应。
    sharedInformers := informers.NewSharedInformerFactory(client, 10*time.Minute)

    // 这里生成了另外一个Client
    // TODO：需要了解一下clientgoclientset 与 internelclientset的区别在哪里
    clientgoExternalClient, err := clientgoclientset.NewForConfig(kubeClientConfig)
    if err != nil {
        return nil, nil, nil, nil, nil, fmt.Errorf("failed to create real external clientset: %v", err)
    }
    versionedInformers := clientgoinformers.NewSharedInformerFactory(clientgoExternalClient, 10*time.Minute)

    var serviceResolver aggregatorapiserver.ServiceResolver // ServiceResolver让我们找到Srevice的URI
    if s.EnableAggregatorRouting {  // EnableAggregatorRouting似乎是让所有的Service，都基于Endpoint Ip来访问，而不基于Cluster Ip
        serviceResolver = aggregatorapiserver.NewEndpointServiceResolver(
            versionedInformers.Core().V1().Services().Lister(),
            versionedInformers.Core().V1().Endpoints().Lister(),
        )
    } else {
        serviceResolver = aggregatorapiserver.NewClusterIPServiceResolver(
            versionedInformers.Core().V1().Services().Lister(),
        )
    }

    // 创建认证实例
    // 注意：可能有多种认证，最终会合并成一个对象（合并方法：在union包中，func union.New(authRequestHandlers ...authenticator.Request) authenticator.Request）
    // 支持的认证：
    // 1. 请求Header认证
    // 2. Auth文件认证
    // 3. KeyStone认证，应该是KMS吧。
    // 4. CA认证认证（也就是常用的双向认证） 【 一般的使用方式 】
    // 5. Token文件认证
    // 6. ServiceAccount认证
    // 7. Bootstrap认证。。。。。          【一般使用，生成客户端证书】
    genericConfig.Authenticator, genericConfig.OpenAPIConfig.SecurityDefinitions, err = BuildAuthenticator(s, storageFactory, client, sharedInformers)
    if err != nil {
        return nil, nil, nil, nil, nil, fmt.Errorf("invalid authentication config: %v", err)
    }
      .....

    return genericConfig, sharedInformers, versionedInformers, insecureServingOptions, serviceResolver, nil
}

存储工厂实例

存储这块会在《API Server之 Etcd存储》中讲解

构建master.Config

master.Config是API Server服务的配置。

preparedGenericAPIServer.Run -> preparedGenericAPIServer.NonBlockingRun -> GenericAPIServer.serveSecurely -> RunServer

启动服务

API Server运行了安全服务和非安全服务，不过过程都差不多，这里只分析安全服务的运行过程，最终的执行我们直到，是运行Aggrator Server实例的Run方法，在这个方法中的调用链为：