在 Logstash 级别读取 CSV 并根据提取的数据进行过滤

Question

我正在使用 Metricbeat 获取流程级数据并使用 Logstash 将其推送到 Elastic Search。

现在，我们的目标是将进程分为 2 个标签，即正在运行的进程要么是浏览器，要么是其他东西。

我可以使用这段代码静态地做到这一点：

input {
  beats {
    port => 5044
  }
}
filter{
    if [process][name]=="firefox.exe" or [process][name]=="chrome.exe" {
        mutate {
            add_field => { "process.type" => "browsers" }
            convert => {
            "process.type" => "string"
            }
        }
    }
    else {
        mutate {
            add_field => { "process.type" => "other" }
        } 
    }
}

output {
  elasticsearch {
    hosts => "localhost:9200"
    # manage_template => false
    index => "metricbeatlogstash"
  }
}

但是当我尝试通过从 CSV 读取进程列表来使条件动态化时，我在 Kibana 中没有得到任何有效结果，在我的 LogStash 级别上也没有错误。

CSV 配置文件代码如下：

input {
  beats {
    port => 5044
  }
  file{
        path=>"filePath"
        start_position=>"beginning"
        sincedb_path=>"NULL"
    }
}
filter{
    csv{
        separator=>","
        columns=>["processList","IT"]
    }
    if [process][name] in [processList] {
        mutate {
            add_field => { "process.type" => "browsers" }
            convert => {
            "process.type" => "string"
            }
        }
    }
    else {
        mutate {
            add_field => { "process.type" => "other" }
        } 
    }
}

output {
  elasticsearch {
    hosts => "localhost:9200"
    # manage_template => false
    index => "metricbeatlogstash2"
  }
}

Answer 1

您尝试做的事情在 Logstash 中不起作用，logstash 管道中的事件彼此独立。

您的beats 输入接收到的事件不知道您的csv 输入接收到的事件，因此您不能在条件中使用来自不同事件的字段。

要做你想做的事，你可以使用translate 过滤器和以下配置。

translate {
    field => "[process][name]"
    destination => "[process][type]"
    dictionary_path => "process.csv"
    fallback => "others"
    refresh_interval => 300
}

此过滤器将对照字典检查字段[process][name]的值，从文件process.csv加载到内存中，字典是一个.csv文件，有两列，第一列是浏览器进程的名称第二个总是browser。

chrome.exe,browser
firefox.exe,browser

如果过滤器匹配，它将使用第二列中的值填充字段[process][type]（不是process.type），在这种情况下，总是browser，如果没有匹配，它将填充字段[process][type] 与fallback 配置的值，在本例中为others，它还将每300 秒（5 分钟）重新加载process.csv 文件的内容