fluent-bit: Fluent-bit fails to send logs to an elasticsearch datastream

Bug Report

Describe the bug Fluentbit fails to send logs to an elasticsearch datastream index using the elasticsearch output plugin. This is caused because in datastreams is not allowed to use the operation “index” when a bulk API call is issued to elasticsearch, which is what fluent-bit does. For more info you can see why this is not allowed here: Use of datastreams. Datastreams were introduced in Elasticsearch version 7.9 and it is the recommended way to store logs from now on.

To Reproduce

• Use the elasticsearch output plugin of fluent-bit to send logs to an index in elasticsearch that is configured as a datastream
• Example error message:

"
PS C:\Users\Pablo\fluentbit> C:\"Program Files"\td-agent-bit\bin\fluent-bit.exe -c .\fluent-bit.conf
Fluent Bit v1.5.7                                                                                                       * Copyright (C) 2019-2020 The Fluent Bit Authors
* Copyright (C) 2015-2018 Treasure Data
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2020/10/08 00:11:51] [ info] Configuration:
[2020/10/08 00:11:51] [ info]  flush time     | 5.000000 seconds
[2020/10/08 00:11:51] [ info]  grace          | 5 seconds
[2020/10/08 00:11:51] [ info]  daemon         | 0
[2020/10/08 00:11:51] [ info] ___________
[2020/10/08 00:11:51] [ info]  inputs:
[2020/10/08 00:11:51] [ info]      winlog
[2020/10/08 00:11:51] [ info] ___________
[2020/10/08 00:11:51] [ info]  filters:
[2020/10/08 00:11:51] [ info] ___________
[2020/10/08 00:11:51] [ info]  outputs:
[2020/10/08 00:11:51] [ info]      es.0
[2020/10/08 00:11:51] [ info] ___________
[2020/10/08 00:11:51] [ info]  collectors:
[2020/10/08 00:11:51] [ info] [engine] started (pid=35696)
[2020/10/08 00:11:51] [debug] [engine] coroutine stack size: 98302 bytes (96.0K)
[2020/10/08 00:11:51] [debug] [storage] [cio stream] new stream registered: winlog.0
[2020/10/08 00:11:51] [ info] [storage] version=1.0.5, initializing...
[2020/10/08 00:11:51] [ info] [storage] in-memory
[2020/10/08 00:11:51] [ info] [storage] normal synchronization mode, checksum disabled, max_chunks_up=128
[2020/10/08 00:11:51] [debug] [input:winlog:winlog.0] load channel<Windows PowerShell record=50737 time=1599082024>
[2020/10/08 00:11:51] [debug] [output:es:es.0] host=127.0.0.1 port=9200 uri=/_bulk index=my_index type=create
[2020/10/08 00:11:51] [debug] [router] match rule winlog.0:es.0
[2020/10/08 00:11:51] [ info] [sp] stream processor started
[2020/10/08 00:12:01] [debug] [input:winlog:winlog.0] read 32432 bytes from 'Windows PowerShell'
[2020/10/08 00:12:01] [debug] [input:winlog:winlog.0] save channel<Windows PowerShell record=50767 time=1599082024>
[2020/10/08 00:12:06] [debug] [task] created task=00000144C03816F0 id=0 OK
[2020/10/08 00:12:06] [debug] [output:es:es.0] HTTP Status=200 URI=/_bulk
[2020/10/08 00:12:06] [error] [output:es:es.0] could not pack/validate JSON response
{"took":0,"errors":true,"items":[{"index":{"_index":"my_index","_type":"create","_id":"ca11aa38-1df5-6400-da10-856a55d17c1c","status":400,"error":{"type":"illegal_argument_exception","reason":"only write ops with an op_type of create are allowed in data streams"}}},
errors":true,"items":[{"index":{"_index":"my_index","_type":"create","_id":"ca11aa38-1df5-6400-da10-856a55d17c1c","status":400,"error":{"type":"illegal_argument_exception","reason":"only write ops with an op_type of create are allowed in data streams"

Expected behavior Logs should be received successfully in Elasticsearch

Your Environment

• Version used: 1.5.7 windows version installed with .exe
• Configuration: default fluentbit.conf Windows configuration plus: [INPUT] Name winlog Channels Windows PowerShell Interval_Sec 10 DB winlog.sqlite

[OUTPUT] Name es Match ** Host 127.0.0.1 Port 9200 Index my_index type create

• Operating System and version: Windows
• Filters and plugins: elasticsearch output plugin