Skip to content

Commit

Permalink
support split table for internal storage
Browse files Browse the repository at this point in the history 
Signed-off-by: calvin <[email protected]>
  • Loading branch information
calvin committed Mar 5, 2024
1 parent 29c0306 commit ffb80ad
Show file tree
Hide file tree
Showing 5 changed files with 164 additions and 59 deletions.
2 changes: 1 addition & 1 deletion examples/pediacluster.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ metadata:
name: cluster-example
spec:
apiserver: "https://10.30.43.43:6443"
caData:
caData:
tokenData:
certData:
keyData:
Expand Down
7 changes: 6 additions & 1 deletion pkg/storage/internalstorage/register.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,12 @@ func NewStorageFactory(configPath string) (storage.StorageFactory, error) {
sqlDB.SetMaxOpenConns(connPool.MaxOpenConns)
sqlDB.SetConnMaxLifetime(connPool.ConnMaxLifetime)

return &StorageFactory{db}, nil
return &StorageFactory{
db: db,
AutoMigration: cfg.AutoMigration,
DivisionPolicy: cfg.DivisionPolicy,
Mapper: cfg.Mapper,
}, nil
}

func newLogger(cfg *Config) (logger.Interface, error) {
Expand Down
63 changes: 35 additions & 28 deletions pkg/storage/internalstorage/resource_storage.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,9 +27,9 @@ import (
)

type ResourceStorage struct {
db *gorm.DB
codec runtime.Codec

db *gorm.DB
codec runtime.Codec
model interface{}
storageGroupResource schema.GroupResource
storageVersion schema.GroupVersion
memoryVersion schema.GroupVersion
Expand Down Expand Up @@ -116,14 +116,18 @@ func (s *ResourceStorage) Update(ctx context.Context, cluster string, obj runtim
updatedResource["deleted_at"] = sql.NullTime{Time: deletedAt.Time, Valid: true}
}

result := s.db.WithContext(ctx).Model(&Resource{}).Where(map[string]interface{}{
"cluster": cluster,
"group": s.storageGroupResource.Group,
"version": s.storageVersion.Version,
"resource": s.storageGroupResource.Resource,
"namespace": metaobj.GetNamespace(),
"name": metaobj.GetName(),
}).Updates(updatedResource)
result := s.db.WithContext(ctx).
Model(s.model).
Where(map[string]interface{}{
"cluster": cluster,
"group": s.storageGroupResource.Group,
"version": s.storageVersion.Version,
"resource": s.storageGroupResource.Resource,
"namespace": metaobj.GetNamespace(),
"name": metaobj.GetName(),
}).
Updates(updatedResource)

return InterpretResourceDBError(cluster, metaobj.GetName(), result.Error)
}

Expand All @@ -144,14 +148,16 @@ func (s *ResourceStorage) ConvertDeletedObject(obj interface{}) (runtime.Object,
}

func (s *ResourceStorage) deleteObject(cluster, namespace, name string) *gorm.DB {
return s.db.Model(&Resource{}).Where(map[string]interface{}{
"cluster": cluster,
"group": s.storageGroupResource.Group,
"version": s.storageVersion.Version,
"resource": s.storageGroupResource.Resource,
"namespace": namespace,
"name": name,
}).Delete(&Resource{})
return s.db.Model(s.model).
Where(map[string]interface{}{
"cluster": cluster,
"group": s.storageGroupResource.Group,
"version": s.storageVersion.Version,
"resource": s.storageGroupResource.Resource,
"namespace": namespace,
"name": name,
}).
Delete(&Resource{})
}

func (s *ResourceStorage) Delete(ctx context.Context, cluster string, obj runtime.Object) error {
Expand All @@ -167,14 +173,15 @@ func (s *ResourceStorage) Delete(ctx context.Context, cluster string, obj runtim
}

func (s *ResourceStorage) genGetObjectQuery(ctx context.Context, cluster, namespace, name string) *gorm.DB {
return s.db.WithContext(ctx).Model(&Resource{}).Select("object").Where(map[string]interface{}{
"cluster": cluster,
"group": s.storageGroupResource.Group,
"version": s.storageVersion.Version,
"resource": s.storageGroupResource.Resource,
"namespace": namespace,
"name": name,
})
return s.db.WithContext(ctx).Model(s.model).Select("object").
Where(map[string]interface{}{
"cluster": cluster,
"group": s.storageGroupResource.Group,
"version": s.storageVersion.Version,
"resource": s.storageGroupResource.Resource,
"namespace": namespace,
"name": name,
})
}

func (s *ResourceStorage) Get(ctx context.Context, cluster, namespace, name string, into runtime.Object) error {
Expand All @@ -199,7 +206,7 @@ func (s *ResourceStorage) genListObjectsQuery(ctx context.Context, opts *interna
result = &ResourceMetadataList{}
}

query := s.db.WithContext(ctx).Model(&Resource{})
query := s.db.WithContext(ctx).Model(s.model)
query = query.Where(map[string]interface{}{
"group": s.storageGroupResource.Group,
"version": s.storageVersion.Version,
Expand Down
127 changes: 99 additions & 28 deletions pkg/storage/internalstorage/storage.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@ package internalstorage
import (
"context"
"fmt"
"strings"
"sync"

"gorm.io/gorm"
"k8s.io/apimachinery/pkg/runtime/schema"
Expand All @@ -11,6 +13,8 @@ import (
"github.com/clusterpedia-io/clusterpedia/pkg/storage"
)

var mutex sync.Mutex

type StorageFactory struct {
db *gorm.DB
AutoMigration *bool
Expand All @@ -19,35 +23,65 @@ type StorageFactory struct {
}

func (s *StorageFactory) AutoMigrate() error {
return nil
}

func (s *StorageFactory) GetSupportedRequestVerbs() []string {
return []string{"get", "list"}
}

func (s *StorageFactory) NewResourceStorage(config *storage.ResourceStorageConfig) (storage.ResourceStorage, error) {
mutex.Lock()
defer mutex.Unlock()

var table string
var model interface{}

if s.AutoMigration != nil && *s.AutoMigration {
table = "resource"
model = &Resource{}

switch s.DivisionPolicy {
if err := s.db.AutoMigrate(&Resource{}); err != nil {
return err
case DivisionPolicyNone:
if exist := s.db.Migrator().HasTable(table); !exist {
if err := s.db.AutoMigrate(&Resource{}); err != nil {
return nil, err
}
}
case "", DivisionPolicyNone:
case DivisionPolicyGroupResource:
model = &GroupVersionResource{}

}
gvr := schema.GroupVersionResource{
Group: config.StorageGroupResource.Group,
Version: config.StorageVersion.Version,
Resource: config.StorageGroupResource.Resource,
}

table = GenerateTableFor(gvr)

if s.DivisionPolicy == "" || s.DivisionPolicy == DivisionPolicyNone {
if err := s.db.AutoMigrate(&Resource{}); err != nil {
return err
if exist := s.db.Migrator().HasTable(table); !exist {
if err := s.db.AutoMigrate(&GroupVersionResource{}); err != nil {
return nil, err
}

err := s.db.Migrator().RenameTable("groupversionresource", table)
if err != nil {
return nil, err
}
}
default:
if exist := s.db.Migrator().HasTable(table); !exist {
if err := s.db.AutoMigrate(&Resource{}); err != nil {
return nil, err
}
}
}
}

return nil
}

func (s *StorageFactory) GetSupportedRequestVerbs() []string {
return []string{"get", "list"}
}

func (s *StorageFactory) NewResourceStorage(config *storage.ResourceStorageConfig) (storage.ResourceStorage, error) {
return &ResourceStorage{
db: s.db,
codec: config.Codec,

db: s.db.Table(table),
model: model,
codec: config.Codec,
storageGroupResource: config.StorageGroupResource,
storageVersion: config.StorageVersion,
memoryVersion: config.MemoryVersion,
Expand All @@ -65,7 +99,8 @@ func (s *StorageFactory) NewCollectionResourceStorage(cr *internal.CollectionRes

func (s *StorageFactory) GetResourceVersions(ctx context.Context, cluster string) (map[schema.GroupVersionResource]map[string]interface{}, error) {
var resources []Resource
result := s.db.WithContext(ctx).Select("group", "version", "resource", "namespace", "name", "resource_version").
result := s.db.WithContext(ctx).
Select("group", "version", "resource", "namespace", "name", "resource_version").
Where(map[string]interface{}{"cluster": cluster}).
Find(&resources)
if result.Error != nil {
Expand All @@ -91,18 +126,44 @@ func (s *StorageFactory) GetResourceVersions(ctx context.Context, cluster string
}

func (s *StorageFactory) CleanCluster(ctx context.Context, cluster string) error {
result := s.db.WithContext(ctx).Where(map[string]interface{}{"cluster": cluster}).Delete(&Resource{})
return InterpretDBError(cluster, result.Error)
mutex.Lock()
tables, err := s.db.Migrator().GetTables()
if err != nil {
mutex.Unlock()
return err
}
mutex.Unlock()

for _, table := range tables {
result := s.db.WithContext(ctx).Table(table).Where(map[string]interface{}{"cluster": cluster}).Delete(&Resource{})
if result.Error != nil {
return InterpretDBError(cluster, result.Error)
}
}

return nil
}

func (s *StorageFactory) CleanClusterResource(ctx context.Context, cluster string, gvr schema.GroupVersionResource) error {
result := s.db.WithContext(ctx).Where(map[string]interface{}{
"cluster": cluster,
"group": gvr.Group,
"version": gvr.Version,
"resource": gvr.Resource,
}).Delete(&Resource{})
return InterpretDBError(fmt.Sprintf("%s/%s", cluster, gvr), result.Error)
err := s.db.Transaction(func(db *gorm.DB) error {
result := s.db.WithContext(ctx).
Table(GenerateTableFor(gvr)).
Where(map[string]interface{}{
"cluster": cluster,
"group": gvr.Group,
"version": gvr.Version,
"resource": gvr.Resource,
}).
Delete(&Resource{})

if result.Error != nil {
return result.Error
}

return nil
})

return InterpretDBError(fmt.Sprintf("%s/%s", cluster, gvr), err)
}

func (s *StorageFactory) GetCollectionResources(ctx context.Context) ([]*internal.CollectionResource, error) {
Expand All @@ -116,3 +177,13 @@ func (s *StorageFactory) GetCollectionResources(ctx context.Context) ([]*interna
func (s *StorageFactory) PrepareCluster(cluster string) error {
return nil
}

// GenerateTableFor return table name using gvr string
func GenerateTableFor(gvr schema.GroupVersionResource) string {
if gvr.Group == "" {
return fmt.Sprintf("%s_%s", gvr.Version, gvr.Resource)
}

group := strings.ReplaceAll(gvr.Group, ".", "_")
return fmt.Sprintf("%s_%s_%s", group, gvr.Version, gvr.Resource)
}
24 changes: 23 additions & 1 deletion pkg/storage/internalstorage/types.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ type Resource struct {
ID uint `gorm:"primaryKey"`

Group string `gorm:"size:63;not null;uniqueIndex:uni_group_version_resource_cluster_namespace_name;index:idx_group_version_resource_namespace_name;index:idx_group_version_resource_name"`
Version string `gorm:"size:15;not null;uniqueIndex:uni_group_version_resource_cluster_namespace_name;index:idx_group_version_resource_namespace_name;index:idx_group_version_resource_name"`
Version string `gorm:"size:14;not null;uniqueIndex:uni_group_version_resource_cluster_namespace_name;index:idx_group_version_resource_namespace_name;index:idx_group_version_resource_name"`
Resource string `gorm:"size:63;not null;uniqueIndex:uni_group_version_resource_cluster_namespace_name;index:idx_group_version_resource_namespace_name;index:idx_group_version_resource_name"`
Kind string `gorm:"size:63;not null"`

Expand Down Expand Up @@ -99,6 +99,28 @@ func (res Resource) ConvertTo(codec runtime.Codec, object runtime.Object) (runti
return obj, err
}

type GroupVersionResource struct {
ID uint `gorm:"primaryKey"`

Group string `gorm:"size:63;not null"`
Version string `gorm:"size:14;not null"`
Resource string `gorm:"size:63;not null"`
Kind string `gorm:"size:63;not null"`

Cluster string `gorm:"size:253;not null;uniqueIndex:uni_group_version_resource_cluster_namespace_name,length:100;index:idx_cluster"`
Namespace string `gorm:"size:253;not null;uniqueIndex:uni_group_version_resource_cluster_namespace_name,length:50;index:idx_group_version_resource_namespace_name"`
Name string `gorm:"size:253;not null;uniqueIndex:uni_group_version_resource_cluster_namespace_name,length:100;index:idx_group_version_resource_namespace_name;index:idx_group_version_resource_name"`
OwnerUID types.UID `gorm:"column:owner_uid;size:36;not null;default:''"`
UID types.UID `gorm:"size:36;not null"`
ResourceVersion string `gorm:"size:30;not null"`

Object datatypes.JSON `gorm:"not null"`

CreatedAt time.Time `gorm:"not null"`
SyncedAt time.Time `gorm:"not null;autoUpdateTime"`
DeletedAt sql.NullTime
}

type ResourceMetadata struct {
ResourceType `gorm:"embedded"`

Expand Down

0 comments on commit ffb80ad

Please sign in to comment.