从零开始在你的服务器集群上搭建elk-stack(日志系统)

作者: 章鱼奇Josefa | 来源:发表于2017-11-30 22:00 被阅读107次

开始入门Logstash(上)

2017.11.28

写这篇文章的时候,我的Logstash,Filebeat,Elasticsearch,kibana都是6.0版本.本文的适用对象:像我一样没有学过运维的小白,另外,这篇文章其实就是翻译官网上的文档+我自己在使用中的备注,本人英语六级水平,有错误的地方请轻喷.

Elk技术栈是分别由Elasticsearch,Logstash,Kibana组成,随着技术的推进也衍生了更多性能更好的工具,比如收集文件更快的Filebeat,在有生之年我希望能把整个技术栈都展现在大家眼前..

本文的内容将会指引你入门Logstash,从最简单的应用到创建多条管道(pipeline)将apcahe的日志作为输入,然后解析日志,最后将日志输出到elasticsearch.最后你将组装多个输入以及输出插件从不同的源头处最终获取到统一的数据.

本文将循序渐进的带你了解Logstash

安装Logstash
运行第一个实例
通过Logstash解析日志
整合多输入,输出插件(下篇文章)

安装Logstash

Logstash 是动态数据收集管道，拥有可扩展的插件生态系统，能够与 Elasticsearch 产生强大的协同作用。

官网上有全部软件的下载地址:The Elastic Stack Download,我的安装方式很简单,首先,你需要安装java环境,然后再下载软件

yum install yum install java-1.8.0
wget https://artifacts.elastic.co/downloads/logstash/logstash-6.0.0.tar.gz
tar zxf logstash-6.0.0.tar.gz

之后我会改名,然后删除压缩包,对于其他软件也是这样处理:

mv logstash-6.0.0 logstash
rm logstash-6.0.0.tar.gz

这样做完后,我的用户目录会比较清晰:

>ll
drwxr-xr-x  7 j j      4096 Nov 11 02:42 elastic
drwxr-xr-x  6 j j      4096 Nov 28 17:36 filebeat
drwxrwxr-x 12 j j      4096 Nov 26 22:39 kibana
drwxrwxr-x 12 j j      4096 Nov 29 09:59 logstash

另外官网上还提供了其他各种各样的安装方式:Installing Logstash,如果你跟我一样是个初学者,可以就用我的方法来安装

运行你的第一个实例

一个Logstash需要两个最近基本的元素,输入以及输出,以及一个可选的元素,过滤.输入组件负责从数据源获取数据,过滤组件负责按需修饰数据(过滤,格式化等),输出组件负责将数据写入指定的目的地.

模型图
现在,来运行一条最基本的命令来测试你的Logstash是否安装成功:

cd logstash
bin/logstash -e 'input { stdin { } } output { stdout {} }'

如果你看到Pipeline stared的日志输出在命令行中,就代表运行成功啦
> bin目录的地址会根据不同的安装环境(.zip,.tar.gz,rpm,Docker,)而改变,具体可以看这里=>
Logstash目录布局

-e参数可以直接在命令行中配置参数,可以让你快速的测试配置而不用编辑一个配置文件.实例中的管道使用标准输入stdin作为输入,并将其以一定的格式作为标准输出stdou

在你看到"Pipeline main started"后,在命令行中输入Hello world

Hello world
2017-11-28T08:11:35.811Z localhost Hello world

Logstash在日志中添加了时间戳以及IP地址的信息,想要退出的话可以执行CTRL-D来退出.

恭喜您呐!你已经创建并且跑了一个基础的Logstash管道.准备一下迎接真实世界的黑暗吧.

用Logstash来修饰日志

在上一节中,你跑了一个最基础的Logstash管道来测试你的Logstash.在真实环境中,一个Logstash管道会复杂一些:往往会有不止一个的输入,过滤,以及输出的组件.

在这一节中,你将使用Filebeat获取Apache的网络日志并将其作为输入,将这些日志解析成明确的信息,并将解析后的数据输入Elasticsearch集群.这次你将使用文件来配置管道.

开始前点击这里来下载本次教程的示例数据.解压下载后得到文件.

配置Filebeat将日志传输给Logstash

在你创建Logstash管道前,你将配置Filebeat将日志传输给Logstash.Filebeat是一个轻量,资源友好的从服务器生成的文件中收集日志的工具,并将日志发送给你的Logstash实例进行处理.Filebeat被设计成稳定,高效(低延迟),在主机上占用很少的资源,Beats input插件最小化了对Logstash实例所需要的资源.

在一个典型的部署中,Filebeat将在你运行Logstash实例以外的多个机器中收集日志.为了这次示例,Logstash与Filebeat将会在同一台机器上运行

Logstash默认安装后会带有Beats input插件,该组件使Logstash可以从Elastic Beats 框架中接收事件.这意味着任何一个使用Beat框架写的服务都能将事件数据传输给Logstash,比如Packetbeat和Metricbeat

首先呢,你需要一个Filebeat.

安装完Filebeat后,你需要配置它.打开安装目录的filebeat.yml文件,替换如下的内容,要保证paths指向了之前下载的示例数据logstash-tutorial.log(文件名可能不一样,自行改名):

filebeat.prospectors:

type: log
paths:

/path/to/file/logstash-tutorial.log
output.logstash:
hosts: ["localhost:5043"]

请使用绝对路径,output.logstash前不能有空格.

保存你的编辑
为了简化这次的配置,不用指定TLS/SSL这些在真实环境中需要的配置

在数据源机器,用如下命令启动Filebeat

sudo ./filebeat -e -c filebeat.yml -d "publish"

如果使用root账号运行Filebeat,需要修改配置文件的拥有者,这里建议为elk创建新的用户以及用户组,因为Elasticsearch也是不能使用root用户运行的.

Filebeat会尝试去连接5043端口,直到Logstash与Beats插件启动前都不会在该端口上有响应,所以现在你看到的应该都是连接端口失败的消息.

配置Logstash使用Filebeat作为输入

接下来,你会创建一个使用Beats输入插件来接收Beats事件的Logstash管道.

下面的文字代表了这次管道配置的骨架信息:

The # character at the beginning of a line indicates a comment.Use # comments to describe your configuration.
input {
}
# The filter part of this file is commented out to indicate that it is
# optional.
# filter {
#
# }
output {
}

这个骨架配置不会起效,因为没有任何有效的配置

开始前,将这段骨架配置复制到你自己的配置中,并命名文件为first-pipeline.conf,保存在你的Logstash目录中

接下来,配置你的Logstash实例使用Beats input插件,在first-pipeline.conf文件的input部分加入如下文本

beats {
port => "5043"
}

在未来你会使用Logstash推送到Elasticsearch,不过现在,先使用标准输出output来打印信息:

stdout { codec => rubydebug }

现在,你的first-pipeline.conf文件应该像这样:

input {
beats {
port => "5043"
}
}
# The filter part of this file is commented out to indicate that it is
# optional.
# filter {
#
# }
output {
stdout { codec => rubydebug }
}

用这个命令来验证你的配置是否正确:

bin/logstash -f first-pipeline.conf --config.test_and_exit

--config.test_and_exit选项将会解析并且提供配置文件的错误信息,如果配置通过了配置测试,运行这个命令来启动Logstash:

bin/logstash -f first-pipeline.conf --config.reload.automatic

--config.reload.automatic选项将会自动读取配置文件,这样你就不用再每次修改配置后都要停止再启动Logstash了

Logstash启动后,你可能会看到多条关于Logstash忽略了pipelines.yml文件的警告信息,你可以放心的忽略这些告警.pipelines.yml文件是用来配置一个Logstash实例运行多个管道,本次示例中,你只运行一个管道.

如果你的管道正常工作,你将会收到多条事件:

{
    "@timestamp" => 2017-11-09T01:44:20.071Z,
        "offset" => 325,
      "@version" => "1",
          "beat" => {
            "name" => "My-MacBook-Pro.local",
        "hostname" => "My-MacBook-Pro.local",
         "version" => "6.0.0"
    },
          "host" => "My-MacBook-Pro.local",
    "prospector" => {
        "type" => "log"
    },
        "source" => "/path/to/file/logstash-tutorial.log",
       "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
          "tags" => [
        [0] "beats_input_codec_plain_applied"
    ]
}
...

通过Grok Filter插件来解析网络日志

现在呢,你有一条通过Filebeat来读取日志的管道了.然后应该注意到现在的日志信息并不是理想中的格式.你想讲日志解析成一条清晰明了的信息,这样的话,就需要使用到grok过滤插件

grok过滤插件是Logstash支持的默认插件之一,想知道管理Logstash插件的更多情况,可以来这里:reference documentation

grok插件让你能够将可读性差,非结构化的日志数据解析成某种结构可查询的数据.

一条具有代表性的网络服务日志如下:

83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png
HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel
Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"

开头的ip地址很容易识别,括号里的时间戳也是.为了解析数据,你可以使用%{COMBINEDAPACHELOG}grok 表达式,讲会通过如下表格来组织文本:

信息	字段
IP Address	`clientip`
User ID	`ident`
User Authentication	`auth`
timestmp	`timestamp`
HTTP Verb	`verb`
Request body	`request`
HTTP Version	`httpversion`
HTTP Status Code	`response`
Bytes served	`bytes`
Referrer URL	`referrer`
User agent	`agent`

如果字构建grok表达式时遇到问题,可以尝试使用Grok Debugger.

修改first-pipeline.conf文件替换filter的内容:

filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}

修改完后,你的first-pipeline.conf应该是这个样子的:

input {
    beats {
        port => "5043"
    }
}
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}
output {
    stdout { codec => rubydebug }
}

保存文件,因为之前你打开了自动重载配置文件的设置,所以不需要重启Logstash以生效.然而,还是要强制Filebeat重头开始读取日志.你需要做的事很简单,找到运行着Filebeat的命令行,按下Ctrl-C关闭Filebeat,删除Filebeat的注册文件:

sudo rm data/registry

正是因为Filebeat讲文件的收录状态记录在了注册文件中,所以当你删除了注册文件后,就能强制Filebeat从头读取这些文件啦

接下来,重新运行Filebeat:

sudo ./filebeat -e -c filebeat.yml -d "publish"

Logstash重新载入config时会有些许的延迟,当你清除注册文件后可能得等一会儿

在Logstash的grok表达式生效后,会收到如下的JSON格式的事件:

{
        "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png",
          "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
         "offset" => 325,
           "auth" => "-",
          "ident" => "-",
           "verb" => "GET",
     "prospector" => {
        "type" => "log"
    },
         "source" => "/path/to/file/logstash-tutorial.log",
        "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
           "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
       "referrer" => "\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"",
     "@timestamp" => 2017-11-09T02:51:12.416Z,
       "response" => "200",
          "bytes" => "203023",
       "clientip" => "83.149.9.216",
       "@version" => "1",
           "beat" => {
            "name" => "My-MacBook-Pro.local",
        "hostname" => "My-MacBook-Pro.local",
         "version" => "6.0.0"
    },
           "host" => "My-MacBook-Pro.local",
    "httpversion" => "1.1",
      "timestamp" => "04/Jan/2015:05:13:42 +0000"
}

注意到事件同时包括了原始信息,日志被细分成了特定的字段.

用Geoip Filter插件里优化你的数据

为了使你的数据更加便于搜索,过滤组件可以从现有的数据中获取额外的信息.例如,geoip插件查询ip,从地址获取地理位置,并添加位置信息到日志中.

来吧,添加geoip插件的设置到你的配置文件first-pipeline.conf中去:

 geoip {
        source => "clientip"
    }

保存设置,像之前一样强制Filebeat读取日志.停止运行Filebeat(按下Ctrl-C),删除注册文件,然后运行Filebeat:

sudo ./filebeat -e -c filebeat.yml -d "publish"

注意到推送的事件中已经包含了地理信息:

{
        "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png",
          "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
          "geoip" => {
              "timezone" => "Europe/Moscow",
                    "ip" => "83.149.9.216",
              "latitude" => 55.7485,
        "continent_code" => "EU",
             "city_name" => "Moscow",
          "country_name" => "Russia",
         "country_code2" => "RU",
         "country_code3" => "RU",
           "region_name" => "Moscow",
              "location" => {
            "lon" => 37.6184,
            "lat" => 55.7485
        },
           "postal_code" => "101194",
           "region_code" => "MOW",
             "longitude" => 37.6184
    },
    ...

将你的数据推给Elasticsearch

经过上述的步骤,你的日志已经被细分成很多字段了,Logstash管道可以将数据编入Elasticsearch集群.修改first-pipeline.conf文件,替换如下内容:

output {
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}

设置了这个内容后,Logstash会使用http协议去连接Elasticsearch.上面的示例假设Logstash以及Elasticsearch在同一台设备上运行,你可以通过hosts设置指定一台远程的Elasticsearch实例,比如hosts => ["es-machine:9092"].

现在,你的first-pipeline.conf配置文件的输入,过滤,输出都有了正确的设置,看起来是这个样子的:

input {
    beats {
        port => "5043"
    }
}
 filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}
output {
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}

我还是保存了标准输出的output,为了检测是否有正常的输出

保存设置,初始化你的Filebeat.Ctrl-C,删除注册文件,启动Filebeat

sudo ./filebeat -e -c filebeat.yml -d "publish"

测试你的管♂道

现在Logstash管道已经设置成讲数据添加至Elasticsearch集群的索引了,你可以查询Elasticsearch

试着通过grok插件创建的字段来查询Elasticsearch.讲命令中的$DATE以YYYY.MM.DD的格式替换成当前的日期:

curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=response=200'

比如我得url就是这样的:curl -XGET 'localhost:9200/logstash-2017.11.28/_search?pretty&q=response=200'

你应该会得到多条反馈:

{
  "took": 50,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 98,
    "max_score": 2.793642,
    "hits": [
      {
        "_index": "logstash-2017.11.09",
        "_type": "doc",
        "_id": "3IzDnl8BW52sR0fx5wdV",
        "_score": 2.793642,
        "_source": {
          "request": "/presentations/logstash-monitorama-2013/images/frontend-response-codes.png",
          "agent": """"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"""",
          "geoip": {
            "timezone": "Europe/Moscow",
            "ip": "83.149.9.216",
            "latitude": 55.7485,
            "continent_code": "EU",
            "city_name": "Moscow",
            "country_name": "Russia",
            "country_code2": "RU",
            "country_code3": "RU",
            "region_name": "Moscow",
            "location": {
              "lon": 37.6184,
              "lat": 55.7485
            },
            "postal_code": "101194",
            "region_code": "MOW",
            "longitude": 37.6184
          },
          "offset": 2932,
          "auth": "-",
          "ident": "-",
          "verb": "GET",
          "prospector": {
            "type": "log"
          },
          "source": "/path/to/file/logstash-tutorial.log",
          "message": """83.149.9.216 - - [04/Jan/2015:05:13:45 +0000] "GET /presentations/logstash-monitorama-2013/images/frontend-response-codes.png HTTP/1.1" 200 52878 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"""",
          "tags": [
            "beats_input_codec_plain_applied"
          ],
          "referrer": """"http://semicomplete.com/presentations/logstash-monitorama-2013/"""",
          "@timestamp": "2017-11-09T03:11:35.304Z",
          "response": "200",
          "bytes": "52878",
          "clientip": "83.149.9.216",
          "@version": "1",
          "beat": {
            "name": "My-MacBook-Pro.local",
            "hostname": "My-MacBook-Pro.local",
            "version": "6.0.0"
          },
          "host": "My-MacBook-Pro.local",
          "httpversion": "1.1",
          "timestamp": "04/Jan/2015:05:13:45 +0000"
        }
      },
    ...

现在试一下用ip获取的地理位置来搜索.同样是替换$DATE以YYYY.MM.DD的格式为当前的时间:

curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=geoip.city_name=Buffalo'

只有几条数据来自Buffalo,所以查询结果日下:

{
  "took": 9,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 2,
    "max_score": 2.6390574,
    "hits": [
      {
        "_index": "logstash-2017.11.09",
        "_type": "doc",
        "_id": "L4zDnl8BW52sR0fx5whY",
        "_score": 2.6390574,
        "_source": {
          "request": "/blog/geekery/disabling-battery-in-ubuntu-vms.html?utm_source=feedburner&utm_medium=feed&utm_campaign=Feed%3A+semicomplete%2Fmain+%28semicomplete.com+-+Jordan+Sissel%29",
          "agent": """"Tiny Tiny RSS/1.11 (http://tt-rss.org/)"""",
          "geoip": {
            "timezone": "America/New_York",
            "ip": "198.46.149.143",
            "latitude": 42.8864,
            "continent_code": "NA",
            "city_name": "Buffalo",
            "country_name": "United States",
            "country_code2": "US",
            "dma_code": 514,
            "country_code3": "US",
            "region_name": "New York",
            "location": {
              "lon": -78.8781,
              "lat": 42.8864
            },
            "postal_code": "14202",
            "region_code": "NY",
            "longitude": -78.8781
          },
          "offset": 22795,
          "auth": "-",
          "ident": "-",
          "verb": "GET",
          "prospector": {
            "type": "log"
          },
          "source": "/path/to/file/logstash-tutorial.log",
          "message": """198.46.149.143 - - [04/Jan/2015:05:29:13 +0000] "GET /blog/geekery/disabling-battery-in-ubuntu-vms.html?utm_source=feedburner&utm_medium=feed&utm_campaign=Feed%3A+semicomplete%2Fmain+%28semicomplete.com+-+Jordan+Sissel%29 HTTP/1.1" 200 9316 "-" "Tiny Tiny RSS/1.11 (http://tt-rss.org/)"""",
          "tags": [
            "beats_input_codec_plain_applied"
          ],
          "referrer": """"-"""",
          "@timestamp": "2017-11-09T03:11:35.321Z",
          "response": "200",
          "bytes": "9316",
          "clientip": "198.46.149.143",
          "@version": "1",
          "beat": {
            "name": "My-MacBook-Pro.local",
            "hostname": "My-MacBook-Pro.local",
            "version": "6.0.0"
          },
          "host": "My-MacBook-Pro.local",
          "httpversion": "1.1",
          "timestamp": "04/Jan/2015:05:29:13 +0000"
        }
      },
     ...

如果你在使用kibana视觉化你的数据,同样可以在kibana上研究FIlebeat数据

kibana
通过这里来了解通过Filebeat来读取Kibana索引模式: Filebeat getting started docs

至此,给自己鼓个掌吧,你已经成功的创建了一个将apache的网络日志作为输入,经过插件解析成指定的字段,并且将其写入了Elasticsearch群.接下来,你还会学习到如何创建一个有多输入以及输出组件的管道.

也给自己鼓个掌..

从零开始在你的服务器集群上搭建elk-stack(日志系统)

开始入门Logstash(上)

安装Logstash

运行你的第一个实例

用Logstash来修饰日志

配置Filebeat将日志传输给Logstash

配置Logstash使用Filebeat作为输入

通过Grok Filter插件来解析网络日志

用Geoip Filter插件里优化你的数据

将你的数据推给Elasticsearch

测试你的管♂道

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

架构服务器数据库

Elk技术栈

程序员

从零开始在你的服务器集群上搭建elk-stack(日志系统)

开始入门Logstash(上)

安装Logstash

运行你的第一个实例

用Logstash来修饰日志

配置Filebeat将日志传输给Logstash

配置Logstash使用Filebeat作为输入

通过Grok Filter插件来解析网络日志

用Geoip Filter插件里优化你的数据

将你的数据推给Elasticsearch

测试你的管♂道

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

架构 服务器 数据库

Elk技术栈

程序员

架构服务器数据库