POC/Provisioning: Update client streaming to use chunked support from backend_srv (#99032)

.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 <Trans />", "6"],
       [0, 0, 0, "No untranslated strings. Wrap text with <Trans />", "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 <Trans /> or use t()", "0"],
       [0, 0, 0, "No untranslated strings in text props. Wrap text with <Trans /> or use t()", "1"],

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

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);

public/app/core/utils/fetch.ts

@@ -115,7 +115,7 @@ export const parseBody = (options: BackendSrvRequest, isAppJson: boolean) => {
 
 export async function parseResponseBody<T>(
   response: Response,
-  responseType?: 'json' | 'text' | 'arraybuffer' | 'blob' | 'stream'
+  responseType?: 'json' | 'text' | 'arraybuffer' | 'blob'
 ): Promise<T> {
   if (responseType) {
     switch (responseType) {
@@ -138,9 +138,6 @@ export async function parseResponseBody<T>(
         }
         return await response.json();
 
-      case 'stream':
-        return Promise.resolve(response.body) as Promise<T>;
-
       case 'text':
         // this specifically returns a Promise<string>
         // TODO refactor this function to remove the type assertions

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<T = object, S = object, K = string> implements
   }
 
   public watch(name?: string, resourceVersion?: string): Observable<ResourceEvent<T, S, K>> {
-    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<ReadableStream<Uint8Array>>({
-      url: name ? `${this.url}/${name}` : this.url,
-      params: {
-        watch: true,
-        resourceVersion,
-      },
-      responseType: 'stream',
-    });
-  }
-
-  private handleWatchStream(result: { data: ReadableStream<Uint8Array> }): Observable<ResourceEvent<T, S, K>> {
-    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<ResourceEvent<T, S, K>> = [];
-
-        // 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<T, S, K>;
-            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<ResourceEvent<T, S, K>> = [];
+          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<S>(name: string, path: string): Promise<S> {
@@ -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<T extends unknown>(
-  stream: ReadableStream<T>,
-  signal?: AbortSignal,
-  queueStrategy?: QueuingStrategy,
-  throwEndAsError = false
-): Observable<T> {
-  function createStream(subscriber: Subscriber<T>) {
-    return new WritableStream<T>(
-      {
-        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<T>((subscriber) => {
-    stream
-      .pipeTo(createStream(subscriber), { signal })
-      .then(() => {
-        return !subscriber.closed && subscriber.complete();
-      })
-      .catch((error) => subscriber.error(error));
-
-    return () => !stream.locked && stream.cancel();
-  });
-}