[Salesforce] Add pagination support (#34065)

* Add pagination support

CHANGELOG.next.asciidoc

@@ -177,6 +177,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 - Add metrics for TCP packet processing. {pull}34333[34333]
 - Add metrics for unix socket packet processing. {pull}34335[34335]
 - Add beta `take over` mode for `filestream` for simple migration from `log` inputs {pull}34292[34292]
+- Add pagination support for Salesforce module. {issue}34057[34057] {pull}34065[34065]
 
 *Auditbeat*
 

x-pack/filebeat/module/salesforce/apex-rest/config/apex-rest.yml

@@ -30,18 +30,21 @@ request.proxy_url: {{ .proxy_url }}
 request.transforms:
   - set:
       target: url.params.q
-      value: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND CreatedDate > [[.cursor.last_published_apex]] AND (EventType = 'ApexCallout' OR EventType = 'ApexExecution' OR EventType = 'ApexRestApi' OR EventType = 'ApexSoap' OR EventType = 'ApexTrigger' OR EventType = 'ExternalCustomApexCallout') ORDER BY CreatedDate ASC NULLS FIRST"
-      default: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND (EventType = 'ApexCallout' OR EventType = 'ApexExecution' OR EventType = 'ApexRestApi' OR EventType = 'ApexSoap' OR EventType = 'ApexTrigger' OR EventType = 'ExternalCustomApexCallout') ORDER BY CreatedDate ASC NULLS FIRST"
-response.split:
-  target: body.records
+      value: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND LogDate > [[.cursor.last_published_apex]] AND (EventType = 'ApexCallout' OR EventType = 'ApexExecution' OR EventType = 'ApexRestApi' OR EventType = 'ApexSoap' OR EventType = 'ApexTrigger' OR EventType = 'ExternalCustomApexCallout') ORDER BY LogDate ASC NULLS FIRST"
+      default: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND (EventType = 'ApexCallout' OR EventType = 'ApexExecution' OR EventType = 'ApexRestApi' OR EventType = 'ApexSoap' OR EventType = 'ApexTrigger' OR EventType = 'ExternalCustomApexCallout') ORDER BY LogDate ASC NULLS FIRST"
+response.pagination:
+  - set:
+      target: url.value
+      value: '[[if (ne .last_response.body.done true)]]{{ .url }}[[.last_response.body.nextRecordsUrl]][[end]]'
+      fail_on_template_error: true
 chain:
   - step:
       request.url: {{ .url }}/services/data/v52.0/sobjects/EventLogFile/$.records[:].Id/LogFile
       request.method: GET
       replace: $.records[:].Id
 cursor:
   last_published_apex:
-    value: '[[.last_event.CreatedDate]]'
+    value: '[[(formatDate ((parseDate .last_event.TIMESTAMP_DERIVED "RFC3339").Add (parseDuration "-1h")))]]'
 
 {{ else if eq .input "file" }}
 

x-pack/filebeat/module/salesforce/login-rest/config/login-rest.yml

@@ -30,18 +30,21 @@ request.proxy_url: {{ .proxy_url }}
 request.transforms:
   - set:
       target: url.params.q
-      value: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND EventType = 'Login' AND CreatedDate > [[.cursor.last_published_login]] ORDER BY CreatedDate ASC NULLS FIRST"
-      default: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND EventType = 'Login' ORDER BY CreatedDate ASC NULLS FIRST"
-response.split:
-  target: body.records
+      value: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND EventType = 'Login' AND LogDate > [[.cursor.last_published_login]] ORDER BY LogDate ASC NULLS FIRST"
+      default: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND EventType = 'Login' ORDER BY LogDate ASC NULLS FIRST"
+response.pagination:
+  - set:
+      target: url.value
+      value: '[[if (ne .last_response.body.done true)]]{{ .url }}[[.last_response.body.nextRecordsUrl]][[end]]'
+      fail_on_template_error: true
 chain:
   - step:
       request.url: {{ .url }}/services/data/v52.0/sobjects/EventLogFile/$.records[:].Id/LogFile
       request.method: GET
       replace: $.records[:].Id
 cursor:
   last_published_login:
-    value: '[[.last_event.CreatedDate]]'
+    value: '[[(formatDate ((parseDate .last_event.TIMESTAMP_DERIVED "RFC3339").Add (parseDuration "-1h")))]]'
 
 {{ else if eq .input "file" }}
 

x-pack/filebeat/module/salesforce/logout-rest/config/logout-rest.yml

@@ -29,18 +29,21 @@ request.proxy_url: {{ .proxy_url }}
 request.transforms:
   - set:
       target: url.params.q
-      value: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND EventType = 'Logout' AND CreatedDate > [[.cursor.last_published_logout]] ORDER BY CreatedDate ASC NULLS FIRST"
-      default: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND EventType = 'Logout' ORDER BY CreatedDate ASC NULLS FIRST"
-response.split:
-  target: body.records
+      value: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND EventType = 'Logout' AND LogDate > [[.cursor.last_published_logout]] ORDER BY LogDate ASC NULLS FIRST"
+      default: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE Interval = 'Hourly' AND EventType = 'Logout' ORDER BY LogDate ASC NULLS FIRST"
+response.pagination:
+  - set:
+      target: url.value
+      value: '[[if (ne .last_response.body.done true)]]{{ .url }}[[.last_response.body.nextRecordsUrl]][[end]]'
+      fail_on_template_error: true
 chain:
   - step:
       request.url: {{ .url }}/services/data/v52.0/sobjects/EventLogFile/$.records[:].Id/LogFile
       request.method: GET
       replace: $.records[:].Id
 cursor:
   last_published_logout:
-    value: '[[.last_event.CreatedDate]]'
+    value: '[[(formatDate ((parseDate .last_event.TIMESTAMP_DERIVED "RFC3339").Add (parseDuration "-1h")))]]'
 
 {{ else if eq .input "file" }}
 

x-pack/filebeat/module/salesforce/setupaudittrail-rest/config/setupaudittrail-rest.yml

@@ -34,6 +34,11 @@ request.transforms:
       default: "SELECT Action,CreatedByContext,CreatedById,CreatedByIssuer,CreatedDate,DelegateUser,Display,Id,ResponsibleNamespacePrefix,Section FROM SetupAuditTrail ORDER BY CreatedDate ASC NULLS FIRST"
 response.split:
   target: body.records
+response.pagination:
+- set:
+    target: url.value
+    value: '[[if (ne .last_response.body.done true)]]{{ .url }}[[.last_response.body.nextRecordsUrl]][[end]]'
+    fail_on_template_error: true
 cursor:
   last_published_setupaudittrail:
     value: '[[.last_event.CreatedDate]]'