ELK(12)：ELK+kafka(日志量中等)

超、凢脫俗 2021-10-29 15:12 327阅读 0赞

# ELK(12)：ELK+kafka(日志不太多) #

我们先用logstash读取Nginx日志和系统日志写入kafka，再用logstash读取出来写入elasticsearch，适合日志量不是太多的架构。

海量日志建议采用filebeat。

其实用redis也可以，redis没必要开快照和持久化，数据写入es后redis的作用就完成了。当然很耗redis内存，一般8-16G。

后端可能几十台logstash往kafka写入，如果kafka内存居高不下，也就是前端的logstash读的太慢，要加logstash。直到平衡。

# Nginx日志 #

## 1. Logstash把nginx日志写入kafka ##

### 1.1 配置logstash ###

#[admin@ris-1 conf.d]$ pwd
    #/etc/logstash/conf.d
    #[admin@ris-1 conf.d]$ cat nginx_kafka.conf
    input {
        file {
            type => "ris-api-nginx-1"
            path => "/home/admin/webserver/logs/api/api.log"
            start_position => "beginning"
            stat_interval => "2"
            codec => "json"
        }
    }
    
    output {
        if  [type] == "ris-api-nginx-1" {
        kafka {
          bootstrap_servers => "10.6.76.27:9092"
          topic_id => "ris-api-nginx-1"
          batch_size => "5"
          codec => "json"
          }
         stdout {
             codec => "rubydebug"
            }
      }
    }

### 1.2 测试配置文件 ###

sudo /usr/share/logstash/bin/logstash -f nginx_kafka.conf –t

### 1.3 前台启动logstash ###

sudo /usr/share/logstash/bin/logstash -f nginx_kafka.conf

### 1.4 查看kafka的topic ###

[admin@pe-jira ~]$ /home/admin/elk/kafka/bin/kafka-topics.sh --list --zookeeper kafka1:2181, kafka2:2181, kafka3:2181
    __consumer_offsets
    messagetest
    ris-api-nginx-1
    [admin@pe-jira ~]$

### 1.5 前台logstash日志 ###

{
               "@version" => "1",
             "@timestamp" => 2019-07-18T09:16:29.000Z,
               "clientip" => "10.6.0.11",
                    "xff" => "-",
           "responsetime" => 0.001,
           "upstreamhost" => "10.6.75.172:8080",
         "request_method" => "GET",
                 "domain" => "api.erp.zhaonongzi.com",
                 "status" => "200",
                   "host" => "10.6.75.171",
        "http_user_agent" => "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.11; rv:67.0) Gecko/20100101 Firefox/67.0",
           "upstreamtime" => "0.001",
                    "url" => "/APICenter/purchase_snapshot_single.wn",
                "referer" => "-",
                   "type" => "ris-api-nginx-1",
                   "size" => 66,
            "remote_user" => "-",
                   "path" => "/home/admin/webserver/logs/api/api.log"
    
    }

### 1.6 再取消前台输出 ###

注释
         stdout {
             codec => "rubydebug"
            }

### 1.7 通过服务方式启动 ###

sudo systemctl start logstash

## 2. 配置logstash从kafka读取nginx日志写入elasticsearch ##

### 2.1 配置logstash ###

#[admin@pe-jira conf.d]$ pwd
    #/etc/logstash/conf.d
    #[admin@pe-jira conf.d]$ cat kafka-es.conf
    input{
        kafka {
            bootstrap_servers => "10.6.76.27:9092" #kafka服务器地址
            topics => "ris-api-nginx-1"
            group_id => "ris-api-nginx-logs"
            decorate_events => true #kafka标记
            consumer_threads => 1
            codec => "json" #写入的时候使用json编码，因为logstash收集后会转换成json格式
        }
    }
    output{
        stdout {
            codec => "rubydebug"
      }
    #    if [type] == "ris-api-nginx-1"{
    #        elasticsearch {
    #           hosts => ["10.6.76.27:9200"]
    #           index => " logstash-ris-api-nginx-1-%{+YYYY.MM.dd}"
    #        }
    #    }
    }

### 2.2 测试配置文件 ###

sudo /usr/share/logstash/bin/logstash -f nginx_kafka.conf –t

### 2.3 前台启动logstash ###

sudo /usr/share/logstash/bin/logstash -f kafka-es.conf

### 2.4 前台logstash日志 ###

{
                 "domain" => "XXXX.com",
           "upstreamtime" => "0.001",
                   "host" => "10.6.75.171",
            "remote_user" => "-",
                 "status" => "200",
                    "xff" => "-",
         "request_method" => "GET",
             "@timestamp" => 2019-07-19T02:10:46.000Z,
           "upstreamhost" => "10.6.75.172:8080",
                   "path" => "/home/admin/webserver/logs/api/api.log",
        "http_user_agent" => "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.11; rv:67.0) Gecko/20100101 Firefox/67.0",
           "responsetime" => 0.001,
                   "size" => 66,
               "@version" => "1",
                   "type" => "ris-api-nginx-1",
                "referer" => "-",
               "clientip" => "10.6.0.11",
                    "url" => "/APICenter/purchase_snapshot_single.wn"
    }

### 2.5 再取消前台输出，直接写入elasticsearch ###

input{
          kafka {
            bootstrap_servers => "10.6.76.27:9092" #kafka服务器地址
            topics => "ris-api-nginx-1"
            group_id => "ris-api-nginx-logs"
            decorate_events => true #kafka标记
            consumer_threads => 1
            codec => "json" #写入的时候使用json编码，因为logstash收集后会转换成json格式
        }
    }
    output{
    #    stdout {
    #       codec => "rubydebug"
    #  }
        if [type] == "ris-api-nginx-1"{
            elasticsearch {
               hosts => ["10.6.76.27:9200"]
               index => " logstash-ris-api-nginx-1-%{+YYYY.MM.dd}"  #必须logstash开头，地图展示需要
            }
        }
    }

###  ###

### 2.6 通过服务方式启动 ###

sudo systemctl start logstash

### 2.7 添加kibana ###

![1131047-20190724123748532-617886608.png][]

![1131047-20190724123758050-1567932954.png][]

# 系统日志 #

## 3. Logstash把系统日志写入kafka ##

### 3.1 修改日志权限 ###

sudo chmod 644 /var/log/messages

### 3.2 agent-logstash配置 ###

#[admin@ris-1 conf.d]$ pwd
    #/etc/logstash/conf.d
    #[admin@ris-1 conf.d]$ cat syslog-kafka.conf
    input {
       file {
        type => "ris-1-systemlog"
        path => "/var/log/messages"
        start_position => "beginning"
        stat_interval => "2"
      }
    }
    output {
        if [type] == "ris-1-systemlog" {
        # stdout {
        #     codec => "rubydebug"
        #    }
        kafka {
         bootstrap_servers => "10.6.76.27:9092"
         topic_id => "ris-1-systemlog"
         batch_size => "5"
         codec => "json"
         }
      }
    }

### 3.3 服务启动 ###

sudo systemctl restart logstash

## 4. 配置logstash从kafka读取系统日志写入elasticsearch ##

### 4.1 lostash配置 ###

[admin@pe-jira conf.d]$ cat sys-kafka-es.conf
    input{
          kafka {
            bootstrap_servers => "10.6.76.27:9092" #kafka服务器地址
            topics => "ris-1-systemlog"
            group_id => "ris-1-systemlog"
            decorate_events => true #kafka标记
            consumer_threads => 1
            codec => "json" #写入的时候使用json编码，因为logstash收集后会转换成json格式
        }
    }
    output{
        stdout {
            codec => "rubydebug"
      }
        if [type] == "ris-1-systemlog"{
            elasticsearch {
               hosts => ["10.6.76.27:9200"]
               index => "logstash-ris-1-systemlog-%{+YYYY.MM.dd}"
            }
        }
    }

### 4.2 服务启动 ###

sudo systemctl restart logstash

### 4.3 添加kibana ###

![1131047-20190724123938556-1530028076.png][]

![1131047-20190724123946778-154626006.png][]

![1131047-20190724123951515-1479918881.png][]

转载于:https://www.cnblogs.com/wangxu01/articles/11237277.html

[1131047-20190724123748532-617886608.png]: /images/20211029/c4a061770f95446db581b0bbc241077d.png
[1131047-20190724123758050-1567932954.png]: /images/20211029/9f71410a4de2415f97cd06ad26befa7a.png
[1131047-20190724123938556-1530028076.png]: /images/20211029/55626ce26ee64a5ca17664d9f4da63be.png
[1131047-20190724123946778-154626006.png]: /images/20211029/bbff0f9827a6420ba2e5e19b5d7cfe82.png
[1131047-20190724123951515-1479918881.png]: /images/20211029/f5493b159bee43a89179060739180bc6.png