[Ingest Manager] Handle Legacy ES client errors (#76865)

## Summary

closes #75862

 1. Use an HTTP status code if ES client error provides one. extended #75862 (comment) to all 4xx-5xx errors
 1. Format Error message as described in #75862 (comment) & agreed to #75862 (comment)

### example Request/Response:
```
> curl --user elastic:changeme -X POST http://localhost:5601/api/ingest_manager/epm/packages/aws-0.2.7 -H 'kbn-xsrf: xyz'
{
  "statusCode":400,
  "error":"Bad Request",
  "message":"[parse_exception] Failed to parse content to map from ES at /_ingest/pipeline/logs-aws.cloudtrail-0.2.7: {\"error\":{\"root_cause\":[{\"type\":\"parse_exception\",\"reason\":\"Failed to parse content to map\"}],\"type\":\"parse_exception\",\"reason\":\"Failed to parse content to map\",\"caused_by\":{\"type\":\"json_parse_exception\",\"reason\":\"Duplicate field 'ListGroupsForUser'\\n at [Source: (byte[])\\\"---\\ndescription: Pipeline for AWS CloudTrail Logs\\nprocessors:\\n  - set:\\n      field: event.ingested\\n      value: '{{_ingest.timestamp}}'\\n  - rename:\\n      field: \\\"message\\\"\\n      target_field: \\\"event.original\\\"\\n  - json:\\n      field: \\\"event.original\\\"\\n      target_field: \\\"json\\\"\\n  - date:\\n      field: \\\"json.eventTime\\\"\\n      target_field: \\\"@timestamp\\\"\\n      ignore_failure: true\\n      formats:\\n        - ISO8601\\n  - rename:\\n      field: \\\"json.eventVersion\\\"\\n      target_field: \\\"aws.cloudtrail.event_versi\\\"[truncated 16425 bytes]; line: 489, column: 26]\"}},\"status\":400}"
}
```

### example Kibana Logs:
`[parse_exception] Failed to parse content to map`

<details><summary>Used `test.each` to generate tests for each 4xx - 5xx error. Call each error 3 different ways.</summary>

```
  defaultIngestErrorHandler
    use the HTTP error status code provided by LegacyESErrors
      ✓ 400 - with path & response (12ms)
      ✓ 401 - with path & response (5ms)
      ✓ 402 - with path & response (5ms)
      ✓ 403 - with path & response (6ms)
      ✓ 404 - with path & response (5ms)
      ✓ 405 - with path & response (17ms)
      ✓ 406 - with path & response (2ms)
      ✓ 407 - with path & response (3ms)
      ✓ 408 - with path & response (6ms)
      ✓ 409 - with path & response (5ms)
      ✓ 410 - with path & response (1ms)
      ✓ 411 - with path & response (1ms)
      ✓ 412 - with path & response (1ms)
      ✓ 413 - with path & response (1ms)
      ✓ 414 - with path & response (1ms)
      ✓ 415 - with path & response (1ms)
      ✓ 416 - with path & response (1ms)
      ✓ 417 - with path & response (9ms)
      ✓ 418 - with path & response (1ms)
      ✓ 421 - with path & response (1ms)
      ✓ 426 - with path & response (1ms)
      ✓ 429 - with path & response (1ms)
      ✓ 450 - with path & response (1ms)
      ✓ 494 - with path & response (1ms)
      ✓ 497 - with path & response (1ms)
      ✓ 499 - with path & response (3ms)
      ✓ 500 - with path & response (2ms)
      ✓ 501 - with path & response (1ms)
      ✓ 502 - with path & response (2ms)
      ✓ 503 - with path & response (1ms)
      ✓ 504 - with path & response (1ms)
      ✓ 505 - with path & response (8ms)
      ✓ 506 - with path & response (2ms)
      ✓ 510 - with path & response (1ms)
      ✓ 400 - with other metadata (1ms)
      ✓ 401 - with other metadata (1ms)
      ✓ 402 - with other metadata (1ms)
      ✓ 403 - with other metadata (1ms)
      ✓ 404 - with other metadata (2ms)
      ✓ 405 - with other metadata (1ms)
      ✓ 406 - with other metadata (2ms)
      ✓ 407 - with other metadata (1ms)
      ✓ 408 - with other metadata (1ms)
      ✓ 409 - with other metadata (1ms)
      ✓ 410 - with other metadata (10ms)
      ✓ 411 - with other metadata (2ms)
      ✓ 412 - with other metadata (1ms)
      ✓ 413 - with other metadata (1ms)
      ✓ 414 - with other metadata (1ms)
      ✓ 415 - with other metadata (1ms)
      ✓ 416 - with other metadata (7ms)
      ✓ 417 - with other metadata (2ms)
      ✓ 418 - with other metadata (1ms)
      ✓ 421 - with other metadata (1ms)
      ✓ 426 - with other metadata (1ms)
      ✓ 429 - with other metadata (1ms)
      ✓ 450 - with other metadata (1ms)
      ✓ 494 - with other metadata (11ms)
      ✓ 497 - with other metadata (1ms)
      ✓ 499 - with other metadata (1ms)
      ✓ 500 - with other metadata (1ms)
      ✓ 501 - with other metadata (1ms)
      ✓ 502 - with other metadata (1ms)
      ✓ 503 - with other metadata (1ms)
      ✓ 504 - with other metadata (2ms)
      ✓ 505 - with other metadata (2ms)
      ✓ 506 - with other metadata (1ms)
      ✓ 510 - with other metadata (1ms)
      ✓ 400 - without metadata (1ms)
      ✓ 401 - without metadata (1ms)
      ✓ 402 - without metadata (10ms)
      ✓ 403 - without metadata (1ms)
      ✓ 404 - without metadata (1ms)
      ✓ 405 - without metadata (1ms)
      ✓ 406 - without metadata (1ms)
      ✓ 407 - without metadata (1ms)
      ✓ 408 - without metadata (1ms)
      ✓ 409 - without metadata (1ms)
      ✓ 410 - without metadata (1ms)
      ✓ 411 - without metadata (1ms)
      ✓ 412 - without metadata (1ms)
      ✓ 413 - without metadata (1ms)
      ✓ 414 - without metadata (1ms)
      ✓ 415 - without metadata (12ms)
      ✓ 416 - without metadata (1ms)
      ✓ 417 - without metadata (2ms)
      ✓ 418 - without metadata (1ms)
      ✓ 421 - without metadata (1ms)
      ✓ 426 - without metadata (2ms)
      ✓ 429 - without metadata (2ms)
      ✓ 450 - without metadata (3ms)
      ✓ 494 - without metadata (2ms)
      ✓ 497 - without metadata (2ms)
      ✓ 499 - without metadata (1ms)
      ✓ 500 - without metadata (1ms)
      ✓ 501 - without metadata (2ms)
      ✓ 502 - without metadata (1ms)
      ✓ 503 - without metadata (10ms)
      ✓ 504 - without metadata (2ms)
      ✓ 505 - without metadata (1ms)
      ✓ 506 - without metadata (2ms)
      ✓ 510 - without metadata (1ms)
```
</details>


### Checklist

- [x] [Unit or functional tests](https://www.elastic.co/guide/en/kibana/master/development-tests.html) were updated or added to match the most common scenarios


### Manual testing
<details><summary><strong>Manual testing</strong></summary>

#### Checkout the branch from elastic/package-storage#370

```
git clone https://github.com/elastic/package-storage.git
cd package-storage/
git switch -C update-aws-0.2.7-1598281986 origin/update-aws-0.2.7-1598281986
```

#### start the stack using the registry from elastic/package-storage#370
```
cd testing/environments/
docker-compose -f snapshot.yml pull
docker-compose -f snapshot.yml -f local.yml up --force-recreate
```

#### Try to install the broken package
```
> curl --user elastic:changeme -X POST http://localhost:5601/api/ingest_manager/epm/packages/aws-0.2.7 -H 'kbn-xsrf: xyz'
{"statusCode":500,"error":"Internal Server Error","message":"Bad Request"}
```
_observe the same error as #75862_

#### start only local registry with the broken package
```
# CTRL-C the stack (shell where you ran `docker-compose`)
cd ../.. # back to package-storage root
docker build .
docker run -p 8080:8080 id_from_prior_step
```

#### start the stack from this PR, pointing at the local registry from prior step
```
yarn start --no-base-path --xpack.ingestManager.registryUrl=http://localhost:8080
yarn es snapshot --license=trial  -E xpack.security.authc.api_key.enabled=true
```

#### Try to install the broken package
```
curl --user elastic:changeme -X POST http://localhost:5601/api/ingest_manager/epm/packages/aws-0.2.7 -H 'kbn-xsrf: xyz'
{
  "statusCode": 400,
  "error": "Bad Request",
  "message": "[parse_exception] Failed to parse content to map response from /_ingest/pipeline/logs-aws.cloudtrail-0.2.7: {\"error\":{\"root_cause\":[{\"type\":\"parse_exception\",\"reason\":\"Failed to parse content to map\"}],\"type\":\"parse_exception\",\"reason\":\"Failed to parse content to map\",\"caused_by\":{\"type\":\"json_parse_exception\",\"reason\":\"Duplicate field 'ListGroupsForUser'\\n at [Source: (byte[])\\\"---\\ndescription: Pipeline for AWS CloudTrail Logs\\nprocessors:\\n  - set:\\n      field: event.ingested\\n      value: '{{_ingest.timestamp}}'\\n  - rename:\\n      field: \\\"message\\\"\\n      target_field: \\\"event.original\\\"\\n  - json:\\n      field: \\\"event.original\\\"\\n      target_field: \\\"json\\\"\\n  - date:\\n      field: \\\"json.eventTime\\\"\\n      target_field: \\\"@timestamp\\\"\\n      ignore_failure: true\\n      formats:\\n        - ISO8601\\n  - rename:\\n      field: \\\"json.eventVersion\\\"\\n      target_field: \\\"aws.cloudtrail.event_versi\\\"[truncated 16425 bytes]; line: 489, column: 26]\"}},\"status\":400}"
}
```
_observe new error format_
</details>

x-pack/plugins/ingest_manager/server/errors.test.ts → x-pack/plugins/ingest_manager/server/errors/handlers.test.ts

@@ -5,16 +5,19 @@
  */
 
 import Boom from 'boom';
+import { errors } from 'elasticsearch';
 import { httpServerMock } from 'src/core/server/mocks';
-import { createAppContextStartContractMock } from './mocks';
-
+import { createAppContextStartContractMock } from '../mocks';
+import { appContextService } from '../services';
 import {
   IngestManagerError,
   RegistryError,
   PackageNotFoundError,
   defaultIngestErrorHandler,
-} from './errors';
-import { appContextService } from './services';
+} from './index';
+
+const LegacyESErrors = errors as Record<string, any>;
+type ITestEsErrorsFnParams = [errorCode: string, error: any, expectedMessage: string];
 
 describe('defaultIngestErrorHandler', () => {
   let mockContract: ReturnType<typeof createAppContextStartContractMock>;
@@ -29,6 +32,55 @@ describe('defaultIngestErrorHandler', () => {
     appContextService.stop();
   });
 
+  async function testEsErrorsFn(...args: ITestEsErrorsFnParams) {
+    const [, error, expectedMessage] = args;
+    jest.clearAllMocks();
+    const response = httpServerMock.createResponseFactory();
+    await defaultIngestErrorHandler({ error, response });
+
+    // response
+    expect(response.ok).toHaveBeenCalledTimes(0);
+    expect(response.customError).toHaveBeenCalledTimes(1);
+    expect(response.customError).toHaveBeenCalledWith({
+      statusCode: error.status,
+      body: { message: expectedMessage },
+    });
+
+    // logging
+    expect(mockContract.logger?.error).toHaveBeenCalledTimes(1);
+    expect(mockContract.logger?.error).toHaveBeenCalledWith(expectedMessage);
+  }
+
+  describe('use the HTTP error status code provided by LegacyESErrors', () => {
+    const statusCodes = Object.keys(LegacyESErrors).filter((key) => /^\d+$/.test(key));
+    const errorCodes = statusCodes.filter((key) => parseInt(key, 10) >= 400);
+    const casesWithPathResponse: ITestEsErrorsFnParams[] = errorCodes.map((errorCode) => [
+      errorCode,
+      new LegacyESErrors[errorCode]('the root message', {
+        path: '/path/to/call',
+        response: 'response is here',
+      }),
+      'the root message response from /path/to/call: response is here',
+    ]);
+    const casesWithOtherMeta: ITestEsErrorsFnParams[] = errorCodes.map((errorCode) => [
+      errorCode,
+      new LegacyESErrors[errorCode]('the root message', {
+        other: '/path/to/call',
+        props: 'response is here',
+      }),
+      'the root message',
+    ]);
+    const casesWithoutMeta: ITestEsErrorsFnParams[] = errorCodes.map((errorCode) => [
+      errorCode,
+      new LegacyESErrors[errorCode]('some message'),
+      'some message',
+    ]);
+
+    test.each(casesWithPathResponse)('%d - with path & response', testEsErrorsFn);
+    test.each(casesWithOtherMeta)('%d - with other metadata', testEsErrorsFn);
+    test.each(casesWithoutMeta)('%d - without metadata', testEsErrorsFn);
+  });
+
   describe('IngestManagerError', () => {
     it('502: RegistryError', async () => {
       const error = new RegistryError('xyz');

x-pack/plugins/ingest_manager/server/errors.ts → x-pack/plugins/ingest_manager/server/errors/handlers.ts

@@ -4,33 +4,46 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-/* eslint-disable max-classes-per-file */
 import Boom, { isBoom } from 'boom';
 import {
   RequestHandlerContext,
   KibanaRequest,
   IKibanaResponse,
   KibanaResponseFactory,
 } from 'src/core/server';
-import { appContextService } from './services';
+import { errors as LegacyESErrors } from 'elasticsearch';
+import { appContextService } from '../services';
+import { IngestManagerError, RegistryError, PackageNotFoundError } from './index';
 
 type IngestErrorHandler = (
   params: IngestErrorHandlerParams
 ) => IKibanaResponse | Promise<IKibanaResponse>;
-
 interface IngestErrorHandlerParams {
   error: IngestManagerError | Boom | Error;
   response: KibanaResponseFactory;
   request?: KibanaRequest;
   context?: RequestHandlerContext;
 }
+// unsure if this is correct. would prefer to use something "official"
+// this type is based on BadRequest values observed while debugging https://github.com/elastic/kibana/issues/75862
 
-export class IngestManagerError extends Error {
-  constructor(message?: string) {
-    super(message);
-    this.name = this.constructor.name; // for stack traces
-  }
+interface LegacyESClientError {
+  message: string;
+  stack: string;
+  status: number;
+  displayName: string;
+  path?: string;
+  query?: string | undefined;
+  body?: {
+    error: object;
+    status: number;
+  };
+  statusCode?: number;
+  response?: string;
 }
+export const isLegacyESClientError = (error: any): error is LegacyESClientError => {
+  return error instanceof LegacyESErrors._Abstract;
+};
 
 const getHTTPResponseCode = (error: IngestManagerError): number => {
   if (error instanceof RegistryError) {
@@ -48,6 +61,22 @@ export const defaultIngestErrorHandler: IngestErrorHandler = async ({
   response,
 }: IngestErrorHandlerParams): Promise<IKibanaResponse> => {
   const logger = appContextService.getLogger();
+  if (isLegacyESClientError(error)) {
+    // there was a problem communicating with ES (e.g. via `callCluster`)
+    // only log the message
+    const message =
+      error?.path && error?.response
+        ? // if possible, return the failing endpoint and its response
+          `${error.message} response from ${error.path}: ${error.response}`
+        : error.message;
+
+    logger.error(message);
+
+    return response.customError({
+      statusCode: error?.statusCode || error.status,
+      body: { message },
+    });
+  }
 
   // our "expected" errors
   if (error instanceof IngestManagerError) {
@@ -76,9 +105,3 @@ export const defaultIngestErrorHandler: IngestErrorHandler = async ({
     body: { message: error.message },
   });
 };
-
-export class RegistryError extends IngestManagerError {}
-export class RegistryConnectionError extends RegistryError {}
-export class RegistryResponseError extends RegistryError {}
-export class PackageNotFoundError extends IngestManagerError {}
-export class PackageOutdatedError extends IngestManagerError {}

x-pack/plugins/ingest_manager/server/errors/index.ts

@@ -0,0 +1,20 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+/* eslint-disable max-classes-per-file */
+export { defaultIngestErrorHandler } from './handlers';
+
+export class IngestManagerError extends Error {
+  constructor(message?: string) {
+    super(message);
+    this.name = this.constructor.name; // for stack traces
+  }
+}
+export class RegistryError extends IngestManagerError {}
+export class RegistryConnectionError extends RegistryError {}
+export class RegistryResponseError extends RegistryError {}
+export class PackageNotFoundError extends IngestManagerError {}
+export class PackageOutdatedError extends IngestManagerError {}

x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/ingest_pipeline/install.ts

@@ -156,7 +156,12 @@ async function installPipeline({
     body: pipeline.contentForInstallation,
   };
   if (pipeline.extension === 'yml') {
-    callClusterParams.headers = { ['Content-Type']: 'application/yaml' };
+    callClusterParams.headers = {
+      // pipeline is YAML
+      'Content-Type': 'application/yaml',
+      // but we want JSON responses (to extract error messages, status code, or other metadata)
+      Accept: 'application/json',
+    };
   }
 
   // This uses the catch-all endpoint 'transport.request' because we have to explicitly