[8.6] [Logs UI] Prevent summary request from piling up (#148670) (#14…

…8792)

# Backport

This will backport the following commits from `main` to `8.6`:
- [[Logs UI] Prevent summary request from piling up
(#148670)](#148670)

<!--- Backport version: 8.9.7 -->

### Questions ?
Please refer to the [Backport tool
documentation](https://github.com/sqren/backport)

<!--BACKPORT [{"author":{"name":"Felix
Stürmer","email":"[email protected]"},"sourceCommit":{"committedDate":"2023-01-12T11:04:05Z","message":"[Logs
UI] Prevent summary request from piling up (#148670)\n\nCo-authored-by:
kibanamachine
<[email protected]>","sha":"f01a61b81ab0a4756dfe40233aa61a196aa64dbe","branchLabelMapping":{"^v8.7.0$":"main","^v(\\d+).(\\d+).\\d+$":"$1.$2"}},"sourcePullRequest":{"labels":["bug","release_note:fix","Feature:Logs
UI","Team:Infra Monitoring
UI","backport:prev-minor","v8.7.0"],"number":148670,"url":"https://github.com/elastic/kibana/pull/148670","mergeCommit":{"message":"[Logs
UI] Prevent summary request from piling up (#148670)\n\nCo-authored-by:
kibanamachine
<[email protected]>","sha":"f01a61b81ab0a4756dfe40233aa61a196aa64dbe"}},"sourceBranch":"main","suggestedTargetBranches":[],"targetPullRequestStates":[{"branch":"main","label":"v8.7.0","labelRegex":"^v8.7.0$","isSourceBranch":true,"state":"MERGED","url":"https://github.com/elastic/kibana/pull/148670","number":148670,"mergeCommit":{"message":"[Logs
UI] Prevent summary request from piling up (#148670)\n\nCo-authored-by:
kibanamachine
<[email protected]>","sha":"f01a61b81ab0a4756dfe40233aa61a196aa64dbe"}}]}]
BACKPORT-->

Co-authored-by: Felix Stürmer <[email protected]>

x-pack/plugins/infra/public/containers/logs/log_summary/log_summary.test.tsx

@@ -19,9 +19,12 @@ import { datemathToEpochMillis } from '../../../utils/datemath';
 jest.mock('./api/fetch_log_summary', () => ({ fetchLogSummary: jest.fn() }));
 const fetchLogSummaryMock = fetchLogSummary as jest.MockedFunction<typeof fetchLogSummary>;
 
-jest.mock('../../../hooks/use_kibana', () => ({
-  useKibanaContextForPlugin: () => ({ services: mockCoreMock.createStart() }),
-}));
+jest.mock('../../../hooks/use_kibana', () => {
+  const services = mockCoreMock.createStart();
+  return {
+    useKibanaContextForPlugin: () => ({ services }),
+  };
+});
 
 describe('useLogSummary hook', () => {
   beforeEach(() => {
@@ -161,6 +164,34 @@ describe('useLogSummary hook', () => {
       expect.anything()
     );
   });
+
+  it("doesn't query for new summary buckets when the previous request is still in flight", async () => {
+    fetchLogSummaryMock.mockResolvedValueOnce(createMockResponse([]));
+
+    const firstRange = createMockDateRange();
+    const { waitForNextUpdate, rerender } = renderHook(
+      ({ startTimestamp, endTimestamp }) =>
+        useLogSummary('SOURCE_ID', startTimestamp, endTimestamp, null),
+      {
+        initialProps: firstRange,
+      }
+    );
+
+    const secondRange = createMockDateRange('now-20s', 'now');
+
+    // intentionally don't wait for an update to test the throttling
+    rerender(secondRange);
+    await waitForNextUpdate();
+
+    expect(fetchLogSummaryMock).toHaveBeenCalledTimes(1);
+    expect(fetchLogSummaryMock).toHaveBeenLastCalledWith(
+      expect.objectContaining({
+        startTimestamp: firstRange.startTimestamp,
+        endTimestamp: firstRange.endTimestamp,
+      }),
+      expect.anything()
+    );
+  });
 });
 
 const createMockResponse = (

x-pack/plugins/infra/public/containers/logs/log_summary/log_summary.tsx

@@ -5,11 +5,12 @@
  * 2.0.
  */
 
-import { useState } from 'react';
-
-import { useCancellableEffect } from '../../../utils/cancellable_effect';
+import { useEffect } from 'react';
+import { exhaustMap, map, Observable } from 'rxjs';
+import { HttpHandler } from '@kbn/core-http-browser';
+import { useObservableState, useReplaySubject } from '../../../utils/use_observable';
 import { fetchLogSummary } from './api/fetch_log_summary';
-import { LogEntriesSummaryResponse } from '../../../../common/http_api';
+import { LogEntriesSummaryRequest, LogEntriesSummaryResponse } from '../../../../common/http_api';
 import { useBucketSize } from './bucket_size';
 import { useKibanaContextForPlugin } from '../../../hooks/use_kibana';
 
@@ -22,36 +23,51 @@ export const useLogSummary = (
   filterQuery: string | null
 ) => {
   const { services } = useKibanaContextForPlugin();
-  const [logSummaryBuckets, setLogSummaryBuckets] = useState<LogSummaryBuckets>([]);
   const bucketSize = useBucketSize(startTimestamp, endTimestamp);
 
-  useCancellableEffect(
-    (getIsCancelled) => {
-      if (startTimestamp === null || endTimestamp === null || bucketSize === null) {
-        return;
-      }
-
-      fetchLogSummary(
-        {
-          sourceId,
-          startTimestamp,
-          endTimestamp,
-          bucketSize,
-          query: filterQuery,
-        },
-        services.http.fetch
-      ).then((response) => {
-        if (!getIsCancelled()) {
-          setLogSummaryBuckets(response.data.buckets);
-        }
-      });
-    },
-    [sourceId, filterQuery, startTimestamp, endTimestamp, bucketSize]
-  );
+  const [logSummaryBuckets$, pushLogSummaryBucketsArgs] = useReplaySubject(fetchLogSummary$);
+  const { latestValue: logSummaryBuckets } = useObservableState(logSummaryBuckets$, NO_BUCKETS);
+
+  useEffect(() => {
+    if (startTimestamp === null || endTimestamp === null || bucketSize === null) {
+      return;
+    }
+
+    pushLogSummaryBucketsArgs([
+      {
+        sourceId,
+        startTimestamp,
+        endTimestamp,
+        bucketSize,
+        query: filterQuery,
+      },
+      services.http.fetch,
+    ]);
+  }, [
+    bucketSize,
+    endTimestamp,
+    filterQuery,
+    pushLogSummaryBucketsArgs,
+    services.http.fetch,
+    sourceId,
+    startTimestamp,
+  ]);
 
   return {
     buckets: logSummaryBuckets,
     start: startTimestamp,
     end: endTimestamp,
   };
 };
+
+const NO_BUCKETS: LogSummaryBuckets = [];
+
+type FetchLogSummaryArgs = [args: LogEntriesSummaryRequest, fetch: HttpHandler];
+
+const fetchLogSummary$ = (
+  fetchArguments$: Observable<FetchLogSummaryArgs>
+): Observable<LogSummaryBuckets> =>
+  fetchArguments$.pipe(
+    exhaustMap(([args, fetch]) => fetchLogSummary(args, fetch)),
+    map(({ data: { buckets } }) => buckets)
+  );

x-pack/plugins/infra/public/utils/use_observable.ts

@@ -6,7 +6,14 @@
  */
 
 import { useEffect, useRef, useState } from 'react';
-import { BehaviorSubject, Observable, OperatorFunction, PartialObserver, Subscription } from 'rxjs';
+import {
+  BehaviorSubject,
+  Observable,
+  OperatorFunction,
+  PartialObserver,
+  ReplaySubject,
+  Subscription,
+} from 'rxjs';
 import { switchMap } from 'rxjs/operators';
 
 export const useLatest = <Value>(value: Value) => {
@@ -42,9 +49,29 @@ export const useBehaviorSubject = <
   deriveObservableOnce: (input$: Observable<InputValue>) => OutputObservable,
   createInitialValue: () => InputValue
 ) => {
-  const [subject$] = useState(() => new BehaviorSubject<InputValue>(createInitialValue()));
+  const [[subject$, next], _] = useState(() => {
+    const newSubject$ = new BehaviorSubject<InputValue>(createInitialValue());
+    const newNext = newSubject$.next.bind(newSubject$);
+    return [newSubject$, newNext] as const;
+  });
+  const [output$] = useState(() => deriveObservableOnce(subject$));
+  return [output$, next] as const;
+};
+
+export const useReplaySubject = <
+  InputValue,
+  OutputValue,
+  OutputObservable extends Observable<OutputValue>
+>(
+  deriveObservableOnce: (input$: Observable<InputValue>) => OutputObservable
+) => {
+  const [[subject$, next], _] = useState(() => {
+    const newSubject$ = new ReplaySubject<InputValue>();
+    const newNext = newSubject$.next.bind(newSubject$);
+    return [newSubject$, newNext] as const;
+  });
   const [output$] = useState(() => deriveObservableOnce(subject$));
-  return [output$, subject$.next.bind(subject$)] as const;
+  return [output$, next] as const;
 };
 
 export const useObservableState = <State, InitialState>(