package server import ( "context" "strings" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta" dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1" "github.com/tidbcloud/cluster-service-ng/domain/entity" "github.com/tidbcloud/cluster-service-ng/domain/workflow" bractionmocks "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br/mocks" "github.com/tidbcloud/cluster-service-ng/domain/workflow/dag" clustergatewaymocks "github.com/tidbcloud/cluster-service-ng/pkg/clustergateway/mocks" dataplanemgrmocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplanemgr/mocks" "github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions" "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor" executormocks "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor/mocks" "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema" repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks" "github.com/tidbcloud/cluster-service-ng/pkg/utils" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" ) func setupAllDAGMocks(dagExecutor *executormocks.RemoteExecutor) { // Mock DAG queries for mergeSharedPoolStatusFromDagTask and other operations dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanCreateSharedPool, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanDeleteSharedPool, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanPauseSharedPool, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanResumeSharedPool, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanUpdateComponent, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanApplyComponent, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanCreateComponent, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanDeleteTiKVGroup, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanSetSharedPoolPDDeploymentModeToMonolithic, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanSetSharedPoolPDDeploymentModeToMicroservice, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanCreateTiDBCluster, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanUpdateTiDBCluster, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanDeleteTiDBCluster, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanCreateTiDBGroup, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanUpdateTiDBGroup, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanDeleteTiDBGroup, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanEnableCoprocessorWorker, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanDisableCoprocessorWorker, mock.AnythingOfType("executor.MatchFilter")).Return(executor.NewExecTaskIdStatus(), nil).Maybe() } // setupAllDAGMocksWithConflict mirrors setupAllDAGMocks, except conflictPlan // returns conflictStatus instead of an empty result. Each plan gets its own // distinct expectation (not a shared wildcard) so conflictPlan's non-empty // result cannot be shadowed by - or shadow - the other plans' empty results. func setupAllDAGMocksWithConflict(dagExecutor *executormocks.RemoteExecutor, conflictPlan string, conflictStatus executor.ExecTaskIdStatus) { allPlans := []string{ dag.PlanCreateSharedPool, dag.PlanDeleteSharedPool, dag.PlanPauseSharedPool, dag.PlanResumeSharedPool, dag.PlanUpdateComponent, dag.PlanApplyComponent, dag.PlanCreateComponent, dag.PlanDeleteTiKVGroup, dag.PlanSetSharedPoolPDDeploymentModeToMonolithic, dag.PlanSetSharedPoolPDDeploymentModeToMicroservice, dag.PlanCreateTiDBCluster, dag.PlanUpdateTiDBCluster, dag.PlanDeleteTiDBCluster, dag.PlanCreateTiDBGroup, dag.PlanUpdateTiDBGroup, dag.PlanDeleteTiDBGroup, dag.PlanEnableCoprocessorWorker, dag.PlanDisableCoprocessorWorker, } for _, planName := range allPlans { result := executor.NewExecTaskIdStatus() if planName == conflictPlan { result = conflictStatus } dagExecutor.On("QueryRunnings", mock.Anything, planName, mock.AnythingOfType("executor.MatchFilter")).Return(result, nil).Maybe() } } func TestCreateTiDBCluster(t *testing.T) { tests := []struct { name string tidbCluster *entity.TiDBCluster setupMocks func(*repomnocks.TiDBClusterRepo, *repomnocks.SharedPoolRepo, *repomnocks.LockRepo, *repomnocks.RuConfigRepo, *executormocks.RemoteExecutor, *clustergatewaymocks.Client, *dataplanemgrmocks.Client) expectedError string expectedCode codes.Code validate func(*testing.T, *entity.TiDBCluster, error) }{ { name: "successful_create_tidb_cluster", tidbCluster: &entity.TiDBCluster{ Id: "testclusterid", TenantId: "testtenant", SharedPoolId: "testsharedpool", RequestUnit: &apiv1.Cluster_RequestUnit{ Min: 10, Max: 100, }, }, setupMocks: func(tidbClusterRepo *repomnocks.TiDBClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor, clusterGatewayClient *clustergatewaymocks.Client, dataplaneClient *dataplanemgrmocks.Client) { // Setup all DAG mocks first setupAllDAGMocks(dagExecutor) // Setup all DAG mocks first setupAllDAGMocks(dagExecutor) // Mock check for existing cluster (called in checkTiDBCluster with force=true) tidbClusterRepo.On("Load", mock.Anything, "testclusterid", true).Return(nil, status.Error(codes.NotFound, "not found")) // Mock shared pool existence check (called in selectSharedPool) sharedPool := &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpool", Phase: apiv1.SharedPool_AVAILABLE, DataplaneName: "test-dataplane", Version: "v7.1.0", }, } sharedPoolRepo.On("Load", mock.Anything, "testsharedpool", false).Return(sharedPool, nil) // Mock lock operations lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) // Mock RU config ruConfig := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tiproxy": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, }, }, } ruConfigRepo.On("GetRuConfig", mock.Anything, "testsharedpool", apiv1.RuConfig_SHARED_POOL).Return(ruConfig, nil) // Mock tidb cluster save tidbClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil) // Mock DAG execution submission dagExecutor.On("Submit", mock.Anything, dag.PlanCreateTiDBCluster, mock.Anything, mock.Anything).Return(nil) // Mock cluster gateway client clusterGatewayClient.On("GetKeySpace", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) // Mock dataplane client for ListNodePools (called in genNodeSpecMap) nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{ NodePools: []*dplanemgrv1.NodePool{ { CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePools", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.ListNodePoolsRequest) bool { return req.DataPlaneId == "test-dataplane" })).Return(nodePoolResp, nil) }, }, { name: "shared_pool_not_found", tidbCluster: &entity.TiDBCluster{ Id: "testclusterid", TenantId: "testtenant", SharedPoolId: "nonexistent", RequestUnit: &apiv1.Cluster_RequestUnit{ Min: 10, Max: 100, }, }, setupMocks: func(tidbClusterRepo *repomnocks.TiDBClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor, clusterGatewayClient *clustergatewaymocks.Client, dataplaneClient *dataplanemgrmocks.Client) { // Setup all DAG mocks first setupAllDAGMocks(dagExecutor) // Mock DAG running check (called first in checkTiDBCluster) // Mock check for existing cluster (called in checkTiDBCluster with force=true) tidbClusterRepo.On("Load", mock.Anything, "testclusterid", true).Return(nil, status.Error(codes.NotFound, "not found")) // Mock lock operations lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) // Mock shared pool not found (called in selectSharedPool) sharedPoolRepo.On("Load", mock.Anything, "nonexistent", false).Return(nil, status.Error(codes.NotFound, "sharedPool not found")) }, expectedError: "sharedPool nonexistent not found", expectedCode: codes.NotFound, }, { name: "successful_create_with_custom_gateway_identity", tidbCluster: &entity.TiDBCluster{ Id: "testclusterid", TenantId: "testtenant", SharedPoolId: "testsharedpool", GatewayIdentity: "custom-gateway-id", RequestUnit: &apiv1.Cluster_RequestUnit{ Min: 10, Max: 100, }, }, setupMocks: func(tidbClusterRepo *repomnocks.TiDBClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor, clusterGatewayClient *clustergatewaymocks.Client, dataplaneClient *dataplanemgrmocks.Client) { // Setup all DAG mocks first setupAllDAGMocks(dagExecutor) // Mock DAG running check (called first in checkTiDBCluster) // Mock check for existing cluster (called in checkTiDBCluster with force=true) tidbClusterRepo.On("Load", mock.Anything, "testclusterid", true).Return(nil, status.Error(codes.NotFound, "not found")) // Mock shared pool existence check (called in selectSharedPool) sharedPool := &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpool", Phase: apiv1.SharedPool_AVAILABLE, DataplaneName: "test-dataplane", Version: "v7.1.0", }, } sharedPoolRepo.On("Load", mock.Anything, "testsharedpool", false).Return(sharedPool, nil) // Mock lock operations lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) // Mock RU config ruConfig := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tiproxy": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, }, }, } ruConfigRepo.On("GetRuConfig", mock.Anything, "testsharedpool", apiv1.RuConfig_SHARED_POOL).Return(ruConfig, nil) // Mock tidb cluster save tidbClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil) // Mock DAG execution submission dagExecutor.On("Submit", mock.Anything, dag.PlanCreateTiDBCluster, mock.Anything, mock.Anything).Return(nil) // Mock cluster gateway client clusterGatewayClient.On("GetKeySpace", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) // Mock dataplane client for ListNodePools (called in genNodeSpecMap) nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{ NodePools: []*dplanemgrv1.NodePool{ { CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePools", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.ListNodePoolsRequest) bool { return req.DataPlaneId == "test-dataplane" })).Return(nodePoolResp, nil) }, validate: func(t *testing.T, result *entity.TiDBCluster, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) assert.Equal(t, "custom-gateway-id", result.GetGatewayIdentity()) }, }, { name: "successful_create_with_auto_generated_gateway_identity", tidbCluster: &entity.TiDBCluster{ Id: "testclusterid", TenantId: "testtenant", SharedPoolId: "testsharedpool", // GatewayIdentity not provided, should be auto-generated RequestUnit: &apiv1.Cluster_RequestUnit{ Min: 10, Max: 100, }, }, setupMocks: func(tidbClusterRepo *repomnocks.TiDBClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor, clusterGatewayClient *clustergatewaymocks.Client, dataplaneClient *dataplanemgrmocks.Client) { // Setup all DAG mocks first setupAllDAGMocks(dagExecutor) // Mock DAG running check (called first in checkTiDBCluster) // Mock check for existing cluster (called in checkTiDBCluster with force=true) tidbClusterRepo.On("Load", mock.Anything, "testclusterid", true).Return(nil, status.Error(codes.NotFound, "not found")) // Mock shared pool existence check (called in selectSharedPool) sharedPool := &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpool", Phase: apiv1.SharedPool_AVAILABLE, DataplaneName: "test-dataplane", Version: "v7.1.0", }, } sharedPoolRepo.On("Load", mock.Anything, "testsharedpool", false).Return(sharedPool, nil) // Mock lock operations lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) // Mock RU config ruConfig := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tiproxy": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, }, }, } ruConfigRepo.On("GetRuConfig", mock.Anything, "testsharedpool", apiv1.RuConfig_SHARED_POOL).Return(ruConfig, nil) // Mock tidb cluster save tidbClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil) // Mock DAG execution submission dagExecutor.On("Submit", mock.Anything, dag.PlanCreateTiDBCluster, mock.Anything, mock.Anything).Return(nil) // Mock cluster gateway client clusterGatewayClient.On("GetKeySpace", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) // Mock dataplane client for ListNodePools (called in genNodeSpecMap) nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{ NodePools: []*dplanemgrv1.NodePool{ { CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePools", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.ListNodePoolsRequest) bool { return req.DataPlaneId == "test-dataplane" })).Return(nodePoolResp, nil) }, validate: func(t *testing.T, result *entity.TiDBCluster, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) assert.NotEmpty(t, result.GetGatewayIdentity(), "GatewayIdentity should be auto-generated when not provided") // Verify the auto-generated identity follows the expected pattern (random chars + timestamp) assert.Greater(t, len(result.GetGatewayIdentity()), 6, "Auto-generated GatewayIdentity should have reasonable length") }, }, { name: "gateway_identity_with_special_characters", tidbCluster: &entity.TiDBCluster{ Id: "testclusterid", TenantId: "testtenant", SharedPoolId: "testsharedpool", GatewayIdentity: "test-gateway-123", RequestUnit: &apiv1.Cluster_RequestUnit{ Min: 10, Max: 100, }, }, setupMocks: func(tidbClusterRepo *repomnocks.TiDBClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor, clusterGatewayClient *clustergatewaymocks.Client, dataplaneClient *dataplanemgrmocks.Client) { // Setup all DAG mocks first setupAllDAGMocks(dagExecutor) // Mock DAG running check (called first in checkTiDBCluster) // Mock check for existing cluster (called in checkTiDBCluster with force=true) tidbClusterRepo.On("Load", mock.Anything, "testclusterid", true).Return(nil, status.Error(codes.NotFound, "not found")) // Mock shared pool existence check (called in selectSharedPool) sharedPool := &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpool", Phase: apiv1.SharedPool_AVAILABLE, DataplaneName: "test-dataplane", Version: "v7.1.0", }, } sharedPoolRepo.On("Load", mock.Anything, "testsharedpool", false).Return(sharedPool, nil) // Mock lock operations lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) // Mock RU config ruConfig := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tiproxy": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, }, }, } ruConfigRepo.On("GetRuConfig", mock.Anything, "testsharedpool", apiv1.RuConfig_SHARED_POOL).Return(ruConfig, nil) // Mock tidb cluster save tidbClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil) // Mock DAG execution submission dagExecutor.On("Submit", mock.Anything, dag.PlanCreateTiDBCluster, mock.Anything, mock.Anything).Return(nil) // Mock cluster gateway client clusterGatewayClient.On("GetKeySpace", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) // Mock dataplane client for ListNodePools (called in genNodeSpecMap) nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{ NodePools: []*dplanemgrv1.NodePool{ { CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePools", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.ListNodePoolsRequest) bool { return req.DataPlaneId == "test-dataplane" })).Return(nodePoolResp, nil) }, validate: func(t *testing.T, result *entity.TiDBCluster, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) assert.Equal(t, "test-gateway-123", result.GetGatewayIdentity()) }, }, { name: "empty_gateway_identity_should_auto_generate", tidbCluster: &entity.TiDBCluster{ Id: "testclusterid", TenantId: "testtenant", SharedPoolId: "testsharedpool", GatewayIdentity: "", // Explicitly empty RequestUnit: &apiv1.Cluster_RequestUnit{ Min: 10, Max: 100, }, }, setupMocks: func(tidbClusterRepo *repomnocks.TiDBClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor, clusterGatewayClient *clustergatewaymocks.Client, dataplaneClient *dataplanemgrmocks.Client) { // Setup all DAG mocks first setupAllDAGMocks(dagExecutor) // Mock DAG running check (called first in checkTiDBCluster) // Mock check for existing cluster (called in checkTiDBCluster with force=true) tidbClusterRepo.On("Load", mock.Anything, "testclusterid", true).Return(nil, status.Error(codes.NotFound, "not found")) // Mock shared pool existence check (called in selectSharedPool) sharedPool := &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpool", Phase: apiv1.SharedPool_AVAILABLE, DataplaneName: "test-dataplane", Version: "v7.1.0", }, } sharedPoolRepo.On("Load", mock.Anything, "testsharedpool", false).Return(sharedPool, nil) // Mock lock operations lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) // Mock RU config ruConfig := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tiproxy": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, }, }, } ruConfigRepo.On("GetRuConfig", mock.Anything, "testsharedpool", apiv1.RuConfig_SHARED_POOL).Return(ruConfig, nil) // Mock tidb cluster save tidbClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil) // Mock DAG execution submission dagExecutor.On("Submit", mock.Anything, dag.PlanCreateTiDBCluster, mock.Anything, mock.Anything).Return(nil) // Mock cluster gateway client clusterGatewayClient.On("GetKeySpace", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) // Mock dataplane client for ListNodePools (called in genNodeSpecMap) nodePoolResp := &dplanemgrv1.ListNodePoolsResponse{ NodePools: []*dplanemgrv1.NodePool{ { CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePools", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.ListNodePoolsRequest) bool { return req.DataPlaneId == "test-dataplane" })).Return(nodePoolResp, nil) }, validate: func(t *testing.T, result *entity.TiDBCluster, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) assert.NotEmpty(t, result.GetGatewayIdentity(), "Empty GatewayIdentity should be auto-generated") assert.Greater(t, len(result.GetGatewayIdentity()), 6, "Auto-generated GatewayIdentity should have reasonable length") }, }, { name: "invalid_gateway_identity_format", tidbCluster: &entity.TiDBCluster{ Id: "testclusterid", TenantId: "testtenant", SharedPoolId: "testsharedpool", GatewayIdentity: "Invalid_Format", RequestUnit: &apiv1.Cluster_RequestUnit{ Min: 10, Max: 100, }, }, setupMocks: func(tidbClusterRepo *repomnocks.TiDBClusterRepo, sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dagExecutor *executormocks.RemoteExecutor, clusterGatewayClient *clustergatewaymocks.Client, dataplaneClient *dataplanemgrmocks.Client) { // Setup all DAG mocks first setupAllDAGMocks(dagExecutor) // No mocks needed for validation failure }, expectedError: "dns identity must contain only lowercase letters, numbers, and hyphens, and cannot start or end with hyphen", expectedCode: codes.InvalidArgument, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { // Create mocks tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t) sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) dagExecutor := executormocks.NewRemoteExecutor(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterGatewayClient := clustergatewaymocks.NewClient(t) scaleManager := bractionmocks.NewScaleManager(t) // Setup scale manager mock scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe() // Setup mocks tt.setupMocks(tidbClusterRepo, sharedPoolRepo, lockRepo, ruConfigRepo, dagExecutor, clusterGatewayClient, dataplaneClient) // Create server implementation server := &serverImpl{ tidbClusterRepo: tidbClusterRepo, sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, // Add minimal mocks for other required dependencies dataplaneClient: dataplaneClient, clustergatewayCli: clusterGatewayClient, scaleManager: scaleManager, } // Execute test result, err := server.CreateTiDBCluster(context.Background(), tt.tidbCluster) switch { case tt.expectedError != "": require.Error(t, err) assert.Contains(t, err.Error(), tt.expectedError) if tt.expectedCode != codes.OK { assert.Equal(t, tt.expectedCode, status.Code(err)) } assert.Nil(t, result) case tt.validate != nil: tt.validate(t, result, err) default: require.NoError(t, err) require.NotNil(t, result) } }) } } // TestCreateTiDBCluster_OnlyWhitelistComponentsSubmitted verifies that when a // TiDB cluster is created with RU-based scaling enabled, only whitelist // SharedPool components (from RcuScaleComponents) plus the per-cluster TIPROXY // are included in the DAG componentIds. Non-whitelist components like PD must // NOT appear in the submitted action. func TestCreateTiDBCluster_OnlyWhitelistComponentsSubmitted(t *testing.T) { tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t) sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) dagExecutor := executormocks.NewRemoteExecutor(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterGatewayClient := clustergatewaymocks.NewClient(t) scaleManager := bractionmocks.NewScaleManager(t) setupAllDAGMocks(dagExecutor) scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe() tidbClusterRepo.On("Load", mock.Anything, "testclusterid", true).Return(nil, status.Error(codes.NotFound, "not found")) // Build a shared pool with whitelist and non-whitelist components. defaultCompId := utils.FnvHash("testsharedpoolid") sharedPool := &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolid", DataplaneName: "test-dataplane", Phase: apiv1.SharedPool_AVAILABLE, Version: "v7.1.0", ComponentGroups: []*apiv1.Component{ // Whitelist: TIKV (SharedPool scope) { Id: defaultCompId, ComponentType: apiv1.ComponentType_TIKV, Phase: apiv1.Component_AVAILABLE, NodeQuota: &apiv1.NodeQuota{ NodeSpec: &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16}, NodeCount: 3, MinNodeCount: 3, MaxNodeCount: 15, }, Storages: []*apiv1.Component_Storage{ {Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 500, Iops: 3000}, {Name: apiv1.Component_Storage_RAFTLOG, Type: "gp3", SizeGi: 50, Iops: 3000}, }, }, // Whitelist: TIDB_WORKER (SharedPool scope) { Id: defaultCompId, ComponentType: apiv1.ComponentType_TIDB_WORKER, Phase: apiv1.Component_AVAILABLE, NodeQuota: &apiv1.NodeQuota{ NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 1, MinNodeCount: 1, MaxNodeCount: 15, }, }, // Non-whitelist: PD { Id: defaultCompId, ComponentType: apiv1.ComponentType_PD, Phase: apiv1.Component_AVAILABLE, NodeQuota: &apiv1.NodeQuota{ NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 3, MinNodeCount: 3, MaxNodeCount: 3, }, }, // Non-whitelist: PD_TSO { Id: defaultCompId, ComponentType: apiv1.ComponentType_PD_TSO, Phase: apiv1.Component_AVAILABLE, NodeQuota: &apiv1.NodeQuota{ NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 2, MinNodeCount: 2, MaxNodeCount: 2, }, }, // Non-whitelist: PD_SCHEDULING { Id: defaultCompId, ComponentType: apiv1.ComponentType_PD_SCHEDULING, Phase: apiv1.Component_AVAILABLE, NodeQuota: &apiv1.NodeQuota{ NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 2, MinNodeCount: 2, MaxNodeCount: 2, }, }, // Per-cluster: TIDB — should NOT be in shared pool list { Id: defaultCompId, ComponentType: apiv1.ComponentType_TIDB, Phase: apiv1.Component_AVAILABLE, NodeQuota: &apiv1.NodeQuota{ NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 2, MinNodeCount: 1, MaxNodeCount: 10, }, Storages: []*apiv1.Component_Storage{ {Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 100}, }, }, // TiFlash Compute — must pre-exist with storages { Id: defaultCompId, ComponentType: apiv1.ComponentType_TIFLASH_COMPUTE, Phase: apiv1.Component_AVAILABLE, NodeQuota: &apiv1.NodeQuota{ NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 1, MaxNodeCount: 12, }, Storages: []*apiv1.Component_Storage{ {Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 100}, }, }, // TiFlash Write { Id: defaultCompId, ComponentType: apiv1.ComponentType_TIFLASH_WRITE, Phase: apiv1.Component_AVAILABLE, NodeQuota: &apiv1.NodeQuota{ NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, NodeCount: 1, MaxNodeCount: 12, }, Storages: []*apiv1.Component_Storage{ {Name: apiv1.Component_Storage_DATA, Type: "gp3", SizeGi: 100}, }, }, }, }, } sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", false).Return(sharedPool, nil) lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) // getRequestUnitTotal calls tidbClusterRepo.List tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{}, nil) tidbClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil) clusterGatewayClient.On("GetKeySpace", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) dataplaneClient.On("ListNodePools", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.ListNodePoolsRequest) bool { return req.DataPlaneId == "test-dataplane" })).Return(&dplanemgrv1.ListNodePoolsResponse{ NodePools: []*dplanemgrv1.NodePool{ {CpuCore: 4, MemoryGi: 8, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 3500, AvailableMemoryMi: 7000}, {CpuCore: 8, MemoryGi: 16, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000}, {CpuCore: 16, MemoryGi: 64, NodeType: dplanemgrv1.NodeType_Dedicated, AvailableCpuMilliCore: 14000, AvailableMemoryMi: 60000}, }, }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, "testsharedpoolid", apiv1.RuConfig_SHARED_POOL).Return(&entity.RuConfig{ RuConfig: apiv1.RuConfig{ FeatureFlags: &apiv1.RuConfig_FeatureFlags{ EnableScaleSharedPoolByRu: proto.Bool(true), }, Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tikv": {MaxNodeCount: 36, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8C16G": {Min: 0, Max: 1500}, "16C64G": {Min: 1501, Max: 200000}}}, "tidb": {MaxNodeCount: 12, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}}, "tidb_worker": {MaxNodeCount: 15, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 1500}, "8C16G": {Min: 1501, Max: 200000}}}, "coprocessor_worker": {MaxNodeCount: 15, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8C16G": {Min: 0, Max: 200000}}}, "tikv_worker": {MaxNodeCount: 15, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8C16G": {Min: 0, Max: 200000}}}, "pd": {MaxNodeCount: 3, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}}, "pd_tso": {MaxNodeCount: 2, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}}, "pd_scheduling": {MaxNodeCount: 2, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}}, "tiflash_compute": {MaxNodeCount: 12, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}}, "tiflash_write": {MaxNodeCount: 12, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"4C8G": {Min: 0, Max: 200000}}}, "tiproxy": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8C16G": {Min: 0, Max: 200000}}}, }, }, }, nil) // Capture the submitted DAG args and verify component filtering. var capturedComponentIDs, capturedComponentTypes string dagExecutor.On("Submit", mock.Anything, dag.PlanCreateTiDBCluster, mock.MatchedBy(func(args actions.ActionArgs) bool { capturedComponentIDs, _ = args.GetString(workflow.ComponentIdsKey) capturedComponentTypes, _ = args.GetString(workflow.ComponentTypesKey) return true // always match, we assert below }), mock.Anything).Return(nil) server := &serverImpl{ tidbClusterRepo: tidbClusterRepo, sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, dataplaneClient: dataplaneClient, clustergatewayCli: clusterGatewayClient, scaleManager: scaleManager, } result, err := server.CreateTiDBCluster(context.Background(), &entity.TiDBCluster{ Id: "testclusterid", TenantId: "testtenant", SharedPoolId: "testsharedpoolid", RequestUnit: &apiv1.Cluster_RequestUnit{ Min: 200, Max: 2000, }, }) require.NoError(t, err) require.NotNil(t, result) dagExecutor.AssertCalled(t, "Submit", mock.Anything, dag.PlanCreateTiDBCluster, mock.Anything, mock.Anything) // Parse the submitted component IDs and types. ids := strings.Split(capturedComponentIDs, ",") types := strings.Split(capturedComponentTypes, ",") // Build a set of submitted component types for assertion. submittedTypes := make(map[string]bool) for _, ct := range types { if ct != "" { submittedTypes[ct] = true } } // Whitelist SharedPool components should be present. assert.Contains(t, submittedTypes, "tikv", "TIKV (whitelist) should be submitted") assert.Contains(t, submittedTypes, "worker-tidb", "TIDB_WORKER (whitelist) should be submitted") // TIPROXY is always added for the new cluster. assert.Contains(t, submittedTypes, "tiproxy", "TIPROXY (per-cluster) should always be submitted in create") // Non-whitelist components must NOT be present. assert.NotContains(t, submittedTypes, "pd", "PD (non-whitelist) must not be submitted") assert.NotContains(t, submittedTypes, "tso", "PD_TSO (non-whitelist) must not be submitted") assert.NotContains(t, submittedTypes, "scheduling", "PD_SCHEDULING (non-whitelist) must not be submitted") // IDs and types must be the same length. assert.Len(t, types, len(ids), "componentIds and componentTypes must have same length") // Verify component IDs: shared pool whitelist components use fnvHash(sharedPoolId), // TIPROXY uses the cluster ID. submittedIDs := make(map[string]bool) for _, id := range ids { if id != "" { submittedIDs[id] = true } } assert.Contains(t, submittedIDs, defaultCompId, "shared pool component ID (fnvHash) should be submitted") assert.Contains(t, submittedIDs, "testclusterid", "TIPROXY component ID (cluster ID) should be submitted") } func TestCreateTiDBClusterDoesNotResizeEssentialV2SharedPoolByRU(t *testing.T) { tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t) sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) dagExecutor := executormocks.NewRemoteExecutor(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterGatewayClient := clustergatewaymocks.NewClient(t) scaleManager := bractionmocks.NewScaleManager(t) setupAllDAGMocks(dagExecutor) scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe() tidbClusterRepo.On("Load", mock.Anything, "testclusterid", true).Return(nil, status.Error(codes.NotFound, "not found")) tidbClusterRepo.On("Save", mock.Anything, mock.Anything).Return(nil) worker := &entity.Component{ Id: utils.FnvHash("essential-v2-pool"), ComponentType: apiv1.ComponentType_TIKV_WORKER, Phase: apiv1.Component_AVAILABLE, NodeQuota: &entity.NodeQuota{ NodeSpec: &entity.NodeSpec{Cpu: 4, MemoryGi: 16}, NodeCount: 1, MinNodeCount: 1, MaxNodeCount: 15, }, } sharedPool := &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "essential-v2-pool", DataplaneName: "test-dataplane", Phase: apiv1.SharedPool_AVAILABLE, Version: "v7.1.0", ComponentGroups: []*entity.Component{worker}, }, DeployType: schema.SharedPoolDeployTypeSharedTiProxy, } sharedPoolRepo.On("Load", mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil) lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) clusterGatewayClient.On("GetKeySpace", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, sharedPool.GetId(), apiv1.RuConfig_SHARED_POOL).Return(&entity.RuConfig{ RuConfig: apiv1.RuConfig{ FeatureFlags: &apiv1.RuConfig_FeatureFlags{EnableScaleSharedPoolByRu: proto.Bool(true)}, Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tikv_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 0}, }}, }, }, }, nil) var capturedArgs actions.ActionArgs dagExecutor.On("Submit", mock.Anything, dag.PlanCreateTiDBCluster, mock.Anything, mock.Anything). Run(func(args mock.Arguments) { capturedArgs = args.Get(2).(actions.ActionArgs) }). Return(nil) server := &serverImpl{ tidbClusterRepo: tidbClusterRepo, sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, dataplaneClient: dataplaneClient, clustergatewayCli: clusterGatewayClient, scaleManager: scaleManager, } _, err := server.CreateTiDBCluster(context.Background(), &entity.TiDBCluster{ Id: "testclusterid", TenantId: "testtenant", SharedPoolId: sharedPool.GetId(), RequestUnit: &apiv1.Cluster_RequestUnit{Min: 2000, Max: 2100}, }) require.NoError(t, err) require.NotNil(t, capturedArgs) sharedPoolJSON, err := capturedArgs.GetString(workflow.SharedPoolKey) require.NoError(t, err) var submittedSharedPool entity.SharedPool require.NoError(t, protojson.Unmarshal([]byte(sharedPoolJSON), &submittedSharedPool)) require.Len(t, submittedSharedPool.GetComponentGroups(), 1) assert.Equal(t, uint32(4), submittedSharedPool.GetComponentGroups()[0].GetNodeQuota().GetNodeSpec().GetCpu()) assert.Equal(t, uint32(16), submittedSharedPool.GetComponentGroups()[0].GetNodeQuota().GetNodeSpec().GetMemoryGi()) assert.Empty(t, mustGetActionArgString(t, capturedArgs, workflow.ComponentIdsKey)) assert.Empty(t, mustGetActionArgString(t, capturedArgs, workflow.ComponentTypesKey)) assert.Equal(t, workflow.SkipComponentHPAApplyValue, mustGetActionArgString(t, capturedArgs, workflow.SkipComponentHPAApplyKey)) assert.Equal(t, workflow.SkipSharedTiProxyApplyValue, mustGetActionArgString(t, capturedArgs, workflow.SkipSharedTiProxyApplyKey)) dataplaneClient.AssertNotCalled(t, "ListNodePools", mock.Anything, mock.Anything) tidbClusterRepo.AssertNotCalled(t, "List", mock.Anything, mock.Anything) } // tkrwCreateRuConfig returns a ru config with EnableScaleSharedPoolByRu // enabled, sufficient to drive CreateTiDBCluster into the branch that calls // guardTiKVRemoteWorkersTopologyStable. The tests below never reach // generatedSharedPoolComponent (the guard rejects first), so the ranges only // need to exist, not be exercised. func tkrwCreateRuConfig() *entity.RuConfig { return &entity.RuConfig{ RuConfig: apiv1.RuConfig{ FeatureFlags: &apiv1.RuConfig_FeatureFlags{ EnableScaleSharedPoolByRu: proto.Bool(true), }, Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tikv": {MaxNodeCount: 36, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8C16G": {Min: 0, Max: 200000}}}, }, }, } } func TestCreateTiDBCluster_GuardBlocksOnRunningTiKVRemoteWorkerDAG(t *testing.T) { tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t) sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) dagExecutor := executormocks.NewRemoteExecutor(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterGatewayClient := clustergatewaymocks.NewClient(t) scaleManager := bractionmocks.NewScaleManager(t) setupAllDAGMocksWithConflict(dagExecutor, dag.PlanEnableCoprocessorWorker, runningDagStatus(tkrwKey(t, tkrwEnabledBoth(t)))) scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe() tidbClusterRepo.On("Load", mock.Anything, "testclusterid", true).Return(nil, status.Error(codes.NotFound, "not found")) sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3) sharedPool.Version = "v7.1.0" sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).Return(sharedPool, nil) lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) clusterGatewayClient.On("GetKeySpace", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(tkrwNodePoolConfigsResponse(), nil) ruConfigRepo.On("GetRuConfig", mock.Anything, testTKRWSharedPoolID, apiv1.RuConfig_SHARED_POOL).Return(tkrwCreateRuConfig(), nil) tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{}, nil) server := &serverImpl{ tidbClusterRepo: tidbClusterRepo, sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, dataplaneClient: dataplaneClient, clustergatewayCli: clusterGatewayClient, scaleManager: scaleManager, } result, err := server.CreateTiDBCluster(context.Background(), &entity.TiDBCluster{ Id: "testclusterid", TenantId: "testtenant", SharedPoolId: testTKRWSharedPoolID, RequestUnit: &apiv1.Cluster_RequestUnit{ Min: 200, Max: 2000, }, }) require.Error(t, err) require.Nil(t, result) statusErr, ok := status.FromError(err) require.True(t, ok) assert.Equal(t, codes.FailedPrecondition, statusErr.Code()) dagExecutor.AssertNotCalled(t, "Submit", mock.Anything, dag.PlanCreateTiDBCluster, mock.Anything, mock.Anything) } func TestCreateTiDBCluster_GuardBlocksOnUnconvergedTiKVRemoteWorkerTopology(t *testing.T) { tidbClusterRepo := repomnocks.NewTiDBClusterRepo(t) sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) dagExecutor := executormocks.NewRemoteExecutor(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterGatewayClient := clustergatewaymocks.NewClient(t) scaleManager := bractionmocks.NewScaleManager(t) setupAllDAGMocks(dagExecutor) scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe() tidbClusterRepo.On("Load", mock.Anything, "testclusterid", true).Return(nil, status.Error(codes.NotFound, "not found")) // Persisted target already equals the guard's normalized target, so the // guard reaches the live-CR check - which sees no CR at all (zero objects // seeded), i.e. genuine drift rather than a legacy-mismatch shortcut. sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3) sharedPool.Version = "v7.1.0" sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).Return(sharedPool, nil) lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) clusterGatewayClient.On("GetKeySpace", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(tkrwNodePoolConfigsResponse(), nil) ruConfigRepo.On("GetRuConfig", mock.Anything, testTKRWSharedPoolID, apiv1.RuConfig_SHARED_POOL).Return(tkrwCreateRuConfig(), nil) tidbClusterRepo.On("List", mock.Anything, mock.Anything).Return([]*entity.TiDBCluster{}, nil) server := &serverImpl{ tidbClusterRepo: tidbClusterRepo, sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, dataplaneClient: dataplaneClient, clustergatewayCli: clusterGatewayClient, scaleManager: scaleManager, dataplaneManager: newTKRWDataplaneManager(t), } result, err := server.CreateTiDBCluster(context.Background(), &entity.TiDBCluster{ Id: "testclusterid", TenantId: "testtenant", SharedPoolId: testTKRWSharedPoolID, RequestUnit: &apiv1.Cluster_RequestUnit{ Min: 200, Max: 2000, }, }) require.Error(t, err) require.Nil(t, result) statusErr, ok := status.FromError(err) require.True(t, ok) assert.Equal(t, codes.FailedPrecondition, statusErr.Code()) dagExecutor.AssertNotCalled(t, "Submit", mock.Anything, dag.PlanCreateTiDBCluster, mock.Anything, mock.Anything) } func TestValidateDNSIdentity(t *testing.T) { tests := []struct { name string dnsIdentity string expectedError string expectedCode codes.Code }{ { name: "empty_identity_valid", dnsIdentity: "", expectedError: "", expectedCode: codes.OK, }, { name: "valid_lowercase_with_numbers", dnsIdentity: "test123", expectedError: "", expectedCode: codes.OK, }, { name: "valid_with_hyphens", dnsIdentity: "test-gateway-123", expectedError: "", expectedCode: codes.OK, }, { name: "valid_single_character", dnsIdentity: "a", expectedError: "", expectedCode: codes.OK, }, { name: "valid_max_length", dnsIdentity: "a12345678901234567890123", expectedError: "", expectedCode: codes.OK, }, { name: "invalid_uppercase", dnsIdentity: "TestGateway", expectedError: "dns identity must contain only lowercase letters, numbers, and hyphens", expectedCode: codes.InvalidArgument, }, { name: "invalid_starts_with_hyphen", dnsIdentity: "-testgateway", expectedError: "dns identity must contain only lowercase letters, numbers, and hyphens", expectedCode: codes.InvalidArgument, }, { name: "invalid_ends_with_hyphen", dnsIdentity: "testgateway-", expectedError: "dns identity must contain only lowercase letters, numbers, and hyphens", expectedCode: codes.InvalidArgument, }, { name: "invalid_too_long", dnsIdentity: "a1234567890123456789012345", expectedError: "dns identity must be no more than 24 characters", expectedCode: codes.InvalidArgument, }, { name: "invalid_underscore", dnsIdentity: "test_gateway", expectedError: "dns identity must contain only lowercase letters, numbers, and hyphens", expectedCode: codes.InvalidArgument, }, { name: "invalid_special_characters", dnsIdentity: "test@gateway", expectedError: "dns identity must contain only lowercase letters, numbers, and hyphens", expectedCode: codes.InvalidArgument, }, { name: "invalid_spaces", dnsIdentity: "test gateway", expectedError: "dns identity must contain only lowercase letters, numbers, and hyphens", expectedCode: codes.InvalidArgument, }, { name: "invalid_consecutive_hyphens", dnsIdentity: "test--gateway", expectedError: "", expectedCode: codes.OK, // consecutive hyphens are actually allowed in DNS labels }, { name: "valid_numbers_only", dnsIdentity: "123456", expectedError: "", expectedCode: codes.OK, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { err := validateDNSIdentity(tt.dnsIdentity) if tt.expectedError == "" { assert.NoError(t, err, "Expected no error for gateway identity: %s", tt.dnsIdentity) } else { require.Error(t, err, "Expected error for gateway identity: %s", tt.dnsIdentity) assert.Contains(t, err.Error(), tt.expectedError, "Error message should contain expected text") statusErr, ok := status.FromError(err) assert.True(t, ok, "Error should be a gRPC status error") assert.Equal(t, tt.expectedCode, statusErr.Code(), "Error code should match expected") } }) } }