浅谈Elasticsearch监控和日志分析

from datetime import datetime
from elasticsearch import Elasticsearch
 
# 假设Elasticsearch运行在本地，端口是9200
es = Elasticsearch(["http://localhost:9200"])
 
# 查询Elasticsearch日志索引的文档数量
def get_log_count(index_name):
    response = es.count(index=index_name)
    return response['count']
 
# 获取最近一小时内的错误日志数量
def get_error_logs_last_hour(index_name):
    end_time = datetime.now()
    start_time = end_time - timedelta(hours=1)
    query = {
        "query": {
            "bool": {
                "must": [
                    {"match": {"log_level": "ERROR"}}
                ],
                "filter": [
                    {"range": {
                        "timestamp": {
                            "gte": start_time,
                            "lt": end_time
                        }
                    }}
                ]
            }
        }
    }
    response = es.search(index=index_name, body=query)
    return response['hits']['total']['value']
 
# 使用函数
index_name = "logstash-*"
error_logs_last_hour = get_error_logs_last_hour(index_name)
print(f"Error logs in the last hour: {error_logs_last_hour}")

这段代码使用了Elasticsearch的Python客户端库，连接到本地运行的Elasticsearch实例，并查询了日志索引中的文档数量以及最近一小时内的错误日志数量。这为开发者提供了一个简单的例子，展示了如何使用Elasticsearch进行基本的查询操作。