Added tests and logger

x-pack/plugins/data_enhanced/server/collectors/fetch.test.ts

@@ -0,0 +1,94 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import {
+  SharedGlobalConfig,
+  ElasticsearchClient,
+  SavedObjectsErrorHelpers,
+  Logger,
+} from '../../../../../src/core/server';
+import { BehaviorSubject } from 'rxjs';
+import { fetchProvider } from './fetch';
+import { elasticsearchServiceMock } from '../../../../../src/core/server/mocks';
+
+describe('fetchProvider', () => {
+  let fetchFn: any;
+  let esClient: jest.Mocked<ElasticsearchClient>;
+  let mockLogger: Logger;
+
+  beforeEach(async () => {
+    const config$ = new BehaviorSubject<SharedGlobalConfig>({
+      kibana: {
+        index: '123',
+      },
+    } as any);
+    mockLogger = {
+      warn: jest.fn(),
+      debug: jest.fn(),
+    } as any;
+    esClient = elasticsearchServiceMock.createElasticsearchClient();
+    fetchFn = fetchProvider(config$, mockLogger);
+  });
+
+  test('returns when ES returns no results', async () => {
+    esClient.search.mockResolvedValue({
+      statusCode: 200,
+      body: {
+        aggregations: {
+          persisted: {
+            buckets: [],
+          },
+        },
+      },
+    } as any);
+
+    const collRes = await fetchFn({ esClient });
+    expect(collRes.transientCount).toBe(0);
+    expect(collRes.persistedCount).toBe(0);
+    expect(collRes.totalCount).toBe(0);
+    expect(mockLogger.warn).not.toBeCalled();
+  });
+
+  test('returns when ES throws an error', async () => {
+    esClient.search.mockRejectedValue(
+      SavedObjectsErrorHelpers.createTooManyRequestsError('a', 'b')
+    );
+
+    const collRes = await fetchFn({ esClient });
+    expect(collRes.transientCount).toBe(0);
+    expect(collRes.persistedCount).toBe(0);
+    expect(collRes.totalCount).toBe(0);
+    expect(mockLogger.warn).toBeCalledTimes(1);
+  });
+
+  test('returns when ES returns full buckets', async () => {
+    esClient.search.mockResolvedValue({
+      statusCode: 200,
+      body: {
+        aggregations: {
+          persisted: {
+            buckets: [
+              {
+                key_as_string: 'true',
+                doc_count: 10,
+              },
+              {
+                key_as_string: 'false',
+                doc_count: 7,
+              },
+            ],
+          },
+        },
+      },
+    } as any);
+
+    const collRes = await fetchFn({ esClient });
+    expect(collRes.transientCount).toBe(7);
+    expect(collRes.persistedCount).toBe(10);
+    expect(collRes.totalCount).toBe(17);
+  });
+});

x-pack/plugins/data_enhanced/server/collectors/fetch.ts

@@ -8,7 +8,7 @@
 import { Observable } from 'rxjs';
 import { first } from 'rxjs/operators';
 import { SearchResponse } from 'elasticsearch';
-import { SharedGlobalConfig } from 'kibana/server';
+import { SharedGlobalConfig, Logger } from 'kibana/server';
 import { CollectorFetchContext } from '../../../../../src/plugins/usage_collection/server';
 import { SEARCH_SESSION_TYPE } from '../../common';
 import { ReportedUsage } from './register';
@@ -18,36 +18,42 @@ interface SessionPersistedTermsBucket {
   doc_count: number;
 }
 
-export function fetchProvider(config$: Observable<SharedGlobalConfig>) {
+export function fetchProvider(config$: Observable<SharedGlobalConfig>, logger: Logger) {
   return async ({ esClient }: CollectorFetchContext): Promise<ReportedUsage> => {
-    const config = await config$.pipe(first()).toPromise();
-    const { body: esResponse } = await esClient.search<SearchResponse<unknown>>({
-      index: config.kibana.index,
-      body: {
-        size: 0,
-        aggs: {
-          persisted: {
-            terms: {
-              field: `${SEARCH_SESSION_TYPE}.persisted`,
+    try {
+      const config = await config$.pipe(first()).toPromise();
+      const { body: esResponse } = await esClient.search<SearchResponse<unknown>>({
+        index: config.kibana.index,
+        body: {
+          size: 0,
+          aggs: {
+            persisted: {
+              terms: {
+                field: `${SEARCH_SESSION_TYPE}.persisted`,
+              },
             },
           },
         },
-      },
-    });
+      });
 
-    const { buckets } = esResponse.aggregations.persisted;
-    if (!buckets.length) {
+      const { buckets } = esResponse.aggregations.persisted;
+      if (!buckets.length) {
+        return { transientCount: 0, persistedCount: 0, totalCount: 0 };
+      }
+
+      const { transientCount = 0, persistedCount = 0 } = buckets.reduce(
+        (usage: Partial<ReportedUsage>, bucket: SessionPersistedTermsBucket) => {
+          const key = bucket.key_as_string === 'false' ? 'transientCount' : 'persistedCount';
+          return { ...usage, [key]: bucket.doc_count };
+        },
+        {}
+      );
+      const totalCount = transientCount + persistedCount;
+      logger.debug(`fetchProvider | ${persistedCount} persisted | ${transientCount} transient`);
+      return { transientCount, persistedCount, totalCount };
+    } catch (e) {
+      logger.warn(`fetchProvider | error | ${e.message}`);
       return { transientCount: 0, persistedCount: 0, totalCount: 0 };
     }
-
-    const { transientCount = 0, persistedCount = 0 } = buckets.reduce(
-      (usage: Partial<ReportedUsage>, bucket: SessionPersistedTermsBucket) => {
-        const key = bucket.key_as_string === 'false' ? 'transientCount' : 'persistedCount';
-        return { ...usage, [key]: bucket.doc_count };
-      },
-      {}
-    );
-    const totalCount = transientCount + persistedCount;
-    return { transientCount, persistedCount, totalCount };
   };
 }

x-pack/plugins/data_enhanced/server/collectors/register.ts

@@ -5,7 +5,7 @@
  * 2.0.
  */
 
-import { PluginInitializerContext } from 'kibana/server';
+import { PluginInitializerContext, Logger } from 'kibana/server';
 import { UsageCollectionSetup } from 'src/plugins/usage_collection/server';
 import { fetchProvider } from './fetch';
 
@@ -17,13 +17,14 @@ export interface ReportedUsage {
 
 export async function registerUsageCollector(
   usageCollection: UsageCollectionSetup,
-  context: PluginInitializerContext
+  context: PluginInitializerContext,
+  logger: Logger
 ) {
   try {
     const collector = usageCollection.makeUsageCollector<ReportedUsage>({
       type: 'search-session',
       isReady: () => true,
-      fetch: fetchProvider(context.config.legacy.globalConfig$),
+      fetch: fetchProvider(context.config.legacy.globalConfig$, logger),
       schema: {
         transientCount: { type: 'long' },
         persistedCount: { type: 'long' },

x-pack/plugins/data_enhanced/server/plugin.ts

@@ -88,7 +88,7 @@ export class EnhancedDataServerPlugin
     });
 
     if (deps.usageCollection) {
-      registerUsageCollector(deps.usageCollection, this.initializerContext);
+      registerUsageCollector(deps.usageCollection, this.initializerContext, this.logger);
     }
   }