improve logstash parsing

[oom.git] / kubernetes / clamp / charts / clamp-dash-logstash / resources / config / pipeline.conf

diff --git a/kubernetes/clamp/charts/clamp-dash-logstash/resources/config/pipeline.conf b/kubernetes/clamp/charts/clamp-dash-logstash/resources/config/pipeline.conf
index aa087e3..f88e40d 100644 (file)
--- a/kubernetes/clamp/charts/clamp-dash-logstash/resources/config/pipeline.conf
+++ b/kubernetes/clamp/charts/clamp-dash-logstash/resources/config/pipeline.conf
@@ -1,3 +1,16 @@
+# Copyright © 2018  AT&T, Amdocs, Bell Canada Intellectual Property.  All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
 input {
   http_poller {
         urls => {
@@ -8,6 +21,7 @@ input {
                     Accept => "application/json"
                 }
                 add_field => { "topic" => "${event_topic}" }
+                type => "dmaap_event"
             }
             notification_queue => {
                 method => get
@@ -16,6 +30,7 @@ input {
                     Accept => "application/json"
                 }
                 add_field => { "topic" => "${notification_topic}" }
+                type => "dmaap_notification"
             }
             request_queue => {
                 method => get
@@ -24,6 +39,7 @@ input {
                     Accept => "application/json"
                 }
                 add_field => { "topic" => "${request_topic}" }
+                type => "dmaap_request"
             }
         }
         socket_timeout => 30
@@ -34,26 +50,39 @@ input {
 }
 
 filter {
-    # avoid noise if no entry in the list
-    if [message] == "[]" {
-       drop { }
-    }
+       if [type] != "dmaap_log" {
+        # avoid noise if no entry in the list
+        if [message] == "[]" {
+           drop { }
+        }
 
-    # parse json, split  the list into multiple events, and parse each event
-    json {
-         source => "[message]"
-         target => "message"
-    }
-    split {
-          field => "message"
-    }
-    json {
-         source => "message"
+        # parse json, split  the list into multiple events, and parse each event
+        json {
+             source => "[message]"
+             target => "message"
+        }
+        split {
+              field => "message"
+                 add_field => {
+                       "type" => "%{type}"
+                       "topic" => "%{topic}"
+                 }
+        }
+        json {
+             source => "message"
+        }
+        mutate { remove_field => [ "message" ] }
     }
-    mutate { remove_field => [ "message" ] }
+    
     # express timestamps in milliseconds instead of microseconds
     ruby {
-        code => "event.set('closedLoopAlarmStart', Integer(event.get('closedLoopAlarmStart')))"
+            code => "
+                     if event.get('closedLoopAlarmStart').to_s.to_i(10) > 9999999999999
+                       event.set('closedLoopAlarmStart', event.get('closedLoopAlarmStart').to_s.to_i(10) / 1000)
+                     else
+                       event.set('closedLoopAlarmStart', event.get('closedLoopAlarmStart').to_s.to_i(10))
+                     end
+                    "
     }
     date {
         match => [ "closedLoopAlarmStart", UNIX_MS ]
@@ -62,7 +91,13 @@ filter {
 
     if [closedLoopAlarmEnd] {
         ruby {
-            code => "event.set('closedLoopAlarmEnd', Integer(event.get('closedLoopAlarmEnd')))"
+            code => "
+                    if event.get('closedLoopAlarmEnd').to_s.to_i(10) > 9999999999999  
+                      event.set('closedLoopAlarmEnd', event.get('closedLoopAlarmEnd').to_s.to_i(10) / 1000)
+                    else
+                      event.set('closedLoopAlarmEnd', event.get('closedLoopAlarmEnd').to_s.to_i(10))
+                    end
+                    "
         }
         date {
             match => [ "closedLoopAlarmEnd", UNIX_MS ]