elk收集log日志——logstash配置

繼續上面的步驟來，已經用filebeat將log日志寫入到了redis，現在需要從redis中取出log日志寫入logstash

logstash配置文件目錄

/etc/logstash/conf.d

寫一個名為nginx_access_log.conf配置文件，內容如下，redis的配置對應的是filebeat寫入redis的配置

input {
  redis {
    type => "nginx_access_log" 
    host => "10.x.53.x"//這里對應的是redis的地址，可以是互通的內網地址
    port => "8417"
    data_type => "list"
    key => "nginx_access_log" //這里要對應的redis的key
  }
}

filter {
     json {
       source => "message"
       remove_field => "message"
     }
}

output {
if [type] == "nginx_access_log"{
  elasticsearch {
    hosts => ["http://127.0.0.1:9200"]//這里是將日志輸入到es  
    index => "nginx_access_log" //對應的索引
    #index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    user => "elastic" //這里是對應的用戶名和密碼 沒有設置的話 這兩項可以先注釋  如何設置請查看后面文檔
    password => "elasticSh51699890"
  }
}
}

重啟logstash服務

這時候可以進入redis，可以看到對應的key正在被消費

kibana創建索引規則

進入到kibana中，創建對應的索引規則，並選擇時間區分

app/management/kibana/indexPatterns

創建成功之后，在創建的索引里可以看到對應的內容

多個配置文件的問題

/etc/logstash/conf.d，配置目錄里可以寫多個.conf結尾的配置文件

舉個例子，如果還有一個配置文件nginx_access_log2.conf

input {
  redis {
    type => "nginx_access_log2" 
    host => "10.x.53.x"//這里對應的是redis的地址，可以是互通的內網地址
    port => "8417"
    data_type => "list"
    key => "nginx_access_log2" //這里要對應的redis的key
  }
}

filter {
     json {
       source => "message"
       remove_field => "message"
     }
}

output {
if [type] == "nginx_access_log2"{
  elasticsearch {
    hosts => ["http://127.0.0.1:9200"]//這里是將日志輸入到es  
    index => "nginx_access_log2" //對應的索引
    #index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    user => "elastic" //這里是對應的用戶名和密碼 沒有設置的話 這兩項可以先注釋  如何設置請查看后面文檔
    password => "elasticSh51699890"
  }
}
}

如果沒有output的type類型判斷，那么在索引nginx_access_log2和nginx_access_log里面會出現兩個input的內容，索引這里要加個type來加以區分。

在別處看到的粘貼一下

logstash如何讀取多個配置文件

我們知道在啟動logstash的時候，只要加上-f /you_path_to_config_file就可以加載配置文件了，如果我們需要加載多個配置文件，只需要-f /you_path_to_config_directory就可以了。簡單說，就是在-f后面加上目錄就可以。 注意：目錄后面不能加 * 號，否則只會讀取一個文件，但是在讀取日志文件時，可以匹配所有，比如sys.log可以匹配所有以sys.log開頭的日志文件，如sys.log1,sys.log2等。

示例如下：

//比如 /home/husen/config/目錄下有
//in1.conf、in2.conf、filter1.conf、filter2.conf、out.conf這5個文件

//我們使用 /logstash-5.5.1/bin/logstash -f /home/husen/config啟動logtstash
//logstash會自動加載這個5個配置文件，並合並成1個整體的配置文件

logstash多個配置文件里的input、filter、output是否相互獨立

比如：

## in1.conf內容如下：
input{
    file{
        path=>[
            "/home/husen/log/sys.log"
        ]
    }
}

## in2.conf內容如下：
input{
    file{
        path=>[
            "/home/husen/log/error.log"
        ]
    }
}

## out1.conf如下
elasticsearch {
        action => "index"          
        hosts  => "localhost:9200" 
        index  => "from_sys_log"
        codec => "json"
    }

## out2.conf如下
elasticsearch {
        action => "index"          
        hosts  => "localhost:9200" 
        index  => "from_error_log"
        codec => "json"
    }
//這幾個配置文件的目的是：
//想把in1.conf讀進來的sys.log的索引建立為from_sys_log
//把in.conf讀進來的error.log的索引建立為femo_error_log

//logstash-5.5.1/bin/logstash -f /home/husen/config

//啟動之后，會發現in1.conf的日志被輸出了兩次，in2.conf讀進來的日志也被輸出了兩次

//結論：logstash讀取多個配置文件只是簡單的將所有配置文件整合到了一起！
//如果要彼此獨立，需要自己加字段，然后判斷一下
//比如讀取來不同不同服務器的同樣格式的日志，那么filter是可以共用的
//但是輸出的索引需要分別建立，以提高辨識度

logstash讀取多個配置文件建議的配置方法

如果要在配置文件中，獨立一些部分，又要共用一些部分，比如我上門提高同樣的日志來自不同的服務器，需要用同樣的filter，但是建立不同的索引的問題，該怎么辦？ 建議使用tags或者type這兩個特殊字段，即在讀取文件的時候，添加標識符在tags中或者定義type變量。

示例如下：

## in1.conf內容如下：
input{
    file{
        path=>[
            "/home/husen/log/sys.log"
        ]
        type => "from_sys"
        #tags => ["from_sys"]
    }
}

## in2.conf內容如下：
input{
    file{
        path=>[
            "/home/husen/log/error.log"
        ]
        type => "from_error"
        #tags => ["from_sys"]
    }
}

## out1.conf如下
if [type] == "from_sys"{
#if "from_sys" in [tags]
    elasticsearch {
        action => "index"          
        hosts  => "localhost:9200" 
        index  => "from_sys_log"
        codec => "json"
    }
}

## out2.conf如下
if [type] == "from_error"{
#if "from_error" in [tags]
    elasticsearch {
        action => "index"          
        hosts  => "localhost:9200" 
        index  => "from_error_log"
        codec => "json"
    }
}

#特別地，如果要針對不同的類型日志用不同filter來grok解析，
#也可以通過類似的方法判斷