logstash 配置

input {
    file {
        path => "/logs/*.log" #日志路径
        codec => multiline {
             pattern => "^%{TIMESTAMP_ISO8601}"
             negate => true
             what => "previous"
        }
    }
}

filter {
    if [path] =~ "access" {
        mutate { replace => { type => "access" } }
        grok {
            match => {
                "message" => "%{TIMESTAMP_ISO8601:timestamp} %{INT:cost} %{IP:remoteIp}:%{POSINT:remotePort} %{IP:localIp}:%{POSINT:localPort} %{PATH:uri} %{INT:httpCode}"
            }
            remove_field => ["message"]
        }
    } else if [path] =~ "server" {
        mutate { replace => { type => "server" } }
        grok {
            match => {
                "message" => "%{TIMESTAMP_ISO8601:timestamp} \[%{LOGLEVEL:level}\] \[%{JAVACLASS:class}\] \[%{DATA:thread}\] - %{GREEDYDATA:content}"
            }
            remove_field => ["message"]
        }
    } else {
        mutate { replace => { type => "random_logs" } }
    }
    date {
        match => [ "timestamp", "yyyy-MM-dd HH:mm:ss.SSS" ]
    }
}

output {
    elasticsearch {
      hosts => ["http://192.168.201.37:9200"]
      index => "local_test"
    }
}

说明：
1 . multiline 处理一个事件由多行日志构成的情况，用时间戳标记新事件。
2 . =~ 正则匹配日志名。
3 . mutate 替换默认属性type的值
4 . remove_field 删除原日志
5 . date 用业务时间戳替换日志写入时间戳

日志举例

1. access-log

2017-04-13 09:23:52.725 6 127.0.0.1:53289 127.0.0.1:9092 /user/item/11 200

2. server-log

2017-04-13 11:13:33.766 [ERROR] [com.chengying.web.UserController] [http-nio-9092-exec-7] - item id 11
com.netflix.hystrix.exception.HystrixRuntimeException: ResourceQuery#queryResourceItem(String) failed and no fallback available.
    at com.netflix.hystrix.AbstractCommand$22.call(AbstractCommand.java:805)
    at com.netflix.hystrix.AbstractCommand$22.call(AbstractCommand.java:790)
    at com.netflix.hystrix.AbstractCommand$DeprecatedOnFallbackHoo
    at rx.observers.Subscribers$5.onError(Subscribers.java:230)