diff --git a/packages/grafana-runtime/src/services/backendSrv.ts b/packages/grafana-runtime/src/services/backendSrv.ts index 25512f3f76b..896e4cd3b63 100644 --- a/packages/grafana-runtime/src/services/backendSrv.ts +++ b/packages/grafana-runtime/src/services/backendSrv.ts @@ -68,7 +68,7 @@ export type BackendSrvRequest = { * * By default values are json parsed from text */ - responseType?: 'json' | 'text' | 'arraybuffer' | 'blob'; + responseType?: 'json' | 'text' | 'arraybuffer' | 'blob' | 'stream'; /** * The credentials read-only property of the Request interface indicates whether the user agent should send cookies from the other domain in the case of cross-origin requests. diff --git a/public/app/core/utils/fetch.ts b/public/app/core/utils/fetch.ts index 027ae66ca0e..b1aab3dd48d 100644 --- a/public/app/core/utils/fetch.ts +++ b/public/app/core/utils/fetch.ts @@ -114,7 +114,7 @@ export const parseBody = (options: BackendSrvRequest, isAppJson: boolean) => { export async function parseResponseBody( response: Response, - responseType?: 'json' | 'text' | 'arraybuffer' | 'blob' + responseType?: 'json' | 'text' | 'arraybuffer' | 'blob' | 'stream' ): Promise { if (responseType) { switch (responseType) { @@ -135,9 +135,11 @@ export async function parseResponseBody( console.warn(`${response.url} returned an invalid JSON`); return {} as T; } - return await response.json(); + case 'stream': + return Promise.resolve(response.body) as Promise + case 'text': // this specifically returns a Promise // TODO refactor this function to remove the type assertions diff --git a/public/app/features/apiserver/client.ts b/public/app/features/apiserver/client.ts index 8ac5b531095..2a2edb65600 100644 --- a/public/app/features/apiserver/client.ts +++ b/public/app/features/apiserver/client.ts @@ -1,3 +1,5 @@ +import { Observable, Subscriber, map, switchMap, from, filter, retry, catchError } from 'rxjs'; + import { config, getBackendSrv } from '@grafana/runtime'; import { contextSrv } from 'app/core/core'; @@ -13,6 +15,7 @@ import { ObjectMeta, K8sAPIGroupList, AnnoKeySavedFromUI, + ResourceEvent, } from './types'; export interface GroupVersionResource { @@ -34,6 +37,107 @@ export class ScopedResourceClient implements return getBackendSrv().get>(`${this.url}/${name}`); } + public watch(name?: string, resourceVersion?: string): Observable> { + return this.createWatchRequest(name, resourceVersion).pipe( + switchMap((response) => this.handleWatchStream(response)), + retry({ count: 3, delay: 1000 }), + catchError((error) => { + console.error('Watch stream error:', error); + throw error; + }) + ); + } + + private createWatchRequest(name?: string, resourceVersion?: string) { + return getBackendSrv().fetch>({ + url: name ? `${this.url}/${name}` : this.url, + params: { + watch: true, + resourceVersion, + }, + responseType: 'stream', + }); + } + + private handleWatchStream(result: { data: ReadableStream }): Observable> { + const decoder = new TextDecoder(); + let buffer = ''; + + return fromReadableStream(result.data).pipe( + map((chunk) => decoder.decode(chunk, { stream: true })), + map((text) => { + buffer += text; + const events: Array> = []; + + // Find complete JSON objects in the buffer + let startIndex = buffer.indexOf('{'); + while (startIndex !== -1) { + try { + const endIndex = this.findJsonEnd(buffer, startIndex); + if (endIndex === -1) { + break; + } + + const jsonStr = buffer.substring(startIndex, endIndex + 1); + const event = JSON.parse(jsonStr) as ResourceEvent; + events.push(event); + + // Remove processed data from buffer + buffer = buffer.substring(endIndex + 1); + startIndex = buffer.indexOf('{'); + } catch (e) { + // If JSON.parse fails, move to next potential object + buffer = buffer.substring(startIndex + 1); + startIndex = buffer.indexOf('{'); + } + } + + return events; + }), + filter((events) => events.length > 0), + switchMap((events) => from(events)) + ); + } + + private findJsonEnd(str: string, startIndex: number): number { + let brackets = 0; + let inString = false; + let escapeNext = false; + + for (let i = startIndex; i < str.length; i++) { + const char = str[i]; + + if (escapeNext) { + escapeNext = false; + continue; + } + + if (char === '\\' && inString) { + escapeNext = true; + continue; + } + + if (char === '"' && !escapeNext) { + inString = !inString; + continue; + } + + if (!inString) { + if (char === '{') { + brackets++; + } + if (char === '}') { + brackets--; + if (brackets === 0) { + return i; + } + } + } + } + + return -1; + } + public async subresource(name: string, path: string): Promise { return getBackendSrv().get(`${this.url}/${name}/${path}`); } @@ -138,3 +242,48 @@ export const parseListOptionsSelector = (selector: ListOptionsLabelSelector | Li .filter(Boolean) .join(','); }; + +/** + * Creates an Observable source from a ReadableStream source that will emit any + * values emitted by the stream. + * + * https://github.com/rxjs-ninja/rxjs-ninja/blob/main/libs/rxjs/utility/src/lib/from-readable-stream.ts + */ +function fromReadableStream( + stream: ReadableStream, + signal?: AbortSignal, + queueStrategy?: QueuingStrategy, + throwEndAsError = false +): Observable { + function createStream(subscriber: Subscriber) { + return new WritableStream( + { + write: (value) => subscriber.next(value), + abort: (error) => { + if (throwEndAsError) { + subscriber.error(error); + } else if (!subscriber.closed) { + subscriber.complete(); + } + }, + close: () => { + if (!subscriber.closed) { + subscriber.complete(); + } + }, + }, + queueStrategy + ); + } + + return new Observable((subscriber) => { + stream + .pipeTo(createStream(subscriber), { signal }) + .then(() => { + return !subscriber.closed && subscriber.complete(); + }) + .catch((error) => subscriber.error(error)); + + return () => !stream.locked && stream.cancel(); + }); +} diff --git a/public/app/features/apiserver/types.ts b/public/app/features/apiserver/types.ts index d90393f0bf1..c65a0f683fa 100644 --- a/public/app/features/apiserver/types.ts +++ b/public/app/features/apiserver/types.ts @@ -8,6 +8,8 @@ * */ +import { Observable } from 'rxjs'; + /** The object type and version */ export interface TypeMeta { apiVersion: string; @@ -169,9 +171,15 @@ export interface MetaStatus { details?: object; } +export interface ResourceEvent { + type: 'ADDED' | 'DELETED' | 'MODIFIED'; + object: Resource; +} + export interface ResourceClient { create(obj: ResourceForCreate): Promise>; get(name: string): Promise>; + watch(name?: string, resourceVersion?: string): Observable>; subresource(name: string, path: string): Promise; list(opts?: ListOptions): Promise>; update(obj: ResourceForCreate): Promise>; diff --git a/public/app/features/provisioning/RepositoryListPage.tsx b/public/app/features/provisioning/RepositoryListPage.tsx index f89eaabfffd..7191a72b4a7 100644 --- a/public/app/features/provisioning/RepositoryListPage.tsx +++ b/public/app/features/provisioning/RepositoryListPage.tsx @@ -18,12 +18,13 @@ import { Page } from 'app/core/components/Page/Page'; import { DeleteRepositoryButton } from './DeleteRepositoryButton'; import { SyncRepository } from './SyncRepository'; -import { Repository, useGetRepositoryStatusQuery } from './api'; +import { Repository } from './api'; import { NEW_URL, PROVISIONING_URL } from './constants'; import { useRepositoryList } from './hooks'; export default function RepositoryListPage() { const [items, isLoading] = useRepositoryList(); + return ( @@ -91,7 +92,6 @@ function RepositoryListPageContent({ items }: { items?: Repository[] }) { ); } - break; case 'local': @@ -106,7 +106,7 @@ function RepositoryListPageContent({ items }: { items?: Repository[] }) { - {item.spec?.title} {name && } + {item.spec?.title} {item.spec?.description} @@ -121,8 +121,6 @@ function RepositoryListPageContent({ items }: { items?: Repository[] }) { {healthy && } - {/* - */} @@ -136,10 +134,12 @@ function RepositoryListPageContent({ items }: { items?: Repository[] }) { ); } -function StatusBadge({ name }: { name: string }) { - const statusQuery = useGetRepositoryStatusQuery({ name }, { pollingInterval: 5000 }); - - const state = statusQuery.data?.status?.sync?.state; +interface StatusBadgeProps { + repo: Repository; + name: string; +} +function StatusBadge({ repo, name }: StatusBadgeProps) { + const state = repo.status?.sync?.state; if (!state) { return null; diff --git a/public/app/features/provisioning/api/endpoints.ts b/public/app/features/provisioning/api/endpoints.ts index a2c2eacb4bf..12e27d5323b 100644 --- a/public/app/features/provisioning/api/endpoints.ts +++ b/public/app/features/provisioning/api/endpoints.ts @@ -1,3 +1,7 @@ +import { Subscription } from 'rxjs'; + +import { ScopedResourceClient } from '../../apiserver/client'; + import { baseAPI as api } from './baseAPI'; export const addTagTypes = ['Job', 'Repository'] as const; const injectedRtkApi = api @@ -7,7 +11,7 @@ const injectedRtkApi = api .injectEndpoints({ endpoints: (build) => ({ listJob: build.query({ - query: (queryArg) => ({ + query: () => ({ url: `/jobs`, }), providesTags: ['Job'], @@ -22,9 +26,60 @@ const injectedRtkApi = api providesTags: ['Job'], }), listRepository: build.query({ - query: (queryArg) => ({ + query: () => ({ url: `/repositories`, }), + async onCacheEntryAdded(arg, { updateCachedData, cacheDataLoaded, cacheEntryRemoved }) { + const client = new ScopedResourceClient({ + group: 'provisioning.grafana.app', + version: 'v0alpha1', + resource: 'repositories', + }); + let subscription: Subscription | null = null; + try { + // wait for the initial query to resolve before proceeding + const response = await cacheDataLoaded; + const resourceVersion = response.data.metadata?.resourceVersion; + + subscription = client.watch(undefined, resourceVersion).subscribe((event) => { + updateCachedData((draft: RepositoryList) => { + if (!draft.items) { + draft.items = []; + } + + const existingIndex = draft.items.findIndex( + (item) => item.metadata?.name === event.object.metadata.name + ); + + if (event.type === 'ADDED') { + // Add the new item + //@ts-expect-error TODO Fix types + draft.items.push(event.object); + } else if (event.type === 'MODIFIED') { + // Update the existing item if it exists + if (existingIndex !== -1) { + //@ts-expect-error TODO Fix types + draft.items[existingIndex] = event.object; + } + } else if (event.type === 'DELETED') { + // Remove the item if it exists + if (existingIndex !== -1) { + draft.items.splice(existingIndex, 1); + } + } + }); + }); + } catch (error) { + // no-op in case `cacheEntryRemoved` resolves before `cacheDataLoaded`, + // in which case `cacheDataLoaded` will throw + console.error('Error in onCacheEntryAdded:', error); + } + // cacheEntryRemoved will resolve when the cache subscription is no longer active + await cacheEntryRemoved; + // perform cleanup steps once the `cacheEntryRemoved` promise resolves + + subscription?.unsubscribe(); + }, providesTags: ['Repository'], }), createRepository: build.mutation({