请教个 ELK 中, logstash 根据不同主机的 `tags` 来指定索引的问题

现在有多台主机, 需要把不同的日志分到不同的 es 索引上. 目前的 logstash 配置是:

 input {
       beats {
         port => 9090
         client_inactivity_timeout => 3600
      }
    }
    output {
      if [tags] in "oa" {
        elasticsearch {
          action => "index"
          hosts => ["http://es:9200"]
          index  => "oa-log-%{+YYYY-MM-dd}"
          user => "user"
          password => "user_pwd"
        }
      }
      if [tags] in "kernel-log" {
        elasticsearch {
          action => "index"
          hosts => ["http://es:9200"]
          index  => "kernel-log-%{+YYYY-MM-dd}"
          user => "user"
          password => "user_pwd"
        }
  }
     } 

其中一台主机的 filebeat 配置:

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/oa/*.log
  tags: ["oa-1-192-ACCESS"]
  fields:
    index: oa-log-%{+YYYY-MM-dd}
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: true
output.logstash:
  hosts: ["logstash:9090"]
  enabled: true

没有根据 tags 指定索引的时候, 是正常的.

现在添加 if 字段后, 发现 es 上没有自动创建这个索引. es 中也允许自动创建 oa*, kernel* 这俩索引了 ES 新手. 麻烦请教下这个问题