Elkstack Logstash - 電子メールでしきい値アラートを送信するには

Question

Logstashにいくつかの助けが必要です。私は現在、以下のLogstash設定を持っています。 [メッセージ]タグに「トークンの検証に失敗しました」というタグがある場合は、認証の問題を示すメールを送信します。

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 

メッセージタグ「トークンの検証に失敗しました」が1分で10回しか送信されないようにする方法を知りたいと思います。エントリが9つ以下の場合、電子メールは送信されません。これを動作させるために設定する必要があるのは何ですか？

Answer 1

これを達成する方法はいくつかあります。で説明したようにあなたが追跡し、カウントするためにaggregate Logstash filterを使用することができますthis answer

B.で説明したようにあなたがXPackアラート（旧ウォッチャーと呼ばれる）またはElastAlertを使用することができます

A.メッセージを「トークンの検証に失敗しました」 this answer。あなたは、単にその後、あなたはあなたがメッセージが発生しました「トークンの検証に失敗しました」回数をカウントし、キャッシュするためにruby Logstash filterを使用することができますあなたのメールのみif [token_failed]

C.を送ることができます

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

する必要があります。基本的にはBと同じですが、Rubyコードでロジックを実装することによって実現します。

D. metrics Logstash filterを使用すると、メッセージフィールドに「トークンの検証に失敗しました」というイベントの割合を計算できます。溶液BおよびCであなたがmore than one worker（すなわち-w 1）でLogstashを起動することができないこと

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

注：

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

は、その後、あなたの出力には、単にこのような計量された情報を使用することができます。その問題を「修正」するためにenhancement requestを提出しましたが、Logstashチームには既にTODOの巨大なパイプラインがあるので、何が起こるかを見ていきます。