Add Server Side batching for UI Metric Collectors (#6721) (#6819)

* Add Server Side batching for UI Metric Collectors * Changeset file for PR #6721 created/updated This PR introduces Server Side batching of UI metric reports sent by clients to reduce the number of updates taking place at OpenSearch. Issues Resolved #6375 --------- (cherry picked from commit 972cf10) Signed-off-by: Suchit Sahoo <[email protected]> Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: opensearch-changeset-bot[bot] <154024398+opensearch-changeset-bot[bot]@users.noreply.github.com>
opensearch-project · May 23, 2024 · ecfaed5 · ecfaed5
1 parent 27627a4
commit ecfaed5
Show file tree

Hide file tree

Showing 10 changed files with 120 additions and 17 deletions.
diff --git a/changelogs/fragments/6721.yml b/changelogs/fragments/6721.yml
@@ -0,0 +1,2 @@
+feat:
+- Add Server Side Batching for UI Metric Colector ([#6721](https://github.com/opensearch-project/OpenSearch-Dashboards/pull/6721))
diff --git a/src/core/server/saved_objects/service/lib/repository.ts b/src/core/server/saved_objects/service/lib/repository.ts
@@ -1542,20 +1542,22 @@ export class SavedObjectsRepository {
   }
 
   /**
-   * Increases a counter field by one. Creates the document if one doesn't exist for the given id.
+   * Increases a counter field by incrementValue which by default is 1. Creates the document if one doesn't exist for the given id.
    *
    * @param {string} type
    * @param {string} id
    * @param {string} counterFieldName
    * @param {object} [options={}]
+   * @param {number} [incrementValue=1]
    * @property {object} [options.migrationVersion=undefined]
    * @returns {promise}
    */
   async incrementCounter(
     type: string,
     id: string,
     counterFieldName: string,
-    options: SavedObjectsIncrementCounterOptions = {}
+    options: SavedObjectsIncrementCounterOptions = {},
+    incrementValue: number = 1
   ): Promise<SavedObject> {
     if (typeof type !== 'string') {
       throw new Error('"type" argument must be a string');
@@ -1579,19 +1581,17 @@ export class SavedObjectsRepository {
     } else if (this._registry.isMultiNamespace(type)) {
       savedObjectNamespaces = await this.preflightGetNamespaces(type, id, namespace);
     }
-
     const migrated = this._migrator.migrateDocument({
       id,
       type,
       ...(savedObjectNamespace && { namespace: savedObjectNamespace }),
       ...(savedObjectNamespaces && { namespaces: savedObjectNamespaces }),
-      attributes: { [counterFieldName]: 1 },
+      attributes: { [counterFieldName]: incrementValue },
       migrationVersion,
       updated_at: time,
     });
 
     const raw = this._serializer.savedObjectToRaw(migrated as SavedObjectSanitizedDoc);
-
     const { body } = await this.client.update<SavedObjectsRawDocSource>({
       id: raw._id,
       index: this.getIndexForType(type),
@@ -1610,7 +1610,7 @@ export class SavedObjectsRepository {
             `,
           lang: 'painless',
           params: {
-            count: 1,
+            count: incrementValue,
             time,
             type,
             counterFieldName,
@@ -1619,7 +1619,6 @@ export class SavedObjectsRepository {
         upsert: raw._source,
       },
     });
-
     const { originId } = body.get?._source ?? {};
     return {
       id,

diff --git a/src/plugins/usage_collection/common/constants.ts b/src/plugins/usage_collection/common/constants.ts
@@ -30,3 +30,4 @@
 
 export const OPENSEARCH_DASHBOARDS_STATS_TYPE = 'opensearch_dashboards_stats';
 export const DEFAULT_MAXIMUM_WAIT_TIME_FOR_ALL_COLLECTORS_IN_S = 60;
+export const DEFAULT_BATCHING_INTERVAL_FOR_UI_METRIC_IN_S = 60;
diff --git a/src/plugins/usage_collection/server/config.ts b/src/plugins/usage_collection/server/config.ts
@@ -30,12 +30,19 @@
 
 import { schema, TypeOf } from '@osd/config-schema';
 import { PluginConfigDescriptor } from 'opensearch-dashboards/server';
-import { DEFAULT_MAXIMUM_WAIT_TIME_FOR_ALL_COLLECTORS_IN_S } from '../common/constants';
+import {
+  DEFAULT_BATCHING_INTERVAL_FOR_UI_METRIC_IN_S,
+  DEFAULT_MAXIMUM_WAIT_TIME_FOR_ALL_COLLECTORS_IN_S,
+} from '../common/constants';
 
 export const configSchema = schema.object({
   uiMetric: schema.object({
     enabled: schema.boolean({ defaultValue: false }),
     debug: schema.boolean({ defaultValue: schema.contextRef('dev') }),
+    batchingIntervalInS: schema.number({
+      min: 0,
+      defaultValue: DEFAULT_BATCHING_INTERVAL_FOR_UI_METRIC_IN_S,
+    }),
   }),
   maximumWaitTimeForAllCollectorsInS: schema.number({
     defaultValue: DEFAULT_MAXIMUM_WAIT_TIME_FOR_ALL_COLLECTORS_IN_S,

diff --git a/src/plugins/usage_collection/server/plugin.ts b/src/plugins/usage_collection/server/plugin.ts
@@ -75,6 +75,7 @@ export class UsageCollectionPlugin implements Plugin<CollectorSet> {
         opensearchDashboardsVersion: this.initializerContext.env.packageInfo.version,
         server: core.http.getServerInfo(),
         uuid: this.initializerContext.env.instanceUuid,
+        batchingInterval: config.uiMetric.batchingIntervalInS,
       },
       metrics: core.metrics,
       overallStatus$: core.status.overall$,

diff --git a/src/plugins/usage_collection/server/report/store_report.test.ts b/src/plugins/usage_collection/server/report/store_report.test.ts
@@ -80,7 +80,9 @@ describe('store_report', () => {
     expect(savedObjectClient.incrementCounter).toHaveBeenCalledWith(
       'ui-metric',
       'test-app-name:test-event-name',
-      'count'
+      'count',
+      {},
+      3
     );
     expect(savedObjectClient.bulkCreate).toHaveBeenCalledWith([
       {

diff --git a/src/plugins/usage_collection/server/report/store_report.ts b/src/plugins/usage_collection/server/report/store_report.ts
@@ -57,11 +57,17 @@ export async function storeReport(
       };
     }),
     ...uiStatsMetrics.map(async ([key, metric]) => {
-      const { appName, eventName } = metric;
+      const { appName, eventName, stats } = metric;
       const savedObjectId = `${appName}:${eventName}`;
       return {
         saved_objects: [
-          await internalRepository.incrementCounter('ui-metric', savedObjectId, 'count'),
+          await internalRepository.incrementCounter(
+            'ui-metric',
+            savedObjectId,
+            'count',
+            {},
+            stats.sum
+          ),
         ],
       };
     }),

diff --git a/src/plugins/usage_collection/server/routes/index.ts b/src/plugins/usage_collection/server/routes/index.ts
@@ -56,11 +56,12 @@ export function setupRoutes({
       hostname: string;
       port: number;
     };
+    batchingInterval: number;
   };
   collectorSet: CollectorSet;
   metrics: MetricsServiceSetup;
   overallStatus$: Observable<ServiceStatus>;
 }) {
-  registerUiMetricRoute(router, getSavedObjects);
+  registerUiMetricRoute(router, getSavedObjects, rest.config.batchingInterval);
   registerStatsRoute({ router, ...rest });
 }
diff --git a/src/plugins/usage_collection/server/routes/report_metrics.ts b/src/plugins/usage_collection/server/routes/report_metrics.ts
@@ -31,11 +31,16 @@
 import { schema } from '@osd/config-schema';
 import { IRouter, ISavedObjectsRepository } from 'opensearch-dashboards/server';
 import { storeReport, reportSchema } from '../report';
+import { BatchReport } from '../types';
+import { ReportSchemaType } from '../report/schema';
 
 export function registerUiMetricRoute(
   router: IRouter,
-  getSavedObjects: () => ISavedObjectsRepository | undefined
+  getSavedObjects: () => ISavedObjectsRepository | undefined,
+  batchingInterval: number
 ) {
+  let batchReport = { report: {}, startTimestamp: 0 } as BatchReport;
+  const batchingIntervalInMs = batchingInterval * 1000;
   router.post(
     {
       path: '/api/ui_metric/report',
@@ -48,15 +53,84 @@ export function registerUiMetricRoute(
     async (context, req, res) => {
       const { report } = req.body;
       try {
-        const internalRepository = getSavedObjects();
-        if (!internalRepository) {
-          throw Error(`The saved objects client hasn't been initialised yet`);
+        const currTime = Date.now();
+
+        // Add the current report to batchReport
+        batchReport.report = combineReports(report, batchReport.report);
+        // If the time duration since the batchReport startTime is greater than batchInterval then write it to the savedObject
+        if (currTime - batchReport.startTimestamp >= batchingIntervalInMs) {
+          const prevReport = batchReport;
+
+          batchReport = {
+            report: {},
+            startTimestamp: currTime,
+          }; // reseting the batchReport and updating the startTimestamp to current TimeStamp
+
+          if (prevReport) {
+            // Write the previously batched Report to the saved object
+            const internalRepository = getSavedObjects();
+            if (!internalRepository) {
+              throw Error(`The saved objects client hasn't been initialised yet`);
+            }
+            await storeReport(internalRepository, prevReport.report);
+          }
         }
-        await storeReport(internalRepository, report);
+
         return res.ok({ body: { status: 'ok' } });
       } catch (error) {
         return res.ok({ body: { status: 'fail' } });
       }
     }
   );
 }
+
+function combineReports(report1: ReportSchemaType, report2: ReportSchemaType) {
+  // Combines report2 onto the report1 and returns the updated report1
+
+  // Combining User Agents
+  const combinedUserAgent = { ...report2.userAgent, ...report1.userAgent };
+
+  // Combining UI metrics
+  const combinedUIMetric = { ...report1.uiStatsMetrics };
+  if (report2.uiStatsMetrics !== undefined) {
+    for (const key of Object.keys(report2.uiStatsMetrics)) {
+      if (report2.uiStatsMetrics[key]?.stats?.sum === undefined) {
+        continue;
+      } else if (report1.uiStatsMetrics?.[key] === undefined) {
+        combinedUIMetric[key] = report2.uiStatsMetrics[key];
+      } else {
+        const { stats, ...rest } = combinedUIMetric[key];
+        const combinedStats = { ...stats };
+        combinedStats.sum += report2.uiStatsMetrics[key].stats.sum; // Updating the sum since it is field we will be using to update the saved Object
+        combinedUIMetric[key] = { ...rest, stats: combinedStats };
+      }
+    }
+  }
+
+  // Combining Application Usage
+  const combinedApplicationUsage = { ...report1.application_usage };
+  if (report2.application_usage !== undefined) {
+    for (const key of Object.keys(report2.application_usage)) {
+      if (
+        report2.application_usage[key]?.numberOfClicks === undefined ||
+        report2.application_usage[key]?.minutesOnScreen === undefined
+      ) {
+        continue;
+      } else if (report1.application_usage?.[key] === undefined) {
+        combinedApplicationUsage[key] = report2.application_usage[key];
+      } else {
+        const combinedUsage = { ...combinedApplicationUsage[key] };
+        combinedUsage.numberOfClicks += report2.application_usage[key]?.numberOfClicks || 0;
+        combinedUsage.minutesOnScreen += report2.application_usage[key]?.minutesOnScreen || 0;
+        combinedApplicationUsage[key] = combinedUsage;
+      }
+    }
+  }
+
+  return {
+    reportVersion: report1.reportVersion,
+    userAgent: combinedUserAgent,
+    uiStatsMetrics: combinedUIMetric,
+    application_usage: combinedApplicationUsage,
+  } as ReportSchemaType;
+}
diff --git a/src/plugins/usage_collection/server/types.ts b/src/plugins/usage_collection/server/types.ts
@@ -0,0 +1,10 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { ReportSchemaType } from './report/schema';
+export interface BatchReport {
+  report: ReportSchemaType;
+  startTimestamp: number;
+}