diff --git a/src/fixtures/telemetry_collectors/nested_collector.ts b/src/fixtures/telemetry_collectors/nested_collector.ts
index bde89fe4a7060..ce563b46b0c4e 100644
--- a/src/fixtures/telemetry_collectors/nested_collector.ts
+++ b/src/fixtures/telemetry_collectors/nested_collector.ts
@@ -29,7 +29,7 @@ interface Usage {
 }
 
 export class NestedInside {
-  collector?: UsageCollector<Usage, Usage>;
+  collector?: UsageCollector<Usage>;
   createMyCollector() {
     this.collector = collectorSet.makeUsageCollector<Usage>({
       type: 'my_nested_collector',
diff --git a/src/plugins/data/server/server.api.md b/src/plugins/data/server/server.api.md
index bb7a8f58c926c..ba87453b0769c 100644
--- a/src/plugins/data/server/server.api.md
+++ b/src/plugins/data/server/server.api.md
@@ -27,15 +27,15 @@ import { ExpressionAstFunction } from 'src/plugins/expressions/common';
 import { ExpressionFunctionDefinition } from 'src/plugins/expressions/common';
 import { ExpressionsServerSetup } from 'src/plugins/expressions/server';
 import { ExpressionValueBoxed } from 'src/plugins/expressions/common';
-import { ISavedObjectsRepository } from 'kibana/server';
+import { ISavedObjectsRepository } from 'src/core/server';
 import { IScopedClusterClient } from 'src/core/server';
 import { ISearchOptions as ISearchOptions_2 } from 'src/plugins/data/public';
 import { ISearchSource } from 'src/plugins/data/public';
 import { IUiSettingsClient } from 'src/core/server';
 import { KibanaRequest } from 'src/core/server';
-import { LegacyAPICaller } from 'kibana/server';
-import { Logger } from 'kibana/server';
-import { Logger as Logger_2 } from 'src/core/server';
+import { LegacyAPICaller } from 'src/core/server';
+import { Logger } from 'src/core/server';
+import { Logger as Logger_2 } from 'kibana/server';
 import { LoggerFactory } from '@kbn/logging';
 import { Moment } from 'moment';
 import moment from 'moment';
diff --git a/src/plugins/kibana_usage_collection/server/collectors/application_usage/telemetry_application_usage_collector.test.ts b/src/plugins/kibana_usage_collection/server/collectors/application_usage/telemetry_application_usage_collector.test.ts
index c1457c64080a6..6cb104416ef58 100644
--- a/src/plugins/kibana_usage_collection/server/collectors/application_usage/telemetry_application_usage_collector.test.ts
+++ b/src/plugins/kibana_usage_collection/server/collectors/application_usage/telemetry_application_usage_collector.test.ts
@@ -19,7 +19,7 @@
 
 import { savedObjectsRepositoryMock, loggingSystemMock } from '../../../../../core/server/mocks';
 import {
-  CollectorOptions,
+  Collector,
   createUsageCollectionSetupMock,
 } from '../../../../usage_collection/server/usage_collection.mock';
 
@@ -40,11 +40,11 @@ describe('telemetry_application_usage', () => {
 
   const logger = loggingSystemMock.createLogger();
 
-  let collector: CollectorOptions;
+  let collector: Collector<unknown, unknown>;
 
   const usageCollectionMock = createUsageCollectionSetupMock();
   usageCollectionMock.makeUsageCollector.mockImplementation((config) => {
-    collector = config;
+    collector = new Collector(logger, config);
     return createUsageCollectionSetupMock().makeUsageCollector(config);
   });
 
diff --git a/src/plugins/kibana_usage_collection/server/collectors/core/index.test.ts b/src/plugins/kibana_usage_collection/server/collectors/core/index.test.ts
index e8efa9997c459..e31437a744e29 100644
--- a/src/plugins/kibana_usage_collection/server/collectors/core/index.test.ts
+++ b/src/plugins/kibana_usage_collection/server/collectors/core/index.test.ts
@@ -18,20 +18,22 @@
  */
 
 import {
-  CollectorOptions,
+  Collector,
   createUsageCollectionSetupMock,
 } from '../../../../usage_collection/server/usage_collection.mock';
 import { createCollectorFetchContextMock } from 'src/plugins/usage_collection/server/mocks';
 import { registerCoreUsageCollector } from '.';
-import { coreUsageDataServiceMock } from '../../../../../core/server/mocks';
+import { coreUsageDataServiceMock, loggingSystemMock } from '../../../../../core/server/mocks';
 import { CoreUsageData } from 'src/core/server/';
 
+const logger = loggingSystemMock.createLogger();
+
 describe('telemetry_core', () => {
-  let collector: CollectorOptions;
+  let collector: Collector<unknown, unknown>;
 
   const usageCollectionMock = createUsageCollectionSetupMock();
   usageCollectionMock.makeUsageCollector.mockImplementation((config) => {
-    collector = config;
+    collector = new Collector(logger, config);
     return createUsageCollectionSetupMock().makeUsageCollector(config);
   });
 
diff --git a/src/plugins/kibana_usage_collection/server/collectors/csp/csp_collector.test.ts b/src/plugins/kibana_usage_collection/server/collectors/csp/csp_collector.test.ts
index 03184d7385861..2851382f7559a 100644
--- a/src/plugins/kibana_usage_collection/server/collectors/csp/csp_collector.test.ts
+++ b/src/plugins/kibana_usage_collection/server/collectors/csp/csp_collector.test.ts
@@ -19,8 +19,13 @@
 
 import { CspConfig, ICspConfig } from '../../../../../core/server';
 import { createCspCollector } from './csp_collector';
-import { httpServiceMock } from '../../../../../core/server/mocks';
-import { createCollectorFetchContextMock } from 'src/plugins/usage_collection/server/mocks';
+import { httpServiceMock, loggingSystemMock } from '../../../../../core/server/mocks';
+import {
+  Collector,
+  createCollectorFetchContextMock,
+} from 'src/plugins/usage_collection/server/mocks';
+
+const logger = loggingSystemMock.createLogger();
 
 describe('csp collector', () => {
   let httpMock: ReturnType<typeof httpServiceMock.createSetupContract>;
@@ -36,7 +41,7 @@ describe('csp collector', () => {
   });
 
   test('fetches whether strict mode is enabled', async () => {
-    const collector = createCspCollector(httpMock);
+    const collector = new Collector(logger, createCspCollector(httpMock));
 
     expect((await collector.fetch(mockedFetchContext)).strict).toEqual(true);
 
@@ -45,7 +50,7 @@ describe('csp collector', () => {
   });
 
   test('fetches whether the legacy browser warning is enabled', async () => {
-    const collector = createCspCollector(httpMock);
+    const collector = new Collector(logger, createCspCollector(httpMock));
 
     expect((await collector.fetch(mockedFetchContext)).warnLegacyBrowsers).toEqual(true);
 
@@ -54,7 +59,7 @@ describe('csp collector', () => {
   });
 
   test('fetches whether the csp rules have been changed or not', async () => {
-    const collector = createCspCollector(httpMock);
+    const collector = new Collector(logger, createCspCollector(httpMock));
 
     expect((await collector.fetch(mockedFetchContext)).rulesChangedFromDefault).toEqual(false);
 
@@ -63,7 +68,7 @@ describe('csp collector', () => {
   });
 
   test('does not include raw csp rules under any property names', async () => {
-    const collector = createCspCollector(httpMock);
+    const collector = new Collector(logger, createCspCollector(httpMock));
 
     // It's important that we do not send the value of csp.rules here as it
     // can be customized with values that can be identifiable to given
diff --git a/src/plugins/kibana_usage_collection/server/collectors/kibana/index.test.ts b/src/plugins/kibana_usage_collection/server/collectors/kibana/index.test.ts
index 88ccb2016d420..16a60eca60f47 100644
--- a/src/plugins/kibana_usage_collection/server/collectors/kibana/index.test.ts
+++ b/src/plugins/kibana_usage_collection/server/collectors/kibana/index.test.ts
@@ -17,20 +17,25 @@
  * under the License.
  */
 
-import { pluginInitializerContextConfigMock } from '../../../../../core/server/mocks';
 import {
-  CollectorOptions,
+  loggingSystemMock,
+  pluginInitializerContextConfigMock,
+} from '../../../../../core/server/mocks';
+import {
+  Collector,
   createUsageCollectionSetupMock,
 } from '../../../../usage_collection/server/usage_collection.mock';
 import { createCollectorFetchContextMock } from '../../../../usage_collection/server/mocks';
 import { registerKibanaUsageCollector } from './';
 
+const logger = loggingSystemMock.createLogger();
+
 describe('telemetry_kibana', () => {
-  let collector: CollectorOptions;
+  let collector: Collector<unknown, unknown>;
 
   const usageCollectionMock = createUsageCollectionSetupMock();
   usageCollectionMock.makeUsageCollector.mockImplementation((config) => {
-    collector = config;
+    collector = new Collector(logger, config);
     return createUsageCollectionSetupMock().makeUsageCollector(config);
   });
 
diff --git a/src/plugins/kibana_usage_collection/server/collectors/management/index.test.ts b/src/plugins/kibana_usage_collection/server/collectors/management/index.test.ts
index e671f739ee083..0aafee01cf49d 100644
--- a/src/plugins/kibana_usage_collection/server/collectors/management/index.test.ts
+++ b/src/plugins/kibana_usage_collection/server/collectors/management/index.test.ts
@@ -17,21 +17,23 @@
  * under the License.
  */
 
-import { uiSettingsServiceMock } from '../../../../../core/server/mocks';
+import { loggingSystemMock, uiSettingsServiceMock } from '../../../../../core/server/mocks';
 import {
-  CollectorOptions,
+  Collector,
   createUsageCollectionSetupMock,
   createCollectorFetchContextMock,
 } from '../../../../usage_collection/server/usage_collection.mock';
 
 import { registerManagementUsageCollector } from './';
 
+const logger = loggingSystemMock.createLogger();
+
 describe('telemetry_application_usage_collector', () => {
-  let collector: CollectorOptions;
+  let collector: Collector<unknown, unknown>;
 
   const usageCollectionMock = createUsageCollectionSetupMock();
   usageCollectionMock.makeUsageCollector.mockImplementation((config) => {
-    collector = config;
+    collector = new Collector(logger, config);
     return createUsageCollectionSetupMock().makeUsageCollector(config);
   });
 
diff --git a/src/plugins/kibana_usage_collection/server/collectors/ops_stats/index.test.ts b/src/plugins/kibana_usage_collection/server/collectors/ops_stats/index.test.ts
index 61990730812cc..8db7010e64026 100644
--- a/src/plugins/kibana_usage_collection/server/collectors/ops_stats/index.test.ts
+++ b/src/plugins/kibana_usage_collection/server/collectors/ops_stats/index.test.ts
@@ -19,20 +19,23 @@
 
 import { Subject } from 'rxjs';
 import {
-  CollectorOptions,
+  Collector,
   createUsageCollectionSetupMock,
   createCollectorFetchContextMock,
 } from '../../../../usage_collection/server/usage_collection.mock';
 
 import { registerOpsStatsCollector } from './';
 import { OpsMetrics } from '../../../../../core/server';
+import { loggingSystemMock } from '../../../../../core/server/mocks';
+
+const logger = loggingSystemMock.createLogger();
 
 describe('telemetry_ops_stats', () => {
-  let collector: CollectorOptions;
+  let collector: Collector<unknown, unknown>;
 
   const usageCollectionMock = createUsageCollectionSetupMock();
   usageCollectionMock.makeStatsCollector.mockImplementation((config) => {
-    collector = config;
+    collector = new Collector(logger, config);
     return createUsageCollectionSetupMock().makeStatsCollector(config);
   });
 
diff --git a/src/plugins/kibana_usage_collection/server/collectors/ui_metric/index.test.ts b/src/plugins/kibana_usage_collection/server/collectors/ui_metric/index.test.ts
index 48e4e0d99d3cd..90e3b7110e1dc 100644
--- a/src/plugins/kibana_usage_collection/server/collectors/ui_metric/index.test.ts
+++ b/src/plugins/kibana_usage_collection/server/collectors/ui_metric/index.test.ts
@@ -17,21 +17,23 @@
  * under the License.
  */
 
-import { savedObjectsRepositoryMock } from '../../../../../core/server/mocks';
+import { loggingSystemMock, savedObjectsRepositoryMock } from '../../../../../core/server/mocks';
 import {
-  CollectorOptions,
+  Collector,
   createUsageCollectionSetupMock,
   createCollectorFetchContextMock,
 } from '../../../../usage_collection/server/usage_collection.mock';
 
 import { registerUiMetricUsageCollector } from './';
 
+const logger = loggingSystemMock.createLogger();
+
 describe('telemetry_ui_metric', () => {
-  let collector: CollectorOptions;
+  let collector: Collector<unknown, unknown>;
 
   const usageCollectionMock = createUsageCollectionSetupMock();
   usageCollectionMock.makeUsageCollector.mockImplementation((config) => {
-    collector = config;
+    collector = new Collector(logger, config);
     return createUsageCollectionSetupMock().makeUsageCollector(config);
   });
 
diff --git a/src/plugins/telemetry/server/telemetry_collection/get_kibana.ts b/src/plugins/telemetry/server/telemetry_collection/get_kibana.ts
index 654c5435650cf..207a467ca5fd0 100644
--- a/src/plugins/telemetry/server/telemetry_collection/get_kibana.ts
+++ b/src/plugins/telemetry/server/telemetry_collection/get_kibana.ts
@@ -21,6 +21,7 @@ import { omit } from 'lodash';
 import { UsageCollectionSetup } from 'src/plugins/usage_collection/server';
 import {
   ISavedObjectsRepository,
+  KibanaRequest,
   LegacyAPICaller,
   SavedObjectsClientContract,
 } from 'kibana/server';
@@ -89,8 +90,14 @@ export async function getKibana(
   usageCollection: UsageCollectionSetup,
   callWithInternalUser: LegacyAPICaller,
   asInternalUser: ElasticsearchClient,
-  soClient: SavedObjectsClientContract | ISavedObjectsRepository
+  soClient: SavedObjectsClientContract | ISavedObjectsRepository,
+  kibanaRequest: KibanaRequest | undefined // intentionally `| undefined` to enforce providing the parameter
 ): Promise<KibanaUsageStats> {
-  const usage = await usageCollection.bulkFetch(callWithInternalUser, asInternalUser, soClient);
+  const usage = await usageCollection.bulkFetch(
+    callWithInternalUser,
+    asInternalUser,
+    soClient,
+    kibanaRequest
+  );
   return usageCollection.toObject(usage);
 }
diff --git a/src/plugins/telemetry/server/telemetry_collection/get_local_stats.test.ts b/src/plugins/telemetry/server/telemetry_collection/get_local_stats.test.ts
index 9298b36ac3ea6..6231fd29e7d3d 100644
--- a/src/plugins/telemetry/server/telemetry_collection/get_local_stats.test.ts
+++ b/src/plugins/telemetry/server/telemetry_collection/get_local_stats.test.ts
@@ -24,7 +24,7 @@ import {
   usageCollectionPluginMock,
   createCollectorFetchContextMock,
 } from '../../../usage_collection/server/mocks';
-import { elasticsearchServiceMock } from '../../../../../src/core/server/mocks';
+import { elasticsearchServiceMock, httpServerMock } from '../../../../../src/core/server/mocks';
 
 function mockUsageCollection(kibanaUsage = {}) {
   const usageCollection = usageCollectionPluginMock.createSetupContract();
@@ -87,6 +87,7 @@ function mockStatsCollectionConfig(clusterInfo: any, clusterStats: any, kibana:
     ...createCollectorFetchContextMock(),
     esClient: mockGetLocalStats(clusterInfo, clusterStats),
     usageCollection: mockUsageCollection(kibana),
+    kibanaRequest: httpServerMock.createKibanaRequest(),
     timestamp: Date.now(),
   };
 }
diff --git a/src/plugins/telemetry/server/telemetry_collection/get_local_stats.ts b/src/plugins/telemetry/server/telemetry_collection/get_local_stats.ts
index 4aeefb1d81d6a..a3666683a05a1 100644
--- a/src/plugins/telemetry/server/telemetry_collection/get_local_stats.ts
+++ b/src/plugins/telemetry/server/telemetry_collection/get_local_stats.ts
@@ -62,16 +62,16 @@ export type TelemetryLocalStats = ReturnType<typeof handleLocalStats>;
 
 /**
  * Get statistics for all products joined by Elasticsearch cluster.
- * @param {Array} cluster uuids
- * @param {Object} config contains the new esClient already scoped contains usageCollection, callCluster, esClient, start, end
+ * @param {Array} cluster uuids array of cluster uuid's
+ * @param {Object} config contains the usageCollection, callCluster (deprecated), the esClient and Saved Objects client scoped to the request or the internal repository, and the kibana request
  * @param {Object} StatsCollectionContext contains logger and version (string)
  */
 export const getLocalStats: StatsGetter<{}, TelemetryLocalStats> = async (
-  clustersDetails, // array of cluster uuid's
-  config, // contains the new esClient already scoped contains usageCollection, callCluster, esClient, start, end and the saved objects client scoped to the request or the internal repository
-  context // StatsCollectionContext contains logger and version (string)
+  clustersDetails,
+  config,
+  context
 ) => {
-  const { callCluster, usageCollection, esClient, soClient } = config;
+  const { callCluster, usageCollection, esClient, soClient, kibanaRequest } = config;
 
   return await Promise.all(
     clustersDetails.map(async (clustersDetail) => {
@@ -79,7 +79,7 @@ export const getLocalStats: StatsGetter<{}, TelemetryLocalStats> = async (
         getClusterInfo(esClient), // cluster info
         getClusterStats(esClient), // cluster stats (not to be confused with cluster _state_)
         getNodesUsage(esClient), // nodes_usage info
-        getKibana(usageCollection, callCluster, esClient, soClient),
+        getKibana(usageCollection, callCluster, esClient, soClient, kibanaRequest),
         getDataTelemetry(esClient),
       ]);
       return handleLocalStats(
diff --git a/src/plugins/telemetry_collection_manager/server/plugin.ts b/src/plugins/telemetry_collection_manager/server/plugin.ts
index e1e1379097adf..2cd06f13a8855 100644
--- a/src/plugins/telemetry_collection_manager/server/plugin.ts
+++ b/src/plugins/telemetry_collection_manager/server/plugin.ts
@@ -157,7 +157,10 @@ export class TelemetryCollectionManagerPlugin
     const soClient = config.unencrypted
       ? collectionSoService.getScopedClient(config.request)
       : collectionSoService.createInternalRepository();
-    return { callCluster, timestamp, usageCollection, esClient, soClient };
+    // Provide the kibanaRequest so opted-in plugins can scope their custom clients only if the request is not encrypted
+    const kibanaRequest = config.unencrypted ? request : void 0;
+
+    return { callCluster, timestamp, usageCollection, esClient, soClient, kibanaRequest };
   }
 
   private async getOptInStats(optInStatus: boolean, config: StatsGetterConfig) {
diff --git a/src/plugins/telemetry_collection_manager/server/types.ts b/src/plugins/telemetry_collection_manager/server/types.ts
index 0100fdbbb3970..7d25b8c8261c4 100644
--- a/src/plugins/telemetry_collection_manager/server/types.ts
+++ b/src/plugins/telemetry_collection_manager/server/types.ts
@@ -79,6 +79,7 @@ export interface StatsCollectionConfig {
   timestamp: number;
   esClient: ElasticsearchClient;
   soClient: SavedObjectsClientContract | ISavedObjectsRepository;
+  kibanaRequest: KibanaRequest | undefined; // intentionally `| undefined` to enforce providing the parameter
 }
 
 export interface BasicStatsPayload {
diff --git a/src/plugins/usage_collection/README.md b/src/plugins/usage_collection/README.md
index 2ac3de510f8ae..5e6ed901c7647 100644
--- a/src/plugins/usage_collection/README.md
+++ b/src/plugins/usage_collection/README.md
@@ -31,7 +31,7 @@ Then you need to make the Telemetry service aware of the collector by registerin
     ```ts
     // server/plugin.ts
     import { UsageCollectionSetup } from 'src/plugins/usage_collection/server';
-    import { CoreSetup, CoreStart } from 'kibana/server';
+    import { CoreSetup, CoreStart } from 'src/core/server';
 
     class Plugin {
       public setup(core: CoreSetup, plugins: { usageCollection?: UsageCollectionSetup }) {
@@ -46,7 +46,7 @@ Then you need to make the Telemetry service aware of the collector by registerin
     ```ts
     // server/collectors/register.ts
     import { UsageCollectionSetup, CollectorFetchContext } from 'src/plugins/usage_collection/server';
-    import { APICluster } from 'kibana/server';
+    import { APICluster } from 'src/core/server';
 
     interface Usage {
       my_objects: {
@@ -95,8 +95,8 @@ Some background:
 
 - `isReady` (added in v7.2.0 and v6.8.4) is a way for a usage collector to announce that some async process must finish first before it can return data in the `fetch` method (e.g. a client needs to ne initialized, or the task manager needs to run a task first). If any collector reports that it is not ready when we call its `fetch` method, we reset a flag to try again and, after a set amount of time, collect data from those collectors that are ready and skip any that are not. This means that if a collector returns `true` for `isReady` and it actually isn't ready to return data, there won't be telemetry data from that collector in that telemetry report (usually once per day). You should consider what it means if your collector doesn't return data in the first few documents when Kibana starts or, if we should wait for any other reason (e.g. the task manager needs to run your task first). If you need to tell telemetry collection to wait, you should implement this function with custom logic. If your `fetch` method can run without the need of any previous dependencies, then you can return true for `isReady` as shown in the example below.
 
-- The `fetch` method needs to support multiple contexts in which it is called. For example, when stats are pulled from a Kibana Metricbeat module, the Beat calls Kibana's stats API to invoke usage collection.
-In this case, the `fetch` method is called as a result of an HTTP API request and `callCluster` wraps `callWithRequest` or `esClient` wraps `asCurrentUser`, where the request headers are expected to have read privilege on the entire `.kibana' index. The `fetch` method also exposes the saved objects client that will have the correct scope when the collectors' `fetch` method is called.
+- The `fetch` method needs to support multiple contexts in which it is called. For example, when a user requests the example of what we collect in the **Kibana>Advanced Settings>Usage data** section, the clients provided in the context of the function (`CollectorFetchContext`) are scoped to that user's privileges. The reason is to avoid exposing via telemetry any data that user should not have access to (i.e.: if the user does not have access to certain indices, they shouldn't be allowed to see the number of documents that exists in it). In this case, the `fetch` method receives the clients `callCluster`, `esClient` and `soClient` scoped to the user who performed the HTTP API request. Alternatively, when requesting the usage data to be reported to the Remote Telemetry Service, the clients are scoped to the internal Kibana user (`kibana_system`). Please, mind it might have lower-level access than the default super-admin `elastic` test user.   
+In some scenarios, your collector might need to maintain its own client. An example of that is the `monitoring` plugin, that maintains a connection to the Remote Monitoring Cluster to push its monitoring data. If that's the case, your plugin can opt-in to receive the additional `kibanaRequest` parameter by adding `extendFetchContext.kibanaRequest: true` to the collector's config: it will be appended to the context of the `fetch` method only if the request needs to be scoped to a user other than Kibana Internal, so beware that your collector will need to work for both scenarios (especially for the scenario when `kibanaRequest` is missing).
 
 Note: there will be many cases where you won't need to use the `callCluster`, `esClient` or `soClient` function that gets passed in to your `fetch` method at all. Your feature might have an accumulating value in server memory, or read something from the OS.
 
@@ -105,7 +105,7 @@ In the case of using a custom SavedObjects client, it is up to the plugin to ini
 ```ts
 // server/plugin.ts
 import { UsageCollectionSetup } from 'src/plugins/usage_collection/server';
-import { CoreSetup, CoreStart } from 'kibana/server';
+import { CoreSetup, CoreStart } from 'src/core/server';
 
 class Plugin {
   private savedObjectsRepository?: ISavedObjectsRepository;
diff --git a/src/plugins/usage_collection/server/collector/collector.ts b/src/plugins/usage_collection/server/collector/collector.ts
index c04b087d4adf5..797fdaa06a620 100644
--- a/src/plugins/usage_collection/server/collector/collector.ts
+++ b/src/plugins/usage_collection/server/collector/collector.ts
@@ -23,7 +23,8 @@ import {
   ElasticsearchClient,
   ISavedObjectsRepository,
   SavedObjectsClientContract,
-} from 'kibana/server';
+  KibanaRequest,
+} from 'src/core/server';
 
 export type CollectorFormatForBulkUpload<T, U> = (result: T) => { type: string; payload: U };
 
@@ -46,26 +47,71 @@ export type MakeSchemaFrom<Base> = {
     : RecursiveMakeSchemaFrom<Required<Base>[Key]>;
 };
 
-export interface CollectorFetchContext {
+/**
+ * The context for the `fetch` method: It includes the most commonly used clients in the collectors (ES and SO clients).
+ * Both are scoped based on the request and the context:
+ * - When users are requesting a sample of data, it is scoped to their role to avoid exposing data they shouldn't read
+ * - When building the telemetry data payload to report to the remote cluster, the requests are scoped to the `kibana` internal user
+ *
+ * @remark Bear in mind when testing your collector that your user has the same privileges as the Kibana Internal user to ensure the expected data is sent to the remote cluster.
+ */
+export type CollectorFetchContext<WithKibanaRequest extends boolean | undefined = false> = {
   /**
-   * @depricated Scoped Legacy Elasticsearch client: use esClient instead
+   * @deprecated Scoped Legacy Elasticsearch client: use esClient instead
    */
   callCluster: LegacyAPICaller;
   /**
-   * Request-scoped Elasticsearch client:
-   * - When users are requesting a sample of data, it is scoped to their role to avoid exposing data they should't read
-   * - When building the telemetry data payload to report to the remote cluster, the requests are scoped to the `kibana` internal user
+   * Request-scoped Elasticsearch client
+   * @remark Bear in mind when testing your collector that your user has the same privileges as the Kibana Internal user to ensure the expected data is sent to the remote cluster (more info: {@link CollectorFetchContext})
    */
   esClient: ElasticsearchClient;
   /**
-   * Request-scoped Saved Objects client:
-   * - When users are requesting a sample of data, it is scoped to their role to avoid exposing data they should't read
-   * - When building the telemetry data payload to report to the remote cluster, the requests are scoped to the `kibana` internal user
+   * Request-scoped Saved Objects client
+   * @remark Bear in mind when testing your collector that your user has the same privileges as the Kibana Internal user to ensure the expected data is sent to the remote cluster (more info: {@link CollectorFetchContext})
    */
   soClient: SavedObjectsClientContract | ISavedObjectsRepository;
+} & (WithKibanaRequest extends true
+  ? {
+      /**
+       * The KibanaRequest that can be used to scope the requests:
+       * It is provided only when your custom clients need to be scoped. If not available, you should use the Internal Client.
+       * More information about when scoping is needed: {@link CollectorFetchContext}
+       * @remark You should only use this if you implement your collector to deal with both scenarios: when provided and, especially, when not provided. When telemetry payload is sent to the remote service the `kibanaRequest` will not be provided.
+       */
+      kibanaRequest?: KibanaRequest;
+    }
+  : {});
+
+export type CollectorFetchMethod<
+  WithKibanaRequest extends boolean | undefined,
+  TReturn,
+  ExtraOptions extends object = {}
+> = (
+  this: Collector<TReturn, unknown> & ExtraOptions, // Specify the context of `this` for this.log and others to become available
+  context: CollectorFetchContext<WithKibanaRequest>
+) => Promise<TReturn> | TReturn;
+
+export interface ICollectorOptionsFetchExtendedContext<WithKibanaRequest extends boolean> {
+  /**
+   * Set to `true` if your `fetch` method requires the `KibanaRequest` object to be added in its context {@link CollectorFetchContextWithRequest}.
+   * @remark You should fully understand acknowledge that by using the `KibanaRequest` in your collector, you need to ensure it should specially work without it because it won't be provided when building the telemetry payload actually sent to the remote telemetry service.
+   */
+  kibanaRequest?: WithKibanaRequest;
 }
 
-export interface CollectorOptions<T = unknown, U = T> {
+export type CollectorOptionsFetchExtendedContext<
+  WithKibanaRequest extends boolean
+> = ICollectorOptionsFetchExtendedContext<WithKibanaRequest> &
+  (WithKibanaRequest extends true // If enforced to true via Types, the config must be expected
+    ? Required<Pick<ICollectorOptionsFetchExtendedContext<WithKibanaRequest>, 'kibanaRequest'>>
+    : {});
+
+export type CollectorOptions<
+  TFetchReturn = unknown,
+  UFormatBulkUploadPayload = TFetchReturn, // TODO: Once we remove bulk_uploader's dependency on usageCollection, we'll be able to remove this type
+  WithKibanaRequest extends boolean = boolean,
+  ExtraOptions extends object = {}
+> = {
   /**
    * Unique string identifier for the collector
    */
@@ -78,23 +124,42 @@ export interface CollectorOptions<T = unknown, U = T> {
   /**
    * Schema definition of the output of the `fetch` method.
    */
-  schema?: MakeSchemaFrom<T>;
-  fetch: (collectorFetchContext: CollectorFetchContext) => Promise<T> | T;
-  /*
+  schema?: MakeSchemaFrom<TFetchReturn>;
+  /**
+   * The method that will collect and return the data in the final format.
+   * @param collectorFetchContext {@link CollectorFetchContext}
+   */
+  fetch: CollectorFetchMethod<WithKibanaRequest, TFetchReturn, ExtraOptions>;
+  /**
    * A hook for allowing the fetched data payload to be organized into a typed
    * data model for internal bulk upload. See defaultFormatterForBulkUpload for
    * a generic example.
    * @deprecated Used only by the Legacy Monitoring collection (to be removed in 8.0)
    */
-  formatForBulkUpload?: CollectorFormatForBulkUpload<T, U>;
-}
+  formatForBulkUpload?: CollectorFormatForBulkUpload<TFetchReturn, UFormatBulkUploadPayload>;
+} & ExtraOptions &
+  (WithKibanaRequest extends true // If enforced to true via Types, the config must be enforced
+    ? {
+        extendFetchContext: CollectorOptionsFetchExtendedContext<WithKibanaRequest>;
+      }
+    : {
+        extendFetchContext?: CollectorOptionsFetchExtendedContext<WithKibanaRequest>;
+      });
 
-export class Collector<T = unknown, U = T> {
-  public readonly type: CollectorOptions<T, U>['type'];
-  public readonly init?: CollectorOptions<T, U>['init'];
-  public readonly fetch: CollectorOptions<T, U>['fetch'];
-  private readonly _formatForBulkUpload?: CollectorFormatForBulkUpload<T, U>;
-  public readonly isReady: CollectorOptions<T, U>['isReady'];
+export class Collector<
+  TFetchReturn,
+  UFormatBulkUploadPayload = TFetchReturn,
+  ExtraOptions extends object = {}
+> {
+  public readonly extendFetchContext: CollectorOptionsFetchExtendedContext<any>;
+  public readonly type: CollectorOptions<TFetchReturn, UFormatBulkUploadPayload, any>['type'];
+  public readonly init?: CollectorOptions<TFetchReturn, UFormatBulkUploadPayload, any>['init'];
+  public readonly fetch: CollectorFetchMethod<any, TFetchReturn, ExtraOptions>;
+  public readonly isReady: CollectorOptions<TFetchReturn, UFormatBulkUploadPayload, any>['isReady'];
+  private readonly _formatForBulkUpload?: CollectorFormatForBulkUpload<
+    TFetchReturn,
+    UFormatBulkUploadPayload
+  >;
   /*
    * @param {Object} logger - logger object
    * @param {String} options.type - property name as the key for the data
@@ -105,8 +170,16 @@ export class Collector<T = unknown, U = T> {
    * @param {Function} options.rest - optional other properties
    */
   constructor(
-    protected readonly log: Logger,
-    { type, init, fetch, formatForBulkUpload, isReady, ...options }: CollectorOptions<T, U>
+    public readonly log: Logger,
+    {
+      type,
+      init,
+      fetch,
+      formatForBulkUpload,
+      isReady,
+      extendFetchContext = {},
+      ...options
+    }: CollectorOptions<TFetchReturn, UFormatBulkUploadPayload, any, ExtraOptions>
   ) {
     if (type === undefined) {
       throw new Error('Collector must be instantiated with a options.type string property');
@@ -126,10 +199,11 @@ export class Collector<T = unknown, U = T> {
     this.init = init;
     this.fetch = fetch;
     this.isReady = typeof isReady === 'function' ? isReady : () => true;
+    this.extendFetchContext = extendFetchContext;
     this._formatForBulkUpload = formatForBulkUpload;
   }
 
-  public formatForBulkUpload(result: T) {
+  public formatForBulkUpload(result: TFetchReturn) {
     if (this._formatForBulkUpload) {
       return this._formatForBulkUpload(result);
     } else {
@@ -137,10 +211,10 @@ export class Collector<T = unknown, U = T> {
     }
   }
 
-  protected defaultFormatterForBulkUpload(result: T) {
+  protected defaultFormatterForBulkUpload(result: TFetchReturn) {
     return {
       type: this.type,
-      payload: (result as unknown) as U,
+      payload: (result as unknown) as UFormatBulkUploadPayload,
     };
   }
 }
diff --git a/src/plugins/usage_collection/server/collector/collector_set.test.ts b/src/plugins/usage_collection/server/collector/collector_set.test.ts
index 359a2d214f991..fc17ce131430c 100644
--- a/src/plugins/usage_collection/server/collector/collector_set.test.ts
+++ b/src/plugins/usage_collection/server/collector/collector_set.test.ts
@@ -47,6 +47,7 @@ describe('CollectorSet', () => {
     const mockCallCluster = jest.fn().mockResolvedValue({ passTest: 1000 });
     const mockEsClient = elasticsearchServiceMock.createClusterClient().asInternalUser;
     const mockSoClient = savedObjectsRepositoryMock.create();
+    const req = void 0; // No need to instantiate any KibanaRequest in these tests
 
     it('should throw an error if non-Collector type of object is registered', () => {
       const collectors = new CollectorSet({ logger });
@@ -93,7 +94,7 @@ describe('CollectorSet', () => {
         })
       );
 
-      const result = await collectors.bulkFetch(mockCallCluster, mockEsClient, mockSoClient);
+      const result = await collectors.bulkFetch(mockCallCluster, mockEsClient, mockSoClient, req);
       expect(loggerSpies.debug).toHaveBeenCalledTimes(1);
       expect(loggerSpies.debug).toHaveBeenCalledWith(
         'Fetching data from MY_TEST_COLLECTOR collector'
@@ -118,7 +119,7 @@ describe('CollectorSet', () => {
 
       let result;
       try {
-        result = await collectors.bulkFetch(mockCallCluster, mockEsClient, mockSoClient);
+        result = await collectors.bulkFetch(mockCallCluster, mockEsClient, mockSoClient, req);
       } catch (err) {
         // Do nothing
       }
@@ -136,7 +137,7 @@ describe('CollectorSet', () => {
         })
       );
 
-      const result = await collectors.bulkFetch(mockCallCluster, mockEsClient, mockSoClient);
+      const result = await collectors.bulkFetch(mockCallCluster, mockEsClient, mockSoClient, req);
       expect(result).toStrictEqual([
         {
           type: 'MY_TEST_COLLECTOR',
@@ -154,7 +155,7 @@ describe('CollectorSet', () => {
         } as any)
       );
 
-      const result = await collectors.bulkFetch(mockCallCluster, mockEsClient, mockSoClient);
+      const result = await collectors.bulkFetch(mockCallCluster, mockEsClient, mockSoClient, req);
       expect(result).toStrictEqual([
         {
           type: 'MY_TEST_COLLECTOR',
@@ -177,7 +178,7 @@ describe('CollectorSet', () => {
         })
       );
 
-      const result = await collectors.bulkFetch(mockCallCluster, mockEsClient, mockSoClient);
+      const result = await collectors.bulkFetch(mockCallCluster, mockEsClient, mockSoClient, req);
       expect(result).toStrictEqual([
         {
           type: 'MY_TEST_COLLECTOR',
@@ -274,4 +275,272 @@ describe('CollectorSet', () => {
       expect(collectors.isUsageCollector(void 0)).toEqual(false);
     });
   });
+
+  describe('makeStatsCollector', () => {
+    const collectorSet = new CollectorSet({ logger });
+    test('TS should hide kibanaRequest when not opted-in', () => {
+      collectorSet.makeStatsCollector({
+        type: 'MY_TEST_COLLECTOR',
+        isReady: () => true,
+        schema: { test: { type: 'long' } },
+        fetch: (ctx) => {
+          // @ts-expect-error
+          const { kibanaRequest } = ctx;
+          return { test: kibanaRequest ? 1 : 0 };
+        },
+      });
+    });
+
+    test('TS should hide kibanaRequest when not opted-in (explicit false)', () => {
+      collectorSet.makeStatsCollector({
+        type: 'MY_TEST_COLLECTOR',
+        isReady: () => true,
+        schema: { test: { type: 'long' } },
+        fetch: (ctx) => {
+          // @ts-expect-error
+          const { kibanaRequest } = ctx;
+          return { test: kibanaRequest ? 1 : 0 };
+        },
+        extendFetchContext: {
+          kibanaRequest: false,
+        },
+      });
+    });
+
+    test('TS should allow using kibanaRequest when opted-in (explicit true)', () => {
+      collectorSet.makeStatsCollector({
+        type: 'MY_TEST_COLLECTOR',
+        isReady: () => true,
+        schema: { test: { type: 'long' } },
+        fetch: (ctx) => {
+          const { kibanaRequest } = ctx;
+          return { test: kibanaRequest ? 1 : 0 };
+        },
+        extendFetchContext: {
+          kibanaRequest: true,
+        },
+      });
+    });
+
+    test('fetch can use the logger (TS allows it)', () => {
+      const collector = collectorSet.makeStatsCollector({
+        type: 'MY_TEST_COLLECTOR',
+        isReady: () => true,
+        schema: { test: { type: 'long' } },
+        fetch() {
+          this.log.info("I can use the Collector's class logger!");
+          return { test: 1 };
+        },
+      });
+      expect(
+        collector.fetch(
+          // @ts-expect-error: the test implementation is not using it
+          {}
+        )
+      ).toStrictEqual({ test: 1 });
+    });
+  });
+
+  describe('makeUsageCollector', () => {
+    const collectorSet = new CollectorSet({ logger });
+    describe('TS validations', () => {
+      describe('when types are inferred', () => {
+        test('TS should hide kibanaRequest when not opted-in', () => {
+          collectorSet.makeUsageCollector({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              // @ts-expect-error
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+          });
+        });
+
+        test('TS should hide kibanaRequest when not opted-in (explicit false)', () => {
+          collectorSet.makeUsageCollector({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              // @ts-expect-error
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+            extendFetchContext: {
+              kibanaRequest: false,
+            },
+          });
+        });
+
+        test('TS should allow using kibanaRequest when opted-in (explicit true)', () => {
+          collectorSet.makeUsageCollector({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+            extendFetchContext: {
+              kibanaRequest: true,
+            },
+          });
+        });
+      });
+
+      describe('when types are explicit', () => {
+        test('TS should hide `kibanaRequest` from ctx when undefined or false', () => {
+          collectorSet.makeUsageCollector<{ test: number }>({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              // @ts-expect-error
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+          });
+          collectorSet.makeUsageCollector<{ test: number }, unknown, false>({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              // @ts-expect-error
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+            extendFetchContext: {
+              kibanaRequest: false,
+            },
+          });
+          collectorSet.makeUsageCollector<{ test: number }, unknown, false>({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              // @ts-expect-error
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+          });
+        });
+        test('TS should not allow `true` when types declare false', () => {
+          // false is the default when at least 1 type is specified
+          collectorSet.makeUsageCollector<{ test: number }>({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              // @ts-expect-error
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+            extendFetchContext: {
+              // @ts-expect-error
+              kibanaRequest: true,
+            },
+          });
+          collectorSet.makeUsageCollector<{ test: number }, unknown, false>({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              // @ts-expect-error
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+            extendFetchContext: {
+              // @ts-expect-error
+              kibanaRequest: true,
+            },
+          });
+        });
+
+        test('TS should allow `true` when types explicitly declare `true` and do not allow `false` or undefined', () => {
+          // false is the default when at least 1 type is specified
+          collectorSet.makeUsageCollector<{ test: number }, unknown, true>({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+            extendFetchContext: {
+              kibanaRequest: true,
+            },
+          });
+          collectorSet.makeUsageCollector<{ test: number }, unknown, true>({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+            extendFetchContext: {
+              // @ts-expect-error
+              kibanaRequest: false,
+            },
+          });
+          collectorSet.makeUsageCollector<{ test: number }, unknown, true>({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+            extendFetchContext: {
+              // @ts-expect-error
+              kibanaRequest: undefined,
+            },
+          });
+          collectorSet.makeUsageCollector<{ test: number }, unknown, true>({
+            type: 'MY_TEST_COLLECTOR',
+            isReady: () => true,
+            schema: { test: { type: 'long' } },
+            fetch: (ctx) => {
+              const { kibanaRequest } = ctx;
+              return { test: kibanaRequest ? 1 : 0 };
+            },
+            // @ts-expect-error
+            extendFetchContext: {},
+          });
+          collectorSet.makeUsageCollector<{ test: number }, unknown, true>(
+            // @ts-expect-error
+            {
+              type: 'MY_TEST_COLLECTOR',
+              isReady: () => true,
+              schema: { test: { type: 'long' } },
+              fetch: (ctx) => {
+                const { kibanaRequest } = ctx;
+                return { test: kibanaRequest ? 1 : 0 };
+              },
+            }
+          );
+        });
+      });
+    });
+
+    test('fetch can use the logger (TS allows it)', () => {
+      const collector = collectorSet.makeUsageCollector({
+        type: 'MY_TEST_COLLECTOR',
+        isReady: () => true,
+        schema: { test: { type: 'long' } },
+        fetch() {
+          this.log.info("I can use the Collector's class logger!");
+          return { test: 1 };
+        },
+      });
+      expect(
+        collector.fetch(
+          // @ts-expect-error: the test implementation is not using it
+          {}
+        )
+      ).toStrictEqual({ test: 1 });
+    });
+  });
 });
diff --git a/src/plugins/usage_collection/server/collector/collector_set.ts b/src/plugins/usage_collection/server/collector/collector_set.ts
index c52830cda6513..fe4f3536ffed6 100644
--- a/src/plugins/usage_collection/server/collector/collector_set.ts
+++ b/src/plugins/usage_collection/server/collector/collector_set.ts
@@ -24,50 +24,79 @@ import {
   ElasticsearchClient,
   ISavedObjectsRepository,
   SavedObjectsClientContract,
-} from 'kibana/server';
+  KibanaRequest,
+} from 'src/core/server';
 import { Collector, CollectorOptions } from './collector';
 import { UsageCollector, UsageCollectorOptions } from './usage_collector';
 
+type AnyCollector = Collector<any, any>;
+type AnyUsageCollector = UsageCollector<any, any>;
+
 interface CollectorSetConfig {
   logger: Logger;
   maximumWaitTimeForAllCollectorsInS?: number;
-  collectors?: Array<Collector<any, any>>;
+  collectors?: AnyCollector[];
 }
 
 export class CollectorSet {
   private _waitingForAllCollectorsTimestamp?: number;
   private readonly logger: Logger;
   private readonly maximumWaitTimeForAllCollectorsInS: number;
-  private readonly collectors: Map<string, Collector<any, any>>;
+  private readonly collectors: Map<string, AnyCollector>;
   constructor({ logger, maximumWaitTimeForAllCollectorsInS, collectors = [] }: CollectorSetConfig) {
     this.logger = logger;
     this.collectors = new Map(collectors.map((collector) => [collector.type, collector]));
     this.maximumWaitTimeForAllCollectorsInS = maximumWaitTimeForAllCollectorsInS || 60;
   }
 
+  /**
+   * Instantiates a stats collector with the definition provided in the options
+   * @param options Definition of the collector {@link CollectorOptions}
+   */
   public makeStatsCollector = <
-    T,
-    U,
-    O extends CollectorOptions<T, U> = CollectorOptions<T, U> // Used to allow extra properties (the Collector constructor extends the class with the additional options provided)
+    TFetchReturn,
+    TFormatForBulkUpload,
+    WithKibanaRequest extends boolean,
+    ExtraOptions extends object = {}
   >(
-    options: O
+    options: CollectorOptions<TFetchReturn, TFormatForBulkUpload, WithKibanaRequest, ExtraOptions>
   ) => {
-    return new Collector(this.logger, options);
+    return new Collector<TFetchReturn, TFormatForBulkUpload, ExtraOptions>(this.logger, options);
   };
+
+  /**
+   * Instantiates an usage collector with the definition provided in the options
+   * @param options Definition of the collector {@link CollectorOptions}
+   */
   public makeUsageCollector = <
-    T,
-    U = T,
-    O extends UsageCollectorOptions<T, U> = UsageCollectorOptions<T, U>
+    TFetchReturn,
+    TFormatForBulkUpload = { usage: { [key: string]: TFetchReturn } },
+    // TODO: Right now, users will need to explicitly claim `true` for TS to allow `kibanaRequest` usage.
+    //  If we improve `telemetry-check-tools` so plugins do not need to specify TFetchReturn,
+    //  we'll be able to remove the type defaults and TS will successfully infer the config value as provided in JS.
+    WithKibanaRequest extends boolean = false,
+    ExtraOptions extends object = {}
   >(
-    options: O
+    options: UsageCollectorOptions<
+      TFetchReturn,
+      TFormatForBulkUpload,
+      WithKibanaRequest,
+      ExtraOptions
+    >
   ) => {
-    return new UsageCollector(this.logger, options);
+    return new UsageCollector<TFetchReturn, TFormatForBulkUpload, ExtraOptions>(
+      this.logger,
+      options
+    );
   };
 
-  /*
-   * @param collector {Collector} collector object
+  /**
+   * Registers a collector to be used when collecting all the usage and stats data
+   * @param collector Collector to be added to the set (previously created via `makeUsageCollector` or `makeStatsCollector`)
    */
-  public registerCollector = <T, U>(collector: Collector<T, U>) => {
+  public registerCollector = <TFetchReturn, TFormatForBulkUpload, ExtraOptions extends object>(
+    collector: Collector<TFetchReturn, TFormatForBulkUpload, ExtraOptions>
+  ) => {
     // check instanceof
     if (!(collector instanceof Collector)) {
       throw new Error('CollectorSet can only have Collector instances registered');
@@ -89,7 +118,7 @@ export class CollectorSet {
     return [...this.collectors.values()].find((c) => c.type === type);
   };
 
-  public isUsageCollector = (x: UsageCollector | any): x is UsageCollector => {
+  public isUsageCollector = (x: AnyUsageCollector | any): x is AnyUsageCollector => {
     return x instanceof UsageCollector;
   };
 
@@ -144,15 +173,22 @@ export class CollectorSet {
     callCluster: LegacyAPICaller,
     esClient: ElasticsearchClient,
     soClient: SavedObjectsClientContract | ISavedObjectsRepository,
-    collectors: Map<string, Collector<any, any>> = this.collectors
+    kibanaRequest: KibanaRequest | undefined, // intentionally `| undefined` to enforce providing the parameter
+    collectors: Map<string, AnyCollector> = this.collectors
   ) => {
     const responses = await Promise.all(
       [...collectors.values()].map(async (collector) => {
         this.logger.debug(`Fetching data from ${collector.type} collector`);
         try {
+          const context = {
+            callCluster,
+            esClient,
+            soClient,
+            ...(collector.extendFetchContext.kibanaRequest && { kibanaRequest }),
+          };
           return {
             type: collector.type,
-            result: await collector.fetch({ callCluster, esClient, soClient }),
+            result: await collector.fetch(context),
           };
         } catch (err) {
           this.logger.warn(err);
@@ -169,7 +205,7 @@ export class CollectorSet {
   /*
    * @return {new CollectorSet}
    */
-  public getFilteredCollectorSet = (filter: (col: Collector) => boolean) => {
+  public getFilteredCollectorSet = (filter: (col: AnyCollector) => boolean) => {
     const filtered = [...this.collectors.values()].filter(filter);
     return this.makeCollectorSetFromArray(filtered);
   };
@@ -177,13 +213,15 @@ export class CollectorSet {
   public bulkFetchUsage = async (
     callCluster: LegacyAPICaller,
     esClient: ElasticsearchClient,
-    savedObjectsClient: SavedObjectsClientContract | ISavedObjectsRepository
+    savedObjectsClient: SavedObjectsClientContract | ISavedObjectsRepository,
+    kibanaRequest: KibanaRequest | undefined // intentionally `| undefined` to enforce providing the parameter
   ) => {
     const usageCollectors = this.getFilteredCollectorSet((c) => c instanceof UsageCollector);
     return await this.bulkFetch(
       callCluster,
       esClient,
       savedObjectsClient,
+      kibanaRequest,
       usageCollectors.collectors
     );
   };
@@ -239,7 +277,7 @@ export class CollectorSet {
     return [...this.collectors.values()].some(someFn);
   };
 
-  private makeCollectorSetFromArray = (collectors: Collector[]) => {
+  private makeCollectorSetFromArray = (collectors: AnyCollector[]) => {
     return new CollectorSet({
       logger: this.logger,
       maximumWaitTimeForAllCollectorsInS: this.maximumWaitTimeForAllCollectorsInS,
diff --git a/src/plugins/usage_collection/server/collector/usage_collector.ts b/src/plugins/usage_collection/server/collector/usage_collector.ts
index 5bfc36537e0b0..a042ea113d5cc 100644
--- a/src/plugins/usage_collection/server/collector/usage_collector.ts
+++ b/src/plugins/usage_collection/server/collector/usage_collector.ts
@@ -22,25 +22,39 @@ import { KIBANA_STATS_TYPE } from '../../common/constants';
 import { Collector, CollectorOptions } from './collector';
 
 // Enforce the `schema` property for UsageCollectors
-export type UsageCollectorOptions<T = unknown, U = T> = CollectorOptions<T, U> &
-  Required<Pick<CollectorOptions<T, U>, 'schema'>>;
+export type UsageCollectorOptions<
+  TFetchReturn = unknown,
+  UFormatBulkUploadPayload = { usage: { [key: string]: TFetchReturn } },
+  WithKibanaRequest extends boolean = false,
+  ExtraOptions extends object = {}
+> = CollectorOptions<TFetchReturn, UFormatBulkUploadPayload, WithKibanaRequest, ExtraOptions> &
+  Required<Pick<CollectorOptions<TFetchReturn, UFormatBulkUploadPayload, boolean>, 'schema'>>;
 
-export class UsageCollector<T = unknown, U = { usage: { [key: string]: T } }> extends Collector<
-  T,
-  U
-> {
-  constructor(protected readonly log: Logger, collectorOptions: UsageCollectorOptions<T, U>) {
+export class UsageCollector<
+  TFetchReturn,
+  UFormatBulkUploadPayload = { usage: { [key: string]: TFetchReturn } },
+  ExtraOptions extends object = {}
+> extends Collector<TFetchReturn, UFormatBulkUploadPayload, ExtraOptions> {
+  constructor(
+    public readonly log: Logger,
+    collectorOptions: UsageCollectorOptions<
+      TFetchReturn,
+      UFormatBulkUploadPayload,
+      any,
+      ExtraOptions
+    >
+  ) {
     super(log, collectorOptions);
   }
 
-  protected defaultFormatterForBulkUpload(result: T) {
+  protected defaultFormatterForBulkUpload(result: TFetchReturn) {
     return {
       type: KIBANA_STATS_TYPE,
       payload: ({
         usage: {
           [this.type]: result,
         },
-      } as unknown) as U,
+      } as unknown) as UFormatBulkUploadPayload,
     };
   }
 }
diff --git a/src/plugins/usage_collection/server/config.ts b/src/plugins/usage_collection/server/config.ts
index 76379d9385cff..09b0e05025e63 100644
--- a/src/plugins/usage_collection/server/config.ts
+++ b/src/plugins/usage_collection/server/config.ts
@@ -18,7 +18,7 @@
  */
 
 import { schema, TypeOf } from '@kbn/config-schema';
-import { PluginConfigDescriptor } from 'kibana/server';
+import { PluginConfigDescriptor } from 'src/core/server';
 import { DEFAULT_MAXIMUM_WAIT_TIME_FOR_ALL_COLLECTORS_IN_S } from '../common/constants';
 
 export const configSchema = schema.object({
diff --git a/src/plugins/usage_collection/server/index.ts b/src/plugins/usage_collection/server/index.ts
index f7a08fdb5e9dd..48ea9afa13976 100644
--- a/src/plugins/usage_collection/server/index.ts
+++ b/src/plugins/usage_collection/server/index.ts
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-import { PluginInitializerContext } from 'kibana/server';
+import { PluginInitializerContext } from 'src/core/server';
 import { UsageCollectionPlugin } from './plugin';
 
 export {
diff --git a/src/plugins/usage_collection/server/mocks.ts b/src/plugins/usage_collection/server/mocks.ts
index d08db1eaec0e1..3d89380f629dc 100644
--- a/src/plugins/usage_collection/server/mocks.ts
+++ b/src/plugins/usage_collection/server/mocks.ts
@@ -20,7 +20,7 @@
 import { loggingSystemMock } from '../../../core/server/mocks';
 import { UsageCollectionSetup } from './plugin';
 import { CollectorSet } from './collector';
-export { createCollectorFetchContextMock } from './usage_collection.mock';
+export { Collector, createCollectorFetchContextMock } from './usage_collection.mock';
 
 const createSetupContract = () => {
   return {
diff --git a/src/plugins/usage_collection/server/plugin.ts b/src/plugins/usage_collection/server/plugin.ts
index 74e70d5ea9d35..9a8876446d01e 100644
--- a/src/plugins/usage_collection/server/plugin.ts
+++ b/src/plugins/usage_collection/server/plugin.ts
@@ -25,7 +25,7 @@ import {
   CoreStart,
   ISavedObjectsRepository,
   Plugin,
-} from 'kibana/server';
+} from 'src/core/server';
 import { ConfigType } from './config';
 import { CollectorSet } from './collector';
 import { setupRoutes } from './routes';
diff --git a/src/plugins/usage_collection/server/report/store_report.ts b/src/plugins/usage_collection/server/report/store_report.ts
index c40622831eeee..d9aac23fd1ff0 100644
--- a/src/plugins/usage_collection/server/report/store_report.ts
+++ b/src/plugins/usage_collection/server/report/store_report.ts
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-import { ISavedObjectsRepository, SavedObject } from 'kibana/server';
+import { ISavedObjectsRepository, SavedObject } from 'src/core/server';
 import { ReportSchemaType } from './schema';
 
 export async function storeReport(
diff --git a/src/plugins/usage_collection/server/routes/index.ts b/src/plugins/usage_collection/server/routes/index.ts
index b367ddc184be7..15d408ff3723b 100644
--- a/src/plugins/usage_collection/server/routes/index.ts
+++ b/src/plugins/usage_collection/server/routes/index.ts
@@ -22,7 +22,7 @@ import {
   ISavedObjectsRepository,
   MetricsServiceSetup,
   ServiceStatus,
-} from 'kibana/server';
+} from 'src/core/server';
 import { Observable } from 'rxjs';
 import { CollectorSet } from '../collector';
 import { registerUiMetricRoute } from './report_metrics';
diff --git a/src/plugins/usage_collection/server/routes/report_metrics.ts b/src/plugins/usage_collection/server/routes/report_metrics.ts
index a72222968eabf..590c3340697b8 100644
--- a/src/plugins/usage_collection/server/routes/report_metrics.ts
+++ b/src/plugins/usage_collection/server/routes/report_metrics.ts
@@ -18,7 +18,7 @@
  */
 
 import { schema } from '@kbn/config-schema';
-import { IRouter, ISavedObjectsRepository } from 'kibana/server';
+import { IRouter, ISavedObjectsRepository } from 'src/core/server';
 import { storeReport, reportSchema } from '../report';
 
 export function registerUiMetricRoute(
diff --git a/src/plugins/usage_collection/server/routes/stats/stats.ts b/src/plugins/usage_collection/server/routes/stats/stats.ts
index d38250067053c..16a1c2c742f04 100644
--- a/src/plugins/usage_collection/server/routes/stats/stats.ts
+++ b/src/plugins/usage_collection/server/routes/stats/stats.ts
@@ -27,6 +27,7 @@ import {
   ElasticsearchClient,
   IRouter,
   ISavedObjectsRepository,
+  KibanaRequest,
   LegacyAPICaller,
   MetricsServiceSetup,
   SavedObjectsClientContract,
@@ -67,9 +68,15 @@ export function registerStatsRoute({
   const getUsage = async (
     callCluster: LegacyAPICaller,
     esClient: ElasticsearchClient,
-    savedObjectsClient: SavedObjectsClientContract | ISavedObjectsRepository
+    savedObjectsClient: SavedObjectsClientContract | ISavedObjectsRepository,
+    kibanaRequest: KibanaRequest
   ): Promise<any> => {
-    const usage = await collectorSet.bulkFetchUsage(callCluster, esClient, savedObjectsClient);
+    const usage = await collectorSet.bulkFetchUsage(
+      callCluster,
+      esClient,
+      savedObjectsClient,
+      kibanaRequest
+    );
     return collectorSet.toObject(usage);
   };
 
@@ -115,7 +122,7 @@ export function registerStatsRoute({
         }
 
         const usagePromise = shouldGetUsage
-          ? getUsage(callCluster, asCurrentUser, savedObjectsClient)
+          ? getUsage(callCluster, asCurrentUser, savedObjectsClient, req)
           : Promise.resolve({});
         const [usage, clusterUuid] = await Promise.all([
           usagePromise,
diff --git a/src/plugins/usage_collection/server/usage_collection.mock.ts b/src/plugins/usage_collection/server/usage_collection.mock.ts
index c31756c60e32d..05dae8fa85164 100644
--- a/src/plugins/usage_collection/server/usage_collection.mock.ts
+++ b/src/plugins/usage_collection/server/usage_collection.mock.ts
@@ -19,13 +19,17 @@
 
 import {
   elasticsearchServiceMock,
+  httpServerMock,
+  loggingSystemMock,
   savedObjectsRepositoryMock,
 } from '../../../../src/core/server/mocks';
 
-import { CollectorOptions } from './collector/collector';
+import { CollectorOptions, Collector, UsageCollector } from './collector';
 import { UsageCollectionSetup, CollectorFetchContext } from './index';
 
-export { CollectorOptions };
+export { CollectorOptions, Collector };
+
+const logger = loggingSystemMock.createLogger();
 
 export const createUsageCollectionSetupMock = () => {
   const usageCollectionSetupMock: jest.Mocked<UsageCollectionSetup> = {
@@ -37,13 +41,13 @@ export const createUsageCollectionSetupMock = () => {
     // @ts-ignore jest.fn doesn't play nice with type guards
     isUsageCollector: jest.fn(),
     makeCollectorSetFromArray: jest.fn(),
-    makeStatsCollector: jest.fn(),
     map: jest.fn(),
     maximumWaitTimeForAllCollectorsInS: 0,
     some: jest.fn(),
     toApiFieldNames: jest.fn(),
     toObject: jest.fn(),
-    makeUsageCollector: jest.fn(),
+    makeStatsCollector: jest.fn().mockImplementation((cfg) => new Collector(logger, cfg)),
+    makeUsageCollector: jest.fn().mockImplementation((cfg) => new UsageCollector(logger, cfg)),
     registerCollector: jest.fn(),
   };
 
@@ -51,11 +55,23 @@ export const createUsageCollectionSetupMock = () => {
   return usageCollectionSetupMock;
 };
 
-export function createCollectorFetchContextMock(): jest.Mocked<CollectorFetchContext> {
-  const collectorFetchClientsMock: jest.Mocked<CollectorFetchContext> = {
+export function createCollectorFetchContextMock(): jest.Mocked<CollectorFetchContext<false>> {
+  const collectorFetchClientsMock: jest.Mocked<CollectorFetchContext<false>> = {
+    callCluster: elasticsearchServiceMock.createLegacyClusterClient().callAsInternalUser,
+    esClient: elasticsearchServiceMock.createClusterClient().asInternalUser,
+    soClient: savedObjectsRepositoryMock.create(),
+  };
+  return collectorFetchClientsMock;
+}
+
+export function createCollectorFetchContextWithKibanaMock(): jest.Mocked<
+  CollectorFetchContext<true>
+> {
+  const collectorFetchClientsMock: jest.Mocked<CollectorFetchContext<true>> = {
     callCluster: elasticsearchServiceMock.createLegacyClusterClient().callAsInternalUser,
     esClient: elasticsearchServiceMock.createClusterClient().asInternalUser,
     soClient: savedObjectsRepositoryMock.create(),
+    kibanaRequest: httpServerMock.createKibanaRequest(),
   };
   return collectorFetchClientsMock;
 }
diff --git a/src/plugins/vis_type_vega/server/usage_collector/register_vega_collector.test.ts b/src/plugins/vis_type_vega/server/usage_collector/register_vega_collector.test.ts
index e092fc8acfd71..fbef55df39719 100644
--- a/src/plugins/vis_type_vega/server/usage_collector/register_vega_collector.test.ts
+++ b/src/plugins/vis_type_vega/server/usage_collector/register_vega_collector.test.ts
@@ -59,9 +59,9 @@ describe('registerVegaUsageCollector', () => {
   it('makeUsageCollector config.fetch calls getStats', async () => {
     const mockCollectorSet = createUsageCollectionSetupMock();
     registerVegaUsageCollector(mockCollectorSet, mockConfig, mockDeps);
-    const usageCollectorConfig = mockCollectorSet.makeUsageCollector.mock.calls[0][0];
+    const usageCollector = mockCollectorSet.makeUsageCollector.mock.results[0].value;
     const mockedCollectorFetchContext = createCollectorFetchContextMock();
-    const fetchResult = await usageCollectorConfig.fetch(mockedCollectorFetchContext);
+    const fetchResult = await usageCollector.fetch(mockedCollectorFetchContext);
     expect(mockGetStats).toBeCalledTimes(1);
     expect(mockGetStats).toBeCalledWith(
       mockedCollectorFetchContext.callCluster,
diff --git a/src/plugins/visualizations/server/usage_collector/register_visualizations_collector.test.ts b/src/plugins/visualizations/server/usage_collector/register_visualizations_collector.test.ts
index 7789e3de13e5a..380a86e15aa51 100644
--- a/src/plugins/visualizations/server/usage_collector/register_visualizations_collector.test.ts
+++ b/src/plugins/visualizations/server/usage_collector/register_visualizations_collector.test.ts
@@ -58,9 +58,9 @@ describe('registerVisualizationsCollector', () => {
   it('makeUsageCollector config.fetch calls getStats', async () => {
     const mockCollectorSet = createUsageCollectionSetupMock();
     registerVisualizationsCollector(mockCollectorSet, mockConfig);
-    const usageCollectorConfig = mockCollectorSet.makeUsageCollector.mock.calls[0][0];
+    const usageCollector = mockCollectorSet.makeUsageCollector.mock.results[0].value;
     const mockCollectorFetchContext = createCollectorFetchContextMock();
-    const fetchResult = await usageCollectorConfig.fetch(mockCollectorFetchContext);
+    const fetchResult = await usageCollector.fetch(mockCollectorFetchContext);
     expect(mockGetStats).toBeCalledTimes(1);
     expect(mockGetStats).toBeCalledWith(mockCollectorFetchContext.callCluster, mockIndex);
     expect(fetchResult).toBe(mockStats);
diff --git a/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_settings_collector.ts b/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_settings_collector.ts
index 0dd5ce291f972..2b81f1078ad0a 100644
--- a/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_settings_collector.ts
+++ b/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_settings_collector.ts
@@ -44,15 +44,23 @@ interface EmailSettingData {
   xpack: { default_admin_email: string | null };
 }
 
-export interface KibanaSettingsCollector extends Collector<EmailSettingData | undefined> {
+export interface KibanaSettingsCollectorExtraOptions {
   getEmailValueStructure(email: string | null): EmailSettingData;
 }
 
+export type KibanaSettingsCollector = Collector<EmailSettingData | undefined> &
+  KibanaSettingsCollectorExtraOptions;
+
 export function getSettingsCollector(
   usageCollection: UsageCollectionSetup,
   config: MonitoringConfig
 ) {
-  return usageCollection.makeStatsCollector({
+  return usageCollection.makeStatsCollector<
+    EmailSettingData | undefined,
+    unknown,
+    false,
+    KibanaSettingsCollectorExtraOptions
+  >({
     type: KIBANA_SETTINGS_TYPE,
     isReady: () => true,
     schema: {
@@ -60,7 +68,7 @@ export function getSettingsCollector(
         default_admin_email: { type: 'text' },
       },
     },
-    async fetch(this: KibanaSettingsCollector) {
+    async fetch() {
       let kibanaSettingsData;
       const defaultAdminEmail = await checkForEmailValue(config);
 
diff --git a/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_usage_collector.test.ts b/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_usage_collector.test.ts
index 2f63a878b0cde..8a2283420ac95 100644
--- a/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_usage_collector.test.ts
+++ b/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_usage_collector.test.ts
@@ -6,6 +6,7 @@
 
 import { getMonitoringUsageCollector } from './get_usage_collector';
 import { fetchClusters } from '../../lib/alerts/fetch_clusters';
+import { elasticsearchServiceMock } from '../../../../../../src/core/server/mocks';
 
 jest.mock('../../lib/alerts/fetch_clusters', () => ({
   fetchClusters: jest.fn().mockImplementation(() => {
@@ -57,7 +58,7 @@ jest.mock('./lib/fetch_license_type', () => ({
 }));
 
 describe('getMonitoringUsageCollector', () => {
-  const callCluster = jest.fn();
+  const esClient = elasticsearchServiceMock.createLegacyClusterClient();
   const config: any = {
     ui: {
       ccs: {
@@ -70,7 +71,7 @@ describe('getMonitoringUsageCollector', () => {
     const usageCollection: any = {
       makeUsageCollector: jest.fn(),
     };
-    await getMonitoringUsageCollector(usageCollection, config, callCluster);
+    await getMonitoringUsageCollector(usageCollection, config, esClient);
 
     const mock = (usageCollection.makeUsageCollector as jest.Mock).mock;
 
@@ -120,11 +121,11 @@ describe('getMonitoringUsageCollector', () => {
       makeUsageCollector: jest.fn(),
     };
 
-    await getMonitoringUsageCollector(usageCollection, config, callCluster);
+    await getMonitoringUsageCollector(usageCollection, config, esClient);
     const mock = (usageCollection.makeUsageCollector as jest.Mock).mock;
     const args = mock.calls[0];
 
-    const result = await args[0].fetch();
+    const result = await args[0].fetch({});
     expect(result).toStrictEqual({
       hasMonitoringData: true,
       clusters: [
@@ -147,7 +148,7 @@ describe('getMonitoringUsageCollector', () => {
       makeUsageCollector: jest.fn(),
     };
 
-    await getMonitoringUsageCollector(usageCollection, config, callCluster);
+    await getMonitoringUsageCollector(usageCollection, config, esClient);
     const mock = (usageCollection.makeUsageCollector as jest.Mock).mock;
     const args = mock.calls[0];
 
@@ -155,7 +156,27 @@ describe('getMonitoringUsageCollector', () => {
       return [];
     });
 
-    const result = await args[0].fetch();
+    const result = await args[0].fetch({});
+    expect(result).toStrictEqual({
+      hasMonitoringData: false,
+      clusters: [],
+    });
+  });
+
+  it('should handle scoped data', async () => {
+    const usageCollection: any = {
+      makeUsageCollector: jest.fn(),
+    };
+
+    await getMonitoringUsageCollector(usageCollection, config, esClient);
+    const mock = (usageCollection.makeUsageCollector as jest.Mock).mock;
+    const args = mock.calls[0];
+
+    (fetchClusters as jest.Mock).mockImplementation(() => {
+      return [];
+    });
+
+    const result = await args[0].fetch({ kibanaRequest: {} });
     expect(result).toStrictEqual({
       hasMonitoringData: false,
       clusters: [],
diff --git a/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_usage_collector.ts b/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_usage_collector.ts
index 278a6c163c0ad..038042f109817 100644
--- a/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_usage_collector.ts
+++ b/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/get_usage_collector.ts
@@ -5,7 +5,7 @@
  */
 
 import { UsageCollectionSetup } from 'src/plugins/usage_collection/server';
-import { LegacyAPICaller } from 'src/core/server';
+import { ILegacyClusterClient } from 'src/core/server';
 import { MonitoringConfig } from '../../config';
 import { fetchAvailableCcs } from '../../lib/alerts/fetch_available_ccs';
 import { getStackProductsUsage } from './lib/get_stack_products_usage';
@@ -18,9 +18,9 @@ import { fetchClusters } from '../../lib/alerts/fetch_clusters';
 export function getMonitoringUsageCollector(
   usageCollection: UsageCollectionSetup,
   config: MonitoringConfig,
-  callCluster: LegacyAPICaller
+  legacyEsClient: ILegacyClusterClient
 ) {
-  return usageCollection.makeUsageCollector<MonitoringUsage>({
+  return usageCollection.makeUsageCollector<MonitoringUsage, unknown, true>({
     type: 'monitoring',
     isReady: () => true,
     schema: {
@@ -97,7 +97,13 @@ export function getMonitoringUsageCollector(
         },
       },
     },
-    fetch: async () => {
+    extendFetchContext: {
+      kibanaRequest: true,
+    },
+    fetch: async ({ kibanaRequest }) => {
+      const callCluster = kibanaRequest
+        ? legacyEsClient.asScoped(kibanaRequest).callAsCurrentUser
+        : legacyEsClient.callAsInternalUser;
       const usageClusters: MonitoringClusterStackProductUsage[] = [];
       const availableCcs = config.ui.ccs.enabled ? await fetchAvailableCcs(callCluster) : [];
       const elasticsearchIndex = getCcsIndexPattern(INDEX_PATTERN_ELASTICSEARCH, availableCcs);
diff --git a/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/index.ts b/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/index.ts
index 47ad78b29962c..25e243656898c 100644
--- a/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/index.ts
+++ b/x-pack/plugins/monitoring/server/kibana_monitoring/collectors/index.ts
@@ -4,7 +4,7 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { LegacyAPICaller } from 'src/core/server';
+import { ILegacyClusterClient } from 'src/core/server';
 import { UsageCollectionSetup } from 'src/plugins/usage_collection/server';
 import { getSettingsCollector } from './get_settings_collector';
 import { getMonitoringUsageCollector } from './get_usage_collector';
@@ -15,10 +15,10 @@ export { KibanaSettingsCollector } from './get_settings_collector';
 export function registerCollectors(
   usageCollection: UsageCollectionSetup,
   config: MonitoringConfig,
-  callCluster: LegacyAPICaller
+  legacyEsClient: ILegacyClusterClient
 ) {
   usageCollection.registerCollector(getSettingsCollector(usageCollection, config));
   usageCollection.registerCollector(
-    getMonitoringUsageCollector(usageCollection, config, callCluster)
+    getMonitoringUsageCollector(usageCollection, config, legacyEsClient)
   );
 }
diff --git a/x-pack/plugins/monitoring/server/plugin.ts b/x-pack/plugins/monitoring/server/plugin.ts
index d3e5028d72fcc..41b501d88af99 100644
--- a/x-pack/plugins/monitoring/server/plugin.ts
+++ b/x-pack/plugins/monitoring/server/plugin.ts
@@ -173,7 +173,7 @@ export class Plugin {
         },
       });
 
-      registerCollectors(plugins.usageCollection, config, cluster.callAsInternalUser);
+      registerCollectors(plugins.usageCollection, config, cluster);
     }
 
     // Always create the bulk uploader
diff --git a/x-pack/plugins/monitoring/server/telemetry_collection/get_all_stats.test.ts b/x-pack/plugins/monitoring/server/telemetry_collection/get_all_stats.test.ts
index 099f6915611cb..a119686afe663 100644
--- a/x-pack/plugins/monitoring/server/telemetry_collection/get_all_stats.test.ts
+++ b/x-pack/plugins/monitoring/server/telemetry_collection/get_all_stats.test.ts
@@ -180,6 +180,7 @@ describe('get_all_stats', () => {
             esClient: esClient as any,
             soClient: soClient as any,
             usageCollection: {} as any,
+            kibanaRequest: undefined,
             timestamp,
           },
           {
@@ -206,6 +207,7 @@ describe('get_all_stats', () => {
             esClient: esClient as any,
             soClient: soClient as any,
             usageCollection: {} as any,
+            kibanaRequest: undefined,
             timestamp,
           },
           {
diff --git a/x-pack/plugins/monitoring/server/telemetry_collection/get_cluster_uuids.test.ts b/x-pack/plugins/monitoring/server/telemetry_collection/get_cluster_uuids.test.ts
index 0acdb9968bc03..b296ff090aedd 100644
--- a/x-pack/plugins/monitoring/server/telemetry_collection/get_cluster_uuids.test.ts
+++ b/x-pack/plugins/monitoring/server/telemetry_collection/get_cluster_uuids.test.ts
@@ -13,6 +13,7 @@ import {
 } from './get_cluster_uuids';
 
 describe('get_cluster_uuids', () => {
+  const kibanaRequest = undefined;
   const callCluster = sinon.stub();
   const esClient = elasticsearchServiceMock.createClusterClient().asInternalUser;
   const soClient = savedObjectsRepositoryMock.create();
@@ -33,7 +34,7 @@ describe('get_cluster_uuids', () => {
       callCluster.withArgs('search').returns(Promise.resolve(response));
       expect(
         await getClusterUuids(
-          { callCluster, esClient, soClient, timestamp, usageCollection: {} as any },
+          { callCluster, esClient, soClient, timestamp, kibanaRequest, usageCollection: {} as any },
           {
             maxBucketSize: 1,
           } as any
@@ -47,7 +48,7 @@ describe('get_cluster_uuids', () => {
       callCluster.returns(Promise.resolve(response));
       expect(
         await fetchClusterUuids(
-          { callCluster, esClient, soClient, timestamp, usageCollection: {} as any },
+          { callCluster, esClient, soClient, timestamp, kibanaRequest, usageCollection: {} as any },
           {
             maxBucketSize: 1,
           } as any
diff --git a/x-pack/plugins/reporting/server/usage/reporting_usage_collector.test.ts b/x-pack/plugins/reporting/server/usage/reporting_usage_collector.test.ts
index 4cecc2e24867f..fff18353c58b0 100644
--- a/x-pack/plugins/reporting/server/usage/reporting_usage_collector.test.ts
+++ b/x-pack/plugins/reporting/server/usage/reporting_usage_collector.test.ts
@@ -74,7 +74,7 @@ describe('license checks', () => {
     let usageStats: any;
     beforeAll(async () => {
       const plugins = getPluginsMock({ license: 'basic' });
-      const { fetch } = getReportingUsageCollector(
+      const collector = getReportingUsageCollector(
         mockCore,
         plugins.usageCollection,
         getLicenseMock('basic'),
@@ -83,7 +83,7 @@ describe('license checks', () => {
           return Promise.resolve(true);
         }
       );
-      usageStats = await fetch(getMockFetchClients(getResponseMock()));
+      usageStats = await collector.fetch(getMockFetchClients(getResponseMock()));
     });
 
     test('sets enables to true', async () => {
@@ -103,7 +103,7 @@ describe('license checks', () => {
     let usageStats: any;
     beforeAll(async () => {
       const plugins = getPluginsMock({ license: 'none' });
-      const { fetch } = getReportingUsageCollector(
+      const collector = getReportingUsageCollector(
         mockCore,
         plugins.usageCollection,
         getLicenseMock('none'),
@@ -112,7 +112,7 @@ describe('license checks', () => {
           return Promise.resolve(true);
         }
       );
-      usageStats = await fetch(getMockFetchClients(getResponseMock()));
+      usageStats = await collector.fetch(getMockFetchClients(getResponseMock()));
     });
 
     test('sets enables to true', async () => {
@@ -132,7 +132,7 @@ describe('license checks', () => {
     let usageStats: any;
     beforeAll(async () => {
       const plugins = getPluginsMock({ license: 'platinum' });
-      const { fetch } = getReportingUsageCollector(
+      const collector = getReportingUsageCollector(
         mockCore,
         plugins.usageCollection,
         getLicenseMock('platinum'),
@@ -141,7 +141,7 @@ describe('license checks', () => {
           return Promise.resolve(true);
         }
       );
-      usageStats = await fetch(getMockFetchClients(getResponseMock()));
+      usageStats = await collector.fetch(getMockFetchClients(getResponseMock()));
     });
 
     test('sets enables to true', async () => {
@@ -161,7 +161,7 @@ describe('license checks', () => {
     let usageStats: any;
     beforeAll(async () => {
       const plugins = getPluginsMock({ license: 'basic' });
-      const { fetch } = getReportingUsageCollector(
+      const collector = getReportingUsageCollector(
         mockCore,
         plugins.usageCollection,
         getLicenseMock('basic'),
@@ -170,7 +170,7 @@ describe('license checks', () => {
           return Promise.resolve(true);
         }
       );
-      usageStats = await fetch(getMockFetchClients({}));
+      usageStats = await collector.fetch(getMockFetchClients({}));
     });
 
     test('sets enables to true', async () => {
@@ -193,7 +193,7 @@ describe('data modeling', () => {
   });
   test('with normal looking usage data', async () => {
     const plugins = getPluginsMock();
-    const { fetch } = getReportingUsageCollector(
+    const collector = getReportingUsageCollector(
       mockCore,
       plugins.usageCollection,
       getLicenseMock(),
@@ -237,13 +237,13 @@ describe('data modeling', () => {
         },
       } as SearchResponse) // prettier-ignore
     );
-    const usageStats = await fetch(collectorFetchContext);
+    const usageStats = await collector.fetch(collectorFetchContext);
     expect(usageStats).toMatchSnapshot();
   });
 
   test('with sparse data', async () => {
     const plugins = getPluginsMock();
-    const { fetch } = getReportingUsageCollector(
+    const collector = getReportingUsageCollector(
       mockCore,
       plugins.usageCollection,
       getLicenseMock(),
@@ -287,13 +287,13 @@ describe('data modeling', () => {
         },
       } as SearchResponse) // prettier-ignore
     );
-    const usageStats = await fetch(collectorFetchContext);
+    const usageStats = await collector.fetch(collectorFetchContext);
     expect(usageStats).toMatchSnapshot();
   });
 
   test('with empty data', async () => {
     const plugins = getPluginsMock();
-    const { fetch } = getReportingUsageCollector(
+    const collector = getReportingUsageCollector(
       mockCore,
       plugins.usageCollection,
       getLicenseMock(),
@@ -337,7 +337,7 @@ describe('data modeling', () => {
       },
     } as SearchResponse) // prettier-ignore
     );
-    const usageStats = await fetch(collectorFetchContext);
+    const usageStats = await collector.fetch(collectorFetchContext);
     expect(usageStats).toMatchSnapshot();
   });
 
diff --git a/x-pack/plugins/spaces/server/usage_collection/spaces_usage_collector.test.ts b/x-pack/plugins/spaces/server/usage_collection/spaces_usage_collector.test.ts
index 864c91c583e82..1a377d2f801a0 100644
--- a/x-pack/plugins/spaces/server/usage_collection/spaces_usage_collector.test.ts
+++ b/x-pack/plugins/spaces/server/usage_collection/spaces_usage_collector.test.ts
@@ -44,7 +44,7 @@ function setup({
   return {
     licensing,
     features: featuresSetup,
-    usageCollecion: {
+    usageCollection: {
       makeUsageCollector: (options: any) => new MockUsageCollector(options),
     },
   };
@@ -77,23 +77,23 @@ const getMockFetchContext = (mockedCallCluster: jest.Mock) => {
 
 describe('error handling', () => {
   it('handles a 404 when searching for space usage', async () => {
-    const { features, licensing, usageCollecion } = setup({
+    const { features, licensing, usageCollection } = setup({
       license: { isAvailable: true, type: 'basic' },
     });
-    const { fetch: getSpacesUsage } = getSpacesUsageCollector(usageCollecion as any, {
+    const collector = getSpacesUsageCollector(usageCollection as any, {
       kibanaIndexConfig$: Rx.of({ kibana: { index: '.kibana' } }),
       features,
       licensing,
     });
 
-    await getSpacesUsage(getMockFetchContext(jest.fn().mockRejectedValue({ status: 404 })));
+    await collector.fetch(getMockFetchContext(jest.fn().mockRejectedValue({ status: 404 })));
   });
 
   it('throws error for a non-404', async () => {
-    const { features, licensing, usageCollecion } = setup({
+    const { features, licensing, usageCollection } = setup({
       license: { isAvailable: true, type: 'basic' },
     });
-    const { fetch: getSpacesUsage } = getSpacesUsageCollector(usageCollecion as any, {
+    const collector = getSpacesUsageCollector(usageCollection as any, {
       kibanaIndexConfig$: Rx.of({ kibana: { index: '.kibana' } }),
       features,
       licensing,
@@ -103,7 +103,7 @@ describe('error handling', () => {
     for (const statusCode of statusCodes) {
       const error = { status: statusCode };
       await expect(
-        getSpacesUsage(getMockFetchContext(jest.fn().mockRejectedValue(error)))
+        collector.fetch(getMockFetchContext(jest.fn().mockRejectedValue(error)))
       ).rejects.toBe(error);
     }
   });
@@ -112,15 +112,15 @@ describe('error handling', () => {
 describe('with a basic license', () => {
   let usageStats: UsageStats;
   beforeAll(async () => {
-    const { features, licensing, usageCollecion } = setup({
+    const { features, licensing, usageCollection } = setup({
       license: { isAvailable: true, type: 'basic' },
     });
-    const { fetch: getSpacesUsage } = getSpacesUsageCollector(usageCollecion as any, {
+    const collector = getSpacesUsageCollector(usageCollection as any, {
       kibanaIndexConfig$: pluginInitializerContextConfigMock({}).legacy.globalConfig$,
       features,
       licensing,
     });
-    usageStats = await getSpacesUsage(getMockFetchContext(defaultCallClusterMock));
+    usageStats = await collector.fetch(getMockFetchContext(defaultCallClusterMock));
 
     expect(defaultCallClusterMock).toHaveBeenCalledWith('search', {
       body: {
@@ -162,13 +162,13 @@ describe('with a basic license', () => {
 describe('with no license', () => {
   let usageStats: UsageStats;
   beforeAll(async () => {
-    const { features, licensing, usageCollecion } = setup({ license: { isAvailable: false } });
-    const { fetch: getSpacesUsage } = getSpacesUsageCollector(usageCollecion as any, {
+    const { features, licensing, usageCollection } = setup({ license: { isAvailable: false } });
+    const collector = getSpacesUsageCollector(usageCollection as any, {
       kibanaIndexConfig$: pluginInitializerContextConfigMock({}).legacy.globalConfig$,
       features,
       licensing,
     });
-    usageStats = await getSpacesUsage(getMockFetchContext(defaultCallClusterMock));
+    usageStats = await collector.fetch(getMockFetchContext(defaultCallClusterMock));
   });
 
   test('sets enabled to false', () => {
@@ -191,15 +191,15 @@ describe('with no license', () => {
 describe('with platinum license', () => {
   let usageStats: UsageStats;
   beforeAll(async () => {
-    const { features, licensing, usageCollecion } = setup({
+    const { features, licensing, usageCollection } = setup({
       license: { isAvailable: true, type: 'platinum' },
     });
-    const { fetch: getSpacesUsage } = getSpacesUsageCollector(usageCollecion as any, {
+    const collector = getSpacesUsageCollector(usageCollection as any, {
       kibanaIndexConfig$: pluginInitializerContextConfigMock({}).legacy.globalConfig$,
       features,
       licensing,
     });
-    usageStats = await getSpacesUsage(getMockFetchContext(defaultCallClusterMock));
+    usageStats = await collector.fetch(getMockFetchContext(defaultCallClusterMock));
   });
 
   test('sets enabled to true', () => {