Merge branch 'main' into support/migrate-config-manager-to-kbagent

Author: leon-ape

cmd/manager/main.go

@@ -148,7 +148,7 @@ func init() {
 	viper.SetDefault("KUBEBLOCKS_SERVICEACCOUNT_NAME", "kubeblocks")
 	viper.SetDefault(constant.CfgKeyCtrlrMgrNS, "default")
 	viper.SetDefault(constant.CfgHostPortConfigMapName, "kubeblocks-host-ports")
-	viper.SetDefault(constant.CfgHostPortIncludeRanges, "1025-65536")
+	viper.SetDefault(constant.CfgHostPortIncludeRanges, "55000-59999")
 	viper.SetDefault(constant.CfgHostPortExcludeRanges, "6443,10250,10257,10259,2379-2380,30000-32767")
 	viper.SetDefault(constant.KubernetesClusterDomainEnv, constant.DefaultDNSDomain)
 	viper.SetDefault(instanceset.MaxPlainRevisionCount, 1024)

controllers/operations/opsdefinition_controller.go

@@ -23,8 +23,12 @@ import (
 	"context"
 	"text/template"
 
+	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/apiextensions-apiserver/pkg/apiserver/validation"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/kube-openapi/pkg/validation/spec"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
@@ -74,13 +78,19 @@ func (r *OpsDefinitionReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 			continue
 		}
 		if _, err = template.New("opsDefTemplate").Parse(v.Rule.Expression); err != nil {
-			if patchErr := r.updateStatusUnavailable(reqCtx, opsDef, err); patchErr != nil {
-				return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "")
-			}
-			return intctrlutil.Reconciled()
+			return r.updateStatusUnavailable(reqCtx, opsDef, err)
+		}
+	}
+	if opsDef.Spec.ParametersSchema != nil {
+		out := &apiextensions.JSONSchemaProps{}
+		if err = apiextensionsv1.Convert_v1_JSONSchemaProps_To_apiextensions_JSONSchemaProps(opsDef.Spec.ParametersSchema.OpenAPIV3Schema, out, nil); err != nil {
+			return r.updateStatusUnavailable(reqCtx, opsDef, err)
+		}
+		openapiSchema := &spec.Schema{}
+		if err = validation.ConvertJSONSchemaPropsWithPostProcess(out, openapiSchema, validation.StripUnsupportedFormatsPostProcess); err != nil {
+			return r.updateStatusUnavailable(reqCtx, opsDef, err)
 		}
 	}
-
 	// TODO: check serviceKind, connectionCredentialName and serviceName
 	statusPatch := client.MergeFrom(opsDef.DeepCopy())
 	opsDef.Status.ObservedGeneration = opsDef.Generation
@@ -92,12 +102,15 @@ func (r *OpsDefinitionReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 	return intctrlutil.Reconciled()
 }
 
-func (r *OpsDefinitionReconciler) updateStatusUnavailable(reqCtx intctrlutil.RequestCtx, opsDef *opsv1alpha1.OpsDefinition, err error) error {
+func (r *OpsDefinitionReconciler) updateStatusUnavailable(reqCtx intctrlutil.RequestCtx, opsDef *opsv1alpha1.OpsDefinition, err error) (ctrl.Result, error) {
 	statusPatch := client.MergeFrom(opsDef.DeepCopy())
 	opsDef.Status.Phase = opsv1alpha1.UnavailablePhase
 	opsDef.Status.ObservedGeneration = opsDef.Generation
 	opsDef.Status.Message = err.Error()
-	return r.Client.Status().Patch(reqCtx.Ctx, opsDef, statusPatch)
+	if err = r.Client.Status().Patch(reqCtx.Ctx, opsDef, statusPatch); err != nil {
+		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "")
+	}
+	return intctrlutil.Reconciled()
 }
 
 // SetupWithManager sets up the controller with the Manager.

deploy/helm/values.yaml

@@ -541,10 +541,21 @@ hostPorts:
   # The TCP/IP port numbers below 1024 are special in that normal users are not allowed to run servers on them.
   # This is a security feaure, in that if you connect to a service on one of these ports you can be fairly sure
   # that you have the real thing, and not a fake which some hacker has put up for you.
+  # =================================================================================
+  # Default ports update in KubeBlocks 1.1
+  # Ports 49152–65535 is defined by IANA as the private (dynamic) port range,
+  # but this range may conflict with the node’s ephemeral ports.
+  # Therefore KubeBlocks reserves 55000–59999 as a dedicated hostPort range
+  # to reduce port collision risks and improve cluster stability.
+  # Users are recommended to update include/exclude ranges on demand
+  # =================================================================================
   include:
-  - "1025-65536"
+  - "55000-59999"
   # https://kubernetes.io/docs/reference/networking/ports-and-protocols/
   # exclude ports used by kubernetes
+  # =================================================================================
+  # Following ports are not in the 'include' range but are still kept as examples
+  # =================================================================================
   exclude:
   - "6443"
   - "10250"

pkg/controller/kubebuilderx/controller.go

@@ -164,7 +164,7 @@ func (c *controller) emitFailureEvent() {
 		return
 	}
 	// TODO(free6om): make error message user-friendly
-	c.tree.EventRecorder.Eventf(c.tree.GetRoot(), corev1.EventTypeWarning, "FailedReconcile", "reconcile failed: %s", c.err.Error())
+	c.tree.EventRecorder.Eventf(c.tree.GetRoot(), corev1.EventTypeWarning, "FailedReconcile", "%s", c.err.Error())
 }
 
 func NewController(ctx context.Context, cli client.Client, req ctrl.Request, recorder record.EventRecorder, logger logr.Logger) Controller {

pkg/controller/kubebuilderx/plan_builder.go

@@ -30,9 +30,9 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
@@ -265,8 +265,8 @@ func (b *PlanBuilder) defaultWalkFunc(v graph.Vertex) error {
 
 func (b *PlanBuilder) createObject(ctx context.Context, vertex *model.ObjectVertex) error {
 	err := b.cli.Create(ctx, vertex.Obj, clientOption(vertex))
-	if err != nil && !apierrors.IsAlreadyExists(err) {
-		return err
+	if err != nil {
+		return fmt.Errorf("create %T %s failed: %w", vertex.Obj, klog.KObj(vertex.Obj), err)
 	}
 	b.emitEvent(vertex.Obj, "SuccessfulCreate", model.CREATE)
 	return nil
@@ -282,8 +282,8 @@ func (b *PlanBuilder) updateObject(ctx context.Context, vertex *model.ObjectVert
 		err = b.cli.Update(ctx, vertex.Obj, clientOption(vertex))
 		reason = "SuccessfulUpdate"
 	}
-	if err != nil && !apierrors.IsNotFound(err) {
-		return err
+	if err != nil {
+		return fmt.Errorf("update %T %s failed: %w", vertex.Obj, klog.KObj(vertex.Obj), err)
 	}
 	b.emitEvent(vertex.Obj, reason, model.UPDATE)
 	return nil
@@ -300,8 +300,8 @@ func (b *PlanBuilder) patchObject(ctx context.Context, vertex *model.ObjectVerte
 		err = b.cli.Patch(ctx, vertex.Obj, patch, clientOption(vertex))
 		reason = "SuccessfulPatch"
 	}
-	if err != nil && !apierrors.IsNotFound(err) {
-		return err
+	if err != nil {
+		return fmt.Errorf("patch %T %s failed: %w", vertex.Obj, klog.KObj(vertex.Obj), err)
 	}
 	b.emitEvent(vertex.Obj, reason, model.PATCH)
 	return nil
@@ -314,15 +314,14 @@ func (b *PlanBuilder) deleteObject(ctx context.Context, vertex *model.ObjectVert
 	}
 	if len(finalizer) > 0 && controllerutil.RemoveFinalizer(vertex.Obj, finalizer) {
 		err := b.cli.Update(ctx, vertex.Obj, clientOption(vertex))
-		if err != nil && !apierrors.IsNotFound(err) {
-			b.transCtx.logger.Error(err, fmt.Sprintf("delete %T error: %s", vertex.Obj, vertex.Obj.GetName()))
-			return err
+		if err != nil {
+			return fmt.Errorf("delete %T %s failed: %w", vertex.Obj, klog.KObj(vertex.Obj), err)
 		}
 	}
 	if !model.IsObjectDeleting(vertex.Obj) {
 		err := b.cli.Delete(ctx, vertex.Obj, clientOption(vertex))
-		if err != nil && !apierrors.IsNotFound(err) {
-			return err
+		if err != nil {
+			return fmt.Errorf("delete %T %s failed: %w", vertex.Obj, klog.KObj(vertex.Obj), err)
 		}
 		b.emitEvent(vertex.Obj, "SuccessfulDelete", model.DELETE)
 	}
@@ -331,7 +330,7 @@ func (b *PlanBuilder) deleteObject(ctx context.Context, vertex *model.ObjectVert
 
 func (b *PlanBuilder) statusObject(ctx context.Context, vertex *model.ObjectVertex) error {
 	if err := b.cli.Status().Update(ctx, vertex.Obj, clientOption(vertex)); err != nil {
-		return err
+		return fmt.Errorf("status %T %s failed: %w", vertex.Obj, klog.KObj(vertex.Obj), err)
 	}
 	return nil
 }