[8.6][ML Inference] New API to fetch ML inference errors (#142799)

* Add ML inference PL creation flow

* [CI] Auto-commit changed files from 'node scripts/eslint --no-cache --fix'

* Add exists check, clean up code a bit

* Fix dest name

* [CI] Auto-commit changed files from 'node scripts/eslint --no-cache --fix'

* Separate concerns

* [CI] Auto-commit changed files from 'node scripts/precommit_hook.js --ref HEAD~1..HEAD --fix'

* Remove i18n due to linter error, fix src field ref

* [CI] Auto-commit changed files from 'node scripts/precommit_hook.js --ref HEAD~1..HEAD --fix'

* Add/update unit tests

* [CI] Auto-commit changed files from 'node scripts/precommit_hook.js --ref HEAD~1..HEAD --fix'

* Refactor error handling

* [CI] Auto-commit changed files from 'node scripts/eslint --no-cache --fix'

* Add sub-pipeline to parent ML PL

* Add unit tests and docs

* [CI] Auto-commit changed files from 'node scripts/precommit_hook.js --ref HEAD~1..HEAD --fix'

* [CI] Auto-commit changed files from 'node scripts/eslint --no-cache --fix'

* Refactor error handling

* Wrap logic into higher level function

* Add route test

* [CI] Auto-commit changed files from 'node scripts/precommit_hook.js --ref HEAD~1..HEAD --fix'

* [CI] Auto-commit changed files from 'node scripts/eslint --no-cache --fix'

* API to fetch inference errors

* Minor style changes

* Add unit tests

* [CI] Auto-commit changed files from 'node scripts/precommit_hook.js --ref HEAD~1..HEAD --fix'

* [CI] Auto-commit changed files from 'node scripts/eslint --no-cache --fix'

Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>

x-pack/plugins/enterprise_search/server/lib/ml_inference_pipeline/get_inference_errors.test.ts

@@ -0,0 +1,87 @@
+/*
+ * 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 { getMlInferenceErrors } from './get_inference_errors';
+
+describe('getMlInferenceErrors', () => {
+  const indexName = 'my-index';
+
+  const mockClient = {
+    search: jest.fn(),
+  };
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('should fetch aggregations and transform them', async () => {
+    mockClient.search.mockImplementation(() =>
+      Promise.resolve({
+        aggregations: {
+          errors: {
+            buckets: [
+              {
+                key: 'Error message 1',
+                doc_count: 100,
+                max_error_timestamp: {
+                  value: 1664977836100,
+                  value_as_string: '2022-10-05T13:50:36.100Z',
+                },
+              },
+              {
+                key: 'Error message 2',
+                doc_count: 200,
+                max_error_timestamp: {
+                  value: 1664977836200,
+                  value_as_string: '2022-10-05T13:50:36.200Z',
+                },
+              },
+            ],
+          },
+        },
+      })
+    );
+
+    const actualResult = await getMlInferenceErrors(
+      indexName,
+      mockClient as unknown as ElasticsearchClient
+    );
+
+    expect(actualResult).toEqual([
+      {
+        message: 'Error message 1',
+        doc_count: 100,
+        timestamp: '2022-10-05T13:50:36.100Z',
+      },
+      {
+        message: 'Error message 2',
+        doc_count: 200,
+        timestamp: '2022-10-05T13:50:36.200Z',
+      },
+    ]);
+    expect(mockClient.search).toHaveBeenCalledTimes(1);
+  });
+
+  it('should return an empty array if there are no aggregates', async () => {
+    mockClient.search.mockImplementation(() =>
+      Promise.resolve({
+        aggregations: {
+          errors: [],
+        },
+      })
+    );
+
+    const actualResult = await getMlInferenceErrors(
+      indexName,
+      mockClient as unknown as ElasticsearchClient
+    );
+
+    expect(actualResult).toEqual([]);
+  });
+});

x-pack/plugins/enterprise_search/server/lib/ml_inference_pipeline/get_inference_errors.ts

@@ -0,0 +1,81 @@
+/*
+ * 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 {
+  AggregationsMultiBucketAggregateBase,
+  AggregationsStringRareTermsBucketKeys,
+} from '@elastic/elasticsearch/lib/api/types';
+
+import { ElasticsearchClient } from '@kbn/core/server';
+
+export interface MlInferenceError {
+  message: string;
+  doc_count: number;
+  timestamp: string | undefined; // Date string
+}
+
+export interface ErrorAggregationBucket extends AggregationsStringRareTermsBucketKeys {
+  max_error_timestamp: {
+    value: number | null;
+    value_as_string?: string;
+  };
+}
+
+/**
+ * Fetches an aggregate of distinct ML inference errors from the target index, along with the most
+ * recent error's timestamp and affected document count for each bucket.
+ * @param indexName the index to get the errors from.
+ * @param esClient the Elasticsearch Client to use to fetch the errors.
+ */
+export const getMlInferenceErrors = async (
+  indexName: string,
+  esClient: ElasticsearchClient
+): Promise<MlInferenceError[]> => {
+  const searchResult = await esClient.search<
+    unknown,
+    {
+      errors: AggregationsMultiBucketAggregateBase<ErrorAggregationBucket>;
+    }
+  >({
+    index: indexName,
+    body: {
+      aggs: {
+        errors: {
+          terms: {
+            field: '_ingest.inference_errors.message.enum',
+            order: {
+              max_error_timestamp: 'desc',
+            },
+            size: 20,
+          },
+          aggs: {
+            max_error_timestamp: {
+              max: {
+                field: '_ingest.inference_errors.timestamp',
+              },
+            },
+          },
+        },
+      },
+      size: 0,
+    },
+  });
+
+  const errorBuckets = searchResult.aggregations?.errors.buckets;
+  if (!errorBuckets) {
+    return [];
+  }
+
+  // Buckets are either in an array or in a Record, we transform them to an array
+  const buckets = Array.isArray(errorBuckets) ? errorBuckets : Object.values(errorBuckets);
+
+  return buckets.map((bucket) => ({
+    message: bucket.key,
+    doc_count: bucket.doc_count,
+    timestamp: bucket.max_error_timestamp?.value_as_string,
+  }));
+};

x-pack/plugins/enterprise_search/server/routes/enterprise_search/indices.test.ts

@@ -23,10 +23,14 @@ jest.mock('../../lib/indices/delete_ml_inference_pipeline', () => ({
 jest.mock('../../lib/indices/exists_index', () => ({
   indexOrAliasExists: jest.fn(),
 }));
+jest.mock('../../lib/ml_inference_pipeline/get_inference_errors', () => ({
+  getMlInferenceErrors: jest.fn(),
+}));
 
 import { deleteMlInferencePipeline } from '../../lib/indices/delete_ml_inference_pipeline';
 import { indexOrAliasExists } from '../../lib/indices/exists_index';
 import { fetchMlInferencePipelineProcessors } from '../../lib/indices/fetch_ml_inference_pipeline_processors';
+import { getMlInferenceErrors } from '../../lib/ml_inference_pipeline/get_inference_errors';
 import { createAndReferenceMlInferencePipeline } from '../../utils/create_ml_inference_pipeline';
 import { ElasticsearchResponseError } from '../../utils/identify_exceptions';
 
@@ -40,6 +44,7 @@ describe('Enterprise Search Managed Indices', () => {
         putPipeline: jest.fn(),
         simulate: jest.fn(),
       },
+      search: jest.fn(),
     },
   };
 
@@ -47,6 +52,64 @@ describe('Enterprise Search Managed Indices', () => {
     elasticsearch: { client: mockClient },
   };
 
+  describe('GET /internal/enterprise_search/indices/{indexName}/ml_inference/errors', () => {
+    beforeEach(() => {
+      const context = {
+        core: Promise.resolve(mockCore),
+      } as unknown as jest.Mocked<RequestHandlerContext>;
+
+      mockRouter = new MockRouter({
+        context,
+        method: 'get',
+        path: '/internal/enterprise_search/indices/{indexName}/ml_inference/errors',
+      });
+
+      registerIndexRoutes({
+        ...mockDependencies,
+        router: mockRouter.router,
+      });
+    });
+
+    it('fails validation without index_name', () => {
+      const request = {
+        params: {},
+      };
+      mockRouter.shouldThrow(request);
+    });
+
+    it('fetches ML inference errors', async () => {
+      const errorsResult = [
+        {
+          message: 'Error message 1',
+          doc_count: 100,
+          timestamp: '2022-10-05T13:50:36.100Z',
+        },
+        {
+          message: 'Error message 2',
+          doc_count: 200,
+          timestamp: '2022-10-05T13:50:36.200Z',
+        },
+      ];
+
+      (getMlInferenceErrors as jest.Mock).mockImplementationOnce(() => {
+        return Promise.resolve(errorsResult);
+      });
+
+      await mockRouter.callRoute({
+        params: { indexName: 'my-index-name' },
+      });
+
+      expect(getMlInferenceErrors).toHaveBeenCalledWith('my-index-name', mockClient.asCurrentUser);
+
+      expect(mockRouter.response.ok).toHaveBeenCalledWith({
+        body: {
+          errors: errorsResult,
+        },
+        headers: { 'content-type': 'application/json' },
+      });
+    });
+  });
+
   describe('GET /internal/enterprise_search/indices/{indexName}/ml_inference/pipeline_processors', () => {
     beforeEach(() => {
       const context = {

x-pack/plugins/enterprise_search/server/routes/enterprise_search/indices.ts

@@ -9,6 +9,7 @@ import {
   IngestPutPipelineRequest,
   IngestSimulateRequest,
 } from '@elastic/elasticsearch/lib/api/types';
+
 import { schema } from '@kbn/config-schema';
 
 import { i18n } from '@kbn/i18n';
@@ -27,6 +28,7 @@ import { fetchIndex } from '../../lib/indices/fetch_index';
 import { fetchIndices } from '../../lib/indices/fetch_indices';
 import { fetchMlInferencePipelineProcessors } from '../../lib/indices/fetch_ml_inference_pipeline_processors';
 import { generateApiKey } from '../../lib/indices/generate_api_key';
+import { getMlInferenceErrors } from '../../lib/ml_inference_pipeline/get_inference_errors';
 import { createIndexPipelineDefinitions } from '../../lib/pipelines/create_pipeline_definitions';
 import { getCustomPipelines } from '../../lib/pipelines/get_custom_pipelines';
 import { getPipeline } from '../../lib/pipelines/get_pipeline';
@@ -524,6 +526,30 @@ export function registerIndexRoutes({
     })
   );
 
+  router.get(
+    {
+      path: '/internal/enterprise_search/indices/{indexName}/ml_inference/errors',
+      validate: {
+        params: schema.object({
+          indexName: schema.string(),
+        }),
+      },
+    },
+    elasticsearchErrorHandler(log, async (context, request, response) => {
+      const indexName = decodeURIComponent(request.params.indexName);
+      const { client } = (await context.core).elasticsearch;
+
+      const errors = await getMlInferenceErrors(indexName, client.asCurrentUser);
+
+      return response.ok({
+        body: {
+          errors,
+        },
+        headers: { 'content-type': 'application/json' },
+      });
+    })
+  );
+
   router.put(
     {
       path: '/internal/enterprise_search/indices/{indexName}/ml_inference/pipeline_processors/{pipelineName}',