feat: date-based Elasticsearch index names via strftime (#27)

README.md

@@ -122,6 +122,31 @@ Notes:
 * Malformed JSON and non-string values are left untouched, so enabling the option
   never corrupts non-JSON data.
 
+## Date-based index names
+
+`mysql_replicator_elasticsearch` resolves the target index name from the tag
+(via `tag_format`). If that index-name segment contains `strftime` tokens such
+as `%Y%m%d`, they are expanded using the record's event time, so you can create
+Logstash-style dated indices like `myindex-20180831`.
+
+Put the tokens in the index-name part of the input plugin's `tag` (the segment
+must not contain `.`, so use `%Y%m%d` or `%Y-%m-%d`):
+
+```
+<source>
+  @type mysql_replicator
+  # ...
+  tag   myindex-%Y%m%d.mytype.${event}.${primary_key}
+</source>
+```
+
+Index names that contain no `%` are left unchanged, so this is fully backward
+compatible.
+
+> **Note on deletions:** delete events target the index computed from the delete
+> event's own time, so date-rotated indices are best suited to insert-only data
+> (a record inserted on a previous day lives in that day's index).
+
 ## Output example
 
 It is a example when detecting insert/update/delete events.

lib/fluent/plugin/out_mysql_replicator_elasticsearch.rb

@@ -66,7 +66,7 @@ def write(chunk)
 
     chunk.msgpack_each do |tag, time, record|
       tag_parts = tag.match(@tag_format)
-      target_index = tag_parts['index_name']
+      target_index = resolve_index_name(tag_parts['index_name'], time)
       target_type = tag_parts['type_name']
       id_key = tag_parts['primary_key']
 
@@ -105,6 +105,17 @@ def new_http
     http
   end
 
+  # Expand strftime tokens (e.g. "%Y%m%d") in the index name using the record's
+  # event time, enabling date-based indices such as "myindex-20180831". Index
+  # names without a "%" are returned unchanged.
+  def resolve_index_name(index_name, time)
+    return index_name unless index_name && index_name.include?('%')
+    Time.at(time.to_i).strftime(index_name)
+  rescue => e
+    log.warn "mysql_replicator_elasticsearch: failed to expand index name '#{index_name}': #{e.message}"
+    index_name
+  end
+
   # Mapping types were removed in Elasticsearch 8.x and deprecated in 7.x.
   # Detect the major version once and omit "_type" for 7.x and later.
   def detect_type_suppression

test/plugin/test_out_mysql_replicator_elasticsearch.rb

@@ -61,6 +61,24 @@ def test_writes_to_speficied_index
     assert_equal('myindex', index_cmds.first['index']['_index'])
   end
 
+  def test_expands_strftime_tokens_in_index_name
+    stub_elastic
+    time = Time.utc(2018, 8, 31, 12, 0, 0).to_i
+    driver.run(default_tag: 'myindex-%Y%m%d.mytype.insert.id') do
+      driver.feed(time, sample_record)
+    end
+    expected = "myindex-#{Time.at(time).strftime('%Y%m%d')}"
+    assert_equal(expected, index_cmds.first['index']['_index'])
+  end
+
+  def test_index_name_without_token_is_unchanged
+    stub_elastic
+    driver.run(default_tag: 'plainindex.mytype.insert.id') do
+      driver.feed(sample_record)
+    end
+    assert_equal('plainindex', index_cmds.first['index']['_index'])
+  end
+
   def test_writes_to_speficied_type
     driver.configure("type_name mytype\n")
     stub_elastic