一、前期准备：打造系统的基石

• 1.ELK日志平台：我们的信息收纳大师
• 2.飞书机器人：我们的贴心小助手
• 3.deepseek api_key：我们的智慧担当
• 4.python环境：我们的全能选手具体搭建部署不在本次文章范围内。

二、原始日志数据准备：“制造” 测试样本

为了更好地验证故障匹配效果，我手动“制造”了几条日志数据。这些日志记录了Redis连接状态的变化，mq请求异常和网络连接异常。

2025-03-11 10:35:05,100 [lettuce-epollEventLoop-4-3] INFO  io.lettuce.core.protocol.ReconnectionHandler:177 - Reconnected to 192.168.6.15:6379
2025-03-11 10:36:03,197 [lettuce-eventExecutorLoop-1-2] INFO  io.lettuce.core.protocol.ConnectionWatchdog:173 - Reconnecting, last destination was 192.168.6.13/192.168.6.13:6379
2025-03-11 10:36:03,200 [lettuce-epollEventLoop-4-4] INFO  io.lettuce.core.protocol.ReconnectionHandler:177 - Reconnected to 192.168.6.11:6379
2025-03-11 10:36:43,096 [lettuce-eventExecutorLoop-1-3] INFO  io.lettuce.core.protocol.ConnectionWatchdog:173 - Reconnecting, last destination was 192.168.6.13/192.168.6.13:6379
2025-03-11 10:36:43,099 [lettuce-epollEventLoop-4-1] INFO  io.lettuce.core.protocol.ReconnectionHandler:177 - Reconnected to 192.168.6.13:6379

2025-03-11 10:37:39,272 [SimpleAsyncTaskExecutor-5] ERROR  cn.com.easypay.aspect.WebLogAspect:39 - mq请求异常开始
2025-03-11 10:37:39,272 [SimpleAsyncTaskExecutor-5] ERROR  cn.com.easypay.aspect.WebLogAspect:39 - mq请求异常结束
2025-03-11 10:37:48,139 网络连接异常开始
2025-03-11 10:37:48,139 网络连接异常结束

三、代码准备：构建智能系统的 “骨架”

1.创建配置文件config.ini

把Elasticsearch、DeepSeek、飞书等关键信息都集中放在config.ini这个“百宝箱”里。这样，管理和修改配置就变得超级方便，一目了然。

[elasticsearch]
host = yout_es_ip
port = 9200
username = your_elastic_user
password = your_elastic_password

[deepseek]
api_key = sk-fc5c4a54XXXXXXXXXX61a25eb #你自己的api_key，可以白嫖也可以充值个10块钱。当然也可以调用本地私有部署大模型，训练后效果更佳。
deepseek_api_url = https://api.deepseek.com/chat/completions

[feishu]
webhook_url = https://open.feishu.cn/open-apis/bot/v2/hook/bee9699d-9490-4226-97bd-xxxxxxx  #你的飞书机器人Webhook 地址

[elasticsearch_index]
index_name = your_es_index_name

2.创建elasticsearch_utils.py文件

在这个文件里，封装了与日志相关的操作。并且设置了查询范围为最近1小时内的最新10条日志，就像给时间和数量设定了“筛选器”。当然啦，大家可以根据实际需求随意调整。

3.创建feishu_utils.py文件

这个文件专门用来封装飞书消息发送的操作。建议使用飞书消息卡片格式，这样展示的内容更丰富，交互体验也更好。

import logging
import requests

class FeishuClient:
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url

    def send_message(self, log, analysis_result):
        headers = {
            "Content-Type": "application/json"
        }
        # 飞书消息纯文本格式，markdown展示不友好
        #text = f"日志内容: {log}\n分析结果: {analysis_result}"
        #data = {
        #    "msg_type": "text",
        #    "content": {
        #        "text": text
        #    }
        #}
        
        # 调整使用飞书消息卡片格式，可以展示更丰富的内容和交互
        data = {
            "msg_type": "interactive",
            "card": {
                "header": {
                    "title": {
                        "tag": "plain_text",
                        "content": "DeepSeek日志分析告警推送"
                    }
                },
                "elements": [
                    {
                        "tag": "div",
                        "text": {
                            "tag": "lark_md",
                            "content": f"**来自Deepseek的消息**:\n{log}"
                        }
                    },
                    {
                        "tag": "div",
                        "text": {
                            "tag": "lark_md",
                            "content": f"**deepseek分析结果**:\n{analysis_result}"
                        }
                    }
                ]
            }
        }
        try:
            response = requests.post(self.webhook_url, headers=headers, json=data)
            response.raise_for_status()
            logging.info("消息已成功发送到飞书。")
        except requests.RequestException as e:
            logging.error(f"请求出错: {e}")
        except requests.HTTPError as e:
            logging.error(f"发送消息到飞书失败，状态码: {response.status_code}，响应内容: {response.text}")

4.创建主函数文件main.py

这里是整个系统的“指挥中心”。为了方便展示，我这里对故障匹配的historical_fault_db采用了简单配置。大家也可以大展身手，调整代码，调用更强大的故障管理系统或数据库来优化，让系统变得更智能。

import logging
import configparser
from elasticsearch_utils import ElasticsearchClient
from feishu_utils import FeishuClient
import requests
import json
import time

# 配置日志记录
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 读取配置文件
config = configparser.ConfigParser()
config.read('config.ini')

es_host = config.get('elasticsearch', 'host')
es_port = config.getint('elasticsearch', 'port')
es_username = config.get('elasticsearch', 'username')
es_password = config.get('elasticsearch', 'password')
deepseek_api_url = config.get('deepseek', 'api_url', fallback="https://api.deepseek.com/chat/completions")

# 初始化 Elasticsearch 客户端
es_client = ElasticsearchClient(es_host, es_port, es_username, es_password)

# 初始化飞书客户端
feishu_client = FeishuClient(config.get('feishu', 'webhook_url'))

# 模拟历史故障匹配,内容仅作为举例使用。
historical_fault_db = {
    "Redis连接池耗尽": ["redis connection pool exhausted", "redis pool full"],
    "数据库连接超时": ["database connection timeout", "db connect timed out"],
    "2024年10月28日 网络连接异常历史故障匹配验证": ["网络连接异常","故障根因：专线丢包，交换机版本bug"]
}

def analyze_logs_with_deepseek(logs):
    headers = {
        "Authorization": f"Bearer {config.get('deepseek', 'api_key')}",
        "Content-Type": "application/json"
    }
    combined_logs = "\n".join(logs)
    data = {
        "model": "deepseek-chat",
        "messages": [
            {
                "role": "user",
                "content": f"作为拥有30年经验的资深运维专家,把日志按照故障类型'数据库异常'、'程序异常'、'网络异常'、'redis异常'、'mq异常'等进行分类,并评估影响范围(P0-P3)给出建议：\n{combined_logs}"
            }
        ]
    }
    max_retries = 3
    retries = 0
    while retries < max_retries:
        try:
            response = requests.post(deepseek_api_url, headers=headers, json=data)
            response.raise_for_status()  # 检查 HTTP 状态码，如果不是 200，抛出异常
            result = response.json()
            return result
        except requests.RequestException as e:
            logging.error(f"请求出错: {e}")
            retries += 1
            time.sleep(2)  # 等待 2 秒后重试
        except json.JSONDecodeError as e:
            logging.error(f"JSON 解析出错: {e}")
            break
        except requests.HTTPError as e:
            if response.status_code == 401:
                logging.error("权限验证失败，请检查授权密钥。")
                break
            elif response.status_code == 400:
                logging.error("请求参数有误，请检查请求数据。")
                break
            else:
                logging.error(f"请求出错，状态码: {response.status_code}，响应内容: {response.text}")
                retries += 1
                time.sleep(2)  # 等待 2 秒后重试
    return None


def match_with_historical_db(logs):
    """
    将日志与历史故障库进行匹配
    :param logs: 日志列表
    :return: 匹配结果
    """
    matches = []
    forlogin logs:
        for fault, keywords in historical_fault_db.items():
            for keyword in keywords:
                if keyword.lower() in log.lower():
                    matches.append((log, fault))
    return matches


# 主函数
def main():
    index_name = config.get('elasticsearch_index', 'index_name')
    logs = es_client.get_logs_from_es(index_name)
    if logs:
        # 用 deepseek 分类打标
        result = analyze_logs_with_deepseek(logs)
        if result:
            try:
                message = result["choices"][0]["message"]["content"]
                logging.info(f"整合日志分析结果: {message}")
                # 与历史故障库进行匹配
                historical_matches = match_with_historical_db(logs)

                historical_match_message = ""
                if historical_matches:
                    historical_match_message = "<font color='red'>**历史故障库匹配结果:**</font>\n"
                    forlog, fault in historical_matches:
                        historical_match_message += f"日志内容: {log}\n 匹配故障: <font color='red'>**{fault}**</font>\n"
                else:
                    historical_match_message = "未找到与历史故障库匹配的日志。"

                logging.info(historical_match_message)

                # 整合 deepseek 分析结果和历史故障匹配结果
                full_message = f"整合日志分类:\n{message}\n\n{historical_match_message}"

                feishu_client.send_message("整合日志分析结果", full_message)
            except KeyError as e:
                logging.error(f"返回结果中缺少必要的键: {e}")
        else:
            logging.error("日志分析失败。")
    else:
        logging.warning("未从Elasticsearch获取到日志。")


if __name__ == "__main__":
    main()

四、见证奇迹：代码运行结果

系统从ELK收集的日志中筛选数据，借助DeepSeek强大的分析能力进行分类打标，再与历史故障库进行匹配，最后通过飞书机器人将详细的分析结果和匹配情况精准推送给我们。

推送到飞书的消息效果

• 截图：

DeepSeek日志分析告警推送
来自Deepseek的消息:
整合日志分析结果
deepseek分析结果:
整合日志分类:
根据提供的日志内容，我将按照故障类型进行分类，并评估影响范围（P0-P3），同时给出相应的建议。

### 1. 网络异常
   - 日志内容:
     ```
     2025-03-11 10:37:48,139 网络连接异常开始
     2025-03-11 10:37:48,139 网络连接异常结束
     ```
   - 影响范围: P2
     - 解释: 网络连接异常可能会导致服务短暂中断或延迟，但日志显示异常持续时间较短，且没有明确的服务中断报告，因此影响范围评估为P2（中等影响）。
   - 建议:
     - 检查网络设备的健康状况，确保网络设备（如交换机、路由器等）正常运行。
     - 监控网络流量，排查是否存在网络拥塞或异常流量。
     - 如果频繁出现网络连接异常，建议进一步排查网络配置或与网络服务提供商联系。

### 2. Redis异常
   - 日志内容:
     ```
     2025-03-11 10:36:03,200 [lettuce-epollEventLoop-4-4] INFO  io.lettuce.core.protocol.ReconnectionHandler:177 - Reconnected to 192.168.103.191:6379
     2025-03-11 10:36:03,197 [lettuce-eventExecutorLoop-1-2] INFO  io.lettuce.core.protocol.ConnectionWatchdog:173 - Reconnecting, last destination was 192.168.103.193/192.168.103.193:6379
     2025-03-11 10:36:43,096 [lettuce-eventExecutorLoop-1-3] INFO  io.lettuce.core.protocol.ConnectionWatchdog:173 - Reconnecting, last destination was 192.168.103.193/192.168.103.193:6379
     2025-03-11 10:35:05,100 [lettuce-epollEventLoop-4-3] INFO  io.lettuce.core.protocol.ReconnectionHandler:177 - Reconnected to 192.168.103.195:6379
     2025-03-11 10:36:43,099 [lettuce-epollEventLoop-4-1] INFO  io.lettuce.core.protocol.ReconnectionHandler:177 - Reconnected to 192.168.103.193:6379
     2025-03-11 10:31:02,100 [lettuce-epollEventLoop-4-1] INFO  io.lettuce.core.protocol.ReconnectionHandler:177 - Reconnected to 192.168.103.193:6379
     ```
   - 影响范围: P1
     - 解释: Redis连接频繁断开并重连，可能会导致缓存服务短暂不可用，影响依赖Redis的服务。虽然重连成功，但频繁的重连可能会影响系统性能，因此评估为P1（较高影响）。
   - 建议:
     - 检查Redis服务器的负载情况，确保Redis服务器有足够的资源（CPU、内存等）。
     - 检查Redis客户端配置，确保连接池配置合理，避免频繁重连。
     - 监控Redis服务器的网络连接，排查是否存在网络抖动或延迟问题。
     - 如果Redis集群中有多个节点，检查集群的健康状态，确保所有节点正常运行。

### 3. MQ异常
   - 日志内容:
     ```
     2025-03-11 10:37:39,272 [SimpleAsyncTaskExecutor-5] ERROR  cn.com.easypay.aspect.WebLogAspect:39 - mq请求异常开始
     2025-03-11 10:37:39,272 [SimpleAsyncTaskExecutor-5] ERROR  cn.com.easypay.aspect.WebLogAspect:39 - mq请求异常结束
     ```
   - 影响范围: P2
     - 解释: MQ请求异常可能会导致消息处理延迟或失败，但日志显示异常持续时间较短，且没有明确的服务中断报告，因此评估为P2（中等影响）。
   - 建议:
     - 检查MQ服务器的健康状况，确保MQ服务器正常运行。
     - 检查MQ客户端的配置，确保连接和重试机制配置合理。
     - 监控MQ的队列积压情况，确保消息能够及时处理。
     - 如果MQ集群中有多个节点，检查集群的健康状态，确保所有节点正常运行。

### 总结：
- 网络异常：影响范围P2，建议检查网络设备和流量。
- Redis异常：影响范围P1，建议检查Redis服务器和客户端配置。
- MQ异常：影响范围P2，建议检查MQ服务器和客户端配置。

建议优先处理Redis异常，因为其影响范围较大，可能会对系统性能产生较大影响。网络异常和MQ异常虽然影响较小，但也需要持续监控和排查，避免问题扩大。

历史故障库匹配结果:
日志内容: 2025-03-11 10:37:48,139 网络连接异常结束
 匹配故障: 2024年10月28日 网络连接异常历史故障匹配验证
日志内容: 2025-03-11 10:37:48,139 网络连接异常开始
 匹配故障: 2024年10月28日 网络连接异常历史故障匹配验证

说在最后

这个方案仅仅只是提供一种利用DeepSeek分析日志的思路。当然，或许还不够完美，但我相信，它能起到抛砖引玉的作用。