Read source filename from beats and redirect output to different Elastic Search index

Elastic Stack Logstash
1

I want to read the source of the csv data and redirect the output to a different ElasticSearch index.
I tried as below (if condition with source in output) by following various discussions but it did not work. It is always going to else part.

Here is my logstash config:

input {
	beats {
		port => "5044"
	}
}
filter {
	csv {
		autodetect_column_names => true
		separator => ","
		remove_field => ["message"]
	}
}

output {
	**if [source] == 'E:\\test_ingest\\test.csv'** {
		elasticsearch {
			hosts => "localhost:9200"
			manage_template => false
			index => "test_index"
			document_type => "nasuni_log"
		}
	} else {
              elasticsearch {
			hosts => "localhost:9200"
			manage_template => false
			index => "b_index"
			document_type => "nasuni_log"
		}
	}
}
2

The approach seems reasonable. What does one of the events look like? Copy and paste from the JSON tab in Kibana Discover.

3

Here is the input example from csv with header:

timestamp(UTC),category,event type,path/from,new path/to,user,group,sid,share/export name,volume type,client IP,snapshot timestamp(UTC),shared link
2018-04-12 12:16:57.846484,Write,Truncate File,/Design/CCH/t3est/review comments.docx,,EMEA\kandrgan,EMEA\domain users,S-1-5-21-1409082233-362288127-725345543-319468,Design,CIFS,10.123.3.58,,

Here is the json from Kibana:
I have only hidden the business related values

{
  "_index": "b_index",
  "_type": "nasuni_log",
  "_id": "AWK_97z0iUaCMb2pr7XB",
  "_version": 2,
  "_score": null,
  "_source": {
    "shared link": null,
    "new path/to": null,
    "offset": 363965,
    "volume type": "CIFS",
    "path/from": "/Design/CCH/t3est/review comments.docx",
    "input_type": "log",
    "source": "E:\\test_ingest\\test1.csv",
    "type": "log",
    "tags": [
      "beats_input_codec_plain_applied",
      "_dateparsefailure"
    ],
    "sid": "S-1-5-21-1409082233-362288127-725345543-319468",
    "@timestamp": "2018-04-13T17:05:01.399Z",
    "event type": "Truncate File",
    "share/export name": "Design",
    "@version": "1",
    "beat": {
      "hostname": "DEHEREMAP9769",
      "name": "DEHEREMAP9769",
      "version": "5.4.0"
    },
    "host": "DEHEREMAP9769",
    "timestamp(UTC)": "2018-04-12 19:27:32.456324",
    "snapshot timestamp(UTC)": null,
    "category": "Write",
    "client IP": "10.143.11.26",
    "user": "EMEA\\kandrgan",
    "group": "EMEA\\domain users"
  },
  "fields": {
    "@timestamp": [
      1523639101399
    ]
  },
  "sort": [
    1523639101399
  ]
}
4

If that is what you see in Kibana then I believe this, with single backslashes, will compare true:

if [source] == 'E:\test_ingest\test.csv'
5

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.