ELK学習2-logstash深化

私はelk stock構造を目指しているので、elasticsearch+redis+logstash+kibanaを中心に次の内容を書きます.
『logstashベストプラクティス』で勉強する
パイププロファイル本体の内容
管理プロファイルは、主にタスクを開始するために使用されます.入力(input)、処理(filter)、出力(output).
input
これは主にそれらのファイルまたは出力を傍受することを指定します.私のelk stockアーキテクチャでは、ファイルとredisの2つのタイプしかありません.

input {
    # redis
    redis {
        host => "127.0.0.1"
        port => 6379
        password => "123456"
        key => "logstash-queue"
        data_type => "list"
        db => 0
    }

    #   
    file {
        type => "nginx-access"
        path => "/usr/local/nginx/logs/access.log"
        start_position => beginning
        sincedb_path => "/var/log/logstash/sincedb/nginx"
        codec => multiline {
            pattern => "^\d+"
            negate => true
            what => "previous"
        }
    }
}

note: input.file.codecというログの内容が複数行表示される場合は^d+で分割でき、改行はnに変換されます
fileter
常用マッチング方式grok(正規マッチング)
logstash-7.4.0/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.1.2/patternsディレクトリの下には、事前に定義された正規マッチングがあります.使用方法(%{IPORHOST:client}など)
満足できない場合は、自分で正則を書いてマッチングすることができます.正則が正しいかどうかはkibanaの開発ツール(Dev)>Grokデバッガ(Grok Debugger)で検証できる.http://grokdebug.herokuapp.com/で検証することもできます.

filter {
    if [type] == "nginx-access" {
        grok {
            match => {
                "message" => "%{COMBINEDAPACHELOG}+%{GREEDYDATA:extra_fields}"
            }
        }
    } else if [type] == "nginx-error" {
        grok {
            match => ["message" , "(?%{YEAR}[./-]%{MONTHNUM}[./-]%{MONTHDAY}[- ]%{TIME}) \[%{LOGLEVEL:severity}\] %{POSINT:pid}#%{NUMBER}: %{GREEDYDATA:errormessage}(?:, client: (?%{IP}|%{HOSTNAME}))(?:, server: %{IPORHOST:server}?)(?:, request: %{QS:request})?(?:, upstream: (?\"%{URI}\"|%{QS}))?(?:, host: %{QS:request_host})?(?:, referrer: \"%{URI:referrer}\")?"]
        }
    }
}

最適化シナリオ

json形式のログに直接転送

ダイレクト・ログは、ログ・コンテンツのマッチングの一部のリソースの使用量を節約します.しかし、すべてのソフトウェアログが構成できるわけではありません.鶏の肋骨がある.
output

output {
    redis {
        host => "127.0.0.1"
        port => 6379
        password => "123456"
        key => "logstash-queue"
        data_type => "list"
        db => 4
    }
    elasticsearch {
        hosts => ["http://localhost:9200"]
        index => "logstash-%{+YYYY.MM.dd}"
    }
}

esインポート時にフィールドタイプを設定する
Esでは全文インデックスがサポートされていますが、デフォルトでは英語がサポートされています.私たちのニーズに合わないので、ik分詞プラグインを借りる必要があります.
リファレンス
FQA
1、1本のデータには多くの行の処理方法がある
inputを使用codecはnginxのデフォルトフォーマットログを例にマージされます.

2019/09/23 10:39:01 [error] 4130#0: *1 FastCGI sent in stderr: "PHP message: PHP Warning:  require(/var/www/study/tp5-study/public/../thinkphp/base.php): failed to open stream: No such file or directory in /var/www/study/tp5-study/public/index.php on line 16
PHP message: PHP Stack trace:
PHP message: PHP   1. {main}() /var/www/study/tp5-study/public/index.php:0
PHP message: PHP Fatal error:  require(): Failed opening required '/var/www/study/tp5-study/public/../thinkphp/base.php' (include_path='.:') in /var/www/study/tp5-study/public/index.php on line 16
PHP message: PHP Stack trace:
PHP message: PHP   1. {main}() /var/www/study/tp5-study/public/index.php:0" while reading response header from upstream, client: 192.168.33.1, server: tp5.study.me, request: "GET /favicon.ico HTTP/1.1", upstream: "fastcgi://127.0.0.1:9000", host: "tp5.study.me", referrer: "http://tp5.study.me/"
2019/09/23 10:40:14 [error] 4130#0: *7 FastCGI sent in stderr: "PHP message: PHP Warning:  require(/var/www/study/tp5-study/public/../thinkphp/base.php): failed to open stream: No such file or directory in /var/www/study/tp5-study/public/index.php on line 16
PHP message: PHP Stack trace:
PHP message: PHP   1. {main}() /var/www/study/tp5-study/public/index.php:0
PHP message: PHP Fatal error:  require(): Failed opening required '/var/www/study/tp5-study/public/../thinkphp/base.php' (include_path='.:') in /var/www/study/tp5-study/public/index.php on line 16
PHP message: PHP Stack trace:
PHP message: PHP   1. {main}() /var/www/study/tp5-study/public/index.php:0" while reading response header from upstream, client: 192.168.33.1, server: tp5.study.me, request: "GET /favicon.ico HTTP/1.1", upstream: "fastcgi://127.0.0.1:9000", host: "tp5.study.me"

以上から分かるように、第1のログの先頭は日付で構成されています.そこで、数値で始まるログ分割を行います.すぐ

input {
    stdin {
        codec => multiline {
            pattern => "^\d+"
            negate => true
            what => "previous"
        }
    }
}

2、ログにはデフォルトでメッセージが付いています.このメッセージはデータが一致していないログです.すでにコンテンツが提出されているので、元のデータが存在する必要はありません.

filter {
        grok {
            match => ["message" , "(?%{YEAR}[./-]%{MONTHNUM}[./-]%{MONTHDAY}[- ]%{TIME}) \[%{LOGLEVEL:severity}\] %{POSINT:pid}#%{NUMBER}: %{GREEDYDATA:message}(?:, client: (?%{IP}|%{HOSTNAME}))(?:, server: %{IPORHOST:server}?)(?:, request: %{QS:request})?(?:, upstream: (?\"%{URI}\"|%{QS}))?(?:, host: %{QS:request_host})?(?:, referrer: \"%{URI:referrer}\")?"]
            overwrite => ["message"]
        }
}

overwriteで書き換えます.overwriteはfilterに必要です.grokの中
3、ログキャプチャに@timestampがあります.古いデータの時間をこの時間に書いてほしいです.
注:これはlogstashが持参したもので、修正はお勧めしませんので、timestampで代用しますが、違いはこれがマッチングで得られる時間です