1. 模块概述

KibanaModule 是一个基于 elk的日志查询模块，用于实现与 elk服务的交互，支持日志查询、数据检索和资源管理。以下是该模块的主要功能和方法说明：

2. 主要功能

   • 日志查询: 通过 KQL 查询语法获取日志数据。
   • 时间范围构建: 自动构建查询的时间范围。
   • KQL 解析: 解析 KQL 语法并生成查询条件。
   • 响应解析: 解析 Kibana API 响应并转换时间戳。
   • 会话管理: 管理 HTTP 会话，包括基本认证和会话 ID。

3. 方法说明

   1.   初始化

   • init(config: Dict[str, Any] = None): 初始化 Kibana 模块，可以传入配置字典。
   • init_from_config(config: Dict[str, Any]): 从配置字典初始化参数。

   2.   日志查询

     search_by_kql(**kwargs)

     描述: 使用 KQL 查询日志。

   参数:

   kql (str): KQL 查询字符串。

   minutes (int, 可选): 查询时间范围（分钟），默认为 15 分钟。

   size (int, 可选): 返回的日志数量，默认为 100 条。

   fields (List[str], 可选): 返回的字段列表，例如 ['@timestamp', 'message', 'app_name', 'log_level']。

   saveAsVariable (str, 可选): 将查询结果保存到指定变量。

   3.   会话管理

     _init_session()

     描述: 初始化 HTTP 会话，包括基本认证和会话 ID。

   4.   时间范围构建

     _build_time_range(minutes: int = 15)

     描述: 构建查询的时间范围。

   参数:

   minutes (int, 可选): 查询时间范围（分钟），默认为 15 分钟。

   5.   KQL 解析

     _parse_kql_to_conditions(kql: str) -> List[Dict[str, Any]]

     描述: 解析 KQL 语法并生成查询条件。

   参数:

   kql (str): KQL 查询字符串。

   6.   响应解析

     _parse_search_response(response_data: Dict[str, Any]) -> Dict[str, Any]

     描述: 解析 Kibana API 响应并转换时间戳。

   参数:

   response_data (Dict[str, Any]): Kibana API 响应数据。

   7.   清理资源

     cleanup() -> Dict[str, Any]

     描述: 清理资源，关闭会话。

4. 环境准备

安装依赖

auto-test-framework-common==0.1.62

5. 初始化模块

6. 示例

import yaml
from kibana import KibanaModule

management = InstanceManagement()
kibana: KibanaModule = management.get_or_create("kabana_module")
# 执行查询
specific_kql = 'app_name.keyword : "icc-agent-portal"  and log_level.keyword : "ERROR"'
result3 = kibana.search_by_kql(
    kql=specific_kql,
    minutes=5,
    size=20
)
print(f"查询结果: {result}")