package server import ( "context" "encoding/json" "errors" "fmt" "testing" "github.com/samber/lo" "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/config" "github.com/tidbcloud/cluster-service-ng/domain/entity" "github.com/tidbcloud/cluster-service-ng/domain/workflow" "github.com/tidbcloud/cluster-service-ng/domain/workflow/dag" clusterconfigmocks "github.com/tidbcloud/cluster-service-ng/pkg/clusterconfig/mocks" dataplanemgrmocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplanemgr/mocks" actions "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" "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo" repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" ) // mockTransactioner is a simple mock implementation of db.Transactioner type mockTransactioner struct{} func (m *mockTransactioner) Transaction(ctx context.Context, fn func(context.Context) error) error { return fn(ctx) } func withSchemaManagerNodeSpec(resp *dplanemgrv1.ListNodePoolConfigsResponse) *dplanemgrv1.ListNodePoolConfigsResponse { resp.NodePoolConfigs = append(resp.NodePoolConfigs, &dplanemgrv1.NodePoolConfig{ CpuCore: 2, MemoryGi: 8, AvailableCpuMilliCore: 2000, AvailableMemoryMi: 8000, NodeType: dplanemgrv1.NodeType_Dedicated, }) return resp } func validResourcePoolCreateInitConfig() *entity.InitSharedPoolConfig { initConfig := generatedSharedPoolInitConfig() initConfig.NetworkCidr = "10.0.0.0/16" return initConfig } // assertColumnarSharedPoolComponents asserts a newly created pool uses the // TiFlash Columnar architecture and starts Schema Manager when // TiFlash Compute has an initial replica. func assertColumnarSharedPoolComponents(t *testing.T, sharedPool *entity.SharedPool) { t.Helper() assert.Equal(t, schema.SharedPoolTiflashModeColumnar, sharedPool.TiflashMode) tiflashComputeComp, ok := lo.Find(sharedPool.GetComponentGroups(), func(component *apiv1.Component) bool { return component.GetComponentType() == apiv1.ComponentType_TIFLASH_COMPUTE }) assert.True(t, ok, "expected TIFLASH_COMPUTE component") schemaManagerComp, ok := lo.Find(sharedPool.GetComponentGroups(), func(component *apiv1.Component) bool { return component.GetComponentType() == apiv1.ComponentType_SCHEMA_MANAGER }) assert.True(t, ok, "expected SCHEMA_MANAGER component") if ok && tiflashComputeComp != nil { wantSchemaManagerNodeCount := uint32(0) if tiflashComputeComp.GetNodeQuota().GetNodeCount() > 0 { wantSchemaManagerNodeCount = 1 } assert.Equal(t, wantSchemaManagerNodeCount, schemaManagerComp.GetNodeQuota().GetNodeCount()) } assert.NotContains(t, lo.Map(sharedPool.GetComponentGroups(), func(component *apiv1.Component, _ int) apiv1.ComponentType { return component.GetComponentType() }), apiv1.ComponentType_TIFLASH_WRITE) } func assertLegacyTiFlashSharedPoolComponents(t *testing.T, sharedPool *entity.SharedPool) { t.Helper() assert.Empty(t, sharedPool.TiflashMode) componentTypes := lo.Map(sharedPool.GetComponentGroups(), func(component *apiv1.Component, _ int) apiv1.ComponentType { return component.GetComponentType() }) assert.Contains(t, componentTypes, apiv1.ComponentType_TIFLASH_WRITE) assert.Contains(t, componentTypes, apiv1.ComponentType_TIFLASH_COMPUTE) assert.NotContains(t, componentTypes, apiv1.ComponentType_SCHEMA_MANAGER) } func TestGenNodeSpecMapFiltersInactiveNodeInstanceGroups(t *testing.T) { ctx := context.Background() dataplaneClient := dataplanemgrmocks.NewClient(t) nodeGroupRepo := repomnocks.NewNodeInstanceGroupRepo(t) dataplaneNodeGroupRepo := repomnocks.NewDataplaneNodeInstanceGroupRepo(t) dataplaneClient.On("ListNodePools", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.ListNodePoolsRequest) bool { return req.GetDataPlaneId() == "dp-1" })).Return(&dplanemgrv1.ListNodePoolsResponse{ NodePools: []*dplanemgrv1.NodePool{ { CpuCore: 8, MemoryGi: 16, InstanceType: "m5.2xlarge", AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, }, { CpuCore: 4, MemoryGi: 8, InstanceType: "m5.xlarge", NodeType: dplanemgrv1.NodeType_Dedicated, }, { CpuCore: 2, MemoryGi: 4, InstanceType: "m5.large", NodeType: dplanemgrv1.NodeType_Dedicated, }, { CpuCore: 12, MemoryGi: 24, InstanceType: "m5.3xlarge", NodeType: dplanemgrv1.NodeType_Dedicated, }, { CpuCore: 16, MemoryGi: 32, InstanceType: "m5.4xlarge", NodeType: dplanemgrv1.NodeType_Dedicated, }, { CpuCore: 32, MemoryGi: 64, InstanceType: "m5.8xlarge", AvailableCpuMilliCore: 31000, AvailableMemoryMi: 63000, NodeType: dplanemgrv1.NodeType_Dedicated, }, { CpuCore: 6, MemoryGi: 12, InstanceType: "m5.custom", AvailableCpuMilliCore: 5000, AvailableMemoryMi: 11000, NodeType: dplanemgrv1.NodeType_Dedicated, }, { CpuCore: 64, MemoryGi: 128, InstanceType: "m5.16xlarge", NodeType: dplanemgrv1.NodeType_Shared, }, }, }, nil).Once() dataplaneNodeGroupRepo.On("ListByDataplane", mock.Anything, "dp-1").Return([]*entity.DataplaneNodeInstanceGroup{ { DataplaneName: "dp-1", NodeSpec: &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32}, PrimaryInstanceType: "m5.4xlarge", Status: apiv1.DataplaneNodeInstanceGroup_DEPRECATED, }, { DataplaneName: "dp-1", NodeSpec: &apiv1.NodeSpec{Cpu: 32, MemoryGi: 64}, PrimaryInstanceType: "m5.8xlarge", Status: apiv1.DataplaneNodeInstanceGroup_ACTIVE, }, }, nil).Once() nodeGroupRepo.On("ListAll", mock.Anything).Return([]*entity.NodeInstanceGroup{ { NodeSpec: &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16}, PrimaryInstanceType: "m5.2xlarge", Status: apiv1.NodeInstanceGroup_ACTIVE, }, { NodeSpec: &apiv1.NodeSpec{Cpu: 4, MemoryGi: 8}, PrimaryInstanceType: "m5.xlarge", Status: apiv1.NodeInstanceGroup_DEPRECATED, }, { NodeSpec: &apiv1.NodeSpec{Cpu: 2, MemoryGi: 4}, PrimaryInstanceType: "m5.large", Status: apiv1.NodeInstanceGroup_PENDING, }, { NodeSpec: &apiv1.NodeSpec{Cpu: 12, MemoryGi: 24}, PrimaryInstanceType: "m5.3xlarge", Status: apiv1.NodeInstanceGroup_RETIRING, }, { NodeSpec: &apiv1.NodeSpec{Cpu: 16, MemoryGi: 32}, PrimaryInstanceType: "m5.4xlarge", Status: apiv1.NodeInstanceGroup_ACTIVE, }, { NodeSpec: &apiv1.NodeSpec{Cpu: 32, MemoryGi: 64}, PrimaryInstanceType: "m5.8xlarge", Status: apiv1.NodeInstanceGroup_DEPRECATED, }, }, nil).Once() server := &serverImpl{ dataplaneClient: dataplaneClient, nodeGroupRepo: nodeGroupRepo, dataplaneNodeGroupRepo: dataplaneNodeGroupRepo, } nodeSpecMap, err := server.genNodeSpecMap(ctx, &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "sp-1", DataplaneName: "dp-1", Phase: apiv1.SharedPool_AVAILABLE, }, }) require.NoError(t, err) require.Len(t, nodeSpecMap, 3) require.Contains(t, nodeSpecMap, "8C16G") require.Contains(t, nodeSpecMap, "32C64G") require.Contains(t, nodeSpecMap, "6C12G") assert.NotContains(t, nodeSpecMap, "4C8G") assert.NotContains(t, nodeSpecMap, "2C4G") assert.NotContains(t, nodeSpecMap, "12C24G") assert.NotContains(t, nodeSpecMap, "16C32G") assert.EqualValues(t, 7, nodeSpecMap["8C16G"][0].GetAvailableCpu()) assert.EqualValues(t, 31, nodeSpecMap["32C64G"][0].GetAvailableCpu()) assert.EqualValues(t, 5, nodeSpecMap["6C12G"][0].GetAvailableCpu()) nodeGroupRepo.AssertNotCalled(t, "LoadBySpecAndPrimary", mock.Anything, mock.Anything, mock.Anything) dataplaneNodeGroupRepo.AssertNotCalled(t, "LoadByDataplaneAndSpec", mock.Anything, mock.Anything, mock.Anything, mock.Anything) } func TestGenNodeSpecMapReturnsErrorWhenNodeInstanceGroupLoadFails(t *testing.T) { ctx := context.Background() dataplaneClient := dataplanemgrmocks.NewClient(t) nodeGroupRepo := repomnocks.NewNodeInstanceGroupRepo(t) dataplaneNodeGroupRepo := repomnocks.NewDataplaneNodeInstanceGroupRepo(t) dataplaneNodeGroupRepo.On("ListByDataplane", mock.Anything, "dp-1").Return([]*entity.DataplaneNodeInstanceGroup{}, nil).Once() nodeGroupRepo.On("ListAll", mock.Anything).Return(nil, errors.New("db unavailable")).Once() server := &serverImpl{ dataplaneClient: dataplaneClient, nodeGroupRepo: nodeGroupRepo, dataplaneNodeGroupRepo: dataplaneNodeGroupRepo, } nodeSpecMap, err := server.genNodeSpecMap(ctx, &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "sp-1", DataplaneName: "dp-1", Phase: apiv1.SharedPool_AVAILABLE, }, }) require.Error(t, err) assert.Contains(t, err.Error(), "failed to list node instance groups") assert.Nil(t, nodeSpecMap) dataplaneClient.AssertNotCalled(t, "ListNodePools", mock.Anything, mock.Anything) } func TestCreateSharedPool(t *testing.T) { ruCfg := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tikv": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tikv_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "coprocessor_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tidb": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tidb_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "pd": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "pd_tso": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "pd_scheduling": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tiflash_write": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tiflash_compute": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tiproxy": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "resource_manager": { MaxNodeCount: 2, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "2c4g": { Min: 0, Max: 0, }, }, }, }, }, } tests := []struct { name string sharedPool *entity.SharedPool requestUnit *entity.RequestUnit initConfig *apiv1.CreateSharedPoolRequest_InitSharedPoolConfig options []*CreateSharedPoolOptions setupMocks func(*repomnocks.SharedPoolRepo, *repomnocks.LockRepo, *repomnocks.RuConfigRepo, *dataplanemgrmocks.Client, *clusterconfigmocks.Client, *executormocks.RemoteExecutor) expectedError string expectedCode codes.Code expectedReason string validate func(*testing.T, *entity.SharedPool, error) }{ { name: "bootstrap capacity rejects caller request units before DPM access", sharedPool: &entity.SharedPool{SharedPool: apiv1.SharedPool{ Id: "bootstrapwithrequestunits", Provider: &apiv1.Provider{Name: "aws", Type: apiv1.ProviderType_AWS}, }, TenantId: "test-tenant"}, requestUnit: &entity.RequestUnit{Min: 1000, Max: 2000}, initConfig: validResourcePoolCreateInitConfig(), options: []*CreateSharedPoolOptions{{BootstrapCapacity: &SharedPoolBootstrapCapacity{ TiKV: &BootstrapComponentCapacity{NodeSpec: &entity.NodeSpec{Cpu: 8, MemoryGi: 16}, NodeCount: 3}, }}}, setupMocks: func(_ *repomnocks.SharedPoolRepo, _ *repomnocks.LockRepo, _ *repomnocks.RuConfigRepo, _ *dataplanemgrmocks.Client, _ *clusterconfigmocks.Client, _ *executormocks.RemoteExecutor) { }, expectedError: "bootstrap_capacity must not be combined with request_unit", expectedCode: codes.InvalidArgument, expectedReason: reasonInvalidBootstrapCapacityMode, }, { name: "create rejects blank cloud resource tag value before writes", sharedPool: &entity.SharedPool{SharedPool: apiv1.SharedPool{ Id: "invalidcreatetags", Provider: &apiv1.Provider{Name: "aws", Type: apiv1.ProviderType_AWS}, CloudResourceTags: map[string]string{"owner": " "}, }, TenantId: "test-tenant"}, requestUnit: &entity.RequestUnit{Min: 1000, Max: 2000}, initConfig: validResourcePoolCreateInitConfig(), setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, _ *repomnocks.LockRepo, _ *repomnocks.RuConfigRepo, _ *dataplanemgrmocks.Client, _ *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "invalidcreatetags", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) }, expectedError: "cloud_resource_tags contains empty value", expectedCode: codes.InvalidArgument, }, { name: "enforced create fails closed without O11Y configuration", sharedPool: &entity.SharedPool{SharedPool: apiv1.SharedPool{ Id: "missingo11yconfiguration", Provider: &apiv1.Provider{Name: "aws", Type: apiv1.ProviderType_AWS}, }, TenantId: "test-tenant"}, requestUnit: &entity.RequestUnit{Min: 1000, Max: 2000}, initConfig: validResourcePoolCreateInitConfig(), options: []*CreateSharedPoolOptions{{EnforceO11yNetworkNonOverlap: true}}, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, _ *repomnocks.LockRepo, _ *repomnocks.RuConfigRepo, _ *dataplanemgrmocks.Client, _ *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "missingo11yconfiguration", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) }, expectedError: "o11y cluster cidr is not configured", expectedCode: codes.FailedPrecondition, expectedReason: reasonO11yCIDRConfigurationMissing, }, { name: "successful creation with all default values", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolid", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, Pca: &apiv1.SharedPool_PCA{ CaIdentifier: "arn:aws:acm-pca", }, VcpusLimit: 106, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1000, Iops: 10000, ThroughputMbps: 500, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 5000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 2000, Iops: 20000, ThroughputMbps: 1000, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1500, Iops: 15000, ThroughputMbps: 750, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { // Mock shared pool not found (new creation) sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", true).Return(nil, status.Error(codes.NotFound, "not found")) // Mock lock/unlock operations lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolid" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) // Mock dataplane client for node pool configs nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: true, }, { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) // Mock cluster config client clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) // Mock DAG executor dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) tikvGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_TIKV }) assert.True(t, ok) assert.False(t, tikvGroup.GetNodeQuota().GetNodeSpec().GetWithLocalDisk()) tikvDataStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_DATA }) assert.True(t, ok) assert.Equal(t, entity.GP3, tikvDataStorage.Type) assert.Positive(t, tikvDataStorage.Iops) assert.Equal(t, "testsharedpoolid", result.GetId()) assert.EqualValues(t, 106, result.GetVcpusLimit()) assert.Equal(t, apiv1.SharedPool_CREATING, result.GetPhase()) assertColumnarSharedPoolComponents(t, result) assert.Equal(t, "v8.0.0", result.GetVersion()) assert.NotNil(t, result.GetObjectStorage()) assert.NotNil(t, result.GetInternalStorage()) // Bucket names are now set during creation assert.Equal(t, "testsharedpoolid-us-west-2-data", result.GetObjectStorage().GetBucket()) assert.Equal(t, "testsharedpoolid-us-west-2-internal", result.GetInternalStorage().GetBucket()) // Verify component groups are created assert.GreaterOrEqual(t, len(result.GetComponentGroups()), 8) // Should have all component types }, }, { name: "successful creation with local disk and pca", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolid", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.LocalSSD, SizeGi: 0, // LocalSSD should not have size set, will use instance local ssd size Iops: 0, // LocalSSD does not support setting IOPS ThroughputMbps: 0, // LocalSSD does not support setting throughput }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 5000, ThroughputMbps: 0, // IO2 does not support setting throughput }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.LocalSSD, SizeGi: 0, // LocalSSD should not have size set, will use instance local ssd size Iops: 0, ThroughputMbps: 0, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 500, Iops: 5000, ThroughputMbps: 250, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { // Mock shared pool not found (new creation) sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", true).Return(nil, status.Error(codes.NotFound, "not found")) // Mock lock/unlock operations lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolid" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) // Mock dataplane client for node pool configs nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: true, LocalDiskSizeGi: 1000, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) // Mock cluster config client clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) // Mock DAG executor dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) tikvGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_TIKV }) assert.True(t, ok) assert.True(t, tikvGroup.GetNodeQuota().GetNodeSpec().GetWithLocalDisk()) tikvDataStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_DATA }) assert.True(t, ok) assert.Equal(t, entity.LocalSSD, tikvDataStorage.Type) assert.Equal(t, uint32(0), tikvDataStorage.Iops) assert.Equal(t, "testsharedpoolid", result.GetId()) assert.Equal(t, apiv1.SharedPool_CREATING, result.GetPhase()) assert.Equal(t, "v8.0.0", result.GetVersion()) assert.NotNil(t, result.GetObjectStorage()) assert.NotNil(t, result.GetInternalStorage()) // Bucket names are now set during creation assert.Equal(t, "testsharedpoolid-us-west-2-data", result.GetObjectStorage().GetBucket()) assert.Equal(t, "testsharedpoolid-us-west-2-internal", result.GetInternalStorage().GetBucket()) // Verify component groups are created assert.GreaterOrEqual(t, len(result.GetComponentGroups()), 8) // Should have all component types }, }, { name: "successful creation with resource manager enabled", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolwithrm", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1000, Iops: 10000, ThroughputMbps: 500, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 5000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 2000, Iops: 20000, ThroughputMbps: 1000, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1500, Iops: 15000, ThroughputMbps: 750, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolwithrm", true).Return(nil, status.Error(codes.NotFound, "not found")) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolwithrm" })).Return(func() error { return nil }, nil) // Return ruConfig with EnableResourceManager=true ruCfgWithRM := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ FeatureFlags: &apiv1.RuConfig_FeatureFlags{ EnableResourceManager: true, }, Components: ruCfg.Components, }, } ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfgWithRM, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, { CpuCore: 2, MemoryGi: 4, AvailableCpuMilliCore: 2000, AvailableMemoryMi: 4000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) // Verify resource manager component is created resourceManagerGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_RESOURCE_MANAGER }) assert.True(t, ok, "resource manager component should be created when feature flag is enabled") assert.NotNil(t, resourceManagerGroup) assert.Equal(t, apiv1.Component_CREATING, resourceManagerGroup.GetPhase()) // Verify resource manager quota is fixed at 2 assert.Equal(t, uint32(2), resourceManagerGroup.GetNodeQuota().GetNodeCount()) assert.Equal(t, uint32(2), resourceManagerGroup.GetNodeQuota().GetMaxNodeCount()) assert.Equal(t, uint32(2), resourceManagerGroup.GetNodeQuota().GetMinNodeCount()) // Verify node spec is 2C4G assert.Equal(t, uint32(2), resourceManagerGroup.GetNodeQuota().GetNodeSpec().GetCpu()) assert.Equal(t, uint32(4), resourceManagerGroup.GetNodeQuota().GetNodeSpec().GetMemoryGi()) }, }, { name: "missing shared pool id", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "", }, }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, setupMocks: func(*repomnocks.SharedPoolRepo, *repomnocks.LockRepo, *repomnocks.RuConfigRepo, *dataplanemgrmocks.Client, *clusterconfigmocks.Client, *executormocks.RemoteExecutor) { }, expectedError: "sharedPool id is required", expectedCode: codes.InvalidArgument, }, { name: "invalid config center meta profile id", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolid", }, }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ ConfigCenterMetaProfileId: "not-an-id", }, setupMocks: func(*repomnocks.SharedPoolRepo, *repomnocks.LockRepo, *repomnocks.RuConfigRepo, *dataplanemgrmocks.Client, *clusterconfigmocks.Client, *executormocks.RemoteExecutor) { }, expectedError: "invalid config center meta profile id", expectedCode: codes.InvalidArgument, }, { name: "RCU columnar creation requires a 2C8G schema manager node spec", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolmissing2c8g", CapacityMode: apiv1.CapacityMode_RCU_DRIVEN, Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{Min: 1000, Max: 2000}, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1000, Iops: 10000, ThroughputMbps: 500, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 5000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 2000, Iops: 20000, ThroughputMbps: 1000, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1500, Iops: 15000, ThroughputMbps: 750, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, _ *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolmissing2c8g", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolmissing2c8g" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil).Maybe() dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{{ CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, }}, }, nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.Error(t, err) assert.Equal(t, codes.InvalidArgument, status.Code(err)) assert.Nil(t, result) }, }, { name: "invalid request unit - min greater than max", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolid", }, }, requestUnit: &entity.RequestUnit{ Min: 2000, Max: 1000, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) }, expectedError: "request unit min 2000 cannot be greater than max 1000", expectedCode: codes.InvalidArgument, }, { name: "running dag instance already exists", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolid", }, }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", true).Return(nil, status.Error(codes.NotFound, "not found")) // Mock running DAG instance found runningDags := executor.NewExecTaskIdStatus() runningDags["existingdagid"] = &executor.ExecTaskStatusEx{ State: executor.ExecTaskStateTypeRunningOK, } dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(runningDags, nil) }, expectedError: "shared pool creation is already running", expectedCode: codes.AlreadyExists, }, { name: "lock failure", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolid", }, }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1000, Iops: 10000, ThroughputMbps: 500, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 5000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 2000, Iops: 20000, ThroughputMbps: 1000, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1500, Iops: 15000, ThroughputMbps: 750, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) // Mock lock failure lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolid" })).Return(func() error { return nil }, fmt.Errorf("lock failed")) }, expectedError: "lock failed", }, { name: "cluster config client failure", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolid", }, }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1000, Iops: 10000, ThroughputMbps: 500, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 5000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 2000, Iops: 20000, ThroughputMbps: 1000, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1500, Iops: 15000, ThroughputMbps: 750, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolid" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 3800, AvailableMemoryMi: 7600, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) // Mock cluster config client failure clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("", errors.New("config error")) }, expectedError: "failed to get default version: config error", expectedCode: codes.Internal, }, { name: "dag executor submit failure", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolid", }, }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1000, Iops: 10000, ThroughputMbps: 500, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 5000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 2000, Iops: 20000, ThroughputMbps: 1000, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1500, Iops: 15000, ThroughputMbps: 750, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolid" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 3800, AvailableMemoryMi: 7600, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) // Mock DAG executor submit failure dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(errors.New("submit failed")) }, expectedError: "failed to submit create sharedpool dag: submit failed", expectedCode: codes.Internal, }, { name: "successful creation with TiFlash compute local disk enabled", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpooltiflashlocal", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant-tiflash", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 500, Iops: 5000, ThroughputMbps: 250, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 50, Iops: 3000, ThroughputMbps: 125, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.LocalSSD, SizeGi: 0, // LocalSSD should not have size set, will use instance local ssd size }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 400, Iops: 4000, ThroughputMbps: 200, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpooltiflashlocal", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpooltiflashlocal" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: true, LocalDiskSizeGi: 2000, }, { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) // Verify TiFlash compute component uses local disk tiflashComputeGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_TIFLASH_COMPUTE }) assert.True(t, ok) assert.True(t, tiflashComputeGroup.GetNodeQuota().GetNodeSpec().GetWithLocalDisk()) tiflashComputeDataStorage, ok := lo.Find(tiflashComputeGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_DATA }) assert.True(t, ok) assert.Equal(t, entity.LocalSSD, tiflashComputeDataStorage.Type) assert.Equal(t, uint32(0), tiflashComputeDataStorage.Iops) assertColumnarSharedPoolComponents(t, result) }, }, { name: "successful Essential V2 RCU creation with custom TiKV raft log storage", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpooltikvraftlog", CapacityMode: apiv1.CapacityMode_RCU_DRIVEN, Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "", DeployType: schema.SharedPoolDeployTypeSharedTiProxy, }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 500, Iops: 5000, ThroughputMbps: 250, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 3000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 400, Iops: 4000, ThroughputMbps: 200, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 300, Iops: 3000, ThroughputMbps: 150, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpooltikvraftlog", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpooltikvraftlog" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) tikvGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_TIKV }) assert.True(t, ok) // Verify TiKV raft log storage configuration tikvRaftLogStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_RAFTLOG }) assert.True(t, ok) assert.Equal(t, entity.IO2, tikvRaftLogStorage.Type) assert.Equal(t, uint32(100), tikvRaftLogStorage.SizeGi) assert.Equal(t, uint32(3000), tikvRaftLogStorage.Iops) assert.Equal(t, schema.SharedPoolDeployTypeSharedTiProxy, result.DeployType) assertColumnarSharedPoolComponents(t, result) }, }, { name: "unsupported TiKV version keeps legacy TiFlash topology", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpooltiflashwn", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 400, Iops: 4000, ThroughputMbps: 200, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 50, Iops: 3000, ThroughputMbps: 125, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 300, Iops: 3000, ThroughputMbps: 150, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 500, Iops: 16000, ThroughputMbps: 1000, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpooltiflashwn", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpooltiflashwn" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) clusterConfigClient.On("GetEffectiveComponentImageTag", mock.Anything, mock.Anything, apiv1.ComponentType_TIKV).Return("v26.3.4-nextgen", nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) assertLegacyTiFlashSharedPoolComponents(t, result) }, }, { name: "successful creation with comprehensive storage configuration", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolcomprehensive", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1000, Iops: 10000, ThroughputMbps: 500, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 5000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 2000, Iops: 20000, ThroughputMbps: 1000, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1500, Iops: 15000, ThroughputMbps: 750, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolcomprehensive", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolcomprehensive" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) // Verify TiKV data storage tikvGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_TIKV }) assert.True(t, ok) tikvDataStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_DATA }) assert.True(t, ok) assert.Equal(t, entity.GP3, tikvDataStorage.Type) assert.Equal(t, uint32(1000), tikvDataStorage.SizeGi) assert.Equal(t, uint32(10000), tikvDataStorage.Iops) assert.Equal(t, uint32(500), tikvDataStorage.ThroughputMbs) // Verify TiKV raft log storage tikvRaftLogStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_RAFTLOG }) assert.True(t, ok) assert.Equal(t, entity.IO2, tikvRaftLogStorage.Type) assert.Equal(t, uint32(100), tikvRaftLogStorage.SizeGi) assert.Equal(t, uint32(5000), tikvRaftLogStorage.Iops) // Verify TiFlash compute storage tiflashComputeGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_TIFLASH_COMPUTE }) assert.True(t, ok) tiflashComputeDataStorage, ok := lo.Find(tiflashComputeGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_DATA }) assert.True(t, ok) assert.Equal(t, entity.GP3, tiflashComputeDataStorage.Type) assert.Equal(t, uint32(2000), tiflashComputeDataStorage.SizeGi) assert.Equal(t, uint32(20000), tiflashComputeDataStorage.Iops) assert.Equal(t, uint32(1000), tiflashComputeDataStorage.ThroughputMbs) assertColumnarSharedPoolComponents(t, result) }, }, { name: "successful creation with mixed storage types", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolmixed", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.LocalSSD, SizeGi: 0, // LocalSSD should not have size set, will use instance local ssd size }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 50, Iops: 2000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.LocalSSD, SizeGi: 0, // LocalSSD should not have size set, will use instance local ssd size }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 800, Iops: 8000, ThroughputMbps: 400, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolmixed", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolmixed" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: true, LocalDiskSizeGi: 1000, }, { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) // Verify TiKV uses local disk for data storage tikvGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_TIKV }) assert.True(t, ok) assert.True(t, tikvGroup.GetNodeQuota().GetNodeSpec().GetWithLocalDisk()) tikvDataStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_DATA }) assert.True(t, ok) assert.Equal(t, entity.LocalSSD, tikvDataStorage.Type) // Verify TiKV raft log uses IO2 tikvRaftLogStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_RAFTLOG }) assert.True(t, ok) assert.Equal(t, entity.IO2, tikvRaftLogStorage.Type) assert.Equal(t, uint32(50), tikvRaftLogStorage.SizeGi) assert.Equal(t, uint32(2000), tikvRaftLogStorage.Iops) // Verify TiFlash compute uses local disk tiflashComputeGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_TIFLASH_COMPUTE }) assert.True(t, ok) assert.True(t, tiflashComputeGroup.GetNodeQuota().GetNodeSpec().GetWithLocalDisk()) tiflashComputeDataStorage, ok := lo.Find(tiflashComputeGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_DATA }) assert.True(t, ok) assert.Equal(t, entity.LocalSSD, tiflashComputeDataStorage.Type) assertColumnarSharedPoolComponents(t, result) }, }, { name: "successful creation with service CIDR", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolservicecidr", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", ServiceCidr: "172.16.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 500, Iops: 5000, ThroughputMbps: 250, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 50, Iops: 3000, ThroughputMbps: 125, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 400, Iops: 4000, ThroughputMbps: 200, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 300, Iops: 3000, ThroughputMbps: 150, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolservicecidr", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolservicecidr" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) assert.Equal(t, "testsharedpoolservicecidr", result.GetId()) assert.Equal(t, apiv1.SharedPool_CREATING, result.GetPhase()) }, }, { name: "invalid service CIDR - overlaps with network CIDR", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpooloverlap", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", ServiceCidr: "10.0.100.0/24", // Overlaps with NetworkCidr TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 500, Iops: 5000, ThroughputMbps: 250, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 50, Iops: 3000, ThroughputMbps: 125, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 400, Iops: 4000, ThroughputMbps: 200, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 300, Iops: 3000, ThroughputMbps: 150, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpooloverlap", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) }, expectedError: "service cidr 10.0.100.0/24 and network cidr 10.0.0.0/16 cannot overlap", expectedCode: codes.InvalidArgument, }, { name: "invalid service CIDR - network CIDR contains service CIDR", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolcontains", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "192.168.0.0/16", ServiceCidr: "192.168.1.0/24", // Contained in NetworkCidr TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 500, Iops: 5000, ThroughputMbps: 250, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 50, Iops: 3000, ThroughputMbps: 125, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 400, Iops: 4000, ThroughputMbps: 200, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 300, Iops: 3000, ThroughputMbps: 150, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolcontains", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) }, expectedError: "service cidr 192.168.1.0/24 and network cidr 192.168.0.0/16 cannot overlap", expectedCode: codes.InvalidArgument, }, { name: "invalid service CIDR format", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolinvalidformat", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", ServiceCidr: "invalid-cidr", // Invalid format TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 500, Iops: 5000, ThroughputMbps: 250, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 50, Iops: 3000, ThroughputMbps: 125, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 400, Iops: 4000, ThroughputMbps: 200, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 300, Iops: 3000, ThroughputMbps: 150, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolinvalidformat", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) }, expectedError: "service cidr invalid-cidr is invalid", expectedCode: codes.InvalidArgument, }, { name: "successful creation with AliCloud CloudAuto storage", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolalicloudauto", Provider: &apiv1.Provider{ Name: "alicloud", Type: apiv1.ProviderType_ALICLOUD, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 500, Iops: 0, // CloudAuto doesn't require IOPS }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 50, Iops: 0, // CloudAuto doesn't require IOPS }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 400, Iops: 0, // CloudAuto doesn't require IOPS }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 300, Iops: 0, // CloudAuto doesn't require IOPS }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolalicloudauto", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolalicloudauto" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) // Verify TiKV storage configuration tikvGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_TIKV }) assert.True(t, ok) tikvDataStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_DATA }) assert.True(t, ok) assert.Equal(t, entity.CloudAuto, tikvDataStorage.Type) assert.Equal(t, uint32(500), tikvDataStorage.SizeGi) assert.Equal(t, uint32(0), tikvDataStorage.Iops) // CloudAuto doesn't use IOPS }, }, { name: "successful creation with AliCloud LocalSSD storage", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolalicloudlocalssd", Provider: &apiv1.Provider{ Name: "alicloud", Type: apiv1.ProviderType_ALICLOUD, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.LocalSSD, SizeGi: 0, // LocalSSD doesn't allow setting size Iops: 0, // LocalSSD doesn't support IOPS ThroughputMbps: 0, // LocalSSD doesn't support throughput }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 50, Iops: 0, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.LocalSSD, SizeGi: 0, // LocalSSD doesn't allow setting size Iops: 0, // LocalSSD doesn't support IOPS ThroughputMbps: 0, // LocalSSD doesn't support throughput }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 300, Iops: 0, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolalicloudlocalssd", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolalicloudlocalssd" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: true, LocalDiskSizeGi: 1862, // i3.2xlarge has 1862 GiB NVMe SSD }, { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, // For TiFlash WN without LocalDisk }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) }, validate: func(t *testing.T, result *entity.SharedPool, err error) { t.Helper() require.NoError(t, err) require.NotNil(t, result) // Verify TiKV storage configuration tikvGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_TIKV }) assert.True(t, ok) tikvDataStorage, ok := lo.Find(tikvGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_DATA }) assert.True(t, ok) assert.Equal(t, entity.LocalSSD, tikvDataStorage.Type) assert.Equal(t, uint32(1862), tikvDataStorage.SizeGi) // Should be set from LocalDiskSizeGi assert.Equal(t, uint32(0), tikvDataStorage.Iops) // LocalSSD doesn't use IOPS // Verify TiFlash CN storage configuration tiflashCnGroup, ok := lo.Find(result.ComponentGroups, func(item *apiv1.Component) bool { return item.ComponentType == apiv1.ComponentType_TIFLASH_COMPUTE }) assert.True(t, ok) tiflashCnStorage, ok := lo.Find(tiflashCnGroup.Storages, func(item *apiv1.Component_Storage) bool { return item.Name == apiv1.Component_Storage_DATA }) assert.True(t, ok) assert.Equal(t, entity.LocalSSD, tiflashCnStorage.Type) assert.Equal(t, uint32(1862), tiflashCnStorage.SizeGi) // Should be set from LocalDiskSizeGi assert.Equal(t, uint32(0), tiflashCnStorage.Iops) // LocalSSD doesn't use IOPS }, }, { name: "AliCloud LocalSSD with invalid iops set", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolalicloudlocalssd", Provider: &apiv1.Provider{ Name: "alicloud", Type: apiv1.ProviderType_ALICLOUD, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.LocalSSD, SizeGi: 0, Iops: 3000, // Invalid: LocalSSD doesn't support IOPS }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 50, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 400, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 300, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolalicloudlocalssd", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) }, expectedError: "tikv data local-ssd storage does not support setting iops", expectedCode: codes.InvalidArgument, }, { name: "AliCloud LocalSSD with invalid size set", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolalicloudlocalssdsize", Provider: &apiv1.Provider{ Name: "alicloud", Type: apiv1.ProviderType_ALICLOUD, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.LocalSSD, SizeGi: 500, // Invalid: LocalSSD doesn't allow setting size Iops: 0, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 50, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 400, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 300, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolalicloudlocalssdsize", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) }, expectedError: "tikv data local-ssd storage does not support setting size (size is determined by node's local disk)", expectedCode: codes.InvalidArgument, }, { name: "AliCloud LocalSSD with invalid throughput set", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolalicloudlocalssdthroughput", Provider: &apiv1.Provider{ Name: "alicloud", Type: apiv1.ProviderType_ALICLOUD, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.LocalSSD, SizeGi: 0, Iops: 0, ThroughputMbps: 125, // Invalid: LocalSSD doesn't support throughput }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 50, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 400, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 300, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolalicloudlocalssdthroughput", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) }, expectedError: "tikv data local-ssd storage does not support setting throughput", expectedCode: codes.InvalidArgument, }, { name: "AliCloud CloudAuto with invalid throughput set", sharedPool: &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolalicloudautothroughput", Provider: &apiv1.Provider{ Name: "alicloud", Type: apiv1.ProviderType_ALICLOUD, }, }, TenantId: "test-tenant", }, requestUnit: &entity.RequestUnit{ Min: 1000, Max: 2000, }, initConfig: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 500, ThroughputMbps: 125, // Invalid: CloudAuto doesn't support throughput }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 50, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 400, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.CloudAuto, SizeGi: 300, }, }, setupMocks: func(sharedPoolRepo *repomnocks.SharedPoolRepo, lockRepo *repomnocks.LockRepo, ruConfigRepo *repomnocks.RuConfigRepo, dataplaneClient *dataplanemgrmocks.Client, clusterConfigClient *clusterconfigmocks.Client, dagExecutor *executormocks.RemoteExecutor) { sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolalicloudautothroughput", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) }, expectedError: "tikv data cloud-auto storage does not support setting throughput", expectedCode: codes.InvalidArgument, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { // Create mocks sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterConfigClient := clusterconfigmocks.NewClient(t) dagExecutor := executormocks.NewRemoteExecutor(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) // Setup mocks tt.setupMocks(sharedPoolRepo, lockRepo, ruConfigRepo, dataplaneClient, clusterConfigClient, dagExecutor) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything). Return("v8.0.0", nil). Maybe() clusterConfigClient.On("GetEffectiveComponentImageTag", mock.Anything, mock.Anything, apiv1.ComponentType_TIKV). Return("v26.3.5-nextgen", nil). Maybe() // Create a simple mock transactioner that just executes the function mockTransactioner := &mockTransactioner{} // Determine cloudProvider from test case cloudProvider := "AWS" region := "us-west-2" if tt.sharedPool != nil && tt.sharedPool.Provider != nil && tt.sharedPool.Provider.Type == apiv1.ProviderType_ALICLOUD { cloudProvider = "alicloud" region = "cn-hangzhou" } // Create server implementation server := &serverImpl{ cloudProvider: cloudProvider, region: region, sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dataplaneClient: dataplaneClient, clusterConfigClient: clusterConfigClient, dagExecutor: dagExecutor, transactioner: mockTransactioner, bizCfg: &config.BizConfig{ DefaultDataplaneAccount: "test-account", }, } // Execute test result, err := server.CreateSharedPool(context.Background(), tt.sharedPool, tt.requestUnit, tt.initConfig, tt.options...) 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)) } if tt.expectedReason != "" { assert.Equal(t, tt.expectedReason, errorInfoReason(err)) } assert.Nil(t, result) case tt.validate != nil: tt.validate(t, result, err) default: require.NoError(t, err) require.NotNil(t, result) } // Verify all mocks were called as expected (mocks with cleanup don't need manual assertion) }) } } // TestCreateSharedPoolInheritsFeatureFlagsFromTenant verifies that new shared // pools faithfully inherit the tenant's feature flags without overriding them. // proto3 zero-values (false) from the tenant DB row pass through as-is; // no service-layer or repo-layer override is applied. func TestCreateSharedPoolInheritsFeatureFlagsFromTenant(t *testing.T) { sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterConfigClient := clusterconfigmocks.NewClient(t) dagExecutor := executormocks.NewRemoteExecutor(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) tenantRuConfig := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ FeatureFlags: &apiv1.RuConfig_FeatureFlags{ EnableScaleSharedPoolByRu: proto.Bool(false), EnableScaleTidbGroupByRu: proto.Bool(false), EnableAutoscaleTikvCpuScaleOut: proto.Bool(false), EnableAutoscaleDiskExpand: proto.Bool(false), EnableResourceManager: true, }, Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tikv": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tikv_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "coprocessor_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tidb": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tidb_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "pd": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "pd_tso": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "pd_scheduling": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tiflash_write": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tiflash_compute": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tiproxy": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "resource_manager": { MaxNodeCount: 2, NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "2c4g": {Min: 0, Max: 0}, }, }, }, }, } sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolflags", true).Return(nil, status.Error(codes.NotFound, "not found")) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolflags" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, "test-tenant", apiv1.RuConfig_TENANT).Return(tenantRuConfig, nil) var savedRuConfig *entity.RuConfig ruConfigRepo.On("SetRuConfig", mock.Anything, "testsharedpoolflags", apiv1.RuConfig_SHARED_POOL, mock.Anything). Run(func(args mock.Arguments) { savedRuConfig = args.Get(3).(*entity.RuConfig) }). Return(nil) var savedSharedPoolProfileIDs []string sharedPoolRepo.On("Save", mock.Anything, mock.Anything). Run(func(args mock.Arguments) { savedSharedPoolProfileIDs = append( savedSharedPoolProfileIDs, args.Get(1).(*entity.SharedPool).ConfigCenterMetaProfileId, ) }). Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, { CpuCore: 2, MemoryGi: 4, AvailableCpuMilliCore: 2000, AvailableMemoryMi: 4000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, "test-tenant").Return("v8.0.0", nil) clusterConfigClient.On("GetEffectiveComponentImageTag", mock.Anything, mock.Anything, apiv1.ComponentType_TIKV).Return("v26.3.5-nextgen", nil) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) server := &serverImpl{ cloudProvider: "AWS", region: "us-west-2", sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dataplaneClient: dataplaneClient, clusterConfigClient: clusterConfigClient, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, bizCfg: &config.BizConfig{ DefaultDataplaneAccount: "test-account", }, } result, err := server.CreateSharedPool(context.Background(), &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolflags", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, &entity.RequestUnit{ Min: 1000, Max: 2000, }, &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", ConfigCenterMetaProfileId: "1001", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1000, Iops: 10000, ThroughputMbps: 500, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 5000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 2000, Iops: 20000, ThroughputMbps: 1000, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1500, Iops: 15000, ThroughputMbps: 750, }, }) require.NoError(t, err) require.NotNil(t, result) require.NotNil(t, savedRuConfig) assert.Equal(t, "1001", result.ConfigCenterMetaProfileId) require.NotEmpty(t, savedSharedPoolProfileIDs) assert.Equal(t, "1001", savedSharedPoolProfileIDs[0]) // Saved RuConfig should faithfully inherit the tenant's flags. assert.False(t, savedRuConfig.GetFeatureFlags().GetEnableScaleSharedPoolByRu()) assert.False(t, savedRuConfig.GetFeatureFlags().GetEnableScaleTidbGroupByRu()) assert.False(t, savedRuConfig.GetFeatureFlags().GetEnableAutoscaleTikvCpuScaleOut()) assert.False(t, savedRuConfig.GetFeatureFlags().GetEnableAutoscaleDiskExpand()) assert.True(t, savedRuConfig.GetFeatureFlags().GetEnableResourceManager()) require.Len(t, savedRuConfig.GetComponents(), len(tenantRuConfig.GetComponents())) require.Contains(t, savedRuConfig.GetComponents(), "resource_manager") assert.Equal(t, uint32(2), savedRuConfig.GetComponents()["resource_manager"].GetMaxNodeCount()) require.Contains(t, savedRuConfig.GetComponents(), "tikv") assert.Equal(t, uint32(100000000), savedRuConfig.GetComponents()["tikv"].GetNodeSpecRuRange()["8c16g"].GetMax()) // Tenant config must not be mutated. assert.False(t, tenantRuConfig.GetFeatureFlags().GetEnableScaleSharedPoolByRu()) assert.False(t, tenantRuConfig.GetFeatureFlags().GetEnableScaleTidbGroupByRu()) assert.False(t, tenantRuConfig.GetFeatureFlags().GetEnableAutoscaleTikvCpuScaleOut()) assert.False(t, tenantRuConfig.GetFeatureFlags().GetEnableAutoscaleDiskExpand()) assert.True(t, tenantRuConfig.GetFeatureFlags().GetEnableResourceManager()) } // TestCreateSharedPoolResolvesFlagsFromBundleTable verifies the new // `resolve.PriorityResolveForSharedPoolFromBundle` entry point is // wired into CreateSharedPool and that the DEFAULT bundle row's // policy flags reach the saved SHARED_POOL ru_configs.FeatureFlags // even when the tenant's `ru_configs.FeatureFlags` row says nothing // about them. This is the canonical scenario the double-resolution // refactor targets — a DEFAULT-scope policy set via // `CreateOrReplaceAutoscalePolicy(SCOPE_DEFAULT, ...)` must reach // every newly created shared pool in one pass, not be silently // ignored for tenants without a per-tenant ru_configs row. func TestCreateSharedPoolResolvesFlagsFromBundleTable(t *testing.T) { sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterConfigClient := clusterconfigmocks.NewClient(t) dagExecutor := executormocks.NewRemoteExecutor(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) bundleRepo := repomnocks.NewAutoscalePolicyBundleRepo(t) const tenantID = "test-tenant-bundle" // Tenant's `ru_configs.FeatureFlags` row is intentionally empty // (no 5 autoscale flags set). The OLD code path required this // row to be populated, otherwise DEFAULT-scope policies would be // silently lost. The NEW path reads the bundle table directly. tenantRuConfig := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ FeatureFlags: nil, Components: minimalRuConfigComponents(), }, } // DEFAULT bundle row: TIKV CPU scale-out ENABLED + disk-expand ENABLED. defaultBundle := &repo.AutoscalePolicyBundleRow{ Scope: apiv1.AutoscalePolicyScope_SCOPE_DEFAULT, SubjectID: "__default__", Specs: []*entity.AutoscalePolicySpec{ { TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_ENABLED, }, { TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_DISK_SIZE, Trigger: apiv1.AutoscalePolicy_DISK, State: apiv1.AutoscalePolicy_ENABLED, }, }, } // TENANT bundle row: TIKV CPU scale-out explicitly DISABLED. // Expected: TIKV CPU = false (TENANT wins), disk-expand = true // (DEFAULT fills in), other 3 flags fall through to server default. tenantBundle := &repo.AutoscalePolicyBundleRow{ Scope: apiv1.AutoscalePolicyScope_SCOPE_TENANT, SubjectID: tenantID, Specs: []*entity.AutoscalePolicySpec{ { TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_DISABLED, }, }, } sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolbundle", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolbundle" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, tenantID, apiv1.RuConfig_TENANT).Return(tenantRuConfig, nil) // Server default has all 5 autoscale flags enabled. ruConfigRepo.On("GetServerDefaultRuConfig", mock.Anything).Return(&entity.RuConfig{ RuConfig: apiv1.RuConfig{ FeatureFlags: &apiv1.RuConfig_FeatureFlags{ EnableScaleSharedPoolByRu: proto.Bool(true), EnableScaleTidbGroupByRu: proto.Bool(true), EnableAutoscaleTikvCpuScaleOut: proto.Bool(true), EnableAutoscaleDiskExpand: proto.Bool(true), EnableAutoscaleIopsThroughputExpand: proto.Bool(true), EnableAutoscaleTiproxyScaleIn: proto.Bool(true), }, }, }, nil) // Bundle repo reads — DEFAULT first, then TENANT. bundleRepo.On("GetBundle", mock.Anything, apiv1.AutoscalePolicyScope_SCOPE_DEFAULT, "__default__"). Return(defaultBundle, nil).Once() bundleRepo.On("GetBundle", mock.Anything, apiv1.AutoscalePolicyScope_SCOPE_TENANT, tenantID). Return(tenantBundle, nil).Once() var savedRuConfig *entity.RuConfig ruConfigRepo.On("SetRuConfig", mock.Anything, "testsharedpoolbundle", apiv1.RuConfig_SHARED_POOL, mock.Anything). Run(func(args mock.Arguments) { savedRuConfig = args.Get(3).(*entity.RuConfig) }). Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, tenantID).Return("v8.0.0", nil) clusterConfigClient.On("GetEffectiveComponentImageTag", mock.Anything, mock.Anything, apiv1.ComponentType_TIKV).Return("v26.3.5-nextgen", nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) server := &serverImpl{ cloudProvider: "AWS", region: "us-west-2", sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, autoscalePolicyBundleRepo: bundleRepo, dataplaneClient: dataplaneClient, clusterConfigClient: clusterConfigClient, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, bizCfg: &config.BizConfig{ DefaultDataplaneAccount: "test-account", AutoscalePolicy: config.AutoscalePolicyConfig{ Mode: config.AutoscalePolicyModeEnforce, AllowAllTenants: true, }, }, } result, err := server.CreateSharedPool(context.Background(), &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolbundle", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: tenantID, }, &entity.RequestUnit{Min: 1000, Max: 2000}, &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1000, Iops: 10000, ThroughputMbps: 500, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 5000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 2000, Iops: 20000, ThroughputMbps: 1000, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1500, Iops: 15000, ThroughputMbps: 750, }, }) require.NoError(t, err) require.NotNil(t, result) require.NotNil(t, savedRuConfig) ff := savedRuConfig.GetFeatureFlags() require.NotNil(t, ff) // TENANT DISABLED wins over DEFAULT ENABLED for TIKV CPU. assert.False(t, ff.GetEnableAutoscaleTikvCpuScaleOut(), "TENANT DISABLED must override DEFAULT ENABLED for TIKV CPU scale-out") // DEFAULT ENABLED fills in the disk-expand slot — TENANT did not set it. assert.True(t, ff.GetEnableAutoscaleDiskExpand(), "DEFAULT ENABLED must fill in for disk-expand when TENANT is unset") // Other 3 flags fall through to server default (all enabled). assert.True(t, ff.GetEnableScaleSharedPoolByRu()) assert.True(t, ff.GetEnableScaleTidbGroupByRu()) assert.True(t, ff.GetEnableAutoscaleIopsThroughputExpand()) assert.True(t, ff.GetEnableAutoscaleTiproxyScaleIn()) // Bundle repo MUST be read for both DEFAULT and TENANT — the // whole point of the refactor is that the double-resolution // case is no longer reachable. bundleRepo.AssertExpectations(t) } // minimalRuConfigComponents returns the minimal Components map needed // for `generatedSharedPoolComponent` to produce a valid shared pool. // Shared by the bundle-aware CreateSharedPool test. func minimalRuConfigComponents() map[string]*apiv1.RuConfig_ComponentsMap { return map[string]*apiv1.RuConfig_ComponentsMap{ "tikv": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tikv_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "coprocessor_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tidb": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tidb_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "pd": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "pd_tso": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "pd_scheduling": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tiflash_write": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tiflash_compute": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tiproxy": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, } } func TestCreateSharedPoolWithTiKVRemoteWorkerOnlyTopology(t *testing.T) { sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterConfigClient := clusterconfigmocks.NewClient(t) dagExecutor := executormocks.NewRemoteExecutor(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) tenantRuConfig := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tikv": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tikv_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "coprocessor_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tidb": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tidb_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "pd": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "pd_tso": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "pd_scheduling": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tiflash_write": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tiflash_compute": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, "tiproxy": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": {Min: 0, Max: 100000000}, }, }, }, }, } sharedPoolRepo.On("Load", mock.Anything, "testworkeronly", true).Return(nil, status.Error(codes.NotFound, "not found")) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testworkeronly" })).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, "test-tenant", apiv1.RuConfig_TENANT).Return(tenantRuConfig, nil) ruConfigRepo.On("SetRuConfig", mock.Anything, "testworkeronly", apiv1.RuConfig_SHARED_POOL, mock.Anything).Return(nil) var savedSharedPool *entity.SharedPool sharedPoolRepo.On("Save", mock.Anything, mock.Anything). Run(func(args mock.Arguments) { savedSharedPool = args.Get(1).(*entity.SharedPool) }). Return(nil) dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(&dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, }), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, "test-tenant").Return("v8.0.0", nil) clusterConfigClient.On("GetEffectiveComponentImageTag", mock.Anything, mock.Anything, apiv1.ComponentType_TIKV).Return("v26.3.5-nextgen", nil) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) server := &serverImpl{ cloudProvider: "AWS", region: "us-west-2", sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dataplaneClient: dataplaneClient, clusterConfigClient: clusterConfigClient, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, bizCfg: &config.BizConfig{ DefaultDataplaneAccount: "test-account", }, } result, err := server.CreateSharedPool(context.Background(), &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testworkeronly", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, EnabledTikvRemoteWorkers: []apiv1.ComponentType{ apiv1.ComponentType_TIKV_WORKER, }, }, TenantId: "test-tenant", }, &entity.RequestUnit{ Min: 1000, Max: 2000, }, &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1000, Iops: 10000, ThroughputMbps: 500, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.IO2, SizeGi: 100, Iops: 5000, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 2000, Iops: 20000, ThroughputMbps: 1000, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: entity.GP3, SizeGi: 1500, Iops: 15000, ThroughputMbps: 750, }, }) require.NoError(t, err) require.NotNil(t, result) require.NotNil(t, savedSharedPool) for _, sharedPool := range []*entity.SharedPool{result, savedSharedPool} { assert.Equal(t, []apiv1.ComponentType{apiv1.ComponentType_TIKV_WORKER}, sharedPool.GetEnabledTikvRemoteWorkers()) assert.True(t, lo.ContainsBy(sharedPool.GetComponentGroups(), func(component *apiv1.Component) bool { return component.GetComponentType() == apiv1.ComponentType_TIKV_WORKER })) assert.False(t, lo.ContainsBy(sharedPool.GetComponentGroups(), func(component *apiv1.Component) bool { return component.GetComponentType() == apiv1.ComponentType_COPROCESSOR_WORKER })) assertColumnarSharedPoolComponents(t, sharedPool) } } func TestCreateSharedPoolTopologyDrivenTiflashMode(t *testing.T) { component := func(componentType apiv1.ComponentType) *apiv1.Component { return &apiv1.Component{ ComponentType: componentType, NodeQuota: &apiv1.NodeQuota{ NodeCount: 1, NodeSpec: &apiv1.NodeSpec{Cpu: 8, MemoryGi: 16}, }, } } tests := []struct { name string id string requestedMode apiv1.TiflashMode assertResult func(*testing.T, *entity.SharedPool) }{ {name: "shared TiProxy AUTO stays legacy", id: "topologyauto", assertResult: assertLegacyTiFlashSharedPoolComponents}, {name: "explicit COLUMNAR uses columnar", id: "topologycolumnar", requestedMode: apiv1.TiflashMode_TIFLASH_MODE_COLUMNAR, assertResult: assertColumnarSharedPoolComponents}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterConfigClient := clusterconfigmocks.NewClient(t) dagExecutor := executormocks.NewRemoteExecutor(t) sharedPoolRepo.On("Load", mock.Anything, tt.id, true).Return(nil, status.Error(codes.NotFound, "not found")) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == tt.id })).Return(func() error { return nil }, nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{{ CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 7000, AvailableMemoryMi: 15000, NodeType: dplanemgrv1.NodeType_Dedicated, }}} dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(nodePoolConfigs, nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, "shared-multi-tenant").Return("v8.0.0", nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) server := &serverImpl{ cloudProvider: "AWS", region: "us-west-2", sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, dataplaneClient: dataplaneClient, clusterConfigClient: clusterConfigClient, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, bizCfg: &config.BizConfig{DefaultDataplaneAccount: "test-account"}, } topologyComponents := []*apiv1.Component{ component(apiv1.ComponentType_PD), component(apiv1.ComponentType_TIKV), component(apiv1.ComponentType_TIKV_WORKER), component(apiv1.ComponentType_TIDB_WORKER), } if tt.requestedMode == apiv1.TiflashMode_TIFLASH_MODE_COLUMNAR { schemaManager := component(apiv1.ComponentType_SCHEMA_MANAGER) schemaManager.Storages = []*apiv1.Component_Storage{{ Name: apiv1.Component_Storage_DATA, Type: entity.GP3, SizeGi: 42, Iops: 4000, ThroughputMbs: 200, }} topologyComponents = append(topologyComponents, component(apiv1.ComponentType_TIFLASH_COMPUTE), schemaManager, ) } result, err := server.CreateSharedPool(context.Background(), &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: tt.id, CapacityMode: apiv1.CapacityMode_TOPOLOGY_DRIVEN, Provider: &apiv1.Provider{Name: "aws", Type: apiv1.ProviderType_AWS}, TopologySpec: &apiv1.TopologySpec{Components: topologyComponents}, }, TenantId: "shared-multi-tenant", DeployType: schema.SharedPoolDeployTypeSharedTiProxy, RequestedTiflashMode: tt.requestedMode, }, nil, &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 1000, Iops: 10000, ThroughputMbps: 500}, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.IO2, SizeGi: 100, Iops: 5000}, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 2000, Iops: 20000, ThroughputMbps: 1000}, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 1500, Iops: 15000, ThroughputMbps: 750}, }) require.NoError(t, err) require.NotNil(t, result) tt.assertResult(t, result) if tt.requestedMode == apiv1.TiflashMode_TIFLASH_MODE_COLUMNAR { schemaManager, ok := lo.Find(result.GetComponentGroups(), func(component *apiv1.Component) bool { return component.GetComponentType() == apiv1.ComponentType_SCHEMA_MANAGER }) require.True(t, ok) require.Len(t, schemaManager.GetStorages(), 1) assert.Equal(t, uint32(42), schemaManager.GetStorages()[0].GetSizeGi()) assert.Equal(t, uint32(4000), schemaManager.GetStorages()[0].GetIops()) assert.Equal(t, uint32(200), schemaManager.GetStorages()[0].GetThroughputMbs()) } else { tiflashWriteGroup, ok := lo.Find(result.GetComponentGroups(), func(component *apiv1.Component) bool { return component.GetComponentType() == apiv1.ComponentType_TIFLASH_WRITE }) require.True(t, ok) dataStorage, ok := lo.Find(tiflashWriteGroup.GetStorages(), func(storage *apiv1.Component_Storage) bool { return storage.GetName() == apiv1.Component_Storage_DATA }) require.True(t, ok) assert.Equal(t, entity.GP3, dataStorage.GetType()) assert.Equal(t, uint32(1500), dataStorage.GetSizeGi()) assert.Equal(t, uint32(15000), dataStorage.GetIops()) assert.Equal(t, uint32(750), dataStorage.GetThroughputMbs()) } }) } } // TestCreateSharedPoolPlacementGroupGatePublicPath is the regression test // for the public CreateSharedPool gRPC path. fgksgf review (round 2) // noted that the existing TestCreateSharedPoolWithProviderTypes injects // Provider.NodeZones directly into the entity, which the public request // cannot do (Provider is OUTPUT_ONLY and the registered app adapter does // not populate it). Earlier PR #3040 derivations (pre-round 1) read // sharedPool.Provider.NodeZones, so the public AWS ZONAL request always // saw zero zones and persisted Provider.PlacementGroupEnabled=false. // // This test exercises the actual public-request shape on AWS: caller // only sets Provider.Name + Provider.Type, and the canonical workload // topology arrives via initConfig.NodeZones. // // PR #3040 round 1 derives Provider.PlacementGroupEnabled from // initConfig.NodeZones after HA resolution. Round 4 made the derivation // HA-aware (REGIONAL → false; ZONAL with 0 or 1 caller zone → follow // provider) and added an upfront shape validation that rejects illegal // combinations (REGIONAL+1 caller zone, ZONAL+>1 caller zones) with // codes.InvalidArgument, mirroring DPM resolveNodeZonesForCreate. // // ZONAL+0 caller zone now persists gate=true for AWS because DPM // auto-fills from regionConfig.nodeZones (single zone) and the effective // topology is single-AZ. The earlier "gate stays off" expectation was // the round 3 follow-up correctness issue the reviewer flagged: CSNG // persisted false while the effective topology was single-AZ, leaving // the actual cluster without a placement group. // // Illegal-shape coverage lives in // TestCreateSharedPoolPlacementGroupGateRejectsIllegalShape below. func TestCreateSharedPoolPlacementGroupGatePublicPath(t *testing.T) { tests := []struct { name string haType apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_HighAvailabilityType callerNodeZones []string expectedPlacementGate bool }{ { // fgksgf regression: AWS ZONAL public request must enable the // gate even though Provider.NodeZones is empty. name: "AWS ZONAL with 1 caller zone enables gate (canonical public input)", haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, callerNodeZones: []string{"us-west-2a"}, expectedPlacementGate: true, }, { // AWS ZONAL with no caller zones: DPM auto-fills from // regionConfig.nodeZones to a single zone, so effective // topology is single-AZ and the gate follows. PR #3040 round // 4 derives the gate from the post-HA effective topology, so // this case persists true for AWS (the previous round 3 // expectation of false was a follow-up correctness issue // called out by the reviewer: CSNG would persist false while // the effective topology was single-AZ, leaving the cluster // without a placement group). name: "AWS ZONAL with no caller zones enables gate (DPM auto-fills single-AZ from regionConfig.nodeZones)", haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, callerNodeZones: nil, expectedPlacementGate: true, }, { // AWS REGIONAL with no caller zones: DPM auto-fills from // regionConfig.zones (multi-AZ), so effective topology is // multi-AZ and the gate stays off. This is the canonical // public-path shape: caller does not pin a specific zone list // for regional clusters; the bring-up layer sources it from // the region config. name: "AWS REGIONAL with no caller zones keeps gate off (DPM auto-fills multi-AZ from regionConfig.zones)", haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_REGIONAL, callerNodeZones: nil, expectedPlacementGate: false, }, { // AWS REGIONAL with 2+ caller zones: caller-specified // multi-AZ layout. Effective topology is multi-AZ; gate off. name: "AWS REGIONAL with 3 caller zones keeps gate off (caller-specified multi-AZ)", haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_REGIONAL, callerNodeZones: []string{"us-west-2a", "us-west-2b", "us-west-2c"}, expectedPlacementGate: false, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterConfigClient := clusterconfigmocks.NewClient(t) dagExecutor := executormocks.NewRemoteExecutor(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) rc := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tikv": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tikv_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "coprocessor_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tidb": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tidb_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd_tso": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd_scheduling": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiflash_write": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiflash_compute": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiproxy": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, }, }, } sharedPoolRepo.On("Load", mock.Anything, mock.Anything, true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(rc, nil) dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(&dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ {CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 3800, AvailableMemoryMi: 7600, NodeType: dplanemgrv1.NodeType_Dedicated}, }, }), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) clusterConfigClient.On("GetEffectiveComponentImageTag", mock.Anything, mock.Anything, apiv1.ComponentType_TIKV).Return("v8.0.0", nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) server := &serverImpl{ cloudProvider: "AWS", region: "us-west-2", sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dataplaneClient: dataplaneClient, clusterConfigClient: clusterConfigClient, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, bizCfg: &config.BizConfig{ DefaultDataplaneAccount: "test-account", }, } // Public request shape: Provider has Name + Type only (NodeZones // is OUTPUT_ONLY and the gRPC adapter does not populate it). // Workload topology arrives via initConfig.NodeZones. HA type // must be explicit so caller zones are not dropped by the // REGIONAL auto-fill path. result, err := server.CreateSharedPool(context.Background(), &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testpublicpathpg", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, &entity.RequestUnit{Min: 1000, Max: 2000}, &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", HighAvailabilityType: tt.haType, NodeZones: tt.callerNodeZones, TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 100, Iops: 3000, ThroughputMbps: 125}, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 50, Iops: 3000, ThroughputMbps: 125}, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 100, Iops: 3000, ThroughputMbps: 125}, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 100, Iops: 3000, ThroughputMbps: 125}, }) require.NoError(t, err) require.NotNil(t, result) assert.Equal(t, tt.expectedPlacementGate, result.GetProvider().GetPlacementGroupEnabled(), "public-path gate must derive from initConfig.NodeZones + HA, not Provider.NodeZones") }) } } // TestCreateSharedPoolPlacementGroupGateRejectsIllegalShape pins the // upfront shape validation in CreateSharedPool. The HA + caller-zone // combinations below are illegal under the same contract that DPM // resolveNodeZonesForCreate enforces at dataplaneservice.go:438-441 // (REGIONAL + 1 caller zone) and 484-488 (ZONAL + >1 caller zones). // CSNG rejects them with codes.InvalidArgument before any SharedPool // record is persisted or any DAG is submitted, so the durable gate is // never written for these cases and the error message names the // caller-side cause instead of bubbling up as a confusing regionConfig // error from DPM. func TestCreateSharedPoolPlacementGroupGateRejectsIllegalShape(t *testing.T) { tests := []struct { name string haType apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_HighAvailabilityType callerNodeZones []string expectedErrMsg string }{ { // Round 4 reviewer blocker: REGIONAL + 1 caller zone was // previously accepted at CSNG (gate derived as true, caller // zones dropped before DAG, DPM auto-fills multi-AZ), then // wedged at CreateNodePool time with wire=true on a // multi-AZ dataplane. CSNG now rejects it upfront. name: "AWS REGIONAL with 1 caller zone is rejected upfront", haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_REGIONAL, callerNodeZones: []string{"us-west-2a"}, expectedErrMsg: "1 zone means a zonal deployment", }, { // Mirror DPM's ZONAL must-be-exactly-1 rejection so CSNG // fails fast on a shape the public request can never // produce a usable dataplane for. name: "AWS ZONAL with 2 caller zones is rejected upfront", haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, callerNodeZones: []string{"us-west-2a", "us-west-2b"}, expectedErrMsg: "must contain at most 1 zone for zonal", }, { name: "AWS ZONAL with 3 caller zones is rejected upfront", haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, callerNodeZones: []string{"us-west-2a", "us-west-2b", "us-west-2c"}, expectedErrMsg: "must contain at most 1 zone for zonal", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterConfigClient := clusterconfigmocks.NewClient(t) dagExecutor := executormocks.NewRemoteExecutor(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) rc := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tikv": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tikv_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "coprocessor_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tidb": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tidb_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd_tso": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd_scheduling": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiflash_write": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiflash_compute": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiproxy": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, }, }, } // No mocks for Save / Submit are registered: the validation // must short-circuit before either runs. The test fails // noisily if the validation regresses and the request // actually reaches the DAG layer. Pre-validation mocks are // marked Maybe so the strict mode doesn't fail when the // rejection happens before they are reached. sharedPoolRepo.On("Load", mock.Anything, mock.Anything, true).Maybe().Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Maybe().Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.Anything).Maybe().Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Maybe().Return(rc, nil) dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Maybe().Return(withSchemaManagerNodeSpec(&dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ {CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 3800, AvailableMemoryMi: 7600, NodeType: dplanemgrv1.NodeType_Dedicated}, }, }), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Maybe().Return("v8.0.0", nil) clusterConfigClient.On("GetEffectiveComponentImageTag", mock.Anything, mock.Anything, apiv1.ComponentType_TIKV).Maybe().Return("v8.0.0", nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Maybe().Return(nil) server := &serverImpl{ cloudProvider: "AWS", region: "us-west-2", sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dataplaneClient: dataplaneClient, clusterConfigClient: clusterConfigClient, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, bizCfg: &config.BizConfig{ DefaultDataplaneAccount: "test-account", }, } _, err := server.CreateSharedPool(context.Background(), &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testillegalpg", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, &entity.RequestUnit{Min: 1000, Max: 2000}, &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", HighAvailabilityType: tt.haType, NodeZones: tt.callerNodeZones, TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 100, Iops: 3000, ThroughputMbps: 125}, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 50, Iops: 3000, ThroughputMbps: 125}, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 100, Iops: 3000, ThroughputMbps: 125}, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 100, Iops: 3000, ThroughputMbps: 125}, }) require.Error(t, err, "illegal HA + zone shape must be rejected upfront") st, ok := status.FromError(err) require.True(t, ok, "rejection must be a gRPC status") assert.Equal(t, codes.InvalidArgument, st.Code(), "illegal-shape rejection must be InvalidArgument, not Internal") assert.Contains(t, st.Message(), tt.expectedErrMsg) }) } } func TestCreateSharedPoolWithProviderTypes(t *testing.T) { // placementGroupEnabled is derived from the shared pool's topology and // provider at CreateSharedPool time, not from the user-set field. The // test pins the contract: single-AZ AWS gets the gate on; single-AZ // Azure/GCP/ACK (and multi-AZ on any provider) succeed with the gate // off. PG is an EKS-only abstraction, so non-AWS pools are never // rejected for being single-AZ — they just stay outside the partition // path. Legacy shared pools (created before this change) keep their // user-set value; CreateSharedPool only runs for new pools. tests := []struct { name string cloudProvider string providerType apiv1.ProviderType providerName string storageType string nodeZones []string haType apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_HighAvailabilityType storageIops uint32 storageThroughputMbps uint32 expectedError string expectedCode codes.Code expectedPlacementGate bool }{ { name: "AWS provider with single-AZ topology enables placement group", cloudProvider: "AWS", providerType: apiv1.ProviderType_AWS, storageType: entity.GP3, nodeZones: []string{"us-west-2a"}, // PR #3040 round 4 requires HA=ZONAL for single-AZ pools. // Earlier rounds let REGIONAL+1 zone reach the gate derivation // (a request shape DPM later rejected as provisioning failure); // the upfront shape validation now rejects it before the // SharedPool is persisted, so the test must pass ZONAL here. haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, storageIops: 3000, storageThroughputMbps: 125, expectedPlacementGate: true, }, { name: "AWS existing provider with single-AZ topology enables placement group", cloudProvider: "AWS", providerType: apiv1.ProviderType_AWS, providerName: "existing-dataplane", storageType: entity.GP3, nodeZones: []string{"us-west-2a"}, haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, storageIops: 3000, storageThroughputMbps: 125, expectedPlacementGate: true, }, { // GCP isn't yet supported at all; the gate check passes (single-AZ // non-AWS just sets PlacementGroupEnabled=false) and the request // hits the "GCP not implemented" branch when the dataplane is // auto-created. PG itself stays off for the GCP case. name: "GCP provider with single-AZ topology is still rejected as not supported", cloudProvider: "GCP", providerType: apiv1.ProviderType_GCP, nodeZones: []string{"us-west-2a"}, haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, expectedError: "GCP cloud provider is not supported yet", expectedCode: codes.Unimplemented, }, { name: "GCP provider with multi-AZ topology is not supported", cloudProvider: "GCP", providerType: apiv1.ProviderType_GCP, // haType: UNSPECIFIED (default) → resolves to REGIONAL. expectedError: "GCP cloud provider is not supported yet", expectedCode: codes.Unimplemented, }, { // Single-AZ Azure is allowed: PG is EKS-only, so the gate stays // off and the cloud's native availability-set handles intra-AZ // placement. name: "Azure provider with single-AZ topology succeeds with PG off", cloudProvider: "AZURE", providerType: apiv1.ProviderType_AZURE, storageType: entity.PremiumSSDv2, nodeZones: []string{"us-west-2a"}, haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, storageIops: 3000, storageThroughputMbps: 125, expectedPlacementGate: false, }, { name: "Azure provider with multi-AZ topology succeeds", cloudProvider: "AZURE", providerType: apiv1.ProviderType_AZURE, storageType: entity.PremiumSSDv2, nodeZones: []string{"us-west-2a", "us-west-2b", "us-west-2c"}, // haType: UNSPECIFIED (default) → resolves to REGIONAL. storageIops: 3000, storageThroughputMbps: 125, }, { // ACK single-AZ is allowed for the same reason as Azure: ACK's // vswitch/zone model handles intra-AZ placement natively and the // PG abstraction is EKS-only. name: "Alicloud provider with single-AZ topology succeeds with PG off", cloudProvider: "ALICLOUD", providerType: apiv1.ProviderType_ALICLOUD, storageType: entity.CloudAuto, nodeZones: []string{"cn-hangzhou-h"}, haType: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, storageIops: 3000, storageThroughputMbps: 0, // cloud-auto does not accept throughput expectedPlacementGate: false, }, { name: "Unknown provider - not supported", cloudProvider: "UNKNOWN", expectedError: "cloud provider UNKNOWN is not supported", expectedCode: codes.Unimplemented, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { // Create mocks sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterConfigClient := clusterconfigmocks.NewClient(t) dagExecutor := executormocks.NewRemoteExecutor(t) ruCfg := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tikv": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tikv_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "coprocessor_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tidb": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tidb_worker": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "pd": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "pd_tso": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "pd_scheduling": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tiflash_write": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tiflash_compute": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, "tiproxy": { NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{ "8c16g": { Min: 0, Max: 100000000, }, }, }, }, }, } ruConfigRepo := repomnocks.NewRuConfigRepo(t) if tt.expectedError == "" { // Setup success mocks for AWS sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool { return lock.Key == "testsharedpoolid" })).Return(func() error { return nil }, nil) nodePoolConfigs := &dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ { CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 3800, AvailableMemoryMi: 7600, NodeType: dplanemgrv1.NodeType_Dedicated, WithLocalDisk: false, }, }, } ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(ruCfg, nil) dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(nodePoolConfigs), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) clusterConfigClient.On("GetEffectiveComponentImageTag", mock.Anything, mock.Anything, apiv1.ComponentType_TIKV).Return("v26.3.5-nextgen", nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) } else if tt.expectedCode != codes.InvalidArgument { // Setup basic mocks for errors reached after initial request validation. sharedPoolRepo.On("Load", mock.Anything, "testsharedpoolid", true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) } server := &serverImpl{ cloudProvider: tt.cloudProvider, region: "us-west-2", sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dataplaneClient: dataplaneClient, clusterConfigClient: clusterConfigClient, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, bizCfg: &config.BizConfig{ DefaultDataplaneAccount: "test-account", }, } sharedPool := &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolid", Provider: &apiv1.Provider{Name: tt.providerName, Type: tt.providerType, NodeZones: tt.nodeZones}, }, } requestUnit := &entity.RequestUnit{ Min: 1000, Max: 2000, } // Provide initSharedPoolConfig for all providers since it's now required. // Per-case storageIops / storageThroughputMbps let providers with // stricter storage schemas (e.g. ACK cloud-auto: no throughput) run // their own storage-validation path. NodeZones drives the durable // Provider.PlacementGroupEnabled gate (PR #3040's "is AWS single-AZ" // check) — the public request's canonical input. initConfig := &entity.InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", HighAvailabilityType: tt.haType, NodeZones: tt.nodeZones, TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: tt.storageType, SizeGi: 100, Iops: tt.storageIops, ThroughputMbps: tt.storageThroughputMbps, }, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: tt.storageType, SizeGi: 50, Iops: tt.storageIops, ThroughputMbps: tt.storageThroughputMbps, }, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: tt.storageType, SizeGi: 100, Iops: tt.storageIops, ThroughputMbps: tt.storageThroughputMbps, }, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{ Type: tt.storageType, SizeGi: 100, Iops: tt.storageIops, ThroughputMbps: tt.storageThroughputMbps, }, } // All providers now require initConfig to pass initial validation // AWS provider succeeds with valid initConfig, others fail with provider-specific errors result, err := server.CreateSharedPool(context.Background(), sharedPool, requestUnit, initConfig) if tt.expectedError != "" { require.Error(t, err) assert.Contains(t, err.Error(), tt.expectedError) assert.Equal(t, tt.expectedCode, status.Code(err)) assert.Nil(t, result) } else { require.NoError(t, err) require.NotNil(t, result) if tt.providerType != apiv1.ProviderType_PROVIDER_TYPE_UNSPECIFIED { assert.Equal(t, tt.providerType, result.GetProvider().GetType()) } assert.Equal(t, tt.expectedPlacementGate, result.GetProvider().GetPlacementGroupEnabled()) } }) } } // TestCreateSharedPoolSetsHighAvailabilityTypeOnResponse is a regression test for the // immediate CreateSharedPool gRPC response. The canonical embedded // apiv1.SharedPool.HighAvailabilityType (output enum) must reflect the resolved value so // the response never reports HIGH_AVAILABILITY_TYPE_UNSPECIFIED to API callers when the // HA preference is known. resolveCreateSharedPoolHAType converts the request enum to the // canonical output enum and the server writes that value into the embedded field once. func TestCreateSharedPoolSetsHighAvailabilityTypeOnResponse(t *testing.T) { tests := []struct { name string inputHA apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_HighAvailabilityType expectedEmbedded apiv1.SharedPool_HighAvailabilityType }{ { name: "explicit ZONAL is mirrored to response", inputHA: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, expectedEmbedded: apiv1.SharedPool_ZONAL, }, { name: "explicit REGIONAL is mirrored to response", inputHA: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_REGIONAL, expectedEmbedded: apiv1.SharedPool_REGIONAL, }, { name: "default (omitted) resolves to REGIONAL on response", inputHA: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_HIGH_AVAILABILITY_TYPE_UNSPECIFIED, expectedEmbedded: apiv1.SharedPool_REGIONAL, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterConfigClient := clusterconfigmocks.NewClient(t) dagExecutor := executormocks.NewRemoteExecutor(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) rc := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tikv": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tikv_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "coprocessor_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tidb": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tidb_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd_tso": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd_scheduling": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiflash_write": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiflash_compute": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiproxy": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, }, }, } sharedPoolRepo.On("Load", mock.Anything, mock.Anything, true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(rc, nil) dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(&dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ {CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 3800, AvailableMemoryMi: 7600, NodeType: dplanemgrv1.NodeType_Dedicated}, }, }), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) clusterConfigClient.On("GetEffectiveComponentImageTag", mock.Anything, mock.Anything, apiv1.ComponentType_TIKV).Return("v8.0.0", nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything).Return(nil) server := &serverImpl{ cloudProvider: "AWS", region: "us-west-2", sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dataplaneClient: dataplaneClient, clusterConfigClient: clusterConfigClient, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, bizCfg: &config.BizConfig{ DefaultDataplaneAccount: "test-account", }, } result, err := server.CreateSharedPool(context.Background(), &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolha", Provider: &apiv1.Provider{ Name: "aws", Type: apiv1.ProviderType_AWS, }, }, TenantId: "test-tenant", }, &entity.RequestUnit{Min: 1000, Max: 2000}, &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", HighAvailabilityType: tt.inputHA, TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 100, Iops: 3000, ThroughputMbps: 125}, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 50, Iops: 3000, ThroughputMbps: 125}, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 100, Iops: 3000, ThroughputMbps: 125}, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.GP3, SizeGi: 100, Iops: 3000, ThroughputMbps: 125}, }) require.NoError(t, err) require.NotNil(t, result) // The canonical embedded apiv1.SharedPool.HighAvailabilityType must reflect the // resolved value. Create writes the canonical value once via // sharedPool.HighAvailabilityType = resolveCreateSharedPoolHAType(...). assert.Equal(t, tt.expectedEmbedded, result.GetHighAvailabilityType(), "canonical embedded HighAvailabilityType on response must reflect resolved value") assert.NotEqual(t, apiv1.SharedPool_HIGH_AVAILABILITY_TYPE_UNSPECIFIED, result.GetHighAvailabilityType(), "response must never carry UNSPECIFIED — caller cannot distinguish from missing") }) } } // TestCreateSharedPoolNodeZonesArgByHAType is a regression test for the REGIONAL HA // zone-handling contract. The DPM pre-validator (dataplaneservice.go:298) rejects REGIONAL // SharedPools when the caller supplied more than 1 node_zone, but the REGIONAL HA contract // says caller must send 0 zones and DPM auto-fills from regionConfig.zones. The HA-type // resolver resolves the HA type but didn't update the create_sharedpool.go arg-build block // to drop caller zones for REGIONAL HA, so the executor DAG passed them through verbatim. // This test pins the contract: DataplaneNodeZonesKey must be ABSENT from the DAG args iff // resolved HA == REGIONAL, regardless of what caller sent. // // PR #3040 round 4 adds upfront shape validation in CreateSharedPool that rejects // REGIONAL+1 and ZONAL+>1 with codes.InvalidArgument (mirrors DPM // resolveNodeZonesForCreate). Illegal-shape coverage lives in // TestCreateSharedPoolPlacementGroupGateRejectsIllegalShape; this test only covers the // legal shapes that reach the DAG arg-build block. func TestCreateSharedPoolNodeZonesArgByHAType(t *testing.T) { tests := []struct { name string inputHA apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_HighAvailabilityType callerZones []string // what the caller sent in CreateSharedPoolRequest.InitSharedPoolConfig.NodeZones expectZonesArg bool // whether DataplaneNodeZonesKey should appear in submitted DAG args cloudProvider string providerName string providerType apiv1.ProviderType region string storageType string }{ { name: "REGIONAL with 3 caller zones is dropped from DAG args", inputHA: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_REGIONAL, callerZones: []string{"us-west-2a", "us-west-2b", "us-west-2c"}, expectZonesArg: false, }, { name: "REGIONAL with no caller zones omits DAG arg (would-be-default auto-fill path)", inputHA: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_REGIONAL, callerZones: nil, expectZonesArg: false, }, { name: "ZONAL with 1 caller zone passes through to DAG args (regression guard)", inputHA: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, callerZones: []string{"us-west-2a"}, expectZonesArg: true, }, { name: "ALICLOUD ZONAL with 1 caller zone passes through to DAG args", inputHA: apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_ZONAL, callerZones: []string{"cn-hangzhou-i"}, expectZonesArg: true, cloudProvider: "ALICLOUD", providerName: "alicloud", providerType: apiv1.ProviderType_ALICLOUD, region: "cn-hangzhou", storageType: entity.CloudAuto, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { cloudProvider := tt.cloudProvider if cloudProvider == "" { cloudProvider = "AWS" } providerName := tt.providerName if providerName == "" { providerName = "aws" } providerType := tt.providerType if providerType == apiv1.ProviderType_PROVIDER_TYPE_UNSPECIFIED { providerType = apiv1.ProviderType_AWS } region := tt.region if region == "" { region = "us-west-2" } storageType := tt.storageType if storageType == "" { storageType = entity.GP3 } iops := uint32(0) throughputMbps := uint32(0) if storageType == entity.GP3 { iops = 3000 throughputMbps = 125 } sharedPoolRepo := repomnocks.NewSharedPoolRepo(t) lockRepo := repomnocks.NewLockRepo(t) dataplaneClient := dataplanemgrmocks.NewClient(t) clusterConfigClient := clusterconfigmocks.NewClient(t) dagExecutor := executormocks.NewRemoteExecutor(t) ruConfigRepo := repomnocks.NewRuConfigRepo(t) rc := &entity.RuConfig{ RuConfig: apiv1.RuConfig{ Components: map[string]*apiv1.RuConfig_ComponentsMap{ "tikv": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tikv_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "coprocessor_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tidb": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tidb_worker": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd_tso": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "pd_scheduling": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiflash_write": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiflash_compute": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, "tiproxy": {NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{"8c16g": {Min: 0, Max: 100000000}}}, }, }, } sharedPoolRepo.On("Load", mock.Anything, mock.Anything, true).Return(nil, status.Error(codes.NotFound, "not found")) dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil) lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil) ruConfigRepo.On("GetRuConfig", mock.Anything, mock.Anything, mock.Anything).Return(rc, nil) dataplaneClient.On("ListNodePoolConfigs", mock.Anything, mock.Anything).Return(withSchemaManagerNodeSpec(&dplanemgrv1.ListNodePoolConfigsResponse{ NodePoolConfigs: []*dplanemgrv1.NodePoolConfig{ {CpuCore: 8, MemoryGi: 16, AvailableCpuMilliCore: 3800, AvailableMemoryMi: 7600, NodeType: dplanemgrv1.NodeType_Dedicated}, }, }), nil) clusterConfigClient.On("GetDefaultVersion", mock.Anything, mock.Anything).Return("v8.0.0", nil) clusterConfigClient.On("GetEffectiveComponentImageTag", mock.Anything, mock.Anything, apiv1.ComponentType_TIKV).Return("v8.0.0", nil) ruConfigRepo.On("SetRuConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) sharedPoolRepo.On("Save", mock.Anything, mock.Anything).Return(nil) // Capture the args map passed to Submit so we can assert on DataplaneNodeZonesKey. var submittedArgs actions.ActionArgs dagExecutor.On("Submit", mock.Anything, dag.PlanCreateSharedPool, mock.Anything, mock.Anything). Run(func(args mock.Arguments) { submittedArgs = args.Get(2).(actions.ActionArgs) }). Return(nil) server := &serverImpl{ cloudProvider: cloudProvider, region: region, sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, ruConfigRepo: ruConfigRepo, dataplaneClient: dataplaneClient, clusterConfigClient: clusterConfigClient, dagExecutor: dagExecutor, transactioner: &mockTransactioner{}, bizCfg: &config.BizConfig{ DefaultDataplaneAccount: "test-account", }, } _, err := server.CreateSharedPool(context.Background(), &entity.SharedPool{ SharedPool: apiv1.SharedPool{ Id: "testsharedpoolzones", Provider: &apiv1.Provider{ Name: providerName, Type: providerType, }, }, TenantId: "test-tenant", }, &entity.RequestUnit{Min: 1000, Max: 2000}, &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig{ NetworkCidr: "10.0.0.0/16", HighAvailabilityType: tt.inputHA, NodeZones: tt.callerZones, TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: storageType, SizeGi: 100, Iops: iops, ThroughputMbps: throughputMbps}, TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: storageType, SizeGi: 50, Iops: iops, ThroughputMbps: throughputMbps}, TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: storageType, SizeGi: 100, Iops: iops, ThroughputMbps: throughputMbps}, TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: storageType, SizeGi: 100, Iops: iops, ThroughputMbps: throughputMbps}, }) require.NoError(t, err) require.NotNil(t, submittedArgs, "Submit should have been called and args captured") _, present := submittedArgs[workflow.DataplaneNodeZonesKey] assert.Equal(t, tt.expectZonesArg, present, "DataplaneNodeZonesKey presence in DAG args must match the resolved HA contract; "+ "REGIONAL drops caller zones (auto-fill from regionConfig.zones), ZONAL passes them through") if tt.expectZonesArg { nodeZones, err := submittedArgs.GetString(workflow.DataplaneNodeZonesKey) require.NoError(t, err) expectedNodeZones, err := json.Marshal(tt.callerZones) require.NoError(t, err) assert.Equal(t, string(expectedNodeZones), nodeZones) } // DataplaneHATypeKey must always carry the resolved HA string for DPM. _, haPresent := submittedArgs[workflow.DataplaneHATypeKey] require.True(t, haPresent, "DataplaneHATypeKey must always be set so DPM receives the resolved HA type") haStr, err := submittedArgs.GetString(workflow.DataplaneHATypeKey) require.NoError(t, err) assert.NotEqual(t, "HIGH_AVAILABILITY_TYPE_UNSPECIFIED", haStr, "DPM must never receive an UNSPECIFIED HA type") }) } }