Splitting a string to fields and values

[mhensen (Michael Hensen)]

I am using a kubernetes platform an have setup central elasticsearch logging with fluent..
Data comes in nicely but now I want to setup grafana to filter and present any needed logging
So no logstash with a grok option..

But the data I am after is in a field 'log' that is a combined string seperated by '|'.
How can I split this to seperate fields? so that I can use it for easier presentation?

here is a sample document
{
"_index": "fluent-2019.01.31",
"_type": "flb_type",
"_id": "6pCPpWgBjd-H6BwdavgR",
"_version": 1,
"_score": null,
"_source": {
"@timestamp": "2019-01-31T20:17:45.411Z",
"log": "2019-01-31 20:17:45.4112||DEBUG|INow.Kernel.Classes.LogQueue|Correct send message from queue: callinfodata to HTTP://inow-logger-service in 1ms \n",
"stream": "stdout",
"time": "2019-01-31T20:17:45.411593143Z",
"kubernetes": {
"pod_name": "inow-rest-service-5475bbd85-757wp",
"namespace_name": "development",
"pod_id": "5c0ec433-253d-11e9-b98a-005056011bd5",
"labels": {
"k8s-app": "inow-rest-service",
"k8s-domain": "nominowservices.com",
"pod-template-hash": "5475bbd85"
},
"annotations": {
"cni.projectcalico.org/podIP": "192.168.3.208/32"
},
"host": "worker2",
"container_name": "inow-rest-service",
"docker_id": "124a91c646324941fa4e8070b5370d65adfc78f36f9733f53b942c4703d1c339"
}
},
"fields": {
"@timestamp": [
"2019-01-31T20:17:45.411Z"
],
"time": [
"2019-01-31T20:17:45.411Z"
]
},
"highlight": {
"kubernetes.labels.k8s-app": [
"@kibana-highlighted-field@inow@/kibana-highlighted-field@-@kibana-highlighted-field@rest@/kibana-highlighted-field@-@kibana-highlighted-field@service@/kibana-highlighted-field@"
]
},
"sort": [
1548965865411
]
}

[Christian_Dahlqvist]

If you do not want to use Logstash (which would be a solution) I can see the following options:

1. Perform the processing in FluentD. I have not used FluentD, so do not know if this is possible nor how easy it is.
2. Create an ingest pipeline in Elasticsearch and have FluentD direct data to this. I do not know if FluentD is capable of this, but if you are on the latest version of Elasticsearch you can declare a default pipeline per index, which might also you to get around that.
3. Switch from FluentD to Filebeat, which does allow the use of ingest node pipelines in Elasticsearch.
4. Index exactly as you do now, but periodically use the reindex API together with one or more ingest node pipelines to reindex new data and add structure to it. This is not real-time and quite ugly, but might be worth mentioning anyway.

[mhensen (Michael Hensen)]

I managed to create a pipeline and set it as default..

This is working great and also acts as a filter for unwanted data..
Thanks for you reply then set me on the right track!