ELK+FileBeat+Log4Net搭建日志系统

共计 5501 个字符，预计需要花费 14 分钟才能阅读完成。

项目中之前都是采用数据库来记录日志，虽然记录还算挺方便，但是每次都要到数据库来查询，如果日志在单独的数据库还好，只是有点麻烦。如果记录的日志数据库和生产正式库在一起，不仅会影响生产库的正常使用，也会带来安全隐患。
项目早期没有统一规划，也是时间仓促，没有做好日志的规划，所有日志都记录到数据库中。的确也遇到了性能问题，因此在了解 ELK 的基础上，使用其作为日志采集、处理和检索的几本框架。

大体框架

日志数据流如下，应用将日志落地在本地文件，部署在每台服务器上的 FileBeat 负责收集日志，然后将日志发送给 LogStash；LogStash 将日志进行处理之后，比如 parse 等；然后将处理后的 Json 对象传递给 ElasticSearch，进行落地并进行索引处理；最后通过 Kibana 来提供 web 界面，来查看日志等。因为 ES 是基于 Lucene 的，所以 Kibana 支持 Lucene 查询语法。

对于日志数据流特别大的情况，LogStash 会造成拥堵，这个时候可以使用消息队列来进行缓冲。同时，日志一旦进过 LogStash 之后，会不方面一些流处理程序来读取。这个时候使用 kafka 就比较好了，因为 kafka 是将消息持久化在本地，流处理应用可以从消息的 offset 初始的地方来读取。加入 kafka 的后的流程如下：

配置过程

Log4Net 配置

首先，最基本的引用 Log4Net 程序集，补多少。
其次，要在项目的 AssemblyInfo.cs 添加如下代码, 这样配置才能给启作用。
[assembly: log4net.Config.XmlConfigurator(Watch = true)]
最后，也是最重要的就是在 web.config（或 app.config）中配置了。
在 configSections 中添加如下代码
<configSections> <section name="log4net" type="log4net.Config.Log4NetConfigurationSectionHandler, log4net" /> </configSections>
添加 log4net 节点

<log4net>
    <root>
      <level value="ALL"/>
      <appender-ref ref="rollingFile"/>
      <appender-ref ref="ErrorRollingFile"/>
    </root>
    <appender name="rollingFile" type="log4net.Appender.RollingFileAppender,log4net">
      <filter type="log4net.Filter.LevelRangeFilter">
        <levelMin value="DEBUG"/>
        <levelMax value="WARN"/>
      </filter>
      <!-- 日志的路径 -->
      <param name="File" type="" value="D://WebLog//Log4NetTest.App//"/>
       <param name="Encoding" value="UTF-8"/>
      <!-- 是否覆盖，默认是追加 true-->
      <param name="AppendToFile" value="true"/>
      <param name="RollingStyle" value="Date"/>
      <!-- 文件名称 -->
      <param name="DatePattern" value="yyyy-MM-dd'.Debug.log'"/>
      <!-- 设置无限备份 =-1，最大备份数为 1000-->
      <param name="MaxSizeRollBackups" value="1000"/>
      <!-- 每个文件的大小 -->
      <param name="MaximumFileSize" value="102KB"/>
      <!-- 名称是否可以更改为 false 为可以更改 -->
      <param name="StaticLogFileName" value="false"/>
      <layout type="log4net.Layout.PatternLayout,log4net">
        <!-- 输出格式 -->
        <param name="ConversionPattern" value="[%date] [%thread] %-5level Log4NetTest %logger %method [%message%exception]%n"/>
      </layout>
    </appender>
    <appender name="ErrorRollingFile" type="log4net.Appender.RollingFileAppender,log4net">
      <filter type="log4net.Filter.LevelRangeFilter">
        <levelMin value="ERROR"/>
        <levelMax value="FATAL"/>
      </filter>
      <!-- 日志的路径 -->
      <param name="File" type="" value="D://WebLog//Log4NetTest.App//"/>
       <param name="Encoding" value="UTF-8"/>
      <!-- 是否覆盖，默认是追加 true-->
      <param name="AppendToFile" value="true"/>
      <param name="RollingStyle" value="Date"/>
      <!-- 文件名称 -->
      <param name="DatePattern" value="yyyy-MM-dd'.Error.log'"/>
      <!-- 设置无限备份 =-1，最大备份数为 1000-->
      <param name="MaxSizeRollBackups" value="1000"/>
      <!-- 每个文件的大小 -->
      <param name="MaximumFileSize" value="102KB"/>
      <!-- 名称是否可以更改为 false 为可以更改 -->
      <param name="StaticLogFileName" value="false"/>
      <layout type="log4net.Layout.PatternLayout,log4net">
        <!-- 输出格式 -->
        <param name="ConversionPattern" value="[%date] [%thread] %-5level Log4NetTest %l [%message%n%exception]%n"/>
      </layout>
    </appender>
  </log4net>

针对两种不同类型的应用日志，分别使用两种 pattern，也分别记录到 Debug.lg 和 Error.log 文件中。

• 日志类型为 Debug,Info,Warn 的日志，使用[%date] [%thread] %-5level Log4NetTest %logger %method [%message%exception]%n 模式，分别记录下时间，线程，日志等级，应用名称，日志记录类属性，日志记录方法，日志信息和异常
• 日志类型为 Error 和 Fatal 的日志，使用[%date] [%thread] %-5level Log4NetTest %l [%message%n%exception]%n，分别是时间，线程，日志等级，应用名称，出错位置（包含具体文件，以及所在行，需要 PDB 文件才能到行），日志信息和异常

分两类的主要原因就在于是否记录了出错位置，这个是比较耗性能的操作，所以对于常规的 Debug,Info,Warn 可以不记录位置。
除此之外，没有记录 host，因为 FileBeat 在采集日志时候会自动记录 hostname。这样 log4net 的基本配置就完成了。

FileBeat 配置

只需简单的配置就即可使用，当然也可以配置的很复杂。配置文件 filebeat.yml
一个 input_type 代表一个输入源，可选值只有 log 和 stdin。
paths 是日志采集路径，可以使用通配符。
document_type 是可以用来表示日志类型。输入到 logstash 中对应[type]，可以据此使用不同 grok 语法来 parse，这里分为 errorLog 和 debugLog。

  multiline.pattern: '^\['
  multiline.negate: true
  multiline.match: after

上面这三个使用了将换行的日志或异常信息聚合为一个事件，因为默认 FileBeat 是按行来读取日志，然后传输给 LogStash，如果没这个设置，就会造成日志分割为多个事件。
output.logstash 就是输出的目标了。
然后直接运行 filebeat.exe 即可启动。当然也可以以服务方式启动。

filebeat.prospectors:
-input_type: log
  # Paths that should be crawled and fetched. Glob based paths.
paths:
    #- /var/log/*.log
-D:\WebLog\*\*.Error.log
document_type: errorLog
  multiline.pattern: '^\['
  multiline.negate: true
  multiline.match: after
-input_type: log
 
  # Paths that should be crawled and fetched. Glob based paths.
paths:
    #- /var/log/*.log
-D:\WebLog\*\*.Debug.log
document_type: debugLog
 
 
#----------------------------- Logstash output --------------------------------
output.logstash:
  # The Logstash hosts
hosts: ["localhost:5044"]

LogStash 配置

在 config 文件夹添加配置文件 first-pipeline.conf。

• input: 指定输入来源
• filter: 是指定如何对日志进行处理。这里 [type] 就是来自 filebeat 中 document_type。然后就是 grok 语法了。
• overwrite：是将原有 message 覆盖掉。如果将原有 message 完全 match 出来的话，是可以这样做的，可以节省空间。
• output：就是输出地址了。
• 运行
  bin/logstash -f first-pipeline.conf –config.test_and_exit 测试配置文件
  bin/logstash -f first-pipeline.conf –config.reload.automatic 自动加载配置文件的修改

input {
    beats {port => "5044"
    }
}
# The filter part of this file is commented out to indicate that it is
# optional.
filter {if [type] == "debugLog" {grok {match => {"message" => "\[(?<datetime>\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2},\d{3})\]\s+\[(?<thread>.*)\]\s+(?<level>\w*)\s+(?<appname>\S*)\s+(?<class>\S*)\s+(?<method>\S*)\s+\[(?<message>.*)\]\s*"
             } 
              overwrite => ["message"]
        }
    }
    if [type] == "errorLog" {grok {match => {"message" => "\[(?<datetime>\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2},\d{3})\]\s+\[(?<thread>.*)\]\s+(?<level>\w*)\s+(?<appname>\S*)\s+(?<location>\S*)\s+\[(?<message>.*)\]\s*"
             } 
              overwrite => ["message"]
        }
    }
}
output {elasticsearch { hosts => ["localhost:9200"] }
    stdout {codec => rubydebug}
}

ElasticSearch 配置

默认���需要配置，监听 9200 端口。直接运行即可

Kibana 配置

elasticsearch.url: “http://localhost:9200”
默认连接 es 地址，如果本机测试无需修改。正式环境中连接到对应服务器就好。
server.port: 5601
监听端口 5601，可修改到合适的端口。

然后直接运行就可启动。
初次进入要指定 index pattern。一般默认使用如下配置即可。

Linux 部署 ELK 日志分析系统与简单测试 http://www.linuxidc.com/Linux/2016-10/135849.htm

ELK stack 实战之 Filebeat 的架构分析、配置解释与示例 http://www.linuxidc.com/Linux/2016-10/135850.htm

日志分析工具 ELK 配置详解  http://www.linuxidc.com/Linux/2016-11/137641.htm

本文永久更新链接地址：http://www.linuxidc.com/Linux/2016-12/138657.htm