0、架构

如下图所示。k8s作为集群的入口，是唯一能够和etcd打交道的模块。其作用主要如下：
1、作为Kubernetes API的服务端，为集群内的节点以及kubectl工具提供API服务
2、与ETCD打交道，实现对k8s内部obj的增删改查
3、保证在分布式存储系统(etcd)中的Kubernetes API对象的状态一致**
image.png

1、配置解析/启动

1.1、option(会删除部分代码，不影响理解)

  1. type ServerRunOptions struct {
  2.     GenericServerRunOptions *genericoptions.ServerRunOptions
  3.     Etcd                    *genericoptions.EtcdOptions
  4.     SecureServing           *genericoptions.SecureServingOptionsWithLoopback
  5.     InsecureServing         *genericoptions.DeprecatedInsecureServingOptionsWithLoopback
  6.     Audit                   *genericoptions.AuditOptions
  7.     Features                *genericoptions.FeatureOptions
  8.     Admission               *kubeoptions.AdmissionOptions
  9.     Authentication          *kubeoptions.BuiltInAuthenticationOptions
  10.     Authorization           *kubeoptions.BuiltInAuthorizationOptions
  11.     CloudProvider           *kubeoptions.CloudProviderOptions
  12.     APIEnablement           *genericoptions.APIEnablementOptions
  13.     EgressSelector          *genericoptions.EgressSelectorOptions
  15.     AllowPrivileged                  bool
  16.     EnableLogsHandler                bool
  17.     EventTTL                         time.Duration
  18.     KubeletConfig                    kubeletclient.KubeletClientConfig
  19.     KubernetesServiceNodePort        int
  20.     MaxConnectionBytesPerSec         int64
  21.     ServiceClusterIPRanges           string
  22.     PrimaryServiceClusterIPRange     net.IPNet
  23.     SecondaryServiceClusterIPRange   net.IPNet
  24.     ServiceNodePortRange             utilnet.PortRange
  25.     SSHKeyfile                       string
  26.     SSHUser                          string
  27.     ProxyClientCertFile              string
  28.     ProxyClientKeyFile               string
  29.     EnableAggregatorRouting          bool
  30.     MasterCount                      int
  31.     EndpointReconcilerType           string
  32.     ServiceAccountSigningKeyFile     string
  33.     ServiceAccountIssuer             serviceaccount.TokenGenerator
  34.     ServiceAccountTokenMaxExpiration time.Duration
  35.     ShowHiddenMetricsForVersion      string
  36. }

1.1.1、GenericServerRunOptions

  1. // 运行generic server的options
  2. type ServerRunOptions struct {
  3.     AdvertiseAddress net.IP
  5.     CorsAllowedOriginList       []string
  6.     ExternalHost                string
  7.     MaxRequestsInFlight         int
  8.     MaxMutatingRequestsInFlight int
  9.     RequestTimeout              time.Duration
  10.     GoawayChance                float64
  11.     LivezGracePeriod            time.Duration
  12.     MinRequestTimeout           int
  13.     ShutdownDelayDuration       time.Duration
  14.     // We intentionally did not add a flag for this option. Users of the
  15.     // apiserver library can wire it to a flag.
  16.     JSONPatchMaxCopyBytes int64
  17.     // The limit on the request body size that would be accepted and
  18.     // decoded in a write request. 0 means no limit.
  19.     // We intentionally did not add a flag for this option. Users of the
  20.     // apiserver library can wire it to a flag.
  21.     MaxRequestBodyBytes       int64
  22.     TargetRAMMB               int
  23.     EnablePriorityAndFairness bool
  24. }
  26. // 默认值
  27. func NewServerRunOptions() *ServerRunOptions {
  28.     defaults := server.NewConfig(serializer.CodecFactory{})
  29.     return &ServerRunOptions{
  30.         MaxRequestsInFlight:         defaults.MaxRequestsInFlight,
  31.         MaxMutatingRequestsInFlight: defaults.MaxMutatingRequestsInFlight,
  32.         RequestTimeout:              defaults.RequestTimeout,
  33.         LivezGracePeriod:            defaults.LivezGracePeriod,
  34.         MinRequestTimeout:           defaults.MinRequestTimeout,
  35.         ShutdownDelayDuration:       defaults.ShutdownDelayDuration,
  36.         JSONPatchMaxCopyBytes:       defaults.JSONPatchMaxCopyBytes,
  37.         MaxRequestBodyBytes:         defaults.MaxRequestBodyBytes,
  38.         EnablePriorityAndFairness:   true,
  39.     }
  40. }
  42. // 此处为NewConfig函数
  43. func NewConfig(codecs serializer.CodecFactory) *Config {
  44.     defaultHealthChecks := []healthz.HealthChecker{healthz.PingHealthz, healthz.LogHealthz}
  45.     return &Config{
  46.         Serializer:                  codecs,
  47.         BuildHandlerChainFunc:       DefaultBuildHandlerChain,
  48.         HandlerChainWaitGroup:       new(utilwaitgroup.SafeWaitGroup),
  49.         LegacyAPIGroupPrefixes:      sets.NewString(DefaultLegacyAPIPrefix),
  50.         DisabledPostStartHooks:      sets.NewString(),
  51.         PostStartHooks:              map[string]PostStartHookConfigEntry{},
  52.         HealthzChecks:               append([]healthz.HealthChecker{}, defaultHealthChecks...),
  53.         ReadyzChecks:                append([]healthz.HealthChecker{}, defaultHealthChecks...),
  54.         LivezChecks:                 append([]healthz.HealthChecker{}, defaultHealthChecks...),
  55.         EnableIndex:                 true,
  56.         EnableDiscovery:             true,
  57.         EnableProfiling:             true,
  58.         EnableMetrics:               true,
  59.         MaxRequestsInFlight:         400,
  60.         MaxMutatingRequestsInFlight: 200,
  61.         RequestTimeout:              time.Duration(60) * time.Second,
  62.         MinRequestTimeout:           1800,
  63.         LivezGracePeriod:            time.Duration(0),
  64.         ShutdownDelayDuration:       time.Duration(0),
  65.         // 1.5MB is the default client request size in bytes
  66.         // the etcd server should accept. See
  67.         // https://github.com/etcd-io/etcd/blob/release-3.4/embed/config.go#L56.
  68.         // A request body might be encoded in json, and is converted to
  69.         // proto when persisted in etcd, so we allow 2x as the largest size
  70.         // increase the "copy" operations in a json patch may cause.
  71.         JSONPatchMaxCopyBytes: int64(3 * 1024 * 1024),
  72.         // 1.5MB is the recommended client request size in byte
  73.         // the etcd server should accept. See
  74.         // https://github.com/etcd-io/etcd/blob/release-3.4/embed/config.go#L56.
  75.         // A request body might be encoded in json, and is converted to
  76.         // proto when persisted in etcd, so we allow 2x as the largest request
  77.         // body size to be accepted and decoded in a write request.
  78.         MaxRequestBodyBytes: int64(3 * 1024 * 1024),
  80.         // Default to treating watch as a long-running operation
  81.         // Generic API servers have no inherent long-running subresources
  82.         LongRunningFunc: genericfilters.BasicLongRunningRequestCheck(sets.NewString("watch"), sets.NewString()),
  83.     }
  84. }

1.1.2、Etcd

  1. // 后端存储类型
  2. var storageTypes = sets.NewString(
  3.     storagebackend.StorageTypeETCD3,
  4. )
  6. // etcd配置文件
  7. type EtcdOptions struct {
  8.     // The value of Paging on StorageConfig will be overridden by the
  9.     // calculated feature gate value.
  10.     StorageConfig                    storagebackend.Config
  11.     EncryptionProviderConfigFilepath string
  13.     EtcdServersOverrides []string
  15.     // To enable protobuf as storage format, it is enough
  16.     // to set it to "application/vnd.kubernetes.protobuf".
  17.     DefaultStorageMediaType string
  18.     DeleteCollectionWorkers int
  19.     EnableGarbageCollection bool
  21.     // Set EnableWatchCache to false to disable all watch caches
  22.     EnableWatchCache bool
  23.     // Set DefaultWatchCacheSize to zero to disable watch caches for those resources that have no explicit cache size set
  24.     DefaultWatchCacheSize int
  25.     // WatchCacheSizes represents override to a given resource
  26.     WatchCacheSizes []string
  27. }
  29. func (s *EtcdOptions) ApplyTo(c *server.Config) error {
  30.     if s == nil {
  31.         return nil
  32.     }
  33.     if err := s.addEtcdHealthEndpoint(c); err != nil {
  34.         return err
  35.     }
  36.     c.RESTOptionsGetter = &SimpleRestOptionsFactory{Options: *s}
  37.     return nil
  38. }
  40. func (s *EtcdOptions) ApplyWithStorageFactoryTo(factory serverstorage.StorageFactory, c *server.Config) error {
  41.     if err := s.addEtcdHealthEndpoint(c); err != nil {
  42.         return err
  43.     }
  44.     c.RESTOptionsGetter = &StorageFactoryRestOptionsFactory{Options: *s, StorageFactory: factory}
  45.     return nil
  46. }
  48. func (s *EtcdOptions) addEtcdHealthEndpoint(c *server.Config) error {
  49.     healthCheck, err := storagefactory.CreateHealthCheck(s.StorageConfig)
  50.     if err != nil {
  51.         return err
  52.     }
  53.     c.AddHealthChecks(healthz.NamedCheck("etcd", func(r *http.Request) error {
  54.         return healthCheck()
  55.     }))
  57.     if s.EncryptionProviderConfigFilepath != "" {
  58.         kmsPluginHealthzChecks, err := encryptionconfig.GetKMSPluginHealthzCheckers(s.EncryptionProviderConfigFilepath)
  59.         if err != nil {
  60.             return err
  61.         }
  62.         c.AddHealthChecks(kmsPluginHealthzChecks...)
  63.     }
  65.     return nil
  66. }
  68. type SimpleRestOptionsFactory struct {
  69.     Options EtcdOptions
  70. }
  72. func (f *SimpleRestOptionsFactory) GetRESTOptions(resource schema.GroupResource) (generic.RESTOptions, error) {
  73.     ret := generic.RESTOptions{
  74.         StorageConfig:           &f.Options.StorageConfig,
  75.         Decorator:               generic.UndecoratedStorage,
  76.         EnableGarbageCollection: f.Options.EnableGarbageCollection,
  77.         DeleteCollectionWorkers: f.Options.DeleteCollectionWorkers,
  78.         ResourcePrefix:          resource.Group + "/" + resource.Resource,
  79.         CountMetricPollPeriod:   f.Options.StorageConfig.CountMetricPollPeriod,
  80.     }
  81.     if f.Options.EnableWatchCache {
  82.         sizes, err := ParseWatchCacheSizes(f.Options.WatchCacheSizes)
  83.         if err != nil {
  84.             return generic.RESTOptions{}, err
  85.         }
  86.         cacheSize, ok := sizes[resource]
  87.         if !ok {
  88.             cacheSize = f.Options.DefaultWatchCacheSize
  89.         }
  90.         // depending on cache size this might return an undecorated storage
  91.         ret.Decorator = genericregistry.StorageWithCacher(cacheSize)
  92.     }
  93.     return ret, nil
  94. }
  96. type StorageFactoryRestOptionsFactory struct {
  97.     Options        EtcdOptions
  98.     StorageFactory serverstorage.StorageFactory
  99. }
  101. func (f *StorageFactoryRestOptionsFactory) GetRESTOptions(resource schema.GroupResource) (generic.RESTOptions, error) {
  102.     storageConfig, err := f.StorageFactory.NewConfig(resource)
  103.     if err != nil {
  104.         return generic.RESTOptions{}, fmt.Errorf("unable to find storage destination for %v, due to %v", resource, err.Error())
  105.     }
  107.     ret := generic.RESTOptions{
  108.         StorageConfig:           storageConfig,
  109.         Decorator:               generic.UndecoratedStorage,
  110.         DeleteCollectionWorkers: f.Options.DeleteCollectionWorkers,
  111.         EnableGarbageCollection: f.Options.EnableGarbageCollection,
  112.         ResourcePrefix:          f.StorageFactory.ResourcePrefix(resource),
  113.         CountMetricPollPeriod:   f.Options.StorageConfig.CountMetricPollPeriod,
  114.     }
  115.     if f.Options.EnableWatchCache {
  116.         sizes, err := ParseWatchCacheSizes(f.Options.WatchCacheSizes)
  117.         if err != nil {
  118.             return generic.RESTOptions{}, err
  119.         }
  120.         cacheSize, ok := sizes[resource]
  121.         if !ok {
  122.             cacheSize = f.Options.DefaultWatchCacheSize
  123.         }
  124.         // depending on cache size this might return an undecorated storage
  125.         ret.Decorator = genericregistry.StorageWithCacher(cacheSize)
  126.     }
  128.     return ret, nil
  129. }
  131. // ParseWatchCacheSizes turns a list of cache size values into a map of group resources
  132. // to requested sizes.
  133. func ParseWatchCacheSizes(cacheSizes []string) (map[schema.GroupResource]int, error) {
  134.     watchCacheSizes := make(map[schema.GroupResource]int)
  135.     for _, c := range cacheSizes {
  136.         tokens := strings.Split(c, "#")
  137.         if len(tokens) != 2 {
  138.             return nil, fmt.Errorf("invalid value of watch cache size: %s", c)
  139.         }
  141.         size, err := strconv.Atoi(tokens[1])
  142.         if err != nil {
  143.             return nil, fmt.Errorf("invalid size of watch cache size: %s", c)
  144.         }
  145.         if size < 0 {
  146.             return nil, fmt.Errorf("watch cache size cannot be negative: %s", c)
  147.         }
  148.         watchCacheSizes[schema.ParseGroupResource(tokens[0])] = size
  149.     }
  150.     return watchCacheSizes, nil
  151. }
  153. // WriteWatchCacheSizes turns a map of cache size values into a list of string specifications.
  154. func WriteWatchCacheSizes(watchCacheSizes map[schema.GroupResource]int) ([]string, error) {
  155.     var cacheSizes []string
  157.     for resource, size := range watchCacheSizes {
  158.         if size < 0 {
  159.             return nil, fmt.Errorf("watch cache size cannot be negative for resource %s", resource)
  160.         }
  161.         cacheSizes = append(cacheSizes, fmt.Sprintf("%s#%d", resource.String(), size))
  162.     }
  163.     return cacheSizes, nil
  164. }

1.2、config

1.3、启动流程

2、REST

2.1、apiserver提供的api

可以看到apiserver提供的api主要分为restful api和非restful api。
restful api主要是对apiserver核心obj的增删改查，非restful api主要是健康检查、监控等api。
其中：
/api/开头的api控制apiserver核心对象，比如pod、node等
/apis/开头的api，增加了group概念，并通过group/kind/version确定唯一一个obj

image.png

2.2、请求流程

如下图，一个完整的请求会有以下路径(以Create为例)：

2.1、经过内置filter的过滤(也即认证、授权、准入)，所有filter如下

image.png

2.2、解析namespace/name

image.png

2.3、获取此url的gvk信息，基于请求的content获取对应格式的序列化器，同时基于HubGV获取对应的解码器。同时解码请求体数据

PS：
1、HubGV一般为内部版本，即group/__internal
2、为了支持多版本，apiserver会讲请求对象解码为实际obj，再转换为内部版本，最后存储为存储版本
image.png
image.png

2.4、将obj的内部版本对象转换为存储版本对象(一般为v1)并存储etcd。利用etcd的事务保证创建幂等

image.png
image.png

2.3、api多版本

概念参照https://www.jianshu.com/p/86130b138a0f

2.4、聚合/CRD

参见学习笔记
k8s源码解析/apiserver/apiextensions-apiserver
k8s源码解析/apiserver/kube-aggregator

2.5、创建HPA全流程debug例子

2.5.0、说明

1、本次debug**通过api而非kubectl
2、debug关闭了部分filter
3、主要针对版本转换代码
4、trace/log可直接跳过
image.png**

2.5.1、在如下文件处打断点，并用debug模式启动apiserver

kubernetes/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/create.go
image.png

2.5.2、利用v2beta1 type创建json文件

  1. {
  2.   "apiVersion": "autoscaling/v2beta1",
  3.   "kind": "HorizontalPodAutoscaler",
  4.   "metadata": {
  5.     "name": "test"
  6.   },
  7.   "spec": {
  8.     "scaleTargetRef": {
  9.       "apiVersion": "extensions/v2beta1",
  10.       "kind": "Deployment",
  11.       "name": "test"
  12.     },
  13.     "minReplicas": 2,
  14.     "maxReplicas": 10,
  15.     "metrics": [
  16.       {
  17.         "type": "Resource",
  18.         "resource": {
  19.           "name": "cpu",
  20.           "targetAverageUtilization": 60
  21.         }
  22.       },
  23.       {
  24.         "type": "Resource",
  25.         "resource": {
  26.           "name": "mem",
  27.           "targetAverageUtilization": 50
  28.         }
  29.       }
  30.     ]
  31.   }
  32. }

2.5.3、调用api创建hpa

image.png

2.5.4、解析ns/name

image.png

2.5.5、获取相应的gv信息和序列化器

image.png
PS：全局序列化器为kubernetes/pkg/api/legacyscheme/scheme.go(s.Serializer即Codecs)
image.png

2.5.6、解析option

option是通过query参数传递进来的，由于本案例没有传如有效option，因此此段代码跳过
image.png

2.5.7、准备解码参数

PS：
r.New()代码位于下图，defaultGVK就是此路径的gvk信息
image.png
image.png

2.5.8、开始解码并转换

2.5.8.1、先将请求体转换为原本的obj(通常情况下此处会转换为defaultGVK对应的obj)

image.png
如下图，可以看到obj为我们v2beta1的obj
image.png

2.5.8.2、转换为内部版本

image.png
可以看到into此时为内部版本对象
image.png

2.5.8.2、转换函数(通常对于内置对象，apiserver注册了默认的转换函数)，v2beta1 -> internal的转换函数为

image.png
函数实现为如下，就是将v2beta1的值赋值到内部版本(内部版本包含了所有的字段)
image.png
image.png
image.png

2.5.9、准备存储

image.png
可以看到再进入存储之前obj已经为内部版本(通过其路径)。
gvk之所以为空是因为apiserver认为内存版本(即内部版本gvk信息多余，因为已经知道是哪个具体的结构体，因此不需要在存储gvk信息，参见：https://github.com/kubernetes/kubernetes/issues/3030)
image.png

2.5.10、开始存储

image.png
可以看到在真正的encode之前会先将obj转换为encodeVersion对应版本的obj
image.png
可以看到转换之前obj为内部版本，转换后的out为v1版本(存储版本)，然后在进行encode，encodeVersion其实就是对应的v1版本，如下图
image.png
image.png

2.5.11、存储到etcd

如下图，当encode结束后，对byte切片进行一些额外的处理，利用etcd的事务进行幂等创建
image.png

2.5.12、创建成功返回

可以看到返回之前需要进行解码和填充版本号的处理
image.png
实现逻辑如下：
可以看到会将data再次转换为内部版本并设置版本号
image.png
image.png

2.5.13、重新转换为v2beta1

可以看到，再最后返回给用户之前，重新获取目标版本，也就是路径版本，并进行转换
image.png
可以看到最终的编码器版本为v2beta1
image.png
image.png
至此整个流程结束

2.5.14、总结

1、期间obj版本经历了v2beta1->internal->v1->internal->v2beta1
2、我们知道v1和v2beta1的字段是不一样的，比如v2beta1有metric而v1没有。如下图1/2。则当转换为v1的时候是不是属性就丢了呢。其实不会。当内部版本转换为v1版本的时候，对于v1版本没有的字段会序列化之后放入到注解里面。当内部版本转换到v2beta1的时候，会反序列化并填充到对应的结构体里面。参见图3
image.png
image.png
image.png

3、搭建debug环境(mac)