abnormal_security: allow configuration of a grace period for ai_security_mailbox_not_analyzed (#14235)

The "Not Analyzed" data source may have a delayed availability for
messages, with advice from Abnormal Security being that 95% of messages
are ready by 60s, but that a 5-10 minute delay would likely be sufficient
to account for most messages transitioning from "Not Analyzed" to
successfully scanned. Given this range of expectations, provide a
mechanism for the user to tune the agent's response. The default
behaviour is to retain no grace period, the current behaviour.

Author: efd6

packages/abnormal_security/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "1.8.1"
+  changes:
+    - description: Prevent loss of recent unprocessed messages by the `ai_security_mailbox_not_analyzed` data stream.
+      type: bugfix
+      link: https://github.com/elastic/integrations/pull/14235
 - version: "1.8.0"
   changes:
     - description: Enrich threat events with attachment and link details.

packages/abnormal_security/data_stream/ai_security_mailbox_not_analyzed/agent/stream/cel.yml.hbs

@@ -18,21 +18,30 @@ state:
   initial_interval: {{initial_interval}}
   access_token: {{access_token}}
   want_more: false
+{{#if wait_interval}}
+  wait: {{wait_interval}}
+{{/if}}
 redact:
   fields:
     - access_token
 program: |
-  (
-    state.with({
-      "start_time": state.?cursor.last_timestamp.orValue((now - duration(state.initial_interval)).format(time_layout.RFC3339)),
-      "end_time": now.format(time_layout.RFC3339),
-    })
-  ).as(state, state.with(
+  {
+    "start_time": state.?cursor.last_timestamp.optMap(t,
+      t.parse_time(time_layout.RFC3339)
+    ).orValue(
+      now - duration(state.initial_interval)
+    ),
+    "end_time": now - duration(state.?wait.orValue("0s")),
+  }.as(query_range, state.with(
     request(
       "GET",
       state.url.trim_right("/") + "/v1/abuse_mailbox/not_analyzed?" + {
-        "start": [state.start_time],
-        "end": [state.end_time],
+        // Condition the start of the request interval so that start is never
+        // after end. Sacrifice a null interval request in the case that the
+        // last start is still within the wait interval.
+        // TODO: replace list min with diadic min when stack version is at least v8.18.
+        "start": [min([query_range.start_time, query_range.end_time]).format(time_layout.RFC3339)],
+        "end": [query_range.end_time.format(time_layout.RFC3339)],
       }.format_query()
     ).with({
       "Header":{
@@ -47,10 +56,14 @@ program: |
               }
             ),
             "cursor": {
-              ?"last_timestamp": has(body.results) && body.results.size() > 0 ?
-                optional.of(body.results.map(r, timestamp(r.reported_datetime)).max().format(time_layout.RFC3339))
+              "last_timestamp": has(body.results) && body.results.size() > 0 ?
+                body.results.map(r, timestamp(r.reported_datetime)).max().format(time_layout.RFC3339)
+              : has(state.?cursor.last_timestamp) ?
+                state.cursor.last_timestamp
               :
-                state.?cursor.last_timestamp
+                // If we did not get any messages, keep the start as
+                // the next cursor.
+                query_range.start_time
             },
           }
         :

packages/abnormal_security/data_stream/ai_security_mailbox_not_analyzed/manifest.yml

@@ -14,6 +14,13 @@ streams:
         show_user: true
         default: 2160h
         description: How far back to pull the AI Security Mailbox Not Analyzed messages from Abnormal Security API. Defaults to 90 days (2160h) before end. Supported units for this parameter are h/m/s.
+      - name: wait_interval
+        type: text
+        title: Recent Message Grace Interval 
+        multi: false
+        required: true
+        show_user: true
+        description: How long to wait before attempting to collect recent messages. This option allows the Abnormal Security API to complete analysis of messages before the agent attempts to collect them. This should not be greater than the initial interval. Supported units for this parameter are h/m/s.
       - name: interval
         type: text
         title: Interval

packages/abnormal_security/manifest.yml

@@ -1,7 +1,7 @@
 format_version: 3.2.1
 name: abnormal_security
 title: Abnormal Security
-version: "1.8.0"
+version: "1.8.1"
 description: Collect logs from Abnormal Security with Elastic Agent.
 type: integration
 categories: