Logstash：在 Logstash 管道中的定制的 Elasticsearch update by query

Posted 2023-03-05 Elastic 中国社区官方博客

我们知道 Elasticsearch output plugin 为我们在 Logstash 的 pipeline 中向 Elasticsearch 的写入提供了可能。我们可以使用如下的格式向 Elasticsearch 写入数据：

  elasticsearch 
    hosts => ["https://localhost:9200"]
    index => "data-%+YYYY.MM.dd"
    user => "elastic"
    password => "NtC7cM-GKQWOxqamHd1R"
    ssl => true
    ca_trusted_fingerprint => "d464eed5d00a20908318b6a1de38f88daf3a867177123def4c34aa2272571aaf"
  

在向 Elasticsearch 写入数据的时候，目前它有四种操作：

• index：索引文档（来自 Logstash 的事件）。
• delete：通过 id 删除文档（此操作需要一个id）
• create：索引文档，如果索引中已存在该 id 的文档，则失败。
• update：通过 id 更新文档。 Update 有一个特殊情况，你可以 upsert — 更新文档（如果不存在）。 请参阅 doc_as_upsert 选项。 注意：这在 Elasticsearch 1.x 中不起作用且不受支持。 请升级到 ES 2.x 或更高版本以将此功能与 Logstash 一起使用！

一个 sprintf 样式的字符串，用于根据事件的内容更改操作。 值 %[foo] 将使用 foo 字段进行操作。 如果 resolved action 不在 [index, delete, create, update] 中，事件将不会发送到 Elasticsearch。 相反，事件将被发送到管道的死信队列 (DLQ)（如果启用），或者将被记录并删除。

在实际的使用中，假如我们的操作不是 index，delete create 或 update 其中的一种，那么我们该怎么办呢？比如我们想根据一定的条件来更新文档，就像 update by query 那样？我们该怎么办呢？

幸运的是，Logstash 提供了一个叫做 HTTP output plugin。它可以帮我解决这个问题。

准备数据

首先，我们来创建如下的一个索引：

PUT customer/_doc/2

  "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,
  "offer": false 

我们在 Kibana 中输入上面的命令来创建一个叫做 customer 的索引。它的 id 为 2。

更新数据

接下来，我们需要按照一定的条件来更新我们的数据。比如，我们想把 paymentType 为 Visa，并且年龄大于或等于 55 岁的人的 offer 设置为 true。在 Kibana 中正常的命令是这样的：

POST customer/_update_by_query

  "query": 
    "bool": 
      "must": [
        
          "match": 
            "paymentType.keyword": "Visa"
          
        ,
        
          "range": 
            "age": 
              "gte": 50
            
          
        
      ]
    
  ,
  "script": 
    "source": "ctx._source.offer = params.offer",
    "lang": "painless",
    "params": 
      "offer": true
    
  

我们可以对 Logsthash 做如下的 pipeline 设计：

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
  
  http 
      url => "https://localhost:9200/customer/_update_by_query"
      user => "elastic"
      password => "Y+6tv9jejPl=W4IGrTD="
      http_method => "post"
      format => "message"
      content_type => "application/json"
      message => '"query":"bool":"must":["match":"paymentType.keyword":"%paymentType","range":"age":"gte":"%age"],"script":"source":"ctx._source.offer = params.offer","lang":"painless","params":"offer":true' 
      cacert => "/Users/liuxg/elastic/elasticsearch-8.6.1/config/certs/http_ca.crt"
  

在上面，我们在 message 中通过一个查询，匹配到 paymentType.keyword 为 Visa，并且 age 为大于等于 55 的文档，我们设置该用户为促销对象。把他的 offer 值设置为 true。这个在实际的使用中，依据自己的条件来进行配置。在上面，cacert 为我们的 Elasticsearch 的证书文件位置。具体使用，请参考文档。

我们接下来运行 Logstash 的 pipeline：

./bin/logstash -f logstash.conf

 

在上面我们可以看出来信息的输出。我们在 Kibana 中使用如下的命令来检查更新后的文档：

GET customer/_search

  "took": 0,
  "timed_out": false,
  "_shards": 
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  ,
  "hits": 
    "total": 
      "value": 1,
      "relation": "eq"
    ,
    "max_score": 1,
    "hits": [
      
        "_index": "customer",
        "_id": "2",
        "_score": 1,
        "_source": 
          "offer": true,
          "country": "Poland",
          "gender": "Female",
          "purpose": "Shoes",
          "name": "Darby Dacks",
          "id": 2,
          "ip_address": "77.72.239.47",
          "age": 55,
          "timestamp": "2019-08-11T17:55:56Z",
          "paymentType": "Visa"
        
      
    ]
  

很显然，我们上面的 offer 值现在变为 true，而不是之前的 false。