最常见的日志收集架构（ELK Stack）

> 本文由 [简悦 SimpRead](http://ksria.com/simpread/) 转码， 原文地址 [www.tuicool.com](https://www.tuicool.com/articles/JbmuErv)

![](https://img2.tuicool.com/n6vymuJ.jpg!web)

日志存在于不同的机器不同的目录，所以首先机器上面要有收集日志的 Agent，这些 Agent 被生动的叫做：Shippers（直译：发货商），好理解，将日志像货物一样发送出去。

我第一次听到 ELK 的时候，以为它就是 Elasticsearch 的缩写，谁知道它代表了三个组件：

*   E - Elasticsearch（简称：ES）
*   L - Logstash
*   K - Kibana

我理解也不是很深刻，Logstash 主要用来收集各个 Shipper 发过来的日志，并做一个过滤，然后发给 Elasticsearch 保存，Elasticsearch 保存日志，建索引，然后提供接口，Kibana 作为前端调接口提供可配置的可视化。

这就是从机器上的日志，到可视化一个过程，蛮清晰的。

日志数据可以这样：

Filebeat（Shipper） -> Elasticsearch

索引（动词）之前如果要做过滤拆分字段就加个 Logstash：

Filebeat（Shipper） -> Logstash -> Elasticsearch

如果 Agent 量比较大，中间再加一个消息队列：

Filebeat（Shipper） -> 消息队列 -> Logstash -> Elasticsearch

这一套搭建下来很容易，遇到的问题我列一下，希望对你也有帮助：

**问：如何保证 ES 中的数据有序？**

答：Logstash 的 filter 流程中的 date 插件可以解析业务时间并替换 `@timestamp` 字段，ES 默认是按照 `@timestamp` 字段排序的。

```
filter {
  if [beat][name] == "xxx-backend" {
 grok {
 match => { "message" => "%{GREEDYDATA:time} \[%{DATA:log_level}\] \[%{DATA:golang}\] %{GREEDYDATA:message}" }
 overwrite => [ "message" ]
 }
 date {
 match => ["time", "yyyy/MM/dd HH:mm:ss.SSS"]
 }
  }
}
```

判断是不是某个业务，如果是，那我们知道格式，然后使用 grok 插件，解析并拆分字段，其中的 time 字段给 date 插件解析并替换 `@timestamp` 。

**问：Kafka 中的数据格式是什么样的，如何消费？**

答：Filebeat 打到 kafka 的日志信息是 json 格式的，Logstash 消费的时候可以使用 filter 流程中的 json 插件来解析。

```
filter {
  if [type] == 'from-kafka' {
    json {
      source => "message"
    }
  }
}
```

**问：如何区分不同的数据来源？**

答：input 块中，type 字段标明即可。

```
input {
  beats {
    port => 5044
    type => "from-beat"
  }
  kafka {
    bootstrap_servers => "kafkahost1:9092,kafkahost2:9092,kafkahost3:9092"
    topics => ["pdd"]
    type => "from-kafka"
  }
}
```

**问：如何标记不同的业务日志？**

答：使用 Shipper 的 name 或者 tag 来标记。

```
name: xxx-backend
filebeat.inputs:
- type: log
 enabled: true
 paths:
 - /data/xxx-logs/*.log
 tags: ["正义之剑"]
```

**问：如何在 ES 中为不同的业务创建不同的 index？**

答：上面我们为 Shipper 起了名字，我们可以用名字作为索引的一部分。

```
output {
  elasticsearch {
    hosts => "http://EShost:9200"
    index => "xxx-%{[beat][name]}-%{+YYYY.MM.dd}"
    user => "xxx"
    password => "yyy"
  }
}
```

**问：如何在 Logstash 层配置多个业务的 filter？**

答：和 nginx 的配置类似，有个 conf.d 的文件夹，配置都放在下面，然后开启自动 reload 即可。

*   在 logstash.yml 配置文件中启用：

```
config.reload.automatic: true
config.reload.interval: 10s
```

*   观察 pipelines.yml 中配置：

```
- pipeline.id: main
  path.config: "/etc/logstash/conf.d/*.conf"
```

只启用一个管道，配置 watch `path.config` 目录 。

先写到这，遇到解决了再写，大家有问题也可以问，一起探讨。