diff --git a/pkg/apiserver/rest/dualwriter.go b/pkg/apiserver/rest/dualwriter.go index 31ca7e9c97e..6cd37cb8740 100644 --- a/pkg/apiserver/rest/dualwriter.go +++ b/pkg/apiserver/rest/dualwriter.go @@ -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() -} diff --git a/pkg/apiserver/rest/dualwriter_mode1.go b/pkg/apiserver/rest/dualwriter_mode1.go deleted file mode 100644 index 7395fc5d13b..00000000000 --- a/pkg/apiserver/rest/dualwriter_mode1.go +++ /dev/null @@ -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) -} diff --git a/pkg/apiserver/rest/dualwriter_mode2.go b/pkg/apiserver/rest/dualwriter_mode2.go deleted file mode 100644 index 17aae84e2fb..00000000000 --- a/pkg/apiserver/rest/dualwriter_mode2.go +++ /dev/null @@ -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 -} diff --git a/pkg/apiserver/rest/dualwriter_mode3.go b/pkg/apiserver/rest/dualwriter_mode3.go deleted file mode 100644 index db61f0ed7f7..00000000000 --- a/pkg/apiserver/rest/dualwriter_mode3.go +++ /dev/null @@ -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) -} diff --git a/pkg/apiserver/rest/metrics.go b/pkg/apiserver/rest/metrics.go index 0b15a7a4ff5..9380e948f75 100644 --- a/pkg/apiserver/rest/metrics.go +++ b/pkg/apiserver/rest/metrics.go @@ -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) diff --git a/pkg/apiserver/rest/storage_mocks_test.go b/pkg/apiserver/rest/storage_mocks_test.go index 3baaa14c526..a85a372c5e1 100644 --- a/pkg/apiserver/rest/storage_mocks_test.go +++ b/pkg/apiserver/rest/storage_mocks_test.go @@ -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{} } diff --git a/pkg/services/apiserver/builder/helper.go b/pkg/services/apiserver/builder/helper.go index 5fdf11c51a8..cb6d2626061 100644 --- a/pkg/services/apiserver/builder/helper.go +++ b/pkg/services/apiserver/builder/helper.go @@ -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) } } diff --git a/pkg/storage/legacysql/dualwrite/dualwriter.go b/pkg/storage/legacysql/dualwrite/dualwriter.go new file mode 100644 index 00000000000..242b401b856 --- /dev/null +++ b/pkg/storage/legacysql/dualwrite/dualwriter.go @@ -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) +} diff --git a/pkg/apiserver/rest/dualwriter_mode1_test.go b/pkg/storage/legacysql/dualwrite/dualwriter_mode1_test.go similarity index 57% rename from pkg/apiserver/rest/dualwriter_mode1_test.go rename to pkg/storage/legacysql/dualwrite/dualwriter_mode1_test.go index 398b25a3acc..74a73a573c2 100644 --- a/pkg/apiserver/rest/dualwriter_mode1_test.go +++ b/pkg/storage/legacysql/dualwrite/dualwriter_mode1_test.go @@ -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) - }) - } -} diff --git a/pkg/apiserver/rest/dualwriter_mode2_test.go b/pkg/storage/legacysql/dualwrite/dualwriter_mode2_test.go similarity index 94% rename from pkg/apiserver/rest/dualwriter_mode2_test.go rename to pkg/storage/legacysql/dualwrite/dualwriter_mode2_test.go index b0d653053df..cd1e87b0b9a 100644 --- a/pkg/apiserver/rest/dualwriter_mode2_test.go +++ b/pkg/storage/legacysql/dualwrite/dualwriter_mode2_test.go @@ -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{}) diff --git a/pkg/apiserver/rest/dualwriter_mode3_test.go b/pkg/storage/legacysql/dualwrite/dualwriter_mode3_test.go similarity index 87% rename from pkg/apiserver/rest/dualwriter_mode3_test.go rename to pkg/storage/legacysql/dualwrite/dualwriter_mode3_test.go index 832c8957aeb..c53be55b64c 100644 --- a/pkg/apiserver/rest/dualwriter_mode3_test.go +++ b/pkg/storage/legacysql/dualwrite/dualwriter_mode3_test.go @@ -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{}) diff --git a/pkg/storage/legacysql/dualwrite/runtime.go b/pkg/storage/legacysql/dualwrite/runtime.go index 33896e669e2..8d9d5eb872d 100644 --- a/pkg/storage/legacysql/dualwrite/runtime.go +++ b/pkg/storage/legacysql/dualwrite/runtime.go @@ -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 } diff --git a/pkg/storage/legacysql/dualwrite/runtime_test.go b/pkg/storage/legacysql/dualwrite/runtime_test.go index 9c148eaf07f..a4b5af5223c 100644 --- a/pkg/storage/legacysql/dualwrite/runtime_test.go +++ b/pkg/storage/legacysql/dualwrite/runtime_test.go @@ -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) diff --git a/pkg/storage/legacysql/dualwrite/static.go b/pkg/storage/legacysql/dualwrite/static.go index d74e2a5b914..db71d338377 100644 --- a/pkg/storage/legacysql/dualwrite/static.go +++ b/pkg/storage/legacysql/dualwrite/static.go @@ -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. diff --git a/pkg/storage/legacysql/dualwrite/storage_mocks_test.go b/pkg/storage/legacysql/dualwrite/storage_mocks_test.go index b3905d1c3d3..409980f59c1 100644 --- a/pkg/storage/legacysql/dualwrite/storage_mocks_test.go +++ b/pkg/storage/legacysql/dualwrite/storage_mocks_test.go @@ -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{} }