Skip to content

Commit da628a7

Browse files
author
calvin
committed
support split table for internal storage
Signed-off-by: calvin <[email protected]>
1 parent 89b8466 commit da628a7

File tree

4 files changed

+125
-51
lines changed

4 files changed

+125
-51
lines changed

‎examples/pediacluster.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ metadata:
44
name: cluster-example
55
spec:
66
apiserver: "https://10.30.43.43:6443"
7-
caData:
7+
caData:
88
tokenData:
99
certData:
1010
keyData:

‎pkg/storage/internalstorage/register.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,12 @@ func NewStorageFactory(configPath string) (storage.StorageFactory, error) {
9393
sqlDB.SetMaxOpenConns(connPool.MaxOpenConns)
9494
sqlDB.SetConnMaxLifetime(connPool.ConnMaxLifetime)
9595

96-
return &StorageFactory{db}, nil
96+
return &StorageFactory{
97+
db: db,
98+
AutoMigration: cfg.AutoMigration,
99+
DivisionPolicy: cfg.DivisionPolicy,
100+
Mapper: cfg.Mapper,
101+
}, nil
97102
}
98103

99104
func newLogger(cfg *Config) (logger.Interface, error) {

‎pkg/storage/internalstorage/resource_storage.go‎

Lines changed: 36 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ import (
2929
type ResourceStorage struct {
3030
db *gorm.DB
3131
codec runtime.Codec
32+
table string
3233

3334
storageGroupResource schema.GroupResource
3435
storageVersion schema.GroupVersion
@@ -83,7 +84,7 @@ func (s *ResourceStorage) Create(ctx context.Context, cluster string, obj runtim
8384
resource.DeletedAt = sql.NullTime{Time: deletedAt.Time, Valid: true}
8485
}
8586

86-
result := s.db.WithContext(ctx).Create(&resource)
87+
result := s.db.WithContext(ctx).Table(s.table).Create(&resource)
8788
return InterpretResourceDBError(cluster, metaobj.GetName(), result.Error)
8889
}
8990

@@ -116,14 +117,19 @@ func (s *ResourceStorage) Update(ctx context.Context, cluster string, obj runtim
116117
updatedResource["deleted_at"] = sql.NullTime{Time: deletedAt.Time, Valid: true}
117118
}
118119

119-
result := s.db.WithContext(ctx).Model(&Resource{}).Where(map[string]interface{}{
120-
"cluster": cluster,
121-
"group": s.storageGroupResource.Group,
122-
"version": s.storageVersion.Version,
123-
"resource": s.storageGroupResource.Resource,
124-
"namespace": metaobj.GetNamespace(),
125-
"name": metaobj.GetName(),
126-
}).Updates(updatedResource)
120+
result := s.db.WithContext(ctx).
121+
Model(&Resource{}).
122+
Where(map[string]interface{}{
123+
"cluster": cluster,
124+
"group": s.storageGroupResource.Group,
125+
"version": s.storageVersion.Version,
126+
"resource": s.storageGroupResource.Resource,
127+
"namespace": metaobj.GetNamespace(),
128+
"name": metaobj.GetName(),
129+
}).
130+
Table(s.table).
131+
Updates(updatedResource)
132+
127133
return InterpretResourceDBError(cluster, metaobj.GetName(), result.Error)
128134
}
129135

@@ -144,14 +150,17 @@ func (c *ResourceStorage) ConvertDeletedObject(obj interface{}) (runtime.Object,
144150
}
145151

146152
func (s *ResourceStorage) deleteObject(cluster, namespace, name string) *gorm.DB {
147-
return s.db.Model(&Resource{}).Where(map[string]interface{}{
148-
"cluster": cluster,
149-
"group": s.storageGroupResource.Group,
150-
"version": s.storageVersion.Version,
151-
"resource": s.storageGroupResource.Resource,
152-
"namespace": namespace,
153-
"name": name,
154-
}).Delete(&Resource{})
153+
return s.db.Model(&Resource{}).
154+
Where(map[string]interface{}{
155+
"cluster": cluster,
156+
"group": s.storageGroupResource.Group,
157+
"version": s.storageVersion.Version,
158+
"resource": s.storageGroupResource.Resource,
159+
"namespace": namespace,
160+
"name": name,
161+
}).
162+
Table(s.table).
163+
Delete(&Resource{})
155164
}
156165

157166
func (s *ResourceStorage) Delete(ctx context.Context, cluster string, obj runtime.Object) error {
@@ -167,14 +176,15 @@ func (s *ResourceStorage) Delete(ctx context.Context, cluster string, obj runtim
167176
}
168177

169178
func (s *ResourceStorage) genGetObjectQuery(ctx context.Context, cluster, namespace, name string) *gorm.DB {
170-
return s.db.WithContext(ctx).Model(&Resource{}).Select("object").Where(map[string]interface{}{
171-
"cluster": cluster,
172-
"group": s.storageGroupResource.Group,
173-
"version": s.storageVersion.Version,
174-
"resource": s.storageGroupResource.Resource,
175-
"namespace": namespace,
176-
"name": name,
177-
})
179+
return s.db.WithContext(ctx).Model(&Resource{}).Table(s.table).Select("object").
180+
Where(map[string]interface{}{
181+
"cluster": cluster,
182+
"group": s.storageGroupResource.Group,
183+
"version": s.storageVersion.Version,
184+
"resource": s.storageGroupResource.Resource,
185+
"namespace": namespace,
186+
"name": name,
187+
})
178188
}
179189

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

202-
query := s.db.WithContext(ctx).Model(&Resource{})
212+
query := s.db.WithContext(ctx).Model(&Resource{}).Table(s.table)
203213
query = query.Where(map[string]interface{}{
204214
"group": s.storageGroupResource.Group,
205215
"version": s.storageVersion.Version,

‎pkg/storage/internalstorage/storage.go‎

Lines changed: 82 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,8 @@ package internalstorage
33
import (
44
"context"
55
"fmt"
6+
"strings"
7+
"sync"
68

79
"gorm.io/gorm"
810
"k8s.io/apimachinery/pkg/runtime/schema"
@@ -11,6 +13,8 @@ import (
1113
"github.com/clusterpedia-io/clusterpedia/pkg/storage"
1214
)
1315

16+
var mutex sync.Mutex
17+
1418
type StorageFactory struct {
1519
db *gorm.DB
1620
AutoMigration *bool
@@ -19,34 +23,54 @@ type StorageFactory struct {
1923
}
2024

2125
func (s *StorageFactory) AutoMigrate() error {
26+
return nil
27+
}
28+
29+
func (s *StorageFactory) GetSupportedRequestVerbs() []string {
30+
return []string{"get", "list"}
31+
}
32+
33+
func (s *StorageFactory) NewResourceStorage(config *storage.ResourceStorageConfig) (storage.ResourceStorage, error) {
34+
mutex.Lock()
35+
defer mutex.Unlock()
36+
37+
var table string
2238
if s.AutoMigration != nil && *s.AutoMigration {
2339
switch s.DivisionPolicy {
24-
if err := s.db.AutoMigrate(&Resource{}); err != nil {
25-
return err
26-
}
2740
case "", DivisionPolicyNone:
41+
table = "resources"
42+
43+
if exist := s.db.Migrator().HasTable(table); !exist {
44+
if err := s.db.AutoMigrate(&Resource{}); err != nil {
45+
return nil, err
46+
}
47+
}
2848
case DivisionPolicyGroupResource:
49+
gvr := schema.GroupVersionResource{
50+
Group: config.StorageGroupResource.Group,
51+
Version: config.StorageVersion.Version,
52+
Resource: config.StorageGroupResource.Resource,
53+
}
2954

30-
}
55+
table = GenerateTableFor(gvr)
56+
57+
if exist := s.db.Migrator().HasTable(table); !exist {
58+
if err := s.db.AutoMigrate(&Resource{}); err != nil {
59+
return nil, err
60+
}
3161

32-
if s.DivisionPolicy == "" || s.DivisionPolicy == DivisionPolicyNone {
33-
if err := s.db.AutoMigrate(&Resource{}); err != nil {
34-
return err
62+
err := s.db.Migrator().RenameTable("resources", table)
63+
if err != nil {
64+
return nil, err
65+
}
3566
}
3667
}
3768
}
3869

39-
return nil
40-
}
41-
42-
func (s *StorageFactory) GetSupportedRequestVerbs() []string {
43-
return []string{"get", "list"}
44-
}
45-
46-
func (s *StorageFactory) NewResourceStorage(config *storage.ResourceStorageConfig) (storage.ResourceStorage, error) {
4770
return &ResourceStorage{
4871
db: s.db,
4972
codec: config.Codec,
73+
table: table,
5074

5175
storageGroupResource: config.StorageGroupResource,
5276
storageVersion: config.StorageVersion,
@@ -65,11 +89,23 @@ func (s *StorageFactory) NewCollectionResourceStorage(cr *internal.CollectionRes
6589

6690
func (f *StorageFactory) GetResourceVersions(ctx context.Context, cluster string) (map[schema.GroupVersionResource]map[string]interface{}, error) {
6791
var resources []Resource
68-
result := f.db.WithContext(ctx).Select("group", "version", "resource", "namespace", "name", "resource_version").
69-
Where(map[string]interface{}{"cluster": cluster}).
70-
Find(&resources)
71-
if result.Error != nil {
72-
return nil, InterpretDBError(cluster, result.Error)
92+
mutex.Lock()
93+
tables, err := f.db.Migrator().GetTables()
94+
if err != nil {
95+
mutex.Unlock()
96+
return nil, err
97+
}
98+
mutex.Unlock()
99+
for _, table := range tables {
100+
var tableResources []Resource
101+
result := f.db.WithContext(ctx).Table(table).Select("group", "version", "resource", "namespace", "name", "resource_version").
102+
Where(map[string]interface{}{"cluster": cluster}).
103+
Find(&tableResources)
104+
if result.Error != nil {
105+
return nil, InterpretDBError(cluster, result.Error)
106+
}
107+
108+
resources = append(resources, tableResources...)
73109
}
74110

75111
resourceversions := make(map[schema.GroupVersionResource]map[string]interface{})
@@ -91,12 +127,25 @@ func (f *StorageFactory) GetResourceVersions(ctx context.Context, cluster string
91127
}
92128

93129
func (f *StorageFactory) CleanCluster(ctx context.Context, cluster string) error {
94-
result := f.db.WithContext(ctx).Where(map[string]interface{}{"cluster": cluster}).Delete(&Resource{})
95-
return InterpretDBError(cluster, result.Error)
130+
mutex.Lock()
131+
tables, err := f.db.Migrator().GetTables()
132+
if err != nil {
133+
mutex.Unlock()
134+
return err
135+
}
136+
mutex.Unlock()
137+
138+
for _, table := range tables {
139+
result := f.db.WithContext(ctx).Table(table).Where(map[string]interface{}{"cluster": cluster}).Delete(&Resource{})
140+
if result.Error != nil {
141+
return InterpretDBError(cluster, result.Error)
142+
}
143+
}
144+
return nil
96145
}
97146

98147
func (s *StorageFactory) CleanClusterResource(ctx context.Context, cluster string, gvr schema.GroupVersionResource) error {
99-
result := s.db.WithContext(ctx).Where(map[string]interface{}{
148+
result := s.db.WithContext(ctx).Table(GenerateTableFor(gvr)).Where(map[string]interface{}{
100149
"cluster": cluster,
101150
"group": gvr.Group,
102151
"version": gvr.Version,
@@ -116,3 +165,13 @@ func (s *StorageFactory) GetCollectionResources(ctx context.Context) ([]*interna
116165
func (s *StorageFactory) PrepareCluster(cluster string) error {
117166
return nil
118167
}
168+
169+
// GenerateTableFor return table name using gvr string
170+
func GenerateTableFor(gvr schema.GroupVersionResource) string {
171+
if gvr.Group == "" {
172+
return fmt.Sprintf("%s_%s", gvr.Version, gvr.Resource)
173+
}
174+
175+
group := strings.ReplaceAll(gvr.Group, ".", "_")
176+
return fmt.Sprintf("%s_%s_%s", group, gvr.Version, gvr.Resource)
177+
}

0 commit comments

Comments
 (0)