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

   1 {{/*
   2 # Copyright (c) 2018 AT&T Intellectual Property.  All rights reserved.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #       http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 */}}
  16 input {
  17     http_poller {
  18         urls => {
  19             event_queue => {
  20                 method => get
  21                 url => "${dmaap_base_url}/events/${event_topic}/${dmaap_consumer_group}/${dmaap_consumer_id}?timeout=15000"
  22                 headers => {
  23                     Accept => "application/json"
  24                 }
  25                 topic => "${event_topic}"
  26                 tags => [ "dmaap_source" ]
  27             }
  28             notification_queue => {
  29                 method => get
  30                 url => "${dmaap_base_url}/events/${notification_topic}/${dmaap_consumer_group}/${dmaap_consumer_id}?timeout=15000"
  31                 headers => {
  32                     Accept => "application/json"
  33                 }
  34                 topic => "${notification_topic}"
  35                 tags => [ "dmaap_source" ]
  36             }
  37             request_queue => {
  38                 method => get
  39                 url => "${dmaap_base_url}/events/${request_topic}/${dmaap_consumer_group}/${dmaap_consumer_id}?timeout=15000"
  40                 headers => {
  41                     Accept => "application/json"
  42                 }
  43                 topic => "${request_topic}"
  44                 tags => [ "dmaap_source" ]
  45             }
  46         }
  47         socket_timeout => 30
  48         request_timeout => 30
  49         schedule => { "every" => "1m" }
  50         codec => "plain"
  51 {{- if .Values.global.aafEnabled }}
  52         cacert => "{{ .Values.certInitializer.credsPath }}/{{ .Values.certInitializer.clamp_ca_certs_pem }}"
  53 {{- else }}
  54         cacert => "/certs.d/aafca.pem"
  55 {{- end }}
  56     }
  57 }
  58
  59
  60 filter {
  61     # avoid noise if no entry in the list
  62     if [message] == "[]" {
  63         drop { }
  64     }
  65
  66     if [http_request_failure] or [@metadata][code] != 200 {
  67        mutate {
  68               add_tag => [ "error" ]
  69        }
  70     }
  71
  72     if "dmaap_source" in [@metadata][request][tags] {
  73         #
  74         # Dmaap provides a json list, whose items are Strings containing the event
  75         # provided to Dmaap, which itself is an escaped json.
  76         #
  77         # We first need to parse the json as we have to use the plaintext as it cannot
  78         # work with list of events, then split that list into multiple string events,
  79         # that we then transform into json.
  80         #
  81         json {
  82             source => "[message]"
  83             target => "message"
  84         }
  85
  86         split {
  87             field => "message"
  88         }
  89         json {
  90             source => "message"
  91         }
  92         mutate {
  93             remove_field => [ "message" ]
  94         }
  95     }
  96
  97     #
  98     # Some timestamps are expressed as milliseconds, some are in microseconds
  99     #
 100     if [closedLoopAlarmStart] {
 101         ruby {
 102             code => "
 103             if event.get('closedLoopAlarmStart').to_s.to_i(10) > 9999999999999
 104               event.set('closedLoopAlarmStart', event.get('closedLoopAlarmStart').to_s.to_i(10) / 1000)
 105             else
 106               event.set('closedLoopAlarmStart', event.get('closedLoopAlarmStart').to_s.to_i(10))
 107             end
 108             "
 109         }
 110         date {
 111             match => [ "closedLoopAlarmStart", UNIX_MS ]
 112             target => "closedLoopAlarmStart"
 113         }
 114     }
 115
 116     if [closedLoopAlarmEnd] {
 117         ruby {
 118             code => "
 119             if event.get('closedLoopAlarmEnd').to_s.to_i(10) > 9999999999999
 120               event.set('closedLoopAlarmEnd', event.get('closedLoopAlarmEnd').to_s.to_i(10) / 1000)
 121             else
 122               event.set('closedLoopAlarmEnd', event.get('closedLoopAlarmEnd').to_s.to_i(10))
 123             end
 124             "
 125         }
 126         date {
 127             match => [ "closedLoopAlarmEnd", UNIX_MS ]
 128             target => "closedLoopAlarmEnd"
 129         }
 130
 131     }
 132
 133
 134     #
 135     # Notification time are expressed under the form "yyyy-MM-dd HH:mm:ss", which
 136     # is close to ISO8601, but lacks of T as spacer: "yyyy-MM-ddTHH:mm:ss"
 137     #
 138     if [notificationTime] {
 139         mutate {
 140             gsub => [
 141                 "notificationTime", " ", "T"
 142                 ]
 143         }
 144         date {
 145             match => [ "notificationTime", ISO8601 ]
 146             target => "notificationTime"
 147         }
 148     }
 149
 150
 151     #
 152     # Renaming some fields for readability
 153     #
 154         if [AAI][generic-vnf.vnf-name] {
 155             mutate {
 156                 add_field => { "vnfName" => "%{[AAI][generic-vnf.vnf-name]}" }
 157             }
 158         }
 159         if [AAI][generic-vnf.vnf-type] {
 160             mutate {
 161                 add_field => { "vnfType" => "%{[AAI][generic-vnf.vnf-type]}" }
 162             }
 163         }
 164         if [AAI][vserver.vserver-name] {
 165             mutate {
 166                 add_field => { "vmName" => "%{[AAI][vserver.vserver-name]}" }
 167             }
 168         }
 169         if [AAI][complex.city] {
 170             mutate {
 171                 add_field => { "locationCity" => "%{[AAI][complex.city]}" }
 172             }
 173         }
 174         if [AAI][complex.state] {
 175             mutate {
 176                 add_field => { "locationState" => "%{[AAI][complex.state]}" }
 177             }
 178         }
 179
 180
 181     #
 182     # Adding some flags to ease aggregation
 183     #
 184     if [closedLoopEventStatus] =~ /(?i)ABATED/ {
 185         mutate {
 186             add_field => { "flagAbated" => "1" }
 187         }
 188     }
 189     if [notification] =~ /^.*?(?:\b|_)FINAL(?:\b|_).*?(?:\b|_)FAILURE(?:\b|_).*?$/ {
 190         mutate {
 191             add_field => { "flagFinalFailure" => "1" }
 192         }
 193     }
 194
 195
 196     if "error" not in [@metadata][request][tags]{
 197         #
 198         # Creating data for a secondary index
 199         #
 200         clone {
 201             clones => [ "event-cl-aggs" ]
 202             add_tag => [ "event-cl-aggs" ]
 203         }
 204
 205         if  "event-cl-aggs" in [@metadata][request][tags]{
 206             #
 207             # we only need a few fields for aggregations; remove all fields from clone except :
 208             #   vmName,vnfName,vnfType,requestID,closedLoopAlarmStart, closedLoopControlName,closedLoopAlarmEnd,abated,nbrDmaapevents,finalFailure
 209             #
 210             prune {
 211                 whitelist_names => ["^@.*$","^topic$","^type$","^tags$","^flagFinalFailure$","^flagAbated$","^locationState$","^locationCity$","^vmName$","^vnfName$","^vnfType$","^requestID$","^closedLoopAlarmStart$","^closedLoopControlName$","^closedLoopAlarmEnd$","^target$","^target_type$","^triggerSourceName$","^policyScope$","^policyName$","^policyVersion$"]
 212             }
 213
 214         }
 215     }
 216 }
 217
 218
 219 output {
 220     stdout {
 221         codec => rubydebug
 222     }
 223
 224     if "error" in [tags] {
 225         elasticsearch {
 226             ilm_enabled => false
 227             codec => "json"
 228 {{- if .Values.global.aafEnabled }}
 229             cacert => "{{ .Values.certInitializer.credsPath }}/{{ .Values.certInitializer.clamp_ca_certs_pem }}"
 230 {{- else }}
 231             cacert => "/clamp-cert/ca-certs.pem"
 232 {{- end }}
 233             ssl_certificate_verification => false
 234             hosts => ["${elasticsearch_base_url}"]
 235             user => ["${logstash_user}"]
 236             password => ["${logstash_pwd}"]
 237             index => "errors-%{+YYYY.MM.DD}"
 238             doc_as_upsert => true
 239         }
 240
 241     } else if "event-cl-aggs" in [tags] {
 242         elasticsearch {
 243             ilm_enabled => false
 244             codec => "json"
 245             hosts => ["${elasticsearch_base_url}"]
 246 {{- if .Values.global.aafEnabled }}
 247             cacert => "{{ .Values.certInitializer.credsPath }}/{{ .Values.certInitializer.clamp_ca_certs_pem }}"
 248 {{- else }}
 249             cacert => "/clamp-cert/ca-certs.pem"
 250 {{- end }}
 251             ssl_certificate_verification => false
 252             user => ["${logstash_user}"]
 253             password => ["${logstash_pwd}"]
 254             document_id => "%{requestID}"
 255             index => "events-cl-%{+YYYY.MM.DD}" # creates daily indexes for control loop
 256             doc_as_upsert => true
 257             action => "update"
 258         }
 259
 260     } else {
 261         elasticsearch {
 262             ilm_enabled => false
 263             codec => "json"
 264             hosts => ["${elasticsearch_base_url}"]
 265 {{- if .Values.global.aafEnabled }}
 266             cacert => "{{ .Values.certInitializer.credsPath }}/{{ .Values.certInitializer.clamp_ca_certs_pem }}"
 267 {{- else }}
 268             cacert => "/clamp-cert/ca-certs.pem"
 269 {{- end }}
 270             ssl_certificate_verification => false
 271             user => ["${logstash_user}"]
 272             password => ["${logstash_pwd}"]
 273             index => "events-%{+YYYY.MM.DD}" # creates daily indexes
 274             doc_as_upsert => true
 275         }
 276     }
 277 }