Storage: Remove entity store (#91233)
This commit is contained in:
Ryan McKinley
@@ -1,90 +0,0 @@
|
||||
// SPDX-License-Identifier: AGPL-3.0-only
|
||||
|
||||
package entity
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"path"
|
||||
"time"
|
||||
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apiserver/pkg/registry/generic"
|
||||
"k8s.io/apiserver/pkg/storage"
|
||||
"k8s.io/apiserver/pkg/storage/storagebackend"
|
||||
"k8s.io/apiserver/pkg/storage/storagebackend/factory"
|
||||
flowcontrolrequest "k8s.io/apiserver/pkg/util/flowcontrol/request"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
|
||||
entityStore "github.com/grafana/grafana/pkg/services/store/entity"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
)
|
||||
|
||||
var _ generic.RESTOptionsGetter = (*RESTOptionsGetter)(nil)
|
||||
|
||||
type RESTOptionsGetter struct {
|
||||
cfg *setting.Cfg
|
||||
store entityStore.EntityStoreClient
|
||||
Codec runtime.Codec
|
||||
}
|
||||
|
||||
func NewRESTOptionsGetter(cfg *setting.Cfg, store entityStore.EntityStoreClient, codec runtime.Codec) *RESTOptionsGetter {
|
||||
return &RESTOptionsGetter{
|
||||
cfg: cfg,
|
||||
store: store,
|
||||
Codec: codec,
|
||||
}
|
||||
}
|
||||
|
||||
func (f *RESTOptionsGetter) GetRESTOptions(resource schema.GroupResource) (generic.RESTOptions, error) {
|
||||
// build connection string to uniquely identify the storage backend
|
||||
connectionInfo, err := json.Marshal(f.cfg.SectionWithEnvOverrides("entity_api").KeysHash())
|
||||
if err != nil {
|
||||
return generic.RESTOptions{}, err
|
||||
}
|
||||
|
||||
storageConfig := &storagebackend.ConfigForResource{
|
||||
Config: storagebackend.Config{
|
||||
Type: "custom",
|
||||
Prefix: "",
|
||||
Transport: storagebackend.TransportConfig{
|
||||
ServerList: []string{
|
||||
string(connectionInfo),
|
||||
},
|
||||
},
|
||||
Codec: f.Codec,
|
||||
EncodeVersioner: nil,
|
||||
Transformer: nil,
|
||||
CompactionInterval: 0,
|
||||
CountMetricPollPeriod: 0,
|
||||
DBMetricPollInterval: 0,
|
||||
HealthcheckTimeout: 0,
|
||||
ReadycheckTimeout: 0,
|
||||
StorageObjectCountTracker: nil,
|
||||
},
|
||||
GroupResource: resource,
|
||||
}
|
||||
|
||||
ret := generic.RESTOptions{
|
||||
StorageConfig: storageConfig,
|
||||
Decorator: func(
|
||||
config *storagebackend.ConfigForResource,
|
||||
resourcePrefix string,
|
||||
keyFunc func(obj runtime.Object) (string, error),
|
||||
newFunc func() runtime.Object,
|
||||
newListFunc func() runtime.Object,
|
||||
getAttrsFunc storage.AttrFunc,
|
||||
trigger storage.IndexerFuncs,
|
||||
indexers *cache.Indexers,
|
||||
) (storage.Interface, factory.DestroyFunc, error) {
|
||||
return NewStorage(config, resource, f.store, f.Codec, keyFunc, newFunc, newListFunc, getAttrsFunc)
|
||||
},
|
||||
DeleteCollectionWorkers: 0,
|
||||
EnableGarbageCollection: false,
|
||||
ResourcePrefix: path.Join(storageConfig.Prefix, resource.Group, resource.Resource),
|
||||
CountMetricPollPeriod: 1 * time.Second,
|
||||
StorageObjectCountTracker: flowcontrolrequest.NewStorageObjectCountTracker(),
|
||||
}
|
||||
|
||||
return ret, nil
|
||||
}
|
||||
@@ -1,82 +0,0 @@
|
||||
package entity
|
||||
|
||||
import (
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
"k8s.io/apimachinery/pkg/labels"
|
||||
"k8s.io/apimachinery/pkg/selection"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||
)
|
||||
|
||||
const SortByKey = "grafana.app/sortBy"
|
||||
const ListDeletedKey = "grafana.app/listDeleted"
|
||||
const ListHistoryKey = "grafana.app/listHistory"
|
||||
|
||||
type Requirements struct {
|
||||
// Equals folder
|
||||
Folder *string
|
||||
// SortBy is a list of fields to sort by
|
||||
SortBy []string
|
||||
// ListDeleted is a flag to list deleted entities
|
||||
ListDeleted bool
|
||||
// ListHistory is a resource name to list the history of
|
||||
ListHistory string
|
||||
}
|
||||
|
||||
func ReadLabelSelectors(selector labels.Selector) (Requirements, labels.Selector, error) {
|
||||
requirements := Requirements{}
|
||||
newSelector := labels.NewSelector()
|
||||
|
||||
if selector == nil {
|
||||
return requirements, newSelector, nil
|
||||
}
|
||||
|
||||
labelSelectors, _ := selector.Requirements()
|
||||
|
||||
for _, r := range labelSelectors {
|
||||
switch r.Key() {
|
||||
case utils.AnnoKeyFolder:
|
||||
if (r.Operator() != selection.Equals) && (r.Operator() != selection.DoubleEquals) {
|
||||
return requirements, newSelector, apierrors.NewBadRequest(utils.AnnoKeyFolder + " label selector only supports equality")
|
||||
}
|
||||
folder := r.Values().List()[0]
|
||||
requirements.Folder = &folder
|
||||
case SortByKey:
|
||||
if r.Operator() != selection.In {
|
||||
return requirements, newSelector, apierrors.NewBadRequest(SortByKey + " label selector only supports in")
|
||||
}
|
||||
requirements.SortBy = r.Values().List()
|
||||
case ListDeletedKey:
|
||||
if r.Operator() != selection.Equals {
|
||||
return requirements, newSelector, apierrors.NewBadRequest(ListDeletedKey + " label selector only supports equality")
|
||||
}
|
||||
if len(r.Values().List()) != 1 {
|
||||
return requirements, newSelector, apierrors.NewBadRequest(ListDeletedKey + " label selector only supports one value")
|
||||
}
|
||||
if r.Values().List()[0] != "true" && r.Values().List()[0] != "false" {
|
||||
return requirements, newSelector, apierrors.NewBadRequest(ListDeletedKey + " label selector only supports true or false")
|
||||
}
|
||||
requirements.ListDeleted = r.Values().List()[0] == "true"
|
||||
case ListHistoryKey:
|
||||
if r.Operator() != selection.Equals {
|
||||
return requirements, newSelector, apierrors.NewBadRequest(ListHistoryKey + " label selector only supports equality")
|
||||
}
|
||||
if len(r.Values().List()) != 1 {
|
||||
return requirements, newSelector, apierrors.NewBadRequest(ListHistoryKey + " label selector only supports one value")
|
||||
}
|
||||
if r.Values().List()[0] == "" {
|
||||
return requirements, newSelector, apierrors.NewBadRequest(ListHistoryKey + " label selector must not be empty")
|
||||
}
|
||||
requirements.ListHistory = r.Values().List()[0]
|
||||
// add all unregonized label selectors to the new selector list, these will be processed by the entity store
|
||||
default:
|
||||
newSelector = newSelector.Add(r)
|
||||
}
|
||||
}
|
||||
|
||||
if requirements.ListDeleted && requirements.ListHistory != "" {
|
||||
return requirements, newSelector, apierrors.NewBadRequest("cannot list deleted and history at the same time")
|
||||
}
|
||||
|
||||
return requirements, newSelector, nil
|
||||
}
|
||||
@@ -1,753 +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 entity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"reflect"
|
||||
"strconv"
|
||||
|
||||
grpcCodes "google.golang.org/grpc/codes"
|
||||
grpcStatus "google.golang.org/grpc/status"
|
||||
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/selection"
|
||||
"k8s.io/apimachinery/pkg/watch"
|
||||
"k8s.io/apiserver/pkg/storage"
|
||||
"k8s.io/apiserver/pkg/storage/storagebackend"
|
||||
"k8s.io/apiserver/pkg/storage/storagebackend/factory"
|
||||
"k8s.io/klog/v2"
|
||||
|
||||
grafanaregistry "github.com/grafana/grafana/pkg/apiserver/registry/generic"
|
||||
entityStore "github.com/grafana/grafana/pkg/services/store/entity"
|
||||
)
|
||||
|
||||
var _ storage.Interface = (*Storage)(nil)
|
||||
|
||||
// Storage implements storage.Interface and stores resources in unified storage
|
||||
type Storage struct {
|
||||
config *storagebackend.ConfigForResource
|
||||
store entityStore.EntityStoreClient
|
||||
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 entityStore.EntityStoreClient,
|
||||
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
|
||||
}
|
||||
|
||||
// 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 := grafanaregistry.ParseKey(key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := s.Versioner().PrepareObjectForStorage(obj); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
e, err := resourceToEntity(obj, *k, s.codec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
req := &entityStore.CreateEntityRequest{
|
||||
Entity: e,
|
||||
}
|
||||
|
||||
rsp, err := s.store.Create(ctx, req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if rsp.Status != entityStore.CreateEntityResponse_CREATED {
|
||||
return fmt.Errorf("this was not a create operation... (%s)", rsp.Status.String())
|
||||
}
|
||||
|
||||
err = EntityToRuntimeObject(rsp.Entity, out, s.codec)
|
||||
if err != nil {
|
||||
return apierrors.NewInternalError(err)
|
||||
}
|
||||
|
||||
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 := grafanaregistry.ParseKey(key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
previousVersion := int64(0)
|
||||
if preconditions != nil && preconditions.ResourceVersion != nil {
|
||||
previousVersion, _ = strconv.ParseInt(*preconditions.ResourceVersion, 10, 64)
|
||||
}
|
||||
|
||||
rsp, err := s.store.Delete(ctx, &entityStore.DeleteEntityRequest{
|
||||
Key: k.String(),
|
||||
PreviousVersion: previousVersion,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = EntityToRuntimeObject(rsp.Entity, out, s.codec)
|
||||
if err != nil {
|
||||
return apierrors.NewInternalError(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) {
|
||||
k, err := grafanaregistry.ParseKey(key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if opts.Predicate.Field != nil {
|
||||
// check for metadata.name field selector
|
||||
if v, ok := opts.Predicate.Field.RequiresExactMatch("metadata.name"); ok && k.Name == "" {
|
||||
// just watch the specific key if we have a name field selector
|
||||
k.Name = v
|
||||
}
|
||||
|
||||
// check for metadata.namespace field selector
|
||||
if v, ok := opts.Predicate.Field.RequiresExactMatch("metadata.namespace"); ok && k.Namespace == "" {
|
||||
// just watch the specific namespace if we have a namespace field selector
|
||||
k.Namespace = v
|
||||
}
|
||||
}
|
||||
|
||||
// translate grafana.app/* label selectors into field requirements
|
||||
requirements, newSelector, err := ReadLabelSelectors(opts.Predicate.Label)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Update the selector to remove the unneeded requirements
|
||||
opts.Predicate.Label = newSelector
|
||||
|
||||
// if we got a listHistory label selector, watch the specified resource
|
||||
if requirements.ListHistory != "" {
|
||||
if k.Name != "" && k.Name != requirements.ListHistory {
|
||||
return nil, apierrors.NewBadRequest("name field selector does not match listHistory")
|
||||
}
|
||||
k.Name = requirements.ListHistory
|
||||
}
|
||||
|
||||
req := &entityStore.EntityWatchRequest{
|
||||
Action: entityStore.EntityWatchRequest_START,
|
||||
Key: []string{
|
||||
k.String(),
|
||||
},
|
||||
Labels: map[string]string{},
|
||||
WithBody: true,
|
||||
WithStatus: true,
|
||||
SendInitialEvents: false,
|
||||
AllowWatchBookmarks: opts.Predicate.AllowWatchBookmarks,
|
||||
}
|
||||
|
||||
if opts.ResourceVersion != "" {
|
||||
rv, err := strconv.ParseInt(opts.ResourceVersion, 10, 64)
|
||||
if err != nil {
|
||||
return nil, apierrors.NewBadRequest(fmt.Sprintf("invalid resource version: %s", opts.ResourceVersion))
|
||||
}
|
||||
|
||||
req.Since = rv
|
||||
}
|
||||
|
||||
if opts.SendInitialEvents == nil && req.Since == 0 {
|
||||
req.SendInitialEvents = true
|
||||
} else if opts.SendInitialEvents != nil {
|
||||
req.SendInitialEvents = *opts.SendInitialEvents
|
||||
}
|
||||
|
||||
if requirements.Folder != nil {
|
||||
req.Folder = *requirements.Folder
|
||||
}
|
||||
|
||||
// translate "equals" label selectors to storage label conditions
|
||||
labelRequirements, selectable := opts.Predicate.Label.Requirements()
|
||||
if !selectable {
|
||||
return nil, apierrors.NewBadRequest("label selector is not selectable")
|
||||
}
|
||||
|
||||
for _, r := range labelRequirements {
|
||||
if r.Operator() == selection.Equals {
|
||||
req.Labels[r.Key()] = r.Values().List()[0]
|
||||
}
|
||||
}
|
||||
|
||||
client, err := s.store.Watch(ctx)
|
||||
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
|
||||
}
|
||||
|
||||
err = client.Send(req)
|
||||
if err != nil {
|
||||
err2 := client.CloseSend()
|
||||
if err2 != nil {
|
||||
klog.Errorf("watch close failed: %s\n", err2)
|
||||
}
|
||||
|
||||
// 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 := &Decoder{
|
||||
client: client,
|
||||
newFunc: s.newFunc,
|
||||
opts: opts,
|
||||
codec: s.codec,
|
||||
}
|
||||
|
||||
w := watch.NewStreamWatcher(decoder, reporter)
|
||||
|
||||
return w, nil
|
||||
}
|
||||
|
||||
// Get unmarshals 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 {
|
||||
k, err := grafanaregistry.ParseKey(key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
resourceVersion := int64(0)
|
||||
if opts.ResourceVersion != "" {
|
||||
resourceVersion, err = strconv.ParseInt(opts.ResourceVersion, 10, 64)
|
||||
if err != nil {
|
||||
return apierrors.NewBadRequest(fmt.Sprintf("invalid resource version: %s", opts.ResourceVersion))
|
||||
}
|
||||
}
|
||||
|
||||
rsp, err := s.store.Read(ctx, &entityStore.ReadEntityRequest{
|
||||
Key: k.String(),
|
||||
WithBody: true,
|
||||
WithStatus: true,
|
||||
ResourceVersion: resourceVersion,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if rsp.Key == "" {
|
||||
if opts.IgnoreNotFound {
|
||||
return nil
|
||||
}
|
||||
|
||||
return apierrors.NewNotFound(s.gr, k.Name)
|
||||
}
|
||||
|
||||
err = EntityToRuntimeObject(rsp, objPtr, s.codec)
|
||||
if err != nil {
|
||||
return apierrors.NewInternalError(err)
|
||||
}
|
||||
|
||||
return 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 {
|
||||
k, err := grafanaregistry.ParseKey(key)
|
||||
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
|
||||
}
|
||||
|
||||
// translate grafana.app/* label selectors into field requirements
|
||||
requirements, newSelector, err := ReadLabelSelectors(opts.Predicate.Label)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Update the selector to remove the unneeded requirements
|
||||
opts.Predicate.Label = newSelector
|
||||
|
||||
if requirements.ListHistory != "" {
|
||||
k.Name = requirements.ListHistory
|
||||
|
||||
req := &entityStore.EntityHistoryRequest{
|
||||
Key: k.String(),
|
||||
WithBody: true,
|
||||
WithStatus: true,
|
||||
NextPageToken: opts.Predicate.Continue,
|
||||
Limit: opts.Predicate.Limit,
|
||||
Sort: requirements.SortBy,
|
||||
}
|
||||
|
||||
rsp, err := s.store.History(ctx, req)
|
||||
if err != nil {
|
||||
return apierrors.NewInternalError(err)
|
||||
}
|
||||
|
||||
for _, r := range rsp.Versions {
|
||||
res := s.newFunc()
|
||||
|
||||
err := EntityToRuntimeObject(r, res, s.codec)
|
||||
if err != nil {
|
||||
return apierrors.NewInternalError(err)
|
||||
}
|
||||
|
||||
// apply any predicates not handled in storage
|
||||
matches, err := opts.Predicate.Matches(res)
|
||||
if err != nil {
|
||||
return apierrors.NewInternalError(err)
|
||||
}
|
||||
if !matches {
|
||||
continue
|
||||
}
|
||||
|
||||
v.Set(reflect.Append(v, reflect.ValueOf(res).Elem()))
|
||||
}
|
||||
|
||||
listAccessor, err := meta.ListAccessor(listObj)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if rsp.NextPageToken != "" {
|
||||
listAccessor.SetContinue(rsp.NextPageToken)
|
||||
}
|
||||
|
||||
listAccessor.SetResourceVersion(strconv.FormatInt(rsp.ResourceVersion, 10))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
req := &entityStore.EntityListRequest{
|
||||
Key: []string{
|
||||
k.String(),
|
||||
},
|
||||
WithBody: true,
|
||||
WithStatus: true,
|
||||
NextPageToken: opts.Predicate.Continue,
|
||||
Limit: opts.Predicate.Limit,
|
||||
Labels: map[string]string{},
|
||||
}
|
||||
|
||||
if requirements.Folder != nil {
|
||||
req.Folder = *requirements.Folder
|
||||
}
|
||||
if len(requirements.SortBy) > 0 {
|
||||
req.Sort = requirements.SortBy
|
||||
}
|
||||
if requirements.ListDeleted {
|
||||
req.Deleted = true
|
||||
}
|
||||
|
||||
// translate "equals" label selectors to storage label conditions
|
||||
labelRequirements, selectable := opts.Predicate.Label.Requirements()
|
||||
if !selectable {
|
||||
return apierrors.NewBadRequest("label selector is not selectable")
|
||||
}
|
||||
|
||||
for _, r := range labelRequirements {
|
||||
if r.Operator() == selection.Equals {
|
||||
req.Labels[r.Key()] = r.Values().List()[0]
|
||||
}
|
||||
}
|
||||
|
||||
rsp, err := s.store.List(ctx, req)
|
||||
if err != nil {
|
||||
return apierrors.NewInternalError(err)
|
||||
}
|
||||
|
||||
for _, r := range rsp.Results {
|
||||
res := s.newFunc()
|
||||
|
||||
err := EntityToRuntimeObject(r, res, s.codec)
|
||||
if err != nil {
|
||||
return apierrors.NewInternalError(err)
|
||||
}
|
||||
|
||||
// apply any predicates not handled in storage
|
||||
matches, err := opts.Predicate.Matches(res)
|
||||
if err != nil {
|
||||
return apierrors.NewInternalError(err)
|
||||
}
|
||||
if !matches {
|
||||
continue
|
||||
}
|
||||
|
||||
v.Set(reflect.Append(v, reflect.ValueOf(res).Elem()))
|
||||
}
|
||||
|
||||
listAccessor, err := meta.ListAccessor(listObj)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if rsp.NextPageToken != "" {
|
||||
listAccessor.SetContinue(rsp.NextPageToken)
|
||||
}
|
||||
|
||||
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 := grafanaregistry.ParseKey(key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
getErr := s.Get(ctx, k.String(), storage.GetOptions{}, destination)
|
||||
if getErr != nil {
|
||||
if ignoreNotFound && apierrors.IsNotFound(getErr) {
|
||||
// destination is already set to zero value
|
||||
// we'll create the resource
|
||||
} else {
|
||||
return getErr
|
||||
}
|
||||
}
|
||||
|
||||
accessor, err := meta.Accessor(destination)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
previousVersion, _ := strconv.ParseInt(accessor.GetResourceVersion(), 10, 64)
|
||||
if preconditions != nil && preconditions.ResourceVersion != nil {
|
||||
previousVersion, _ = strconv.ParseInt(*preconditions.ResourceVersion, 10, 64)
|
||||
}
|
||||
|
||||
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()))
|
||||
}
|
||||
|
||||
e, err := resourceToEntity(updatedObj, *k, s.codec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// if we have a non-nil getErr, then we've ignored a not found error
|
||||
if getErr != nil {
|
||||
// object does not exist, create it
|
||||
req := &entityStore.CreateEntityRequest{
|
||||
Entity: e,
|
||||
}
|
||||
|
||||
rsp, err := s.store.Create(ctx, req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = EntityToRuntimeObject(rsp.Entity, destination, s.codec)
|
||||
if err != nil {
|
||||
return apierrors.NewInternalError(err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// update the existing object
|
||||
req := &entityStore.UpdateEntityRequest{
|
||||
Entity: e,
|
||||
PreviousVersion: previousVersion,
|
||||
}
|
||||
|
||||
rsp, err := s.store.Update(ctx, req)
|
||||
if err != nil {
|
||||
return err // continue???
|
||||
}
|
||||
|
||||
if rsp.Status == entityStore.UpdateEntityResponse_UNCHANGED {
|
||||
return nil // destination is already set
|
||||
}
|
||||
|
||||
err = EntityToRuntimeObject(rsp.Entity, destination, s.codec)
|
||||
if err != nil {
|
||||
return apierrors.NewInternalError(err)
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
type Decoder struct {
|
||||
client entityStore.EntityStore_WatchClient
|
||||
newFunc func() runtime.Object
|
||||
opts storage.ListOptions
|
||||
codec runtime.Codec
|
||||
}
|
||||
|
||||
func (d *Decoder) Decode() (action watch.EventType, object runtime.Object, err error) {
|
||||
decode:
|
||||
for {
|
||||
err := d.client.Context().Err()
|
||||
if err != nil {
|
||||
klog.Errorf("client: context error: %s\n", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
|
||||
resp, err := d.client.Recv()
|
||||
if errors.Is(err, io.EOF) {
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
|
||||
if grpcStatus.Code(err) == grpcCodes.Canceled {
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
klog.Errorf("client: error receiving result: %s", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
|
||||
if resp.Entity == nil {
|
||||
klog.Errorf("client: received nil entity\n")
|
||||
continue decode
|
||||
}
|
||||
|
||||
obj := d.newFunc()
|
||||
|
||||
if resp.Entity.Action == entityStore.Entity_BOOKMARK {
|
||||
// here k8s expects an empty object with just resource version and k8s.io/initial-events-end annotation
|
||||
accessor, err := meta.Accessor(obj)
|
||||
if err != nil {
|
||||
klog.Errorf("error getting object accessor: %s", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
|
||||
accessor.SetResourceVersion(fmt.Sprintf("%d", resp.Entity.ResourceVersion))
|
||||
accessor.SetAnnotations(map[string]string{"k8s.io/initial-events-end": "true"})
|
||||
return watch.Bookmark, obj, nil
|
||||
}
|
||||
|
||||
err = EntityToRuntimeObject(resp.Entity, obj, d.codec)
|
||||
if err != nil {
|
||||
klog.Errorf("error decoding entity: %s", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
|
||||
var watchAction watch.EventType
|
||||
switch resp.Entity.Action {
|
||||
case entityStore.Entity_CREATED:
|
||||
// apply any predicates not handled in storage
|
||||
matches, err := d.opts.Predicate.Matches(obj)
|
||||
if err != nil {
|
||||
klog.Errorf("error matching object: %s", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
if !matches {
|
||||
continue decode
|
||||
}
|
||||
|
||||
watchAction = watch.Added
|
||||
case entityStore.Entity_UPDATED:
|
||||
watchAction = watch.Modified
|
||||
|
||||
// apply any predicates not handled in storage
|
||||
matches, err := d.opts.Predicate.Matches(obj)
|
||||
if err != nil {
|
||||
klog.Errorf("error matching object: %s", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
|
||||
// if we have a previous object, check if it matches
|
||||
prevMatches := false
|
||||
prevObj := d.newFunc()
|
||||
if resp.Previous != nil {
|
||||
err = EntityToRuntimeObject(resp.Previous, prevObj, d.codec)
|
||||
if err != nil {
|
||||
klog.Errorf("error decoding entity: %s", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
|
||||
// apply any predicates not handled in storage
|
||||
prevMatches, err = d.opts.Predicate.Matches(prevObj)
|
||||
if err != nil {
|
||||
klog.Errorf("error matching object: %s", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
}
|
||||
|
||||
if !matches {
|
||||
if !prevMatches {
|
||||
continue decode
|
||||
}
|
||||
|
||||
// if the object didn't match, send a Deleted event
|
||||
watchAction = watch.Deleted
|
||||
|
||||
// here k8s expects the previous object but with the new resource version
|
||||
obj = prevObj
|
||||
|
||||
accessor, err := meta.Accessor(obj)
|
||||
if err != nil {
|
||||
klog.Errorf("error getting object accessor: %s", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
|
||||
accessor.SetResourceVersion(fmt.Sprintf("%d", resp.Entity.ResourceVersion))
|
||||
} else if !prevMatches {
|
||||
// if the object didn't previously match, send an Added event
|
||||
watchAction = watch.Added
|
||||
}
|
||||
case entityStore.Entity_DELETED:
|
||||
watchAction = watch.Deleted
|
||||
|
||||
// if we have a previous object, return that in the deleted event
|
||||
if resp.Previous != nil {
|
||||
err = EntityToRuntimeObject(resp.Previous, obj, d.codec)
|
||||
if err != nil {
|
||||
klog.Errorf("error decoding entity: %s", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
|
||||
// here k8s expects the previous object but with the new resource version
|
||||
accessor, err := meta.Accessor(obj)
|
||||
if err != nil {
|
||||
klog.Errorf("error getting object accessor: %s", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
|
||||
accessor.SetResourceVersion(fmt.Sprintf("%d", resp.Entity.ResourceVersion))
|
||||
}
|
||||
|
||||
// apply any predicates not handled in storage
|
||||
matches, err := d.opts.Predicate.Matches(obj)
|
||||
if err != nil {
|
||||
klog.Errorf("error matching object: %s", err)
|
||||
return watch.Error, nil, err
|
||||
}
|
||||
if !matches {
|
||||
continue decode
|
||||
}
|
||||
default:
|
||||
watchAction = watch.Error
|
||||
}
|
||||
|
||||
return watchAction, obj, nil
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Decoder) Close() {
|
||||
err := d.client.CloseSend()
|
||||
if err != nil {
|
||||
klog.Errorf("error closing watch stream: %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
var _ watch.Decoder = (*Decoder)(nil)
|
||||
@@ -1,376 +0,0 @@
|
||||
// SPDX-License-Identifier: AGPL-3.0-only
|
||||
// Provenance-includes-location: https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go
|
||||
// Provenance-includes-license: Apache-2.0
|
||||
// Provenance-includes-copyright: The Kubernetes Authors.
|
||||
|
||||
package test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"k8s.io/apimachinery/pkg/api/apitesting"
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/runtime/serializer"
|
||||
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
||||
"k8s.io/apiserver/pkg/apis/example"
|
||||
examplev1 "k8s.io/apiserver/pkg/apis/example/v1"
|
||||
"k8s.io/apiserver/pkg/storage"
|
||||
"k8s.io/apiserver/pkg/storage/storagebackend"
|
||||
"k8s.io/apiserver/pkg/storage/storagebackend/factory"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apimachinery/identity"
|
||||
storagetesting "github.com/grafana/grafana/pkg/apiserver/storage/testing"
|
||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||
"github.com/grafana/grafana/pkg/services/apiserver/storage/entity"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/services/sqlstore"
|
||||
entityStore "github.com/grafana/grafana/pkg/services/store/entity"
|
||||
"github.com/grafana/grafana/pkg/services/store/entity/db/dbimpl"
|
||||
"github.com/grafana/grafana/pkg/services/store/entity/sqlstash"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/tests/testinfra"
|
||||
"github.com/grafana/grafana/pkg/tests/testsuite"
|
||||
)
|
||||
|
||||
var scheme = runtime.NewScheme()
|
||||
var codecs = serializer.NewCodecFactory(scheme)
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
testsuite.Run(m)
|
||||
}
|
||||
|
||||
func createTestContext(t *testing.T) (entityStore.EntityStoreClient, factory.DestroyFunc) {
|
||||
t.Helper()
|
||||
|
||||
grafDir, cfgPath := testinfra.CreateGrafDir(t, testinfra.GrafanaOpts{
|
||||
EnableFeatureToggles: []string{
|
||||
featuremgmt.FlagGrpcServer,
|
||||
featuremgmt.FlagUnifiedStorage,
|
||||
},
|
||||
AppModeProduction: false, // required for migrations to run
|
||||
GRPCServerAddress: "127.0.0.1:0", // :0 for choosing the port automatically
|
||||
})
|
||||
|
||||
cfg, err := setting.NewCfgFromArgs(setting.CommandLineArgs{Config: cfgPath, HomePath: grafDir})
|
||||
assert.NoError(t, err)
|
||||
|
||||
featureManager, err := featuremgmt.ProvideManagerService(cfg)
|
||||
assert.NoError(t, err)
|
||||
|
||||
featureToggles := featuremgmt.ProvideToggles(featureManager)
|
||||
|
||||
db := sqlstore.InitTestDBWithMigration(t, nil, sqlstore.InitTestDBOpt{EnsureDefaultOrgAndUser: false})
|
||||
require.NoError(t, err)
|
||||
|
||||
eDB, err := dbimpl.ProvideEntityDB(db, cfg, featureToggles, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
err = eDB.Init()
|
||||
require.NoError(t, err)
|
||||
|
||||
traceConfig, err := tracing.ParseTracingConfig(cfg)
|
||||
require.NoError(t, err)
|
||||
tracer, err := tracing.ProvideService(traceConfig)
|
||||
require.NoError(t, err)
|
||||
store, err := sqlstash.ProvideSQLEntityServer(eDB, tracer)
|
||||
require.NoError(t, err)
|
||||
|
||||
client := entityStore.NewEntityStoreClientLocal(store)
|
||||
|
||||
return client, func() { store.Stop() }
|
||||
}
|
||||
|
||||
func init() {
|
||||
metav1.AddToGroupVersion(scheme, metav1.SchemeGroupVersion)
|
||||
utilruntime.Must(example.AddToScheme(scheme))
|
||||
utilruntime.Must(examplev1.AddToScheme(scheme))
|
||||
}
|
||||
|
||||
type setupOptions struct {
|
||||
codec runtime.Codec
|
||||
newFunc func() runtime.Object
|
||||
newListFunc func() runtime.Object
|
||||
prefix string
|
||||
resourcePrefix string
|
||||
groupResource schema.GroupResource
|
||||
}
|
||||
|
||||
type setupOption func(*setupOptions, *testing.T)
|
||||
|
||||
func withDefaults(options *setupOptions, t *testing.T) {
|
||||
options.codec = apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)
|
||||
options.newFunc = newPod
|
||||
options.newListFunc = newPodList
|
||||
options.prefix = t.TempDir()
|
||||
options.resourcePrefix = "/resource/pods"
|
||||
options.groupResource = schema.GroupResource{Resource: "pods"}
|
||||
}
|
||||
|
||||
var _ setupOption = withDefaults
|
||||
|
||||
func testSetup(t *testing.T, opts ...setupOption) (context.Context, storage.Interface, factory.DestroyFunc, error) {
|
||||
setupOpts := setupOptions{}
|
||||
opts = append([]setupOption{withDefaults}, opts...)
|
||||
for _, opt := range opts {
|
||||
opt(&setupOpts, t)
|
||||
}
|
||||
|
||||
config := storagebackend.NewDefaultConfig(setupOpts.prefix, setupOpts.codec)
|
||||
|
||||
client, destroyFunc := createTestContext(t)
|
||||
|
||||
store, _, err := entity.NewStorage(
|
||||
config.ForResource(setupOpts.groupResource),
|
||||
setupOpts.groupResource,
|
||||
client,
|
||||
setupOpts.codec,
|
||||
func(obj runtime.Object) (string, error) {
|
||||
accessor, err := meta.Accessor(obj)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return storagetesting.KeyFunc(accessor.GetNamespace(), accessor.GetName()), nil
|
||||
},
|
||||
setupOpts.newFunc,
|
||||
setupOpts.newListFunc,
|
||||
storage.DefaultNamespaceScopedAttr,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
|
||||
// Test with an admin identity
|
||||
ctx := identity.WithRequester(context.Background(), &identity.StaticRequester{
|
||||
Type: identity.TypeUser,
|
||||
Login: "testuser",
|
||||
UserID: 123,
|
||||
UserUID: "u123",
|
||||
OrgRole: identity.RoleAdmin,
|
||||
IsGrafanaAdmin: true, // can do anything
|
||||
})
|
||||
|
||||
return ctx, store, destroyFunc, nil
|
||||
}
|
||||
|
||||
func TestIntegrationWatch(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
t.Skip("In maintenance")
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestWatch(ctx, t, store)
|
||||
}
|
||||
|
||||
func TestIntegrationClusterScopedWatch(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
t.Skip("In maintenance")
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestClusterScopedWatch(ctx, t, store)
|
||||
}
|
||||
|
||||
func TestIntegrationNamespaceScopedWatch(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
t.Skip("In maintenance")
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestNamespaceScopedWatch(ctx, t, store)
|
||||
}
|
||||
|
||||
func TestIntegrationDeleteTriggerWatch(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
t.Skip("In maintenance")
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestDeleteTriggerWatch(ctx, t, store)
|
||||
}
|
||||
|
||||
func TestIntegrationWatchFromZero(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
t.Skip("In maintenance")
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestWatchFromZero(ctx, t, store, nil)
|
||||
}
|
||||
|
||||
// TestWatchFromNonZero tests that
|
||||
// - watch from non-0 should just watch changes after given version
|
||||
func TestIntegrationWatchFromNonZero(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
t.Skip("In maintenance")
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestWatchFromNonZero(ctx, t, store)
|
||||
}
|
||||
|
||||
/*
|
||||
// TODO this times out, we need to buffer events
|
||||
func TestIntegrationDelayedWatchDelivery(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestDelayedWatchDelivery(ctx, t, store)
|
||||
}
|
||||
*/
|
||||
|
||||
/* func TestIntegrationWatchError(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
ctx, store, _ := testSetup(t)
|
||||
storagetesting.RunTestWatchError(ctx, t, &storeWithPrefixTransformer{store})
|
||||
} */
|
||||
|
||||
func TestIntegrationWatchContextCancel(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
t.Skip("In maintenance")
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestWatchContextCancel(ctx, t, store)
|
||||
}
|
||||
|
||||
func TestIntegrationWatcherTimeout(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
t.Skip("In maintenance")
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestWatcherTimeout(ctx, t, store)
|
||||
}
|
||||
|
||||
func TestIntegrationWatchDeleteEventObjectHaveLatestRV(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
t.Skip("In maintenance")
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestWatchDeleteEventObjectHaveLatestRV(ctx, t, store)
|
||||
}
|
||||
|
||||
// TODO: enable when we support flow control and priority fairness
|
||||
/* func TestIntegrationWatchInitializationSignal(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestWatchInitializationSignal(ctx, t, store)
|
||||
} */
|
||||
|
||||
/* func TestIntegrationProgressNotify(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunOptionalTestProgressNotify(ctx, t, store)
|
||||
} */
|
||||
|
||||
// TestWatchDispatchBookmarkEvents makes sure that
|
||||
// setting allowWatchBookmarks query param against
|
||||
// etcd implementation doesn't have any effect.
|
||||
func TestIntegrationWatchDispatchBookmarkEvents(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
t.Skip("In maintenance")
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunTestWatchDispatchBookmarkEvents(ctx, t, store, false)
|
||||
}
|
||||
|
||||
func TestIntegrationSendInitialEventsBackwardCompatibility(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunSendInitialEventsBackwardCompatibility(ctx, t, store)
|
||||
}
|
||||
|
||||
// TODO this test times out
|
||||
func TestIntegrationEtcdWatchSemantics(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
t.Skip("In maintenance")
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunWatchSemantics(ctx, t, store)
|
||||
}
|
||||
|
||||
/*
|
||||
// TODO this test times out
|
||||
func TestIntegrationEtcdWatchSemanticInitialEventsExtended(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
ctx, store, destroyFunc, err := testSetup(t)
|
||||
defer destroyFunc()
|
||||
assert.NoError(t, err)
|
||||
storagetesting.RunWatchSemanticInitialEventsExtended(ctx, t, store)
|
||||
}
|
||||
*/
|
||||
|
||||
func newPod() runtime.Object {
|
||||
return &example.Pod{}
|
||||
}
|
||||
|
||||
func newPodList() runtime.Object {
|
||||
return &example.PodList{}
|
||||
}
|
||||
@@ -1,176 +0,0 @@
|
||||
package entity
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||
grafanaregistry "github.com/grafana/grafana/pkg/apiserver/registry/generic"
|
||||
entityStore "github.com/grafana/grafana/pkg/services/store/entity"
|
||||
)
|
||||
|
||||
func EntityToRuntimeObject(rsp *entityStore.Entity, res runtime.Object, codec runtime.Codec) error {
|
||||
var err error
|
||||
|
||||
// Read the body first -- it includes old resourceVersion!
|
||||
if len(rsp.Body) > 0 {
|
||||
decoded, _, err := codec.Decode(rsp.Body, &schema.GroupVersionKind{Group: rsp.Group, Version: rsp.GroupVersion}, res)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
res = decoded
|
||||
}
|
||||
|
||||
metaAccessor, err := meta.Accessor(res)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if len(rsp.Meta) > 0 {
|
||||
err = json.Unmarshal(rsp.Meta, res)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
metaAccessor.SetName(rsp.Name)
|
||||
metaAccessor.SetNamespace(rsp.Namespace)
|
||||
metaAccessor.SetUID(types.UID(rsp.Guid))
|
||||
metaAccessor.SetResourceVersion(fmt.Sprintf("%d", rsp.ResourceVersion))
|
||||
metaAccessor.SetCreationTimestamp(metav1.Unix(rsp.CreatedAt/1000, rsp.CreatedAt%1000*1000000))
|
||||
|
||||
grafanaAccessor, err := utils.MetaAccessor(metaAccessor)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if rsp.Folder != "" {
|
||||
grafanaAccessor.SetFolder(rsp.Folder)
|
||||
}
|
||||
if rsp.CreatedBy != "" {
|
||||
grafanaAccessor.SetCreatedBy(rsp.CreatedBy)
|
||||
}
|
||||
if rsp.UpdatedBy != "" {
|
||||
grafanaAccessor.SetUpdatedBy(rsp.UpdatedBy)
|
||||
}
|
||||
if rsp.UpdatedAt != 0 {
|
||||
updatedAt := time.UnixMilli(rsp.UpdatedAt).UTC()
|
||||
grafanaAccessor.SetUpdatedTimestamp(&updatedAt)
|
||||
}
|
||||
grafanaAccessor.SetSlug(rsp.Slug)
|
||||
|
||||
if rsp.Origin != nil {
|
||||
originTime := time.UnixMilli(rsp.Origin.Time).UTC()
|
||||
grafanaAccessor.SetOriginInfo(&utils.ResourceOriginInfo{
|
||||
Name: rsp.Origin.Source,
|
||||
Path: rsp.Origin.Key, // Using "key" in the
|
||||
// Path: rsp.Origin.Path,
|
||||
Timestamp: &originTime,
|
||||
})
|
||||
}
|
||||
|
||||
if len(rsp.Labels) > 0 {
|
||||
metaAccessor.SetLabels(rsp.Labels)
|
||||
}
|
||||
|
||||
// TODO fields?
|
||||
|
||||
if len(rsp.Status) > 0 {
|
||||
status := reflect.ValueOf(res).Elem().FieldByName("Status")
|
||||
if status != (reflect.Value{}) && status.CanSet() {
|
||||
err = json.Unmarshal(rsp.Status, status.Addr().Interface())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func resourceToEntity(res runtime.Object, k grafanaregistry.Key, codec runtime.Codec) (*entityStore.Entity, error) {
|
||||
metaAccessor, err := meta.Accessor(res)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
grafanaAccessor, err := utils.MetaAccessor(metaAccessor)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rv, _ := strconv.ParseInt(metaAccessor.GetResourceVersion(), 10, 64)
|
||||
|
||||
// add the object's name to the provided key
|
||||
k.Name = metaAccessor.GetName()
|
||||
|
||||
rsp := &entityStore.Entity{
|
||||
Group: k.Group,
|
||||
GroupVersion: res.GetObjectKind().GroupVersionKind().Version,
|
||||
Resource: k.Resource,
|
||||
Namespace: k.Namespace,
|
||||
Key: k.String(),
|
||||
Name: k.Name,
|
||||
Guid: string(metaAccessor.GetUID()),
|
||||
ResourceVersion: rv,
|
||||
Folder: grafanaAccessor.GetFolder(),
|
||||
CreatedAt: metaAccessor.GetCreationTimestamp().Time.UnixMilli(),
|
||||
CreatedBy: grafanaAccessor.GetCreatedBy(),
|
||||
UpdatedBy: grafanaAccessor.GetUpdatedBy(),
|
||||
Slug: grafanaAccessor.GetSlug(),
|
||||
Title: grafanaAccessor.FindTitle(metaAccessor.GetName()),
|
||||
Origin: &entityStore.EntityOriginInfo{
|
||||
Source: grafanaAccessor.GetOriginName(),
|
||||
// Deprecated: Keeping "key" in the protobuf to avoid migrations while a bigger one is in place
|
||||
Key: grafanaAccessor.GetOriginPath(),
|
||||
},
|
||||
Labels: metaAccessor.GetLabels(),
|
||||
}
|
||||
|
||||
t, err := grafanaAccessor.GetUpdatedTimestamp()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if t != nil {
|
||||
rsp.UpdatedAt = t.UnixMilli()
|
||||
}
|
||||
|
||||
t, err = grafanaAccessor.GetOriginTimestamp()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if t != nil {
|
||||
rsp.Origin.Time = t.UnixMilli()
|
||||
}
|
||||
|
||||
rsp.Meta, err = json.Marshal(meta.AsPartialObjectMetadata(metaAccessor))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var buf bytes.Buffer
|
||||
err = codec.Encode(res, &buf)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rsp.Body = buf.Bytes()
|
||||
|
||||
status := reflect.ValueOf(res).Elem().FieldByName("Status")
|
||||
if status != (reflect.Value{}) {
|
||||
rsp.Status, err = json.Marshal(status.Interface())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return rsp, nil
|
||||
}
|
||||
@@ -1,219 +0,0 @@
|
||||
package entity
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/serializer"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apis/playlist/v0alpha1"
|
||||
grafanaregistry "github.com/grafana/grafana/pkg/apiserver/registry/generic"
|
||||
entityStore "github.com/grafana/grafana/pkg/services/store/entity"
|
||||
)
|
||||
|
||||
func TestResourceToEntity(t *testing.T) {
|
||||
createdAt := metav1.Now()
|
||||
createdAtStr := createdAt.UTC().Format(time.RFC3339)
|
||||
|
||||
// truncate here because RFC3339 doesn't support millisecond precision
|
||||
// consider updating accessor to use RFC3339Nano to encode timestamps
|
||||
updatedAt := createdAt.Add(time.Hour).Truncate(time.Second)
|
||||
updatedAtStr := updatedAt.UTC().Format(time.RFC3339)
|
||||
|
||||
Scheme := runtime.NewScheme()
|
||||
Scheme.AddKnownTypes(v0alpha1.PlaylistResourceInfo.GroupVersion(), &v0alpha1.Playlist{})
|
||||
Codecs := serializer.NewCodecFactory(Scheme)
|
||||
|
||||
testCases := []struct {
|
||||
key grafanaregistry.Key
|
||||
resource runtime.Object
|
||||
codec runtime.Codec
|
||||
expectedKey string
|
||||
expectedGroupVersion string
|
||||
expectedName string
|
||||
expectedNamespace string
|
||||
expectedTitle string
|
||||
expectedGuid string
|
||||
expectedVersion string
|
||||
expectedFolder string
|
||||
expectedCreatedAt int64
|
||||
expectedUpdatedAt int64
|
||||
expectedCreatedBy string
|
||||
expectedUpdatedBy string
|
||||
expectedSlug string
|
||||
expectedOrigin *entityStore.EntityOriginInfo
|
||||
expectedLabels map[string]string
|
||||
expectedMeta []byte
|
||||
expectedBody []byte
|
||||
}{
|
||||
{
|
||||
key: grafanaregistry.Key{
|
||||
Group: "playlist.grafana.app",
|
||||
Resource: "playlists",
|
||||
Namespace: "default",
|
||||
Name: "test-name",
|
||||
},
|
||||
resource: &v0alpha1.Playlist{
|
||||
TypeMeta: metav1.TypeMeta{
|
||||
APIVersion: "playlist.grafana.app/v0alpha1",
|
||||
Kind: "Playlist",
|
||||
},
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
CreationTimestamp: createdAt,
|
||||
Labels: map[string]string{"label1": "value1", "label2": "value2"},
|
||||
Name: "test-name",
|
||||
ResourceVersion: "1",
|
||||
UID: "test-uid",
|
||||
Annotations: map[string]string{
|
||||
"grafana.app/createdBy": "test-created-by",
|
||||
"grafana.app/updatedBy": "test-updated-by",
|
||||
"grafana.app/updatedTimestamp": updatedAtStr,
|
||||
"grafana.app/folder": "test-folder",
|
||||
"grafana.app/slug": "test-slug",
|
||||
},
|
||||
},
|
||||
Spec: v0alpha1.Spec{
|
||||
Title: "A playlist",
|
||||
Interval: "5m",
|
||||
Items: []v0alpha1.Item{
|
||||
{Type: v0alpha1.ItemTypeDashboardByTag, Value: "panel-tests"},
|
||||
{Type: v0alpha1.ItemTypeDashboardByUid, Value: "vmie2cmWz"},
|
||||
},
|
||||
},
|
||||
},
|
||||
expectedKey: "/group/playlist.grafana.app/resource/playlists/namespace/default/name/test-name",
|
||||
expectedGroupVersion: "v0alpha1",
|
||||
expectedName: "test-name",
|
||||
expectedNamespace: "default",
|
||||
expectedTitle: "A playlist",
|
||||
expectedGuid: "test-uid",
|
||||
expectedVersion: "1",
|
||||
expectedFolder: "test-folder",
|
||||
expectedCreatedAt: createdAt.UnixMilli(),
|
||||
expectedUpdatedAt: updatedAt.UnixMilli(),
|
||||
expectedCreatedBy: "test-created-by",
|
||||
expectedUpdatedBy: "test-updated-by",
|
||||
expectedSlug: "test-slug",
|
||||
expectedOrigin: &entityStore.EntityOriginInfo{Source: "", Key: ""},
|
||||
expectedLabels: map[string]string{"label1": "value1", "label2": "value2"},
|
||||
expectedMeta: []byte(fmt.Sprintf(`{"metadata":{"name":"test-name","uid":"test-uid","resourceVersion":"1","creationTimestamp":%q,"labels":{"label1":"value1","label2":"value2"},"annotations":{"grafana.app/createdBy":"test-created-by","grafana.app/folder":"test-folder","grafana.app/slug":"test-slug","grafana.app/updatedBy":"test-updated-by","grafana.app/updatedTimestamp":%q}}}`, createdAtStr, updatedAtStr)),
|
||||
expectedBody: []byte(fmt.Sprintf(`{"kind":"Playlist","apiVersion":"playlist.grafana.app/v0alpha1","metadata":{"name":"test-name","uid":"test-uid","resourceVersion":"1","creationTimestamp":%q,"labels":{"label1":"value1","label2":"value2"},"annotations":{"grafana.app/createdBy":"test-created-by","grafana.app/folder":"test-folder","grafana.app/slug":"test-slug","grafana.app/updatedBy":"test-updated-by","grafana.app/updatedTimestamp":%q}},"spec":{"title":"A playlist","interval":"5m","items":[{"type":"dashboard_by_tag","value":"panel-tests"},{"type":"dashboard_by_uid","value":"vmie2cmWz"}]}}`, createdAtStr, updatedAtStr)),
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.resource.GetObjectKind().GroupVersionKind().Kind+" to entity conversion should succeed", func(t *testing.T) {
|
||||
entity, err := resourceToEntity(tc.resource, tc.key, Codecs.LegacyCodec(v0alpha1.PlaylistResourceInfo.GroupVersion()))
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, tc.expectedKey, entity.Key)
|
||||
assert.Equal(t, tc.expectedName, entity.Name)
|
||||
assert.Equal(t, tc.expectedNamespace, entity.Namespace)
|
||||
assert.Equal(t, tc.expectedTitle, entity.Title)
|
||||
assert.Equal(t, tc.expectedGroupVersion, entity.GroupVersion)
|
||||
assert.Equal(t, tc.expectedName, entity.Name)
|
||||
assert.Equal(t, tc.expectedGuid, entity.Guid)
|
||||
assert.Equal(t, tc.expectedFolder, entity.Folder)
|
||||
assert.Equal(t, tc.expectedCreatedAt, entity.CreatedAt)
|
||||
assert.Equal(t, tc.expectedUpdatedAt, entity.UpdatedAt)
|
||||
assert.Equal(t, tc.expectedCreatedBy, entity.CreatedBy)
|
||||
assert.Equal(t, tc.expectedUpdatedBy, entity.UpdatedBy)
|
||||
assert.Equal(t, tc.expectedSlug, entity.Slug)
|
||||
assert.Equal(t, tc.expectedOrigin, entity.Origin)
|
||||
assert.Equal(t, tc.expectedLabels, entity.Labels)
|
||||
assert.Equal(t, tc.expectedMeta, entity.Meta)
|
||||
assert.Equal(t, tc.expectedBody, entity.Body[:len(entity.Body)-1]) // remove trailing newline
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestEntityToResource(t *testing.T) {
|
||||
createdAt := metav1.Now()
|
||||
createdAtStr := createdAt.UTC().Format(time.RFC3339)
|
||||
|
||||
updatedAt := createdAt.Add(time.Hour)
|
||||
updatedAtStr := updatedAt.UTC().Format(time.RFC3339)
|
||||
|
||||
Scheme := runtime.NewScheme()
|
||||
Scheme.AddKnownTypes(v0alpha1.PlaylistResourceInfo.GroupVersion(), &v0alpha1.Playlist{})
|
||||
Codecs := serializer.NewCodecFactory(Scheme)
|
||||
|
||||
testCases := []struct {
|
||||
entity *entityStore.Entity
|
||||
codec runtime.Codec
|
||||
expectedApiVersion string
|
||||
expectedCreationTimestamp metav1.Time
|
||||
expectedLabels map[string]string
|
||||
expectedName string
|
||||
expectedResourceVersion string
|
||||
expectedUid string
|
||||
expectedTitle string
|
||||
expectedAnnotations map[string]string
|
||||
expectedSpec any
|
||||
}{
|
||||
{
|
||||
entity: &entityStore.Entity{
|
||||
Key: "/group/playlist.grafana.app/resource/playlists/namespaces/default/name/test-uid",
|
||||
GroupVersion: "v0alpha1",
|
||||
Name: "test-uid",
|
||||
Title: "A playlist",
|
||||
Guid: "test-guid",
|
||||
Folder: "test-folder",
|
||||
CreatedBy: "test-created-by",
|
||||
CreatedAt: createdAt.UnixMilli(),
|
||||
UpdatedAt: updatedAt.UnixMilli(),
|
||||
UpdatedBy: "test-updated-by",
|
||||
Slug: "test-slug",
|
||||
Origin: &entityStore.EntityOriginInfo{},
|
||||
Labels: map[string]string{"label1": "value1", "label2": "value2"},
|
||||
Meta: []byte(fmt.Sprintf(`{"metadata":{"name":"test-name","uid":"test-uid","resourceVersion":"1","creationTimestamp":%q,"labels":{"label1":"value1","label2":"value2"},"annotations":{"grafana.app/createdBy":"test-created-by","grafana.app/folder":"test-folder","grafana.app/slug":"test-slug","grafana.app/updatedTimestamp":%q,"grafana.app/updatedBy":"test-updated-by"}}}`, createdAtStr, updatedAtStr)),
|
||||
Body: []byte(fmt.Sprintf(`{"kind":"Playlist","apiVersion":"playlist.grafana.app/v0alpha1","metadata":{"name":"test-name","uid":"test-uid","resourceVersion":"1","creationTimestamp":%q,"labels":{"label1":"value1","label2":"value2"},"annotations":{"grafana.app/createdBy":"test-created-by","grafana.app/folder":"test-folder","grafana.app/slug":"test-slug","grafana.app/updatedBy":"test-updated-by","grafana.app/updatedTimestamp":%q}},"spec":{"title":"A playlist","interval":"5m","items":[{"type":"dashboard_by_tag","value":"panel-tests"},{"type":"dashboard_by_uid","value":"vmie2cmWz"}]}}`, createdAtStr, updatedAtStr)),
|
||||
ResourceVersion: 1,
|
||||
Action: entityStore.Entity_CREATED,
|
||||
},
|
||||
codec: runtime.Codec(nil),
|
||||
expectedApiVersion: "playlist.grafana.app/v0alpha1",
|
||||
expectedCreationTimestamp: createdAt,
|
||||
expectedLabels: map[string]string{"label1": "value1", "label2": "value2"},
|
||||
expectedName: "test-uid",
|
||||
expectedTitle: "test-name",
|
||||
expectedResourceVersion: "1",
|
||||
expectedUid: "test-guid",
|
||||
expectedAnnotations: map[string]string{
|
||||
"grafana.app/createdBy": "test-created-by",
|
||||
"grafana.app/folder": "test-folder",
|
||||
"grafana.app/slug": "test-slug",
|
||||
"grafana.app/updatedBy": "test-updated-by",
|
||||
"grafana.app/updatedTimestamp": updatedAtStr,
|
||||
},
|
||||
expectedSpec: v0alpha1.Spec{
|
||||
Title: "A playlist",
|
||||
Interval: "5m",
|
||||
Items: []v0alpha1.Item{
|
||||
{Type: v0alpha1.ItemTypeDashboardByTag, Value: "panel-tests"},
|
||||
{Type: v0alpha1.ItemTypeDashboardByUid, Value: "vmie2cmWz"},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.entity.Key+" to resource conversion should succeed", func(t *testing.T) {
|
||||
var p v0alpha1.Playlist
|
||||
err := EntityToRuntimeObject(tc.entity, &p, Codecs.LegacyCodec(v0alpha1.PlaylistResourceInfo.GroupVersion()))
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, tc.expectedApiVersion, p.TypeMeta.APIVersion)
|
||||
assert.Equal(t, tc.expectedCreationTimestamp.Unix(), p.ObjectMeta.CreationTimestamp.Unix())
|
||||
assert.Equal(t, tc.expectedLabels, p.ObjectMeta.Labels)
|
||||
assert.Equal(t, tc.expectedName, p.ObjectMeta.Name)
|
||||
assert.Equal(t, tc.expectedResourceVersion, p.ObjectMeta.ResourceVersion)
|
||||
assert.Equal(t, tc.expectedUid, string(p.ObjectMeta.UID))
|
||||
assert.Equal(t, tc.expectedAnnotations, p.ObjectMeta.Annotations)
|
||||
assert.Equal(t, tc.expectedSpec, p.Spec)
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user