title: Set fluentD elastic-search index dynamically
body: <p>I'm trying to forward logs to elastic-search and got stuck with setting the index dynamically (by field in the input data).</p>
  
  <p>My input data format is JSON and always have the key "es_idx". I wish to forward to elasticsearch by that key and add it timestamp, I use <code>logstash_format</code> true to achieve the timestamp feature and <code>logstash_prefix</code> to set the index name other than "fluentd"</p>
  
  <p>This is how my fluentd config looks like:</p>
  
  <pre><code># fluentd/conf/fluent.conf
  &lt;source&gt;
    type stdin
    # Input pattern. It depends on Parser plugin
    format json
  
    # Optional. default is stdin.events
  &lt;/source&gt;
  
  &lt;match *.**&gt;
    @type copy
    &lt;store&gt;
      @type stdout
    &lt;/store&gt;
    &lt;store&gt;
      @type elasticsearch
      host &lt;es-host&gt;
      port&lt;es-port&gt;
      logstash_format true
      logstash_prefix ${$record["es_idx"]}
      type_name fluentd
      flush_interval 5s
    &lt;/store&gt;
  
  &lt;/match&gt;
  
  </code></pre>
  
  <p>When using the following input {"tenant_id":"test","es_idx":"blabla"}, I'm getting the following error:</p>
  
  <pre><code>2020-05-27 10:38:06 +0300 [warn]: #0 dump an error event: error_class=Fluent::Plugin::ElasticsearchErrorHandler::ElasticsearchError error="400 - Rejected by Elasticsearch" location=nil tag="stdin.events" time=2020-05-27 10:37:59.498450000 +0300 record={"tenant_id"=&gt;"test", "es_idx"=&gt;"blabla"}
  </code></pre>
  
  <p>If I'm setting the logstash_pattern to other string like this: "logstash_pattern blabla" it works fine. </p>
  
  <p>Does anyone have a clue what can be the issue?</p>