ELK(ElasticSearch+Logstash+Kibana)+redis日志收集分析系统

日志主要包括系统日志、应用程序日志和安全日志。系统运维和开发人员可以通过日志了解服务器软硬件信息、检查配置过程中的错误及错误发生的原因。经常分析日志可以了解服务器的负荷，性能安全性，从而及时采取措施纠正错误。

通常，日志被分散的储存不同的设备上。如果你管理数十上百台服务器，你还在使用依次登录每台机器的传统方法查阅日志。这样是不是感觉很繁琐和效率低下。当务之急我们使用集中化的日志管理，例如：开源的syslog，将所有服务器上的日志收集汇总。

集中化管理日志后，日志的统计和检索又成为一件比较麻烦的事情，一般我们使用grep、awk和wc等Linux命令能实现检索和统计，但是对于要求更高的查询、排序和统计等要求和庞大的机器数量依然使用这样的方法难免有点力不从心。

开源实时日志分析ELK平台能够完美的解决我们上述的问题，ELK由ElasticSearch、Logstash和Kiabana三个开源工具组成。官方网站：https://www.elastic.co

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

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

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

开源实时日志分析ELK平台部署流程：

1、安装Logstash依赖包JDK

 Logstash的运行依赖于Java运行环境， Logstash 1.5以上版本不低于java 7推荐使用最新版本的Java。由于我们只是运行Java程序，而不是开发，下载JRE即可。首先，在Oracle官方下载新版jre，下 载地址：http://www.oracle.com/technetwork/java/javase/downloads/jre8-downloads-2133155.html

# tar -zxf jdk-8u45-linux-x64.tar.gz -C /usr/local/
设置环境变量 /etc/profile，添加一下代码
  export JAVA_HOME=/usr/local/jdk1.8.0_45
  export PATH=$PATH:$JAVA_HOME/bin
  exportCLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar:$CLASSPATH
 # source /etc/profile
测试是否安装成功
 # java -version

2、安装Logstash （日志收集、分析，并将其存储供以后使用）

下载并安装Logstash，安装logstash只需将它解压的对应目录即可，例如：/usr/local下：

下载地址https://www.elastic.co/downloads/logstash

  # https://download.elastic.co/logstash/logstash/logstash-2.4.0.tar.gz
  # tar –zxf logstash-2.4.0.tar.gz -C /usr/local/

安装完成后运行如下命令启动logstash：

• 普通方式启动

# logstash -e 'input { stdin { } } output { stdout {} }'
 Settings: Default pipeline workers: 2
 Pipeline main started
 等待输入: hello world
 2016-09-06T06:22:50.326Z localhost.localdomain hello world 
 我们可以看到，我们输入什么内容logstash按照某种格式输出，其中-e参数参数允许Logstash直接通过命令行接受设置。这点尤其快速的帮助我们反复的测试配置是否正确而不用写配置文件。使用CTRL-C命令可以退出之前运行的Logstash。

 使用-e参数在命令行中指定配置是很常用的方式，不过如果需要配置更多设置则需要很长的内容。

 这种情况，我们首先创建一个简单的配置文件，并且指定logstash使用这个配置文件。

 例如：在logstash安装目录下创建一个“基本配置”测试文件logstash-test.conf，文件内容如下：

 # cat logstash-test.conf
     input { stdin { }}
     output { stdout { codec=> rubydebug }}

  Logstash使用input和output定义收集日志时的输入和输出的相关配置，本例中input定义了一个叫"stdin"的input，output定义一个叫"stdout"的output。

 无论我们输入什么字符，Logstash都会按照某种格式来返回我们输入的字符，其中output被定义为"stdout"并使用了codec参数来指定logstash输出格式。 

• 开启debug模式启动

   使用logstash的-f参数来读取配置文件然后启动，执行如下开始进行测试：(debug方式启动)

# /usr/local/logstash-2.3.4/bin/logstash agent -f /usr/local/logstash-2.3.4/conf/logstash-test.conf
   Logstash startup completed
   Tue Jul 14 18:07:07 EDT 2015 hello World   #该行是执行echo “`date`hello World” 后输出的结果，直接粘贴到该位置，然后会输出下面结果
    {
         "message" => "Tue Sep  6 14:25:52 CST 2016 hello world",
         "@version" => "1",
         "@timestamp" => "2016-09-06T06:26:31.270Z",
         "host" => "localhost.localdomain"
     }

2.1、logstash输出信息到redis数据库中

 刚才我们是将信息直接显示在屏幕上了，现在我们将logstash的输出信息保存到redis数据库中,

  logstash输出信息存储到redis数据库中,redis其实就充当一个消息队列，并不作为存储，等待elasticsearch的消费

  前提是本地安装有redis数据库，那么下一步我们就是安装redis数据库.

# cat /usr/local/logstash-2.3.4/conf/logstash_index_redis.conf  将信息存储到redis
      input { stdin { } }
      output {
          stdout { codec => rubydebug }
          redis {
              host => '10.2.8.45'
              port => 6379
              data_type => 'list'
              key => 'logstash'
          }
      }

  logstash的默认端口号9301

   # netstat -tnlp |grep 9301

    服务端的redis收集日志配置文件

# cat logstash_index_redis.conf 从redis读取信息，并且发送到elasticsearch
  input {                       
      redis {
          host => "10.2.8.45"
          port => 6379
          data_type => "list"
          key => "logstash"
          codec => "json"
      }
  }
  output {
      stdout { codec => rubydebug }
      elasticsearch { hosts => "10.2.8.45" }
  }

 客户端将收集的信息存储到redis的配置文件

# cat logstash_agent.conf
  input {
      file {
          path => [ "/var/log/messages", "/var/log/*.log" ]
          type => "system"
          start_position => beginning
      }
  }
  output {
      stdout { codec => rubydebug }
      redis {
          host => "10.2.8.45"
          port => 6379
          data_type => "list"
          key => "logstash"
      }
  #    elasticsearch { hosts => "10.2.8.45" }
  }

3、安装Elasticsearch   

 下载地址 https://www.elastic.co/downloads/elasticsearch

 建立启动elasticsearch的普通用户(默认情况下不能使root用户启动elasticsearch)

# groupadd elk
# useradd elasticsearch -g elk

 安装

 # tar zxvf https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.4.0/elasticsearch-2.4.0.tar.gz -C /usr/local/
 # 修改配置文件 /usr/local/elasticsearch-2.3.5/config/elasticsearch.yml
   network.host: 10.2.8.45   # 端口绑定ip地址
   http.port: 9200

 启动(注意! 默认是不能使用root用户启动，要使用普通用户)

# su - elasticsearch
# nohup /usr/local/elasticsearch-2.3.5/bin/elasticsearch > nohup &
# netstat -anp |grep :9200
# curl http://localhost:9200  #查看当前elastcisearch状态

 接下来我们在logstash安装目录下创建一个用于测试logstash使用elasticsearch作为logstash的后端的测试文件logstash_elasticsearch.conf，

 该文件中定义了stdout和elasticsearch作为output，这样的“多重输出”即保证输出结果显示到屏幕上，同时也输出到elastisearch中。

 # cat /user/local/logstash-2.3.4/conf.d/logstash_elasticsearch.conf
   input { stdin { } }   # 手动输入内容
   output {
      elasticsearch {hosts => "localhost" } # 输出到elasticsearch
      stdout { codec=> rubydebug }         # 输出到屏幕上
   }

执行如下命令

# logstash agent -f /user/local/logstash-2.3.4/conf.d/logstash_elasticsearch.conf
 hello logstash  # 输入
 输出如下结果
   {
      "message" => "hello logstash",
      "@version" => "1",
      "@timestamp" => "2016-09-06T06:49:39.654Z",
      "host" => "localhost.localdomain"
   }

我们可以使用curl命令发送请求来查看ES是否接收到了数据：

# curl http://localhost:9200/_search?pretty

至此，已经成功利用Elasticsearch和Logstash来收集日志数据了

3.1、安装elasticsearch插件

  Elasticsearch-kopf插件可以查询Elasticsearch中的数据，安装elasticsearch-kopf，只要在你安装Elasticsearch的目录中执行以下命令即可：

# cd /usr/local/elasticsearch-2.3.5/bin
# ./plugin install lmenezes/elasticsearch-kopf

执行插件安装后会提示失败，很有可能是网络等情况..
若下载失败可以手动下载该软件，不通过插件安装命令...
 # cd /usr/local/elasticsearch-2.3.5/plugins
 # wget https://github.com/lmenezes/elasticsearch-kopf/archive/master.zip
 # unzip master.zip
 # mv elasticsearch-kopf-master kopf
安装完成后在plugins目录下可以看到 kopf 目录
浏览器访问kopf页面访问elasticsearch保存的数据 http://10.2.8.45:9200/_plugin/kopf/

4、安装Kibana

# wget https://download.elastic.co/kibana/kibana/kibana-4.6.0-linux-x86_64.tar.gz
修改配置文件 /usr/local/kibana-4.6.0-linux-x86_64/config/kibana.yml
    elasticsearch.url: "
 启动(后台启动)
   # nohup /usr/local/kibana-4.6.0-linux-x86_64/bin/kibana > nohup.out &
 默认端口 5601 
   # netstat -tnlp |grep 5601

ELK默认端口
logstash  9301
elasticsearch  9200 9300
kibana   5601

浏览器访问 http://10.2.8.45:5601

 页面错误提示： 

1、This version of Kibana requires Elasticsearch ^2.4.0 on all nodes. I found the following incompatible nodes in your cluster: Elasticsearch v2.3.5 @ 10.2.8.45:9200 (10.2.8.45)
   说明是Kibana版本和elasticsearch的版本不匹配，查看官网，安装匹配版本
2、unable to fetch mapping. 这就说明logstash没有把日志写入到elasticsearch。
   检查logstash与elasticsearch之间的通讯是否有问题，一般问题就在这。修改完成后重启应用

使用 http://kibanaServerIP:5601访问Kibana，登录后，首先，配置一个索引，默认，Kibana的数据被指向Elasticsearch，使用默认的logstash-*的索引名称，并且是基于时间的，点击“Create”即可

然后点击“Discover”，可以搜索和浏览Elasticsearch中的数据，默认搜索的是最近15分钟的数据。可以自定义选择时间。

到此，说明你的ELK平台安装部署完成。

客户端安装

客户端只需要安装logstash即可，安装方法如上1和2（必须要安装jdk）

然后在/usr/local/logstash-2.3.4/conf/添加logstash_agent.conf配置文件(可按需修改)

# cat logstash_agent.conf
  input {
      file {
          path => [ "/var/log/messages", "/var/log/*.log" ]  # 要收集的日志文件
          type => "system"                 # 自定义类型，会在kibana中显示
          start_position => beginning
      }
  }
  output {
      stdout { codec => rubydebug }
      redis {
          host => "10.2.8.45"
          port => 6379
          data_type => "list"
          key => "logstash"
      }
  #    elasticsearch { hosts => "10.2.8.45" }
  }
后台运行
 # nohup /usr/local/logstash-2.3.4/bin/logstash agent -f /usr/local/logstash-2.3.4/conf/logstash_agent.conf > /usr/local/logstash-2.3.4/logstash.log 2>&1 &
查看进程
 # ps -ef|grep java