《ELK Stack权威指南（第2版）》一 2.2　编解码配置

本节书摘来自华章出版社《ELK Stack权威指南（第2版）》一书中的第2章，第2.2节，作者 饶琛琳  更多章节内容可以访问云栖社区“华章计算机”公众号查看。 

2.2　编解码配置

Codec是Logstash从1.3.0版开始新引入的概念（Codec来自Coder/decoder两个单词的首字母缩写）。

在此之前，Logstash只支持纯文本形式输入，然后以过滤器处理它。但现在，我们可以在输入期处理不同类型的数据，这全是因为有了Codec设置。

所以，这里需要纠正之前的一个概念。Logstash不只是一个input | filter | output的数据流，而是一个input | decode | filter | encode | output的数据流！Codec就是用来decode、encode事件的。

Codec的引入，使得Logstash可以更好、更方便地与其他有自定义数据格式的运维产品共存，比如graphite、fluent、netflow、collectd，以及使用msgpack、json、edn等通用数据格式的其他产品等。

事实上，我们在第一个“Hello World”用例中就已经用过Codec了—rubydebug就是一种Codec！虽然它一般只会用在stdout插件中，作为配置测试或者调试的工具。

这个五段式的流程说明源自Perl版的Logstash（后来改名叫Message::Passing模块）的设计。本书稍后5.8节会对该模块稍作介绍。

2.2.1　JSON编解码

在早期的版本中，有一种降低Logstash过滤器的CPU负载消耗的做法盛行于社区（在当时的Cookbook上有专门的一节介绍）：直接输入预定义好的JSON数据，这样就可以省略掉filter/grok配置！

这个建议依然有效，不过在当前版本中需要稍微做一点配置变动，因为现在有专门的Codec设置。

1.配置示例

2.运行结果
下面访问一下用Nginx发布的Web页面，然后你会看到Logstash进程输出类似下面这样的内容：
{
"@timestamp" =>"2014-03-21T18:52:25.000+08:00",
"@version" =>"1",
"host" =>"raochenlindeMacBook-Air.local",
"client" =>"123.125.74.53",
"size" =>8096,
"responsetime" =>0.04,
"domain" =>"www.domain.com",
"url" =>"/path/to/file.suffix",
"status" =>"200"
}




3. Nginx代理服务的日志格式问题

对于一个Web服务器的访问日志，看起来已经可以很好的工作了。不过如果Nginx是作为一个代理服务器运行的话，访问日志里有些变量，比如说$upstream_response_time，可能不会一直是数字，它也可能是一个“-”字符串！这会直接导致Logstash对输入数据验证报异常。

有两个办法解决这个问题：

1）用sed在输入之前先替换-成0。运行Logstash进程时不再读取文件而是标准输入，这样命令就成了下面这个样子：

tail -F /var/log/nginx/proxy_access.log_json \

    | sed 's/upstreamtime":-/upstreamtime":0/' \

    | /usr/local/logstash/bin/logstash -f /usr/local/logstash/etc/proxylog.conf

2）日志格式中统一记录为字符串格式（即都带上双引号），然后再在Logstash中用filter/mutate插件来变更应该是数值类型的字符字段的值类型。

有关LogStash::Filters::Mutate的内容，本书稍后会有介绍。

2.2.2　多行事件编码

有些时候，应用程序调试日志会包含非常丰富的内容，为一个事件打印出很多行内容。这种日志通常都很难通过命令行解析的方式做分析。

而Logstash正为此准备好了codec/multiline插件！当然，multiline插件也可以用于其他类似的堆栈式信息，比如Linux的内核日志。

配置示例如下：

input {

    stdin {

        codec => multiline {

            pattern =>"^\["

            negate => true

            what =>"previous"

        }

    }

}

运行Logstash进程，然后在等待输入的终端中输入如下几行数据：

[Aug/08/08 14:54:03] hello world

[Aug/08/09 14:54:04] hello logstash

    hello best practice

    hello raochenlin

[Aug/08/10 14:54:05] the end

你会发现Logstash输出下面这样的返回：




{
"@timestamp" =>"2014-08-09T13:32:03.368Z",
"message" =>"[Aug/08/08 14:54:03] hello world\n",
"@version" =>"1",
"host" =>"raochenlindeMacBook-Air.local"
}
{
"@timestamp" =>"2014-08-09T13:32:24.359Z",
"message" =>"[Aug/08/09 14:54:04] hello logstash\n\n    hello best practice\n\n
            hello raochenlin\n",
"@version" =>"1",
    "tags" => [
        [0] "multiline"
    ],
"host" =>"raochenlindeMacBook-Air.local"
}




你看，后面这个事件，在“message”字段里存储了三行数据！

输出的事件中没有最后一行的"the end"字符串，这是因为你最后输入的回车符\n并不匹配设定的^\[正则表达式，Logstash还得等下一行数据直到匹配成功后才会输出这个事件。

其实这个插件的原理很简单，就是把当前行的数据添加到前面一行后面，直到新进的当前行匹配^\[正则为止。这个正则还可以用grok表达式，稍后你就会学习这方面的内容。具体的Java日志正则见：https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/java

说到应用程序日志，Log4j肯定是第一个被大家想到的，使用codec/multiline也确实是一个办法。

不过，如果你本身就是开发人员，或者可以推动程序修改变更的话，Logstash还提供了另一种处理Log4j的方式：input/log4j。与codec/multiline不同，这个插件是直接调用了org.apache.log4j.spi.LoggingEvent处理TCP端口接收的数据。后面3.6节会详细讲述Log4j的用法。

2.2.3　网络流编码

NetFlow是Cisco发明的一种数据交换方式。NetFlow提供网络流量的会话级视图，记录下每个TCP/IP事务的信息。它的目的不是像tcpdump那样提供网络流量的完整记录，而是汇集起来形成更易于管理和易读的流向和容量的分析监控。

Cisco上配置NetFlow的方法，请参照具体的设备说明，主要是设定采集服务器的地址和端口，为运行Logstash服务的主机地址和端口（示例中为9995）。

采集NetFlow数据的Logstash服务配置示例如下：




input {
    udp {
        port => 9995
        codec => netflow {
            definitions =>"/opt/logstash-1.4.2/lib/logstash/codecs/netflow/netflow.yaml"
            versions => [5]
        }
    }
}
output {
    elasticsearch {
        index =>"logstash_netflow5-%{+YYYY.MM.dd}"
        host =>"localhost"
    }
}
由于该插件生成的字段较多，所以建议对应的Elasticsesarch索引模板也需要单独提交：
# curl -XPUT localhost:9200/_template/logstash_netflow5 -d '{
"template" : "logstash_netflow5-*",
"settings": {
        "index.refresh_interval": "5s"
},
"mappings" : {
        "_default_" : {
        "_all" : {"enabled" : false},
        "properties" : {
            "@version": { "index": "analyzed", "type": "integer" },
            "@timestamp": { "index": "analyzed", "type": "date" },
            "netflow": {
            "dynamic": true,
            "type": "object",
            "properties": {
                "version": { "index": "analyzed", "type": "integer" },
                "flow_seq_num": { "index": "not_analyzed", "type": "long" },
                "engine_type": { "index": "not_analyzed", "type": "integer" },
                "engine_id": { "index": "not_analyzed", "type": "integer" },
                "sampling_algorithm": { "index": "not_analyzed", "type": "integer" },
                "sampling_interval": { "index": "not_analyzed", "type": "integer" },
                "flow_records": { "index": "not_analyzed", "type": "integer" },
                "ipv4_src_addr": { "index": "analyzed", "type": "ip" },
                "ipv4_dst_addr": { "index": "analyzed", "type": "ip" },
                "ipv4_next_hop": { "index": "analyzed", "type": "ip" },
                "input_snmp": { "index": "not_analyzed", "type": "long" },
                "output_snmp": { "index": "not_analyzed", "type": "long" },
                "in_pkts": { "index": "analyzed", "type": "long" },
                "in_bytes": { "index": "analyzed", "type": "long" },
                "first_switched": { "index": "not_analyzed", "type": "date" },
                "last_switched": { "index": "not_analyzed", "type": "date" },
                "l4_src_port": { "index": "analyzed", "type": "long" },
                "l4_dst_port": { "index": "analyzed", "type": "long" },
                "tcp_flags": { "index": "analyzed", "type": "integer" },
                "protocol": { "index": "analyzed", "type": "integer" },
                "src_tos": { "index": "analyzed", "type": "integer" },
                "src_as": { "index": "analyzed", "type": "integer" },
                "dst_as": { "index": "analyzed", "type": "integer" },
                "src_mask": { "index": "analyzed", "type": "integer" },
                "dst_mask": { "index": "analyzed", "type": "integer" }
              }
          }
       }
    }
  }
}'
Elasticsearch索引模板的功能，本书稍后12.6节会有详细介绍。

2.2.4　collectd输入

collectd是一个守护（daemon）进程，用来收集系统性能和提供各种存储方式来存储不同值的机制。它会在系统运行和存储信息时周期性的统计系统的相关统计信息。利用这些信息有助于查找当前系统性能瓶颈（如作为性能分析performance analysis）和预测系统未来的load（如能力部署capacity planning）等

下面简单介绍一下:collectd的部署以及与Logstash对接的相关配置实例。

1. collectd的安装

collectd的安装同样有两种方式，使用官方软件仓库安装或源代码安装。

使用官方软件仓库安装（推荐）

collectd官方有一个隐藏的软件仓库：https://pkg.ci.collectd.org，构建有RHEL/CentOS （rpm），Debian/Ubuntu（deb）的软件包，如果你使用的操作系统属于上述的，那么推荐使用软件仓库安装。

目前collectd官方维护3个版本：5.4、5.5、5.6。根据需要选择合适的版本仓库。下面示例安装5.5版本的方法。

Debian/Ubuntu仓库安装：

echo "deb http://pkg.ci.collectd.org/deb $(lsb_release -sc) collectd-5.5" | sudo tee /etc/apt/sources.list.d/collectd.list

curl -s https://pkg.ci.collectd.org/pubkey.asc | sudo apt-key add -

sudo apt-get update && sudo apt-get install -y collectd

注意，Debian/Ubuntu软件仓库自带有collectd软件包，如果软件仓库自带的版本足够你使用，那么可以不用添加仓库，直接通过apt-get install collectd即可。

RHEL/CentOS仓库安装：

cat > /etc/yum.repos.d/collectd.repo <<EOF

[collectd-5.5]

name=collectd-5.5

baseurl=http://pkg.ci.collectd.org/rpm/collectd-5.5/epel-\$releasever-\$basearch/

gpgcheck=1

gpgkey=http://pkg.ci.collectd.org/pubkey.asc

EOF

yum install -y collectd

你如果需要使用其他collectd插件，此时也可一并安装对应的collectd-xxxx软件包。

源码安装collectd

collectd目前维护3个版本：5.4、5.5、5.6。源代码编译安装时同样可以根据自己需要选择对应版本的源码下载：

wget http://collectd.org/files/collectd-5.4.1.tar.gz

tar zxvf collectd-5.4.1.tar.gz

cd collectd-5.4.1

./configure --prefix=/usr --sysconfdir=/etc --localstatedir=/var --libdir=/usr/lib --mandir=/usr/share/man --enable-all-plugins

make && make install

源代码编译安装需要预先解决好各种环境依赖，在RedHat平台上要提前安装如下软件包：

rpm -ivh "http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm"

yum -y install libcurl libcurl-devel rrdtool rrdtool-devel perl-rrdtool rrdtool-prel libgcrypt-devel gcc make gcc-c++ liboping liboping-devel perl-CPAN net-snmp net-snmp-devel

安装启动脚本如下：

cp contrib/redhat/init.d-collectd /etc/init.d/collectd

chmod +x /etc/init.d/collectd

启动collectd如下：

service collectd start

2. collectd的配置

以下配置可以实现对服务器基本的CPU、内存、网卡流量、磁盘I/O以及磁盘空间占用情况的监控：




Hostname "host.example.com"
LoadPlugin interface
LoadPlugin cpu
LoadPlugin memory
LoadPlugin network
LoadPlugin df
LoadPlugin disk
<Plugin interface>
    Interface "eth0"
    IgnoreSelected false
</Plugin>
<Plugin network>
<Server "10.0.0.1""25826"> ## Logstash 的 IP 地址和 collectd 的数据接收端口号
</Server>
</Plugin>




3. Logstash的配置

以下配置实现通过Logstash监听25826端口，接收从collectd发送过来的各项检测数据。

注意，logstash-filter-collectd插件本身需要单独安装，logstash插件安装说明之前已经讲过：

bin/logstash-plugin install logstash-filter-collectd

Logstash默认自带有collectd的codec编码插件。

推荐配置示例如下：