Python到elasticsearch查询不同时间段的日志

Question

我们拥有包含数千个应用程序的弹性搜索集群。我们想通过 python 脚本从弹性集群读取日志。

用例：

1. 获取最近 1 小时或 1 天的日志
2. 如果出现错误 500 或 404

我们希望从集群中获取此类详细信息。

能够连接到集群并在正文中提供具有匹配条件的查询。

但是我很震惊地提供了日志错误过滤器并获取最后 1 小时的数据

Answer 1

from datetime import datetime, timedelta
from elasticsearch import Elasticsearch

es = Elasticsearch(['your_elasticsearch_host'], port=9200)

index_pattern = "your_index_pattern*"


query = {
  "query": {
    "bool": {
      "must": [
        {
          "range": {
            "@timestamp": {
              "gte": "now-1h"  # Logs from the last 1 hour
            }
          }
        },
        {
          "terms": {
            "status_code": [500, 404]  # Filter logs based on status code 500 or 404
          }
        }
      ]
    }
  }
}


results = es.search(index=index_pattern, body=query)

//response generation.
for hit in results['hits']['hits']:
    print(hit['_source'])

过滤过去 1 小时的日志（“gte”：“now-1h”）

参考：https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-range-query.html