日志存在于不同的机器不同的目录，所以首先机器上面要有收集日志的 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 目录 。