## ------------------------------------------
## Retrieve messages from Voltha alarm topic
## ------------------------------------------

input {
  	kafka {
		bootstrap_servers => "kafka:9092"
		topics => ["voltha.alarms"]
		tags => [ "alarm" ]
		type => "voltha_alarms"
  	}
}

## ------------------------------------------
## The message field is a JSON block.
## Break it down in multiple fields
## for easy extraction
## ------------------------------------------

filter {
    json{
        source => "message"
    }
}

## ------------------------------------------
## Intercept alarm messages and manipulate 
## the content prior to injection in OpenNMS 
## ------------------------------------------

filter {
    if "alarm" in [tags] {
        mutate {
            add_field => { "[@metadata][alarm_id]" => "%{id}" }
            add_field => { "[@metadata][resource_id]" => "%{resource_id}" }
            add_field => { "[@metadata][description]" => "%{description}" }
            add_field => { "[@metadata][type]" => "%{type}" }
            add_field => { "[@metadata][category]" => "%{category}" }
            add_field => { "[@metadata][state]" => "%{state}" }
            add_field => { "[@metadata][raised_time]" => "%{raised_ts}" }
            add_field => { "[@metadata][reported_time]" => "%{reported_ts}" }
            add_field => { "[@metadata][changed_time]" => "%{changed_ts}" }
            add_field => { "[@metadata][context]" => "%{context}" }
            add_field => { "[@metadata][service]" => "voltha" }
            add_field => { "[@metadata][interface]" => "localhost" }
            add_tag   => "opennms"
        }

        if [state] == "RAISED" {
            mutate {
                add_field => { "[@metadata][uei]" => "uei.opennms.org/vendor/voltha/alarmRaised" }
                add_field => { "[@metadata][severity]" => "%{severity}" }
            }
        } else if [state] == "CLEARED" {
            mutate {
                add_field => { "[@metadata][uei]" => "uei.opennms.org/vendor/voltha/alarmCleared" }
                add_field => { "[@metadata][severity]" => "Normal" }
            }
        }
    }
}

## ------------------------------------------
## View incoming messages (for debugging)
## ------------------------------------------

output {
    stdout {
        codec  => rubydebug { metadata => true }
    }
}

## ------------------------------------------
## Forward alarm messages to OpenNMS
## ------------------------------------------

output {
    if "opennms" in [tags] {
        opennms {
            opennms_server_hosts      => {
                "10.100.198.220" => 5817
            }

            opennms_event_timezone    => "utc"
            opennms_event_service     => "[@metadata][service]"
            opennms_event_nodeid      => 0
            opennms_event_interface   => "[@metadata][interface]"
            opennms_event_uei         => "[@metadata][uei]"
            opennms_event_severity    => "[@metadata][severity]"
            opennms_event_description => "[@metadata][description]"
            opennms_event_logmsg      => "logmsg"

            opennms_event_parms => {
                "AlarmId"       => "[@metadata][alarm_id]"
                "ResourceId"    => "[@metadata][resource_id]"
                "Description"   => "[@metadata][description]"
                "Type"          => "[@metadata][type]"
                "Category"      => "[@metadata][category]"
                "State"         => "[@metadata][state]"
                "Severity"      => "[@metadata][severity]"
                "RaisedTime"    => "[@metadata][raised_time]"
                "ReportedTime"  => "[@metadata][reported_time]"
                "ChangedTime"   => "[@metadata][changed_time]"
                "Context"       => "[@metadata][context]"
            }

            opennms_event_parms_order => [ "AlarmId", "ResourceId", "Description", "Type", "Category", "State", "Severity", "RaisedTime", "ReportedTime", "ChangedTime", "Context" ]
        }
    }
}