[8.x] [streams] lifecycle - ingestion and total docs metadata (#210301)

x-pack/platform/plugins/shared/dataset_quality/common/api_types.ts

@@ -32,6 +32,7 @@ export const dataStreamStatRt = rt.intersection([
     lastActivity: rt.number,
     integration: rt.string,
     totalDocs: rt.number,
+    creationDate: rt.number,
   }),
 ]);
 

x-pack/platform/plugins/shared/dataset_quality/public/index.ts

@@ -9,8 +9,12 @@ import type { PluginInitializerContext } from '@kbn/core/public';
 import { DatasetQualityConfig } from '../common/plugin_config';
 import { DatasetQualityPlugin } from './plugin';
 
+export type { DataStreamStatServiceResponse } from '../common/data_streams_stats';
 export type { DatasetQualityPluginSetup, DatasetQualityPluginStart } from './types';
 
+export { DataStreamsStatsService } from './services/data_streams_stats/data_streams_stats_service';
+export type { IDataStreamsStatsClient } from './services/data_streams_stats/types';
+
 export function plugin(context: PluginInitializerContext<DatasetQualityConfig>) {
   return new DatasetQualityPlugin(context);
 }

x-pack/platform/plugins/shared/dataset_quality/public/services/data_streams_stats/data_streams_stats_client.ts

@@ -41,12 +41,15 @@ export class DataStreamsStatsClient implements IDataStreamsStatsClient {
   public async getDataStreamsStats(
     params: GetDataStreamsStatsQuery
   ): Promise<DataStreamStatServiceResponse> {
-    const types = params.types.length === 0 ? KNOWN_TYPES : params.types;
+    const types =
+      'types' in params
+        ? rison.encodeArray(params.types.length === 0 ? KNOWN_TYPES : params.types)
+        : undefined;
     const response = await this.http
       .get<GetDataStreamsStatsResponse>('/internal/dataset_quality/data_streams/stats', {
         query: {
           ...params,
-          types: rison.encodeArray(types),
+          types,
         },
       })
       .catch((error) => {

x-pack/platform/plugins/shared/dataset_quality/server/routes/data_streams/get_data_streams_creation_date.ts

@@ -0,0 +1,45 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { ElasticsearchClient } from '@kbn/core/server';
+import { dataStreamService } from '../../services';
+
+export async function getDataStreamsCreationDate({
+  esClient,
+  dataStreams,
+}: {
+  esClient: ElasticsearchClient;
+  dataStreams: string[];
+}) {
+  const matchingStreams = await dataStreamService.getMatchingDataStreams(esClient, dataStreams);
+  const streamByIndex = matchingStreams.reduce((acc, { name, indices }) => {
+    if (indices[0]) acc[indices[0].index_name] = name;
+    return acc;
+  }, {} as Record<string, string>);
+
+  const indices = Object.keys(streamByIndex);
+  if (indices.length === 0) {
+    return {};
+  }
+  // While _cat api is not recommended for application use this is the only way
+  // to retrieve the creation date in serverless for now. We should change this
+  // once a proper approach exists (see elastic/elasticsearch-serverless#3010)
+  const catIndices = await esClient.cat.indices({
+    index: indices,
+    h: ['creation.date', 'index'],
+    format: 'json',
+  });
+
+  return catIndices.reduce((acc, index) => {
+    const creationDate = index['creation.date'];
+    const indexName = index.index!;
+    const stream = streamByIndex[indexName];
+
+    acc[stream] = creationDate ? Number(creationDate) : undefined;
+    return acc;
+  }, {} as Record<string, number | undefined>);
+}

x-pack/platform/plugins/shared/dataset_quality/server/routes/data_streams/routes.ts

@@ -6,6 +6,7 @@
  */
 
 import * as t from 'io-ts';
+import { toBooleanRt } from '@kbn/io-ts-utils';
 import {
   CheckAndLoadIntegrationResponse,
   DataStreamDetails,
@@ -38,15 +39,14 @@ import { getDegradedFieldValues } from './get_degraded_field_values';
 import { getDegradedFields } from './get_degraded_fields';
 import { getNonAggregatableDataStreams } from './get_non_aggregatable_data_streams';
 import { updateFieldLimit } from './update_field_limit';
+import { getDataStreamsCreationDate } from './get_data_streams_creation_date';
 
 const statsRoute = createDatasetQualityServerRoute({
   endpoint: 'GET /internal/dataset_quality/data_streams/stats',
   params: t.type({
     query: t.intersection([
-      t.type({ types: typesRt }),
-      t.partial({
-        datasetQuery: t.string,
-      }),
+      t.union([t.type({ types: typesRt }), t.type({ datasetQuery: t.string })]),
+      t.partial({ includeCreationDate: toBooleanRt }),
     ]),
   }),
   options: {
@@ -81,22 +81,33 @@ const statsRoute = createDatasetQualityServerRoute({
       return dataStream.userPrivileges.canMonitor;
     });
 
-    const dataStreamsStats = isServerless
-      ? await getDataStreamsMeteringStats({
-          esClient: esClientAsSecondaryAuthUser,
-          dataStreams: privilegedDataStreams.map((stream) => stream.name),
-        })
-      : await getDataStreamsStats({
-          esClient,
-          dataStreams: privilegedDataStreams.map((stream) => stream.name),
-        });
+    const dataStreamsNames = privilegedDataStreams.map((stream) => stream.name);
+    const [dataStreamsStats, dataStreamsCreationDate] = await Promise.all([
+      isServerless
+        ? getDataStreamsMeteringStats({
+            esClient: esClientAsSecondaryAuthUser,
+            dataStreams: dataStreamsNames,
+          })
+        : getDataStreamsStats({
+            esClient,
+            dataStreams: dataStreamsNames,
+          }),
+
+      params.query.includeCreationDate
+        ? getDataStreamsCreationDate({
+            esClient: esClientAsSecondaryAuthUser,
+            dataStreams: dataStreamsNames,
+          })
+        : ({} as Record<string, number | undefined>),
+    ]);
 
     return {
       datasetUserPrivileges,
       dataStreamsStats: dataStreams.map((dataStream: DataStreamStat) => {
         dataStream.size = dataStreamsStats[dataStream.name]?.size;
         dataStream.sizeBytes = dataStreamsStats[dataStream.name]?.sizeBytes;
         dataStream.totalDocs = dataStreamsStats[dataStream.name]?.totalDocs;
+        dataStream.creationDate = dataStreamsCreationDate[dataStream.name];
 
         return dataStream;
       }),

x-pack/platform/plugins/shared/dataset_quality/server/services/data_stream.ts

@@ -15,7 +15,7 @@ import { reduceAsyncChunks } from '../utils/reduce_async_chunks';
 class DataStreamService {
   public async getMatchingDataStreams(
     esClient: ElasticsearchClient,
-    datasetName: string
+    datasetName: string | string[]
   ): Promise<IndicesDataStream[]> {
     try {
       const { data_streams: dataStreamsInfo } = await esClient.indices.getDataStream({

x-pack/solutions/observability/plugins/streams_app/.storybook/get_mock_streams_app_context.tsx

@@ -15,6 +15,7 @@ import type { SharePublicStart } from '@kbn/share-plugin/public/plugin';
 import { NavigationPublicStart } from '@kbn/navigation-plugin/public/types';
 import type { SavedObjectTaggingPluginStart } from '@kbn/saved-objects-tagging-plugin/public';
 import { fieldsMetadataPluginPublicMock } from '@kbn/fields-metadata-plugin/public/mocks';
+import { DataStreamsStatsClient } from '@kbn/dataset-quality-plugin/public/services/data_streams_stats/data_streams_stats_client';
 import type { StreamsAppKibanaContext } from '../public/hooks/use_kibana';
 
 export function getMockStreamsAppContext(): StreamsAppKibanaContext {
@@ -38,7 +39,7 @@ export function getMockStreamsAppContext(): StreamsAppKibanaContext {
       },
     },
     services: {
-      query: jest.fn(),
+      dataStreamsClient: Promise.resolve({} as unknown as DataStreamsStatsClient),
     },
     isServerless: false,
   };

x-pack/solutions/observability/plugins/streams_app/kibana.jsonc

@@ -19,6 +19,7 @@
       "savedObjectsTagging",
       "navigation",
       "fieldsMetadata",
+      "datasetQuality"
     ],
     "requiredBundles": [
       "kibanaReact"

x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_lifecycle/helpers/format_bytes.ts

@@ -0,0 +1,10 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { formatNumber } from '@elastic/eui';
+
+export const formatBytes = (value: number) => formatNumber(value, '0.0 b');

x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_lifecycle/helpers/ingestion_rate_query.ts

@@ -0,0 +1,55 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import datemath from '@kbn/datemath';
+
+export const ingestionRateQuery = ({
+  index,
+  start,
+  end,
+  timestampField = '@timestamp',
+  bucketCount = 10,
+}: {
+  index: string;
+  start: string;
+  end: string;
+  timestampField?: string;
+  bucketCount?: number;
+}) => {
+  const startDate = datemath.parse(start);
+  const endDate = datemath.parse(end);
+  if (!startDate || !endDate) {
+    throw new Error(`Expected a valid start and end date but got [start: ${start} | end: ${end}]`);
+  }
+
+  const intervalInSeconds = Math.max(
+    Math.round(endDate.diff(startDate, 'seconds') / bucketCount),
+    1
+  );
+
+  return {
+    index,
+    track_total_hits: false,
+    body: {
+      size: 0,
+      query: {
+        bool: {
+          filter: [{ range: { [timestampField]: { gte: start, lte: end } } }],
+        },
+      },
+      aggs: {
+        docs_count: {
+          date_histogram: {
+            field: timestampField,
+            fixed_interval: `${intervalInSeconds}s`,
+            min_doc_count: 0,
+          },
+        },
+      },
+    },
+  };
+};

x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_lifecycle/hooks/use_data_stream_stats.tsx

@@ -0,0 +1,58 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import moment from 'moment';
+import { IngestStreamGetResponse } from '@kbn/streams-schema';
+import { DataStreamStatServiceResponse } from '@kbn/dataset-quality-plugin/public';
+import { useKibana } from '../../../hooks/use_kibana';
+import { useStreamsAppFetch } from '../../../hooks/use_streams_app_fetch';
+
+export type DataStreamStats = DataStreamStatServiceResponse['dataStreamsStats'][number] & {
+  bytesPerDoc: number;
+  bytesPerDay: number;
+};
+
+export const useDataStreamStats = ({ definition }: { definition?: IngestStreamGetResponse }) => {
+  const {
+    services: { dataStreamsClient },
+  } = useKibana();
+
+  const statsFetch = useStreamsAppFetch(async () => {
+    if (!definition) {
+      return;
+    }
+
+    const client = await dataStreamsClient;
+    const {
+      dataStreamsStats: [dsStats],
+    } = await client.getDataStreamsStats({
+      datasetQuery: definition.stream.name,
+      includeCreationDate: true,
+    });
+
+    if (!dsStats || !dsStats.creationDate || !dsStats.sizeBytes) {
+      return undefined;
+    }
+    const daysSinceCreation = Math.max(
+      1,
+      Math.round(moment().diff(moment(dsStats.creationDate), 'days'))
+    );
+
+    return {
+      ...dsStats,
+      bytesPerDay: dsStats.sizeBytes / daysSinceCreation,
+      bytesPerDoc: dsStats.totalDocs ? dsStats.sizeBytes / dsStats.totalDocs : 0,
+    };
+  }, [dataStreamsClient, definition]);
+
+  return {
+    stats: statsFetch.value,
+    isLoading: statsFetch.loading,
+    refresh: statsFetch.refresh,
+    error: statsFetch.error,
+  };
+};