Logstash中日志级别（level）判断条件失效问题求助

问题描述 #

Node.js应用通过Winston将日志发送到Logstash（未使用Filebeat），目标是在Elastic Cloud中根据日志的level（info/error）分别存储到不同索引，但Logstash配置中的level判断条件始终走else分支。仅判断message是否为JSON对象的逻辑正常，level字段的判断完全失效。

核心原因 #

Logstash的TCP输入接收的是JSON格式的字符串，虽然第一个条件判断了message是JSON结构，但没有将其解析为结构化字段，导致[level]字段实际不存在，自然无法匹配判断条件。

解决方案 #

1. 修正Logstash配置 #

在filter中添加json过滤器，将message字段的JSON字符串解析为结构化数据，这样level字段才能被正确引用。同时修正原配置中索引命名的逻辑错误（info日志错误指向了error-logs索引）：

input {
    tcp {
        port => 6000
        codec => json_lines # 可选，直接按JSON行解析，避免后续手动判断
    }
}

filter {
    # 如果不用codec => json_lines，保留以下判断和解析步骤
    # if [message] =~ /^{.*}$/ {
        json {
            source => "message" # 将message字段的JSON字符串解析为结构化字段
            remove_field => "message" # 可选，解析后移除原始message字段
        }
        
        if [level] == 'info' {
            mutate {
                add_field => { "index" => "api-logs-%{+YYYY.MM.dd}" }
            }
        } else if [level] == 'error' {
            mutate {
                add_field => { "index" => "error-logs-%{+YYYY.MM.dd}" }
            }
        } else {
            mutate {
                add_field => { "index" => "other-logs-%{+YYYY.MM.dd}" }
            }
        }
    # } else {
    #     drop {}
    # }
}

output {
    elasticsearch { 
        cloud_id => "cloud_id" 
        cloud_auth => "credentials for cloud"
        index => "%{index}"
    }
    stdout { codec => rubydebug }
}

优化建议：直接在tcp输入中配置codec => json_lines，Logstash会自动解析每行的JSON数据，无需手动判断和解析，简化配置。

2. 优化Node.js的Winston配置（可选但推荐） #

• 无需创建两个logger实例，一个logger即可通过info()/error()方法输出不同level的日志
• 修正代码中的拼写错误：messsage → message
• 推荐使用ecsFormat，让日志格式符合Elastic Common Schema，更适配Elastic生态

修改后的Node.js代码：

const express = require('express');
const winston = require('winston');
const {ecsFormat} = require('@elastic/ecs-winston-format');
const {ElasticsearchTransport} = require('winston-elasticsearch');
const app = express();
app.use(express.json());

// 单个logger实例，支持多level日志
const logger = winston.createLogger({
    level: 'info', // 最低日志级别
    format: ecsFormat(), // 使用ECS格式，更适配Elastic生态
    transports: [
        new ElasticsearchTransport({
            clientOpts: {node: 'http://localhost:6000'},
        })
    ]
});

const signup = async function (req, res) {
    try {
        const { firstName, lastName, phoneNumber, emailId, password } = req.body;
        res.status(200).json({message: "Request successful!"});
        if (req.body) {
            logger.info('New signup request', {
                message: 'New signup request', 
                request_header: req.headers,
                request_body: req.body
            })
        }
    } catch (error) {
        console.log(error);
        logger.error(`Error: ${error.message}`, {
            request_header : req.headers, 
            request_body : req.body
        });
        res.status(500).json({ message: error.message });
    }
};

app.post("/new", signup);

app.listen(7000, console.log("Server is running on port 7000!"))

验证方法 #

启动Logstash和Node.js应用后，发送请求测试：

1. 正常请求会触发info级日志，应写入api-logs-YYYY.MM.dd索引
2. 构造错误请求（比如不传必填参数）触发error级日志，应写入error-logs-YYYY.MM.dd索引
3. 通过Logstash的stdout输出（rubydebug codec）可以看到解析后的结构化字段，确认level字段是否存在

内容的提问来源于stack exchange，提问作者Akhil