ELK 框架

• Beats: Beats只做一件事，收集并传输数据。它本身是一个轻量级的 agent，通过收集本机的数据，并发往 Logstash 和 Elasticsearch。这其中又包括：
  
• FILEBEAT：使用 backpressure-sensitive 协议，当 Logstash 或者 Elasticsearch 无法承受负荷时，FILEBEAT 会减慢发送速率知道下游恢复。
• METRICBEAT：收集系统指标，从 CPU 到内存，也可以是 Redis, Nginx, Apache, MySQL, ZooKeeper。
• PACKETBEAT：收集网络数据，包括 HTTP，Thrift-RPC, DNS, MySQL, Redis 等
• WINLOGBEAT（好奇怪的感觉）。
• Logstash：收集不同的数据，并做清洗和过滤，建立 Mapping 并将数据送往。
• Elasticsearch。Elasticsearch：分布式查询和分析引擎，其实也是一个存储引擎。提供 TESTful API。
• Kibana：图形化展示你想要看到的数据。
  

为什么选用ELK

1. 开源，意味着更多新的特性和想法可以提前获取
2. 从收集到展示非常全面，并且相关参考资料也很全面，并且现在版本全面升级，性能有恨到提高
3. 处理方式灵活，插件丰富，检索高效。

配置和使用

• Log 收集
• 这里我们使用了 Filebeat。首先从官网上下载最新的 Filebeat，配置 filebeat.yml 文件，基本配置如下：

• filebeat.prospectors:
  - input_type: log
    document_type: hello_world
    paths:
    - /data/log/*.log
  
  output.logstash:
    hosts:["localhost:5044"]
  

这里需要注意的一点是 document_type， 它的作用是在于 Indexer 的时候，通过这个 字段来匹配相应的 filter 和 output。

• Indexer
  
• 这里使用 Logstash 做为 Indexer, 我们采用对不同的输入类型的 Log 采用不同的 config 方式，通过 Log 的 type 来做匹配，这样，就可以动态扩展配置，并且不会影响到现有的逻辑。启动命令入下：

• bin/logstash -f logstash_configuration/conf.d --config.reload.automatic
  

  这里，采用 automatic 的方式，部署后会重启 logstash, conf.d 目录里面保存所有的配置文件，基本格式如下：

• input {
      beats {
          port => 5044
          codec => plain
      }
  }
  
  filter {
      if [type] == "halo_api" {
          grok {
              match => {
                  "message" => "\[(?<log_level>\w+)\s+(?<year>\d\d)%{MONTHNUM:month}%{MONTHDAY:day}\s%{TIME:time}\s+%{DATA:trace_id}\s+%{DATA:module}:%{NUMBER:line_no:int}\s+%{DATA:hostname}:%{NUMBER:process:int}\]\s+(?<log_message>[\s\w\W]+)"            }
          }
          translate {
              field => "log_level"
              destination => "log_level"
              override => "true"
              dictionary => ["A", "ALERT", "T", "TRACE", "D", "DEBUG",
                  "N", "NOTICE", "W", "WARN", "I", "INFO", "E", "ERROR",
                  "C", "CRITICAL", "F", "FATAL"]
          }
  
          mutate {
              add_field => {
                  "log_timestamp" => "20%{year}-%{month}-%{day} %{time}"
              }
          }
  
          date {
              match => ["log_timestamp", "yyyy-MM-dd HH:mm:ss"]
              timezone => "Etc/GMT"
              target => "log_timestamp"
          }
      }
  }
  output {
      if [type] == "halo_api" {
          elasticsearch {
              hosts => ["halo-es01:9200"]
              index => "%{type}-%{+yyyy.MM.dd}"
              template => "/data/apps/logstash_configuration/templates/halo_api.json"
              template_name => "halo_api.json"
              template_overwrite => true
          }
      }
  }
  


有用的插件， 可以通过 logstash-plugin install 安装：
• logstash-filter-translate：将原 field 转换为你所想的 field
• logstash-filter-mutate：功能很强大，比如做拼接， 替换字符，生成新的 field等等
• logstash-filter-date：处理时间，可以转换成你想要的格式
• 更多配置
• workers：运行 filter 和 output 的 pipeline 线程数量。默认是 CPU 核数。
• batch：
• size：每个 Logstash pipeline 线程，在执行具体的 filter 和 output 函数之前，最多能累积的日志条数。默认是 125 条。越大性能越好，同样也会消耗越多的 JVM 内存。
• delay：每个 Logstash pipeline 线程，在打包批量日志的时候，最多等待几毫秒。默认是 5 ms。

• pipeline:
      workers: 24
      batch:
          size: 125
          delay: 5
  

  
  
• 坑：
• 当你需要多个 Mapping 的时候，记住一定要在 output 中增加 template_name 和 template_overwrite 字段。
• 对于 Mapping 来讲，ES 默认的 filter 是 lowercase 的，因此需要自己定义一个 filter：

•  "settings": {
      "index.refresh_interval": "15s",
      "analysis": {
          "analyzer": {
              "my_custom_analyzer": {
                  "type": "custom",
                  "tokenizer": "standard",
                  "filter": "standard"
              }
         }
      }
    },
  

  
  
• Elasticsearch
• 没有做更多的优化，按照网上的教程配置。