sentinel_one: fix duplicate event ingestion in threat_event and application_risk

efd6 · efd6 · commit 5e2e95f0f388 · 2026-07-20T17:40:05.000+09:30
Add a time watermark (updatedAt__gte) to the threat_event CEL program
so each interval only fetches threats updated since the last completed
walk, rather than re-walking the entire catalog. The high-water mark
is tracked in cursor.last_timestamp and seeded from a new
initial_interval manifest variable on the first run. The default poll
interval drops from 24h to 5m since the watermark eliminates full
re-walks.

Add fingerprint processors to both the threat_event and
application_risk ingest pipelines so that re-emitted documents with
the same id are rejected within a backing index rather than
accumulating as duplicates.
diff --git a/packages/sentinel_one/changelog.yml b/packages/sentinel_one/changelog.yml
@@ -1,4 +1,12 @@
 # newer versions go on top
+- version: "2.11.1"
+  changes:
+    - description: Add time watermark and fingerprint deduplication to the threat event data stream to prevent duplicate ingestion.
+      type: bugfix
+      link: https://github.com/elastic/integrations/pull/20211
+    - description: Add fingerprint deduplication to the application risk data stream to prevent duplicate ingestion.
+      type: bugfix
+      link: https://github.com/elastic/integrations/pull/20211
 - version: "2.11.0"
   changes:
     - description: |
diff --git a/packages/sentinel_one/data_stream/application_risk/elasticsearch/ingest_pipeline/default.yml b/packages/sentinel_one/data_stream/application_risk/elasticsearch/ingest_pipeline/default.yml
@@ -26,6 +26,13 @@ processors:
       field: event.original
       tag: json_event_original
       target_field: json
+  - fingerprint:
+      tag: fingerprint_application_risk_dedup
+      fields:
+        - json.id
+        - json.status
+      target_field: _id
+      ignore_missing: true
   - set:
       field: event.kind
       tag: set_event_kind
diff --git a/packages/sentinel_one/data_stream/threat_event/_dev/test/policy/test-all.expected b/packages/sentinel_one/data_stream/threat_event/_dev/test/policy/test-all.expected
@@ -33,8 +33,17 @@ inputs:
                     state.url.trim_right("/") + "/web/api/v2.1/threats?" + {
                       "skipCount": ["true"],
                       "limit": [string(state.batch_size)],
+                      "sortBy": ["updatedAt"],
+                      "sortOrder": ["asc"],
+                      "updatedAt__gte": [
+                        state.?cursor.?list_updated_at_gte.orValue(
+                          state.?cursor.?last_timestamp.orValue(
+                            (now - duration(state.initial_interval)).format(time_layout.RFC3339)
+                          )
+                        )
+                      ],
                       ?"siteIds": state.?site_ids.optMap(v, [string(v)]),
-                      ?"cursor": state.?cursor.next_page.token.optMap(v, [v]),
+                      ?"cursor": state.?cursor.?next_page.token.optMap(v, [v]),
                     }.format_query()
                   ).with(
                     {
@@ -51,6 +60,12 @@ inputs:
                             ?"token": (body.?pagination.nextCursor.orValue(null) != null) ? optional.of(body.pagination.nextCursor) : optional.none(),
                           },
                           "fetch_more": body.?pagination.nextCursor.orValue(null) != null,
+                          "list_updated_at_gte": state.?cursor.?list_updated_at_gte.orValue(
+                            state.?cursor.?last_timestamp.orValue(
+                              (now - duration(state.initial_interval)).format(time_layout.RFC3339)
+                            )
+                          ),
+                          ?"last_timestamp": state.?cursor.last_timestamp,
                         },
                       }
                     )
@@ -69,7 +84,6 @@ inputs:
                         },
                       },
                       "want_more": false,
-                      "cursor": {},
                     }
                   )
                 )
@@ -119,6 +133,11 @@ inputs:
                                 ?"token": has_more_events ? optional.of(body.pagination.nextCursor) : optional.none(),
                               },
                               "fetch_more": state.?cursor.fetch_more.orValue(false),
+                              "list_updated_at_gte": state.cursor.list_updated_at_gte,
+                              ?"last_timestamp": has_more_events ?
+                                state.?cursor.last_timestamp
+                              :
+                                state.cursor.worklist.data[0].?threatInfo.updatedAt.or(state.?cursor.last_timestamp),
                             },
                           }
                         )
@@ -145,7 +164,6 @@ inputs:
                   {
                     "events": [],
                     "want_more": false,
-                    "cursor": {},
                   }
               )
             )
@@ -242,6 +260,7 @@ inputs:
           state:
             api_token: ${SECRET_0}
             batch_size: 100
+            initial_interval: 1h
             site_ids: "123"
           tags:
             - preserve_original_event
diff --git a/packages/sentinel_one/data_stream/threat_event/_dev/test/policy/test-all.yml b/packages/sentinel_one/data_stream/threat_event/_dev/test/policy/test-all.yml
@@ -83,6 +83,7 @@ vars:
       -----END PRIVATE KEY-----
 data_stream:
   vars:
+    initial_interval: 1h
     interval: 30s
     batch_size: 100
     max_executions: 1000
diff --git a/packages/sentinel_one/data_stream/threat_event/_dev/test/policy/test-default.expected b/packages/sentinel_one/data_stream/threat_event/_dev/test/policy/test-default.expected
@@ -9,7 +9,7 @@ inputs:
         - config_version: 2
           data_stream:
             dataset: sentinel_one.threat_event
-          interval: 24h
+          interval: 5m
           max_executions: 1000
           program: |-
             (
@@ -22,8 +22,17 @@ inputs:
                     state.url.trim_right("/") + "/web/api/v2.1/threats?" + {
                       "skipCount": ["true"],
                       "limit": [string(state.batch_size)],
+                      "sortBy": ["updatedAt"],
+                      "sortOrder": ["asc"],
+                      "updatedAt__gte": [
+                        state.?cursor.?list_updated_at_gte.orValue(
+                          state.?cursor.?last_timestamp.orValue(
+                            (now - duration(state.initial_interval)).format(time_layout.RFC3339)
+                          )
+                        )
+                      ],
                       ?"siteIds": state.?site_ids.optMap(v, [string(v)]),
-                      ?"cursor": state.?cursor.next_page.token.optMap(v, [v]),
+                      ?"cursor": state.?cursor.?next_page.token.optMap(v, [v]),
                     }.format_query()
                   ).with(
                     {
@@ -40,6 +49,12 @@ inputs:
                             ?"token": (body.?pagination.nextCursor.orValue(null) != null) ? optional.of(body.pagination.nextCursor) : optional.none(),
                           },
                           "fetch_more": body.?pagination.nextCursor.orValue(null) != null,
+                          "list_updated_at_gte": state.?cursor.?list_updated_at_gte.orValue(
+                            state.?cursor.?last_timestamp.orValue(
+                              (now - duration(state.initial_interval)).format(time_layout.RFC3339)
+                            )
+                          ),
+                          ?"last_timestamp": state.?cursor.last_timestamp,
                         },
                       }
                     )
@@ -58,7 +73,6 @@ inputs:
                         },
                       },
                       "want_more": false,
-                      "cursor": {},
                     }
                   )
                 )
@@ -108,6 +122,11 @@ inputs:
                                 ?"token": has_more_events ? optional.of(body.pagination.nextCursor) : optional.none(),
                               },
                               "fetch_more": state.?cursor.fetch_more.orValue(false),
+                              "list_updated_at_gte": state.cursor.list_updated_at_gte,
+                              ?"last_timestamp": has_more_events ?
+                                state.?cursor.last_timestamp
+                              :
+                                state.cursor.worklist.data[0].?threatInfo.updatedAt.or(state.?cursor.last_timestamp),
                             },
                           }
                         )
@@ -134,7 +153,6 @@ inputs:
                   {
                     "events": [],
                     "want_more": false,
-                    "cursor": {},
                   }
               )
             )
@@ -152,6 +170,7 @@ inputs:
           state:
             api_token: ${SECRET_0}
             batch_size: 1000
+            initial_interval: 24h
           tags:
             - forwarded
             - sentinel_one-threat_event
diff --git a/packages/sentinel_one/data_stream/threat_event/_dev/test/policy/test-default.yml b/packages/sentinel_one/data_stream/threat_event/_dev/test/policy/test-default.yml
@@ -3,7 +3,8 @@ vars:
   api_token: test_api_token
 data_stream:
   vars:
-    interval: 24h
+    initial_interval: 24h
+    interval: 5m
     batch_size: 1000
     max_executions: 1000
     enable_request_tracer: false
diff --git a/packages/sentinel_one/data_stream/threat_event/agent/stream/cel.yml.hbs b/packages/sentinel_one/data_stream/threat_event/agent/stream/cel.yml.hbs
@@ -18,6 +18,7 @@ resource.url: {{url}}
 state:
   api_token: {{api_token}}
   batch_size: {{batch_size}}
+  initial_interval: {{initial_interval}}
 {{#if site_ids }}
   site_ids: !!str {{site_ids}}
 {{/if}}
@@ -35,8 +36,17 @@ program: |-
           state.url.trim_right("/") + "/web/api/v2.1/threats?" + {
             "skipCount": ["true"],
             "limit": [string(state.batch_size)],
+            "sortBy": ["updatedAt"],
+            "sortOrder": ["asc"],
+            "updatedAt__gte": [
+              state.?cursor.?list_updated_at_gte.orValue(
+                state.?cursor.?last_timestamp.orValue(
+                  (now - duration(state.initial_interval)).format(time_layout.RFC3339)
+                )
+              )
+            ],
             ?"siteIds": state.?site_ids.optMap(v, [string(v)]),
-            ?"cursor": state.?cursor.next_page.token.optMap(v, [v]),
+            ?"cursor": state.?cursor.?next_page.token.optMap(v, [v]),
           }.format_query()
         ).with(
           {
@@ -53,6 +63,12 @@ program: |-
                   ?"token": (body.?pagination.nextCursor.orValue(null) != null) ? optional.of(body.pagination.nextCursor) : optional.none(),
                 },
                 "fetch_more": body.?pagination.nextCursor.orValue(null) != null,
+                "list_updated_at_gte": state.?cursor.?list_updated_at_gte.orValue(
+                  state.?cursor.?last_timestamp.orValue(
+                    (now - duration(state.initial_interval)).format(time_layout.RFC3339)
+                  )
+                ),
+                ?"last_timestamp": state.?cursor.last_timestamp,
               },
             }
           )
@@ -71,7 +87,6 @@ program: |-
               },
             },
             "want_more": false,
-            "cursor": {},
           }
         )
       )
@@ -121,6 +136,11 @@ program: |-
                       ?"token": has_more_events ? optional.of(body.pagination.nextCursor) : optional.none(),
                     },
                     "fetch_more": state.?cursor.fetch_more.orValue(false),
+                    "list_updated_at_gte": state.cursor.list_updated_at_gte,
+                    ?"last_timestamp": has_more_events ?
+                      state.?cursor.last_timestamp
+                    :
+                      state.cursor.worklist.data[0].?threatInfo.updatedAt.or(state.?cursor.last_timestamp),
                   },
                 }
               )
@@ -147,7 +167,6 @@ program: |-
         {
           "events": [],
           "want_more": false,
-          "cursor": {},
         }
     )
   )
diff --git a/packages/sentinel_one/data_stream/threat_event/elasticsearch/ingest_pipeline/default.yml b/packages/sentinel_one/data_stream/threat_event/elasticsearch/ingest_pipeline/default.yml
@@ -45,6 +45,13 @@ processors:
       field: event.original
       tag: json_event_original
       target_field: json
+  - fingerprint:
+      tag: fingerprint_threat_event_dedup
+      fields:
+        - json.id
+        - json.createdAt
+      target_field: _id
+      ignore_missing: true
   - rename:
       field: json.id
       tag: rename_id
diff --git a/packages/sentinel_one/data_stream/threat_event/manifest.yml b/packages/sentinel_one/data_stream/threat_event/manifest.yml
@@ -8,11 +8,19 @@ streams:
     enabled: false
     template_path: cel.yml.hbs
     vars:
+      - name: initial_interval
+        type: text
+        title: Initial Interval
+        description: How far back to look for threats on the first run. Supported units for this parameter are h/m/s.
+        default: 24h
+        multi: false
+        required: true
+        show_user: true
       - name: interval
         type: text
         title: Interval
         description: Duration between requests to the SentinelOne API. Supported units for this parameter are h/m/s.
-        default: 24h
+        default: 5m
         multi: false
         required: true
         show_user: true
diff --git a/packages/sentinel_one/manifest.yml b/packages/sentinel_one/manifest.yml
@@ -1,7 +1,7 @@
 format_version: "3.4.0"
 name: sentinel_one
 title: SentinelOne
-version: "2.11.0"
+version: "2.11.1"
 description: Collect logs from SentinelOne with Elastic Agent.
 type: integration
 categories:

Original file line number	Diff line number	Diff line change
`@@ -33,8 +33,17 @@ inputs:`
`33`	`33`	`state.url.trim_right("/") + "/web/api/v2.1/threats?" + {`
`34`	`34`	`"skipCount": ["true"],`
`35`	`35`	`"limit": [string(state.batch_size)],`
	`36`	`+ "sortBy": ["updatedAt"],`
	`37`	`+ "sortOrder": ["asc"],`
	`38`	`+ "updatedAt__gte": [`
	`39`	`+ state.?cursor.?list_updated_at_gte.orValue(`
	`40`	`+ state.?cursor.?last_timestamp.orValue(`
	`41`	`+ (now - duration(state.initial_interval)).format(time_layout.RFC3339)`
	`42`	`+ )`
	`43`	`+ )`
	`44`	`+ ],`
`36`	`45`	`?"siteIds": state.?site_ids.optMap(v, [string(v)]),`
`37`		`- ?"cursor": state.?cursor.next_page.token.optMap(v, [v]),`
	`46`	`+ ?"cursor": state.?cursor.?next_page.token.optMap(v, [v]),`
`38`	`47`	`}.format_query()`
`39`	`48`	`).with(`
`40`	`49`	`{`
`@@ -51,6 +60,12 @@ inputs:`
`51`	`60`	`?"token": (body.?pagination.nextCursor.orValue(null) != null) ? optional.of(body.pagination.nextCursor) : optional.none(),`
`52`	`61`	`},`
`53`	`62`	`"fetch_more": body.?pagination.nextCursor.orValue(null) != null,`
	`63`	`+ "list_updated_at_gte": state.?cursor.?list_updated_at_gte.orValue(`
	`64`	`+ state.?cursor.?last_timestamp.orValue(`
	`65`	`+ (now - duration(state.initial_interval)).format(time_layout.RFC3339)`
	`66`	`+ )`
	`67`	`+ ),`
	`68`	`+ ?"last_timestamp": state.?cursor.last_timestamp,`
`54`	`69`	`},`
`55`	`70`	`}`
`56`	`71`	`)`
`@@ -69,7 +84,6 @@ inputs:`
`69`	`84`	`},`
`70`	`85`	`},`
`71`	`86`	`"want_more": false,`
`72`		`- "cursor": {},`
`73`	`87`	`}`
`74`	`88`	`)`
`75`	`89`	`)`
`@@ -119,6 +133,11 @@ inputs:`
`119`	`133`	`?"token": has_more_events ? optional.of(body.pagination.nextCursor) : optional.none(),`
`120`	`134`	`},`
`121`	`135`	`"fetch_more": state.?cursor.fetch_more.orValue(false),`
	`136`	`+ "list_updated_at_gte": state.cursor.list_updated_at_gte,`
	`137`	`+ ?"last_timestamp": has_more_events ?`
	`138`	`+ state.?cursor.last_timestamp`
	`139`	`+ :`
	`140`	`+ state.cursor.worklist.data[0].?threatInfo.updatedAt.or(state.?cursor.last_timestamp),`
`122`	`141`	`},`
`123`	`142`	`}`
`124`	`143`	`)`
`@@ -145,7 +164,6 @@ inputs:`
`145`	`164`	`{`
`146`	`165`	`"events": [],`
`147`	`166`	`"want_more": false,`
`148`		`- "cursor": {},`
`149`	`167`	`}`
`150`	`168`	`)`
`151`	`169`	`)`
`@@ -242,6 +260,7 @@ inputs:`
`242`	`260`	`state:`
`243`	`261`	`api_token: ${SECRET_0}`
`244`	`262`	`batch_size: 100`
	`263`	`+ initial_interval: 1h`
`245`	`264`	`site_ids: "123"`
`246`	`265`	`tags:`
`247`	`266`	`- preserve_original_event`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@ resource.url: {{url}}`
`18`	`18`	`state:`
`19`	`19`	`api_token: {{api_token}}`
`20`	`20`	`batch_size: {{batch_size}}`
	`21`	`+ initial_interval: {{initial_interval}}`
`21`	`22`	`{{#if site_ids }}`
`22`	`23`	`site_ids: !!str {{site_ids}}`
`23`	`24`	`{{/if}}`
`@@ -35,8 +36,17 @@ program: \|-`
`35`	`36`	`state.url.trim_right("/") + "/web/api/v2.1/threats?" + {`
`36`	`37`	`"skipCount": ["true"],`
`37`	`38`	`"limit": [string(state.batch_size)],`
	`39`	`+ "sortBy": ["updatedAt"],`
	`40`	`+ "sortOrder": ["asc"],`
	`41`	`+ "updatedAt__gte": [`
	`42`	`+ state.?cursor.?list_updated_at_gte.orValue(`
	`43`	`+ state.?cursor.?last_timestamp.orValue(`
	`44`	`+ (now - duration(state.initial_interval)).format(time_layout.RFC3339)`
	`45`	`+ )`
	`46`	`+ )`
	`47`	`+ ],`
`38`	`48`	`?"siteIds": state.?site_ids.optMap(v, [string(v)]),`
`39`		`- ?"cursor": state.?cursor.next_page.token.optMap(v, [v]),`
	`49`	`+ ?"cursor": state.?cursor.?next_page.token.optMap(v, [v]),`
`40`	`50`	`}.format_query()`
`41`	`51`	`).with(`
`42`	`52`	`{`
`@@ -53,6 +63,12 @@ program: \|-`
`53`	`63`	`?"token": (body.?pagination.nextCursor.orValue(null) != null) ? optional.of(body.pagination.nextCursor) : optional.none(),`
`54`	`64`	`},`
`55`	`65`	`"fetch_more": body.?pagination.nextCursor.orValue(null) != null,`
	`66`	`+ "list_updated_at_gte": state.?cursor.?list_updated_at_gte.orValue(`
	`67`	`+ state.?cursor.?last_timestamp.orValue(`
	`68`	`+ (now - duration(state.initial_interval)).format(time_layout.RFC3339)`
	`69`	`+ )`
	`70`	`+ ),`
	`71`	`+ ?"last_timestamp": state.?cursor.last_timestamp,`
`56`	`72`	`},`
`57`	`73`	`}`
`58`	`74`	`)`
`@@ -71,7 +87,6 @@ program: \|-`
`71`	`87`	`},`
`72`	`88`	`},`
`73`	`89`	`"want_more": false,`
`74`		`- "cursor": {},`
`75`	`90`	`}`
`76`	`91`	`)`
`77`	`92`	`)`
`@@ -121,6 +136,11 @@ program: \|-`
`121`	`136`	`?"token": has_more_events ? optional.of(body.pagination.nextCursor) : optional.none(),`
`122`	`137`	`},`
`123`	`138`	`"fetch_more": state.?cursor.fetch_more.orValue(false),`
	`139`	`+ "list_updated_at_gte": state.cursor.list_updated_at_gte,`
	`140`	`+ ?"last_timestamp": has_more_events ?`
	`141`	`+ state.?cursor.last_timestamp`
	`142`	`+ :`
	`143`	`+ state.cursor.worklist.data[0].?threatInfo.updatedAt.or(state.?cursor.last_timestamp),`
`124`	`144`	`},`
`125`	`145`	`}`
`126`	`146`	`)`
`@@ -147,7 +167,6 @@ program: \|-`
`147`	`167`	`{`
`148`	`168`	`"events": [],`
`149`	`169`	`"want_more": false,`
`150`		`- "cursor": {},`
`151`	`170`	`}`
`152`	`171`	`)`
`153`	`172`	`)`