[Monitoring/Telemetry] Force collectors to indicate when they are ready (#36153)

* Initial code to force collectors to indicate when they are ready

* Add and fix tests

* Remove debug

* Add ready check in api call

* Fix prettier complaints

* Return 503 if not all collectors are ready

* PR feedback

* Add retry logic for  usage collection in the reporting tests

* Fix incorrect boomify usage

* Fix more issues with the tests

* Just add debug I guess

* More debug

* Try and handle this exception

* Try and make the tests more defensive and remove console logs

* Retry logic here too

* Debug for the reporting tests failure

* I don't like this, but lets see if it works

* Move the retry logic into the collector set directly

* Add support for this new collector

* Localize this

* This shouldn't be static on the class, but rather static for the entire runtime

Author: chrisronline

src/legacy/core_plugins/kibana/server/lib/kql_usage_collector/make_kql_usage_collector.js

@@ -25,6 +25,7 @@ export function makeKQLUsageCollector(server) {
   const kqlUsageCollector = server.usage.collectorSet.makeUsageCollector({
     type: 'kql',
     fetch,
+    isReady: () => true,
   });
 
   server.usage.collectorSet.register(kqlUsageCollector);

src/legacy/core_plugins/ui_metric/server/usage/collector.ts

@@ -52,6 +52,7 @@ export function registerUiMetricUsageCollector(server: any) {
 
       return uiMetricsByAppName;
     },
+    isReady: () => true,
   });
 
   server.usage.collectorSet.register(collector);

src/legacy/server/config/schema.js

@@ -175,6 +175,10 @@ export default () => Joi.object({
     pollInterval: Joi.number().default(1500),
   }).default(),
 
+  stats: Joi.object({
+    maximumWaitTimeForAllCollectorsInS: Joi.number().default(60)
+  }).default(),
+
   optimize: Joi.object({
     enabled: Joi.boolean().default(true),
     bundleFilter: Joi.string().default('!tests'),

src/legacy/server/sample_data/usage/collector.ts

@@ -36,6 +36,7 @@ export function makeSampleDataUsageCollector(server: KbnServer) {
     server.usage.collectorSet.makeUsageCollector({
       type: 'sample-data',
       fetch: fetchProvider(index),
+      isReady: () => true,
     })
   );
 }

src/legacy/server/status/collectors/get_ops_stats_collector.js

@@ -45,6 +45,7 @@ export function getOpsStatsCollector(server, kbnServer) {
         ...kbnServer.metrics // latest metrics captured from the ops event listener in src/legacy/server/status/index
       };
     },
+    isReady: () => true,
     ignoreForInternalUploader: true, // Ignore this one from internal uploader. A different stats collector is used there.
   });
 }

src/legacy/server/status/routes/api/register_stats.js

@@ -18,10 +18,15 @@
  */
 
 import Joi from 'joi';
-import { boomify } from 'boom';
+import boom from 'boom';
+import { i18n }  from '@kbn/i18n';
 import { wrapAuthConfig } from '../../wrap_auth_config';
 import { KIBANA_STATS_TYPE } from '../../constants';
 
+const STATS_NOT_READY_MESSAGE = i18n.translate('server.stats.notReadyMessage', {
+  defaultMessage: 'Stats are not ready yet. Please try again later.',
+});
+
 /*
  * API for Kibana meta info and accumulated operations stats
  * Including ?extended in the query string fetches Elasticsearch cluster_uuid and server.usage.collectorSet data
@@ -69,6 +74,11 @@ export function registerStatsApi(kbnServer, server, config) {
         if (isExtended) {
           const { callWithRequest } = req.server.plugins.elasticsearch.getCluster('admin');
           const callCluster = (...args) => callWithRequest(req, ...args);
+          const collectorsReady = await collectorSet.areAllCollectorsReady();
+
+          if (shouldGetUsage && !collectorsReady) {
+            return boom.serverUnavailable(STATS_NOT_READY_MESSAGE);
+          }
 
           const usagePromise = shouldGetUsage ? getUsage(callCluster) : Promise.resolve();
           try {
@@ -77,7 +87,6 @@ export function registerStatsApi(kbnServer, server, config) {
               getClusterUuid(callCluster),
             ]);
 
-
             let modifiedUsage = usage;
             if (isLegacy) {
               // In an effort to make telemetry more easily augmented, we need to ensure
@@ -123,14 +132,17 @@ export function registerStatsApi(kbnServer, server, config) {
               });
             }
           } catch (e) {
-            throw boomify(e);
+            throw boom.boomify(e);
           }
         }
 
         /* kibana_stats gets singled out from the collector set as it is used
          * for health-checking Kibana and fetch does not rely on fetching data
          * from ES */
         const kibanaStatsCollector = collectorSet.getCollectorByType(KIBANA_STATS_TYPE);
+        if (!await kibanaStatsCollector.isReady()) {
+          return boom.serverUnavailable(STATS_NOT_READY_MESSAGE);
+        }
         let kibanaStats = await kibanaStatsCollector.fetch();
         kibanaStats = collectorSet.toApiFieldNames(kibanaStats);
 

src/legacy/server/usage/classes/collector.js

@@ -28,7 +28,7 @@ export class Collector {
    * @param {Function} options.formatForBulkUpload - optional
    * @param {Function} options.rest - optional other properties
    */
-  constructor(server, { type, init, fetch, formatForBulkUpload = null, ...options } = {}) {
+  constructor(server, { type, init, fetch, formatForBulkUpload = null, isReady = null, ...options } = {}) {
     if (type === undefined) {
       throw new Error('Collector must be instantiated with a options.type string property');
     }
@@ -49,6 +49,9 @@ export class Collector {
 
     const defaultFormatterForBulkUpload = result => ({ type, payload: result });
     this._formatForBulkUpload = formatForBulkUpload || defaultFormatterForBulkUpload;
+    if (typeof isReady === 'function') {
+      this.isReady = isReady;
+    }
   }
 
   /*
@@ -69,4 +72,8 @@ export class Collector {
   formatForBulkUpload(result) {
     return this._formatForBulkUpload(result);
   }
+
+  isReady() {
+    throw `isReady() must be implemented in ${this.type} collector`;
+  }
 }

src/legacy/server/usage/classes/collector_set.js

@@ -23,18 +23,19 @@ import { getCollectorLogger } from '../lib';
 import { Collector } from './collector';
 import { UsageCollector } from './usage_collector';
 
+let _waitingForAllCollectorsTimestamp = null;
+
 /*
  * A collector object has types registered into it with the register(type)
  * function. Each type that gets registered defines how to fetch its own data
  * and optionally, how to combine it into a unified payload for bulk upload.
  */
 export class CollectorSet {
-
   /*
    * @param {Object} server - server object
    * @param {Array} collectors to initialize, usually as a result of filtering another CollectorSet instance
    */
-  constructor(server, collectors = []) {
+  constructor(server, collectors = [], config = null) {
     this._log = getCollectorLogger(server);
     this._collectors = collectors;
 
@@ -44,7 +45,9 @@ export class CollectorSet {
      */
     this.makeStatsCollector = options => new Collector(server, options);
     this.makeUsageCollector = options => new UsageCollector(server, options);
-    this._makeCollectorSetFromArray = collectorsArray => new CollectorSet(server, collectorsArray);
+    this._makeCollectorSetFromArray = collectorsArray => new CollectorSet(server, collectorsArray, config);
+
+    this._maximumWaitTimeForAllCollectorsInS = config ? config.get('stats.maximumWaitTimeForAllCollectorsInS') : 60;
   }
 
   /*
@@ -73,6 +76,40 @@ export class CollectorSet {
     return x instanceof UsageCollector;
   }
 
+  async areAllCollectorsReady(collectorSet = this) {
+    if (!(collectorSet instanceof CollectorSet)) {
+      throw new Error(`areAllCollectorsReady method given bad collectorSet parameter: ` + typeof collectorSet);
+    }
+
+    const collectorTypesNotReady = [];
+    let allReady = true;
+    await collectorSet.asyncEach(async collector => {
+      if (!await collector.isReady()) {
+        allReady = false;
+        collectorTypesNotReady.push(collector.type);
+      }
+    });
+
+    if (!allReady && this._maximumWaitTimeForAllCollectorsInS >= 0) {
+      const nowTimestamp = +new Date();
+      _waitingForAllCollectorsTimestamp = _waitingForAllCollectorsTimestamp || nowTimestamp;
+      const timeWaitedInMS = nowTimestamp - _waitingForAllCollectorsTimestamp;
+      const timeLeftInMS = (this._maximumWaitTimeForAllCollectorsInS * 1000) - timeWaitedInMS;
+      if (timeLeftInMS <= 0) {
+        this._log.debug(`All collectors are not ready (waiting for ${collectorTypesNotReady.join(',')}) `
+        + `but we have waited the required `
+        + `${this._maximumWaitTimeForAllCollectorsInS}s and will return data from all collectors that are ready.`);
+        return true;
+      } else {
+        this._log.debug(`All collectors are not ready. Waiting for ${timeLeftInMS}ms longer.`);
+      }
+    } else {
+      _waitingForAllCollectorsTimestamp = null;
+    }
+
+    return allReady;
+  }
+
   /*
    * Call a bunch of fetch methods and then do them in bulk
    * @param {CollectorSet} collectorSet - a set of collectors to fetch. Default to all registered collectors
@@ -155,4 +192,14 @@ export class CollectorSet {
   map(mapFn) {
     return this._collectors.map(mapFn);
   }
+
+  some(someFn) {
+    return this._collectors.some(someFn);
+  }
+
+  async asyncEach(eachFn) {
+    for (const collector of this._collectors) {
+      await eachFn(collector);
+    }
+  }
 }

src/legacy/server/usage/index.js

@@ -19,8 +19,8 @@
 
 import { CollectorSet } from './classes';
 
-export function usageMixin(kbnServer, server) {
-  const collectorSet = new CollectorSet(server);
+export function usageMixin(kbnServer, server, config) {
+  const collectorSet = new CollectorSet(server, undefined, config);
 
   /*
    * expose the collector set object on the server

x-pack/plugins/apm/server/lib/apm_telemetry/make_apm_usage_collector.ts

@@ -37,7 +37,8 @@ export function makeApmUsageCollector(core: CoreSetupWithUsageCollector) {
       } catch (err) {
         return createApmTelementry();
       }
-    }
+    },
+    isReady: () => true
   });
   server.usage.collectorSet.register(apmUsageCollector);
 }