Promtail pipeline_stagesを完全理解したい

promtailのmultilineの設定がよくわからなかったので調べてたら、pipeline_stagesからちゃんと理解する必要がありました
ので、一応最後のmultilineがメインです

ドキュメント

処理順序

設定した順番で処理されます
stageの種類で処理の優先度、とかはないです

ログエントリのstructure

pipeline処理時に内部で保持している構造物
なんとなく頭に入ってると色々理解しやすいです

• Line
  • 流れてきた生ログ
  • 最終的なアウトプットになるもの
  • string
• Timestamp
  • Lokiで読み取る際に使用するtimestamp（のはず）
  • デフォルトはログエントリが作られた時刻が入る？（未確認）
  • time.Time
• Labels
  • アウトプットするログに付与するlabel
  • keyとvalueのmap
• Extracted
  • 各stageでparseしたものを保持するところ
  • keyとvalueのmap

stages

（個人的に）よく使うものを掘っていきます

json

文字列をjsonにparseして、任意のフィールドをExtractedに格納する
parseしたjsonのkeyとExtractedに格納するkeyを同じにする場合は省略できる
これは他のstageでも共通

- json:
    expressions:
      extracted_key: json_key
      common_key:

parseする文字列はデフォルトでLineで、sourceを指定すればExtractedの内容をparseできるので、入れ子のjsonをparseできる

例えば、以下のjson文字列を
{"one": {"two": {"three":"goal"}}}

pipeline_stages:
- json: 
    expressions:
      one:
- json:
    expressions:
      two:
    source: one
- json:
    expressions
      three:
    source: two
- output:
    source: three

こんな流れで完全にparseできる

regex

parseしてExtractedに格納する流れはjson stageと同じで、parseにjsonでなく正規表現を使用する
処理対象はデフォルトでLine、sourceで指定可能
expression: '(?P<keyname>.*)'という形で、Extractedにkeyにkeyname、valueに正規表現でキャプチャした値（この場合は全文）が入る

labels

Extractedの内容をlabelで使用する
指定したkeyがExtractedに存在しない場合は何も起きない（エラーにはならない）

timestamp

ログエントリのTimestampを設定する
設定する値は、sourceでExtracted内のフィールドを指定するので、事前に何かしらのparseをしておく必要がある
フォーマットはプリセットから選ぶ必要があって、自由記述できない

output

Extractedの中から、ログとして出力するフィールドを決める
内部ではLineを置換している
sourceで指定したものがExtractedに存在しない場合は何もしない
なので、outputステージを設定しなかったり、指定してもそれがExtractedに存在しなかったら、生ログをそのまま出力する

docker

以下のstagesのwrapperになっている

- json:
    output: log
    stream: stream
    timestamp: time
- labels:
    stream:
- timestamp:
    source: timestamp
    format: RFC3339Nano
- output:
    source: output

cri

以下のstagesのwrapperになっている

- regex:
    expression: "^(?s)(?P<time>\\S+?) (?P<stream>stdout|stderr) (?P<flags>\\S+?) (?P<content>.*)$",
- labels:
    stream:
- timestamp:
    source: time
    format: RFC3339Nano
- output:
    source: content

multiline

ちゃんと理解するためにだいぶ長くなる
複数行のログの集合体blockを作り、最後に繋げてひとつのログにする

設定項目

• firstline
  • 最初の行であるという判定を行う正規表現
• max_wait_times
  • 後続のログを待つ時間
  • あまり長く設定すると、最新のログが長時間ここで待ちになるのでその間の閲覧ができない
  • 最初の行が出てからではなく、前の行が出てからの間隔
  • デフォルトは3秒
• max_lines
  • ひとつのblockの最大行数
  • デフォルトは128

内部ステータス

multiline専用の内部ステータスがある

• buffer
  • 現在のblockで保持してるログのバッファ
• startLineEntry
  • firstlineに該当したログのログエントリstructure
• currentLines
  • 現在のblockで保持しているログの行数

処理の流れ

• ログが来る
• firstlineに該当するか判定
  • 判定の対象になるのはLine
    • よって、output stageの前か後かで挙動が変わる
  • 該当する場合、flush処理（後述）をして現在のログエントリをまるっとstartLineEntryに入れる
• bufferに Line を追加して、currentLinesを+1する
  • ログは\n（改行）で繋ぐ
  • firstlineに該当したエントリもこの処理は行う
• currentLinesがmax_linesに達している場合、flush処理
• 後続のログを待ち、max_wait_timesを超えるとflush処理
• 先頭へ

flush処理とは

• bufferが空の場合は何もしない
• ログエントリを作成
  • Timestamp Labels Extracted はstartLineEntryのものを使用
  • Lineはbufferを文字列にしたもの
• 作成したログエントリを後続のstageに渡す
• bufferとcurrentLinesをリセットする
• startLineEntryはリセットしない
  • おそらくmax_linesやmax_wait_timesでflushされた際に次のblockでまだ使うため

ユニーク性

上記の処理は、ログエントリのlabelごとに独立して行われる

ポイント

• Lineに対して先頭行の判定を行ったり繋げたり
• Line以外の情報は先頭行のものを使用
• k8sの場合、dockerやcriで1回parseしてからoutputでLineを置き換えてからmultilineで処理する必要がある
• 複数のログを改行で繋いでる
  • これをなくしたいならこの後replaceする必要があり、元のログに改行がある場合は一緒に消えてしまう

おまけ

そもそも今回multilneでやりたかったのは、dockerやcontainerdのログが16kbで分割されるやつの再結合だったんですが
これは現状のmultiline stageでは不可能なようです

dockerの場合

ログ分割が起きてるかどうかを見分ける方法は、本文の終端が\nで終わっているかどうか（だったはず）
終端が\nで終わっていない場合は、ログが分割されていてまだ続きがある（3分割以上の可能性もあるので、先頭とは限らない）
終わっている場合は、分割されたログの最後
-> multilineで判定できるのは先頭行のみなので不可能（最終行で判定できればよかった）

containerdの場合

containerdのログ分割が起きている場合はタグに入るP/Fで判定
-> ログ本文以外での判定はできないので不可能