引言

上周领导在周会上问了一个问题："我们的日活用户到底有多少？哪个接口被调得最多？"团队沉默了几秒——产品有埋点方案，但排期在两个月后；数据组说要做完整的 BI 体系；风控那边倒是有数据，但要走数据同步流程。

散会后我打开 Kibana 看了看我们早已在采集的 API 访问日志，突然意识到：答案其实一直躺在日志里。每条 API 日志里有 userId、接口路径、响应码、耗时、时间戳——DAU 就是 userId 的去重计数，接口 TOP10 就是一次 terms 聚合，异常率就是 status≥500 的占比，慢接口排行就是耗时字段的 top_hits。不需要埋点、不需要加 SDK、不需要排期，把日志结构化好，Kibana 拖拖拽拽就是业务大屏。

这篇文章把这套"穷人版 BI"完整讲一遍：

• 链路：Logback 结构化日志（JSON）→ Filebeat 采集 → Elasticsearch 存储 → Kibana 可视化，每个环节怎么配
• 四个实战指标从日志里"算"出来：DAU、接口调用 TOP10、异常率趋势、慢接口排行
• 完整可抄的配置：Logback JSON 配置 + Filebeat 配置 + ES 索引模板 + Kibana 仪表盘清单
• 和埋点方案的边界：哪些指标日志做不了，什么时候该上真埋点

一、为什么日志能干业务监控的活

1.1 一条 API 日志里藏着什么

先看一条精心设计的 API 访问日志，每个字段都是潜在的"业务指标原料"：

{
  "ts": "2024-08-22T14:32:05.123+08:00",
  "traceId": "a1b2c3d4e5f67890",
  "userId": "U5001",
  "method": "POST",
  "uri": "/api/order/create",
  "status": 200,
  "costMs": 187,
  "clientIp": "223.104.5.87",
  "userAgent": "Mozilla/5.0 (iPhone...)",
  "appId": "com.example.app",
  "errMsg": null
}

这些字段本来就在请求上下文里，只是大多数团队的日志没把它们打出来。补一行拦截器的代码，一切就有了。

1.2 和埋点方案的边界

日志监控不是要替代埋点，两者边界清晰：

结论：服务端行为分析用日志，前端转化分析用埋点。日志方案是"今天就要看数据"的最快路径，也是永远可信的数据底座（埋点丢失时用日志对账）。

1.3 整体链路

应用 JVM                    采集                 存储                可视化
┌──────────────┐      ┌─────────────┐      ┌──────────────┐   ┌─────────────┐
│ Logback       │      │ Filebeat     │      │ Elasticsearch │   │ Kibana       │
│ JSON 结构化日志 │ ───► │ 监听日志文件   │ ───► │ 日志索引        │──►│ 仪表盘/大屏   │
│ (API 拦截器打) │ 文件  │ 批量发送      │ HTTP │ (ILM 生命周期) │   │ (拖拽聚合)   │
└──────────────┘      └─────────────┘      └──────────────┘   └─────────────┘
     ①结构化            ②解耦采集             ③索引存储           ④免开发可视化

选这套链路的理由：ELK 栈大概率已经在用了（错误日志排查都靠它），增量成本只有"把日志打对 + 建个索引模板 + 搭个 Dashboard"。如果公司用 Loki，文末 FAQ 给了等效方案。

二、第一步：让日志可分析——Logback 结构化改造

2.1 API 拦截器：一次请求一行 JSON

改造的核心是一个拦截器：请求进来计时，响应后把关键字段打成一行 JSON 日志：

/**
 * API 访问日志拦截器：一次请求一行结构化日志
 * 这是整个监控方案的数据源头——字段打不全，后面全是巧妇难为无米之炊
 */
@Component
@RequiredArgsConstructor
public class ApiAccessLogInterceptor implements HandlerInterceptor {

    private static final Logger accessLog = LoggerFactory.getLogger("API_ACCESS");
    private static final String ATTR_START = "api_start_ts";

    @Override
    public boolean preHandle(HttpServletRequest request,
                             HttpServletResponse response, Object handler) {
        request.setAttribute(ATTR_START, System.currentTimeMillis());
        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request,
                                HttpServletResponse response,
                                Object handler, Exception ex) {
        Long start = (Long) request.getAttribute(ATTR_START);
        if (start == null) {
            return;
        }
        long costMs = System.currentTimeMillis() - start;

        // userId：从认证上下文取（匿名请求为 null）
        String userId = CurrentUserHolder.getUserIdOrNull();

        // 错误信息：业务异常摘要，不打堆栈（堆栈另走 error 日志）
        String errMsg = null;
        if (ex instanceof BizException biz) {
            errMsg = biz.getCode() + ": " + biz.getMessage();
        } else if (ex != null) {
            errMsg = ex.getClass().getSimpleName();
        }

        // 用占位符写参，SLF4J + LogstashEncoder 负责 JSON 组装
        accessLog.info("{}|{}|{}|{}|{}|{}|{}|{}|{}",
                userId,
                request.getMethod(),
                request.getRequestURI(),
                response.getStatus(),
                costMs,
                request.getRemoteAddr(),
                request.getHeader("User-Agent"),
                request.getHeader("X-App-Id"),
                errMsg);
    }
}

// WebMvcConfig 注册（拦截 API，排除静态资源）
@Configuration
@RequiredArgsConstructor
public class WebMvcConfig implements WebMvcConfigurer {

    private final ApiAccessLogInterceptor apiAccessLogInterceptor;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(apiAccessLogInterceptor)
                .addPathPatterns("/api/**")
                .excludePathPatterns("/api/health", "/api/metrics");
    }
}

两个设计决策：

1. 独立的 API_ACCESS logger——访问日志和错误日志分开走，Filebeat 采集时天然分流，不用在采集端做正则过滤；
2. 一次请求一行——绝对不打多行、不打堆栈进访问日志，一行一个 JSON 是所有日志分析的前提。

2.2 Logback JSON 配置：LogstashEncoder 一行搞定

<!-- logback-spring.xml -->
<configuration>
    <!-- 访问日志 appender：JSON 格式输出到独立文件 -->
    <appender name="API_ACCESS_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>/var/log/app/api-access.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
            <!-- 按天滚动 + 单文件 200MB，保留 7 天（采集后本地日志就是临时缓冲） -->
            <fileNamePattern>/var/log/app/api-access.%d{yyyy-MM-dd}.%i.log.gz</fileNamePattern>
            <maxFileSize>200MB</maxFileSize>
            <maxHistory>7</maxHistory>
            <totalSizeCap>5GB</totalSizeCap>
        </rollingPolicy>
        <encoder class="net.logstash.logback.encoder.LogstashEncoder">
            <!-- 自定义字段分隔符：我们用 "|" 把业务字段并进 message，
                 也可以改成 Map 结构（见下方说明） -->
            <fieldNames>
                <timestamp>ts</timestamp>
                <message>[ignore]</message>
            </fieldNames>
            <customFields>{"app":"order-service","host":"${HOSTNAME}"}</customFields>
        </encoder>
    </appender>

    <!-- API_ACCESS logger 单独走 JSON appender -->
    <logger name="API_ACCESS" level="INFO" additivity="false">
        <appender-ref ref="API_ACCESS_FILE"/>
    </logger>

    <!-- 常规日志照旧 -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{HH:mm:ss.SSS} [%X{trace_id}] %-5level %logger - %msg%n</pattern>
        </encoder>
    </appender>
    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
    </root>
</configuration>

依赖（LogstashEncoder）：

<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>7.4</version>
</dependency>

更优雅的变体：直接打 Map 结构（推荐新项目用，省去采集端解析 | 分隔符）：

// 用 StructuredArguments 把字段作为 JSON 的顶层字段输出
import static net.logstash.logback.argument.StructuredArguments.kv;

accessLog.info("api access",
        kv("userId", userId),
        kv("method", request.getMethod()),
        kv("uri", request.getRequestURI()),
        kv("status", response.getStatus()),
        kv("costMs", costMs),
        kv("clientIp", request.getRemoteAddr()),
        kv("appId", request.getHeader("X-App-Id")),
        kv("errMsg", errMsg));
// 输出：{"ts":"...","message":"api access","userId":"U5001","method":"POST",
//        "uri":"/api/order/create","status":200,"costMs":187,...}
// 每个字段都是 ES 里的独立字段，Kibana 可直接聚合，无需解析

三、第二步：Filebeat 采集配置

3.1 为什么用 Filebeat 而不是 Logstash 直采

Filebeat 的注册表（registry）机制保证至少一次投递——ES 不可用期间日志堆积在文件里，恢复后自动补采。

3.2 filebeat.yml 完整配置

# filebeat.yml
filebeat.inputs:
  - type: filestream                    # 7.16+ 推荐 filestream（替代 log 类型）
    id: api-access                      # 唯一 ID：registry 断点续传的依据
    enabled: true
    paths:
      - /var/log/app/api-access*.log    # 匹配滚动产生的文件
    parsers:
      - ndjson:                         # 一行一个 JSON，自动展开为顶层字段
          overwrite_keys: true
          add_error_key: true
    fields:                             # 打上环境标签（多环境检索用）
      env: production
      app: order-service
    fields_under_root: true

# ── 处理：解析 "|" 分隔的 message，拆成独立字段 ──
processors:
  - dissect:
      tokenizer: "%{userId}|%{method}|%{uri}|%{status}|%{costMs}|%{clientIp}|%{userAgent}|%{appId}|%{errMsg}"
      field: "message"
      target_prefix: ""
      ignore_failure: true
  # 数字字段转类型（否则 ES 里是 text 无法做数值聚合）
  - convert:
      fields:
        - {from: status, type: integer}
        - {from: costMs, type: integer}
      ignore_missing: true
  # errMsg 空值置 null（便于 Kibana 过滤"无错误"）
  - drop_fields:
      fields: ["ecs", "agent", "log", "input"]   # 减肥：去掉无用的元数据字段

# ── 输出 ──
output.elasticsearch:
  hosts: ["http://es-node-1:9200", "http://es-node-2:9200"]
  index: "api-access-%{+yyyy.MM.dd}"     # 按天建索引
  bulk_max_size: 2048                    # 批量发送大小
  worker: 2

# 索引模板引用（见 3.3）
setup.template.name: "api-access"
setup.template.pattern: "api-access-*"
setup.ilm.enabled: false                # 简化：自己按天滚动（也可开 ILM）

# ── 可靠性 ──
queue.mem:
  events: 8192                           # 内存队列
  flush.min_events: 512
  flush.timeout: 5s
logging.level: info

启动：

filebeat test config -c filebeat.yml    # 配置校验
filebeat -e -c filebeat.yml             # 前台启动观察

dissect 处理器的位置：如果 2.2 用的是 StructuredArguments（Map 结构），这里整段 dissect 删掉——字段已经是顶层 JSON，直接进 convert 转类型即可。两种打日志方式选一种，前后端配置对应好。

3.3 ES 索引模板：字段类型定错，聚合全废

这是整个方案最容易翻车的点：ES 对新字段的动态映射，字符串默认建成 text + keyword 复合——text 可以全文检索但不能做 terms 聚合的精确去重；数字若被打成 text，avg/max/percentiles 全部不可用。DAU 要算去重，userId 必须是 keyword；costMs 要算分位，必须是 integer：

PUT _index_template/api-access
{
  "index_patterns": ["api-access-*"],
  "priority": 100,
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "refresh_interval": "5s",
      "index.lifecycle.name": "api-access-ilm"
    },
    "mappings": {
      "properties": {
        "ts":        { "type": "date" },
        "traceId":   { "type": "keyword", "ignore_above": 64 },
        "userId":    { "type": "keyword" },
        "method":    { "type": "keyword" },
        "uri":       { "type": "keyword", "ignore_above": 256 },
        "status":    { "type": "integer" },
        "costMs":    { "type": "integer" },
        "clientIp":  { "type": "ip" },
        "userAgent": { "type": "keyword", "ignore_above": 512 },
        "appId":     { "type": "keyword" },
        "errMsg":    { "type": "keyword", "ignore_above": 512 },
        "env":       { "type": "keyword" },
        "app":       { "type": "keyword" },
        "host":      { "type": "keyword" }
      }
    }
  }
}

原则速记：能枚举/要精确匹配/要去重的字段一律 keyword；要做数值计算的字段一律 integer/long；uri 这种高基数字段用 keyword + ignore_above 防炸索引。模板建好后第二天的新索引自动生效（当天已创建的索引要手动改 mapping 或删除重采）。

四、实战：四个指标从日志里"算"出来

4.1 指标一：DAU（日活跃用户数）

Kibana Lens 三步：选 api-access-* 数据视图 → 画布类型选 Metric → 指标选 Unique count of userId，过滤器 userId IS NOT NULL。左侧时间范围选 Today——DAU 就出来了。

等价的 ES 查询（程序化取数时用）：

POST api-access-*/_search
{
  "size": 0,
  "query": {
    "bool": {
      "filter": [
        { "range": { "ts": { "gte": "now/d", "time_zone": "+08:00" } } },
        { "exists": { "field": "userId" } }
      ]
    }
  },
  "aggs": {
    "dau": { "cardinality": { "field": "userId", "precision_threshold": 40000 } }
  }
}

注意 cardinality 是近似算法（HyperLogLog）：千万级用户下误差 1% 以内可接受，precision_threshold 调到 40000 精度更好但更耗内存。报表口径写清楚"近似去重"，和埋点数据对账时差异就有了解释。

进阶玩法——近 30 天 DAU 趋势：Lens 里把时间字段拖到横轴（interval=1d），Unique count userId 拖到纵轴，30 秒出图。

4.2 指标二：接口调用 TOP10

Lens → 画布选 Table → 行维度 uri（Top 10 values，按 count 排序）→ 指标 Count。可以叠加第二列"错误数"（filter: status >= 500 的 count）一次看流量和健康度。

Tsvb（Timelion）版更适合做成"TOP10 排行榜"视觉：

// Timelion：每 5 分钟刷新的实时接口调用排行
.es(index=api-access-*, metric=count, field=uri, split=uri.top=10,
    metric_agg=terms, terms_order_by=_count)
.label("接口名")
.legend(columns=1, position=ne)

实用变体：按 appId 分组的 TOP10——回答"流量主要来自哪个端"的问题；按 userId 的 TOP10——秒级识别刷子/爬虫（某用户单日调用 8 万次 API）。

4.3 指标三：异常率趋势

Lens → 画布 Line → 横轴时间 → 纵轴两个指标：

• 总请求数：Count
• 异常数：Count + 过滤器 status >= 500
• 异常率：Tsvb 或 Lens 公式 (异常数 / 总请求数) * 100

// 异常率（按小时分桶）+ 按接口细分定位异常源
POST api-access-*/_search
{
  "size": 0,
  "query": { "range": { "ts": { "gte": "now-24h" } } },
  "aggs": {
    "by_hour": {
      "date_histogram": { "field": "ts", "fixed_interval": "1h", "time_zone": "+08:00" },
      "aggs": {
        "total":      { "filter": { "match_all": {} } },
        "errors":     { "filter": { "range": { "status": { "gte": 500 } } } },
        "error_ratio": {
          "bucket_script": {
            "buckets_path": { "e": "errors>_count", "t": "total>_count" },
            "script": "params.e / params.t * 100"
          }
        },
        "top_err_uri": {
          "terms": { "field": "uri", "size": 3,
                     "order": { "err_count": "desc" } },
          "aggs": {
            "err_count": { "filter": { "range": { "status": { "gte": 500 } } } }
          }
        }
      }
    }
  }
}

异常率曲线配上"top_err_uri"钻取，异常率一抬，鼠标一点就知道是哪个接口在炸——这是比"看告警再翻日志"快得多的排查路径。

4.4 指标四：慢接口排行

两块面板配合：

面板 A：慢接口 TOP 排行（Table）——按 uri 分组，指标用 P95 of costMs 降序取 Top 10：

POST api-access-*/_search
{
  "size": 0,
  "query": {
    "bool": {
      "filter": { "range": { "ts": { "gte": "now-1d" } } }
    }
  },
  "aggs": {
    "slow_top10": {
      "terms": {
        "field": "uri", "size": 10,
        "order": { "p95_cost": "desc" }
      },
      "aggs": {
        "p95_cost": {
          "percentiles": { "field": "costMs", "percents": [50, 95, 99] }
        },
        "call_count": { "value_count": { "field": "costMs" } }
      }
    }
  }
}

面板 B：耗时分布直方图（Histogram）——横轴 costMs 间隔 50ms，一眼看出整体耗时形态和长尾。

两个实践提醒：① 慢接口排行不要只看平均值——平均值被大响应掩盖，P95/P99 才是用户体感；② 排行榜要结合调用量看——P95 高但日调用 3 次的接口，优先级低于 P95 高且日调用 10 万次的接口，表格里同时放 count 列。

4.5 大屏组装：一块电视屏放什么

按"先看大盘、再看细节"的层次组织 Kibana Dashboard：

┌────────────────────────────────────────────────────────────────┐
│  业务监控大屏                    时间范围: Today   自动刷新: 1min  │
├──────────────┬──────────────┬──────────────┬────────────────────┤
│  DAU          │ 总请求量      │ 异常率        │ P95 耗时           │
│  (Metric)    │  (Metric)    │  (Metric)    │  (Metric)          │
├──────────────┴──────────────┼──────────────┴────────────────────┤
│  近 24h 请求量 + 异常率趋势    │  近 30 天 DAU 趋势                 │
│  (Line, 双轴)               │  (Line)                           │
├─────────────────────────────┼───────────────────────────────────┤
│  接口调用 TOP10 (Table)       │  慢接口 P95 排行 (Table)           │
├─────────────────────────────┼───────────────────────────────────┤
│  各端流量占比 appId (Pie)     │  异常 Top 错误信息 (Table)          │
└─────────────────────────────┴───────────────────────────────────┘

大屏模式（Dashboard → 满屏 + 自动刷新 1 分钟）投到会议室电视上，数据自己说话，周会再也不用临时跑数。

五、Kibana 仪表盘的导出与复用

5.1 Saved Objects 导入导出

Kibana 的仪表盘是 Saved Object，原生支持导出 JSON 分发给其他环境：

Stack Management → Saved Objects → 导出：
  勾选 Dashboard + 其依赖的 Visualization / Data View / Tsvb 面板
  → 导出 ndjson 文件（约 50~200KB）

导入（新环境）：
  Stack Management → Saved Objects → Import
  → 勾选 "Automatically overwrite conflicts"（同名覆盖）
  → 注意：导入后检查 Data View（原 index pattern）的 ID 是否匹配
    不匹配时手动把面板指向新环境的数据视图即可

团队协作姿势：把导出的 ndjson 放进基础设施仓库（和 Grafana Dashboard 的 JSON 同等待遇），环境重建/新团队接入直接导入——大屏配置也是代码资产。

5.2 Data View 命名规范

命名：biz-access-{env}（如 biz-access-prod）
时间字段：ts
字段过滤前缀：只保留 mappings 里定义的业务字段（隐藏元数据噪音）

多环境多应用共用一套 Dashboard 的技巧：面板的过滤器用变量占位（Kibana 的 Control Visualization 放一个 env/appId 下拉框在 Dashboard 顶部），一套大屏切换环境用。

六、成本控制与常见坑

6.1 日志量与存储成本

API 访问日志是全量高频日志，量的估算和成本控制必须提前算：

四个成本杠杆：

采样代码（拦截器里三行）：

// 高频心跳类接口采样：只打 10% 的日志，指标统计时按 10 倍还原
if (isHighFrequencyPolling(uri) && ThreadLocalRandom.current().nextInt(10) != 0) {
    return;
}

6.2 六个高频坑

七、常见问题

7.1 我们用 Loki 不用 ES，能做这套吗？

能，但形态不同。Loki 的强项是日志检索和 LogQL 聚合（sum by (uri) (rate({job="api-access"}[5m]))），Grafana 里也能做 DAU/TOP10/异常率面板（LogQL 的 count_over_time + unwrap costMs）。差距在两点：① Loki 的基数处理不如 ES，userId 去重（DAU）只有近似实现，大规模用户下精度和性能不如 ES 的 cardinality；② 没有类似 Lens 的拖拽分析，指标定义都要写 LogQL。建议：排查向的日志走 Loki（可观测性那篇的体系），业务分析向的访问日志走 ES——两者不冲突，API_ACCESS logger 分流即可。

7.2 用 ES 的 Kafka 中间再加一层（Filebeat→Kafka→Logstash→ES）有必要吗？

看量级。日均千万条以内，Filebeat 直发 ES 完全够；日均亿级以上、或 ES 有明显峰谷压力时，加 Kafka 做削峰缓冲才值得。中间层不是免费的：多两个组件的运维、配置、故障点。先直发，量到了再上 Kafka，不要为了架构图好看加层。

7.3 userId 为空（匿名流量）怎么处理？

三步：① 日志照打（userId=null），匿名流量本身就是分析对象（爬虫/未登录用户的转化空间）；② DAU 只统计 userId IS NOT NULL；③ 匿名流量用 clientIp 近似去重（口径叫"独立 IP"，别和 DAU 混淆）。更进一步的"设备维度 DAU"需要前端配合打 deviceId 到 header——这就进入了埋点方案的地盘，按 1.2 的边界决定。

7.4 接口路径带 ID 怎么归一化？

拦截器里把真实 URI 归一化成 pattern：/api/order/10086 → /api/order/{id}。Spring MVC 下最简单的做法是用 HandlerMethod 拿到匹配的 @RequestMapping 路径模板：

// afterCompletion 里：
if (handler instanceof HandlerMethod hm) {
    String pattern = hm.getMethodMapping().toString();   // 如 /api/order/{orderId}
    // 用 pattern 替代 requestURI 打日志
}

归一化后的 uri 才有聚合价值——这一步漏了，TOP10 会全是具体订单号的碎片。

7.5 这套和 Prometheus/Grafana 监控什么关系？

互补关系，数据源和分析范式都不同：Prometheus 是拉模型 + 预聚合时序，适合"固定的、连续的"指标（QPS/P99/错误率告警），成本极低；ES 日志方案是存原始事件 + 按需聚合，适合"事后想切就切"的探索式分析（任意维度组合、TopN、下钻）。我们的分工：Prometheus 做告警（必须先定义指标），日志大屏做业务探索（想到哪切到哪）——可观测性那篇的 HTTP 指标和本篇的 API 日志字段高度重合，埋一次拦截器两边都受益。

7.6 Kibana 大屏很慢/超时怎么办？

按顺序检查：① 时间范围是不是太大（大屏固定 Today）；② 聚合字段是否高基数（userAgent/裸 uri）；③ 索引分片是否过碎（每天一个小索引分 10 片就是浪费，单索引 3~5 片够用）；④ 热节点内存是否打满（ILM 及时转移温数据）；⑤ 最后手段：定时任务预聚合——每小时把上一小时的统计结果写入一个小索引，大屏直接读预聚合结果，查询成本降一个数量级。

八、总结

链路速查卡

┌─────────────┬──────────────────────────────────────────────────┐
│ 环节         │ 关键配置                                          │
├─────────────┼──────────────────────────────────────────────────┤
│ 日志结构化   │ ApiAccessLogInterceptor + LogstashEncoder        │
│             │ 一次请求一行 JSON；userId/uri/status/costMs 必打   │
│ 采集        │ Filebeat filestream + ndjson parser + registry   │
│             │ 解耦采集、断点续传、至少一次投递                    │
│ 存储        │ ES 索引模板：聚合字段 keyword、数值字段 integer     │
│             │ ILM 热温分层控成本                                 │
│ 可视化       │ Lens 拖拽四指标 → Dashboard 大屏 → ndjson 导出复用 │
├─────────────┼──────────────────────────────────────────────────┤
│ 边界        │ 服务端行为分析用日志；前端转化分析用埋点；           │
│             │ 告警用 Prometheus；探索式分析用日志大屏              │
└─────────────┴──────────────────────────────────────────────────┘

四指标实现一览

关键数据（我们的真实落地）

• 接入成本：1 个拦截器 + 3 份配置，一个下午完成
• 大屏上线当天回答了老板的 DAU/TOP10 问题，后续月均支撑 20+ 次临时取数需求
• 日志量：5000 万请求/天 → 采样后 ES 存储 ~10GB/天，保留 30 天
• 临时取数响应：从"提需求等排期"变成"打开 Kibana 切两下，30 秒"

一句话

业务监控的门槛不在工具，而在"日志里有没有那几个字段"——userId、uri、status、costMs 四个字段打对了，DAU、TOP10、异常率、慢接口全是现成的聚合函数。日志方案不会替代埋点，但它让"看数据"这件事从排期清单变成了自助餐：零埋点、零 SDK、零额外成本，现有 ELK 栈一个下午就能跑起来。

给团队的建议

互动话题：你们团队的 DAU 是怎么统计的？有没有被"提个数要等三天"折磨过？日志大屏在你公司落地后最意外的一个发现是什么？评论区聊聊。更多技术文章欢迎关注我的公众号：服务端技术精选。

参考资料

• logstash-logback-encoder（LogstashEncoder）
• Filebeat 官方文档：filestream input
• Elasticsearch cardinality 聚合文档
• ES 索引模板与动态映射
• ES ILM 生命周期管理
• Kibana Lens 可视化
• Kibana Saved Objects 导入导出
• OWASP：日志安全（脱敏注意事项）