ELK生态：Logstash增量读取log文件数据，导入到Elasticsearch

╰半橙微兮° 2022-10-02 04:57 176阅读 0赞

### 简介 ###

1.  ELK生态之Logstash导入数据到Elasticsearch；
2.  数据源：log格式文件，内容为log日志；
3.  Elasticsearch和Logstash版本：5.6.1；
4.  前提环境：Elasticsearch单机或集群；Logstash客户端；

### 实践 ###

*  log文件内容：

01-Apr-2019 08:17:04.787 SEVERE [ajp-nio-8009-exec-7] org.apache.coyote.ajp.AjpMessage.processHeader Invalid message received with signature [5635]
    01-Apr-2019 08:17:05.704 SEVERE [ajp-nio-8009-exec-9] org.apache.coyote.ajp.AjpMessage.processHeader Invalid message received with signature [18245]
    01-Apr-2019 11:20:36.635 INFO [http-nio-8080-exec-6] org.apache.coyote.http11.Http11Processor.service Error parsing HTTP request header
     Note: further occurrences of HTTP request parsing errors will be logged at DEBUG level.
     java.lang.IllegalArgumentException: Invalid character found in method name. HTTP method names must be tokens
    	at org.apache.coyote.http11.Http11InputBuffer.parseRequestLine(Http11InputBuffer.java:410)
    	at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:291)
    	at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:66)
    	at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:754)
    	at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1376)
    	at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49)
    	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    	at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
    	at java.lang.Thread.run(Thread.java:748)

*  log文件内容：

#读取log文件
    input{
      file{
            #设置log文件路径，多个文件路径可设置成数组[]，模糊匹配用*
            #指定单一文件
            path => "/data/es/logstash-5.6.1/files/test.log"
            #指定数组文件
            #path => ["/data/es/logstash-5.6.1/files/test-1.log","/data/es/logstash-5.6.1/files/test-2.log"]
            #指定同级目录模糊匹配
            #path => "/data/es/logstash-5.6.1/files/test*.log"
            #指定多级目录模糊匹配
            #path => "/data/es/logstash-5.6.1/files/**/test*.log"
          
            #可设置成begining或end，begining表示从头开始读取文件，end表示读取最新数据，可和ignore_older一起使用
            #begining只针对首次启动是否需要读取所有的历史数据，而当文件修改了之后，同样会自动增量更新新数据
            start_position =>"beginning"
          
            #设置输入规则
            codec => multiline {
                #利用正则匹配规则，匹配每一行开始的位置，这里匹配每一行开始的位置为数字
                pattern => "^[0-9]"
         
                #true表示不匹配正则表达式，false为匹配正则表达式，默认false
                #如果不匹配，则会结合what参数，进行合并操作
                negate => true
                
                #what可设置previous和next，previous则表示将所有不匹配的数据都合并到上一个正则事件
                #而next则相反，将所有的不匹配的数据都合并到下一个正则事件
                what => "previous"
    
                #表示当多长时间没有新的数据，最后一个正则匹配积累的多行数据都归属为最后一个事件，这里的10表示10秒
                #auto_flush_interval => 10
           }
           
           #当存在多个文件的时候可使用type指定输入输出路径
           type=>"log_index"
           
      }
    }
        
    #输出log文件
    output{
        #输出控制台
        stdout {
            #codec => json
        }
        
        #输出到es
        if[type] == "log_index"{
            elasticsearch{
                hosts=>["127.0.0.1:9200"]
                
                #以当前的日期作为index和type
                index=>"log-%{+YYYY.MM.dd}"
                document_type=>"log-%{+YYYY.MM.dd}"
                
                #覆盖模板，不需要可注释掉，通用模板下载：https://download.csdn.net/download/alan_liuyue/11241484
                #template_overwrite=>true
                #template=>"/data/es/logstash-5.6.1/template/logstash.json"
            }
        }
    }

*  补充file-input字段说明：

codec => #可选项，默认是plain，可设置其他编码方式；
    
    discover_interval => #可选项，logstash多久检查一下path下有新文件，默认15s；
    
    exclude => #可选项，排除path下不想监听的文件；
    
    sincedb_path => #可选项，记录文件以及文件读取信息位置的数据文件；
    
    sincedb_write_interval => #可选项，logstash多久写一次sincedb文件，默认15s；
    
    stat_interval => #可选项，logstash多久检查一次被监听文件的变化，默认1s；
    
    start_position => #可选项，表示从哪个位置读取文件数据，初次导入为：beginning，最新数据为：end
    
    path => #必选项，配置文件路径，可定义多个，也可模糊匹配；
    
    tags => #可选项，在数据处理过程中，由具体的插件来添加或者删除的标记；
    
    type => #可选项，当有多个file的时候，可用于一对一匹配输入或者输出；

### 注意事项 ###

1.  logstash启动之后，进程会一直处于运行状态，若log文件被修改，程序会自动监听，导入新数据；
2.  若需要重新导入数据，则需要删除logstash安装目录下 \\plugins\\inputs\\file 下的文件(该文件属于隐藏文件)，所以直接删除该目录即可，之后目录会重新生成；
3.  如果使用了模板覆盖，需要将模板的message字段设置成分词，否则无法有效进行分词全局搜索；

### 总结 ###

1.  Logstash读取log文件内容导入Elasticsearch，可以实时将日志保存以及展示，达到快速搜索日志的效果；
2.  实践是检验认识真理性的唯一标准，自己动手丰衣足食~