ELK-LogStash6.5.4

Dear 丶 2022-03-14 10:50 214阅读 0赞

## ELK-LogStash6.5.4 ##

# 前言 #

与Elasticsearch、Kibana不同，Logstash默认并不作为系统服务安装，我们也不建议作为服务启动。主要原因为：

大多数情况下，Elasticsearch、Kibana在每台服务器上只有一个实例，可以作为服务启动；而一个Logstash实例一般只服务于一个数据流，因此一台服务器上经常运行多个实例； Elasticsearch、Kibana是服务器端，而Logstash是客户端。

### 学习资料 ###

[Logstash 最佳实践][Logstash]  
[elk学习资料][elk]，提取码c0l1  
从学习资料里学习基本语法。

插件：  
[插件安装][Link 1]  
[插件api][api]

## logstash一些文件位置 ##

1.  sincedb数据库文件位置：./data/plugins/inputs/file/.sincedb\_452905a167cf4509fd08acb964fdb20c
2.  Grok默认提供的正则表达式位置:/usr/local/elk/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-patterns-core-4.1.2/patterns  
    这些默认的[正则表达式也可从github查看][github]

## logstash命令一些可选参数 ##

### 加载指定配置文件 ###

使用–config 或 -f，例子

logstash –f apache.config

### 配置自动加载 ###

使用–config.reload.automatic。例子

logstash –f apache.config --config.reload.automatic

### 测试配置文件正确性 ###

使用–config.test\_and\_exit或\-t。例子

bin/logstash –f apache.config --config.test_and_exit

或

logstash –f apache.config --t

### log日志路径 ###

使用–log 或 -l，例子

logstash –f apache.config -l /usr/local/elk/logstash/logs

## 插件安装 ##

[logstash支持的插件][Link 1],可从路径里搜索支持的插件。  
比如安装 合并多行数据的codec/multiline 插件

logstash-plugin install codec/multiline

其它可选命令

1.  install:安装
2.  uninstall:卸载
3.  update:更新
4.  list：列出所有已经安装插件

## 输入插件 ##

[输入插件所有类型][Link 2]

### 原理 ###

Logstash 使用一个名叫 FileWatch 的 Ruby Gem 库来监听文件变化。这个库支持  
glob 展开文件路径，而且会记录一个叫 .sincedb 的数据库文件来跟踪被监听的日  
志文件的当前读取位置。FileWatch 只支持文件的绝对路径，而且会不自动递归目录  
windows 平台上，推荐考虑使用 nxlog 作为收集端

可参考[Logstash简单介绍][Logstash 1]，里面有部分内容介绍不同插件可配置的字段。

### 标准输入(Stdin) ###

忽略，只有学习阶段使用

### 文件输入 ###

下面配置从头开始读文件，每15秒检查目录下是否有新文件，每1秒检查一次被监听文件状态（是否有更新）

input {
    	file {
    	path => ["/var/log/*.log", "/var/log/messages"]
    	type => "system"
    	start_position => "beginning"
    	}
    }

### TCP输入(网络数据) ###

打开514端口接收消息，  
multiline是输入插件，用于合并多行数据，遇到开头是日期的才视为一行。  
type随便写。

tcp {
    		port => 514
    		codec => multiline{
    			pattern => "^\d{4}(\-|\/|.)\d{1,2}\1\d{1,2}"
    			negate => true
    			what => "previous"
    		}
    		type=>"test243"
    		ssl_enable => false
       	}

### UPD输入 ###

不使用它，建议使用TCP协议。  
UDP 监听器只用了一个线程，而 TCP 监听器会在接收每个连接  
的时候都启动新的线程来处理后续步骤

### codec配置 ###

input配置的编码插件，用于编码接收或采集的数据。  
常用的插件

1.  接收json格式数据的json插件

input {
    	file {
    		path => "/var/log/nginx/access.log_json"
    		codec => "json"
    	}
    }

配置nginx输出json格式日志。在nxinx.conf配置文件的http块里添加

log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                          '$status $body_bytes_sent "$http_referer" '
                          '"$http_user_agent" "$http_x_forwarded_for"';
    					  
    	log_format json '{"@timestamp":"$time_iso8601",'
                               '"@version":"1",'
                               '"client":"$remote_addr",'
                               '"url":"$uri",'
                               '"status":"$status",'
                               '"domain":"$host",'
                               '"host":"$server_addr",'
                               '"size":$body_bytes_sent,'
                               '"responsetime":$request_time,'
                               '"referer": "$http_referer",'
                               '"ua": "$http_user_agent"'
                   '}';

在server模块里添加

access_log  logs/access_json.log  json;

1.  合并多行数据插件multiline  
    这个插件需要安装  
    执行命令：logstash-plugin install codec/multiline  
    使用例子

## 过滤器 ##

[过滤器插件所有类型][api]。

我们可以在/usr/local/elk/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-patterns-core-4.1.2/patterns事先定义好一些默认的正则表达式(当然logstask事件已经默认定义了一些)，然后再logstash配置文件里使用使用，比如%\{IP:ip\},大写的IP其实真正的正则表达式是(?:%\{IPV6\}|%\{IPV4\}),它的定义在[https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns][https_github.com_logstash-plugins_logstash-patterns-core_blob_master_patterns_grok-patterns]

### 所有插件都支持的api ###

1.  add\_field：动态增加字段  
    默认值是\{\}  
    可以使用字段引用,比如下面的%\{somefield\}

filter {
      grok {
        add_field => { "foo_%{somefield}" => "Hello world, from %{host}" }
      }
    }

1.  add\_tag：增加tag  
    默认值是\[\]

filter {
      grok {
        add_tag => [ "foo_%{somefield}", "taggedy_tag"]
      }
    }

1.  remove\_field：删除属性

filter {
      grok {
        remove_field => [ "foo_%{somefield}" ]
      }
    }

1.  remove\_tag:删除tag

filter {
      grok {
        remove_tag => [ "foo_%{somefield}", "sad_unwanted_tag"]
      }
    }

1.  id：给插件指定唯一的id，用于区分插件，如果不指定，logstash默认生成一个  
    在有多个插件时，用于区别插件类型

filter {
      grok {
        id => "ABC"
      }
    }

1.  enable\_metric：默认true,作用不知

### Grok插件 ###

[logstash grok插件语法介绍][logstash grok]  
[grok调试工具][grok]

### mutate插件 ###

更多可访问[Mutate filter plugin][]  
数据修改插件,提供一些函数帮助处理数据

1.  数据类型转换：covert  
    可以设置的转换类型包括：“integer”，“float” 和 “string”。示例可以设置的转换类型包括：“integer”，“float” 和 “string”。示例

filter {
    	mutate {
    		convert => ["request_time", "float"]
    	}
    }

1.  字符串分割成数组：split  
    示例，遇到|就分割字符串

filter {
    	mutate {
    		split => ["message", "|"]
    	}
    }

1.  连接字符串：join,仅对数组类型字段有效  
    示例

filter {
    	mutate {
    		split => ["message", "|"]
    	}
    	mutate {
    		join => ["message", ","]
    	}
    }

1.  合并2个数组成一个数组：merge  
    如果 src 字段是字符串，会自动先转换成一个单元素的数组再合并

filter {
    	mutate {
    		split => ["message", "|"]
    	}
    	mutate {
    		merge => ["message", "message"]
    	}
    }

1.  重命名字段：rename  
    重命名某个字段，如果目的字段已经存在，会被覆盖掉

filter {
    	mutate {
    		rename => ["syslog_host", "host"]
    	}
    }

1.  增加字段：add\_field

filter {
    	mutate {
    		add_field =>["name","cy"]
    		split => ["hostname", "."]
            add_field => { "shortHostname" => "%{hostname[0]}" }
    	}
    }

1.  部分替换:gsub  
    公支持数组和字符串，其它类型没有动作

filter {
      mutate {
        gsub => [
          # 替换/为——
          "fieldname", "/", "_",
          # 支持正则表达式
          "fieldname2", "[\\?#-]", "."
        ]
      }
    }

1.  整体替换：replace

filter {
      mutate {
        replace => { "message" => "%{source_host}: My new message" }
      }
    }

1.  大小写转换，lowercase,uppercase  
    公支持字符串

filter {
      mutate {
         lowercase => [ "fieldname" ]
         uppercase => [ "fieldname" ]
      }
    }

1.  去掉首尾空白字符：strip  
    注意，只过滤首尾空白

filter {
      mutate {
         strip => ["field1", "field2"]
      }
    }

1.  复制一成存在的字段值给另一个存在的字段

filter {
      mutate {
         copy => { "source_field" => "dest_field" }
      }
    }

1.  更新字段值：update

filter {
      mutate {
        update => { "sample" => "My new message" }
      }
    }

## 输出插件output ##

[输出插件所有类型][Link 3]

### 标准输出 ###

output {
        stdout {
            codec => rubydebug
            workers => 2
        }
    }

### file输出 ###

在这里插入代码片

### elasticsearch输出 ###

output{
    	elasticsearch {
    			hosts => ["127.0.0.1:9200"]
    			index => "securitylog-%{+YYYY.MM.dd}"
    		}
    }

[Logstash]: https://doc.yonyoucloud.com/doc/logstash-best-practice-cn/index.html
[elk]: https://pan.baidu.com/s/1CzMx46cRGsRZ-unDd0nc5Q
[Link 1]: https://github.com/logstash-plugins?utf8=%E2%9C%93&q=&type=&language=
[api]: https://www.elastic.co/guide/en/logstash/current/filter-plugins.html
[github]: https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns
[Link 2]: https://www.elastic.co/guide/en/logstash/current/input-plugins.html
[Logstash 1]: https://blog.csdn.net/chenleiking/article/details/73563930
[https_github.com_logstash-plugins_logstash-patterns-core_blob_master_patterns_grok-patterns]: https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns
[logstash grok]: https://blog.csdn.net/qq_34021712/article/details/79746413
[grok]: https://grokdebug.herokuapp.com/
[Mutate filter plugin]: https://www.elastic.co/guide/en/logstash/current/plugins-filters-mutate.html
[Link 3]: https://www.elastic.co/guide/en/logstash/current/output-plugins.html