如何用elasticsearch-dsl-py复刻Kibana查询获取主机最新字段值

需要编写一个函数，接收主机名列表和字段列表作为参数，动态生成高效的Elasticsearch查询，获取每个传入主机对应字段的最新值，实现与Kibana可视化表格相同的效果。

Kibana实现效果 #

以表格形式展示指定主机的目标字段最新值，每行对应一个主机，列对应传入的各个字段，确保每个主机仅展示最新时间戳对应的字段值。

Kibana生成的查询语句 #

{
  "aggs": {
    "2": {
      "terms": {
        "field": "host.hostname",
        "order": {
          "_key": "desc"
        },
        "size": 5
      },
      "aggs": {
        "1": {
          "top_hits": {
            "fields": [
              {
                "field": "host.ip"
              }
            ],
            "_source": false,
            "size": 1,
            "sort": [
              {
                "@timestamp": {
                  "order": "desc"
                }
              }
            ]
          }
        },
        "3": {
          "top_hits": {
            "fields": [
              {
                "field": "source.ip"
              }
            ],
            "_source": false,
            "size": 1,
            "sort": [
              {
                "@timestamp": {
                  "order": "desc"
                }
              }
            ]
          }
        }
      }
    }
  },
  "size": 0,
  "script_fields": {},
  "stored_fields": [
    "*"
  ],
  "runtime_mappings": {},
  "query": {
    "bool": {
      "must": [],
      "filter": [
        {
          "bool": {
            "should": [
              {
                "bool": {
                  "should": [
                    {
                      "match_phrase": {
                        "host.hostname": "p-hostname-a"
                      }
                    }
                  ],
                  "minimum_should_match": 1
                }
              },
              {
                "bool": {
                  "should": [
                    {
                      "match_phrase": {
                        "host.hostname": "P-hostname-H"
                      }
                    }
                  ],
                  "minimum_should_match": 1
                }
              }
            ],
            "minimum_should_match": 1
          }
        },
        {
          "range": {
            "@timestamp": {
              "format": "strict_date_optional_time",
              "gte": "2022-09-20T07:57:26.189Z",
              "lte": "2022-09-21T07:57:26.189Z"
            }
          }
        }
      ],
      "should": [],
      "must_not": []
    }
  }
}

我的Python实现思路 #

def get_data_from_elastic(self, fields_to_get, set_of_host_hostname):
    s = Search().using(client=self.es_con).index(self.METRICBEAT_INDEX_NAME)
    s = s.filter("range", ** {'@timestamp': {'gte': 'now-3600m/m'}})
    set_of_host_hostname = list(set_of_host_hostname)

    set_of_host_hostname = set_of_host_hostname[0:2]

    s.aggs.bucket(name=f"main_bucket", agg_type="terms", field="host.hostname", size=1)
    for field in fields_to_get:
        s.aggs["main_bucket"].metric(name=f"name_{field}", agg_type="top_hits", fields=[{"field":field}])

    s.source(fields=fields_to_get)
    main_q = Q("bool", minimum_should_match=1)
    for host_hostname in set_of_host_hostname:
        q = Q("match", **{"host.hostname":host_hostname})
        main_q.should.append(q)
    s = s.query(main_q)
    # print("\n\n\n")
    pp(s.to_dict())
    print("\n\n\n")
    res = s.execute()
    for hit in res:
        print(hit)
        print(hit.host.ip)

执行结果 #

生成的查询结构 #

{    
    "aggs":{
        "main_bucket":{
            "aggs":{
                "name_host.ip":{
                    "top_hits":{
                        "fields":[
                            {
                                "field":"host.ip"
                            }
                        ]
                    }
                },
                "name_host.os.type":{
                    "top_hits":{
                        "fields":[
                            {
                                "field":"host.os.type"
                            }
                        ]
                    }
                }
            },
            "terms":{
                "field":"host.hostname",
                "size":1
            }
        }    
    },    
    "query":{
        "bool":{
            "filter":[
                {
                    "range":{
                        "@timestamp":{
                            "gte":"now-3600m/m"
                        }
                    }
                }
            ],
            "minimum_should_match":1,
            "should":[
                {
                    "match":{
                        "host.hostname":"P-hostname-2"
                    }
                },
                {
                    "match":{
                        "host.hostname":"P-hostname-1"
                    }
                }
            ]
        }    
    } 
}

查询返回结果 #

<Hit(metricbeat-7.17.1-system-2022.38/AEESYIMB2-liSoZOlAYu): {'agent': {'version': '7.17.1'}, '@timestamp': '2022-09-21T1...}>
['1thesameip0']
<Hit(metricbeat-7.17.1-system-2022.38/AUESYIMB2-liSoZOlAYu): {'agent': {'version': '7.17.1'}, '@timestamp': '2022-09-21T1...}>
['1thesameip0']
<Hit(metricbeat-7.17.1-system-2022.38/AkESYIMB2-liSoZOlAYu): {'agent': {'version': '7.17.1'}, '@timestamp': '2022-09-21T1...}>
['1thesameip0']
<Hit(metricbeat-7.17.1-system-2022.38/A0ESYIMB2-liSoZOlAYu): {'agent': {'version': '7.17.1'}, '@timestamp': '2022-09-21T1...}>
['1thesameip0']
<Hit(metricbeat-7.17.1-system-2022.38/BEESYIMB2-liSoZOlAYu): {'agent': {'version': '7.17.1'}, '@timestamp': '2022-09-21T1...}>
['1thesameip0']
<Hit(metricbeat-7.17.1-system-2022.38/BUESYIMB2-liSoZOlAYu): {'agent': {'version': '7.17.1'}, '@timestamp': '2022-09-21T1...}>
['1thesameip0']
<Hit(metricbeat-7.17.1-system-2022.38/BkESYIMB2-liSoZOlAYu): {'agent': {'version': '7.17.1'}, '@timestamp': '2022-09-21T1...}>
['1thesameip0']
<Hit(metricbeat-7.17.1-system-2022.38/B0ESYIMB2-liSoZOlAYu): {'agent': {'version': '7.17.1'}, '@timestamp': '2022-09-21T1...}>
['1thesameip0']
<Hit(metricbeat-7.17.1-system-2022.38/CEESYIMB2-liSoZOlAYu): {'agent': {'version': '7.17.1'}, '@timestamp': '2022-09-21T1...}>
['1thesameip0']
<Hit(metricbeat-7.17.1-system-2022.38/CUESYIMB2-liSoZOlAYu): {'agent': {'version': '7.17.1'}, '@timestamp': '2022-09-21T1...}>
['1thesameip0']

问题 #

尝试了多种变体，但仍无法复刻Kibana生成的查询结构，请求指导如何实现该需求。使用的elasticsearch-dsl版本为7.4。

内容的提问来源于stack exchange，提问作者Adrian