diff --git a/packages/kbn-dev-utils/src/tooling_log/__snapshots__/tooling_log_text_writer.test.ts.snap b/packages/kbn-dev-utils/src/tooling_log/__snapshots__/tooling_log_text_writer.test.ts.snap
index 7ff982acafbe..5fa074d4c773 100644
--- a/packages/kbn-dev-utils/src/tooling_log/__snapshots__/tooling_log_text_writer.test.ts.snap
+++ b/packages/kbn-dev-utils/src/tooling_log/__snapshots__/tooling_log_text_writer.test.ts.snap
@@ -170,6 +170,14 @@ exports[`level:warning/type:warning snapshots: output 1`] = `
 "
 `;
 
+exports[`never ignores write messages from the kibana elasticsearch.deprecation logger context 1`] = `
+"         │[elasticsearch.deprecation]
+         │{ foo: { bar: { '1': [Array] } }, bar: { bar: { '1': [Array] } } }
+         │
+         │Infinity
+"
+`;
+
 exports[`throws error if created with invalid level 1`] = `"Invalid log level \\"foo\\" (expected one of silent,error,warning,success,info,debug,verbose)"`;
 
 exports[`throws error if writeTo config is not defined or doesn't have a write method 1`] = `"ToolingLogTextWriter requires the \`writeTo\` option be set to a stream (like process.stdout)"`;
diff --git a/packages/kbn-dev-utils/src/tooling_log/tooling_log_text_writer.test.ts b/packages/kbn-dev-utils/src/tooling_log/tooling_log_text_writer.test.ts
index b4668f29b6e2..fbccfdcdf6ac 100644
--- a/packages/kbn-dev-utils/src/tooling_log/tooling_log_text_writer.test.ts
+++ b/packages/kbn-dev-utils/src/tooling_log/tooling_log_text_writer.test.ts
@@ -88,3 +88,55 @@ it('formats %s patterns and indents multi-line messages correctly', () => {
   const output = write.mock.calls.reduce((acc, chunk) => `${acc}${chunk}`, '');
   expect(output).toMatchSnapshot();
 });
+
+it('does not write messages from sources in ignoreSources', () => {
+  const write = jest.fn();
+  const writer = new ToolingLogTextWriter({
+    ignoreSources: ['myIgnoredSource'],
+    level: 'debug',
+    writeTo: {
+      write,
+    },
+  });
+
+  writer.write({
+    source: 'myIgnoredSource',
+    type: 'success',
+    indent: 10,
+    args: [
+      '%s\n%O\n\n%d',
+      'foo bar',
+      { foo: { bar: { 1: [1, 2, 3] } }, bar: { bar: { 1: [1, 2, 3] } } },
+      Infinity,
+    ],
+  });
+
+  const output = write.mock.calls.reduce((acc, chunk) => `${acc}${chunk}`, '');
+  expect(output).toEqual('');
+});
+
+it('never ignores write messages from the kibana elasticsearch.deprecation logger context', () => {
+  const write = jest.fn();
+  const writer = new ToolingLogTextWriter({
+    ignoreSources: ['myIgnoredSource'],
+    level: 'debug',
+    writeTo: {
+      write,
+    },
+  });
+
+  writer.write({
+    source: 'myIgnoredSource',
+    type: 'write',
+    indent: 10,
+    args: [
+      '%s\n%O\n\n%d',
+      '[elasticsearch.deprecation]',
+      { foo: { bar: { 1: [1, 2, 3] } }, bar: { bar: { 1: [1, 2, 3] } } },
+      Infinity,
+    ],
+  });
+
+  const output = write.mock.calls.reduce((acc, chunk) => `${acc}${chunk}`, '');
+  expect(output).toMatchSnapshot();
+});
diff --git a/packages/kbn-dev-utils/src/tooling_log/tooling_log_text_writer.ts b/packages/kbn-dev-utils/src/tooling_log/tooling_log_text_writer.ts
index 660dae3fa1f5..4fe33241cf77 100644
--- a/packages/kbn-dev-utils/src/tooling_log/tooling_log_text_writer.ts
+++ b/packages/kbn-dev-utils/src/tooling_log/tooling_log_text_writer.ts
@@ -92,7 +92,15 @@ export class ToolingLogTextWriter implements Writer {
     }
 
     if (this.ignoreSources && msg.source && this.ignoreSources.includes(msg.source)) {
-      return false;
+      if (msg.type === 'write') {
+        const txt = format(msg.args[0], ...msg.args.slice(1));
+        // Ensure that Elasticsearch deprecation log messages from Kibana aren't ignored
+        if (!/elasticsearch\.deprecation/.test(txt)) {
+          return false;
+        }
+      } else {
+        return false;
+      }
     }
 
     const prefix = has(MSG_PREFIXES, msg.type) ? MSG_PREFIXES[msg.type] : '';
diff --git a/packages/kbn-pm/dist/index.js b/packages/kbn-pm/dist/index.js
index 36b81ca60c09..d4f3dc726d8b 100644
--- a/packages/kbn-pm/dist/index.js
+++ b/packages/kbn-pm/dist/index.js
@@ -6639,7 +6639,15 @@ class ToolingLogTextWriter {
     }
 
     if (this.ignoreSources && msg.source && this.ignoreSources.includes(msg.source)) {
-      return false;
+      if (msg.type === 'write') {
+        const txt = (0, _util.format)(msg.args[0], ...msg.args.slice(1)); // Ensure that Elasticsearch deprecation log messages from Kibana aren't ignored
+
+        if (!/elasticsearch\.deprecation/.test(txt)) {
+          return false;
+        }
+      } else {
+        return false;
+      }
     }
 
     const prefix = has(MSG_PREFIXES, msg.type) ? MSG_PREFIXES[msg.type] : '';
diff --git a/src/core/server/elasticsearch/client/configure_client.test.ts b/src/core/server/elasticsearch/client/configure_client.test.ts
index 35fcb5819d01..e89317617892 100644
--- a/src/core/server/elasticsearch/client/configure_client.test.ts
+++ b/src/core/server/elasticsearch/client/configure_client.test.ts
@@ -6,21 +6,18 @@
  * Side Public License, v 1.
  */
 
-import { Buffer } from 'buffer';
-import { Readable } from 'stream';
-
-import { RequestEvent, errors } from '@elastic/elasticsearch';
 import type { Client } from '@elastic/elasticsearch';
-import type {
-  TransportRequestOptions,
-  TransportRequestParams,
-  RequestBody,
-} from '@elastic/elasticsearch/lib/Transport';
+
+jest.mock('./log_query_and_deprecation.ts', () => ({
+  __esModule: true,
+  instrumentEsQueryAndDeprecationLogger: jest.fn(),
+}));
 
 import { parseClientOptionsMock, ClientMock } from './configure_client.test.mocks';
 import { loggingSystemMock } from '../../logging/logging_system.mock';
 import type { ElasticsearchClientConfig } from './client_config';
 import { configureClient } from './configure_client';
+import { instrumentEsQueryAndDeprecationLogger } from './log_query_and_deprecation';
 
 const createFakeConfig = (
   parts: Partial<ElasticsearchClientConfig> = {}
@@ -36,40 +33,9 @@ const createFakeClient = () => {
   const client = new actualEs.Client({
     nodes: ['http://localhost'], // Enforcing `nodes` because it's mandatory
   });
-  jest.spyOn(client, 'on');
   return client;
 };
 
-const createApiResponse = <T>({
-  body,
-  statusCode = 200,
-  headers = {},
-  warnings = [],
-  params,
-  requestOptions = {},
-}: {
-  body: T;
-  statusCode?: number;
-  headers?: Record<string, string>;
-  warnings?: string[];
-  params?: TransportRequestParams;
-  requestOptions?: TransportRequestOptions;
-}): RequestEvent<T> => {
-  return {
-    body,
-    statusCode,
-    headers,
-    warnings,
-    meta: {
-      body,
-      request: {
-        params: params!,
-        options: requestOptions,
-      } as any,
-    } as any,
-  };
-};
-
 function getProductCheckValue(client: Client) {
   const tSymbol = Object.getOwnPropertySymbols(client.transport || client).filter(
     (symbol) => symbol.description === 'product check'
@@ -92,6 +58,7 @@ describe('configureClient', () => {
   afterEach(() => {
     parseClientOptionsMock.mockReset();
     ClientMock.mockReset();
+    jest.clearAllMocks();
   });
 
   it('calls `parseClientOptions` with the correct parameters', () => {
@@ -121,11 +88,14 @@ describe('configureClient', () => {
     expect(client).toBe(ClientMock.mock.results[0].value);
   });
 
-  it('listens to client on `response` events', () => {
+  it('calls instrumentEsQueryAndDeprecationLogger', () => {
     const client = configureClient(config, { logger, type: 'test', scoped: false });
-
-    expect(client.on).toHaveBeenCalledTimes(1);
-    expect(client.on).toHaveBeenCalledWith('response', expect.any(Function));
+    expect(instrumentEsQueryAndDeprecationLogger).toHaveBeenCalledTimes(1);
+    expect(instrumentEsQueryAndDeprecationLogger).toHaveBeenCalledWith({
+      logger,
+      client,
+      type: 'test',
+    });
   });
 
   describe('Product check', () => {
@@ -145,360 +115,4 @@ describe('configureClient', () => {
       expect(getProductCheckValue(asScoped)).toBe(2);
     });
   });
-
-  describe('Client logging', () => {
-    function createResponseWithBody(body?: RequestBody) {
-      return createApiResponse({
-        body: {},
-        statusCode: 200,
-        params: {
-          method: 'GET',
-          path: '/foo',
-          querystring: { hello: 'dolly' },
-          body,
-        },
-      });
-    }
-
-    describe('logs each query', () => {
-      it('creates a query logger context based on the `type` parameter', () => {
-        configureClient(createFakeConfig(), { logger, type: 'test123' });
-        expect(logger.get).toHaveBeenCalledWith('query', 'test123');
-      });
-
-      it('when request body is an object', () => {
-        const client = configureClient(createFakeConfig(), { logger, type: 'test', scoped: false });
-
-        const response = createResponseWithBody({
-          seq_no_primary_term: true,
-          query: {
-            term: { user: 'kimchy' },
-          },
-        });
-
-        client.emit('response', null, response);
-        expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
-                  Array [
-                    Array [
-                      "200
-                  GET /foo?hello=dolly
-                  {\\"seq_no_primary_term\\":true,\\"query\\":{\\"term\\":{\\"user\\":\\"kimchy\\"}}}",
-                      undefined,
-                    ],
-                  ]
-              `);
-      });
-
-      it('when request body is a string', () => {
-        const client = configureClient(createFakeConfig(), { logger, type: 'test', scoped: false });
-
-        const response = createResponseWithBody(
-          JSON.stringify({
-            seq_no_primary_term: true,
-            query: {
-              term: { user: 'kimchy' },
-            },
-          })
-        );
-
-        client.emit('response', null, response);
-        expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
-                  Array [
-                    Array [
-                      "200
-                  GET /foo?hello=dolly
-                  {\\"seq_no_primary_term\\":true,\\"query\\":{\\"term\\":{\\"user\\":\\"kimchy\\"}}}",
-                      undefined,
-                    ],
-                  ]
-              `);
-      });
-
-      it('when request body is a buffer', () => {
-        const client = configureClient(createFakeConfig(), { logger, type: 'test', scoped: false });
-
-        const response = createResponseWithBody(
-          Buffer.from(
-            JSON.stringify({
-              seq_no_primary_term: true,
-              query: {
-                term: { user: 'kimchy' },
-              },
-            })
-          )
-        );
-
-        client.emit('response', null, response);
-        expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
-          Array [
-            Array [
-              "200
-          GET /foo?hello=dolly
-          [buffer]",
-              undefined,
-            ],
-          ]
-        `);
-      });
-
-      it('when request body is a readable stream', () => {
-        const client = configureClient(createFakeConfig(), { logger, type: 'test', scoped: false });
-
-        const response = createResponseWithBody(
-          Readable.from(
-            JSON.stringify({
-              seq_no_primary_term: true,
-              query: {
-                term: { user: 'kimchy' },
-              },
-            })
-          )
-        );
-
-        client.emit('response', null, response);
-        expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
-          Array [
-            Array [
-              "200
-          GET /foo?hello=dolly
-          [stream]",
-              undefined,
-            ],
-          ]
-        `);
-      });
-
-      it('when request body is not defined', () => {
-        const client = configureClient(createFakeConfig(), { logger, type: 'test', scoped: false });
-
-        const response = createResponseWithBody();
-
-        client.emit('response', null, response);
-        expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
-          Array [
-            Array [
-              "200
-          GET /foo?hello=dolly",
-              undefined,
-            ],
-          ]
-        `);
-      });
-
-      it('properly encode queries', () => {
-        const client = configureClient(createFakeConfig(), { logger, type: 'test', scoped: false });
-
-        const response = createApiResponse({
-          body: {},
-          statusCode: 200,
-          params: {
-            method: 'GET',
-            path: '/foo',
-            querystring: { city: 'Münich' },
-          },
-        });
-
-        client.emit('response', null, response);
-
-        expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
-                  Array [
-                    Array [
-                      "200
-                  GET /foo?city=M%C3%BCnich",
-                      undefined,
-                    ],
-                  ]
-              `);
-      });
-
-      it('logs queries even in case of errors', () => {
-        const client = configureClient(createFakeConfig(), { logger, type: 'test', scoped: false });
-
-        const response = createApiResponse({
-          statusCode: 500,
-          body: {
-            error: {
-              type: 'internal server error',
-            },
-          },
-          params: {
-            method: 'GET',
-            path: '/foo',
-            querystring: { hello: 'dolly' },
-            body: {
-              seq_no_primary_term: true,
-              query: {
-                term: { user: 'kimchy' },
-              },
-            },
-          },
-        });
-        client.emit('response', new errors.ResponseError(response), response);
-
-        expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
-          Array [
-            Array [
-              "500
-          GET /foo?hello=dolly
-          {\\"seq_no_primary_term\\":true,\\"query\\":{\\"term\\":{\\"user\\":\\"kimchy\\"}}} [internal server error]: internal server error",
-              undefined,
-            ],
-          ]
-        `);
-      });
-
-      it('logs debug when the client emits an @elastic/elasticsearch error', () => {
-        const client = configureClient(createFakeConfig(), { logger, type: 'test', scoped: false });
-
-        const response = createApiResponse({ body: {} });
-        client.emit('response', new errors.TimeoutError('message', response), response);
-
-        expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
-                  Array [
-                    Array [
-                      "[TimeoutError]: message",
-                      undefined,
-                    ],
-                  ]
-              `);
-      });
-
-      it('logs debug when the client emits an ResponseError returned by elasticsearch', () => {
-        const client = configureClient(createFakeConfig(), { logger, type: 'test', scoped: false });
-
-        const response = createApiResponse({
-          statusCode: 400,
-          headers: {},
-          params: {
-            method: 'GET',
-            path: '/_path',
-            querystring: { hello: 'dolly' },
-          },
-          body: {
-            error: {
-              type: 'illegal_argument_exception',
-              reason: 'request [/_path] contains unrecognized parameter: [name]',
-            },
-          },
-        });
-        client.emit('response', new errors.ResponseError(response), response);
-
-        expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
-          Array [
-            Array [
-              "400
-          GET /_path?hello=dolly [illegal_argument_exception]: request [/_path] contains unrecognized parameter: [name]",
-              undefined,
-            ],
-          ]
-        `);
-      });
-
-      it('logs default error info when the error response body is empty', () => {
-        const client = configureClient(createFakeConfig(), { logger, type: 'test', scoped: false });
-
-        let response: RequestEvent<any, any> = createApiResponse({
-          statusCode: 400,
-          headers: {},
-          params: {
-            method: 'GET',
-            path: '/_path',
-          },
-          body: {
-            error: {},
-          },
-        });
-        client.emit('response', new errors.ResponseError(response), response);
-
-        expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
-          Array [
-            Array [
-              "400
-          GET /_path [undefined]: {\\"error\\":{}}",
-              undefined,
-            ],
-          ]
-        `);
-
-        logger.debug.mockClear();
-
-        response = createApiResponse({
-          statusCode: 400,
-          headers: {},
-          params: {
-            method: 'GET',
-            path: '/_path',
-          },
-          body: undefined,
-        });
-        client.emit('response', new errors.ResponseError(response), response);
-
-        expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
-          Array [
-            Array [
-              "400
-          GET /_path [undefined]: Response Error",
-              undefined,
-            ],
-          ]
-        `);
-      });
-
-      it('adds meta information to logs', () => {
-        const client = configureClient(createFakeConfig(), { logger, type: 'test', scoped: false });
-
-        let response = createApiResponse({
-          statusCode: 400,
-          headers: {},
-          params: {
-            method: 'GET',
-            path: '/_path',
-          },
-          requestOptions: {
-            opaqueId: 'opaque-id',
-          },
-          body: {
-            error: {},
-          },
-        });
-        client.emit('response', null, response);
-
-        expect(loggingSystemMock.collect(logger).debug[0][1]).toMatchInlineSnapshot(`
-          Object {
-            "http": Object {
-              "request": Object {
-                "id": "opaque-id",
-              },
-            },
-          }
-        `);
-
-        logger.debug.mockClear();
-
-        response = createApiResponse({
-          statusCode: 400,
-          headers: {},
-          params: {
-            method: 'GET',
-            path: '/_path',
-          },
-          requestOptions: {
-            opaqueId: 'opaque-id',
-          },
-          body: {} as any,
-        });
-        client.emit('response', new errors.ResponseError(response), response);
-
-        expect(loggingSystemMock.collect(logger).debug[0][1]).toMatchInlineSnapshot(`
-          Object {
-            "http": Object {
-              "request": Object {
-                "id": "opaque-id",
-              },
-            },
-          }
-        `);
-      });
-    });
-  });
 });
diff --git a/src/core/server/elasticsearch/client/configure_client.ts b/src/core/server/elasticsearch/client/configure_client.ts
index 93c404593af3..a295d2a53d1a 100644
--- a/src/core/server/elasticsearch/client/configure_client.ts
+++ b/src/core/server/elasticsearch/client/configure_client.ts
@@ -6,16 +6,15 @@
  * Side Public License, v 1.
  */
 
-import { Buffer } from 'buffer';
-import { stringify } from 'querystring';
-import { ApiError, Client, RequestEvent, errors, Transport } from '@elastic/elasticsearch';
+import { Client, Transport } from '@elastic/elasticsearch';
+
 import type {
-  RequestBody,
   TransportRequestParams,
   TransportRequestOptions,
 } from '@elastic/elasticsearch/lib/Transport';
 import { Logger } from '../../logging';
 import { parseClientOptions, ElasticsearchClientConfig } from './client_config';
+import { instrumentEsQueryAndDeprecationLogger } from './log_query_and_deprecation';
 
 const noop = () => undefined;
 
@@ -47,7 +46,6 @@ export const configureClient = (
   }
 
   const client = new Client({ ...clientOptions, Transport: KibanaTransport });
-  addLogging(client, logger.get('query', type));
 
   // --------------------------------------------------------------------------------- //
   // Hack to disable the "Product check" only in the scoped clients while we           //
@@ -55,90 +53,9 @@ export const configureClient = (
   if (scoped) skipProductCheck(client);
   // --------------------------------------------------------------------------------- //
 
-  return client;
-};
-
-const convertQueryString = (qs: string | Record<string, any> | undefined): string => {
-  if (qs === undefined || typeof qs === 'string') {
-    return qs ?? '';
-  }
-  return stringify(qs);
-};
-
-function ensureString(body: RequestBody): string {
-  if (typeof body === 'string') return body;
-  if (Buffer.isBuffer(body)) return '[buffer]';
-  if ('readable' in body && body.readable && typeof body._read === 'function') return '[stream]';
-  return JSON.stringify(body);
-}
-
-/**
- * Returns a debug message from an Elasticsearch error in the following format:
- * [error type] error reason
- */
-export function getErrorMessage(error: ApiError): string {
-  if (error instanceof errors.ResponseError) {
-    return `[${error.meta.body?.error?.type}]: ${error.meta.body?.error?.reason ?? error.message}`;
-  }
-  return `[${error.name}]: ${error.message}`;
-}
-
-/**
- * returns a string in format:
- *
- * status code
- * method URL
- * request body
- *
- * so it could be copy-pasted into the Dev console
- */
-function getResponseMessage(event: RequestEvent): string {
-  const errorMeta = getRequestDebugMeta(event);
-  const body = errorMeta.body ? `\n${errorMeta.body}` : '';
-  return `${errorMeta.statusCode}\n${errorMeta.method} ${errorMeta.url}${body}`;
-}
+  instrumentEsQueryAndDeprecationLogger({ logger, client, type });
 
-/**
- * Returns stringified debug information from an Elasticsearch request event
- * useful for logging in case of an unexpected failure.
- */
-export function getRequestDebugMeta(event: RequestEvent): {
-  url: string;
-  body: string;
-  statusCode: number | null;
-  method: string;
-} {
-  const params = event.meta.request.params;
-  // definition is wrong, `params.querystring` can be either a string or an object
-  const querystring = convertQueryString(params.querystring);
-  return {
-    url: `${params.path}${querystring ? `?${querystring}` : ''}`,
-    body: params.body ? `${ensureString(params.body)}` : '',
-    method: params.method,
-    statusCode: event.statusCode,
-  };
-}
-
-const addLogging = (client: Client, logger: Logger) => {
-  client.on('response', (error, event) => {
-    if (event) {
-      const opaqueId = event.meta.request.options.opaqueId;
-      const meta = opaqueId
-        ? {
-            http: { request: { id: event.meta.request.options.opaqueId } },
-          }
-        : undefined; // do not clutter logs if opaqueId is not present
-      if (error) {
-        if (error instanceof errors.ResponseError) {
-          logger.debug(`${getResponseMessage(event)} ${getErrorMessage(error)}`, meta);
-        } else {
-          logger.debug(getErrorMessage(error), meta);
-        }
-      } else {
-        logger.debug(getResponseMessage(event), meta);
-      }
-    }
-  });
+  return client;
 };
 
 /**
diff --git a/src/core/server/elasticsearch/client/index.ts b/src/core/server/elasticsearch/client/index.ts
index 29f8b8569519..2d7076454ac3 100644
--- a/src/core/server/elasticsearch/client/index.ts
+++ b/src/core/server/elasticsearch/client/index.ts
@@ -20,5 +20,6 @@ export type { IScopedClusterClient } from './scoped_cluster_client';
 export type { ElasticsearchClientConfig } from './client_config';
 export { ClusterClient } from './cluster_client';
 export type { IClusterClient, ICustomClusterClient } from './cluster_client';
-export { configureClient, getRequestDebugMeta, getErrorMessage } from './configure_client';
+export { configureClient } from './configure_client';
+export { getRequestDebugMeta, getErrorMessage } from './log_query_and_deprecation';
 export { retryCallCluster, migrationRetryCallCluster } from './retry_call_cluster';
diff --git a/src/core/server/elasticsearch/client/log_query_and_deprecation.test.ts b/src/core/server/elasticsearch/client/log_query_and_deprecation.test.ts
new file mode 100644
index 000000000000..86655556bd16
--- /dev/null
+++ b/src/core/server/elasticsearch/client/log_query_and_deprecation.test.ts
@@ -0,0 +1,619 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { Buffer } from 'buffer';
+import { Readable } from 'stream';
+
+import { Client, errors } from '@elastic/elasticsearch';
+
+import { parseClientOptionsMock, ClientMock } from './configure_client.test.mocks';
+import { loggingSystemMock } from '../../logging/logging_system.mock';
+import { instrumentEsQueryAndDeprecationLogger } from './log_query_and_deprecation';
+import {
+  RequestBody,
+  RequestEvent,
+  TransportRequestOptions,
+  TransportRequestParams,
+} from '@elastic/elasticsearch/lib/Transport';
+
+const createApiResponse = <T>({
+  body,
+  statusCode = 200,
+  headers = {},
+  warnings = null,
+  params,
+  requestOptions = {},
+}: {
+  body: T;
+  statusCode?: number;
+  headers?: Record<string, string>;
+  warnings?: string[] | null;
+  params?: TransportRequestParams & { headers?: Record<string, string> };
+  requestOptions?: TransportRequestOptions;
+}): RequestEvent<T> => {
+  return {
+    body,
+    statusCode,
+    headers,
+    warnings,
+    meta: {
+      body,
+      request: {
+        params: params!,
+        options: requestOptions,
+      } as any,
+    } as any,
+  };
+};
+
+const createFakeClient = () => {
+  const actualEs = jest.requireActual('@elastic/elasticsearch');
+  const client = new actualEs.Client({
+    nodes: ['http://localhost'], // Enforcing `nodes` because it's mandatory
+  });
+  jest.spyOn(client, 'on');
+  return client as Client;
+};
+
+describe('instrumentQueryAndDeprecationLogger', () => {
+  let logger: ReturnType<typeof loggingSystemMock.createLogger>;
+  const client = createFakeClient();
+
+  beforeEach(() => {
+    logger = loggingSystemMock.createLogger();
+    parseClientOptionsMock.mockReturnValue({});
+    ClientMock.mockImplementation(() => createFakeClient());
+  });
+
+  afterEach(() => {
+    parseClientOptionsMock.mockReset();
+    ClientMock.mockReset();
+    jest.clearAllMocks();
+  });
+
+  function createResponseWithBody(body?: RequestBody) {
+    return createApiResponse({
+      body: {},
+      statusCode: 200,
+      params: {
+        method: 'GET',
+        path: '/foo',
+        querystring: { hello: 'dolly' },
+        body,
+      },
+    });
+  }
+
+  it('creates a query logger context based on the `type` parameter', () => {
+    instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test123' });
+    expect(logger.get).toHaveBeenCalledWith('query', 'test123');
+  });
+
+  describe('logs each query', () => {
+    it('when request body is an object', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createResponseWithBody({
+        seq_no_primary_term: true,
+        query: {
+          term: { user: 'kimchy' },
+        },
+      });
+
+      client.emit('response', null, response);
+      expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
+        Array [
+          Array [
+            "200
+        GET /foo?hello=dolly
+        {\\"seq_no_primary_term\\":true,\\"query\\":{\\"term\\":{\\"user\\":\\"kimchy\\"}}}",
+            undefined,
+          ],
+        ]
+      `);
+    });
+
+    it('when request body is a string', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createResponseWithBody(
+        JSON.stringify({
+          seq_no_primary_term: true,
+          query: {
+            term: { user: 'kimchy' },
+          },
+        })
+      );
+
+      client.emit('response', null, response);
+      expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
+        Array [
+          Array [
+            "200
+        GET /foo?hello=dolly
+        {\\"seq_no_primary_term\\":true,\\"query\\":{\\"term\\":{\\"user\\":\\"kimchy\\"}}}",
+            undefined,
+          ],
+        ]
+      `);
+    });
+
+    it('when request body is a buffer', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createResponseWithBody(
+        Buffer.from(
+          JSON.stringify({
+            seq_no_primary_term: true,
+            query: {
+              term: { user: 'kimchy' },
+            },
+          })
+        )
+      );
+
+      client.emit('response', null, response);
+      expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
+        Array [
+          Array [
+            "200
+        GET /foo?hello=dolly
+        [buffer]",
+            undefined,
+          ],
+        ]
+      `);
+    });
+
+    it('when request body is a readable stream', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createResponseWithBody(
+        Readable.from(
+          JSON.stringify({
+            seq_no_primary_term: true,
+            query: {
+              term: { user: 'kimchy' },
+            },
+          })
+        )
+      );
+
+      client.emit('response', null, response);
+      expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
+        Array [
+          Array [
+            "200
+        GET /foo?hello=dolly
+        [stream]",
+            undefined,
+          ],
+        ]
+      `);
+    });
+
+    it('when request body is not defined', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createResponseWithBody();
+
+      client.emit('response', null, response);
+      expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
+        Array [
+          Array [
+            "200
+        GET /foo?hello=dolly",
+            undefined,
+          ],
+        ]
+      `);
+    });
+
+    it('properly encode queries', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createApiResponse({
+        body: {},
+        statusCode: 200,
+        params: {
+          method: 'GET',
+          path: '/foo',
+          querystring: { city: 'Münich' },
+        },
+      });
+
+      client.emit('response', null, response);
+
+      expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
+        Array [
+          Array [
+            "200
+        GET /foo?city=M%C3%BCnich",
+            undefined,
+          ],
+        ]
+      `);
+    });
+
+    it('logs queries even in case of errors', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createApiResponse({
+        statusCode: 500,
+        body: {
+          error: {
+            type: 'internal server error',
+          },
+        },
+        params: {
+          method: 'GET',
+          path: '/foo',
+          querystring: { hello: 'dolly' },
+          body: {
+            seq_no_primary_term: true,
+            query: {
+              term: { user: 'kimchy' },
+            },
+          },
+        },
+      });
+      client.emit('response', new errors.ResponseError(response), response);
+
+      expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
+        Array [
+          Array [
+            "500
+        GET /foo?hello=dolly
+        {\\"seq_no_primary_term\\":true,\\"query\\":{\\"term\\":{\\"user\\":\\"kimchy\\"}}} [internal server error]: internal server error",
+            undefined,
+          ],
+        ]
+      `);
+    });
+
+    it('logs debug when the client emits an @elastic/elasticsearch error', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createApiResponse({ body: {} });
+      client.emit('response', new errors.TimeoutError('message', response), response);
+
+      expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
+        Array [
+          Array [
+            "[TimeoutError]: message",
+            undefined,
+          ],
+        ]
+      `);
+    });
+
+    it('logs debug when the client emits an ResponseError returned by elasticsearch', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createApiResponse({
+        statusCode: 400,
+        headers: {},
+        params: {
+          method: 'GET',
+          path: '/_path',
+          querystring: { hello: 'dolly' },
+        },
+        body: {
+          error: {
+            type: 'illegal_argument_exception',
+            reason: 'request [/_path] contains unrecognized parameter: [name]',
+          },
+        },
+      });
+      client.emit('response', new errors.ResponseError(response), response);
+
+      expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
+        Array [
+          Array [
+            "400
+        GET /_path?hello=dolly [illegal_argument_exception]: request [/_path] contains unrecognized parameter: [name]",
+            undefined,
+          ],
+        ]
+      `);
+    });
+
+    it('logs default error info when the error response body is empty', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      let response: RequestEvent<any, any> = createApiResponse({
+        statusCode: 400,
+        headers: {},
+        params: {
+          method: 'GET',
+          path: '/_path',
+        },
+        body: {
+          error: {},
+        },
+      });
+      client.emit('response', new errors.ResponseError(response), response);
+
+      expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
+        Array [
+          Array [
+            "400
+        GET /_path [undefined]: {\\"error\\":{}}",
+            undefined,
+          ],
+        ]
+      `);
+
+      logger.debug.mockClear();
+
+      response = createApiResponse({
+        statusCode: 400,
+        headers: {},
+        params: {
+          method: 'GET',
+          path: '/_path',
+        },
+        body: undefined,
+      });
+      client.emit('response', new errors.ResponseError(response), response);
+
+      expect(loggingSystemMock.collect(logger).debug).toMatchInlineSnapshot(`
+        Array [
+          Array [
+            "400
+        GET /_path [undefined]: Response Error",
+            undefined,
+          ],
+        ]
+      `);
+    });
+
+    it('adds meta information to logs', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      let response = createApiResponse({
+        statusCode: 400,
+        headers: {},
+        params: {
+          method: 'GET',
+          path: '/_path',
+        },
+        requestOptions: {
+          opaqueId: 'opaque-id',
+        },
+        body: {
+          error: {},
+        },
+      });
+      client.emit('response', null, response);
+
+      expect(loggingSystemMock.collect(logger).debug[0][1]).toMatchInlineSnapshot(`
+        Object {
+          "http": Object {
+            "request": Object {
+              "id": "opaque-id",
+            },
+          },
+        }
+      `);
+
+      logger.debug.mockClear();
+
+      response = createApiResponse({
+        statusCode: 400,
+        headers: {},
+        params: {
+          method: 'GET',
+          path: '/_path',
+        },
+        requestOptions: {
+          opaqueId: 'opaque-id',
+        },
+        body: {} as any,
+      });
+      client.emit('response', new errors.ResponseError(response), response);
+
+      expect(loggingSystemMock.collect(logger).debug[0][1]).toMatchInlineSnapshot(`
+        Object {
+          "http": Object {
+            "request": Object {
+              "id": "opaque-id",
+            },
+          },
+        }
+      `);
+    });
+  });
+
+  describe('deprecation warnings from response headers', () => {
+    it('does not log when no deprecation warning header is returned', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createApiResponse({
+        statusCode: 200,
+        warnings: null,
+        params: {
+          method: 'GET',
+          path: '/_path',
+          querystring: { hello: 'dolly' },
+        },
+        body: {
+          hits: [
+            {
+              _source: 'may the source be with you',
+            },
+          ],
+        },
+      });
+      client.emit('response', new errors.ResponseError(response), response);
+
+      // One debug log entry from 'elasticsearch.query' context
+      expect(loggingSystemMock.collect(logger).debug.length).toEqual(1);
+      expect(loggingSystemMock.collect(logger).info).toEqual([]);
+    });
+
+    it('does not log when warning header comes from a warn-agent that is not elasticsearch', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createApiResponse({
+        statusCode: 200,
+        warnings: [
+          '299 nginx/2.3.1 "GET /_path is deprecated"',
+          '299 nginx/2.3.1 "GET hello query param is deprecated"',
+        ],
+        params: {
+          method: 'GET',
+          path: '/_path',
+          querystring: { hello: 'dolly' },
+        },
+        body: {
+          hits: [
+            {
+              _source: 'may the source be with you',
+            },
+          ],
+        },
+      });
+      client.emit('response', new errors.ResponseError(response), response);
+
+      // One debug log entry from 'elasticsearch.query' context
+      expect(loggingSystemMock.collect(logger).debug.length).toEqual(1);
+      expect(loggingSystemMock.collect(logger).info).toEqual([]);
+    });
+
+    it('logs error when the client receives an Elasticsearch error response for a deprecated request originating from a user', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createApiResponse({
+        statusCode: 400,
+        warnings: ['299 Elasticsearch-8.1.0 "GET /_path is deprecated"'],
+        params: {
+          method: 'GET',
+          path: '/_path',
+          querystring: { hello: 'dolly' },
+        },
+        body: {
+          error: {
+            type: 'illegal_argument_exception',
+            reason: 'request [/_path] contains unrecognized parameter: [name]',
+          },
+        },
+      });
+      client.emit('response', new errors.ResponseError(response), response);
+
+      expect(loggingSystemMock.collect(logger).info).toEqual([]);
+      // Test debug[1] since theree is one log entry from 'elasticsearch.query' context
+      expect(loggingSystemMock.collect(logger).debug[1][0]).toMatch(
+        'Elasticsearch deprecation: 299 Elasticsearch-8.1.0 "GET /_path is deprecated"'
+      );
+      expect(loggingSystemMock.collect(logger).debug[1][0]).toMatch('Origin:user');
+      expect(loggingSystemMock.collect(logger).debug[1][0]).toMatch(
+        /Query:\n.*400\n.*GET \/_path\?hello\=dolly \[illegal_argument_exception\]: request \[\/_path\] contains unrecognized parameter: \[name\]/
+      );
+    });
+
+    it('logs warning when the client receives an Elasticsearch error response for a deprecated request originating from kibana', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createApiResponse({
+        statusCode: 400,
+        warnings: ['299 Elasticsearch-8.1.0 "GET /_path is deprecated"'],
+        params: {
+          method: 'GET',
+          path: '/_path',
+          querystring: { hello: 'dolly' },
+          // Set the request header to indicate to Elasticsearch that this is a request over which users have no control
+          headers: { 'x-elastic-product-origin': 'kibana' },
+        },
+        body: {
+          error: {
+            type: 'illegal_argument_exception',
+            reason: 'request [/_path] contains unrecognized parameter: [name]',
+          },
+        },
+      });
+      client.emit('response', new errors.ResponseError(response), response);
+
+      // One debug log entry from 'elasticsearch.query' context
+      expect(loggingSystemMock.collect(logger).debug.length).toEqual(1);
+      expect(loggingSystemMock.collect(logger).info[0][0]).toMatch(
+        'Elasticsearch deprecation: 299 Elasticsearch-8.1.0 "GET /_path is deprecated"'
+      );
+      expect(loggingSystemMock.collect(logger).info[0][0]).toMatch('Origin:kibana');
+      expect(loggingSystemMock.collect(logger).info[0][0]).toMatch(
+        /Query:\n.*400\n.*GET \/_path\?hello\=dolly \[illegal_argument_exception\]: request \[\/_path\] contains unrecognized parameter: \[name\]/
+      );
+    });
+
+    it('logs error when the client receives an Elasticsearch success response for a deprecated request originating from a user', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createApiResponse({
+        statusCode: 200,
+        warnings: ['299 Elasticsearch-8.1.0 "GET /_path is deprecated"'],
+        params: {
+          method: 'GET',
+          path: '/_path',
+          querystring: { hello: 'dolly' },
+        },
+        body: {
+          hits: [
+            {
+              _source: 'may the source be with you',
+            },
+          ],
+        },
+      });
+      client.emit('response', null, response);
+
+      expect(loggingSystemMock.collect(logger).info).toEqual([]);
+      // Test debug[1] since theree is one log entry from 'elasticsearch.query' context
+      expect(loggingSystemMock.collect(logger).debug[1][0]).toMatch(
+        'Elasticsearch deprecation: 299 Elasticsearch-8.1.0 "GET /_path is deprecated"'
+      );
+      expect(loggingSystemMock.collect(logger).debug[1][0]).toMatch('Origin:user');
+      expect(loggingSystemMock.collect(logger).debug[1][0]).toMatch(
+        /Query:\n.*200\n.*GET \/_path\?hello\=dolly/
+      );
+    });
+
+    it('logs warning when the client receives an Elasticsearch success response for a deprecated request originating from kibana', () => {
+      instrumentEsQueryAndDeprecationLogger({ logger, client, type: 'test type' });
+
+      const response = createApiResponse({
+        statusCode: 200,
+        warnings: ['299 Elasticsearch-8.1.0 "GET /_path is deprecated"'],
+        params: {
+          method: 'GET',
+          path: '/_path',
+          querystring: { hello: 'dolly' },
+          // Set the request header to indicate to Elasticsearch that this is a request over which users have no control
+          headers: { 'x-elastic-product-origin': 'kibana' },
+        },
+        body: {
+          hits: [
+            {
+              _source: 'may the source be with you',
+            },
+          ],
+        },
+      });
+      client.emit('response', null, response);
+
+      // One debug log entry from 'elasticsearch.query' context
+      expect(loggingSystemMock.collect(logger).debug.length).toEqual(1);
+      expect(loggingSystemMock.collect(logger).info[0][0]).toMatch(
+        'Elasticsearch deprecation: 299 Elasticsearch-8.1.0 "GET /_path is deprecated"'
+      );
+      expect(loggingSystemMock.collect(logger).info[0][0]).toMatch('Origin:kibana');
+      expect(loggingSystemMock.collect(logger).info[0][0]).toMatch(
+        /Query:\n.*200\n.*GET \/_path\?hello\=dolly/
+      );
+    });
+  });
+});
diff --git a/src/core/server/elasticsearch/client/log_query_and_deprecation.ts b/src/core/server/elasticsearch/client/log_query_and_deprecation.ts
new file mode 100644
index 000000000000..85d0518d9bae
--- /dev/null
+++ b/src/core/server/elasticsearch/client/log_query_and_deprecation.ts
@@ -0,0 +1,140 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { Buffer } from 'buffer';
+import { stringify } from 'querystring';
+import { errors, Client, ApiError, RequestEvent } from '@elastic/elasticsearch';
+import { RequestBody } from '@elastic/elasticsearch/lib/Transport';
+import { Logger } from '../../logging';
+
+const convertQueryString = (qs: string | Record<string, any> | undefined): string => {
+  if (qs === undefined || typeof qs === 'string') {
+    return qs ?? '';
+  }
+  return stringify(qs);
+};
+
+function ensureString(body: RequestBody): string {
+  if (typeof body === 'string') return body;
+  if (Buffer.isBuffer(body)) return '[buffer]';
+  if ('readable' in body && body.readable && typeof body._read === 'function') return '[stream]';
+  return JSON.stringify(body);
+}
+
+/**
+ * Returns a debug message from an Elasticsearch error in the following format:
+ * [error type] error reason
+ */
+export function getErrorMessage(error: ApiError): string {
+  if (error instanceof errors.ResponseError) {
+    return `[${error.meta.body?.error?.type}]: ${error.meta.body?.error?.reason ?? error.message}`;
+  }
+  return `[${error.name}]: ${error.message}`;
+}
+
+/**
+ * returns a string in format:
+ *
+ * status code
+ * method URL
+ * request body
+ *
+ * so it could be copy-pasted into the Dev console
+ */
+function getResponseMessage(event: RequestEvent): string {
+  const errorMeta = getRequestDebugMeta(event);
+  const body = errorMeta.body ? `\n${errorMeta.body}` : '';
+  return `${errorMeta.statusCode}\n${errorMeta.method} ${errorMeta.url}${body}`;
+}
+
+/**
+ * Returns stringified debug information from an Elasticsearch request event
+ * useful for logging in case of an unexpected failure.
+ */
+export function getRequestDebugMeta(event: RequestEvent): {
+  url: string;
+  body: string;
+  statusCode: number | null;
+  method: string;
+} {
+  const params = event.meta.request.params;
+  // definition is wrong, `params.querystring` can be either a string or an object
+  const querystring = convertQueryString(params.querystring);
+  return {
+    url: `${params.path}${querystring ? `?${querystring}` : ''}`,
+    body: params.body ? `${ensureString(params.body)}` : '',
+    method: params.method,
+    statusCode: event.statusCode,
+  };
+}
+
+/** HTTP Warning headers have the following syntax:
+ * <warn-code> <warn-agent> <warn-text> (where warn-code is a three digit number)
+ * This function tests if a warning comes from an Elasticsearch warn-agent
+ * */
+const isEsWarning = (warning: string) => /\d\d\d Elasticsearch-/.test(warning);
+
+export const instrumentEsQueryAndDeprecationLogger = ({
+  logger,
+  client,
+  type,
+}: {
+  logger: Logger;
+  client: Client;
+  type: string;
+}) => {
+  const queryLogger = logger.get('query', type);
+  const deprecationLogger = logger.get('deprecation');
+  client.on('response', (error, event) => {
+    if (event) {
+      const opaqueId = event.meta.request.options.opaqueId;
+      const meta = opaqueId
+        ? {
+            http: { request: { id: event.meta.request.options.opaqueId } },
+          }
+        : undefined; // do not clutter logs if opaqueId is not present
+      let queryMsg = '';
+      if (error) {
+        if (error instanceof errors.ResponseError) {
+          queryMsg = `${getResponseMessage(event)} ${getErrorMessage(error)}`;
+        } else {
+          queryMsg = getErrorMessage(error);
+        }
+      } else {
+        queryMsg = getResponseMessage(event);
+      }
+
+      queryLogger.debug(queryMsg, meta);
+
+      if (event.warnings && event.warnings.filter(isEsWarning).length > 0) {
+        // Plugins can explicitly mark requests as originating from a user by
+        // removing the `'x-elastic-product-origin': 'kibana'` header that's
+        // added by default. User requests will be shown to users in the
+        // upgrade assistant UI as an action item that has to be addressed
+        // before they upgrade.
+        // Kibana requests will be hidden from the upgrade assistant UI and are
+        // only logged to help developers maintain their plugins
+
+        // @ts-expect-error Elasticsearch typings don't include headers field
+        const headers = event.meta.request.params.headers;
+        const requestOrigin =
+          (headers != null && (headers['x-elastic-product-origin'] as unknown as string)) ===
+          'kibana'
+            ? 'kibana'
+            : 'user';
+
+        const deprecationMsg = `Elasticsearch deprecation: ${event.warnings}\nOrigin:${requestOrigin}\nQuery:\n${queryMsg}`;
+        if (requestOrigin === 'kibana') {
+          deprecationLogger.info(deprecationMsg);
+        } else {
+          deprecationLogger.debug(deprecationMsg);
+        }
+      }
+    }
+  });
+};
diff --git a/test/common/config.js b/test/common/config.js
index 142c8ebd5cbc..d1a08d40f708 100644
--- a/test/common/config.js
+++ b/test/common/config.js
@@ -53,6 +53,12 @@ export default function () {
         `--plugin-path=${path.join(__dirname, 'fixtures', 'plugins', 'newsfeed')}`,
         `--newsfeed.service.urlRoot=${servers.kibana.protocol}://${servers.kibana.hostname}:${servers.kibana.port}`,
         `--newsfeed.service.pathTemplate=/api/_newsfeed-FTS-external-service-simulators/kibana/v{VERSION}.json`,
+        // Log deprecation warnings on a single line so that they aren't cut off on Buildkite logs
+        '--logging.appenders.deprecation.type=console',
+        '--logging.appenders.deprecation.layout.type=json',
+        '--logging.loggers[0].name=elasticsearch.deprecation',
+        '--logging.loggers[0].level=all',
+        '--logging.loggers[0].appenders[0]=deprecation',
       ],
     },
     services,