使用logstash和elasticsearch实现日志链路（一）

elasticsearch：logstash：

大志哥123

496人浏览 · 2026-01-27 09:23:04

大志哥123 · 2026-01-27 09:23:04 发布

1.安装环境

elasticsearch：

logstash：

2.需求

2.1.需求描述

对于中台多个微服务的一个交易链路，能够全部串联查询出来。在中台前端显示。

2.2.日志样例

[linkservera] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:47.309] > [INFO] [http-nio-13000-exec-1] - ========================================== 开始请求 ==========================================
[linkservera] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:47.318] > [INFO] [http-nio-13000-exec-1] - 请求URL: http://localhost:13000/api/common/used
[linkservera] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:47.318] > [INFO] [http-nio-13000-exec-1] - 调用方法: POST
[linkservera] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:47.318] > [INFO] [http-nio-13000-exec-1] - 请求IP: 127.0.0.1
[linkservera] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:47.319] > [INFO] [http-nio-13000-exec-1] - 请求类方法: com.gg.midend.controller.api.CommonApiController.getProcessMemory
[linkservera] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:47.320] > [INFO] [http-nio-13000-exec-1] - 请求参数: {“tradeTime”:“2026-03-22 08:18:00”,“name”:“老王”,“patientId”:“123456”}
[linkservera] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:47.955] > [INFO] [http-nio-13000-exec-1] - 耗时: 634毫秒
[linkservera] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:47.956] > [INFO] [http-nio-13000-exec-1] - 返回结果: {“tradeTime”:“2026-03-22 08:18:00”,“name”:“老王”,“patientId”:“123456”}
[linkservera] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:47.956] > [INFO] [http-nio-13000-exec-1] - ========================================== 请求结束 ==========================================

[linkservera] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:49.491] > [INFO] [http-nio-13000-exec-2] - ========================================== 开始请求 ==========================================
[linkservera] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:49.491] > [INFO] [http-nio-13000-exec-2] - 请求URL: http://localhost:13000/api/common/used
[linkservera] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:49.491] > [INFO] [http-nio-13000-exec-2] - 调用方法: POST
[linkservera] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:49.491] > [INFO] [http-nio-13000-exec-2] - 请求IP: 127.0.0.1
[linkservera] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:49.491] > [INFO] [http-nio-13000-exec-2] - 请求类方法: com.gg.midend.controller.api.CommonApiController.getProcessMemory
[linkservera] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:49.491] > [INFO] [http-nio-13000-exec-2] - 请求参数: {“tradeTime”:“2026-03-22 08:18:00”,“name”:“小王”,“patientId”:“123456”}
[linkservera] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:49.541] > [INFO] [http-nio-13000-exec-2] - 耗时: 50毫秒
[linkservera] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:49.541] > [INFO] [http-nio-13000-exec-2] - 返回结果: {“tradeTime”:“2026-03-22 08:18:00”,“name”:“小王”,“patientId”:“123456”}
[linkservera] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:49.541] > [INFO] [http-nio-13000-exec-2] - ========================================== 请求结束 ==========================================

[linkserverb] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.309] > [INFO] [http-nio-14000-exec-1] - ========================================== 开始请求 ==========================================
[linkserverb] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.318] > [INFO] [http-nio-14000-exec-1] - 请求URL: http://localhost:14000/api/common/used
[linkserverb] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.318] > [INFO] [http-nio-14000-exec-1] - 调用方法: POST
[linkserverb] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.318] > [INFO] [http-nio-14000-exec-1] - 请求IP: 127.0.0.1
[linkserverb] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.319] > [INFO] [http-nio-14000-exec-1] - 请求类方法: com.gg.midend.controller.api.CommonApiController.getProcessMemory
[linkserverb] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.320] > [INFO] [http-nio-14000-exec-1] - 请求参数: {“tradeTime”:“2026-03-22 08:18:00”,“name”:“老王”,“patientId”:“123456”}
[linkserverb] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.955] > [INFO] [http-nio-14000-exec-1] - 耗时: 634毫秒
[linkserverb] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.956] > [INFO] [http-nio-14000-exec-1] - 返回结果: {“tradeTime”:“2026-03-22 08:18:00”,“name”:“老王”,“patientId”:“123456”}
[linkserverb] [695ca6b346e9e8d63bf96eda44f09c6e,3bf96eda44f09c6e] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.956] > [INFO] [http-nio-14000-exec-1] - ========================================== 请求结束 ==========================================

[linkserverb] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.491] > [INFO] [http-nio-14000-exec-2] - ========================================== 开始请求 ==========================================
[linkserverb] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.491] > [INFO] [http-nio-14000-exec-2] - 请求URL: http://localhost:14000/api/common/used
[linkserverb] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.491] > [INFO] [http-nio-14000-exec-2] - 调用方法: POST
[linkserverb] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.491] > [INFO] [http-nio-14000-exec-2] - 请求IP: 127.0.0.1
[linkserverb] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.491] > [INFO] [http-nio-14000-exec-2] - 请求类方法: com.gg.midend.controller.api.CommonApiController.getProcessMemory
[linkserverb] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.491] > [INFO] [http-nio-14000-exec-2] - 请求参数: {“tradeTime”:“2026-03-22 08:18:00”,“name”:“小王”,“patientId”:“123456”}
[linkserverb] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.541] > [INFO] [http-nio-14000-exec-2] - 耗时: 50毫秒
[linkserverb] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.541] > [INFO] [http-nio-14000-exec-2] - 返回结果: {“tradeTime”:“2026-03-22 08:18:00”,“name”:“小王”,“patientId”:“123456”}
[linkserverb] [695ca6b528140ca7802052bea3187450,802052bea3187450] [com.gg.core.aspect.MethodAspect] [2026-01-06 14:07:48.541] > [INFO] [http-nio-14000-exec-2] - ========================================== 请求结束 ==========================================

3.环境配置

3.1.logstash配置

input {
  file {
    path => "/home/geit/midend-center/service3/linkservera/logs/servera.log"
    type => "linkservera"
    start_position => "beginning"
    codec => plain { charset => "UTF-8" }
  }
  file {
    path => "/home/geit/midend-center/service3/linkserverb/logs/serverb.log"
    type => "linkserverb"
    start_position => "beginning"
    codec => plain { charset => "UTF-8" }
  }
}

filter {
  # 1. Grok 解析固定前缀
  grok {
    match => {
      "message" => "\[%{DATA:app}\] \[%{DATA:traceId},%{DATA:spanId}\] \[%{DATA:logger_name}\] \[%{TIMESTAMP_ISO8601:log_time}\] > \[%{LOGLEVEL:level}\] \[%{DATA:thread_name}\] - %{GREEDYDATA:msg_raw}"
    }
  }

  mutate {
    rename => { "msg_raw" => "message" }
  }

  # 2. 解析 "请求参数:" 后面的 JSON 字符串
  if [message] =~ /^请求参数: / {
    grok {
      match => {
        "message" => "^请求参数: %{GREEDYDATA:json_str}"
      }
    }

    json {
      source => "json_str"
      target => "biz_params"
      remove_field => [ "json_str" ]
    }

    # 3. 【关键】将 JSON 里的字段提升到根层级
    # 这样无论 JSON 里有什么字段，都会直接成为 ES 的可搜索字段
    mutate {
      rename => { "[biz_params][tradeTime]" => "tradeTime" }
      rename => { "[biz_params][patientId]" => "patientId" }
      rename => { "[biz_params][name]" => "name" }
      # 如果有其他字段，可以继续加，或者写 ruby 脚本自动遍历 biz_params 提取
      # rename => { "[biz_params][orderId]" => "orderId" }
    }
  }

  # 4. 时间处理
  date {
    match => ["log_time", "yyyy-MM-dd HH:mm:ss.SSS"]
    target => "@timestamp"
  }
  
  mutate {
    remove_field => [ "log_time", "msg_raw" ]
  }
}

output {
  if [type] == "linkservera" or [type] == "linkserverb" {
    elasticsearch {
      hosts => ["http://127.0.0.1:9200"]
      index => "microservice-logs-%{+YYYY.MM.dd}"
    }
  }
}

3.2.设置启动

去除zipkin启动

必须使用geit用户（非root)启动，然后，报错：

java.io.FileNotFoundException: /home/geit/midend-center/service3/elasticsearch/logs/geit-es-cluster_server.json (权限不够)

修复：

4.写Java客户端接口

引入依赖：

        <!--新增-->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.13.5</version>
        </dependency>

        <!-- Jakarta JSON API（避免 ClassNotFoundException: jakarta.json.spi.JsonProvider） -->
        <dependency>
            <groupId>jakarta.json</groupId>
            <artifactId>jakarta.json-api</artifactId>
            <version>2.0.1</version>
        </dependency>

        <!-- JSON 实现（Parsson） -->
        <dependency>
            <groupId>org.eclipse.parsson</groupId>
            <artifactId>parsson</artifactId>
            <version>1.0.0</version>
        </dependency>

        <!-- Elasticsearch Java API Client 7.17（新版，官方推荐） -->
        <dependency>
            <groupId>co.elastic.clients</groupId>
            <artifactId>elasticsearch-java</artifactId>
            <version>7.17.29</version>
        </dependency>

        <!-- 底层 low-level REST Client -->
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.17.29</version>
        </dependency>

部分代码：

测试：

返回空，查看logstash日志：

这里强调一下，一定要仔细确认logstash和ES的启动状体正常。

接口代码：

package com.gg.midend.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import com.gg.midend.config.GlobalConfig;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Service
public class FlexibleTraceService {

    private final ElasticsearchClient esClient;

    // 修正后的正则：匹配 [serviceName] 和 [traceId,spanId]
    private static final Pattern LOG_PATTERN = Pattern.compile(
            "\\[([^\\[\\]]+)\\]\\s+\\[([a-fA-F0-9]+),[a-fA-F0-9]+\\]"
    );

    public FlexibleTraceService(ElasticsearchClient esClient) {
        this.esClient = esClient;
    }

    public List<Map<String, Object>> searchTraces(String tradeDate, String criteriaStr) throws Exception {
        GlobalConfig.log_api.info(">>> [ES Query] 开始查询。tradeDate: [" + tradeDate + "], criteria: [" + criteriaStr + "]");

        BoolQuery.Builder boolBuilder = new BoolQuery.Builder();

        // 1. 处理 tradeDate
        if (tradeDate != null && !tradeDate.trim().isEmpty()) {
            tradeDate = tradeDate.trim();
            if (!tradeDate.matches("\\d{4}-\\d{2}-\\d{2}")) {
                throw new IllegalArgumentException("tradeTime 必须为 yyyy-MM-dd 格式，当前值: " + tradeDate);
            }
            String wildcardPattern = tradeDate + "*";
            boolBuilder.must(Query.of(q -> q
                    .wildcard(w -> w.field("tradeTime.keyword").value(wildcardPattern))
            ));
            GlobalConfig.log_api.info(">>> [ES Query] 添加日期匹配条件: tradeTime.keyword LIKE '" + wildcardPattern + "'");
        }

        // 2. 处理 criteria
        if (criteriaStr != null && !criteriaStr.trim().isEmpty()) {
            criteriaStr = criteriaStr
                    .replace("“", "")
                    .replace("”", "")
                    .replace("\"", "")
                    .replace("'", "")
                    .trim();

            if (!criteriaStr.isEmpty()) {
                String[] values = criteriaStr.split("\\|");
                BoolQuery.Builder shouldBool = new BoolQuery.Builder();

                for (String value : values) {
                    value = value.trim();
                    if (!value.isEmpty()) {
                        String searchValue = value;
                        shouldBool.should(Query.of(q -> q.term(t -> t.field("patientId.keyword").value(searchValue))));
                        shouldBool.should(Query.of(q -> q.term(t -> t.field("name.keyword").value(searchValue))));
                        shouldBool.should(Query.of(q -> q.term(t -> t.field("orderId.keyword").value(searchValue))));
                        shouldBool.should(Query.of(q -> q.term(t -> t.field("phone.keyword").value(searchValue))));
                        shouldBool.should(Query.of(q -> q.term(t -> t.field("idCard.keyword").value(searchValue))));
                    }
                }

                shouldBool.minimumShouldMatch("1");
                boolBuilder.must(Query.of(q -> q.bool(shouldBool.build())));
                GlobalConfig.log_api.info(">>> [ES Query] 添加关键字搜索: [" + criteriaStr + "]，字段: [patientId, name, orderId, phone, idCard]");
            }
        }

        // 3. 执行查询
        SearchRequest request = SearchRequest.of(s -> s
                .index("microservice-logs-*")
                .query(Query.of(q -> q.bool(boolBuilder.build())))
                .sort(sort -> sort.field(f -> f.field("@timestamp").order(SortOrder.Asc)))
                .source(src -> src.filter(f -> f.includes(
                        "tradeTime",
                        "patientId", "name", "orderId", "phone", "idCard",
                        "@timestamp", "content"
                )))
                .size(100)
        );

        GlobalConfig.log_api.info(">>> [ES Query] 执行查询请求...");
        SearchResponse<Map> response = esClient.search(request, Map.class);

        long totalHits = response.hits().total().value();
        GlobalConfig.log_api.info(">>> [ES Query] ES 命中文档总数: " + totalHits);

        // 4. 组装结果，并提取 serviceName 和 traceId
        List<Map<String, Object>> results = new ArrayList<>();
        for (Hit<Map> hit : response.hits().hits()) {
            Map<String, Object> source = hit.source();
            if (source != null) {
                String content = (String) source.get("content");
                String serviceName = "";
                String traceId = "";

                if (content != null) {
                    Matcher matcher = LOG_PATTERN.matcher(content);
                    if (matcher.find()) {
                        serviceName = matcher.group(1); // 第一个 [] 内容
                        traceId = matcher.group(2);     // 第二个 [] 中的第一个 ID
                    }
                }

                source.put("serviceName", serviceName);
                source.put("traceId", traceId);

                results.add(source);
            }
        }

        return results;
    }
}

logstash的pipeline.conf配置（现阶段后面还有很多优化）

input {
  file {
    path => ["/home/geit/midend-center/service3/linkservera/logs/servera.log"]
    start_position => "beginning"
    codec => plain { charset => "UTF-8" }
    add_field => { "service" => "linkservera" }
  }

  file {
    path => ["/home/geit/midend-center/service3/linkserverb/logs/serverb.log"]
    start_position => "beginning"
    codec => plain { charset => "UTF-8" }
    add_field => { "service" => "linkserverb" }
  }
}

filter {
  # 第一步：解析主日志格式
  grok {
    match => {
      "message" => "^  $ %{DATA:tmp_service} $     $ %{DATA:trace_id} $     $ %{DATA:logger} $     $ %{TIMESTAMP_ISO8601:log_timestamp} $   >   $ %{LOGLEVEL:level} $     $ %{DATA:thread} $   - %{GREEDYDATA:content}"
    }
  }

  # 【关键修复】将 trace_id 重命名为 traceId，供 ES 查询使用
  mutate {
    rename => { "trace_id" => "traceId" }
  }

  # 如果 grok 失败，保留原始 message
  if ![content] {
    mutate {
      rename => { "message" => "content" }
    }
  }

  # 第二步：从 content 中提取请求参数 JSON
  if [content] {
    grok {
      match => {
        "content" => "请求参数[:：]\s*%{GREEDYDATA:request_payload}"
      }
      tag_on_failure => []
    }
  }

  # 第三步：清洗并解析 JSON
  if [request_payload] {
    mutate {
      strip => ["request_payload"]
      gsub => [
        "request_payload", "“", '"',
        "request_payload", "”", '"',
        "request_payload", "[\r\n\t]", ""
      ]
    }

    json {
      source => "request_payload"
      target => "req"
    }

    if [req] {
      mutate {
        rename => { "[req][tradeTime]" => "tradeTime" }
        rename => { "[req][name]"     => "name" }
        rename => { "[req][patientId]" => "patientId" }
      }
    }
  }

  # 第四步：时间戳处理
  if [log_timestamp] {
    date {
      match => ["log_timestamp", "yyyy-MM-dd HH:mm:ss.SSS", "ISO8601"]
    }
  }
}

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

  stdout {
    codec => rubydebug
  }
}

5.第一个接口测试

测试：

后续开发第二个接口，通过tradeId，查询整个链路返回

腾讯云开发者社区

腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。

更多推荐

Elasticsearch复杂数据类型终极指南：从入门到精通

Elasticsearch作为功能强大的搜索引擎，支持多种复杂数据类型，让开发者能够灵活处理各种结构化和非结构化数据。本文将带你全面了解Elasticsearch中的复杂数据类型，从基础概念到实际应用，助你轻松掌握数据建模的核心技巧。## 内部对象：构建层级化数据结构在Elasticsearch中，对象类型（Object）是最基础的复杂数据类型之一，用于表示具有嵌套关系的数据。例如，我们可

腾讯云开发者社区

终极指南：Flink SQL连接器版本管理从混乱到有序的升级之路

Apache Flink作为流处理领域的佼佼者，其SQL连接器的版本管理一直是开发者面临的核心挑战。本文将系统讲解Flink SQL连接器版本管理的最佳实践，帮助你轻松应对版本兼容性问题，实现从混乱到有序的升级之旅。## 连接器版本管理的常见痛点 😫在Flink应用开发中，连接器版本管理常常让开发者头疼不已。不同版本的连接器可能导致各种兼容性问题，例如API变更、功能差异甚至运行时错误。

腾讯云开发者社区

如何快速搭建Neon无服务器PostgreSQL：面向初学者的完整指南

Neon是一款革命性的无服务器PostgreSQL解决方案，它通过分离存储和计算层，实现了自动扩缩容、类代码式数据库分支以及零级扩展能力。本指南将帮助你从零开始搭建Neon开发环境，体验这款创新数据库的强大功能。## 准备工作：环境要求与依赖项在开始搭建Neon环境前，请确保你的系统满足以下要求：- Linux操作系统（推荐Ubuntu 20.04+或Debian 11+）- Git