kubernetes/clamp/charts/clamp-dash-logstash/resources/config/pipeline.conf

   1 # Copyright © 2018  AT&T, Amdocs, Bell Canada Intellectual Property.  All rights reserved.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #       http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 input {
  15   http_poller {
  16         urls => {
  17             event_queue => {
  18                 method => get
  19                 url => "${dmaap_base_url}/events/${event_topic}/${dmaap_consumer_group}/${dmaap_consumer_id}?timeout=15000"
  20                 headers => {
  21                     Accept => "application/json"
  22                 }
  23                 add_field => { "topic" => "${event_topic}" }
  24             }
  25             notification_queue => {
  26                 method => get
  27                 url => "${dmaap_base_url}/events/${notification_topic}/${dmaap_consumer_group}/${dmaap_consumer_id}?timeout=15000"
  28                 headers => {
  29                     Accept => "application/json"
  30                 }
  31                 add_field => { "topic" => "${notification_topic}" }
  32             }
  33             request_queue => {
  34                 method => get
  35                 url => "${dmaap_base_url}/events/${request_topic}/${dmaap_consumer_group}/${dmaap_consumer_id}?timeout=15000"
  36                 headers => {
  37                     Accept => "application/json"
  38                 }
  39                 add_field => { "topic" => "${request_topic}" }
  40             }
  41         }
  42         socket_timeout => 30
  43         request_timeout => 30
  44         interval => 60
  45         codec => "plain"
  46   }
  47 }
  48
  49 filter {
  50     # avoid noise if no entry in the list
  51     if [message] == "[]" {
  52        drop { }
  53     }
  54
  55     # parse json, split  the list into multiple events, and parse each event
  56     json {
  57          source => "[message]"
  58          target => "message"
  59     }
  60     split {
  61           field => "message"
  62     }
  63     json {
  64          source => "message"
  65     }
  66     mutate { remove_field => [ "message" ] }
  67     # express timestamps in milliseconds instead of microseconds
  68     ruby {
  69         code => "event.set('closedLoopAlarmStart', Integer(event.get('closedLoopAlarmStart')))"
  70     }
  71     date {
  72         match => [ "closedLoopAlarmStart", UNIX_MS ]
  73         target => "closedLoopAlarmStart"
  74     }
  75
  76     if [closedLoopAlarmEnd] {
  77         ruby {
  78             code => "event.set('closedLoopAlarmEnd', Integer(event.get('closedLoopAlarmEnd')))"
  79         }
  80         date {
  81             match => [ "closedLoopAlarmEnd", UNIX_MS ]
  82             target => "closedLoopAlarmEnd"
  83         }
  84
  85     }
  86     #"yyyy-MM-dd HH:mm:ss"
  87     if [notificationTime] {
  88        mutate {
  89               gsub => [
  90                    "notificationTime", " ", "T"
  91               ]
  92        }
  93        date {
  94             match => [ "notificationTime", ISO8601 ]
  95             target => "notificationTime"
  96        }
  97     }
  98 }
  99 output {
 100     stdout {
 101         codec => rubydebug
 102     }
 103
 104     if [http_request_failure] {
 105         elasticsearch {
 106             codec => "json"
 107             hosts => ["${elasticsearch_base_url}"]
 108             index => "errors-%{+YYYY.MM.DD}"
 109             doc_as_upsert => true
 110         }
 111     } else {
 112         elasticsearch {
 113             codec => "json"
 114             hosts => ["${elasticsearch_base_url}"]
 115             index => "logstash-%{+YYYY.MM.DD}" # creates daily indexes
 116             doc_as_upsert => true
 117
 118         }
 119     }
 120
 121 }