proofpoint_on_demand: add sinceTime cursor to message stream

efd6 · efd6 · commit 1289a1d75e85 · 2026-02-23T11:43:19.000+10:30
Track the ts timestamp from each message event and pass it as the sinceTime query parameter on WebSocket reconnection. This allows the agent to resume from its last position rather than relying on the API's default replay behaviour, which only covers the most recent hour of data [1]. The ts values have microsecond resolution while the API documentation specifies millisecond resolution for sinceTime. We pass the microsecond value through unchanged because at least one other integration (logrhythm-proofpoint-on-demand [2]) does the same successfully. Only the message data stream is changed here. The mail and audit streams use the same API endpoint and likely support sinceTime too, but there is no direct evidence confirming that events are delivered in ascending ts order for those stream types. [1] https://docs.cyderes.cloud/files/proofpoint-on-demand-log-api-rev-c.pdf [2] https://github.com/jpsutton/logrhythm-proofpoint-on-demand
diff --git a/packages/proofpoint_on_demand/changelog.yml b/packages/proofpoint_on_demand/changelog.yml
@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "1.9.0"
+  changes:
+    - description: Use sinceTime cursor for message data stream to resume from last position on reconnection.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/17508
 - version: "1.8.1"
   changes:
     - description: Remove duplicate security-solution-default tag references
diff --git a/packages/proofpoint_on_demand/data_stream/message/agent/stream/websocket.yml.hbs b/packages/proofpoint_on_demand/data_stream/message/agent/stream/websocket.yml.hbs
@@ -1,9 +1,17 @@
 url: {{url}}/v1/stream?cid={{cluster_id}}&type=message
+url_program: |
+  has(state.?cursor.last_timestamp) ?
+    state.url+"&sinceTime="+state.cursor.last_timestamp
+  :
+    state.url
 auth.bearer_token: {{access_token}}
 redact:
   fields: ~
 program: |
   bytes(state.response).decode_json().as(body,{
+    "cursor": {
+      "last_timestamp": body.ts,
+    },
     "events": {
       "message":  body.encode_json(),
     }
diff --git a/packages/proofpoint_on_demand/manifest.yml b/packages/proofpoint_on_demand/manifest.yml
@@ -1,7 +1,7 @@
 format_version: 3.1.4
 name: proofpoint_on_demand
 title: Proofpoint On Demand
-version: "1.8.1"
+version: "1.9.0"
 description: Collect logs from Proofpoint On Demand with Elastic Agent.
 type: integration
 categories:
