# 实现Controller
#### Controller实现背景
在创建Kafka实例时，Kafka-Operator需要创建的Kubernetes资源如下：
- 1个StatefulSet，包含3个Pod分别启动ZooKeeper；
- 1个Service，用来暴露ZooKeeper访问地址；
- 1个StatefulSet，包含3个Pod分别启动Hwfka broker；
- 1个Service，用来暴露Hwfka访问地址；
- 1个Deployment，包含1个Pod启动KafkaManager；
- 1个Service，用来暴露KafkaManager访问地址。
在创建API的时候，SDK已自动创建controllers/hwfka_controller.go，为SetupWithManager添加对Service、StatefulSet、Deployment的监听和管理。
```
import (
"context"
"fmt"
"github.com/go-logr/logr"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
osctestv1 "hwfka-operator/api/v1"
)
func (r *HwfkaReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&osctestv1.Hwfka{}).
Owns(&corev1.Service{}).
Owns(&appsv1.StatefulSet{}).
Owns(&appsv1.Deployment{}).
Complete(r)
}
```
#### 调度逻辑
对于Kubernetes集群内任何Hwfka或Service、StatefulSet、Deployment的变化，Controller都会监听到，并生成事件，触发Reconcile()方法。需要在Reconcile()方法中实现协调逻辑，创建Service、StatefulSet等资源，并更新应用实例状态，实现的过程中，可以参考：
1. **查询资源：** 控制器使用controller-runtime库中的Client实现对Kubernetes资源的增查改删，示例代码参见：[example_test.go](https://github.com/kubernetes-sigs/controller-runtime/blob/master/pkg/client/example_test.go)。
2. **创建资源：** 使用Go语言调用Kubernetes API创建资源，可参考[Kubernetes API Reference](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.23/)。
3. **设置关联：** 为创建的Kubernetes资源设置ownerReferences，以便其能在CR删除时被级联删除，可参考如下代码。
   ```
   // 为指定命名空间和名称的 secret 设置 owner，返回 secret
   func (k *K8sClient) SetSecretOwner(cr *oscv1.Kafka, secretName string) (*corev1.Secret, error) {
   ctx := context.Background()
   secret, err := k.GetSecretByNamespaceName(cr.Namespace, secretName)
   if err != nil {
   return nil, err
   }
   if secret.OwnerReferences == nil {
   patch := client.MergeFrom(secret.DeepCopy())
   secret.OwnerReferences = []metav1.OwnerReference{
   *metav1.NewControllerRef(cr, schema.GroupVersionKind{
   Group:   cr.GroupVersionKind().Group,
   Version: cr.GroupVersionKind().Version,
   Kind:    cr.Kind,
   }),
   }
   if err := k.Patch(ctx, secret, patch); err != nil {
   return nil, err
   }
   }
   return secret, nil
   }
   ```
   
4. 如果一个动作的处理时间较长，为了避免Reconcile阻塞，需要使请求返回并重新排队，有四种方法：
   ```
   // 请求成功，不再排队
   return ctrl.Result{}, nil
   // 请求失败，重新加入队列
   return ctrl.Result{}, err
   // 请求因某种原因需要重新加入队列
   return ctrl.Result{Requeue: true}, nil
   // 请求因某种原因，需要在指定时间后重新加入队列
   return ctrl.Result{RequeueAfter: time.Second*5}, nil
   ```
   
5. 当删除CR时，如果需要预先清理Kubernetes之外的资源，此时仅凭ownerReferences无法实现，可以利用finalizers特性。利用finalizers延时删除资源的方法请参见[Using Finalizers](https://book.kubebuilder.io/reference/using-finalizers.html)。
   ![](https://support.huaweicloud.com/devg-osc/public_sys-resources/note_3.0-zh-cn.png)
   示例代码请参见[Controller实现](https://support.huaweicloud.com/devg-osc/osc_ss_0055.html)。
   
 
#### 挂载存储
应用实例可以使用本地磁盘存储（HostPath），也可以使用网络存储或云存储（**推荐** ），在Kubernetes中，通过PersistentVolume(PV)方式挂载存储，具体请参见[挂载存储](https://support.huaweicloud.com/devg-osc/osc_ss_0056.html) 。
#### 云存储类型
云存储类型主要指标包括：
- IO读写时延：连续两次进行读写操作所需要的最小时间间隔。
- IOPS：每秒进行读写的操作次数。
- 吞吐量：每秒读取和写入的数据量。
![](https://support.huaweicloud.com/devg-osc/public_sys-resources/note_3.0-zh-cn.png)
- 更多信息请参见[云硬盘EVS：磁盘类型及性能介绍](https://support.huaweicloud.com/productdesc-evs/zh-cn_topic_0044524691.html)，云磁盘计费方式请参见[云硬盘计费说明](https://support.huaweicloud.com/productdesc-evs/evs_01_0124.html)。
- 通过在集群中创建PVC的方式创建存储，创建方式请参见[创建存储](https://support.huaweicloud.com/devg-osc/osc_ss_0057.html)。
 
#### RBAC权限管理
Operator使用基于角色的访问控制机制对Kubernetes中的资源进行访问，因此，要保证Operator正确的运行，需要使用管理员权限的账号对Operator进行授权，即预先创建对应的role和rolebinding。
在Reconcile()上添加markers，增加Operator对于Service、StatefulSet、Deployment、Pod的管理权限，SDK即可自动生成对应的role和rolebinding等资源描述文件。
```
// +kubebuilder:rbac:groups=osctest.huawei.com,resources=hwfkas,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=osctest.huawei.com,resources=hwfkas/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch
// +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims,verbs=get;list;watch;update
func (r *HwfkaReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
    // ...
}
```
如果Operator为namespace级别的，则需要还添加namespace，如：
```
// +kubebuilder:rbac:groups=osctest.huawei.com,namespace=example,resources=hwfkas,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=osctest.huawei.com,namespace=example,resources=hwfkas/status,verbs=get;update;patch
```