前提・実現したいこと

fluentd×Logstash×Elasticsearch×Kibana環境で、サーバーのアクセスログ監視をしています。
Logstashのfilter機能を使って、同ログにおけるフィールド間の差分を計算したいです。

fluentedから転送されたアクセスログに、接客開始時間(start_time: string)、接客終了時間(end_time: string)フィールドが存在します。Logstashのfilter機能を用いて、これらのフィールド間の差分（接客終了時間 - 接客開始時間 = 接客時間）を計算、その値「接客時間(service_time:date)」を新フィールドに追加した状態で、Elasticsearchにデータ転送したいです。

ログデータ加工の流れ

サンプルログ

00001~~問い合わせ~~2019-11-05 10:00:00~2019-11-05 10:10:00

データ加工前


{
 
  "number": "00001",                 // 受付番号
  "business_type": "問い合わせ",　    // 業務種別
  "start_time": "2019-11-05 15:06:24",　// 接客開始時間
  "end_time": "2026-09-28 15:13:39"　　 // 接客終了時間
}

データ加工後


{
 
  "number": 00001,　　    // 受付番号
  "business_type": "問い合わせ",　// 業務種別　
  "start_time": 2019-11-05 15:06:24,   // 接客開始時間
  "end_time": 2026-09-28 15:13:39　　　 // 接客終了時間
  "service_time: 00:10:00"　　　　　　　 // 接客時間
}

やったこと・対応中

【済】接客開始時間(start_time)と接客終了時間(end_time)のデータ型を、stringからdate型に変更する
【済】アクセスログに新規フィールド「接客時間(service_time:date)」を追加
【調査中】接客開始時間と接客終了時間の差分を計算して、その値をservice_timeフィールドに入れたい

logstash
1
2// 接客開始時間(start_time)と接客終了時間(end_time)のデータ型を、stringからdate型に変更する
3    date {
4      match => [ "start_time", "YYYY-MM-dd HH:mm:ss", "YYYY-MM-dd HH:mm:ss" ]
5      timezone => "Asia/Tokyo"
6      target => "start_time"
7    }
8
9    date {
10      match => [ "end_time", "YYYY-MM-dd HH:mm:ss", "YYYY-MM-dd HH:mm:ss" ]
11      timezone => "Asia/Tokyo"
12      target => "end_time"
13    }
14
15// アクセスログに新規フィールド「接客時間(service_time:date)」を追加
16// 【調査中】接客開始時間と接客終了時間の差分を計算して、その値をservice_timeフィールドに入れたい
17
18    mutate {
19     add_field => {
20       "service_time" => "???"   ※新規フィールドは追加できましたが、フィールドの値に"接客時間"を入れる方法が不明です
21     }
22    }
23