Logging Windows event log information to Logstash using nxlog and JSON transport

logstash.conf

input {
    tcp {
        type => "syslog"
        host => "127.0.0.1"
        port => 3514
    }
    tcp {
        type   => "eventlog"
        host   => "10.1.1.2"
        port   => 3515
        format => 'json'
    }
}

# Details at http://cookbook.logstash.net/recipes/syslog-pri/
filter {

# Incoming data from rsyslog
    grok {
        type      => "syslog"
        pattern   => [ "<%{POSINT:syslog_pri}>(?:%{SYSLOGTIMESTAMP:syslog_timestamp}|%{TIMESTAMP_ISO8601:syslog_timestamp8601}) %{SYSLOGHOST:syslog_hostname} %{PROG:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" ]
        add_field => [ "received_at", "%{@timestamp}" ]
        add_field => [ "received_from", "%{@source_host}" ]
    }
    syslog_pri {
        type => "syslog"
    }
    date {
        type                 => "syslog"
        syslog_timestamp8601 => "ISO8601" # RSYSLOG_ForwardFormat
        syslog_timestamp     => [ "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
    mutate {
        type         => "syslog"
        exclude_tags => "_grokparsefailure"
        replace      => [ "@source_host", "%{syslog_hostname}" ]
        replace      => [ "@message", "%{syslog_message}" ]
    }
    mutate {
        type   => "syslog"
        remove => [ "syslog_hostname", "syslog_message", "syslog_timestamp", "syslog_timestamp8601" ]
    }

# Incoming Windows Event logs from nxlog
    # The EventReceivedTime field must contain only digits, or it is an invalid message
    grep {
        type              => "eventlog"
        EventReceivedTime => "\d+"
    }
    mutate {
        # Lowercase some values that are always in uppercase
        type      => "eventlog"
        lowercase => [ "EventType", "FileName", "Hostname", "Severity" ]
    }
    mutate {
        # Set source to what the message says
        type   => "eventlog"
        rename => [ "Hostname", "@source_host" ]
    }
    date {
        # Convert timestamp from integer in UTC
        type              => "eventlog"
        EventReceivedTime => "UNIX"
    }
    mutate {
        # Rename some fields into something more useful
        type   => "eventlog"
        rename => [ "Message", "@message" ]
        rename => [ "Severity", "eventlog_severity" ]
        rename => [ "SeverityValue", "eventlog_severity_code" ]
        rename => [ "Channel", "eventlog_channel" ]
        rename => [ "SourceName", "eventlog_program" ]
        rename => [ "SourceModuleName", "nxlog_input" ]
        rename => [ "Category", "eventlog_category" ]
        rename => [ "EventID", "eventlog_id" ]
        rename => [ "RecordNumber", "eventlog_record_number" ]
        rename => [ "ProcessID", "eventlog_pid" ]
    }
    mutate {
        # Remove redundant fields
        type   => "eventlog"
        remove => [ "SourceModuleType", "EventTimeWritten", "EventTime", "EventReceivedTime", "EventType" ]
    }
}

output {
    elasticsearch {
        embedded => true
    }
    graphite {
        # Ping the graphite server every time a syslog message is received
        type => "syslog"
        port => 2023     # carbon-aggregator
        metrics => [ "syslog.received.%{@source_host}.count", "1" ]
    }
    graphite {
        # Ping the graphite server every time an eventlog message is received
        type => "eventlog"
        port => 2023     # carbon-aggregator
        metrics => [ "eventlog.received.%{@source_host}.count", "1" ]
    }
}

nxlog.conf

define ROOT C:\Program Files (x86)\nxlog

Moduledir %ROOT%\modules
CacheDir %ROOT%\data
Pidfile %ROOT%\data\nxlog.pid
SpoolDir %ROOT%\data
LogFile %ROOT%\data\nxlog.log

<Extension json>
    Module      xm_json
</Extension>

<Input internal>
    Module      im_internal
</Input>

<Input eventlog>
    Module      im_msvistalog
    # this kinda works for me, put * to get everything
    Query       <QueryList>\
                    <Query Id="0">\
                        <Select Path="Application">*</Select>\
                        <Select Path="System">*</Select>\
                        <Select Path="Security">*</Select>\
                    </Query>\
                </QueryList>
</Input>

<Output out>
    Module      om_tcp
    Host        10.1.1.2
    Port        3515
    Exec        $EventReceivedTime = integer($EventReceivedTime) / 1000000; \
                to_json();
</Output>

<Route 1>
    Path        eventlog, internal => out
</Route>

Thank you for sharing this gist.

Is nxlog should be installed on Windows host in order to pull data from local Windows Event Log?

According to your configuration it looks like Logstash opens connection to nxlog. Could you correct me if this is not a case?

Thank you in advance for your answers.

I'm not the author, but here's how I understand the setting:

Is nxlog should be installed on Windows host in order to pull data from local Windows Event Log?

Yes.

According to your configuration it looks like Logstash opens connection to nxlog. Could you correct me if this is not a case?

No, nxlog sends the logs to logstash which in turn is configured to only listen on a specific host address (see http://logstash.net/docs/1.1.9/inputs/tcp#setting_host for details). Logstash stores the logs in an elasticsearch database.

Is 10.1.1.2 the IP of logstash?

Im confused about the input section at the top.
You have 127.0.0.1 configured as well as 10.1.1.2 on a different port.

Can you explain your setup a little?

Thanks for the help.

@c0mputernick (note that @simon04 has given a similar answer). The problem these config files are trying to solve is shipping Windows Eventlog events to a Logstash instance

The Logstash instance is (usually, but not necessarily) different from the Windows host. The IP 1.0.1.1.2 belogs to the Logstash host, it is the interface where logstash is bound at port 3515 (logstash.conf at line 10) listening for incoming messages.