Handle ES client errors in defaultIngestHandler

convert only EPM hanlder that wasn't using defaultIngestHandler

From discussions in elastic#75862 (comment)

```
> curl --user elastic:changeme -X POST http://localhost:5601/api/ingest_manager/epm/packages/aws-0.2.7 -H 'kbn-xsrf: xyz' | jq
{
  "statusCode": 400,
  "error": "Bad Request",
  "message": "HTTP 400 response from ES at /_ingest/pipeline/logs-aws.cloudtrail-0.2.7: [parse_exception] Failed to parse content to map"
}
```

and logged this in Kibana

```
server    log   [17:15:44.961] [error][ingestManager][plugins] { Error: [parse_exception] Failed to parse content to map
    at respond (/Users/jfsiii/work/kibana/node_modules/elasticsearch/src/lib/transport.js:349:15)
    at checkRespForFailure (/Users/jfsiii/work/kibana/node_modules/elasticsearch/src/lib/transport.js:306:7)
    at HttpConnector.<anonymous> (/Users/jfsiii/work/kibana/node_modules/elasticsearch/src/lib/connectors/http.js:173:7)
    at IncomingMessage.wrapper (/Users/jfsiii/work/kibana/node_modules/lodash/lodash.js:4949:19)
    at IncomingMessage.emit (events.js:203:15)
    at endReadableNT (_stream_readable.js:1145:12)
    at process._tickCallback (internal/process/next_tick.js:63:19)
  status: 400,
  displayName: 'BadRequest',
  message: '[parse_exception] Failed to parse content to map',
  path: '/_ingest/pipeline/logs-aws.cloudtrail-0.2.7',
  query: undefined,
  body:
   { error:
      { root_cause: [Array],
        type: 'parse_exception',
        reason: 'Failed to parse content to map',
        caused_by: [Object] },
     status: 400 },
  statusCode: 400,
  response:
   '{"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: (org.elasticsearch.common.bytes.AbstractBytesReference$MarkSupportingStreamInputWrapper); line: 489, column: 26]"}},"status":400}',
  toString: [Function],
  toJSON: [Function] }
```

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

@@ -14,14 +14,34 @@ import {
 } from 'src/core/server';
 import { appContextService } from './services';
 
+interface ESClientError {
+  statusCode: number;
+  message: string;
+  body: string;
+  path: string;
+  query: string | undefined;
+  response: string;
+}
+
+const isESClientError = (error: any): error is ESClientError => {
+  return (
+    'statusCode' in error &&
+    'message' in error &&
+    'body' in error &&
+    'path' in error &&
+    'query' in error &&
+    'response' in error
+  );
+};
+
 export class IngestManagerError extends Error {
   constructor(message?: string) {
     super(message);
     this.name = this.constructor.name; // for stack traces
   }
 }
 
-export const getHTTPResponseCode = (error: IngestManagerError): number => {
+export const getHTTPResponseCode = (error: any): number => {
   switch (true) {
     case error instanceof RegistryConnectionError:
       return 502; // Bad Gateway
@@ -51,6 +71,15 @@ export interface IngestErrorHandlerParams {
 
 export const defaultIngestErrorHandler: IngestErrorHandler = async ({ error, response }) => {
   const logger = appContextService.getLogger();
+  if (isESClientError(error)) {
+    logger.error(error);
+    const message = `HTTP ${error.statusCode} response from ES at ${error.path}: ${error.message}`;
+    return response.customError({
+      statusCode: error.statusCode,
+      body: { message },
+    });
+  }
+
   if (error instanceof IngestManagerError) {
     logger.error(error.message);
     return response.customError({
@@ -65,6 +94,7 @@ export const defaultIngestErrorHandler: IngestErrorHandler = async ({ error, res
       body: { message: error.output.payload.message },
     });
   }
+
   logger.error(error);
   return response.customError({
     statusCode: 500,

x-pack/plugins/ingest_manager/server/routes/epm/handlers.ts

@@ -32,7 +32,7 @@ import {
   getLimitedPackages,
   getInstallationObject,
 } from '../../services/epm/packages';
-import { IngestManagerError, getHTTPResponseCode, defaultIngestErrorHandler } from '../../errors';
+import { IngestManagerError, defaultIngestErrorHandler } from '../../errors';
 import { splitPkgKey } from '../../services/epm/registry';
 
 export const getCategoriesHandler: RequestHandler<
@@ -155,14 +155,10 @@ export const installPackageHandler: RequestHandler<
     };
     return response.ok({ body });
   } catch (e) {
+    const defaultResult = await defaultIngestErrorHandler({ error: e, response });
     if (e instanceof IngestManagerError) {
-      logger.error(e);
-      return response.customError({
-        statusCode: getHTTPResponseCode(e),
-        body: { message: e.message },
-      });
+      return defaultResult;
     }
-
     // if there is an unknown server error, uninstall any package assets
     try {
       const installedPkg = await getInstallationObject({ savedObjectsClient, pkgName });
@@ -173,11 +169,7 @@ export const installPackageHandler: RequestHandler<
     } catch (error) {
       logger.error(`could not remove failed installation ${error}`);
     }
-    logger.error(e);
-    return response.customError({
-      statusCode: 500,
-      body: { message: e.message },
-    });
+    return defaultResult;
   }
 };
 

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

@@ -156,7 +156,10 @@ async function installPipeline({
     body: pipeline.contentForInstallation,
   };
   if (pipeline.extension === 'yml') {
-    callClusterParams.headers = { ['Content-Type']: 'application/yaml' };
+    callClusterParams.headers = {
+      'Content-Type': 'application/yaml',
+      Accept: 'application/json',
+    };
   }
 
   // This uses the catch-all endpoint 'transport.request' because we have to explicitly

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

@@ -146,11 +146,12 @@ export async function installPackage({
   }
   const installIndexPatternPromise = installIndexPatterns(savedObjectsClient, pkgName, pkgVersion);
   const kibanaAssets = await getKibanaAssets(paths);
-  if (installedPkg)
+  if (installedPkg) {
     await deleteKibanaSavedObjectsAssets(
       savedObjectsClient,
       installedPkg.attributes.installed_kibana
     );
+  }
   // save new kibana refs before installing the assets
   const installedKibanaAssetsRefs = await saveKibanaAssetsRefs(
     savedObjectsClient,

x-pack/plugins/ingest_manager/server/services/epm/registry/requests.ts

@@ -53,8 +53,12 @@ export async function getResponse(url: string): Promise<Response> {
     });
     return response;
   } catch (error) {
-    // throw same error type whether isSystemError (after max retry attempts) or something else
-    throw new RegistryConnectionError(error);
+    // isSystemError here means we didn't succeed after max retries
+    if (isSystemError(error)) {
+      throw new RegistryConnectionError(error);
+    } else {
+      throw error;
+    }
   }
 }