robocanic commented on code in PR #1349: URL: https://github.com/apache/dubbo-admin/pull/1349#discussion_r2486393116
########## pkg/store/postgres/postgres.go: ########## Review Comment: suggestion: 这个文件是不是和mysql的大部分是一样的? 看能不能再抽象一下，把mysql和postgres这两种用dialector来区分，其他增删改查的部分复用同一份代码 ########## pkg/store/postgres/pool.go: ########## @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package postgres + +import ( + "sync" + + "gorm.io/driver/postgres" + + "github.com/apache/dubbo-admin/pkg/core/logger" + "github.com/apache/dubbo-admin/pkg/store/dbcommon" +) + +var ( + // postgresPool is the shared PostgreSQL connection pool instance + postgresPool *dbcommon.ConnectionPool + // postgresOnce ensures the pool is initialized only once + postgresOnce sync.Once + // poolMutex protects access to the pool during creation and reuse + poolMutex sync.RWMutex +) + +// GetOrCreatePostgresPool returns or creates a PostgreSQL connection pool +// It implements a singleton pattern with reference counting to allow pool reuse across multiple stores +// If a pool already exists for the same address, it increments the reference count and returns the existing pool +// Otherwise, it creates a new pool with the PostgreSQL driver configured +func GetOrCreatePostgresPool(address string, config *dbcommon.ConnectionPoolConfig) (*dbcommon.ConnectionPool, error) { Review Comment: suggestion: 这个文件是不是和mysql的大部分是一样的? 看能不能再抽象一下，把mysql和postgres这两种用dialector来区分，其他增删改查的部分复用同一份代码 ########## pkg/store/mysql/mysql.go: ########## @@ -17,4 +17,705 @@ package mysql -// TODO implement memory resource store, refer to GORM https://gorm.io/docs/ +import ( + "errors" + "fmt" + "reflect" + "sort" + "sync" + + storecfg "github.com/apache/dubbo-admin/pkg/config/store" + "gorm.io/gorm" + "k8s.io/client-go/tools/cache" + + "github.com/apache/dubbo-admin/pkg/common/bizerror" + "github.com/apache/dubbo-admin/pkg/core/logger" + "github.com/apache/dubbo-admin/pkg/core/resource/model" + "github.com/apache/dubbo-admin/pkg/core/runtime" + "github.com/apache/dubbo-admin/pkg/core/store" + "github.com/apache/dubbo-admin/pkg/store/dbcommon" +) + +func init() { + store.RegisterFactory(&mysqlStoreFactory{}) +} + +// mysqlStoreFactory is the factory for creating MySQL store instances +type mysqlStoreFactory struct{} + +var _ store.Factory = &mysqlStoreFactory{} + +// Support checks if this factory supports the given store type +func (f *mysqlStoreFactory) Support(s storecfg.Type) bool { + return s == storecfg.MySQL +} + +// New creates a new MySQL store instance for the specified resource kind +func (f *mysqlStoreFactory) New(kind model.ResourceKind, cfg *storecfg.Config) (store.ManagedResourceStore, error) { + return NewMySQLStore(kind, cfg.Address) +} + +// mysqlStore is a MySQL-backed store implementation for Dubbo resources +// It uses GORM for database operations and maintains in-memory indices for fast lookups +type mysqlStore struct { + pool *dbcommon.ConnectionPool // Shared connection pool with reference counting + kind model.ResourceKind // The resource kind this store manages + address string // MySQL connection address + indexers cache.Indexers // Index functions for creating indices + indexerLock sync.RWMutex // Protects indexers map + indices map[string]map[string]map[string]struct{} // In-memory index: map[indexName]map[indexedValue]set[resourceKey] + indicesLock sync.RWMutex // Protects indices map + stopCh chan struct{} // Channel for signaling shutdown +} + +var _ store.ManagedResourceStore = &mysqlStore{} + +// NewMySQLStore creates a new MySQL store for the specified resource kind +// The store is not initialized until Init() is called +func NewMySQLStore(kind model.ResourceKind, address string) (store.ManagedResourceStore, error) { + return &mysqlStore{ + kind: kind, + address: address, + indexers: cache.Indexers{}, + indices: make(map[string]map[string]map[string]struct{}), + stopCh: make(chan struct{}), + }, nil +} + +// Init initializes the MySQL store by creating/reusing a connection pool and migrating the schema +func (ms *mysqlStore) Init(_ runtime.BuilderContext) error { + // Get or create MySQL connection pool + pool, err := GetOrCreateMySQLPool(ms.address, dbcommon.DefaultConnectionPoolConfig()) + if err != nil { + return fmt.Errorf("failed to initialize mysql connection pool: %w", err) + } + ms.pool = pool + + // Perform table migration + db := ms.pool.GetDB() + modelForMigration := &dbcommon.ResourceModel{ResourceKind: ms.kind.ToString()} + if err := db.AutoMigrate(modelForMigration); err != nil { + return fmt.Errorf("failed to migrate schema for %s: %w", ms.kind.ToString(), err) + } + + logger.Infof("MySQL store initialized for resource kind: %s", ms.kind.ToString()) + return nil +} + +// Start starts the MySQL store and monitors for shutdown signal +func (ms *mysqlStore) Start(_ runtime.Runtime, stopCh <-chan struct{}) error { + logger.Infof("MySQL store started for resource kind: %s", ms.kind.ToString()) + + // Monitor stop channel for graceful shutdown in a goroutine + go func() { + <-stopCh + logger.Infof("MySQL store for %s received stop signal, initiating graceful shutdown", ms.kind.ToString()) + + // Close the internal stop channel to signal any ongoing operations + close(ms.stopCh) + + // Decrement the reference count and potentially close the connection pool + if ms.pool != nil { + if err := ms.pool.Close(); err != nil { + logger.Errorf("Failed to close MySQL connection pool for %s: %v", ms.kind.ToString(), err) + } else { + logger.Infof("MySQL store for %s shutdown completed", ms.kind.ToString()) + } + } + }() + + return nil +} + +// Add inserts a new resource into the MySQL database +// Returns an error if the resource already exists +func (ms *mysqlStore) Add(obj interface{}) error { + resource, ok := obj.(model.Resource) + if !ok { + return bizerror.NewAssertionError("Resource", reflect.TypeOf(obj).Name()) + } + + if resource.ResourceKind() != ms.kind { + return fmt.Errorf("resource kind mismatch: expected %s, got %s", ms.kind, resource.ResourceKind()) + } + + var count int64 + db := ms.pool.GetDB() + err := db.Model(&dbcommon.ResourceModel{}). + Where("resource_key = ?", resource.ResourceKey()). + Count(&count).Error + if err != nil { + return err + } + if count > 0 { + return store.ErrorResourceAlreadyExists( + resource.ResourceKind().ToString(), + resource.ResourceMeta().Name, + resource.MeshName(), + ) + } + + m, err := dbcommon.FromResource(resource) + if err != nil { + return err + } + + if err := db.Create(m).Error; err != nil { + return err + } + + // Update indices after successful DB operation + ms.updateIndicesForResource(resource, nil) + + return nil +} + +// Update modifies an existing resource in the MySQL database +// Returns an error if the resource does not exist +func (ms *mysqlStore) Update(obj interface{}) error { + resource, ok := obj.(model.Resource) + if !ok { + return bizerror.NewAssertionError("Resource", reflect.TypeOf(obj).Name()) + } + + if resource.ResourceKind() != ms.kind { + return fmt.Errorf("resource kind mismatch: expected %s, got %s", ms.kind, resource.ResourceKind()) + } + + // Get old resource for index update + oldResource, exists, err := ms.GetByKey(resource.ResourceKey()) + if err != nil { + return err + } + if !exists { + return store.ErrorResourceNotFound( + resource.ResourceKind().ToString(), + resource.ResourceMeta().Name, + resource.MeshName(), + ) + } + + m, err := dbcommon.FromResource(resource) + if err != nil { + return err + } + + db := ms.pool.GetDB() + result := db.Model(&dbcommon.ResourceModel{}). + Where("resource_key = ?", resource.ResourceKey()). + Updates(map[string]interface{}{ + "data": m.Data, + "updated_at": m.UpdatedAt, + }) + + if result.Error != nil { + return result.Error + } + + if result.RowsAffected == 0 { + return store.ErrorResourceNotFound( + resource.ResourceKind().ToString(), + resource.ResourceMeta().Name, + resource.MeshName(), + ) + } + + // Update indices: remove old and add new + ms.updateIndicesForResource(resource, oldResource.(model.Resource)) + + return nil +} + +// Delete removes a resource from the MySQL database +// Returns an error if the resource does not exist +func (ms *mysqlStore) Delete(obj interface{}) error { + resource, ok := obj.(model.Resource) + if !ok { + return bizerror.NewAssertionError("Resource", reflect.TypeOf(obj).Name()) + } + + db := ms.pool.GetDB() + result := db.Where("resource_key = ?", resource.ResourceKey()). + Delete(&dbcommon.ResourceModel{}) + + if result.Error != nil { + return result.Error + } + + if result.RowsAffected == 0 { + return store.ErrorResourceNotFound( + resource.ResourceKind().ToString(), + resource.ResourceMeta().Name, + resource.MeshName(), + ) + } + + // Remove from indices + ms.removeFromIndices(resource) + + return nil +} + +// List returns all resources of the configured kind from the MySQL database +func (ms *mysqlStore) List() []interface{} { + var models []dbcommon.ResourceModel + db := ms.pool.GetDB() + if err := db.Where("resource_kind = ?", ms.kind.ToString()).Find(&models).Error; err != nil { + logger.Errorf("failed to list resources: %v", err) + return []interface{}{} + } + + result := make([]interface{}, 0, len(models)) + for _, m := range models { + resource, err := m.ToResource() + if err != nil { + logger.Errorf("failed to deserialize resource: %v", err) + continue + } + result = append(result, resource) + } + return result +} + +// ListKeys returns all resource keys of the configured kind from the MySQL database +func (ms *mysqlStore) ListKeys() []string { + var keys []string + db := ms.pool.GetDB() + db.Model(&dbcommon.ResourceModel{}). + Where("resource_kind = ?", ms.kind.ToString()). + Pluck("resource_key", &keys) + return keys +} + +// Get retrieves a resource by its object reference +func (ms *mysqlStore) Get(obj interface{}) (item interface{}, exists bool, err error) { + resource, ok := obj.(model.Resource) + if !ok { + return nil, false, bizerror.NewAssertionError("Resource", reflect.TypeOf(obj).Name()) + } + return ms.GetByKey(resource.ResourceKey()) +} + +// GetByKey retrieves a resource by its unique key +func (ms *mysqlStore) GetByKey(key string) (item interface{}, exists bool, err error) { + var m dbcommon.ResourceModel + db := ms.pool.GetDB() + result := db.Where("resource_key = ? AND resource_kind = ?", key, ms.kind.ToString()). Review Comment: suggestion: 按照resource kind进行隔离后，这里是不是可以少一个where条件？ ########## pkg/store/mysql/mysql.go: ########## @@ -17,4 +17,705 @@ package mysql -// TODO implement memory resource store, refer to GORM https://gorm.io/docs/ +import ( + "errors" + "fmt" + "reflect" + "sort" + "sync" + + storecfg "github.com/apache/dubbo-admin/pkg/config/store" Review Comment: correction: please follow [here](https://github.com/apache/dubbo-admin/wiki/%E6%96%B0%E7%89%88Dubbo-Admin%E5%BC%80%E5%8F%91%E8%80%85%E6%8C%87%E5%8D%97#goland-import%E9%85%8D%E7%BD%AE) to correct the order of import ########## pkg/store/mysql/mysql.go: ########## @@ -17,4 +17,705 @@ package mysql -// TODO implement memory resource store, refer to GORM https://gorm.io/docs/ +import ( + "errors" + "fmt" + "reflect" + "sort" + "sync" + + storecfg "github.com/apache/dubbo-admin/pkg/config/store" + "gorm.io/gorm" + "k8s.io/client-go/tools/cache" + + "github.com/apache/dubbo-admin/pkg/common/bizerror" + "github.com/apache/dubbo-admin/pkg/core/logger" + "github.com/apache/dubbo-admin/pkg/core/resource/model" + "github.com/apache/dubbo-admin/pkg/core/runtime" + "github.com/apache/dubbo-admin/pkg/core/store" + "github.com/apache/dubbo-admin/pkg/store/dbcommon" +) + +func init() { + store.RegisterFactory(&mysqlStoreFactory{}) +} + +// mysqlStoreFactory is the factory for creating MySQL store instances +type mysqlStoreFactory struct{} + +var _ store.Factory = &mysqlStoreFactory{} + +// Support checks if this factory supports the given store type +func (f *mysqlStoreFactory) Support(s storecfg.Type) bool { + return s == storecfg.MySQL +} + +// New creates a new MySQL store instance for the specified resource kind +func (f *mysqlStoreFactory) New(kind model.ResourceKind, cfg *storecfg.Config) (store.ManagedResourceStore, error) { + return NewMySQLStore(kind, cfg.Address) +} + +// mysqlStore is a MySQL-backed store implementation for Dubbo resources +// It uses GORM for database operations and maintains in-memory indices for fast lookups +type mysqlStore struct { + pool *dbcommon.ConnectionPool // Shared connection pool with reference counting + kind model.ResourceKind // The resource kind this store manages + address string // MySQL connection address + indexers cache.Indexers // Index functions for creating indices + indexerLock sync.RWMutex // Protects indexers map + indices map[string]map[string]map[string]struct{} // In-memory index: map[indexName]map[indexedValue]set[resourceKey] Review Comment: suggestion: 这里可以考虑为索引定义一个单独的数据结构Index，现在这种多层的map结构易读性不太好，另外更新索引的原子操作可以放在这个Index里面完成，不用在外层加锁。其次是set这个结构可以考虑使用lancet的"github.com/duke-git/lancet/v2/datastructure/set"。 ########## pkg/store/mysql/mysql.go: ########## @@ -17,4 +17,705 @@ package mysql -// TODO implement memory resource store, refer to GORM https://gorm.io/docs/ +import ( + "errors" + "fmt" + "reflect" + "sort" + "sync" + + storecfg "github.com/apache/dubbo-admin/pkg/config/store" + "gorm.io/gorm" + "k8s.io/client-go/tools/cache" + + "github.com/apache/dubbo-admin/pkg/common/bizerror" + "github.com/apache/dubbo-admin/pkg/core/logger" + "github.com/apache/dubbo-admin/pkg/core/resource/model" + "github.com/apache/dubbo-admin/pkg/core/runtime" + "github.com/apache/dubbo-admin/pkg/core/store" + "github.com/apache/dubbo-admin/pkg/store/dbcommon" +) + +func init() { + store.RegisterFactory(&mysqlStoreFactory{}) +} + +// mysqlStoreFactory is the factory for creating MySQL store instances +type mysqlStoreFactory struct{} + +var _ store.Factory = &mysqlStoreFactory{} + +// Support checks if this factory supports the given store type +func (f *mysqlStoreFactory) Support(s storecfg.Type) bool { + return s == storecfg.MySQL +} + +// New creates a new MySQL store instance for the specified resource kind +func (f *mysqlStoreFactory) New(kind model.ResourceKind, cfg *storecfg.Config) (store.ManagedResourceStore, error) { + return NewMySQLStore(kind, cfg.Address) +} + +// mysqlStore is a MySQL-backed store implementation for Dubbo resources +// It uses GORM for database operations and maintains in-memory indices for fast lookups +type mysqlStore struct { + pool *dbcommon.ConnectionPool // Shared connection pool with reference counting + kind model.ResourceKind // The resource kind this store manages + address string // MySQL connection address + indexers cache.Indexers // Index functions for creating indices + indexerLock sync.RWMutex // Protects indexers map + indices map[string]map[string]map[string]struct{} // In-memory index: map[indexName]map[indexedValue]set[resourceKey] + indicesLock sync.RWMutex // Protects indices map + stopCh chan struct{} // Channel for signaling shutdown +} + +var _ store.ManagedResourceStore = &mysqlStore{} + +// NewMySQLStore creates a new MySQL store for the specified resource kind +// The store is not initialized until Init() is called +func NewMySQLStore(kind model.ResourceKind, address string) (store.ManagedResourceStore, error) { + return &mysqlStore{ + kind: kind, + address: address, + indexers: cache.Indexers{}, + indices: make(map[string]map[string]map[string]struct{}), + stopCh: make(chan struct{}), + }, nil +} + +// Init initializes the MySQL store by creating/reusing a connection pool and migrating the schema +func (ms *mysqlStore) Init(_ runtime.BuilderContext) error { + // Get or create MySQL connection pool + pool, err := GetOrCreateMySQLPool(ms.address, dbcommon.DefaultConnectionPoolConfig()) + if err != nil { + return fmt.Errorf("failed to initialize mysql connection pool: %w", err) + } + ms.pool = pool + + // Perform table migration + db := ms.pool.GetDB() + modelForMigration := &dbcommon.ResourceModel{ResourceKind: ms.kind.ToString()} + if err := db.AutoMigrate(modelForMigration); err != nil { + return fmt.Errorf("failed to migrate schema for %s: %w", ms.kind.ToString(), err) + } + + logger.Infof("MySQL store initialized for resource kind: %s", ms.kind.ToString()) + return nil +} + +// Start starts the MySQL store and monitors for shutdown signal +func (ms *mysqlStore) Start(_ runtime.Runtime, stopCh <-chan struct{}) error { + logger.Infof("MySQL store started for resource kind: %s", ms.kind.ToString()) + + // Monitor stop channel for graceful shutdown in a goroutine + go func() { + <-stopCh + logger.Infof("MySQL store for %s received stop signal, initiating graceful shutdown", ms.kind.ToString()) + + // Close the internal stop channel to signal any ongoing operations + close(ms.stopCh) + + // Decrement the reference count and potentially close the connection pool + if ms.pool != nil { + if err := ms.pool.Close(); err != nil { + logger.Errorf("Failed to close MySQL connection pool for %s: %v", ms.kind.ToString(), err) + } else { + logger.Infof("MySQL store for %s shutdown completed", ms.kind.ToString()) + } + } + }() + + return nil +} + +// Add inserts a new resource into the MySQL database +// Returns an error if the resource already exists +func (ms *mysqlStore) Add(obj interface{}) error { + resource, ok := obj.(model.Resource) + if !ok { + return bizerror.NewAssertionError("Resource", reflect.TypeOf(obj).Name()) + } + + if resource.ResourceKind() != ms.kind { + return fmt.Errorf("resource kind mismatch: expected %s, got %s", ms.kind, resource.ResourceKind()) + } + + var count int64 + db := ms.pool.GetDB() + err := db.Model(&dbcommon.ResourceModel{}). + Where("resource_key = ?", resource.ResourceKey()). + Count(&count).Error + if err != nil { + return err + } + if count > 0 { + return store.ErrorResourceAlreadyExists( + resource.ResourceKind().ToString(), + resource.ResourceMeta().Name, + resource.MeshName(), + ) + } + + m, err := dbcommon.FromResource(resource) + if err != nil { + return err + } + + if err := db.Create(m).Error; err != nil { + return err + } + + // Update indices after successful DB operation + ms.updateIndicesForResource(resource, nil) + + return nil +} + +// Update modifies an existing resource in the MySQL database +// Returns an error if the resource does not exist +func (ms *mysqlStore) Update(obj interface{}) error { + resource, ok := obj.(model.Resource) + if !ok { + return bizerror.NewAssertionError("Resource", reflect.TypeOf(obj).Name()) + } + + if resource.ResourceKind() != ms.kind { + return fmt.Errorf("resource kind mismatch: expected %s, got %s", ms.kind, resource.ResourceKind()) + } + + // Get old resource for index update + oldResource, exists, err := ms.GetByKey(resource.ResourceKey()) + if err != nil { + return err + } + if !exists { + return store.ErrorResourceNotFound( + resource.ResourceKind().ToString(), + resource.ResourceMeta().Name, + resource.MeshName(), + ) + } + + m, err := dbcommon.FromResource(resource) + if err != nil { + return err + } + + db := ms.pool.GetDB() + result := db.Model(&dbcommon.ResourceModel{}). + Where("resource_key = ?", resource.ResourceKey()). + Updates(map[string]interface{}{ + "data": m.Data, + "updated_at": m.UpdatedAt, + }) + + if result.Error != nil { + return result.Error + } + + if result.RowsAffected == 0 { + return store.ErrorResourceNotFound( + resource.ResourceKind().ToString(), + resource.ResourceMeta().Name, + resource.MeshName(), + ) + } + + // Update indices: remove old and add new + ms.updateIndicesForResource(resource, oldResource.(model.Resource)) + + return nil +} + +// Delete removes a resource from the MySQL database +// Returns an error if the resource does not exist +func (ms *mysqlStore) Delete(obj interface{}) error { + resource, ok := obj.(model.Resource) + if !ok { + return bizerror.NewAssertionError("Resource", reflect.TypeOf(obj).Name()) + } + + db := ms.pool.GetDB() + result := db.Where("resource_key = ?", resource.ResourceKey()). + Delete(&dbcommon.ResourceModel{}) + + if result.Error != nil { + return result.Error + } + + if result.RowsAffected == 0 { + return store.ErrorResourceNotFound( + resource.ResourceKind().ToString(), + resource.ResourceMeta().Name, + resource.MeshName(), + ) + } + + // Remove from indices + ms.removeFromIndices(resource) + + return nil +} + +// List returns all resources of the configured kind from the MySQL database +func (ms *mysqlStore) List() []interface{} { + var models []dbcommon.ResourceModel + db := ms.pool.GetDB() + if err := db.Where("resource_kind = ?", ms.kind.ToString()).Find(&models).Error; err != nil { + logger.Errorf("failed to list resources: %v", err) + return []interface{}{} + } + + result := make([]interface{}, 0, len(models)) + for _, m := range models { + resource, err := m.ToResource() + if err != nil { + logger.Errorf("failed to deserialize resource: %v", err) + continue + } + result = append(result, resource) + } + return result +} + +// ListKeys returns all resource keys of the configured kind from the MySQL database +func (ms *mysqlStore) ListKeys() []string { + var keys []string + db := ms.pool.GetDB() + db.Model(&dbcommon.ResourceModel{}). + Where("resource_kind = ?", ms.kind.ToString()). + Pluck("resource_key", &keys) + return keys +} + +// Get retrieves a resource by its object reference +func (ms *mysqlStore) Get(obj interface{}) (item interface{}, exists bool, err error) { + resource, ok := obj.(model.Resource) + if !ok { + return nil, false, bizerror.NewAssertionError("Resource", reflect.TypeOf(obj).Name()) + } + return ms.GetByKey(resource.ResourceKey()) +} + +// GetByKey retrieves a resource by its unique key +func (ms *mysqlStore) GetByKey(key string) (item interface{}, exists bool, err error) { + var m dbcommon.ResourceModel + db := ms.pool.GetDB() + result := db.Where("resource_key = ? AND resource_kind = ?", key, ms.kind.ToString()). + First(&m) + + if result.Error != nil { + if errors.Is(result.Error, gorm.ErrRecordNotFound) { + return nil, false, nil + } + return nil, false, result.Error + } + + resource, err := m.ToResource() + if err != nil { + return nil, false, err + } + + return resource, true, nil +} + +// Replace atomically replaces all resources in the database with the provided list +// This operation is performed within a transaction to ensure atomicity +func (ms *mysqlStore) Replace(list []interface{}, _ string) error { + db := ms.pool.GetDB() + return db.Transaction(func(tx *gorm.DB) error { + // Delete all existing records for this resource kind + if err := tx.Where("resource_kind = ?", ms.kind.ToString()).Delete(&dbcommon.ResourceModel{}).Error; err != nil { + return err + } + + // Clear all indices + ms.clearIndices() + + // Return early if list is empty + if len(list) == 0 { + return nil + } + + // Convert all resources to dbcommon.ResourceModel + models := make([]*dbcommon.ResourceModel, 0, len(list)) + resources := make([]model.Resource, 0, len(list)) + for _, obj := range list { + resource, ok := obj.(model.Resource) + if !ok { + return bizerror.NewAssertionError("Resource", reflect.TypeOf(obj).Name()) + } + + m, err := dbcommon.FromResource(resource) + if err != nil { + return err + } + models = append(models, m) + resources = append(resources, resource) + } + + // Batch insert all models at once + // GORM will automatically split into multiple batches if needed + if err := tx.CreateInBatches(models, 100).Error; err != nil { + return err + } + + // Rebuild indices for all resources + for _, resource := range resources { + ms.updateIndicesForResource(resource, nil) + } + + return nil + }) +} + +func (ms *mysqlStore) Resync() error { + return nil +} + +func (ms *mysqlStore) Index(indexName string, obj interface{}) ([]interface{}, error) { + ms.indexerLock.RLock() + indexFunc, exists := ms.indexers[indexName] + ms.indexerLock.RUnlock() + + if !exists { + return nil, fmt.Errorf("index %s does not exist", indexName) + } + + indexValues, err := indexFunc(obj) + if err != nil { + return nil, err + } + + if len(indexValues) == 0 { + return []interface{}{}, nil + } + + return ms.findByIndex(indexName, indexValues[0]) +} + +func (ms *mysqlStore) IndexKeys(indexName, indexedValue string) ([]string, error) { + ms.indexerLock.RLock() + _, exists := ms.indexers[indexName] + ms.indexerLock.RUnlock() + + if !exists { + return nil, fmt.Errorf("index %s does not exist", indexName) + } + + resources, err := ms.findByIndex(indexName, indexedValue) + if err != nil { + return nil, err + } + + keys := make([]string, 0, len(resources)) + for _, obj := range resources { + if resource, ok := obj.(model.Resource); ok { + keys = append(keys, resource.ResourceKey()) + } + } + + return keys, nil +} + +func (ms *mysqlStore) ListIndexFuncValues(indexName string) []string { Review Comment: correction：这个方法的业务逻辑是获取indexName的所有的indexValue值，这里搞复杂了，可以直接把indices[indexName]的所有值拿过来就可以，可以参考client-go里cache的方法实现 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
