「kubefed 00」从入门到入坑
· 14 min read
Kubernetes 多集群搞了那么多年了，貌似也没倒腾出比较 stable 的项目，看看现在 kubefed 怎么样。本篇是 kubefed 的开篇。
FBI WARNING: 不喜欢看代码分析的就直接去最后一篇看结论就好
前言
当需要考虑到 Kubernetes 多集群的时候，我们往往是什么样的场景？
- 集群容灾：当某个集群异常的时候迁移到另一个集群中
- 多云容灾：当前在业界的另一个解释是不让云厂控制「你」
- 边缘计算：边缘计算中常常会有多地自治域的存在，所以常常是边缘上有一坨的 Kubernetes 集群（没用 Kubernetes 当我没说）
- 普普通通的多集群：真就很普通的多集群，服务需要部署多集群（虽然也是容灾，但不是为了 Kubernetes 的容灾而是服务的容灾）
k8s 在这个 SIG 的路确实走的有点久，从最开始的 Kubernetes Federation 再到推倒重写 kubefed，虽然浩浩荡荡但是还没有 GA。有些厂提供了 kubefed 的能力个用户，也有些许大厂自己做多集群，比如 Anthos、Rancher 等等。 所以究竟 kubefed 是怎么样的一个东西，值得去用吗？不知道，但里面一定有值得学的。
本文基于 kubefed v0.5.1 8a2f4a47 进行分析，因为有代码解读所以会又臭又长，如果不想看代码就直接拉到最后
代码解读
以下忽略几个内容
- DNS 相关部分
概念一口闷
其实这些概念 repo README 也有拉，不过为了阅读方便就搬过来翻译下
Type configuration：对应 FederatedTypeConfig CRD，主要是用来描述什么 API Type 需要被 federated（然后这个 CRD 里主要有一个字段 Propagation 用来标明是否开启传播）
Cluster Configuration：不严格的对应 KubefedCluster CRD，主要是一个集群的集合，但实际上概念中这个还包括了操作对应集群的 KubeConfig 等信息
Federated Type：这货是当我们对某个对象开启 Federate 的时候由 kubefed 生成的一个 CRD，比如给 Deployment 开 Federated，那就会生成一个 FederatedDeployment。在一众的 FederatedType 中不变的就是三件套
- Template（动态生成）：实际我们需要创建的 Object 的 Schema，比如 FederatedDeployment 中的 Template 就是个 Deployment。
- Placement（固定）：指定需要放置的集群，比如通过名字或者通过标签
- Override（固定）：针对于各个集群对 Template 的 patch/override
Scheduling：通过 SchedulingPreference 实现控制各个集群的调度情况，比如当前提供了 Deployment、ReplicaSet 的实例数调度支持，以控制不同集群调度的配比。
当前多集群的结构是：一个 Host Cluster 负责主持大局，然后有无数的 Member Cluster 注册到 Host Cluster 被管理，当然 Host Cluster 可以将自己注册为 Member Cluster。 概括起来就是：一主（Cluster）多从，一挂全挂，横屏 kubefed？
基于上述的概念，咱们可以猜猜是怎么实现的？大概是
- 实现 FederatedCRD（FCRD） 里面包括了我们期望创建的 API Type
- 当我们创建这个 FCRD 时帮我们调度下然后发布到对应的集群上
- 采集点 status
那是事实是？就是这样没错
上代码
咱们看看代码结构吧，不重要的包或者目录我就扔掉了
├── cmd 一切 binary 的入口
│ ├── controller-manager
│ │ └── app
│ │ ├── leaderelection
│ │ └── options
│ ├── hyperfed
│ ├── kubefedctl
│ └── webhook 启动 webhook
│ └── app
├── pkg
│ ├── apis 定义 CRD 的地方
│ │ ├── core
│ │ │ ├── common
│ │ │ ├── typeconfig
│ │ │ ├── v1alpha1
│ │ │ └── v1beta1
│ │ │ ├── defaults
│ │ │ └── validation
│ │ ├── multiclusterdns
│ │ │ └── v1alpha1
│ │ └── scheduling
│ │ └── v1alpha1
│ ├── client 实现的一个基于 runtime.Object receiver 的 Kubernetes client
│ │ └── generic
│ │ └── scheme
│ ├── controller ControllerManager 的主要组成
│ │ ├── dnsendpoint
│ │ ├── federatedtypeconfig
│ │ ├── ingressdns
│ │ ├── kubefedcluster
│ │ ├── schedulingmanager
│ │ ├── schedulingpreference
│ │ ├── servicedns
│ │ ├── status
│ │ ├── sync 核心同步逻辑
│ │ │ ├── dispatch
│ │ │ ├── status
│ │ │ └── version
│ │ ├── testdata
│ │ │ └── fixtures
│ │ ├── util
│ │ │ ├── finalizers
│ │ │ ├── planner
│ │ │ └── podanalyzer
│ │ └── webhook
│ │ ├── federatedtypeconfig
│ │ ├── kubefedcluster
│ │ └── kubefedconfig
│ ├── features 存放 Feature 的基本信息和开关
│ ├── kubefedctl kubefedctl 的实际实现包
│ │ ├── enable
│ │ ├── federate
│ │ ├── options
│ │ ├── orphaning
│ │ └── util
│ ├── metrics 各项暴露的监测指标
│ └── schedulingtypes 给 SchedulingManager 提供各种 API Type 做 SchedulingPreference 计算的小弟
以上基本可以看到整个代码组成的结构：
- kubefedctl：提供 cli 来控制各种 CRD 创建修改
- webhook：Kubernetes Admission Webhook，当前用了 mutating 和 validating 的 hook，接管的对象有 federatedTypeConfig、kubefedCluster、kuebfedConfig
- controller manager：kubefed 核心所在
咱们就不扯 kuebfedctl 和 webhook 了，直接上 controller 部分，也就是所谓的 kubefed control-panel。不过在这之前，先来看下 kubefed 的数据结构和一些名词定义（比如缩写省得写一大串）
主要数据结构
metav1.APIResource
这货是定义一个 Type 的基本元信息（就不翻译了）
// APIResource specifies the name of a resource and whether it is namespaced.
type APIResource struct {
// name is the plural name of the resource.
Name string `json:"name" protobuf:"bytes,1,opt,name=name"`
// singularName is the singular name of the resource. This allows clients to handle plural and singular opaquely.
// The singularName is more correct for reporting status on a single item and both singular and plural are allowed
// from the kubectl CLI interface.
SingularName string `json:"singularName" protobuf:"bytes,6,opt,name=singularName"`
// namespaced indicates if a resource is namespaced or not.
Namespaced bool `json:"namespaced" protobuf:"varint,2,opt,name=namespaced"`
// group is the preferred group of the resource. Empty implies the group of the containing resource list.
// For subresources, this may have a different value, for example: Scale".
Group string `json:"group,omitempty" protobuf:"bytes,8,opt,name=group"`
// version is the preferred version of the resource. Empty implies the version of the containing resource list
// For subresources, this may have a different value, for example: v1 (while inside a v1beta1 version of the core resource's group)".
Version string `json:"version,omitempty" protobuf:"bytes,9,opt,name=version"`
// kind is the kind for the resource (e.g. 'Foo' is the kind for a resource 'foo')
Kind string `json:"kind" protobuf:"bytes,3,opt,name=kind"`
// verbs is a list of supported kube verbs (this includes get, list, watch, create,
// update, patch, delete, deletecollection, and proxy)
Verbs Verbs `json:"verbs" protobuf:"bytes,4,opt,name=verbs"`
// shortNames is a list of suggested short names of the resource.
ShortNames []string `json:"shortNames,omitempty" protobuf:"bytes,5,rep,name=shortNames"`
// categories is a list of the grouped resources this resource belongs to (e.g. 'all')
Categories []string `json:"categories,omitempty" protobuf:"bytes,7,rep,name=categories"`
// The hash value of the storage version, the version this resource is
// converted to when written to the data store. Value must be treated
// as opaque by clients. Only equality comparison on the value is valid.
// This is an alpha feature and may change or be removed in the future.
// The field is populated by the apiserver only if the
// StorageVersionHash feature gate is enabled.
// This field will remain optional even if it graduates.
// +optional
StorageVersionHash string `json:"storageVersionHash,omitempty" protobuf:"bytes,10,opt,name=storageVersionHash"`
}
然后 kubefed 自己撸了一个 APIResource 两者在关键信息上差别不大
// APIResource defines how to configure the dynamic client for an API resource.
type APIResource struct {
// metav1.GroupVersion is not used since the json annotation of
// the fields enforces them as mandatory.
// Group of the resource.
// +optional
Group string `json:"group,omitempty"`
// Version of the resource.
Version string `json:"version"`
// Camel-cased singular name of the resource (e.g. ConfigMap)
Kind string `json:"kind"`
// Lower-cased plural name of the resource (e.g. configmaps). If
// not provided, it will be computed by lower-casing the kind and
// suffixing an 's'.
PluralName string `json:"pluralName"`
// Scope of the resource.
Scope apiextv1b1.ResourceScope `json:"scope"`
}
FederatedTypeConfig
缩写 FTC，用于生成 FederatedType
type FederatedTypeConfig struct {
metav1.TypeMeta `json:",inline"`
metav1.ObjectMeta `json:"metadata,omitempty"`
Spec FederatedTypeConfigSpec `json:"spec"`
// +optional
Status FederatedTypeConfigStatus `json:"status,omitempty"`
}
// 主要的定义都在这
// FederatedTypeConfigSpec defines the desired state of FederatedTypeConfig.
type FederatedTypeConfigSpec struct {
// The configuration of the target type. If not set, the pluralName and
// groupName fields will be set from the metadata.name of this resource. The
// kind field must be set.
// 指定我们期望 Federated 的目标类型
TargetType APIResource `json:"targetType"`
// Whether or not propagation to member clusters should be enabled.
// 表明是否需要 sync，当前只有 Enable 和 Disable
Propagation PropagationMode `json:"propagation"`
// Configuration for the federated type that defines (via
// template, placement and overrides fields) how the target type
// should appear in multiple cluster.
// 指定我们生成的 APIResource 是什么样的，说白就是生成的 FederatedCRD 中名字、group、version 等信息
FederatedType APIResource `json:"federatedType"`
// Configuration for the status type that holds information about which type
// holds the status of the federated resource. If not provided, the group
// and version will default to those provided for the federated type api
// resource.
// +optional
// target type 的 Status 类型
StatusType *APIResource `json:"statusType,omitempty"`
// Whether or not Status object should be populated.
// +optional
// 是否进行 Target Type Status 做采集
StatusCollection *StatusCollectionMode `json:"statusCollection,omitempty"`
}
KubefedCluster
这个结构体是 member cluster 注册到 Host cluster 时的产物。在 Spec 中主要存储对应集群的 APIServer 和访问凭证， 而在 Status 中主要存储的是集群的健康状态以及如包含的 Region、Zone 等。
type KubeFedCluster struct {
metav1.TypeMeta `json:",inline"`
metav1.ObjectMeta `json:"metadata,omitempty"`
Spec KubeFedClusterSpec `json:"spec"`
// +optional
Status KubeFedClusterStatus `json:"status,omitempty"`
}
// KubeFedClusterSpec defines the desired state of KubeFedCluster
type KubeFedClusterSpec struct {
// The API endpoint of the member cluster. This can be a hostname,
// hostname:port, IP or IP:port.
APIEndpoint string `json:"apiEndpoint"`
// CABundle contains the certificate authority information.
// +optional
CABundle []byte `json:"caBundle,omitempty"`
// Name of the secret containing the token required to access the
// member cluster. The secret needs to exist in the same namespace
// as the control plane and should have a "token" key.
SecretRef LocalSecretReference `json:"secretRef"`
// DisabledTLSValidations defines a list of checks to ignore when validating
// the TLS connection to the member cluster. This can be any of *, SubjectName, or ValidityPeriod.
// If * is specified, it is expected to be the only option in list.
// +optional
DisabledTLSValidations []TLSValidation `json:"disabledTLSValidations,omitempty"`
}
// KubeFedClusterStatus contains information about the current status of a
// cluster updated periodically by cluster controller.
type KubeFedClusterStatus struct {
// Conditions is an array of current cluster conditions.
Conditions []ClusterCondition `json:"conditions"`
// Zones are the names of availability zones in which the nodes of the cluster exist, e.g. 'us-east1-a'.
// +optional
Zones []string `json:"zones,omitempty"`
// Region is the name of the region in which all of the nodes in the cluster exist. e.g. 'us-east1'.
// +optional
Region *string `json:"region,omitempty"`
}
KubefedConfig
这个是 kubefed 的配置，不太像深入，这个大家各自看代码就好，本身涉及不多
FederatedType 是怎么创建的
最开始，我们会手动或者通过 kubefed client 或者什么其他姿势，反正就是创建了一个 FederatedTypeConfig 就像下面
apiVersion: core.kubefed.io/v1beta1
kind: FederatedTypeConfig
metadata:
annotations:
meta.helm.sh/release-name: kubefed
meta.helm.sh/release-namespace: kube-federation-system
finalizers:
- core.kubefed.io/federated-type-config
generation: 1
labels:
app.kubernetes.io/managed-by: Helm
name: deployments.apps
namespace: kube-federation-system
spec:
federatedType:
group: types.kubefed.io
kind: FederatedDeployment
pluralName: federateddeployments
scope: Namespaced
version: v1beta1
propagation: Enabled
targetType:
group: apps
kind: Deployment
pluralName: deployments
scope: Namespaced
version: v1
将 apps/v1.Deployment federated 了，federatedType 表明最后生成的 FCRD 的基本信息，targetType 就是我们想要 federated 的 Type 信息，propagation 这边设为开启。当这样一个 Object 被创建或者更新时，FederatedTypeConfigController 也就对应 repo 中的 pkg/controller/federatedtypeconfig 就会开始工作（如果启动了的话，他由 cmd 里的 controller-manager 拉起，不赘述）。
FederatedTypeConfig 的创建
注意：FTC 的 name 是为 targetType 的 name.group 组成（core group 可省略 group，毕竟其可为「空串」）
启动时在 cmd 中 controller-manager 初始化 FederatedTypeConfigController
// newController returns a new controller to manage FederatedTypeConfig objects.
func newController(config *util.ControllerConfig) (*Controller, error) {
...创建 genericClient...
c := &Controller{
controllerConfig: config,
client: genericclient,
stopChannels: make(map[string]chan struct{}),
}
// 创建了一个延迟队列包裹的 c.reconcile，后续我们通过往 worker 塞东西后入延迟队列再到 c.reconcile 处理
c.worker = util.NewReconcileWorker(c.reconcile, util.WorkerTiming{})
// 根据用户设定的 `KubeFedNamespace` 去起 FTC 的 informer，这边之所以是特定而非全部 namespace
// 主要是因为 kubefed 可以是 namespace level 的 control，如果是直接监听全部的话当是 namespace scope 的时候会有权限问题
// 当 informer 有新东西的时候就会通过 c.worker.EnqueueObject 入到 c.reconcile 去处理。
//
// **其他几个 controller 也是类似的套路，如果没有特别的就不赘述了**
//
// 这边的 store 是 `cache.Store` 也就是原生 cache 存储的地方
// c.controller 实际为 InformerController 可以不用过多关注
c.store, c.controller, err = util.NewGenericInformer(
kubeConfig,
config.KubeFedNamespace,
&corev1b1.FederatedTypeConfig{},
util.NoResyncPeriod,
c.worker.EnqueueObject,
)
if err != nil {
return nil, err
}
return c, nil
}
初始化后通过 Controller.Run(stopCh) 启动这个 Controller。
当我们创建或者修改了一个 FederatedTypeConfig 的时候，informer 会发现并通过 EnqueueObject 塞到 FederatedTypeConfigController.concile 做处理，那就看看这货现在怎么处理 FTC 的吧。
func (c *Controller) reconcile(qualifiedName util.QualifiedName) util.ReconciliationStatus {
// 省略一坨从 cache 拿 FTC 的过程，没啥特别的
syncEnabled := typeConfig.GetPropagationEnabled()
statusEnabled := typeConfig.GetStatusEnabled()
limitedScope := c.controllerConfig.TargetNamespace != metav1.NamespaceAll
// 对于开了 sync 并且当前 kubefed 限制在特定 namespace 的，然后 target type 又是非 namesapced。这种情况就是非预期，需要关闭他的 sync。
if limitedScope && syncEnabled && !typeConfig.GetNamespaced() {
// 如果没取到 stop ch 则构造一个
// 问题点：没有 get 到为什么需要造一个，因为从总体语义来说，如果能 get 到说明对应 sync Contoller是 running 的
_, ok := c.getStopChannel(typeConfig.Name)
if !ok {
// 塞个新的到 StopChannel
}
// 更新 FTC 的 status 并 return 终止流程
}
statusKey := typeConfig.Name + "/status"
syncStopChan, syncRunning := c.getStopChannel(typeConfig.Name)
statusStopChan, statusRunning := c.getStopChannel(statusKey)
deleted := typeConfig.DeletionTimestamp != nil
if deleted {
// 如果 sync controller 和 status controller 开着，就关掉它们
// 如果 federatedNamespace 要被删除的话，按总体的设计来说，namespace 内的 Namespaced FederatedTypes 需要全部重新回炉下
if typeConfig.IsNamespace() {
// 这个函数会将 非 Namespace 对象但是 namespaced 的 FederatedTypes 全部重新入 reconcile 处理下
c.reconcileOnNamespaceFTCUpdate()
}
// 移除 finalizer
err := c.removeFinalizer(typeConfig)
return util.StatusAllOK
}
// 确保 finalizer 都设置到位了
updated, err := c.ensureFinalizer(typeConfig)
if err != nil {
return util.StatusError
} else if updated && typeConfig.IsNamespace() {
// 这边是新建的 Namespace FTC 的时候会进入的流程，这时候会把之前 Namespaced 的 FTC 都过一把，
// 因为早前因为没有 Namespace 的存在，所以他们都不会创建 sync controller，这时候需要全部拉起来跑。
// 问题点：但其实这货根据这个作为条件也很奇怪，因为我如果手动去掉 FTC 的 finalizer 也会触发 updated。。。
c.reconcileOnNamespaceFTCUpdate()
}
startNewSyncController := !syncRunning && syncEnabled
// 如果 sync controller 在运行，但是不需要再开启了或者当前是个 namespaced 的 type 而且对应的 namespace FTC 不存在，也需要把 sync controller 停掉
stopSyncController := syncRunning && (!syncEnabled || (typeConfig.GetNamespaced() && !c.namespaceFTCExists()))
if startNewSyncController {
// 通过 c.startSyncController(typeConfig) 启动 sync controller
} else if stopSyncController {
// 通过 c.stopController(typeConfig.Name, syncStopChan) 关停 sync controller
}
startNewStatusController := !statusRunning && statusEnabled
stopStatusController := statusRunning && !statusEnabled
if startNewStatusController {
// c.startStatusController(statusKey, typeConfig) 启动 status controller
} else if stopStatusController {
c.stopController(statusKey, statusStopChan)
}
// sync controller 已经开着了，但是 FTC status 的 ObservedGeneration 跟 FTC 的 Generation 不一致，
// 那么说明可能 out-of-sync，所以重启下 sync controller
if !startNewSyncController && !stopSyncController &&
typeConfig.Status.ObservedGeneration != typeConfig.Generation {
if err := c.refreshSyncController(typeConfig); err != nil {
runtime.HandleError(err)
return util.StatusError
}
}
// 更新 SyncController 和 StatusController 到 FTC
return util.StatusAllOK
}
KubeFedSyncController
在上文中，当有一个 FTC 存在且需要 sync 的时候，就会通过 startSyncController(tc *corev1b1.FederatedTypeConfig) 去启动一个 SyncController，接下来看看 SyncController 的启动和工作内容。
func (c *Controller) startSyncController(tc *corev1b1.FederatedTypeConfig) error {
ftc := tc.DeepCopyObject().(*corev1b1.FederatedTypeConfig)
kind := ftc.Spec.FederatedType.Kind
// 如果是一个 namespaced target type，那么我们需要在启动前确保 FederatedNamespace 已经开启了，不然无法进行
var fedNamespaceAPIResource *metav1.APIResource
if ftc.GetNamespaced() {
var err error
fedNamespaceAPIResource, err = c.getFederatedNamespaceAPIResource()
if err != nil {
return errors.Wrapf(err, "Unable to start sync controller for %q due to missing FederatedTypeConfig for namespaces", kind)
}
}
stopChan := make(chan struct{})
// 实际启动一个 SyncController
err := synccontroller.StartKubeFedSyncController(c.controllerConfig, stopChan, ftc, fedNamespaceAPIResource)
if err != nil {
close(stopChan)
return errors.Wrapf(err, "Error starting sync controller for %q", kind)
}
klog.Infof("Started sync controller for %q", kind)
// 塞到 stopChannels 标明这个 FTC 已经有了 running 的 SyncController
c.lock.Lock()
defer c.lock.Unlock()
c.stopChannels[ftc.Name] = stopChan
return nil
}
// pkg/controller/sync/controller.go
以下
func StartKubeFedSyncController(controllerConfig *util.ControllerConfig, stopChan <-chan struct{}, typeConfig typeconfig.Interface, fedNamespaceAPIResource *metav1.APIResource) error {
controller, err := newKubeFedSyncController(controllerConfig, typeConfig, fedNamespaceAPIResource)
if err != nil {
return err
}
// 问题点：这是一个非常诡异的东西，从注释看仅仅为了测试。。。
if controllerConfig.MinimizeLatency {
controller.minimizeLatency()
}
klog.Infof(fmt.Sprintf("Starting sync controller for %q", typeConfig.GetFederatedType().Kind))
controller.Run(stopChan)
return nil
}
// newKubeFedSyncController returns a new sync controller for the configuration
func newKubeFedSyncController(controllerConfig *util.ControllerConfig, typeConfig typeconfig.Interface, fedNamespaceAPIResource *metav1.APIResource) (*KubeFedSyncController, error) {
// 初始化 generic client 和一个 event recorder
...
s := &KubeFedSyncController{
clusterAvailableDelay: controllerConfig.ClusterAvailableDelay,
clusterUnavailableDelay: controllerConfig.ClusterUnavailableDelay,
smallDelay: time.Second * 3,
eventRecorder: recorder,
typeConfig: typeConfig,
hostClusterClient: client,
skipAdoptingResources: controllerConfig.SkipAdoptingResources,
limitedScope: controllerConfig.LimitedScope(),
}
s.worker = util.NewReconcileWorker(s.reconcile, util.WorkerTiming{
ClusterSyncDelay: s.clusterAvailableDelay,
})
// Build deliverer for triggering cluster reconciliations.
s.clusterDeliverer = util.NewDelayingDeliverer()
targetAPIResource := typeConfig.GetTargetType()
// 起一个 informer watch 在 HostCluster 的 TargetType 上，适时的如 worker 队列。
// 重点：并处理集群的健康状态变更，当 member cluster 有状态变更的时候，直接重新过一遍全部的 Target Type Object
var err error
s.informer, err = util.NewFederatedInformer(
controllerConfig,
client,
&targetAPIResource,
func(obj pkgruntime.Object) {
qualifiedName := util.NewQualifiedName(obj)
s.worker.EnqueueForRetry(qualifiedName)
},
&util.ClusterLifecycleHandlerFuncs{
ClusterAvailable: func(cluster *fedv1b1.KubeFedCluster) {
// When new cluster becomes available process all the target resources again.
s.clusterDeliverer.DeliverAt(allClustersKey, nil, time.Now().Add(s.clusterAvailableDelay))
},
// When a cluster becomes unavailable process all the target resources again.
ClusterUnavailable: func(cluster *fedv1b1.KubeFedCluster, _ []interface{}) {
s.clusterDeliverer.DeliverAt(allClustersKey, nil, time.Now().Add(s.clusterUnavailableDelay))
},
},
)
if err != nil {
return nil, err
}
// 重点
s.fedAccessor, err = NewFederatedResourceAccessor(
controllerConfig, typeConfig, fedNamespaceAPIResource,
client, s.worker.EnqueueObject, recorder)
if err != nil {
return nil, err
}
return s, nil
}
在 FTCController 部分的 SyncController 启动逻辑不会太难理解，总体可以理解为：如果是 Namespaced Type 那么需要保证 FederatedNamespace 存在，不然就不进行创建。如果只是普通的 Type 则直接开启，最后在 channel 注册下。
FederatedResourceAccessor
// pkg/controller/sync/accessor.go
type FederatedResourceAccessor interface {
Run(stopChan <-chan struct{})
HasSynced() bool
FederatedResource(qualifiedName util.QualifiedName) (federatedResource FederatedResource, possibleOrphan bool, err error)
VisitFederatedResources(visitFunc func(obj interface{}))
}
func NewFederatedResourceAccessor(
controllerConfig *util.ControllerConfig,
typeConfig typeconfig.Interface,
fedNamespaceAPIResource *metav1.APIResource,
client genericclient.Client,
enqueueObj func(pkgruntime.Object),
eventRecorder record.EventRecorder) (FederatedResourceAccessor, error) {
// 这个函数的基本流程会是
// 获取 FederatedType 的 client 和 informer
// 如果 TargetType 是个 Namespace 就再初始化下 对应 namespace 的 client 和 informer
// 接着，如果 TargetType 是个 Namespaced Type 那么会再初始化一下 KubeFedNamespace 的 client 和 informer 在 FederatedNamespace 对象变动的时候需要整个 namespace 内的 FederatedResource 全部重新进队列处理
// 再起一个 Version Manager
}
Version Manager 这边暂时不细展开代码，主要的工作是
- 当创建 FederatedObject 的时候，会去创建一个
PropagatedVersion其中记录了对应 resource（target resource） 和 template、override 的 version - 基于此 version 我们可以判定一个完整的 FederatedObject 是否需要更新
回到 Accessor，在初始化完后会将 versionManager 及相关的 informer 启动起来。以下是几个 Accessor 的方法
以下判断一个 FederatedType 是否已同步
func (a *resourceAccessor) HasSynced() bool {
kind := a.typeConfig.GetFederatedType().Kind
if !a.versionManager.HasSynced() {
return false
}
if !a.federatedController.HasSynced() {
return false
}
if a.namespaceController != nil && !a.namespaceController.HasSynced() {
return false
}
if a.fedNamespaceController != nil && !a.fedNamespaceController.HasSynced() {
return false
}
return true
}
// 从 eventSource 中获取对应的 FederatedSource 也就是 FederatedType 的实际对象
func (a *resourceAccessor) FederatedResource(eventSource util.QualifiedName) (FederatedResource, bool, error) {
// 问题点：忽略了系统的 namespace，暂时没 get 到为啥
if a.targetIsNamespace && a.isSystemNamespace(eventSource.Name) {
klog.V(7).Infof("Ignoring system namespace %q", eventSource.Name)
return nil, false, nil
}
// Most federated resources have the same name as their targets.
// NamespaceForResource 当前实际是直接 return eventSource.Namespace 所以两个 name 是一样的
targetName := util.QualifiedName{
Namespace: eventSource.Namespace,
Name: eventSource.Name,
}
federatedName := util.QualifiedName{
Namespace: util.NamespaceForResource(eventSource.Namespace, a.fedNamespace),
Name: eventSource.Name,
}
// 这边对 namespace 类型的 FederatedType 做了一个特殊处理。
// A federated type for namespace "foo" is namespaced
// (e.g. "foo/foo"). An event sourced from a namespace in the host
// or member clusters will have the name "foo", and an event
// sourced from a federated resource will have the name "foo/foo".
// In order to ensure object retrieval from the informers, it is
// necessary to derive the target name and federated name from the
// event source.
if a.targetIsNamespace {
// eventSource 是否来自 core.Namespace（就是没有 federated 的那个 object），因为那么 namespace 是非 namespaced，所以这边通过判断 namespace 为空确认
eventSourceIsTarget := eventSource.Namespace == ""
if eventSourceIsTarget {
// Ensure the federated name is namespace qualified.
// 因为当前 eventSource 是 core.Namespace 所以其 federated 的对象所在 namespace 就是其下属
federatedName.Namespace = federatedName.Name
} else {
// Ensure the target name is not namespace qualified.
// 当前 eventSource 是非 core.Namespace 也就是 FederatedNamespace。他的 target 是 core.Namespace 非 namespaced 所以需要置空
targetName.Namespace = ""
}
// 这边为啥不直接就覆盖掉的了？为了凑代码吗？
}
key := federatedName.String()
resource, err := util.ObjFromCache(a.federatedStore, kind, key)
if err != nil {
return nil, false, err
}
// 如果从 federatedType Cache 里取不到并且不是 FederatedNamespace 那么则有可能是 orphan
// 疑问：为啥是 FNS 就不会是 orphan？？？
if resource == nil {
sourceIsFederatedNamespace := a.targetIsNamespace && eventSource.Namespace != ""
possibleOrphan := !sourceIsFederatedNamespace
return nil, possibleOrphan, nil
}
var namespace *unstructured.Unstructured
if a.targetIsNamespace {
// 这边会确认下 host cluster 中对应需要 propagate 的 namespace 存在
}
var fedNamespace *unstructured.Unstructured
if a.typeConfig.GetNamespaced() {
// 对于 namespaced 的需要确认下 federated namespace 是否开启，不开也没法进行
fedNamespaceName := util.QualifiedName{Namespace: federatedName.Namespace, Name: federatedName.Namespace}
fedNamespace, err = util.ObjFromCache(a.fedNamespaceStore, a.fedNamespaceAPIResource.Kind, fedNamespaceName.String())
if err != nil {
return nil, false, err
}
// If fedNamespace is nil, the resources in member clusters
// will be removed.
}
return &federatedResource{...}, false, nil
}
KubeFedSyncController.reconcile
终于回来了，继续理 sync controller 的 reconcile
func (s *KubeFedSyncController) reconcile(qualifiedName util.QualifiedName) util.ReconciliationStatus {
if !s.isSynced() {
return util.StatusNotSynced
}
kind := s.typeConfig.GetFederatedType().Kind
fedResource, possibleOrphan, err := s.fedAccessor.FederatedResource(qualifiedName)
if err != nil {
runtime.HandleError(errors.Wrapf(err, "Error creating FederatedResource helper for %s %q", kind, qualifiedName))
return util.StatusError
}
// 可能是 orphan ，在所有集群理把 kubefed.io/managed=true 的标签删除
if possibleOrphan {
apiResource := s.typeConfig.GetTargetType()
gvk := apiResourceToGVK(&apiResource)
klog.V(2).Infof("Ensuring the removal of the label %q from %s %q in member clusters.", util.ManagedByKubeFedLabelKey, gvk.Kind, qualifiedName)
err = s.removeManagedLabel(gvk, qualifiedName)
if err != nil {
wrappedErr := errors.Wrapf(err, "failed to remove the label %q from %s %q in member clusters", util.ManagedByKubeFedLabelKey, gvk.Kind, qualifiedName)
runtime.HandleError(wrappedErr)
return util.StatusError
}
return util.StatusAllOK
}
if fedResource == nil {
return util.StatusAllOK
}
key := fedResource.FederatedName().String()
klog.V(4).Infof("Starting to reconcile %s %q", kind, key)
startTime := time.Now()
defer func() {
klog.V(4).Infof("Finished reconciling %s %q (duration: %v)", kind, key, time.Since(startTime))
metrics.ReconcileFederatedResourcesDurationFromStart(startTime)
}()
if fedResource.Object().GetDeletionTimestamp() != nil {
return s.ensureDeletion(fedResource)
}
// 确保下对应的 FederatedObject 有 finalizer
err = s.ensureFinalizer(fedResource)
if err != nil {
fedResource.RecordError("EnsureFinalizerError", errors.Wrap(err, "Failed to ensure finalizer"))
return util.StatusError
}
// 推到各个集群
return s.syncToClusters(fedResource)
}
删除部分需要特别说明下，因为这边涉及到了 kubefed.io/orphan 标签
func (s *KubeFedSyncController) ensureDeletion(fedResource FederatedResource) util.ReconciliationStatus {
fedResource.DeleteVersions()
key := fedResource.FederatedName().String()
kind := fedResource.FederatedKind()
klog.V(2).Infof("Ensuring deletion of %s %q", kind, key)
obj := fedResource.Object()
// 没有 finalizer 就不管了，丢那了
finalizers := sets.NewString(obj.GetFinalizers()...)
if !finalizers.Has(FinalizerSyncController) {
klog.V(2).Infof("%s %q does not have the %q finalizer. Nothing to do.", kind, key, FinalizerSyncController)
return util.StatusAllOK
}
// 如果设定了 kubefed.io/orphan=true 的 annotation 就不会做实际的删除，但是会去掉 finalizer 和 managed 的标签
if util.IsOrphaningEnabled(obj) {
klog.V(2).Infof("Found %q annotation on %s %q. Removing the finalizer.",
util.OrphanManagedResourcesAnnotation, kind, key)
err := s.removeFinalizer(fedResource)
if err != nil {
wrappedErr := errors.Wrapf(err, "failed to remove finalizer %q from %s %q", FinalizerSyncController, kind, key)
runtime.HandleError(wrappedErr)
return util.StatusError
}
klog.V(2).Infof("Initiating the removal of the label %q from resources previously managed by %s %q.", util.ManagedByKubeFedLabelKey, kind, key)
err = s.removeManagedLabel(fedResource.TargetGVK(), fedResource.TargetName())
if err != nil {
wrappedErr := errors.Wrapf(err, "failed to remove the label %q from all resources previously managed by %s %q", util.ManagedByKubeFedLabelKey, kind, key)
runtime.HandleError(wrappedErr)
return util.StatusError
}
return util.StatusAllOK
}
// 真的需要删除的。
// 这边之所以会有 recheck 是因为有些资源如果已经在 deletion 了（DeletionTimestamp != nil） 那就不会再删除。
// deleteFromClusters 里面还有个特殊处理是，如果是删 namespaced 不会需要 recheck，也就是只有删 federatedNamespace 需要
klog.V(2).Infof("Deleting resources managed by %s %q from member clusters.", kind, key)
recheckRequired, err := s.deleteFromClusters(fedResource)
if err != nil {
wrappedErr := errors.Wrapf(err, "failed to delete %s %q", kind, key)
runtime.HandleError(wrappedErr)
return util.StatusError
}
if recheckRequired {
return util.StatusNeedsRecheck
}
return util.StatusAllOK
}
总算要推到各个集群了，这部分我觉得是写的最干脆的。
// syncToClusters ensures that the state of the given object is
// synchronized to member clusters.
func (s *KubeFedSyncController) syncToClusters(fedResource FederatedResource) util.ReconciliationStatus {
clusters, err := s.informer.GetClusters()
if err != nil {
fedResource.RecordError(string(status.ClusterRetrievalFailed), errors.Wrap(err, "Failed to retrieve list of clusters"))
return s.setFederatedStatus(fedResource, status.ClusterRetrievalFailed, nil)
}
selectedClusterNames, err := fedResource.ComputePlacement(clusters)
if err != nil {
fedResource.RecordError(string(status.ComputePlacementFailed), errors.Wrap(err, "Failed to compute placement"))
return s.setFederatedStatus(fedResource, status.ComputePlacementFailed, nil)
}
// 上述两个步骤取了所有集群和相关 FederatedObject 中 placement 涉及的集群
kind := fedResource.TargetKind()
key := fedResource.TargetName().String()
klog.V(4).Infof("Ensuring %s %q in clusters: %s", kind, key, strings.Join(selectedClusterNames.List(), ","))
dispatcher := dispatch.NewManagedDispatcher(s.informer.GetClientForCluster, fedResource, s.skipAdoptingResources)
// Range 所有集群，因为我们需要做回收，所以不能单单 range selectedCluster
for _, cluster := range clusters {
clusterName := cluster.Name
selectedCluster := selectedClusterNames.Has(clusterName)
// 糟糕，选到的集群 NotReady 了，记个错误
if !util.IsClusterReady(&cluster.Status) {
if selectedCluster {
// Cluster state only needs to be reported in resource
// status for clusters selected for placement.
err := errors.New("Cluster not ready")
dispatcher.RecordClusterError(status.ClusterNotReady, clusterName, err)
}
continue
}
rawClusterObj, _, err := s.informer.GetTargetStore().GetByKey(clusterName, key)
if err != nil {
wrappedErr := errors.Wrap(err, "Failed to retrieve cached cluster object")
dispatcher.RecordClusterError(status.CachedRetrievalFailed, clusterName, wrappedErr)
continue
}
var clusterObj *unstructured.Unstructured
if rawClusterObj != nil {
clusterObj = rawClusterObj.(*unstructured.Unstructured)
}
// 如果没选上的集群里有，Target Object 那就干掉
// 潜在风险：多集群加入 kubefed 的时候如果正好命中，那就 GG 了
// Resource should not exist in the named cluster
if !selectedCluster {
if clusterObj == nil {
// Resource does not exist in the cluster
continue
}
if clusterObj.GetDeletionTimestamp() != nil {
// Resource is marked for deletion
dispatcher.RecordStatus(clusterName, status.WaitingForRemoval)
continue
}
if fedResource.IsNamespaceInHostCluster(clusterObj) {
// Host cluster namespace needs to have the managed
// label removed so it won't be cached anymore.
dispatcher.RemoveManagedLabel(clusterName, clusterObj)
} else {
dispatcher.Delete(clusterName)
}
continue
}
// Resource should appear in the named cluster
// 创建或者更新对象
// TODO(marun) Consider waiting until the result of resource
// creation has reached the target store before attempting
// subsequent operations. Otherwise the object won't be found
// but an add operation will fail with AlreadyExists.
if clusterObj == nil {
dispatcher.Create(clusterName)
} else {
dispatcher.Update(clusterName, clusterObj)
}
}
// 等 dispatcher 刚刚的工作都完成了，然后更新下 version map
_, timeoutErr := dispatcher.Wait()
if timeoutErr != nil {
fedResource.RecordError("OperationTimeoutError", timeoutErr)
}
// Write updated versions to the API.
updatedVersionMap := dispatcher.VersionMap()
err = fedResource.UpdateVersions(selectedClusterNames.List(), updatedVersionMap)
if err != nil {
// Versioning of federated resources is an optimization to
// avoid unnecessary updates, and failure to record version
// information does not indicate a failure of propagation.
runtime.HandleError(err)
}
// 更新下传播状态 -> PropagationStatus
collectedStatus := dispatcher.CollectedStatus()
return s.setFederatedStatus(fedResource, status.AggregateSuccess, &collectedStatus)
}
从上述这段代码来说，这边没有处理集群挂掉时候的 failover（比如对于 Deployment、ReplicaSet 这类 ReplicaedObject），这部分会留到 SchedulingPreference 那边处理。
Dispatcher
// ManagedDispatcher dispatches operations to member clusters for resources
// managed by a federated resource.
type ManagedDispatcher interface {
UnmanagedDispatcher
Create(clusterName string)
Update(clusterName string, clusterObj *unstructured.Unstructured)
VersionMap() map[string]string
CollectedStatus() status.CollectedPropagationStatus
RecordClusterError(propStatus status.PropagationStatus, clusterName string, err error)
RecordStatus(clusterName string, propStatus status.PropagationStatus)
}
// UnmanagedDispatcher dispatches operations to member clusters for
// resources that are no longer managed by a federated resource.
type UnmanagedDispatcher interface {
OperationDispatcher
Delete(clusterName string)
RemoveManagedLabel(clusterName string, clusterObj *unstructured.Unstructured)
}
// OperationDispatcher provides an interface to wait for operations
// dispatched to member clusters.
type OperationDispatcher interface {
// Wait returns true for ok if all operations completed
// successfully and false if only some operations completed
// successfully. An error is returned on timeout.
Wait() (ok bool, timeoutErr error)
}
来看看 Dispatcher 都定义了什么 func
- 增、删、改
- 采集状态、版本
- 等待 dispatch 完成
- 移除 ManagedLabel
咱们主要关注点在增删和 wait ，其他就不看了
Dispatcher.Create
func (d *managedDispatcherImpl) Create(clusterName string) {
// Default the status to an operation-specific timeout. Otherwise
// when a timeout occurs it won't be possible to determine which
// operation timed out. The timeout status will be cleared by
// Wait() if a timeout does not occur.
// 开篇先来个骚动作。
// 以后咱们对于超时的这种处理状态反馈，可以先默认超时，后面成了再改回去。万一漏改了没事，下次再改
d.RecordStatus(clusterName, status.CreationTimedOut)
start := time.Now()
d.dispatcher.incrementOperationsInitiated()
const op = "create"
go d.dispatcher.clusterOperation(clusterName, op, func(client generic.Client) util.ReconciliationStatus {
d.recordEvent(clusterName, op, "Creating")
obj, err := d.fedResource.ObjectForCluster(clusterName)
if err != nil {
return d.recordOperationError(status.ComputeResourceFailed, clusterName, op, err)
}
err = d.fedResource.ApplyOverrides(obj, clusterName)
if err != nil {
return d.recordOperationError(status.ApplyOverridesFailed, clusterName, op, err)
}
// 以上两个步骤：根据 FederatedObject 获取 Spec 然后 apply 下对应集群的 overrides。然后创建一波我们要的 Object
err = client.Create(context.Background(), obj)
if err == nil {
version := util.ObjectVersion(obj)
d.recordVersion(clusterName, version)
metrics.DispatchOperationDurationFromStart("create", start)
return util.StatusAllOK
}
// TODO(marun) Figure out why attempting to create a namespace that
// already exists indicates ServerTimeout instead of AlreadyExists.
// 因为 controller-runtime 的 client 实现关系，所以这边的 namespace 重复创建会反馈为 timeout
alreadyExists := apierrors.IsAlreadyExists(err) || d.fedResource.TargetKind() == util.NamespaceKind && apierrors.IsServerTimeout(err)
if !alreadyExists {
return d.recordOperationError(status.CreationFailed, clusterName, op, err)
}
// Attempt to update the existing resource to ensure that it
// is labeled as a managed resource.
err = client.Get(context.Background(), obj, obj.GetNamespace(), obj.GetName())
if err != nil {
wrappedErr := errors.Wrapf(err, "failed to retrieve object potentially requiring adoption")
return d.recordOperationError(status.RetrievalFailed, clusterName, op, wrappedErr)
}
// 完全不知道 adoptingResource 啥意思啊。。。盲猜：已存在的资源。
// 跟其他同学聊了下，这货比较准确应该翻译为：从别处接管来的资源。
if d.skipAdoptingResources && !d.fedResource.IsNamespaceInHostCluster(obj) {
_ = d.recordOperationError(status.AlreadyExists, clusterName, op, errors.Errorf("Resource pre-exist in cluster"))
return util.StatusAllOK
}
d.recordError(clusterName, op, errors.Errorf("An update will be attempted instead of a creation due to an existing resource"))
d.Update(clusterName, obj)
metrics.DispatchOperationDurationFromStart("update", start)
return util.StatusAllOK
})
}
Dispatcher.Update
func (d *managedDispatcherImpl) Update(clusterName string, clusterObj *unstructured.Unstructured) {
d.RecordStatus(clusterName, status.UpdateTimedOut)
d.dispatcher.incrementOperationsInitiated()
const op = "update"
go d.dispatcher.clusterOperation(clusterName, op, func(client generic.Client) util.ReconciliationStatus {
if util.IsExplicitlyUnmanaged(clusterObj) {
err := errors.Errorf("Unable to manage the object which has label %s: %s", util.ManagedByKubeFedLabelKey, util.UnmanagedByKubeFedLabelValue)
return d.recordOperationError(status.ManagedLabelFalse, clusterName, op, err)
}
obj, err := d.fedResource.ObjectForCluster(clusterName)
if err != nil {
return d.recordOperationError(status.ComputeResourceFailed, clusterName, op, err)
}
// 保留了原有的 Version、finalizer、annotation 和实例数，以及部分保留 service 和 service account
err = RetainClusterFields(d.fedResource.TargetKind(), obj, clusterObj, d.fedResource.Object())
if err != nil {
wrappedErr := errors.Wrapf(err, "failed to retain fields")
return d.recordOperationError(status.FieldRetentionFailed, clusterName, op, wrappedErr)
}
err = d.fedResource.ApplyOverrides(obj, clusterName)
if err != nil {
return d.recordOperationError(status.ApplyOverridesFailed, clusterName, op, err)
}
// 比对版本确认是否需要更新，避免盲目更新
version, err := d.fedResource.VersionForCluster(clusterName)
if err != nil {
return d.recordOperationError(status.VersionRetrievalFailed, clusterName, op, err)
}
if !util.ObjectNeedsUpdate(obj, clusterObj, version) {
// Resource is current
return util.StatusAllOK
}
// Only record an event if the resource is not current
d.recordEvent(clusterName, op, "Updating")
err = client.Update(context.Background(), obj)
if err != nil {
return d.recordOperationError(status.UpdateFailed, clusterName, op, err)
}
d.setResourcesUpdated()
version = util.ObjectVersion(obj)
d.recordVersion(clusterName, version)
return util.StatusAllOK
})
}
Dispatcher.Wait
func (d *managedDispatcherImpl) Wait() (bool, error) {
...
Wait 主要是干两件事情，等待 update、delete 等等操作结束，然后更新下 status
}
这部分没啥坑，不赘述
小结
以上基本是可以 cover 整个 kubefed 的在多集群传播工作（propagation）原理了，然后还剩下调度部分、潜在坑、场景和一些感受留到下篇写。
整体来说除了一些蜜汁代码倒还好，有些 namespace、cluster level 的下层 object reconcile 比较蛋疼不过也确实没有特别好的办法