在公司的一周时间中，整理了一下我们自己的日志方案

需求

分为内网和外网
分为实时日志和非实时日志
实时日志通过消息队列通道传到内网中，处理后存储到ES中
非实时日志上传到腾讯云OSS，然后同步到内网中，作为后续的日志分析，处理后存储到ES和内网中CEPH对象存储中

技术选型

系统 ubuntu18.04，其他带有rsyslog8.x的系统即可，没有也可以安装一个

利用系统服务：rsyslog 8.4.20

logstash 7.13.0

elasticsearch 7.8.0

日志格式我们使用了

HOSTNAME PROGRAME JSON字符串
复制代码

为什么使用JSON字符串，肯定方便后续的程序去处理它，虽然增加了一些空间。

工作流程

程序写rsyslog日志 -> rsyslog转发到logstash -> logstash对实时日志和非实时日志分流到消息队列和对象存储

一个简单的日志格式

data yq-data: {"type": "off-line", "time": "2021-06-23T07:03:55.122584Z", "msg": "I am a log message", "level": "DEBUG"}
复制代码

配置`rsyslog`

vim /etc/rsyslog.conf

增加下面配置

local7.* @@logstash-host:5514
复制代码

logstash-host:5514 为logstash启动syslog的IP和端口
local7为rsyslog的facility

增加一个rsyslog的配置，在/etc/rsyslog.d下面

vim /etc/rsyslog.d/20-default.conf,填入下面内容

template(name="DynaFile" type="string" string="/var/log/%programname%/%programname%.log")

local7.* action(type="omfile" dynaFile="DynaFile")
复制代码

配置完成后重启systemctl restart rsyslog

什么意思呢，就是从local7发过来的日志，都会存储到/var/log/日志格式中的PROGRAME/日志格式中的PROGRAME.log文件

为什么要这样做呢？是为了不仅仅我们将日志发到logstash，还在我们本地也存放一份，毕竟不能保证logstash完全百分百分一直运转，当他出错的时候我们可以通过filebeats从新抽取一遍，或者直接丢给我们的日志程序去处理一下

以python logging为例

# address rsyslog的IP和端口
SysLogHandler(facility=SysLogHandler.LOG_LOCAL7, address=("172.16.102.227", 514))
复制代码

再来配置`logrotate`

顾名思义，就是日志轮转，我们配置每天轮转，保存30天的日志，具体的去看下logrotate的文档就行了

vim /etc/logrotate/custom-log

/var/log/yq*/*.log
{
    rotate 30
    daily
    missingok
    notifempty
    delaycompress
    nocompress
    postrotate
        systemctl kill -s HUP rsyslog.service
    endscript
}
复制代码

这里为什么用yq*，因为项目太多了，所以我们为所有的项目定一个前缀，当然你也可以自己处理它

配置 `logstash`

vim /etc/logstash/conf.d/my.conf

input {
    syslog {
        port => 5514
    }
}

filter {
    json {
        source => "message"
    }
    prune {
        whitelist_names => [ "msg", "logsource", "program", "time", "level", "type" ]
    }
    mutate {
        remove_field => ["@timestamp", "timestamp"]
    }
}

output {
    # 对于内网直接存储到ES
    if [type] == "off-line" {
        elasticsearch {
            hosts => ["ES-HOST"]
            index => "my-log"
        }
    }
    # 对于外网直接存储到腾讯云OSS 同上面的两者取其一
    if [type] == "off-line" {
        s3 {
            endpoint => "https://cos.ap-nanjing.myqcloud.com"
            access_key_id => "xxxx"
            secret_access_key => "xxxx"
            region => "ap-nanjing"
            bucket => "xxxx"
            # 10分钟
            time_file => 10
            codec => "json_lines"
            canned_acl => "public-read"
        }
    }
    
    if [type] == "real-time" {
        rabbitmq {
            exchange => "my-exchange"
            host => "localhost"
            exchange_type => "direct"
            key => "logstash"
            user => "admin"
            password => "admin"
        }
    }
    stdout {}
}
复制代码

调试模式 /user/share/logstash/bin/logstash -f /etc/logstash/conf.d/my.conf

生产环境下去掉stdout {}

systemctl restart logstash