Hi.
I want to create an Alert Rule using the elasticsearch query which contains an ES QL query with 4 fork branches, each fork running a different query and collecting STATS on different fields.
This works all well.
Now I only want to send an email via the action mail connector IF the count of errors in FORK1 is higher than 100, and then rows for all 4 FORK branches should be included in the email body.
But I cannot seem to get this right.
If I add a WHERE clause at the very end to determine if error5xx count is higher than my defined value, it will send the email but only contains the rows for FORK1.
Below is the full snippet of my ES QL query.
I would greatly appreciate if anyone can recommend or advice on how to accomplish conditional email action?
FROM eslive*:logs-*
| FORK
( // FORK query for http error 5xx
**WHERE** http.response.status_code >= 500
| **WHERE** iitech.saxotrader.data.url NOT LIKE "\*openapi/platform/v1/maintenance/notices"
| **WHERE** url.domain != "employee.saxotrader.mid.dom"
| **STATS** count = COUNT(\*), error5xx = COUNT(\*) **BY** event.action
| **SORT** count DESC
| **LIMIT** 5
)
( // FORK query for service.namespace has error
**WHERE** TO_LOWER(log.level) == "error"
| **WHERE** message NOT LIKE "Health check Custom Readiness Checks\*"
| **WHERE** data_stream.namespace NOT IN ("business_activity_monitor_grafana","baffle","flux_system","elastic", "system", "supaphly_svc","af","unified_client_registration","devops_releasevalidation_logs","monitoring","ssoc","marketdata_mdmdashboard_svc","fo_riskconfig","morningstar_documentfeed_svc","internal_chatbot","openapi_pce","kube_system","supaphly_pcs","semcmsweb_svc","mcs_svc","semtrading_svc","mcs_callstats","fo_fos","oapiliquiditymgmt","tradingplatform","crm_svc","openapi_svc","fo_foecs", "oapidevportal")
| **WHERE** log.logger IS NULL OR log.logger NOT LIKE "\*ObjectPoolStatisticsLogger\*"
| **WHERE** data_stream.dataset NOT IN ("system.application", "system.security", "system.system")
| **WHERE** url.domain IS NULL OR url.domain != "employee.saxotrader.mid.dom"
| **WHERE** message NOT LIKE "\*No FxVolatilitySurface found in Price Feed\*"
| **WHERE** message NOT LIKE "\*Failed to load the requested model name\*"
| **WHERE** log.logger IS NULL OR log.logger NOT LIKE "\*ObjectPoolStatisticsLogger\*"
| **STATS** count = COUNT(message), service_error = COUNT(\*) **BY** data_stream.namespace, message, service.name, process.name
| **SORT** count DESC
| **LIMIT** 10
)
( // FORK query for service.namespace has warnings with error message
**WHERE** data_stream.namespace NOT IN ("business_activity_monitor_grafana","baffle","flux_system","elastic", "system", "supaphly_svc","af", "platformknowledgebase","unified_client_registration","devops_releasevalidation_logs","monitoring","ssoc","marketdata_mdmdashboard_svc","fo_riskconfig","morningstar_documentfeed_svc","internal_chatbot","openapi_pce","kube_system","supaphly_pcs","semcmsweb_svc","mcs_svc","semtrading_svc","mcs_callstats","fo_fos","oapiliquiditymgmt","tradingplatform","crm_svc","openapi_svc","fo_foecs","oapiacchub", "datahubkafka", "cloudgraph","order_management","assetmanagement", "oapidevportal")
| **WHERE** log.logger IS NULL OR log.logger NOT LIKE "\*ObjectPoolStatisticsLogger\*"
| **WHERE** data_stream.dataset NOT IN ("system.application", "system.security", "system.system")
| **WHERE** url.domain IS NULL OR url.domain != "employee.saxotrader.mid.dom"
| **WHERE** message NOT LIKE "\*No FxVolatilitySurface found in Price Feed\*"
| **WHERE** message NOT LIKE "\*Failed to load the requested model name\*"
| **WHERE** TO_LOWER(log.level) == "warn" OR TO_LOWER(log.level) == "warning"
| **WHERE** error.message IS NOT NULL
| **STATS** count = COUNT(\*), service_warning = COUNT(\*) **BY** data_stream.namespace, message, service.name, process.name, error.message
| **RENAME** error.message AS Service_Warning_Message
| **SORT** count DESC
| **LIMIT** 10
)
( // FORK query for REDIS connection errors
**WHERE** error.type == "StackExchange.Redis.RedisConnectionException" AND message NOT LIKE "\*Valkey ConnectionFailure\*"
| **STATS** count = COUNT(\*), redis_error = COUNT(\*) **BY** kubernetes.namespace, kubernetes.container.name, kubernetes.pod.name, message, url.domain, url.path
)
| EVAL source = CASE(_fork == "fork1", "status_code>500", _fork == "fork2", "Service_Errors",_fork == "fork3", "Service_Warnings", _fork == "fork4", "redis error", "unknown"), is_fork1 = CASE(_fork == "fork1", 1, 0), is_fork2 = CASE(_fork == "fork2", 1, 0),is_fork3 = CASE(_fork == "fork3", 1, 0), is_fork4 = CASE(_fork == "fork4", 1, 0)
// --- CUSTOM PADDING OPERATORS APPLIED HERE ---
| EVAL pad_action = LEFT(CONCAT(event.action, SPACE(100)), 100),
pad_namespace = **LEFT**(CONCAT(data_stream.namespace, SPACE(40)), 40),
pad_error = **LEFT**(CONCAT(message, SPACE(100)), 100),
pad_service_name = **LEFT**(CONCAT(service.name, SPACE(50)), 50),
pad_process_name = **LEFT**(CONCAT(process.name, SPACE(50)), 50),
pad_service_warning = **LEFT**(CONCAT(Service_Warning_Message, SPACE(100)), 100),
pad_k8namespace = **LEFT**(CONCAT(kubernetes.namespace, SPACE(40)), 40),
pad_k8container = **LEFT**(CONCAT(kubernetes.container.name, SPACE(40)), 40),
pad_k8pod = **LEFT**(CONCAT(kubernetes.pod.name, SPACE(40)), 40)
| KEEP source, pad_action, pad_namespace, pad_error, pad_service_name, pad_process_name, pad_k8namespace, pad_k8container, pad_k8pod, is_fork1, is_fork2, is_fork3, is_fork4, pad_service_warning, count, error5xx, service_error, service_warning, redis_error
| RENAME source AS http_error | RENAME pad_namespace AS Service_Namespace | RENAME pad_action AS Event_Action | RENAME pad_error AS Error_Message | RENAME pad_k8namespace AS K8_namespace | RENAME pad_k8container AS K8_container | RENAME pad_k8pod AS K8_pod | RENAME pad_service_name AS Service_Name | RENAME pad_process_name AS Process_Name | RENAME pad_service_warning AS Service_Warning
| SORT http_error, count DESC
| LIMIT 100