Merging events

Some source can generate many events that needs to be correlated. Loghub can identify correlated events using an index and merge them. The syntax for merging is:

merge {
    if: a groovy expression
    index: "a string",
    seeds: {
        "astring": specialvalue,
        ...
    },
    forward: boolean,
    expiration: integer,
    default: specialvalue,
}

Metadata fields are not merged.

The field if define a groovy expression, if it returns true, a merge will be tried. If missing, all events passing to that step will be merged.

The field index define how to identify correlated index. It's a string formatted using Extended pattern.

The field expiration define the windows of time for merging events, in seconds.

The field forward define if events that matches the index are also forwarded, it defaults to false.

If non-null expression doFire will be tested against the constructed event. If it evaluates to true, the merged event will be sent to the pipeline onFire.

The pipeline onExpiration if present is executed if the timeout is reached.

The fields seeds, default and defaultMeta defines how fields are merged ; seeds contains a set of attribute for the event and define how each of them will be merged and default defines the merge comportement for not given attributes.

The values possible for default and seeds attributes:

• '<' only the first value is kept
• '>' only the last value is kep
• 'c' count the number of values for that field
• true values will be merged by a boolean AND
• false values will be merged by a boolean OR
• 0 values will be added together as integer
• 1 values will be multiplied as integer
• 0.0 values will be added together as double
• 1.0 values will be multiplied as double
• [] values will be added in a list
• {} values that are maps will be merged in a unique map; in case of multiple occurrence of a key, they will be merged in a list
• "anystring" values will be merged as a string using the given string as a separator ; it can be empty

For example, sendmail generate many lines of line for each message like:

Jan 26 16:25:01 mailer sendmail[12097]: v0QFP1F2012097: from=<nagios@monitoring.example.com>, size=991, class=0, nrcpts=1, msgid=<201701261525.v0QFP1Oa038905@monitoring.example.com>, proto=ESMTP, daemon=MTA, relay=monitoring.example.com [10.0.0.1]
Jan 26 16:25:01 mailer sendmail[12099]: v0QFP1F2012097: to=<equipe@example.com>, delay=00:00:00, xdelay=00:00:00, mailer=esmtp, pri=120991, relay= mailhostext.example.com. [10.0.0.2], dsn=2.0.0, stat=Sent (<201701261525.v0QFP1Oa038905@monitoring.example.com> [InternalId=179728146] Queued mail for delivery)

So following a mail can be hard. An example configuration for that can be

pipeline[syslog] {
...
    | [program] == "sendmail" ? $sendmail
...
}
pipeline[sendmail] {
    loghub.processors.Grok {
        pattern: "^(?:(?<sendmailmsgid>[0-9A-Za-z]+): )?%{GREEDYDATA:message}",
        field: "message",
    }
    | loghub.processors.Grok {
        pattern: " ?Milter add: header: %{GREEDYDATA:milterheader}",
        field: "message",
        success: { [message]- }
    }
    | loghub.processors.VarExtractor { 
        parser: "(?<name>\\p{Alnum}+)=(?<value>.+?)(?:(?:, )|$)",
        path: "sendmailvars",
        field: ".message",
    }
    | merge {
        index: "${sendmailmsgid%s}",
        seeds: {
            "sendmailvars": {},
             "message": [],
             "milterheader": [],
             "pid": [],
        },
        forward: false,
        timeout: 10,
        default: '<',
        inPipeline: "main"
    }
}

The examples lines will generate the following event:

{
    "date": "2017-01-26 16:25:01",
    "@timestamp": "2017-01-26T16:25:01.361+0100",
    "sendmailmsgid": "v0QFP1F2012097",
    "host": "mail400",
    "pid": "12097",
    "program": "sendmail",
    "logsource": "mailer",
    "syslog_pri": {
        "severity": "informational",
        "facility": "mail"
    },
    "sendmailvars": {
        "stat": "Sent (<201701261525.v0QFP1Oa038905@monitoring.example.com> [InternalId=179728146] Queued mail for delivery)",
        "pri": "120991",
        "relay": [
            "monitoring.example.com [10.0.0.1]",
            "mailhostext.example.com. [10.0.0.2]"
        ],
        "msgid": "<201701261525.v0QFP1Oa038905@monitoring.example.com>",
        "mailer": "esmtp",
        "daemon": "MTA",
        "delay": "00:00:00",
        "size": "991",
        "proto": "ESMTP",
        "from": "<nagios@monitoring.example.com>",
        "nrcpts": "1",
        "to": "<equipe@example.com>",
        "class": "0",
        "xdelay": "00:00:00",
        "dsn": "2.0.0"
    }
}