ELK学习笔记（三）——使用Filebeat8.15.0收集日志

使用Filebeat收集日志

前面教程已经把ElasticSearch和Kibana部署完毕，接着我们就要使用filebeat去收集我们的java服务日志，这里首先介绍一下ELK和EFK的区别。

一、ELK和EFK的区别

在收集和处理日志时，使用 ELK（Elasticsearch, Logstash, Kibana）和 EFK（Elasticsearch, Filebeat, Kibana）堆栈都有其独特的优势和应用场景。两者之间的主要区别在于日志收集和传输组件的选择：Logstash vs. Filebeat。下面是两者的比较和使用 Filebeat 的一些优势：

ELK（Elasticsearch, Logstash, Kibana）的优势

强大的数据处理能力：
Logstash 拥有丰富的数据处理和过滤插件，能够处理复杂的数据转换和增强，例如解析复杂的日志格式、添加字段、重命名字段、地理定位等。Logstash 支持高级的数据管道操作，如聚合、拆分、条件过滤等。 支持更多的数据源：
Logstash 可以从各种数据源收集数据，不仅仅是日志文件，还包括数据库、消息队列、HTTP 端点等。对于需要集成来自多个不同数据源的数据环境，Logstash 提供了更灵活的选择。 灵活的插件系统：
Logstash 提供了一个插件系统，可以轻松地添加输入、过滤和输出插件，以适应不同的应用场景。

EFK（Elasticsearch, Filebeat, Kibana）的优势

轻量级和高效：
Filebeat 是一个轻量级的日志收集器，专门设计用于收集和转发日志。它消耗的资源比 Logstash 少，因此在性能和效率上更优，尤其适合在资源受限的环境中使用（例如 Kubernetes 环境中的容器）。Filebeat 更容易配置和部署，不需要像 Logstash 那样处理复杂的配置文件和插件管理。 简单易用：
Filebeat 的配置文件相对简单，易于维护和管理。对于基本的日志收集场景，配置成本较低。Filebeat 有内置的模块（例如 system，nginx，apache 等），这些模块已经预先配置好收集和解析某些常见服务的日志，减少了手动配置的工作量。 更适合 Kubernetes 环境：
Filebeat 与 Kubernetes 原生集成得很好，支持自动发现容器日志，可以轻松将每个容器的日志收集并发送到 Elasticsearch。它能自动添加 Kubernetes 元数据（例如 pod 名称、命名空间、标签等），方便在 Kibana 中进行基于 Kubernetes 元数据的日志搜索和分析。在 Kubernetes 环境中，直接部署 Filebeat 作为 DaemonSet 可以高效收集集群中所有节点的日志。 可靠的数据传输：
Filebeat 有内置的背压机制和可靠的传输协议（如使用 ACK 确认机制的 Elasticsearch 输出），确保日志在传输过程中不会丢失。

什么时候选择 EFK（Filebeat）

资源受限的环境：如 Kubernetes 集群或需要在多台机器上部署日志采集器的环境。主要是日志收集，而不需要复杂的日志处理：如果你只需要将日志采集并发送到 Elasticsearch。希望减少复杂性和管理开销：Filebeat 更容易部署和维护。

什么时候选择 ELK（Logstash）

需要复杂的日志处理和增强：如正则表达式解析、数据过滤、条件处理、多步骤处理等。日志源种类繁多：不仅是日志文件，还有其他数据源，如数据库、消息队列等。系统资源不是主要问题：在高性能服务器或数据中心环境中部署。

总之，Filebeat 在轻量、简洁、高效的场景下表现更好，而 Logstash 在需要复杂日志处理的场景下更为适用。根据你的需求选择适合的工具可以显著提高日志收集系统的效率和性能。

前面，我们已经在k8s环境中部署好了ElasticSearch和Kibana。并且我们只是简单的收集一下后端服务产生的打印日志，所以这里我选择使用filebeat去收集日志。

二、部署Filebeat服务

2.1、创建目录

mkdir -p /home/ec2-user/k8s/elk/filebeat
cd /home/ec2-user/k8s/elk/filebeat

12

2.2、下载filebeat-8.15.0压缩文件

wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.15.0-linux-x86_64.tar.gz

1

2.3、解压

tar -zxvf filebeat-8.15.0-linux-x86_64.tar.gz
#重命名
mv filebeat-8.15.0-linux-x86_64 filebeat-8.15.0


1234

2.4、配置filebeat.yml

vim filebeat.yml
# ============================== Filebeat inputs ===============================

filebeat.inputs:

- type: log

  # Unique ID among all inputs, an ID is required.
  id: new-erp-filestream-id

  # Change to true to enable this input configuration.
  enabled: true
  encoding: utf-8
  paths:
    # 配置要收集日志的目录
    - /home/ec2-user/k8s/nfs/data/share/renpho-erp-dev-gateway-logs-data-erp-gateway-dev-*/gateway-logs/*.log
    - /home/ec2-user/k8s/nfs/data/share/renpho-erp-dev-smc-log-data-erp-smc-dev-*/smc-logs/*.log
    - /home/ec2-user/k8s/nfs/data/share/renpho-erp-dev-uac-log-data-erp-uac-dev-*/uac-logs/*.log

  #匹配日志规则：日志文件中以如下规则开头的则为一行
  multiline.pattern: '^d{4}-d{2}-d{2} d{2}:d{2}:d{2}'
  multiline.negate: true
  multiline.match: after
  #排除以.gz结尾的压缩包文件
  prospector.scanner.exclude_files: ['.gz$']
  fields:
    tag: erp-log
    #
    type: erp-logdata-pipeline 
    source: common

# ============================== Filebeat modules ==============================

filebeat.config.modules:
  # Glob pattern for configuration loading
  path: ${path.config}/modules.d/*.yml

  # Set to true to enable config reloading
  reload.enabled: false

# ======================= Elasticsearch template setting =======================

setup.template.settings:
  index.number_of_shards: 1
setup.template.name: "erp-log"
setup.template.pattern: "erp-log-*"

# =================================== Kibana ===================================
setup.kibana:
  host: "https://192.168.6.220:30091"
  #可以到kibana控制台，创建一个账号
  username: "kibanaerp"
  password: "replace your password"


# ================================== Outputs ===================================

output.elasticsearch:
  # Array of hosts to connect to.
  hosts: ["https://192.168.6.220:30092"]
  protocol: "https"
  ssl.verification_mode: "none"
  # ssl.certificate_authorities: ["/home/ec2-user/k8s/elk/kibana/certs/elasticsearch-ca.pem"]
  # Authentication credentials - either API key or username/password.
  #api_key: "id:api_key"
  username: "elastic"
  password: "replace your password"
  #数据流
  index: "erp-log-%{+yyyy.MM.dd}"
  #索引
  indices:
   - index: "erp-logindex-%{+yyyy.MM.dd}"
  pipelines:
   #定义一个管道，用于解析自定义日志格式，稍后会在下面kibana中用到
   - pipeline: erp-logdata-pipeline

logging.level: debug


processors:
  # 将自带的一些字段可以去除
  # - add_host_metadata:
  #     when.not.contains.tags: forwarded
  # - add_cloud_metadata: ~
  # - add_docker_metadata: ~
  # - add_kubernetes_metadata: ~
  # 在这里可以设置要去除的字段
  - drop_fields:
     # when: 可以设置去除的条件
     #   condition
     fields: ["log","host","input","agent","ecs"]
     ignore_missing: false


1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192

2.5、Kibana后台创建管道pipeline

使用管理员账号elastic，登录到Kibana后台，

进到采集管道页面

创建新管道

按顺序依次点击下图按钮

添加处理器，并保存。

上图中第9步的值为

%{TIMESTAMP_ISO8601:log_date} %{DATA:springAppName} [%{DATA:tlog_trace_id}] %{LOGLEVEL:level} [%{DATA:thread_name}] [%{DATA:x_b3_span_id}] %{DATA:class} – %{GREEDYDATA:content}

我的项目中logback.xml中配置的的日志输出格式如下图所示，

<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60 seconds" debug="false">
    <include resource="org/springframework/boot/logging/logback/defaults.xml" />
    <include resource="org/springframework/boot/logging/logback/console-appender.xml" />
    <conversionRule conversionWord="clr" converterClass="org.springframework.boot.logging.logback.ColorConverter" />

    <!-- 配合apollo app.id, 因为log日志启动先于 apollo的启动，所以用这样的方式进行命名区别 -->
    <springProperty name="springAppName" scope="context" source="spring.application.name" />
    <!-- 日志存放路径 -->
    <property name="logPath" value="/home/erp/logs/${springAppName}"/>
    <!-- 日志输出格式 -->
    <property name="LogPatternConsole" value="%d{yyyy-MM-dd HH:mm:ss.SSS} %clr(${springAppName}) %clr(${PID:- }){magenta} %clr(-%5p [%X{tlogTraceId:-},%X{X-B3-SpanId:-}]) %clr(-){faint} %clr([%t]){faint} %clr(%logger){cyan} %clr(:){faint} %msg%n" />
    <property name="LogPattern" value="%d{yyyy-MM-dd HH:mm:ss.SSS} ${springAppName} [%X{tlogTraceId:-}] %level{5} [%thread] [%X{X-B3-SpanId:-}] %logger{20} - %msg%n"/>

    <property name="logfile" value="${LOG_FILE:-${logPath}/info.log}" />
    <property name="errorLogfile" value="${LOG_FILE:-${logPath}/error.log}" />


    <logger name="org.apache.catalina.startup.DigesterFactory" level="ERROR" />
    <logger name="org.apache.sshd.common.util.SecurityUtils" level="WARN" />
    <logger name="org.apache.tomcat.util.net.NioSelectorPool" level="WARN" />
    <logger name="org.crsh.plugin" level="WARN" />
    <logger name="org.crsh.ssh" level="WARN" />
    <logger name="com.trace.ssm" level="INFO" />
    <!-- 系统模块日志级别控制  -->
    <logger name="com.renpho" level="info"/>
    <!-- Spring日志级别控制  -->
    <logger name="org.springframework" level="warn"/>

    <appender name="Console" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>${LogPatternConsole}</pattern>
            <charset>utf8</charset>
        </encoder>
    </appender>

    <appender name="InfoLogFile" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <encoder>
            <pattern>${LogPattern}</pattern>
            <charset>utf8</charset>
        </encoder>
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <!-- 过滤的级别 -->
            <level>INFO</level>
            <!-- 匹配时的操作：接收（记录） -->
            <onMatch>ACCEPT</onMatch>
            <!-- 不匹配时的操作：拒绝（不记录） -->
            <onMismatch>DENY</onMismatch>
        </filter>
        <file>${logfile}</file>
        <!-- 循环政策：基于时间创建日志文件 -->
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${logPath}/info.%d{yyyy-MM-dd}.%i.log.gz</fileNamePattern>
            <!-- 日志最大的历史 30天 -->
            <maxHistory>30</maxHistory>
            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <!-- maxFileSize:这是当前活动日志文件的大小，自动分割日志，默认值是10MB，测试时可改成1KB看效果 -->
                <maxFileSize>50MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
        </rollingPolicy>
    </appender>
    <!-- error的日志 -->
    <appender name="ErrorLogFile" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <encoder>
            <pattern>${LogPattern}</pattern>
            <charset>utf8</charset>
        </encoder>
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <!-- 过滤的级别 -->
            <level>ERROR</level>
            <!-- 匹配时的操作：接收（记录） -->
            <onMatch>ACCEPT</onMatch>
            <!-- 不匹配时的操作：拒绝（不记录） -->
            <onMismatch>DENY</onMismatch>
        </filter>
        <file>${errorLogfile}</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${logPath}/error.%d{yyyy-MM-dd}.%i.log.gz</fileNamePattern>
            <maxHistory>30</maxHistory>
            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <maxFileSize>50MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
        </rollingPolicy>
    </appender>

    <appender name="AsyncRollingLogFile" class="ch.qos.logback.classic.AsyncAppender">
        <!-- 异步日志 https://blog.csdn.net/u011943534/article/details/119427147 -->
        <discardingThreshold>0</discardingThreshold>
        <queueSize>12345</queueSize>
        <appender-ref ref="InfoLogFile" />
    </appender>
    <appender name="AsyncErrorLogFile" class="ch.qos.logback.classic.AsyncAppender">
        <!-- 本文设置为0说明永远不会丢弃日志level -->
        <discardingThreshold>0</discardingThreshold>
        <queueSize>12345</queueSize>
        <appender-ref ref="ErrorLogFile" />
    </appender>

    <root level="INFO">
        <appender-ref ref="Console" />
        <appender-ref ref="AsyncRollingLogFile" />
        <appender-ref ref="AsyncErrorLogFile" />
    </root>

    <jmxConfigurator />
</configuration>


123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106

三、启动Filebeat

cd /home/ec2-user/k8s/elk/filebeat-8.15.0
#启动filebeat
sudo nohup ./filebeat -e -c filebeat.yml > /dev/null 2>&1 &

123

四、查看索引

至此，filebeat部署成功！