$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")"%msg:2:$%\n"  #接收日志的格式(去掉开头的空格）"/data/pglogs/%hostname%/%$YEAR%-%$MONTH%-%$DAY%.csv""pg_5432" "htmlcode">

cat /etc/rsyslog.d/10-pg.conf"imuxsock")
module(load="imklog")
module(load="imfile")"imudp")
#input(type="imudp" port="514")"imtcp" MaxSessions="500")
input(type="imtcp" port="514")"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
}"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名称
)


四、logstash配置
在日志服务器上编辑好配置文件后，启动logstash即可。配置文件如下：


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"
  }
 }
}
","
  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
 }
"log_time" => " CST" }
   add_field => { "log_time_tmp" => "%{[log_time][0]}" }
   remove_field => ["log_time"]
 }
"log_time_tmp", "yyyy-MM-dd HH:mm:ss.SSS"]
   target => "@timestamp"
   locale => "cn"
   remove_field => ["log_time_tmp"]
 }
"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" }
 }
}"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"
  }
 }
}


五、此处省略了Elasticsearch存储与Kibana展示，这里不多介绍
补充：PostgreSQL开发者模式错误反馈与日志设置
####when何时记录


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


####where记录到哪里


#log_destination = 'stderr'
logging_collector = on  #打开日志收集
log_directory = 'pg_log' #日志目录
log_filename = 'postgresql-%Y-%m-%d_%H%M%S.log'  


####what写什么日志


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 #语句执行时间，对于分析



以上为个人经验，希望能给大家一个参考，也希望大家多多支持。如有错误或未考虑完全的地方，望不吝赐教。