From 6e39f2458890d158f5ed49fdfeb39c675f117727 Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Thu, 18 Jul 2024 07:47:47 -0700 Subject: [PATCH] Storage: Update storage.Interface for backend (#90382) --- pkg/apiserver/registry/generic/key.go | 7 +- pkg/apiserver/registry/generic/key_test.go | 6 + pkg/apiserver/storage/file/restoptions.go | 80 --- pkg/apiserver/storage/file/util.go | 121 ---- pkg/apiserver/storage/testing/utils.go | 6 +- .../storage/testing/watcher_tests.go | 10 +- pkg/registry/apis/dashboard/legacy_storage.go | 4 +- .../apiserver/aggregator/aggregator.go | 2 +- pkg/services/apiserver/options/aggregator.go | 14 +- pkg/services/apiserver/service.go | 15 +- pkg/storage/unified/apistore/prepare.go | 4 + pkg/storage/unified/apistore/restoptions.go | 101 +++- pkg/storage/unified/apistore/storage.go | 537 ----------------- .../unified/apistore/store.go} | 560 ++++++++++-------- .../unified/apistore/store_test.go} | 16 +- pkg/storage/unified/apistore/stream.go | 16 +- pkg/storage/unified/apistore/util.go | 163 +++++ .../unified/apistore}/watcher_test.go | 48 +- .../unified/apistore}/watchset.go | 2 +- pkg/storage/unified/resource/validation.go | 2 +- .../unified/resource/validation_test.go | 29 + 21 files changed, 682 insertions(+), 1061 deletions(-) delete mode 100644 pkg/apiserver/storage/file/restoptions.go delete mode 100644 pkg/apiserver/storage/file/util.go delete mode 100644 pkg/storage/unified/apistore/storage.go rename pkg/{apiserver/storage/file/file.go => storage/unified/apistore/store.go} (60%) rename pkg/{apiserver/storage/file/file_test.go => storage/unified/apistore/store_test.go} (95%) create mode 100644 pkg/storage/unified/apistore/util.go rename pkg/{apiserver/storage/file => storage/unified/apistore}/watcher_test.go (86%) rename pkg/{apiserver/storage/file => storage/unified/apistore}/watchset.go (99%) create mode 100644 pkg/storage/unified/resource/validation_test.go diff --git a/pkg/apiserver/registry/generic/key.go b/pkg/apiserver/registry/generic/key.go index f257cc03894..01ce8eb89ea 100644 --- a/pkg/apiserver/registry/generic/key.go +++ b/pkg/apiserver/registry/generic/key.go @@ -32,6 +32,11 @@ func ParseKey(raw string) (*Key, error) { for i := 0; i < len(parts); i += 2 { k := parts[i] if i+1 >= len(parts) { + // Kube aggregator just appends the name to a key + if key.Group != "" && key.Resource != "" && key.Namespace == "" && key.Name == "" { + key.Name = k + return key, nil + } return nil, fmt.Errorf("invalid key: %s", raw) } v := parts[i+1] @@ -45,7 +50,7 @@ func ParseKey(raw string) (*Key, error) { case "name": key.Name = v default: - return nil, fmt.Errorf("invalid key name: %s", key) + return nil, fmt.Errorf("invalid key part: %s", raw) } } diff --git a/pkg/apiserver/registry/generic/key_test.go b/pkg/apiserver/registry/generic/key_test.go index c000bd58784..cbf5b78c9c2 100644 --- a/pkg/apiserver/registry/generic/key_test.go +++ b/pkg/apiserver/registry/generic/key_test.go @@ -54,6 +54,12 @@ func TestParseKey(t *testing.T) { expected: nil, wantErr: true, }, + { + name: "Support kube-aggregator format", + raw: "/group/test-group/resource/test-resource/test-name", + expected: &Key{Group: "test-group", Resource: "test-resource", Name: "test-name"}, + wantErr: false, + }, } for _, tt := range tests { diff --git a/pkg/apiserver/storage/file/restoptions.go b/pkg/apiserver/storage/file/restoptions.go deleted file mode 100644 index 69d0ca71d87..00000000000 --- a/pkg/apiserver/storage/file/restoptions.go +++ /dev/null @@ -1,80 +0,0 @@ -// SPDX-License-Identifier: AGPL-3.0-only - -package file - -import ( - "os" - "path/filepath" - "time" - - "k8s.io/apimachinery/pkg/runtime/schema" - "k8s.io/apiserver/pkg/registry/generic" - "k8s.io/apiserver/pkg/storage/storagebackend" - flowcontrolrequest "k8s.io/apiserver/pkg/util/flowcontrol/request" -) - -var _ generic.RESTOptionsGetter = (*RESTOptionsGetter)(nil) - -type RESTOptionsGetter struct { - path string - original storagebackend.Config -} - -// Optionally, this constructor allows specifying directories -// for resources that are required to be read/watched on startup and there -// won't be any write operations that initially bootstrap their directories -func NewRESTOptionsGetter(path string, - originalStorageConfig storagebackend.Config, - createResourceDirs ...string) (*RESTOptionsGetter, error) { - if path == "" { - path = filepath.Join(os.TempDir(), "grafana-apiserver") - } - - if err := initializeDirs(path, createResourceDirs); err != nil { - return nil, err - } - - return &RESTOptionsGetter{path: path, original: originalStorageConfig}, nil -} - -func initializeDirs(root string, createResourceDirs []string) error { - for _, dir := range createResourceDirs { - if err := ensureDir(filepath.Join(root, dir)); err != nil { - return err - } - } - return nil -} - -func (r *RESTOptionsGetter) GetRESTOptions(resource schema.GroupResource) (generic.RESTOptions, error) { - storageConfig := &storagebackend.ConfigForResource{ - Config: storagebackend.Config{ - Type: "file", - Prefix: r.path, - Transport: storagebackend.TransportConfig{}, - Codec: r.original.Codec, - EncodeVersioner: r.original.EncodeVersioner, - Transformer: r.original.Transformer, - CompactionInterval: 0, - CountMetricPollPeriod: 0, - DBMetricPollInterval: 0, - HealthcheckTimeout: 0, - ReadycheckTimeout: 0, - StorageObjectCountTracker: flowcontrolrequest.NewStorageObjectCountTracker(), - }, - GroupResource: resource, - } - - ret := generic.RESTOptions{ - StorageConfig: storageConfig, - Decorator: NewStorage, - DeleteCollectionWorkers: 0, - EnableGarbageCollection: false, - // k8s expects forward slashes here, we'll convert them to os path separators in the storage - ResourcePrefix: "/group/" + resource.Group + "/resource/" + resource.Resource, - CountMetricPollPeriod: 1 * time.Second, - StorageObjectCountTracker: storageConfig.Config.StorageObjectCountTracker, - } - - return ret, nil -} diff --git a/pkg/apiserver/storage/file/util.go b/pkg/apiserver/storage/file/util.go deleted file mode 100644 index 1ce9bf18dbe..00000000000 --- a/pkg/apiserver/storage/file/util.go +++ /dev/null @@ -1,121 +0,0 @@ -// SPDX-License-Identifier: AGPL-3.0-only -// Provenance-includes-location: https://github.com/kubernetes-sigs/apiserver-runtime/blob/main/pkg/experimental/storage/filepath/jsonfile_rest.go -// Provenance-includes-license: Apache-2.0 -// Provenance-includes-copyright: The Kubernetes Authors. - -package file - -import ( - "bytes" - "errors" - "os" - "path/filepath" - "strings" - "sync/atomic" - - "k8s.io/apimachinery/pkg/runtime" -) - -func (s *Storage) filePath(key string) string { - // Replace backslashes with underscores to avoid creating bogus subdirectories - key = strings.Replace(key, "\\", "_", -1) - fileName := filepath.Join(s.root, filepath.Clean(key+".json")) - return fileName -} - -// this is for constructing dirPath in a sanitized way provided you have -// already calculated the key. In order to go in the other direction, from a file path -// key to its dir, use the go standard library: filepath.Dir -func (s *Storage) dirPath(key string) string { - return dirPath(s.root, key) -} - -func writeFile(codec runtime.Codec, path string, obj runtime.Object) error { - buf := new(bytes.Buffer) - if err := codec.Encode(obj, buf); err != nil { - return err - } - return os.WriteFile(path, buf.Bytes(), 0600) -} - -var fileReadCount uint64 = 0 - -//func getReadsAndReset() uint64 { -//return atomic.SwapUint64(&fileReadCount, 0) -//} - -func readFile(codec runtime.Codec, path string, newFunc func() runtime.Object) (runtime.Object, error) { - atomic.AddUint64(&fileReadCount, 1) - content, err := os.ReadFile(filepath.Clean(path)) - if err != nil { - return nil, err - } - newObj := newFunc() - decodedObj, _, err := codec.Decode(content, nil, newObj) - if err != nil { - return nil, err - } - return decodedObj, nil -} - -func readDirRecursive(codec runtime.Codec, path string, newFunc func() runtime.Object) ([]runtime.Object, error) { - var objs []runtime.Object - err := filepath.Walk(path, func(path string, info os.FileInfo, err error) error { - if err != nil { - return err - } - if info.IsDir() || filepath.Ext(path) != ".json" { - return nil - } - obj, err := readFile(codec, path, newFunc) - if err != nil { - return err - } - objs = append(objs, obj) - return nil - }) - if err != nil { - if errors.Is(err, os.ErrNotExist) { - return objs, nil - } - return nil, err - } - return objs, nil -} - -func deleteFile(path string) error { - return os.Remove(path) -} - -func exists(filepath string) bool { - _, err := os.Stat(filepath) - return err == nil -} - -func dirPath(root string, key string) string { - // Replace backslashes with underscores to avoid creating bogus subdirectories - key = strings.Replace(key, "\\", "_", -1) - dirName := filepath.Join(root, filepath.Clean(key)) - return dirName -} - -func ensureDir(dirname string) error { - if !exists(dirname) { - return os.MkdirAll(dirname, 0700) - } - return nil -} - -func isUnchanged(codec runtime.Codec, obj runtime.Object, newObj runtime.Object) (bool, error) { - buf := new(bytes.Buffer) - if err := codec.Encode(obj, buf); err != nil { - return false, err - } - - newBuf := new(bytes.Buffer) - if err := codec.Encode(newObj, newBuf); err != nil { - return false, err - } - - return bytes.Equal(buf.Bytes(), newBuf.Bytes()), nil -} diff --git a/pkg/apiserver/storage/testing/utils.go b/pkg/apiserver/storage/testing/utils.go index 8dd8c2fdf1f..e2134b18eb4 100644 --- a/pkg/apiserver/storage/testing/utils.go +++ b/pkg/apiserver/storage/testing/utils.go @@ -29,6 +29,10 @@ import ( grafanaregistry "github.com/grafana/grafana/pkg/apiserver/registry/generic" ) +var NewContext = func() context.Context { + return context.Background() +} + // KeyFunc is a function that generates keys for tests. // All tests use the "pods" resource, so the resource is hardcoded to "pods". var KeyFunc = func(namespace, name string) string { @@ -97,7 +101,7 @@ func testPropagateStore(ctx context.Context, t *testing.T, store storage.Interfa } setOutput := &example.Pod{} if err := store.Create(ctx, key, obj, setOutput, 0); err != nil { - t.Fatalf("Set failed: %v", err) + t.Fatalf("Create failed: %v", err) } return key, setOutput } diff --git a/pkg/apiserver/storage/testing/watcher_tests.go b/pkg/apiserver/storage/testing/watcher_tests.go index 97d8ca6d1e1..4bbcf1b2ce5 100644 --- a/pkg/apiserver/storage/testing/watcher_tests.go +++ b/pkg/apiserver/storage/testing/watcher_tests.go @@ -14,7 +14,6 @@ import ( "time" "github.com/stretchr/testify/require" - apiequality "k8s.io/apimachinery/pkg/api/equality" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -1053,7 +1052,7 @@ func RunTestWatchDispatchBookmarkEvents(ctx context.Context, t *testing.T, store t.Run(tt.name, func(t *testing.T) { pred := storage.Everything pred.AllowWatchBookmarks = tt.allowWatchBookmarks - ctx, cancel := context.WithTimeout(context.Background(), tt.timeout) + ctx, cancel := context.WithTimeout(NewContext(), tt.timeout) defer cancel() watcher, err := store.Watch(ctx, key, storage.ListOptions{ResourceVersion: startRV, Predicate: pred}) @@ -1112,7 +1111,7 @@ func RunTestOptionalWatchBookmarksWithCorrectResourceVersion(ctx context.Context pred := storage.Everything pred.AllowWatchBookmarks = true - ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + ctx, cancel := context.WithTimeout(NewContext(), 3*time.Second) defer cancel() watcher, err := store.Watch(ctx, key, storage.ListOptions{ResourceVersion: startRV, Predicate: pred, Recursive: true}) @@ -1462,7 +1461,7 @@ func RunWatchSemantics(ctx context.Context, t *testing.T, store storage.Interfac opts.ResourceVersion = scenario.resourceVersion } - w, err := store.Watch(context.Background(), KeyFunc(ns, ""), opts) + w, err := store.Watch(NewContext(), KeyFunc(ns, ""), opts) require.NoError(t, err, "failed to create watch: %v") defer w.Stop() @@ -1481,6 +1480,7 @@ func RunWatchSemantics(ctx context.Context, t *testing.T, store storage.Interfac createdPods = append(createdPods, out) } testCheckResultsInStrictOrder(t, w, scenario.expectedEventsAfterEstablishingWatch(createdPods)) + testCheckNoMoreResults(t, w) }) } @@ -1524,7 +1524,7 @@ func RunWatchSemanticInitialEventsExtended(ctx context.Context, t *testing.T, st opts.SendInitialEvents = &trueVal opts.Predicate.AllowWatchBookmarks = true - w, err := store.Watch(context.Background(), KeyFunc(ns, ""), opts) + w, err := store.Watch(NewContext(), KeyFunc(ns, ""), opts) require.NoError(t, err, "failed to create watch: %v") defer w.Stop() diff --git a/pkg/registry/apis/dashboard/legacy_storage.go b/pkg/registry/apis/dashboard/legacy_storage.go index 746d3fcc32d..48ff52c95e7 100644 --- a/pkg/registry/apis/dashboard/legacy_storage.go +++ b/pkg/registry/apis/dashboard/legacy_storage.go @@ -43,8 +43,8 @@ func (s *dashboardStorage) newStore(scheme *runtime.Scheme, defaultOptsGetter ge return nil, err } client := resource.NewLocalResourceStoreClient(server) - optsGetter := apistore.NewRESTOptionsGetter(client, - defaultOpts.StorageConfig.Codec, + optsGetter := apistore.NewRESTOptionsGetterForClient(client, + defaultOpts.StorageConfig.Config, ) strategy := grafanaregistry.NewStrategy(scheme) diff --git a/pkg/services/apiserver/aggregator/aggregator.go b/pkg/services/apiserver/aggregator/aggregator.go index d8998ddced8..b13b2ee6256 100644 --- a/pkg/services/apiserver/aggregator/aggregator.go +++ b/pkg/services/apiserver/aggregator/aggregator.go @@ -134,7 +134,7 @@ func CreateAggregatorConfig(commandOptions *options.Options, sharedConfig generi }, } - if err := commandOptions.AggregatorOptions.ApplyTo(aggregatorConfig, commandOptions.RecommendedOptions.Etcd, commandOptions.StorageOptions.DataPath); err != nil { + if err := commandOptions.AggregatorOptions.ApplyTo(aggregatorConfig, commandOptions.RecommendedOptions.Etcd); err != nil { return nil, err } diff --git a/pkg/services/apiserver/options/aggregator.go b/pkg/services/apiserver/options/aggregator.go index 7ff47ce76d6..24ee8196ad8 100644 --- a/pkg/services/apiserver/options/aggregator.go +++ b/pkg/services/apiserver/options/aggregator.go @@ -1,8 +1,6 @@ package options import ( - servicev0alpha1 "github.com/grafana/grafana/pkg/apis/service/v0alpha1" - filestorage "github.com/grafana/grafana/pkg/apiserver/storage/file" "github.com/spf13/pflag" v1 "k8s.io/api/apps/v1" "k8s.io/apimachinery/pkg/runtime" @@ -15,6 +13,9 @@ import ( apiregistrationv1beta1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1beta1" aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver" aggregatorscheme "k8s.io/kube-aggregator/pkg/apiserver/scheme" + + servicev0alpha1 "github.com/grafana/grafana/pkg/apis/service/v0alpha1" + "github.com/grafana/grafana/pkg/storage/unified/apistore" ) // AggregatorServerOptions contains the state for the aggregator apiserver @@ -51,7 +52,7 @@ func (o *AggregatorServerOptions) Validate() []error { return nil } -func (o *AggregatorServerOptions) ApplyTo(aggregatorConfig *aggregatorapiserver.Config, etcdOpts *options.EtcdOptions, dataPath string) error { +func (o *AggregatorServerOptions) ApplyTo(aggregatorConfig *aggregatorapiserver.Config, etcdOpts *options.EtcdOptions) error { genericConfig := aggregatorConfig.GenericConfig genericConfig.PostStartHooks = map[string]genericapiserver.PostStartHookConfigEntry{} @@ -79,11 +80,8 @@ func (o *AggregatorServerOptions) ApplyTo(aggregatorConfig *aggregatorapiserver. if err := etcdOptions.ApplyTo(&genericConfig.Config); err != nil { return err } - // override the RESTOptionsGetter to use the file storage options getter - restOptionsGetter, err := filestorage.NewRESTOptionsGetter(dataPath, etcdOptions.StorageConfig, - "/group/apiregistration.k8s.io/resource/apiservices", - "/group/service.grafana.app/resource/externalnames", - ) + // override the RESTOptionsGetter to use the in memory storage options + restOptionsGetter, err := apistore.NewRESTOptionsGetterMemory(etcdOptions.StorageConfig) if err != nil { return err } diff --git a/pkg/services/apiserver/service.go b/pkg/services/apiserver/service.go index 6a48d0820e0..7e456bd0a1d 100644 --- a/pkg/services/apiserver/service.go +++ b/pkg/services/apiserver/service.go @@ -24,7 +24,6 @@ import ( "github.com/grafana/grafana/pkg/api/routing" grafanaresponsewriter "github.com/grafana/grafana/pkg/apiserver/endpoints/responsewriter" - filestorage "github.com/grafana/grafana/pkg/apiserver/storage/file" "github.com/grafana/grafana/pkg/infra/appcontext" "github.com/grafana/grafana/pkg/infra/db" "github.com/grafana/grafana/pkg/infra/kvstore" @@ -297,8 +296,9 @@ func (s *service) start(ctx context.Context) error { if err != nil { return err } - serverConfig.Config.RESTOptionsGetter = apistore.NewRESTOptionsGetterForServer(server, - o.RecommendedOptions.Etcd.StorageConfig.Codec) + client := resource.NewLocalResourceStoreClient(server) + serverConfig.Config.RESTOptionsGetter = apistore.NewRESTOptionsGetterForClient(client, + o.RecommendedOptions.Etcd.StorageConfig) case grafanaapiserveroptions.StorageTypeUnifiedNextGrpc: if !s.features.IsEnabledGlobally(featuremgmt.FlagUnifiedStorage) { @@ -312,7 +312,7 @@ func (s *service) start(ctx context.Context) error { // Create a client instance client := resource.NewResourceStoreClientGRPC(conn) - serverConfig.Config.RESTOptionsGetter = apistore.NewRESTOptionsGetter(client, o.RecommendedOptions.Etcd.StorageConfig.Codec) + serverConfig.Config.RESTOptionsGetter = apistore.NewRESTOptionsGetterForClient(client, o.RecommendedOptions.Etcd.StorageConfig) case grafanaapiserveroptions.StorageTypeUnified, grafanaapiserveroptions.StorageTypeUnifiedGrpc: var client entity.EntityStoreClient @@ -347,8 +347,9 @@ func (s *service) start(ctx context.Context) error { if err != nil { return err } - serverConfig.Config.RESTOptionsGetter = apistore.NewRESTOptionsGetterForServer(server, - o.RecommendedOptions.Etcd.StorageConfig.Codec) + client := resource.NewLocalResourceStoreClient(server) + serverConfig.Config.RESTOptionsGetter = apistore.NewRESTOptionsGetterForClient(client, + o.RecommendedOptions.Etcd.StorageConfig) } else { serverConfig.Config.RESTOptionsGetter = entitystorage.NewRESTOptionsGetter(s.cfg, client, o.RecommendedOptions.Etcd.StorageConfig.Codec) @@ -357,7 +358,7 @@ func (s *service) start(ctx context.Context) error { case grafanaapiserveroptions.StorageTypeLegacy: fallthrough case grafanaapiserveroptions.StorageTypeFile: - restOptionsGetter, err := filestorage.NewRESTOptionsGetter(o.StorageOptions.DataPath, o.RecommendedOptions.Etcd.StorageConfig) + restOptionsGetter, err := apistore.NewRESTOptionsGetterForFile(o.StorageOptions.DataPath, o.RecommendedOptions.Etcd.StorageConfig) if err != nil { return err } diff --git a/pkg/storage/unified/apistore/prepare.go b/pkg/storage/unified/apistore/prepare.go index 931c6235251..0a8dc35377a 100644 --- a/pkg/storage/unified/apistore/prepare.go +++ b/pkg/storage/unified/apistore/prepare.go @@ -7,6 +7,7 @@ import ( "time" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apiserver/pkg/storage" "github.com/grafana/grafana/pkg/apimachinery/identity" "github.com/grafana/grafana/pkg/apimachinery/utils" @@ -26,6 +27,9 @@ func (s *Storage) prepareObjectForStorage(ctx context.Context, newObject runtime if obj.GetName() == "" { return nil, fmt.Errorf("new object must have a name") } + if obj.GetResourceVersion() != "" { + return nil, storage.ErrResourceVersionSetOnCreate + } obj.SetGenerateName("") // Clear the random name field obj.SetResourceVersion("") obj.SetSelfLink("") diff --git a/pkg/storage/unified/apistore/restoptions.go b/pkg/storage/unified/apistore/restoptions.go index 3346331094a..21a2f5c00fa 100644 --- a/pkg/storage/unified/apistore/restoptions.go +++ b/pkg/storage/unified/apistore/restoptions.go @@ -3,9 +3,13 @@ package apistore import ( - "path" + "context" + "os" + "path/filepath" "time" + "gocloud.dev/blob/fileblob" + "gocloud.dev/blob/memblob" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apiserver/pkg/registry/generic" @@ -21,43 +25,85 @@ import ( var _ generic.RESTOptionsGetter = (*RESTOptionsGetter)(nil) type RESTOptionsGetter struct { - client resource.ResourceStoreClient - Codec runtime.Codec + client resource.ResourceStoreClient + original storagebackend.Config } -func NewRESTOptionsGetterForServer(server resource.ResourceServer, codec runtime.Codec) *RESTOptionsGetter { +func NewRESTOptionsGetterForClient(client resource.ResourceStoreClient, original storagebackend.Config) *RESTOptionsGetter { return &RESTOptionsGetter{ - client: resource.NewLocalResourceStoreClient(server), - Codec: codec, + client: client, + original: original, } } -func NewRESTOptionsGetter(client resource.ResourceStoreClient, codec runtime.Codec) *RESTOptionsGetter { - return &RESTOptionsGetter{ - client: client, - Codec: codec, +func NewRESTOptionsGetterMemory(originalStorageConfig storagebackend.Config) (*RESTOptionsGetter, error) { + backend, err := resource.NewCDKBackend(context.Background(), resource.CDKBackendOptions{ + Bucket: memblob.OpenBucket(&memblob.Options{}), + }) + if err != nil { + return nil, err } + server, err := resource.NewResourceServer(resource.ResourceServerOptions{ + Backend: backend, + }) + if err != nil { + return nil, err + } + return NewRESTOptionsGetterForClient( + resource.NewLocalResourceStoreClient(server), + originalStorageConfig, + ), nil } -func (f *RESTOptionsGetter) GetRESTOptions(resource schema.GroupResource) (generic.RESTOptions, error) { +// Optionally, this constructor allows specifying directories +// for resources that are required to be read/watched on startup and there +// won't be any write operations that initially bootstrap their directories +func NewRESTOptionsGetterForFile(path string, + originalStorageConfig storagebackend.Config) (*RESTOptionsGetter, error) { + if path == "" { + path = filepath.Join(os.TempDir(), "grafana-apiserver") + } + + bucket, err := fileblob.OpenBucket(filepath.Join(path, "resource"), &fileblob.Options{ + CreateDir: true, + Metadata: fileblob.MetadataDontWrite, // skip + }) + if err != nil { + return nil, err + } + backend, err := resource.NewCDKBackend(context.Background(), resource.CDKBackendOptions{ + Bucket: bucket, + }) + if err != nil { + return nil, err + } + server, err := resource.NewResourceServer(resource.ResourceServerOptions{ + Backend: backend, + }) + if err != nil { + return nil, err + } + return NewRESTOptionsGetterForClient( + resource.NewLocalResourceStoreClient(server), + originalStorageConfig, + ), nil +} + +func (r *RESTOptionsGetter) GetRESTOptions(resource schema.GroupResource) (generic.RESTOptions, error) { storageConfig := &storagebackend.ConfigForResource{ Config: storagebackend.Config{ - Type: "custom", - Prefix: "", - Transport: storagebackend.TransportConfig{ - ServerList: []string{ - // ??? string(connectionInfo), - }, - }, - Codec: f.Codec, - EncodeVersioner: nil, - Transformer: nil, + Type: "resource", + Prefix: "resource/", // Not actually used + Transport: storagebackend.TransportConfig{}, + Codec: r.original.Codec, + EncodeVersioner: r.original.EncodeVersioner, + Transformer: r.original.Transformer, CompactionInterval: 0, CountMetricPollPeriod: 0, DBMetricPollInterval: 0, HealthcheckTimeout: 0, ReadycheckTimeout: 0, - StorageObjectCountTracker: nil, + StorageObjectCountTracker: flowcontrolrequest.NewStorageObjectCountTracker(), }, GroupResource: resource, } @@ -74,13 +120,14 @@ func (f *RESTOptionsGetter) GetRESTOptions(resource schema.GroupResource) (gener trigger storage.IndexerFuncs, indexers *cache.Indexers, ) (storage.Interface, factory.DestroyFunc, error) { - return NewStorage(config, resource, f.client, f.Codec, keyFunc, newFunc, newListFunc, getAttrsFunc) + return NewStorage(config, r.client, keyFunc, nil, newFunc, newListFunc, getAttrsFunc, trigger, indexers) }, - DeleteCollectionWorkers: 0, - EnableGarbageCollection: false, - ResourcePrefix: path.Join(storageConfig.Prefix, resource.Group, resource.Resource), + DeleteCollectionWorkers: 0, + EnableGarbageCollection: false, + // k8s expects forward slashes here, we'll convert them to os path separators in the storage + ResourcePrefix: "/group/" + resource.Group + "/resource/" + resource.Resource, CountMetricPollPeriod: 1 * time.Second, - StorageObjectCountTracker: flowcontrolrequest.NewStorageObjectCountTracker(), + StorageObjectCountTracker: storageConfig.Config.StorageObjectCountTracker, } return ret, nil diff --git a/pkg/storage/unified/apistore/storage.go b/pkg/storage/unified/apistore/storage.go deleted file mode 100644 index 30406ba9623..00000000000 --- a/pkg/storage/unified/apistore/storage.go +++ /dev/null @@ -1,537 +0,0 @@ -// SPDX-License-Identifier: AGPL-3.0-only -// Provenance-includes-location: https://github.com/kubernetes-sigs/apiserver-runtime/blob/main/pkg/experimental/storage/filepath/jsonfile_rest.go -// Provenance-includes-license: Apache-2.0 -// Provenance-includes-copyright: The Kubernetes Authors. - -package apistore - -import ( - "context" - "errors" - "fmt" - "io" - "reflect" - "strconv" - - apierrors "k8s.io/apimachinery/pkg/api/errors" - "k8s.io/apimachinery/pkg/api/meta" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/conversion" - "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/runtime/schema" - "k8s.io/apimachinery/pkg/types" - "k8s.io/apimachinery/pkg/watch" - "k8s.io/apiserver/pkg/storage" - "k8s.io/apiserver/pkg/storage/storagebackend" - "k8s.io/apiserver/pkg/storage/storagebackend/factory" - - "github.com/grafana/grafana/pkg/apimachinery/utils" - grafanaregistry "github.com/grafana/grafana/pkg/apiserver/registry/generic" - "github.com/grafana/grafana/pkg/storage/unified/resource" -) - -var _ storage.Interface = (*Storage)(nil) - -// Storage implements storage.Interface and stores resources in unified storage -type Storage struct { - config *storagebackend.ConfigForResource - store resource.ResourceStoreClient - gr schema.GroupResource - codec runtime.Codec - keyFunc func(obj runtime.Object) (string, error) - newFunc func() runtime.Object - newListFunc func() runtime.Object - getAttrsFunc storage.AttrFunc - // trigger storage.IndexerFuncs - // indexers *cache.Indexers -} - -func NewStorage( - config *storagebackend.ConfigForResource, - gr schema.GroupResource, - store resource.ResourceStoreClient, - codec runtime.Codec, - keyFunc func(obj runtime.Object) (string, error), - newFunc func() runtime.Object, - newListFunc func() runtime.Object, - getAttrsFunc storage.AttrFunc, -) (storage.Interface, factory.DestroyFunc, error) { - return &Storage{ - config: config, - gr: gr, - codec: codec, - store: store, - keyFunc: keyFunc, - newFunc: newFunc, - newListFunc: newListFunc, - getAttrsFunc: getAttrsFunc, - }, nil, nil -} - -func errorWrap(status *resource.ErrorResult) error { - if status != nil { - err := &apierrors.StatusError{ErrStatus: metav1.Status{ - Status: metav1.StatusFailure, - Code: status.Code, - Reason: metav1.StatusReason(status.Reason), - Message: status.Message, - }} - if status.Details != nil { - err.ErrStatus.Details = &metav1.StatusDetails{ - Group: status.Details.Group, - Kind: status.Details.Kind, - Name: status.Details.Name, - UID: types.UID(status.Details.Uid), - RetryAfterSeconds: status.Details.RetryAfterSeconds, - } - for _, c := range status.Details.Causes { - err.ErrStatus.Details.Causes = append(err.ErrStatus.Details.Causes, metav1.StatusCause{ - Type: metav1.CauseType(c.Reason), - Message: c.Message, - Field: c.Field, - }) - } - } - return err - } - return nil -} - -func getKey(val string) (*resource.ResourceKey, error) { - k, err := grafanaregistry.ParseKey(val) - if err != nil { - return nil, err - } - // if k.Group == "" { - // return nil, apierrors.NewInternalError(fmt.Errorf("missing group in request")) - // } - if k.Resource == "" { - return nil, apierrors.NewInternalError(fmt.Errorf("missing resource in request")) - } - return &resource.ResourceKey{ - Namespace: k.Namespace, - Group: k.Group, - Resource: k.Resource, - Name: k.Name, - }, err -} - -// Create adds a new object at a key unless it already exists. 'ttl' is time-to-live -// in seconds (0 means forever). If no error is returned and out is not nil, out will be -// set to the read value from database. -func (s *Storage) Create(ctx context.Context, key string, obj runtime.Object, out runtime.Object, ttl uint64) error { - k, err := getKey(key) - if err != nil { - return err - } - - value, err := s.prepareObjectForStorage(ctx, obj) - if err != nil { - return err - } - cmd := &resource.CreateRequest{ - Key: k, - Value: value, - } - - rsp, err := s.store.Create(ctx, cmd) - if err != nil { - return err - } - err = errorWrap(rsp.Error) - if err != nil { - return err - } - - if rsp.Error != nil { - return fmt.Errorf("error in status %+v", rsp.Error) - } - - // Decode into the result (can we just copy?) - _, _, err = s.codec.Decode(cmd.Value, nil, out) - if err != nil { - return err - } - after, err := utils.MetaAccessor(out) - if err != nil { - return err - } - after.SetResourceVersionInt64(rsp.ResourceVersion) - return nil -} - -// Delete removes the specified key and returns the value that existed at that spot. -// If key didn't exist, it will return NotFound storage error. -// If 'cachedExistingObject' is non-nil, it can be used as a suggestion about the -// current version of the object to avoid read operation from storage to get it. -// However, the implementations have to retry in case suggestion is stale. -func (s *Storage) Delete(ctx context.Context, key string, out runtime.Object, preconditions *storage.Preconditions, validateDeletion storage.ValidateObjectFunc, cachedExistingObject runtime.Object) error { - k, err := getKey(key) - if err != nil { - return err - } - - // if validateDeletion != nil { - // return fmt.Errorf("not supported (validate deletion)") - // } - - cmd := &resource.DeleteRequest{Key: k} - if preconditions != nil { - if preconditions.ResourceVersion != nil { - cmd.ResourceVersion, err = strconv.ParseInt(*preconditions.ResourceVersion, 10, 64) - if err != nil { - return err - } - } - if preconditions.UID != nil { - cmd.Uid = string(*preconditions.UID) - } - } - - rsp, err := s.store.Delete(ctx, cmd) - if err != nil { - return err - } - err = errorWrap(rsp.Error) - if err != nil { - return err - } - return nil -} - -// Watch begins watching the specified key. Events are decoded into API objects, -// and any items selected by 'p' are sent down to returned watch.Interface. -// resourceVersion may be used to specify what version to begin watching, -// which should be the current resourceVersion, and no longer rv+1 -// (e.g. reconnecting without missing any updates). -// If resource version is "0", this interface will get current object at given key -// and send it in an "ADDED" event, before watch starts. -func (s *Storage) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) { - listopts, _, err := toListRequest(key, opts) - if err != nil { - return nil, err - } - if listopts == nil { - return watch.NewEmptyWatch(), nil - } - - cmd := &resource.WatchRequest{ - Since: listopts.ResourceVersion, - Options: listopts.Options, - SendInitialEvents: false, - AllowWatchBookmarks: opts.Predicate.AllowWatchBookmarks, - } - if opts.SendInitialEvents != nil { - cmd.SendInitialEvents = *opts.SendInitialEvents - } - - client, err := s.store.Watch(ctx, cmd) - if err != nil { - // if the context was canceled, just return a new empty watch - if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, io.EOF) { - return watch.NewEmptyWatch(), nil - } - return nil, err - } - - reporter := apierrors.NewClientErrorReporter(500, "WATCH", "") - decoder := &streamDecoder{ - client: client, - newFunc: s.newFunc, - opts: opts, - codec: s.codec, - } - - return watch.NewStreamWatcher(decoder, reporter), nil -} - -// Get decodes object found at key into objPtr. On a not found error, will either -// return a zero object of the requested type, or an error, depending on 'opts.ignoreNotFound'. -// Treats empty responses and nil response nodes exactly like a not found error. -// The returned contents may be delayed, but it is guaranteed that they will -// match 'opts.ResourceVersion' according 'opts.ResourceVersionMatch'. -func (s *Storage) Get(ctx context.Context, key string, opts storage.GetOptions, objPtr runtime.Object) error { - var err error - req := &resource.ReadRequest{} - req.Key, err = getKey(key) - if err != nil { - return err - } - - if opts.ResourceVersion != "" { - req.ResourceVersion, err = strconv.ParseInt(opts.ResourceVersion, 10, 64) - if err != nil { - return err - } - } - - rsp, err := s.store.Read(ctx, req) - if err != nil { - return err - } - err = errorWrap(rsp.Error) - if err != nil { - return err - } - - _, _, err = s.codec.Decode(rsp.Value, &schema.GroupVersionKind{}, objPtr) - if err != nil { - return err - } - obj, err := utils.MetaAccessor(objPtr) - if err != nil { - return err - } - obj.SetResourceVersionInt64(rsp.ResourceVersion) - return nil -} - -func toListRequest(key string, opts storage.ListOptions) (*resource.ListRequest, storage.SelectionPredicate, error) { - predicate := opts.Predicate - k, err := getKey(key) - if err != nil { - return nil, predicate, err - } - req := &resource.ListRequest{ - Limit: opts.Predicate.Limit, - Options: &resource.ListOptions{ - Key: k, - }, - NextPageToken: predicate.Continue, - } - - if opts.Predicate.Label != nil && !opts.Predicate.Label.Empty() { - requirements, selectable := opts.Predicate.Label.Requirements() - if !selectable { - return nil, predicate, nil // not selectable - } - - for _, r := range requirements { - v := r.Key() - - req.Options.Labels = append(req.Options.Labels, &resource.Requirement{ - Key: v, - Operator: string(r.Operator()), - Values: r.Values().List(), - }) - } - } - - if opts.Predicate.Field != nil && !opts.Predicate.Field.Empty() { - requirements := opts.Predicate.Field.Requirements() - for _, r := range requirements { - requirement := &resource.Requirement{Key: r.Field, Operator: string(r.Operator)} - if r.Value != "" { - requirement.Values = append(requirement.Values, r.Value) - } - req.Options.Labels = append(req.Options.Labels, requirement) - } - } - - if opts.ResourceVersion != "" { - rv, err := strconv.ParseInt(opts.ResourceVersion, 10, 64) - if err != nil { - return nil, predicate, apierrors.NewBadRequest(fmt.Sprintf("invalid resource version: %s", opts.ResourceVersion)) - } - req.ResourceVersion = rv - } - - switch opts.ResourceVersionMatch { - case "", metav1.ResourceVersionMatchNotOlderThan: - req.VersionMatch = resource.ResourceVersionMatch_NotOlderThan - case metav1.ResourceVersionMatchExact: - req.VersionMatch = resource.ResourceVersionMatch_Exact - default: - return nil, predicate, apierrors.NewBadRequest( - fmt.Sprintf("unsupported version match: %v", opts.ResourceVersionMatch), - ) - } - - return req, predicate, nil -} - -// GetList unmarshalls objects found at key into a *List api object (an object -// that satisfies runtime.IsList definition). -// If 'opts.Recursive' is false, 'key' is used as an exact match. If `opts.Recursive' -// is true, 'key' is used as a prefix. -// The returned contents may be delayed, but it is guaranteed that they will -// match 'opts.ResourceVersion' according 'opts.ResourceVersionMatch'. -func (s *Storage) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error { - req, predicate, err := toListRequest(key, opts) - if err != nil { - return err - } - - rsp, err := s.store.List(ctx, req) - if err != nil { - return err - } - - listPtr, err := meta.GetItemsPtr(listObj) - if err != nil { - return err - } - v, err := conversion.EnforcePtr(listPtr) - if err != nil { - return err - } - - for _, item := range rsp.Items { - tmp := s.newFunc() - - tmp, _, err = s.codec.Decode(item.Value, nil, tmp) - if err != nil { - return err - } - obj, err := utils.MetaAccessor(tmp) - if err != nil { - return err - } - obj.SetResourceVersionInt64(item.ResourceVersion) - - // apply any predicates not handled in storage - matches, err := predicate.Matches(tmp) - if err != nil { - return apierrors.NewInternalError(err) - } - if !matches { - continue - } - - v.Set(reflect.Append(v, reflect.ValueOf(tmp).Elem())) - } - - listAccessor, err := meta.ListAccessor(listObj) - if err != nil { - return err - } - if rsp.NextPageToken != "" { - listAccessor.SetContinue(rsp.NextPageToken) - } - if rsp.RemainingItemCount > 0 { - listAccessor.SetRemainingItemCount(&rsp.RemainingItemCount) - } - if rsp.ResourceVersion > 0 { - listAccessor.SetResourceVersion(strconv.FormatInt(rsp.ResourceVersion, 10)) - } - return nil -} - -// GuaranteedUpdate keeps calling 'tryUpdate()' to update key 'key' (of type 'destination') -// retrying the update until success if there is index conflict. -// Note that object passed to tryUpdate may change across invocations of tryUpdate() if -// other writers are simultaneously updating it, so tryUpdate() needs to take into account -// the current contents of the object when deciding how the update object should look. -// If the key doesn't exist, it will return NotFound storage error if ignoreNotFound=false -// else `destination` will be set to the zero value of it's type. -// If the eventual successful invocation of `tryUpdate` returns an output with the same serialized -// contents as the input, it won't perform any update, but instead set `destination` to an object with those -// contents. -// If 'cachedExistingObject' is non-nil, it can be used as a suggestion about the -// current version of the object to avoid read operation from storage to get it. -// However, the implementations have to retry in case suggestion is stale. -func (s *Storage) GuaranteedUpdate( - ctx context.Context, - key string, - destination runtime.Object, - ignoreNotFound bool, - preconditions *storage.Preconditions, - tryUpdate storage.UpdateFunc, - cachedExistingObject runtime.Object, -) error { - k, err := getKey(key) - if err != nil { - return err - } - - // Get the current version - err = s.Get(ctx, key, storage.GetOptions{}, destination) - if err != nil { - if ignoreNotFound && apierrors.IsNotFound(err) { - // destination is already set to zero value - // we'll create the resource - } else { - return err - } - } - - accessor, err := utils.MetaAccessor(destination) - if err != nil { - return err - } - - // Early optimistic locking failure - previousVersion, _ := strconv.ParseInt(accessor.GetResourceVersion(), 10, 64) - if preconditions != nil { - if preconditions.ResourceVersion != nil { - rv, err := strconv.ParseInt(*preconditions.ResourceVersion, 10, 64) - if err != nil { - return err - } - if rv != previousVersion { - return fmt.Errorf("optimistic locking mismatch (previousVersion mismatch)") - } - } - - if preconditions.UID != nil { - if accessor.GetUID() != *preconditions.UID { - return fmt.Errorf("optimistic locking mismatch (UID mismatch)") - } - } - } - - res := &storage.ResponseMeta{} - updatedObj, _, err := tryUpdate(destination, *res) - if err != nil { - var statusErr *apierrors.StatusError - if errors.As(err, &statusErr) { - // For now, forbidden may come from a mutation handler - if statusErr.ErrStatus.Reason == metav1.StatusReasonForbidden { - return statusErr - } - } - return apierrors.NewInternalError( - fmt.Errorf("could not successfully update object. key=%s, err=%s", k.String(), err.Error()), - ) - } - - value, err := s.prepareObjectForUpdate(ctx, updatedObj, destination) - if err != nil { - return err - } - - req := &resource.UpdateRequest{Key: k, Value: value} - rsp, err := s.store.Update(ctx, req) - if err != nil { - return err - } - err = errorWrap(rsp.Error) - if err != nil { - return err - } - - // Decode into the response (can we just copy?) - _, _, err = s.codec.Decode(value, nil, destination) - if err != nil { - return err - } - accessor, err = utils.MetaAccessor(destination) - if err != nil { - return err - } - accessor.SetResourceVersionInt64(rsp.ResourceVersion) - return nil -} - -// Count returns number of different entries under the key (generally being path prefix). -func (s *Storage) Count(key string) (int64, error) { - return 0, nil -} - -func (s *Storage) Versioner() storage.Versioner { - return &storage.APIObjectVersioner{} -} - -func (s *Storage) RequestWatchProgress(ctx context.Context) error { - return nil -} diff --git a/pkg/apiserver/storage/file/file.go b/pkg/storage/unified/apistore/store.go similarity index 60% rename from pkg/apiserver/storage/file/file.go rename to pkg/storage/unified/apistore/store.go index e4526707922..f81f3b2e198 100644 --- a/pkg/apiserver/storage/file/file.go +++ b/pkg/storage/unified/apistore/store.go @@ -3,16 +3,16 @@ // Provenance-includes-license: Apache-2.0 // Provenance-includes-copyright: The Kubernetes Authors. -package file +package apistore import ( "context" "errors" "fmt" - "path/filepath" + "io" + "net/http" "reflect" - "strings" - "sync" + "strconv" "time" apierrors "k8s.io/apimachinery/pkg/api/errors" @@ -28,7 +28,9 @@ import ( "k8s.io/client-go/tools/cache" "k8s.io/klog/v2" + "github.com/grafana/grafana/pkg/apimachinery/utils" grafanaregistry "github.com/grafana/grafana/pkg/apiserver/registry/generic" + "github.com/grafana/grafana/pkg/storage/unified/resource" ) const MaxUpdateAttempts = 30 @@ -37,21 +39,17 @@ var _ storage.Interface = (*Storage)(nil) // Storage implements storage.Interface and storage resources as JSON files on disk. type Storage struct { - root string - resourcePrefix string - gr schema.GroupResource - codec runtime.Codec - keyFunc func(obj runtime.Object) (string, error) - newFunc func() runtime.Object - newListFunc func() runtime.Object - getAttrsFunc storage.AttrFunc - trigger storage.IndexerFuncs - indexers *cache.Indexers + gr schema.GroupResource + codec runtime.Codec + keyFunc func(obj runtime.Object) (string, error) + newFunc func() runtime.Object + newListFunc func() runtime.Object + getAttrsFunc storage.AttrFunc + trigger storage.IndexerFuncs + indexers *cache.Indexers - // rvMutex provides synchronization between Get and GetList+Watch+CUD methods - // with access to resource version generation for the latter group - rvMutex sync.RWMutex - currentRV uint64 + store resource.ResourceStoreClient + getKey func(string) (*resource.ResourceKey, error) watchSet *WatchSet versioner storage.Versioner @@ -66,78 +64,59 @@ var ErrNamespaceNotExists = errors.New("namespace does not exist") // NewStorage instantiates a new Storage. func NewStorage( config *storagebackend.ConfigForResource, - resourcePrefix string, + store resource.ResourceStoreClient, keyFunc func(obj runtime.Object) (string, error), + keyParser func(key string) (*resource.ResourceKey, error), newFunc func() runtime.Object, newListFunc func() runtime.Object, getAttrsFunc storage.AttrFunc, trigger storage.IndexerFuncs, indexers *cache.Indexers, ) (storage.Interface, factory.DestroyFunc, error) { - root := config.Prefix - if err := ensureDir(root); err != nil { - return nil, func() {}, fmt.Errorf("could not establish a writable directory at path=%s", root) - } - s := &Storage{ - root: root, - resourcePrefix: resourcePrefix, - gr: config.GroupResource, - codec: config.Codec, - keyFunc: keyFunc, - newFunc: newFunc, - newListFunc: newListFunc, - getAttrsFunc: getAttrsFunc, - trigger: trigger, - indexers: indexers, + store: store, + gr: config.GroupResource, + codec: config.Codec, + keyFunc: keyFunc, + newFunc: newFunc, + newListFunc: newListFunc, + getAttrsFunc: getAttrsFunc, + trigger: trigger, + indexers: indexers, watchSet: NewWatchSet(), + getKey: keyParser, versioner: &storage.APIObjectVersioner{}, } - // Initialize the RV stored in storage - s.getCurrentResourceVersion() + // The key parsing callback allows us to support the hardcoded paths from upstream tests + if s.getKey == nil { + s.getKey = func(key string) (*resource.ResourceKey, error) { + k, err := grafanaregistry.ParseKey(key) + if err != nil { + return nil, err + } + if k.Group == "" { + return nil, apierrors.NewInternalError(fmt.Errorf("missing group in request")) + } + if k.Resource == "" { + return nil, apierrors.NewInternalError(fmt.Errorf("missing resource in request")) + } + return &resource.ResourceKey{ + Namespace: k.Namespace, + Group: k.Group, + Resource: k.Resource, + Name: k.Name, + }, err + } + } return s, func() { s.watchSet.cleanupWatchers() }, nil } -func (s *Storage) getNewResourceVersion() uint64 { - s.currentRV += 1 - return s.currentRV -} - -func (s *Storage) getCurrentResourceVersion() uint64 { - if s.currentRV != 0 { - return s.currentRV - } - - objs, err := readDirRecursive(s.codec, s.root, s.newFunc) - if err != nil { - s.currentRV = 1 - return s.currentRV - } - - for _, obj := range objs { - currentVersion, err := s.versioner.ObjectResourceVersion(obj) - if err != nil && s.currentRV == 0 { - s.currentRV = 1 - continue - } - if currentVersion > s.currentRV { - s.currentRV = currentVersion - } - } - - if s.currentRV == 0 { - s.currentRV = 1 - } - - return s.currentRV -} - func (s *Storage) Versioner() storage.Versioner { return s.versioner } @@ -146,37 +125,36 @@ func (s *Storage) Versioner() storage.Versioner { // in seconds (0 means forever). If no error is returned and out is not nil, out will be // set to the read value from database. func (s *Storage) Create(ctx context.Context, key string, obj runtime.Object, out runtime.Object, ttl uint64) error { - s.rvMutex.Lock() - defer s.rvMutex.Unlock() - - fpath := s.filePath(key) - if exists(fpath) { - return storage.NewKeyExistsError(key, 0) - } - - dirname := filepath.Dir(fpath) - if err := ensureDir(dirname); err != nil { - return err - } - - generatedRV := s.getNewResourceVersion() - - metaObj, err := meta.Accessor(obj) + var err error + req := &resource.CreateRequest{} + req.Value, err = s.prepareObjectForStorage(ctx, obj) if err != nil { return err } - metaObj.SetSelfLink("") - if metaObj.GetResourceVersion() != "" { - return storage.ErrResourceVersionSetOnCreate + req.Key, err = s.getKey(key) + if err != nil { + return err + } + rsp, err := s.store.Create(ctx, req) + if err != nil { + return err + } + if rsp.Error != nil { + if rsp.Error.Code == http.StatusConflict { + return storage.NewKeyExistsError(key, 0) + } + return fmt.Errorf("other error %+v", rsp.Error) } - if err := s.versioner.UpdateObject(obj, generatedRV); err != nil { + if err := copyModifiedObjectToDestination(obj, out); err != nil { return err } - if err := writeFile(s.codec, fpath, obj); err != nil { + meta, err := utils.MetaAccessor(out) + if err != nil { return err } + meta.SetResourceVersionInt64(rsp.ResourceVersion) // set a timer to delete the file after ttl seconds if ttl > 0 { @@ -187,10 +165,6 @@ func (s *Storage) Create(ctx context.Context, key string, obj runtime.Object, ou }) } - if err := s.Get(ctx, key, storage.GetOptions{}, out); err != nil { - return err - } - s.watchSet.notifyWatchers(watch.Event{ Object: out.DeepCopyObject(), Type: watch.Added, @@ -212,30 +186,45 @@ func (s *Storage) Delete( validateDeletion storage.ValidateObjectFunc, _ runtime.Object, ) error { - s.rvMutex.Lock() - defer s.rvMutex.Unlock() - - fpath := s.filePath(key) if err := s.Get(ctx, key, storage.GetOptions{}, out); err != nil { return err } + k, err := s.getKey(key) + if err != nil { + return err + } + cmd := &resource.DeleteRequest{Key: k} + if preconditions != nil { if err := preconditions.Check(key, out); err != nil { return err } + + if preconditions.ResourceVersion != nil { + cmd.ResourceVersion, err = strconv.ParseInt(*preconditions.ResourceVersion, 10, 64) + if err != nil { + return err + } + } + if preconditions.UID != nil { + cmd.Uid = string(*preconditions.UID) + } } - generatedRV := s.getNewResourceVersion() - if err := s.versioner.UpdateObject(out, generatedRV); err != nil { - return err - } - + // ?? this was after delete before if err := validateDeletion(ctx, out); err != nil { return err } - - if err := deleteFile(fpath); err != nil { + rsp, err := s.store.Delete(ctx, cmd) + if err != nil { + return err + } + err = errorWrap(rsp.Error) + if err != nil { + return err + } + if err := s.versioner.UpdateObject(out, uint64(rsp.ResourceVersion)); err != nil { return err } @@ -246,6 +235,48 @@ func (s *Storage) Delete( return nil } +// This version is not yet passing the watch tests +func (s *Storage) WatchNEXT(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) { + k, err := s.getKey(key) + if err != nil { + return watch.NewEmptyWatch(), nil + } + + req, predicate, err := toListRequest(k, opts) + if err != nil { + return watch.NewEmptyWatch(), nil + } + + cmd := &resource.WatchRequest{ + Since: req.ResourceVersion, + Options: req.Options, + SendInitialEvents: false, + AllowWatchBookmarks: opts.Predicate.AllowWatchBookmarks, + } + if opts.SendInitialEvents != nil { + cmd.SendInitialEvents = *opts.SendInitialEvents + } + + client, err := s.store.Watch(ctx, cmd) + if err != nil { + // if the context was canceled, just return a new empty watch + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, io.EOF) { + return watch.NewEmptyWatch(), nil + } + return nil, err + } + + reporter := apierrors.NewClientErrorReporter(500, "WATCH", "") + decoder := &streamDecoder{ + client: client, + newFunc: s.newFunc, + predicate: predicate, + codec: s.codec, + } + + return watch.NewStreamWatcher(decoder, reporter), nil +} + // Watch begins watching the specified key. Events are decoded into API objects, // and any items selected by the predicate are sent down to returned watch.Interface. // resourceVersion may be used to specify what version to begin watching, @@ -254,26 +285,28 @@ func (s *Storage) Delete( // If resource version is "0", this interface will get current object at given key // and send it in an "ADDED" event, before watch starts. func (s *Storage) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) { - p := opts.Predicate + k, err := s.getKey(key) + if err != nil { + return watch.NewEmptyWatch(), nil + } + + req, predicate, err := toListRequest(k, opts) + if err != nil { + return watch.NewEmptyWatch(), nil + } + listObj := s.newListFunc() - // Parses to 0 for opts.ResourceVersion == 0 - requestedRV, err := s.versioner.ParseResourceVersion(opts.ResourceVersion) - if err != nil { - return nil, apierrors.NewBadRequest(fmt.Sprintf("invalid resource version: %v", err)) - } - - parsedkey, err := grafanaregistry.ParseKey(key) - if err != nil { - return nil, err - } - var namespace *string - if parsedkey.Namespace != "" { - namespace = &parsedkey.Namespace + if k.Namespace != "" { + namespace = &k.Namespace } - if (opts.SendInitialEvents == nil && requestedRV == 0) || (opts.SendInitialEvents != nil && *opts.SendInitialEvents) { + if ctx.Err() != nil { + return watch.NewEmptyWatch(), nil + } + + if (opts.SendInitialEvents == nil && req.ResourceVersion == 0) || (opts.SendInitialEvents != nil && *opts.SendInitialEvents) { if err := s.GetList(ctx, key, opts, listObj); err != nil { return nil, err } @@ -290,7 +323,7 @@ func (s *Storage) Watch(ctx context.Context, key string, opts storage.ListOption return nil, err } - jw := s.watchSet.newWatch(ctx, maybeUpdatedRV, p, s.versioner, namespace) + jw := s.watchSet.newWatch(ctx, maybeUpdatedRV, predicate, s.versioner, namespace) initEvents := make([]watch.Event, 0) listPtr, err := meta.GetItemsPtr(listObj) @@ -314,7 +347,7 @@ func (s *Storage) Watch(ctx context.Context, key string, opts storage.ListOption }) } - if p.AllowWatchBookmarks && len(initEvents) > 0 { + if predicate.AllowWatchBookmarks && len(initEvents) > 0 { listRV, err := s.versioner.ParseResourceVersion(listAccessor.GetResourceVersion()) if err != nil { return nil, fmt.Errorf("could not get last init event's revision for bookmark: %v", err) @@ -341,13 +374,23 @@ func (s *Storage) Watch(ctx context.Context, key string, opts storage.ListOption return jw, nil } - maybeUpdatedRV := requestedRV + maybeUpdatedRV := uint64(req.ResourceVersion) if maybeUpdatedRV == 0 { - s.rvMutex.RLock() - maybeUpdatedRV = s.getCurrentResourceVersion() - s.rvMutex.RUnlock() + rsp, err := s.store.List(ctx, &resource.ListRequest{ + Options: &resource.ListOptions{ + Key: k, + }, + Limit: 1, // we ignore the results, just look at the RV + }) + if err != nil { + return nil, err + } + maybeUpdatedRV = uint64(rsp.ResourceVersion) + if maybeUpdatedRV < 1 { + return nil, fmt.Errorf("expecting a non-zero resource version") + } } - jw := s.watchSet.newWatch(ctx, maybeUpdatedRV, p, s.versioner, namespace) + jw := s.watchSet.newWatch(ctx, maybeUpdatedRV, predicate, s.versioner, namespace) jw.Start() return jw, nil @@ -359,39 +402,42 @@ func (s *Storage) Watch(ctx context.Context, key string, opts storage.ListOption // The returned contents may be delayed, but it is guaranteed that they will // match 'opts.ResourceVersion' according 'opts.ResourceVersionMatch'. func (s *Storage) Get(ctx context.Context, key string, opts storage.GetOptions, objPtr runtime.Object) error { - // No RV generation locking in single item get since its read from the disk - fpath := s.filePath(key) - - rv, err := s.versioner.ParseResourceVersion(opts.ResourceVersion) - if err != nil { - return err - } - - // Since it's a get, check if the dir exists and return early as needed - dirname := filepath.Dir(fpath) - if !exists(dirname) { - return storage.NewKeyNotFoundError(key, int64(rv)) - } - - obj, err := readFile(s.codec, fpath, func() runtime.Object { - return s.newFunc() - }) + var err error + req := &resource.ReadRequest{} + req.Key, err = s.getKey(key) if err != nil { if opts.IgnoreNotFound { return runtime.SetZeroValue(objPtr) } - return storage.NewKeyNotFoundError(key, int64(rv)) + return storage.NewKeyNotFoundError(key, 0) } - if err := copyModifiedObjectToDestination(obj, objPtr); err != nil { + if opts.ResourceVersion != "" { + req.ResourceVersion, err = strconv.ParseInt(opts.ResourceVersion, 10, 64) + if err != nil { + return err + } + } + + rsp, err := s.store.Read(ctx, req) + if err != nil { return err } - - if err = s.validateMinimumResourceVersion(opts.ResourceVersion, s.getCurrentResourceVersion()); err != nil { - return err + if rsp.Error != nil { + if rsp.Error.Code == http.StatusNotFound { + if opts.IgnoreNotFound { + return runtime.SetZeroValue(objPtr) + } + return storage.NewKeyNotFoundError(key, req.ResourceVersion) + } + return errorWrap(rsp.Error) } - return nil + _, _, err = s.codec.Decode(rsp.Value, nil, objPtr) + if err != nil { + return err + } + return s.versioner.UpdateObject(objPtr, uint64(rsp.ResourceVersion)) } // GetList unmarshalls objects found at key into a *List api object (an object @@ -401,51 +447,22 @@ func (s *Storage) Get(ctx context.Context, key string, opts storage.GetOptions, // The returned contents may be delayed, but it is guaranteed that they will // match 'opts.ResourceVersion' according 'opts.ResourceVersionMatch'. func (s *Storage) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error { - resourceVersionInt := uint64(0) - - // read state protected by mutex - objs, err := func() ([]runtime.Object, error) { - s.rvMutex.Lock() - defer s.rvMutex.Unlock() - - resourceVersionInt = s.getCurrentResourceVersion() - - var fpath string - dirpath := s.dirPath(key) - // Since it's a get, check if the dir exists and return early as needed - if !exists(dirpath) { - // We may have gotten the key targeting an individual item - dirpath = filepath.Dir(dirpath) - if !exists(dirpath) { - // ensure we return empty list in listObj instead of a not found error - return []runtime.Object{}, nil - } - fpath = s.filePath(key) - } - - var objs []runtime.Object - if fpath != "" { - obj, err := readFile(s.codec, fpath, func() runtime.Object { - return s.newFunc() - }) - if err == nil { - objs = append(objs, obj) - } - } else { - var err error - objs, err = readDirRecursive(s.codec, dirpath, s.newFunc) - if err != nil { - return nil, err - } - } - return objs, nil - }() - + k, err := s.getKey(key) if err != nil { return err } - if err := s.validateMinimumResourceVersion(opts.ResourceVersion, resourceVersionInt); err != nil { + req, predicate, err := toListRequest(k, opts) + if err != nil { + return err + } + + rsp, err := s.store.List(ctx, req) + if err != nil { + return err + } + + if err := s.validateMinimumResourceVersion(opts.ResourceVersion, uint64(rsp.ResourceVersion)); err != nil { return err } @@ -462,8 +479,15 @@ func (s *Storage) GetList(ctx context.Context, key string, opts storage.ListOpti v.Set(reflect.MakeSlice(v.Type(), 0, 0)) } - remainingItems := (*int64)(nil) - for i, obj := range objs { + for _, item := range rsp.Items { + obj, _, err := s.codec.Decode(item.Value, nil, s.newFunc()) + if err != nil { + return err + } + if err := s.versioner.UpdateObject(obj, uint64(item.ResourceVersion)); err != nil { + return err + } + if opts.ResourceVersionMatch == metaV1.ResourceVersionMatchExact { currentVersion, err := s.versioner.ObjectResourceVersion(obj) if err != nil { @@ -478,22 +502,19 @@ func (s *Storage) GetList(ctx context.Context, key string, opts storage.ListOpti } } - ok, err := opts.Predicate.Matches(obj) + ok, err := predicate.Matches(obj) if err == nil && ok { v.Set(reflect.Append(v, reflect.ValueOf(obj).Elem())) } - - if int64(v.Len()) >= opts.Predicate.Limit && opts.Predicate.Limit > 0 { - remaining := int64(len(objs) - i - 1) - remainingItems = &remaining - break - } } - if err := s.versioner.UpdateList(listObj, resourceVersionInt, "", remainingItems); err != nil { + var remainingItems *int64 + if rsp.RemainingItemCount > 0 { + remainingItems = &rsp.RemainingItemCount + } + if err := s.versioner.UpdateList(listObj, uint64(rsp.ResourceVersion), rsp.NextPageToken, remainingItems); err != nil { return err } - return nil } @@ -510,6 +531,7 @@ func (s *Storage) GetList(ctx context.Context, key string, opts storage.ListOpti // If 'cachedExistingObject' is non-nil, it can be used as a suggestion about the // current version of the object to avoid read operation from storage to get it. // However, the implementations have to retry in case suggestion is stale. +// nolint:gocyclo func (s *Storage) GuaranteedUpdate( ctx context.Context, key string, @@ -522,40 +544,64 @@ func (s *Storage) GuaranteedUpdate( var ( res storage.ResponseMeta updatedObj runtime.Object - objFromDisk runtime.Object + existingObj runtime.Object created bool - fpath = s.filePath(key) - dirpath = filepath.Dir(fpath) + err error ) + req := &resource.UpdateRequest{} + req.Key, err = s.getKey(key) + if err != nil { + return err + } + if preconditions != nil && preconditions.ResourceVersion != nil { + req.ResourceVersion, err = strconv.ParseInt(*preconditions.ResourceVersion, 10, 64) + if err != nil { + return err + } + } for attempt := 1; attempt <= MaxUpdateAttempts; attempt = attempt + 1 { - var err error + // Read the latest value + rsp, err := s.store.Read(ctx, &resource.ReadRequest{Key: req.Key}) + if err != nil { + return err + } - if !exists(dirpath) { - if err := ensureDir(dirpath); err != nil { + if rsp.Error != nil { + if rsp.Error.Code == http.StatusNotFound { + if !ignoreNotFound { + return apierrors.NewNotFound(s.gr, req.Key.Name) + } + } else { + return fmt.Errorf("read error %+v", rsp.Error) + } + } + + created = true + existingObj = s.newFunc() + if len(rsp.Value) > 0 { + created = false + _, _, err = s.codec.Decode(rsp.Value, nil, existingObj) + if err != nil { return err } - } - - if !exists(fpath) && !ignoreNotFound { - return apierrors.NewNotFound(s.gr, s.nameFromKey(key)) - } - - objFromDisk, err = readFile(s.codec, fpath, s.newFunc) - if err != nil { - // fallback to new object if the file is not found - objFromDisk = s.newFunc() - created = true - } - - if err := preconditions.Check(key, objFromDisk); err != nil { - if attempt >= MaxUpdateAttempts { - return fmt.Errorf("precondition failed: %w", err) + mmm, err := utils.MetaAccessor(existingObj) + if err != nil { + return err } - continue + mmm.SetResourceVersionInt64(rsp.ResourceVersion) + + if err := preconditions.Check(key, existingObj); err != nil { + if attempt >= MaxUpdateAttempts { + return fmt.Errorf("precondition failed: %w", err) + } + continue + } + } else if !ignoreNotFound { + return apierrors.NewNotFound(s.gr, req.Key.Name) } - updatedObj, _, err = tryUpdate(objFromDisk, res) + updatedObj, _, err = tryUpdate(existingObj, res) if err != nil { if attempt >= MaxUpdateAttempts { return err @@ -565,7 +611,7 @@ func (s *Storage) GuaranteedUpdate( break } - unchanged, err := isUnchanged(s.codec, objFromDisk, updatedObj) + unchanged, err := isUnchanged(s.codec, existingObj, updatedObj) if err != nil { return err } @@ -577,15 +623,39 @@ func (s *Storage) GuaranteedUpdate( return nil } - s.rvMutex.Lock() - generatedRV := s.getNewResourceVersion() - s.rvMutex.Unlock() - - if err := s.versioner.UpdateObject(updatedObj, generatedRV); err != nil { - return err + rv := int64(0) + if created { + value, err := s.prepareObjectForStorage(ctx, updatedObj) + if err != nil { + return err + } + rsp2, err := s.store.Create(ctx, &resource.CreateRequest{ + Key: req.Key, + Value: value, + }) + if err != nil { + return err + } + if rsp2.Error != nil { + return fmt.Errorf("backend update error: %+v", rsp2.Error) + } + rv = rsp2.ResourceVersion + } else { + req.Value, err = s.prepareObjectForUpdate(ctx, updatedObj, existingObj) + if err != nil { + return err + } + rsp2, err := s.store.Update(ctx, req) + if err != nil { + return err + } + if rsp2.Error != nil { + return fmt.Errorf("backend update error: %+v", rsp2.Error) + } + rv = rsp2.ResourceVersion } - if err := writeFile(s.codec, fpath, updatedObj); err != nil { + if err := s.versioner.UpdateObject(updatedObj, uint64(rv)); err != nil { return err } @@ -593,21 +663,17 @@ func (s *Storage) GuaranteedUpdate( return err } - eventType := watch.Modified if created { - eventType = watch.Added s.watchSet.notifyWatchers(watch.Event{ Object: destination.DeepCopyObject(), - Type: eventType, + Type: watch.Added, }, nil) - return nil + } else { + s.watchSet.notifyWatchers(watch.Event{ + Object: destination.DeepCopyObject(), + Type: watch.Modified, + }, existingObj.DeepCopyObject()) } - - s.watchSet.notifyWatchers(watch.Event{ - Object: destination.DeepCopyObject(), - Type: eventType, - }, objFromDisk.DeepCopyObject()) - return nil } @@ -652,10 +718,6 @@ func (s *Storage) validateMinimumResourceVersion(minimumResourceVersion string, return nil } -func (s *Storage) nameFromKey(key string) string { - return strings.Replace(key, s.resourcePrefix+"/", "", 1) -} - func copyModifiedObjectToDestination(updatedObj runtime.Object, destination runtime.Object) error { u, err := conversion.EnforcePtr(updatedObj) if err != nil { diff --git a/pkg/apiserver/storage/file/file_test.go b/pkg/storage/unified/apistore/store_test.go similarity index 95% rename from pkg/apiserver/storage/file/file_test.go rename to pkg/storage/unified/apistore/store_test.go index 8204fea0b1d..5c9c12745bf 100644 --- a/pkg/apiserver/storage/file/file_test.go +++ b/pkg/storage/unified/apistore/store_test.go @@ -3,7 +3,7 @@ // Provenance-includes-license: Apache-2.0 // Provenance-includes-copyright: The Kubernetes Authors. -package file +package apistore import ( "context" @@ -20,6 +20,7 @@ import ( examplev1 "k8s.io/apiserver/pkg/apis/example/v1" "k8s.io/apiserver/pkg/storage" + "github.com/grafana/grafana/pkg/apimachinery/identity" storagetesting "github.com/grafana/grafana/pkg/apiserver/storage/testing" ) @@ -27,6 +28,19 @@ func init() { metav1.AddToGroupVersion(scheme, metav1.SchemeGroupVersion) utilruntime.Must(example.AddToScheme(scheme)) utilruntime.Must(examplev1.AddToScheme(scheme)) + + // Make sure there is a user in every context + storagetesting.NewContext = func() context.Context { + testUserA := &identity.StaticRequester{ + Namespace: identity.NamespaceUser, + Login: "testuser", + UserID: 123, + UserUID: "u123", + OrgRole: identity.RoleAdmin, + IsGrafanaAdmin: true, // can do anything + } + return identity.WithRequester(context.Background(), testUserA) + } } // GetPodAttrs returns labels and fields of a given object for filtering purposes. diff --git a/pkg/storage/unified/apistore/stream.go b/pkg/storage/unified/apistore/stream.go index e5c66b59b22..a425279185a 100644 --- a/pkg/storage/unified/apistore/stream.go +++ b/pkg/storage/unified/apistore/stream.go @@ -17,10 +17,10 @@ import ( ) type streamDecoder struct { - client resource.ResourceStore_WatchClient - newFunc func() runtime.Object - opts storage.ListOptions - codec runtime.Codec + client resource.ResourceStore_WatchClient + newFunc func() runtime.Object + predicate storage.SelectionPredicate + codec runtime.Codec } func (d *streamDecoder) toObject(w *resource.WatchEvent_Resource) (runtime.Object, error) { @@ -95,7 +95,7 @@ decode: switch evt.Type { case resource.WatchEvent_ADDED: // apply any predicates not handled in storage - matches, err := d.opts.Predicate.Matches(obj) + matches, err := d.predicate.Matches(obj) if err != nil { klog.Errorf("error matching object: %s", err) return watch.Error, nil, err @@ -109,7 +109,7 @@ decode: watchAction = watch.Modified // apply any predicates not handled in storage - matches, err := d.opts.Predicate.Matches(obj) + matches, err := d.predicate.Matches(obj) if err != nil { klog.Errorf("error matching object: %s", err) return watch.Error, nil, err @@ -126,7 +126,7 @@ decode: } // apply any predicates not handled in storage - prevMatches, err = d.opts.Predicate.Matches(prevObj) + prevMatches, err = d.predicate.Matches(prevObj) if err != nil { klog.Errorf("error matching object: %s", err) return watch.Error, nil, err @@ -177,7 +177,7 @@ decode: } // apply any predicates not handled in storage - matches, err := d.opts.Predicate.Matches(obj) + matches, err := d.predicate.Matches(obj) if err != nil { klog.Errorf("error matching object: %s", err) return watch.Error, nil, err diff --git a/pkg/storage/unified/apistore/util.go b/pkg/storage/unified/apistore/util.go new file mode 100644 index 00000000000..90eb40412e8 --- /dev/null +++ b/pkg/storage/unified/apistore/util.go @@ -0,0 +1,163 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Provenance-includes-location: https://github.com/kubernetes-sigs/apiserver-runtime/blob/main/pkg/experimental/storage/filepath/jsonfile_rest.go +// Provenance-includes-license: Apache-2.0 +// Provenance-includes-copyright: The Kubernetes Authors. + +package apistore + +import ( + "bytes" + "fmt" + "strconv" + "strings" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apiserver/pkg/storage" + + grafanaregistry "github.com/grafana/grafana/pkg/apiserver/registry/generic" + "github.com/grafana/grafana/pkg/storage/unified/resource" +) + +func errorWrap(status *resource.ErrorResult) error { + if status != nil { + err := &apierrors.StatusError{ErrStatus: metav1.Status{ + Status: metav1.StatusFailure, + Code: status.Code, + Reason: metav1.StatusReason(status.Reason), + Message: status.Message, + }} + if status.Details != nil { + err.ErrStatus.Details = &metav1.StatusDetails{ + Group: status.Details.Group, + Kind: status.Details.Kind, + Name: status.Details.Name, + UID: types.UID(status.Details.Uid), + RetryAfterSeconds: status.Details.RetryAfterSeconds, + } + for _, c := range status.Details.Causes { + err.ErrStatus.Details.Causes = append(err.ErrStatus.Details.Causes, metav1.StatusCause{ + Type: metav1.CauseType(c.Reason), + Message: c.Message, + Field: c.Field, + }) + } + } + return err + } + return nil +} + +func toListRequest(k *resource.ResourceKey, opts storage.ListOptions) (*resource.ListRequest, storage.SelectionPredicate, error) { + predicate := opts.Predicate + req := &resource.ListRequest{ + Limit: opts.Predicate.Limit, + Options: &resource.ListOptions{ + Key: k, + }, + NextPageToken: predicate.Continue, + } + + if opts.Predicate.Label != nil && !opts.Predicate.Label.Empty() { + requirements, selectable := opts.Predicate.Label.Requirements() + if !selectable { + return nil, predicate, nil // not selectable + } + + for _, r := range requirements { + v := r.Key() + + req.Options.Labels = append(req.Options.Labels, &resource.Requirement{ + Key: v, + Operator: string(r.Operator()), + Values: r.Values().List(), + }) + } + } + + if opts.Predicate.Field != nil && !opts.Predicate.Field.Empty() { + requirements := opts.Predicate.Field.Requirements() + for _, r := range requirements { + requirement := &resource.Requirement{Key: r.Field, Operator: string(r.Operator)} + if r.Value != "" { + requirement.Values = append(requirement.Values, r.Value) + } + req.Options.Labels = append(req.Options.Labels, requirement) + } + } + + if opts.ResourceVersion != "" { + rv, err := strconv.ParseInt(opts.ResourceVersion, 10, 64) + if err != nil { + return nil, predicate, apierrors.NewBadRequest(fmt.Sprintf("invalid resource version: %s", opts.ResourceVersion)) + } + req.ResourceVersion = rv + } + + switch opts.ResourceVersionMatch { + case "", metav1.ResourceVersionMatchNotOlderThan: + req.VersionMatch = resource.ResourceVersionMatch_NotOlderThan + case metav1.ResourceVersionMatchExact: + req.VersionMatch = resource.ResourceVersionMatch_Exact + default: + return nil, predicate, apierrors.NewBadRequest( + fmt.Sprintf("unsupported version match: %v", opts.ResourceVersionMatch), + ) + } + + return req, predicate, nil +} + +func isUnchanged(codec runtime.Codec, obj runtime.Object, newObj runtime.Object) (bool, error) { + buf := new(bytes.Buffer) + if err := codec.Encode(obj, buf); err != nil { + return false, err + } + + newBuf := new(bytes.Buffer) + if err := codec.Encode(newObj, newBuf); err != nil { + return false, err + } + + return bytes.Equal(buf.Bytes(), newBuf.Bytes()), nil +} + +func testKeyParser(val string) (*resource.ResourceKey, error) { + k, err := grafanaregistry.ParseKey(val) + if err != nil { + if strings.HasPrefix(val, "pods/") { + parts := strings.Split(val, "/") + if len(parts) == 2 { + err = nil + k = &grafanaregistry.Key{ + Resource: parts[0], // pods + Name: parts[1], + } + } else if len(parts) == 3 { + err = nil + k = &grafanaregistry.Key{ + Resource: parts[0], // pods + Namespace: parts[1], + Name: parts[2], + } + } + } + } + if err != nil { + return nil, err + } + if k.Group == "" { + k.Group = "example.apiserver.k8s.io" + } + if k.Resource == "" { + return nil, apierrors.NewInternalError(fmt.Errorf("missing resource in request")) + } + return &resource.ResourceKey{ + Namespace: k.Namespace, + Group: k.Group, + Resource: k.Resource, + Name: k.Name, + }, err +} diff --git a/pkg/apiserver/storage/file/watcher_test.go b/pkg/storage/unified/apistore/watcher_test.go similarity index 86% rename from pkg/apiserver/storage/file/watcher_test.go rename to pkg/storage/unified/apistore/watcher_test.go index 8651bc24244..38510a088b3 100644 --- a/pkg/apiserver/storage/file/watcher_test.go +++ b/pkg/storage/unified/apistore/watcher_test.go @@ -3,16 +3,18 @@ // Provenance-includes-license: Apache-2.0 // Provenance-includes-copyright: The Kubernetes Authors. -package file +package apistore import ( "context" - "io/fs" + "fmt" "os" - "path" "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gocloud.dev/blob/fileblob" + "gocloud.dev/blob/memblob" "k8s.io/apimachinery/pkg/api/apitesting" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -27,6 +29,7 @@ import ( "k8s.io/apiserver/pkg/storage/storagebackend/factory" storagetesting "github.com/grafana/grafana/pkg/apiserver/storage/testing" + "github.com/grafana/grafana/pkg/storage/unified/resource" ) var scheme = runtime.NewScheme() @@ -67,10 +70,39 @@ func testSetup(t testing.TB, opts ...setupOption) (context.Context, storage.Inte opt(&setupOpts, t) } + setupOpts.groupResource = schema.GroupResource{ + Group: "example.apiserver.k8s.io", + Resource: "pods", + } + + bucket := memblob.OpenBucket(nil) + if true { + tmp, err := os.MkdirTemp("", "xxx-*") + require.NoError(t, err) + + bucket, err = fileblob.OpenBucket(tmp, &fileblob.Options{ + CreateDir: true, + Metadata: fileblob.MetadataDontWrite, // skip + }) + require.NoError(t, err) + fmt.Printf("ROOT: %s\n\n", tmp) + } + ctx := storagetesting.NewContext() + backend, err := resource.NewCDKBackend(ctx, resource.CDKBackendOptions{ + Bucket: bucket, + }) + require.NoError(t, err) + + server, err := resource.NewResourceServer(resource.ResourceServerOptions{ + Backend: backend, + }) + require.NoError(t, err) + client := resource.NewLocalResourceStoreClient(server) + config := storagebackend.NewDefaultConfig(setupOpts.prefix, setupOpts.codec) store, destroyFunc, err := NewStorage( config.ForResource(setupOpts.groupResource), - setupOpts.resourcePrefix, + client, func(obj runtime.Object) (string, error) { accessor, err := meta.Accessor(obj) if err != nil { @@ -78,22 +110,16 @@ func testSetup(t testing.TB, opts ...setupOption) (context.Context, storage.Inte } return storagetesting.KeyFunc(accessor.GetNamespace(), accessor.GetName()), nil }, + testKeyParser, // will fallback to hardcoded /pods/... keys setupOpts.newFunc, setupOpts.newListFunc, storage.DefaultNamespaceScopedAttr, make(map[string]storage.IndexerFunc, 0), nil, ) - - // Some tests may start reading before writing - if err := os.MkdirAll(path.Join(setupOpts.prefix, storagetesting.KeyFunc("test-ns", "")), fs.ModePerm); err != nil { - return nil, nil, nil, err - } - if err != nil { return nil, nil, nil, err } - ctx := context.Background() return ctx, store, destroyFunc, nil } diff --git a/pkg/apiserver/storage/file/watchset.go b/pkg/storage/unified/apistore/watchset.go similarity index 99% rename from pkg/apiserver/storage/file/watchset.go rename to pkg/storage/unified/apistore/watchset.go index 98c028727e0..9c9d214b4b6 100644 --- a/pkg/apiserver/storage/file/watchset.go +++ b/pkg/storage/unified/apistore/watchset.go @@ -3,7 +3,7 @@ // Provenance-includes-license: Apache-2.0 // Provenance-includes-copyright: The Kubernetes Authors. -package file +package apistore import ( "context" diff --git a/pkg/storage/unified/resource/validation.go b/pkg/storage/unified/resource/validation.go index 746bea81ded..edb41ae745b 100644 --- a/pkg/storage/unified/resource/validation.go +++ b/pkg/storage/unified/resource/validation.go @@ -5,7 +5,7 @@ import ( "regexp" ) -var validNameCharPattern = `a-zA-Z0-9\-\_` +var validNameCharPattern = `a-zA-Z0-9\-\_\.` var validNamePattern = regexp.MustCompile(`^[` + validNameCharPattern + `]*$`).MatchString func validateName(name string) error { diff --git a/pkg/storage/unified/resource/validation_test.go b/pkg/storage/unified/resource/validation_test.go new file mode 100644 index 00000000000..5ab0c2a3e75 --- /dev/null +++ b/pkg/storage/unified/resource/validation_test.go @@ -0,0 +1,29 @@ +package resource + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNameValidation(t *testing.T) { + require.Error(t, validateName("")) // too short + require.Error(t, validateName( // too long (max 64) + "0123456789012345678901234567890123456789012345678901234567890123456789", + )) + + // OK + require.NoError(t, validateName("hello-world")) + require.NoError(t, validateName("hello.world")) + require.NoError(t, validateName("hello_world")) + + // Bad characters + require.Error(t, validateName("hello world")) + require.Error(t, validateName("hello!")) + require.Error(t, validateName("hello~")) + require.Error(t, validateName("hello ")) + require.Error(t, validateName("hello*")) + require.Error(t, validateName("hello+")) + require.Error(t, validateName("hello=")) + require.Error(t, validateName("hello%")) +}