[8.15] [Synthetics] Monitors sync request, retry on huge payload !! (#…

…202467) (#203665) # Backport This will backport the following commits from `main` to `8.15`: - [[Synthetics] Monitors sync request, retry on huge payload !! (#202467)](#202467)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport) \n\n### Questions ?\nPlease refer to the [Backport tool\ndocumentation](https://github.com/sqren/backport)\n\n\n\n---------\n\nCo-authored-by: Shahzad <[email protected]>"}},{"branch":"8.16","label":"v8.16.3","branchLabelMappingKey":"^v(\\d+).(\\d+).\\d+$","isSourceBranch":false,"state":"NOT_CREATED"},{"branch":"8.17","label":"v8.17.1","branchLabelMappingKey":"^v(\\d+).(\\d+).\\d+$","isSourceBranch":false,"state":"NOT_CREATED"}]}] BACKPORT--> --------- Co-authored-by: Shahzad <[email protected]>
elastic · Dec 10, 2024 · 844645d · 844645d
1 parent ab59825
commit 844645d
Show file tree

Hide file tree

Showing 3 changed files with 98 additions and 18 deletions.
diff --git a/...ns/observability_solution/synthetics/server/synthetics_service/service_api_client.test.ts b/...ns/observability_solution/synthetics/server/synthetics_service/service_api_client.test.ts
@@ -470,6 +470,61 @@ describe('callAPI', () => {
       url: 'https://service.dev/monitors/sync',
     });
   });
+
+  it('splits the payload into multiple requests if the payload is too large', async () => {
+    const requests: number[] = [];
+    const axiosSpy = (axios as jest.MockedFunction<typeof axios>).mockImplementation((req: any) => {
+      requests.push(req.data.monitors.length);
+      if (req.data.monitors.length > 100) {
+        // throw 413 error
+        return Promise.reject({ response: { status: 413 } });
+      }
+
+      return Promise.resolve({} as any);
+    });
+
+    const apiClient = new ServiceAPIClient(
+      logger,
+      {
+        manifestUrl: 'http://localhost:8080/api/manifest',
+        tls: { certificate: 'test-certificate', key: 'test-key' } as any,
+      },
+      {
+        isDev: true,
+        stackVersion: '8.7.0',
+        cloud: { cloudId: 'test-id', deploymentId: 'deployment-id' },
+      } as SyntheticsServerSetup
+    );
+
+    apiClient.locations = testLocations;
+
+    const output = { hosts: ['https://localhost:9200'], api_key: '12345' };
+
+    const monitors = new Array(250).fill({
+      ...request1[0],
+      locations: [
+        {
+          id: 'us_central',
+          isServiceManaged: true,
+        },
+      ],
+    });
+
+    await apiClient.syncMonitors({
+      monitors,
+      output,
+      license: licenseMock.license,
+      location: {
+        id: 'us_central',
+        url: 'https://service.dev',
+        label: 'Test location',
+        isServiceManaged: true,
+      },
+    });
+
+    expect(axiosSpy).toHaveBeenCalledTimes(7);
+    expect(requests).toEqual([250, 125, 125, 63, 62, 63, 62]);
+  });
 });
 
 const testLocations: PublicLocations = [

diff --git a/...plugins/observability_solution/synthetics/server/synthetics_service/service_api_client.ts b/...plugins/observability_solution/synthetics/server/synthetics_service/service_api_client.ts
@@ -6,7 +6,7 @@
  */
 
 import axios, { AxiosError, AxiosRequestConfig, AxiosResponse } from 'axios';
-import { forkJoin, from as rxjsFrom, Observable, of } from 'rxjs';
+import { concat, forkJoin, from as rxjsFrom, Observable, of } from 'rxjs';
 import { catchError, tap } from 'rxjs';
 import * as https from 'https';
 import { SslConfig } from '@kbn/server-http-tools';
@@ -211,21 +211,47 @@ export class ServiceAPIClient {
 
     const monitorsByLocation = this.processServiceData(serviceData);
 
-    monitorsByLocation.forEach(({ location: { url, id }, monitors, data }) => {
-      const promise = this.callServiceEndpoint(data, method, url, endpoint);
-      promises.push(
-        rxjsFrom(promise).pipe(
+    monitorsByLocation.forEach(({ location: { url, id }, data }) => {
+      const sendRequest = (payload: ServicePayload): Observable<any> => {
+        const promise = this.callServiceEndpoint(payload, method, url, endpoint);
+        return rxjsFrom(promise).pipe(
           tap((result) => {
-            this.logSuccessMessage(url, method, monitors.length, result);
+            this.logSuccessMessage(url, method, payload.monitors.length, result);
           }),
           catchError((err: AxiosError<{ reason: string; status: number }>) => {
+            if (err.response?.status === 413 && payload.monitors.length > 1) {
+              // If payload is too large, split it and retry
+              const mid = Math.ceil(payload.monitors.length / 2);
+              const firstHalfMonitors = payload.monitors.slice(0, mid);
+              const secondHalfMonitors = payload.monitors.slice(mid);
+
+              this.logger.debug(
+                `Payload of ${payload.monitors.length} monitors is too large for location ${id}, splitting in half, in chunks of ${mid}`
+              );
+
+              return concat(
+                sendRequest({
+                  ...payload,
+                  monitors: firstHalfMonitors,
+                }), // Retry with the first half
+                sendRequest({
+                  ...payload,
+                  monitors: secondHalfMonitors,
+                }) // Retry with the second half
+              );
+            }
+
             pushErrors.push({ locationId: id, error: err.response?.data! });
-            this.logServiceError(err, url, method, monitors.length);
-            // we don't want to throw an unhandled exception here
+            this.logServiceError(err, url, method, payload.monitors.length);
+
+            // Return an empty observable to prevent unhandled exceptions
             return of(true);
           })
-        )
-      );
+        );
+      };
+
+      // Start with the initial data payload
+      promises.push(sendRequest(data));
     });
 
     const result = await forkJoin(promises).toPromise();

diff --git a/...plugins/observability_solution/synthetics/server/synthetics_service/synthetics_service.ts b/...plugins/observability_solution/synthetics/server/synthetics_service/synthetics_service.ts
@@ -152,7 +152,9 @@ export class SyntheticsService {
       service.locations = result.locations;
       service.apiClient.locations = result.locations;
       this.logger.debug(
-        `Fetched ${service.locations} Synthetics service locations from manifest: ${this.config.manifestUrl}`
+        `Fetched ${service.locations
+          .map((loc) => loc.id)
+          .join(',')} Synthetics service locations from manifest: ${this.config.manifestUrl}`
       );
     } catch (e) {
       this.logger.error(e);
@@ -167,7 +169,7 @@ export class SyntheticsService {
       [SYNTHETICS_SERVICE_SYNC_MONITORS_TASK_TYPE]: {
         title: 'Synthetics Service - Sync Saved Monitors',
         description: 'This task periodically pushes saved monitors to Synthetics Service.',
-        timeout: '1m',
+        timeout: '2m',
         maxAttempts: 3,
 
         createTaskRunner: ({ taskInstance }: { taskInstance: ConcreteTaskInstance }) => {
@@ -670,22 +672,19 @@ export class SyntheticsService {
 
       if (lastRunAt) {
         // log if it has missed last schedule
-        const diff = moment(lastRunAt).diff(current, 'minutes');
+        const diff = moment(current).diff(lastRunAt, 'minutes');
         const syncInterval = Number((this.config.syncInterval ?? '5m').split('m')[0]);
         if (diff > syncInterval) {
-          const message = `Synthetics monitor sync task has missed its schedule, it last ran ${diff} ago.`;
+          const message = `Synthetics monitor sync task has missed its schedule, it last ran ${diff} minutes ago.`;
           this.logger.warn(message);
           sendErrorTelemetryEvents(this.logger, this.server.telemetry, {
             message,
             reason: 'Failed to run synthetics sync task on schedule',
             type: 'syncTaskMissedSchedule',
             stackVersion: this.server.stackVersion,
           });
-        } else {
-          this.logger.debug(
-            `Synthetics monitor sync task is running as expected, it last ran ${diff} minutes ago.`
-          );
         }
+        this.logger.debug(`Synthetics monitor sync task last ran ${diff} minutes ago.`);
       }
       state.lastRunAt = current.toISOString();
     } catch (e) {