AIops-容器自动重启项目(二)-北京尧图网络

1.前情提要预期目的一个容器化的智能运维AIOps平台能自动监控业务和容器指标通过异常检测与时间序列预测模型自动执行运维动作如重启容器、扩容、告警形成“采集-分析-决策-执行”闭环。达成目的1.已经完成了GrafanaprometheuscAdvisor的部署和使用。2.实现容器报警-容器cpu使用率突破80%告警。试验容器名为nginx-web3.指定重启策略俩种初步重启和深度重启。根据实际情况选择指定。详情可见容器自动重启项目(一)后续计划将报错日志写入Loki 实现日志聚合与异常关联。并完成轻量级异常检测模型的构建。本期目的1.部署 Loki和 Promtai。2.在 Grafana Explore 中创建 Loki 数据源并推送日志。3.将提取的日志片段写入 Elasticsearch 或数据库作为 AIOps 知识库。2.作用介绍1.Loki 在该项目中的作用Loki 是一个日志聚合存储引擎它的核心任务是接收、存储和查询 Promtail 推送的日志。在你的项目中Loki 接收来自 Promtail 的日志数据后会按标签如container、job建立索引以实现快速检索同时将原始日志压缩存储到磁盘以节省空间。此外Loki 提供了 HTTP API 查询接口你的 Python 脚本loki_to_es.py正是通过这个接口定时查询包含error、warn等关键词的错误日志然后将这些日志写入 Elasticsearch 形成知识库供后续故障排查使用。2.Promtail 在该项目中的作用Promtail 是一个日志采集代理它的核心任务是从 Docker 容器中读取日志并推送到 Loki。在你的项目中Promtail 通过挂载的/var/run/docker.sock自动发现所有运行中的容器如 nginx-web、prometheus、grafana然后读取这些容器的标准输出日志存储在/var/lib/docker/containers/目录下。在读取过程中Promtail 会为每条日志打上标签如containernginx-web、imagenginx:latest最终将这些带标签的日志通过 HTTP API 推送到 Loki 服务。3.详细部署策略目标将容器日志与告警、重启事件关联支持快速根因定位。1.具体动作部署 Loki Promtail或 Grafana Agent收集所有容器日志。在 Grafana Explore 中创建 Loki 数据源。为你的 Webhook 服务增加一个日志查询接口当容器因 CPU 高被重启时自动拉取重启前 5 分钟的该容器日志通过 Loki API提取 ERROR、WARN 或异常堆栈。将提取的日志片段写入 Elasticsearch 或数据库作为 AIOps 知识库。2.产出在 Grafana 中可通过{containernginx-web} | error快速检索。每次重启自动附带“可疑日志片段”减少人工排查时间。3.实验方法1.部署loki和grafana的文件docker-compose.ymlversion: 3.8 services: loki: image: grafana/loki:latest container_name: loki user: root ports: - 3100:3100 volumes: - /root/loki/loki-config.yaml:/etc/loki/loki-config.yaml - ./loki-data:/loki - /var/log:/var/log:ro - /var/lib/docker/containers:/var/lib/docker/containers:ro - /var/run/docker.sock:/var/run/docker.sock - /root/loki/promtail-config.yaml:/etc/promtail/promtail-config.yaml command: -config.file/etc/loki/loki-config.yaml restart: unless-stopped promtail: image: grafana/promtail:latest container_name: promtail volumes: - /var/log:/var/log:ro - /var/lib/docker/containers:/var/lib/docker/containers:ro - /var/run/docker.sock:/var/run/docker.sock - ./promtail-config.yaml:/etc/promtail/promtail-config.yaml command: -config.file/etc/promtail/promtail-config.yaml restart: unless-stoppedpromtail-config.yamlserver: http_listen_port: 9080 grpc_listen_port: 0 positions: filename: /tmp/positions.yaml clients: - url: http://loki:3100/loki/api/v1/push scrape_configs: - job_name: docker docker_sd_configs: - host: unix:///var/run/docker.sock refresh_interval: 5s relabel_configs: - source_labels: [__meta_docker_container_name] regex: /(.*) target_label: container - source_labels: [__meta_docker_container_label_com_docker_compose_service] target_label: compose_service - source_labels: [__meta_docker_container_image] target_label: imageloki-config.yaml:auth_enabled: false server: http_listen_port: 3100 common: instance_addr: 127.0.0.1 path_prefix: /loki storage: filesystem: chunks_directory: /loki/chunks rules_directory: /loki/rules replication_factor: 1 ring: kvstore: store: inmemory schema_config: configs: - from: 2020-10-24 store: tsdb object_store: filesystem schema: v13 index: prefix: index_ period: 24h同一文件夹下。执行命令。运行镜像有容器docker-compose up -d2.在 Grafana Explore 中创建 Loki 数据源接入新的数据源了。确保ip:端口正确。loki默认暴露3100端口。3.elasticsearch部署将提取的日志片段写入 Elasticsearch 或数据库作为 AIOps 知识库。安装es,并派出容器,为防止后续认证报错禁用es安全认证。:开放9300端口。4.日志聚合1.查询日志条数该语句旨在统计一段时间内出现错误的条数当日志错误达到一定条数可以触发报警。可以圈定固定error后触发报警。2.查询日志内容该语句旨在统计一段时间内loki是否接受到日志可当日志错误达到一定条数可以触发报警。可以圈定固定error后触发报警。该语句主要是为了判断接受日志。是否触发报警请谨慎选择。5.日志推送5.1.脚本日志推送此前已经完成错误日志的聚合现在将错误日志推送至elasticsearch做为aiops的知识库。推送脚本如下loki_to_es.py#!/usr/bin/env python3 独立日志同步服务从 Loki 拉取错误日志写入 Elasticsearch 支持定时执行、断点续传、按容器过滤 import os import time import json import logging from datetime import datetime, timedelta from typing import List, Dict, Any, Optional import aiohttp import asyncio from elasticsearch import Elasticsearch, helpers from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.interval import IntervalTrigger from apscheduler.triggers.cron import CronTrigger # 配置通过环境变量或直接修改 LOKI_URL os.getenv(LOKI_URL, http:/ip:3100) ES_URL os.getenv(ES_URL, http://IP:9200) ES_INDEX os.getenv(ES_INDEX, container-logs) CONTAINER_NAMES os.getenv(CONTAINER_NAMES, nginx-web,prometheus,grafana).split(,) SYNC_INTERVAL_MINUTES int(os.getenv(SYNC_INTERVAL_MINUTES, 5)) # 每5分钟同步一次 LOOKBACK_MINUTES int(os.getenv(LOOKBACK_MINUTES, 10)) # 每次查询过去10分钟的日志 LOG_LEVEL os.getenv(LOG_LEVEL, INFO) # 状态文件记录每个容器最后同步的时间戳实现断点续传 STATE_FILE /data/sync_state.json logging.basicConfig( levelgetattr(logging, LOG_LEVEL), format%(asctime)s - %(name)s - %(levelname)s - %(message)s ) logger logging.getLogger(__name__) # 状态管理 class SyncState: 管理每个容器的同步状态断点续传 def __init__(self, state_file: str STATE_FILE): self.state_file state_file self.state: Dict[str, str] {} self._load() def _load(self): 从文件加载状态 try: if os.path.exists(self.state_file): with open(self.state_file, r) as f: self.state json.load(f) logger.info(f加载状态文件成功: {len(self.state)} 个容器) else: logger.info(状态文件不存在将创建新文件) except Exception as e: logger.error(f加载状态文件失败: {e}) self.state {} def _save(self): 保存状态到文件 try: # 确保目录存在 os.makedirs(os.path.dirname(self.state_file), exist_okTrue) with open(self.state_file, w) as f: json.dump(self.state, f, indent2) except Exception as e: logger.error(f保存状态文件失败: {e}) def get_last_sync(self, container: str) - Optional[datetime]: 获取某个容器的最后同步时间 if container in self.state: try: return datetime.fromisoformat(self.state[container]) except: return None return None def update_last_sync(self, container: str, sync_time: datetime): 更新某个容器的最后同步时间 self.state[container] sync_time.isoformat() self._save() def get_or_default(self, container: str, default_minutes: int 10) - datetime: 获取最后同步时间如果没有则返回当前时间减去默认分钟数 last self.get_last_sync(container) if last: return last return datetime.now() - timedelta(minutesdefault_minutes) # Loki 查询 async def query_loki( container: str, start_time: datetime, end_time: datetime, max_logs: int 500 ) - List[Dict[str, Any]]: 从 Loki 查询指定容器的错误日志返回格式[{timestamp: 2024-01-01T00:00:00Z, message: error log, labels: {...}}] # LogQL 查询过滤错误/警告日志不区分大小写 # 可以自定义关键词 keywords (?i)(error|warn|exception|failed|timeout|panic|fatal|critical|unable|denied|refused) query f{{container{container}}} |~ {keywords} # 转换为 Unix 纳秒 start_ns int(start_time.timestamp() * 1e9) end_ns int(end_time.timestamp() * 1e9) url f{LOKI_URL}/loki/api/v1/query_range params { query: query, start: start_ns, end: end_ns, limit: max_logs, direction: FORWARD } try: async with aiohttp.ClientSession(timeoutaiohttp.ClientTimeout(total30)) as session: async with session.get(url, paramsparams) as resp: if resp.status ! 200: error_text await resp.text() logger.error(fLoki 查询失败 [{resp.status}]: {error_text}) return [] data await resp.json() results data.get(data, {}).get(result, []) logs [] for stream in results: stream_labels stream.get(stream, {}) values stream.get(values, []) for ts_ns, msg in values: logs.append({ timestamp: datetime.fromtimestamp(int(ts_ns) / 1e9).isoformat(), message: msg.strip(), labels: stream_labels }) logger.info(f从 Loki 查询到 {len(logs)} 条日志 [容器: {container}]) return logs except asyncio.TimeoutError: logger.error(fLoki 查询超时 [容器: {container}]) return [] except Exception as e: logger.error(fLoki 查询异常: {e}) return [] # Elasticsearch 写入 def create_es_index_if_not_exists(es_client: Elasticsearch, index: str): 创建 ES 索引如果不存在 if es_client.indices.exists(indexindex): logger.info(fES 索引 {index} 已存在) return mapping { mappings: { properties: { container: {type: keyword}, timestamp: {type: date}, message: {type: text, fields: {keyword: {type: keyword, ignore_above: 256}}}, level: {type: keyword}, labels: {type: object, enabled: False}, # 保留原始标签 synced_at: {type: date}, source: {type: keyword, index: True} } }, settings: { number_of_shards: 1, number_of_replicas: 0, index.refresh_interval: 5s } } try: es_client.indices.create(indexindex, bodymapping) logger.info(f✅ 创建 ES 索引 {index} 成功) except Exception as e: logger.error(f创建 ES 索引失败: {e}) def bulk_write_to_es( es_client: Elasticsearch, index: str, logs: List[Dict[str, Any]], container: str ) - int: 批量写入日志到 Elasticsearch返回成功写入数量 if not logs: return 0 actions [] for log in logs: actions.append({ _index: index, _source: { container: container, timestamp: log.get(timestamp), message: log.get(message), labels: log.get(labels, {}), level: detect_log_level(log.get(message, )), synced_at: datetime.now().isoformat(), source: loki_sync } }) try: success, failed helpers.bulk(es_client, actions, stats_onlyTrue, raise_on_errorFalse) logger.info(f✅ ES 写入成功: {success} 条, 失败: {failed} 条 [容器: {container}]) return success except Exception as e: logger.error(fES 写入失败: {e}) return 0 def detect_log_level(message: str) - str: 根据日志内容检测日志级别 msg_lower message.lower() if error in msg_lower or exception in msg_lower or fatal in msg_lower: return error elif warn in msg_lower or warning in msg_lower: return warn elif critical in msg_lower or panic in msg_lower: return critical elif info in msg_lower: return info elif debug in msg_lower: return debug return unknown # 核心同步任务 async def sync_container_logs( es_client: Elasticsearch, state: SyncState, container: str, lookback_minutes: int 10 ): 同步单个容器的日志 logger.info(f开始同步容器日志: {container}) # 获取最后同步时间断点续传 last_sync state.get_or_default(container, lookback_minutes) end_time datetime.now() # 如果距离上次同步太近 1分钟跳过避免重复 if (end_time - last_sync).total_seconds() 60: logger.debug(f容器 {container} 上次同步在 {last_sync}, 跳过) return # 从 Loki 查询 logs await query_loki(container, last_sync, end_time) if logs: # 写入 ES count bulk_write_to_es(es_client, ES_INDEX, logs, container) if count 0: logger.info(f容器 {container} 同步完成: {count} 条新日志) else: logger.info(f容器 {container} 无新日志) else: logger.info(f容器 {container} 无错误日志) # 更新状态即使没有日志也更新避免重复查询 state.update_last_sync(container, end_time) async def sync_all_containers(): 同步所有容器的日志 logger.info( * 50) logger.info(开始同步所有容器日志...) # 初始化 ES 客户端每次同步重新创建避免连接过期 es_client Elasticsearch( [ES_URL], request_timeout30, retry_on_timeoutTrue, max_retries3 ) try: # 检查 ES 连接 if not es_client.ping(): logger.error(ES 连接失败) return # 确保索引存在 create_es_index_if_not_exists(es_client, ES_INDEX) # 状态管理 state SyncState() # 同步每个容器 for container in CONTAINER_NAMES: container container.strip() if not container: continue try: await sync_container_logs( es_client, state, container, LOOKBACK_MINUTES ) except Exception as e: logger.error(f同步容器 {container} 失败: {e}) except Exception as e: logger.error(f同步过程异常: {e}) finally: es_client.close() logger.info(所有容器同步完成) logger.info( * 50) # 调度器设置 def setup_scheduler(): 配置定时任务 scheduler AsyncIOScheduler() # 每 SYNC_INTERVAL_MINUTES 分钟执行一次 scheduler.add_job( sync_all_containers, triggerIntervalTrigger(minutesSYNC_INTERVAL_MINUTES), idsync_loki_to_es, replace_existingTrue ) # 可选每天凌晨 2 点全量同步清理重复数据 # scheduler.add_job( # sync_all_containers, # triggerCronTrigger(hour2, minute0), # iddaily_full_sync # ) return scheduler # 主程序 async def main(): logger.info( Loki → Elasticsearch 日志同步服务启动) logger.info(f配置:) logger.info(f - Loki URL: {LOKI_URL}) logger.info(f - ES URL: {ES_URL}) logger.info(f - ES Index: {ES_INDEX}) logger.info(f - 监控容器: {CONTAINER_NAMES}) logger.info(f - 同步间隔: {SYNC_INTERVAL_MINUTES} 分钟) logger.info(f - 查询回溯: {LOOKBACK_MINUTES} 分钟) # 启动调度器 scheduler setup_scheduler() scheduler.start() logger.info(✅ 调度器已启动) # 启动时立即执行一次同步 logger.info(执行首次同步...) await sync_all_containers() # 保持运行 try: while True: await asyncio.sleep(60) except KeyboardInterrupt: logger.info( 收到中断信号正在关闭...) scheduler.shutdown() logger.info(服务已关闭) if __name__ __main__: asyncio.run(main())执行后脚本开始推送保证环境正确和包的安装。执行成功后情况如下5.2容器日志推送5.2.1文件准备如需要容器日志推送给出以该脚本功能制作的容器方法。该目录下下辖文件如下ocker-compose.ymlversion: 3.8 services: loki-to-es: build: . container_name: loki-to-es restart: unless-stopped # 环境变量配置 environment: - LOKI_URL${LOKI_URL:-http://loki:3100} - ES_HOST${ES_HOST:-http://elasticsearch:9200} - ES_USERNAME${ES_USERNAME:-elastic} - ES_PASSWORD${ES_PASSWORD:-} - ES_INDEX${ES_INDEX:-loki-logs} - INTERVAL${INTERVAL:-60} - BATCH_SIZE${BATCH_SIZE:-100} - LOG_LEVEL${LOG_LEVEL:-INFO} - TZAsia/Shanghai # 使用环境变量文件 env_file: - .env # 挂载配置文件可选 volumes: - ./config:/app/config:ro - ./logs:/app/logs # 网络配置 networks: - elastic - loki # 日志配置 logging: driver: json-file options: max-size: 10m max-file: 3 # 依赖服务如果使用外部服务可取消注释 # depends_on: # - elasticsearch # - loki # 网络定义如果使用外部网络取消注释并调整 networks: elastic: external: true name: elastic loki: external: true name: lokidockfile:FROM python:3.9-alpine # 设置工作目录 WORKDIR /app # 安装运行时依赖不需要 gcc RUN apk add --no-cache \ gcc \ musl-dev \ libffi-dev \ rm -rf /var/cache/apk/* # 复制依赖文件并安装 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple # 复制脚本 COPY loki_to_es.py . # 创建非 root 用户 RUN adduser -D -u 1000 appuser chown -R appuser:appuser /app USER appuser CMD [python, loki_to_es.py].env# Loki 配置 LOKI_URLhttp://IP:3100 # Elasticsearch 配置 ES_HOSThttp://IP:9200 ES_USERNAMEelastic ES_PASSWORDyour_password_here # 索引配置 ES_INDEXloki-logs # 运行参数 INTERVAL60 BATCH_SIZE100 LOG_LEVELINFO # 时区 TZAsia/Shanghairequirements.txt:aiohttp3.8.0 elasticsearch8.0.0,9.0.0 python-dotenv1.0.0加上loki_to_es.py合计5个文件。建议这个五个文件放在同一目录下且只此5个文件。且该文件夹暂且命名为loki_to_es。5.2.2构建镜像在loki_to_es文件夹下执行该命令构建镜像docker build -t loki-to-es:latest .注意源文件选择否则构建镜像时间过长。验证镜像是否构建成功docker images存在loki-to-es镜像就即可。6.总结完成目的完成lokipromtailelasticsearch的部署。向elasticsearch推送容器错误日志。此错误日志将作为AI的知识库使用也可作为总结内容。后续目的AI轻量化模型构建和预测。结语感谢观看欢迎指教。如写的不好请嘴下留情。最后美女镇楼水淼aqua

AIops-容器自动重启项目(二)

相关推荐

Go语言后端数据库表设计规范与最佳实践

OpenAI低延迟AI技术解析：WebSocket与硬件加速

AI Agent时代，程序员最该吃透的底层思维，状态机与二维矩阵表

更多资讯

2026 年门窗智能控制系统的网络安全与数据主权：等保合规、加密传输与零信任架构的十大技术品牌

Mastra Server 部署验证测试指南：`--test server` 从部署到 API 冒烟验证

大数据时代的数据质量保障体系设计与实践

外贸精准获客难？星谷云AI赋能B2B制造企业出海营销新范式

MySQL 8.0 JSON字段与函数索引在SpringBoot中的实践

中国汽车空气悬架技术突破与国际供应链机遇

今日更新

Linux 内核 ARM 虚拟内存布局全解析：从 4GB 地址空间划分到源码级验证

AIGC降重工具测评与教育应用指南

数据容灾核心指标与实战方案解析

本周精选

超人会飞不算本事：系统稳定依赖清晰规则与边界设计

超人VS蜘蛛侠：拆解超级IP的影响力与传播方法论

基于CNN的调制信号识别：MATLAB实现时频图分类实战

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？