业务目的是能够分析nginx和apache每天产生的日志，对url、ip、rest接口等信息进行监控，并将数据发送到elasticsearch服务。
从日志文件中获取信息
file { path => "/home/keepgostudio/demo/logs/test.log" start_position => "beginning" }目前是logstash中把非标准化的日志数据转换成标准化并且可搜索数据最好的方式。Logstash默认提供了能分析包括java堆栈日志、apache日志在内的120种形式。点击查看
如没有特别的需求的话，使用默认的apache日志形式就能达到想要的效果，如下。
grok{ match => {"message" => ["%{COMBINEDAPACHELOG}"]} }但如果想要监控更多的信息，比如url上的参数，那么默认的表达式将没办法满足我们的需求，这时我们就需要自己动手去编写一些符合我们业务需要的表达式，并告诉logstash以某种期望的方式进行数据转换。
首先，在logstash的根目录下创建一个patterns文件夹，这个文件夹默认是没有的。
其次，在patterns文件夹中创建文件test_pattern（这里为了方便所以没有按照pattern的功能对文件进行命名，在实际应用中最好按照功能来对文件命名，至于原因你懂的）。在test_pattern文件中可以按照“名称 正则表达式”这样的格式自定义一些正则表达式，以便在grok中进行使用。
最后，在使用的时候一定要把pattern_dir这个参数带上，否则logstash无法识别你自定义的这些正则表达式。
grok { patterns_dir => ["/home/keepgostudio/download/logstash-5.2.0/patterns"] match => { "message" => ["%{PARAMS_APACHELOG}", "%{NO_PARAMS_APACHELOG}"] } remove_field => ["host", "timestamp", "httpversion", "@version"] }将数据源转换成键值对，并创建相对的field。比如传入“a=111&b=2222&c=3333”，输出的时候，a，b，c会被创建成三个field，这样做的好处是，当需要查询某一个参数的时候可直接查询，而不是把一个字符串搜索出来再做解析。
kv { source => "field_name" field_split => "&?" }这个从字面上就能看出他的功能，根据ip查出相应的地理信息，比如城市，省份，国家，经纬度等。这个ip信息是在logstash中的一个数据源中进行搜索查找，而不是进行网络搜索。
geoip { source => "field_name" fields => ["country_name", "region_name", "city_name", "latitude", "longitude"] target => "location" }drop可以跳过某些不想统计的日志信息，当某条日志信息符合if规则时，该条信息则不会在out中出现，logstash将直接进行下一条日志的解析。
if [field_name] == "value" { drop {} }logstash输出到elasticsearch时，会随机生成一个id给每一条记录。建议在刚开始接触的时候，可以采用手动输入的方式。这样结合elasticsearch的服务，更容易理解整个实现的流程。
elasticsearch { hosts => ["192.168.1.44:9200"] index => "logstash-test-%{type}-%{host}" }