logstashで現在より古い日付のelasticsearchのインデックス名を指定する方法

やりたいこと

elasticsearchのインデックスを、
logstash-%{+YYYYMMdd}という日付を入れた名称にして、
curatorでデータのローテーションをしている。

以前書いた
elasticsearchのlogstashで同一IDのデータを更新/マージする方法（updateとdoc_as_upsert）
でデータ処理をしていたところ、
更新する分のデータについて、現在のインデックスではなく、
１日前の日付のインデックスに更新を行いたいシチュエーションがあった。

現在ではない日付のindex名をlogstashで指定しようとしたところ、
調べた結果、一工夫必要だったので、メモとして残す。

logstashの一般的なインデックス名の指定方法

logstashのoutputを下記のように設定して、
indexの箇所のYYYYMMddで現在の日付（UTC）の付与されたindex名になる。

output {
  elasticsearch {
    hosts => [ "ElasticsearchのIPアドレス:9200" ]
    index => "logstash-%{+YYYYMMdd}"
  }
}

インデックス名を古い日付を指定する方法

上記の設定だと現在の日付(UTC)になるが、データの更新で１世代前の日付にしたい。
以下を参考に調べた。

Elasticsearch output pluginのindexの説明
https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-index

LS uses Joda to format the index pattern from event timestamp.

YYYYMMddの箇所は、elasticsearchの@timestampから取られている。
システム時間から取られているわけではない。
フォーマットは書いてあっても、時間をずらす方法は記載なし。

Logstash output Indexname with old Date
https://discuss.elastic.co/t/logstash-output-indexname-with-old-date/123433
こちらにもYYYYMMddの箇所は、elasticsearchの@timestampから取られている説明あり。
古い日付のindexを指定するには、@timestampを上書きするか、@metadataを使う方法が提示されている。
後者はプラグインとして、logstash-filter-date_formatterというのリンクがある。

Elasticsearchのタイムゾーン問題
https://qiita.com/hssh2_bin/items/0c4cbd1157ea230549a9
@metadataを使った設定例で、ruby filter pluginで日付を生成して、@metadataに投入している。
実雪堤としてはこちらのサイトの設定をほぼ流用した。

実際の設定例

現在の日付の１時間前の日付のインデックス名を指定する。
ruby filterでrubyのコードを直接書くことができるので、それで１時間前の日付を取得し、@metadataを利用して、outputのindexにする。

Elasticsearchのタイムゾーン問題
https://qiita.com/hssh2_bin/items/0c4cbd1157ea230549a9
からの説明の引用と、設定例の補足。

• ruby filterのevent.setで(field,value)を作成しています。(rubyのputs的な処理)
• field nameは[@metadata][local_time]としていますが、[@metadata]を付与することでElasticsearchにIndexingされないfieldにすることができます。
• strftime('%Y-%m-%d')で取得した時刻を文字列に変換しています。

補足１：下記の設定例では[@metadata][delay_date]に指定しています
補足２：下記の設定例では('%Y%m%d')に指定しています。strftimeの日付の記法は、こちらのリファレンスを参照

下記で使用したrubyのTimeのリファレンス
https://docs.ruby-lang.org/ja/latest/class/Time.html
リファレンスだとTimeはローカルタイムを返すようだったので、
下記の例では、UTCで合わせるためにgetutcをstrftimeの前に置いて変換している。
※現在の方のインデックスであるlogstash-%{+YYYYMMdd}"はUTCなので。

filter {
  ruby {
    code => "
     t = Time.now;
     t2 = t - ( 60 * 60 );
     event.set('[@metadata][delay_date]',t2.getutc.strftime('%Y%m%d'))"
  }
}

output {
  elasticsearch {
    hosts => [ "ElasticsearchのIPアドレス:9200" ]
    index => "logstash-%{[@metadata][delay_date]}"
  }
}