grokフィルターは、Logstashに届いた1行のテキストログを正規表現ベースのパターンで分解し、IPアドレスやステータスコードなどのフィールドに変える仕組みです。この記事では、Logstash 9.5.5(2026年10月6日リリース)の公式Dockerイメージで主な設定例を実行し、出力を確かめたうえで、構文の読み方、Logstash 8以降の既定でフィールド名が変わる点、カスタムパターン、_grokparsefailureの切り分け、DissectとRubyフィルターの書き方と使い分けを説明します。
まとめ
- 構文は
%{パターン名:フィールド名}。%{NUMBER:bytes:int}のように型を付けられるが、変換できるのはintとfloatだけ。 - Logstash 8以降は
ecs_compatibilityの既定がv8。%{COMBINEDAPACHELOG}の出力はclientipやresponseではなく[source][address]や[http][response][status_code]になる。旧名が要るならecs_compatibility => disabled。 - 独自パターンは、そのフィルター内だけなら
pattern_definitions、複数のフィルターで共有するならpatterns_dir。 - grokを複数段に重ねるときは
tag_on_failureを段ごとに別名にする。既定の_grokparsefailureのままだと、どの段で落ちたか分からない。 - 区切りが毎行同じログはDissect、行ごとに形が変わるログはgrok、値の計算や条件付き加工はRubyフィルター。前半をDissect、残りをgrokで切る併用もできる。
まず、grokがパイプラインのどこで何をするかを押さえます。
grokフィルターの役割とパイプライン内の位置
Logstashは、入力(input)→加工(filter)→出力(output)の3段でイベントを流すパイプラインです。grokはfilter段のプラグインで、matchで指定したフィールド(多くは元の1行が入るmessage)にパターンを当て、マッチした部分を名前付きのフィールドとして取り出します。Logstashを含めた収集経路全体の設計はElasticsearchのログ収集とELKスタックの収集経路、Kafkaから取り込む場合のinput設定はLogstash Kafka連携の設計で扱っています。
Elastic公式のgrokプラグイン文書によると、正規表現ライブラリはOnigurumaで、Logstashには約120個のパターンが同梱されています。組み込みパターンは名前を付けた正規表現の部品で、生の正規表現もそのまま書けます。正規表現そのものの記号の読み方は正規表現とは?基本記号の読み方と言語ごとの方言にまとめています。
grok構文の書き方|%{SYNTAX:SEMANTIC}と型変換
grokパターンの最小単位は%{SYNTAX:SEMANTIC}です。SYNTAXはマッチに使うパターン名、SEMANTICは取り出した値に付けるフィールド名です。%{IPV4:client_ip}なら「IPv4アドレスに当たる部分をclient_ipとして取り出す」という意味になります。フィールド名には[http][status]のような入れ子の参照も書けます。
型変換で使えるintとfloat
取り出した値は既定で文字列です。数値にしたい場合は3つ目に型を付け、%{NUMBER:status:int}や%{NUMBER:took_ms:float}と書きます。公式文書は「Currently the only supported conversions are int and float」としており、日付や真偽値への変換はgrokではできません。@timestampへ入れたい日時は、grokで文字列として取り出したあとdateフィルターで変換します。
よく使う組み込みパターン
パターンの定義はlogstash-patterns-coreのpatternsディレクトリにあり、ECS用(ecs-v1)と旧来用(legacy)の2系統があります。
| パターン名 | マッチ対象 | 使いどころ |
|---|---|---|
| IPORHOST | IPアドレスかホスト名 | クライアント・送信元 |
| NUMBER / INT | 小数を含む数値 / 整数 | バイト数・ステータス |
| WORD | 英数字と_の連続 | HTTPメソッド |
| NOTSPACE | 空白以外の連続 | URLパス・ID |
| DATA | 最短一致の任意文字 | 区切りの間の値 |
| GREEDYDATA | 最長一致の任意文字 | 行末の本文だけ |
| TIMESTAMP_ISO8601 | ISO 8601の日時 | アプリログの時刻 |
| SYSLOGTIMESTAMP | Oct 28 10:15:31 形式 | syslogの時刻 |
| LOGLEVEL | INFO・WARN・ERROR等 | ログレベル |
| HTTPD_COMBINEDLOG | Apache/nginxの結合形式 | アクセスログ1行 |
よく見かけるCOMBINEDAPACHELOGは、ECS用・旧来用どちらの定義ファイルでも# Deprecatedの節に置かれたHTTPD_COMBINEDLOGの別名です。今も動きますが、新しく書く設定ではHTTPD_COMBINEDLOGを使うほうが定義ファイルと読み合わせやすくなります。
grokフィルターの設定と実行確認
パターンを書いたら、標準入力から1行流して標準出力で結果を見るのが最短の確認方法です。次の設定は、Javaアプリのログとsyslog形式の2種類を1つのgrokで受け、続く2段目のgrokで本文の中身をさらに分解します。
input { stdin {} }
filter {
grok {
match => {
"message" => [
"^%{TIMESTAMP_ISO8601:ts} %{LOGLEVEL:level} \[%{DATA:thread}\] %{NOTSPACE:logger} - %{GREEDYDATA:msg}$",
"^%{SYSLOGTIMESTAMP:ts} %{SYSLOGHOST:host_name} %{PROG:program}(?:\[%{POSINT:pid:int}\])?: %{GREEDYDATA:msg}$"
]
}
tag_on_failure => ["_grok_app_failure"]
}
grok {
pattern_definitions => { "QUEUEID" => "[0-9A-F]{10,11}" }
patterns_dir => ["/etc/logstash/patterns"]
match => { "msg" => "^%{QUEUEID:queue_id}: took=%{NUMBER:took_ms:float}ms status=%{NUMBER:status:int} user=%{USERNAME_JP:user}$" }
tag_on_failure => ["_grok_detail_failure"]
}
}
output { stdout { codec => rubydebug } }
Dockerなら、後述のUSERNAME_JPを書いたpatterns/customを先に作り、設定とパターンの両方をマウントしてdocker run --rm -i -v "$PWD":/t -v "$PWD/patterns":/etc/logstash/patterns docker.elastic.co/logstash/logstash:9.5.5 -f /t/basic.conf < basic.txtで実行できます。標準入力が終わるとパイプラインも止まります。次の3行を流しました。
2026-10-28T10:15:30.123+09:00 ERROR [http-nio-8080-exec-3] com.example.OrderService - BEF25A72965: took=12.5ms status=500 user=sato_k
Oct 28 10:15:31 web01 postfix/cleanup[21403]: BEF25A72965: took=3ms status=200 user=suzuki
this line matches nothing
1行目はtook_ms => 12.5(float)、status => 500(int)、2行目はpid => 21403まで取れました。3行目には_grok_app_failureと_grok_detail_failureの両方が付きます。2段目は対象のmsgフィールドが存在しない場合にも失敗タグを付けるので、1段目の失敗が2段目へ連鎖して見えます。どこで落ちたかは、最初に付いたタグで判断します。
複数パターンとbreak_on_match
matchに配列でパターンを渡すと、先頭から順に試します。既定のbreak_on_match => trueでは最初にマッチした時点で打ち切るので、実際に多い形式を先頭に置くほど平均の照合回数が減ります。falseにするとすべてのパターンを評価するので遅くなり、同じフィールド名を取り出すパターンが複数マッチすると、値は上書きされずに配列へ追加されます(overwriteで指定したフィールドを除く)。
主要オプションと既定値
grokプラグインv4.4.4(2026年2月11日リリース)の公式文書に載っている既定値です。
| オプション | 既定値 | 役割 |
|---|---|---|
| match | {} | 対象フィールドとパターン |
| break_on_match | true | 初回マッチで打ち切る |
| ecs_compatibility | パイプライン設定に従う | ECS用か旧来のパターンか |
| named_captures_only | true | 名前付きの取得だけ残す |
| keep_empty_captures | false | 空の取得もフィールドにする |
| overwrite | [] | 既存フィールドを上書きする |
| target | なし | 取得結果をまとめる親フィールド |
| tag_on_failure | [“_grokparsefailure”] | 不一致時のタグ |
| tag_on_timeout | “_groktimeout” | タイムアウト時のタグ |
| timeout_millis | 30000 | 照合のタイムアウト |
| timeout_scope | “pattern” | タイムアウトをパターン単位かイベント単位か |
overwrite => ["message"]は、messageから本文だけを取り出して同じ名前に入れ直したいときに使います。指定しないと元の値と新しい値が配列で並びます。target => "parsed"を付けると、取得したフィールドが[parsed][...]の下にまとまり、既存のフィールドとぶつかりません。
Logstash 8以降のECS互換の既定とフィールド名の変化
古い記事の設定を写して「clientipが出てこない」となる原因の多くはこれです。Elastic公式のECS解説は「In Logstash 8, all plugins are run in ECS compatibility v8 mode by default」としており、9.5.5のconfig/logstash.ymlもpipeline.ecs_compatibilityの既定をv8と記しています。grokはこの設定に従い、組み込みの複合パターンをECS(Elastic Common Schema)準拠のフィールド名で出力します。
同じ1行を既定とdisabledで解析した結果
Apacheの結合形式1行を%{COMBINEDAPACHELOG}だけで解析し、出力されたフィールドを並べました。
| 値 | 既定(v8) | disabled(旧来) |
|---|---|---|
| 203.0.113.7 | [source][address] | clientip |
| GET | [http][request][method] | verb |
| /api/orders?id=10 | [url][original] | request |
| 1.1 | [http][version] | httpversion |
| 200 | [http][response][status_code](数値) | response(文字列) |
| 5120 | [http][response][body][bytes](数値) | bytes(文字列) |
| リファラー | [http][request][referrer] | referrer(”付き) |
| User-Agent | [user_agent][original] | agent(”付き) |
名前のほかに型も違います。ECS用のパターンはステータスコードとバイト数を:intで定義しているので数値で入り、旧来のパターンでは文字列のままです。旧来側はリファラーとUser-AgentをQS(引用符付き文字列)で取るため、値の前後に"が残ります。Elasticsearch側のマッピングやKibanaのダッシュボードは、このどちらの名前で作られているかに合わせる必要があります。
旧来のフィールド名に戻す3つの単位
既存のダッシュボードが旧来の名前で組まれているなら、戻す範囲を選べます。
# 1. そのgrokだけ(pipeline.conf)
grok {
ecs_compatibility => disabled
match => { "message" => "%{HTTPD_COMBINEDLOG}" }
}
# 2. パイプライン単位(config/pipelines.yml)
- pipeline.id: legacy-apache
path.config: "/etc/logstash/conf.d/apache.conf"
pipeline.ecs_compatibility: disabled
# 3. プロセス全体(config/logstash.yml)
pipeline.ecs_compatibility: disabled
プロセス全体をdisabledにすると、これから作るパイプラインまで旧来の挙動になります。移行中の古いパイプラインだけをpipelines.ymlで個別に戻し、新しいものは既定のECSで作るのが無難です。なお、9.5.5で既定のまま起動すると「ECS v8 support is a preview of the unreleased ECS v8, and uses the v1 patterns」という警告が出ます。grokのv8は中身がecs-v1のパターンであり、警告が出ても解析結果は上の表のとおりです。
カスタムGrokパターンの定義場所と共有範囲
組み込みにない形式は、自分でパターンを定義します。書き方は「パターン名、半角スペース、正規表現」の1行で、置き場所の違いで使える範囲が変わります。
フィルター内だけで使うpattern_definitions
上の設定例の"QUEUEID" => "[0-9A-F]{10,11}"がこれです。公式文書のとおり、この定義はそのgrokフィルターの中でしか使えず、別のgrokからは参照できません。1回きりの値なら、名前を付けずにOnigurumaの名前付きキャプチャ(?<queue_id>[0-9A-F]{10,11})をパターン内へ直接書く方法もあります。
複数のフィルターで共有するpatterns_dir
同じ定義を複数のパイプラインで使うなら、ファイルにまとめてpatterns_dirで読み込みます。読み込むファイル名はpatterns_files_glob(既定"*")で絞れます。
# /etc/logstash/patterns/custom
USERNAME_JP [a-z][a-z0-9_]{2,15}
上の実行例では、このファイルのUSERNAME_JPとpattern_definitionsのQUEUEIDを同じgrokで併用し、user => "sato_k"まで取り出せています。定義ファイルをパイプラインの設定と分けてGitで管理すると、パターンの修正だけをレビューできます。
Grok Debuggerでの抽出確認とlogstash -tでの設定検査
パターンは一度で通ることが少ないので、本番に入れる前に手元で試します。KibanaのGrok Debuggerは、ナビゲーションメニューかグローバル検索から開く開発者ツールの1つで、サンプル行とパターンを貼って「Simulate」を押すと抽出結果が表示されます。Elasticのセキュリティ機能を有効にしている環境ではmanage_pipeline権限が必要です。Custom Patterns欄に入れた定義は保存されず、そのセッション限りです。
Kibanaが無い環境では、上で示した stdin → rubydebug の構成がいちばん確実です。本番と同じバージョンとパターン定義で評価できます。設定の文法だけを確かめるならbin/logstash -f pipeline.conf -t(--config.test_and_exit)を使います。パターンの組み立ては、まず%{GREEDYDATA:rest}で行全体を受け、左から1つずつ確定させていくと、どこで崩れたかを見失いません。
grokの高速化|アンカー・GREEDYDATA・timeout_scope
grokが遅くなる大きな原因の1つは、正規表現のバックトラッキング(照合に失敗したときの後戻り)です。
どの書き方がどれだけ効くかは、パターンとログで変わります。そこで、Apacheの結合形式1行(マッチする行)とJSON 1行(マッチしない行)をそれぞれ10万件流し、Logstashの統計APIでフィルターごとの処理時間を測りました。各フィルターにidを付けておくと、curl -s localhost:9600/_node/stats/pipelines?prettyのplugins.filtersに、そのフィルターのevents.inとduration_in_millisが出ます。
| フィルター | マッチする行 | マッチしない行 |
|---|---|---|
| grok %{HTTPD_COMBINEDLOG} | 413〜467 | 335〜390 |
| grok ^%{HTTPD_COMBINEDLOG}$ | 209〜244 | 154〜182 |
| grok GREEDYDATA×5 | 176〜214 | 150〜217 |
| dissect | 57〜72 | 333〜393 |
単位は1件あたりのマイクロ秒で、3回測った最小〜最大です。環境はcolimaのx86_64仮想マシン上のLogstash 9.5.5、pipeline.workers: 1で、2本のパイプラインを同時に動かしました。環境や入力が変われば値も比率も変わるので、自分のログで測るときの目安として見てください。フィルターの並び順を入れ替えた回も傾向は同じでした。
行頭・行末のアンカー
同じHTTPD_COMBINEDLOGでも、^と$で挟んだだけで、マッチする行でもしない行でも処理時間がほぼ半分になりました。アンカーが無いと、行頭で外れても照合の開始位置をずらして探し続けます。ログ全体の形が決まっているなら、両端のアンカーは付けておきます。
GREEDYDATA・DATAの置き換えと効果の測り方
GREEDYDATA(実体は.*)とDATA(.*?)は、後続のパターンとの兼ね合いで後戻りが増えるため、空白を含まない値はNOTSPACE、数値はNUMBERのように狭めるのが定石です。ただし上の計測では、短い1行にGREEDYDATAを5つ並べたパターンは、アンカー付きのHTTPD_COMBINEDLOGと同程度の速さでした。区切りが少なく行が短いログでは差が出にくいということです。置き換えの効果は、統計APIで自分のログを測って確かめます。GREEDYDATAを行末の本文1か所に限るのは、速さよりも、どこまでを1つの値として受けるかを読み違えないためでもあります。
timeout_scope => eventによるタイムアウトの集約
timeout_millisの既定値30,000ミリ秒は照合を打ち切る上限の設定値で(公式文書は厳密ではなく近似的な判定としています)、超えた行には_groktimeoutが付きます。既定のtimeout_scope => "pattern"ではこの上限をパターン1本ごとに管理するため、配列で多数のパターンを並べると、その管理の負担が積み上がります。変更履歴によるとgrok v4.2.0で入ったtimeout_scope => eventにすると、1イベントで試すパターン全体に1つのタイムアウトをかけるので、公式文書の説明では同じ安全策をより小さい負担で得られます。_groktimeoutが付く行が出たら、上限を延ばすのではなくパターンを狭めるほうを先に検討します。
_grokparsefailureの原因と失敗行の扱い
_grokparsefailureは、パターンにマッチしなかったイベントに付くタグです。エラーで止まるわけではなく、イベントは素のmessageのまま出力先まで流れます。Elasticsearchへ送っているなら、tags:_grokparsefailureで検索して落ちた行を集め、パターンを足すのが基本の流れです。
原因として多いのは次の4つです。
- パターンの空白の数が実際のログと違う(syslogの日付は1桁の日に空白が2つ入る)
- 前段のgrokが失敗し、後段が参照するフィールドがそもそも無い
- ECS既定の環境で旧来のフィールド名を前提に後段の条件分岐を書いている
- 複数行のスタックトレースが1行ずつ別イベントとして届いている
2つ目は上の実行例で再現したとおりで、段ごとにtag_on_failureを別名にしておけば一目で分かります。複数行のログは、grokより前のinput段でmultilineコーデックやFilebeatの複数行設定でまとめる必要があります。解析できなかった行を残したくない場合は、条件分岐で落とします。
filter {
grok {
match => { "message" => "^%{HTTPD_COMBINEDLOG}$" }
tag_on_failure => ["_grok_access_failure"]
}
if "_grok_access_failure" in [tags] {
drop { }
}
}
ただし、捨てた行はあとから調べられません。運用を始めたばかりの時期はdropせず、失敗した行だけを別のインデックスへ送り分けて件数を見るほうが、パターン漏れに早く気づけます。
Rubyフィルター(logstash ruby filter)の書き方
grokは文字列を切り出すところまでで、取り出した値の計算や条件付きの加工はできません。そこを埋めるのがRubyフィルター(9.5.5の同梱版はv3.1.8)で、イベントごとに任意のRubyコードを実行できます。値の読み書きはevent.get('[a][b]')とevent.set('[a][b]', 値)で行います。
codeとinitによるインライン記述
codeはイベントごとに、initは起動時に1回だけ実行されます。公式文書はcodeの中でメソッドを定義するとスループットが大きく落ちるとしており、ライブラリの読み込みや閾値の設定はinitに置きます。閾値は大文字で始まる定数ではなく、initで一度だけ設定するインスタンス変数にします(定数はプラグインの外まで影響するため、公式文書が避けるよう求めています)。
filter {
grok { match => { "message" => "^%{WORD:method} %{NOTSPACE:path} %{NUMBER:bytes:int} %{NUMBER:duration_ms:float}$" } }
if "_grokparsefailure" not in [tags] {
ruby {
init => "require 'uri'; @slow_ms = 500.0"
code => "
d = event.get('duration_ms')
event.set('[perf][slow]', d >= @slow_ms)
begin
q = URI.parse(event.get('path')).query
event.set('[url][params]', URI.decode_www_form(q).to_h) if q
rescue URI::InvalidURIError
event.tag('_url_parse_failure')
end
event.set('kb', (event.get('bytes') / 1024.0).round(1))
"
}
}
}
GET /search?q=logstash&page=2 20480 812.4を流すと、[perf][slow] => true、[url][params] => {"q"=>"logstash", "page"=>"2"}、kb => 20.0が付きました。grokのNOTSPACEは空白以外を何でも受けるので、URIとして正しいとは限りません。未エンコードの日本語を含む/検索?q=ログではURI::InvalidURIErrorになるため、rescueで_url_parse_failureタグを付けて先へ流しています。
pathで読み込む外部スクリプトとtestブロック
コードが数行を超えたら.rbファイルに切り出し、pathで読み込みます。ファイル側には、script_paramsを受け取るregister(params)(任意)と、イベントを受けてイベントの配列を返すfilter(event)(必須)を定義します。空配列を返せばそのイベントは破棄され、複数要素を返せば分割になります。
# /etc/logstash/ruby/slow_split.rb
def register(params)
@threshold = params.fetch("threshold_ms").to_f
end
def filter(event)
d = event.get("duration_ms")
return [] if d.nil?
event.set("[perf][slow]", d >= @threshold)
[event]
end
test "閾値以上は slow=true" do
parameters { { "threshold_ms" => 500 } }
in_event { { "duration_ms" => 812.4 } }
expect("slow が true") { |events| events.size == 1 && events.first.get("[perf][slow]") == true }
end
test "duration_ms が無いイベントは破棄" do
parameters { { "threshold_ms" => 500 } }
in_event { { "message" => "no duration" } }
expect("0件になる") { |events| events.empty? }
end
filter {
ruby {
path => "/etc/logstash/ruby/slow_split.rb"
script_params => { "threshold_ms" => 500 }
}
}
ファイル末尾のtestブロックは、パイプラインの起動時と-tでの設定テスト時に実行され、1件でも失敗するとパイプラインは起動しません。9.5.5で-tを付けて実行すると、次の行が出てからConfiguration OKで終わりました。
[INFO ][logstash.filters.ruby.script] Test run complete {script_path: "/t/slow_split.rb", results: {passed: 2, failed: 0, errored: 0}}
実際に3行を流すと、duration_msを取れなかったbad lineは空配列が返って破棄され、出力は2件でした。
params.fetchにしているのは、キー名の書き間違いを起動時に止めるためです。公式文書の例は、設定で'drop_percentage' => 0.5を渡しながらスクリプトではparams["percentage"]を読んでおり、そのまま写すと閾値がnilのまま動き出します。params[...]で読むと、この食い違いはイベントが届くまで表に出ません。
例外時の_rubyexceptionと並行実行の注意
コードが例外を投げると、イベントは止まらずに_rubyexceptionタグ(tag_on_exceptionで変更可)を付けて先へ流れます。ifで囲まず、1行目をevent.set('[perf][slow]', !d.nil? && d >= @slow_ms)とnilを許す書き方にした版でbad lineを流すと、grokが失敗してbytesが無いため、最終行の割り算でNoMethodErrorになり、タグは_grokparsefailureと_rubyexceptionの2つでした。しかも例外の前に設定した[perf][slow] => falseは残っており、途中まで加工された状態で出力されます。上の例のようにif "_grokparsefailure" not in [tags]で囲み、grokが成功した行だけをRubyに通すのが確実です。
並行実行にも注意が要ります。ワーカーが複数あると、1つのRubyフィルターを複数スレッドが同時に使います。@で始まるインスタンス変数は全ワーカーで共有されるため、registerやinitで設定するのは安全でも、イベント処理中に書き換えると競合します。件数を数えるような状態を持たせたい場合は、Rubyフィルターではなくmetricsフィルターや出力先の集計に任せます。
Grok・Dissect・Rubyフィルターの使い分け
grokの公式文書は、Dissectを「正規表現を使わず高速」で「データが確実に繰り返されるときに向く」、grokを「行ごとに構造が変わるときに向く」と説明しています。
| フィルター | 方式 | 向くログ | 失敗時のタグ |
|---|---|---|---|
| dissect | 区切り文字で分割 | 形が毎行同じ | _dissectfailure |
| grok | 正規表現 | 形が行ごとに変わる | _grokparsefailure |
| ruby | 任意のコード | 計算・条件付き加工 | _rubyexception |
判断の順番は、まずDissectで切れるかを考え、切れない部分だけをgrokに回し、それでも書けない加工をRubyに回す、です。Rubyは何でも書ける反面、テストを付けないと例外時の挙動が上のとおり分かりにくくなります。JSONで出力できるアプリなら、そもそもjsonフィルターで済むので、ログを出す側を構造化ログに寄せられないかも検討します。
前半をDissect、残りをgrokで切る併用
先頭の日時・レベル・PIDは固定で、後半だけ形が変わるログの例です。Dissect(v1.3.0)の%{+date}は直前の区切り文字ごと前のフィールドへ追記し、%{?ignored}は値を捨てます。
filter {
dissect {
mapping => { "message" => "%{date} %{+date} [%{level}] %{?ignored} pid=%{pid} %{rest}" }
convert_datatype => { "pid" => "int" }
}
if "_dissectfailure" not in [tags] {
grok {
match => { "rest" => "^%{WORD:verb} %{URIPATHPARAM:path}(?: took=%{NUMBER:took:float}ms)?$" }
}
}
}
2026-10-28 10:15:30 [WARN] worker pid=4123 GET /api/orders?id=10 took=35.2msはdate => "2026-10-28 10:15:30"、pid => 4123、took => 35.2に、tookの無い行はtookなしで分解されました。区切りの無い行には_dissectfailureに加えて_dataconversionnullvalue_pid_intも付きます。Dissectが失敗してもconvert_datatypeは走るので、失敗の判定には_dissectfailureだけを見ます。
Dissectが速いのは区切りが合う行に限られます。上の計測でマッチしない行が成功行の5倍以上遅かったのは、Dissectが失敗した行ごとに「Dissector mapping, pattern not found」のWARNを1行ずつログへ出すためで、10万件の失敗で10万行のWARNが出ました。形の違う行が混ざる入力では、Dissectの前にif [message] =~ /^\d{4}-/のような条件で対象を絞ります。
LogstashのJVMの重さが問題なら、収集側をFluent BitやVectorに替えて変換もそちらで行う選択肢もあります。Vectorの変換言語VRLもparse_grokでgrokパターンを読めるので、移行の際はVectorでログ収集する手順を参照してください。ログ基盤の構築やCI/CDに組み込んだ設定テストの運用は、DevOps・CI/CD導入支援でも支援しています。
よくある質問
grokフィルターとは何ですか?
Logstashのfilter段で使うプラグインで、1行のテキストログに正規表現ベースのパターンを当て、IPアドレスやステータスコードなどのフィールドに分解します。構文は%{パターン名:フィールド名}です。
%{COMBINEDAPACHELOG}でclientipが出ないのはなぜですか?
Logstash 8以降はECS互換が既定のためです。送信元IPは[source][address]、ステータスコードは[http][response][status_code]に入ります。旧来の名前が必要なら、そのgrokにecs_compatibility => disabledを付けるか、pipelines.ymlでパイプライン単位に戻します。
GREEDYDATAとNOTSPACEはどう使い分けますか?
GREEDYDATAは残りをすべて受けるため後戻りが増えます。空白を含まない値はNOTSPACEのように狭いパターンにし、GREEDYDATAは行末の本文を受ける1か所に限ります。
_grokparsefailureが出るのはなぜですか?
パターンにマッチしなかったイベントに付くタグです。空白の数の違い、前段のgrokの失敗で対象フィールドが無い、複数行ログが分割されている、が典型です。段ごとにtag_on_failureを別名にすると、どこで落ちたかを特定できます。
LogstashのRubyフィルターはどう書きますか?
短い処理はcodeにインラインで書き、起動時に1回だけ必要な処理はinitに置きます。長くなったらpathで.rbファイルに切り出し、filter(event)がイベントの配列を返すように書きます。ファイルにtestブロックを書くと、パイプライン起動時と-tで検証されます。