Elasticsearch not able to identify the time field defined in Fluentd

[saurabh122]

Issue: I have the logs getting logged into my log file and each log entry has the time information. The time field in the log is not getting recognized in Elasticssearch and hence I am not able to select this time information while creating my Kibana indexes.

**Environment:**
Ubuntu 18.04
Fluentd 1.12.3
td-agent 4 (curl -L https://toolbelt.treasuredata.com/sh/install-ubuntu-bionic-td-agent4.sh | sh)
elasticsearch-7.12.1
kibana-7.12.1

**td-agent.conf:**
<source>
@type tail
path /var/log/sau.log
pos_file /var/log/td-agent/sau.pos
read_from_head true
tag log
<parse>
@type regexp
expression /^\[(?<logtime>[^\]]*)\] (?<name>[^ ]*) (?<title>[^ ]*) (?<id>\d*)$/
time_key logtime
keep_time_key true
time_format %Y-%m-%d %H:%M:%S %z
types id:integer
</parse>
</source>
<match *.**>
@type elasticsearch
host localhost
port 9200
index_name sauindx1
</match>

log entry sample: [2013-02-28 12:00:00 +0900] sau engineer 1
log entry simulation command: echo “[2013-02-28 12:00:00 +0900] sau engineer 1” | tee -a /var/log/sau.log
Official doc referred: regexp - Fluentd

log entry in stdout through fluentd:
2013-02-28 03:00:00.000000000 +0000 log: {“logtime”:“2013-02-28 12:00:00 +0900”,“name”:“sau”,“title”:“engineer”,“id”:1}

As shown in the above line, when I do an stdout print from Fluentd, I get the above line which clearly indicates that the time field defined by me is recognized properly.
However when I divert the logs to Elasticsearch and then create the Kibana index, I am not able to get the time field recognized.

Is there anything I am doing wrong here? Needed some guidance on this please.

Reference links:
Elasticsearch installation – Install Elasticsearch with Debian Package | Elasticsearch Guide [7.12] | Elastic

Kibana installation – Install Kibana with Debian package | Kibana Guide [7.12] | Elastic

Fluentd installation – Install by DEB Package (Debian/Ubuntu) - Fluentd

Regex expression – regexp - Fluentd

[warkolm (Mark Walkom)]

What is the mapping for the index that fluentd is creating in Elasticsearch?

[saurabh122]

Hi Mark,

It is showing up as a string for the time field 'logtime' marking it as searchable. The field _index takes the proper name as defined in the .conf file.
I have attached the same.
Regards, Saurabh

[warkolm (Mark Walkom)]

I am not familiar with fluentd, but I know that Logstash has the ability to set data as a specific type. eg using the date filter to convert the value into ISO8601 which Elasticsearch then automatically understands.

Does it have that ability?
If not then you will need to create a custom index template that correctly sets the format before you ingest the data.

[saurabh122]

So you are saying I need to convert the string format here to a date format in fluentd before it is ingested to elasticsearch?

[warkolm (Mark Walkom)]

Ideally, yes. Can it do that?

[saurabh122]

Hi Mark,
I was able to make it work thanks to the pointers that you had suggested. I did the format conversion to ISO8601 using the below snippet in the configuration file:

<filter *.**>
  @type record_transformer
  enable_ruby
  auto_typecast true
  <record>
    datetime_received ${require 'time'; Time.parse(record["logtime"].to_s + " +0530").iso8601.to_s}
  </record>
</filter>

Wanted to thank you for all the help you have extended. Appreciate it.

Regards,
Saurabh