Elkstack Logstash - How to send a threshold alert by email

Question

I need some help with Logstash. I currently have the below Logstash config which works. When the [message] tag has "Token validation failed" in it it sends an email out saying auth issue.

 input {

  tcp {
    codec => "json"
    port => 5144
    tags => ["windows","nxlog"]
    type => "nxlog-json"
  }

} # end input

filter {

  if [type] == "nxlog-json" {
    date {
      match => ["[EventTime]", "YYYY-MM-dd HH:mm:ss"]
      timezone => "Europe/London"
    }
    mutate {
        rename => [ "AccountName", "user" ]
        rename => [ "AccountType", "[eventlog][account_type]" ]
        rename => [ "ActivityId", "[eventlog][activity_id]" ]
        rename => [ "Address", "ip6" ]
        rename => [ "ApplicationPath", "[eventlog][application_path]" ]
        rename => [ "AuthenticationPackageName", "[eventlog][authentication_package_name]" ]
        rename => [ "Category", "[eventlog][category]" ]
        rename => [ "Channel", "[eventlog][channel]" ]
        rename => [ "Domain", "domain" ]
        rename => [ "EventID", "[eventlog][event_id]" ]
        rename => [ "EventType", "[eventlog][event_type]" ]
        rename => [ "File", "[eventlog][file_path]" ]
        rename => [ "Guid", "[eventlog][guid]" ]
        rename => [ "Hostname", "hostname" ]
        rename => [ "Interface", "[eventlog][interface]" ]
        rename => [ "InterfaceGuid", "[eventlog][interface_guid]" ]
        rename => [ "InterfaceName", "[eventlog][interface_name]" ]
        rename => [ "IpAddress", "ip" ]
        rename => [ "IpPort", "port" ]
        rename => [ "Key", "[eventlog][key]" ]
        rename => [ "LogonGuid", "[eventlog][logon_guid]" ]
        rename => [ "Message", "message" ]
        rename => [ "ModifyingUser", "[eventlog][modifying_user]" ]
        rename => [ "NewProfile", "[eventlog][new_profile]" ]
        rename => [ "OldProfile", "[eventlog][old_profile]" ]
        rename => [ "Port", "port" ]
        rename => [ "PrivilegeList", "[eventlog][privilege_list]" ]
        rename => [ "ProcessID", "pid" ]
        rename => [ "ProcessName", "[eventlog][process_name]" ]
        rename => [ "ProviderGuid", "[eventlog][provider_guid]" ]
        rename => [ "ReasonCode", "[eventlog][reason_code]" ]
        rename => [ "RecordNumber", "[eventlog][record_number]" ]
        rename => [ "ScenarioId", "[eventlog][scenario_id]" ]
        rename => [ "Severity", "level" ]
        rename => [ "SeverityValue", "[eventlog][severity_code]" ]
        rename => [ "SourceModuleName", "nxlog_input" ]
        rename => [ "SourceName", "[eventlog][program]" ]
        rename => [ "SubjectDomainName", "[eventlog][subject_domain_name]" ]
        rename => [ "SubjectLogonId", "[eventlog][subject_logonid]" ]
        rename => [ "SubjectUserName", "[eventlog][subject_user_name]" ]
        rename => [ "SubjectUserSid", "[eventlog][subject_user_sid]" ]
        rename => [ "System", "[eventlog][system]" ]
        rename => [ "TargetDomainName", "[eventlog][target_domain_name]" ]
        rename => [ "TargetLogonId", "[eventlog][target_logonid]" ]
        rename => [ "TargetUserName", "[eventlog][target_user_name]" ]
        rename => [ "TargetUserSid", "[eventlog][target_user_sid]" ]
        rename => [ "ThreadID", "thread" ]

    }
    mutate {
        remove_field => [
                    "CurrentOrNextState",
                    "Description",
                    "EventReceivedTime",
                    "EventTime",
                    "EventTimeWritten",
                    "IPVersion",
                    "KeyLength",
                    "Keywords",
                    "LmPackageName",
                    "LogonProcessName",
                    "LogonType",
                    "Name",
                    "Opcode",
                    "OpcodeValue",
                    "PolicyProcessingMode",
                    "Protocol",
                    "ProtocolType",
                    "SourceModuleType",
                    "State",
                    "Task",
                    "TransmittedServices",
                    "Type",
                    "UserID",
                    "Version"
                    ]
    }
  }

}

output {
   elasticsearch {
    hosts => ["localhost:9200"]
    }

if "Token validation failed" in [message]  {

email {

address => "smtp01.domain.com"
to => "example@domain.com"
from => "Sender@domain.com"
subject => "Auth Issue"
body => "Auth Issue"   
port => 25
use_tls => false
via => "smtp" 

}   
}

} # end output

I would like to know how to get the email to send only if the message tag "Token validation failed" 10 times in one minute. If it has 9 or below entries it will not send any emails. What config do I need to setup to get this to work?

score 0 · Answer 1 · edited May 23 '17 at 11:55

There are a few ways to achieve that.

A. You can use XPack Alerting (formerly called Watcher) or ElastAlert as described in this answer

B. You can use the aggregate Logstash filter in order to keep track and count the "Token validation failed" messages as described in this answer. You simply need to

  aggregate {
    task_id => "%{[eventlog][target_logonid]}"
    code => "map['failed_count'] ||= 0; map['failed_count'] += 1;"
    push_map_as_event_on_timeout => true
    timeout => 60 # 1 minute timeout
    timeout_tags => ['_aggregatetimeout']
    timeout_code => "event.set('token_failed', event.get('failed_count') >= 10)"
  }

Then you can send your email only if [token_failed]

C. You can use the ruby Logstash filter in order to count and cache the number of times the "Token validation failed" message has occurred. It's basically the same as B but by implementing the logic yourself in Ruby code.

D. You can use the metrics Logstash filter in order to compute the rate of events having "Token validation failed" in the message field.

  metrics {
    meter => [ "message" ]
    rates => [ 1 ]
    add_tag => "metric"
  }

Then in your output you can simply use the metered info like this:

  if "metric" in [tags] and [Token validation failed][count] >= 10 {
    email {
      ...
    }
  }

Note that with solutions B and C you cannot launch Logstash with more than one worker (i.e. -w 1). I've filed an enhancement request to "fix" that issue, but since the Logstash team already has a huge pipeline of TODOs, we'll see what happens.

Elkstack Logstash - How to send a threshold alert by email

1 Answers1