[fix] Reconcile exception

pkg/common/utils/k8s/client.go

@@ -46,13 +46,20 @@ type StatefulSetEqual func(st1 *appv1.StatefulSet, st2 *appv1.StatefulSet) bool
 func ApplyService(ctx context.Context, k8sclient client.Client, svc *corev1.Service, equal ServiceEqual) error {
 	// As stated in the RetryOnConflict's documentation, the returned error shouldn't be wrapped.
 	var esvc corev1.Service
+	//avoid clusterIps Invalid value failed.
+	svc.Spec.ClusterIPs = nil
 	err := k8sclient.Get(ctx, types.NamespacedName{Name: svc.Name, Namespace: svc.Namespace}, &esvc)
 	if err != nil && apierrors.IsNotFound(err) {
-		return CreateClientObject(ctx, k8sclient, svc)
-	} else if err != nil {
+		//avoid client version not match k8s version will result resourceVersion is not "" and response "resourceVersion should not set"  failed.
+		svc.ResourceVersion = ""
+		if err = CreateClientObject(ctx, k8sclient, svc); err == nil || apierrors.IsAlreadyExists(err) {
+			return nil
+		}
+
+		return err
+	} else if err != nil && apierrors.IsNotFound(err) {
 		return err
 	}
-
 	if equal(svc, &esvc) {
 		klog.Info("CreateOrUpdateService service Name, Ports, Selector, ServiceType, Labels have not change ", "namespace ", svc.Namespace, " name ", svc.Name)
 		return nil
@@ -69,7 +76,7 @@ func ApplyStatefulSet(ctx context.Context, k8sclient client.Client, st *appv1.St
 	err := k8sclient.Get(ctx, types.NamespacedName{Namespace: st.Namespace, Name: st.Name}, &est)
 	if err != nil && apierrors.IsNotFound(err) {
 		return CreateClientObject(ctx, k8sclient, st)
-	} else if err != nil {
+	} else if err != nil && !apierrors.IsNotFound(err) {
 		return err
 	}
 
@@ -80,7 +87,11 @@ func ApplyStatefulSet(ctx context.Context, k8sclient client.Client, st *appv1.St
 	}
 
 	st.ResourceVersion = est.ResourceVersion
-	return PatchClientObject(ctx, k8sclient, st)
+	err = PatchClientObject(ctx, k8sclient, st)
+	if err == nil || apierrors.IsConflict(err) {
+		return nil
+	}
+	return err
 }
 
 func CreateClientObject(ctx context.Context, k8sclient client.Client, object client.Object) error {
@@ -100,7 +111,7 @@ func UpdateClientObject(ctx context.Context, k8sclient client.Client, object cli
 }
 
 func CreateOrUpdateClientObject(ctx context.Context, k8sclient client.Client, object client.Object) error {
-	klog.V(4).Infof("create or update resource namespace=%s,name=%s,kind=%s.", object.GetNamespace(), object.GetName(), object.GetObjectKind())
+	klog.Infof("create or update resource namespace=%s,name=%s,kind=%s.", object.GetNamespace(), object.GetName(), object.GetObjectKind())
 	if err := k8sclient.Update(ctx, object); apierrors.IsNotFound(err) {
 		return k8sclient.Create(ctx, object)
 	} else if err != nil {
@@ -112,7 +123,7 @@ func CreateOrUpdateClientObject(ctx context.Context, k8sclient client.Client, ob
 
 // PatchClientObject patch object when the object exist. if not return error.
 func PatchClientObject(ctx context.Context, k8sclient client.Client, object client.Object) error {
-	klog.V(4).Infof("patch resource namespace=%s,name=%s,kind=%s.", object.GetNamespace(), object.GetName(), object.GetObjectKind())
+	klog.Infof("patch resource namespace=%s,name=%s,kind=%s.", object.GetNamespace(), object.GetName(), object.GetObjectKind())
 	if err := k8sclient.Patch(ctx, object, client.Merge); err != nil {
 		return err
 	}

pkg/common/utils/resource/statefulset.go

@@ -89,7 +89,7 @@ func NewStatefulSet(dcr *v1.DorisCluster, componentType v1.ComponentType) appv1.
 
 // StatefulSetDeepEqual judge two statefulset equal or not.
 func StatefulSetDeepEqual(new *appv1.StatefulSet, old *appv1.StatefulSet, excludeReplicas bool) bool {
-	equal := StatefulsetDeepEqualWithOmitKey(new, old, v1.ComponentResourceHash, false, excludeReplicas)
+	equal := StatefulsetDeepEqualWithKey(new, old, v1.ComponentResourceHash, excludeReplicas)
 	if !equal {
 		return clear_config_env_path_numbers_alwaysEquals(new, old, v1.ComponentResourceHash, excludeReplicas)
 
@@ -179,14 +179,14 @@ func clear_config_env_path_numbers_alwaysEquals(new *appv1.StatefulSet, old *app
 	}
 }
 
-func StatefulsetDeepEqualWithOmitKey(new, old *appv1.StatefulSet, annoKey string, omit bool, excludeReplicas bool) bool {
-	if omit {
+func StatefulsetDeepEqualWithKey(new, old *appv1.StatefulSet, annoKey string, excludeReplicas bool) bool {
+	/*	if omit {
 		newHso := statefulSetHashObject(new, excludeReplicas)
 		newHashv := hash.HashObject(newHso)
 		oldHso := statefulSetHashObject(old, excludeReplicas)
 		oldHashv := hash.HashObject(oldHso)
 		return new.Namespace == old.Namespace && newHashv == oldHashv
-	}
+	}*/
 	var newHashv, oldHashv string
 	if annoKey == "" {
 		annoKey = v1.ComponentResourceHash

pkg/controller/controllers_utils.go

@@ -25,7 +25,7 @@ import (
 )
 
 func disAggregatedInconsistentStatus(ests *dv1.DorisDisaggregatedClusterStatus, ddc *dv1.DorisDisaggregatedCluster) bool {
-	return reflect.DeepEqual(ests, ddc.Status)
+	return reflect.DeepEqual(ests, &ddc.Status)
 }
 
 func inconsistentStatus(status *v1.DorisClusterStatus, dcr *v1.DorisCluster) bool {

pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go

@@ -176,7 +176,7 @@ func (dcgs *DisaggregatedComputeGroupsController) reconcileStatefulset(ctx conte
 	scaleType := getScaleType(st, &est, cgStatus.Phase)
 
 	if err := k8s.ApplyStatefulSet(ctx, dcgs.K8sclient, st, func(st, est *appv1.StatefulSet) bool {
-		return resource.StatefulsetDeepEqualWithOmitKey(st, est, dv1.DisaggregatedSpecHashValueAnnotation, true, false)
+		return resource.StatefulsetDeepEqualWithKey(st, est, dv1.DisaggregatedSpecHashValueAnnotation, false)
 	}); err != nil {
 		klog.Errorf("disaggregatedComputeGroupsController reconcileStatefulset apply statefulset namespace=%s name=%s failed, err=%s", st.Namespace, st.Name, err.Error())
 		return &sc.Event{Type: sc.EventWarning, Reason: sc.CGApplyResourceFailed, Message: err.Error()}, err

pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe/controller.go

@@ -283,7 +283,7 @@ func (dfc *DisaggregatedFEController) reconcileStatefulset(ctx context.Context,
 
 	// apply fe StatefulSet
 	if err := k8s.ApplyStatefulSet(ctx, dfc.K8sclient, st, func(st, est *appv1.StatefulSet) bool {
-		return resource.StatefulsetDeepEqualWithOmitKey(st, est, v1.DisaggregatedSpecHashValueAnnotation, true, false)
+		return resource.StatefulsetDeepEqualWithKey(st, est, v1.DisaggregatedSpecHashValueAnnotation, false)
 	}); err != nil {
 		klog.Errorf("disaggregatedFEController reconcileStatefulset apply statefulset namespace=%s name=%s failed, err=%s", st.Namespace, st.Name, err.Error())
 		return &sc.Event{Type: sc.EventWarning, Reason: sc.FEApplyResourceFailed, Message: err.Error()}, err

pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe/service.go

@@ -117,9 +117,9 @@ func (dfc *DisaggregatedFEController) newInternalService(ddc *dv1.DorisDisaggreg
 }
 
 func getInternalServicePort(config map[string]interface{}) corev1.ServicePort {
-	httpPort := resource.GetPort(config, resource.HTTP_PORT)
+	httpPort := resource.GetPort(config, resource.QUERY_PORT)
 	return corev1.ServicePort{
-		Port: httpPort, TargetPort: intstr.FromInt(int(httpPort)), Name: resource.GetPortKey(resource.HTTP_PORT),
+		Port: httpPort, TargetPort: intstr.FromInt(int(httpPort)), Name: resource.GetPortKey(resource.QUERY_PORT),
 	}
 }
 

pkg/controller/sub_controller/disaggregated_cluster/metaservice/controller.go

@@ -149,7 +149,7 @@ func (dms *DisaggregatedMSController) reconcileStatefulset(ctx context.Context,
 	}
 
 	if err := k8s.ApplyStatefulSet(ctx, dms.K8sclient, st, func(st, est *appv1.StatefulSet) bool {
-		return resource.StatefulsetDeepEqualWithOmitKey(st, est, v1.DisaggregatedSpecHashValueAnnotation, true, false)
+		return resource.StatefulsetDeepEqualWithKey(st, est, v1.DisaggregatedSpecHashValueAnnotation, false)
 	}); err != nil {
 		klog.Errorf("dms controller reconcileStatefulset apply statefulset namespace=%s name=%s failed, err=%s", st.Namespace, st.Name, err.Error())
 		return &sc.Event{Type: sc.EventWarning, Reason: sc.CGApplyResourceFailed, Message: err.Error()}, err

pkg/controller/sub_controller/disaggregated_subcontroller.go

@@ -32,7 +32,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -206,17 +205,6 @@ func (d *DisaggregatedSubDefaultController) newDefaultAffinity(matchKey, value s
 
 // the common logic to apply service, will used by fe,be,ms.
 func (d *DisaggregatedSubDefaultController) DefaultReconcileService(ctx context.Context, svc *corev1.Service) (*Event, error) {
-	var esvc corev1.Service
-	if err := d.K8sclient.Get(ctx, types.NamespacedName{Namespace: svc.Namespace, Name: svc.Name}, &esvc); apierrors.IsNotFound(err) {
-		if err = k8s.CreateClientObject(ctx, d.K8sclient, svc); err != nil {
-			klog.Errorf("disaggregatedSubDefaultController reconcileService create service namespace=%s name=%s failed, err=%s", svc.Namespace, svc.Name, err.Error())
-			return &Event{Type: EventWarning, Reason: ServiceApplyedFailed, Message: err.Error()}, err
-		}
-	} else if err != nil {
-		klog.Errorf("disaggregatedSubDefaultController reconcileService get service failed, namespace=%s name=%s failed, err=%s", svc.Namespace, svc.Name, err.Error())
-		return nil, err
-	}
-
 	if err := k8s.ApplyService(ctx, d.K8sclient, svc, func(nsvc, osvc *corev1.Service) bool {
 		return resource.ServiceDeepEqualWithAnnoKey(nsvc, osvc, v1.DisaggregatedSpecHashValueAnnotation)
 	}); err != nil {