关于PostgreSQL错误日志与慢查询日志收集

$PreserveFQDN on    #用于正确的获取主机名

$FileOwner root   #存储的文件属主

$FileGroup root   #文件属主

$FileCreateMode 0644 #生成的文件权限

$DirCreateMode 0755 #生成的目录权限

$Umask 0022

$PrivDropToUser root  #可以删除日志的用户

$PrivDropToGroup root  #可以删除日志的用户组

module( load = "imuxsock" )

module( load = "imklog" )

module( load = "imudp" )

#input(type= "imudp" port= "514" )

module( load = "imtcp" MaxSessions= "500" )

input(type= "imtcp" port= "514" )​

$template linefmt, "%msg:2:$%\n"   #接收日志的格式(去掉开头的空格）​

$template pgloglocation, "/data/pglogs/%hostname%/%$YEAR%-%$MONTH%-%$DAY%.csv" ​

:rawmsg, contains , "pg_5432" ?pgloglocation;linefmt

##变量：%fromhost-ip%

cat /etc/rsyslog.d/10-pg.conf​

module( load = "imuxsock" )

module( load = "imklog" )

module( load = "imfile" )​

#module( load = "imudp" )

#input(type= "imudp" port= "514" )​

module( load = "imtcp" MaxSessions= "500" )

input(type= "imtcp" port= "514" )​

ruleset( name = "remote" ){

     action (type= "omfwd"

             target= "x.x.x.x"   #日志服务器IP地址

             port= "514"      #端口

             protocol= "tcp"    #使用协议

             queue.type= "linkedList" #使用异步处理

             queue.spoolDirectory= "/var/log/rsyslog" #队列目录

             queue.fileName= "pglog"          #队列名称

             queue.maxDiskSpace= "1g"          #队列占最大磁盘空间

             queue.saveOnShutdown= "on"   #保存内存数据如果rsyslog关闭

             action .resumeRetryCount= "-1" #无限重试插入失败

        )

     stop

}​

​

input(

  type= "imfile"

  File= "/pg/data/log/*.csv" #PG服务器日志路径

  Facility= "local1"

  Severity= "info"

  Tag= "pg_5432"        #定义日志标签，重要，服务端根据这个标签可以识别日志

  PersistStateInterval= "1"   #回写偏移量数据到文件间隔时间(秒)，根据实际情况而定

  deleteStateOnFileDelete= "on"

  reopenOnTruncate= "on"

  Ruleset= "remote"       #rsyslog.conf中定义的 rule 名称

)

input {

  file {

   path => [ "/data/pglogs/*/*.csv" ]

   start_position => "end"

   codec => multiline {

    pattern => "^20[0-9]{2}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}"

    negate => true

    what => "previous"

   }

  }

}

​

filter {

  csv {

   separator => ","

   columns => [ "log_time" , "user_name" , "database_name" , "process_id" , "connection_from" , "session_id" , "session_line_num" , "command_tag" , "session_start_time" , "virtual_transaction_id" , "transaction_id" , "error_severity" , "sql_state_code" , "message" , "detail" , "hint" , "internal_query" , "internal_query_pos" , "context" , "query" , "query_pos" , "location" , "application_name" ]

   convert => {

    "process_id" => "integer"

    "session_line_num" => "integer"

    "transaction_id" => "integer"

    "internal_query_pos" => "integer"

    "query_pos" => "integer"

   }

   skip_empty_columns => true

  }

​

  mutate{

    split => { "log_time" => " CST" }

    add_field => { "log_time_tmp" => "%{[log_time][0]}" }

    remove_field => [ "log_time" ]

  }

​

  date {

    match => [ "log_time_tmp" , "yyyy-MM-dd HH:mm:ss.SSS" ]

    target => "@timestamp"

    locale => "cn"

    remove_field => [ "log_time_tmp" ]

  }

​

  if "duration:" in [message] and "ms" in [message] and "statement:" in [message] {

   grok{

    match => { "message" => "duration: %{NUMBER:duration} ms" }

   }

   mutate{

    split => { "message" => "statement: " }

    add_field => { "statement" => "%{[message][1]}" }

    remove_field => [ "message" ]

   }

  }

  mutate{

   split => { "path" => "/" }

   add_field => { "db_host" => "%{[path][3]}" }

   remove_field => [ "path" , "host" ]

   convert => { "duration" => "float" }

  }

}​

​

output {

  if [error_severity] == "ERROR" or [error_severity] == "FATAL" or [error_severity] == "PANIC" {

   elasticsearch {

    hosts => [ "x.x.x.x:x" , "x.x.x.x:x" ]

    index => "pg_error"

    id => "elasticsearch_pg_error"

   }

  } else if [duration] and [statement] {

   elasticsearch {

    hosts => [ "x.x.x.x:x" , "x.x.x.x:x" ]

    index => "pg_slow"

    id => "elasticsearch_pg_slow"

   }

  }

}

#client_min_messages = notice

log_min_messages = debug5 #debug级别是提供给开发人员使用的，这个可以看到程序调用的信息以及SQL转化为数据结构的信息，每分钟的级别

#log_destination = 'stderr'

logging_collector = on   #打开日志收集

log_directory = 'pg_log' #日志目录

log_filename = 'postgresql-%Y-%m-%d_%H%M%S.log'

debug_print_parse = on #解析树

debug_print_rewritten = on #查询重写后的SQL

debug_print_plan = on   #执行计划详细

debug_pretty_print = on #对debug_print_parse,debug_print_rewritten,debug_print_plan可读性格式化

#log_checkpoints = off #如果是研究pg的磁盘IO，这个需要设置为 on

log_connections = on #连接日志

log_disconnection = on #断开连接日志

#log_duration= on #语句执行时间，对于分析