使用logstash的http-poller input插件获取的数据怎么保存到 elasticsearch

我用的是7.12.0版本

我的配置文件整体上是这样的
1、在http-poller里调用了 elasticsearch 的 sql api，获取了一些统计数据

input {
         http_poller {
                   urls => {
                             item => {
                                        method => post
                                        url => "http不允许发布站外链接://localhost:9205/_sql?format=csv"
                                        body => '{"query": "SELECT \u0027test\u0027 AS data_type, time, sum(count) AS count FROM test group by time"}'
                                        headers => {
                                                     "content-type" => "application/json"
                                        }
                             }
                  }
                  codec => "plain"
                  schedule => { cron => "*/2 * * * * *"}
        }
}

能获取到数据
data_type,time,count
test,2021-08-10,1
test,2021-08-11,2

2、我想在过滤器里去匹配出每一行，并且保存到es里，我的配置是这样的，但是是不行的

filter {
            grok {
                      match => { "message" => "test,%{TIMESTAMP:time},%{NUMBER:count}" }
                      add_field => {
                                 "time" => "%{time}"
                                 "count" => "%{count}"
                      }
            }
            mutate {
                     # 删除默认生成的字段
                     remove_field => ["@timestamp", "@version"]
            }
}

3、output配置

elasticsearch {
             ecs_compatibility => disabled
             action => "update"
             doc_as_upsert => true
             hosts => ["localhost:9205"]
             index => "demo"
             document_id => "%{time}"
}