【日志高级】搭配logstash

上一节是把一台机器内多个worker中的日志合并

这里把多台机器的日志，进行统一合并，并且索引到elasticsearch，使用logstash进行日志索引管理

注意: 千万不要把敏感的数据打印出来，这里用的插件并没有ssl加密传输的功能，如果走的是内网倒还好，最怕的是tcp连接走了外网

安装
```
npm i -S km-log4js-logstash-tcp
```
- 最新版本 1.0.4
添加两个appender到log4js-conf.json
```
"logstash_access_log": {
    "type": "km-log4js-logstash-tcp",
    "layout": { "type": "pattern", "pattern": "[%d{ISO8601}] [%p] %c - %h %z %m%n" },
    "host": "localhost",
    "port": 5050,
    "service": {
        "application": "app",
        "category": "access_log"
    }
},
"logstash_log": {
    "type": "km-log4js-logstash-tcp",
    "layout": { "type": "pattern", "pattern": "[%d{ISO8601}] [%p] %c - %h %z %m%n" },
    "host": "localhost",
    "port": 5050,
    "service": {
        "application": "app",
        "category": "log"
    }
}
```
- 可以将
  - "logstash_access_log"补充到"access_log"的"appenders"中
  - "logstash_log"补充到"log"的"appenders"中
- 这里手动配置"service"下的
  - "application"为app
  - "category"为"access_log"或者"log"
  - 这样就可以根据不同的项目，不同的日志类型，在下面logstash的条件判断中，做不同的解构，并写入不同的索引中
    - 对于app的access_log，会写入app-access_log-
    - 对于app的log，会写入app-log-
- 这里手动配置layout，采用pattern模式，
  - 自己指定发送给logstash的message内容
    - %d会自动填日期
    - %p会自动填日志类型
    - %c会自动填来自的logger
    - %h会自动填发送主机名(服务器)
    - %z会自动填发送的进程的pid
    - %m会自动填log.error(xxx)中间的xxx
    - %n会自动填\n用于输出换行
  - 这样后面logstash要用同样的模式去解构
    - 可以得到更多信息!!!
  - 虽然tcp已经附带了@timestamp，但我们还是要传日期，因为要用这个做计算替换掉@timestamp

配置logstash

input {
    tcp {
        type => "from_tcp"
        codec => "json"
        port => 5050
    }
}
filter {
    if [type] == "from_tcp" and [service][application] == "app" and [service][category] == "access_log" {
        grok {
            match => { "message" => "\[%{TIMESTAMP_ISO8601:server_time}\] \[%{LOGLEVEL:level}\] %{WORD:category} - %{USERNAME:hostname} %{NUMBER:pid} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:status} %{NUMBER:duration}ms %{USERNAME:user_uuid} %{IP:clientIp}" }
        }
        geoip {
            source => "clientIp"
        }
        mutate{
            add_field => { "log_time" => "%{@timestamp}"}
        }
        date {
            match => [ "server_time", "ISO8601"]
            target => "@timestamp"
            timezone =>"+00:00"
        }
        mutate{
            remove_field => ["message", "server_time"]
        }
    }
    if [type] == "from_tcp" and [service][application] == "app" and [service][category] == "log" {
        grok {
            match => { "message" => "\[%{TIMESTAMP_ISO8601:server_time}\] \[%{LOGLEVEL:level}\] %{WORD:category} - %{USERNAME:hostname} %{NUMBER:pid} %{GREEDYDATA:content}" }
        }
        mutate{
            add_field => { "log_time" => "%{@timestamp}"}
        }
        date {
            match => [ "server_time", "ISO8601"]
            target => "@timestamp"
            timezone =>"+00:00"
        }
        mutate{
            remove_field => ["message", "server_time"]
        }
    }
}
output {
    if [type] == "from_tcp" and [service][application] == "app" and [service][category] == "access_log" {
        elasticsearch {
            hosts => ["http://localhost:9200"]
            index => "app-access_log-%{+YYYY.MM.dd}"
        }
    }
    if [type] == "from_tcp" and [service][application] == "app" and [service][category] == "log" {
        elasticsearch {
            hosts => ["http://localhost:9200"]
            index => "app-log-%{+YYYY.MM.dd}"
        }
    }
}

这里的grok需要按上报的message的内容，书写规则
- 写规则容易出错，难排查，所以要用下面的网站边检验边写
- Grok Debugger
内容保存为logstash安装目录下的./conf.d/app.conf
开启elasticsearch
- 进入安装的目录，./bin/elasticsearch

执行测试

./bin/logstash -f ./conf.d/app.conf --config.test_and_exit

测试通过之后，正式运行

./bin/logstash -f ./conf.d/app.conf --config.reload.automatic

开启kibana
- 进入安装的目录，./bin/kibana

运行代码框架

如果控制台输出下面的配置，代表设置成功

log4js - config =  {
  type: 'km-log4js-logstash-tcp',
  host: 'localhost',
  port: 5050,
  service: { application: 'app' }
}
{
  appenders: { out: { type: 'stdout' } },
  categories: { default: { appenders: [Array], level: 'OFF' } }
}

发送路由

log4js - loggingEvent =  LoggingEvent {
  startTime: 2021-12-17T11:06:47.919Z,
  categoryName: 'access_log',
  data: [ 'Mangoum.local 12399 GET /api_v1 304 1ms 192.168.0.222\n' ],
  level: Level { level: 20000, levelStr: 'INFO', colour: 'green' },
  context: {},
  pid: 12399
}

如果控制台输出上面的消息事件，代表发送到logstash成功

上面这些输出，都是插件的作者遗漏在里面的调试代码，可以自己到km-log4js-logstash-tcp包的index.js文件中注释掉
访问Elastic索引管理后台
- 看有没有索引
  - 有的话，设置一个索引模式app-access_log-
  - 有的话，设置另一个索引模式app-log-
- 新建索引模式后，会发现grok用模式解析出来的字段，并没有被索引？
  - 所以要刷新索引模式
    - 点击相应的索引模式，然后右上角的刷新按钮，这样以后这个模式新建的索引，都会索引这些grok解析出来的字段
    - 只要删除掉就会发现，后面都有t字表示索引字段
普及关于时区的问题
- T表示分隔符，Z表示的是UTC
- UTC：世界标准时间，在标准时间上加上8小时，即东八区时间，也就是北京时间。
- 举例
  - 北京时间：2020-01-14 00:00:00
  - 对应的国际标准时间格式为：2020-01-13T16:00:00.000Z
为了能让logstash每次都按照东八区的0-24点生成一个完整索引，而不是用国际UTC时间(这样每天北京时间8点才会生成新索引，也就是UTC的0点生成新索引)，需要做如下处理
- 新建一个log_time存储真正的@timestamp，这是我们用来查询的真实北京时间
- 我们自己额外传一个server_time
- 把server_time用date插件强行指定timezone为"+00:00"，进行转化，然后赋回给"@timestamp"，这样就能偷天换日
- 最终的效果是
```
"@timestamp": "2021-12-21T18:01:38.592Z",
"log_time": "2021-12-21T10:01:38.592Z",
"server_time": "2021-12-21T18:01:38.592"
```
  - 上面前两个均是UTC时间
  - 最后一个没有Z后缀，所以才能偷天换日
  - 现在是晚上6点
  - 机器看 @timestamp 来安排索引的，但是对我们来说是不准的，因为实际上这个时间已经是北京时间第二天2点了，但机器是按照UTC前面的日期来安排索引的
  - 运维人员看log_time，是准的，这个UTC时间会被kibana+8小时，转换为北京时间
- 这样我们建立索引模式时，就不能用@timestamp来排序，而是要指定为用log_time，这样符合人类的习惯
- 最后记得要删掉server_time

Keyboard shortcuts

【日志高级】搭配logstash