【星海出品】LogStash

智谱清言
ZhipuAI

LogStash | Fluentd

Logstash 依赖于JAVA
Oracle 修复了 Java 超过一半的问题。所以我们到Oracle 官方网站上下载 java 的JDK，安装方法见之前的文档

LogStash 的核心价值就在于它将业务系统和数据展示系统隔离开来，屏蔽了各自系统变化对彼此的影响，使系统之间的依赖降低并可独自进化发展。
LogStash 可以从多个数据源提取数据，然后再清洗过滤并发送到指定的目标数据源。目标数据源也可以是多个，而且只要修改 LogStash 管道配置就可以轻松扩展数据的源头和目标。

LogStash 读取大文件主要通过其内部的一个关键组件——FileWatch Ruby Gem库。
这个库使得LogStash能够监听文件变化，并高效地读取大文件内容。

LogStash 使用 FileWatch 库来监听指定路径下的文件变化。
FileWatch 支持 glob 展开文件路径，这意味着可以指定通配符（如*.log）来匹配多个文件。

为了避免在读取大文件时漏掉数据，FileWatch 会记录一个 .sincedb 的数据库文件。
.sincedb 文件中记录了每个被监听的文件的 inode、major number、minor number 和当前读取位置（pos）。

LogStash 可以选择 tail 模式或 read 模式来读取文件。
Tail 模式：当文件有新数据写入时，LogStash 会继续从上一次读取的位置开始读取。
Read 模式：LogStash 会一次性读取整个文件内容，并在读取完成后关闭文件。
通过设置 close_older 参数，可以指定在读取文件后多久关闭文件。这有助于管理打开的文件句柄数量。

logstash可以通过设置 close_older 参数，可以指定在读取文件后多久关闭文件。这有助于管理打开的文件句柄数量。

input {  file {  path => "/var/log/myapp/*.log"  start_position => "beginning"  sincedb_path => "/dev/null" # 或者其他路径来存储sincedb文件  close_older => "5 minutes" # 5分钟后没有新日志则关闭文件句柄# close_older参数默认值是3600秒（即一小时）# discover_interval 默认15秒}  
}  filter {  # ... 你的过滤配置 ...  
}  output {  # ... 你的输出配置 ...  
}

%{IPORHOST:clientip} - - \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response:int} %{NUMBER:bytes:int} "%{GREEDYDATA:referrer}" "%{GREEDYDATA:agent}"Xxx/logstash-xxx/vendor/bundle/jruby/xxx/gems/logstash-patterns-core-4.1.2/patternsUSER_IPADDRESS   ([0-9\.]+)\s+
DATETIME  ([0-9\-]+\s[0-9\:]+)\s+
METHOD  ([A-Z]+)\s+
URL ([\/A-Za-z0-9\.]+)\s+
STATUS  ([0-9]+)\s+
REQUEST_SEND  ([0-9]+)\s+
REQUEST_TIME  ([0-9\.]+)

grok {Patterns_dir => [“./patterns”]Match => {“message” => “%{USER_IPADDRESS:user_ip} %{DATETIME:date}  %{METHOD:method}    ”}
}

性能优化：

对于大文件，性能是关键。LogStash 提供了一系列配置选项来优化读取性能，如设置 discover_interval 来控制多久检查一次新文件，使用 exclude 参数来排除不需要的文件等。
还可以通过调整 JVM 堆内存大小、工作线程数等参数来进一步提高 LogStash 的性能。

Ibana

基础使用Management Dev Tools 的 Console

GET _search
{"query": {"match_all": {}}
}
#查看索引列表
GET _cat/indices#查看索引
GET /_cat/indices?v# 查看所有的log开头的索引信息
GET log*/_search  
{  "query": {  "match_all": {}  }  
}#查看特定的log索引信息
GET logstash-20240621/_search
{  "query": {  "match_all": {}  }  
}GET kibana_sample_data_logs/_search
{  "query": {  "match_all": {}  }  
}PUT http://localhost:9200/{索引名}
#创建索引#创建索引为logstash-20240620名字的索引
PUT /logstash-20240620# 查看索引的设置
GET /index_search/_settings# 查看整个集群健康状态
GET _cat/health

精确查询
{
“query”: {
“term”: {
“field_name”: “exact_value”
}
}
}
多匹配
{
“query”: {
“terms”: {
“field_name”: [“value1”, “value2”]
}
}
}

Ibana 使用 Management -> Dev Tools -> Grok Debugger 进行调试。

Sample Data

172.16.1.60 - - [20/Jun/2024:02:51:00 +0000] “GET / HTTP/1.1” 304 0 “-” “Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36”

Grok Pattern

%{IPORHOST:clientip} - - [%{HTTPDATE:timestamp}] “%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}” %{NUMBER:response:int} %{NUMBER:bytes:int} “%{GREEDYDATA:referrer}” “%{GREEDYDATA:agent}”

Result

{
“request”: “/”,
“agent”: “Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36”,
“verb”: “GET”,
“referrer”: “-”,
“response”: 304,
“bytes”: 0,
“clientip”: “172.16.1.60”,
“httpversion”: “1.1”,
“timestamp”: “20/Jun/2024:02:51:00 +0000”
}

Custom

USER_IPADDRESS ([0-9.]+)\s+
DATETIME ([0-9-]+\s[0-9:]+)\s+
METHOD ([A-Z]+)\s+
URL ([/A-Za-z0-9.]+)\s+
STATUS ([0-9]+)\s+
REQUEST_SEND ([0-9]+)\s+
REQUEST_TIME ([0-9.]+)

#grok示例：