使用Filebeat+Kafka+Logstash+Elasticsearch构建日志分析系统

操作流程

1. 准备工作
   完成环境准备，包括创建实例、安装Filebeat等。

   注意 安装Filebeat的ECS、阿里云ES、阿里云Logstash、阿里云消息队列Kafka必须在同一VPC下。
2. 步骤一：配置Filebeat
   配置Filebeat的input为系统日志，output为Kafka，将日志数据采集到Kafka的指定Topic中。
3. 步骤二：配置Logstash管道
   配置Logstash管道的input为Kafka，output为阿里云ES，使用Logstash消费Topic中的数据并传输到阿里云ES中。
4. 步骤三：查看日志消费状态
   在消息队列Kafka中查看日志数据的消费的状态，验证日志数据是否采集成功。
5. 步骤四：通过Kibana过滤日志数据
   在Kibana控制台的Discover页面，通过Filter过滤出Kafka相关的日志。

准备工作

1. 创建阿里云ES实例，并开启实例的自动创建索引功能。
   具体操作步骤请参见创建阿里云Elasticsearch实例和配置YML参数，本文以6.7版本为例。
2. 创建阿里云Logstash实例。要求该实例与阿里云ES实例的版本相同，并且在同一专有网络VPC（Virtual Private Cloud）下。
   具体操作步骤请参见创建阿里云Logstash实例。
3. 购买并部署阿里云消息队列Kafka版实例、创建Topic和Consumer Group。
   本文使用VPC实例，并要求该实例与阿里云ES实例在同一VPC下，具体操作步骤请参见VPC接入。

   创建Topic和Consumer Group的具体步骤请参见步骤三：创建资源。

4. 创建阿里云ECS实例，并且该ECS实例与阿里云ES实例和Logstash实例处于同一VPC下。

   具体操作步骤请参见使用向导创建实例。

   注意 该ECS实例用来安装Filebeat，由于Filebeat目前仅支持Aliyun Linux、RedHat和CentOS这三种操作系统，因此在创建时请选择其中一种操作系统。
5. 在ECS实例上安装Filebeat。
   具体操作步骤请参见Install Filebeat。本文使用6.8.5版本。

步骤一：配置Filebeat

1. 连接安装了Filebeat的ECS服务器。
   具体操作步骤请参见连接实例。
2. 进入Filebeat安装目录，执行以下命令，创建并配置filebeat.kafka.yml文件。

   cd filebeat-6.8.5-linux-x86_64
   vi filebeat.kafka.yml

   filebeat.kafka.yml配置如下。

   filebeat.prospectors:
     - type: log
       enabled: true
       paths:
           - /var/log/*.log
   
   output.kafka:
       hosts: ["172.16.**.**:9092","172.16.**.**:9092","172.16.**.**:9092"]
       topic: estest
       version: 0.10.2

   参数	说明
   type	输入类型。设置为log，表示输入源为日志。
   enabled	设置配置是否生效。true表示生效，false表示不生效。
   paths	需要监控的日志文件的路径。多个日志可在当前路径下另起一行写入日志文件路径。
   hosts	消息队列Kafka实例的接入点，可在实例详情页面获取，详情请参见查看接入点。由于本文使用的是VPC实例，因此使用默认接入点。
   topic	日志输出到消息队列Kafka的Topic，请指定为您已创建的Topic。
   version	Kafka的版本，可在消息队列Kafka的实例详情页面获取。 注意 不配置此参数会报错。

3. 启动Filebeat。

   ./filebeat -e -c filebeat.kafka.yml

步骤二：配置Logstash管道

1. 登录阿里云Elasticsearch控制台。
2. 进入目标实例。
   1. 在顶部菜单栏处，选择地域。
   2. 在左侧导航栏，单击Logstash实例，然后在Logstash实例中单击目标实例ID。
3. 在左侧导航栏，单击管道管理。
4. 在管道列表区域，单击创建管道。
   
5. 在创建管道任务页面，输入管道ID并配置管道。
   本文使用的管道配置如下。

   input {
     kafka {
       bootstrap_servers => ["172.**.**.92:9092,172.16.**.**:9092,172.16.**.**:9092"]
       group_id => "es-test"
       topics => ["estest"] 
       codec => json
    }
   }
   filter {
   
   }
   output {
     elasticsearch {
       hosts => "http://es-cn-n6w1o1x0w001c****.elasticsearch.aliyuncs.com:9200"
       user =>"elastic"
       password =>"<your_password>"
       index => "kafka‐%{+YYYY.MM.dd}"
    }
   }

   表 1. input参数说明

   参数	说明
   bootstrap_servers	消息队列Kafka实例的接入点，可在实例详情页面获取，详情请参见查看接入点。由于本文使用的是VPC实例，因此使用默认接入点。
   group_id	指定为您已创建的Consumer Group的名称。
   topics	指定为您已创建的Topic的名称，需要与Filebeat中配置的Topic名称保持一致。
   codec	设置为json，表示解析JSON格式的字段，便于在Kibana中分析。

   表 2. output参数说明

   参数	说明
   hosts	阿里云ES的访问地址，取值为http://<阿里云ES实例的内网地址>:9200。 说明 您可在阿里云ES实例的基本信息页面获取其内网地址，详情请参见查看实例的基本信息。
   user	访问阿里云ES的用户名，默认为elastic。您也可以使用自建用户，详情请参见创建用户。
   password	访问阿里云ES的密码，在创建实例时设置。如果忘记密码，可进行重置，重置密码的注意事项及操作步骤请参见重置实例访问密码。
   index	索引名称。设置为kafka‐%{+YYYY.MM.dd}表示索引名称以kafka为前缀，以日期为后缀，例如kafka-2020.05.27。

   更多Config配置详情请参见Logstash配置文件说明。

6. 单击下一步，配置管道参数。
   

   表 3. 管道配置参数说明

   参数	说明
   管道工作线程	并行执行管道的Filter和Output的工作线程数量。当事件出现积压或CPU未饱和时，请考虑增大线程数，更好地使用CPU处理能力。默认值：实例的CPU核数。
   管道批大小	单个工作线程在尝试执行Filter和Output前，可以从Input收集的最大事件数目。较大的管道批大小可能会带来较大的内存开销。您可以设置LS_HEAP_SIZE变量，来增大JVM堆大小，从而有效使用该值。默认值：125。
   管道批延迟	创建管道事件批时，将过小的批分派给管道工作线程之前，要等候每个事件的时长，单位为毫秒。默认值：50ms。
   队列类型	用于事件缓冲的内部排队模型。可选值： MEMORY：默认值。基于内存的传统队列。 PERSISTED：基于磁盘的ACKed队列（持久队列）。
   队列最大字节数	请确保该值小于您的磁盘总容量。默认值：1024 MB。
   队列检查点写入数	启用持久性队列时，在强制执行检查点之前已写入事件的最大数目。设置为0，表示无限制。默认值：1024。

   警告 配置完成后，需要保存并部署才能生效。保存并部署操作会触发实例重启，请在不影响业务的前提下，继续执行以下步骤。
7. 单击保存或者保存并部署。
   • 保存：将管道信息保存在Logstash里并触发实例变更，配置不会生效。保存后，系统会返回管道管理页面。可在管道列表区域，单击操作列下的立即部署，触发实例重启，使配置生效。
   • 保存并部署：保存并且部署后，会触发实例重启，使配置生效。

步骤三：查看日志消费状态

1. 进入消息队列Kafka控制台。
2. 在左侧导航栏，单击Consumer Group管理。
3. 在Consumer Group管理页面，单击目标消息队列Kafka实例。
4. 单击对应Consumer Group右侧操作列下的消费状态。
   
5. 在消费状态对话框中，单击对应Topic右侧操作列下的详情，查看详细消费状态。
   正常情况下，返回结果如下。

步骤四：通过Kibana过滤日志数据

1. 登录目标阿里云ES实例的Kibana控制台。
   具体步骤请参见登录Kibana控制台。
2. 创建一个索引模式。
   1. 在左侧导航栏，单击Management。
   2. 在Kibana区域，单击Index Patterns。
   3. 单击Create index pattern。
   4. 输入Index pattern（本文使用kafka-*），单击Next step。
      
   5. 选择Time Filter field name（本文选择@timestamp），单击Create index pattern。
      
3. 在左侧导航栏，单击Discover。
4. 从页面左侧的下拉列表中，选择您已创建的索引模式（kafka-*）。
5. 在页面右上角，选择一段时间，查看对应时间段内的Filebeat采集的日志数据。
   
6. 单击Add a filter，在Add filter页面中设置过滤条件，查看符合对应过滤条件的日志数据。
   

常见问题

Q：同步日志数据出现问题，管道一直在生效中，无法将数据导入Elasticsearch，如何解决？

更多问题原因及解决方法，请参见管道创建后，进程卡住了，实例变更进度不变，如何处理？。