从ELK走到Loki：日志管理笔记

作者: 醉月思发布日期: 2022-09-12 分类: 技术实践

从ELK走到Loki：日志管理笔记一旦进项目，好看的架构图就没那么管用了。

那次之后才开始认真做集中式日志：先上 ELK，存储成本上来后又试了 Loki。

最初的问题

分散的日志文件

每个服务都有独立的日志文件，分布在不同的服务器上。

# 查看日志需要登录每台服务器
ssh server1 tail -f /var/log/myapp/app.log
ssh server2 tail -f /var/log/myapp/app.log

问题：

排查困难
没有统一的视图
日志容易丢失

ELK Stack

架构组成

graph TB A[应用日志] --> B[Filebeat] B --> C[Logstash] C --> D[Elasticsearch] D --> E[Kibana] style A fill:#FFD700 style B fill:#90EE90 style C fill:#90EE90 style D fill:#87CEEB style E fill:#FFB6C1

Filebeat：轻量级日志收集器 Logstash：日志处理和转换 Elasticsearch：日志存储和搜索 Kibana：日志可视化

部署 ELK

使用 Docker 部署：

# Elasticsearch
docker run -d --name elasticsearch \
  -p 9200:9200 \
  -e "discovery.type=single-node" \
  -e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \
  elasticsearch:7.14.0

# Logstash
docker run -d --name logstash \
  --link elasticsearch:elasticsearch \
  -p 5044:5044 \
  logstash:7.14.0

# Kibana
docker run -d --name kibana \
  --link elasticsearch:elasticsearch \
  -p 5601:5601 \
  kibana:7.14.0

# Filebeat（在应用服务器上）
docker run -d --name filebeat \
  --link logstash:logstash \
  -v /var/log:/var/log \
  elastic/filebeat:7.14.0

Filebeat 配置

# filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/myapp/*.log
  fields:
    app: myapp
    env: production

output.logstash:
  hosts: ["logstash:5044"]

processors:
  - add_cloud_metadata: ~
  - add_host_metadata: ~

Logstash 配置

# logstash.conf
input {
  beats {
    port => 5044
  }
}

filter {
  # 解析 JSON 日志
  json {
    source => "message"
  }

  # 提取时间
  date {
    match => ["timestamp", "ISO8601"]
  }

  # 添加元数据
  mutate {
    add_field => {
      "environment" => "production"
      "region" => "us-west-1"
    }
  }
}

output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "myapp-%{+YYYY.MM.dd}"
  }
}

结构化日志

// 使用结构化日志
const winston = require('winston');

const logger = winston.createLogger({
  level: 'info',
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.json()
  ),
  transports: [
    new winston.transports.File({ filename: '/var/log/myapp/app.log' })
  ]
});

// 记录日志
logger.info('User login', {
  userId: 123,
  username: 'alice',
  ip: '192.168.1.1',
  userAgent: 'Mozilla/5.0...'
});

logger.error('Database error', {
  error: 'Connection timeout',
  query: 'SELECT * FROM users WHERE id = 123',
  duration: 5000
});

Kibana 查询

// 查询特定用户的日志
{
  "query": {
    "bool": {
      "must": [
        { "term": { "userId": 123 } },
        { "range": { "@timestamp": { "gte": "now-1h" } } }
      ]
    }
  }
}

// 查询错误日志
{
  "query": {
    "term": { "level": "error" }
  }
}

// 聚合统计
{
  "size": 0,
  "aggs": {
    "by_level": {
      "terms": {
        "field": "level"
      }
    },
    "by_app": {
      "terms": {
        "field": "app"
      }
    }
  }
}

Loki

为什么用 Loki

ELK 的资源消耗太高，不适合大规模部署。

Loki 的优势：

资源消耗低
部署简单
与 Prometheus 集成好
支持标签查询

部署 Loki

使用 Docker 部署：

# Loki
docker run -d --name loki \
  -p 3100:3100 \
  grafana/loki:latest \
  -config.file=/etc/loki/local-config.yaml

# Promtail（在应用服务器上）
docker run -d --name promtail \
  -v /var/log:/var/log \
  -v /etc/promtail:/etc/promtail \
  grafana/promtail:latest \
  -config.file=/etc/promtail/config.yml

# Grafana
docker run -d --name grafana \
  -p 3000:3000 \
  --link loki:loki \
  grafana/grafana:latest

Promtail 配置

# promtail/config.yml
server:
  http_listen_port: 9080

positions:
  filename: /tmp/positions.yaml

clients:
  - url: http://loki:3100/loki/api/v1/push

scrape_configs:
- job_name: myapp
  static_configs:
  - targets:
      - localhost
    labels:
      job: myapp
      env: production
      __path__: /var/log/myapp/*.log

  pipeline_stages:
  - json:
      expressions:
        level: level
        userId: userId
        timestamp: timestamp
  - labels:
      level:
  - timestamp:
      source: timestamp
      format: RFC3339

Grafana 查询

# 查询特定应用的日志
{job="myapp"}

# 查询错误日志
{job="myapp", level="error"}

# 查询特定用户的日志
{job="myapp"} |= "userId:123"

# 聚合统计
count_over_time({job="myapp"}[5m])

# 计算错误率
sum(rate({job="myapp", level="error"}[5m])) / sum(rate({job="myapp"}[5m]))

日志分级

日志级别

const logger = winston.createLogger({
  levels: {
    error: 0,
    warn: 1,
    info: 2,
    debug: 3,
    trace: 4
  }
});

// 错误日志：需要立即处理
logger.error('Service unavailable', {
  error: 'Connection refused',
  host: 'db.example.com'
});

// 警告日志：潜在问题
logger.warn('High memory usage', {
  memory: '90%',
  threshold: '80%'
});

// 信息日志：重要事件
logger.info('User registered', {
  userId: 123,
  email: '[email protected]'
});

// 调试日志：开发调试
logger.debug('Query executed', {
  query: 'SELECT * FROM users WHERE id = 123',
  duration: 50
});

// 追踪日志：详细的执行流程
logger.trace('Function called', {
  function: 'getUser',
  params: { id: 123 }
});

日志采样

高流量场景，可以采样日志。

const SAMPLE_RATE = 0.1;  // 10% 采样

function logDebug(message, data) {
  if (Math.random() < SAMPLE_RATE) {
    logger.debug(message, data);
  }
}

function logInfo(message, data) {
  // 重要日志不采样
  logger.info(message, data);
}

日志轮转

Logrotate 配置

# /etc/logrotate.d/myapp
/var/log/myapp/*.log {
    daily
    rotate 7
    compress
    delaycompress
    missingok
    notifempty
    create 0644 myapp myapp
    sharedscripts
    postrotate
        systemctl reload myapp
    endscript
}

应用层轮转

const winston = require('winston');
const winstonDailyRotateFile = require('winston-daily-rotate-file');

const transport = new winstonDailyRotateFile({
  filename: '/var/log/myapp/app-%DATE%.log',
  datePattern: 'YYYY-MM-DD',
  maxSize: '20m',
  maxFiles: '7d'
});

const logger = winston.createLogger({
  transports: [transport]
});

踩过的坑

坑一：日志格式不统一

不同服务的日志格式不一样，难以查询。

解决：统一使用结构化日志。

// 统一格式
logger.info('event', {
  timestamp: new Date().toISOString(),
  level: 'info',
  app: 'myapp',
  service: 'user-service',
  message: 'User logged in',
  data: {
    userId: 123,
    username: 'alice'
  }
});

坑二：日志丢失

应用重启后，日志丢失。

解决：使用缓冲和批量发送。

const buffer = [];
const MAX_BUFFER_SIZE = 100;
const FLUSH_INTERVAL = 5000;  // 5 秒

function logToLoki(data) {
  buffer.push(data);

  if (buffer.length >= MAX_BUFFER_SIZE) {
    flushBuffer();
  }
}

function flushBuffer() {
  if (buffer.length === 0) return;

  fetch('http://loki:3100/loki/api/v1/push', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ streams: buffer })
  }).then(() => {
    buffer.length = 0;  // 清空缓冲区
  });
}

setInterval(flushBuffer, FLUSH_INTERVAL);

// 进程退出时刷新
process.on('exit', flushBuffer);

坑三：性能问题

日志写入太慢，影响应用性能。

解决：

异步写入
使用消息队列
采样日志

const queue = new PQueue({ concurrency: 5 });

async function logAsync(data) {
  await queue.add(() => {
    return fetch('http://loki:3100/loki/api/v1/push', {
      method: 'POST',
      body: JSON.stringify({ streams: [data] })
    });
  });
}

写在最后

日志管理这东西，不是技术问题，是运维问题。

解决了：

排查困难
统一视图
日志分析

带来了：

资源消耗
运维成本
复杂度增加

实施之前先评估：

日志量
查询需求
资源预算
团队能力

不是所有场景都需要 ELK，有时候简单的日志聚合就够用。

这次日志管理改造花了两周，从分散的日志文件到 ELK，再到 Loki。改造完成后，排查问题的时间从 30 分钟降到 5 分钟，运维效率提升明显。

版权声明: 本文首发于指尖魔法屋-从ELK走到Loki：日志管理笔记（https://blog.thinkmoon.cn/post/66-log-management-elk-loki-architecture/）转载或引用必须申明原指尖魔法屋来源及源地址！

利用 Cursor ACP 模式搭建飞书助手机器人，无需任何大模型接口

用 lark-cli 收飞书消息，Python 桥接 Cursor ACP（agent acp），项目代码里不集成任何大模型 API；适合已有 Cursor 订阅、想在本机跑 Agent 的场景。

📁 技术实践

📅 2026-07-07

监控告警踩坑记录

凌晨三点被邮件吵醒，原来是磁盘脚本误报——同一条告警连发七封，值班同事已经麻木了。