diff --git a/README.md b/README.md index f4c9d23..34a02b3 100755 --- a/README.md +++ b/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`): + +``` + + @type mysql_replicator + # ... + tag myindex-%Y%m%d.mytype.${event}.${primary_key} + +``` + +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. diff --git a/lib/fluent/plugin/out_mysql_replicator_elasticsearch.rb b/lib/fluent/plugin/out_mysql_replicator_elasticsearch.rb index f7afb32..47cc363 100644 --- a/lib/fluent/plugin/out_mysql_replicator_elasticsearch.rb +++ b/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 diff --git a/test/plugin/test_out_mysql_replicator_elasticsearch.rb b/test/plugin/test_out_mysql_replicator_elasticsearch.rb index 3d6f36f..069ea26 100755 --- a/test/plugin/test_out_mysql_replicator_elasticsearch.rb +++ b/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