ini 使用nxlog和JSON传输将Windows事件日志信息记录到Logstash

Posted 2021-05-24

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>

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" ]
    }
}