merge main

pkg/apiserver/go.mod

@@ -4,6 +4,7 @@ go 1.21.10
 
 require (
 	github.com/bwmarrin/snowflake v0.3.0
+	github.com/google/go-cmp v0.6.0
 	github.com/grafana/grafana/pkg/apimachinery v0.0.0-20240701135906-559738ce6ae1
 	github.com/prometheus/client_golang v1.19.1
 	github.com/stretchr/testify v1.9.0
@@ -11,7 +12,9 @@ require (
 	k8s.io/apimachinery v0.29.3
 	k8s.io/apiserver v0.29.2
 	k8s.io/client-go v0.29.3
+	k8s.io/component-base v0.29.2
 	k8s.io/klog/v2 v2.120.1
+	k8s.io/utils v0.0.0-20230726121419-3b25d923346b
 )
 
 require (
@@ -34,7 +37,6 @@ require (
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/gnostic-models v0.6.8 // indirect
-	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/pprof v0.0.0-20240416155748-26353dc0451f // indirect
 	github.com/google/uuid v1.6.0 // indirect
@@ -84,9 +86,7 @@ require (
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/api v0.29.3 // indirect
-	k8s.io/component-base v0.29.2 // indirect
 	k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 // indirect
-	k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect
 	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.28.0 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect

pkg/services/apiserver/service.go

@@ -285,12 +285,8 @@ func (s *service) start(ctx context.Context) error {
 			return err
 		}
 
-		// TODO: determine when to close the connection, we cannot defer it here
-		// defer conn.Close()
-
 		// Create a client instance
 		client := resource.NewResourceStoreClientGRPC(conn)
-
 		serverConfig.Config.RESTOptionsGetter = apistore.NewRESTOptionsGetter(client, o.RecommendedOptions.Etcd.StorageConfig.Codec)
 
 	case grafanaapiserveroptions.StorageTypeUnified, grafanaapiserveroptions.StorageTypeUnifiedGrpc:

pkg/services/apiserver/storage/entity/test/watch_test.go

@@ -0,0 +1,376 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+// Provenance-includes-location: https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go
+// Provenance-includes-license: Apache-2.0
+// Provenance-includes-copyright: The Kubernetes Authors.
+
+package test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/api/apitesting"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apiserver/pkg/apis/example"
+	examplev1 "k8s.io/apiserver/pkg/apis/example/v1"
+	"k8s.io/apiserver/pkg/storage"
+	"k8s.io/apiserver/pkg/storage/storagebackend"
+	"k8s.io/apiserver/pkg/storage/storagebackend/factory"
+
+	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	storagetesting "github.com/grafana/grafana/pkg/apiserver/storage/testing"
+	"github.com/grafana/grafana/pkg/infra/tracing"
+	"github.com/grafana/grafana/pkg/services/apiserver/storage/entity"
+	"github.com/grafana/grafana/pkg/services/featuremgmt"
+	"github.com/grafana/grafana/pkg/services/sqlstore"
+	entityStore "github.com/grafana/grafana/pkg/services/store/entity"
+	"github.com/grafana/grafana/pkg/services/store/entity/db/dbimpl"
+	"github.com/grafana/grafana/pkg/services/store/entity/sqlstash"
+	"github.com/grafana/grafana/pkg/setting"
+	"github.com/grafana/grafana/pkg/tests/testinfra"
+	"github.com/grafana/grafana/pkg/tests/testsuite"
+)
+
+var scheme = runtime.NewScheme()
+var codecs = serializer.NewCodecFactory(scheme)
+
+func TestMain(m *testing.M) {
+	testsuite.Run(m)
+}
+
+func createTestContext(t *testing.T) (entityStore.EntityStoreClient, factory.DestroyFunc) {
+	t.Helper()
+
+	grafDir, cfgPath := testinfra.CreateGrafDir(t, testinfra.GrafanaOpts{
+		EnableFeatureToggles: []string{
+			featuremgmt.FlagGrpcServer,
+			featuremgmt.FlagUnifiedStorage,
+		},
+		AppModeProduction: false,         // required for migrations to run
+		GRPCServerAddress: "127.0.0.1:0", // :0 for choosing the port automatically
+	})
+
+	cfg, err := setting.NewCfgFromArgs(setting.CommandLineArgs{Config: cfgPath, HomePath: grafDir})
+	assert.NoError(t, err)
+
+	featureManager, err := featuremgmt.ProvideManagerService(cfg)
+	assert.NoError(t, err)
+
+	featureToggles := featuremgmt.ProvideToggles(featureManager)
+
+	db := sqlstore.InitTestDBWithMigration(t, nil, sqlstore.InitTestDBOpt{EnsureDefaultOrgAndUser: false})
+	require.NoError(t, err)
+
+	eDB, err := dbimpl.ProvideEntityDB(db, cfg, featureToggles, nil)
+	require.NoError(t, err)
+
+	err = eDB.Init()
+	require.NoError(t, err)
+
+	traceConfig, err := tracing.ParseTracingConfig(cfg)
+	require.NoError(t, err)
+	tracer, err := tracing.ProvideService(traceConfig)
+	require.NoError(t, err)
+	store, err := sqlstash.ProvideSQLEntityServer(eDB, tracer)
+	require.NoError(t, err)
+
+	client := entityStore.NewEntityStoreClientLocal(store)
+
+	return client, func() { store.Stop() }
+}
+
+func init() {
+	metav1.AddToGroupVersion(scheme, metav1.SchemeGroupVersion)
+	utilruntime.Must(example.AddToScheme(scheme))
+	utilruntime.Must(examplev1.AddToScheme(scheme))
+}
+
+type setupOptions struct {
+	codec          runtime.Codec
+	newFunc        func() runtime.Object
+	newListFunc    func() runtime.Object
+	prefix         string
+	resourcePrefix string
+	groupResource  schema.GroupResource
+}
+
+type setupOption func(*setupOptions, *testing.T)
+
+func withDefaults(options *setupOptions, t *testing.T) {
+	options.codec = apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)
+	options.newFunc = newPod
+	options.newListFunc = newPodList
+	options.prefix = t.TempDir()
+	options.resourcePrefix = "/resource/pods"
+	options.groupResource = schema.GroupResource{Resource: "pods"}
+}
+
+var _ setupOption = withDefaults
+
+func testSetup(t *testing.T, opts ...setupOption) (context.Context, storage.Interface, factory.DestroyFunc, error) {
+	setupOpts := setupOptions{}
+	opts = append([]setupOption{withDefaults}, opts...)
+	for _, opt := range opts {
+		opt(&setupOpts, t)
+	}
+
+	config := storagebackend.NewDefaultConfig(setupOpts.prefix, setupOpts.codec)
+
+	client, destroyFunc := createTestContext(t)
+
+	store, _, err := entity.NewStorage(
+		config.ForResource(setupOpts.groupResource),
+		setupOpts.groupResource,
+		client,
+		setupOpts.codec,
+		func(obj runtime.Object) (string, error) {
+			accessor, err := meta.Accessor(obj)
+			if err != nil {
+				return "", err
+			}
+			return storagetesting.KeyFunc(accessor.GetNamespace(), accessor.GetName()), nil
+		},
+		setupOpts.newFunc,
+		setupOpts.newListFunc,
+		storage.DefaultNamespaceScopedAttr,
+	)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	// Test with an admin identity
+	ctx := identity.WithRequester(context.Background(), &identity.StaticRequester{
+		Namespace:      identity.NamespaceUser,
+		Login:          "testuser",
+		UserID:         123,
+		UserUID:        "u123",
+		OrgRole:        identity.RoleAdmin,
+		IsGrafanaAdmin: true, // can do anything
+	})
+
+	return ctx, store, destroyFunc, nil
+}
+
+func TestIntegrationWatch(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	t.Skip("In maintenance")
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestWatch(ctx, t, store)
+}
+
+func TestIntegrationClusterScopedWatch(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	t.Skip("In maintenance")
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestClusterScopedWatch(ctx, t, store)
+}
+
+func TestIntegrationNamespaceScopedWatch(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	t.Skip("In maintenance")
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestNamespaceScopedWatch(ctx, t, store)
+}
+
+func TestIntegrationDeleteTriggerWatch(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	t.Skip("In maintenance")
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestDeleteTriggerWatch(ctx, t, store)
+}
+
+func TestIntegrationWatchFromZero(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	t.Skip("In maintenance")
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestWatchFromZero(ctx, t, store, nil)
+}
+
+// TestWatchFromNonZero tests that
+// - watch from non-0 should just watch changes after given version
+func TestIntegrationWatchFromNonZero(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	t.Skip("In maintenance")
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestWatchFromNonZero(ctx, t, store)
+}
+
+/*
+// TODO this times out, we need to buffer events
+func TestIntegrationDelayedWatchDelivery(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestDelayedWatchDelivery(ctx, t, store)
+}
+*/
+
+/* func TestIntegrationWatchError(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	ctx, store, _ := testSetup(t)
+	storagetesting.RunTestWatchError(ctx, t, &storeWithPrefixTransformer{store})
+} */
+
+func TestIntegrationWatchContextCancel(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	t.Skip("In maintenance")
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestWatchContextCancel(ctx, t, store)
+}
+
+func TestIntegrationWatcherTimeout(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	t.Skip("In maintenance")
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestWatcherTimeout(ctx, t, store)
+}
+
+func TestIntegrationWatchDeleteEventObjectHaveLatestRV(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	t.Skip("In maintenance")
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestWatchDeleteEventObjectHaveLatestRV(ctx, t, store)
+}
+
+// TODO: enable when we support flow control and priority fairness
+/* func TestIntegrationWatchInitializationSignal(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestWatchInitializationSignal(ctx, t, store)
+} */
+
+/* func TestIntegrationProgressNotify(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunOptionalTestProgressNotify(ctx, t, store)
+} */
+
+// TestWatchDispatchBookmarkEvents makes sure that
+// setting allowWatchBookmarks query param against
+// etcd implementation doesn't have any effect.
+func TestIntegrationWatchDispatchBookmarkEvents(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	t.Skip("In maintenance")
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunTestWatchDispatchBookmarkEvents(ctx, t, store, false)
+}
+
+func TestIntegrationSendInitialEventsBackwardCompatibility(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunSendInitialEventsBackwardCompatibility(ctx, t, store)
+}
+
+// TODO this test times out
+func TestIntegrationEtcdWatchSemantics(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	t.Skip("In maintenance")
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunWatchSemantics(ctx, t, store)
+}
+
+/*
+// TODO this test times out
+func TestIntegrationEtcdWatchSemanticInitialEventsExtended(t *testing.T) {
+		if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	ctx, store, destroyFunc, err := testSetup(t)
+	defer destroyFunc()
+	assert.NoError(t, err)
+	storagetesting.RunWatchSemanticInitialEventsExtended(ctx, t, store)
+}
+*/
+
+func newPod() runtime.Object {
+	return &example.Pod{}
+}
+
+func newPodList() runtime.Object {
+	return &example.PodList{}
+}

pkg/storage/unified/resource/rv.go

@@ -1,16 +0,0 @@
-package resource
-
-import "sync/atomic"
-
-// The kubernetes storage.Interface tests expect this to be a sequential progression
-// SnowflakeIDs do not pass the off-the-shelf k8s tests, although they provide totally
-// acceptable values.
-type NextResourceVersion = func() int64
-
-func newResourceVersionCounter(start int64) NextResourceVersion {
-	var counter atomic.Int64
-	_ = counter.Swap(start + 1)
-	return func() int64 {
-		return counter.Add(1)
-	}
-}

pkg/storage/unified/resource/testdata/01_create_playlist.json

@@ -1,23 +0,0 @@
-{
-    "apiVersion": "playlist.grafana.app/v0alpha1",
-    "kind": "Playlist",
-    "metadata": {
-        "name": "fdgsv37qslr0ga",
-        "namespace": "default",
-        "annotations": {
-            "grafana.app/originName": "elsewhere",
-            "grafana.app/originPath": "path/to/item",
-            "grafana.app/originTimestamp": "2024-02-02T00:00:00Z"
-        }
-    },
-    "spec": {
-        "title": "hello",
-        "interval": "5m",
-        "items": [
-            {
-                "type": "dashboard_by_uid",
-                "value": "vmie2cmWz"
-            }
-        ]
-    }
-}

pkg/tests/apis/dashboard/dashboards_test.go

@@ -47,7 +47,7 @@ func TestIntegrationDashboardsApp(t *testing.T) {
 
 	t.Run("Check discovery client", func(t *testing.T) {
 		disco := helper.GetGroupVersionInfoJSON("dashboard.grafana.app")
-		// fmt.Printf("%s", disco)
+		//fmt.Printf("%s", string(disco))
 
 		require.JSONEq(t, `[
 			{
@@ -77,10 +77,10 @@ func TestIntegrationDashboardsApp(t *testing.T) {
 					{
 						"responseKind": {
 						"group": "",
-              "kind": "PartialObjectMetadataList",
+						"kind": "DashboardVersionList",
 						"version": ""
 						},
-            "subresource": "history",
+						"subresource": "versions",
 						"verbs": [
 						"get"
 						]
@@ -93,13 +93,12 @@ func TestIntegrationDashboardsApp(t *testing.T) {
 					"get",
 					"list",
 					"patch",
-          "update",
+					"update"
-          "watch"
 					]
 				}
 				],
 				"version": "v0alpha1"
 			}
-]`, disco)
+			]`, disco)
 	})
 }