fix(cortex-xdr): the alerts endpoint never answered — wrong request dialect
get_alerts posted the incidents body, {filters, search_from, search_to, sort},
to alerts/get_alerts_by_filter_data/. That endpoint serves the alerts GRID and
speaks another dialect entirely, so every call — before this branch as much as
after it — came back a bare HTTP 500 with no hint as to why.
Shape taken from the reference client (demisto/content, Packs/ApiModules/
Scripts/CoreIRApiModule, get_alerts_by_filter_command):
request_data.filter_data = {
sort: [{FIELD, ORDER}], # a list, uppercase keys
paging: {from, to}, # not search_from/search_to
filter: {AND: [{SEARCH_FIELD, SEARCH_TYPE, SEARCH_VALUE}]},
}
Severity is an enum there (SEV_040_HIGH), and several severities are OR'd, not
passed as a list. The watermark is a RANGE, since the grid has no gte operator;
its upper bound carries five minutes of slack, because our clock and the
tenant's are not the same clock. A filterless query is bounded to the last
thirty days rather than sent empty — the reference client refuses one outright,
and the grid is not meant to be asked for a whole retention.
The response needed as much work as the request. Rows arrive wrapped as
{alert_fields, incident_fields}, and mapping through that wrapper would put an
alert_fields. prefix on every expression an operator writes, so each row is
unwrapped. Two of its fields cannot be mapped as they stand: severity is the
enum code, and status.progress carries a dot INSIDE the key, which no mapping
path can express. Both are derived into severity_name and status_progress.
The mapper follows the grid's own vocabulary — internal_id, alert_name,
agent_hostname, agent_ip_addresses — and dedup moves to internal_id, since
alert_id belongs to the other API. case_id is kept as the correlation UID: it
is the join back to the incident feed.
Verified end to end against the vendor's own recorded response
(test_data/get_alerts_by_filter_results.json): 33 of 54 OCSF entries resolve on
it, severity lands on 3, the detection anchor is set, and the paging walks
0-100, 100-200, 200-250 with the truncation flag raised only when the ceiling,
not the window, ended the fetch.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -2,7 +2,7 @@ id: cortex_xdr
|
||||
name: Cortex XDR
|
||||
version: 1.3.0
|
||||
description: "Palo Alto Cortex XDR (public API v1) — incident and alert ingestion + write-back, endpoint isolation/scan/delete/tagging, RTR scripts, hash block/allow lists, file quarantine/restore/retrieval, alert exclusions, external alert push (parsed/CEF), device-control violations, audits, distributions and RBAC/risk."
|
||||
changelog: "1.3.0 — Alert ingestion: get_alerts is now a fetch command (results path reply.alerts, dedup on alert_id, incremental on source_insert_ts) with a bundled OCSF mapper and a Cortex XDR Alert incident type, so detections can be ingested alongside — or instead of — incidents. The incident mapper is fixed on the way past: its severity expression compared strings, which the mapping engine cannot do, so every ingested incident silently took the rule's default severity; it also now carries a detection anchor so MTTD is measurable. It pages past the API's 100-results-per-call ceiling, and sorts oldest-first on an incremental fetch so a truncated window drops the alerts the next poll can still see. 1.2.1 — Connection troubleshooting: the URL is normalised to the tenant host (a pasted /public_api/v1 or console path no longer breaks the call), a non-JSON reply reports the status, content type and body instead of a bare JSON parse error, missing key/key ID is caught up front, nonce and timestamp are sent in both auth modes as the reference client does, and test_connection now probes get_incidents. 1.2.0 — Incident write-back (update_incident: status/severity/assignment/resolve comment) and external alert push (insert_parsed_alerts, insert_cef_alerts). 1.1.0 — Full command coverage: added delete/alias/tag endpoints, abort scan, original alerts, script metadata/code/snippet/exec-status, file retrieval (+details), alert exclusions, device-control violations, audits, distribution url/status/create and RBAC (users, roles, groups, risk score, risky users/hosts). 1.0.0 — Initial release: incident ingestion (get_incidents) with OCSF mapper, endpoints, isolate/unisolate, scan, hash blocklist/allowlist, quarantine/restore, run script + results, alerts retrieval, distributions and action status. Standard or Advanced API authentication."
|
||||
changelog: "1.3.0 — Alert ingestion, and the alerts endpoint answers at last: get_alerts was sending the incidents dialect ({filters, search_from, search_to, sort}) to a grid endpoint that speaks request_data.filter_data (SEARCH_FIELD/SEARCH_TYPE/SEARCH_VALUE blocks, paging.from/to, sort as a list), and every call came back HTTP 500. Body rebuilt from the reference client, rows unwrapped out of their alert_fields envelope, severity code and the dotted status.progress key derived into readable fields. Alert ingestion: get_alerts is now a fetch command (results path reply.alerts, dedup on alert_id, incremental on source_insert_ts) with a bundled OCSF mapper and a Cortex XDR Alert incident type, so detections can be ingested alongside — or instead of — incidents. The incident mapper is fixed on the way past: its severity expression compared strings, which the mapping engine cannot do, so every ingested incident silently took the rule's default severity; it also now carries a detection anchor so MTTD is measurable. It pages past the API's 100-results-per-call ceiling, and sorts oldest-first on an incremental fetch so a truncated window drops the alerts the next poll can still see. 1.2.1 — Connection troubleshooting: the URL is normalised to the tenant host (a pasted /public_api/v1 or console path no longer breaks the call), a non-JSON reply reports the status, content type and body instead of a bare JSON parse error, missing key/key ID is caught up front, nonce and timestamp are sent in both auth modes as the reference client does, and test_connection now probes get_incidents. 1.2.0 — Incident write-back (update_incident: status/severity/assignment/resolve comment) and external alert push (insert_parsed_alerts, insert_cef_alerts). 1.1.0 — Full command coverage: added delete/alias/tag endpoints, abort scan, original alerts, script metadata/code/snippet/exec-status, file retrieval (+details), alert exclusions, device-control violations, audits, distribution url/status/create and RBAC (users, roles, groups, risk score, risky users/hosts). 1.0.0 — Initial release: incident ingestion (get_incidents) with OCSF mapper, endpoints, isolate/unisolate, scan, hash blocklist/allowlist, quarantine/restore, run script + results, alerts retrieval, distributions and action status. Standard or Advanced API authentication."
|
||||
category: endpoint
|
||||
|
||||
# Per-instance configuration. The base URL is the tenant API root, e.g.
|
||||
@@ -83,18 +83,20 @@ commands:
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_alerts
|
||||
name: cortex-xdr-get-alerts
|
||||
description: "Fetch Cortex XDR alerts for ingestion (get_alerts_by_filter_data). Returns {reply:{alerts:[...]}}; use reply.alerts as the alert rule results path. Alerts are the detection layer under incidents: ingest them alongside get_incidents when the SOC works detections, not only aggregates."
|
||||
description: "Fetch Cortex XDR alerts for ingestion (get_alerts_by_filter_data). Returns {reply:{alerts:[...]}}; use reply.alerts as the alert rule results path. Each row is unwrapped out of the API's alert_fields envelope and carries a readable severity_name and status_progress, so alerts-grid field names (internal_id, alert_name, agent_hostname) are what mapping expressions see. Alerts are the detection layer under incidents: ingest them alongside get_incidents when the SOC works detections, not only aggregates."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
severity: { type: string, description: "Comma-separated severities (informational, low, medium, high, critical)" }
|
||||
created_after: { type: string, description: "Lower bound on alert source_insert_ts, ISO8601 or epoch ms (incremental fetch watermark)" }
|
||||
limit: { type: number, description: "Maximum alerts to fetch (default 100). The API serves 100 per call at most; above that the script pages until the limit is reached." }
|
||||
# Left unfiltered, the call is bounded to the last 30 days: the alerts
|
||||
# grid is not meant to be asked for a tenant's whole retention.
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
ingest:
|
||||
results_path: reply.alerts
|
||||
dedup_key: alert_id
|
||||
dedup_key: internal_id
|
||||
incremental_field: created_after
|
||||
- id: insert_parsed_alerts
|
||||
name: cortex-xdr-insert-parsed-alerts
|
||||
|
||||
@@ -1,76 +1,87 @@
|
||||
name: "Cortex XDR Alerts → OCSF"
|
||||
description: "Maps a Cortex XDR alert (alerts/get_alerts_by_filter_data/, results_path = reply.alerts) to OCSF Detection Finding fields. Alerts are the per-detection layer under incidents: one incident aggregates many alerts, so a tenant ingesting both feeds will hold the same detection twice, once as an aggregate and once on its own."
|
||||
description: "Maps one Cortex XDR alert (alerts/get_alerts_by_filter_data/, results_path = reply.alerts) to OCSF Detection Finding fields. Field names are the alerts-grid ones (internal_id, alert_name, agent_hostname…), not the incident ones; the script unwraps the API's alert_fields envelope and derives severity_name and status_progress, which the raw payload cannot express. Alerts are the detection layer under incidents: a tenant ingesting both feeds holds each detection twice, once inside an aggregate and once on its own."
|
||||
field_mappings:
|
||||
title: "name"
|
||||
description: "description"
|
||||
# The raw string is enough: Riposte reads critical/high/medium/low/informational
|
||||
# onto its 1-5 scale itself, and the mapping engine cannot compare strings.
|
||||
severity: "severity"
|
||||
title: "alert_name"
|
||||
description: "alert_description"
|
||||
# severity_name, not severity: the API sends an enum code (SEV_040_HIGH) that
|
||||
# no severity scale can read, so the script carries the plain name alongside it.
|
||||
severity: "severity_name"
|
||||
# Which sensor fired: "XDR Agent", "PAN NGFW", "XDR Analytics"…
|
||||
source: "source"
|
||||
source: "alert_source"
|
||||
# results_path = reply.alerts; source_path is JSONata over ONE alert object.
|
||||
# Paths absent from a given alert are skipped at ingestion, so entries for fields
|
||||
# a tenant never emits are safe. Where two entries target the same OCSF field,
|
||||
# the LAST non-empty one wins — that is how the fallbacks below are ordered.
|
||||
# the LAST non-empty one wins — that is how the events[] fallbacks are ordered.
|
||||
ocsf:
|
||||
# ── Finding ───────────────────────────────────────────────────────
|
||||
- { source_path: "alert_id", ocsf_field: "finding_info.uid" }
|
||||
- { source_path: "internal_id", ocsf_field: "finding_info.uid" }
|
||||
- { source_path: "external_id", ocsf_field: "finding_info.uid_alt" }
|
||||
- { source_path: "name", ocsf_field: "finding_info.title" }
|
||||
- { source_path: "description", ocsf_field: "finding_info.desc" }
|
||||
- { source_path: "detection_timestamp", ocsf_field: "finding_info.created_time" }
|
||||
- { source_path: "alert_name", ocsf_field: "finding_info.title" }
|
||||
- { source_path: "alert_description", ocsf_field: "finding_info.desc" }
|
||||
- { source_path: "source_insert_ts", ocsf_field: "finding_info.created_time" }
|
||||
- { source_path: "local_insert_ts", ocsf_field: "finding_info.modified_time" }
|
||||
- { source_path: "category", ocsf_field: "finding_info.analytic.category" }
|
||||
- { source_path: "name", ocsf_field: "finding_info.analytic.name" }
|
||||
- { source_path: "filter_rule_id", ocsf_field: "finding_info.analytic.uid" }
|
||||
- { source_path: "alert_category", ocsf_field: "finding_info.analytic.category" }
|
||||
- { source_path: "alert_name", ocsf_field: "finding_info.analytic.name" }
|
||||
- { source_path: "matching_service_rule_id", ocsf_field: "finding_info.analytic.uid" }
|
||||
# ── Detection time: the MTTD anchor ───────────────────────────────
|
||||
# `time` is what Riposte measures detection-to-ingestion against. Ingestion
|
||||
# time (source_insert_ts) is the honest fallback when the sensor sent no
|
||||
# detection timestamp of its own — it still beats leaving MTTD empty.
|
||||
# `time` is what Riposte measures detection-to-ingestion against. The grid
|
||||
# exposes when the tenant took the alert in (source_insert_ts); a sensor-side
|
||||
# detection timestamp, when the tenant sends one, is the better anchor and
|
||||
# comes last so it wins.
|
||||
- { source_path: "source_insert_ts", ocsf_field: "time" }
|
||||
- { source_path: "detection_timestamp", ocsf_field: "time" }
|
||||
# ── Alert state ───────────────────────────────────────────────────
|
||||
- { source_path: "severity", ocsf_field: "severity" }
|
||||
- { source_path: "alert_type", ocsf_field: "activity_name" }
|
||||
- { source_path: "action_pretty", ocsf_field: "action" }
|
||||
- { source_path: "action", ocsf_field: "action" }
|
||||
- { source_path: "resolution_status", ocsf_field: "status" }
|
||||
- { source_path: "resolution_comment", ocsf_field: "status_detail" }
|
||||
- { source_path: "severity_name", ocsf_field: "severity" }
|
||||
- { source_path: "alert_domain", ocsf_field: "activity_name" }
|
||||
- { source_path: "alert_action_status", ocsf_field: "action" }
|
||||
- { source_path: "status_progress", ocsf_field: "status" }
|
||||
- { source_path: "matching_status", ocsf_field: "status_detail" }
|
||||
- { source_path: "events_length", ocsf_field: "count" }
|
||||
# The XDR case this alert was folded into — the join back to the incident feed.
|
||||
- { source_path: "case_id", ocsf_field: "metadata.correlation_uid" }
|
||||
# ── Product identity ──────────────────────────────────────────────
|
||||
- { source_path: "'Cortex XDR'", ocsf_field: "metadata.product.name" }
|
||||
- { source_path: "'Palo Alto Networks'", ocsf_field: "metadata.product.vendor_name" }
|
||||
- { source_path: "source", ocsf_field: "metadata.log_source" }
|
||||
- { source_path: "'Cortex XDR'", ocsf_field: "metadata.product.name" }
|
||||
- { source_path: "'Palo Alto Networks'", ocsf_field: "metadata.product.vendor_name" }
|
||||
- { source_path: "alert_source", ocsf_field: "metadata.log_source" }
|
||||
# ── MITRE ATT&CK ──────────────────────────────────────────────────
|
||||
# Both fields arrive as a list on most tenants and as a bare string on some;
|
||||
# [0] reads the first element either way.
|
||||
- { source_path: "mitre_tactic_id_and_name[0]", ocsf_field: "attacks.tactic.name" }
|
||||
- { source_path: "mitre_technique_id_and_name[0]", ocsf_field: "attacks.technique.name" }
|
||||
# ── Affected endpoint ─────────────────────────────────────────────
|
||||
- { source_path: "host_name", ocsf_field: "device.hostname" }
|
||||
- { source_path: "host_ip[0]", ocsf_field: "device.ip" }
|
||||
- { source_path: "mac", ocsf_field: "device.mac" }
|
||||
- { source_path: "endpoint_id", ocsf_field: "device.uid" }
|
||||
- { source_path: "agent_device_domain", ocsf_field: "device.domain" }
|
||||
- { source_path: "agent_os_type", ocsf_field: "device.os.type" }
|
||||
- { source_path: "agent_os_sub_type", ocsf_field: "device.os.name" }
|
||||
- { source_path: "agent_hostname", ocsf_field: "device.hostname" }
|
||||
- { source_path: "agent_ip_addresses[0]", ocsf_field: "device.ip" }
|
||||
- { source_path: "agent_id", ocsf_field: "device.uid" }
|
||||
- { source_path: "agent_os_type", ocsf_field: "device.os.type" }
|
||||
# Mirrored onto src_endpoint so routers and pre-processing rules written for
|
||||
# the incident feed (which maps hosts there) match alerts unchanged.
|
||||
- { source_path: "host_name", ocsf_field: "src_endpoint.hostname" }
|
||||
- { source_path: "host_ip[0]", ocsf_field: "src_endpoint.ip" }
|
||||
- { source_path: "user_name", ocsf_field: "user.name" }
|
||||
# ── First triggering event (events[] carries the per-event detail) ─
|
||||
- { source_path: "agent_hostname", ocsf_field: "src_endpoint.hostname" }
|
||||
- { source_path: "agent_ip_addresses[0]", ocsf_field: "src_endpoint.ip" }
|
||||
- { source_path: "actor_effective_username", ocsf_field: "user.name" }
|
||||
# ── What actually happened ────────────────────────────────────────
|
||||
# Grid columns first, then the same detail read off the first event, which is
|
||||
# where a tenant that does not flatten these columns puts them.
|
||||
- { source_path: "actor_process_image_name", ocsf_field: "process.name" }
|
||||
- { source_path: "actor_process_command_line", ocsf_field: "process.cmd_line" }
|
||||
- { source_path: "actor_process_image_sha256", ocsf_field: "process.file.hashes.sha256" }
|
||||
- { source_path: "causality_actor_process_command_line", ocsf_field: "process.parent_process.cmd_line" }
|
||||
- { source_path: "action_file_path", ocsf_field: "file.path" }
|
||||
- { source_path: "action_file_sha256", ocsf_field: "file.hashes.sha256" }
|
||||
- { source_path: "action_file_md5", ocsf_field: "file.hashes.md5" }
|
||||
- { source_path: "action_registry_key_name", ocsf_field: "reg_key.path" }
|
||||
- { source_path: "action_registry_data", ocsf_field: "reg_value.data" }
|
||||
- { source_path: "action_local_ip", ocsf_field: "src_endpoint.ip" }
|
||||
- { source_path: "action_local_port", ocsf_field: "src_endpoint.port" }
|
||||
- { source_path: "action_remote_ip", ocsf_field: "dst_endpoint.ip" }
|
||||
- { source_path: "action_remote_port", ocsf_field: "dst_endpoint.port" }
|
||||
- { source_path: "dst_action_external_hostname", ocsf_field: "dst_endpoint.hostname" }
|
||||
- { source_path: "events[0].actor_process_image_name", ocsf_field: "process.name" }
|
||||
- { source_path: "events[0].actor_process_command_line", ocsf_field: "process.cmd_line" }
|
||||
- { source_path: "events[0].actor_process_image_path", ocsf_field: "process.path" }
|
||||
- { source_path: "events[0].actor_process_os_pid", ocsf_field: "process.pid" }
|
||||
- { source_path: "events[0].actor_process_image_sha256", ocsf_field: "process.file.hashes.sha256" }
|
||||
- { source_path: "events[0].causality_actor_process_image_name", ocsf_field: "process.parent_process.name" }
|
||||
- { source_path: "events[0].causality_actor_process_command_line", ocsf_field: "process.parent_process.cmd_line" }
|
||||
- { source_path: "events[0].causality_actor_process_image_name", ocsf_field: "process.parent_process.name" }
|
||||
- { source_path: "events[0].action_file_path", ocsf_field: "file.path" }
|
||||
- { source_path: "events[0].action_file_sha256", ocsf_field: "file.hashes.sha256" }
|
||||
- { source_path: "events[0].action_file_md5", ocsf_field: "file.hashes.md5" }
|
||||
- { source_path: "events[0].action_remote_ip", ocsf_field: "dst_endpoint.ip" }
|
||||
- { source_path: "events[0].action_remote_port", ocsf_field: "dst_endpoint.port" }
|
||||
- { source_path: "events[0].action_external_hostname", ocsf_field: "dst_endpoint.hostname" }
|
||||
- { source_path: "events[0].action_local_ip", ocsf_field: "src_endpoint.ip" }
|
||||
- { source_path: "events[0].action_local_port", ocsf_field: "src_endpoint.port" }
|
||||
|
||||
@@ -76,51 +76,120 @@ def to_ms(v):
|
||||
return None
|
||||
|
||||
|
||||
# The API serves at most 100 alerts per call. Ingestion is what makes that a
|
||||
# real ceiling: an alert feed easily carries more than a hundred detections
|
||||
# between two polls, and anything a single page leaves behind is never fetched
|
||||
# again — the next run's watermark has already moved past it.
|
||||
# This endpoint speaks the alerts-grid dialect, NOT the incidents one: a body of
|
||||
# {filters, search_from, search_to, sort} — what incidents/get_incidents/ takes —
|
||||
# is answered with a bare HTTP 500. It wants request_data.filter_data with
|
||||
# SEARCH_FIELD/SEARCH_TYPE/SEARCH_VALUE blocks, paging.from/to and a sort LIST.
|
||||
# Shape taken from the reference client (demisto/content,
|
||||
# Packs/ApiModules/Scripts/CoreIRApiModule — get_alerts_by_filter_command).
|
||||
PAGE = 100
|
||||
|
||||
# Severity travels as an enum code both ways. Riposte reads plain names onto its
|
||||
# 1-5 scale, so alerts carry `severity_name` alongside the raw code.
|
||||
SEVERITY_CODE_TO_NAME = {
|
||||
"SEV_010_INFO": "informational",
|
||||
"SEV_020_LOW": "low",
|
||||
"SEV_030_MEDIUM": "medium",
|
||||
"SEV_040_HIGH": "high",
|
||||
"SEV_050_CRITICAL": "critical",
|
||||
}
|
||||
SEVERITY_NAME_TO_CODE = dict((v, k) for k, v in SEVERITY_CODE_TO_NAME.items())
|
||||
SEVERITY_NAME_TO_CODE["info"] = "SEV_010_INFO"
|
||||
|
||||
# Our clock and the tenant's are not the same clock. A range that ends exactly
|
||||
# now silently drops alerts the tenant stamped a few seconds ahead of us.
|
||||
SKEW_MS = 5 * 60 * 1000
|
||||
|
||||
# Window applied when the caller passes no filter at all — see main().
|
||||
DEFAULT_LOOKBACK_MS = 30 * 24 * 60 * 60 * 1000
|
||||
|
||||
|
||||
def severity_block(value):
|
||||
"""One EQ block per severity, OR'd together (the reference client's array rule)."""
|
||||
blocks = [
|
||||
{"SEARCH_FIELD": "severity", "SEARCH_TYPE": "EQ",
|
||||
"SEARCH_VALUE": SEVERITY_NAME_TO_CODE.get(s.lower(), s.upper())}
|
||||
for s in csv(value)
|
||||
]
|
||||
if not blocks:
|
||||
return None
|
||||
return blocks[0] if len(blocks) == 1 else {"OR": blocks}
|
||||
|
||||
|
||||
def flatten(item):
|
||||
"""One grid row -> one flat alert.
|
||||
|
||||
The API wraps every row as {alert_fields, incident_fields}. Mapping through
|
||||
that wrapper would put an `alert_fields.` prefix on every expression an
|
||||
operator writes, so the row is unwrapped here and the two fields Riposte
|
||||
cannot express are derived: `status.progress` carries a dot INSIDE the key
|
||||
(unusable as a mapping path) and severity is an enum code.
|
||||
"""
|
||||
fields = item.get("alert_fields")
|
||||
alert = dict(fields) if isinstance(fields, dict) else dict(item)
|
||||
alert.pop("incident_fields", None)
|
||||
if "status.progress" in alert:
|
||||
alert["status_progress"] = alert.pop("status.progress")
|
||||
name = SEVERITY_CODE_TO_NAME.get(alert.get("severity"))
|
||||
if name:
|
||||
alert["severity_name"] = name
|
||||
incident = item.get("incident_fields")
|
||||
if isinstance(incident, dict):
|
||||
alert["incident_fields"] = incident
|
||||
return alert
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
limit = max(1, int(inputs.get("limit") or 100))
|
||||
filters = []
|
||||
if inputs.get("severity"):
|
||||
filters.append({"field": "severity", "operator": "in", "value": csv(inputs["severity"])})
|
||||
conditions = []
|
||||
sev = severity_block(inputs.get("severity"))
|
||||
if sev:
|
||||
conditions.append(sev)
|
||||
created_ms = to_ms(inputs.get("created_after"))
|
||||
if created_ms is not None:
|
||||
filters.append({"field": "source_insert_ts", "operator": "gte", "value": created_ms})
|
||||
conditions.append({
|
||||
"SEARCH_FIELD": "source_insert_ts",
|
||||
"SEARCH_TYPE": "RANGE",
|
||||
"SEARCH_VALUE": {"from": created_ms, "to": int(time.time() * 1000) + SKEW_MS},
|
||||
})
|
||||
if not conditions:
|
||||
# The reference client refuses a filterless query outright, and an
|
||||
# unbounded scan of the whole alerts grid is not what the API is for.
|
||||
# A recent window is a better default than an empty filter the tenant
|
||||
# may well answer with a 500.
|
||||
now_ms = int(time.time() * 1000)
|
||||
conditions.append({
|
||||
"SEARCH_FIELD": "source_insert_ts",
|
||||
"SEARCH_TYPE": "RANGE",
|
||||
"SEARCH_VALUE": {"from": now_ms - DEFAULT_LOOKBACK_MS, "to": now_ms + SKEW_MS},
|
||||
})
|
||||
# Oldest first on an incremental fetch, so that a window holding more alerts
|
||||
# than `limit` drops its most RECENT ones — the only ones the next poll can
|
||||
# still see. Newest first otherwise, which is what an operator running the
|
||||
# command by hand is asking for.
|
||||
keyword = "asc" if created_ms is not None else "desc"
|
||||
order = "ASC" if created_ms is not None else "DESC"
|
||||
|
||||
alerts, total = [], None
|
||||
alerts, truncated = [], False
|
||||
while len(alerts) < limit:
|
||||
rd = {
|
||||
"search_from": len(alerts),
|
||||
"search_to": min(len(alerts) + PAGE, limit),
|
||||
"sort": {"field": "source_insert_ts", "keyword": keyword},
|
||||
}
|
||||
if filters:
|
||||
rd["filters"] = filters
|
||||
rd = {"filter_data": {
|
||||
"sort": [{"FIELD": "source_insert_ts", "ORDER": order}],
|
||||
"paging": {"from": len(alerts), "to": min(len(alerts) + PAGE, limit)},
|
||||
"filter": {"AND": conditions},
|
||||
}}
|
||||
reply = (post("/alerts/get_alerts_by_filter_data/", rd) or {}).get("reply") or {}
|
||||
page = reply.get("alerts") or []
|
||||
if total is None:
|
||||
total = reply.get("total_count")
|
||||
alerts.extend(page)
|
||||
alerts.extend(flatten(a) for a in page)
|
||||
if len(page) < PAGE:
|
||||
break
|
||||
# Stopped on the ceiling rather than on an exhausted window: whatever is
|
||||
# left is not coming back on the next poll, and a silent cap reads like
|
||||
# a quiet feed.
|
||||
truncated = len(alerts) >= limit
|
||||
|
||||
out = {"result_count": len(alerts), "alerts": alerts}
|
||||
if total is not None:
|
||||
out["total_count"] = total
|
||||
# Say it when the window was larger than the limit: those alerts are not
|
||||
# coming back on the next poll, and a silent cap reads like a quiet feed.
|
||||
out["truncated"] = total > len(alerts)
|
||||
if truncated:
|
||||
out["truncated"] = True
|
||||
print(json.dumps({"reply": out}))
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user