logstash抽取oracle慢sql和alert日志

create or replace view slow_sql_view as
select  to_char(sysdate-1/24,'yyyy-mm-dd hh24') as snapshot_time,--快照时间
        v_1.sql_id,
        v_1.elapsed_time,--一小时内累计耗时
        v_1.cpu_time,--一小时内累计CPU时间
        v_1.iowait_time,--一小时内累计io等待时间
        v_1.gets,--一小时内累逻辑读
        v_1.reads,--一小时内累计物理读
        v_1.rws,--一小时内累计返回行数
        v_1.clwait_time,--一小时内累计集群等待时间
        v_1.execs,--一小时内累计执行次数
        v_1.elpe,--平均每条SQL消耗时间
        nvl(v_2.machine,'null') as machine,--客户服务器名称
        nvl(v_2.username,'null') as username,--客户连接用户名
        to_char(substr(v_1.sqt,1,3000)) as sql
 
from
  (select s.sql_id,
          round(elapsed_time  1000000,2) elapsed_time,
          round(cpu_time / 1000000,2) cpu_time,
          round(iowait_time / 1000000,2) iowait_time,
          gets,
          reads,
          rws,
          round(clwait_time / 1000000,2) clwait_time,
          execs,
          st.sql_text sqt,
          round(elapsed_time / 1000000 / decode(execs, 0, null, execs),2) elpe
     from (select *
             from (select sql_id,
                          sum(executions_delta) execs,
                          sum(buffer_gets_delta) gets,
                          sum(disk_reads_delta) reads,
                          sum(rows_processed_delta) rws,
                          sum(cpu_time_delta) cpu_time,
                          sum(elapsed_time_delta) elapsed_time,
                          sum(clwait_delta) clwait_time,
                          sum(iowait_delta) iowait_time
                     from dba_hist_sqlstat
                    where snap_id =(select max(snap_id) from dba_hist_snapshot)
                    group by sql_id
                    order by sum(elapsed_time_delta) desc)
            where rownum <= 20) s,
          dba_hist_sqltext  st
    where st.sql_id = s.sql_id) v_1
    left join
  (select distinct a.sql_id, a.machine, b.username
     from dba_hist_active_sess_history a
     left join dba_users b
       on a.user_id = b.user_id
    where a.snap_id = (select max(snap_id) from dba_hist_snapshot)) v_2
      on v_1.sql_id = v_2.sql_id
      where v_1.elpe >=1  --平均执行时间大于1s的sql过滤出来
      order by elpe desc
;

vim oracle_slow.py 
 
#!/usr/bin/python
#coding=utf-8
import os
import cx_Oracle
 
os.environ['ORACLE_HOME'] = '/u01/app/oracle/product/11.2.0/db_1'
os.environ['ORACLE_SID'] = 'orcl'
os.environ['PATH']
 
#连接数据库，查视图
def slow_sql_qurey():
    conn = cx_Oracle.connect('slow_user','******','')
    cursor=conn.cursor()
    lists = []
    try:
        cursor.execute ("select * from slow_sql_view")  
        #print("连接成功！")
        lists = cursor.fetchall()  
    except Exception:
        print("connenct oracle error,dblink error!",Exception)
    finally:
        cursor.close()  
        conn.close()  
    #print(lists)
  
    msg = ''
    if len(lists):
        for i in lists:
            msg = msg +  ",".join(map(str, i)) + '\n'
        #print(msg)
        #将文件输出到文件
        f=open('/u01/app/slow_log/slow.log','w+')
        f.write(msg)
        f.close()
 
  
def main():
    slow_sql_qurey()
if __name__ == "__main__":
    main()

10 * * * * /usr/bin/python /u01/app/slow_log/oracle_slow.py > /u01/app/slow_log/exec_qurey_slow.log 2>&1

wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.3.0-linux-x86_64.tar.gz
tar xvf filebeat-6.3.0-linux-x86_64.tar.gz 
mv filebeat-6.3.0-linux-x86_64 /usr/local/filebeat
cd /usr/local/filebeat
 
vim  filebeat.yml
 
filebeat.inputs:
 
#oracle_alert.log 收集Oracle告警日志
- type: log
  enabled: true
  paths:
    - /u01/app/oracle/diag/rdbms/orcldg2/orcl/trace/alert_orcl.log 
  tags: ["oracle-log"]
  multiline.pattern:  '(Mon|Tue|Wed|Thu|Fri|Sat|Sun)\s(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\s(\d{2})\s(\d{2}):(\d{2}):(\d{2})\s(\d{4})'
  multiline.negate: true
  multiline.match: after
  fields:
      log_source: oracle-alert-log
 
#oracle_slow.log 收集慢SQL(每小时会自动触发一次)
- type: log
  enabled: true
  paths:
    - /u01/app/slow_log/slow.log
  tags: ["oracle-slow-log"]
  multiline.pattern:  '\d{4}-\d{2}-\d{2}'
  multiline.negate: true
  multiline.match: after
  fields:
      log_source: oracle-slow-log
 
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
setup.template.settings:
  index.number_of_shards: 3
setup.kibana:
 
output.logstash:
  hosts: ["10.30.1.12:55051"]

nohup /usr/local/filebeat/filebeat -e -c /usr/local/filebeat/filebeat.yml >/dev/null 2>&1 &

vim /usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-patterns-core-4.1.2/patterns/grok-patterns
 
添加
 
ORACLE_TIME (Mon|Tue|Wed|Thu|Fri|Sat|Sun)\s(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\s(\d{2})\s(\d{2}):(\d{2}):(\d{2})\s(\d{4})
DB_LOGCONTENT .*
SNAPSHOT_TIME \d{4}-\d{2}-\d{2}\s\d+
SQL_ID [A-Za-z0-9]+
ORA_MACHINE .*?
ORA_USER .*?
ORA_SQL .*

vim oracle-log.conf
 
input {
  beats {
    port => 55051
  }
}
 
filter{
  if "oracle-log" in [tags] {
    mutate { add_field => {  "clienthost" => "%{[beat][hostname]}" } }
    grok {
        match => {
            "message" =>  "%{ORACLE_TIME:datetime}\n%{DB_LOGCONTENT:logmessage}"
        }
    }
  }
 
 if "oracle-slow-log" in [tags] {
 grok {
        match => [ "message" , "%{SNAPSHOT_TIME:snapshot_time},%{SQL_ID:sql_id},%{NUMBER:elapsed_time:float},%{NUMBER:cpu_time:float},%{NUMBER:iowait_time:float},%{NUMBER:gets:int},%{NUMBER:reads:int},%{NUMBER:rows:int},%{NUMBER:cluster_wait_time:float},%{NUMBER:execs:int},%{NUMBER:elpe_time:float},%{ORA_MACHINE:machine},%{ORA_USER:username},%{ORA_SQL:sql}" ]
        
 
    }
 }
 
}
 
 
output {
  if "oracle-log" in [tags] {
    elasticsearch {
      hosts => ["dbloges.e6niu.com:30351"]
      manage_template => false
      index => "oracle-log-%{+YYYY.MM}"
   }}
  if "oracle-slow-log" in [tags] {
    elasticsearch {
      hosts => ["dbloges.e6niu.com:30351"]
      manage_template => false
      index => "oracle-slow-log-%{+YYYY.MM}"
   }
 }
}

添加索引
oracle-log-*
oracle-slow-log-*