Rules (XQL)
/*--------------------------------------------------------------------------------------
Cloud NSS Feeds Ingestion Rule
------------------------------------------------------------------------------------*/
/* This rule supports ingestion for Zscaler ZIA cloud NSS feeds, handling both cases
where the the raw input is a multiple event record array (JSON Array Notation enabled),
and the case where the raw input is a single JSON record (JSON Array Notation disabled).
The _time assignment is implemented only for event records which contain either an epoch
value (in seconds), or a combination of a formatted datetime string, along with it's
corresponding timezone, which has to be a valid IANA DB timezone.
If an event record lacks both an epoch value and a valid timezone / formatted datetime
string, then the log would be ingested, but without an explicit assignment to _time.
----------------------------------------------------------------------------------------*/
[INGEST:vendor="zscaler", product="cloudnss", target_dataset="zscaler_cloudnss_raw", no_hit=keep]
// use case 1: JSON Array Notation enabled, with a valid epoch / timestamp & timezone.
filter _raw_log -> [] != null
| call ZSCALER_NSS_FLATTEN_ARRAY_RECORDS
| call ZSCALER_NSS_EXTRACT_TIMESTAMP_FIELDS
| call ZSCALER_NSS_ASSERT_VALID_TIMESTAMP
| call ZSCALER_NSS_PARSE_TIMESTAMP
| call ZSCALER_NSS_REMOVE_REDUNDANT_FIELDS;
// use case 2: JSON Array Notation enabled, without a valid epoch / timestamp & timezone.
filter _raw_log -> [] != null
| call ZSCALER_NSS_FLATTEN_ARRAY_RECORDS
| call ZSCALER_NSS_EXTRACT_TIMESTAMP_FIELDS
| call ZSCALER_NSS_ASSERT_INVALID_TIMESTAMP
| call ZSCALER_NSS_REMOVE_REDUNDANT_FIELDS;
// use case 3: JSON Array Notation disabled, with a valid epoch / timestamp & timezone.
filter _raw_log -> [] = null
| alter _raw_log = to_string(to_json_string(event))
| call ZSCALER_NSS_EXTRACT_TIMESTAMP_FIELDS
| call ZSCALER_NSS_ASSERT_VALID_TIMESTAMP
| call ZSCALER_NSS_PARSE_TIMESTAMP
| call ZSCALER_NSS_REMOVE_REDUNDANT_FIELDS;
// use case 4: JSON Array Notation disabled, without a valid epoch / timestamp & timezone.
filter _raw_log -> [] = null
| alter _raw_log = to_string(to_json_string(event))
| call ZSCALER_NSS_EXTRACT_TIMESTAMP_FIELDS
| call ZSCALER_NSS_ASSERT_INVALID_TIMESTAMP
| call ZSCALER_NSS_REMOVE_REDUNDANT_FIELDS;
/*----------------------------------------------------------
Constants - regex patterns for supported datetime formats
----------------------------------------------------------*/
[CONST]
EPOCH_REGEX = "\d{10}"; // e.g. 1716797554 (seconds)
TIMESTAMP1_REGEX = "\d{4}\-\d{2}\-\d{2}\s+\d{2}\:\d{2}\:\d{2}"; // e.g. 2024-05-27 08:12:34
TIMESTAMP1_REGEX_CAPTURE = "(\d{4}\-\d{2}\-\d{2}\s+\d{2}\:\d{2}\:\d{2})";
TIMESTAMP2_REGEX = "\w{3}\s+\d{2}\s+\d{2}\:\d{2}\:\d{2}\s+\d{4}"; // e.g. "May 27 08:12:34 2024"
TIMESTAMP2_REGEX_CAPTURE = "(\w{3}\s+\d{2}\s+\d{2}\:\d{2}\:\d{2}\s+\d{4})";
/*---------
RULES
--------*/
[RULE:ZSCALER_NSS_FLATTEN_ARRAY_RECORDS]
alter _raw_log = _raw_log -> []
| arrayexpand _raw_log
| alter
sourcetype = _raw_log -> sourcetype,
_raw_log = to_string(_raw_log -> event{});
[RULE:ZSCALER_NSS_EXTRACT_TIMESTAMP_FIELDS]
alter // timestamp could be populated either as an epoch value (in seconds), or as a formatted string
tmp_tz = _raw_log -> tz,
tmp_epochtime = to_string(_raw_log -> epochtime),
tmp_datetime = to_string(coalesce(_raw_log -> time, _raw_log -> datetime))
| alter tmp_datetime = coalesce( // get rid of redundant preceding/trailing datetime parts such as day of the week
arrayindex(regextract(tmp_datetime, $TIMESTAMP1_REGEX_CAPTURE), 0),
arrayindex(regextract(tmp_datetime, $TIMESTAMP2_REGEX_CAPTURE), 0));
[RULE:ZSCALER_NSS_PARSE_TIMESTAMP]
alter _time = if(
tmp_epochtime != null, to_timestamp(to_integer(tmp_epochtime), "SECONDS"),
tmp_datetime ~= $TIMESTAMP1_REGEX, parse_timestamp("%F %T", tmp_datetime, tmp_tz),
tmp_datetime ~= $TIMESTAMP2_REGEX, parse_timestamp("%b %d %T %Y", tmp_datetime, tmp_tz));
[RULE:ZSCALER_NSS_REMOVE_REDUNDANT_FIELDS]
fields - tmp_*, event;
[RULE:ZSCALER_NSS_ASSERT_VALID_TIMESTAMP]
filter (tmp_epochtime ~= $EPOCH_REGEX and len(tmp_epochtime) = 10) or (tmp_tz != null and tmp_datetime != null);
[RULE:ZSCALER_NSS_ASSERT_INVALID_TIMESTAMP]
filter (tmp_epochtime !~= $EPOCH_REGEX or len(tmp_epochtime) != 10) and (tmp_tz = null or tmp_datetime = null);