🌊 Streams: Handle missing permissions for fetching failure store information on listing page (elastic#235918)

This PR makes sure that the failure store is only queried if the logged
in user has permissions to query it. It extends `internal/streams` to
also return failure store read permissions - if available, it queries
it, otherwise it shows a warning icon next to the affected columns:

Old without permissions:
<img width="1253" height="225" alt="Screenshot 2025-09-22 at 09 40 11"
src="https://github.com/user-attachments/assets/0148d567-1316-4c65-b401-c590435d461d"
/>

New without permissions:
<img width="735" height="492" alt="Screenshot 2025-09-22 at 09 36 16"
src="https://github.com/user-attachments/assets/d8c0627e-8ffa-4026-b4d5-0d5b2940eac9"
/>

New with permissions (same as before)
<img width="1020" height="434" alt="Screenshot 2025-09-22 at 09 37 01"
src="https://github.com/user-attachments/assets/f62be846-f075-4e75-ac1e-f5bdb8aef0c4"
/>

Author: flash1293

x-pack/platform/packages/shared/kbn-streams-schema/src/models/ingest/base.ts

@@ -28,6 +28,8 @@ interface IngestStreamPrivileges {
   simulate: boolean;
   // User can get data information using the text structure API (e.g. to detect the structure of a message)
   text_structure: boolean;
+  // User can read from the failure store
+  read_failure_store: boolean;
 }
 
 const ingestStreamPrivilegesSchema: z.Schema<IngestStreamPrivileges> = z.object({
@@ -36,6 +38,7 @@ const ingestStreamPrivilegesSchema: z.Schema<IngestStreamPrivileges> = z.object(
   lifecycle: z.boolean(),
   simulate: z.boolean(),
   text_structure: z.boolean(),
+  read_failure_store: z.boolean(),
 });
 
 export interface IngestBase {

x-pack/platform/packages/shared/kbn-streams-schema/src/models/ingest/classic.test.ts

@@ -99,6 +99,7 @@ describe('ClassicStream', () => {
           monitor: true,
           simulate: true,
           text_structure: true,
+          read_failure_store: true,
         },
         data_stream_exists: true,
         ...emptyAssets,

x-pack/platform/packages/shared/kbn-streams-schema/src/models/ingest/wired.test.ts

@@ -95,6 +95,7 @@ describe('WiredStream', () => {
           monitor: true,
           simulate: true,
           text_structure: true,
+          read_failure_store: true,
         },
         effective_lifecycle: {
           dsl: {},

x-pack/platform/plugins/shared/streams/server/lib/streams/client.ts

@@ -492,11 +492,16 @@ export class StreamsClient {
   }
 
   /**
-   * Checks whether the user has the required privileges to manage the stream.
+   * Checks whether the user has the required privileges to manage the stream (or streams).
    * Managing a stream means updating the stream properties. It does not
    * include the dashboard links.
+   *
+   * In case multiple streams are provided, it checks whether the user has
+   * the required privileges on all streams, and returns the least-privileged
+   * result.
    */
-  async getPrivileges(name: string) {
+  async getPrivileges(nameOrNames: string | string[]) {
+    const names = Array.isArray(nameOrNames) ? nameOrNames : [nameOrNames];
     const isServerless = this.dependencies.isServerless;
     const REQUIRED_MANAGE_PRIVILEGES = [
       'manage_index_templates',
@@ -516,6 +521,7 @@ export class StreamsClient {
       'manage',
       'monitor',
       'manage_data_stream_lifecycle',
+      'read_failure_store',
     ];
     if (!isServerless) {
       REQUIRED_INDEX_PRIVILEGES.push('manage_ilm');
@@ -526,7 +532,7 @@ export class StreamsClient {
         cluster: REQUIRED_MANAGE_PRIVILEGES,
         index: [
           {
-            names: [name],
+            names,
             privileges: REQUIRED_INDEX_PRIVILEGES,
           },
         ],
@@ -535,15 +541,23 @@ export class StreamsClient {
     return {
       manage:
         REQUIRED_MANAGE_PRIVILEGES.every((privilege) => privileges.cluster[privilege] === true) &&
-        Object.values(privileges.index[name]).every((privilege) => privilege === true),
-      monitor: privileges.index[name].monitor,
+        names.every((name) =>
+          Object.values(privileges.index[name]).every((privilege) => privilege === true)
+        ),
+      monitor: names.every((name) => privileges.index[name].monitor),
       // on serverless, there is no ILM, so we map lifecycle to true if the user has manage_data_stream_lifecycle
       lifecycle: isServerless
-        ? privileges.index[name].manage_data_stream_lifecycle
-        : privileges.index[name].manage_data_stream_lifecycle && privileges.index[name].manage_ilm,
-      simulate: privileges.cluster.read_pipeline && privileges.index[name].create,
+        ? names.every((name) => privileges.index[name].manage_data_stream_lifecycle)
+        : names.every(
+            (name) =>
+              privileges.index[name].manage_data_stream_lifecycle &&
+              privileges.index[name].manage_ilm
+          ),
+      simulate:
+        privileges.cluster.read_pipeline && names.every((name) => privileges.index[name].create),
       // text structure is always available for the internal user, but not for the current user
       text_structure: isServerless ? true : privileges.cluster.monitor_text_structure,
+      read_failure_store: names.every((name) => privileges.index[name].read_failure_store),
     };
   }
 

x-pack/platform/plugins/shared/streams/server/routes/internal/streams/crud/route.ts

@@ -32,15 +32,29 @@ export const listStreamsRoute = createServerRoute({
       requiredPrivileges: [STREAMS_API_PRIVILEGES.read],
     },
   },
-  handler: async ({ request, getScopedClients }): Promise<{ streams: ListStreamDetail[] }> => {
+  handler: async ({
+    request,
+    getScopedClients,
+  }): Promise<{ streams: ListStreamDetail[]; canReadFailureStore: boolean }> => {
     const { streamsClient, scopedClusterClient } = await getScopedClients({ request });
     const streams = await streamsClient.listStreamsWithDataStreamExistence();
 
     const streamNames = streams.filter(({ exists }) => exists).map(({ stream }) => stream.name);
 
-    const dataStreams = await processAsyncInChunks(streamNames, (streamNamesChunk) =>
-      scopedClusterClient.asCurrentUser.indices.getDataStream({ name: streamNamesChunk })
-    );
+    let canReadFailureStore = true;
+
+    const dataStreams = await processAsyncInChunks(streamNames, async (streamNamesChunk) => {
+      const [{ read_failure_store: readFailureStore }, dataStreamsChunk] = await Promise.all([
+        streamsClient.getPrivileges(streamNamesChunk),
+        scopedClusterClient.asCurrentUser.indices.getDataStream({ name: streamNamesChunk }),
+      ]);
+
+      if (!readFailureStore) {
+        canReadFailureStore = false;
+      }
+
+      return dataStreamsChunk;
+    });
 
     const enrichedStreams = streams.reduce<ListStreamDetail[]>((acc, { stream }) => {
       if (Streams.GroupStream.Definition.is(stream)) {
@@ -57,7 +71,7 @@ export const listStreamsRoute = createServerRoute({
       return acc;
     }, []);
 
-    return { streams: enrichedStreams };
+    return { streams: enrichedStreams, canReadFailureStore };
   },
 });
 

x-pack/platform/plugins/shared/streams_app/public/components/stream_list_view/data_quality_column.tsx

@@ -10,7 +10,7 @@ import { mapPercentageToQuality } from '@kbn/dataset-quality-plugin/common';
 import { DatasetQualityIndicator, calculatePercentage } from '@kbn/dataset-quality-plugin/public';
 import useAsync from 'react-use/lib/useAsync';
 import { esqlResultToTimeseries } from '../../util/esql_result_to_timeseries';
-import type { StreamHistogramFetch } from '../../hooks/use_streams_histogram_fetch';
+import type { StreamHistogramFetch } from '../../hooks/use_doc_count_fetch';
 
 export function DataQualityColumn({
   histogramQueryFetch,

x-pack/platform/plugins/shared/streams_app/public/components/stream_list_view/documents_column.tsx

@@ -31,7 +31,7 @@ import { esqlResultToTimeseries } from '../../util/esql_result_to_timeseries';
 import type { useTimefilter } from '../../hooks/use_timefilter';
 import { TooltipOrPopoverIcon } from '../tooltip_popover_icon/tooltip_popover_icon';
 import { getFormattedError } from '../../util/errors';
-import type { StreamHistogramFetch } from '../../hooks/use_streams_histogram_fetch';
+import type { StreamHistogramFetch } from '../../hooks/use_doc_count_fetch';
 
 export function DocumentsColumn({
   indexPattern,

x-pack/platform/plugins/shared/streams_app/public/components/stream_list_view/index.tsx

@@ -61,12 +61,10 @@ export function StreamListView() {
 
   const { timeState } = useTimefilter();
   const streamsListFetch = useStreamsAppFetch(
-    async ({ signal }) => {
-      const { streams } = await streamsRepositoryClient.fetch('GET /internal/streams', {
+    async ({ signal }) =>
+      streamsRepositoryClient.fetch('GET /internal/streams', {
         signal,
-      });
-      return streams;
-    },
+      }),
     // time state change is used to trigger a refresh of the listed
     // streams metadata but we operate on stale data if we don't
     // also refresh the streams
@@ -91,7 +89,7 @@ export function StreamListView() {
         <StreamsAppContextProvider context={context}>
           <GroupStreamModificationFlyout
             client={streamsRepositoryClient}
-            streamsList={streamsListFetch.value}
+            streamsList={streamsListFetch.value?.streams}
             refresh={() => {
               streamsListFetch.refresh();
               overlayRef.current?.close();
@@ -185,11 +183,15 @@ export function StreamListView() {
           <StreamsListEmptyPrompt onAddData={handleAddData} />
         ) : (
           <>
-            <StreamsTreeTable loading={streamsListFetch.loading} streams={streamsListFetch.value} />
+            <StreamsTreeTable
+              loading={streamsListFetch.loading}
+              streams={streamsListFetch.value?.streams}
+              canReadFailureStore={streamsListFetch.value?.canReadFailureStore}
+            />
             {groupStreams?.enabled && (
               <>
                 <EuiSpacer size="l" />
-                <GroupStreamsCards streams={streamsListFetch.value} />
+                <GroupStreamsCards streams={streamsListFetch.value?.streams} />
               </>
             )}
           </>

x-pack/platform/plugins/shared/streams_app/public/components/stream_list_view/translations.ts

@@ -11,6 +11,14 @@ export const NAME_COLUMN_HEADER = i18n.translate('xpack.streams.streamsTreeTable
   defaultMessage: 'Name',
 });
 
+export const FAILURE_STORE_PERMISSIONS_ERROR = i18n.translate(
+  'xpack.streams.streamsTreeTable.failureStorePermissionsError',
+  {
+    defaultMessage:
+      'Does not include failed documents - user does not have access to failure store',
+  }
+);
+
 export const DOCUMENTS_COLUMN_HEADER = i18n.translate(
   'xpack.streams.streamsTreeTable.documentsColumnName',
   { defaultMessage: 'Documents' }

x-pack/platform/plugins/shared/streams_app/public/components/stream_list_view/tree_table.tsx

@@ -15,6 +15,7 @@ import {
   EuiInMemoryTable,
   useEuiTheme,
   EuiHighlight,
+  EuiIconTip,
   EuiButtonIcon,
 } from '@elastic/eui';
 import { css } from '@emotion/css';
@@ -33,7 +34,7 @@ import { StreamsAppSearchBar } from '../streams_app_search_bar';
 import { DocumentsColumn } from './documents_column';
 import { DataQualityColumn } from './data_quality_column';
 import { useStreamsAppRouter } from '../../hooks/use_streams_app_router';
-import { useStreamHistogramFetch } from '../../hooks/use_streams_histogram_fetch';
+import { useDocCountFetch } from '../../hooks/use_doc_count_fetch';
 import { useTimefilter } from '../../hooks/use_timefilter';
 import { RetentionColumn } from './retention_column';
 import {
@@ -45,14 +46,17 @@ import {
   NO_STREAMS_MESSAGE,
   DATA_QUALITY_COLUMN_HEADER,
   DOCUMENTS_COLUMN_HEADER,
+  FAILURE_STORE_PERMISSIONS_ERROR,
 } from './translations';
 import { DiscoverBadgeButton } from '../stream_badges';
 
 export function StreamsTreeTable({
   loading,
   streams = [],
+  canReadFailureStore = false,
 }: {
   streams?: ListStreamDetail[];
+  canReadFailureStore?: boolean;
   loading?: boolean;
 }) {
   const router = useStreamsAppRouter();
@@ -161,7 +165,7 @@ export function StreamsTreeTable({
 
   const numDataPoints = 25;
 
-  const { getStreamDocCounts } = useStreamHistogramFetch(numDataPoints);
+  const { getStreamDocCounts } = useDocCountFetch({ numDataPoints, canReadFailureStore });
 
   const sorting = {
     sort: {
@@ -283,7 +287,18 @@ export function StreamsTreeTable({
         },
         {
           field: 'documentsCount',
-          name: DOCUMENTS_COLUMN_HEADER,
+          name: (
+            <EuiFlexGroup alignItems="center" gutterSize="s">
+              {!canReadFailureStore && (
+                <EuiIconTip
+                  content={FAILURE_STORE_PERMISSIONS_ERROR}
+                  type="warning"
+                  color="warning"
+                />
+              )}
+              {DOCUMENTS_COLUMN_HEADER}
+            </EuiFlexGroup>
+          ),
           width: '180px',
           sortable: false,
           align: 'right',
@@ -300,7 +315,18 @@ export function StreamsTreeTable({
         },
         {
           field: 'dataQuality',
-          name: DATA_QUALITY_COLUMN_HEADER,
+          name: (
+            <EuiFlexGroup alignItems="center" gutterSize="s">
+              {!canReadFailureStore && (
+                <EuiIconTip
+                  content={FAILURE_STORE_PERMISSIONS_ERROR}
+                  type="warning"
+                  color="warning"
+                />
+              )}
+              {DATA_QUALITY_COLUMN_HEADER}
+            </EuiFlexGroup>
+          ),
           width: '150px',
           sortable: false,
           dataType: 'number',