ELK日志分析学习笔记

（贴一篇之前工作期间整理的elk学习笔记）

ELK官网 https://www.elastic.co

ELK日志分析系统学习笔记

概念：ELK = elasticsearch + logstash + kibana

编程语言分别是 Java、JRuby、Ruby

概括的说，logstash 采集和结构化日志，输入elasticsearch创建索引，kibana查询elasticsearch这个搜索引擎 来完成数据的分析展示。

ELK 介绍：https://yq.aliyun.com/articles/73622

elasticsearch 基本概念，原理

详见：http://www.cnblogs.com/valor-xh/p/6095894.html

基本概念

索引(Index)

ES将数据存储于一个或多个索引中，索引是具有类似特性的文档的集合。类比传统的关系型数据库领域来说，索引相当于SQL中的一个数据库，或者一个数据存储方案(schema)。

举例说：pandora的所有日志可以存放于一个索引之中。

索引由其名称(必须为全小写字符)进行标识，并通过引用此名称完成文档的创建、搜索、更新及删除操作。一个ES集群中可以按需创建任意数目的索引。

类型(Type)

类型是索引内部的逻辑分区(category/partition)，然而其意义完全取决于用户需求。因此，一个索引内部可定义一个或多个类型(type)。一般来说，类型就是为那些拥有相同的域的文档做的预定义。

举例说：pandora这个索引的type可以只有一个：pandoralog, 也可根据pandora的模块分成多个type, 比如登陆、部署、配置...

例如，在索引中，可以定义一个用于存储用户数据的类型，一个存储日志数据的类型，以及一个存储评论数据的类型。类比传统的关系型数据库领域来说，类型相当于“表”。

文档(Document)

文档是Lucene索引和搜索的原子单位，基于JSON格式进行表示。一个文档就是一个保存在 es 中的 JSON 文本，可以把它理解为关系型数据库表中的一行。每个文档都是保存在索引中的，拥有一种type和 id。

Id 是用于标识文档的，一个文档的索引/类型/id 必须是唯一的。文档 id 是自动生成的（如果不指定）。

field 字段

一个文档包含了若干字段，或称之为键值对。字段的值可以是简单（标量）值（例如字符串、整型、日期），也可以是嵌套结构，例如数组或对象。一个字段类似于关系型数据库表中的一列。每个字段的映射都有一个字段类型（不要和文档类型搞混了），它描述了这个字段可以保存的值类型，例如整型、字符串、对象。映射还可以让我们定义一个字段的值如何进行分析。

映射(Mapping)

ES中，所有的文档在存储之前都要首先进行分析。用户可根据需要定义如何将文本分割成token、哪些token应该被过滤掉，以及哪些文本需要进行额外处理等等。

另外，ES还提供了额外功能，例如将域中的内容按需排序。事实上，ES也能自动根据其值确定域的类型。

总结来说， mapping的作用就是执行一系列的指令将输入的数据转成可搜索的索引项。

详见http://blog.csdn.net/lvhong84/article/details/23936697

接下去再说说ES Cluster相关的一些概念。

集群(Cluster)

ES集群是一个或多个节点的集合，它们共同存储了整个数据集，并提供了联合索引以及可跨所有节点的搜索能力。

多节点组成的集群拥有冗余能力，它可以在一个或几个节点出现故障时保证服务的整体可用性。

集群靠其独有的名称进行标识，默认名称为“elasticsearch”。节点靠其集群名称来决定加入哪个ES集群，一个节点只能属一个集群。

如果不考虑冗余能力等特性，仅有一个节点的ES集群一样可以实现所有的存储及搜索功能。

节点(Node)

运行了单个实例的ES主机称为节点，它是集群的一个成员，可以存储数据、参与集群索引及搜索操作。

类似于集群，节点靠其名称进行标识，默认为启动时自动生成的随机Marvel字符名称。

用户可以按需要自定义任何希望使用的名称，但出于管理的目的，此名称应该尽可能有较好的识别性。

节点通过为其配置的ES集群名称确定其所要加入的集群。

分片(Shard)和副本(Replica)

ES的“分片(shard)”机制可将一个索引内部的数据分布地存储于多个节点，它通过将一个索引切分为多个底层物理的Lucene索引完成索引数据的分割存储功能，这每一个物理的Lucene索引称为一个分片(shard)。

每个分片其内部都是一个全功能且独立的索引，因此可由集群中的任何主机存储。创建索引时，用户可指定其分片的数量，默认数量为5个。

Shard有两种类型：primary和replica，即主shard及副本shard。

Primary shard用于文档存储，每个新的索引会自动创建5个Primary shard，当然此数量可在索引创建之前通过配置自行定义，不过，一旦创建完成，其Primary shard的数量将不可更改。

Replica shard是Primary Shard的副本，用于冗余数据及提高搜索性能。

每个Primary shard默认配置了一个Replica shard，但也可以配置多个，且其数量可动态更改。ES会根据需要自动增加或减少这些Replica shard的数量。

ES集群可由多个节点组成，各Shard分布式地存储于这些节点上。

ES可自动在节点间按需要移动shard，例如增加节点或节点故障时。简而言之，分片实现了集群的分布式存储，而副本实现了其分布式处理及冗余功能。

简单的一个演示加深对 index、doucument、type、id 的理解：

10.6.129.101:9200已搭建通过restApi 与es交互

几个很常用的接口：

/_cat 查看可用命令

/_cat/nodes?v：查集群状态

/_cat/health?v ：查询集群健康状态

/_cat/shards?v：查看分片状态

/${index}/${type}/_search?pretty：搜索

/${index}/_search?pretty&q=xxx=xxx ：带条件搜索

/_search?pretty :查询全部文档 或/_count?pretty

（uri后面加?v 表示对查询结果按类型分列展示；?pretty 表示对结果格式化输出）

查看es 可用索引http://10.6.129.101:9200/_cat/indices?v

索引雇员文档

每个雇员索引一个文档，包含该雇员的所有信息。
每个文档都将是 employee 类型。 type
该类型位于索引 megacorp 内。 index
该索引保存在我们的 Elasticsearch 集群中。

实践中这非常简单（尽管看起来有很多步骤），我们可以通过一条命令完成所有这些动作：

（拷贝下列代码到kibana DevTools模块中运行是一个很方便的方法）

PUT /megacorp/employee/1

{

"first_name" : "John",

"last_name" : "Smith",

"age" : 25,

"about" : "I love to go rock climbing",

"interests": [ "sports", "music" ]

}

注意，路径 /megacorp/employee/1 包含了三部分的信息：

megacorp

索引名称

employee

类型名称

特定雇员的ID

接着 再写入两条

PUT /megacorp/employee/2

    "first_name" :  "Jane",

    "last_name" :   "Smith",

    "age" :         32,

    "about" :       "I like to collect rock albums",

    "interests":  [ "music" ]

PUT /megacorp/employee/3

    "first_name" :  "Douglas",

    "last_name" :   "Fir",

    "age" :         35,

    "about":        "I like to build cabinets",

    "interests":  [ "forestry" ]

查询es所有索引http://10.6.129.101:9200/_cat/indices?v 可以看到新创建的megacorp 索引

检索文档

目前我们已经在 Elasticsearch 中存储了一些数据，接下来就能专注于实现应用的业务需求了。

1）第一个需求是可以检索到单个雇员的数据。

GET /megacorp/employee/1

将 HTTP 命令由 PUT 改为 GET 可以用来检索文档，同样的，可以使用 DELETE 命令来删除文档，以及使用 HEAD 指令来检查文档是否存在。如果想更新已存在的文档，只需再次 PUT

2）我们使用下列请求来搜索所有雇员

GET /megacorp/employee/_search

3）接下来，尝试下搜索姓氏为 ``Smith`` 的雇员。为此，我们将使用一个 高亮 搜索，很容易通过命令行完成。这个方法一般涉及到一个 查询字符串 （_query-string_） 搜索

GET /megacorp/employee/_search?q=last_name:Smith

我们仍然在请求路径中使用 _search 端点，并将查询本身赋值给参数 q= 。返回结果给出了所有的 Smith

4）Query-string 搜索通过命令非常方便地进行临时性的即席搜索，但它有自身的局限性（参见轻量搜索）。Elasticsearch 提供一个丰富灵活的查询语言叫做 查询表达式 ，它支持构建更加复杂和健壮的查询。

领域特定语言 （DSL），指定了使用一个 JSON 请求。我们可以像这样重写之前的查询所有 Smith 的搜索：

GET /megacorp/employee/_search

    "query" : {

        "match" : {

            "last_name" : "Smith"

这个请求使用 JSON 构造，并使用了一个 match 查询

5）更复杂的搜索

现在尝试下更复杂的搜索。同样搜索姓氏为 Smith 的雇员，但这次我们只需要年龄大于 30 的。查询需要稍作调整，使用过滤器 filter ，它支持高效地执行一个结构化查询。

GET /megacorp/employee/_search

    "query" : {

        "bool": {

            "must": {

                "match" : {

                    "last_name" : "smith"

},

            "filter": {

                "range" : {

                    "age" : { "gt" : 30 }

	这部分与我们之前使用的 match 查询一样。
	这部分是一个 range 过滤器，它能找到年龄大于 30 的文档，其中 gt 表示_大于(_great than)。

目前无需太多担心语法问题，后续会更详细地介绍。只需明确我们添加了一个 过滤器 用于执行一个范围查询，并复用之前的 match 查询。现在结果只返回了一个雇员，叫 Jane Smith，32 岁。

6）全文搜索

截止目前的搜索相对都很简单：单个姓名，通过年龄过滤。现在尝试下稍微高级点儿的全文搜索——一项传统数据库确实很难搞定的任务。

搜索下所有喜欢攀岩（rock climbing）的雇员：

GET /megacorp/employee/_search

    "query" : {

        "match" : {

            "about" : "rock climbing"

显然我们依旧使用之前的 match 查询在about 属性上搜索 “rock climbing” 。得到两个匹配的文档：

...

   "hits": {

      "total":      2,

      "max_score":  0.16273327,

      "hits": [

...

            "_score":         0.16273327,

            "_source": {

               "first_name":  "John",

               "last_name":   "Smith",

               "age":         25,

               "about":       "I love to go rock climbing",

               "interests": [ "sports", "music" ]

},

...

            "_score":         0.016878016,

            "_source": {

               "first_name":  "Jane",

               "last_name":   "Smith",

               "age":         32,

               "about":       "I like to collect rock albums",

               "interests": [ "music" ]

教程结语

这是一个关于 Elasticsearch 基础描述的教程，且仅仅是浅尝辄止，更多诸如 suggestions、geolocation、percolation、fuzzy 与 partial matching 等特性均被省略，以便保持教程的简洁。但它确实突显了开始构建高级搜索功能多么容易。不需要配置——只需要添加数据并开始搜索！

删除索引

DELETE megacorp

其他的一些操作命令（

按时间查询日志（国际时间）

Try a test query to Elasticsearch based on the fields created by the grok filter plugin. Replace $DATE with the current date, in YYYY.MM.DD format:

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

例如 http://10.6.129.101:9200/logstash-2017.08.12/_search?pretty&q=host=10.33.42.212

其中 host=xxx可替换为host:xxx

按城市名称查询

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

）

Logstash 介绍

Logstash 是一款强大的数据处理工具，它可以实现数据传输，格式处理，格式化输出，还有强大的插件功能，常用于日志处理（运行环境jvm 8）

官网地址:https://www.elastic.co/products/logstash

官方文档：https://www.elastic.co/guide/en/logstash

Logstash管道有两个必需的元素，input、 output，以及一个可选的元素，filter。输入插件消费源数据，过滤器插件会按照你指定的方式修改数据，输出插件将数据写入到目的地(查看logstash安装了哪些插件./bin/logstash-plugin list)。配置模板

input {{

}}

filter { filter {

}}

output {{

}}

示例 参见虚拟机10.6.129.101配置示例。

工作流程

Logstash 工作的三个阶段：

input 数据输入端，可以接收来自任何地方的源数据。

file：从文件中读取
syslog：监听在514端口的系统日志信息，并解析成RFC3164格式。
redis：从redis-server list 中获取
beat：接收来自Filebeat的事件

Filter 数据中转层，主要进行格式处理，数据类型转换、数据过滤、字段添加，修改等，常用的过滤器如下。

grok: 通过正则解析和结构化任何文本。Grok 目前是logstash最好的方式对非结构化日志数据解析成结构化和可查询化。logstash内置了120个匹配模式，满足大部分需求。
mutate: 在事件字段执行一般的转换。可以重命名、删除、替换和修改事件字段。
drop: 完全丢弃事件，如debug事件。
clone: 复制事件，可能添加或者删除字段。
geoip: 添加有关IP地址地理位置信息。

output 是logstash工作的最后一个阶段，负责将数据输出到指定位置，兼容大多数应用，常用的有:

elasticsearch: 发送事件数据到 Elasticsearch，便于查询，分析，绘图。
file: 将事件数据写入到磁盘文件上。
mongodb:将事件数据发送至高性能NoSQL mongodb，便于永久存储，查询，分析，大数据分片。
redis:将数据发送至redis-server，常用于中间层暂时缓存。
graphite: 发送事件数据到graphite，用于存储和绘制指标的流行开源工具。http://graphite.wikidot.com/
statsd: 发送事件数据到 statsd。

Codecs编解码器

编解码器可以作为输入的流过滤器。编解码器使您可以轻松地将邮件的传输与序列化过程分开。流行的编解码器包括json，msgpack和plain （文本）。

简单的两种测试输入输出的配置

测试您的Logstash，运行最基本的Logstash管道。

1）例如：shell标准输入数据

cd logstash-5.5.1

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

2）或 端口输入，stdout输出

input {

    beats {

        port => "5043"

output {

    stdout { codec => rubydebug }

验证配置文件是否正确和配置自动加载

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

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

该--config.reload.automatic选项启用自动配置重新加载，以便您每次修改配置文件时不必停止并重新启动Logstash。

Grok 结构化数据

过滤器插件使您能够将非结构化日志数据进行结构化参见演示平台logstash 配置

filter {

    grok {

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

删除注册文件，让filebeat 重新读取日志

sudo rm data/registry

geoip 补充ip信息需下载geoip数据库，或机器能连网

filter {

    grok {

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

    geoip {

        source => "clientip"

logstash配置文件基础语法:

https://www.elastic.co/guide/en/logstash/5.5/event-dependent-configuration.html

Grok 使用语法

https://www.elastic.co/guide/en/logstash/5.5/plugins-filters-grok.html

解析任意文本和结构。

Grok目前是logstash中将最坏的非结构化日志数据解析为结构化和可查询的最佳方式。

此工具非常适用于syslog日志，apache和其他Web服务器日志，mysql日志以及通常为人类而不是计算机消费而编写的任何日志格式。

默认情况下，Logstash约有120种模式。你可以在这里找到他们：https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns。你可以自己加入。（见patterns_dir设定）

如果您需要帮助构建模式以匹配您的日志，您会发现 http://grokdebug.herokuapp.com和http://grokconstructor.appspot.com/应用程序非常有用！

基本演示：访问http://grokdebug.herokuapp.com/

日志：

55.3.244.1 GET /index.xml 15824 0.043

pattern:

%{IPV4:client} %{WORD:method} %{URIPATH:uri} %{INT:datalength} %{BASE16FLOAT:timecost}

输出：

写成配置：

input {

file {

path => "/var/log/http.log"

}

filter {

grok {

match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }

remove_field => [ "message" ]

}

filebeat 配置 fields type

multiline：将多行文本事件（如java异常和堆栈跟踪）消息合并到一个事件中。

filebeat.prospectors:

- input_type: log

  paths:

    - /var/log/*.log

  fields:

    type: syslog

output.logstash:

  hosts: ["localhost:5043"]

logstash 配置 input tritter ,out put file，其中output：elasticsearch 可以填写集群多个节点，自动负载均衡。端口可以不填，默认9200

input {

    twitter {

        consumer_key => "enter_your_consumer_key_here"

        consumer_secret => "enter_your_secret_here"

        keywords => ["cloud"]

        oauth_token => "enter_your_access_token_here"

        oauth_token_secret => "enter_your_access_token_secret_here"

    beats {

        port => "5043"

file {

path => “/var/log/ apache/ access.log”

type => “apache” }

output {

    elasticsearch {

        hosts => ["IP Address 1:port1", "IP Address 2:port2", "IP Address 3"]

    file {

        path => "/path/to/target/file"

elsaticsearch 查询 上面配置的filed type :

curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=fields.type:syslog'

查询推特信息

curl -XGET 'http://localhost:9200/logstash-$DATE/_search?pretty&q=client:iphone'

后台运行logstash

logstash 启停 /etc/init.d/logstash start

# 通过这种方式启动，logstash会自动加载 /etc/logstash/conf.d/ 下的配置文件

或后台命令行运行：

nohup /usr/share/logstash/bin/logstash -f /usr/share/logstash/bin/logstash.conf &>/dev/null &

哈希

哈希是以格式指定的键值对的集合"field1" => "value1"。请注意，多个键值条目由空格分隔，而不是逗号。

例：

match => { “field1” => “value1” “field2” => “value2” ... }

output  -elasticsearch

https://www.elastic.co/guide/en/logstash/5.5/plugins-outputs-elasticsearch.html

`index`

值类型是字符串
默认值为 "logstash-%{+YYYY.MM.dd}"
值类型是字符串
此设置没有默认值。

`id`

添加一个独特ID的插件配置。如果没有指定ID，Logstash将生成一个。强烈建议您在配置中设置此ID。当您有两个或多个相同类型的插件时，例如，如果您有2个Grok过滤器，则此功能特别有用。在这种情况下添加命名ID将有助于在使用监视API时监视Logstash。

输出{

 stdout {

   id =>“my_plugin_id”

nginx 实战比较全面的一个示例

参考文档http://tchuairen.blog.51cto.com/3848118/1840596/

1）注意事项：

其中nginx 日志改json 格式的方法： 在http 域内 定义名为json的log_format 格式，然后指定日志文件access.log使用该格式：

access_log  logs/access.log  json;

最后重新加载，或重启nginx 以生效

如下：

http {

    log_format json '{"@timestamp":"$time_iso8601",'

                 '"slbip":"$remote_addr",'

                 '"clientip":"$http_x_forwarded_for",'

                 '"serverip":"$server_addr",'

                 '"size":$body_bytes_sent,'

                 '"responsetime":$request_time,'

                 '"domain":"$host",'

                 '"method":"$request_method",'

                 '"requesturi":"$request_uri",'

                 '"url":"$uri",'

                 '"appversion":"$HTTP_APP_VERSION",'

                 '"referer":"$http_referer",'

                 '"agent":"$http_user_agent",'

                 '"status":"$status"}';

access_log  logs/access.log  json;

- input_type: log

  paths:

    - /usr/local/openresty/nginx/logs/access.log

  fields:

    type: nginxacclog

  fields_under_root: true

2）grok匹配pandora 日志示例：

.2017-08-13 12:42:25.112 MsgProcessThread-4 org.springframework.web.servlet.mvc.method INFO - ping from [/10.33.25.215:41542]

%{TIMESTAMP_ISO8601:time} %{USERNAME :thread} %{USERNAME:classpath} %{LOGLEVEL:loglevel} - %{GREEDYDATA:data}

logstash if else 语法

if EXPRESSION {

...

} else if EXPRESSION {

...

} else {

...

}

kibana 基本操作

Discover 对日志进行检索查看

Visuallize 图表可视化配置；对特定规则聚合的数据进行图表分析

Dashboard 将Visualize设置的图表展示成仪表盘，有分组功能

Timelion 默认显示的.es(*) 为 es 文档个数

Dev Tools 一个RestAPI Console 与elasticsearch 进行交互

Management 创建管理索引的地方以及其他的高级设置

ELK日志分析 学习笔记

ELK日志分析系统 学习笔记