From afee775b7a5d91ee856c153ee2ffc82040e5de5b Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Thu, 16 Jan 2025 10:54:16 +0300 Subject: [PATCH] POC/Provisioning: Update client streaming to use chunked support from backend_srv (#99032) --- .betterer.results | 6 +- .../src/services/backendSrv.ts | 2 +- public/app/core/services/backend_srv.ts | 9 +- public/app/core/utils/fetch.ts | 5 +- public/app/features/apiserver/client.ts | 175 ++++-------------- 5 files changed, 41 insertions(+), 156 deletions(-) diff --git a/.betterer.results b/.betterer.results index 039459ba420..b6152fde655 100644 --- a/.betterer.results +++ b/.betterer.results @@ -1328,8 +1328,7 @@ exports[`better eslint`] = { [0, 0, 0, "Do not use any type assertions.", "2"], [0, 0, 0, "Do not use any type assertions.", "3"], [0, 0, 0, "Do not use any type assertions.", "4"], - [0, 0, 0, "Do not use any type assertions.", "5"], - [0, 0, 0, "Unexpected any. Specify a different type.", "6"] + [0, 0, 0, "Unexpected any. Specify a different type.", "5"] ], "public/app/core/utils/object.ts:5381": [ [0, 0, 0, "Do not use any type assertions.", "0"], @@ -3070,9 +3069,6 @@ exports[`better eslint`] = { [0, 0, 0, "No untranslated strings. Wrap text with ", "6"], [0, 0, 0, "No untranslated strings. Wrap text with ", "7"] ], - "public/app/features/apiserver/client.ts:5381": [ - [0, 0, 0, "Do not use any type assertions.", "0"] - ], "public/app/features/auth-config/AuthDrawer.tsx:5381": [ [0, 0, 0, "No untranslated strings in text props. Wrap text with or use t()", "0"], [0, 0, 0, "No untranslated strings in text props. Wrap text with or use t()", "1"], diff --git a/packages/grafana-runtime/src/services/backendSrv.ts b/packages/grafana-runtime/src/services/backendSrv.ts index 6d636c8a60b..14352c6aabb 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' | 'stream'; + responseType?: 'json' | 'text' | 'arraybuffer' | 'blob'; /** * Used to cancel an open connection diff --git a/public/app/core/services/backend_srv.ts b/public/app/core/services/backend_srv.ts index ce7dab8c838..cc667ac6c56 100644 --- a/public/app/core/services/backend_srv.ts +++ b/public/app/core/services/backend_srv.ts @@ -198,13 +198,14 @@ export class BackendSrv implements BackendService { } } process() - .catch((e) => { - console.log(requestId, 'catch', e); - }) // from abort .then(() => { console.log(requestId, 'complete'); observer.complete(); - }); // runs in background + }) // runs in background + .catch((e) => { + console.log(requestId, 'catch', e); + observer.error(e); + }); // from abort }, error: (e) => { observer.error(e); diff --git a/public/app/core/utils/fetch.ts b/public/app/core/utils/fetch.ts index fb2564146d8..237c5d0eb13 100644 --- a/public/app/core/utils/fetch.ts +++ b/public/app/core/utils/fetch.ts @@ -115,7 +115,7 @@ export const parseBody = (options: BackendSrvRequest, isAppJson: boolean) => { export async function parseResponseBody( response: Response, - responseType?: 'json' | 'text' | 'arraybuffer' | 'blob' | 'stream' + responseType?: 'json' | 'text' | 'arraybuffer' | 'blob' ): Promise { if (responseType) { switch (responseType) { @@ -138,9 +138,6 @@ export async function parseResponseBody( } 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 2a2edb65600..c38c3bf4476 100644 --- a/public/app/features/apiserver/client.ts +++ b/public/app/features/apiserver/client.ts @@ -1,4 +1,4 @@ -import { Observable, Subscriber, map, switchMap, from, filter, retry, catchError } from 'rxjs'; +import { Observable, switchMap, from, retry, catchError } from 'rxjs'; import { config, getBackendSrv } from '@grafana/runtime'; import { contextSrv } from 'app/core/core'; @@ -38,104 +38,40 @@ export class ScopedResourceClient implements } 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; + return getBackendSrv() + .chunked({ + url: name ? `${this.url}/${name}` : this.url, + params: { + watch: true, + resourceVersion, + }, }) - ); - } - - 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('{'); + .pipe( + switchMap((response) => { + const events: Array> = []; + if (response.ok && response.data) { + const decoder = new TextDecoder(); + decoder + .decode(response.data) + .split('\n') + .forEach((v) => { + if (v?.length) { + try { + events.push(JSON.parse(v)); + } catch (e) { + console.warn('invalid response', e); + } + } + }); } - } - - 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; + return from(events); + }), + retry({ count: 3, delay: 1000 }), + catchError((error) => { + console.error('Watch stream error:', error); + throw error; + }) + ); } public async subresource(name: string, path: string): Promise { @@ -242,48 +178,3 @@ 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(); - }); -}