Allow headers config to propagate in callAsCurrentUser

src/core/server/elasticsearch/scoped_cluster_client.test.ts

@@ -106,6 +106,17 @@ describe('#callAsCurrentUser', () => {
     );
     scopedAPICaller.mockClear();
 
+    await expect(
+      clusterClient.callAsCurrentUser('security.authenticate', { some: 'some' })
+    ).resolves.toBe(mockResponse);
+    expect(scopedAPICaller).toHaveBeenCalledTimes(1);
+    expect(scopedAPICaller).toHaveBeenCalledWith(
+      'security.authenticate',
+      { some: 'some', headers: { one: '1' } },
+      undefined
+    );
+    scopedAPICaller.mockClear();
+
     await expect(
       clusterClient.callAsCurrentUser('ping', undefined, { wrap401Errors: true })
     ).resolves.toBe(mockResponse);
@@ -120,10 +131,11 @@ describe('#callAsCurrentUser', () => {
     await expect(
       clusterClient.callAsCurrentUser(
         'security.authenticate',
-        { some: 'some', headers: { one: '1' } },
+        { some: 'some' },
         { wrap401Errors: true }
       )
     ).resolves.toBe(mockResponse);
+
     expect(scopedAPICaller).toHaveBeenCalledTimes(1);
     expect(scopedAPICaller).toHaveBeenCalledWith(
       'security.authenticate',
@@ -134,6 +146,31 @@ describe('#callAsCurrentUser', () => {
     expect(internalAPICaller).not.toHaveBeenCalled();
   });
 
+  test('callAsCurrentUser allows passing additional headers', async () => {
+    const mockResponse = { data: 'response' };
+    scopedAPICaller.mockResolvedValue(mockResponse);
+    await expect(
+      clusterClient.callAsCurrentUser('security.authenticate', {
+        some: 'some',
+        headers: { additionalHeader: 'Oh Yes!' },
+      })
+    ).resolves.toBe(mockResponse);
+    expect(scopedAPICaller).toHaveBeenCalledTimes(1);
+    expect(scopedAPICaller).toHaveBeenCalledWith(
+      'security.authenticate',
+      { some: 'some', headers: { one: '1', additionalHeader: 'Oh Yes!' } },
+      undefined
+    );
+  });
+
+  test('callAsCurrentUser cannot override default headers', async () => {
+    const expectedErrorResponse = new Error('Cannot override default header one.');
+    const withHeaderOverride = async () =>
+      clusterClient.callAsCurrentUser('security.authenticate', { headers: { one: 'OVERRIDE' } });
+    await expect(withHeaderOverride()).rejects.toThrowError(expectedErrorResponse);
+    expect(scopedAPICaller).toHaveBeenCalledTimes(0);
+  });
+
   test('properly forwards errors returned by the API caller', async () => {
     const mockErrorResponse = new Error('some-error');
     scopedAPICaller.mockRejectedValue(mockErrorResponse);
@@ -152,8 +189,8 @@ describe('#callAsCurrentUser', () => {
     await expect(clusterClientWithoutHeaders.callAsCurrentUser('ping')).resolves.toBe(mockResponse);
     expect(scopedAPICaller).toHaveBeenCalledTimes(1);
     expect(scopedAPICaller).toHaveBeenCalledWith('ping', {}, undefined);
-    scopedAPICaller.mockClear();
 
+    scopedAPICaller.mockClear();
     await expect(
       clusterClientWithoutHeaders.callAsCurrentUser('security.authenticate', { some: 'some' })
     ).resolves.toBe(mockResponse);

src/core/server/elasticsearch/scoped_cluster_client.ts

@@ -17,6 +17,7 @@
  * under the License.
  */
 
+import { intersection, isObject } from 'lodash';
 import { Headers } from '../http/router';
 import { CallAPIOptions } from './cluster_client';
 
@@ -72,10 +73,20 @@ export class ScopedClusterClient {
     clientParams: Record<string, unknown> = {},
     options?: CallAPIOptions
   ) {
-    if (this.headers !== undefined) {
-      clientParams.headers = this.headers;
-    }
+    const defaultHeaders = this.headers;
+    if (defaultHeaders !== undefined) {
+      const customHeaders: any = clientParams.headers;
+      if (isObject(customHeaders)) {
+        const duplicates = intersection(Object.keys(defaultHeaders), Object.keys(customHeaders));
+        duplicates.forEach(duplicate => {
+          if (defaultHeaders[duplicate] !== (customHeaders as any)[duplicate]) {
+            throw Error(`Cannot override default header ${duplicate}.`);
+          }
+        });
+      }
 
+      clientParams.headers = Object.assign({}, clientParams.headers, this.headers);
+    }
     return this.scopedAPICaller(endpoint, clientParams, options);
   }
 }

x-pack/plugins/xpack_main/common/constants.js

@@ -75,3 +75,9 @@ export const PRIVACY_STATEMENT_URL = `https://www.elastic.co/legal/telemetry-pri
  * @type {string}
  */
 export const KIBANA_LOCALIZATION_STATS_TYPE = 'localization';
+
+/**
+ * The header sent by telemetry service when hitting Elasticsearch to identify query source
+ * @type {string}
+ */
+export const TELEMETRY_QUERY_SOURCE = 'TELEMETRY';

x-pack/plugins/xpack_main/server/lib/telemetry/monitoring/get_high_level_stats.js

@@ -7,7 +7,7 @@
 import { get } from 'lodash';
 import { createQuery } from './create_query';
 import { INDEX_PATTERN_KIBANA, INDEX_PATTERN_BEATS, INDEX_PATTERN_LOGSTASH } from '../../../../../monitoring/common/constants';
-import { KIBANA_SYSTEM_ID, BEATS_SYSTEM_ID, APM_SYSTEM_ID, LOGSTASH_SYSTEM_ID } from '../../../../common/constants';
+import { KIBANA_SYSTEM_ID, BEATS_SYSTEM_ID, APM_SYSTEM_ID, LOGSTASH_SYSTEM_ID, TELEMETRY_QUERY_SOURCE } from '../../../../common/constants';
 
 /**
  * Update a counter associated with the {@code key}.
@@ -206,6 +206,9 @@ export async function fetchHighLevelStats(server, callCluster, clusterUuids, sta
   const params = {
     index: getIndexPatternForStackProduct(product),
     size: config.get('xpack.monitoring.max_bucket_size'),
+    headers: {
+      'X-QUERY-SOURCE': TELEMETRY_QUERY_SOURCE,
+    },
     ignoreUnavailable: true,
     filterPath: [
       'hits.hits._source.cluster_uuid',