Pipeline Health Monitor: Ensuring End-to-End Event Flow in QRadar

In a SOC, "silence" can be a bad thing but it can also be misleading. If logs stop flowing from a managed host, how do you know if it's just a quiet evening, if the network is blocked, or if the QRadar pipeline itself has failed? How do you know if the silence is normal or expected behavior?

To solve this, we can implement a Pipeline Health Monitor. This is a synthetic offense generator designed to validate the end-to-end health of your data pipeline, from ingestion and correlation to offense creation, on a per host basis. This helps confirm if the silence in offenses is from lack of events or the system acting up.

Step-by-Step Implementation Guide

Deploy the Script on Managed Hosts

On each managed host, save the following script (e.g., as /opt/monitor/pipeline_check.sh) and make it executable (with chmod +x):

#!/bin/bash
UNIQUE_ID="monitor_$(cat /proc/sys/kernel/random/uuid)"

printf "<14>%s $(hostname)_check  %s QRadarMonitor: SYNTHETIC_MONITOR_EVENT src=127.0.0.1 user=${UNIQUE_ID} msg=pipeline_health_check\n" \
  "$(date '+%b %d %H:%M:%S')" \
  "${UNIQUE_ID}" \
| nc -u -w1 127.0.0.1 514

What the script does:

  • UUID Generation: Creates a collision proof ID from the kernel on every run so we have something unique to index the offense with
  • Log Source Identifier (LSI): The LSI will be the device's hostname, and _check. The hostname by itself might cause the event to go to the wrong log source, or for logs to incorrectly go to the one we generate below.
  • Custom Property: Embeds the UUID into a specific field that QRadar will extract for offense indexing.

Setup Cronjob:
Add a crontab entry to run every hour (using your filepath):

0 * * * * /opt/monitor/pipeline_check.sh

Create a Custom Event Property (CEP)

You need to tell QRadar how to extract the UUID from the incoming message.

  1. Navigate to Admin → Custom Event Properties → New Property.
  2. Configure the following:
    • Property Name: Pipeline_UUID
    • Property Type: Regex
    • Optimized for rules: [Check]
    • Extraction Regex: user=(monitor_[a-f0-9\-]{36})
    • Capture Group: 1

Create a Log Source for Each Managed Host

To ensure the logs are categorized correctly, you must create a log source for each host you intend to monitor. This might be one EC per domain, or every host.

  1. Navigate to Admin → Log Sources → Add Log Source.
  2. Configure the following (repeat per host):
    • Log Source Name: [Hostname]_Health_Monitor
    • Log Source Type: Universal DSM
    • Protocol: Syslog
    • Log Source Identifier (LSI): This must match the format produced by the script (e.g., hostname_check. In my testing environment, this was qradarcon.local_check but will of course be different in your deployment and per host).

Create the Offense / Event Rule

Finally, create the rule that generates the offense.

  1. Navigate to Offenses → Rules → New Event Rule.
  2. Configure the following:
    • Rule Condition: "When the event(s) were detected by one or more of these log sources: [Select all your monitor log sources]"
      • Ensure the detected event is part of an offense: [Checked]
      • Index offense by: Custom Property → Pipeline UUID

By indexing the offense by the unique UUID, QRadar is prevented from "merging" new events into an old offense. This should guarantee a brand-new offense for every single script execution, the likelihood of duplicate UUIDs is so insanely low it would be impressive if it occurred.

Expected Behavior

Once configured, you have a way of checking pipeline and offense health automatically.

  1. Script Runs: A new UUID is generated and sent to QRadar.
  2. Ingestion: QRadar matches the LSI to the correct host.
  3. Extraction: The CEP extracts the unique UUID.
  4. Alerting: The correlation rule fires and creates a unique offense.

If no offense is generated after the cronjob ran, then something is up.

Share