Logstash file input - output to different indices

Elastic Stack Logstash
1

Hello All,

I am running ELK 7.6.2 stack.

In my current set up, a single file gets ingested in logstash and creates an index successfully as below (only relevant part shown):

input {
  file {
    path => "/opt/gtal/ital/elasticsearch/app/logstash/stage/MQA_PRD_STATS-*.txt"

             - - - - - - - - - - - 
             - - - - - - - - - - - 
             - - - - - - - - - - - 
             - - - - - - - - - - - 

output {
      file {
       path => "/opt/gtal/ital/elasticsearch/logs/mqa/rubydebug.txt"
       codec => rubydebug
     }


    elasticsearch {
     hosts => [ "xxxxxxx:43045","xxxxxxxx:43045","xxxxxxxx:43045","xxxxxxxx:43045" ]
     user => "elastic"
     password => "xxxxxxxx"
     index => "demo-csv-%{+YYYY.MM.dd}"
     doc_as_upsert => true
     action => "update"
     document_id => "%{my_fingerprint}"
  }
}

Now I want to ingest 2 files instead of one in the same logstash file and direct the output to two different indices. The resultant logstash config file would looks something like below:

input {
  file {
    path 	=> 	[
			"/opt/gtal/ital/elasticsearch/app/logstash/stage/MQA_PRD_STATS-*.txt",
			"/opt/gtal/ital/elasticsearch/app/logstash/stage/MQA_DR_STATS-*.txt"
		]				

             - - - - - - - - - - - 
             - - - - - - - - - - - 
             - - - - - - - - - - - 
             - - - - - - - - - - -

How do I modify the output part below which should direct output of "MQA_PRD_STATS-.txt" and "MQA_DR_STATS-.txt" separately in two indices?

output {
      file {
       path => "/opt/gtal/ital/elasticsearch/logs/mqa/rubydebug.txt"
       codec => rubydebug
     }

    elasticsearch {
     hosts => [ "xxxxxxx:43045","xxxxxxxx:43045","xxxxxxxx:43045","xxxxxxxx:43045" ]
     user => "elastic"
     password => "xxxxxxxx"
     index => "demo-csv-%{+YYYY.MM.dd}"
     doc_as_upsert => true
     action => "update"
     document_id => "%{my_fingerprint}"
  }
}

Please guide.

Thanks

2

Hi,

I think, the easiest way to achive what you want is to create a new field who is gonna take the name of the file you are currently reading and put that file name in the index.

input {
  ...
}
filter {
  grok {
    match => {
        # Take the value between a slash and the extention 
        # and put this value in the field filename
        "path" => "^%{GREEDYDATA}/{DATA:filename}[.]{WORD}$"
        # In grok, greedydata take all the values until the last value that follow it
        # So here, it take all the values until the last /
    }
  }
}
output {
      file {
       path => "/opt/gtal/ital/elasticsearch/logs/mqa/rubydebug.txt"
       codec => rubydebug
     }

    elasticsearch {
     hosts => [ "xxxxxxx:43045","xxxxxxxx:43045","xxxxxxxx:43045","xxxxxxxx:43045" ]
     user => "elastic"
     password => "xxxxxxxx"
     #Adding the filename to the index
     index => "demo-csv-%{[filename]}-%{+YYYY.MM.dd}"
     doc_as_upsert => true
     action => "update"
     document_id => "%{my_fingerprint}"
  }
}

Cad.

3

Thanks. I did some research too which assigns a type to the file. Based on the file type requests go to the relevant index. Would the following work?

input {
  
  
    file {
            type => "PRD"
			path => "/opt/gtal/ital/elasticsearch/app/logstash/stage/MQA_PRD_STATS-*.txt"
			
    }
    file {
            type => "DR"
			path => "/opt/gtal/ital/elasticsearch/app/logstash/stage/MQA_DR_STATS-*.txt"
			
    } 
}

      -- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - 

if [type] == "PRD" {
			elasticsearch {
			 hosts => [ "xx-xxx-xxxx:43045","xx-xxx-xxxx:43045","xx-xxx-xxxx:43045","xx-xxx-xxxx:43045","xx-xxx-xxxx:43045" ]
			 user => "xxxxxx"
			 password => "xxxxxxxxxxxxxxxxxxxxxxx"
			 index => "prd-csv-%{+YYYY.MM.dd}"
			 doc_as_upsert => true
			 action => "update"
			 document_id => "%{my_fingerprint}"
  }
    }
    if [type] == "DR" {
			elasticsearch {
			 hosts => ["xx-xxx-xxxx:43045","xx-xxx-xxxx:43045","xx-xxx-xxxx:43045","xx-xxx-xxxx:43045","xx-xxx-xxxx:43045" ]
			 user => "xxxx"
			 password => "xxxxxxxxxxxxxxxxxxxxxxx"
			 index => "dr-csv-%{+YYYY.MM.dd}"
			 doc_as_upsert => true
			 action => "update"
			 document_id => "%{my_fingerprint}"
		}
	}
}

The resultant would be two indices prd-csv-%{+YYYY.MM.dd} and dr-csv-%{+YYYY.MM.dd}

Please guide.

Thanks

4

if you want to do it before the output something like this would actually work

mutate{
     add_field => {"[@metadata][target_index] => "index-%{[filename]}-%{+YYYY.MM.dd}"}}
}
5

Thanks.

Do I need to replace "filename" above with the name of the file e.g. MQA_PRD_STATS-*.txt and the other one in my case?

Also how would the output part look like?