MonkeyCode实现日记分析体系：从日记网络到实时告警

日记是体系的黑匣子

线上出标题，第一反应就是看日记。但日记散落在各台服务器，grep查起来慢，关键错误被沉没在海量日记里……
这篇文章用MonkeyCode构建一个完备的日记分析体系：网络→存储→分析→告警，全流程主动化。
体系架构

1. [应用服务器] --Fluentd--> [Kafka] --> [Logstash] --> [Elasticsearch] --> [Kibana]
2.                                                     |
3.                                                     --> [告警服务] --> [Slack/企微]

复制代码

给MonkeyCode的Prompt

1. 构建日志分析系统，要求：
2. 1. 日志收集：Fluentd配置（收集Nginx、应用日志）
3. 2. 消息队列：Kafka（解耦收集和处理）
4. 3. 日志处理：Logstash（解析、过滤、转换）
5. 4. 存储：Elasticsearch（索引、搜索）
6. 5. 可视化：Kibana仪表板
7. 6. 告警规则：错误率、响应时间、异常关键字
8. 7. 告警通知：Webhook发送到Slack
10. 日志格式：JSON（包含timestamp, level, service, message, trace_id）

复制代码

1. Fluentd设置

1. # fluentd/fluent.conf - MonkeyCode生成
3. # Nginx访问日志
4. <source>
5.   @type tail
6.   path /var/log/nginx/access.log
7.   pos_file /var/log/td-agent/nginx-access.log.pos
8.   tag nginx.access
9.   format nginx
10. </source>
12. # Nginx错误日志
13. <source>
14.   @type tail
15.   path /var/log/nginx/error.log
16.   pos_file /var/log/td-agent/nginx-error.log.pos
17.   tag nginx.error
18.   format regexp
19.   expression /^\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2} \[(\w+)\] (\d+#\d+): (.*)$/
20. </source>
22. # 应用JSON日志
23. <source>
24.   @type tail
25.   path /var/log/app/*.log
26.   pos_file /var/log/td-agent/app.log.pos
27.   tag app
28.   format json
29.   time_key timestamp
30.   time_format %Y-%m-%dT%H:%M:%S.%NZ
31. </source>
33. # 发送到Kafka
34. <match **>
35.   @type kafka2
36.   
37.   brokers kafka1:9092,kafka2:9092,kafka3:9092
38.   topic_key topic
39.   default_topic logs
40.   
41.   <format>
42.     @type json
43.   </format>
44.   
45.   <buffer>
46.     @type file
47.     path /var/log/td-agent/buffer
48.     flush_interval 5s
49.     chunk_limit_size 16MB
50.     queue_limit_length 1024
51.   </buffer>
52. </match>

复制代码

2. Kafka设置

1. # kafka/docker-compose.yml
2. version: '3.8'
3. services:
4.   zookeeper:
5.     image: confluentinc/cp-zookeeper:7.3.0
6.     environment:
7.       ZOOKEEPER_CLIENT_PORT: 2181
8.       ZOOKEEPER_TICK_TIME: 2000
9.     volumes:
10.       - zookeeper-data:/var/lib/zookeeper/data
12.   kafka:
13.     image: confluentinc/cp-kafka:7.3.0
14.     depends_on:
15.       - zookeeper
16.     ports:
17.       - "9092:9092"
18.     environment:
19.       KAFKA_BROKER_ID: 1
20.       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
21.       KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
22.       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
23.       KAFKA_LOG_RETENTION_HOURS: 24
24.       KAFKA_LOG_SEGMENT_BYTES: 1073741824
25.       KAFKA_RETENTION_CHECK_INTERVAL_MS: 300000
26.     volumes:
27.       - kafka-data:/var/lib/kafka/data
29. volumes:
30.   zookeeper-data:
31.   kafka-data:

复制代码

3. Logstash管道

1. # logstash/pipeline/logs.conf - MonkeyCode生成
3. input {
4.   kafka {
5.     bootstrap_servers => "kafka1:9092,kafka2:9092"
6.     topics => ["logs"]
7.     group_id => "logstash"
8.     consumer_threads => 4
9.     decorate_events => true
10.   }
11. }
13. filter {
14.   # 解析Nginx访问日志
15.   if "nginx.access" in [tags] {
16.     grok {
17.       match => { "message" => "%{COMBINEDAPACHELOG}" }
18.     }
19.     date {
20.       match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
21.     }
22.    
23.     # 提取响应时间
24.     grok {
25.       match => { "message" => "rt=(?<response_time>\d+\.\d+)" }
26.     }
27.    
28.     # 地理位置解析
29.     geoip {
30.       source => "clientip"
31.       target => "geoip"
32.     }
33.    
34.     # User-Agent解析
35.     useragent {
36.       source => "agent"
37.       target => "useragent"
38.     }
39.   }
40.   
41.   # 解析应用JSON日志
42.   if [level] {
43.     # 标准化日志级别
44.     mutate {
45.       replace => { "log_level" => "%{level}" }
46.     }
47.    
48.     # 错误日志特殊处理
49.     if [level] == "ERROR" or [level] == "FATAL" {
50.       mutate {
51.         add_tag => ["error"]
52.       }
53.       
54.       # 提取异常栈（如果有）
55.       if [exception] {
56.         grok {
57.           match => { "exception" => "(?<exception_class>\w+\.\w+Exception): (?<exception_message>.*)" }
58.         }
59.       }
60.     }
61.   }
62.   
63.   # 添加服务标识
64.   mutate {
65.     add_field => { "environment" => "production" }
66.     add_field => { "hostname" => "%{host}" }
67.   }
68.   
69.   # 移除不需要的字段
70.   mutate {
71.     remove_field => [ "agent", "auth" ]
72.   }
73. }
75. output {
76.   # 输出到Elasticsearch
77.   elasticsearch {
78.     hosts => ["http://elasticsearch:9200"]
79.     index => "logs-%{[service]}-%{+YYYY.MM.dd}"
80.     template_name => "logs"
81.     template_overwrite => true
82.   }
83.   
84.   # 错误日志同时发送到告警主题
85.   if "error" in [tags] {
86.     kafka {
87.       bootstrap_servers => "kafka1:9092"
88.       topic_id => "alerts"
89.     }
90.   }
91. }

复制代码

4. Elasticsearch索引模板

1. // elasticsearch/templates/logs.json
2. {
3.   "index_patterns": ["logs-*"],
4.   "settings": {
5.     "number_of_shards": 3,
6.     "number_of_replicas": 1,
7.     "refresh_interval": "5s"
8.   },
9.   "mappings": {
10.     "properties": {
11.       "@timestamp": { "type": "date" },
12.       "level": { "type": "keyword" },
13.       "service": { "type": "keyword" },
14.       "trace_id": { "type": "keyword" },
15.       "message": { "type": "text", "analyzer": "standard" },
16.       "exception": { "type": "text", "index": false },
17.       "hostname": { "type": "keyword" },
18.       "environment": { "type": "keyword" },
19.       "response_time": { "type": "float" },
20.       "status": { "type": "integer" },
21.       "request": {
22.         "properties": {
23.           "method": { "type": "keyword" },
24.           "path": { "type": "keyword" },
25.           "params": { "type": "object", "enabled": false }
26.         }
27.       },
28.       "user": {
29.         "properties": {
30.           "id": { "type": "keyword" },
31.           "ip": { "type": "ip" }
32.         }
33.       },
34.       "geoip": {
35.         "properties": {
36.           "country": { "type": "keyword" },
37.           "city": { "type": "keyword" },
38.           "location": { "type": "geo_point" }
39.         }
40.       }
41.     }
42.   }
43. }

复制代码

5. 告警服务

1. # alert_service.py - MonkeyCode生成
2. import json
3. import requests
4. import time
5. from kafka import KafkaConsumer
6. from elasticsearch import Elasticsearch
7. from datetime import datetime, timedelta
8. from dataclasses import dataclass
9. from typing import List, Dict, Any
10. import logging
12. logging.basicConfig(level=logging.INFO)
13. logger = logging.getLogger(__name__)
15. @dataclass
16. class AlertRule:
17.     name: str
18.     condition: dict  # ES查询条件
19.     threshold: int
20.     time_window: int  # 秒
21.     severity: str  # critical/warning/info
22.     channels: List[str]  # ['slack', 'wechat']
24. class AlertService:
25.     """日志告警服务"""
26.    
27.     def __init__(self, es_hosts: list, kafka_servers: list):
28.         self.es = Elasticsearch(es_hosts)
29.         self.consumer = KafkaConsumer(
30.             'alerts',
31.             bootstrap_servers=kafka_servers,
32.             value_deserializer=lambda m: json.loads(m.decode('utf-8'))
33.         )
34.         self.alert_rules = self._load_rules()
35.         self.alert_cooldown = {}  # 防止重复告警
36.    
37.     def _load_rules(self) -> List[AlertRule]:
38.         """加载告警规则"""
39.         return [
40.             # 规则1：5分钟内错误超过100次
41.             AlertRule(
42.                 name='high_error_rate',
43.                 condition={'term': {'level': 'ERROR'}},
44.                 threshold=100,
45.                 time_window=300,
46.                 severity='critical',
47.                 channels=['slack', 'wechat']
48.             ),
49.             # 规则2：5分钟内500响应超过50次
50.             AlertRule(
51.                 name='high_500_rate',
52.                 condition={'term': {'status': 500}},
53.                 threshold=50,
54.                 time_window=300,
55.                 severity='critical',
56.                 channels=['slack']
57.             ),
58.             # 规则3：P99响应时间超过2秒
59.             AlertRule(
60.                 name='slow_response',
61.                 condition={'range': {'response_time': {'gt': 2}}},
62.                 threshold=10,
63.                 time_window=60,
64.                 severity='warning',
65.                 channels=['slack']
66.             ),
67.             # 规则4：特定异常出现
68.             AlertRule(
69.                 name='critical_exception',
70.                 condition={'match': {'exception_class': 'OutOfMemoryError'}},
71.                 threshold=1,
72.                 time_window=60,
73.                 severity='critical',
74.                 channels=['slack', 'wechat']
75.             ),
76.         ]
77.    
78.     def check_rules(self):
79.         """检查所有告警规则"""
80.         now = datetime.utcnow()
81.         
82.         for rule in self.alert_rules:
83.             # 冷却期检查（避免重复告警）
84.             if rule.name in self.alert_cooldown:
85.                 if now < self.alert_cooldown[rule.name]:
86.                     continue
87.             
88.             # 构建ES查询
89.             query = {
90.                 "query": {
91.                     "bool": {
92.                         "must": [
93.                             rule.condition,
94.                             {"range": {"@timestamp": {"gte": f"now-{rule.time_window}s"}}}
95.                         ]
96.                     }
97.                 },
98.                 "size": 0
99.             }
100.             
101.             # 执行查询
102.             result = self.es.search(index="logs-*", body=query)
103.             count = result['hits']['total']['value']
104.             
105.             # 判断是否触发
106.             if count >= rule.threshold:
107.                 self._send_alert(rule, count)
108.                 # 设置冷却期（5分钟内不重复告警）
109.                 self.alert_cooldown[rule.name] = now + timedelta(minutes=5)
110.    
111.     def _send_alert(self, rule: AlertRule, count: int):
112.         """发送告警"""
113.         alert_data = {
114.             "rule_name": rule.name,
115.             "severity": rule.severity,
116.             "count": count,
117.             "threshold": rule.threshold,
118.             "time_window": f"{rule.time_window}秒",
119.             "timestamp": datetime.utcnow().isoformat()
120.         }
121.         
122.         logger.warning(f"Alert triggered: {rule.name}, count={count}")
123.         
124.         if 'slack' in rule.channels:
125.             self._send_to_slack(alert_data)
126.         
127.         if 'wechat' in rule.channels:
128.             self._send_to_wechat(alert_data)
129.    
130.     def _send_to_slack(self, alert: dict):
131.         """发送到Slack"""
132.         webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
133.         
134.         color = {
135.             'critical': 'danger',
136.             'warning': 'warning',
137.             'info': '#439FE0'
138.         }.get(alert['severity'], 'warning')
139.         
140.         payload = {
141.             "attachments": [{
142.                 "color": color,
143.                 "title": f"🚨 Alert: {alert['rule_name']}",
144.                 "fields": [
145.                     {"title": "严重程度", "value": alert['severity'], "short": True},
146.                     {"title": "触发次数", "value": str(alert['count']), "short": True},
147.                     {"title": "阈值", "value": str(alert['threshold']), "short": True},
148.                     {"title": "时间窗口", "value": alert['time_window'], "short": True},
149.                 ],
150.                 "footer": alert['timestamp']
151.             }]
152.         }
153.         
154.         try:
155.             requests.post(webhook_url, json=payload, timeout=5)
156.         except Exception as e:
157.             logger.error(f"Failed to send Slack alert: {e}")
158.    
159.     def _send_to_wechat(self, alert: dict):
160.         """发送到企业微信"""
161.         webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
162.         
163.         payload = {
164.             "msgtype": "markdown",
165.             "markdown": {
166.                 "content": f"""
167. ## 🚨 告警通知
169. **规则**: {alert['rule_name']}
170. **严重程度**: {alert['severity']}
171. **触发次数**: {alert['count']}
172. **阈值**: {alert['threshold']}
173. **时间窗口**: {alert['time_window']}
174. **时间**: {alert['timestamp']}
175. """
176.             }
177.         }
178.         
179.         try:
180.             requests.post(webhook_url, json=payload, timeout=5)
181.         except Exception as e:
182.             logger.error(f"Failed to send WeChat alert: {e}")
183.    
184.     def run(self):
185.         """主循环"""
186.         logger.info("Alert service started")
187.         
188.         while True:
189.             try:
190.                 self.check_rules()
191.                 time.sleep(30)  # 每30秒检查一次
192.             except KeyboardInterrupt:
193.                 break
194.             except Exception as e:
195.                 logger.error(f"Error in alert loop: {e}")
196.                 time.sleep(60)
198. if __name__ == '__main__':
199.     service = AlertService(
200.         es_hosts=['http://elasticsearch:9200'],
201.         kafka_servers=['kafka1:9092']
202.     )
203.     service.run()

复制代码

6. Kibana仪表板设置

1. // kibana/dashboard.json
2. {
3.   "version": "8.0.0",
4.   "objects": [
5.     {
6.       "id": "logs-overview",
7.       "type": "dashboard",
8.       "attributes": {
9.         "title": "Logs Overview",
10.         "panelsJSON": [
11.           {
12.             "id": "error-count",
13.             "type": "visualization",
14.             "gridData": {"x": 0, "y": 0, "w": 6, "h": 4},
15.             "panelConfig": {
16.               "visType": "metric",
17.               "aggs": [
18.                 {"id": "1", "type": "count", "schema": "metric"}
19.               ],
20.               "filters": [
21.                 {"query": "level:ERROR", "label": "Errors"}
22.               ]
23.             }
24.           },
25.           {
26.             "id": "response-time-percentiles",
27.             "type": "visualization",
28.             "gridData": {"x": 6, "y": 0, "w": 6, "h": 4},
29.             "panelConfig": {
30.               "visType": "line",
31.               "aggs": [
32.                 {"id": "1", "type": "percentiles", "field": "response_time", "percents": [50, 95, 99]}
33.               ]
34.             }
35.           },
36.           {
37.             "id": "top-services",
38.             "type": "visualization",
39.             "gridData": {"x": 0, "y": 4, "w": 6, "h": 4},
40.             "panelConfig": {
41.               "visType": "pie",
42.               "aggs": [
43.                 {"id": "1", "type": "terms", "field": "service", "size": 10}
44.               ]
45.             }
46.           },
47.           {
48.             "id": "error-trend",
49.             "type": "visualization",
50.             "gridData": {"x": 6, "y": 4, "w": 6, "h": 4},
51.             "panelConfig": {
52.               "visType": "area",
53.               "aggs": [
54.                 {"id": "1", "type": "count"},
55.                 {"id": "2", "type": "date_histogram", "field": "@timestamp", "interval": "1h"}
56.               ],
57.               "filters": [
58.                 {"query": "level:ERROR"}
59.               ]
60.             }
61.           }
62.         ]
63.       }
64.     }
65.   ]
66. }

复制代码

7. 日记查询工具

1. # log_query.py - MonkeyCode生成
2. from elasticsearch import Elasticsearch
3. from datetime import datetime, timedelta
4. from typing import List, Dict, Any, Optional
5. import json
7. class LogQuery:
8.     """日志查询工具"""
9.    
10.     def __init__(self, es_hosts: list):
11.         self.es = Elasticsearch(es_hosts)
12.    
13.     def search(self, query: str, service: str = None, level: str = None,
14.                start_time: datetime = None, end_time: datetime = None,
15.                size: int = 100) -> List[Dict]:
16.         """通用搜索"""
17.         must = []
18.         
19.         # 关键字搜索
20.         if query:
21.             must.append({"match": {"message": query}})
22.         
23.         # 服务过滤
24.         if service:
25.             must.append({"term": {"service": service}})
26.         
27.         # 日志级别过滤
28.         if level:
29.             must.append({"term": {"level": level}})
30.         
31.         # 时间范围
32.         time_range = {}
33.         if start_time:
34.             time_range["gte"] = start_time.isoformat()
35.         if end_time:
36.             time_range["lte"] = end_time.isoformat()
37.         if time_range:
38.             must.append({"range": {"@timestamp": time_range}})
39.         
40.         # 执行查询
41.         body = {
42.             "query": {"bool": {"must": must}} if must else {"match_all": {}},
43.             "size": size,
44.             "sort": [{"@timestamp": {"order": "desc"}}]
45.         }
46.         
47.         result = self.es.search(index="logs-*", body=body)
48.         
49.         return [hit["_source"] for hit in result["hits"]["hits"]]
50.    
51.     def get_by_trace_id(self, trace_id: str) -> List[Dict]:
52.         """根据trace_id查询完整链路"""
53.         body = {
54.             "query": {"term": {"trace_id": trace_id}},
55.             "size": 1000,
56.             "sort": [{"@timestamp": {"order": "asc"}}]
57.         }
58.         
59.         result = self.es.search(index="logs-*", body=body)
60.         return [hit["_source"] for hit in result["hits"]["hits"]]
61.    
62.     def get_error_context(self, log_id: str, before: int = 10, after: int = 10) -> List[Dict]:
63.         """获取错误日志上下文"""
64.         # 先找到目标日志
65.         target = self.es.get(index="logs-*", id=log_id)
66.         target_time = target["_source"]["@timestamp"]
67.         hostname = target["_source"]["hostname"]
68.         
69.         # 查询前后的日志
70.         body = {
71.             "query": {
72.                 "bool": {
73.                     "must": [
74.                         {"term": {"hostname": hostname}},
75.                         {"range": {
76.                             "@timestamp": {
77.                                 "gte": datetime.fromisoformat(target_time) - timedelta(minutes=5),
78.                                 "lte": datetime.fromisoformat(target_time) + timedelta(minutes=5)
79.                             }
80.                         }}
81.                     ]
82.                 }
83.             },
84.             "size": before + after + 1,
85.             "sort": [{"@timestamp": {"order": "asc"}}]
86.         }
87.         
88.         result = self.es.search(index="logs-*", body=body)
89.         return [hit["_source"] for hit in result["hits"]["hits"]]
90.    
91.     def stats_by_service(self, time_range: str = "1h") -> Dict:
92.         """按服务统计日志数量"""
93.         body = {
94.             "query": {"range": {"@timestamp": {"gte": f"now-{time_range}"}}},
95.             "size": 0,
96.             "aggs": {
97.                 "by_service": {
98.                     "terms": {"field": "service", "size": 20},
99.                     "aggs": {
100.                         "by_level": {
101.                             "terms": {"field": "level"}
102.                         }
103.                     }
104.                 }
105.             }
106.         }
107.         
108.         result = self.es.search(index="logs-*", body=body)
109.         
110.         stats = {}
111.         for bucket in result["aggregations"]["by_service"]["buckets"]:
112.             service = bucket["key"]
113.             stats[service] = {
114.                 "total": bucket["doc_count"],
115.                 "by_level": {
116.                     b["key"]: b["doc_count"]
117.                     for b in bucket["by_level"]["buckets"]
118.                 }
119.             }
120.         
121.         return stats
123. # 使用示例
124. query = LogQuery(['http://localhost:9200'])
126. # 搜索最近1小时的错误日志
127. errors = query.search(
128.     query="OutOfMemoryError",
129.     level="ERROR",
130.     start_time=datetime.utcnow() - timedelta(hours=1)
131. )
133. # 查询trace_id链路
134. trace_logs = query.get_by_trace_id("abc-123-def")
136. # 按服务统计
137. stats = query.stats_by_service("1h")
138. for service, data in stats.items():
139.     print(f"{service}: {data}")

复制代码

8. Docker Compose一键摆设

1. # docker-compose.yml - MonkeyCode生成
2. version: '3.8'
4. services:
5.   elasticsearch:
6.     image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
7.     environment:
8.       - discovery.type=single-node
9.       - xpack.security.enabled=false
10.       - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
11.     ports:
12.       - "9200:9200"
13.     volumes:
14.       - es-data:/usr/share/elasticsearch/data
15.     healthcheck:
16.       test: ["CMD", "curl", "-f", "http://localhost:9200/_cluster/health"]
17.       interval: 10s
18.       timeout: 5s
19.       retries: 5
21.   kibana:
22.     image: docker.elastic.co/kibana/kibana:8.11.0
23.     depends_on:
24.       elasticsearch:
25.         condition: service_healthy
26.     ports:
27.       - "5601:5601"
28.     environment:
29.       - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
31.   logstash:
32.     image: docker.elastic.co/logstash/logstash:8.11.0
33.     depends_on:
34.       - elasticsearch
35.     volumes:
36.       - ./logstash/pipeline:/usr/share/logstash/pipeline
37.     ports:
38.       - "5044:5044"
40.   alert-service:
41.     build: ./alert-service
42.     depends_on:
43.       - elasticsearch
44.       - kafka
45.     environment:
46.       - ES_HOSTS=http://elasticsearch:9200
47.       - KAFKA_SERVERS=kafka:9092
49. volumes:
50.   es-data:

复制代码

用MonkeyCode天生日记分析体系的关键是：把日记流从产生到告警的完备链路形貌清楚。ELK技能栈成熟但设置复杂，AI天生的设置文件可以节省大量时间，但告警规则必要根据业务特点调优。

免责声明：如果侵犯了您的权益，请联系站长及时删除侵权内容，谢谢合作！qidao123.com:ToB企服之家，中国第一个企服评测及软件市场,开放入驻,技术点评得现金.