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 <wen.chen@daocloud.io>
  • Loading branch information
calvin committed Mar 14, 2024
1 parent 29c0306 commit ead917e
Show file tree
Hide file tree
Showing 6 changed files with 185 additions and 92 deletions.
26 changes: 3 additions & 23 deletions pkg/storage/internalstorage/config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,6 @@ import (
"gopkg.in/natefinch/lumberjack.v2"
"gorm.io/gorm/logger"
"k8s.io/klog/v2"

clusterv1alpha2 "github.com/clusterpedia-io/api/cluster/v1alpha2"
)

const (
Expand All @@ -30,7 +28,7 @@ type DivisionPolicy string

const (
DivisionPolicyNone DivisionPolicy = "None"
DivisionPolicyGroupResource DivisionPolicy = "GroupResource"
DivisionPolicyGroupResource DivisionPolicy = "GVR"
DivisionPolicyCustom DivisionPolicy = "Custom"
)

Expand Down Expand Up @@ -58,30 +56,12 @@ type Config struct {

Params map[string]string `yaml:"params"`

AutoMigration *bool `yaml:"autoMigration"` // If set to false, no tables will be created
DivisionPolicy DivisionPolicy `yaml:"divisionPolicy"`
Mapper []ResourceMapper `yaml:"mapper"` // Only DivisionPolicy is DivisionPolicyCustom it need to specify the mapping between resource and table
SkipAutoMigration bool `yaml:"autoMigration"` // If set to false, no tables will be created
DivisionPolicy DivisionPolicy `yaml:"divisionPolicy"`

Log *LogConfig `yaml:"log"`
}

type ResourceMapper struct {
Table *Table `yaml:"table"`
Resources []clusterv1alpha2.ClusterGroupResources `yaml:"resources"`
}

type Table struct {
Name string `yaml:"name"`
ExtraFields []ExtraField `yaml:"extraFields"`
}

type ExtraField struct {
Name string `yaml:"name"`
PlainPath string `yaml:"plainPath"`
Type string `yaml:"type"`
Index string `yaml:"index"`
}

type LogConfig struct {
Stdout bool `yaml:"stdout"`
Level string `yaml:"level"`
Expand Down
6 changes: 5 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,11 @@ func NewStorageFactory(configPath string) (storage.StorageFactory, error) {
sqlDB.SetMaxOpenConns(connPool.MaxOpenConns)
sqlDB.SetConnMaxLifetime(connPool.ConnMaxLifetime)

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

func newLogger(cfg *Config) (logger.Interface, error) {
Expand Down
62 changes: 34 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,15 @@ 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.Where(map[string]interface{}{
"cluster": cluster,

Check failure on line 152 in pkg/storage/internalstorage/resource_storage.go

View workflow job for this annotation

GitHub Actions / Lint with golangci-lint 

File is not `gci`-ed with --skip-generated -s Standard -s Default -s Prefix(github.com/clusterpedia-io) (gci)
"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 +172,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).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 +205,7 @@ func (s *ResourceStorage) genListObjectsQuery(ctx context.Context, opts *interna
result = &ResourceMetadataList{}
}

query := s.db.WithContext(ctx).Model(&Resource{})
query := s.db.WithContext(ctx)
query = query.Where(map[string]interface{}{
"group": s.storageGroupResource.Group,
"version": s.storageVersion.Version,
Expand Down
3 changes: 2 additions & 1 deletion pkg/storage/internalstorage/resource_storage_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -465,7 +465,8 @@ func TestResourceStorage_Update(t *testing.T) {

func newTestResourceStorage(db *gorm.DB, storageGVK schema.GroupVersionResource) *ResourceStorage {
return &ResourceStorage{
db: db,
db: db.Table("resources"),
model: &Resource{},
storageGroupResource: storageGVK.GroupResource(),
storageVersion: storageGVK.GroupVersion(),
}
Expand Down
156 changes: 118 additions & 38 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 @@ -12,30 +14,14 @@ import (
)

type StorageFactory struct {
db *gorm.DB
AutoMigration *bool
DivisionPolicy DivisionPolicy
Mapper []ResourceMapper
sync.Mutex

db *gorm.DB
SkipAutoMigration bool
DivisionPolicy DivisionPolicy
}

func (s *StorageFactory) AutoMigrate() error {
if s.AutoMigration != nil && *s.AutoMigration {
switch s.DivisionPolicy {
if err := s.db.AutoMigrate(&Resource{}); err != nil {
return err
}
case "", DivisionPolicyNone:
case DivisionPolicyGroupResource:

}

if s.DivisionPolicy == "" || s.DivisionPolicy == DivisionPolicyNone {
if err := s.db.AutoMigrate(&Resource{}); err != nil {
return err
}
}
}

return nil
}

Expand All @@ -44,10 +30,47 @@ func (s *StorageFactory) GetSupportedRequestVerbs() []string {
}

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

gvr := schema.GroupVersionResource{
Group: config.StorageGroupResource.Group,
Version: config.StorageVersion.Version,
Resource: config.StorageGroupResource.Resource,
}
table := s.tableName(gvr)

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

err := s.db.Migrator().RenameTable("groupversionresources", table)
if err != nil {
return nil, err
}
}
}
case "", DivisionPolicyNone:
model = &Resource{}
if !s.SkipAutoMigration {
if exist := s.db.Migrator().HasTable(table); !exist {
if err := s.db.AutoMigrate(&Resource{}); err != nil {
return nil, err
}
}
}
}

return &ResourceStorage{
db: s.db.Table(table).Model(model),
model: model,
codec: config.Codec,
storageGroupResource: config.StorageGroupResource,
storageVersion: config.StorageVersion,
memoryVersion: config.MemoryVersion,
Expand All @@ -64,12 +87,24 @@ func (s *StorageFactory) NewCollectionResourceStorage(cr *internal.CollectionRes
}

func (s *StorageFactory) GetResourceVersions(ctx context.Context, cluster string) (map[schema.GroupVersionResource]map[string]interface{}, error) {
s.Lock()
tables, err := s.db.Migrator().GetTables()
if err != nil {
s.Unlock()
return nil, err
}
s.Unlock()

var resources []Resource
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 {
return nil, InterpretDBError(cluster, result.Error)
for _, table := range tables {
result := s.db.WithContext(ctx).
Table(table).
Select("group", "version", "resource", "namespace", "name", "resource_version").
Where(map[string]interface{}{"cluster": cluster}).
Find(&resources)
if result.Error != nil {
return nil, InterpretDBError(cluster, result.Error)
}
}

resourceversions := make(map[schema.GroupVersionResource]map[string]interface{})
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)
s.Lock()
tables, err := s.db.Migrator().GetTables()
if err != nil {
s.Unlock()
return err
}
s.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(s.tableName(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,22 @@ 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)
}

func (s *StorageFactory) tableName(gvr schema.GroupVersionResource) string {
table := "resources"
if s.DivisionPolicy == DivisionPolicyGroupResource {
table = GenerateTableFor(gvr)
}

return table
}
Loading

0 comments on commit ead917e

Please sign in to comment.