Just wanted to share my configuration with the community after refining it for the better part of a week. Being new to the ELK stack it took a bit of time to understand how to get tcp flows into the platform. There was also a bit of a learning curve as some previous examples of how to do this were discussed years ago, and some bugs needed to be worked out as a result. The script consumes the csv lines that come in, maps them to their related field names, creates/caculates/populates a duration field in seconds, and then deletes the original CSV message before sending it on.
Hopefully this is helpful to someone else out there looking to log their SMDR data without having to purchase a call accounting software etc.
input {
tcp {
port => "5500"
}
}
filter{
csv{
skip_empty_columns => true
columns => ["Call Start","Connected Time","Ring Time","Caller","Direction","Called Number","Dialled Number","Account","Is Internal","Call ID","Continuation","Party1Device","Party1Name","Party2Device","Party2Name","Hold Time","Park Time","AuthValid","AuthCode","User Charged","Call Charge","Currency","Amt at Last User Change","Call Units","Units at Last User Chg","Cost per Unit","Mark Up","Ext Targeting Cause","Ext Targeter Id","Ext Targeted Number","Srv IP of caller extn","Unique call id for the caller extension","Server IP address of the called extension","Unique call id for the called extension","UTC time"]
}
mutate{
add_field => { "Duration" => 0 }
}
mutate{
convert => { "Duration" => "integer" }
}
ruby {
code => " h, m, s = event.get('Connected Time').split(':').map{|str| str.to_i};
event.set('Duration', h*3600 + m*60 + s) "
}
mutate{
remove_field => [ "message" ]
}
}
output {
elasticsearch {
hosts => ["localhost:9200"]
manage_template => false
}
}