# 安装 Vector
curl --proto '=https' --tlsv1.2 -sSf https://sh.vector.dev | sh

# 验证安装
vector --version

# ============================================
# 数据源：收集各类日志
# ============================================

# 系统日志
[sources.system_logs]
type = "journald"
include_units = ["nginx", "docker", "postgres", "app"]

# Nginx 访问日志
[sources.nginx_access]
type = "file"
include = ["/var/log/nginx/access.log"]
read_from = "beginning"
data_format = "json"

# Nginx 错误日志
[sources.nginx_error]
type = "file"
include = ["/var/log/nginx/error.log"]
read_from = "beginning"
data_format = "newline_delimited"

# Docker 容器日志
[sources.docker_logs]
type = "docker_logs"

# PostgreSQL 慢查询日志
[sources.postgres_slow]
type = "file"
include = ["/var/log/postgresql/postgresql-*-slow.log"]
read_from = "beginning"

# ============================================
# 转换：丰富日志数据
# ============================================

[transforms.enrich_logs]
type = "remap"
inputs = ["system_logs", "nginx_access", "nginx_error", "docker_logs", "postgre_slow"]
source = '''
  .log_level = default(., "INFO")
  .hostname = gethostname!()
  .collected_at = now()
  
  # 从消息中提取关键信息
  if contains(.message, "error") || contains(.message, "Error") || contains(.message, "ERROR") {
    .log_level = "ERROR"
  } else if contains(.message, "warn") || contains(.message, "Warn") || contains(.message, "WARN") {
    .log_level = "WARN"
  } else if contains(.message, "fatal") || contains(.message, "FATAL") {
    .log_level = "FATAL"
  }
'''

# ============================================
# 输出：发送到多个目的地
# ============================================

# 输出到 Vector 内部 API（供 AI 引擎消费）
[outputs.api_pipe]
type = "socket"
address = "127.0.0.1:9000"
mode = "tcp"
encoding.codec = "json"

# 输出到本地文件备份
[outputs.file_backup]
type = "file"
path = "/var/log/vector-backup/%Y-%m-%d.json"
encoding.codec = "json"

sudo vector --config /etc/vector/vector.toml

# 检查 Vector 状态
curl -s http://localhost:8686/metrics | grep vector_processing

# 测试接收日志
echo '{"message": "test log entry", "level": "INFO"}' \
  | nc localhost 9000

pip install sentence-transformers qdrant-client ollama requests pydantic

#!/usr/bin/env python3
"""
AI 日志向量化服务
将日志转换为向量并存储到 Qdrant
"""

import json
import socket
import logging
from datetime import datetime
from pathlib import Path
from typing import List, Dict
import numpy as np

try:
    from sentence_transformers import SentenceTransformer
except ImportError:
    print("安装 sentence-transformers: pip install sentence-transformers")
    exit(1)

try:
    from qdrant_client import QdrantClient
    from qdrant_client.models import Distance, PointStruct, VectorParams, Filter, FieldCondition, MatchValue
except ImportError:
    print("安装 qdrant-client: pip install qdrant-client")
    exit(1)

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger("log-vectorizer")

# ============================================
# 配置
# ============================================

COLLECTION_NAME = "vps_logs"
EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2"  # 轻量级中文友好
QDRANT_HOST = "localhost"
QDRANT_PORT = 6333
SOCKET_PORT = 9000

# 日志分类映射
LOG_CATEGORIES = {
    "nginx": ["access", "error", "404", "502", "503", "timeout"],
    "docker": ["container", "restart", "oom", "healthcheck"],
    "system": ["kernel", "oom", "disk", "memory", "cpu"],
    "database": ["connection", "slow", "deadlock", "replication"],
    "security": ["authentication", "permission", "denied", "unauthorized"],
}


class LogVectorizer:
    """日志向量化器"""
    
    def __init__(self):
        # 加载嵌入模型
        logger.info(f"加载嵌入模型: {EMBEDDING_MODEL}")
        self.model = SentenceTransformer(EMBEDDING_MODEL)
        
        # 连接 Qdrant
        self.client = QdrantClient(host=QDRANT_HOST, port=QDRANT_PORT)
        
        # 确保集合存在
        self._ensure_collection()
        
        # 统计信息
        self.stats = {
            "total_processed": 0,
            "errors": 0,
            "anomalies_detected": 0,
        }
    
    def _ensure_collection(self):
        """确保 Qdrant 集合存在"""
        collections = [c.name for c in self.client.get_collections().collections]
        if COLLECTION_NAME not in collections:
            self.client.create_collection(
                collection_name=COLLECTION_NAME,
                vectors_config=VectorParams(
                    size=384,  # all-MiniLM-L6-v2 输出维度
                    distance=Distance.COSINE
                ),
            )
            logger.info(f"创建集合: {COLLECTION_NAME}")
    
    def classify_log(self, message: str) -> str:
        """简单分类日志类型"""
        msg_lower = message.lower()
        for category, keywords in LOG_CATEGORIES.items():
            if any(kw in msg_lower for kw in keywords):
                return category
        return "general"
    
    def embed_log(self, log_entry: dict) -> dict:
        """将单条日志转换为向量并存储"""
        try:
            message = log_entry.get("message", "")
            if not message or len(message) < 5:
                return {"status": "skipped", "reason": "too_short"}
            
            # 生成嵌入向量
            embedding = self.model.encode(message).tolist()
            
            # 分类
            category = self.classify_log(message)
            
            # 构造元数据
            payload = {
                "message": message,
                "timestamp": log_entry.get("collected_at", datetime.now().isoformat()),
                "log_level": log_entry.get("log_level", "INFO"),
                "category": category,
                "hostname": log_entry.get("hostname", ""),
            }
            
            # 存入 Qdrant
            point = PointStruct(
                id=self.stats["total_processed"],
                vector=embedding,
                payload=payload,
            )
            
            self.client.upsert(
                collection_name=COLLECTION_NAME,
                points=[point],
            )
            
            self.stats["total_processed"] += 1
            
            # 异常检测（简单规则）
            if self._is_anomalous(log_entry):
                self.stats["anomalies_detected"] += 1
                logger.warning(f"检测到异常日志: {message[:100]}...")
            
            return {"status": "ok", "category": category}
        
        except Exception as e:
            self.stats["errors"] += 1
            logger.error(f"处理日志失败: {e}")
            return {"status": "error", "error": str(e)}
    
    def _is_anomalous(self, log_entry: dict) -> bool:
        """简单异常检测规则"""
        message = log_entry.get("message", "").lower()
        level = log_entry.get("log_level", "INFO").upper()
        
        # 严重级别日志
        if level in ["FATAL", "CRITICAL"]:
            return True
        
        # 关键词匹配
        anomaly_keywords = [
            "out of memory", "oom-killer", "segfault",
            "connection refused", "connection timeout",
            "disk full", "no space left",
            "panic", "kernel panic",
            "too many open files",
            "certificate expired",
        ]
        return any(kw in message for kw in anomaly_keywords)
    
    def run_server(self):
        """启动 TCP 日志接收服务"""
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind(("127.0.0.1", SOCKET_PORT))
        server.listen(100)
        
        logger.info(f"日志接收服务已启动，监听端口 {SOCKET_PORT}")
        logger.info(f"已处理 {self.stats['total_processed']} 条日志，"
                     f"检测到 {self.stats['anomalies_detected']} 条异常")
        
        buffer = b""
        while True:
            try:
                conn, addr = server.accept()
                with conn:
                    while True:
                        data = conn.recv(4096)
                        if not data:
                            break
                        buffer += data
                        while b"\n" in buffer:
                            line, buffer = buffer.split(b"\n", 1)
                            try:
                                log_entry = json.loads(line.decode("utf-8"))
                                result = self.embed_log(log_entry)
                                if result["status"] == "ok":
                                    logger.debug(f"已处理日志 [{result['category']}]")
                            except json.JSONDecodeError:
                                pass
            except Exception as e:
                logger.error(f"连接错误: {e}")


if __name__ == "__main__":
    vectorizer = LogVectorizer()
    vectorizer.run_server()

# 使用 Docker 部署 Qdrant
docker run -d \
  --name qdrant \
  --restart unless-stopped \
  -p 6333:6333 \
  -p 6334:6334 \
  -v $(pwd)/qdrant_storage:/qdrant/storage:z \
  qdrant/qdrant:latest

# 安装 Ollama
curl -fsSL https://ollama.com/install.sh | sh

# 拉取适合分析的模型
ollama pull qwen2.5:7b

# 创建日志分析专用模型
cat > /etc/ollama/modelfiles/log-analyzer << 'EOF'
FROM qwen2.5:7b

SYSTEM """你是一个专业的运维日志分析师。你的职责是：
1. 分析日志内容，识别异常模式和潜在问题
2. 关联多条日志，推断根本原因
3. 按严重程度排序问题
4. 提供可操作的修复建议

输出格式为 JSON：
{
  "summary": "简要总结",
  "issues": [
    {
      "severity": "critical|high|medium|low",
      "type": "问题类型",
      "description": "详细描述",
      "root_cause": "可能的根本原因",
      "recommendation": "修复建议",
      "related_logs": ["相关日志片段"]
    }
  ],
  "action_items": ["需要立即执行的操作"]
}

只输出 JSON，不要其他内容。"""
EOF

ollama create log-analyzer -f /etc/ollama/modelfiles/log-analyzer

#!/usr/bin/env python3
"""
AI 日志分析引擎
提供查询、异常检测和根因分析功能
"""

import json
import os
import time
import logging
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Dict, Optional

try:
    from qdrant_client import QdrantClient
    from qdrant_client.models import (
        Distance, FieldCondition, MatchValue, MatchText,
        Filter, PointStruct, SearchRequest, VectorParams
    )
    import ollama
except ImportError:
    print("安装依赖: pip install qdrant-client ollama")
    exit(1)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("log-analyzer")

COLLECTION_NAME = "vps_logs"
OLLAMA_MODEL = "log-analyzer"


class LogAnalyzer:
    """AI 日志分析器"""
    
    def __init__(self):
        self.qdrant = QdrantClient(host="localhost", port=6333)
        self.query_history = []
    
    # -----------------------------------------------
    # 核心功能 1：自然语言日志查询
    # -----------------------------------------------
    
    def semantic_search(self, query: str, limit: int = 20) -> List[dict]:
        """
        使用语义搜索查找与查询相关的日志
        
        示例用法：
            analyzer = LogAnalyzer()
            results = analyzer.semantic_search("昨天的数据库超时错误")
        """
        try:
            # 这里需要一个嵌入模型来转换查询
            from sentence_transformers import SentenceTransformer
            model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
            query_vector = model.encode(query).tolist()
            
            results = self.qdrant.search(
                collection_name=COLLECTION_NAME,
                query_vector=query_vector,
                limit=limit,
            )
            
            return [
                {
                    "message": r.payload["message"],
                    "timestamp": r.payload["timestamp"],
                    "log_level": r.payload["log_level"],
                    "category": r.payload["category"],
                    "score": r.score,
                }
                for r in results
            ]
        
        except Exception as e:
            logger.error(f"搜索失败: {e}")
            return []
    
    # -----------------------------------------------
    # 核心功能 2：异常日志检测
    # -----------------------------------------------
    
    def detect_anomalies(self, hours: int = 24, threshold: float = 0.85) -> List[dict]:
        """
        检测指定时间范围内的异常日志
        
        通过向量相似度分析，识别偏离正常模式的日志
        """
        try:
            # 获取最近 N 小时的日志
            cutoff = datetime.now() - timedelta(hours=hours)
            
            results = self.qdrant.scroll(
                collection_name=COLLECTION_NAME,
                scroll_filter=Filter(
                    must=[
                        FieldCondition(
                            key="timestamp",
                            range=None  # 实际使用时需设置 datetime 范围
                        )
                    ]
                ),
                limit=1000,
                with_payload=True,
            )
            
            messages = [r.payload["message"] for r in results[0]]
            if not messages:
                return []
            
            # 使用 LLM 分析日志模式
            prompt = f"""请分析以下最近 {hours} 小时的日志，识别异常模式：

{chr(10).join(messages[:50])}

请返回 JSON 格式的分析结果：
{{
  "anomalies": [
    {{
      "pattern": "发现的异常模式描述",
      "affected_services": ["受影响的服务列表"],
      "severity": "critical|high|medium|low",
      "sample_logs": ["代表性日志"],
      "recommendation": "建议操作"
    }}
  ],
  "overall_health": "系统健康评估"
}}"""
            
            response = ollama.generate(
                model=OLLAMA_MODEL,
                prompt=prompt,
                options={"temperature": 0.1}
            )
            
            # 解析 JSON 响应
            output = response.get("response", "")
            json_start = output.find("{")
            json_end = output.rfind("}") + 1
            if json_start >= 0 and json_end > json_start:
                analysis = json.loads(output[json_start:json_end])
                return analysis.get("anomalies", [])
            
            return []
        
        except Exception as e:
            logger.error(f"异常检测失败: {e}")
            return []
    
    # -----------------------------------------------
    # 核心功能 3：根因分析
    # -----------------------------------------------
    
    def root_cause_analysis(self, incident_description: str, 
                           context_hours: int = 4) -> dict:
        """
        对特定事件进行根因分析
        
        示例：
            result = analyzer.root_cause_analysis(
                "Nginx 在下午 2 点开始返回 502 错误",
                context_hours=4
            )
        """
        # 获取上下文日志
        search_results = self.semantic_search(
            f"Nginx 502 error response timeout",
            limit=50
        )
        
        context_logs = [r["message"] for r in search_results]
        
        # 构建分析提示
        prompt = f"""你是一名资深 SRE 工程师。请分析以下事故：

**事件描述**: {incident_description}
**时间范围**: 过去 {context_hours} 小时

**相关日志**:
{chr(10).join(context_logs[:30])}

请进行根因分析，返回 JSON：
{{
  "incident_summary": "事故概述",
  "timeline": [
    {{
      "time": "时间点",
      "event": "事件描述",
      "impact": "影响范围"
    }}
  ],
  "root_cause": "根本原因分析",
  "contributing_factors": [" contributing factors"],
  "immediate_actions": ["立即执行的修复步骤"],
  "preventive_measures": ["长期预防措施"],
  "confidence": 0.0-1.0
}}"""
        
        try:
            response = ollama.generate(
                model=OLLAMA_MODEL,
                prompt=prompt,
                options={"temperature": 0.1, "num_predict": 1000}
            )
            
            output = response.get("response", "")
            json_start = output.find("{")
            json_end = output.rfind("}") + 1
            if json_start >= 0 and json_end > json_start:
                return json.loads(output[json_start:json_end])
            
            return {"error": "Failed to parse analysis"}
        
        except Exception as e:
            return {"error": str(e)}
    
    # -----------------------------------------------
    # 核心功能 4：日志摘要生成
    # -----------------------------------------------
    
    def generate_daily_report(self, date_str: str = None) -> str:
        """
        生成每日日志摘要报告
        
        示例：
            report = analyzer.generate_daily_report("2026-07-11")
            print(report)
        """
        if date_str is None:
            date_str = datetime.now().strftime("%Y-%m-%d")
        
        # 获取当日关键日志
        error_logs = self.semantic_search(
            f"{date_str} error critical fatal",
            limit=30
        )
        
        warning_logs = self.semantic_search(
            f"{date_str} warning warn",
            limit=20
        )
        
        # 构建报告
        report = f"""
# 📊 每日日志分析报告 — {date_str}

## 概览
- 错误日志: {len(error_logs)} 条
- 警告日志: {len(warning_logs)} 条

## 🔴 错误日志摘要
"""
        for log in error_logs[:10]:
            report += f"- [{log['log_level']}] {log['message'][:150]}\n"
        
        report += "\n## ⚠️ 警告日志摘要\n"
        for log in warning_logs[:10]:
            report += f"- [{log['log_level']}] {log['message'][:150]}\n"
        
        report += "\n## 💡 AI 建议\n"
        report += "基于以上日志分析，建议关注以下方面...\n"
        
        return report
    
    # -----------------------------------------------
    # 统计功能
    # -----------------------------------------------
    
    def get_statistics(self) -> dict:
        """获取日志统计信息"""
        count = self.qdrant.count(collection_name=COLLECTION_NAME)
        
        # 按级别统计
        level_stats = {}
        for level in ["INFO", "WARN", "ERROR", "FATAL", "DEBUG"]:
            results = self.qdrant.scroll(
                collection_name=COLLECTION_NAME,
                scroll_filter=Filter(must=[
                    FieldCondition(
                        key="log_level",
                        match=MatchValue(value=level)
                    )
                ]),
                limit=1,
                with_payload=False,
            )
            # 简化：实际应使用 count API
            level_stats[level] = 0
        
        return {
            "total_logs": count.count,
            "collection": COLLECTION_NAME,
            "level_breakdown": level_stats,
            "generated_at": datetime.now().isoformat(),
        }


# CLI 入口
if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="AI 日志分析工具")
    subparsers = parser.add_subparsers(dest="command")
    
    # 搜索子命令
    search_parser = subparsers.add_parser("search", help="语义搜索日志")
    search_parser.add_argument("query", help="搜索查询")
    search_parser.add_argument("--limit", type=int, default=20)
    
    # 异常检测子命令
    anomaly_parser = subparsers.add_parser("anomaly", help="检测异常")
    anomaly_parser.add_argument("--hours", type=int, default=24)
    
    # 根因分析子命令
    rca_parser = subparsers.add_parser("rca", help="根因分析")
    rca_parser.add_argument("incident", help="事件描述")
    rca_parser.add_argument("--hours", type=int, default=4)
    
    # 日报子命令
    report_parser = subparsers.add_parser("report", help="生成日报")
    report_parser.add_argument("--date", help="日期 YYYY-MM-DD")
    
    args = parser.parse_args()
    analyzer = LogAnalyzer()
    
    if args.command == "search":
        results = analyzer.semantic_search(args.query, args.limit)
        print(json.dumps(results, indent=2, ensure_ascii=False))
    
    elif args.command == "anomaly":
        anomalies = analyzer.detect_anomalies(args.hours)
        print(json.dumps(anomalies, indent=2, ensure_ascii=False))
    
    elif args.command == "rca":
        result = analyzer.root_cause_analysis(args.incident, args.hours)
        print(json.dumps(result, indent=2, ensure_ascii=False))
    
    elif args.command == "report":
        report = analyzer.generate_daily_report(args.date)
        print(report)
    
    else:
        parser.print_help()

# Grafana Dashboard JSON 片段
{
  "dashboard": {
    "title": "AI 日志分析总览",
    "panels": [
      {
        "title": "日志级别分布",
        "type": "piechart",
        "targets": [
          {
            "expr": "count by (level) (log_messages_total)"
          }
        ]
      },
      {
        "title": "异常趋势",
        "type": "graph",
        "targets": [
          {
            "expr": "increase(log_anomalies_total[1h])"
          }
        ]
      },
      {
        "title": "最近异常事件",
        "type": "table",
        "datasource": "Qdrant"
      }
    ]
  }
}

#!/usr/bin/env python3
"""
简易日志查询 Web UI
"""

from flask import Flask, request, jsonify
from analyzer import LogAnalyzer
import threading

app = Flask(__name__)
analyzer = LogAnalyzer()


@app.route("/api/search", methods=["POST"])
def search():
    """语义搜索接口"""
    data = request.json
    query = data.get("query", "")
    limit = data.get("limit", 20)
    
    results = analyzer.semantic_search(query, limit)
    return jsonify({"results": results, "count": len(results)})


@app.route("/api/anomaly", methods=["GET"])
def anomalies():
    """异常检测接口"""
    hours = int(request.args.get("hours", 24))
    results = analyzer.detect_anomalies(hours)
    return jsonify({"anomalies": results})


@app.route("/api/rca", methods=["POST"])
def root_cause():
    """根因分析接口"""
    data = request.json
    incident = data.get("incident", "")
    hours = data.get("hours", 4)
    
    result = analyzer.root_cause_analysis(incident, hours)
    return jsonify(result)


@app.route("/api/report", methods=["GET"])
def report():
    """日报生成接口"""
    date = request.args.get("date")
    report_text = analyzer.generate_daily_report(date)
    return jsonify({"report": report_text})


@app.route("/api/stats", methods=["GET"])
def stats():
    """统计信息"""
    return jsonify(analyzer.get_statistics())


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8500, debug=False)

# 安装 Flask
pip install flask

# 启动
python3 /opt/log-analyzer/app.py &

# 查询昨天的数据库错误
curl -X POST http://localhost:8500/api/search \
  -H "Content-Type: application/json" \
  -d '{"query": "数据库连接超时错误", "limit": 10}'

# 查询安全相关事件
curl -X POST http://localhost:8500/api/search \
  -H "Content-Type: application/json" \
  -d '{"query": "SSH 登录失败", "limit": 20}'

# 对 Nginx 502 问题进行根因分析
curl -X POST http://localhost:8500/api/rca \
  -H "Content-Type: application/json" \
  -d '{
    "incident": "Nginx 在下午 2 点到 3 点之间大量返回 502 Bad Gateway",
    "hours": 4
  }'

{
  "incident_summary": "Nginx 在 14:00-15:00 期间出现 502 错误高峰",
  "timeline": [
    {
      "time": "13:45",
      "event": "后端应用重启",
      "impact": "短暂服务中断"
    },
    {
      "time": "14:00",
      "event": "502 错误开始出现",
      "impact": "用户可见错误"
    },
    {
      "time": "14:30",
      "event": "后端进程内存持续增长",
      "impact": "性能下降"
    },
    {
      "time": "15:00",
      "event": "服务恢复正常",
      "impact": "自动恢复"
    }
  ],
  "root_cause": "后端应用存在内存泄漏，导致在高负载下进程崩溃重启",
  "contributing_factors": [
    "未设置合理的内存限制",
    "缺少健康检查自动重启机制",
    "监控告警延迟"
  ],
  "immediate_actions": [
    "检查后端应用的内存使用情况",
    "临时增加后端实例数量分担负载",
    "审查最近代码变更"
  ],
  "preventive_measures": [
    "添加内存使用监控告警（阈值 80%）",
    "实施自动重启策略",
    "定期压力测试"
  ],
  "confidence": 0.87
}

# 每天早上 9 点自动生成日报并发送
crontab -e

# 添加以下行
0 9 * * * /usr/local/bin/generate_log_report.sh

#!/bin/bash
# 生成昨日日志日报
DATE=$(date -d "yesterday" +%Y-%m-%d)

REPORT=$(curl -s http://localhost:8500/api/report?date=$DATE | \
  jq -r '.report')

# 发送到 Telegram / Slack / 邮件
curl -X POST "https://api.telegram.org/bot${TG_BOT_TOKEN}/sendMessage" \
  -d "chat_id=${TG_CHAT_ID}" \
  -d "text=${REPORT}" \
  -d "parse_mode=Markdown"

# 替换为中文优化的嵌入模型
EMBEDDING_MODEL = "shibing624/text2vec-base-chinese"

def feedback_loop(anomaly_result: dict, user_feedback: str):
    """
    用户反馈闭环：标记误报或确认漏报，持续优化检测规则
    
    user_feedback: "false_positive" | "true_positive" | "missed"
    """
    # 记录反馈
    feedback_record = {
        "timestamp": datetime.now().isoformat(),
        "anomaly": anomaly_result,
        "feedback": user_feedback,
    }
    
    # 根据反馈调整检测阈值
    if user_feedback == "false_positive":
        # 降低该类别的敏感度
        adjust_threshold(anomaly_result["category"], delta=-0.05)
    elif user_feedback == "missed":
        # 提高该类别的敏感度
        adjust_threshold(anomaly_result["category"], delta=+0.05)
    
    # 存储反馈用于后续训练
    save_feedback(feedback_record)

def cross_host_analysis(hosts: list, event_timeframe: str):
    """
    跨主机日志关联分析
    
    hosts: ["web-01", "db-01", "cache-01"]
    timeframe: "2026-07-11 14:00:00 to 2026-07-11 15:00:00"
    """
    # 收集所有主机的相关日志
    all_logs = []
    for host in hosts:
        logs = fetch_logs(host, timeframe)
        all_logs.extend(logs)
    
    # 使用 LLM 关联分析
    prompt = f"""分析以下来自多台主机的日志，找出跨主机的关联事件：

{chr(10).join(all_logs[:100])}

请识别：
1. 是否存在跨主机的连锁故障？
2. 哪个主机是最先出问题的？
3. 故障传播路径是什么？"""
    
    analysis = ollama.generate(model=OLLAMA_MODEL, prompt=prompt)
    return analysis