Logstash：在实施之前测试 Logstash 管道/过滤器

Posted 2023-02-23 Elastic 中国社区官方博客

检测解析的日志是否包含单个或多个警告消息，然后添加一个字段来说明这两种情况。在很多的情形下，我们在测试 Logstash 的过滤器时，并不急于把实际的 input 的数据接入到过滤器中来进行测试。我们首先来选择一个比较容易理解的 input 方式，使用一个文档来进行解析，并测试管道。在今天的文章中，我来详细介绍两种常用的方法来如何测试 Logstash 的管道/过滤器。

方法一：使用 generator

方法如下：

logstash.conf

input 
  generator 
    message => '"id":2,"timestamp":"2019-08-11T17:55:56Z","paymentType":"Visa","name":"Darby Dacks","gender":"Female","ip_address":"77.72.239.47","purpose":"Shoes","country":"Poland","age":55'
    count => 1
  

 
filter 
    json 
        source => "message"
    

    if [paymentType] == "Mastercard" 
        drop 
    

    mutate 
        remove_field => ["message", "@timestamp", "path", "host", "@version", "log", "event"]
    


 
output 
  stdout 
    codec => rubydebug
  

在上面，我们使用 generator 的方法来生成一个文档，并让这个文档经过 filter 部分，并最终在 console 中进行展示。我们可以通过如下的命令来运行上面的 Logstash 管道：

$ pwd
/Users/liuxg/elastic/logstash-8.6.1
$ ls logstash.conf
logstash.conf
$ ./bin/logstash -f logstash.conf

   

从上面，我们可以看出来 json filter 工作正常。在本示例中，为了说明问题的方便，我仅使用了几个过滤器。在时间的使用中，我们可以有很多的过滤器来组成这个 pipeline。一旦我们确定了这些过滤器能完成我们所需要的功能，我们可以把所需要的 input 换进来即可，比如：

logstash_filter.conf

input 
  file 
    path => "/Users/liuxg/elastic/logstash-8.6.1/sample.json"
    type    => "applog"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  


filter 
    json 
        source => "message"
    

    if [paymentType] == "Mastercard" 
        drop 
    

    mutate 
        remove_field => ["message", "@timestamp", "path", "host", "@version", "log", "event"]
    



output 
	stdout  
	  codec => rubydebug 
	

我们可以使用诸如如下格式的测试文件来进行测试：

sample.json

"id":1,"timestamp":"2019-09-12T13:43:42Z","paymentType":"Amex","name":"Merrill Duffield","gender":"Female","ip_address":"132.150.218.21","purpose":"Toys","country":"United Arab Emirates","age":33
"id":2,"timestamp":"2019-08-11T17:55:56Z","paymentType":"Visa","name":"Darby Dacks","gender":"Female","ip_address":"77.72.239.47","purpose":"Shoes","country":"Poland","age":55
"id":3,"timestamp":"2019-07-14T04:48:25Z","paymentType":"Visa","name":"Harri Cayette","gender":"Female","ip_address":"227.6.210.146","purpose":"Sports","country":"Canada","age":27
"id":4,"timestamp":"2020-02-29T12:41:59Z","paymentType":"Mastercard","name":"Regan Stockman","gender":"Male","ip_address":"139.224.15.154","purpose":"Home","country":"Indonesia","age":34
"id":5,"timestamp":"2019-08-03T19:37:51Z","paymentType":"Mastercard","name":"Wilhelmina Polle","gender":"Female","ip_address":"252.254.68.68","purpose":"Health","country":"Ukraine","age":51

当然实际的文档可能比这个要长很多。

更多关于 generator 方面的示例，请阅读我之前的文章 “Logstash：Data 转换，分析，提取，丰富及核心操作”。

方法二：使用 stdin input

假设我们有以下代表上述两种情况的日志文件：

$ pwd
/Users/liuxg/elastic/logstash-8.6.1
$ cat multivaluewarn.json 
"waf": "ver": "2.0","warnRules": "3000030;3000057;950001;950109;959073;973335;981173;981244;981318","denyMsg": "Anomaly Score Exceeded for SQL Injection","denyActions": "3","warnMsg": "Basic SQL Authentication Bypass Attempts 3/3;Cross-site Scripting (XSS) common keywords;SQL Injection Attack;Multiple URL Encoding Detected;SQL Injection Attack;IE XSS Filters - Attack Detected;Restricted SQL Character Anomaly Detection Alert - Total # of special characters exceeded;Basic SQL Authentication Bypass Attempts 1/3;SQL Injection Attack: Common Injection Testing Detected" 
$ 
$ cat singlevaluewarn.json 
"waf": "ver": "2.0","warnRules": "681984","policy": "api_89894","warnMsg": "Alert rq without DEVICEID header","warnTags": "DEVICEID_Detection","warnActions": "2"

multivaluewarn.json

"waf": "ver": "2.0","warnRules": "3000030;3000057;950001;950109;959073;973335;981173;981244;981318","denyMsg": "Anomaly Score Exceeded for SQL Injection","denyActions": "3","warnMsg": "Basic SQL Authentication Bypass Attempts 3/3;Cross-site Scripting (XSS) common keywords;SQL Injection Attack;Multiple URL Encoding Detected;SQL Injection Attack;IE XSS Filters - Attack Detected;Restricted SQL Character Anomaly Detection Alert - Total # of special characters exceeded;Basic SQL Authentication Bypass Attempts 1/3;SQL Injection Attack: Common Injection Testing Detected" 

   "waf":
      "ver":"2.0",
      "warnRules":"3000030;3000057;950001;950109;959073;973335;981173;981244;981318",
      "denyMsg":"Anomaly Score Exceeded for SQL Injection",
      "denyActions":"3",
      "warnMsg":"Basic SQL Authentication Bypass Attempts 3/3;Cross-site Scripting (XSS) common keywords;SQL Injection Attack;Multiple URL Encoding Detected;SQL Injection Attack;IE XSS Filters - Attack Detected;Restricted SQL Character Anomaly Detection Alert - Total # of special characters exceeded;Basic SQL Authentication Bypass Attempts 1/3;SQL Injection Attack: Common Injection Testing Detected"
   

singlevaluewarn.json

"waf": "ver": "2.0","warnRules": "681984","policy": "api_89894","warnMsg": "Alert rq without DEVICEID header","warnTags": "DEVICEID_Detection","warnActions": "2"

   "waf":
      "ver":"2.0",
      "warnRules":"681984",
      "policy":"api_89894",
      "warnMsg":"Alert rq without DEVICEID header",
      "warnTags":"DEVICEID_Detection",
      "warnActions":"2"
   

阅读日志我们可以看到字段 [waf][warnMsg] 使用分号分隔警告消息； 在多次警告的情况下。
将收集到的信息转换为 Logstash 管道将导致：

logstash_warning.conf

input 
  stdin  codec => json 


filter 
  if ";" in [waf][warnMsg]
    mutate 
      add_field =>  [ "wafWarningMSG", "multi warnings" ]
    
  
  else 
    mutate 
      add_field =>  [ "wafWarningMSG", "single" ]
    
  


output 
  stdout 
    codec => rubydebug
  

将管道添加到 conf 文件（称为 logstash_warning.conf ），然后使用如下的命令来测试 pipeline：

$ pwd
/Users/liuxg/elastic/logstash-8.6.1
$ ls logstash_warning.conf 
logstash_warning.conf
$ ./bin/logstash -f logstash_warning.conf < multivaluewarn.json

 输出显示一个名为 wafWarningMSG 的新字段，其中包含 "multi warnings"：

当然，我们也可以使用如下的命令来进行测试：

./bin/logstash -f logstash_warning.conf < singlevaluewarn.json 

 

从上面的输出中，我们可以看到 wafWarningMSG 字段的值为 single。

一旦我们测试好 pipeline 中的过滤器，我们就可以直接把 input 部分换成我们想要的格式即可。

希望你觉得它有用，如有任何问题，请随时联系我！