Skip to content

Commit

Permalink
replace storage.ResourceStorageConfig with ResourceConfig
Browse files Browse the repository at this point in the history 
Signed-off-by: Iceber Gu <[email protected]>
  • Loading branch information
@Iceber
Iceber committed Jun 24, 2024
1 parent f1b0b14 commit 3e4a62e
Show file tree
Hide file tree
Showing 13 changed files with 115 additions and 262 deletions.
10 changes: 5 additions & 5 deletions pkg/apiserver/registry/clusterpedia/collectionresources/rest.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,8 @@ import (
internal "github.com/clusterpedia-io/api/clusterpedia"
"github.com/clusterpedia-io/api/clusterpedia/scheme"
"github.com/clusterpedia-io/api/clusterpedia/v1beta1"
resourceconfigfactory "github.com/clusterpedia-io/clusterpedia/pkg/runtime/resourceconfig/factory"
"github.com/clusterpedia-io/clusterpedia/pkg/storage"
"github.com/clusterpedia-io/clusterpedia/pkg/storageconfig"
"github.com/clusterpedia-io/clusterpedia/pkg/utils"
"github.com/clusterpedia-io/clusterpedia/pkg/utils/negotiation"
"github.com/clusterpedia-io/clusterpedia/pkg/utils/request"
Expand All @@ -48,7 +48,7 @@ func NewREST(serializer runtime.NegotiatedSerializer, factory storage.StorageFac

list := &internal.CollectionResourceList{}
storages := make(map[string]storage.CollectionResourceStorage, len(crs))
configFactory := storageconfig.NewStorageConfigFactory()
configFactory := resourceconfigfactory.New()
for _, cr := range crs {
for irt := range cr.ResourceTypes {
rt := &cr.ResourceTypes[irt]
Expand All @@ -59,9 +59,9 @@ func NewREST(serializer runtime.NegotiatedSerializer, factory storage.StorageFac
}

*rt = internal.CollectionResourceType{
Group: config.StorageGroupResource.Group,
Version: config.StorageVersion.Version,
Resource: config.StorageGroupResource.Resource,
Group: config.StorageResource.Group,
Version: config.StorageResource.Version,
Resource: config.StorageResource.Resource,
}
}
}
Expand Down
17 changes: 8 additions & 9 deletions pkg/kube_state_metrics/metrics_store.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,22 +16,22 @@ import (
"k8s.io/kube-state-metrics/v2/pkg/optin"
"k8s.io/kube-state-metrics/v2/pkg/options"

resourceconfigfactory "github.com/clusterpedia-io/clusterpedia/pkg/runtime/resourceconfig/factory"
"github.com/clusterpedia-io/clusterpedia/pkg/runtime/scheme"
"github.com/clusterpedia-io/clusterpedia/pkg/storageconfig"
)

var (
storageConfigFactory = storageconfig.NewStorageConfigFactory()
hubGVRs = make(map[schema.GroupVersionResource]schema.GroupVersionResource)
resourceConfigFactory = resourceconfigfactory.New()
hubGVRs = make(map[schema.GroupVersionResource]schema.GroupVersionResource)
)

func init() {
for gvr := range generators {
config, err := storageConfigFactory.NewLegacyResourceConfig(gvr.GroupResource(), false)
memory, err := resourceConfigFactory.MemoryResource(gvr)
if err != nil {
panic(err)
}
hubGVRs[config.StorageGroupResource.WithVersion(config.MemoryVersion.Version)] = gvr
hubGVRs[memory] = gvr
}
}

Expand Down Expand Up @@ -101,11 +101,10 @@ func (builder *MetricsStoreBuilder) GetMetricStore(cluster string, resource sche
return nil
}

config, err := storageConfigFactory.NewLegacyResourceConfig(resource.GroupResource(), false)
hub, err := resourceConfigFactory.MemoryResource(resource)
if err != nil {
return nil
}
hub := config.StorageGroupResource.WithVersion(config.MemoryVersion.Version)
metricsGVR, ok := hubGVRs[hub]
if !ok {
return nil
Expand Down Expand Up @@ -134,11 +133,11 @@ func (builder *MetricsStoreBuilder) GetMetricStore(cluster string, resource sche
return obj, nil
}

hobj, err := scheme.LegacyResourceScheme.ConvertToVersion(obj.(runtime.Object), config.MemoryVersion)
hobj, err := scheme.LegacyResourceScheme.ConvertToVersion(obj.(runtime.Object), hub.GroupVersion())
if err != nil {
return nil, err
}
if metricsGVR.GroupVersion() == config.MemoryVersion {
if metricsGVR.GroupVersion() == hub.GroupVersion() {
return hobj, nil
}
return scheme.LegacyResourceScheme.ConvertToVersion(hobj, metricsGVR.GroupVersion())
Expand Down
22 changes: 11 additions & 11 deletions pkg/kubeapiserver/restmanager.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,16 +28,16 @@ import (
"github.com/clusterpedia-io/clusterpedia/pkg/kubeapiserver/discovery"
"github.com/clusterpedia-io/clusterpedia/pkg/kubeapiserver/printers"
"github.com/clusterpedia-io/clusterpedia/pkg/kubeapiserver/resourcerest"
resourceconfigfactory "github.com/clusterpedia-io/clusterpedia/pkg/runtime/resourceconfig/factory"
"github.com/clusterpedia-io/clusterpedia/pkg/runtime/scheme"
unstructuredscheme "github.com/clusterpedia-io/clusterpedia/pkg/runtime/scheme/unstructured"
"github.com/clusterpedia-io/clusterpedia/pkg/storage"
"github.com/clusterpedia-io/clusterpedia/pkg/storageconfig"
)

type RESTManager struct {
serializer runtime.NegotiatedSerializer
storageFactory storage.StorageFactory
resourcetSorageConfig *storageconfig.StorageConfigFactory
resourceConfigFactory *resourceconfigfactory.ResourceConfigFactory
equivalentResourceRegistry runtime.EquivalentResourceMapper

lock sync.Mutex
Expand Down Expand Up @@ -87,7 +87,7 @@ func NewRESTManager(serializer runtime.NegotiatedSerializer, storageMediaType st
manager := &RESTManager{
serializer: serializer,
storageFactory: storageFactory,
resourcetSorageConfig: storageconfig.NewStorageConfigFactory(),
resourceConfigFactory: resourceconfigfactory.New(),
equivalentResourceRegistry: runtime.NewEquivalentResourceRegistry(),
requestVerbs: requestVerbs,
}
Expand Down Expand Up @@ -270,12 +270,12 @@ func (m *RESTManager) addRESTResourceInfosLocked(addedInfos map[schema.GroupVers
}

func (m *RESTManager) genLegacyResourceRESTStorage(gvr schema.GroupVersionResource, kind string, namespaced bool) (*resourcerest.RESTStorage, error) {
storageConfig, err := m.resourcetSorageConfig.NewLegacyResourceConfig(gvr.GroupResource(), namespaced)
resourceConfig, err := m.resourceConfigFactory.NewLegacyResourceConfig(gvr.GroupResource(), namespaced)
if err != nil {
return nil, err
}

resourceStorage, err := m.storageFactory.NewResourceStorage(storageConfig)
resourceStorage, err := m.storageFactory.NewResourceStorage(resourceConfig)
if err != nil {
return nil, err
}
Expand All @@ -284,11 +284,11 @@ func (m *RESTManager) genLegacyResourceRESTStorage(gvr schema.GroupVersionResour
DefaultQualifiedResource: gvr.GroupResource(),

NewFunc: func() runtime.Object {
obj, _ := scheme.LegacyResourceScheme.New(storageConfig.MemoryVersion.WithKind(kind))
obj, _ := scheme.LegacyResourceScheme.New(resourceConfig.MemoryResource.GroupVersion().WithKind(kind))
return obj
},
NewListFunc: func() runtime.Object {
obj, _ := scheme.LegacyResourceScheme.New(storageConfig.MemoryVersion.WithKind(kind + "List"))
obj, _ := scheme.LegacyResourceScheme.New(resourceConfig.MemoryResource.GroupVersion().WithKind(kind + "List"))
return obj
},

Expand All @@ -297,25 +297,25 @@ func (m *RESTManager) genLegacyResourceRESTStorage(gvr schema.GroupVersionResour
}

func (m *RESTManager) genUnstructuredRESTStorage(gvr schema.GroupVersionResource, kind string, namespaced bool) (*resourcerest.RESTStorage, error) {
storageConfig, err := m.resourcetSorageConfig.NewUnstructuredConfig(gvr, namespaced)
resourceConfig, err := m.resourceConfigFactory.NewUnstructuredConfig(gvr, namespaced)
if err != nil {
return nil, err
}

resourceStorage, err := m.storageFactory.NewResourceStorage(storageConfig)
resourceStorage, err := m.storageFactory.NewResourceStorage(resourceConfig)
if err != nil {
return nil, err
}

return &resourcerest.RESTStorage{
NewFunc: func() runtime.Object {
obj := &unstructured.Unstructured{}
obj.SetGroupVersionKind(storageConfig.MemoryVersion.WithKind(kind))
obj.SetGroupVersionKind(resourceConfig.MemoryResource.GroupVersion().WithKind(kind))
return obj
},
NewListFunc: func() runtime.Object {
obj := &unstructured.UnstructuredList{}
obj.SetGroupVersionKind(storageConfig.MemoryVersion.WithKind(kind + "List"))
obj.SetGroupVersionKind(resourceConfig.MemoryResource.GroupVersion().WithKind(kind + "List"))
return obj
},

Expand Down
93 changes: 41 additions & 52 deletions pkg/storage/internalstorage/resource_storage.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,24 +23,37 @@ import (
"k8s.io/client-go/tools/cache"

internal "github.com/clusterpedia-io/api/clusterpedia"
"github.com/clusterpedia-io/clusterpedia/pkg/storage"
"github.com/clusterpedia-io/clusterpedia/pkg/runtime/resourceconfig"
)

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

storageGroupResource schema.GroupResource
storageVersion schema.GroupVersion
memoryVersion schema.GroupVersion
db *gorm.DB
config resourceconfig.ResourceConfig
}

func (s *ResourceStorage) GetStorageConfig() *storage.ResourceStorageConfig {
return &storage.ResourceStorageConfig{
Codec: s.codec,
StorageGroupResource: s.storageGroupResource,
StorageVersion: s.storageVersion,
MemoryVersion: s.memoryVersion,
func (s *ResourceStorage) GetResourceConfig() *resourceconfig.ResourceConfig {
config := s.config
return &config
}

func (s *ResourceStorage) gvrKeyMap() map[string]interface{} {
return map[string]interface{}{
"group": s.config.StorageResource.Group,
"version": s.config.StorageResource.Version,
"resource": s.config.StorageResource.Resource,
}
}

func (s *ResourceStorage) resourceKeyMap(cluster, namespace, name string) map[string]interface{} {
return map[string]interface{}{
"cluster": cluster,
"group": s.config.StorageResource.Group,
"version": s.config.StorageResource.Version,
"resource": s.config.StorageResource.Resource,
"namespace": namespace,
"name": name,
}
}

Expand All @@ -61,7 +74,7 @@ func (s *ResourceStorage) Create(ctx context.Context, cluster string, obj runtim
}

var buffer bytes.Buffer
if err := s.codec.Encode(obj, &buffer); err != nil {
if err := s.config.Codec.Encode(obj, &buffer); err != nil {
return err
}

Expand All @@ -71,9 +84,9 @@ func (s *ResourceStorage) Create(ctx context.Context, cluster string, obj runtim
UID: metaobj.GetUID(),
Name: metaobj.GetName(),
Namespace: metaobj.GetNamespace(),
Group: s.storageGroupResource.Group,
Resource: s.storageGroupResource.Resource,
Version: s.storageVersion.Version,
Group: s.config.StorageResource.Group,
Resource: s.config.StorageResource.Resource,
Version: s.config.StorageResource.Version,
Kind: gvk.Kind,
ResourceVersion: metaobj.GetResourceVersion(),
Object: buffer.Bytes(),
Expand All @@ -94,7 +107,7 @@ func (s *ResourceStorage) Update(ctx context.Context, cluster string, obj runtim
}

var buffer bytes.Buffer
if err := s.codec.Encode(obj, &buffer); err != nil {
if err := s.config.Codec.Encode(obj, &buffer); err != nil {
return err
}

Expand All @@ -116,14 +129,9 @@ 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(&Resource{}).
Where(s.resourceKeyMap(cluster, metaobj.GetNamespace(), metaobj.GetName())).
Updates(updatedResource)
return InterpretResourceDBError(cluster, metaobj.GetName(), result.Error)
}

Expand All @@ -144,14 +152,7 @@ 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(&Resource{}).Where(s.resourceKeyMap(cluster, namespace, name)).Delete(&Resource{})
}

func (s *ResourceStorage) Delete(ctx context.Context, cluster string, obj runtime.Object) error {
Expand All @@ -167,14 +168,7 @@ 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(&Resource{}).Select("object").Where(s.resourceKeyMap(cluster, namespace, name))
}

func (s *ResourceStorage) Get(ctx context.Context, cluster, namespace, name string, into runtime.Object) error {
Expand All @@ -183,7 +177,7 @@ func (s *ResourceStorage) Get(ctx context.Context, cluster, namespace, name stri
return InterpretResourceDBError(cluster, namespace+"/"+name, result.Error)
}

obj, _, err := s.codec.Decode(objects[0], nil, into)
obj, _, err := s.config.Codec.Decode(objects[0], nil, into)
if err != nil {
return err
}
Expand All @@ -199,12 +193,7 @@ func (s *ResourceStorage) genListObjectsQuery(ctx context.Context, opts *interna
result = &ResourceMetadataList{}
}

query := s.db.WithContext(ctx).Model(&Resource{})
query = query.Where(map[string]interface{}{
"group": s.storageGroupResource.Group,
"version": s.storageVersion.Version,
"resource": s.storageGroupResource.Resource,
})
query := s.db.WithContext(ctx).Model(&Resource{}).Where(s.gvrKeyMap())
offset, amount, query, err := applyListOptionsToResourceQuery(s.db, query, opts)
return offset, amount, query, result, err
}
Expand All @@ -216,7 +205,7 @@ func (s *ResourceStorage) List(ctx context.Context, listObject runtime.Object, o
}

if err := result.From(query); err != nil {
return InterpretDBError(s.storageGroupResource.String(), err)
return InterpretDBError(s.groupResource.String(), err)
}
objects := result.Items()

Expand Down Expand Up @@ -246,7 +235,7 @@ func (s *ResourceStorage) List(ctx context.Context, listObject runtime.Object, o
unstructuredList.Items = make([]unstructured.Unstructured, 0, len(objects))
for _, object := range objects {
uObj := &unstructured.Unstructured{}
obj, err := object.ConvertTo(s.codec, uObj)
obj, err := object.ConvertTo(s.config.Codec, uObj)
if err != nil {
return err
}
Expand Down Expand Up @@ -283,7 +272,7 @@ func (s *ResourceStorage) List(ctx context.Context, listObject runtime.Object, o
slice := reflect.MakeSlice(v.Type(), len(objects), len(objects))
expected := reflect.New(v.Type().Elem()).Interface().(runtime.Object)
for i, object := range objects {
obj, err := object.ConvertTo(s.codec, expected.DeepCopyObject())
obj, err := object.ConvertTo(s.config.Codec, expected.DeepCopyObject())
if err != nil {
return err
}
Expand All @@ -294,7 +283,7 @@ func (s *ResourceStorage) List(ctx context.Context, listObject runtime.Object, o
}

func (s *ResourceStorage) Watch(_ context.Context, _ *internal.ListOptions) (watch.Interface, error) {
return nil, apierrors.NewMethodNotSupported(s.storageGroupResource, "watch")
return nil, apierrors.NewMethodNotSupported(s.groupResource, "watch")
}

func applyListOptionsToResourceQuery(db *gorm.DB, query *gorm.DB, opts *internal.ListOptions) (int64, *int64, *gorm.DB, error) {
Expand Down
Loading

0 comments on commit 3e4a62e

Please sign in to comment.