ELK搭建环境之配置logstash监听文件并利用grok|ruby插件进行切割利于统计

作者: admin 分类: ELK 发布时间: 2019-01-26 23:49  阅读: 1,005 views

一、配置ELK环境在logstash和elasticsearch之间建立通信,并能查看对应的索引、日志信息

1. 本地下载logstash6.4.2运行包,不会装请移步《(ElasticSearch+LogStash+Kibana)ELK搭建在Mac/Linux系统上6.4.2版本

2. 解压并进行基础配置,配置如下:

默认配置文件在安装目录 logstash6.4.2/config/logstash-sample.conf

# 监听本地目录下的test.log日志文件
input {
  file {
      path => ["/Users/chenhailong/daily-test/test.log"]
  }
}
# 配置日志发送的目标elasticsearch服务
output {
  elasticsearch {
    hosts => ["http://xxxx.xx.xxx.xx:9200"]
    index => "grok-test"
    user => "username"
    password => "password"
  }
}

3.确保elasticsearch服务是可以访问的,(开启代理、翻墙、防火墙、IP不通可能造成无法通信)

在浏览器键入对应地址:http://xxxx.xx.xxx.xx:9200

显示如下:

{
  "name" : "wqbVDEq",
  "cluster_name" : "elasticsearch",
  "cluster_uuid" : "lq8YkTH4Q7eslqvd0pn9Kg",
  "version" : {
    "number" : "6.4.2",
    "build_flavor" : "default",
    "build_type" : "tar",
    "build_hash" : "04711c2",
    "build_date" : "2018-09-26T13:34:09.098244Z",
    "build_snapshot" : false,
    "lucene_version" : "7.4.0",
    "minimum_wire_compatibility_version" : "5.6.0",
    "minimum_index_compatibility_version" : "5.0.0"
  },
  "tagline" : "You Know, for Search"
}

4.在客户端保证有效的日志文件,日志记录有序的生成。

客户端的

日志文件为:/Users/chenhailong/daily-test/test.log

日志格式为:2018-12-31,16:12:1546245470 INFO sdklog &event_id=10003&uuid=90b7786d-a905-48aa-93e9-bc1cf64ffce2&event_type=2&device_id=37037&session_id=1545899195566_37037&uid=2D9CBA4701227157E050840AF2422B52&date=2018-12-27&time=16:26:36&ts=1545899196409&

在需要测试时,执行脚本追加日志记录务必保证日志文件中有新的追加记录,确保监听到了新的数据。否则,没有新数据可能看不到效果

 

5. 启动logstash

#指定配置文件启动

./bin/logstash -f ../conf/logstash-sample.conf

6. 如果一切顺利的话,可以在kibana控制台看到相关信息

a. kibana 》系统管理 》 Index Mangement 》 可以搜索查看到对应的 “grok-test” 索引名称
b. kibana 》系统管理 》 Index Patterns 》可以创建索引前缀
c. kibana 》发现 》 可以搜索到 “grok-test”索引文件下的所有日志记录,如下

logstash索引

 

二:对日志进行切割

1. 前期了解:

https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns

2. 在kibana后台 》开发工具 》 Grok Debugger进行测试,如下图

不会用,请移步《使用kibana的开发工具console做一些elasticsearch的基本查询和设置

ELK的kibana工具台

3. 日志切割请参考,注意格式

%{TIMESTAMP_ISO8601:date} %{LOGLEVEL:level} %{WORD:type} &event_id=%{NUMBER:event_id}&uuid=%{USERNAME:uuid}&event_type=%{INT:event_type}&device_id=%{INT:device_id}&session_id=%{USERNAME:session_id}&uid=%{WORD:uid}&date=%{USERNAME:ymd}&time=%{TIME:time}&ts=%{NUMBER:ts}&url=%{URI:url}

4.修改配置文件,增加grok插件

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  file {
      path => ["/Users/chenhailong/daily-test/test.log"]
  }
}

filter {
  #定义数据的格式
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:date} %{LOGLEVEL:level} %{WORD:type} &event_id=%{NUMBER:event_id}&uuid=%{USERNAME:uuid}&event_type=%{INT:event_type}&device_id=%{INT:device_id}&session_id=%{USERNAME:session_id}&uid=%{WORD:uid}&date=%{USERNAME:ymd}&time=%{TIME:time}&ts=%{NUMBER:ts}&url=%{URI:url}"}
  }
}

output {
  elasticsearch {
    hosts => ["http://xxx.xx.xxx.xx:9200"]
    index => "grok-test"
    user => "username"
    password => "password"
  }
}

5. 打开kibana 》 发现 》 搜索grok-test索引

会发现日志已经按照设定的格式进行切切割。这样就可以进行数据的报表分析,更加直观化。更有效的利用日志数据。

不局限于日常日志,也可以针对性的记录相关业务数据。

利用grok插件解析日志

6. 配置文件中还可以增加多种插件配合处理

具体可以移步查看官方文件,更加详细,更加专业

https://www.elastic.co/guide/en/logstash/current/filter-plugins.html

至此:ELK的学习使用就告一段落,有需要再去细致研究。

ELK环境搭建的相关学习内容

 

补充一:

可以使用正则表达式,自己解析日志

日志格式:asdfasdfadfasdfadsf xxxxxxx

正则写法(?<basss>\b\w+\b) (?<bbb>\b\w+\b)

{
  "bbb": "xxxxxxx",
  "basss": "asdfasdfadfasdfadsf"
}

补充二:

针对于日志,可能有时多个属性,有时少个属性。这种情况下grok可以匹配多个规则,如下

//格式一
filter {
  grok {
    match => {
      "message" , "%{TIMESTAMP_ISO8601:date} %{LOGLEVEL:level}  %{WORD:type} &event_id=%{NUMBER:event_id}&uuid=%{USERNAME:uuid}&event_type=%{INT:event_type}&device_id=%{INT:devi
ce_id}&session_id=%{USERNAME:session_id}&uid=%{WORD:uid}",
      "message" , "%{TIMESTAMP_ISO8601:date} %{LOGLEVEL:level}  %{WORD:type} &uuid=%{USERNAME:uuid}&event_id=%{NUMBER:event_id}&event_type=%{INT:event_type}&device_id=%{INT:devic
e_id}&session_id=%{USERNAME:session_id}&uid=%{WORD:uid}"

    }
  }
} 

//格式二
filter {
    grok {
	match => {
	    "message"=>[
		"%{DATA:hostname}\|%{DATA:tag}\|%{DATA:types}\|%{DATA:uid}\|%{GREEDYDATA:msg}",
		"%{DATA:hostname}\|%{DATA:tag}\|%{GREEDYDATA:msg}"]
	    }
    }
}

比较建议的是用正则或ruby进行处理,减少解析日志时的性能损耗

filter {
    ruby {
        code =>'
        arr = event["message"].split("|")
        if arr.length == 5
            event["hostname"] = arr[0]
            event["tag"] = arr[1]
            event["types"] = arr[2]
            event["uid"] = arr[3]
            event["msg"] = arr[4]
        elsif arr.length == 3
            event["hostname"] = arr[0]
            event["tag"] = arr[1]
            event["msg"] = arr[2]
        end'
       remove_field => ['type','_id','input_type','tags','message','beat','offset']
    }
}

以上的ruby适用于5.0以前的logstash版本,新的版本需要这样写

如:将日志切割为下面的格式

172.16.xxx.xx|
122.224.xxx.xxx|
21|
2019-05-09 16:21:31|
GET|
200|
1427|
/crowdfundingservice2/mall/order/v2/detail?orderNo=2019042416462988710082581&client=wap|
Mozilla/5.0 (iPhone; CPU iPhone OS 11_0 like Mac OS X) AppleWebKit/604.1.38 (KHTML, like Gecko) Version/11.0 Mobile/15A372 Safari/604.1|
fc47279dd37d2841673ac47cc3784ad7|
56C91F37A273CAA0E050A00ACC3C7AFC 257902d4fef6f5bd8c151ec405ef51a9

-服务器名称
-客户端Ip,  X-Forwarded-For代理层级显示
-请求时间,毫秒记
-请求日期
-请求类型 GET/POST
-请求状态
-发送的字节数, 排除http头
-请求的url地址 + url参数【可以拆分为请求项目,请求接口】
-代理
-签名
-认证信息

代码格式如下

filter {
  ruby {
    code =>'
    arr = event.get("message").split("|")
    if arr.length == 11
      event.set("serverIp",arr[0])
      event.set("clientIp",arr[1])
      event.set("mills",arr[2])
      event.set("date",arr[3])
      event.set("httpType",arr[4])
      event.set("httpStatus",arr[5])
      event.set("Bytes",arr[6])
      event.set("requestUrl",arr[7])
      event.set("userAgent",arr[8])
      event.set("sign",arr[9])
      event.set("auth",arr[10])
      url = event.get("requestUrl").split("/")
      event.set("programm",url[1])
      api = event.get("requestUrl").split("?")
      event.set("api", api[0])
    elsif arr.length == 3
    end'
最终数据被索引为下面的格式
{
  "_index": "sdk-test-2019.05.09",
  "_type": "doc",
  "_id": "DtZLnGoB1u2C0dAJw7mz",
  "_version": 1,
  "_score": null,
  "_source": {
    "api": "/crowdfundingservice2/mall/order/checkReturn",
    "httpStatus": "200",
    "path": "/home/admin/logs/access_log.2019-05-09",
    "userAgent": "okhttp/3.8.1",
    "host": "server-test2",
    "requestUrl": "/crowdfundingservice2/mall/order/checkReturn?client=android&clientversion=6.5.0&deviceid=18870&appType=0",
    "sign": "44b45619d0a84fcaa31ebc3f9ce083ee",
    "httpType": "POST",
    "serverIp": "172.16.202.126",
    "@timestamp": "2019-05-09T11:12:52.039Z",
    "message": "xxx|xxx|55|2019-05-09 19:12:45|POST|200|387|/crowdfundingservice2/mall/order/checkReturn?client=android&clientversion=6.5.0&deviceid=18870&appType=0|okhttp/3.8.1|44b45619d0a84fcaa31ebc3f9ce083ee|6F37DB883776A3A6E050A00ACC3C60BC de5856b09b066dd0b3bc2beec1ec9a9e",
    "mills": "55",
    "Bytes": "387",
    "auth": "6F37DB883776A3A6E050A00ACC3C60BC de5856b09b066dd0b3bc2beec1ec9a9e",
    "clientIp": "122.224.233.150",
    "@version": "1",
    "date": "2019-05-09 19:12:45",
    "programm": "crowdfundingservice2"
  },
  "fields": {
    "@timestamp": [
      "2019-05-09T11:12:52.039Z"
    ]
  },
  "sort": [
    1557400372039
  ]
}

参考官方地址:

https://www.elastic.co/guide/en/logstash/current/plugins-filters-ruby.html#plugins-filters-ruby-add_field

 


   原创文章,转载请标明本文链接: ELK搭建环境之配置logstash监听文件并利用grok|ruby插件进行切割利于统计

如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!

发表评论

电子邮件地址不会被公开。 必填项已用*标注

更多阅读