BI 11g: сбор, парсинг и анализ логов с помощью Filebeat + Logstash + ElasticSearch + Kibana

Привет всем, кто еще читает этот блог!

Сегодня хочу рассказать о настройках, необходимых для сбора и парсинга логов системных компонент Oracle BI с помощью стека ELK (ElasticSearch - Logstash - Kibana).
Я не буду подробно останавливаться на шагах развертывания и базовой настройки ELK - информации об этом огромное множество в интернете.
Здесь же опишу лишь настройки сборщика логов Filebeat, который устанавливается на стороне сервера Oracle BI.
А также настройки парсинга логов с помощью Logstash.

В итоге у вас может получиться нечто подобное:





Представленный дашборд Kibana содержит упрощенную аналитику состояния Oracle BI на основании данных о кол-ве ошибок и предупреждений в логах компонент системы.

Для анализа доступны следующие атрибуты лог-сообщений:


Для того, чтобы можно было анализировать лог-сообщения OracleBI в разрезе приведенных атрибутов (дата-время; тип компонента; статус лог-сообщения; текст и детализация лог-сообщения и т.д.) необходимо:
- доставлять на регулярной основе все новые лог-сообщения в индекс ElasticSearch;
- при добавлении новых лог-сообщений в ES следует парсить их по определенным правилам (согласно структуре логов).

Для доставки лог-сообщений применяется Filebeat, со следующим конфигурационным файлом (/etc/filebeat/filebeat.yml):

############################# Filebeat ######################################
filebeat:
  # List of prospectors to fetch data.
  prospectors:

    -
      paths:
        - /var/log/*.log
      document_type: syslog
      scan_frequency: 10s

    -
      paths:
        - /var/log/oracle/obiee/analytics/OracleBIServerComponent/coreapplication_obis1/nqserver*.log
      document_type: bi_server
      scan_frequency: 30s

    -
      paths:
        - /var/log/oracle/obiee/analytics/OracleBIPresentationServicesComponent/coreapplication_obips1/sawlog*.log
        - /var/log/oracle/obiee/analytics/OracleBIPresentationServicesComponent/coreapplication_obips2/sawlog*.log
      document_type: bi_presentation_server
      scan_frequency: 30s

    -
      paths:
        - /var/log/oracle/obiee/analytics/OracleBIJavaHostComponent/coreapplication_obijh1/jh*.log
        - /var/log/oracle/obiee/analytics/OracleBIJavaHostComponent/coreapplication_obijh2/jh*.log
      document_type: bi_javahost
      scan_frequency: 30s

    -
      paths:
        - /var/log/oracle/obiee/analytics/OracleBISchedulerComponent/coreapplication_obisch1/nqscheduler*.log
      document_type: bi_scheduler
      scan_frequency: 30s

    -
      paths:
        - /var/log/oracle/obiee/analytics/OracleBISchedulerComponent/coreapplication_obisch1/Agent*.Log
      document_type: bi_ibots
      scan_frequency: 30s

    -
      paths:
        - /var/log/oracle/obiee/analytics/OracleBIClusterControllerComponent/coreapplication_obiccs1/nqcluster*.log
      document_type: bi_cluster_controller
      scan_frequency: 30s


############################# Output ##########################################

# Configure what outputs to use when sending the data collected by the beat.
# Multiple outputs may be used.
output:

  ### Logstash as output
  logstash:
    # The Logstash hosts
    hosts: ["logstash_server_ip:5044"]

    # Optional TLS. By default is off.
    tls:
      # List of root certificates for HTTPS server verifications
      certificate_authorities: ["/etc/pki/tls/certs/logstash-forwarder.crt"]

############################# Logging #########################################

# There are three options for the log ouput: syslog, file, stderr.
# Under Windos systems, the log files are per default sent to the file output,
# under all other system per default to syslog.
logging:

  # Write all logging output to files. Beats automatically rotate files if rotateeverybytes
  # limit is reached.
  to_files: true

  # To enable logging to files, to_files option has to be set to true
  files:
    # The directory where the log files will written to.
    path: /var/log/mybeat

    # The name of the files where the logs are written to.
    name: filebeat

    # Configure log file size limit. If limit is reached, log file will be
    # automatically rotated
    rotateeverybytes: 10485760 # = 10MB

    # Number of rotated log files to keep. Oldest files will be deleted first.
    keepfiles: 7

  # Sets log level. The default log level is error.
  # Available log levels are: critical, error, warning, info, debug
  level: error

Из важного:
- последовательно перечисляем все лог-файлы (используя маски), содержимое которых хотим пересылать на Logstash-сервер с заданной периодичностью;
- для каждого типа лог-файлов указываем атрибут "document_type";
- указываем, что приемником лог-сообщений будет не ElasticSearch-кластер, а Logstash.

Для приема и парсинга лог-сообщений на стороне сервера Logstash создаем конфигурационные-файлы:
1) для описания правил приема лог-сообщений извне - /etc/logstash/conf.d/02-beats-input.conf

input {
  beats {
    port => 5044
    ssl => true
    ssl_certificate => "/etc/pki/tls/certs/logstash-forwarder.crt"
    ssl_key => "/etc/pki/tls/private/logstash-forwarder.key"
  }
}

2) для описания правил загрузки обработанных лог-сообщений в ElasticSearch - /etc/logstash/conf.d/30-elasticsearch-output.conf

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    sniffing => true
    manage_template => false
    index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
    document_type => "%{[@metadata][type]}"
  }
}

(пересылка идет на localhost, так как ES и Logstash располагаются на одном сервере)

3) и наконец самое важное - правила парсинга лог-сообщений - /etc/logstash/conf.d/10-bi-filter.conf

filter {
  if [type] == "bi_server" {
    multiline {
      pattern => "^\[%{TIMESTAMP_ISO8601}\]"
      negate => true
      what => previous
    }
    grok {
      match => [ "message", "(?m)\[%{TIMESTAMP_ISO8601:odl-timestamp}\]\s+\[%{DATA:odl-component}\]\s+\[%{DATA:odl-msgType}:%{INT:odl-msgTypeLevel}\]\s+\[\]\s+\[%{DATA:odl-callProc}\]\s+\[ecid:\s+%{DATA:odl-ecid}\]\s+\[tid:\s+%{DATA:odl-tid}\]\s+(?:(%{DATA:odl-msg}\.\s*$)|(%{DATA:odl-msg}\.\s*\[\[%{GREEDYDATA:odl-msgDetails}\]\]$))" ]
      add_field => [ "receivedAt", "%{@timestamp}" ]
    }
  }

  if [type] == "bi_presentation_server" {
    multiline {
      pattern => "^\[%{TIMESTAMP_ISO8601}\]"
      negate => true
      what => previous
    }
    grok {
      match => [ "message", "(?m)\[%{TIMESTAMP_ISO8601:odl-timestamp}\]\s+\[%{DATA:odl-component}\]\s+\[%{DATA:odl-msgType}:%{INT:odl-msgTypeLevel}\]\s+\[\]\s+\[%{DATA:odl-callProc}\]\s+\[ecid:\s+%{DATA:odl-ecid}\]\s+\[tid:\s+%{DATA:odl-tid}\]\s+%{DATA:odl-msg}\[\[%{GREEDYDATA:odl-msgDetails}\]\]" ]
      add_field => [ "receivedAt", "%{@timestamp}" ]
    }
    kv {
      source => "msgDetails"
      field_split => "\n"
      value_split => ": "
      exclude_keys => [ "ecid" ]
      prefix => "odl-msgDtl"
    }
  }

  if [type] == "bi_javahost" {
    multiline {
      pattern => "^\[%{TIMESTAMP_ISO8601}\]"
      negate => true
      what => previous
    }
    grok {
      match => [ "message", "(?m)\[%{TIMESTAMP_ISO8601:odl-timestamp}\]\s+\[%{DATA:odl-component}\]\s+\[%{DATA:odl-msgType}\]\s+\[\]\s+\[%{DATA:odl-callProc}\]\s+\[tid:\s+%{DATA:odl-tid}\]\s+\[ecid:\s+%{DATA:odl-ecid}\]\s+%{DATA:odl-msg}\[\[%{GREEDYDATA:odl-msgDetails}\]\]" ]
      add_field => [ "receivedAt", "%{@timestamp}" ]
    }   
  }

  if [type] == "bi_scheduler" {
    multiline {
      pattern => "^\[%{TIMESTAMP_ISO8601}\]"
      negate => true
      what => previous
    }
    grok {
      match => [ "message", "(?m)\[%{TIMESTAMP_ISO8601:odl-timestamp}\]\s+\[%{DATA:odl-component}\]\s+\[%{DATA:odl-msgType}:%{INT:odl-msgTypeLevel}\]\s+\[\]\s+\[%{DATA:odl-callProc}\]\s+\[ecid:\s+%{DATA:odl-ecid}\]\s+\[tid:\s+%{DATA:odl-tid}\]\s+%{DATA:odl-msg}\[\[%{GREEDYDATA:odl-msgDetails}\]\]" ]
      add_field => [ "receivedAt", "%{@timestamp}" ]
    }
  }

  if [type] == "bi_ibots" {
    multiline {
      pattern => "^\[%{TIMESTAMP_ISO8601}\]"
      negate => true
      what => previous
    }
    grok {
      match => [ "message", "(?m)\[%{TIMESTAMP_ISO8601:odl-timestamp}\]\s+\[%{DATA:odl-component}\]\s+\[%{DATA:odl-msgType}:%{INT:odl-msgTypeLevel}\]\s+\[\]\s+\[%{DATA:odl-callProc}\]\s+\[ecid:\s+%{DATA:odl-ecid}\]\s+\[tid:\s+%{DATA:odl-tid}\]\s+%{DATA:odl-msg}\[\[%{GREEDYDATA:odl-msgDetails}\]\]" ]
      add_field => [ "receivedAt", "%{@timestamp}" ]
    }
  }

  if [type] == "bi_cluster_controller" {
    multiline {
      pattern => "^\[%{TIMESTAMP_ISO8601}\]"
      negate => true
      what => previous
    }
    grok {
      match => [ "message", "(?m)\[%{TIMESTAMP_ISO8601:odl-timestamp}\]\s+\[%{DATA:odl-component}\]\s+\[%{DATA:odl-msgType}:%{INT:odl-msgTypeLevel}\]\s+\[\]\s+\[%{DATA:odl-callProc}\]\s+\[ecid:\s+%{DATA:odl-ecid}\]\s+\[tid:\s+%{DATA:odl-tid}\]\s+%{DATA:odl-msg}\[\[%{GREEDYDATA:odl-msgDetails}\]\]" ]
      add_field => [ "receivedAt", "%{@timestamp}" ]
    }
  }
}

P.S. Используется Logstash версии 2.2