ELK 由 ElasticSearch 、 Logstash 和 Kiabana 三个开源工具组成。官方网站： https://www.elastic.co/products

Elasticsearch 是个开源分布式搜索引擎，它的特点有：分布式，零配置，自动发现，索引自动分片，索引副本机制， restful 风格接口，多数据源，自动搜索负载等。

Logstash 是一个完全开源的工具，他可以对你的日志进行收集、分析，并将其存储供以后使用。

kibana 也是一个开源和免费的工具，他 Kibana 可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，可以帮助您汇总、分析和搜索重要数据日志。

elk工作流程：
数据–[输入数据]–>Logstash–[过滤输出数据]–>Elasticsearch–[展示分析数据]–>kibana

这里主要说下Logstash的配置（其他安装配置都比较简单，且文档丰富）
常规日志格式
1、 首先是配置Nginx访问日志的格式，根据日志分析需求配置

 log_format access '$http_host $remote_addr - $remote_user [$time_local] "$request" '
            '$status $body_bytes_sent "$http_referer" '
            '"$http_user_agent" '
            '$request_time $upstream_response_time '
            '"$http_x_forwarded_for"';

2、 Logstash，它主要是连接数据和es的管道，一切数据源，过滤，输出地址均在logstash配置文件中进行配置，写配置文件的时候需要根据elk的版本号对应配置，比如我在配置过程中filter中没有type设置项，执行时就会有报错，可以根据报错信息进行对应修改

 input {
            file {
                    path => "/Users/User/log/access.log"
                    type => "web_nginx_access"
            }
    }
    filter {
    if [type] == "web_nginx_access" {
            grok {
                    match => [
                    "message", "%{IPORHOST:http_host} %{IPORHOST:user_ip} - - \[%{HTTPDATE:timestamp}\] \"(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:httpversion:float})?|%{DATA:rawrequest})\" %{NUMBER:response:int} (?:%{NUMBER:bytes:int}|-) %{QS:referrer} %{QS:useragent} (?:%{NUMBER:request_time:float}|-) (?:%{NUMBER:upstream_time:float}|-)"
                    ]
            }
    }
            geoip {
                    source => "user_ip"
            }
    
    if [type] == "web_nginx_access" {
            date {
                    match => [ "timestamp" , "dd/MMM/YYYY:HH:mm:ss Z" ]
            }
    }
            useragent {
                    target => "ua"
                    source => "useragent"
            }
    }
    output {
        elasticsearch {
             hosts => "127.0.0.1"
            index => "logstash-web_nginx_access-%{+YYYY.MM.dd}"
        }
    }

Json日志格式
1、nginx配置使用json格式，logstash的配置会非常简单

2、logstash配置

input {
        file {
                path => "/Users/User/log/access.log"
                type => "web_nginx_access"
                start_position => beginning
                codec => "json"
        }
}

filter {
}
output {
    elasticsearch {
        hosts => "127.0.0.1"
        index => "logstash-web_nginx_access-%{+YYYY.MM.dd}"
    }
}

自定义格式
1、例如程序记录的日志格式是这样的：[NOTICE] [2017-04-05 12:00:00] [404] [/index.php] [500] [id=123&btime=2017&etime=2018] [路径不存在]；就可以在logstash自定义正则匹配，重命名字段
2、logstash配置

input {
    file {
        path => ["/opt/logs/applog/applog_*.log", "***.log"] #日志路径，可以是数组形式
        start_position => beginning #从日志开始位置导入数据
        sincedb_path => "/etc/logstash/conf.d/sincedb" #存储导入进度，删除后会重新导入所有数据
    }
}
filter {
    grok {
        match => { "message" => "(?<log_level>\[(.*?)\])\s*(?<log_time>\[(.*?)\])\s*(?<log_errorid>\[(.*?)\])\s*(?<log_uri>\[(.*?)\])\s*(?<log_runtime>\[(.*?)\])\s*(?<log_query>\[(.*?)\])\s*(?<log_desc>\[(.*?)\])\s*"} #匹配规则，match可以是多个，自动匹配合适的规则表达式
    }
}
output {
   elasticsearch {
       hosts => "127.0.0.1:9200"
       index => "logstash-%{+YYYY.MM.dd}" #生成es文件名称
   }
}