Merge remote-tracking branch 'upstream/master'

api/v1alpha1/helpers.go

@@ -0,0 +1,77 @@
+/*
+Copyright (c) 2024 OceanBase
+ob-operator is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details.
+*/
+
+package v1alpha1
+
+import (
+	"context"
+
+	"github.com/go-logr/logr"
+	"github.com/pkg/errors"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	oceanbaseconst "github.com/oceanbase/ob-operator/internal/const/oceanbase"
+	clusterstatus "github.com/oceanbase/ob-operator/internal/const/status/obcluster"
+	"github.com/oceanbase/ob-operator/pkg/oceanbase-sdk/connector"
+	"github.com/oceanbase/ob-operator/pkg/oceanbase-sdk/operation"
+)
+
+func readPassword(c client.Client, namespace, secretName string) (string, error) {
+	secret := &corev1.Secret{}
+	err := c.Get(context.Background(), types.NamespacedName{
+		Namespace: namespace,
+		Name:      secretName,
+	}, secret)
+	if err != nil {
+		return "", errors.Wrapf(err, "Get password from secret %s failed", secretName)
+	}
+	return string(secret.Data["password"]), err
+}
+
+func getSysClient(c client.Client, logger *logr.Logger, obcluster *OBCluster, userName, tenantName, secretName string) (*operation.OceanbaseOperationManager, error) {
+	observerList := &OBServerList{}
+	err := c.List(context.Background(), observerList, client.MatchingLabels{
+		oceanbaseconst.LabelRefOBCluster: obcluster.Name,
+	}, client.InNamespace(obcluster.Namespace))
+	if err != nil {
+		return nil, errors.Wrap(err, "Get observer list")
+	}
+	if len(observerList.Items) == 0 {
+		return nil, errors.Errorf("No observer belongs to cluster %s", obcluster.Name)
+	}
+
+	var s *connector.OceanBaseDataSource
+	password, err := readPassword(c, obcluster.Namespace, secretName)
+	if err != nil {
+		return nil, errors.Wrapf(err, "Read password to get oceanbase operation manager of cluster %s", obcluster.Name)
+	}
+	for _, observer := range observerList.Items {
+		address := observer.Status.GetConnectAddr()
+		switch obcluster.Status.Status {
+		case clusterstatus.New:
+			s = connector.NewOceanBaseDataSource(address, oceanbaseconst.SqlPort, oceanbaseconst.RootUser, tenantName, "", "")
+		case clusterstatus.Bootstrapped:
+			s = connector.NewOceanBaseDataSource(address, oceanbaseconst.SqlPort, oceanbaseconst.RootUser, tenantName, "", oceanbaseconst.DefaultDatabase)
+		default:
+			s = connector.NewOceanBaseDataSource(address, oceanbaseconst.SqlPort, userName, tenantName, password, oceanbaseconst.DefaultDatabase)
+		}
+		// if err is nil, db connection is already checked available
+		sysClient, err := operation.GetOceanbaseOperationManager(s)
+		if err == nil && sysClient != nil {
+			sysClient.Logger = logger
+			return sysClient, nil
+		}
+	}
+	return nil, errors.Errorf("Can not get oceanbase operation manager of obcluster %s after checked all servers", obcluster.Name)
+}

api/v1alpha1/obcluster_types.go

@@ -59,6 +59,9 @@ type OBClusterStatus struct {
 //+kubebuilder:subresource:status
 //+kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.status"
 //+kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"
+//+kubebuilder:printcolumn:name="Tasks",type="string",JSONPath=".status.operationContext.tasks",priority=1
+//+kubebuilder:printcolumn:name="Task",type="string",JSONPath=".status.operationContext.task",priority=1
+//+kubebuilder:printcolumn:name="TaskIdx",type="string",JSONPath=".status.operationContext.idx",priority=1
 
 // OBCluster is the Schema for the obclusters API
 type OBCluster struct {

api/v1alpha1/obcluster_webhook.go

@@ -39,6 +39,7 @@ import (
 	apitypes "github.com/oceanbase/ob-operator/api/types"
 	obcfg "github.com/oceanbase/ob-operator/internal/config/operator"
 	oceanbaseconst "github.com/oceanbase/ob-operator/internal/const/oceanbase"
+	clusterstatus "github.com/oceanbase/ob-operator/internal/const/status/obcluster"
 )
 
 // log is for logging in this package.
@@ -50,8 +51,6 @@ func (r *OBCluster) SetupWebhookWithManager(mgr ctrl.Manager) error {
 		Complete()
 }
 
-// TODO(user): EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
-
 //+kubebuilder:webhook:path=/mutate-oceanbase-oceanbase-com-v1alpha1-obcluster,mutating=true,failurePolicy=fail,sideEffects=None,groups=oceanbase.oceanbase.com,resources=obclusters,verbs=create;update,versions=v1alpha1,name=mobcluster.kb.io,admissionReviewVersions=v1
 
 var _ webhook.Defaulter = &OBCluster{}
@@ -186,6 +185,42 @@ func (r *OBCluster) ValidateUpdate(old runtime.Object) (admission.Warnings, erro
 	} else if !oldCluster.SupportStaticIP() && (oldResource.Cpu != newResource.Cpu || oldResource.Memory != newResource.Memory) {
 		return nil, errors.New("forbid to modify cpu or memory quota of non-static-ip cluster")
 	}
+	if newResource.Memory.Cmp(oldResource.Memory) < 0 || newResource.Cpu.Cmp(oldResource.Cpu) < 0 {
+		if r.Status.Status != clusterstatus.Running {
+			return nil, errors.New("forbid to shrink memory size of non-running cluster")
+		}
+		conn, err := getSysClient(clt, &obclusterlog, r, oceanbaseconst.OperatorUser, oceanbaseconst.SysTenant, r.Spec.UserSecrets.Operator)
+		if err != nil {
+			return nil, err
+		}
+		defer conn.Close()
+		gvservers, err := conn.ListGVServers(context.Background())
+		if err != nil {
+			return nil, err
+		}
+		var maxAssignedCPU int64
+		var maxAssignedMemory int64
+		var memoryLimitPercent float64
+		for _, gvserver := range gvservers {
+			if gvserver.MemAssigned > maxAssignedMemory {
+				if oldResource.Memory.Value() < gvserver.MemoryLimit {
+					memoryLimitPercent = 0.9
+				} else {
+					memoryLimitPercent = float64(gvserver.MemoryLimit) / oldResource.Memory.AsApproximateFloat64()
+				}
+				maxAssignedMemory = gvserver.MemAssigned
+			}
+			if gvserver.CPUAssigned > maxAssignedCPU {
+				maxAssignedCPU = gvserver.CPUAssigned
+			}
+		}
+		if newResource.Memory.AsApproximateFloat64()*memoryLimitPercent < float64(maxAssignedMemory) {
+			return nil, errors.New("Assigned memory is larger than new memory size")
+		}
+		if oldResource.Cpu.Value() > 16 && newResource.Cpu.AsApproximateFloat64() < float64(maxAssignedCPU) {
+			return nil, errors.New("Assigned CPU is larger than new CPU size")
+		}
+	}
 	if r.Spec.BackupVolume == nil && oldCluster.Spec.BackupVolume != nil {
 		return nil, errors.New("forbid to remove backup volume")
 	}
@@ -195,27 +230,30 @@ func (r *OBCluster) ValidateUpdate(old runtime.Object) (admission.Warnings, erro
 			err = errors.New("forbid to add backup volume to non-static-ip cluster")
 		}
 	}
-	if r.Spec.OBServerTemplate.Storage.DataStorage.Size.Cmp(oldCluster.Spec.OBServerTemplate.Storage.DataStorage.Size) > 0 {
-		err = errors.Join(err, r.validateStorageClassAllowExpansion(r.Spec.OBServerTemplate.Storage.DataStorage.StorageClass))
+
+	newStorage := r.Spec.OBServerTemplate.Storage
+	oldStorage := oldCluster.Spec.OBServerTemplate.Storage
+	if newStorage.DataStorage.Size.Cmp(oldStorage.DataStorage.Size) > 0 {
+		err = errors.Join(err, r.validateStorageClassAllowExpansion(newStorage.DataStorage.StorageClass))
 	}
-	if r.Spec.OBServerTemplate.Storage.LogStorage.Size.Cmp(oldCluster.Spec.OBServerTemplate.Storage.LogStorage.Size) > 0 {
-		err = errors.Join(err, r.validateStorageClassAllowExpansion(r.Spec.OBServerTemplate.Storage.LogStorage.StorageClass))
+	if newStorage.LogStorage.Size.Cmp(oldStorage.LogStorage.Size) > 0 {
+		err = errors.Join(err, r.validateStorageClassAllowExpansion(newStorage.LogStorage.StorageClass))
 	}
-	if r.Spec.OBServerTemplate.Storage.RedoLogStorage.Size.Cmp(oldCluster.Spec.OBServerTemplate.Storage.RedoLogStorage.Size) > 0 {
-		err = errors.Join(err, r.validateStorageClassAllowExpansion(r.Spec.OBServerTemplate.Storage.RedoLogStorage.StorageClass))
+	if newStorage.RedoLogStorage.Size.Cmp(oldStorage.RedoLogStorage.Size) > 0 {
+		err = errors.Join(err, r.validateStorageClassAllowExpansion(newStorage.RedoLogStorage.StorageClass))
 	}
 	if err != nil {
 		return nil, err
 	}
 
-	if r.Spec.OBServerTemplate.Storage.DataStorage.Size.Cmp(oldCluster.Spec.OBServerTemplate.Storage.DataStorage.Size) < 0 {
-		err = errors.Join(err, field.Invalid(field.NewPath("spec").Child("observer").Child("storage").Child("dataStorage").Child("size"), r.Spec.OBServerTemplate.Storage.DataStorage.Size.String(), "forbid to shrink data storage size"))
+	if newStorage.DataStorage.Size.Cmp(oldStorage.DataStorage.Size) < 0 {
+		err = errors.Join(err, field.Invalid(field.NewPath("spec").Child("observer").Child("storage").Child("dataStorage").Child("size"), newStorage.DataStorage.Size.String(), "forbid to shrink data storage size"))
 	}
-	if r.Spec.OBServerTemplate.Storage.LogStorage.Size.Cmp(oldCluster.Spec.OBServerTemplate.Storage.LogStorage.Size) < 0 {
-		err = errors.Join(err, field.Invalid(field.NewPath("spec").Child("observer").Child("storage").Child("logStorage").Child("size"), r.Spec.OBServerTemplate.Storage.LogStorage.Size.String(), "forbid to shrink log storage size"))
+	if newStorage.LogStorage.Size.Cmp(oldStorage.LogStorage.Size) < 0 {
+		err = errors.Join(err, field.Invalid(field.NewPath("spec").Child("observer").Child("storage").Child("logStorage").Child("size"), newStorage.LogStorage.Size.String(), "forbid to shrink log storage size"))
 	}
-	if r.Spec.OBServerTemplate.Storage.RedoLogStorage.Size.Cmp(oldCluster.Spec.OBServerTemplate.Storage.RedoLogStorage.Size) < 0 {
-		err = errors.Join(err, field.Invalid(field.NewPath("spec").Child("observer").Child("storage").Child("redoLogStorage").Child("size"), r.Spec.OBServerTemplate.Storage.RedoLogStorage.Size.String(), "forbid to shrink redo log storage size"))
+	if newStorage.RedoLogStorage.Size.Cmp(oldStorage.RedoLogStorage.Size) < 0 {
+		err = errors.Join(err, field.Invalid(field.NewPath("spec").Child("observer").Child("storage").Child("redoLogStorage").Child("size"), newStorage.RedoLogStorage.Size.String(), "forbid to shrink redo log storage size"))
 	}
 	if err != nil {
 		return nil, err

api/v1alpha1/obparameter_types.go

@@ -49,6 +49,8 @@ type OBParameterStatus struct {
 //+kubebuilder:printcolumn:name="ClusterName",type="string",JSONPath=".spec.clusterName"
 //+kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.status"
 //+kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"
+//+kubebuilder:printcolumn:name="Key",type="string",JSONPath=".spec.parameter.name"
+//+kubebuilder:printcolumn:name="WantedValue",type="string",JSONPath=".spec.parameter.value"
 
 // OBParameter is the Schema for the obparameters API
 type OBParameter struct {

config/crd/bases/oceanbase.oceanbase.com_obclusters.yaml

@@ -21,6 +21,18 @@ spec:
     - jsonPath: .metadata.creationTimestamp
       name: Age
       type: date
+    - jsonPath: .status.operationContext.tasks
+      name: Tasks
+      priority: 1
+      type: string
+    - jsonPath: .status.operationContext.task
+      name: Task
+      priority: 1
+      type: string
+    - jsonPath: .status.operationContext.idx
+      name: TaskIdx
+      priority: 1
+      type: string
     name: v1alpha1
     schema:
       openAPIV3Schema:

config/crd/bases/oceanbase.oceanbase.com_obparameters.yaml

@@ -24,6 +24,12 @@ spec:
     - jsonPath: .metadata.creationTimestamp
       name: Age
       type: date
+    - jsonPath: .spec.parameter.name
+      name: Key
+      type: string
+    - jsonPath: .spec.parameter.value
+      name: WantedValue
+      type: string
     name: v1alpha1
     schema:
       openAPIV3Schema:

config/manager/kustomization.yaml

@@ -4,5 +4,5 @@ apiVersion: kustomize.config.k8s.io/v1beta1
 kind: Kustomization
 images:
 - name: controller
-  newName: oceanbase/ob-operator
-  newTag: 2.2.1
+  newName: oceanbasedev/ob-operator
+  newTag: 2.2.2-alpha.2

internal/const/status/obcluster/obcluster_status.go

@@ -13,18 +13,19 @@ See the Mulan PSL v2 for more details.
 package obcluster
 
 const (
-	New                 = "new"
-	MigrateFromExisting = "migrate from existing"
-	Running             = "running"
-	AddOBZone           = "add obzone"
-	DeleteOBZone        = "delete obzone"
-	ModifyOBZoneReplica = "modify obzone replica"
-	Upgrade             = "upgrade"
-	ModifyOBParameter   = "modify parameter"
-	Bootstrapped        = "bootstrapped"
-	FinalizerFinished   = "finalizer finished"
-	ScaleUp             = "scale up"
-	ExpandPVC           = "expand pvc"
-	Failed              = "failed"
-	MountBackupVolume   = "mount backup volume"
+	New                    = "new"
+	MigrateFromExisting    = "migrate from existing"
+	Running                = "running"
+	AddOBZone              = "add obzone"
+	DeleteOBZone           = "delete obzone"
+	ModifyOBZoneReplica    = "modify obzone replica"
+	Upgrade                = "upgrade"
+	ModifyOBParameter      = "modify parameter"
+	Bootstrapped           = "bootstrapped"
+	FinalizerFinished      = "finalizer finished"
+	ScaleUp                = "scale up"
+	ExpandPVC              = "expand pvc"
+	Failed                 = "failed"
+	MountBackupVolume      = "mount backup volume"
+	RollingUpdateOBServers = "rolling update observers"
 )

internal/const/status/obzone/obzone_status.go

@@ -13,17 +13,18 @@ See the Mulan PSL v2 for more details.
 package obzone
 
 const (
-	New                 = "new"
-	MigrateFromExisting = "migrate from existing"
-	Maintaining         = "maintaining"
-	Running             = "running"
-	AddOBServer         = "add observer"
-	DeleteOBServer      = "delete observer"
-	Deleting            = "deleting"
-	Upgrade             = "upgrade"
-	BootstrapReady      = "bootstrap ready"
-	FinalizerFinished   = "finalizer finished"
-	ScaleUp             = "scale up"
-	ExpandPVC           = "expand pvc"
-	MountBackupVolume   = "mount backup volume"
+	New                  = "new"
+	MigrateFromExisting  = "migrate from existing"
+	Maintaining          = "maintaining"
+	Running              = "running"
+	AddOBServer          = "add observer"
+	DeleteOBServer       = "delete observer"
+	Deleting             = "deleting"
+	Upgrade              = "upgrade"
+	BootstrapReady       = "bootstrap ready"
+	FinalizerFinished    = "finalizer finished"
+	ScaleUp              = "scale up"
+	ExpandPVC            = "expand pvc"
+	MountBackupVolume    = "mount backup volume"
+	RollingUpdateServers = "rolling update servers"
 )

internal/resource/obcluster/names.go

@@ -30,6 +30,7 @@ const (
 	fScaleUpOBZones                  ttypes.FlowName = "scale up obzones"
 	fExpandPVC                       ttypes.FlowName = "expand pvc for obcluster"
 	fMountBackupVolume               ttypes.FlowName = "mount backup volume for obcluster"
+	fRollingUpdateOBServers          ttypes.FlowName = "rolling update observers"
 )
 
 // obcluster tasks
@@ -64,5 +65,7 @@ const (
 	tExpandPVC                  ttypes.TaskName = "expand pvc"
 	tMountBackupVolume          ttypes.TaskName = "mount backup volume"
 	tCheckEnvironment           ttypes.TaskName = "check environment"
+	tAdjustParameters           ttypes.TaskName = "adjust parameters"
 	tAnnotateOBCluster          ttypes.TaskName = "annotate obcluster"
+	tRollingUpdateOBZones       ttypes.TaskName = "rolling update observers"
 )

internal/resource/obcluster/obcluster_flow.go

@@ -4,8 +4,10 @@ ob-operator is licensed under Mulan PSL v2.
 You can use this software according to the terms and conditions of the Mulan PSL v2.
 You may obtain a copy of Mulan PSL v2 at:
          http://license.coscl.org.cn/MulanPSL2
-THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
-EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS,
+WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED,
+INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details.
 */
@@ -162,8 +164,8 @@ func genScaleUpOBZonesFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 		OperationContext: &tasktypes.OperationContext{
 			Name: fScaleUpOBZones,
 			Tasks: []tasktypes.TaskName{
+				tAdjustParameters,
 				tScaleUpOBZones,
-				tWaitOBZoneRunning,
 			},
 			TargetStatus: clusterstatus.Running,
 		},
@@ -195,3 +197,15 @@ func genMountBackupVolumeFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 		},
 	}
 }
+
+func genRollingUpdateOBZonesFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
+	return &tasktypes.TaskFlow{
+		OperationContext: &tasktypes.OperationContext{
+			Name: fRollingUpdateOBServers,
+			Tasks: []tasktypes.TaskName{
+				tRollingUpdateOBZones,
+			},
+			TargetStatus: clusterstatus.Running,
+		},
+	}
+}

internal/resource/obcluster/obcluster_manager.go

@@ -109,6 +109,8 @@ func (m *OBClusterManager) GetTaskFlow() (*tasktypes.TaskFlow, error) {
 		taskFlow = genExpandPVCFlow(m)
 	case clusterstatus.MountBackupVolume:
 		taskFlow = genMountBackupVolumeFlow(m)
+	case clusterstatus.RollingUpdateOBServers:
+		taskFlow = genRollingUpdateOBZonesFlow(m)
 	default:
 		m.Logger.V(oceanbaseconst.LogLevelTrace).Info("No need to run anything for obcluster", "obcluster", m.OBCluster.Name)
 		return nil, nil
@@ -186,6 +188,10 @@ func (m *OBClusterManager) UpdateStatus() error {
 					m.OBCluster.Status.Status = clusterstatus.ScaleUp
 					break outer
 				}
+				if m.checkIfStorageClassChange(&obzone) {
+					m.OBCluster.Status.Status = clusterstatus.RollingUpdateOBServers
+					break outer
+				}
 				if m.checkIfStorageSizeExpand(&obzone) {
 					m.OBCluster.Status.Status = clusterstatus.ExpandPVC
 					break outer