author：sufei

预备知识

StatefulSet

 在Kubernetes系统中，Pod的管理对象RC、Deployment、DaemonSet和Job都面向无状态的服务。但现实中有很多服务是有状态的，例如MySQL集群、MongoDB集群、Akka集群、ZooKeeper集群等，这些应用集群有4个共同点。

• 每个节点都有固定的身份ID，通过这个ID，集群中的成员可以相互发现并通信。
• 集群中的每个节点都是有状态的，通常会持久化数据到永久存储中。
• 如果磁盘损坏，则集群里的某个节点无法正常运行，集群功能受损。

 如果通过RC或Deployment控制Pod副本数量来实现上述有状态的集群，就会发现：

第1点是无法满足的，因为Pod的名称是随机产生的，Pod的IP地址也是在运行期才确定且可能有变动的，我们事先无法为每个Pod都确定唯一不变的ID。

另外，为了能够在其他节点上恢复某个失败的节点，这种集群中的Pod需要挂接某种共享存储.

 为了解决这个问题Kubernetes引入了StatefulSet，StatefulSet从本质上来说，可以看作Deployment/RC的一个特殊变种，它有如下特性：

• StatefulSet里的每个Pod都有稳定、唯一的网络标识，可以用来发现集群内的其他成员。假设StatefulSet的名称为mysql，那么第1个Pod叫mysql-0，第2个叫mysql-1，以此类推；
• StatefulSet控制的Pod副本的启停顺序是受控的，操作第n个Pod时，前n-1个Pod已经是运行且准备好的状态；
• StatefulSet里的Pod采用稳定的持久化存储卷，通过PV或PVC来实现，删除Pod时默认不会删除与StatefulSet相关的存储卷（为了保证数据的安全）。

 每个StatefulSet还要与Headless Service配合使用。下面说明什么是Headless Service。

Headless Service

<font color=green>什么是k8s Service？</font>

 Kubernetes里的每个Service其实就是我们经常提起的微服务架构中的一个微服务，Pod、RC与Service的逻辑关系如下图：

k8s service

 从上图1.12中可以看到，Kubernetes的Service定义了一个服务的访问入口地址，前端的应用（Pod）通过这个入口地址访问其背后的一组由Pod副本组成的集群实例，Service与其后端Pod副本集群之间则是通过Label Selector来实现无缝对接的，Service一旦被创建，Kubernetes就会自动为它分配一个可用的Cluster IP，而且在Service的整个生命周期内，它的Cluster IP不会发生改变，前台通过访问这个Cluster IP，随机转发到服务后端pod，从而实现负载均衡。RC的作用实际上是保证Service的服务能力和服务质量始终符合预期标准（确保后续pod数）。

<font color=green>什么是Headless Service？</font>

 Headless Service与普通Service的关键区别在于，它没有Cluster IP，如果解析Headless Service的DNS域名，则返回的是该Service对应的全部Pod的Endpoint列表。StatefulSet在Headless Service的基础上又为StatefulSet控制的每个Pod实例都创建了一个DNS域名，这个域名的格式为：

pod_name.service_name

 比如一个3节点的Kafka的StatefulSet集群对应的Headless Service的名称为mysql，StatefulSet的名称为mysql，则StatefulSet里的3个Pod的DNS名称分别为mysql-0.mysql、mysql-1.mysql、mysql-2.mysql，这些DNS名称可以直接在集群的配置文件中固定下来。

一、operator原理

 Operator 是 CoreOS 推出的旨在简化复杂有状态应用管理的框架，它是一个感知应用状态的控制器，通过扩展 Kubernetes API 来自动创建、管理和配置应用实例。

可能上面的说明不太明朗，个人理解就是：

k8s 提过了扩张API接口，用户可以通过该接口定义自己的资源类型（当然这个类型有k8s内部固有资源类型组合而成），然后通过自己注册的资源控制器来管理该资源，具体的管理方式为：观察集群资源状态 -> 对比集群状态与期望状态差别 -> 调用Kubernetes API 消除这些差别。控制器通过上面的循环来保证集群期望状态。

 下面是CoreOS官方给出了有关etcd operator相关说明

etcd

etcd operator 通过下面的三个步骤模拟了管理 etcd 集群的行为：

1. 通过 Kubernetes API 观察集群的当前状态；
2. 分析当前状态与期望状态的差别；
3. 调用 etcd 集群管理 API 或 Kubernetes API 消除这些差别。

 Operator 是一个感知应用状态的控制器，所以实现一个 Operator 最关键的就是把管理应用状态的所有操作封装到配置资源和控制器中。通常来说 Operator 需要包括以下功能：

• Operator 自身以 deployment 的方式部署
• Operator 自动创建一个 Third Party Resources 资源类型，用户可以用该类型创建应用实例
• Operator 应该利用 Kubernetes 内置的 Serivce/ReplicaSet 等管理应用
• Operator 应该向后兼容，并且在 Operator 自身退出或删除时不影响应用的状态
• Operator 应该支持应用版本更新
• Operator 应该测试 Pod 失效、配置错误、网络错误等异常情况

二、mysql-operator源码分析

2.1 创建自定义资源控制器

在mysql-operator启动时，首先会向k8s创建资源类型控制器

    clusterController := cluster.NewController(
        *s,
        mysqlopClient,
        kubeClient,
        operatorInformerFactory.MySQL().V1alpha1().Clusters(), //自定义资源类型
        kubeInformerFactory.Apps().V1beta1().StatefulSets(),   //statefulsets
        kubeInformerFactory.Core().V1().Pods(),                //pods
        kubeInformerFactory.Core().V1().Services(),            //services
        30*time.Second,
        s.Namespace,
    )

从上面可以看出，mysql-operator自定义资源类型由statefulsets，pods和Headless services三个基本资源类型组成，所以在后续监听中就是监听上面几种资源的状态变化事件。具体代码与相关处理函数在NewController函数内，如下：

func NewController(
    opConfig operatoropts.MySQLOperatorOpts,
    opClient clientset.Interface,
    kubeClient kubernetes.Interface,
    clusterInformer informersv1alpha1.ClusterInformer,
    statefulSetInformer appsinformers.StatefulSetInformer,
    podInformer coreinformers.PodInformer,
    serviceInformer coreinformers.ServiceInformer,
    resyncPeriod time.Duration,
    namespace string,
) *MySQLController {
    opscheme.AddToScheme(scheme.Scheme) // TODO: This shouldn't be done here I don't think.

    // Create event broadcaster.
    glog.V(4).Info("Creating event broadcaster")
    eventBroadcaster := record.NewBroadcaster()
    eventBroadcaster.StartLogging(glog.Infof)
    eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")})
    recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})

    m := MySQLController{
        opConfig: opConfig,

        opClient:   opClient,
        kubeClient: kubeClient,
        // 设置关注各种自定义类型的资源
        clusterLister:       clusterInformer.Lister(),
        clusterListerSynced: clusterInformer.Informer().HasSynced,
        clusterUpdater:      newClusterUpdater(opClient, clusterInformer.Lister()),

        serviceLister:       serviceInformer.Lister(),
        serviceListerSynced: serviceInformer.Informer().HasSynced,
        serviceControl:      NewRealServiceControl(kubeClient, serviceInformer.Lister()),

        statefulSetLister:       statefulSetInformer.Lister(),
        statefulSetListerSynced: statefulSetInformer.Informer().HasSynced,
        statefulSetControl:      NewRealStatefulSetControl(kubeClient, statefulSetInformer.Lister()),

        podLister:       podInformer.Lister(),
        podListerSynced: podInformer.Informer().HasSynced,
        podControl:      NewRealPodControl(kubeClient, podInformer.Lister()),

        secretControl: NewRealSecretControl(kubeClient),

        queue:    workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "mysqlcluster"),
        recorder: recorder,
    }
    /*
    监控自定义类型mysqlcluster的变化(增加、更新、删除)，
    这里看一看m.enqueueCluster函数可以发现都只是把发生变化的自定义对象的名称放入工作队列中
    */
    clusterInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: m.enqueueCluster,
        UpdateFunc: func(old, new interface{}) {
            m.enqueueCluster(new)
        },
        DeleteFunc: func(obj interface{}) {
            cluster, ok := obj.(*v1alpha1.Cluster)
            if ok {
                m.onClusterDeleted(cluster.Name)
            }
        },
    })
    //同样监控statefulset变化，handleObject首先查看书否属于哪个自定义类型cluster，然后加入到相应集群
    statefulSetInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: m.handleObject,
        UpdateFunc: func(old, new interface{}) {
            newStatefulSet := new.(*apps.StatefulSet)
            oldStatefulSet := old.(*apps.StatefulSet)
            if newStatefulSet.ResourceVersion == oldStatefulSet.ResourceVersion {
                return
            }

            // If cluster is ready ...
            if newStatefulSet.Status.ReadyReplicas == newStatefulSet.Status.Replicas {
                clusterName, ok := newStatefulSet.Labels[constants.ClusterLabel]
                if ok {
                    m.onClusterReady(clusterName)
                }
            }
            m.handleObject(new)
        },
        DeleteFunc: m.handleObject,
    })

    return &m
}

设置好了资源监听以及相关处理函数（将事件放入处理队列中），下面就是进入控制权处理逻辑

// main函数中开启一个协程来运行控制器核心逻辑
    go func() {
        defer wg.Done()
        clusterController.Run(ctx, 5)
    }()

//Run函数里会启动工作协程处理上述放入工作队列的自定义对象
func (m *MySQLController) Run(ctx context.Context, threadiness int) {
    ……
    for i := 0; i < threadiness; i++ {
        go wait.Until(m.runWorker, time.Second, ctx.Done())
    }
    ……
}

2.2 controller核心逻辑

// syncHandler compares the actual state with the desired, and attempts to
// converge the two. It then updates the Status block of the Cluster
// resource with the current status of the resource.
func (m *MySQLController) syncHandler(key string) error {
    // Convert the namespace/name string into a distinct namespace and name.
    // 获取资源的命名空间与资源名
    namespace, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
        utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
        return nil
    }

    nsName := types.NamespacedName{Namespace: namespace, Name: name}

    // Get the Cluster resource with this namespace/name.
    cluster, err := m.clusterLister.Clusters(namespace).Get(name)
    if err != nil {
        // The Cluster resource may no longer exist, in which case we stop processing.
        if apierrors.IsNotFound(err) {
            utilruntime.HandleError(fmt.Errorf("mysqlcluster '%s' in work queue no longer exists", key))
            return nil
        }
        return err
    }

    cluster.EnsureDefaults()
    // 校验自定义资源对象
    if err = cluster.Validate(); err != nil {
        return errors.Wrap(err, "validating Cluster")
    }
    
    /*
    给自定义资源对象设置一些默认属性
    */
    if cluster.Spec.Repository == "" {
        cluster.Spec.Repository = m.opConfig.Images.DefaultMySQLServerImage
    }

    operatorVersion := buildversion.GetBuildVersion()
    // Ensure that the required labels are set on the cluster.
    sel := combineSelectors(SelectorForCluster(cluster), SelectorForClusterOperatorVersion(operatorVersion))
    if !sel.Matches(labels.Set(cluster.Labels)) {
        glog.V(2).Infof("Setting labels on cluster %s", SelectorForCluster(cluster).String())
        if cluster.Labels == nil {
            cluster.Labels = make(map[string]string)
        }
        cluster.Labels[constants.ClusterLabel] = cluster.Name
        cluster.Labels[constants.MySQLOperatorVersionLabel] = buildversion.GetBuildVersion()
        return m.clusterUpdater.UpdateClusterLabels(cluster.DeepCopy(), labels.Set(cluster.Labels))
    }

    // Create a MySQL root password secret for the cluster if required.
    if cluster.RequiresSecret() {
        err = m.secretControl.CreateSecret(secrets.NewMysqlRootPassword(cluster))
        if err != nil && !apierrors.IsAlreadyExists(err) {
            return errors.Wrap(err, "creating root password Secret")
        }
    }
    
    // 检测services是否存在，不存在则新建重启
    svc, err := m.serviceLister.Services(cluster.Namespace).Get(cluster.Name)
    // If the resource doesn't exist, we'll create it
    if apierrors.IsNotFound(err) {
        glog.V(2).Infof("Creating a new Service for cluster %q", nsName)
        svc = services.NewForCluster(cluster)
        err = m.serviceControl.CreateService(svc)
    }

    // If an error occurs during Get/Create, we'll requeue the item so we can
    // attempt processing again later. This could have been caused by a
    // temporary network failure, or any other transient reason.
    if err != nil {
        return err
    }

    // If the Service is not controlled by this Cluster resource, we should
    // log a warning to the event recorder and return.
    if !metav1.IsControlledBy(svc, cluster) {
        msg := fmt.Sprintf(MessageResourceExists, "Service", svc.Namespace, svc.Name)
        m.recorder.Event(cluster, corev1.EventTypeWarning, ErrResourceExists, msg)
        return errors.New(msg)
    }

    //同样如果自定义类型cluster中statefulset不存在，同样重启一个statefulset
    ss, err := m.statefulSetLister.StatefulSets(cluster.Namespace).Get(cluster.Name)
    // If the resource doesn't exist, we'll create it
    if apierrors.IsNotFound(err) {
        glog.V(2).Infof("Creating a new StatefulSet for cluster %q", nsName)
        ss = statefulsets.NewForCluster(cluster, m.opConfig.Images, svc.Name)
        err = m.statefulSetControl.CreateStatefulSet(ss)
    }

    // If an error occurs during Get/Create, we'll requeue the item so we can
    // attempt processing again later. This could have been caused by a
    // temporary network failure, or any other transient reason.
    if err != nil {
        return err
    }

    // If the StatefulSet is not controlled by this Cluster resource, we
    // should log a warning to the event recorder and return.
    if !metav1.IsControlledBy(ss, cluster) {
        msg := fmt.Sprintf(MessageResourceExists, "StatefulSet", ss.Namespace, ss.Name)
        m.recorder.Event(cluster, corev1.EventTypeWarning, ErrResourceExists, msg)
        return fmt.Errorf(msg)
    }
    //检测mysql-operator版本
    // Upgrade the required component resources the current MySQLOperator version.
    if err := m.ensureMySQLOperatorVersion(cluster, ss, buildversion.GetBuildVersion()); err != nil {
        return errors.Wrap(err, "ensuring MySQL Operator version")
    }
    //检测数据库版本
    // Upgrade the MySQL server version if required.
    if err := m.ensureMySQLVersion(cluster, ss); err != nil {
        return errors.Wrap(err, "ensuring MySQL version")
    }

    // 如果cluster副本数与期望的不一致，则更新statefulset，从而消除差异
    // If this number of the members on the Cluster does not equal the
    // current desired replicas on the StatefulSet, we should update the
    // StatefulSet resource.
    if cluster.Spec.Members != *ss.Spec.Replicas {
        glog.V(4).Infof("Updating %q: clusterMembers=%d statefulSetReplicas=%d",
            nsName, cluster.Spec.Members, ss.Spec.Replicas)
        old := ss.DeepCopy()
        ss = statefulsets.NewForCluster(cluster, m.opConfig.Images, svc.Name)
        if err := m.statefulSetControl.Patch(old, ss); err != nil {
            // Requeue the item so we can attempt processing again later.
            // This could have been caused by a temporary network failure etc.
            return err
        }
    }
    //更新自定义类型状态
    // Finally, we update the status block of the Cluster resource to
    // reflect the current state of the world.
    err = m.updateClusterStatus(cluster, ss)
    if err != nil {
        return err
    }

    m.recorder.Event(cluster, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
    return nil
}

这个控制核心相对简单，就是维护services和stateful与定义在cluster中的一致。

总结

• controller主要维护services是否存在以及statefulset集群是否与期望的一致
• 故障迁移完全依赖statefulset自愈
• 扩缩容也是通过设置statefulset副本数来实现

2.3 mysql-agent

 作为mysql服务器的sidecar，与mysql服务器部署在同一个pod中，主要的作用有：

1、用于实时监控数据库状态，以及操作数据库以完成集群搭建；

2、通过探针，从而保证当mysql集群搭建完成之前或者mysql mgr故障时，设置pod不可用；

下面是构建mysql-agent pod容器源码（statefulset.go文件中）

v1.Container{
        Name:         MySQLAgentName,  //容器名
        Image:        fmt.Sprintf("%s:%s", mysqlAgentImage, agentVersion), //相应镜像
        Args:         []string{"--v=4"},
        VolumeMounts: volumeMounts(cluster), //挂载卷
        Env: []v1.EnvVar{   //传入的相关环境变量，主要是连接同pod下mysql的相关配置
            clusterNameEnvVar(cluster),
            namespaceEnvVar(),
            replicationGroupSeedsEnvVar(replicationGroupSeeds),
            multiMasterEnvVar(cluster.Spec.MultiMaster),
            rootPassword,
            {
                Name: "MY_POD_IP",
                ValueFrom: &v1.EnvVarSource{
                    FieldRef: &v1.ObjectFieldSelector{
                        FieldPath: "status.podIP",
                    },
                },
            },
        },
        // 设置Livesness探针，用于k8s检测pod存活情况
        LivenessProbe: &v1.Probe{  
            Handler: v1.Handler{
                HTTPGet: &v1.HTTPGetAction{
                    Path: "/live",
                    Port: intstr.FromInt(int(agentopts.DefaultMySQLAgentHeathcheckPort)),
                },
            },
        },
        // 设置Readiness探针，用于k8s检测该pod是否可以对外提供服务
        ReadinessProbe: &v1.Probe{
            Handler: v1.Handler{
                HTTPGet: &v1.HTTPGetAction{
                    Path: "/ready",
                    Port: intstr.FromInt(int(agentopts.DefaultMySQLAgentHeathcheckPort)),
                },
            },
        },
        Resources: resourceLimits,
    }

监控数据库状态与集群搭建逻辑

agent监控数据库同步逻辑主要在Sync函数中，在agent主函数中的调用逻辑如下：

// 首先调用Sync函数，通过返回值检测数据库是否启动成功，如果成功，则进入后续逻辑
for !manager.Sync(ctx) {
    time.Sleep(10 * time.Second)
}
wg.Add(1)
go func() {
    defer wg.Done()
    manager.Run(ctx) //run函数启动一个协程，每隔一定的时间运行Sync函数
}()

// Run函数调用wait.Until，启动一个协程，每个15秒运行Sync函数
func (m *ClusterManager) Run(ctx context.Context) {
    wait.Until(func() { m.Sync(ctx) }, time.Second*pollingIntervalSeconds, ctx.Done())

    <-ctx.Done()
}

下面来具体分析Sync函数的实现

/*
检测数据库是否运行，并监控器其在集群中的运行状态，进行相应的操作

*/
func (m *ClusterManager) Sync(ctx context.Context) bool {
    //通过mysqladmin连接数据库，检测数据库是否运行
    if !isDatabaseRunning(ctx) {
        glog.V(2).Infof("Database not yet running. Waiting...")
        return false
    }

    // 通过mysqlsh连接数据库获取集群状态信息，dba.get_cluster('%s').status()
    clusterStatus, err := m.getClusterStatus(ctx)
    if err != nil {
        myshErr, ok := errors.Cause(err).(*mysqlsh.Error)
        if !ok {
            glog.Errorf("Failed to get the cluster status: %+v", err)
            return false
        }

        /*
        如果没有想过信息，说明集群还没有建立，并且本pod编号为0，说明是整个集群第一个启动的数据库
        则调用bootstrap自己建立集群
        */
        if m.Instance.Ordinal == 0 {
            // mysqlsh dba.create_cluster('%s', %s).status()创建集群
            clusterStatus, err = m.bootstrap(ctx, myshErr)
            if err != nil {
                glog.Errorf("Error bootstrapping cluster: %v", err)
                metrics.IncEventCounter(clusterCreateErrorCount)
                return false
            }
            metrics.IncEventCounter(clusterCreateCount)
        } else {
            glog.V(2).Info("Cluster not yet present. Waiting...")
            return false
        }
    }

    // 到这一步，说明以及获得集群状态信息，设置集群状态信息，以便healthckeck协程进行检查
    cluster.SetStatus(clusterStatus)

    if clusterStatus.DefaultReplicaSet.Status == innodb.ReplicaSetStatusNoQuorum {
        glog.V(4).Info("Cluster as seen from this instance is in NO_QUORUM state")
        metrics.IncEventCounter(clusterNoQuorumCount)
    }

    online := false
    // 获取状态信息中本节点的状态
    instanceStatus := clusterStatus.GetInstanceStatus(m.Instance.Name())
    // 下面是根据整体进行相应的处理
    switch instanceStatus {
    case innodb.InstanceStatusOnline:
        metrics.IncStatusCounter(instanceStatusCount, innodb.InstanceStatusOnline)
        glog.V(4).Info("MySQL instance is online")
        online = true

    case innodb.InstanceStatusRecovering:
        metrics.IncStatusCounter(instanceStatusCount, innodb.InstanceStatusRecovering)
        glog.V(4).Info("MySQL instance is recovering")
    /*
    如果状态为missing
    1、首先获得主节点信息GetPrimaryAddr（其是通过节点是否可以读写来判断哪个节点是主）；
    2、调用handleInstanceMissing函数，
            检查是否可以重新加入，主要是查看主节点是否存活
            然后重新加入集群dba.get_cluster('%s').rejoin_instance('%s', %s)
    */
    case innodb.InstanceStatusMissing:
        metrics.IncStatusCounter(instanceStatusCount, innodb.InstanceStatusMissing)
        primaryAddr, err := clusterStatus.GetPrimaryAddr()
        if err != nil {
            glog.Errorf("%v", err)
            return false
        }
        online = m.handleInstanceMissing(ctx, primaryAddr)
        if online {
            metrics.IncEventCounter(instanceRejoinCount)
        } else {
            metrics.IncEventCounter(instanceRejoinErrorCount)
        }
    /*
    如果状态为notfound，也就说明该实例还没有加入集群，
    获取主节点信息，然后调用handleInstanceNotFound添加到集群
    dba.get_cluster('%s').add_instance('%s', %s)
    */
    case innodb.InstanceStatusNotFound:
        metrics.IncStatusCounter(instanceStatusCount, innodb.InstanceStatusNotFound)
        primaryAddr, err := clusterStatus.GetPrimaryAddr()
        if err != nil {
            glog.Errorf("%v", err)
            return false
        }
        online = m.handleInstanceNotFound(ctx, primaryAddr)
        if online {
            metrics.IncEventCounter(instanceAddCount)
        } else {
            metrics.IncEventCounter(instanceAddErrorCount)
        }

    case innodb.InstanceStatusUnreachable:
        metrics.IncStatusCounter(instanceStatusCount, innodb.InstanceStatusUnreachable)

    default:
        metrics.IncStatusCounter(instanceStatusCount, innodb.InstanceStatusUnknown)
        glog.Errorf("Received unrecognised cluster membership status: %q", instanceStatus)
    }

    if online && !m.Instance.MultiMaster {
        m.ensurePrimaryControllerState(ctx, clusterStatus)
    }

    return online
}

总结：

 整个agent集群监控过程就如上述代码，相对简单。

1、集群的主从切换，以及数据一致性完全由mysql group replication来保证，agent仅仅确保集群的搭建和维护；

2、并且通过节点是否可读写来区分和判断集群主节点，然后进行新节点或者旧节点的加入。

Readiness探针

 上面我们说过mysql节点的可用性判断是通过Readiness来实现的。当k8s通过agent设置的Readiness，没有得到相应的结果，则会设置这个pod的ready状态设为不可用，从而services不会将相应的流量导入该节点。

 下面我们来看一下agent相关的健康检查。

• 首先我们在说一小节说明了：在获得集群状态后，保存集群状态信息，以便healthckeck协程进行检查

cluster.SetStatus(clusterStatus)

• 然后根据设置的状态来判断，返回健康检查结果

func NewHealthCheck() (healthcheck.Check, error) {
    instance, err := NewLocalInstance()
    if err != nil {
        return nil, errors.Wrap(err, "getting local mysql instance")
    }

    return func() error {
        s := GetStatus()
        // 如果本节点状态不为online，则返回错误信息
        if s == nil || s.GetInstanceStatus(instance.Name()) != innodb.InstanceStatusOnline {
            return errors.New("database still requires management")
        }
        return nil
    }, nil
}

• 在一个协程中开启http服务，用于k8s探针检查

// Set up healthchecks (liveness and readiness).
checkInCluster, err := cluster.NewHealthCheck()
if err != nil {
    glog.Fatal(err)
}
health := healthcheck.NewHandler()
health.AddReadinessCheck("node-in-cluster", checkInCluster)
//在协程中开启http服务，k8s通过http相关的返回结果设置ready状态，从而反映mysql的可用性
go func() {
    glog.Fatal(http.ListenAndServe(
        net.JoinHostPort(opts.Address, strconv.Itoa(int(opts.HealthcheckPort))),
        health,
    ))
}()

// http服务相应目录的回调函数
func NewHandler() Handler {
    h := &basicHandler{
        livenessChecks:  make(map[string]Check),
        readinessChecks: make(map[string]Check),
    }
    h.Handle("/live", http.HandlerFunc(h.LiveEndpoint))
    h.Handle("/ready", http.HandlerFunc(h.ReadyEndpoint))
    return h
}
// 在相应的回调函数中，如果存在错误，则返回StatusServiceUnavailable错误码，从而通知k8s该节点不可用
func (s *basicHandler) collectChecks(checks map[string]Check, resultsOut map[string]string, statusOut *int) {
    s.checksMutex.RLock()
    defer s.checksMutex.RUnlock()
    for name, check := range checks {
        if err := check(); err != nil {
            *statusOut = http.StatusServiceUnavailable
            resultsOut[name] = err.Error()
        } else {
            resultsOut[name] = "OK"
        }
    }
}

总结：

 mysql健康检查的整个逻辑如下：

1、Sync协程获取了集群状态信息，并保存在全局变量中；

2、healcheck健康检查协程，通过查看本节点集群状态信息，来控制http服务GET相应目录的返回（只有本节点集群状态为online才正常返回）；

3、k8s通过http服务GET相应目录，触发调用相应的回调函数，通过返回结果设置k8s上该pod是否可服务。