Demo: processing syslog

[abbshr]

水文一篇, 我也是近几天才开始搞, 里面很多细节与概念(如syslog, grok, elasticsearch等)没有过多挖掘, 挺多都是摘自官网的内容, 就当给初学者的介绍了~~

syslog简介

syslog是一个广泛用于os,network中的日志协议, 其中规定了日志的格式, 等级, 类型等等.

Linux中通过守护进程rsyslogd实现了syslog的收集存储, rsyslogd可以写入系统日志到本地也可以远程传输到一个网络中的日志收集点. /etc/rsyslogd.conf, /etc/rsyslogd.d/存放了rsyslogd的配置.

Linux中的syslog实体记录在/var/log目录下, 日志按类型等级分类为不同的文件, 如:

├── auth.log
├── boot.log
├── bootstrap.log
├── dist-upgrade
├── dmesg
├── dmesg.0
├── dpkg.log
├── kern.log
├── kern.log.1
├── lastlog
├── memcached.log
├── redis
│   └── redis.log
├── syslog
├── syslog.1
├── udev
└── Xorg.0.log.old

选择syslog做demo的主要原因是系统日志如实记录了系统运行过程中(when)(how)触发的(which)事件, 是运维/开发分析解决问题的终极武器.(其实是实在找不到其他日志了...), 目的就是希望透过这个例子介绍+总结如何使用ELK技术栈处理/分析数据, 让大家有个了解.

从Logstash开始

Logstash是Jordan Sissel于2009年开发的一个日志收集处理框架, 2013年被Elasticsearch公司收购并成为ELK Stack中一员.

Logstash的设计蛮符合Unix设计哲学, 核心高度解耦, 之间通过"事件"通信. 它的数据处理流程分为三大阶段:

1. input
2. filter
3. output

每个阶段表示一个数据处理的过程, 其中可以执行各种处理插件, 比如编码解码, 增删改字段, 模式匹配等等.

你会这一设计和iptables很像, input表示处理数据前的hook, output表示输出给另一个系统之前要做的事, 而最关键的就是filter过程, 各种丰富的插件都主要作用于filter. 因为是数据处理的核心过程, 避免复杂的计算影响Logstash整体性能, filter中的处理都是另开线程独立执行的.

这三个阶段都是选择性配置的, 缺少任何一个阶段都不会对logstash的运行产生影响. 在配置上, logstash采用了类似Ruby语法的DSL, 十分清晰易读写. 下面介绍几个调试/测试常用的插件:

codec

codec作为input/output阶段的子阶段, 负责数据格式的变换. codec有很多插件, 这里说的叫rubydebug, 测试时可以把output阶段产出以color inspection的形式打印到标准输出

# plugin: rubydebug
output {
    stdout {
        codec => rubydebug
    }
}

这是一个使用rubydebug输出的样例:

{
       "message" => "mad",
      "@version" => "1",
    "@timestamp" => "2015-12-03T07:54:43.076Z",
          "host" => "huanran-desktop",
          "show" => "fuck",
     "@metadata" => {
               "test" => "Hello",
        "retry_count" => 0
    }
}

generator

generator插件的作用是生成测试数据, 主要用途就是配合其他工具做性能测试.

input {
    # 产生100W个事件
    generator {
        count => 1000000
        message => "foo-bar"
        codec => plain
    }
}

output {
    stdout {
        # 这里用到了codec的dots插件, 作用是将每个output事件转成一个点"."
        # 以单字节"."产生事件
        codec => dots
    }
}

这样就可以结合pv工具实时监控logstash的处理速度了"

logstash -f config.conf | pv -abt > /dev/null

除此之外, 你也看到了上面使用的stdout, stdin插件, 没错他们用于配置标准输入/输出.

配置与启动

那么在哪里配置input,output,filter呢?
写在外部文件里即可, 启动logstash时可以通过-f参数来指定加载的配置. -f还接受目录形式的参数, logstash在初始化时会遍历这个目录, 按字母表顺序依次加载目录内每个文件, 然后拼接在一起.

logstash启动的比较慢, 因为需要启动JVM虚拟机以及许多初始化操作, 等待七八秒应该能就绪了.

logstash其实就是依赖各式插件实现数据流的处理任务, 这和Unix下的系统管理命令极其相似, 如使用sed,awk,uniq,sort等等等等小工具通过pipeline拼接在一起实现复杂的数据处理任务:

cat in | grep "\^Error\:" | cut -d"." -f1 | sort -rn | uniq | tee out > fifo

到这里你应该明白为什么之前说它"符合Unix涉及哲学"了吧~

从本地文件收集数据

以syslog为例, 这里先创建一个目录config, 里面包含两个文件:

• syslog-from-file.conf

• syslog-to-elasticsearch.conf

  下面配置input阶段:

# syslog-from-file.conf
#
# plugins: file
input {
    file {
        path => ["/var/log/syslog*"]
        type => "syslog"
        start_position => "beginning"
    }
}

每个插件都有自己的可选配置区域, 如file插件可以配置输入路径, 类型, 起始读取位置(默认从最后读取的位置开始). 为了演示, 我们先不配置filter过程, 直接跳到output:

# syslog-to-elasticsearch.conf
#
# plugin: elasticsearch
output {
    elasticsearch {
        hosts => ["127.0.0.1"]
        # 配置index字段, 表示讲输出数据存入elasticsearch中如下索引中
        # 这里的配置是每天一个索引
        index => "logstash-%{type}-%{+YYYY.MM.dd}"
    }
    stdout {
        codec => rubydebug
    }
}

好了, 接下来启动elasticsearch. 对, 你没看错, 就是直接启动不用配置! 这个demo中不需要. 然后启动logstash看看:

logstash -f config/

你应该会看到这样的滚动提示:

Received an event that has a different character encoding than you configured. {.....}

logstash默认接收的UTF-8编码字符, 说明日志里包含非UTF-8字符. (注: 因为path的写法导致logstash在读取日志时会把所有匹配syslog的文件全部读取, 但有些如syslog.1.gz这种属于归档文件, 肯定是binary直接编码的) 那我们就只用纯文本格式的syslog吧.

但是你会发现输出事件的时间戳和预想不太一样, 这些事件的时间戳是当前时间而不是日志每个记录条目的时间戳. 啊, 看来还需要在时间上做些处理呢~ 那就好好配置一下filter阶段.

这里我们使用了logstash DSL提供的逻辑表达式及流程控制语句, 以及新的插件grok, date等等, 不必惊慌, 接下来会介绍他们的:

# syslog-filter.conf
#
# plugin: grok, date
filter {
    # 先判断一下type是不是syslog
    # 因为input阶段都可以有多个输入源, 不同输入源产生的数据格式及其使用目的应该是不同的
    # 所以这里要根据数据类型选择处理逻辑
    if [type] == 'syslog' {

        # 这是一个syslog格式的grok正则
        # 它会从匹配的数据中提取捕获到的值, 并在事件中增加对应字段
        # 这里, 如syslog_timestamp, syslog_hostname, syslog_program, syslog_message, syslog_pid
        grok {
          match => {
            "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}"
          }
          # 匹配了上面正则的事件会进入这部分
          # 这里给事件增加两个信息字段, 包含当前时间和主机名称
          add_field => ["received_at", "%{@timestamp}"]
          add_field => ["received_from", "%{host}"]
        }

        # 提取syslog中的PRI字段, 如果没找到默认设置13
        syslog_pri {}

        # 修改timestamp为日志时间
        # 这里使用了date插件, 默认修改@timestamp为匹配的字段值
        date {
          # 匹配满足如下格式的syslog_timestamp字段
          match => ["syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
        }
    }
}

这回再跑一次看看, 应该有类似这样的数据产出了:

{
                 "message" => "Dec  3 12:06:19 huanran-desktop avahi-daemon[672]: Invalid response packet from host 10.0.1.176.",
                "@version" => "1",
              "@timestamp" => "2015-12-03T04:06:19.000Z",
                    "host" => "huanran-desktop",
                    "path" => "/var/log/syslog.1",
                    "type" => "syslog",
        "syslog_timestamp" => "Dec  3 12:06:19",
         "syslog_hostname" => "huanran-desktop",
          "syslog_program" => "avahi-daemon",
              "syslog_pid" => "672",
          "syslog_message" => "Invalid response packet from host 10.0.1.176.",
             "received_at" => "2015-12-04T03:46:06.009Z",
           "received_from" => "huanran-desktop",
    "syslog_severity_code" => 5,
    "syslog_facility_code" => 1,
         "syslog_facility" => "user-level",
         "syslog_severity" => "notice"
}

数据处理过后就是考虑如何展示/分析他们了, 接下来就可以打开kibana看看我们的数据被整理成什么样子了.

清理测试数据

如果导入了一堆乱七八糟的测试数据, 现在想重来一次怎么办? 像上面从文件导入数据的方式, logstash会在用户主目录创建以.sincedb_ 为前缀的文件, 记录最后一次读取位置的inode, major等信息. 如果要让logstash重头开始读文件, 那么需要删除这几个.sincedb_ 文件. 此外, 也可能需要清除elasticsearch, 做法 => rm -rf data/. 这样elasticsearch就会重新建立索引了.

logstash DSL

看了上面的配置语法, 是不是感觉很简洁明了但是完全不懂怎么写呢~~?

ok, 下面简单介绍一下logstash配置所用的DSL. 正如你所看到的, 它很简单:

# 配置是从前面讲的三个阶段开始的, 比如input, 其他阶段同理:
input {
    # 每个阶段都有个section, 就是这里
    # 这部分可以配置各种插件(plugins), 比如这个file插件:
    file {
        # 每个插件可以很多settings, 这里可以写该插件的详细配置信息
        path => "/var/log"
    }

    # 插件可以配置很多, 这里再增加一个输入源
    stdin {
        # logstash DSL中的值有这么几种: 数组, 布尔, byte, 字符串, 哈希, 数字, 以及codec
        # 每个section其实就是一个ruby的hash对象, 通过`=>`指定每个字段的值
        codec => plain
    }
}

经过每个阶段的数据都以事件对象传递出去, 事件是一个Ruby Hash对象, 可以对字段进行增删改操作.

# 数据以Hash object形式表示, 如何获取其中的字段? 
# DSL中以字段的名字表示其内容, 如root; 如果想引用字段的子节点, 就需要[root][child]的写法:
filter {
    # DSL中还支持流程控制语句和逻辑表达式
    if type == "test" {
        mutate {
            # 可以用%{field}语法在字符串中嵌入字段值:
            add_field => ["received_at", "%{@timestamp}"]
        }
    } else if [testField][name] == 'child' {
        grok {}
    }
}

grok表达式

前面已经简单介绍过grok插件的用法了, 这里说点额外的.

grok还支持把预定义的grok表达式写入到文件中, 这样就不用把各种复杂的正则一股脑写入grok{}了.

表达式示例如下:

USERNAME [a-zA-Z0-9._-]+
USER %{USERNAME}

第一行，用普通的正则表达式来定义一个 grok 表达式；
第二行，通过打印赋值格式，用前面定义好的 grok 表达式来定义另一个 grok 表达式。

grok 表达式的打印赋值格式的完整语法是下面这样的：

%{PATTERN_NAME:capture_name:data_type}

然后把写好的grok表达式放到一个目录里, 然后通过patterns_dir字段指定它:

grok {
    patterns_dir => "./patterns"
    match { "field" => "%{USERNAME}" }
}

这里有一些官方定义好的内置pattern可以使用: https://github.com/logstash-plugins/logstash-patterns-core

geoip

GeoIP 库可以根据 IP 地址提供对应的地域信息，包括国别，省市，经纬度等，对于可视化地图和区域统计非常有用。

geoip {
    source => "message"
}

当message为"183.60.92.253"时会产生类似这样的事件:

{
       "message" => "183.60.92.253",
      "@version" => "1",
    "@timestamp" => "2014-08-07T10:32:55.610Z",
          "host" => "raochenlindeMacBook-Air.local",
         "geoip" => {
                      "ip" => "183.60.92.253",
           "country_code2" => "CN",
           "country_code3" => "CHN",
            "country_name" => "China",
          "continent_code" => "AS",
             "region_name" => "30",
               "city_name" => "Guangzhou",
                "latitude" => 23.11670000000001,
               "longitude" => 113.25,
                "timezone" => "Asia/Chongqing",
        "real_region_name" => "Guangdong",
                "location" => [
            [0] 113.25,
            [1] 23.11670000000001
        ]
    }
}

redis

input阶段的redis插件. 随着系统的扩展, 日志来源的增加, 最好能有一个消息队列负责统一输入并均衡logstash负载. 而Redis是这是官方建议的输入源.

redis插件对于数据类型只有三个可选项:

• list
• channel
• pattern_channel

这个值在data_type字段中配置, 没有默认选项. 如果选择list类型, 那么将会用BLPOP阻塞读取(即同一时间只有一个进程能读到数据, 用于解决并发消费问题). 如果选择channel/pattern_channel类型, 那么所有订阅了相同key模式的客户端都将通过SUBSCRIBE/PSUBSCRIBE接收到相同数据.

如:

input {
    redis {
        # 一次处理一个元素, 即产生一个事件
        batch_count => 1
        # 处理线程数量
        threads => 1
        data_type => "list"
        key => "logstash-list"
        host => "127.0.0.1"
        port => 6379
    }
}

email

output阶段插件, 可以使用本机SMTP服务发送邮件通知, 如:

output {
    email {
        to => "[email protected],[email protected]"
        cc => "[email protected]"
        via => "smtp"
        subject => "Warning: %{title}"
        options => {
            smtpIporHost       => "localhost",
            port               => 25,
            domain             => 'localhost.localdomain',
            userName           => nil,
            password           => nil,
            authenticationType => nil, # (plain, login and cram_md5)
            starttls           => true
        }
        htmlbody => ""
        body => ""
        attachments => ["/path/to/filename"]
    }
}

更多插件及详细介绍见 https://www.elastic.co/guide/en/logstash/current/index.html