diff --git a/pkg/apiserver/go.mod b/pkg/apiserver/go.mod index e1186be8ac2..897519e8a68 100644 --- a/pkg/apiserver/go.mod +++ b/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 diff --git a/pkg/services/apiserver/service.go b/pkg/services/apiserver/service.go index c5ae9a3cd35..1caacecfdf3 100644 --- a/pkg/services/apiserver/service.go +++ b/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: diff --git a/pkg/services/apiserver/storage/entity/test/watch_test.go b/pkg/services/apiserver/storage/entity/test/watch_test.go new file mode 100644 index 00000000000..d826895a226 --- /dev/null +++ b/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{} +} diff --git a/pkg/storage/unified/resource/rv.go b/pkg/storage/unified/resource/rv.go deleted file mode 100644 index 88ec63e4567..00000000000 --- a/pkg/storage/unified/resource/rv.go +++ /dev/null @@ -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) - } -} diff --git a/pkg/storage/unified/resource/testdata/01_create_playlist.json b/pkg/storage/unified/resource/testdata/01_create_playlist.json deleted file mode 100644 index 151435fe1c6..00000000000 --- a/pkg/storage/unified/resource/testdata/01_create_playlist.json +++ /dev/null @@ -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" - } - ] - } -} \ No newline at end of file diff --git a/pkg/tests/apis/dashboard/dashboards_test.go b/pkg/tests/apis/dashboard/dashboards_test.go index fe6b59bc7b4..8ff9569d91e 100644 --- a/pkg/tests/apis/dashboard/dashboards_test.go +++ b/pkg/tests/apis/dashboard/dashboards_test.go @@ -47,59 +47,58 @@ 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, `[ - { - "freshness": "Current", - "resources": [ - { - "resource": "dashboards", - "responseKind": { - "group": "", - "kind": "Dashboard", - "version": "" - }, - "scope": "Namespaced", - "singularResource": "dashboard", - "subresources": [ - { - "responseKind": { - "group": "", - "kind": "DashboardWithAccessInfo", - "version": "" - }, - "subresource": "dto", - "verbs": [ - "get" - ] - }, - { - "responseKind": { - "group": "", - "kind": "PartialObjectMetadataList", - "version": "" - }, - "subresource": "history", - "verbs": [ - "get" - ] - } - ], - "verbs": [ - "create", - "delete", - "deletecollection", - "get", - "list", - "patch", - "update", - "watch" - ] - } - ], - "version": "v0alpha1" - } -]`, disco) + { + "freshness": "Current", + "resources": [ + { + "resource": "dashboards", + "responseKind": { + "group": "", + "kind": "Dashboard", + "version": "" + }, + "scope": "Namespaced", + "singularResource": "dashboard", + "subresources": [ + { + "responseKind": { + "group": "", + "kind": "DashboardWithAccessInfo", + "version": "" + }, + "subresource": "dto", + "verbs": [ + "get" + ] + }, + { + "responseKind": { + "group": "", + "kind": "DashboardVersionList", + "version": "" + }, + "subresource": "versions", + "verbs": [ + "get" + ] + } + ], + "verbs": [ + "create", + "delete", + "deletecollection", + "get", + "list", + "patch", + "update" + ] + } + ], + "version": "v0alpha1" + } + ]`, disco) }) }