zoukankan      html  css  js  c++  java

  • 六、Kibana使用、logstash扩展插件

    1.导入数据
    批量导入数据并查看 
     
    1.1 导入数据
     
    1) 使用POST方式批量导入数据，数据格式为json，url 编码使用data-binary导入含有index配置的json文件
    ]# scp /var/ftp/elk/*.gz 192.168.1.66:/root/
    kibana ~]# gzip -d logs.jsonl.gz 
    kibana ~]# gzip -d accounts.json.gz 
    kibana ~]# gzip -d shakespeare.json.gz
     
    kibana ~]# curl -X POST "http://192.168.1.61:9200/_bulk"  
    --data-binary @shakespeare.json
     
    kibana ~]# curl -X POST "http://192.168.1.61:9200/xixi/haha/_bulk"
    --data-binary @accounts.json  
    //索引是xixi，类型是haha，必须导入索引和类型，没有索引，要加上
     
    kibana ~]# curl -X POST "http://192.168.1.61:9200/_bulk"   
    --data-binary @logs.jsonl
     
    2）使用GET查询结果
    kibana ~]# curl -XGET 'http://192.168.1.61:9200/_mget?pretty' -d '{
         "docs":[
             {
                "_index":"shakespeare",
                "_type:":"act",
                "_id":0
        },
        {
                "_index":"shakespeare",
                "_type:":"line",
                "_id":0
        },
        {
                "_index":"xixi",
                "_type:":"haha",
                "_id":25
        }
        ]
        }'
     
    1.2 使用kibana查看数据是否导入成功
    1）数据导入以后查看logs是否导入成功，如图所示：
    图片
     
    2）kibana导入数据，如图-2所示：
    kibana ~]# firefox http://192.168.1.66:5601
    图片
    3）成功创建会有logstash-*
     
    4）导入成功之后选择Discover，如图-4所示：
    图片
    注意： 这里没有数据的原因是导入日志的时间段不对，默认配置是最近15分钟，在这可以修改一下时间来显示
     
    5）kibana修改时间，选择Lsat 15 miuntes--选择Absolute--选择时间2015-5-15到2015-5-22，返回，查看结果
    图片
    6）除了柱状图，Kibana还支持很多种展示方式，选择visualize--Forms a new Search
    查看前六名访问网页的国家：
    buckets--Split Slices --Terms--Fieled--geo.rsc--Count-Size(6)
    查看访问网页的人都用什么操作系统的机器
    buckets--小按钮折叠上一个--Split Slices --Terms--Fieled--machine.os--Count-Size(前5)
     
    2. 综合练习
    练习插件
    安装一台Apache服务并配置
    使用filebeat收集Apache服务器的日志
    使用grok处理filebeat发送过来的日志
    存入elasticsearch
     
    2.1 安装logstash
    1）配置主机名，ip和yum源，配置/etc/hosts（请把se1-se5和kibana主机配置和logstash一样的/etc/hosts）
    logstash ~]# vim /etc/hosts
    192.168.1.61 se1
    192.168.1.62 se2
    192.168.1.63 se3
    192.168.1.64 se4
    192.168.1.65 se5
    192.168.1.66 kibana
    192.168.1.67 logstash
     
    2）安装java-1.8.0-openjdk和logstash
    logstash ~]# yum -y install java-1.8.0-openjdk
    logstash ~]# yum -y install logstash
    logstash ~]# java -version
    openjdk version "1.8.0_131"
    OpenJDK Runtime Environment (build 1.8.0_131-b12)
    OpenJDK 64-Bit Server VM (build 25.131-b12, mixed mode)
     
    logstash ~]# touch /etc/logstash/logstash.conf
    logstash ~]# /opt/logstash/bin/logstash  --version
    logstash 2.3.4
    logstash ~]# /opt/logstash/bin/logstash-plugin list //查看插件
    ...
    logstash-input-stdin    //标准输入插件
    logstash-output-stdout    //标准输出插件
    ...
     
    logstash ~]# vim /etc/logstash/logstash.conf
    input{
        stdin{
       }
    }
    filter{
    }
    output{
        stdout{
       }
    }
     
    //启动并测试
    logstash ~]# /opt/logstash/bin/logstash -f /etc/logstash/logstash.conf
    Settings: Default pipeline workers: 2
    Pipeline main started
    aa
    2019-03-01T08:36:25.227Z logstash aa
    bb
    2019-03-01T08:36:29.282Z logstash bb
    备注：若不会写配置文件可以找帮助，插件文档的位置：
     
    3）codec类插件
    logstash ~]# vim /etc/logstash/logstash.conf
    input{
        stdin{
        codec => "json"        //输入设置为编码json
       }
    }
    filter{
    }
    output{
        stdout{
        codec => "rubydebug"   //输出设置为rubydebug
       }
    }
     
    logstash ~]# /opt/logstash/bin/logstash -f /etc/logstash/logstash.conf 
    Settings: Default pipeline workers: 2
    Pipeline main started
    {"a":1}(手打）
    {
                 "a" => 1,
          "@version" => "1",
        "@timestamp" => "2019-03-01T08:39:23.978Z",
              "host" => "logstash"
    }
     
    4）file模块插件
    logstash ~]# vim /etc/logstash/logstash.conf
    input{
      file {
        path          => [ "/tmp/a.log", "/var/tmp/b.log" ]
       sincedb_path   => "/var/lib/logstash/sincedb" //记录读取文件的位置
       start_position => "beginning"   //配置第一次读取文件从什么地方开始
       type           => "testlog"  //类型名称
      }
    }
    filter{
    }
    output{
        stdout{
        codec => "rubydebug"
    }
    }
     
    logstash ~]# touch /tmp/a.log
    logstash ~]# touch /var/tmp/b.log
    logstash ~]# /opt/logstash/bin/logstash -f /etc/logstash/logstash.conf
    (已经在检测中）
     
    另开一个终端：写入数据
    logstash ~]# echo a1 > /tmp/a.log 
    logstash ~]# echo b1 > /var/tmp/b.log
     
    显示数据：
    Settings: Default pipeline workers: 2
    Pipeline main started
    {
           "message" => "a1",
          "@version" => "1",
        "@timestamp" => "2019-03-01T08:43:38.633Z",
              "path" => "/tmp/a.log",
              "host" => "logstash",
              "type" => "testlog"
    }
    {
           "message" => "b1",
          "@version" => "1",
        "@timestamp" => "2019-03-01T08:43:52.671Z",
              "path" => "/var/tmp/b.log",
              "host" => "logstash",
              "type" => "testlog"
    }
        
    5）tcp、udp模块插件
    logstash ~]# vim /etc/logstash/logstash.conf
    input{
      file {
        path          => [ "/tmp/a.log", "/var/tmp/b.log" ]
       sincedb_path   => "/var/lib/logstash/sincedb"
       start_position => "beginning"
       type           => "testlog"
      }
      tcp {
         host => "0.0.0.0"
         port => "8888"
         type => "tcplog"
    }
       udp {
         host => "0.0.0.0"
         port => "9999"
         type => "udplog"
    }
    }
    filter{
    }
    output{
        stdout{
        codec => "rubydebug"
    }
    } 
     
    logstash ~]# /opt/logstash/bin/logstash -f /etc/logstash/logstash.conf 
    //启动，监控中
     
    另开一个终端查看，可以看到端口
    logstash tmp]# netstat -antup | grep 8888
    tcp6       0      0 :::8888                 :::*                    LISTEN      22191/java          
    logstash tmp]# netstat -antup | grep 9999
    udp6       0      0 :::9999                 :::*                                22191/java
     
    在另一台主机上写一个脚本，发送数据，使启动的logstash可以接收到数据
    se5 ~]# vim tcp.sh
    function sendmsg(){
      if [[ "$1" == "tcp" ]];then
             exec 9<>/dev/tcp/192.168.1.67/8888
       else
             exec 9<>/dev/udp/192.168.1.67/9999
       fi
          echo "$2" >&9
          exec 9<&-
    }
    se5 ~]# . tcp.sh        //重新载入一下
    se5 ~]# sendmsg udp "is tcp test" 
    se5 ~]# sendmsg udp "is tcp ss"
     
    logstash主机查看结果
    Settings: Default pipeline workers: 2
    Pipeline main started
    {
           "message" => "is tcp test ",
          "@version" => "1",
        "@timestamp" => "2019-03-01T08:50:39.921Z",
              "type" => "udplog",
              "host" => "192.168.1.65"
    }
    {
           "message" => "is tcp ss ",
          "@version" => "1",
        "@timestamp" => "2019-03-01T08:50:55.608Z",
              "type" => "udplog",
              "host" => "192.168.1.65"
    }
     
    6）syslog插件练习
    logstash ~]# systemctl list-unit-files | grep syslog
    rsyslog.service                               enabled 
    syslog.socket                                 static  
     
    logstash ~]# vim /etc/logstash/logstash.conf
    input{
      file {
        path          => [ "/tmp/a.log", "/var/tmp/b.log" ]
      sincedb_path   => "/var/lib/logstash/sincedb"
      start_position => "beginning"
       type           => "testlog"
      }
      tcp {
         host => "0.0.0.0"
         port => "8888"
         type => "tcplog"
    }
       udp {
         host => "0.0.0.0"
         port => "9999"
         type => "udplog"
    }
      syslog {
         port => "514"
         type => "syslog"
      }
    }
    filter{
    }
    output{
        stdout{
        codec => "rubydebug"
    }
    }
    logstash]# /opt/logstash/bin/logstash -f /etc/logstash/logstash.conf
    开启检测
     
    另一个终端查看是否检测到514
    logstash ~]# netstat -antup | grep 514
    cp6    0  0 :::514    :::*   LISTEN  1278/java
    udp6   0  0 :::514    :::*           1278/java
     
    另一台主机上面操作,本地写的日志本地可以查看
    se5 ~]# vim /etc/rsyslog.conf
    local0.info   /var/log/mylog      //自己添加这一行
    se5 ~]# systemctl restart rsyslog //重启rsyslog
    se5 ~]# ll /var/log/mylog        //提示没有那个文件或目录
     
    se5 ~]# logger -p local0.info -t nsd "elk"   //写日志
    se5 ~]# ll /var/log/mylog        //再次查看，有文件
    -rw------- 1 root root 29 3月   1 17:05 /var/log/mylog
    se5 ~]# tail /var/log/mylog   //可以查看到写的日志
    Mar  1 17:05:02 se5 nsd: elk
     
    se5 ~]# tail /var/log/messages  
    //可以查看到写的日志，因为配置文件里有写以.info结尾的可以收到
    ...
    Mar  1 17:05:02 se5 nsd: elk
     
    把本地的日志发送给远程1.67
    se5 ~]# vim /etc/rsyslog.conf
    local0.info  @192.168.1.67:514 (添加这行）
    //写一个@或两个@@都可以，一个@代表udp，两个@@代表tcp
     
    se5 ~]# systemctl restart rsyslog
    se5 ~]# logger -p local0.info -t nds "001 elk"
     
    logstash检测到写的日志
    {
               "message" => "001 elk",
              "@version" => "1",
            "@timestamp" => "2019-03-01T09:11:12.000Z",
                  "type" => "syslog",
                  "host" => "192.168.1.65",
              "priority" => 134,
             "timestamp" => "Mar  1 17:11:12",
             "logsource" => "se5",
               "program" => "nds",
              "severity" => 6,
              "facility" => 16,
        "facility_label" => "local0",
        "severity_label" => "Informational"
    }
     
    rsyslog.conf配置向远程发送数据，远程登陆1.65的时侯，把登陆日志的信息（/var/log/secure）转发给logstash即1.67这台机器
    se5 ~]# vim /etc/rsyslog.conf
    57 authpriv.*    @@192.168.1.67:514
    //57行的/var/log/secure改为@@192.168.1.67:514 
    se5 ~]# systemctl restart rsyslog
     
    logstash ~]# /opt/logstash/bin/logstash -f /etc/logstash/logstash.conf  
    //找一台主机登录1.65，logstash主机会有数据
     
    Settings: Default pipeline workers: 2
    Pipeline main started
    {
               "message" => "Accepted publickey for root from 192.168.1.254 port 40954 ssh2: RSA SHA256:+bNE5YtlrU+NbKSDeszQXuk2C8Dc0si6BMbKVybe4Ig ",
              "@version" => "1",
            "@timestamp" => "2019-03-01T09:15:31.000Z",
                  "type" => "syslog",
                  "host" => "192.168.1.65",
              "priority" => 86,
             "timestamp" => "Mar  1 17:15:31",
             "logsource" => "se5",
               "program" => "sshd",
                   "pid" => "968",
              "severity" => 6,
              "facility" => 10,
        "facility_label" => "security/authorization",
        "severity_label" => "Informational"
    }
    {
               "message" => "pam_unix(sshd:session): session opened for user root by (uid=0) ",
              "@version" => "1",
            "@timestamp" => "2019-03-01T09:15:31.000Z",
                  "type" => "syslog",
                  "host" => "192.168.1.65",
              "priority" => 86,
             "timestamp" => "Mar  1 17:15:31",
             "logsource" => "se5",
               "program" => "sshd",
                   "pid" => "968",
              "severity" => 6,
              "facility" => 10,
        "facility_label" => "security/authorization",
        "severity_label" => "Informational"
    }
     
    7）filter grok插件
    grok插件：
    解析各种非结构化的日志数据插件
    grok使用正则表达式把非结构化的数据结构化
    在分组匹配，正则表达式需要根据具体数据结构编写
    虽然编写困难，但适用性极广
     
    logstash ~]# vim /etc/logstash/logstash.conf
    input{
            stdin{ codec => "json" }
      file {
        path          => [ "/tmp/a.log", "/var/tmp/b.log" ]
       sincedb_path   => "/var/lib/logstash/sincedb"
       start_position => "beginning"
       type           => "testlog"
      }
      tcp {
         host => "0.0.0.0"
         port => "8888"
         type => "tcplog"
    }
       udp {
         host => "0.0.0.0"
         port => "9999"
         type => "udplog"
    }
      syslog {
         port => "514"
         type => "syslog"
      }
    }
    filter{
       grok{
            match => ["message", "(?<key>reg)"]
      }
    }
    output{
        stdout{
        codec => "rubydebug"
    }
    }
     
    logstash ~]# /opt/logstash/bin/logstash -f /etc/logstash/logstash.conf
    开启检测
     
    se5 ~]# yum -y install httpd
    se5 ~]# systemctl restart httpd
    se5 ~]# vim /var/log/httpd/access_log
    192.168.1.254 - - [15/Sep/2018:18:25:46 +0800] "GET / HTTP/1.1" 403 4897 "-" "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:52.0) Gecko/20100101 Firefox/52.0"
     
    新开窗口，复制/var/log/httpd/access_log的日志到logstash下的/tmp/a.log
    logstash ~]# vim /tmp/a.log
    192.168.1.254 - - [15/Sep/2018:18:25:46 +0800] "GET / HTTP/1.1" 403 4897 "-" "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:52.0) Gecko/20100101 Firefox/52.0"
     
    //出现message的日志，但是没有解析是什么意思
    {
           "message" => ".168.1.254 - - [15/Sep/2018:18:25:46 +0800] "GET / HTTP/1.1" 403 4897 "-" "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:52.0) Gecko/20100101 Firefox/52.0"",
          "@version" => "1",
        "@timestamp" => "2019-03-01T09:38:24.428Z",
              "path" => "/tmp/a.log",
              "host" => "logstash",
              "type" => "testlog",
              "tags" => [
            [0] "_grokparsefailure"
        ]
    }
     
    若要解决没有解析的问题，同样的方法把日志复制到/tmp/a.log，logstash.conf配置文件里面修改grok
    查找正则宏路径
    logstash ~]# cd  /opt/logstash/vendor/bundle/
    bundle]# vim grok-patterns  //查找COMBINEDAPACHELOG
    COMBINEDAPACHELOG %{COMMONAPACHELOG} %{QS:referrer} %{QS:agent}
    logstash ~]# vim /etc/logstash/logstash.conf
    ...
    filter{
       grok{
            match => ["message", "%{COMBINEDAPACHELOG}"]
      }
    }
    ...
    解析出的结果
    logstash ~]# /opt/logstash/bin/logstash -f /etc/logstash/logstash.conf  
    Settings: Default pipeline workers: 2
    Pipeline main started
    {
            "message" => "192.168.1.254 - - [15/Sep/2018:18:25:46 +0800] "GET /noindex/css/open-sans.css HTTP/1.1" 200 5081 "http://192.168.1.65/" "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:52.0) Gecko/20100101 Firefox/52.0"",
           "@version" => "1",
         "@timestamp" => "2018-09-15T10:55:57.743Z",
               "path" => "/tmp/a.log",
               "host" => "logstash",
               "type" => "testlog",
           "clientip" => "192.168.1.254",
              "ident" => "-",
               "auth" => "-",
          "timestamp" => "15/Sep/2018:18:25:46 +0800",
               "verb" => "GET",
            "request" => "/noindex/css/open-sans.css",
        "httpversion" => "1.1",
           "response" => "200",
              "bytes" => "5081",
           "referrer" => ""http://192.168.1.65/"",
              "agent" => ""Mozilla/5.0 (Windows NT 6.1; WOW64; rv:52.0) Gecko/20100101 Firefox/52.0""
    }
     
    2.2 安装Apache服务，用filebeat收集Apache服务器的日志，存入elasticsearch
     
    1）在之前安装了Apache的主机上面安装filebeat
    se5 ~]# yum -y install filebeat
    se5 ~]# vim /etc/filebeat/filebeat.yml
    paths:
        - /var/log/httpd/access_log   //日志的路径，短横线加空格代表yml格式
    document_type: apachelog    //文档类型
    elasticsearch:              //加上注释
    hosts: ["localhost:9200"]   //加上注释
    logstash:                   //去掉注释
    hosts: ["192.168.1.67:5044"]     //去掉注释,logstash那台主机的ip
    se5 ~]# systemctl start filebeat
     
    logstash ~]# vim /etc/logstash/logstash.conf
    input{
            stdin{ codec => "json" }
            beats{
                port => 5044
    }
      file {
        path          => [ "/tmp/a.log", "/var/tmp/b.log" ]
       sincedb_path   => "/dev/null"
       start_position => "beginning"
       type           => "testlog"
      }
      tcp {
         host => "0.0.0.0"
         port => "8888"
         type => "tcplog"
    }
       udp {
         host => "0.0.0.0"
         port => "9999"
         type => "udplog"
    }
      syslog {
         port => "514"
         type => "syslog"
      }
    }
    filter{
    if [type] == "apachelog"{
       grok{
            match => ["message", "%{COMBINEDAPACHELOG}"]
      }}
    }
    output{
          stdout{ codec => "rubydebug" }
          if [type] == "filelog"{
          elasticsearch {
              hosts => ["192.168.1.61:9200", "192.168.1.62:9200"]
              index => "filelog"
              flush_size => 2000
              idle_flush_time => 10
          }}
    }
     
    logstash]# /opt/logstash/bin/logstash -f /etc/logstash/logstash.conf
    打开另一终端查看5044是否成功启动
    logstash ~]# netstat -antup | grep 5044
     
    se5 ~]# firefox 192.168.1.65  //ip为安装filebeat的那台机器
    回到原来的终端，有数据
     
    2）修改logstash.conf文件
    logstash]# vim /etc/logstash/logstash.conf
    ...
    output{
          stdout{ codec => "rubydebug" }
          if [type] == "apachelog"{
          elasticsearch {
              hosts => ["192.168.1.61:9200", "192.168.1.62:9200"]
              index => "apachelog"
              flush_size => 2000
              idle_flush_time => 10
          }}
    }
    浏览器访问Elasticsearch，有apachelog
    图片
    ##########################
    知识点整理：

    01:Kibana 数据展示
    02：Logstash 数据采集、加工处理即传输的工具
    所有类型的数据集中处理
    不同模式和模式数据的正常化
    自定义日志格式的迅速扩展
    为自定义数据源轻松添加插件
    input-->filter-->output

    03:Logstash配置扩展插件
    codec,input file,sincedb_path(日志读取起始位置）,syslog（获取系统日志），filter grok 解析各种非结构化的日志数据插件
    www.jinbuguo.com(帮助文档汉化）

    04：filebeat(c语言编写,安装在每台web服务器上）将web日志发送给logstash（的beats模块）
  • 相关阅读:
    变分法浅析
    再生核希尔伯特空间(RKHS)在监督学习(SVM)中的应用
    关于对偶最优化
    两个公式
    Linux系统——账号管理
    Linux系统——源码编译安装
    Linux系统——最小化安装
    Linux系统——本地yum仓库安装
    Linux系统——rpm命令
    Linux系统——vim编辑器
  • 原文地址：https://www.cnblogs.com/mkmkmk/p/12185990.html
Copyright © 2011-2022 走看看