func (r *TestWorkloadReconciler) reconcileRunning(ctx context.Context, workload *naglfarv1.TestWorkload) (ctrl.Result, error) {
	var (
		phaseChangedWorkloads = 0
		succeededWorkloads    = 0
		failedWorkloads       = 0
	)
	for _, item := range workload.Spec.Workloads {
		workloadNode, err := r.getWorkloadRequestNode(ctx, workload.Namespace, item.DockerContainer)
		if err != nil {
			return ctrl.Result{}, err
		}
		if workloadNode == nil {
			err := fmt.Errorf("cannot find the resource %s", item.DockerContainer.ResourceRequest.Name)
			r.Recorder.Event(workload, "Warning", "Inspect", err.Error())
			return ctrl.Result{}, err
		}
		switch workloadNode.Status.State {
		case naglfarv1.ResourcePending, naglfarv1.ResourceDestroy:
			panic(fmt.Sprintf("there's a bug, it shouldn't see the `%s` state", workloadNode.Status.State))
		case naglfarv1.ResourceUninitialized, naglfarv1.ResourceReady, naglfarv1.ResourceFinish:
			if workload.Status.WorkloadStatus == nil {
				workload.Status.WorkloadStatus = make(map[string]naglfarv1.WorkloadStatus)
			}
			if _, exist := workload.Status.WorkloadStatus[item.Name]; !exist {
				phaseChangedWorkloads += 1
				workload.Status.WorkloadStatus[item.Name] = naglfarv1.WorkloadStatus{Phase: naglfarv1.ResourcePhasePending}
			}
			if workload.Status.WorkloadStatus[item.Name].Phase != workloadNode.Status.Phase {
				workload.Status.WorkloadStatus[item.Name] = naglfarv1.WorkloadStatus{Phase: workloadNode.Status.Phase}
				phaseChangedWorkloads += 1
			}
			switch workloadNode.Status.Phase {
			case naglfarv1.ResourcePhaseSucceeded:
				succeededWorkloads += 1
				r.Recorder.Eventf(workload, "Normal", "Workload", "workload %s succeeded", item.Name)
			case naglfarv1.ResourcePhaseFailed:
				failedWorkloads += 1
				r.Recorder.Eventf(workload, "Warning", "Workload", "workload %s failed", item.Name)
			}
		default:
			panic(fmt.Sprintf("it's a bug, we forget to process the `%s` state", workloadNode.Status.State))
		}
	}
	if phaseChangedWorkloads != 0 {
		if succeededWorkloads == len(workload.Spec.Workloads) {
			workload.Status.State = naglfarv1.TestWorkloadStateSucceeded
			r.Recorder.Event(workload, "Normal", "Finish", "all workloads finished")
		} else if succeededWorkloads+failedWorkloads == len(workload.Spec.Workloads) {
			workload.Status.State = naglfarv1.TestWorkloadStateFailed
			r.Recorder.Event(workload, "Normal", "Finish", "all workloads finished")
		}
		err := r.Status().Update(ctx, workload)
		return ctrl.Result{RequeueAfter: 5 * time.Second}, err
	}
	return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
}
1. poll the state of workload resource nodes, if all workloads have finished, set itself to `finish`
package controllers
import (
	"context"
	"fmt"
	"time"
	"github.com/go-logr/logr"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	naglfarv1 "github.com/PingCAP-QE/Naglfar/api/v1"
	"github.com/PingCAP-QE/Naglfar/pkg/ref"
	"github.com/PingCAP-QE/Naglfar/pkg/util"
)
const (
	testWorkloadFinalizer          = "testworkload.naglfar.pingcap.com"
	NaglfarClusterNs               = "NAGLFAR_CLUSTER_NS"
	NaglfarTestResourceRequestName = "NAGLFAR_TEST_RESOURCE_REQUEST_NAME"
	NaglfarTestWorkloadName        = "NAGLFAR_TESTWORKLOAD_NAME"
	NaglfarTestWorkloadItem        = "NAGLFAR_TESTWORKLOAD_WORKLOAD_ITEM"
)
type TestWorkloadReconciler struct {
	client.Client
	Log      logr.Logger
	Scheme   *runtime.Scheme
	Recorder record.EventRecorder
}
func (r *TestWorkloadReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
	ctx := context.Background()
	log := r.Log.WithValues("testworkload", req.NamespacedName)
	workload := new(naglfarv1.TestWorkload)
	if err := r.Get(ctx, req.NamespacedName, workload); err != nil {
		log.Error(err, "unable to fetch testworkload")
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}
	if workload.ObjectMeta.DeletionTimestamp.IsZero() {
		if !util.StringsContains(workload.ObjectMeta.Finalizers, testWorkloadFinalizer) {
			workload.ObjectMeta.Finalizers = append(workload.ObjectMeta.Finalizers, testWorkloadFinalizer)
			if err := r.Update(ctx, workload); err != nil {
				return ctrl.Result{}, err
			}
		}
	} else {
		if util.StringsContains(workload.ObjectMeta.Finalizers, testWorkloadFinalizer) {
			if err := r.uninstallWorkload(ctx, workload); err != nil {
				return ctrl.Result{}, err
			}
		}
		workload.Finalizers = util.StringsRemove(workload.Finalizers, testWorkloadFinalizer)
		if err := r.Update(ctx, workload); err != nil {
			return ctrl.Result{}, err
		}
		return ctrl.Result{}, nil
	}
	switch workload.Status.State {
	case "":
		workload.Status.State = naglfarv1.TestWorkloadStatePending
		workload.Status.WorkloadStatus = make(map[string]naglfarv1.WorkloadStatus)
		err := r.Status().Update(ctx, workload)
		return ctrl.Result{}, err
	case naglfarv1.TestWorkloadStatePending:
		return r.reconcilePending(ctx, workload)
	case naglfarv1.TestWorkloadStateRunning:
		return r.reconcileRunning(ctx, workload)
	case naglfarv1.TestWorkloadStateSucceeded, naglfarv1.TestWorkloadStateFailed:
		return r.reconcileFinish(workload)
	}
	return ctrl.Result{}, nil
}
func (r *TestWorkloadReconciler) reconcilePending(ctx context.Context, workload *naglfarv1.TestWorkload) (ctrl.Result, error) {
	clusterTopologies := make(map[types.NamespacedName]struct{})
	for _, item := range workload.Spec.ClusterTopologiesRefs {
		clusterTopologies[types.NamespacedName{
			Namespace: workload.Namespace,
			Name:      item.Name,
		}] = struct{}{}
	}
	topologies, allReady, err := r.checkTopologiesReady(ctx, clusterTopologies)
	if err != nil {
		return ctrl.Result{}, err
	}
	if !allReady {
		r.Recorder.Event(workload, "Warning", "Precondition", "not all clusters are ready")
		return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
	}
	var resourceList naglfarv1.TestResourceList
	if err := r.List(ctx, &resourceList, client.InNamespace(workload.Namespace)); err != nil {
		return ctrl.Result{}, err
	}
	var installedCount = 0
	for _, item := range workload.Spec.Workloads {
		workloadNode, err := r.getWorkloadRequestNode(ctx, workload.Namespace, item.DockerContainer)
		if err != nil {
			return ctrl.Result{}, err
		}
		if workloadNode == nil {
			err := fmt.Errorf("cannot find the resource %s", item.DockerContainer.ResourceRequest.Name)
			r.Recorder.Event(workload, "Warning", "Precondition", err.Error())
			return ctrl.Result{}, err
		}
		if workloadNode.Status.ClaimRef != nil && *workloadNode.Status.ClaimRef != ref.CreateRef(&workload.ObjectMeta) {
			r.Recorder.Eventf(workload, "Warning", "Precondition", "node %s is occupied by %s",
				ref.CreateRef(&workloadNode.ObjectMeta).Key(),
				workloadNode.Status.ClaimRef.Key())
			return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
		}
		switch workloadNode.Status.State {
		case naglfarv1.ResourceDestroy:
			return ctrl.Result{RequeueAfter: time.Second}, nil
		case naglfarv1.ResourcePending:
			panic(fmt.Sprintf("there's a bug, it shouldn't see the `%s` state", workloadNode.Status.State))
		case naglfarv1.ResourceUninitialized:
			if workloadNode.Status.Image == "" {
				self := ref.CreateRef(&workload.ObjectMeta)
				workloadNode.Status.ClaimRef = &self
				r.setContainerSpec(&workloadNode.Status, workload, &item)
				topologyEnvs, err := r.buildTopologyEnvs(&workload.Spec, topologies, resourceList)
				if err != nil {
					r.Recorder.Event(workload, "Warning", "Precondition", err.Error())
					return ctrl.Result{}, err
				}
				workloadNode.Status.Envs = append(workloadNode.Status.Envs, topologyEnvs...)
				if err := r.Status().Update(ctx, workloadNode); err != nil {
					return ctrl.Result{}, err
				}
				r.Recorder.Event(workload, "Normal", "Install", fmt.Sprintf("preparing the workload: %s", item.Name))
			} else if workloadNode.Status.Image != item.DockerContainer.Image {
				err := fmt.Errorf("resource %s has installed a conflict image", workloadNode.Name)
				r.Recorder.Event(workload, "Warning", "Precondition", err.Error())
				return ctrl.Result{}, err
			}
		case naglfarv1.ResourceReady, naglfarv1.ResourceFinish:
			self := ref.CreateRef(&workload.ObjectMeta)
			if workloadNode.Status.ClaimRef == nil || *workloadNode.Status.ClaimRef != self {
				err := fmt.Errorf("resource %s is used by others now, wait", workloadNode.Name)
				r.Recorder.Eventf(workload, "Warning", "Precondition", err.Error())
				return ctrl.Result{}, err
			}
			installedCount += 1
		default:
			panic(fmt.Sprintf("there's a bug, forget to process the `%s` state", workloadNode.Status.State))
		}
	}
	if installedCount == len(workload.Spec.Workloads) {
		workload.Status.State = naglfarv1.TestWorkloadStateRunning
		if err := r.Status().Update(ctx, workload); err != nil {
			return ctrl.Result{}, err
		}
		r.Recorder.Event(workload, "Normal", "Install", "all workload has been installed")
		return ctrl.Result{}, nil
	}
	return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
}