编写Kubernetes Operator

...
  if err := app.Run(opts); err != nil {
		fmt.Fprintf(os.Stderr, "%v\n", err)
		os.Exit(1)
	}
...	

// Run starts the mysql-operator controllers. This should never exit.
func Run(s *operatoropts.MySQLOperatorOpts) error {
  // 构造kubeconfig以便连接kubernetes的APIServer
	kubeconfig, err := clientcmd.BuildConfigFromFlags(s.Master, s.KubeConfig)
	if err != nil {
		return err
	}
	
	...
	
	// 构造kubeClient、 mysqlopClient, 以便操作Kubernetes里的一些资源
	kubeClient := kubernetes.NewForConfigOrDie(kubeconfig)
	mysqlopClient := clientset.NewForConfigOrDie(kubeconfig)

  // 构造一些共享的informer，以便监听自定义对象及kubernetes里的一些核心资源
	// Shared informers (non namespace specific).
	operatorInformerFactory := informers.NewFilteredSharedInformerFactory(mysqlopClient, resyncPeriod(s)(), s.Namespace, nil)
	kubeInformerFactory := kubeinformers.NewFilteredSharedInformerFactory(kubeClient, resyncPeriod(s)(), s.Namespace, nil)

  var wg sync.WaitGroup

  // 构造自定义类型mysqlcluster的控制器
	clusterController := cluster.NewController(
		*s,
		mysqlopClient,
		kubeClient,
		operatorInformerFactory.MySQL().V1alpha1().Clusters(),
		kubeInformerFactory.Apps().V1beta1().StatefulSets(),
		kubeInformerFactory.Core().V1().Pods(),
		kubeInformerFactory.Core().V1().Services(),
		30*time.Second,
		s.Namespace,
	)
	wg.Add(1)
	go func() {
		defer wg.Done()
		clusterController.Run(ctx, 5)
	}()
	
	// 下面分别为每个自定义类型构造了相应的控制器
	...

// NewController creates a new MySQLController.
func NewController(
	...
) *MySQLController {
  // 构造MySQLController
  m := MySQLController{
		...
	}

  // 监控自定义类型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)
			}
		},
	})

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

func (m *MySQLController) syncHandler(key string) error {
	...

	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
	}

	...

	svc, err := m.serviceLister.Services(cluster.Namespace).Get(cluster.Name)
	// If the resource doesn't exist, we'll create it
	// 如果该自定义资源对象存在，则应该要创建相应的Serivce，如Serivce不存在，则创建
	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)
	}

	ss, err := m.statefulSetLister.StatefulSets(cluster.Namespace).Get(cluster.Name)
	// If the resource doesn't exist, we'll create it
	// 如果该自定义资源对象存在，则应该要创建相应的StatefulSet，如StatefulSet不存在，则创建
	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)
	}

	// Upgrade the required component resources the current MySQLOperator version.
	// 确保StatefulSet上的BuildVersion与自定义资源对象上的一致，如不一致，则修改得一致
	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")
	}

	// 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.
	// 如果StatefulSet的Replicas值与自定义资源对象上配置不一致，则更新StatefulSet
	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
}

			OwnerReferences: []metav1.OwnerReference{
				*metav1.NewControllerRef(cluster, schema.GroupVersionKind{
					Group:   v1alpha1.SchemeGroupVersion.Group,
					Version: v1alpha1.SchemeGroupVersion.Version,
					Kind:    v1alpha1.ClusterCRDResourceKind,
				}),
			},

// Reconcile reads that state of the cluster for a Memcached object and makes changes based on the state read
// and what is in the Memcached.Spec
// TODO(user): Modify this Reconcile function to implement your Controller logic.  This example creates
// a Memcached Deployment for each Memcached CR
// Note:
// The Controller will requeue the Request to be processed again if the returned error is non-nil or
// Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
func (r *ReconcileMemcached) Reconcile(request reconcile.Request) (reconcile.Result, error) {
  ...
}