K8s: Add etcd tests for dual write (local) (#78161)

.gitignore

@@ -80,6 +80,7 @@ public/css/*.min.css
 
 # any certificates generated by grafana apiserver
 apiserver.local.config/
+default.etcd/
 
 # devenv
 /devenv/docker-compose.yaml

pkg/services/grafana-apiserver/rest/dualwriter.go

@@ -38,6 +38,7 @@ type LegacyStorage interface {
 	rest.Scoper
 	rest.SingularNameProvider
 	rest.TableConvertor
+	rest.Getter
 }
 
 // DualWriter is a storage implementation that writes first to LegacyStorage and then to Storage.
@@ -82,8 +83,15 @@ func (d *DualWriter) Create(ctx context.Context, obj runtime.Object, createValid
 		if err != nil {
 			return nil, err
 		}
-		obj = created // write the updated version
+
-		rsp, err := d.Storage.Create(ctx, obj, createValidation, options)
+		accessor, err := meta.Accessor(created)
+		if err != nil {
+			return created, err
+		}
+		accessor.SetResourceVersion("")
+		accessor.SetUID("")
+
+		rsp, err := d.Storage.Create(ctx, created, createValidation, options)
 		if err != nil {
 			d.log.Error("unable to create object in duplicate storage", "error", err)
 		}
@@ -96,7 +104,7 @@ func (d *DualWriter) Create(ctx context.Context, obj runtime.Object, createValid
 // Update overrides the default behavior of the Storage and writes to both the LegacyStorage and Storage.
 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) {
 	if legacy, ok := d.legacy.(rest.Updater); ok {
-		// Will resource version checking work????
+		// Get the previous version from k8s storage (the one)
 		old, err := d.Get(ctx, name, &metav1.GetOptions{})
 		if err != nil {
 			return nil, false, err

pkg/services/grafana-apiserver/service.go

@@ -437,6 +437,10 @@ func getK8sApiserverVersion() (string, error) {
 		return "", fmt.Errorf("debug.ReadBuildInfo() failed")
 	}
 
+	if len(bi.Deps) == 0 {
+		return "v?.?", nil // this is normal while debugging
+	}
+
 	for _, dep := range bi.Deps {
 		if dep.Path == "k8s.io/apiserver" {
 			if !semver.IsValid(dep.Version) {

pkg/services/grafana-apiserver/storage/file/file.go

@@ -31,6 +31,10 @@ const MaxUpdateAttempts = 30
 
 var _ storage.Interface = (*Storage)(nil)
 
+// Replace with: https://github.com/kubernetes/kubernetes/blob/v1.29.0-alpha.3/staging/src/k8s.io/apiserver/pkg/storage/errors.go#L28
+// When we upgrade to 1.29
+var errResourceVersionSetOnCreate = errors.New("resourceVersion should not be set on objects to be created")
+
 // Storage implements storage.Interface and storage resources as JSON files on disk.
 type Storage struct {
 	root         string
@@ -124,6 +128,9 @@ func (s *Storage) Create(ctx context.Context, key string, obj runtime.Object, ou
 		return err
 	}
 	metaObj.SetSelfLink("")
+	if metaObj.GetResourceVersion() != "" {
+		return errResourceVersionSetOnCreate
+	}
 
 	if err := s.Versioner().UpdateObject(obj, *generatedRV); err != nil {
 		return err

pkg/tests/apis/playlist/playlist_test.go

@@ -20,6 +20,12 @@ import (
 	"github.com/grafana/grafana/pkg/tests/testinfra"
 )
 
+var gvr = schema.GroupVersionResource{
+	Group:    "playlist.grafana.app",
+	Version:  "v0alpha1",
+	Resource: "playlists",
+}
+
 func TestPlaylist(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
@@ -76,7 +82,7 @@ func TestPlaylist(t *testing.T) {
 		}))
 	})
 
-	t.Run("with dual write", func(t *testing.T) {
+	t.Run("with dual write (file)", func(t *testing.T) {
 		doPlaylistTests(t, apis.NewK8sTestHelper(t, testinfra.GrafanaOpts{
 			AppModeProduction:    true,
 			DisableAnonymous:     true,
@@ -87,15 +93,34 @@ func TestPlaylist(t *testing.T) {
 			},
 		}))
 	})
+
+	t.Run("with dual write (etcd)", func(t *testing.T) {
+		// NOTE: running local etcd, that will be wiped clean!
+		t.Skip("local etcd testing")
+
+		helper := apis.NewK8sTestHelper(t, testinfra.GrafanaOpts{
+			AppModeProduction:    true,
+			DisableAnonymous:     true,
+			APIServerStorageType: "etcd", // requires etcd running on localhost:2379
+			EnableFeatureToggles: []string{
+				featuremgmt.FlagGrafanaAPIServer,
+				featuremgmt.FlagKubernetesPlaylists, // Required so that legacy calls are also written
+			},
+		})
+
+		// Clear the collection before starting (etcd)
+		client := helper.GetResourceClient(apis.ResourceClientArgs{
+			User: helper.Org1.Admin,
+			GVR:  gvr,
+		})
+		err := client.Resource.DeleteCollection(context.Background(), metav1.DeleteOptions{}, metav1.ListOptions{})
+		require.NoError(t, err)
+
+		doPlaylistTests(t, helper)
+	})
 }
 
 func doPlaylistTests(t *testing.T, helper *apis.K8sTestHelper) *apis.K8sTestHelper {
-	gvr := schema.GroupVersionResource{
-		Group:    "playlist.grafana.app",
-		Version:  "v0alpha1",
-		Resource: "playlists",
-	}
-
 	t.Run("Check direct List permissions from different org users", func(t *testing.T) {
 		// Check view permissions
 		rsp := helper.List(helper.Org1.Viewer, "default", gvr)

pkg/tests/testinfra/testinfra.go

@@ -347,6 +347,12 @@ func CreateGrafDir(t *testing.T, opts ...GrafanaOpts) (string, string) {
 			require.NoError(t, err)
 			_, err = section.NewKey("storage_type", o.APIServerStorageType)
 			require.NoError(t, err)
+
+			// Hardcoded local etcd until this is needed to run in CI
+			if o.APIServerStorageType == "etcd" {
+				_, err = section.NewKey("etcd_servers", "localhost:2379")
+				require.NoError(t, err)
+			}
 		}
 
 		if o.GRPCServerAddress != "" {