logstash.example.conf

input {
    beats {
        port => 5044
    }
    # kafka {
    #     bootstrap_servers => '192.168.10.7:9092'
    #     topics => 'log-collect'
    #     group_id => 'logstash'
    #     codec => json
    #     poll_timeout_ms => "10000"
    #     session_timeout_ms => "120000"
    #     request_timeout_ms => "130000"
    # }
}

filter {
    if [log_type] == "json_log" {
        json {
            skip_on_invalid_json => false
            source               => "message"
            target               => "message"
        }
    }
    else if [log_type] == "nginx_error" {
        grok {
            match => {
                "message" => "(?<timestamp>%{YEAR}[./-]%{MONTHNUM}[./-]%{MONTHDAY}[- ]%{TIME}) \[%{LOGLEVEL:severity}\] %{NUMBER:pid}#%{NUMBER}: %{GREEDYDATA:errormessage}, client: %{IP:client_ip}, server: .*?"
            }
        }

        if "_grokparsefailure" not in [tags] {
            date {
                match => ["timestamp", "yyyy-MM-dd HH:mm:ss"]
                target => "timestamp"
            }

            mutate {
                add_field => {
                    "log_level" => "ERROR"
                }
            }
        }
    }
    # PHP 错误
    else if [log_type] == "php_errors" {
        mutate {
            add_field => {
                "log_level" => "ERROR"
            }
        }
    }

    if [log_level] == "ERROR" {
        mutate {
            add_field => {
                "to_alert" => "adanos"
            }
        }
    }
}

output {
    if [to_alert] == "adanos" {
        http {
            # adanos-alert-agent 服务的 HTTP 监听地址，也可以直接使用
            # adanos-alert-server 的地址（默认端口 19999）
            url          => "http://localhost:29999/api/messages/logstash/"
            http_method  => "post"
            content_type => "json"
            retry_failed => false
            keepalive    => true
        }
    }
}