From 1a00801e6aef96c2dfc66e2ef846999ad30b01cb Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Fri, 21 Mar 2025 17:45:25 +0300 Subject: [PATCH] Provisioning: Merge watch support into live (#102618) --- .betterer.results | 5 + packages/grafana-data/src/types/live.ts | 1 + pkg/api/api.go | 1 + pkg/api/dashboard_test.go | 5 +- pkg/apis/provisioning/v0alpha1/jobs.go | 5 +- .../v0alpha1/zz_generated.openapi.go | 9 +- pkg/services/live/features/watch.go | 214 +++++++++++++ pkg/services/live/live.go | 17 +- pkg/services/live/live_test.go | 6 +- .../provisioning.grafana.app-v0alpha1.json | 15 +- .../api/clients/provisioning/endpoints.gen.ts | 8 +- .../utils/createOnCacheEntryAdded.ts | 12 +- public/app/features/apiserver/client.ts | 47 +-- .../app/features/apiserver/discovery.test.ts | 12 + public/app/features/apiserver/discovery.ts | 84 +++++ .../snapshots/discovery-snapshot.json | 295 ++++++++++++++++++ public/app/features/apiserver/types.ts | 12 + .../plugins/panel/live/LiveChannelEditor.tsx | 33 +- 18 files changed, 723 insertions(+), 58 deletions(-) create mode 100644 pkg/services/live/features/watch.go create mode 100644 public/app/features/apiserver/discovery.test.ts create mode 100644 public/app/features/apiserver/discovery.ts create mode 100644 public/app/features/apiserver/snapshots/discovery-snapshot.json diff --git a/.betterer.results b/.betterer.results index f74ac1484c1..75f00fb1529 100644 --- a/.betterer.results +++ b/.betterer.results @@ -6789,6 +6789,11 @@ exports[`better eslint`] = { "public/app/plugins/panel/histogram/utils.ts:5381": [ [0, 0, 0, "\'@grafana/data/src/transformations/transformers/histogram\' import is restricted from being used by a pattern. Import from the public export instead.", "0"] ], + "public/app/plugins/panel/live/LiveChannelEditor.tsx:5381": [ + [0, 0, 0, "Do not use any type assertions.", "0"], + [0, 0, 0, "Do not use any type assertions.", "1"], + [0, 0, 0, "Unexpected any. Specify a different type.", "2"] + ], "public/app/plugins/panel/live/LivePanel.tsx:5381": [ [0, 0, 0, "Do not use any type assertions.", "0"] ], diff --git a/packages/grafana-data/src/types/live.ts b/packages/grafana-data/src/types/live.ts index 565a6d3fcae..7f4af737f05 100644 --- a/packages/grafana-data/src/types/live.ts +++ b/packages/grafana-data/src/types/live.ts @@ -12,6 +12,7 @@ export enum LiveChannelScope { Plugin = 'plugin', // namespace = plugin name (singleton works for apps too) Grafana = 'grafana', // namespace = feature Stream = 'stream', // namespace = id for the managed data stream + Watch = 'watch', // namespace = k8s group we will watch } /** diff --git a/pkg/api/api.go b/pkg/api/api.go index 3aefe35ad22..0026529cbd8 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -117,6 +117,7 @@ func (hs *HTTPServer) registerRoutes() { r.Get("/admin/orgs/edit/:id", authorizeInOrg(ac.UseGlobalOrg, ac.OrgsAccessEvaluator), hs.Index) r.Get("/admin/stats", authorize(ac.EvalPermission(ac.ActionServerStatsRead)), hs.Index) r.Get("/admin/provisioning", reqOrgAdmin, hs.Index) + r.Get("/admin/provisioning/*", reqOrgAdmin, hs.Index) if hs.Features.IsEnabledGlobally(featuremgmt.FlagOnPremToCloudMigrations) { r.Get("/admin/migrate-to-cloud", authorize(cloudmigration.MigrationAssistantAccess), hs.Index) diff --git a/pkg/api/dashboard_test.go b/pkg/api/dashboard_test.go index b715e17481f..549532806bc 100644 --- a/pkg/api/dashboard_test.go +++ b/pkg/api/dashboard_test.go @@ -135,7 +135,10 @@ func newTestLive(t *testing.T, store db.DB) *live.GrafanaLive { nil, &usagestats.UsageStatsMock{T: t}, nil, - features, acimpl.ProvideAccessControl(features), &dashboards.FakeDashboardService{}, annotationstest.NewFakeAnnotationsRepo(), nil) + features, acimpl.ProvideAccessControl(features), + &dashboards.FakeDashboardService{}, + annotationstest.NewFakeAnnotationsRepo(), + nil, nil) require.NoError(t, err) return gLive } diff --git a/pkg/apis/provisioning/v0alpha1/jobs.go b/pkg/apis/provisioning/v0alpha1/jobs.go index d621bf3411f..1e075fd1c26 100644 --- a/pkg/apis/provisioning/v0alpha1/jobs.go +++ b/pkg/apis/provisioning/v0alpha1/jobs.go @@ -104,16 +104,13 @@ type ExportJobOptions struct { Branch string `json:"branch,omitempty"` // Prefix in target file system - Prefix string `json:"prefix,omitempty"` + Path string `json:"path,omitempty"` // Include the identifier in the exported metadata Identifier bool `json:"identifier"` } type MigrateJobOptions struct { - // Target file prefix - Prefix string `json:"prefix,omitempty"` - // Preserve history (if possible) History bool `json:"history,omitempty"` diff --git a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go b/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go index 571f26b2b27..6ddc883ebda 100644 --- a/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go +++ b/pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go @@ -108,7 +108,7 @@ func schema_pkg_apis_provisioning_v0alpha1_ExportJobOptions(ref common.Reference Format: "", }, }, - "prefix": { + "path": { SchemaProps: spec.SchemaProps{ Description: "Prefix in target file system", Type: []string{"string"}, @@ -818,13 +818,6 @@ func schema_pkg_apis_provisioning_v0alpha1_MigrateJobOptions(ref common.Referenc SchemaProps: spec.SchemaProps{ Type: []string{"object"}, Properties: map[string]spec.Schema{ - "prefix": { - SchemaProps: spec.SchemaProps{ - Description: "Target file prefix", - Type: []string{"string"}, - Format: "", - }, - }, "history": { SchemaProps: spec.SchemaProps{ Description: "Preserve history (if possible)", diff --git a/pkg/services/live/features/watch.go b/pkg/services/live/features/watch.go new file mode 100644 index 00000000000..05a7f875518 --- /dev/null +++ b/pkg/services/live/features/watch.go @@ -0,0 +1,214 @@ +package features + +import ( + "context" + "fmt" + "strings" + "sync" + + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/dynamic" + + provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1" + + "github.com/grafana/authlib/types" + "github.com/grafana/grafana-app-sdk/logging" + "github.com/grafana/grafana-plugin-sdk-go/backend" + "github.com/grafana/grafana-plugin-sdk-go/data/utils/jsoniter" + "github.com/grafana/grafana-plugin-sdk-go/live" + + "github.com/grafana/grafana/pkg/apimachinery/identity" + "github.com/grafana/grafana/pkg/services/apiserver" + "github.com/grafana/grafana/pkg/services/live/model" +) + +// WatchRunner will start a watch task and broadcast results +type WatchRunner struct { + publisher model.ChannelPublisher + configProvider apiserver.RestConfigProvider + + watchingMu sync.Mutex + watching map[string]*watcher +} + +func NewWatchRunner(publisher model.ChannelPublisher, configProvider apiserver.RestConfigProvider) *WatchRunner { + return &WatchRunner{ + publisher: publisher, + configProvider: configProvider, + watching: make(map[string]*watcher), + } +} + +func (b *WatchRunner) GetHandlerForPath(_ string) (model.ChannelHandler, error) { + return b, nil // all dashboards share the same handler +} + +// Valid paths look like: {version}/{resource}[={name}]/{user.uid} +// * v0alpha1/dashboards/u12345 +// * v0alpha1/dashboards=ABCD/u12345 +func (b *WatchRunner) OnSubscribe(ctx context.Context, u identity.Requester, e model.SubscribeEvent) (model.SubscribeReply, backend.SubscribeStreamStatus, error) { + // To make sure we do not share resources across users, in clude the UID in the path + userID := u.GetIdentifier() + if userID == "" { + return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, fmt.Errorf("missing user identity") + } + if !strings.HasSuffix(e.Path, userID) { + return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, fmt.Errorf("path must end with user uid (%s)", userID) + } + + // While testing with provisioning repositories, we will limit this to admin only + if !u.HasRole(identity.RoleAdmin) { + return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, fmt.Errorf("only admin users for now") + } + + b.watchingMu.Lock() + defer b.watchingMu.Unlock() + + current, ok := b.watching[e.Channel] + if ok && !current.done { + return model.SubscribeReply{ + JoinLeave: false, + Presence: false, + Recover: false, + }, backend.SubscribeStreamStatusOK, nil + } + + // Try to start a watcher for this request + gvr, name, err := parseWatchRequest(e.Channel, userID) + if err != nil { + return model.SubscribeReply{}, backend.SubscribeStreamStatusNotFound, err + } + + // Test this with only provisiong support -- then we can evaluate a broader rollout + if gvr.Group != provisioning.GROUP { + return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, + fmt.Errorf("watching provisioned resources is OK allowed (for now)") + } + + requester := types.WithAuthInfo(context.Background(), u) + cfg, err := b.configProvider.GetRestConfig(requester) + if err != nil { + return model.SubscribeReply{}, backend.SubscribeStreamStatusNotFound, err + } + uclient, err := dynamic.NewForConfig(cfg) + if err != nil { + return model.SubscribeReply{}, backend.SubscribeStreamStatusNotFound, err + } + client := uclient.Resource(gvr).Namespace(u.GetNamespace()) + + opts := v1.ListOptions{} + if len(name) > 1 { + opts.FieldSelector = "metadata.name=" + name + } + watch, err := client.Watch(requester, opts) + if err != nil { + return model.SubscribeReply{}, backend.SubscribeStreamStatusNotFound, err + } + + current = &watcher{ + orgId: u.GetOrgID(), + channel: e.Channel, + publisher: b.publisher, + watch: watch, + } + + b.watching[e.Channel] = current + go current.run(ctx) + + return model.SubscribeReply{ + JoinLeave: false, // need unsubscribe envents + Presence: false, + Recover: false, + }, backend.SubscribeStreamStatusOK, nil +} + +func parseWatchRequest(channel string, user string) (gvr schema.GroupVersionResource, name string, err error) { + addr, err := live.ParseChannel(channel) + if err != nil { + return gvr, "", err + } + + parts := strings.Split(addr.Path, "/") + if len(parts) != 3 { + return gvr, "", fmt.Errorf("expecting path: {version}/{resource}={name}/{user}") + } + if parts[2] != user { + return gvr, "", fmt.Errorf("expecting user suffix: %s", user) + } + + resource := strings.Split(parts[1], "=") + gvr = schema.GroupVersionResource{ + Group: addr.Namespace, + Version: parts[0], + Resource: resource[0], + } + if len(resource) > 1 { + name = resource[1] + } + return gvr, name, nil +} + +// OnPublish is called when a client wants to broadcast on the websocket +func (b *WatchRunner) OnPublish(_ context.Context, u identity.Requester, e model.PublishEvent) (model.PublishReply, backend.PublishStreamStatus, error) { + return model.PublishReply{}, backend.PublishStreamStatusNotFound, fmt.Errorf("watch does not support publish") +} + +type watcher struct { + orgId int64 + channel string + publisher model.ChannelPublisher + done bool + watch watch.Interface +} + +func (b *watcher) run(ctx context.Context) { + logger := logging.FromContext(ctx).With("channel", b.channel) + + ch := b.watch.ResultChan() + for { + select { + // This is sent when there are no longer any subscriptions + case <-ctx.Done(): + logger.Info("context done", "channel", b.channel) + b.watch.Stop() + b.done = true + return + + // Each watch event + case event, ok := <-ch: + if !ok { + logger.Info("watch stream broken", "channel", b.channel) + b.watch.Stop() + b.done = true // will force reconnect from the frontend + return + } + + cfg := jsoniter.ConfigCompatibleWithStandardLibrary + stream := cfg.BorrowStream(nil) + defer cfg.ReturnStream(stream) + + // regular json.Marshal() uses upper case + stream.WriteObjectStart() + stream.WriteObjectField("type") + stream.WriteString(string(event.Type)) + stream.WriteMore() + stream.WriteObjectField("object") + stream.WriteVal(event.Object) + stream.WriteObjectEnd() + + buf := stream.Buffer() + data := make([]byte, len(buf)) + copy(data, buf) + + err := b.publisher(b.orgId, b.channel, data) + if err != nil { + logger.Error("publish error", "channel", b.channel, "err", err) + b.watch.Stop() + b.done = true // will force reconnect from the frontend + continue + } + } + } +} diff --git a/pkg/services/live/live.go b/pkg/services/live/live.go index 64efaaa18b6..eb854f99d9c 100644 --- a/pkg/services/live/live.go +++ b/pkg/services/live/live.go @@ -36,6 +36,7 @@ import ( "github.com/grafana/grafana/pkg/plugins" "github.com/grafana/grafana/pkg/services/accesscontrol" "github.com/grafana/grafana/pkg/services/annotations" + "github.com/grafana/grafana/pkg/services/apiserver" contextmodel "github.com/grafana/grafana/pkg/services/contexthandler/model" "github.com/grafana/grafana/pkg/services/dashboards" "github.com/grafana/grafana/pkg/services/datasources" @@ -79,7 +80,7 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r dataSourceCache datasources.CacheService, sqlStore db.DB, secretsService secrets.Service, usageStatsService usagestats.Service, queryDataService query.Service, toggles featuremgmt.FeatureToggles, accessControl accesscontrol.AccessControl, dashboardService dashboards.DashboardService, annotationsRepo annotations.Repository, - orgService org.Service) (*GrafanaLive, error) { + orgService org.Service, configProvider apiserver.RestConfigProvider) (*GrafanaLive, error) { g := &GrafanaLive{ Cfg: cfg, Features: toggles, @@ -191,6 +192,11 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r g.GrafanaScope.Features["dashboard"] = dash g.GrafanaScope.Features["broadcast"] = features.NewBroadcastRunner(g.storage) + // Testing watch with just the provisioning support -- this will be removed when it is well validated + if toggles.IsEnabledGlobally(featuremgmt.FlagProvisioning) { + g.GrafanaScope.Features["watch"] = features.NewWatchRunner(g.Publish, configProvider) + } + g.surveyCaller = survey.NewCaller(managedStreamRunner, node) err = g.surveyCaller.SetupHandlers() if err != nil { @@ -889,6 +895,8 @@ func (g *GrafanaLive) GetChannelHandlerFactory(ctx context.Context, user identit switch scope { case live.ScopeGrafana: return g.handleGrafanaScope(user, namespace) + case "watch": // TODO: live.ScopeWatch: update 275 https://github.com/grafana/grafana-plugin-sdk-go/releases + return g.handleWatchScope() case live.ScopePlugin: return g.handlePluginScope(ctx, user, namespace) case live.ScopeDatasource: @@ -907,6 +915,13 @@ func (g *GrafanaLive) handleGrafanaScope(_ identity.Requester, namespace string) return nil, fmt.Errorf("unknown feature: %q", namespace) } +func (g *GrafanaLive) handleWatchScope() (model.ChannelHandlerFactory, error) { + if p, ok := g.GrafanaScope.Features["watch"]; ok { + return p, nil + } + return nil, fmt.Errorf("watch not registered") +} + func (g *GrafanaLive) handlePluginScope(ctx context.Context, _ identity.Requester, namespace string) (model.ChannelHandlerFactory, error) { streamHandler, err := g.getStreamPlugin(ctx, namespace) if err != nil { diff --git a/pkg/services/live/live_test.go b/pkg/services/live/live_test.go index 447910e1b84..c002ec7c705 100644 --- a/pkg/services/live/live_test.go +++ b/pkg/services/live/live_test.go @@ -36,7 +36,11 @@ func Test_provideLiveService_RedisUnavailable(t *testing.T) { nil, &usagestats.UsageStatsMock{T: t}, nil, - featuremgmt.WithFeatures(), acimpl.ProvideAccessControl(featuremgmt.WithFeatures()), &dashboards.FakeDashboardService{}, annotationstest.NewFakeAnnotationsRepo(), nil) + featuremgmt.WithFeatures(), + acimpl.ProvideAccessControl(featuremgmt.WithFeatures()), + &dashboards.FakeDashboardService{}, + annotationstest.NewFakeAnnotationsRepo(), + nil, nil) // Proceeds without live HA if redis is unavaialble require.NoError(t, err) diff --git a/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json b/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json index d74ca23d0be..5121b7861eb 100644 --- a/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json +++ b/pkg/tests/apis/openapi_snapshots/provisioning.grafana.app-v0alpha1.json @@ -1121,7 +1121,7 @@ "type": "boolean", "default": false }, - "prefix": { + "path": { "description": "Prefix in target file system", "type": "string" } @@ -1130,7 +1130,7 @@ "example": { "folder": "grafan-folder-ref", "branch": "target-branch", - "prefix": "prefix/in/repo/tree", + "path": "path/in/tree", "identifier": false } } @@ -1780,15 +1780,10 @@ "description": "Include the identifier in the exported metadata", "type": "boolean", "default": false - }, - "prefix": { - "description": "Target file prefix", - "type": "string" } } }, "example": { - "prefix": "prefix/in/repo/tree", "history": true, "identifier": false } @@ -2567,7 +2562,7 @@ "type": "boolean", "default": false }, - "prefix": { + "path": { "description": "Prefix in target file system", "type": "string" } @@ -3035,10 +3030,6 @@ "description": "Include the identifier in the exported metadata", "type": "boolean", "default": false - }, - "prefix": { - "description": "Target file prefix", - "type": "string" } } }, diff --git a/public/app/api/clients/provisioning/endpoints.gen.ts b/public/app/api/clients/provisioning/endpoints.gen.ts index 8f22200d13f..a7baef15950 100644 --- a/public/app/api/clients/provisioning/endpoints.gen.ts +++ b/public/app/api/clients/provisioning/endpoints.gen.ts @@ -497,7 +497,7 @@ export type CreateRepositoryExportApiArg = { /** Include the identifier in the exported metadata */ identifier: boolean; /** Prefix in target file system */ - prefix?: string; + path?: string; }; }; export type GetRepositoryFilesApiResponse = /** status 200 OK */ { @@ -587,8 +587,6 @@ export type CreateRepositoryMigrateApiArg = { history?: boolean; /** Include the identifier in the exported metadata */ identifier: boolean; - /** Target file prefix */ - prefix?: string; }; }; export type GetRepositoryRenderWithPathApiResponse = unknown; @@ -748,8 +746,6 @@ export type MigrateJobOptions = { history?: boolean; /** Include the identifier in the exported metadata */ identifier: boolean; - /** Target file prefix */ - prefix?: string; }; export type PullRequestJobOptions = { hash?: string; @@ -772,7 +768,7 @@ export type ExportJobOptions = { /** Include the identifier in the exported metadata */ identifier: boolean; /** Prefix in target file system */ - prefix?: string; + path?: string; }; export type JobSpec = { /** Possible enum values: diff --git a/public/app/api/clients/provisioning/utils/createOnCacheEntryAdded.ts b/public/app/api/clients/provisioning/utils/createOnCacheEntryAdded.ts index 724eb79193d..c3c4f38b8e1 100644 --- a/public/app/api/clients/provisioning/utils/createOnCacheEntryAdded.ts +++ b/public/app/api/clients/provisioning/utils/createOnCacheEntryAdded.ts @@ -48,17 +48,17 @@ export function createOnCacheEntryAdded< } const existingIndex = draft.items.findIndex((item) => item.metadata?.name === event.object.metadata.name); - if (event.type === 'ADDED') { - // Add the new item + if (event.type === 'ADDED' && existingIndex === -1) { // eslint-disable-next-line @typescript-eslint/consistent-type-assertions draft.items.push(event.object as unknown as T); - } else if (event.type === 'MODIFIED' && existingIndex !== -1) { - // Update the existing item if it exists - // eslint-disable-next-line @typescript-eslint/consistent-type-assertions - draft.items[existingIndex] = event.object as unknown as T; } else if (event.type === 'DELETED' && existingIndex !== -1) { // Remove the item if it exists draft.items.splice(existingIndex, 1); + } else if (existingIndex !== -1) { + // Could be ADDED or MODIFIED + // Update the existing item if it exists + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + draft.items[existingIndex] = event.object as unknown as T; } }); }); diff --git a/public/app/features/apiserver/client.ts b/public/app/features/apiserver/client.ts index cb53c57aa65..2af5503674f 100644 --- a/public/app/features/apiserver/client.ts +++ b/public/app/features/apiserver/client.ts @@ -1,6 +1,7 @@ import { Observable, from, retry, catchError, filter, map, mergeMap } from 'rxjs'; -import { BackendSrvRequest, config, getBackendSrv } from '@grafana/runtime'; +import { isLiveChannelMessageEvent, LiveChannelScope } from '@grafana/data'; +import { config, getBackendSrv, getGrafanaLiveSrv } from '@grafana/runtime'; import { contextSrv } from 'app/core/core'; import { getAPINamespace } from '../../api/utils'; @@ -19,20 +20,16 @@ import { K8sAPIGroupList, AnnoKeySavedFromUI, ResourceEvent, + GroupVersionResource, } from './types'; -export interface GroupVersionResource { - group: string; - version: string; - resource: string; -} - export class ScopedResourceClient implements ResourceClient { readonly url: string; + readonly gvr: GroupVersionResource; constructor(gvr: GroupVersionResource, namespaced = true) { const ns = namespaced ? `namespaces/${getAPINamespace()}/` : ''; - + this.gvr = gvr; this.url = `/apis/${gvr.group}/${gvr.version}/${ns}${gvr.resource}`; } @@ -40,26 +37,40 @@ export class ScopedResourceClient implements return getBackendSrv().get>(`${this.url}/${name}`); } - public watch( - params?: WatchOptions, - config?: Pick - ): Observable> { - const decoder = new TextDecoder(); - const { name, ...rest } = params ?? {}; // name needs to be added to fieldSelector + public watch(params?: WatchOptions): Observable> { const requestParams = { - ...rest, watch: true, labelSelector: this.parseListOptionsSelector(params?.labelSelector), fieldSelector: this.parseListOptionsSelector(params?.fieldSelector), }; - if (name) { + if (params?.name) { requestParams.fieldSelector = `metadata.name=${name}`; } + + // For now, watch over live only supports provisioning + if (this.gvr.group === 'provisioning.grafana.app') { + let query = ''; + if (requestParams.fieldSelector?.startsWith('metadata.name=')) { + query = requestParams.fieldSelector.substring('metadata.name'.length); + } + return getGrafanaLiveSrv() + .getStream>({ + scope: LiveChannelScope.Watch, + namespace: this.gvr.group, + path: `${this.gvr.version}/${this.gvr.resource}${query}/${config.bootData.user.uid}`, + }) + .pipe( + filter((event) => isLiveChannelMessageEvent(event)), + map((event) => event.message) + ); + } + + const decoder = new TextDecoder(); return getBackendSrv() .chunked({ url: this.url, params: requestParams, - ...config, + method: 'GET', }) .pipe( filter((response) => response.ok && response.data instanceof Uint8Array), @@ -73,7 +84,7 @@ export class ScopedResourceClient implements try { return JSON.parse(line); } catch (e) { - console.warn('Invalid JSON in watch stream:', e); + console.warn('Invalid JSON in watch stream:', e, line); return null; } }), diff --git a/public/app/features/apiserver/discovery.test.ts b/public/app/features/apiserver/discovery.test.ts new file mode 100644 index 00000000000..5b784197eea --- /dev/null +++ b/public/app/features/apiserver/discovery.test.ts @@ -0,0 +1,12 @@ +import { discoveryResources } from './discovery'; + +const discoverySnapshot = require('./snapshots/discovery-snapshot.json'); + +describe('simple typescript tests', () => { + it('simple', async () => { + const watchable = discoveryResources(discoverySnapshot) + .filter((v) => v.verbs.includes('watch')) + .map((v) => v.resource); + expect(watchable).toEqual(['user-storage', 'dashboards', 'dashboards', 'dashboards']); + }); +}); diff --git a/public/app/features/apiserver/discovery.ts b/public/app/features/apiserver/discovery.ts new file mode 100644 index 00000000000..ece21681d48 --- /dev/null +++ b/public/app/features/apiserver/discovery.ts @@ -0,0 +1,84 @@ +import { lastValueFrom, map } from 'rxjs'; + +import { FetchResponse, getBackendSrv } from '@grafana/runtime'; + +import { GroupVersionKind, ListMeta } from './types'; + +export type GroupDiscoveryResource = { + resource: string; + responseKind: GroupVersionKind; + scope: 'Namespaced' | 'Cluster'; + singularResource: string; + verbs: string[]; + subresources?: GroupDiscoverySubresource[]; +}; + +export type GroupDiscoverySubresource = { + subresource: string; + responseKind: GroupVersionKind; + verbs: string[]; +}; + +export type GroupDiscoveryVersion = { + version: string; + freshness: 'Current' | string; + resources: GroupDiscoveryResource[]; +}; + +export type GroupDiscoveryItem = { + metadata: { + name: string; + }; + versions: GroupDiscoveryVersion[]; +}; + +export type APIGroupDiscoveryList = { + metadata: ListMeta; + items: GroupDiscoveryItem[]; +}; + +export async function getAPIGroupDiscoveryList(): Promise { + return await lastValueFrom( + getBackendSrv() + .fetch({ + method: 'GET', + url: '/apis', + headers: { + Accept: + 'application/json;g=apidiscovery.k8s.io;v=v2;as=APIGroupDiscoveryList,application/json;g=apidiscovery.k8s.io;v=v2beta1;as=APIGroupDiscoveryList,application/json', + }, + }) + .pipe( + map((response: FetchResponse) => { + // Fill in the group+version before returning + for (let api of response.data.items) { + for (let version of api.versions) { + for (let resource of version.resources) { + resource.responseKind.group = api.metadata.name; + resource.responseKind.version = version.version; + if (resource.subresources) { + for (let sub of resource.subresources) { + sub.responseKind.group = api.metadata.name; + sub.responseKind.version = version.version; + } + } + } + } + } + return response.data; + }) + ) + ); +} + +export function discoveryResources(apis: APIGroupDiscoveryList): GroupDiscoveryResource[] { + const resources: GroupDiscoveryResource[] = []; + for (let api of apis.items) { + for (let version of api.versions) { + for (let resource of version.resources) { + resources.push(resource); + } + } + } + return resources; +} diff --git a/public/app/features/apiserver/snapshots/discovery-snapshot.json b/public/app/features/apiserver/snapshots/discovery-snapshot.json new file mode 100644 index 00000000000..36fc7f05aa3 --- /dev/null +++ b/public/app/features/apiserver/snapshots/discovery-snapshot.json @@ -0,0 +1,295 @@ +{ + "kind": "APIGroupDiscoveryList", + "apiVersion": "apidiscovery.k8s.io/v2", + "metadata": {}, + "items": [ + { + "metadata": { "name": "userstorage.grafana.app", "creationTimestamp": null }, + "versions": [ + { + "version": "v0alpha1", + "resources": [ + { + "resource": "user-storage", + "responseKind": { "group": "", "version": "", "kind": "UserStorage" }, + "scope": "Namespaced", + "singularResource": "user-storage", + "verbs": ["create", "delete", "deletecollection", "get", "list", "patch", "update", "watch"] + } + ], + "freshness": "Current" + } + ] + }, + { + "metadata": { "name": "notifications.alerting.grafana.app", "creationTimestamp": null }, + "versions": [ + { + "version": "v0alpha1", + "resources": [ + { + "resource": "receivers", + "responseKind": { "group": "", "version": "", "kind": "Receiver" }, + "scope": "Namespaced", + "singularResource": "receiver", + "verbs": ["create", "delete", "deletecollection", "get", "list", "patch", "update"] + }, + { + "resource": "routingtrees", + "responseKind": { "group": "", "version": "", "kind": "RoutingTree" }, + "scope": "Namespaced", + "singularResource": "routingtree", + "verbs": ["create", "delete", "deletecollection", "get", "list", "patch", "update"] + }, + { + "resource": "templategroups", + "responseKind": { "group": "", "version": "", "kind": "TemplateGroup" }, + "scope": "Namespaced", + "singularResource": "templategroup", + "verbs": ["create", "delete", "deletecollection", "get", "list", "patch", "update"] + }, + { + "resource": "timeintervals", + "responseKind": { "group": "", "version": "", "kind": "TimeInterval" }, + "scope": "Namespaced", + "singularResource": "timeinterval", + "verbs": ["create", "delete", "deletecollection", "get", "list", "patch", "update"] + } + ], + "freshness": "Current" + } + ] + }, + { + "metadata": { "name": "iam.grafana.app", "creationTimestamp": null }, + "versions": [ + { + "version": "v0alpha1", + "resources": [ + { + "resource": "serviceaccounts", + "responseKind": { "group": "", "version": "", "kind": "ServiceAccount" }, + "scope": "Namespaced", + "singularResource": "serviceaccount", + "verbs": ["get", "list"], + "subresources": [ + { + "subresource": "tokens", + "responseKind": { "group": "", "version": "", "kind": "UserTeamList" }, + "verbs": ["get"] + } + ] + }, + { + "resource": "ssosettings", + "responseKind": { "group": "", "version": "", "kind": "SSOSetting" }, + "scope": "Namespaced", + "singularResource": "ssosetting", + "verbs": ["delete", "get", "list", "patch", "update"] + }, + { + "resource": "teambindings", + "responseKind": { "group": "", "version": "", "kind": "TeamBinding" }, + "scope": "Namespaced", + "singularResource": "teambinding", + "verbs": ["get", "list"] + }, + { + "resource": "teams", + "responseKind": { "group": "", "version": "", "kind": "Team" }, + "scope": "Namespaced", + "singularResource": "team", + "verbs": ["get", "list"], + "subresources": [ + { + "subresource": "members", + "responseKind": { "group": "", "version": "", "kind": "TeamMemberList" }, + "verbs": ["get"] + } + ] + }, + { + "resource": "users", + "responseKind": { "group": "", "version": "", "kind": "User" }, + "scope": "Namespaced", + "singularResource": "user", + "verbs": ["get", "list"], + "subresources": [ + { + "subresource": "teams", + "responseKind": { "group": "", "version": "", "kind": "UserTeamList" }, + "verbs": ["get"] + } + ] + } + ], + "freshness": "Current" + } + ] + }, + { + "metadata": { "name": "folder.grafana.app", "creationTimestamp": null }, + "versions": [ + { + "version": "v0alpha1", + "resources": [ + { + "resource": "folders", + "responseKind": { "group": "", "version": "", "kind": "Folder" }, + "scope": "Namespaced", + "singularResource": "folder", + "verbs": ["create", "delete", "deletecollection", "get", "list", "patch", "update"], + "subresources": [ + { + "subresource": "access", + "responseKind": { "group": "", "version": "", "kind": "FolderAccessInfo" }, + "verbs": ["get"] + }, + { + "subresource": "counts", + "responseKind": { "group": "", "version": "", "kind": "DescendantCounts" }, + "verbs": ["get"] + }, + { + "subresource": "parents", + "responseKind": { "group": "", "version": "", "kind": "FolderInfoList" }, + "verbs": ["get"] + } + ] + } + ], + "freshness": "Current" + } + ] + }, + { + "metadata": { "name": "featuretoggle.grafana.app", "creationTimestamp": null }, + "versions": [ + { + "version": "v0alpha1", + "resources": [ + { + "resource": "features", + "responseKind": { "group": "", "version": "", "kind": "Feature" }, + "scope": "Cluster", + "singularResource": "feature", + "verbs": ["get", "list"] + }, + { + "resource": "featuretoggles", + "responseKind": { "group": "", "version": "", "kind": "FeatureToggles" }, + "scope": "Namespaced", + "singularResource": "featuretoggle", + "verbs": ["get", "list"] + } + ], + "freshness": "Current" + } + ] + }, + { + "metadata": { "name": "dashboard.grafana.app", "creationTimestamp": null }, + "versions": [ + { + "version": "v0alpha1", + "resources": [ + { + "resource": "dashboards", + "responseKind": { "group": "", "version": "", "kind": "Dashboard" }, + "scope": "Namespaced", + "singularResource": "dashboard", + "verbs": ["create", "delete", "deletecollection", "get", "list", "patch", "update", "watch"], + "subresources": [ + { + "subresource": "dto", + "responseKind": { "group": "", "version": "", "kind": "DashboardWithAccessInfo" }, + "verbs": ["get"] + } + ] + }, + { + "resource": "librarypanels", + "responseKind": { "group": "", "version": "", "kind": "LibraryPanel" }, + "scope": "Namespaced", + "singularResource": "librarypanel", + "verbs": ["get", "list"] + } + ], + "freshness": "Current" + }, + { + "version": "v1alpha1", + "resources": [ + { + "resource": "dashboards", + "responseKind": { "group": "", "version": "", "kind": "Dashboard" }, + "scope": "Namespaced", + "singularResource": "dashboard", + "verbs": ["create", "delete", "deletecollection", "get", "list", "patch", "update", "watch"], + "subresources": [ + { + "subresource": "dto", + "responseKind": { "group": "", "version": "", "kind": "DashboardWithAccessInfo" }, + "verbs": ["get"] + } + ] + }, + { + "resource": "librarypanels", + "responseKind": { "group": "", "version": "", "kind": "LibraryPanel" }, + "scope": "Namespaced", + "singularResource": "librarypanel", + "verbs": ["get", "list"] + } + ], + "freshness": "Current" + }, + { + "version": "v2alpha1", + "resources": [ + { + "resource": "dashboards", + "responseKind": { "group": "", "version": "", "kind": "Dashboard" }, + "scope": "Namespaced", + "singularResource": "dashboard", + "verbs": ["create", "delete", "deletecollection", "get", "list", "patch", "update", "watch"], + "subresources": [ + { + "subresource": "dto", + "responseKind": { "group": "", "version": "", "kind": "DashboardWithAccessInfo" }, + "verbs": ["get"] + } + ] + }, + { + "resource": "librarypanels", + "responseKind": { "group": "", "version": "", "kind": "LibraryPanel" }, + "scope": "Namespaced", + "singularResource": "librarypanel", + "verbs": ["get", "list"] + } + ], + "freshness": "Current" + } + ] + }, + { + "metadata": { "name": "playlist.grafana.app", "creationTimestamp": null }, + "versions": [ + { + "version": "v0alpha1", + "resources": [ + { + "resource": "playlists", + "responseKind": { "group": "", "version": "", "kind": "Playlist" }, + "scope": "Namespaced", + "singularResource": "playlist", + "verbs": ["create", "delete", "deletecollection", "get", "list", "patch", "update"] + } + ], + "freshness": "Current" + } + ] + } + ] +} diff --git a/public/app/features/apiserver/types.ts b/public/app/features/apiserver/types.ts index 06199f6d677..631f56f505c 100644 --- a/public/app/features/apiserver/types.ts +++ b/public/app/features/apiserver/types.ts @@ -111,6 +111,18 @@ type GrafanaLabels = { [DeprecatedInternalId]?: string; }; +export interface GroupVersionResource { + group: string; + version: string; + resource: string; +} + +export interface GroupVersionKind { + group: string; + version: string; + kind: string; +} + export interface Resource extends TypeMeta { metadata: ObjectMeta; spec: T; diff --git a/public/app/plugins/panel/live/LiveChannelEditor.tsx b/public/app/plugins/panel/live/LiveChannelEditor.tsx index d5ba8f45fe1..3276992eb37 100644 --- a/public/app/plugins/panel/live/LiveChannelEditor.tsx +++ b/public/app/plugins/panel/live/LiveChannelEditor.tsx @@ -9,8 +9,9 @@ import { GrafanaTheme2, parseLiveChannelAddress, } from '@grafana/data'; -import { Select, Alert, Label, stylesFactory } from '@grafana/ui'; +import { Select, Alert, Label, stylesFactory, Combobox } from '@grafana/ui'; import { config } from 'app/core/config'; +import { discoveryResources, getAPIGroupDiscoveryList, GroupDiscoveryResource } from 'app/features/apiserver/discovery'; import { getManagedChannelInfo } from 'app/features/live/info'; import { LivePanelOptions } from './types'; @@ -22,6 +23,7 @@ const scopes: Array> = [ { label: 'Data Sources', value: LiveChannelScope.DataSource, description: 'Data sources with live support' }, { label: 'Plugins', value: LiveChannelScope.Plugin, description: 'Plugins with live support' }, { label: 'Stream', value: LiveChannelScope.Stream, description: 'data streams (eg, influx style)' }, + { label: 'Watch', value: LiveChannelScope.Watch, description: 'Watch k8s style resources' }, ]; export function LiveChannelEditor(props: Props) { @@ -93,6 +95,16 @@ export function LiveChannelEditor(props: Props) { }); }; + const getWatchableResources = async (v: string) => { + const apis = await getAPIGroupDiscoveryList(); + return discoveryResources(apis) + .filter((v) => v.verbs.includes('watch')) + .map((r) => ({ + value: `${r.responseKind.group}/${r.responseKind.version}/${r.resource}`, // must be string | number + resource: r, + })); + }; + const { scope, namespace, path } = props.value; const style = getStyles(config.theme2); @@ -109,6 +121,25 @@ export function LiveChannelEditor(props: Props) {