シナリオ | Logstash | データ加工 | |
フィールド名を変更 | mutate | mutate { rename => {"old_field_name" => "new_field_name" } } | fields_rename("old_field_name","new_field_name" ) |
フィールドを削除 | | mutate { remove_field => ["password_hash"] } | fields_drop("password_hash") |
フィールド値を更新 | | mutate { update => {"status_code" => "Not Found" } } //status_codeをNot Foundに更新します。 | fields_set("status_code", "Not Found") |
キー値の抽出 - Grok | grok | grok { match => { "message" => "%{TIMESTAMP_ISO8601:time} %{LOGLEVEL:level} " }} | ext_grok("message",grok="%{TIMESTAMP_ISO8601:time} %{DATA:level}") |
キー値抽出 - 区切り文字 | split | mutate { split => { "message" => "|" } add_field => { "time" => "%{[message][0]}" "level" => "%{[message][1]}" "taskId" => "%{[message][2]}" "ProcessName" => "%{[message][3]}" "ip" => "%{[message][4]}" } remove_field => ["message"] } | ext_sepstr("message","time,loglevel,taskId,ProcessName,ip",sep="\\|") fields_drop("message") |
キー値抽出 - JSON | json | json { source => "message" target => "parsed_data" } | ext_json("message") |
ログの削除 | drop | if [status] == 404 { //statusが404の場合 drop { } //ログを削除します。 } | log_drop（ログを削除します。 op_eq(v("status"),404)// statusの値が404の場合。 ) |
論理判断 | if else | if [log] //logフィールドが存在する場合 if "Cost" in [message] //messageフィールド値に「Cost」が含まれる場合 | t_if(has_field("log"))//logフィールドが存在する場合。 t_if ( str_exist(v(message), "Cost", ignore_upper=True) )//messageフィールドの値に「Cost」という文字が含まれる場合。 |
| or , and | if "Cost" in [message] or "cost" in [message] | str_exist(v(message), "Cost", ignore_upper=False), str_exist(v(message), "cost", ignore_upper=False) ) |
複数のsink（ターゲット）へのログ配信 | output | if [container] == "scm-pfc" { elasticsearch { hosts => ["xx.xx.x.xxx:9200"] index => "p-k8s" } } else { elasticsearch { hosts => ["xx.xx.x.xx:9200"] index => "p-container" }} | op_str_eq(v("container"),"scm-pfc"), log_output("p-k8s")、//if分岐。 log_output("p-container")//else分岐。 ) |
フィードバック