LogstashでのRubyスクリプト
Logstash パイプラインでの高度なデータ変換を実現する Logstash Ruby フィルター プラグインについて学習します。
Logstash は、複数のソースからデータを取り込み、変換して、選択した宛先に送信するデータ処理パイプラインです。フィルター プラグインはこのプロセスの鍵であり、データがパイプラインを通過するときにデータに対して特定の操作を実行します。
Logstash には、データの解析、拡充、変更などの一般的なタスク用のフィルターがいくつか組み込まれています。しかし、場合によっては、これらの標準フィルターが提供できる範囲を超えたカスタム ロジックが必要なシナリオに遭遇することがあります。ここでRuby フィルター プラグインが役立ちます。
Ruby フィルター プラグインを使用すると、Logstash パイプライン内でカスタム Ruby コードを直接実行できます。標準フィルターでは不十分な場合は、Ruby フィルターを使用すると、複雑なデータ変換を処理したり、カスタム ビジネス ロジックを実装したり、外部システムと統合したりできます。
このブログでは、Ruby フィルターの使い方を、基本的な使い方から高度な使い方まで解説します。
Ruby フィルターはいつ使用すればよいですか?
Elastic のコンサルティング アーキテクトとして、Logstash は現在最先端のデータ処理エンジンではないにもかかわらず、データ処理パイプラインに Logstash を使用している顧客をよく見かけます。複雑なデータ操作やカスタム ロジックに関しては、標準フィルターの制限に悩まされることがよくあります。このような場合、Ruby フィルターはこれらの課題を克服するのに役立ちます。
Ruby フィルターは、標準の Logstash フィルターが特定の要件を満たせない場合に役立ちます。一般的な使用例をいくつか紹介します。
深くネストされたデータ操作: 複雑な JSON 構造、配列内の配列を変更したり、コンテンツに基づいてデータを動的に再構築したりできます。
高度な文字列処理: 非構造化テキストから構造化データを解析して抽出します
複雑なビジネス ロジックの実装: 条件付きロジック、ループ、または複雑な計算を必要とするカスタム変換を作成します。
基本的な使い方
Ruby フィルターがどのように機能するかを理解するために、簡単な例から始めましょう。
Rubyフィルターの設定
Logstash パイプラインを作成するときは、構成ファイルを/etc/logstash/conf.dディレクトリに配置する必要があります。あるいは、Logstash を手動で起動するときに-fオプションを使用して構成ファイルへのパスを指定し、パイプラインを簡単に試すことができます。
$ ./bin/logstash -f /path/to/your_pipeline.conf構成ファイルには.conf拡張子が必要です。
Ruby フィルターを使用するには、Logstash パイプライン構成 (*.conf) ファイルのフィルター セクションでrubyフィルターを定義します。基本的な例は次のとおりです。
filter {
ruby {
code => "
event.set('new_field', 'Hello from Ruby!')
"
}
}このインライン Ruby フィルターは、Logstash 構成内で Ruby フィルター インスタンスを定義します。codeパラメータは、このフィルタによって処理される各イベントに対して Logstash が実行するインライン Ruby スクリプトを提供します。そのスクリプト内には、イベント自体を表すevent変数があります。イベント オブジェクトには、Logstash に送信された元のデータと、Logstash のフィルター ステージ中に作成された追加フィールドが含まれます。これらのフィールドには、 event.get()やevent.set()などの Logstash イベント API を介してアクセスできます。このサンプルコードでは、 event.set('new_field', 'Hello from Ruby!') new_fieldという名前の新しいフィールドを文字列値Hello from Ruby!に設定します。必要に応じて、このcodeブロックに他のコードを追加できます。
このeventオブジェクトはキーバリュー型のデータコンテナとして機能しますが、通常の Ruby のハッシュオブジェクトではないことに注意してください。イベント API の詳細については、この公式ドキュメントをご覧ください。
Rubyスクリプトを外部化する
単純な変換には、インライン Ruby コードが便利です。ただし、複雑なロジックや再利用可能な関数の場合は、コードを外部の Ruby スクリプトに移動することをお勧めします。これにより保守性が向上し、Logstash パイプラインの構成がクリーンな状態に保たれます。
まず、Ruby スクリプトを作成し、 my_ruby_script.rbとして保存します。スクリプトはイベントを処理するfilterメソッドを定義する必要があります。処理中の現在のイベントを表すイベント オブジェクトを引数として受け取ります。filterメソッドは、発行するイベントの配列を返す必要があります。イベントを削除するには、空の配列を返します。
たとえば、次のスクリプトはmessageフィールドを読み取り、その長さを計算し、結果をmessage_lengthという新しいフィールドに格納します。
def register(params)
# This method is called when the plugin is loaded.
# You can use it to initialize any instance variables or perform setup tasks.
end
def filter(event)
message = event.get('message')
if message
event.set('message_length', message.length)
end
return [event]
end次に、 pathオプションを使用してスクリプトを参照するように Ruby フィルター構成を設定します。これにより、Logstash は外部スクリプトを読み込んで実行します。外部スクリプトを使用する場合は、ファイルが存在し、適切な権限があることを確認してください。
filter {
ruby {
path => "/path/to/my_ruby_script.rb"
}
}これで、各イベントはmy_ruby_script.rbのfilterメソッドに渡され、処理されるようになります。
このアプローチにより、複雑なロジックをより効率的に管理できるようになり、Ruby コードのテスト、デバッグ、再利用が容易になります。
高度な使用法
このセクションでは、Logstash で Ruby フィルターを使用する高度な例をいくつか紹介します。これらの例では、Ruby を使用してデータ変換を実行し、イベントを強化し、カスタム ロジックを実装する方法を示します。
ネストされたデータ構造の操作
Logstash イベントは、Logstash が処理するコア データ構造です。配列やハッシュなどのネストされたデータ構造を含むさまざまなフィールドを含めることができます。Ruby フィルターを使用すると、これらのネストされた構造を簡単に操作できます。
Ruby フィルターはハッシュや配列などのネストされたデータ構造を処理できるため、これらの構造内のフィールドを変更したり追加したりできます。これは、JSON のような複雑なデータ形式を扱うときに便利です。
input {
generator {
lines => [
'{"nested": {"key1": "value1", "key2": "value2"}}'
]
count => 1
codec => "json"
ecs_compatibility => "disabled"
}
}
filter {
ruby {
code => "
nested_data = event.get('nested')
if nested_data.is_a?(Hash)
nested_data['key3'] = 'value3'
event.set('nested', nested_data)
end
"
}
}
output {
stdout { codec => rubydebug }
}この例では、入力データにネストされた JSON オブジェクトが含まれています。Ruby フィルターは、新しいキーと値のペアを追加してネストされたデータを変更します。ネストされたデータに対するこの種の操作は標準の Logstash フィルターでは不可能であるため、Ruby フィルターは複雑なデータ構造に便利なオプションになります。
1つのイベントを複数のイベントに分割する
Ruby フィルターを使用して、単一のイベントを複数のイベントに分割することもできます。これは、アイテムの配列を含む単一のイベントがあり、アイテムごとに個別のイベントを作成する場合に便利です。
Elasticsearch の取り込みパイプラインも Beats/Elastic Agent のプロセッサもイベントの分割をサポートしていないことに注意してください。これは Logstash の最も強力な使用例の 1 つです。
分割フィルター付き
splitフィルターを使用すると、指定されたフィールドに基づいてイベントを複数のイベントに分割できます。ただし、分割中に追加の変換やロジックを実行する必要がある場合は、分割フィルターと組み合わせて Ruby フィルターを使用できます。
次の例では、RSS フィードが 1 行の XML テキストとして存在します。複数の<item>要素が含まれています。Ruby フィルターは、XML から<item>要素を抽出し、 itemsという新しいフィールドに保存するために使用されます。次に、分割フィルターを使用して、 itemsフィールドに基づいてイベントを複数のイベントに分割します。
input {
generator {
lines => [
'<rss version="2.0"><channel><title>Sample RSS</title><item><title>Article 1</title><link>http://example.com/1</link><description>Desc 1</description></item><item><title>Article 2</title><link>http://example.com/2</link><description>Desc 2</description></item></channel></rss>'
]
count => 1
codec => "plain"
ecs_compatibility => "disabled"
}
}
filter {
xml {
source => "message"
target => "rss"
store_xml => true
force_array => false
}
ruby {
code => "event.set('items', event.get('[rss][channel][item]')) if event.get('[rss][channel][item]')"
}
split {
field => "items"
}
ruby {
code => "
item = event.get('items')
event.set('title', item['title']) if item['title']
event.set('link', item['link']) if item['link']
event.set('description', item['description']) if item['description']
"
}
mutate {
remove_field => ["@timestamp", "@version", "sequence", "host", "event", "message", "rss", "items"]
}
}
output {
stdout { codec => rubydebug }
}出力は次のようになります:
{
"title" => "Article 1",
"link" => "http://example.com/1",
"description" => "Desc 1"
}
{
"title" => "Article 2",
"link" => "http://example.com/2",
"description" => "Desc 2"
}お気づきかもしれませんが、この場合、 rubyフィルターは必須ではありません。splitフィルターを使用すると、 itemsフィールドに基づいてイベントを複数のイベントに分割できます。また、 mutateフィルターを使用すると、不要なフィールドを削除できます。ただし、分割中に追加の変換やロジックを実行する必要がある場合は、Ruby フィルターを使用できます。
インラインRubyスクリプトを使用する
インライン Ruby スクリプトを使用して、 event.cloneメソッドとnew_event_block variable ( new_event_block.call(new_event)など) を使用することで、単一のイベントを複数のイベントに分割することもできます。これにより、元のイベントのデータを保持しながら、元のイベントに基づいて新しいイベントを作成できます。
Ruby フィルターを使用して 1 つのイベントを複数のイベントに分割する方法の例を次に示します。入力と出力は前の例と同じです。
filter {
xml {
source => "message"
target => "rss"
store_xml => true
force_array => false
}
ruby {
code => "
items = event.get('[rss][channel][item]')
if items.is_a?(Array)
items.each do |item|
new_event = event.clone
new_event.set('title', item['title'])
new_event.set('link', item['link'])
new_event.set('description', item['description'])
new_event_block.call new_event
end
event.cancel
elsif items.is_a?(Hash)
event.set('title', items['title'])
event.set('link', items['link'])
event.set('description', items['description'])
end
"
}
mutate {
remove_field => ["@timestamp", "@version", "sequence", "host", "event", "message", "rss", "items"]
}
}外部Rubyスクリプトを使用する
外部の Ruby スクリプトを使用して、単一のイベントを複数のイベントに分割することもできます。
設定ファイル:
filter {
xml {
source => "message"
target => "rss"
store_xml => true
force_array => false
}
ruby {
path => "path/to/ruby/split_event.rb"
}
mutate {
remove_field => ["@timestamp", "@version", "sequence", "host", "event", "message", "rss", "items"]
}
}Ruby スクリプトはsplit_event.rbとして外部化する必要があります:
def filter(event)
items = event.get('[rss][channel][item]')
events = []
if items.is_a?(Array)
items.each do |item|
new_event = event.clone
new_event.set('title', item['title'])
new_event.set('link', item['link'])
new_event.set('description', item['description'])
events << new_event
end
return events
elsif items.is_a?(Hash)
event.set('title', items['title'])
event.set('link', items['link'])
event.set('description', items['description'])
return [event]
else
return []
end
endfilterメソッドはイベントの配列を返す必要があることに注意してください。受信したイベント オブジェクトを複製して配列に追加することで複数のイベントを返すことも、1 つの要素を持つ配列として 1 つのイベントを返すこともできます。
return events
# or
# return [event]これにより、単一のイベントを複数のイベントに分割できます。
外部コマンドを実行し、その出力を解析する
Logstash exec 入力プラグインを使用すると、外部コマンドを実行でき、その出力は Logstash のイベントになります。コマンドの出力はイベントのmessageフィールドに保存されます。
通常、システム コマンドの出力は人間が判読できますが、Logstash が簡単に解析できる JSON やその他の形式として構造化されていません。これを処理するには、Ruby フィルターを使用して出力を解析し、そこから情報を抽出します。
以下は、 exec入力プラグインを使用して、Unix 系システムで実行中のすべてのプロセスを一覧表示するps -efコマンドを実行する例です。出力は Ruby フィルターによって解析され、各プロセスに関する関連情報が抽出されます。
input {
exec {
command => "ps -ef"
interval => 60
}
}
filter {
ruby {
code => '
processes = []
lines = event.get("message").split("\n")
lines.each_with_index do |line, index|
# Skip header line and empty lines
next if index == 0 || line.strip.empty?
entry = nil
# Use regex to match the ps -ef output format more flexibly
# This pattern accounts for variable spacing and different time formats
if line =~ /^\s*(\S+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\S+)\s+(\S+)\s+([\d:]+\.?\d*)\s+(.+)$/
uid, pid, ppid, c, stime, tty, time, cmd = $1, $2, $3, $4, $5, $6, $7, $8
entry = {
"UID" => uid,
"PID" => pid,
"PPID" => ppid,
"C" => c,
"STIME" => stime,
"TTY" => tty,
"TIME" => time,
"CMD" => cmd.strip
}
elsif line =~ /^\s*(\S+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(.+)$/
# Fallback pattern for lines that might not match the exact format
# Split the remaining part more carefully
uid, pid, ppid, c, remainder = $1, $2, $3, $4, $5
# Split remainder into STIME, TTY, TIME, CMD
parts = remainder.strip.split(/\s+/, 4)
if parts.length >= 4
stime, tty, time, cmd = parts[0], parts[1], parts[2], parts[3]
entry = {
"UID" => uid,
"PID" => pid,
"PPID" => ppid,
"C" => c,
"STIME" => stime,
"TTY" => tty,
"TIME" => time,
"CMD" => cmd
}
end
end
if entry && entry["UID"] == "0"
original_line = line.strip
entry["original_line"] = original_line if original_line.length > 0
processes.push(entry)
end
end
event.set("processes", processes)
event.remove("message")
event.remove("event")
'
}
}
output {
stdout { codec => rubydebug }
}この例では、 exec入力プラグインを使用して、60 秒ごとにps -efコマンドを実行します。Ruby フィルターは出力を処理し、UID、PID、PPID、CPU 使用率 (C)、開始時刻 (STIME)、TTY、合計 CPU 時間 (TIME)、実行されたコマンド (CMD) などの関連フィールドを抽出します。私の macOS 環境では問題なく動作しますが、システムのps -efコマンドの出力形式に合わせて正規表現パターンを調整する必要がある場合があります。
組み込みライブラリを使用する
Ruby フィルター プラグインを使用すると、さまざまなタスクに非常に役立つ組み込みの Ruby ライブラリを使用できます。たとえば、 jsonライブラリを使用して JSON 文字列を解析したり、 dateライブラリを使用して日付を操作したりできます。
以下は、 jsonライブラリを使用して、フィールドに格納されている JSON 文字列を解析する例です。
require 'json'
def filter(event)
json_string = event.get('message')
parsed_json = JSON.parse(json_string)
event.set('parsed_json', parsed_json)
return [event]
end毎回ライブラリを要求するのを避けるには、Ruby フィルター スクリプトの先頭でrequireステートメントを使用できるように Ruby コードを外部化する必要があります。これにより、ライブラリが一度読み込まれ、スクリプトで使用できるようになります。
環境で使用できるライブラリを確認するには、Ruby フィルターで次のコードを実行して、組み込みライブラリを一覧表示できます。
Gem.loaded_specs.sort_by { |name, _| name }.each do |name, spec|
puts "#{name}: #{spec.version}"
end注意:組み込みライブラリは Logstash では正式にサポートされていないため、その動作は変更される可能性があり、将来のバージョンでは利用できなくなる可能性があります。自己責任でご使用ください。
まとめ
Logstash Ruby フィルターを使用すると、Logstash パイプラインの機能をカスタマイズおよび拡張できます。この記事では、Ruby フィルターの基本的な使用方法を説明し、高度な使用例を示しました。
Ruby フィルターを活用することで、カスタム ロジックや高度な操作を必要とする複雑なデータ処理タスクを処理できます。ネストされたデータ構造を操作したり、イベントを分割したり、複雑または非構造化テキストを解析して構造化 JSON に変換したりする場合でも、Ruby フィルターは特定の要件を満たす柔軟性を提供します。
このガイドが、Logstash Ruby フィルターの可能性を最大限に引き出すための知識とインスピレーションを提供できたことを願っています。スクリプト作成を楽しんでください!
