ELK+FileBeat+Log4Net搭建日志系统

464次阅读

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

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

大体框架

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

ELK+FileBeat+Log4Net 搭建日志系统

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

ELK+FileBeat+Log4Net 搭建日志系统

配置过程

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}
}