wenshu/k8s-orchestrator/internal/controller/starrocksvirtualcluster_controller.go

package controller

import (
	"context"
	"fmt"
	"reflect"
	"sort"
	"strconv"
	"strings"
	"time"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"

	orcv1 "github.com/cyrus/spark-orchestrator/api/v1alpha1"
	"github.com/cyrus/spark-orchestrator/internal/k8s"
	"github.com/cyrus/spark-orchestrator/internal/util"
)

const (
	starRocksClusterAPIVersion       = "starrocks.com/v1"
	starRocksClusterKind             = "StarRocksCluster"
	starRocksDefaultFeImage          = "starrocks/fe-ubuntu:3.3.10"
	starRocksDefaultBeImage          = "starrocks/be-ubuntu:3.3.10"
	starRocksNamespaceOwnerKindLabel = "starrocks"
	awsChecksumCalculationWhenNeeded = "WHEN_REQUIRED"
)

type StarRocksVirtualClusterReconciler struct {
	client.Client
	Scheme           *runtime.Scheme
	PolarisBootstrap PolarisCatalogBootstrapConfig
}

func (r *StarRocksVirtualClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx).WithValues("starRocksVirtualCluster", req.Name)

	var cluster orcv1.StarRocksVirtualCluster
	if err := r.Get(ctx, req.NamespacedName, &cluster); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	if cluster.Spec.DesiredState == "" {
		cluster.Spec.DesiredState = orcv1.VirtualClusterDesiredActive
	}
	if cluster.Spec.ScaleInPolicy == "" {
		cluster.Spec.ScaleInPolicy = orcv1.ScaleInPolicyBlockIfRunning
	}

	if cluster.DeletionTimestamp.IsZero() {
		if !controllerutil.ContainsFinalizer(&cluster, util.FinalizerStarRocksCluster) {
			controllerutil.AddFinalizer(&cluster, util.FinalizerStarRocksCluster)
			if err := r.Update(ctx, &cluster); err != nil {
				return ctrl.Result{}, err
			}
		}
	} else {
		if controllerutil.ContainsFinalizer(&cluster, util.FinalizerStarRocksCluster) {
			ns := cluster.Status.Namespace
			if ns == "" {
				ns = resolvedStarRocksNamespace(cluster.Name, cluster.Spec.Namespace)
			}
			if err := r.cleanupCluster(ctx, &cluster, ns); err != nil {
				return ctrl.Result{}, err
			}
			controllerutil.RemoveFinalizer(&cluster, util.FinalizerStarRocksCluster)
			if err := r.Update(ctx, &cluster); err != nil {
				return ctrl.Result{}, err
			}
		}
		return ctrl.Result{}, nil
	}

	ns := cluster.Spec.Namespace
	if ns == "" {
		ns = resolvedStarRocksNamespace(cluster.Name, "")
		if err := k8s.EnsureNamespaceOwnedByCluster(ctx, r.Client, ns, starRocksNamespaceOwnerKindLabel, cluster.Name); err != nil {
			return ctrl.Result{}, err
		}
	} else {
		if err := k8s.EnsureNamespace(ctx, r.Client, ns); err != nil {
			return ctrl.Result{}, err
		}
	}

	if cluster.Spec.DesiredState == orcv1.VirtualClusterDesiredReleased {
		cluster.Status.Phase = orcv1.StarRocksClusterPhaseReleasing
		cluster.Status.ObservedAt = metav1.Now()
		if err := r.Status().Update(ctx, &cluster); err != nil {
			return ctrl.Result{}, err
		}
		if err := r.cleanupCluster(ctx, &cluster, ns); err != nil {
			cluster.Status.Phase = orcv1.StarRocksClusterPhaseFailed
			cluster.Status.LastError = err.Error()
			cluster.Status.ObservedAt = metav1.Now()
			_ = r.Status().Update(ctx, &cluster)
			return ctrl.Result{}, err
		}
		cluster.Status.Phase = orcv1.StarRocksClusterPhaseReleased
		cluster.Status.AssignedFeNodes = nil
		cluster.Status.AssignedBeNodes = nil
		cluster.Status.LastError = ""
		cluster.Status.Namespace = ns
		cluster.Status.ObservedAt = metav1.Now()
		setCondition(&cluster.Status.Conditions, orcv1.NewCondition(orcv1.ConditionReady, "Released", "Cluster released and nodes are idle", metav1.ConditionFalse, cluster.Generation))
		if err := r.Status().Update(ctx, &cluster); err != nil {
			return ctrl.Result{}, err
		}
		return ctrl.Result{}, nil
	}

	targetFeNodes := normalizeNodeList(cluster.Spec.FeNodeRefs)
	targetBeNodes := normalizeNodeList(cluster.Spec.BeNodeRefs)
	if len(targetFeNodes) == 0 {
		return r.failCluster(ctx, &cluster, fmt.Errorf("feNodeRefs must not be empty"))
	}
	if len(targetBeNodes) == 0 {
		return r.failCluster(ctx, &cluster, fmt.Errorf("beNodeRefs must not be empty"))
	}
	if overlap := overlappingNodes(targetFeNodes, targetBeNodes); len(overlap) > 0 {
		return r.failCluster(ctx, &cluster, fmt.Errorf("fe and be nodes cannot overlap: %s", strings.Join(overlap, ",")))
	}

	previousFeNodes := normalizeNodeList(cluster.Status.AssignedFeNodes)
	previousBeNodes := normalizeNodeList(cluster.Status.AssignedBeNodes)
	previousNodes := normalizeNodeList(append(previousFeNodes, previousBeNodes...))
	targetNodes := normalizeNodeList(append(targetFeNodes, targetBeNodes...))
	toRelease := diffNodes(previousNodes, targetNodes)

	for _, nodeID := range toRelease {
		if err := r.releaseNode(ctx, cluster.Name, nodeID); err != nil {
			return r.failCluster(ctx, &cluster, err)
		}
	}
	for _, nodeID := range targetNodes {
		if err := r.allocateNode(ctx, cluster.Name, nodeID); err != nil {
			logger.Error(err, "allocate node failed", "nodeID", nodeID)
			return r.failCluster(ctx, &cluster, err)
		}
	}

	feInventories, err := r.loadNodeInventories(ctx, targetFeNodes)
	if err != nil {
		return r.failCluster(ctx, &cluster, err)
	}
	beInventories, err := r.loadNodeInventories(ctx, targetBeNodes)
	if err != nil {
		return r.failCluster(ctx, &cluster, err)
	}

	cfg, err := resolveStarRocksConfig(&cluster, feInventories, beInventories)
	if err != nil {
		return r.failCluster(ctx, &cluster, err)
	}
	feKubeNodes, err := inventoryNodeNames(feInventories)
	if err != nil {
		return r.failCluster(ctx, &cluster, err)
	}
	beKubeNodes, err := inventoryNodeNames(beInventories)
	if err != nil {
		return r.failCluster(ctx, &cluster, err)
	}

	if err := r.applyStarRocksClusterResource(ctx, cluster.Name, ns, feKubeNodes, beKubeNodes, cfg); err != nil {
		return r.failCluster(ctx, &cluster, err)
	}

	externalPhase, externalReason, err := r.getStarRocksClusterStatus(ctx, cluster.Name, ns)
	if err != nil && !apierrors.IsNotFound(err) {
		return r.failCluster(ctx, &cluster, err)
	}

	cluster.Status.Namespace = ns
	cluster.Status.AssignedFeNodes = targetFeNodes
	cluster.Status.AssignedBeNodes = targetBeNodes
	cluster.Status.LastError = ""
	cluster.Status.ObservedAt = metav1.Now()

	switch strings.ToLower(externalPhase) {
	case "running":
		bootstrapReady, bootstrapMessage, bootstrapErr := r.ensurePolarisCatalogBootstrap(ctx, &cluster, ns)
		if bootstrapErr != nil {
			return r.failCluster(ctx, &cluster, bootstrapErr)
		}
		if bootstrapReady == false {
			cluster.Status.Phase = orcv1.StarRocksClusterPhaseCreating
			if strings.TrimSpace(bootstrapMessage) == "" {
				bootstrapMessage = "Waiting polaris catalog bootstrap"
			}
			setCondition(&cluster.Status.Conditions, orcv1.NewCondition(orcv1.ConditionRunning, "BootstrapPolarisCatalog", bootstrapMessage, metav1.ConditionTrue, cluster.Generation))
			break
		}
		cluster.Status.Phase = orcv1.StarRocksClusterPhaseReady
		setCondition(&cluster.Status.Conditions, orcv1.NewCondition(orcv1.ConditionReady, "ClusterReady", "StarRocks cluster is running", metav1.ConditionTrue, cluster.Generation))
	case "failed":
		cluster.Status.Phase = orcv1.StarRocksClusterPhaseFailed
		cluster.Status.LastError = externalReason
		setCondition(&cluster.Status.Conditions, orcv1.NewCondition(orcv1.ConditionFailed, "ExternalFailed", externalReason, metav1.ConditionTrue, cluster.Generation))
	default:
		if len(toRelease) > 0 ||
			!equalNodeLists(previousFeNodes, targetFeNodes) ||
			!equalNodeLists(previousBeNodes, targetBeNodes) {
			cluster.Status.Phase = orcv1.StarRocksClusterPhaseScaling
		} else {
			cluster.Status.Phase = orcv1.StarRocksClusterPhaseCreating
		}
		setCondition(&cluster.Status.Conditions, orcv1.NewCondition(orcv1.ConditionRunning, "Reconciling", "Waiting StarRocks operator reconcile", metav1.ConditionTrue, cluster.Generation))
	}

	if err := r.Status().Update(ctx, &cluster); err != nil {
		return ctrl.Result{}, err
	}

	if cluster.Status.Phase != orcv1.StarRocksClusterPhaseReady && cluster.Status.Phase != orcv1.StarRocksClusterPhaseFailed {
		return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
	}
	return ctrl.Result{}, nil
}

func (r *StarRocksVirtualClusterReconciler) failCluster(ctx context.Context, cluster *orcv1.StarRocksVirtualCluster, err error) (ctrl.Result, error) {
	cluster.Status.Phase = orcv1.StarRocksClusterPhaseFailed
	cluster.Status.LastError = err.Error()
	cluster.Status.ObservedAt = metav1.Now()
	setCondition(&cluster.Status.Conditions, orcv1.NewCondition(orcv1.ConditionFailed, "ReconcileFailed", err.Error(), metav1.ConditionTrue, cluster.Generation))
	_ = r.Status().Update(ctx, cluster)
	return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
}

func (r *StarRocksVirtualClusterReconciler) cleanupCluster(ctx context.Context, cluster *orcv1.StarRocksVirtualCluster, namespace string) error {
	if err := r.deleteStarRocksClusterResource(ctx, cluster.Name, namespace); err != nil {
		return err
	}

	nodes := normalizeNodeList(append(cluster.Status.AssignedFeNodes, cluster.Status.AssignedBeNodes...))
	if len(nodes) == 0 {
		nodes = normalizeNodeList(append(cluster.Spec.FeNodeRefs, cluster.Spec.BeNodeRefs...))
	}
	for _, nodeID := range nodes {
		if err := r.releaseNode(ctx, cluster.Name, nodeID); err != nil {
			return err
		}
	}
	return nil
}

func (r *StarRocksVirtualClusterReconciler) allocateNode(ctx context.Context, clusterName, nodeID string) error {
	var nodeInv orcv1.NodeInventory
	if err := r.Get(ctx, types.NamespacedName{Name: nodeID}, &nodeInv); err != nil {
		return err
	}
	if nodeInv.Status.KubernetesNodeName == "" {
		return fmt.Errorf("node %s has no kubernetesNodeName yet", nodeID)
	}

	clusterRef := util.StarRocksClusterRef(clusterName)
	if nodeInv.Status.Phase != orcv1.NodePhaseIdle &&
		!(nodeInv.Status.Phase == orcv1.NodePhaseAllocated &&
			(nodeInv.Status.ClusterRef == clusterRef || nodeInv.Status.ClusterRef == clusterName)) {
		return fmt.Errorf("node %s is not idle", nodeID)
	}

	if err := k8s.EnsureStarRocksNodeIsolation(ctx, r.Client, nodeInv.Status.KubernetesNodeName, clusterName); err != nil {
		return err
	}

	nodeInv.Status.Phase = orcv1.NodePhaseAllocated
	nodeInv.Status.ClusterRef = clusterRef
	nodeInv.Status.LastError = ""
	nodeInv.Status.ObservedAt = metav1.Now()
	setCondition(&nodeInv.Status.Conditions, orcv1.NewCondition(orcv1.ConditionReady, "Allocated", fmt.Sprintf("Allocated to starrocks cluster %s", clusterName), metav1.ConditionTrue, nodeInv.Generation))
	return r.Status().Update(ctx, &nodeInv)
}

func (r *StarRocksVirtualClusterReconciler) releaseNode(ctx context.Context, clusterName, nodeID string) error {
	var nodeInv orcv1.NodeInventory
	if err := r.Get(ctx, types.NamespacedName{Name: nodeID}, &nodeInv); err != nil {
		if apierrors.IsNotFound(err) {
			return nil
		}
		return err
	}

	clusterRef := util.StarRocksClusterRef(clusterName)
	if nodeInv.Status.ClusterRef != "" &&
		nodeInv.Status.ClusterRef != clusterName &&
		nodeInv.Status.ClusterRef != clusterRef {
		return fmt.Errorf("node %s is allocated to another cluster: %s", nodeID, nodeInv.Status.ClusterRef)
	}

	if nodeInv.Status.KubernetesNodeName != "" {
		if err := k8s.ClearStarRocksNodeIsolation(ctx, r.Client, nodeInv.Status.KubernetesNodeName); err != nil {
			return err
		}
	}
	nodeInv.Status.Phase = orcv1.NodePhaseIdle
	nodeInv.Status.ClusterRef = ""
	nodeInv.Status.ObservedAt = metav1.Now()
	setCondition(&nodeInv.Status.Conditions, orcv1.NewCondition(orcv1.ConditionReady, "Idle", "Node is idle and available", metav1.ConditionTrue, nodeInv.Generation))
	return r.Status().Update(ctx, &nodeInv)
}

func (r *StarRocksVirtualClusterReconciler) loadNodeInventories(ctx context.Context, nodeIDs []string) ([]orcv1.NodeInventory, error) {
	inventories := make([]orcv1.NodeInventory, 0, len(nodeIDs))
	for _, nodeID := range nodeIDs {
		var nodeInv orcv1.NodeInventory
		if err := r.Get(ctx, types.NamespacedName{Name: nodeID}, &nodeInv); err != nil {
			return nil, err
		}
		inventories = append(inventories, nodeInv)
	}
	return inventories, nil
}

func inventoryNodeNames(inventories []orcv1.NodeInventory) ([]string, error) {
	nodeNames := make([]string, 0, len(inventories))
	for _, inv := range inventories {
		if inv.Status.KubernetesNodeName == "" {
			return nil, fmt.Errorf("node %s has no kubernetesNodeName yet", inv.Name)
		}
		nodeNames = append(nodeNames, inv.Status.KubernetesNodeName)
	}
	sort.Strings(nodeNames)
	return nodeNames, nil
}

func resolveStarRocksConfig(cluster *orcv1.StarRocksVirtualCluster, feNodes, beNodes []orcv1.NodeInventory) (orcv1.StarRocksClusterConfig, error) {
	cfg := cluster.Spec.StarRocksConfig
	if cfg.FeImage == "" {
		cfg.FeImage = starRocksDefaultFeImage
	}
	if cfg.BeImage == "" {
		cfg.BeImage = starRocksDefaultBeImage
	}

	feDerived, err := deriveMinResources(feNodes)
	if err != nil {
		return orcv1.StarRocksClusterConfig{}, fmt.Errorf("derive fe resources: %w", err)
	}
	beDerived, err := deriveMinResources(beNodes)
	if err != nil {
		return orcv1.StarRocksClusterConfig{}, fmt.Errorf("derive be resources: %w", err)
	}

	cfg.FeResources = mergeResourceSpec(feDerived, cfg.FeResources)
	cfg.BeResources = mergeResourceSpec(beDerived, cfg.BeResources)
	return cfg, nil
}

func deriveMinResources(nodes []orcv1.NodeInventory) (*orcv1.StarRocksResourceSpec, error) {
	if len(nodes) == 0 {
		return nil, fmt.Errorf("node list is empty")
	}
	minCPU := nodes[0].Spec.Capacity.CPUCores
	minRAM := nodes[0].Spec.Capacity.RamGB
	for i := 1; i < len(nodes); i++ {
		if nodes[i].Spec.Capacity.CPUCores < minCPU {
			minCPU = nodes[i].Spec.Capacity.CPUCores
		}
		if nodes[i].Spec.Capacity.RamGB < minRAM {
			minRAM = nodes[i].Spec.Capacity.RamGB
		}
	}
	if minCPU <= 0 || minRAM <= 0 {
		return nil, fmt.Errorf("invalid node capacity cpu=%d ram=%d", minCPU, minRAM)
	}
	// Use 75% of CPU/RAM for requests to leave headroom for OS/system processes.
	reqCPU := minCPU * 3 / 4
	if reqCPU < 1 {
		reqCPU = 1
	}
	reqRAM := minRAM * 3 / 4
	if reqRAM < 1 {
		reqRAM = 1
	}
	return &orcv1.StarRocksResourceSpec{
		Requests: orcv1.StarRocksResourcePair{
			CPU:    strconv.Itoa(int(reqCPU)),
			Memory: fmt.Sprintf("%dGi", reqRAM),
		},
		Limits: orcv1.StarRocksResourcePair{
			CPU:    strconv.Itoa(int(minCPU)),
			Memory: fmt.Sprintf("%dGi", minRAM),
		},
	}, nil
}

func mergeResourceSpec(base, override *orcv1.StarRocksResourceSpec) *orcv1.StarRocksResourceSpec {
	if base == nil && override == nil {
		return nil
	}
	if base == nil {
		copy := *override
		return &copy
	}
	out := *base
	if override == nil {
		return &out
	}
	if override.Requests.CPU != "" {
		out.Requests.CPU = override.Requests.CPU
	}
	if override.Requests.Memory != "" {
		out.Requests.Memory = override.Requests.Memory
	}
	if override.Limits.CPU != "" {
		out.Limits.CPU = override.Limits.CPU
	}
	if override.Limits.Memory != "" {
		out.Limits.Memory = override.Limits.Memory
	}
	return &out
}

func overlappingNodes(feNodeIDs, beNodeIDs []string) []string {
	feSet := map[string]struct{}{}
	for _, nodeID := range feNodeIDs {
		feSet[nodeID] = struct{}{}
	}
	out := make([]string, 0)
	for _, nodeID := range beNodeIDs {
		if _, ok := feSet[nodeID]; ok {
			out = append(out, nodeID)
		}
	}
	sort.Strings(out)
	return out
}

func resolvedStarRocksNamespace(clusterName, configuredNamespace string) string {
	if configuredNamespace != "" {
		return configuredNamespace
	}
	return util.SafeName("starrocks", clusterName)
}

func (r *StarRocksVirtualClusterReconciler) applyStarRocksClusterResource(
	ctx context.Context,
	clusterName string,
	namespace string,
	feNodeNames []string,
	beNodeNames []string,
	cfg orcv1.StarRocksClusterConfig,
) error {
	desired := buildStarRocksClusterResource(clusterName, namespace, feNodeNames, beNodeNames, cfg)

	current := &unstructured.Unstructured{}
	current.SetAPIVersion(starRocksClusterAPIVersion)
	current.SetKind(starRocksClusterKind)
	if err := r.Get(ctx, types.NamespacedName{Name: clusterName, Namespace: namespace}, current); err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}
		return r.Create(ctx, desired)
	}

	labels := current.GetLabels()
	if labels == nil {
		labels = map[string]string{}
	}
	updated := false
	for k, v := range desired.GetLabels() {
		if labels[k] != v {
			labels[k] = v
			updated = true
		}
	}
	if !reflect.DeepEqual(current.Object["spec"], desired.Object["spec"]) {
		current.Object["spec"] = desired.Object["spec"]
		updated = true
	}
	if !updated {
		return nil
	}
	current.SetLabels(labels)
	return r.Update(ctx, current)
}

func (r *StarRocksVirtualClusterReconciler) deleteStarRocksClusterResource(ctx context.Context, clusterName, namespace string) error {
	resource := &unstructured.Unstructured{}
	resource.SetAPIVersion(starRocksClusterAPIVersion)
	resource.SetKind(starRocksClusterKind)
	resource.SetName(clusterName)
	resource.SetNamespace(namespace)
	if err := r.Delete(ctx, resource); err != nil && !apierrors.IsNotFound(err) {
		return err
	}
	return nil
}

func (r *StarRocksVirtualClusterReconciler) getStarRocksClusterStatus(ctx context.Context, clusterName, namespace string) (string, string, error) {
	resource := &unstructured.Unstructured{}
	resource.SetAPIVersion(starRocksClusterAPIVersion)
	resource.SetKind(starRocksClusterKind)
	if err := r.Get(ctx, types.NamespacedName{Name: clusterName, Namespace: namespace}, resource); err != nil {
		return "", "", err
	}
	phase, _, _ := unstructured.NestedString(resource.Object, "status", "phase")
	reason, _, _ := unstructured.NestedString(resource.Object, "status", "reason")
	return phase, reason, nil
}

func buildStarRocksClusterResource(
	clusterName string,
	namespace string,
	feNodeNames []string,
	beNodeNames []string,
	cfg orcv1.StarRocksClusterConfig,
) *unstructured.Unstructured {
	return &unstructured.Unstructured{
		Object: map[string]interface{}{
			"apiVersion": starRocksClusterAPIVersion,
			"kind":       starRocksClusterKind,
			"metadata": map[string]interface{}{
				"name":      clusterName,
				"namespace": namespace,
				"labels": map[string]interface{}{
					util.LabelManagedBy:          util.ManagedByValue,
					util.LabelStarRocksClusterID: clusterName,
				},
			},
			"spec": map[string]interface{}{
				"starRocksFeSpec": buildStarRocksComponentSpec(clusterName, "fe", cfg.FeImage, int32(len(feNodeNames)), feNodeNames, cfg.FeResources),
				"starRocksBeSpec": buildStarRocksComponentSpec(clusterName, "be", cfg.BeImage, int32(len(beNodeNames)), beNodeNames, cfg.BeResources),
			},
		},
	}
}

func buildStarRocksComponentSpec(
	clusterName string,
	component string,
	image string,
	replicas int32,
	nodeNames []string,
	resources *orcv1.StarRocksResourceSpec,
) map[string]interface{} {
	componentSpec := map[string]interface{}{
		"image":    image,
		"replicas": replicas,
		"podLabels": map[string]interface{}{
			util.LabelManagedBy:          util.ManagedByValue,
			util.LabelStarRocksClusterID: clusterName,
			util.LabelStarRocksComponent: component,
		},
		"nodeSelector": map[string]interface{}{
			util.LabelStarRocksClusterID: clusterName,
		},
		"tolerations": []interface{}{
			map[string]interface{}{
				"key":      util.TaintKeyStarRocksCluster,
				"operator": "Equal",
				"value":    clusterName,
				"effect":   "NoSchedule",
			},
		},
	}

	if len(nodeNames) > 0 {
		componentSpec["affinity"] = buildStarRocksAffinity(clusterName, component, nodeNames)
	}
	if resources != nil {
		if resources.Requests.CPU != "" || resources.Requests.Memory != "" {
			componentSpec["requests"] = map[string]interface{}{
				"cpu":    resources.Requests.CPU,
				"memory": resources.Requests.Memory,
			}
		}
		if resources.Limits.CPU != "" || resources.Limits.Memory != "" {
			componentSpec["limits"] = map[string]interface{}{
				"cpu":    resources.Limits.CPU,
				"memory": resources.Limits.Memory,
			}
		}
	}
	switch component {
	case "fe":
		componentSpec["feEnvVars"] = buildAWSChecksumEnvVars()
	case "be":
		componentSpec["beEnvVars"] = buildAWSChecksumEnvVars()
	}
	return componentSpec
}

func buildAWSChecksumEnvVars() []interface{} {
	return []interface{}{
		map[string]interface{}{
			"name":  "AWS_REQUEST_CHECKSUM_CALCULATION",
			"value": awsChecksumCalculationWhenNeeded,
		},
		map[string]interface{}{
			"name":  "AWS_RESPONSE_CHECKSUM_VALIDATION",
			"value": awsChecksumCalculationWhenNeeded,
		},
	}
}

func buildStarRocksAffinity(clusterName, component string, nodeNames []string) map[string]interface{} {
	values := make([]interface{}, 0, len(nodeNames))
	for _, nodeName := range nodeNames {
		values = append(values, nodeName)
	}
	return map[string]interface{}{
		"nodeAffinity": map[string]interface{}{
			"requiredDuringSchedulingIgnoredDuringExecution": map[string]interface{}{
				"nodeSelectorTerms": []interface{}{
					map[string]interface{}{
						"matchExpressions": []interface{}{
							map[string]interface{}{
								"key":      "kubernetes.io/hostname",
								"operator": "In",
								"values":   values,
							},
						},
					},
				},
			},
		},
		"podAntiAffinity": map[string]interface{}{
			"requiredDuringSchedulingIgnoredDuringExecution": []interface{}{
				map[string]interface{}{
					"labelSelector": map[string]interface{}{
						"matchLabels": map[string]interface{}{
							util.LabelStarRocksClusterID: clusterName,
							util.LabelStarRocksComponent: component,
						},
					},
					"topologyKey": "kubernetes.io/hostname",
				},
			},
		},
	}
}

func (r *StarRocksVirtualClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&orcv1.StarRocksVirtualCluster{}).
		Complete(r)
}