Logstash eventlog filter

Elastic Stack Logstash
1

I need to filter error messages from eventlog from logstash.

logstash configuration

input {
   tcp {
     port => 5045
     type => 'eventlog'

   }
}
filter{

if [type] == 'eventlog' {
    if [Severity] == "ERROR" {
    mutate {
      add_tag => "error"
    }
  }
}


}
output {

  elasticsearch {
         hosts => ["IP_ADDRSS:9200"]
      }

   if "error" in [tags]{

   stdout { codec => 'rubydebug' }
}
}

But still I am getting thousands of logs from which I can't filter out the error logs.
How to effectively filter error logs from all type of eventlogs?

2

If your goal is to avoid sending events tagged "error" to ES you need to wrap the elasticsearch output in a conditional. Right now you are indeed sending all events to ES.

3

I need to send only error logs . Can you please help me doing this.?
ELK version : Elasticsearch 2.2.x, Logstash 2.2.x, and Kibana 4.4.x

4 
output {
  if "error" in [tags] {
    elasticsearch {
      ...
    }
  }
  ...
}

This assumes that your events are being correctly tagged with "error", but your existing filter looks okay.

5 
@timestamp	  	June 2nd 2016, 17:29:47.728
t@version	  	1
t_id	  	AVUQ-9fn60SmJzRpwVoX
t_index	  	logstash-2016.06.02
_score	  	
t_type	  	eventlog
t host	  	IP_ADDRSS

?message {"EventTime":"2016-06-02 17:29:45","Hostname":"KOZHIKODE.ontashindia.local","Keywords":-9214364837600034816,"EventType":"AUDIT_SUCCESS","SeverityValue":2,"Severity":"INFO","EventID":4634,"SourceName":"Microsoft-Windows-Security-Auditing","ProviderGuid":"{54849625-5478-4994-A5BA-3E3B0328C30D}","Version":0,"Task":12545,"OpcodeValue":0,"RecordNumber":18433232,"ProcessID":556,"ThreadID":10692,"Channel":"Security","Message":"An account was logged off.\r\n\r\nSubject:\r\n\tSecurity ID:\t\tS-1-5-18\r\n\tAccount Name:\t\tKOZHIKODE$\r\n\tAccount Domain:\t\tONTASHINDIA\r\n\tLogon ID:\t\t0x1DBE6BA4F\r\n\r\nLogon Type:\t\t\t3\r\n\r\nThis event is generated when a logon session is destroyed. It may be positively correlated with a logon event using the Logon ID value. Logon IDs are only unique between reboots on the same computer.","Category":"Logoff","Opcode":"Info","TargetUserSid":"S-1-5-18","TargetUserName":"KOZHIKODE$","TargetDomainName":"ONTASHINDIA","TargetLogonId":"0x1dbe6ba4f","LogonType":"3","EventReceivedTime":"2016-06-02 17:29:47","SourceModuleName":"eventlog","SourceModuleType":"im_msvistalog"}

The above is the log I am getting. Still getting lots of 'INFO' logs.
I think the filter
if [Severity] == "ERROR" {
mutate {
add_tag => "error"
}
is not working properly. Any solution ?

6

What does a message with an error severity look like when Logstash is done with it? Does it have an "error" tag?

7

If error happens the ?message field looks like
message:{"EventTime":"2016-06-02 22:00:38","Hostname":"KOZHIKODE.ontashindia.local","Keywords":4611686018427387904,"EventType":"ERROR","SeverityValue":4,"Severity":"ERROR"

8

Oh, now I see. You're not parsing the message field as JSON so its payload is never turned into fields in the event. Add codec => json (or maybe json_lines) to your tcp input.

9

I did parse message field as JSON and got severity field .

INFO logs are still recieving at logstash end.
Actually I am trying to filter out the error logs and send them to hipchat room. I dont want INFO and WARNING logs .

10

I did parse message field as JSON and got severity field .

Not according to anything you've shown us. Your message field contains a JSON string and so does your copy/paste from Kibana.

11

Right. The way he has it coded, it is expecting the data to be sent in Json format. My guess is data is not in that format. Showing the output from message in elastic is one large string in json, not to be confused with the message value inside.

If you are not sending in json (which you need to show a snippet of raw data BEFORE it hits elastic), you could use a grok filter to process raw lines and then get them to fields you can do something with.

My guess is that you are new at this because you have left lots of critical little things out. For example the json codec, and your if statements not encompassing the code to execute. So, your code will only work if the RAW data being sent is already in Json format before it hits logstash. Since you left off the json codec and the output is not working... this means the data is not in Json. Obviously when it gets to elastic... the data is then put into fields consisting of Json data. This is what you are showing us, which is Post logstash processing.

12

Your guess is right . I am new to this. Please excuse me if my questions are below average.

I am using nxlog in windows server to send eventlogs to logstash. And the data being sent is formatted to JSON in output section. See below

<Extension _json>
    Module      xm_json
</Extension>

<Input eventlog>
    Module      im_msvistalog
# For windows 2003 and earlier use the following:
#   Module      im_mseventlog
</Input>

<Output out>
    Module      om_tcp
    Host        LOGSTASH_IP
    Port         LOGSTASH_PORT
    Exec        to_json();
 </Output>

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

In kibana I am able to see Severity fields as

    "Keywords": -9214364837600035000,
    "EventType": "AUDIT_SUCCESS",
    "SeverityValue": 2,
    "Severity": "INFO",
    "EventID": 4634,
    "SourceName": "Microsoft-Windows-Security-Auditing",

Now please tell me where I am wrong, and what should I do to achieve my goal.

13

What does the Logstash config look like now? What version of Logstash are you using?

14

Xxxx

15

Logstash configuration :

input {
tcp {
port => 5045
type => 'eventlog'
codec =>"json"
}
}
filter {
if [type] == "eventlog" and [Severity] == "ERROR"{
mutate {
add_tag => "error"
}
}
}

output {
if "error" in [tags] {
elasticsearch {
hosts => ["172.30.0.206:9200"]
}
}

Logstash version : 2.2.x

16

Please show the full, raw event that you're seeing in Kibana. I want to see whether it gets the "error" tag and if the type field really is "eventlog".

17

Ok .. Here is the full event seeing in kibana

"_index": "logstash-2016.06.09",
  "_type": "eventlog",
  "_id": "AVUzhrZOiZuyFxp6GL_k",
  "_score": null,
  "_source": {
    "EventTime": "2016-06-09 10:28:31",
    "Hostname": "KOZHIKODE.ontashindia.local",
    "Keywords": 4611686018427388000,
    "EventType": "ERROR",
    "SeverityValue": 4,
    "Severity": "ERROR",
    "EventID": 5858,
    "SourceName": "Microsoft-Windows-WMI-Activity",
    "ProviderGuid": "{1418EF04-B0B4-4623-BF7E-D74AB47BBDAA}",
    "Version": 0,
    "Task": 0,
    "OpcodeValue": 0,
    "RecordNumber": 693934,
    "ProcessID": 912,
    "ThreadID": 7484,
    "Channel": "Microsoft-Windows-WMI-Activity/Operational",
    "Domain": "NT AUTHORITY",
    "AccountName": "SYSTEM",
    "UserID": "SYSTEM",
    "AccountType": "User",
    "Message": "Id = {E5ACABE8-4D53-4BDE-B47F-4AC40830204D}; ClientMachine = KOZHIKODE; User = ONTASHINDIA\\Administrator; ClientProcessId = 4304; Component = Provider; Operation = Start IWbemServices::ExecMethod - root\\microsoft\\windows\\servermanager : MSFT_ServerManagerTasks::GetCounterSamplesInTimeRange; ResultCode = 0x80041001; PossibleCause = Unknown",
    "Opcode": "Info",
    "EventReceivedTime": "2016-06-09 10:28:33",
    "SourceModuleName": "eventlog",
    "SourceModuleType": "im_msvistalog",
    "@version": "1",
    "@timestamp": "2016-06-09T04:58:33.337Z",
    "host": "135.219.172.23",
    "port": 58301,
    "type": "eventlog",
    "tags": [
      "error",
      "_grokparsefailure"
    ]
  },
  "fields": {
    "@timestamp": [
      1465448313337
    ]
  },
  "highlight": {
    "tags.raw": [
      "@kibana-highlighted-field@error@/kibana-highlighted-field@"
    ],

   "SourceModuleName": [
      "@kibana-highlighted-field@eventlog@/kibana-highlighted-field@"
    ],
    "SourceModuleName.raw": [
      "@kibana-highlighted-field@eventlog@/kibana-highlighted-field@"
    ],
    "EventType": [
      "@kibana-highlighted-field@ERROR@/kibana-highlighted-field@"
    ],
    "Severity": [
      "@kibana-highlighted-field@ERROR@/kibana-highlighted-field@"
    ],
    "type.raw": [
      "@kibana-highlighted-field@eventlog@/kibana-highlighted-field@"
    ],
    "type": [
      "@kibana-highlighted-field@eventlog@/kibana-highlighted-field@"
    ],
    "tags": [
      "@kibana-highlighted-field@error@/kibana-highlighted-field@"
    ]
  }
,
  "sort": [
    1465448313337
  ]
18

Okay, but what does a non-error event look like? The one that shouldn't be present in the ES index?

19

INFO :

 "_index": "logstash-2016.06.10",
  "_type": "eventlog",
  "_id": "AVU4pSR0yeP15vt6ARUS",
  "_score": null,
  "_source": {
    "EventTime": "2016-06-09 23:13:57",
    "Hostname": "KOZHIKODE.ontashindia.local",
    "Keywords": -9214364837600035000,
    "EventType": "AUDIT_SUCCESS",
    "SeverityValue": 2,
    "Severity": "INFO",
    "EventID": 4634,
    "SourceName": "Microsoft-Windows-Security-Auditing",
    "ProviderGuid": "{54849625-5478-4994-A5BA-3E3B0328C30D}",
    "Version": 0,
    "Task": 12545,
    "OpcodeValue": 0,
    "RecordNumber": 18518828,
    "ProcessID": 548,
    "ThreadID": 580,
    "Channel": "Security",
    "Message": "An account was logged off.\r\n\r\nSubject:\r\n\tSecurity ID:\t\tS-1-5-18\r\n\tAccount Name:\t\tKOZHIKODE$\r\n\tAccount Domain:\t\tONTASHINDIA\r\n\tLogon ID:\t\t0x1DBDAF7B\r\n\r\nLogon Type:\t\t\t3\r\n\r\nThis event is generated when a logon session is destroyed. It may be positively correlated with a logon event using the Logon ID value. Logon IDs are only unique between reboots on the same computer.",
    "Category": "Logoff",
    "Opcode": "Info",
    "TargetUserSid": "S-1-5-18",
    "TargetUserName": "KOZHIKODE$",
    "TargetDomainName": "ONTASHINDIA",
    "TargetLogonId": "0x1dbdaf7b",
    "LogonType": "3",
    "EventReceivedTime": "2016-06-10 10:19:34",
    "SourceModuleName": "eventlog",
    "SourceModuleType": "im_msvistalog",
    "@version": "1",
    "@timestamp": "2016-06-10T04:49:54.762Z",
    "host": "135.219.172.23",
    "port": 57366,
    "type": "eventlog",
    "tags": [
      "_grokparsefailure"
    ]
  },
  "fields": {
    "@timestamp": [
      1465534194762
    ]
  },
  "sort": [
    1465534194762
  ]

WARNING :

"_index": "logstash-2016.06.10",
  "_type": "eventlog",
  "_id": "AVU4qG85yeP15vt6AZY4",
  "_score": null,
  "_source": {
    "EventTime": "2016-06-10 10:23:28",
    "Hostname": "KOZHIKODE.ontashindia.local",
    "Keywords": -9223372036854776000,
    "EventType": "WARNING",
    "SeverityValue": 3,
    "Severity": "WARNING",
    "EventID": 1002,
    "SourceName": "Microsoft-Windows-KnownFolders",
    "ProviderGuid": "{8939299F-2315-4C5C-9B91-ABB86AA0627D}",
    "Version": 0,
    "Task": 0,
    "OpcodeValue": 0,
    "RecordNumber": 2049330,
    "ActivityID": "{417C638D-1C28-4374-A6CA-3DC8784F39D1}",
    "ProcessID": 912,
    "ThreadID": 3564,
    "Channel": "Microsoft-Windows-Known Folders API Service",
    "Domain": "NT AUTHORITY",
    "AccountName": "SYSTEM",
    "UserID": "SYSTEM",
    "AccountType": "User",
    "Message": "Error 0x80070002 occurred while verifying known folder {B97D20BB-F46A-4C97-BA10-5E3608430854} with path 'C:\\Users\\Default\\AppData\\Roaming\\Microsoft\\Windows\\Start Menu\\Programs\\Startup'.",
    "Opcode": "Info",
    "hrError": "0x80070002",
    "FolderId": "{B97D20BB-F46A-4C97-BA10-5E3608430854}",
    "Path": "C:\\Users\\Default\\AppData\\Roaming\\Microsoft\\Windows\\Start Menu\\Programs\\Startup",
    "EventReceivedTime": "2016-06-10 10:23:29",
    "SourceModuleName": "eventlog",
    "SourceModuleType": "im_msvistalog",
    "@version": "1",
    "@timestamp": "2016-06-10T04:53:30.178Z",
    "host": "135.219.172.23",
    "port": 57366,
    "type": "eventlog",
    "tags": [
      "_grokparsefailure"
    ]
  },
  "fields": {
    "@timestamp": [
      1465534410178
    ]
  },
  "highlight": {
    "SourceModuleName": [
      "@kibana-highlighted-field@eventlog@/kibana-highlighted-field@"
    ],
    "SourceModuleName.raw": [
      "@kibana-highlighted-field@eventlog@/kibana-highlighted-field@"
    ],
    "EventType": [
      "@kibana-highlighted-field@WARNING@/kibana-highlighted-field@"
    ],
    "Severity": [
      "@kibana-highlighted-field@WARNING@/kibana-highlighted-field@"
    ],
    "type.raw": [
      "@kibana-highlighted-field@eventlog@/kibana-highlighted-field@"
    ],
    "type": [
      "@kibana-highlighted-field@eventlog@/kibana-highlighted-field@"
    ]
  },
  "sort": [
    1465534410178
  ]
20

Okay, so events don't have an "error" tag yet still end up in ES. That doesn't make sense given the configuration you claim to be running. How are you starting Logstash?