1
2
3
4
5

# 创建主集群
kind create cluster --name main
# 创建子集群
kind create cluster --name sub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

// TestReconciler reconciles a Test object
type TestReconciler struct {
	// 主集群 client
	client.Client
	// 所有集群的客户端列表
	Clients map[string]client.Client
	Scheme *runtime.Scheme
}
// NewTestReconciler ...
func NewTestReconciler(mgr ctrl.Manager, clusters map[string]cluster.Cluster) (*TestReconciler, error) {
	r := TestReconciler{
		Client: mgr.GetClient(),
		Scheme: mgr.GetScheme(),
		Clients: map[string]client.Client{
			"main": mgr.GetClient(),
		},
	}
	for name, cluster := range clusters {
		r.Clients[name] = cluster.GetClient()
	}
	err := r.SetupWithManager(mgr)
	return &r, err
}
func (r *TestReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	var test jobv1.Test
	var res ctrl.Result
	err := r.Get(ctx, req.NamespacedName, &test)
	if err != nil {
		return res, client.IgnoreNotFound(err)
	}
	job := test.Job()
	for _, c := range r.Clients {
		err := c.Create(ctx, job.DeepCopy())
		if err != nil {
			return res, err
		}
	}
	return ctrl.Result{}, nil
}
// SetupWithManager sets up the controller with the Manager.
func (r *TestReconciler) SetupWithManager(mgr ctrl.Manager) error {
	builder := ctrl.NewControllerManagedBy(mgr).
		For(&jobv1.Test{})
	return builder.Complete(r)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

// NewSubClusters 初始化子集群
// 在 ~/.kube/config 文件中需要有这两个 context 集群
func NewSubClusters(mgr ctrl.Manager, clientContexts ...string) map[string]cluster.Cluster {
	clusters := map[string]cluster.Cluster{}
	for _, v := range clientContexts {
		conf, err := config.GetConfigWithContext(v)
		checkErr(err, "get client config fail", "context", v)
		c, err := cluster.New(conf)
		checkErr(err, "new cluster fail", "context", v)
		err = mgr.Add(c)
		checkErr(err, "add cluster in manager", "context", v)
		clusters[v] = c
	}
	return clusters
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

type secretMirrorReconciler struct {
	referenceClusterClient, mirrorClusterClient client.Client
}
func (r *secretMirrorReconciler) Reconcile(r reconcile.Request)(reconcile.Result, error){
	s := &corev1.Secret{}
	if err := r.referenceClusterClient.Get(context.TODO(), r.NamespacedName, s); err != nil {
		if kerrors.IsNotFound{ return reconcile.Result{}, nil }
		return reconcile.Result, err
	}
	if err := r.mirrorClusterClient.Get(context.TODO(), r.NamespacedName, &corev1.Secret); err != nil {
		if !kerrors.IsNotFound(err) {
			return reconcile.Result{}, err
		}
		mirrorSecret := &corev1.Secret{
			ObjectMeta: metav1.ObjectMeta{Namespace: s.Namespace, Name: s.Name},
			Data: s.Data,
		}
		return reconcile.Result{}, r.mirrorClusterClient.Create(context.TODO(), mirrorSecret)
	}
	return nil
}
func NewSecretMirrorReconciler(mgr manager.Manager, mirrorCluster cluster.Cluster) error {
	return ctrl.NewControllerManagedBy(mgr).
		// Watch Secrets in the reference cluster
		For(&corev1.Secret{}).
		// Watch Secrets in the mirror cluster
		Watches(
			source.NewKindWithCache(&corev1.Secret{}, mirrorCluster.GetCache()),
			&handler.EnqueueRequestForObject{},
		).
		Complete(&secretMirrorReconciler{
			referenceClusterClient: mgr.GetClient(),
			mirrorClusterClient:    mirrorCluster.GetClient(),
		})
	}
}
// ... 省略 main 函数

1

Reconcile(ctx context.Context, req ctrl.Request) (res ctrl.Result, err error)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// MuiltClustersEnqueue 多集群入队器
// 将集群名称附加在 Namespace 上
func MuiltClustersEnqueue(clusterName string) handler.EventHandler {
	return handler.EnqueueRequestsFromMapFunc(func(o client.Object) []reconcile.Request {
		return []reconcile.Request{
			{
				NamespacedName: types.NamespacedName{
					Name:      o.GetName(),
					Namespace: clusterName + "/" + o.GetNamespace(),
				},
			},
		}
	})
}
// SetupWithManager sets up the controller with the Manager.
func (r *TestJobReconciler) SetupWithManager(mgr ctrl.Manager, cs map[string]cluster.Cluster) error {
	build := ctrl.NewControllerManagedBy(mgr).
		For(&batchv1.Job{})
		// 监听多个集群
	for name, cluster := range cs {
		build = build.Watches(
			source.NewKindWithCache(&batchv1.Job{}, cluster.GetCache()),
			MuiltClustersEnqueue(name),
		)
	}
	return build.Complete(r)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

func (r *TestJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	var res ctrl.Result
	logger := log.FromContext(ctx)
	var job batchv1.Job
	cluster, ns := GetClusterNameNs(req.Namespace)
	req.Namespace = ns
	logger.Info("get job", "cluster", cluster)
	err := r.GetClient(cluster).Get(ctx, req.NamespacedName, &job)
	if err != nil {
		return res, client.IgnoreNotFound(err)
	}
	if job.Status.CompletionTime.IsZero() {
		return res, nil
	}
	logger.Info("job complete", "cluster", cluster)
	var test jobv1.Test
	err = r.Get(ctx, clusterx.GetOwnerNameNs(&job), &test)
	if err != nil {
		return res, client.IgnoreNotFound(err)
	}
	test.Status.Phase = "finished"
	err = r.Client.Status().Update(ctx, &test)
	return ctrl.Result{}, err
}