使用 log4js UDP 發送數據到 logstash

本文地址 http://www.cnblogs.com/jasonxuli/p/6532723.html

 

因為 nodejs 一般會部署在多台機器，並且每台機器會起多個進程，因此查看日志時往往要人工區分一個完整的請求包含哪些行。如果在日志中添加 服務器名稱和進程id，就比較容易了。

如果在 filebeat 配置中修改正則表達式肯定是可以完成這個工作的，但今天發現 log4js（1.1.1版本） 的模塊 logstashUDP 支持通過 UDP 直接發送數據到 logstash，這就更自由了，這樣就不用在每台機器上跑 filebeat。

 

這里我們假定已經配置好了 ELK。EKL 配置方法參見本博客文章 http://www.cnblogs.com/jasonxuli/p/6397244.html

 

根據 logstashUDP 的代碼，以 log4js.getLogger().info("message", arg1) 為例，第二個參數 arg1 會被作為 kv 結構，與 appender 配置中的 fields 對象合並，然后發送給 logstash。

也就是說，如果 arg1 = {"customField1": "", "customField2": ""}，那么最終在 elasticsearch 中，數據大體結構為： {"fields.customField1": "", "fields.customField2": ""}。

 

appender 配置：

var logLayout = {
                type:"pattern",
                    pattern: "%h %x{pid} [%d] [%p] %c > %m",
                    tokens: {
                    pid: function () {
                        return process.pid
                    }
                }
            };

appenders     : [
        {
            type: "console",
            layout  : logLayout
        },
        {
            type    : "logstashUDP",  // 定義默認的 appender
            level   : "ALL",
            layout  : logLayout, // 自定義 layout
            host    : "192.168.1.123",
            port    : "5050",
            fields : { // fields 會和 第二個參數 arg1 合並，然后發送給 logstash；並且這個 fields 會作為 config.fields 用於之后每次日志事件 machine : os.hostname(), // 指定機器名 processId: process.pid // 指定線程ID }
        }
　　 ],
    replaceConsole: true
}

 

如果想要動態指定某個字段的值，需要包裝一下 logger 的各個方法：

exports.logger = function getLogger (moduleName) {
    var l = log4js.getLogger(moduleName);

    l.trace = wrapper(l.trace);
    l.debug = wrapper(l.debug);
    l.info  = wrapper(l.info);
    l.warn  = wrapper(l.warn);
    l.error = wrapper(l.error);
    l.fatal = wrapper(l.fatal);

    return l;
};

function wrapper(func){
    return function(message){
        var data = {log: Array.prototype.join.call(arguments, ', ')}; // 拼接原始log參數；logger.info("attr1", "attr2"), 那么不會把 "attr2" 當做 data 發給 logstash
        return func.call(this, data.log, data);  // 這里默認傳遞 data 參數給 logstash，不需要在每次 info() 時手動傳遞第二個參數
    }
}

 

logstash 的配置， /etc/logstash/conf.d/first-pipeline.conf

input {
    beats {
        port => "5043"
    }
    udp { port => "5050"
        codec => json // 這一行一定要指定 }
}
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}
output {
    if [fields][category] != "memory" { // 因為 logstashUDP 模塊會將第二個參數合並到 config.fields 中， [][] 的結構用於指定二級字段 fields.category
        elasticsearch {
            hosts => [ "192.168.20.50:9200" ]
            index => "vrslog-%{+YYYY.MM.dd}"
        }
    }
    if [fields][category] == "memory" { // 為 memory 日志生成單獨的 elasticsearch 索引
        elasticsearch {
            hosts => [ "192.168.20.50:9200" ]
            index => "memory-%{+YYYY.MM.dd}"
        }
    }
}