feat(obtenant): added support for inter-obcluster standby tenant setup

oceanbase · Apr 1, 2024 · 66e7fd8 · 66e7fd8
1 parent 113a2c0
commit 66e7fd8
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 6 deletions.
diff --git a/api/v1alpha1/obtenant_webhook.go b/api/v1alpha1/obtenant_webhook.go
@@ -214,25 +214,42 @@ func (r *OBTenant) validateMutation() error {
 			allErrs = append(allErrs, field.Invalid(field.NewPath("spec").Child("tenantRole"), r.Spec.TenantRole, "Standby must have a source option, but both restore and tenantRef are nil now"))
 		} else if r.Spec.Source.Tenant != nil {
 			tenant := &OBTenant{}
+			ns := r.GetNamespace()
+			tenantCR := *r.Spec.Source.Tenant
+			splits := strings.Split(*r.Spec.Source.Tenant, "/")
+			switch len(splits) {
+			case 0, 1:
+			case 2:
+				if splits[0] == "" {
+					return field.Invalid(field.NewPath("spec").Child("source").Child("tenant"), tenantCR, "Given tenant namespace is empty")
+				}
+				if splits[1] == "" {
+					return field.Invalid(field.NewPath("spec").Child("source").Child("tenant"), tenantCR, "Given tenant name is empty")
+				}
+				ns = splits[0]
+				tenantCR = splits[1]
+			default:
+				return field.Invalid(field.NewPath("spec").Child("source").Child("tenant"), tenantCR, "Given tenant name is invalid, it should be namespace/name or name format")
+			}
 			err = tenantClt.Get(context.TODO(), types.NamespacedName{
-				Namespace: r.GetNamespace(),
-				Name:      *r.Spec.Source.Tenant,
+				Namespace: ns,
+				Name:      tenantCR,
 			}, tenant)
 			if err != nil {
 				if apierrors.IsNotFound(err) {
-					allErrs = append(allErrs, field.Invalid(field.NewPath("spec").Child("source").Child("tenant"), r.Spec.Source.Tenant, "Given tenant not found in namespace"+r.GetNamespace()))
+					allErrs = append(allErrs, field.Invalid(field.NewPath("spec").Child("source").Child("tenant"), tenantCR, "Given tenant not found in namespace "+ns))
 				} else {
 					allErrs = append(allErrs, field.InternalError(field.NewPath("spec").Child("source").Child("tenant"), err))
 				}
 			}
 			cluster := &OBCluster{}
 			err = tenantClt.Get(context.Background(), types.NamespacedName{
-				Namespace: r.GetNamespace(),
+				Namespace: ns,
 				Name:      tenant.Spec.ClusterName,
 			}, cluster)
 			if err != nil {
 				if apierrors.IsNotFound(err) {
-					allErrs = append(allErrs, field.Invalid(field.NewPath("spec").Child("source").Child("tenant"), r.Spec.Source.Tenant, "Given tenant not found in namespace"+r.GetNamespace()))
+					allErrs = append(allErrs, field.Invalid(field.NewPath("spec").Child("source").Child("tenant"), tenantCR, "Given tenant not found in namespace "+ns))
 				} else {
 					allErrs = append(allErrs, field.InternalError(field.NewPath("spec").Child("source").Child("tenant"), err))
 				}

diff --git a/internal/resource/obtenant/obtenant_task.go b/internal/resource/obtenant/obtenant_task.go
@@ -16,6 +16,7 @@ package obtenant
 import (
 	"fmt"
 	"reflect"
+	"strings"
 	"time"
 
 	"github.com/pkg/errors"
@@ -219,7 +220,14 @@ func CreateEmptyStandbyTenant(m *OBTenantManager) tasktypes.TaskError {
 	if err != nil {
 		return err
 	}
-	restoreSource, err := resourceutils.GetTenantRestoreSource(m.Ctx, m.Client, m.Logger, con, m.OBTenant.Namespace, *m.OBTenant.Spec.Source.Tenant)
+	ns := m.OBTenant.GetNamespace()
+	tenantCRName := *m.OBTenant.Spec.Source.Tenant
+	splits := strings.Split(*m.OBTenant.Spec.Source.Tenant, "/")
+	if len(splits) == 2 {
+		ns = splits[0]
+		tenantCRName = splits[1]
+	}
+	restoreSource, err := resourceutils.GetTenantRestoreSource(m.Ctx, m.Client, m.Logger, con, ns, tenantCRName)
 	if err != nil {
 		return err
 	}