K8s/DualWriter: Use dualwriter package for all dual writers (#101393)
This commit is contained in:
Ryan McKinley
@@ -8,8 +8,6 @@ import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apiserver/pkg/registry/rest"
|
||||
@@ -101,36 +99,6 @@ const (
|
||||
Mode5
|
||||
)
|
||||
|
||||
// TODO: make this function private as there should only be one public way of setting the dual writing mode
|
||||
// NewDualWriter returns a new DualWriter.
|
||||
func NewDualWriter(
|
||||
mode DualWriterMode,
|
||||
legacy Storage,
|
||||
unified Storage,
|
||||
reg prometheus.Registerer,
|
||||
resource string,
|
||||
) Storage {
|
||||
metrics := &dualWriterMetrics{}
|
||||
metrics.init(reg)
|
||||
switch mode {
|
||||
case Mode0:
|
||||
return legacy
|
||||
case Mode1:
|
||||
// read and write only from legacy storage
|
||||
return newDualWriterMode1(legacy, unified, metrics, resource)
|
||||
case Mode2:
|
||||
// write to both, read from storage but use legacy as backup
|
||||
return newDualWriterMode2(legacy, unified, metrics, resource)
|
||||
case Mode3:
|
||||
// write to both, read from storage only
|
||||
return newDualWriterMode3(legacy, unified, metrics, resource)
|
||||
case Mode4, Mode5:
|
||||
return unified
|
||||
default:
|
||||
return newDualWriterMode1(legacy, unified, metrics, resource)
|
||||
}
|
||||
}
|
||||
|
||||
type NamespacedKVStore interface {
|
||||
Get(ctx context.Context, key string) (string, bool, error)
|
||||
Set(ctx context.Context, key, value string) error
|
||||
@@ -254,15 +222,3 @@ func extractSpec(obj runtime.Object) []byte {
|
||||
}
|
||||
return jsonObj
|
||||
}
|
||||
|
||||
func getName(o runtime.Object) string {
|
||||
if o == nil {
|
||||
return ""
|
||||
}
|
||||
accessor, err := meta.Accessor(o)
|
||||
if err != nil {
|
||||
klog.Error("failed to get object name: ", err)
|
||||
return ""
|
||||
}
|
||||
return accessor.GetName()
|
||||
}
|
||||
|
||||
@@ -1,346 +0,0 @@
|
||||
package rest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apiserver/pkg/registry/rest"
|
||||
"k8s.io/klog/v2"
|
||||
)
|
||||
|
||||
type DualWriterMode1 struct {
|
||||
Legacy Storage
|
||||
Storage Storage
|
||||
*dualWriterMetrics
|
||||
resource string
|
||||
Log klog.Logger
|
||||
}
|
||||
|
||||
const mode1Str = "1"
|
||||
|
||||
// NewDualWriterMode1 returns a new DualWriter in mode 1.
|
||||
// Mode 1 represents writing to and reading from LegacyStorage.
|
||||
func newDualWriterMode1(legacy Storage, storage Storage, dwm *dualWriterMetrics, resource string) *DualWriterMode1 {
|
||||
return &DualWriterMode1{
|
||||
Legacy: legacy,
|
||||
Storage: storage,
|
||||
Log: klog.NewKlogr().WithName("DualWriterMode1").WithValues("mode", mode1Str, "resource", resource),
|
||||
dualWriterMetrics: dwm,
|
||||
resource: resource,
|
||||
}
|
||||
}
|
||||
|
||||
// Mode returns the mode of the dual writer.
|
||||
func (d *DualWriterMode1) Mode() DualWriterMode {
|
||||
return Mode1
|
||||
}
|
||||
|
||||
// Create overrides the behavior of the generic DualWriter and writes only to LegacyStorage.
|
||||
func (d *DualWriterMode1) Create(ctx context.Context, in runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
|
||||
var method = "create"
|
||||
log := d.Log.WithValues("method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
accIn, err := meta.Accessor(in)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if accIn.GetUID() != "" {
|
||||
return nil, fmt.Errorf("UID should not be present:: %v", accIn.GetUID())
|
||||
}
|
||||
|
||||
startLegacy := time.Now()
|
||||
created, err := d.Legacy.Create(ctx, in, createValidation, options)
|
||||
d.recordLegacyDuration(err != nil, mode1Str, d.resource, method, startLegacy)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to create object in legacy storage")
|
||||
return created, err
|
||||
}
|
||||
|
||||
createdCopy := created.DeepCopyObject()
|
||||
|
||||
//nolint:errcheck
|
||||
go d.createOnUnifiedStorage(ctx, createValidation, createdCopy, options)
|
||||
|
||||
return created, err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) createOnUnifiedStorage(ctx context.Context, createValidation rest.ValidateObjectFunc, createdCopy runtime.Object, options *metav1.CreateOptions) error {
|
||||
var method = "create"
|
||||
log := d.Log.WithValues("method", method)
|
||||
|
||||
// Ignores cancellation signals from parent context. Will automatically be canceled after 10 seconds.
|
||||
ctx, cancel := context.WithTimeoutCause(context.WithoutCancel(ctx), time.Second*10, errors.New("storage create timeout"))
|
||||
defer cancel()
|
||||
|
||||
accCreated, err := meta.Accessor(createdCopy)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
accCreated.SetResourceVersion("")
|
||||
|
||||
startStorage := time.Now()
|
||||
storageObj, errObjectSt := d.Storage.Create(ctx, createdCopy, createValidation, options)
|
||||
d.recordStorageDuration(errObjectSt != nil, mode1Str, d.resource, method, startStorage)
|
||||
if errObjectSt != nil {
|
||||
log.Error(errObjectSt, "unable to create object in storage")
|
||||
cancel()
|
||||
}
|
||||
areEqual := Compare(storageObj, createdCopy)
|
||||
d.recordOutcome(mode1Str, getName(createdCopy), areEqual, method)
|
||||
if !areEqual {
|
||||
log.Info("object from legacy and storage are not equal")
|
||||
}
|
||||
|
||||
return errObjectSt
|
||||
}
|
||||
|
||||
// Get overrides the behavior of the generic DualWriter and reads only from LegacyStorage.
|
||||
func (d *DualWriterMode1) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
|
||||
var method = "get"
|
||||
log := d.Log.WithValues("method", method, "name", name)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
startLegacy := time.Now()
|
||||
res, errLegacy := d.Legacy.Get(ctx, name, options)
|
||||
if errLegacy != nil {
|
||||
log.Error(errLegacy, "unable to get object in legacy storage")
|
||||
}
|
||||
d.recordLegacyDuration(errLegacy != nil, mode1Str, d.resource, method, startLegacy)
|
||||
|
||||
//nolint:errcheck
|
||||
go d.getFromUnifiedStorage(ctx, res, name, options)
|
||||
|
||||
return res, errLegacy
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) getFromUnifiedStorage(ctx context.Context, objFromLegacy runtime.Object, name string, options *metav1.GetOptions) error {
|
||||
var method = "get"
|
||||
log := d.Log.WithValues("method", method, "name", name)
|
||||
|
||||
startStorage := time.Now()
|
||||
// Ignores cancellation signals from parent context. Will automatically be canceled after 10 seconds.
|
||||
ctx, cancel := context.WithTimeoutCause(context.WithoutCancel(ctx), time.Second*10, errors.New("storage get timeout"))
|
||||
defer cancel()
|
||||
storageObj, err := d.Storage.Get(ctx, name, options)
|
||||
d.recordStorageDuration(err != nil, mode1Str, d.resource, method, startStorage)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to get object in storage")
|
||||
cancel()
|
||||
}
|
||||
|
||||
areEqual := Compare(storageObj, objFromLegacy)
|
||||
d.recordOutcome(mode1Str, name, areEqual, method)
|
||||
if !areEqual {
|
||||
log.WithValues("name", name).Info("object from legacy and storage are not equal")
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// List overrides the behavior of the generic DualWriter and reads only from LegacyStorage.
|
||||
func (d *DualWriterMode1) List(ctx context.Context, options *metainternalversion.ListOptions) (runtime.Object, error) {
|
||||
var method = "list"
|
||||
log := d.Log.WithValues("resourceVersion", options.ResourceVersion, "method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
startLegacy := time.Now()
|
||||
res, err := d.Legacy.List(ctx, options)
|
||||
d.recordLegacyDuration(err != nil, mode1Str, d.resource, method, startLegacy)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to list object in legacy storage")
|
||||
}
|
||||
|
||||
//nolint:errcheck
|
||||
go d.listFromUnifiedStorage(ctx, options, res)
|
||||
|
||||
return res, err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) listFromUnifiedStorage(ctx context.Context, options *metainternalversion.ListOptions, objFromLegacy runtime.Object) error {
|
||||
var method = "list"
|
||||
log := d.Log.WithValues("resourceVersion", options.ResourceVersion, "method", method)
|
||||
|
||||
startStorage := time.Now()
|
||||
// Ignores cancellation signals from parent context. Will automatically be canceled after 10 seconds.
|
||||
ctx, cancel := context.WithTimeoutCause(context.WithoutCancel(ctx), time.Second*10, errors.New("storage list timeout"))
|
||||
defer cancel()
|
||||
|
||||
storageObj, err := d.Storage.List(ctx, options)
|
||||
d.recordStorageDuration(err != nil, mode1Str, d.resource, method, startStorage)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to list objects from unified storage")
|
||||
cancel()
|
||||
}
|
||||
areEqual := Compare(storageObj, objFromLegacy)
|
||||
d.recordOutcome(mode1Str, getName(objFromLegacy), areEqual, method)
|
||||
if !areEqual {
|
||||
log.Info("object from legacy and storage are not equal")
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
|
||||
var method = "delete"
|
||||
log := d.Log.WithValues("name", name, "method", method, "name", name)
|
||||
ctx = klog.NewContext(ctx, d.Log)
|
||||
|
||||
startLegacy := time.Now()
|
||||
res, async, err := d.Legacy.Delete(ctx, name, deleteValidation, options)
|
||||
d.recordLegacyDuration(err != nil, mode1Str, name, method, startLegacy)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to delete object in legacy storage")
|
||||
return res, async, err
|
||||
}
|
||||
|
||||
//nolint:errcheck
|
||||
go d.deleteFromUnifiedStorage(ctx, res, name, deleteValidation, options)
|
||||
|
||||
return res, async, err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) deleteFromUnifiedStorage(ctx context.Context, res runtime.Object, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) error {
|
||||
var method = "delete"
|
||||
log := d.Log.WithValues("name", name, "method", method, "name", name)
|
||||
|
||||
startStorage := time.Now()
|
||||
// Ignores cancellation signals from parent context. Will automatically be canceled after 10 seconds.
|
||||
ctx, cancel := context.WithTimeoutCause(context.WithoutCancel(ctx), time.Second*10, errors.New("storage delete timeout"))
|
||||
defer cancel()
|
||||
storageObj, _, err := d.Storage.Delete(ctx, name, deleteValidation, options)
|
||||
d.recordStorageDuration(err != nil, mode1Str, d.resource, method, startStorage)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to delete object from unified storage")
|
||||
cancel()
|
||||
}
|
||||
areEqual := Compare(storageObj, res)
|
||||
d.recordOutcome(mode1Str, name, areEqual, method)
|
||||
if !areEqual {
|
||||
log.Info("object from legacy and storage are not equal")
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// DeleteCollection overrides the behavior of the generic DualWriter and deletes only from LegacyStorage.
|
||||
func (d *DualWriterMode1) DeleteCollection(ctx context.Context, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions, listOptions *metainternalversion.ListOptions) (runtime.Object, error) {
|
||||
var method = "delete-collection"
|
||||
log := d.Log.WithValues("resourceVersion", listOptions.ResourceVersion, "method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
startLegacy := time.Now()
|
||||
res, err := d.Legacy.DeleteCollection(ctx, deleteValidation, options, listOptions)
|
||||
d.recordLegacyDuration(err != nil, mode1Str, d.resource, method, startLegacy)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to delete collection in legacy storage")
|
||||
return res, err
|
||||
}
|
||||
|
||||
//nolint:errcheck
|
||||
go d.deleteCollectionFromUnifiedStorage(ctx, res, deleteValidation, options, listOptions)
|
||||
|
||||
return res, err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) deleteCollectionFromUnifiedStorage(ctx context.Context, res runtime.Object, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions, listOptions *metainternalversion.ListOptions) error {
|
||||
var method = "delete-collection"
|
||||
log := d.Log.WithValues("resourceVersion", listOptions.ResourceVersion, "method", method)
|
||||
|
||||
startStorage := time.Now()
|
||||
// Ignores cancellation signals from parent context. Will automatically be canceled after 10 seconds.
|
||||
ctx, cancel := context.WithTimeoutCause(context.WithoutCancel(ctx), time.Second*10, errors.New("storage deletecollection timeout"))
|
||||
defer cancel()
|
||||
storageObj, err := d.Storage.DeleteCollection(ctx, deleteValidation, options, listOptions)
|
||||
d.recordStorageDuration(err != nil, mode1Str, d.resource, method, startStorage)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to delete collection object from unified storage")
|
||||
cancel()
|
||||
}
|
||||
areEqual := Compare(storageObj, res)
|
||||
d.recordOutcome(mode1Str, getName(res), areEqual, method)
|
||||
if !areEqual {
|
||||
log.Info("object from legacy and storage are not equal")
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) Update(ctx context.Context, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error) {
|
||||
var method = "update"
|
||||
log := d.Log.WithValues("name", name, "method", method, "name", name)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
startLegacy := time.Now()
|
||||
objLegacy, async, err := d.Legacy.Update(ctx, name, objInfo, createValidation, updateValidation, forceAllowCreate, options)
|
||||
d.recordLegacyDuration(err != nil, mode1Str, d.resource, method, startLegacy)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to update in legacy storage")
|
||||
return objLegacy, async, err
|
||||
}
|
||||
|
||||
//nolint:errcheck
|
||||
go d.updateOnUnifiedStorageMode1(ctx, objLegacy, name, objInfo, createValidation, updateValidation, forceAllowCreate, options)
|
||||
|
||||
return objLegacy, async, err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) updateOnUnifiedStorageMode1(ctx context.Context, objLegacy runtime.Object, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) error {
|
||||
// The incoming RV is from legacy storage, so we can ignore it
|
||||
ctx = context.WithValue(ctx, dualWriteContextKey{}, true)
|
||||
|
||||
var method = "update"
|
||||
log := d.Log.WithValues("name", name, "method", method, "name", name)
|
||||
|
||||
// Ignores cancellation signals from parent context. Will automatically be canceled after 10 seconds.
|
||||
ctx, cancel := context.WithTimeoutCause(context.WithoutCancel(ctx), time.Second*10, errors.New("storage update timeout"))
|
||||
|
||||
startStorage := time.Now()
|
||||
defer cancel()
|
||||
storageObj, _, err := d.Storage.Update(ctx, name, objInfo, createValidation, updateValidation, forceAllowCreate, options)
|
||||
d.recordStorageDuration(err != nil, mode1Str, d.resource, method, startStorage)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to update object from unified storage")
|
||||
cancel()
|
||||
}
|
||||
areEqual := Compare(storageObj, objLegacy)
|
||||
d.recordOutcome(mode1Str, name, areEqual, method)
|
||||
if !areEqual {
|
||||
log.WithValues("name", name).Info("object from legacy and storage are not equal")
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) Destroy() {
|
||||
d.Storage.Destroy()
|
||||
d.Legacy.Destroy()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) GetSingularName() string {
|
||||
return d.Legacy.GetSingularName()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) NamespaceScoped() bool {
|
||||
return d.Legacy.NamespaceScoped()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) New() runtime.Object {
|
||||
return d.Legacy.New()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) NewList() runtime.Object {
|
||||
return d.Storage.NewList()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode1) ConvertToTable(ctx context.Context, object runtime.Object, tableOptions runtime.Object) (*metav1.Table, error) {
|
||||
return d.Legacy.ConvertToTable(ctx, object, tableOptions)
|
||||
}
|
||||
@@ -1,309 +0,0 @@
|
||||
package rest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apiserver/pkg/registry/rest"
|
||||
"k8s.io/klog/v2"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||
)
|
||||
|
||||
type DualWriterMode2 struct {
|
||||
Storage Storage
|
||||
Legacy Storage
|
||||
*dualWriterMetrics
|
||||
resource string
|
||||
Log klog.Logger
|
||||
}
|
||||
|
||||
const mode2Str = "2"
|
||||
|
||||
// newDualWriterMode2 returns a new DualWriter in mode 2.
|
||||
// Mode 2 represents writing to LegacyStorage first, then to Storage.
|
||||
// When reading, values from LegacyStorage will be returned.
|
||||
func newDualWriterMode2(legacy Storage, storage Storage, dwm *dualWriterMetrics, resource string) *DualWriterMode2 {
|
||||
return &DualWriterMode2{
|
||||
Legacy: legacy,
|
||||
Storage: storage,
|
||||
Log: klog.NewKlogr().WithName("DualWriterMode2").WithValues("mode", mode2Str, "resource", resource),
|
||||
dualWriterMetrics: dwm,
|
||||
resource: resource,
|
||||
}
|
||||
}
|
||||
|
||||
// Mode returns the mode of the dual writer.
|
||||
func (d *DualWriterMode2) Mode() DualWriterMode {
|
||||
return Mode2
|
||||
}
|
||||
|
||||
// Create overrides the behavior of the generic DualWriter and writes to LegacyStorage and Storage.
|
||||
func (d *DualWriterMode2) Create(ctx context.Context, in runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
|
||||
var method = "create"
|
||||
log := d.Log.WithValues("method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
accIn, err := meta.Accessor(in)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if accIn.GetUID() != "" {
|
||||
return nil, fmt.Errorf("UID should be empty: %v", accIn.GetUID())
|
||||
}
|
||||
|
||||
startLegacy := time.Now()
|
||||
createdFromLegacy, err := d.Legacy.Create(ctx, in, createValidation, options)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to create object in legacy storage")
|
||||
d.recordLegacyDuration(true, mode2Str, d.resource, method, startLegacy)
|
||||
return nil, err
|
||||
}
|
||||
d.recordLegacyDuration(false, mode2Str, d.resource, method, startLegacy)
|
||||
|
||||
createdCopy := createdFromLegacy.DeepCopyObject()
|
||||
|
||||
accCreated, err := meta.Accessor(createdCopy)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
accCreated.SetResourceVersion("")
|
||||
|
||||
startStorage := time.Now()
|
||||
createdFromStorage, err := d.Storage.Create(ctx, createdCopy, createValidation, options)
|
||||
if err != nil {
|
||||
log.WithValues("name").Error(err, "unable to create object in storage")
|
||||
d.recordStorageDuration(true, mode2Str, d.resource, method, startStorage)
|
||||
return createdFromStorage, err
|
||||
}
|
||||
d.recordStorageDuration(false, mode2Str, d.resource, method, startStorage)
|
||||
|
||||
go func() {
|
||||
areEqual := Compare(createdFromStorage, createdFromLegacy)
|
||||
d.recordOutcome(mode2Str, getName(createdFromStorage), areEqual, method)
|
||||
if !areEqual {
|
||||
log.Info("object from legacy and storage are not equal")
|
||||
}
|
||||
}()
|
||||
|
||||
return createdFromLegacy, err
|
||||
}
|
||||
|
||||
// Get retrieves an object from Storage if possible, and if not it falls back to LegacyStorage.
|
||||
func (d *DualWriterMode2) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
|
||||
var method = "get"
|
||||
log := d.Log.WithValues("name", name, "resourceVersion", options.ResourceVersion, "method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
startLegacy := time.Now()
|
||||
objLegacy, err := d.Legacy.Get(ctx, name, options)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to fetch object from legacy")
|
||||
d.recordLegacyDuration(true, mode2Str, d.resource, method, startLegacy)
|
||||
return nil, err
|
||||
}
|
||||
d.recordLegacyDuration(false, mode2Str, d.resource, method, startLegacy)
|
||||
|
||||
startStorage := time.Now()
|
||||
objStorage, err := d.Storage.Get(ctx, name, options)
|
||||
d.recordStorageDuration(err != nil, mode2Str, d.resource, method, startStorage)
|
||||
if err != nil {
|
||||
if !apierrors.IsNotFound(err) {
|
||||
log.Error(err, "unable to fetch object from storage")
|
||||
return nil, err
|
||||
}
|
||||
log.Info("object not found in storage, dual write or migration didn't happen yet")
|
||||
}
|
||||
|
||||
go func() {
|
||||
areEqual := Compare(objStorage, objLegacy)
|
||||
d.recordOutcome(mode2Str, name, areEqual, method)
|
||||
if !areEqual {
|
||||
log.Info("object from legacy and storage are not equal")
|
||||
}
|
||||
}()
|
||||
|
||||
return objLegacy, nil
|
||||
}
|
||||
|
||||
// List overrides the behavior of the generic DualWriter.
|
||||
func (d *DualWriterMode2) List(ctx context.Context, options *metainternalversion.ListOptions) (runtime.Object, error) {
|
||||
var method = "list"
|
||||
log := d.Log.WithValues("resourceVersion", options.ResourceVersion, "method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
startLegacy := time.Now()
|
||||
ll, err := d.Legacy.List(ctx, options)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to list objects from legacy storage")
|
||||
d.recordLegacyDuration(true, mode2Str, d.resource, method, startLegacy)
|
||||
return ll, err
|
||||
}
|
||||
d.recordLegacyDuration(false, mode2Str, d.resource, method, startLegacy)
|
||||
|
||||
// Even if we don't compare, we want to fetch from unified storage and check that it doesn't error.
|
||||
startStorage := time.Now()
|
||||
if _, err := d.Storage.List(ctx, options); err != nil {
|
||||
log.Error(err, "unable to list objects from storage")
|
||||
d.recordStorageDuration(true, mode2Str, d.resource, method, startStorage)
|
||||
return nil, err
|
||||
}
|
||||
d.recordStorageDuration(false, mode2Str, d.resource, method, startStorage)
|
||||
|
||||
return ll, nil
|
||||
}
|
||||
|
||||
// DeleteCollection overrides the behavior of the generic DualWriter and deletes from both LegacyStorage and Storage.
|
||||
func (d *DualWriterMode2) DeleteCollection(ctx context.Context, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions, listOptions *metainternalversion.ListOptions) (runtime.Object, error) {
|
||||
var method = "delete-collection"
|
||||
log := d.Log.WithValues("resourceVersion", listOptions.ResourceVersion, "method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
startLegacy := time.Now()
|
||||
deletedLegacy, err := d.Legacy.DeleteCollection(ctx, deleteValidation, options, listOptions)
|
||||
if err != nil {
|
||||
log.WithValues("deleted", deletedLegacy).Error(err, "failed to delete collection successfully from legacy storage")
|
||||
d.recordLegacyDuration(true, mode2Str, d.resource, method, startLegacy)
|
||||
return nil, err
|
||||
}
|
||||
d.recordLegacyDuration(false, mode2Str, d.resource, method, startLegacy)
|
||||
|
||||
startStorage := time.Now()
|
||||
deletedStorage, err := d.Storage.DeleteCollection(ctx, deleteValidation, options, listOptions)
|
||||
if err != nil {
|
||||
log.WithValues("deleted", deletedStorage).Error(err, "failed to delete collection successfully from Storage")
|
||||
d.recordStorageDuration(true, mode2Str, d.resource, method, startStorage)
|
||||
return nil, err
|
||||
}
|
||||
d.recordStorageDuration(false, mode2Str, d.resource, method, startStorage)
|
||||
|
||||
go func() {
|
||||
areEqual := Compare(deletedStorage, deletedLegacy)
|
||||
d.recordOutcome(mode2Str, getName(deletedStorage), areEqual, method)
|
||||
if !areEqual {
|
||||
log.Info("object from legacy and storage are not equal")
|
||||
}
|
||||
}()
|
||||
|
||||
return deletedLegacy, err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode2) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
|
||||
var method = "delete"
|
||||
log := d.Log.WithValues("name", name, "method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
// We should delete from Unified storage first so we can retry if legacy fails.
|
||||
startStorage := time.Now()
|
||||
deletedS, _, err := d.Storage.Delete(ctx, name, deleteValidation, options)
|
||||
d.recordStorageDuration(err != nil, mode2Str, d.resource, method, startStorage)
|
||||
if err != nil {
|
||||
if !apierrors.IsNotFound(err) {
|
||||
log.WithValues("objectList", deletedS).Error(err, "could not delete from unified storage")
|
||||
return nil, false, err
|
||||
}
|
||||
}
|
||||
|
||||
startLegacy := time.Now()
|
||||
deletedLS, async, err := d.Legacy.Delete(ctx, name, deleteValidation, options)
|
||||
d.recordLegacyDuration(err != nil, mode2Str, d.resource, method, startLegacy)
|
||||
// Deleting from legacy should always work in mode two, as legacy is still the primary database and
|
||||
// needs to have all the data.
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
|
||||
go func() {
|
||||
areEqual := Compare(deletedS, deletedLS)
|
||||
d.recordOutcome(mode2Str, name, areEqual, method)
|
||||
if !areEqual {
|
||||
log.WithValues("name", name).Info("object from legacy and storage are not equal")
|
||||
}
|
||||
}()
|
||||
|
||||
return deletedLS, async, err
|
||||
}
|
||||
|
||||
// Update overrides the generic behavior of the Storage and writes first to the legacy storage and then to storage.
|
||||
func (d *DualWriterMode2) Update(ctx context.Context, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error) {
|
||||
var method = "update"
|
||||
log := d.Log.WithValues("name", name, "method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
// The incoming RV is not stable -- it may be from legacy or storage!
|
||||
// This sets a flag in the context and our apistore is more lenient when it exists
|
||||
ctx = context.WithValue(ctx, dualWriteContextKey{}, true)
|
||||
|
||||
startLegacy := time.Now()
|
||||
objFromLegacy, created, err := d.Legacy.Update(ctx, name, objInfo, createValidation, updateValidation, forceAllowCreate, options)
|
||||
if err != nil {
|
||||
log.WithValues("object", objFromLegacy).Error(err, "could not update in legacy storage")
|
||||
d.recordLegacyDuration(true, mode2Str, d.resource, "update", startLegacy)
|
||||
return objFromLegacy, created, err
|
||||
}
|
||||
d.recordLegacyDuration(false, mode2Str, d.resource, "update", startLegacy)
|
||||
|
||||
startStorage := time.Now()
|
||||
objFromStorage, created, err := d.Storage.Update(ctx, name, objInfo, createValidation, updateValidation, forceAllowCreate, options)
|
||||
if err != nil {
|
||||
log.WithValues("object", objFromStorage).Error(err, "could not update in storage")
|
||||
d.recordStorageDuration(true, mode2Str, d.resource, "update", startStorage)
|
||||
return objFromStorage, created, err
|
||||
}
|
||||
|
||||
go func() {
|
||||
areEqual := Compare(objFromStorage, objFromLegacy)
|
||||
d.recordOutcome(mode2Str, name, areEqual, method)
|
||||
if !areEqual {
|
||||
log.WithValues("name", name).Info("object from legacy and storage are not equal")
|
||||
}
|
||||
}()
|
||||
|
||||
return objFromLegacy, created, err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode2) Destroy() {
|
||||
d.Storage.Destroy()
|
||||
d.Legacy.Destroy()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode2) GetSingularName() string {
|
||||
return d.Storage.GetSingularName()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode2) NamespaceScoped() bool {
|
||||
return d.Storage.NamespaceScoped()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode2) New() runtime.Object {
|
||||
return d.Storage.New()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode2) NewList() runtime.Object {
|
||||
return d.Storage.NewList()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode2) ConvertToTable(ctx context.Context, object runtime.Object, tableOptions runtime.Object) (*metav1.Table, error) {
|
||||
return d.Storage.ConvertToTable(ctx, object, tableOptions)
|
||||
}
|
||||
|
||||
func parseList(legacyList []runtime.Object) (map[string]int, error) {
|
||||
indexMap := map[string]int{}
|
||||
|
||||
for i, obj := range legacyList {
|
||||
accessor, err := utils.MetaAccessor(obj)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
indexMap[accessor.GetName()] = i
|
||||
}
|
||||
return indexMap, nil
|
||||
}
|
||||
@@ -1,327 +0,0 @@
|
||||
package rest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/watch"
|
||||
"k8s.io/apiserver/pkg/registry/rest"
|
||||
"k8s.io/klog/v2"
|
||||
)
|
||||
|
||||
type DualWriterMode3 struct {
|
||||
Legacy Storage
|
||||
Storage Storage
|
||||
watchImp rest.Watcher // watch is only available in mode 3 and 4
|
||||
*dualWriterMetrics
|
||||
resource string
|
||||
Log klog.Logger
|
||||
}
|
||||
|
||||
// newDualWriterMode3 returns a new DualWriter in mode 3.
|
||||
// Mode 3 represents writing to LegacyStorage and Storage and reading from Storage.
|
||||
func newDualWriterMode3(legacy Storage, storage Storage, dwm *dualWriterMetrics, resource string) *DualWriterMode3 {
|
||||
return &DualWriterMode3{
|
||||
Legacy: legacy,
|
||||
Storage: storage,
|
||||
Log: klog.NewKlogr().WithName("DualWriterMode3").WithValues("mode", mode3Str, "resource", resource),
|
||||
dualWriterMetrics: dwm,
|
||||
resource: resource,
|
||||
}
|
||||
}
|
||||
|
||||
// Mode returns the mode of the dual writer.
|
||||
func (d *DualWriterMode3) Mode() DualWriterMode {
|
||||
return Mode3
|
||||
}
|
||||
|
||||
const mode3Str = "3"
|
||||
|
||||
// Create overrides the behavior of the generic DualWriter and writes to LegacyStorage and Storage.
|
||||
func (d *DualWriterMode3) Create(ctx context.Context, in runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
|
||||
var method = "create"
|
||||
log := d.Log.WithValues("method", method)
|
||||
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
accIn, err := meta.Accessor(in)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if accIn.GetUID() != "" {
|
||||
return nil, fmt.Errorf("UID should not be: %v", accIn.GetUID())
|
||||
}
|
||||
|
||||
if accIn.GetName() == "" && accIn.GetGenerateName() == "" {
|
||||
return nil, fmt.Errorf("name or generatename have to be set")
|
||||
}
|
||||
|
||||
// create in legacy first, and then unistore. if unistore fails, but legacy succeeds,
|
||||
// will try to cleanup the object in legacy.
|
||||
|
||||
startLegacy := time.Now()
|
||||
createdFromLegacy, err := d.Legacy.Create(ctx, in, createValidation, options)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to create object in legacy storage")
|
||||
d.recordLegacyDuration(true, mode2Str, d.resource, method, startLegacy)
|
||||
return createdFromLegacy, err
|
||||
}
|
||||
d.recordLegacyDuration(false, mode2Str, d.resource, method, startLegacy)
|
||||
|
||||
createdCopy := createdFromLegacy.DeepCopyObject()
|
||||
accCreated, err := meta.Accessor(createdCopy)
|
||||
if err != nil {
|
||||
return createdFromLegacy, err
|
||||
}
|
||||
accCreated.SetResourceVersion("")
|
||||
|
||||
startStorage := time.Now()
|
||||
storageObj, errObjectSt := d.Storage.Create(ctx, createdCopy, createValidation, options)
|
||||
d.recordStorageDuration(errObjectSt != nil, mode3Str, d.resource, method, startStorage)
|
||||
if errObjectSt != nil {
|
||||
log.Error(err, "unable to create object in storage")
|
||||
|
||||
// if we cannot create in unistore, attempt to clean up legacy
|
||||
_, _, err = d.Legacy.Delete(ctx, accCreated.GetName(), nil, &metav1.DeleteOptions{})
|
||||
if err != nil {
|
||||
log.Error(err, "unable to cleanup object in legacy storage")
|
||||
}
|
||||
|
||||
return storageObj, errObjectSt
|
||||
}
|
||||
|
||||
areEqual := Compare(createdFromLegacy, storageObj)
|
||||
d.recordOutcome(mode3Str, getName(storageObj), areEqual, method)
|
||||
if !areEqual {
|
||||
log.Info("object from legacy and storage are not equal")
|
||||
}
|
||||
|
||||
return storageObj, errObjectSt
|
||||
}
|
||||
|
||||
// Get overrides the behavior of the generic DualWriter and retrieves an object from Storage.
|
||||
func (d *DualWriterMode3) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
|
||||
var method = "get"
|
||||
log := d.Log.WithValues("name", name, "method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
startStorage := time.Now()
|
||||
storageObj, err := d.Storage.Get(ctx, name, options)
|
||||
d.recordStorageDuration(err != nil, mode3Str, d.resource, method, startStorage)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to get object in storage")
|
||||
return nil, err
|
||||
}
|
||||
|
||||
//nolint:errcheck
|
||||
go d.getFromLegacyStorage(ctx, storageObj, name, options)
|
||||
|
||||
return storageObj, err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode3) getFromLegacyStorage(ctx context.Context, storageObj runtime.Object, name string, options *metav1.GetOptions) error {
|
||||
var method = "get"
|
||||
log := d.Log.WithValues("method", method, "name", name)
|
||||
|
||||
startLegacy := time.Now()
|
||||
// Ignores cancellation signals from parent context. Will automatically be canceled after 10 seconds.
|
||||
ctx, cancel := context.WithTimeoutCause(context.WithoutCancel(ctx), time.Second*10, errors.New("legacy get timeout"))
|
||||
defer cancel()
|
||||
|
||||
objFromLegacy, err := d.Legacy.Get(ctx, name, options)
|
||||
d.recordLegacyDuration(err != nil, mode3Str, d.resource, method, startLegacy)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to get object in legacy storage")
|
||||
cancel()
|
||||
}
|
||||
|
||||
areEqual := Compare(storageObj, objFromLegacy)
|
||||
d.recordOutcome(mode3Str, name, areEqual, method)
|
||||
if !areEqual {
|
||||
log.WithValues("name", name).Info("object from legacy and storage are not equal")
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// List overrides the behavior of the generic DualWriter and reads only from Unified Store.
|
||||
func (d *DualWriterMode3) List(ctx context.Context, options *metainternalversion.ListOptions) (runtime.Object, error) {
|
||||
var method = "list"
|
||||
log := d.Log.WithValues("resourceVersion", options.ResourceVersion, "method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
startStorage := time.Now()
|
||||
objFromStorage, err := d.Storage.List(ctx, options)
|
||||
d.recordStorageDuration(err != nil, mode3Str, d.resource, method, startStorage)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to list object in storage")
|
||||
}
|
||||
return objFromStorage, err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode3) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
|
||||
var method = "delete"
|
||||
log := d.Log.WithValues("name", name, "method", method)
|
||||
ctx = klog.NewContext(ctx, d.Log)
|
||||
|
||||
// delete from legacy first, and then unistore. Will return a failure if either fails,
|
||||
// unless its a 404.
|
||||
//
|
||||
// we want to delete from legacy first, otherwise if the delete from unistore was successful,
|
||||
// but legacy failed, the user would get a failure, but not be able to retry the delete
|
||||
// as they would not be able to see the object in unistore anymore.
|
||||
startLegacy := time.Now()
|
||||
objFromLegacy, asyncLegacy, err := d.Legacy.Delete(ctx, name, deleteValidation, options)
|
||||
d.recordLegacyDuration(err != nil && !apierrors.IsNotFound(err), mode3Str, d.resource, method, startLegacy)
|
||||
if err != nil {
|
||||
if !apierrors.IsNotFound(err) {
|
||||
log.WithValues("object", objFromLegacy).Error(err, "could not delete from legacy store")
|
||||
return objFromLegacy, asyncLegacy, err
|
||||
}
|
||||
}
|
||||
|
||||
startStorage := time.Now()
|
||||
objFromStorage, asyncStorage, err := d.Storage.Delete(ctx, name, deleteValidation, options)
|
||||
d.recordStorageDuration(err != nil && !apierrors.IsNotFound(err), mode3Str, d.resource, method, startStorage)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
|
||||
areEqual := Compare(objFromStorage, objFromLegacy)
|
||||
d.recordOutcome(mode3Str, name, areEqual, method)
|
||||
if !areEqual {
|
||||
log.WithValues("name", name).Info("object from legacy and storage are not equal")
|
||||
}
|
||||
|
||||
return objFromStorage, asyncStorage, err
|
||||
}
|
||||
|
||||
// Update overrides the behavior of the generic DualWriter and writes first to Storage and then to LegacyStorage.
|
||||
func (d *DualWriterMode3) Update(ctx context.Context, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error) {
|
||||
var method = "update"
|
||||
log := d.Log.WithValues("name", name, "method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
// The incoming RV is not stable -- it may be from legacy or storage!
|
||||
// This sets a flag in the context and our apistore is more lenient when it exists
|
||||
ctx = context.WithValue(ctx, dualWriteContextKey{}, true)
|
||||
|
||||
// update in legacy first, and then unistore. Will return a failure if either fails.
|
||||
//
|
||||
// we want to update in legacy first, otherwise if the update from unistore was successful,
|
||||
// but legacy failed, the user would get a failure, but see the update did apply to the source
|
||||
// of truth, and be less likely to retry to save (and get the stores in sync again)
|
||||
|
||||
startLegacy := time.Now()
|
||||
objFromLegacy, createdLegacy, err := d.Legacy.Update(ctx, name, objInfo, createValidation, updateValidation, forceAllowCreate, options)
|
||||
if err != nil {
|
||||
log.WithValues("object", objFromLegacy).Error(err, "could not update in legacy storage")
|
||||
d.recordLegacyDuration(true, mode2Str, d.resource, "update", startLegacy)
|
||||
return objFromLegacy, createdLegacy, err
|
||||
}
|
||||
d.recordLegacyDuration(false, mode2Str, d.resource, "update", startLegacy)
|
||||
|
||||
startStorage := time.Now()
|
||||
objFromStorage, created, err := d.Storage.Update(ctx, name, objInfo, createValidation, updateValidation, forceAllowCreate, options)
|
||||
if err != nil {
|
||||
log.WithValues("object", objFromStorage).Error(err, "could not update in storage")
|
||||
d.recordStorageDuration(true, mode2Str, d.resource, "update", startStorage)
|
||||
return objFromStorage, created, err
|
||||
}
|
||||
|
||||
areEqual := Compare(objFromStorage, objFromLegacy)
|
||||
d.recordOutcome(mode3Str, name, areEqual, method)
|
||||
if !areEqual {
|
||||
log.WithValues("name", name).Info("object from legacy and storage are not equal")
|
||||
}
|
||||
|
||||
return objFromStorage, created, err
|
||||
}
|
||||
|
||||
// DeleteCollection overrides the behavior of the generic DualWriter and deletes from both LegacyStorage and Storage.
|
||||
func (d *DualWriterMode3) DeleteCollection(ctx context.Context, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions, listOptions *metainternalversion.ListOptions) (runtime.Object, error) {
|
||||
var method = "delete-collection"
|
||||
log := d.Log.WithValues("resourceVersion", listOptions.ResourceVersion, "method", method)
|
||||
ctx = klog.NewContext(ctx, log)
|
||||
|
||||
// delete from legacy first, and anything that is successful can be deleted in unistore too.
|
||||
//
|
||||
// we want to delete from legacy first, otherwise if the delete from unistore was successful,
|
||||
// but legacy failed, the user would get a failure, but not be able to retry the delete
|
||||
// as they would not be able to see the object in unistore anymore.
|
||||
|
||||
startLegacy := time.Now()
|
||||
deletedLegacy, err := d.Legacy.DeleteCollection(ctx, deleteValidation, options, listOptions)
|
||||
if err != nil {
|
||||
log.WithValues("deleted", deletedLegacy).Error(err, "failed to delete collection successfully from legacy storage")
|
||||
d.recordLegacyDuration(true, mode3Str, d.resource, method, startLegacy)
|
||||
return deletedLegacy, err
|
||||
}
|
||||
d.recordLegacyDuration(false, mode3Str, d.resource, method, startLegacy)
|
||||
|
||||
legacyList, err := meta.ExtractList(deletedLegacy)
|
||||
if err != nil {
|
||||
log.Error(err, "unable to extract list from legacy storage")
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Only the items deleted by the legacy DeleteCollection call are selected for deletion by Storage.
|
||||
_, err = parseList(legacyList)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
startStorage := time.Now()
|
||||
deletedStorage, err := d.Storage.DeleteCollection(ctx, deleteValidation, options, listOptions)
|
||||
if err != nil {
|
||||
log.WithValues("deleted", deletedStorage).Error(err, "failed to delete collection successfully from Storage")
|
||||
d.recordStorageDuration(true, mode3Str, d.resource, method, startStorage)
|
||||
return deletedStorage, err
|
||||
}
|
||||
d.recordStorageDuration(false, mode3Str, d.resource, method, startStorage)
|
||||
|
||||
areEqual := Compare(deletedStorage, deletedLegacy)
|
||||
d.recordOutcome(mode3Str, getName(deletedLegacy), areEqual, method)
|
||||
if !areEqual {
|
||||
log.Info("object from legacy and storage are not equal")
|
||||
}
|
||||
|
||||
return deletedStorage, err
|
||||
}
|
||||
|
||||
func (d *DualWriterMode3) Watch(ctx context.Context, options *metainternalversion.ListOptions) (watch.Interface, error) {
|
||||
var method = "watch"
|
||||
d.Log.WithValues("method", method, "mode", mode3Str).Info("starting to watch")
|
||||
return d.watchImp.Watch(ctx, options)
|
||||
}
|
||||
|
||||
func (d *DualWriterMode3) Destroy() {
|
||||
d.Storage.Destroy()
|
||||
d.Legacy.Destroy()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode3) GetSingularName() string {
|
||||
return d.Storage.GetSingularName()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode3) NamespaceScoped() bool {
|
||||
return d.Storage.NamespaceScoped()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode3) New() runtime.Object {
|
||||
return d.Storage.New()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode3) NewList() runtime.Object {
|
||||
return d.Storage.NewList()
|
||||
}
|
||||
|
||||
func (d *DualWriterMode3) ConvertToTable(ctx context.Context, object runtime.Object, tableOptions runtime.Object) (*metav1.Table, error) {
|
||||
return d.Storage.ConvertToTable(ctx, object, tableOptions)
|
||||
}
|
||||
@@ -10,43 +10,10 @@ import (
|
||||
)
|
||||
|
||||
type dualWriterMetrics struct {
|
||||
legacy *prometheus.HistogramVec
|
||||
storage *prometheus.HistogramVec
|
||||
outcome *prometheus.HistogramVec
|
||||
syncer *prometheus.HistogramVec
|
||||
syncerOutcome *prometheus.HistogramVec
|
||||
}
|
||||
|
||||
// DualWriterStorageDuration is a metric summary for dual writer storage duration per mode
|
||||
var DualWriterStorageDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
||||
Name: "dual_writer_storage_duration_seconds",
|
||||
Help: "Histogram for the runtime of dual writer storage duration per mode",
|
||||
Namespace: "grafana",
|
||||
NativeHistogramBucketFactor: 1.1,
|
||||
}, []string{"is_error", "mode", "resource", "method"})
|
||||
|
||||
// DualWriterLegacyDuration is a metric summary for dual writer legacy duration per mode
|
||||
var DualWriterLegacyDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
||||
Name: "dual_writer_legacy_duration_seconds",
|
||||
Help: "Histogram for the runtime of dual writer legacy duration per mode",
|
||||
Namespace: "grafana",
|
||||
NativeHistogramBucketFactor: 1.1,
|
||||
}, []string{"is_error", "mode", "resource", "method"})
|
||||
|
||||
// DualWriterOutcome is a metric summary for dual writer outcome comparison between the 2 stores per mode
|
||||
var DualWriterOutcome = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
||||
Name: "dual_writer_outcome",
|
||||
Help: "Histogram for the runtime of dual writer outcome comparison between the 2 stores per mode",
|
||||
Namespace: "grafana",
|
||||
NativeHistogramBucketFactor: 1.1,
|
||||
}, []string{"mode", "name", "method"})
|
||||
|
||||
var DualWriterReadLegacyCounts = prometheus.NewCounterVec(prometheus.CounterOpts{
|
||||
Name: "dual_writer_read_legacy_count",
|
||||
Help: "Histogram for the runtime of dual writer reads from legacy",
|
||||
Namespace: "grafana",
|
||||
}, []string{"resource", "method"})
|
||||
|
||||
// DualWriterSyncerDuration is a metric summary for dual writer sync duration per mode
|
||||
var DualWriterSyncerDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
||||
Name: "dual_writer_data_syncer_duration_seconds",
|
||||
@@ -65,39 +32,15 @@ var DualWriterDataSyncerOutcome = prometheus.NewHistogramVec(prometheus.Histogra
|
||||
|
||||
func (m *dualWriterMetrics) init(reg prometheus.Registerer) {
|
||||
log := klog.NewKlogr()
|
||||
m.legacy = DualWriterLegacyDuration
|
||||
m.storage = DualWriterStorageDuration
|
||||
m.outcome = DualWriterOutcome
|
||||
m.syncer = DualWriterSyncerDuration
|
||||
m.syncerOutcome = DualWriterDataSyncerOutcome
|
||||
errLegacy := reg.Register(m.legacy)
|
||||
errStorage := reg.Register(m.storage)
|
||||
errOutcome := reg.Register(m.outcome)
|
||||
errSyncer := reg.Register(m.syncer)
|
||||
errSyncerOutcome := reg.Register(m.syncerOutcome)
|
||||
if errLegacy != nil || errStorage != nil || errOutcome != nil || errSyncer != nil || errSyncerOutcome != nil {
|
||||
if errSyncer != nil || errSyncerOutcome != nil {
|
||||
log.Info("cloud migration metrics already registered")
|
||||
}
|
||||
}
|
||||
|
||||
func (m *dualWriterMetrics) recordLegacyDuration(isError bool, mode string, resource string, method string, startFrom time.Time) {
|
||||
duration := time.Since(startFrom).Seconds()
|
||||
m.legacy.WithLabelValues(strconv.FormatBool(isError), mode, resource, method).Observe(duration)
|
||||
}
|
||||
|
||||
func (m *dualWriterMetrics) recordStorageDuration(isError bool, mode string, resource string, method string, startFrom time.Time) {
|
||||
duration := time.Since(startFrom).Seconds()
|
||||
m.storage.WithLabelValues(strconv.FormatBool(isError), mode, resource, method).Observe(duration)
|
||||
}
|
||||
|
||||
func (m *dualWriterMetrics) recordOutcome(mode string, name string, areEqual bool, method string) {
|
||||
var observeValue float64
|
||||
if !areEqual {
|
||||
observeValue = 1
|
||||
}
|
||||
m.outcome.WithLabelValues(mode, name, method).Observe(observeValue)
|
||||
}
|
||||
|
||||
func (m *dualWriterMetrics) recordDataSyncerDuration(isError bool, mode DualWriterMode, resource string, startFrom time.Time) {
|
||||
duration := time.Since(startFrom).Seconds()
|
||||
m.syncer.WithLabelValues(strconv.FormatBool(isError), fmt.Sprintf("%d", mode), resource).Observe(duration)
|
||||
|
||||
@@ -3,14 +3,26 @@ package rest
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/stretchr/testify/mock"
|
||||
metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apiserver/pkg/apis/example"
|
||||
"k8s.io/apiserver/pkg/registry/rest"
|
||||
)
|
||||
|
||||
var now = time.Now()
|
||||
|
||||
var exampleObj = &example.Pod{TypeMeta: metav1.TypeMeta{Kind: "foo"}, ObjectMeta: metav1.ObjectMeta{Name: "foo", ResourceVersion: "1", CreationTimestamp: metav1.Time{}, GenerateName: "foo"}, Spec: example.PodSpec{}, Status: example.PodStatus{StartTime: &metav1.Time{Time: now}}}
|
||||
var anotherObj = &example.Pod{TypeMeta: metav1.TypeMeta{Kind: "foo"}, ObjectMeta: metav1.ObjectMeta{Name: "bar", ResourceVersion: "2", GenerateName: "foo"}, Spec: example.PodSpec{}, Status: example.PodStatus{StartTime: &metav1.Time{Time: now}}}
|
||||
var exampleList = &example.PodList{TypeMeta: metav1.TypeMeta{Kind: "foo"}, ListMeta: metav1.ListMeta{}, Items: []example.Pod{*exampleObj}}
|
||||
var anotherList = &example.PodList{Items: []example.Pod{*anotherObj}}
|
||||
|
||||
var p = prometheus.NewRegistry()
|
||||
|
||||
type storageMock struct {
|
||||
*mock.Mock
|
||||
Storage
|
||||
@@ -104,12 +116,3 @@ func (m storageMock) DeleteCollection(ctx context.Context, deleteValidation rest
|
||||
}
|
||||
return args.Get(0).(runtime.Object), args.Error(1)
|
||||
}
|
||||
|
||||
type updatedObjInfoObj struct{}
|
||||
|
||||
func (u updatedObjInfoObj) UpdatedObject(ctx context.Context, oldObj runtime.Object) (newObj runtime.Object, err error) { // nolint:staticcheck
|
||||
// nolint:staticcheck
|
||||
oldObj = exampleObj
|
||||
return oldObj, nil
|
||||
}
|
||||
func (u updatedObjInfoObj) Preconditions() *metav1.Preconditions { return &metav1.Preconditions{} }
|
||||
|
||||
@@ -372,7 +372,7 @@ func InstallAPIs(
|
||||
if currentMode != mode {
|
||||
klog.Warningf("Requested DualWrite mode: %d, but using %d for %+v", mode, currentMode, gr)
|
||||
}
|
||||
return grafanarest.NewDualWriter(currentMode, legacy, storage, reg, key), nil
|
||||
return dualwrite.NewDualWriter(gr, currentMode, legacy, storage)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,229 @@
|
||||
package dualwrite
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apiserver/pkg/registry/rest"
|
||||
|
||||
"github.com/grafana/grafana-app-sdk/logging"
|
||||
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
)
|
||||
|
||||
var (
|
||||
_ grafanarest.Storage = (*dualWriter)(nil)
|
||||
)
|
||||
|
||||
// dualWriter will write first to legacy, then to unified keeping the same internal ID
|
||||
type dualWriter struct {
|
||||
legacy grafanarest.Storage
|
||||
unified grafanarest.Storage
|
||||
readUnified bool
|
||||
errorIsOK bool // in "mode1" we try writing both -- but don't block on unified write errors
|
||||
log logging.Logger
|
||||
}
|
||||
|
||||
func (d *dualWriter) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
|
||||
// Call get (send read traffic in cloud)
|
||||
unifiedGet, unifiedErr := d.unified.Get(ctx, name, options)
|
||||
if d.readUnified {
|
||||
return unifiedGet, unifiedErr
|
||||
}
|
||||
|
||||
legacyGet, err := d.legacy.Get(ctx, name, options)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if unifiedErr != nil && !apierrors.IsNotFound(unifiedErr) && !d.errorIsOK {
|
||||
return nil, unifiedErr // the unified error
|
||||
}
|
||||
return legacyGet, nil
|
||||
}
|
||||
|
||||
func (d *dualWriter) List(ctx context.Context, options *metainternalversion.ListOptions) (runtime.Object, error) {
|
||||
// Call list (send read traffic in cloud)
|
||||
unifiedList, err := d.unified.List(ctx, options)
|
||||
if d.readUnified {
|
||||
return unifiedList, err
|
||||
}
|
||||
|
||||
if err != nil && !d.errorIsOK {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return d.legacy.List(ctx, options)
|
||||
}
|
||||
|
||||
// Create overrides the behavior of the generic DualWriter and writes to LegacyStorage and Storage.
|
||||
func (d *dualWriter) Create(ctx context.Context, in runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
|
||||
log := d.log.With("method", "Create").WithContext(ctx)
|
||||
|
||||
accIn, err := meta.Accessor(in)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if accIn.GetUID() != "" {
|
||||
return nil, fmt.Errorf("UID should not be: %v", accIn.GetUID())
|
||||
}
|
||||
|
||||
if accIn.GetName() == "" && accIn.GetGenerateName() == "" {
|
||||
return nil, fmt.Errorf("name or generatename have to be set")
|
||||
}
|
||||
|
||||
// create in legacy first, and then unistore. if unistore fails, but legacy succeeds,
|
||||
// will try to cleanup the object in legacy.
|
||||
createdFromLegacy, err := d.legacy.Create(ctx, in, createValidation, options)
|
||||
if err != nil {
|
||||
log.Error("unable to create object in legacy storage", "err", err)
|
||||
return createdFromLegacy, err
|
||||
}
|
||||
|
||||
createdCopy := createdFromLegacy.DeepCopyObject()
|
||||
accCreated, err := meta.Accessor(createdCopy)
|
||||
if err != nil {
|
||||
return createdFromLegacy, err
|
||||
}
|
||||
accCreated.SetResourceVersion("")
|
||||
accCreated.SetUID("")
|
||||
|
||||
storageObj, errObjectSt := d.unified.Create(ctx, createdCopy, createValidation, options)
|
||||
if errObjectSt != nil {
|
||||
log.Error("unable to create object in unified storage", "err", err)
|
||||
if d.errorIsOK {
|
||||
return createdFromLegacy, nil
|
||||
}
|
||||
|
||||
// if we cannot create in unistore, attempt to clean up legacy
|
||||
_, _, err = d.legacy.Delete(ctx, accCreated.GetName(), nil, &metav1.DeleteOptions{})
|
||||
if err != nil {
|
||||
log.Error("unable to cleanup object in legacy storage", "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
if d.readUnified {
|
||||
return storageObj, errObjectSt
|
||||
}
|
||||
return createdFromLegacy, errObjectSt
|
||||
}
|
||||
|
||||
func (d *dualWriter) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
|
||||
// delete from legacy first, and then unistore. Will return a failure if either fails,
|
||||
// unless its a 404.
|
||||
//
|
||||
// we want to delete from legacy first, otherwise if the delete from unistore was successful,
|
||||
// but legacy failed, the user would get a failure, but not be able to retry the delete
|
||||
// as they would not be able to see the object in unistore anymore.
|
||||
objFromLegacy, asyncLegacy, err := d.legacy.Delete(ctx, name, deleteValidation, options)
|
||||
if err != nil && !d.readUnified {
|
||||
return objFromLegacy, asyncLegacy, err
|
||||
}
|
||||
|
||||
objFromStorage, asyncStorage, err := d.unified.Delete(ctx, name, deleteValidation, options)
|
||||
if err != nil && apierrors.IsNotFound(err) || d.errorIsOK {
|
||||
err = nil // clear the error
|
||||
}
|
||||
|
||||
if d.readUnified {
|
||||
return objFromStorage, asyncStorage, err
|
||||
}
|
||||
|
||||
return objFromLegacy, asyncLegacy, err
|
||||
}
|
||||
|
||||
// Update overrides the behavior of the generic DualWriter and writes first to Storage and then to LegacyStorage.
|
||||
func (d *dualWriter) Update(ctx context.Context, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error) {
|
||||
log := d.log.With("method", "Update").WithContext(ctx)
|
||||
|
||||
// The incoming RV is not stable -- it may be from legacy or storage!
|
||||
// This sets a flag in the context and our apistore is more lenient when it exists
|
||||
ctx = grafanarest.WithDualWriteUpdate(ctx)
|
||||
|
||||
// update in legacy first, and then unistore. Will return a failure if either fails.
|
||||
//
|
||||
// we want to update in legacy first, otherwise if the update from unistore was successful,
|
||||
// but legacy failed, the user would get a failure, but see the update did apply to the source
|
||||
// of truth, and be less likely to retry to save (and get the stores in sync again)
|
||||
|
||||
objFromLegacy, createdLegacy, err := d.legacy.Update(ctx, name, objInfo, createValidation, updateValidation, forceAllowCreate, options)
|
||||
if err != nil {
|
||||
log.With("object", objFromLegacy).Error("could not update in legacy storage", "err", err)
|
||||
return objFromLegacy, createdLegacy, err
|
||||
}
|
||||
|
||||
objFromStorage, created, err := d.unified.Update(ctx, name, objInfo, createValidation, updateValidation, forceAllowCreate, options)
|
||||
if err != nil {
|
||||
log.With("object", objFromStorage).Error("could not update in storage", "err", err)
|
||||
if d.errorIsOK {
|
||||
return objFromLegacy, createdLegacy, nil
|
||||
}
|
||||
}
|
||||
|
||||
if d.readUnified {
|
||||
return objFromStorage, created, err
|
||||
}
|
||||
|
||||
return objFromLegacy, createdLegacy, err
|
||||
}
|
||||
|
||||
// DeleteCollection overrides the behavior of the generic DualWriter and deletes from both LegacyStorage and Storage.
|
||||
func (d *dualWriter) DeleteCollection(ctx context.Context, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions, listOptions *metainternalversion.ListOptions) (runtime.Object, error) {
|
||||
log := d.log.With("method", "DeleteCollection", "resourceVersion", listOptions.ResourceVersion).WithContext(ctx)
|
||||
|
||||
// delete from legacy first, and anything that is successful can be deleted in unistore too.
|
||||
//
|
||||
// we want to delete from legacy first, otherwise if the delete from unistore was successful,
|
||||
// but legacy failed, the user would get a failure, but not be able to retry the delete
|
||||
// as they would not be able to see the object in unistore anymore.
|
||||
|
||||
deletedLegacy, err := d.legacy.DeleteCollection(ctx, deleteValidation, options, listOptions)
|
||||
if err != nil {
|
||||
log.With("deleted", deletedLegacy).Error("failed to delete collection successfully from legacy storage", "err", err)
|
||||
return deletedLegacy, err
|
||||
}
|
||||
|
||||
deletedStorage, err := d.unified.DeleteCollection(ctx, deleteValidation, options, listOptions)
|
||||
if err != nil {
|
||||
log.With("deleted", deletedStorage).Error("failed to delete collection successfully from Storage", "err", err)
|
||||
if d.errorIsOK {
|
||||
return deletedLegacy, nil
|
||||
}
|
||||
}
|
||||
|
||||
if d.readUnified {
|
||||
return deletedStorage, err
|
||||
}
|
||||
|
||||
return deletedLegacy, err
|
||||
}
|
||||
|
||||
func (d *dualWriter) Destroy() {
|
||||
d.legacy.Destroy()
|
||||
d.unified.Destroy()
|
||||
}
|
||||
|
||||
func (d *dualWriter) GetSingularName() string {
|
||||
return d.unified.GetSingularName()
|
||||
}
|
||||
|
||||
func (d *dualWriter) NamespaceScoped() bool {
|
||||
return d.unified.NamespaceScoped()
|
||||
}
|
||||
|
||||
func (d *dualWriter) New() runtime.Object {
|
||||
return d.unified.New()
|
||||
}
|
||||
|
||||
func (d *dualWriter) NewList() runtime.Object {
|
||||
return d.unified.NewList()
|
||||
}
|
||||
|
||||
func (d *dualWriter) ConvertToTable(ctx context.Context, object runtime.Object, tableOptions runtime.Object) (*metav1.Table, error) {
|
||||
return d.unified.ConvertToTable(ctx, object, tableOptions)
|
||||
}
|
||||
+28
-376
@@ -1,4 +1,4 @@
|
||||
package rest
|
||||
package dualwrite
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -14,6 +14,8 @@ import (
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apiserver/pkg/apis/example"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
)
|
||||
|
||||
var now = time.Now()
|
||||
@@ -26,7 +28,6 @@ var exampleList = &example.PodList{TypeMeta: metav1.TypeMeta{Kind: "foo"}, ListM
|
||||
var anotherList = &example.PodList{Items: []example.Pod{*anotherObj}}
|
||||
|
||||
var p = prometheus.NewRegistry()
|
||||
var kind = "foo"
|
||||
|
||||
func TestMode1_Create(t *testing.T) {
|
||||
type testCase struct {
|
||||
@@ -60,8 +61,8 @@ func TestMode1_Create(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -73,7 +74,8 @@ func TestMode1_Create(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode1, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, err := dw.Create(context.Background(), tt.input, func(context.Context, runtime.Object) error { return nil }, &metav1.CreateOptions{})
|
||||
|
||||
@@ -90,63 +92,6 @@ func TestMode1_Create(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_CreateOnUnifiedStorage(t *testing.T) {
|
||||
ctxCanceled, cancel := context.WithCancel(context.TODO())
|
||||
cancel()
|
||||
|
||||
type testCase struct {
|
||||
name string
|
||||
input runtime.Object
|
||||
ctx *context.Context
|
||||
setupLegacyFn func(m *mock.Mock)
|
||||
setupStorageFn func(m *mock.Mock)
|
||||
}
|
||||
tests :=
|
||||
[]testCase{
|
||||
{
|
||||
name: "Create on unified storage",
|
||||
input: exampleObj,
|
||||
setupStorageFn: func(m *mock.Mock) {
|
||||
m.On("Create", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(exampleObjNoRV, nil)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Create on unified storage works even if parent context is canceled",
|
||||
input: exampleObj,
|
||||
ctx: &ctxCanceled,
|
||||
setupStorageFn: func(m *mock.Mock) {
|
||||
m.On("Create", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(exampleObjNoRV, nil)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
|
||||
if tt.setupLegacyFn != nil {
|
||||
tt.setupLegacyFn(ls.Mock)
|
||||
}
|
||||
if tt.setupStorageFn != nil {
|
||||
tt.setupStorageFn(us.Mock)
|
||||
}
|
||||
|
||||
ctx := context.TODO()
|
||||
if tt.ctx != nil {
|
||||
ctx = *tt.ctx
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
err := dw.(*DualWriterMode1).createOnUnifiedStorage(ctx, func(context.Context, runtime.Object) error { return nil }, tt.input, &metav1.CreateOptions{})
|
||||
require.NoError(t, err)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_Get(t *testing.T) {
|
||||
type testCase struct {
|
||||
setupLegacyFn func(m *mock.Mock, name string)
|
||||
@@ -190,8 +135,8 @@ func TestMode1_Get(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -203,7 +148,8 @@ func TestMode1_Get(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, name)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode1, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, err := dw.Get(context.Background(), name, &metav1.GetOptions{})
|
||||
|
||||
@@ -220,62 +166,6 @@ func TestMode1_Get(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_GetFromUnifiedStorage(t *testing.T) {
|
||||
ctxCanceled, cancel := context.WithCancel(context.TODO())
|
||||
cancel()
|
||||
|
||||
type testCase struct {
|
||||
setupLegacyFn func(m *mock.Mock, name string)
|
||||
setupStorageFn func(m *mock.Mock, name string)
|
||||
ctx *context.Context
|
||||
name string
|
||||
}
|
||||
tests :=
|
||||
[]testCase{
|
||||
{
|
||||
name: "should succeed when getting an object from UnifiedStorage",
|
||||
setupStorageFn: func(m *mock.Mock, name string) {
|
||||
m.On("Get", mock.Anything, name, mock.Anything).Return(exampleObj, nil)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "should succeed when getting an object from UnifiedStorage even if parent context is canceled",
|
||||
ctx: &ctxCanceled,
|
||||
setupStorageFn: func(m *mock.Mock, name string) {
|
||||
m.On("Get", mock.Anything, name, mock.Anything).Return(exampleObj, nil)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
name := "foo"
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
|
||||
if tt.setupLegacyFn != nil {
|
||||
tt.setupLegacyFn(ls.Mock, name)
|
||||
}
|
||||
if tt.setupStorageFn != nil {
|
||||
tt.setupStorageFn(us.Mock, name)
|
||||
}
|
||||
|
||||
ctx := context.TODO()
|
||||
if tt.ctx != nil {
|
||||
ctx = *tt.ctx
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
err := dw.(*DualWriterMode1).getFromUnifiedStorage(ctx, exampleObj, name, &metav1.GetOptions{})
|
||||
require.NoError(t, err)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_List(t *testing.T) {
|
||||
type testCase struct {
|
||||
setupLegacyFn func(m *mock.Mock)
|
||||
@@ -308,8 +198,8 @@ func TestMode1_List(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -321,9 +211,10 @@ func TestMode1_List(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode1, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err := dw.List(context.Background(), &metainternalversion.ListOptions{})
|
||||
_, err = dw.List(context.Background(), &metainternalversion.ListOptions{})
|
||||
|
||||
if tt.wantErr {
|
||||
require.Error(t, err)
|
||||
@@ -333,64 +224,6 @@ func TestMode1_List(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_ListFromUnifiedStorage(t *testing.T) {
|
||||
ctxCanceled, cancel := context.WithCancel(context.TODO())
|
||||
cancel()
|
||||
|
||||
type testCase struct {
|
||||
ctx *context.Context
|
||||
name string
|
||||
setupLegacyFn func(m *mock.Mock)
|
||||
setupStorageFn func(m *mock.Mock)
|
||||
}
|
||||
tests :=
|
||||
[]testCase{
|
||||
{
|
||||
name: "should succeed when listing from UnifiedStorage",
|
||||
setupStorageFn: func(m *mock.Mock) {
|
||||
m.On("List", mock.Anything, mock.Anything).Return(anotherList, nil)
|
||||
},
|
||||
setupLegacyFn: func(m *mock.Mock) {
|
||||
m.On("List", mock.Anything, mock.Anything).Return(anotherList, nil)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "should succeed when listing from UnifiedStorage even if parent context is canceled",
|
||||
ctx: &ctxCanceled,
|
||||
setupStorageFn: func(m *mock.Mock) {
|
||||
m.On("List", mock.Anything, mock.Anything).Return(anotherList, nil)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
|
||||
if tt.setupLegacyFn != nil {
|
||||
tt.setupLegacyFn(ls.Mock)
|
||||
}
|
||||
if tt.setupStorageFn != nil {
|
||||
tt.setupStorageFn(us.Mock)
|
||||
}
|
||||
|
||||
ctx := context.TODO()
|
||||
if tt.ctx != nil {
|
||||
ctx = *tt.ctx
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
|
||||
err := dw.(*DualWriterMode1).listFromUnifiedStorage(ctx, &metainternalversion.ListOptions{}, anotherList)
|
||||
require.NoError(t, err)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_Delete(t *testing.T) {
|
||||
type testCase struct {
|
||||
setupLegacyFn func(m *mock.Mock, name string)
|
||||
@@ -434,8 +267,8 @@ func TestMode1_Delete(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -447,7 +280,8 @@ func TestMode1_Delete(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, name)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode1, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, _, err := dw.Delete(context.Background(), name, func(ctx context.Context, obj runtime.Object) error { return nil }, &metav1.DeleteOptions{})
|
||||
|
||||
@@ -463,63 +297,6 @@ func TestMode1_Delete(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_DeleteFromUnifiedStorage(t *testing.T) {
|
||||
ctxCanceled, cancel := context.WithCancel(context.TODO())
|
||||
cancel()
|
||||
|
||||
type testCase struct {
|
||||
ctx *context.Context
|
||||
setupLegacyFn func(m *mock.Mock, name string)
|
||||
setupStorageFn func(m *mock.Mock, name string)
|
||||
name string
|
||||
}
|
||||
tests :=
|
||||
[]testCase{
|
||||
{
|
||||
name: "should succeed when deleting an object from UnifiedStorage",
|
||||
setupStorageFn: func(m *mock.Mock, name string) {
|
||||
m.On("Delete", mock.Anything, name, mock.Anything, mock.Anything).Return(exampleObj, false, nil)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "should succeed when deleting an object from UnifiedStorage even if parent context is canceled",
|
||||
ctx: &ctxCanceled,
|
||||
setupStorageFn: func(m *mock.Mock, name string) {
|
||||
m.On("Delete", mock.Anything, name, mock.Anything, mock.Anything).Return(exampleObj, false, nil)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
name := "foo"
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
|
||||
if tt.setupLegacyFn != nil {
|
||||
tt.setupLegacyFn(ls.Mock, name)
|
||||
}
|
||||
if tt.setupStorageFn != nil {
|
||||
tt.setupStorageFn(us.Mock, name)
|
||||
}
|
||||
|
||||
ctx := context.TODO()
|
||||
if tt.ctx != nil {
|
||||
ctx = *tt.ctx
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
|
||||
err := dw.(*DualWriterMode1).deleteFromUnifiedStorage(ctx, exampleObj, name, func(ctx context.Context, obj runtime.Object) error { return nil }, &metav1.DeleteOptions{})
|
||||
require.NoError(t, err)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_DeleteCollection(t *testing.T) {
|
||||
type testCase struct {
|
||||
input *metav1.DeleteOptions
|
||||
@@ -565,8 +342,8 @@ func TestMode1_DeleteCollection(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -578,7 +355,8 @@ func TestMode1_DeleteCollection(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, tt.input)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode1, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, err := dw.DeleteCollection(context.Background(), func(ctx context.Context, obj runtime.Object) error { return nil }, tt.input, &metainternalversion.ListOptions{})
|
||||
|
||||
@@ -594,64 +372,6 @@ func TestMode1_DeleteCollection(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_DeleteCollectionFromUnifiedStorage(t *testing.T) {
|
||||
ctxCanceled, cancel := context.WithCancel(context.TODO())
|
||||
cancel()
|
||||
|
||||
type testCase struct {
|
||||
ctx *context.Context
|
||||
setupLegacyFn func(m *mock.Mock)
|
||||
setupStorageFn func(m *mock.Mock)
|
||||
name string
|
||||
input *metav1.DeleteOptions
|
||||
}
|
||||
tests :=
|
||||
[]testCase{
|
||||
{
|
||||
name: "should succeed when deleting a collection from UnifiedStorage",
|
||||
input: &metav1.DeleteOptions{TypeMeta: metav1.TypeMeta{Kind: "foo"}},
|
||||
setupStorageFn: func(m *mock.Mock) {
|
||||
m.On("DeleteCollection", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(exampleObj, nil)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "should succeed when deleting a collection from UnifiedStorage even if parent context is canceled",
|
||||
input: &metav1.DeleteOptions{TypeMeta: metav1.TypeMeta{Kind: "foo"}},
|
||||
ctx: &ctxCanceled,
|
||||
setupStorageFn: func(m *mock.Mock) {
|
||||
m.On("DeleteCollection", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(exampleObj, nil)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
|
||||
if tt.setupLegacyFn != nil {
|
||||
tt.setupLegacyFn(ls.Mock)
|
||||
}
|
||||
if tt.setupStorageFn != nil {
|
||||
tt.setupStorageFn(us.Mock)
|
||||
}
|
||||
|
||||
ctx := context.TODO()
|
||||
if tt.ctx != nil {
|
||||
ctx = *tt.ctx
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
|
||||
err := dw.(*DualWriterMode1).deleteCollectionFromUnifiedStorage(ctx, exampleObj, func(ctx context.Context, obj runtime.Object) error { return nil }, tt.input, &metainternalversion.ListOptions{})
|
||||
require.NoError(t, err)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_Update(t *testing.T) {
|
||||
type testCase struct {
|
||||
setupLegacyFn func(m *mock.Mock, input string)
|
||||
@@ -695,8 +415,8 @@ func TestMode1_Update(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -708,7 +428,8 @@ func TestMode1_Update(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, name)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode1, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, _, err := dw.Update(context.Background(), name, updatedObjInfoObj{}, func(ctx context.Context, obj runtime.Object) error { return nil }, func(ctx context.Context, obj, old runtime.Object) error { return nil }, false, &metav1.UpdateOptions{})
|
||||
|
||||
@@ -722,72 +443,3 @@ func TestMode1_Update(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_UpdateOnUnifiedStorage(t *testing.T) {
|
||||
ctxCanceled, cancel := context.WithCancel(context.TODO())
|
||||
cancel()
|
||||
|
||||
type testCase struct {
|
||||
ctx *context.Context
|
||||
setupLegacyFn func(m *mock.Mock, input string)
|
||||
setupStorageFn func(m *mock.Mock, input string)
|
||||
setupGetFn func(m *mock.Mock, input string)
|
||||
name string
|
||||
}
|
||||
tests :=
|
||||
[]testCase{
|
||||
{
|
||||
name: "should succeed when updating an object on UnifiedStorage",
|
||||
setupStorageFn: func(m *mock.Mock, input string) {
|
||||
m.On("Update", mock.Anything, input, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(anotherObj, false, nil)
|
||||
},
|
||||
setupGetFn: func(m *mock.Mock, input string) {
|
||||
m.On("Get", mock.Anything, input, mock.Anything).Return(exampleObj, nil)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "should succeed when updating an object on UnifiedStorage even if parent context is canceled",
|
||||
ctx: &ctxCanceled,
|
||||
setupStorageFn: func(m *mock.Mock, input string) {
|
||||
m.On("Update", mock.Anything, input, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(anotherObj, false, nil)
|
||||
},
|
||||
setupGetFn: func(m *mock.Mock, input string) {
|
||||
m.On("Get", mock.Anything, input, mock.Anything).Return(exampleObj, nil)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
name := "foo"
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
|
||||
if tt.setupLegacyFn != nil {
|
||||
tt.setupLegacyFn(ls.Mock, name)
|
||||
}
|
||||
if tt.setupStorageFn != nil {
|
||||
tt.setupStorageFn(us.Mock, name)
|
||||
}
|
||||
|
||||
if tt.setupGetFn != nil {
|
||||
tt.setupGetFn(ls.Mock, name)
|
||||
tt.setupGetFn(us.Mock, name)
|
||||
}
|
||||
|
||||
ctx := context.TODO()
|
||||
if tt.ctx != nil {
|
||||
ctx = *tt.ctx
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode1, ls, us, p, kind)
|
||||
|
||||
err := dw.(*DualWriterMode1).updateOnUnifiedStorageMode1(ctx, exampleObj, name, updatedObjInfoObj{}, func(ctx context.Context, obj runtime.Object) error { return nil }, func(ctx context.Context, obj, old runtime.Object) error { return nil }, false, &metav1.UpdateOptions{})
|
||||
require.NoError(t, err)
|
||||
})
|
||||
}
|
||||
}
|
||||
+27
-19
@@ -1,4 +1,4 @@
|
||||
package rest
|
||||
package dualwrite
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -12,6 +12,8 @@ import (
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
)
|
||||
|
||||
var createFn = func(context.Context, runtime.Object) error { return nil }
|
||||
@@ -54,8 +56,8 @@ func TestMode2_Create(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -67,7 +69,8 @@ func TestMode2_Create(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, tt.input)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode2, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode2, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, err := dw.Create(context.Background(), tt.input, createFn, &metav1.CreateOptions{})
|
||||
|
||||
@@ -138,8 +141,8 @@ func TestMode2_Get(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -151,7 +154,8 @@ func TestMode2_Get(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, tt.input)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode2, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode2, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, err := dw.Get(context.Background(), tt.input, &metav1.GetOptions{})
|
||||
|
||||
@@ -212,8 +216,8 @@ func TestMode2_List(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -225,7 +229,8 @@ func TestMode2_List(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode2, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode2, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, err := dw.List(context.Background(), &metainternalversion.ListOptions{})
|
||||
|
||||
@@ -313,8 +318,8 @@ func TestMode2_Delete(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -326,7 +331,8 @@ func TestMode2_Delete(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, name)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode2, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode2, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, _, err := dw.Delete(context.Background(), name, func(context.Context, runtime.Object) error { return nil }, &metav1.DeleteOptions{})
|
||||
|
||||
@@ -382,8 +388,8 @@ func TestMode2_DeleteCollection(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -395,7 +401,8 @@ func TestMode2_DeleteCollection(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode2, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode2, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, err := dw.DeleteCollection(context.Background(), func(ctx context.Context, obj runtime.Object) error { return nil }, &metav1.DeleteOptions{TypeMeta: metav1.TypeMeta{Kind: name}}, &metainternalversion.ListOptions{})
|
||||
|
||||
@@ -451,8 +458,8 @@ func TestMode2_Update(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -464,7 +471,8 @@ func TestMode2_Update(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, name)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode2, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode2, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, _, err := dw.Update(context.Background(), name, updatedObjInfoObj{}, func(ctx context.Context, obj runtime.Object) error { return nil }, func(ctx context.Context, obj, old runtime.Object) error { return nil }, false, &metav1.UpdateOptions{})
|
||||
|
||||
+27
-73
@@ -1,11 +1,10 @@
|
||||
package rest
|
||||
package dualwrite
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/stretchr/testify/mock"
|
||||
"github.com/stretchr/testify/require"
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
@@ -13,6 +12,8 @@ import (
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
)
|
||||
|
||||
func TestMode3_Create(t *testing.T) {
|
||||
@@ -61,8 +62,8 @@ func TestMode3_Create(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -74,7 +75,8 @@ func TestMode3_Create(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, tt.input)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode3, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode3, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, err := dw.Create(context.Background(), tt.input, createFn, &metav1.CreateOptions{})
|
||||
|
||||
@@ -128,8 +130,8 @@ func TestMode3_Get(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -141,8 +143,8 @@ func TestMode3_Get(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, name)
|
||||
}
|
||||
|
||||
p := prometheus.NewRegistry()
|
||||
dw := NewDualWriter(Mode3, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode3, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, err := dw.Get(context.Background(), name, &metav1.GetOptions{})
|
||||
|
||||
@@ -157,58 +159,6 @@ func TestMode3_Get(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode1_GetFromLegacyStorage(t *testing.T) {
|
||||
ctxCanceled, cancel := context.WithCancel(context.TODO())
|
||||
cancel()
|
||||
|
||||
type testCase struct {
|
||||
setupLegacyFn func(m *mock.Mock, name string)
|
||||
ctx *context.Context
|
||||
name string
|
||||
}
|
||||
tests :=
|
||||
[]testCase{
|
||||
{
|
||||
name: "should succeed when getting an object from the LegacyStorage",
|
||||
setupLegacyFn: func(m *mock.Mock, name string) {
|
||||
m.On("Get", mock.Anything, name, mock.Anything).Return(exampleObj, nil)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "should succeed when getting an object from the LegacyStorage even if parent context is canceled",
|
||||
ctx: &ctxCanceled,
|
||||
setupLegacyFn: func(m *mock.Mock, name string) {
|
||||
m.On("Get", mock.Anything, name, mock.Anything).Return(exampleObj, nil)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
name := "foo"
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
|
||||
if tt.setupLegacyFn != nil {
|
||||
tt.setupLegacyFn(ls.Mock, name)
|
||||
}
|
||||
|
||||
ctx := context.TODO()
|
||||
if tt.ctx != nil {
|
||||
ctx = *tt.ctx
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode3, ls, us, p, kind)
|
||||
err := dw.(*DualWriterMode3).getFromLegacyStorage(ctx, exampleObj, name, &metav1.GetOptions{})
|
||||
require.NoError(t, err)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestMode3_List(t *testing.T) {
|
||||
type testCase struct {
|
||||
setupStorageFn func(m *mock.Mock, options *metainternalversion.ListOptions)
|
||||
@@ -234,8 +184,8 @@ func TestMode3_List(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -244,7 +194,8 @@ func TestMode3_List(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, &metainternalversion.ListOptions{TypeMeta: metav1.TypeMeta{Kind: "foo"}})
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode3, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode3, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
res, err := dw.List(context.Background(), &metainternalversion.ListOptions{TypeMeta: metav1.TypeMeta{Kind: "foo"}})
|
||||
|
||||
@@ -311,8 +262,8 @@ func TestMode3_Delete(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -324,7 +275,8 @@ func TestMode3_Delete(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, name)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode3, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode3, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, _, err := dw.Delete(context.Background(), name, func(context.Context, runtime.Object) error { return nil }, &metav1.DeleteOptions{})
|
||||
|
||||
@@ -380,8 +332,8 @@ func TestMode3_DeleteCollection(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -393,7 +345,8 @@ func TestMode3_DeleteCollection(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode3, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode3, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, err := dw.DeleteCollection(context.Background(), func(ctx context.Context, obj runtime.Object) error { return nil }, &metav1.DeleteOptions{TypeMeta: metav1.TypeMeta{Kind: name}}, &metainternalversion.ListOptions{})
|
||||
|
||||
@@ -449,8 +402,8 @@ func TestMode3_Update(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
l := (Storage)(nil)
|
||||
s := (Storage)(nil)
|
||||
l := (rest.Storage)(nil)
|
||||
s := (rest.Storage)(nil)
|
||||
|
||||
ls := storageMock{&mock.Mock{}, l}
|
||||
us := storageMock{&mock.Mock{}, s}
|
||||
@@ -462,7 +415,8 @@ func TestMode3_Update(t *testing.T) {
|
||||
tt.setupStorageFn(us.Mock, name)
|
||||
}
|
||||
|
||||
dw := NewDualWriter(Mode3, ls, us, p, kind)
|
||||
dw, err := NewDualWriter(kind, rest.Mode3, ls, us)
|
||||
require.NoError(t, err)
|
||||
|
||||
obj, _, err := dw.Update(context.Background(), name, updatedObjInfoObj{}, func(ctx context.Context, obj runtime.Object) error { return nil }, func(ctx context.Context, obj, old runtime.Object) error { return nil }, false, &metav1.UpdateOptions{})
|
||||
|
||||
@@ -11,25 +11,26 @@ import (
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apiserver/pkg/registry/rest"
|
||||
|
||||
"github.com/grafana/grafana-app-sdk/logging"
|
||||
|
||||
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
)
|
||||
|
||||
func (m *service) NewStorage(gr schema.GroupResource,
|
||||
legacy grafanarest.Storage,
|
||||
storage grafanarest.Storage,
|
||||
) (grafanarest.Storage, error) {
|
||||
func (m *service) NewStorage(gr schema.GroupResource, legacy grafanarest.Storage, unified grafanarest.Storage) (grafanarest.Storage, error) {
|
||||
status, err := m.Status(context.Background(), gr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
log := logging.DefaultLogger.With("gr", gr.String())
|
||||
|
||||
if m.enabled && status.Runtime {
|
||||
// Dynamic storage behavior
|
||||
return &runtimeDualWriter{
|
||||
service: m,
|
||||
legacy: legacy,
|
||||
unified: storage,
|
||||
dualwrite: grafanarest.NewDualWriter(grafanarest.Mode3, legacy, storage, m.reg, gr.String()),
|
||||
unified: unified,
|
||||
dualwrite: &dualWriter{legacy: legacy, unified: unified, log: log}, // not used for read
|
||||
gr: gr,
|
||||
}, nil
|
||||
}
|
||||
@@ -37,13 +38,13 @@ func (m *service) NewStorage(gr schema.GroupResource,
|
||||
if status.ReadUnified {
|
||||
if status.WriteLegacy {
|
||||
// Write both, read unified
|
||||
return grafanarest.NewDualWriter(grafanarest.Mode3, legacy, storage, m.reg, gr.String()), nil
|
||||
return &dualWriter{legacy: legacy, unified: unified, log: log, readUnified: true}, nil
|
||||
}
|
||||
return storage, nil
|
||||
return unified, nil
|
||||
}
|
||||
if status.WriteUnified {
|
||||
// Write both, read legacy
|
||||
return grafanarest.NewDualWriter(grafanarest.Mode2, legacy, storage, m.reg, gr.String()), nil
|
||||
return &dualWriter{legacy: legacy, unified: unified, log: log}, nil
|
||||
}
|
||||
return legacy, nil
|
||||
}
|
||||
@@ -55,7 +56,7 @@ type runtimeDualWriter struct {
|
||||
service Service
|
||||
legacy grafanarest.Storage
|
||||
unified grafanarest.Storage
|
||||
dualwrite grafanarest.Storage
|
||||
dualwrite *dualWriter
|
||||
gr schema.GroupResource
|
||||
}
|
||||
|
||||
|
||||
@@ -4,33 +4,20 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/stretchr/testify/mock"
|
||||
"github.com/stretchr/testify/require"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apiserver/pkg/apis/example"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
)
|
||||
|
||||
var now = time.Now()
|
||||
|
||||
var createFn = func(context.Context, runtime.Object) error { return nil }
|
||||
|
||||
var exampleObj = &example.Pod{TypeMeta: metav1.TypeMeta{Kind: "foo"}, ObjectMeta: metav1.ObjectMeta{Name: "foo", ResourceVersion: "1", CreationTimestamp: metav1.Time{}, GenerateName: "foo"}, Spec: example.PodSpec{}, Status: example.PodStatus{StartTime: &metav1.Time{Time: now}}}
|
||||
var exampleObjNoRV = &example.Pod{TypeMeta: metav1.TypeMeta{Kind: "foo"}, ObjectMeta: metav1.ObjectMeta{Name: "foo", ResourceVersion: "", CreationTimestamp: metav1.Time{}, GenerateName: "foo"}, Spec: example.PodSpec{}, Status: example.PodStatus{StartTime: &metav1.Time{Time: now}}}
|
||||
var anotherObj = &example.Pod{TypeMeta: metav1.TypeMeta{Kind: "foo"}, ObjectMeta: metav1.ObjectMeta{Name: "bar", ResourceVersion: "2", GenerateName: "foo"}, Spec: example.PodSpec{}, Status: example.PodStatus{StartTime: &metav1.Time{Time: now}}}
|
||||
var failingObj = &example.Pod{TypeMeta: metav1.TypeMeta{Kind: "foo"}, ObjectMeta: metav1.ObjectMeta{Name: "object-fail", ResourceVersion: "2", GenerateName: "object-fail"}, Spec: example.PodSpec{}, Status: example.PodStatus{}}
|
||||
|
||||
var p = prometheus.NewRegistry()
|
||||
var kind = schema.GroupResource{Group: "g", Resource: "r"}
|
||||
|
||||
func TestManagedMode3_Create(t *testing.T) {
|
||||
func TestRuntime_Create(t *testing.T) {
|
||||
type testCase struct {
|
||||
input runtime.Object
|
||||
setupLegacyFn func(m *mock.Mock, input runtime.Object)
|
||||
@@ -105,7 +92,7 @@ func TestManagedMode3_Create(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagedMode3_Get(t *testing.T) {
|
||||
func TestRuntime_Get(t *testing.T) {
|
||||
type testCase struct {
|
||||
setupLegacyFn func(m *mock.Mock, name string)
|
||||
setupStorageFn func(m *mock.Mock, name string)
|
||||
@@ -184,7 +171,7 @@ func TestManagedMode3_Get(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagedMode3_CreateWhileMigrating(t *testing.T) {
|
||||
func TestRuntime_CreateWhileMigrating(t *testing.T) {
|
||||
type testCase struct {
|
||||
input runtime.Object
|
||||
setupLegacyFn func(m *mock.Mock, input runtime.Object)
|
||||
|
||||
@@ -6,16 +6,58 @@ import (
|
||||
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
|
||||
"github.com/grafana/grafana-app-sdk/logging"
|
||||
"github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
)
|
||||
|
||||
// NewDualWriter -- temporary shim
|
||||
func NewDualWriter(
|
||||
gr schema.GroupResource,
|
||||
mode rest.DualWriterMode,
|
||||
legacy rest.Storage,
|
||||
unified rest.Storage,
|
||||
) (rest.Storage, error) {
|
||||
m := &staticService{}
|
||||
m.SetMode(gr, mode)
|
||||
return m.NewStorage(gr, legacy, unified)
|
||||
}
|
||||
|
||||
type staticService struct {
|
||||
cfg *setting.Cfg
|
||||
}
|
||||
|
||||
func (m *staticService) NewStorage(gr schema.GroupResource, legacy rest.Storage, storage rest.Storage) (rest.Storage, error) {
|
||||
return nil, fmt.Errorf("not implemented")
|
||||
// Used in tests
|
||||
func (m *staticService) SetMode(gr schema.GroupResource, mode rest.DualWriterMode) {
|
||||
if m.cfg == nil {
|
||||
m.cfg = &setting.Cfg{}
|
||||
}
|
||||
if m.cfg.UnifiedStorage == nil {
|
||||
m.cfg.UnifiedStorage = make(map[string]setting.UnifiedStorageConfig)
|
||||
}
|
||||
m.cfg.UnifiedStorage[gr.String()] = setting.UnifiedStorageConfig{
|
||||
DualWriterMode: mode,
|
||||
}
|
||||
}
|
||||
|
||||
func (m *staticService) NewStorage(gr schema.GroupResource, legacy rest.Storage, unified rest.Storage) (rest.Storage, error) {
|
||||
log := logging.DefaultLogger.With("dualwrite", gr.String())
|
||||
|
||||
config := m.cfg.UnifiedStorage[gr.String()]
|
||||
switch config.DualWriterMode {
|
||||
case rest.Mode1:
|
||||
return &dualWriter{log: log, legacy: legacy, unified: unified, errorIsOK: true}, nil
|
||||
case rest.Mode2:
|
||||
return &dualWriter{log: log, legacy: legacy, unified: unified}, nil
|
||||
case rest.Mode3:
|
||||
return &dualWriter{log: log, legacy: legacy, unified: unified, readUnified: true}, nil
|
||||
case rest.Mode4, rest.Mode5:
|
||||
return unified, nil // use unified directly
|
||||
case rest.Mode0:
|
||||
fallthrough
|
||||
default:
|
||||
return legacy, nil
|
||||
}
|
||||
}
|
||||
|
||||
// ReadFromUnified implements Service.
|
||||
|
||||
@@ -106,3 +106,12 @@ func (m storageMock) DeleteCollection(ctx context.Context, deleteValidation rest
|
||||
}
|
||||
return args.Get(0).(runtime.Object), args.Error(1)
|
||||
}
|
||||
|
||||
type updatedObjInfoObj struct{}
|
||||
|
||||
func (u updatedObjInfoObj) UpdatedObject(ctx context.Context, oldObj runtime.Object) (newObj runtime.Object, err error) { // nolint:staticcheck
|
||||
// nolint:staticcheck
|
||||
oldObj = exampleObj
|
||||
return oldObj, nil
|
||||
}
|
||||
func (u updatedObjInfoObj) Preconditions() *metav1.Preconditions { return &metav1.Preconditions{} }
|
||||
|
||||
Reference in New Issue
Block a user