Logstash 过滤器

Logstash 在输入和输出之间的管道中间使用过滤器。 Logstash 度量的过滤器操作和创建诸如 Apache-Access 之类的事件。许多过滤器插件用于管理 Logstash 中的事件。此处，在 Logstash 聚合过滤器的示例中，我们过滤数据库中每个 SQL 事务的持续时间并计算总时间。

安装聚合过滤器插件

使用 Logstash-plugin 实用程序安装聚合过滤器插件。 Logstash 插件是 Logstash 中 bin 文件夹 中 Windows 的批处理文件。

>logstash-plugin install logstash-filter-aggregate

logstash.conf

在此配置中，可以看到三个"if"语句，用于 初始化、递增、和 生成事务的总持续时间，即 sql_duration。聚合插件用于添加 sql_duration，存在于输入日志的每个事件中。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input.log"
   }
} 
filter {
   grok {
      match => [
         "message", "%{LOGLEVEL:loglevel}-
            %{NOTSPACE:taskid}-%{NOTSPACE:logger}-
            %{WORD:label}(-%{INT:duration:int})?" 
      ]
   }
   if [logger] == "TRANSACTION_START" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] = 0"
         map_action => "create"
      }
   }
   if [logger] == "SQL" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] ||= 0 ;
            map['sql_duration'] += event.get('duration')"
      }
   }
   if [logger] == "TRANSACTION_END" {
      aggregate {
         task_id => "%{taskid}"
         code => "event.set('sql_duration', map['sql_duration'])"
         end_of_task => true
         timeout => 120
      }
   }
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"    
   }
}

运行 Logstash

我们可以使用以下命令运行 Logstash。

>logstash –f logstash.conf 

input.log

以下代码块显示了输入日志数据。

INFO-48566-TRANSACTION_START-start
INFO-48566-SQL-transaction1-320
INFO-48566-SQL-transaction1-200
INFO-48566-TRANSACTION_END-end

输出.log

如配置文件中指定的，记录器所在的最后一个‘if’语句——TRANSACTION_END，打印总事务时间或sql_duration。这已在 output.log 中以黄色突出显示。

{
   "path":"C:/tpwork/logstash/bin/log/input.log","@timestamp": "2016-12-22T19:04:37.214Z",
   "loglevel":"INFO","logger":"TRANSACTION_START","@version": "1","host":"wcnlab-PC",
   "message":"8566-TRANSACTION_START-start\r","tags":[]
}
{
   "duration":320,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-22T19:04:38.366Z","loglevel":"INFO","logger":"SQL",
   "@version":"1","host":"wcnlab-PC","label":"transaction1",
   "message":" INFO-48566-SQL-transaction1-320\r","taskid":"48566","tags":[]
}
{
   "duration":200,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-22T19:04:38.373Z","loglevel":"INFO","logger":"SQL",
   "@version":"1","host":"wcnlab-PC","label":"transaction1",
   "message":" INFO-48566-SQL-transaction1-200\r","taskid":"48566","tags":[]
}
{
   "sql_duration":520,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-22T19:04:38.380Z","loglevel":"INFO","logger":"TRANSACTION_END",
   "@version":"1","host":"wcnlab-PC","label":"end",
   "message":" INFO-48566-TRANSACTION_END-end\r","taskid":"48566","tags":[]
}

Logstash 提供了各种插件来转换已解析的日志。这些插件可以在日志中 添加、删除、和 更新字段，以便在输出系统中更好地理解和查询。我们正在使用 Mutate 插件 在输入日志的每一行中添加一个字段名称 use ...