如何通过Elastic API/Python获取Java应用近1小时链路追踪及事务列表

1. 通过API获取事务列表 #

首先确认存储事务数据的Elasticsearch索引（通常命名类似transactions-*），使用Elasticsearch的Search API执行查询即可提取事务列表。

Python实现示例 #

from elasticsearch import Elasticsearch

# 建立Elasticsearch连接（根据实际环境调整地址、认证信息）
es = Elasticsearch("https://localhost:9200", 
                   basic_auth=("username", "password"),  # 有认证需求时添加
                   verify_certs=False)  # 测试环境可关闭证书验证，生产环境建议开启

# 构造查询：可根据需求替换过滤条件，这里以获取所有事务为例
query = {
    "query": {
        "match_all": {}
    },
    "_source": ["transaction_id", "name", "timestamp", "status"]  # 指定返回的事务字段
}

# 执行查询，替换为你的事务索引名
response = es.search(index="transactions-*", body=query, size=1000)  # size按需调整返回数量

# 提取事务列表
transactions = [hit["_source"] for hit in response["hits"]["hits"]]
print("事务列表：", transactions)

2. 通过Python获取近1小时的链路追踪详情 #

链路追踪数据一般存储在类似traces-*的索引中，需构造时间范围查询筛选近1小时数据，再结合业务条件过滤。

Python实现示例 #

from elasticsearch import Elasticsearch
from datetime import datetime, timedelta

# 建立连接
es = Elasticsearch("https://localhost:9200", 
                   basic_auth=("username", "password"),
                   verify_certs=False)

# 计算近1小时的时间范围（适配Elasticsearch的ISO8601格式）
one_hour_ago = (datetime.now() - timedelta(hours=1)).isoformat() + "Z"

# 构造查询：结合业务条件与时间范围筛选
query = {
    "query": {
        "bool": {
            "must": [
                {"match": {"name": "1NN_java"}},  # 你的业务过滤条件
                {"range": {
                    "timestamp": {
                        "gte": one_hour_ago,
                        "lte": "now"
                    }
                }}
            ]
        }
    },
    "_source": ["trace_id", "span_id", "duration", "service_name", "timestamp"]  # 指定链路追踪字段
}

# 执行查询，替换为你的链路追踪索引名
response = es.search(index="traces-*", body=query, size=1000)

# 提取链路追踪详情
trace_details = [hit["_source"] for hit in response["hits"]["hits"]]
print("近1小时链路追踪详情：", trace_details)

注意事项 #

• 替换示例中的索引名、认证信息为实际环境配置
• 根据数据结构调整_source字段和查询条件
• 数据量较大时，建议使用分页（from和size参数）或滚动查询优化性能

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