Storage: Add mode-specific dual writers (#85551)

* Set up skeleton dual writers for each mode
* Add Create functionality to each of the mode-specific DualWriters
* Add switch for selecting DualWriter

pkg/apiserver/rest/dualwriter.go

@@ -8,7 +8,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apiserver/pkg/registry/rest"
-	"k8s.io/klog/v2"
+	"k8s.io/klog"
 )
 
 var (
@@ -65,6 +65,19 @@ type DualWriter struct {
 	legacy LegacyStorage
 }
 
+type DualWriterMode int
+
+const (
+	Mode1 DualWriterMode = iota
+	Mode2
+	Mode3
+	Mode4
+)
+
+var CurrentMode = Mode2
+
+// #TODO make CurrentMode customisable and specific to each entity
+
 // NewDualWriter returns a new DualWriter.
 func NewDualWriter(legacy LegacyStorage, storage Storage) *DualWriter {
 	return &DualWriter{
@@ -190,3 +203,18 @@ func (u *updateWrapper) Preconditions() *metav1.Preconditions {
 func (u *updateWrapper) UpdatedObject(ctx context.Context, oldObj runtime.Object) (newObj runtime.Object, err error) {
 	return u.updated, nil
 }
+
+func SelectDualWriter(mode DualWriterMode, legacy LegacyStorage, storage Storage) Storage {
+	switch mode {
+	case Mode1:
+		return NewDualWriterMode1(legacy, storage)
+	case Mode2:
+		return NewDualWriterMode2(legacy, storage)
+	case Mode3:
+		return NewDualWriterMode3(legacy, storage)
+	case Mode4:
+		return NewDualWriterMode4(legacy, storage)
+	default:
+		return NewDualWriterMode2(legacy, storage)
+	}
+}

pkg/apiserver/rest/dualwriter_mode1.go

@@ -0,0 +1,30 @@
+package rest
+
+import (
+	"context"
+	"fmt"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/registry/rest"
+)
+
+type DualWriterMode1 struct {
+	DualWriter
+}
+
+// NewDualWriterMode1 returns a new DualWriter in mode 1.
+// Mode 1 represents writing to and reading from LegacyStorage.
+func NewDualWriterMode1(legacy LegacyStorage, storage Storage) *DualWriterMode1 {
+	return &DualWriterMode1{*NewDualWriter(legacy, storage)}
+}
+
+// Create overrides the default behavior of the DualWriter and writes only to LegacyStorage.
+func (d *DualWriterMode1) Create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
+	legacy, ok := d.legacy.(rest.Creater)
+	if !ok {
+		return nil, fmt.Errorf("legacy storage rest.Creater is missing")
+	}
+
+	return legacy.Create(ctx, obj, createValidation, options)
+}

pkg/apiserver/rest/dualwriter_mode2.go

@@ -0,0 +1,79 @@
+package rest
+
+import (
+	"context"
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/registry/rest"
+	"k8s.io/klog"
+)
+
+type DualWriterMode2 struct {
+	DualWriter
+}
+
+// NewDualWriterMode2 returns a new DualWriter in mode 2.
+// Mode 2 represents writing to LegacyStorage and Storage and reading from LegacyStorage.
+func NewDualWriterMode2(legacy LegacyStorage, storage Storage) *DualWriterMode2 {
+	return &DualWriterMode2{*NewDualWriter(legacy, storage)}
+}
+
+// Create overrides the default behavior of the DualWriter and writes to LegacyStorage and Storage.
+func (d *DualWriterMode2) Create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
+	legacy, ok := d.legacy.(rest.Creater)
+	if !ok {
+		return nil, fmt.Errorf("legacy storage rest.Creater is missing")
+	}
+
+	created, err := legacy.Create(ctx, obj, createValidation, options)
+	if err != nil {
+		return created, err
+	}
+
+	c, err := enrichObject(obj, created)
+	if err != nil {
+		return created, err
+	}
+
+	accessor, err := meta.Accessor(c)
+	if err != nil {
+		return created, err
+	}
+	accessor.SetResourceVersion("")
+	accessor.SetUID("")
+
+	rsp, err := d.Storage.Create(ctx, c, createValidation, options)
+	if err != nil {
+		klog.Error("unable to create object in duplicate storage", "error", err, "mode", Mode2)
+	}
+	return rsp, err
+}
+
+func enrichObject(orig, copy runtime.Object) (runtime.Object, error) {
+	accessorC, err := meta.Accessor(copy)
+	if err != nil {
+		return nil, err
+	}
+	accessorO, err := meta.Accessor(orig)
+	if err != nil {
+		return nil, err
+	}
+
+	accessorC.SetLabels(accessorO.GetLabels())
+
+	ac := accessorC.GetAnnotations()
+	for k, v := range accessorO.GetAnnotations() {
+		ac[k] = v
+	}
+	accessorC.SetAnnotations(ac)
+
+	// #TODO set resource version and UID when required (Update for example)
+	// accessorC.SetResourceVersion(accessorO.GetResourceVersion())
+
+	// accessorC.SetUID(accessorO.GetUID())
+
+	return copy, nil
+}

pkg/apiserver/rest/dualwriter_mode3.go

@@ -0,0 +1,39 @@
+package rest
+
+import (
+	"context"
+	"fmt"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/registry/rest"
+	"k8s.io/klog"
+)
+
+type DualWriterMode3 struct {
+	DualWriter
+}
+
+// NewDualWriterMode3 returns a new DualWriter in mode 3.
+// Mode 3 represents writing to LegacyStorage and Storage and reading from Storage.
+func NewDualWriterMode3(legacy LegacyStorage, storage Storage) *DualWriterMode3 {
+	return &DualWriterMode3{*NewDualWriter(legacy, storage)}
+}
+
+// Create overrides the default behavior of the DualWriter and writes to LegacyStorage and Storage.
+func (d *DualWriterMode3) Create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
+	legacy, ok := d.legacy.(rest.Creater)
+	if !ok {
+		return nil, fmt.Errorf("legacy storage rest.Creater is missing")
+	}
+
+	created, err := d.Storage.Create(ctx, obj, createValidation, options)
+	if err != nil {
+		return created, err
+	}
+
+	if _, err := legacy.Create(ctx, obj, createValidation, options); err != nil {
+		klog.Error("unable to create object in legacy storage", "error", err)
+	}
+	return created, nil
+}

pkg/apiserver/rest/dualwriter_mode4.go

@@ -0,0 +1,25 @@
+package rest
+
+import (
+	"context"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/registry/rest"
+)
+
+type DualWriterMode4 struct {
+	DualWriter
+}
+
+// NewDualWriterMode4 returns a new DualWriter in mode 4.
+// Mode 4 represents writing and reading from Storage.
+func NewDualWriterMode4(legacy LegacyStorage, storage Storage) *DualWriterMode4 {
+	return &DualWriterMode4{*NewDualWriter(legacy, storage)}
+}
+
+// Create overrides the default behavior of the DualWriter and writes only to Storage.
+// #TODO remove this once we remove the default DualWriter implementation
+func (d *DualWriterMode4) Create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
+	return d.Storage.Create(ctx, obj, createValidation, options)
+}