Logstash
 
简介
 

 用途：
 
 
部署安装
 
 
#检查jdk环境，要求jdk1.8+
java -version
#解压安装包
tar -xvf logstash-6.5.4.tar.gz
#第一个logstash示例
bin/logstash -e 'input { stdin { } } output { stdout {} }'
 
执行效果如下：
 
 
配置详解
 
Logstash的配置有三部分，如下：
 
nput { #输入
stdin { ... } #标准输入
}
filter { #过滤，对数据进行分割、截取等处理
 ...
}
output { #输出
stdout { ... } #标准输出
}
 
输入
 
采集各种样式、大小和来源的数据，数据往往以各种各样的形式，或分散或集中地存在于很多系统中。
Logstash 支持各种输入选择 ，可以在同一时间从众多常用来源捕捉事件。能够以连续的流式传输方式，轻松地从您的日志、指标、Web 应用、数据存储以及各种 AWS 服务采集数据。
 
 
过滤
 
实时解析和转换数据
数据从源传输到存储库的过程中，Logstash 过滤器能够解析各个事件，识别已命名的字段以构建结构，并将它们转换成通用格式，以便更轻松、更快速地分析和实现商业价值。
 
 
输出
 
Logstash 提供众多输出选择，您可以将数据发送到您要指定的地方，并且能够灵活地解锁众多下游用例。
 

第一种：
 
通过官网logstash包下载安装 https://artifacts.elastic.co/downloads/logstash/logstash-7.10.1-linux-aarch64.tar.gz
 
解压在mnt目录下
 
 
 
修改配置文件.conf
 
cd /mnt/logstash-7.10.1
 
bin/logstash -f config/logstash-sample.conf -t   检测配置文件
 
bin/logstash -f config/logstash-sample.conf >/dev/null 2>log & 后台启动
 
日志在 /mnt/logstash-7.10.1/logs
 
 
 
 
 
 
 
 
 
 
 
input { stdin { } }
output {
  elasticsearch { hosts => ["localhost:9200"] }
  stdout { codec => rubydebug }
}
 
 
 
bin/logstash -f logstash-simple.conf
 
 
 
 
 
 
 
 
 
第二种： docker 安装
 
https://www.elastic.co/cn/downloads/logstash
 https://artifacts.elastic.co/downloads/logstash/logstash-7.10.1-linux-aarch64.tar.gz
 yum: command not found
 
使用docker 安装Logstash，来收集文件/var/log/messages
 
https://blog.csdn.net/shykevin/article/details/108251996
 5044 9600
 
docker pull wep4you/logstash:7.9.2
 
docker run -d --name=logstash wep4you/logstash:7.9.2
 

 等待30秒，查看日志
 docker logs -f logstash
 
mkdir -p /mnt/docker/elk7/logstash/config/conf.d
 

 docker cp logstash:/usr/share/logstash/config /mnt/docker/elk7/logstash/
 docker cp logstash:/usr/share/logstash/data /mnt/docker/elk7/logstash/
 docker cp logstash:/usr/share/logstash/pipeline /mnt/docker/elk7/logstash/
 chmod 777 -R /mnt/docker/elk7/logstash
 
vim /mnt/docker/elk7/logstash/config/logstash.yml
 http.host: "0.0.0.0"
 xpack.monitoring.elasticsearch.hosts: [ "http://139.9.53.103:9200" ]
 path.config: /usr/share/logstash/config/conf.d/*.conf
 path.logs: /usr/share/logstash/logs
 
vim /mnt/docker/elk7/logstash/config/conf.d/syslog.conf
 input {
   file {
     #标签
     type => "systemlog-localhost"
     #采集点
     path => "/var/log/messages"
     #开始收集点
     start_position => "beginning"
     #扫描间隔时间，默认是1s，建议5s
     stat_interval => "5"
   }
 }
 
output {
   elasticsearch {
     hosts => ["您的ip:9200"]
     index => "logstash-system-localhost-%{+YYYY.MM.dd}"
  }
 }
 

docker rm -f logstash  强制删除一个运行中的日容器
 
chmod 644 /var/log/messages
 
docker run -d -p 5044:5044 --name=logstash --restart=always -v /mnt/docker/elk7/logstash:/usr/share/logstash -v /var/log/messages:/var/log/messages wep4you/logstash:7.9.2
 
docker run -d --restart=always --log-driver json-file --log-opt max-size=100m --log-opt max-file=2 -p 5044:5044 --name xinyar-logstash -v /data/elk/logstash/logstash.yml:/usr/share/logstash/config/logstash.yml -v /data/elk/logstash/conf.d/:/usr/share/logstash/conf.d/ logstash:7.4.1
   
 
docker exec -it logstash /bin/bash
  

timestamp 用来标记事件的发生时间。因为这个字段涉及到Logstash的内部流转，所以必须是一个joda对象，如果你尝试自己给一个字符串字段重命名为@timestamp的话，Logstash会直接报错。所以，请使用 filter/date插件来管理这个字段。
type 标记事情的唯一类型。
host 标记事情发生在哪里。
tags 标记事件的某方面属性。只是一个数组，一个时间可以有多个标签。
version 标记事件的版本信息
id 标记事件分配的ID信息
index 标记事件所在的索引
score
 
可以随意的给事件添加字段或者从事件里删除字段。事实上事件就是一个Ruby对象。 
 命令行配置Logstash 提供了一个shell脚本叫logstash,支持以下参数：
-e 执行。 bin/logstash -e 是执行一个默认的配置。
–config 或 -f 文件。 
 真实运用中，我们会写很长的配置，甚至可能超过shell锁能支持的1024个字符长度。所以我们必须把配置固话到文件里，然后通过 bin/logstash -f agent.conf 这样的形式来运行。
 
可以直接用 bin/logstash -f /etc/logstash.d/ 来运行，这样logstash会自动读取/etc/logstash.d/目录下所有的*.conf 的文本文件，然后在自己内存里拼接成一个完整的大配置文件，再去执行。
–configtest 或 -t 测试。 
 用来测试Logstash读取到的配置文件语法是否能正常解析。LogStash列出目录下所有文件时，是用字母排序的。
–logs 或 -l 日志。 
 Logstash默认输出日志到标准错误。
–filterworkers 或 -w 工作线程。 
 Logstash会运行多个线程。可以使用 bin/logstash -w 5 这样的方式强制Logstash为过滤插件运行5个线程。在Logstash-2.0之前，该参数默认是1，之后修改为自动适配CPU核数的一半。
–verbose 输出一定的调试日志。
– debug 输出更多的调试日志。
 
过程管道
 
Inputs：你使用Inputs去获取数据到Logstash中，下面是一些常用的inputs类型。
 
  
file：从文件中读取，类似于UNIX命令中的 tail -0f
syslog：监听514端口上的系统日志消息，并且根据RFC3164格式来解析。
redis：从redis服务读取信息，使用redis管道和redis列表。Redis经常用来使用作为一个“broker”在一个集中的Logstash安装，运输Logstash事件从一个远程的Logstash“shippers”。
beats：由Filebeat发送的进程事件。
FIlters：FIlters是在Logstash管道中的中间运输设备。你可以组合过滤条件来对符合一定规则的事件进行操作。
 
  
grok：解析和构建任意文本。Grok是当前在Logstash中去解析未构建的日志数据变成结构化和可查询的最佳方式。在Logstash中内置了120个模式，总有一个会符合你的需求。
mutate：执行对事件字段的一般转换。你可以重命名，删除，替换和修改字段。
drop：丢掉一个事件，例如debug事件。
clone：给事件做一个副本，可能添加或删除字段。 
 -geoip：添加关于Ip地址的地理位置信息。
Outputs：Outputs是Logstash传输途径中的最后解析。一个事件可以有多个输出，但是一旦所有的输出处理完成后，该事件就完成了它的执行。
 
  
elasticsearch：传输数据给Elasticsearch。如果你打算保存数据在一个高效的，方便的，更轻松的可查询结构中。
file： 将事件数据写在磁盘中。
graphite：把数据传输到graphite，一个受欢迎的开源工具用来保存和图表化数据。
statsd：Statsd是一个简单的网络守护进程，基于Node.js平台，通过UDP或者TCP方式侦听各种统计信息，包括计数器和定时器，并发送聚合信息到后端服务。
Codecs：基于流，可以对一部分输入输出做过滤的操作。Codes可以让你轻松的将消息从序列化的过程中分离出来。
 
  
json：编码或解码数据为Json格式。
multiline：合并多行数据事件例如java异常和堆栈信息为一个单独的事件。
 
管道在现在的版本中过滤器和输出都在同一个线程中。在2.2之前的版本中，过滤器和输出都存储在单独的空间，被独立的线程处理。新的管道大大增加了线程活跃度，减少了资源使用，提高了吞吐率。当前的Logstash管道是一个微型batching管道，比一次做一件事的方法更高效。这些效率来自很多地方，有两个比较突出的在于争夺资源的减少和线程活跃度的提高。 
 命令行标识Logstash有许多的命令标识。你可以使用 –help标识来展示这些信息。
 
Nginx : 从1.7版开始，加入了syslog支持。可以通过syslog直接发送日志出来。
 
配置：
 
access_log syslog:server=unix:/data0/rsyslog/nginx.sock locallog;
 
或者直接发送给远程logstash 
 access_log syslog:server=192.168.0.2:5140,facility=local6,tag=ngnix-access,severity=info logstashlog; 
 在这种情况下，Ngnix将使用local7.info等级，ngnix为标签，发送数据。注意，采用syslog发送日志的时候，无法配置buffer=16k选项。 
 Log4J 配置直接通过log4j写入到logstash中。 
 XML配置文件： 
 需要配置Java应用的Log4J设置，启动一个内置的
 
<appender name="LOGSTASH" class="org.apache.log4j.net.SocketAppender">
<param name="RemoteHost" value="logstash_hostname" />
<param name="ReconnectionDelay" value="60000" />
<param name="LocationInfo" value="true" />
<param name="Threshold" value="DEBUG" />
</appender>
 
然后把这个新定义的appender对象加入到 root logger 里，可以跟其他已有logger共存。
 
<root>
<level value="INFO"/>
<appender-ref ref="OTHERPLACE"/>
<appender-ref ref="LOGSTASH"/>
</root>
 
properties文件配置：
 
log4j.rootLogger=DEBUG, logstash
 
SocketAppender
 
log4j.appender.logstash=org.apache.log4j.net.SocketAppender
log4j.appender.logstash.Port=4560
log4j.appender.logstash.RemoteHost=logstash_hostname
log4j.appender.logstash.ReconnectionDelay=60000
log4j.appender.logstash.LocationInfo=true
 
logstash 支持所有主流日志类型，插件支持最丰富，可以灵活DIY，但性能较差，JVM容易导致内存使用量高。
 
Input
 
collected 是一个守护进程，用来收集系统性能和提供各种存储方式来存储不同值的机制。它会在系统运行和存储信息时周期性的统计系统的相关统计信息。利用这些信息有助于查找当前系统性能瓶颈（如作为性能分析）和预测系统未来的load等。
Generator生成测试数据

Logstash 是一个功能强大的工具，可与各种部署集成。 它提供了大量插件，可帮助你解析，丰富，转换和缓冲来自各种来源的数据。 如果你的数据需要 Beats 中没有的其他处理，则需要将 Logstash 添加到部署中。
 
 
Logstash 是 Elastic 栈非常重要的一部分，但是它不仅仅为 Elasticsearch 所使用。它可以介绍广泛的各种数据源。Logstash 可以帮利用它自己的 Filter 帮我们对数据进行解析，丰富，转换等。
 
 
最后，它可以把自己的数据输出到各种需要的数据储存地，这其中包括 Elasticsearch。
 
 
尽管 Logstash 不仅仅处理 Log，但是如果我们以 Log 为例来描述它在 Elastic Stack 中的工作流程，它可以用如下的一张图来进行描述：
 
 
在上面，最原始的 Log 数据，经过 Logstash 的处理，可以把非结构化的数据变成结构化的数据。我们甚至可以使用 Logstash 强大的 Filter 来对数据继续加工。比如在上面，我们甚至可以使用 GeoIP 过滤器来丰富 IP 地址字段，从而能得到具体的的位置信息。我们可以结合外部数据库对数据丰富，转换等等。
 
在今天的文章中，我将讲述如果使用 Elastic Stack 家族的 Logstash。
 

 
 
Elastic Logstash 动手实践
 
 
Elastic 总览 
 
在之前的一些市场活动中，我发现很多的开发者和 Elastics Stack 的使用者都对 ELK 比较熟悉，但是他们对 Elastic 公司并不熟悉。他们很了解 ELK 是一个开源的软件栈，但是不知道在这个软件栈后面还有一个商业的 Elastic 公司的存在。简单的一句话：
 
                                                                                        Elastic 是一个搜索公司
 
 
当大多数人听到搜索这个词时，他们会想到一个搜索框。 在网络上搜索内容或在应用程序上搜索内容。但是搜索不仅仅是搜索框。
 
 
搜索通过图形或图表可视化日志和/或指标数据。也许你正在使用自己喜欢的应用程序通过寻找距离你最近并且可以接受信用卡并且具有4星级评级的餐馆来寻找餐馆。 这是一个过滤器搜索。或者，也许你正在使用 rideshare 应用程序。 连接驾驶员和乘客并找到到达目的地的路线。地图搜索也是搜索。搜索的不仅仅是这些示例。 它适用于许多的应用场景！
 
Elasticsearch 广泛地应用于我们日常生活的很多的场景，比如 Uber，滴滴打车，美团送餐，抖音视频搜索及推荐等等。还有很多著名的网站也使用 Elasticsearch 来做它们的搜索引擎。大家知道 github 上的搜索就是使用 Elasticsearch 来完成的，还有很多的网站，比如 wikipedia 及 linkedin 等。我们中国开发者最喜欢的 CSDN 里的搜索也是使用 Elasticsearch :)
 
我们可以在 Free and Open Search: The Creators of Elasticsearch, ELK & Kibana | Elastic 官方地址找到更多关于 Elastic 公司的介绍。
 
 
Elasticsearch
 
Elastic 公司的核心的产品是 Elasticsearch。关于Elasticsearch的更多描述可以在我之前的文章 “Elasticsearch简介”。简单地说：
 
 
Elasticsearch 是一个分布式的开源搜索和分析引擎，适用于所有类型的数据，包括文本，数字，地理空间，结构化和非结构化。Elasticsearch 基于 Apache Lucene 构建，并于2010年由 Elasticsearch N.V.（现称为 Elastic）首次发布。
 
Elasticsearch 以其简单的 REST API，分布式性质，速度和易扩展性而闻名。Elasticsearch的搜索体验的基本原则是规模（scale），速度（speed），相关性（relevance）。 总之，这三个属性是 Elastic 与其他产品的区别。 这些属性贯穿我们能看到的任何一个商业的示例，并且如果剥离这些层，则通常是它们使用 Elastic 的真正原因。
 
Scale：可扩展性是指摄取和处理PB级数据的能力。Elasticsearch 集群是分布式的，所以它很容根据商业的需求来扩容。如果需要存储更多的数据，我们很容添加更多的服务器来进行满足商业的需求。
 
Speed：快速获得搜索结果的能力， 即使在大规模的情况下。 在中国有一种说法：天下武功唯快不破。Elasticsearch 可以在 PB 级数据情况下，也能获得毫秒级的全文搜索。即使是新数据导入到 Elasticsearch 中，也可以在 1 秒内变为可以搜索，从而实现近实时的搜索。对于有的数据库来说，搜索可能是需要数小时才能完成。
Relevance： 关联性是一种能够以任意方式查询数据并获得相关结果的能力，而不论是查看文本，数字还是地理数据。Elasticsearch 可以根据数据的匹配度来返回数据。每个搜索的结果有一个分数，它表示匹配的相关度。在返回的数据结果中，匹配度最大的结果排在返回的结果的前面。
 
Elastic Stack
 
 
“ELK” 是三个开源项目的缩写：Elasticsearch，Logstash 和 Kibana。 Elasticsearch 是搜索和分析引擎。Elasticsearch 是整个 Elastic Stack 的核心组件。 Logstash 是一个服务器端数据处理管道，它同时从多个源中提取数据，进行转换，然后将其发送到类似 Elasticsearch 的 “存储” 中。Beats 是一些轻量级的数据摄入器的组合，用于将数据发送到 Elasticsearch 或发向 Logstash 做进一步的处理，并最后导入到 Elasticsearch。 Kibana 允许用户在 Elasticsearch 中使用图表将数据可视化。Kibana 也在不断地完善。它可以对 Elastic Stack 进行监控，管理。同时它也集成了许多应用。这些应用包括 Logs, Metrics，机器学习，Maps 等等。
 
 
Elastic 方案
 
 
Elastic 公司围绕 Elastic Stack 创建了许多的开箱即用的方案。对于很多搜索或数据库的公司来说，他们可能有很好的产品，但是运用它们开发一套实现某种方案来说，也是需要很多的精力来组合不同公司的产品来完成这些方案。围绕 Elastic Stack，Elastic 公司推出了 3+1:
 
 
在上面，我们可以看到 Elastic 的三大解决方案：
 
企业搜索
可观测性
安全
上面的三大解决方案基于同一个 Elastic (ELK) Stack：Elasticsearch 及 Kibana。
 
 
Logstash 简介
 
Logstash 是一个数据流引擎：
 
它是用于数据物流的开源流式 ETL（Extract-Transform-Load）引擎
在几分钟内建立数据流管道
具有水平可扩展及韧性且具有自适应缓冲
不可知的数据源
具有 200 多个集成和处理器的插件生态系统
使用 Elastic Stack 监视和管理部署
Logstash 几乎可以摄入各种类别的数据：
 
 
它可以摄入日志，文件，指标或者网路真实数据。经过 Logstash 的处理，变为可以使用的 Web Apps 可以消耗的数据，也可以存储于数据中心，或变为其它的流式数据。
 
最为流行的数据源
 
 
Logstash 可以很方便地和 Beats一起合作，这也是被推荐的方法
Logstash 也可以和那些著名的云厂商的服务一起合作处理它们的数据
它也可以和最为同样的信息消息队列，比如 redis 或 kafka 一起协作
Logstash 也可以使用 JDBC 来访问 RDMS 数据
它也可以和 IoT 设备一起处理它们的数据
Logstash 不仅仅可以把数据传送到 Elasticsearch，而且它还可以把数据发送至很多其它的目的地，并作为它们的输入源做进一步的处理
 
Logstash 在 Elastic Stack 中是如何融入的？
 
到目前为止，有如下的3中方式能够把我们所感兴趣的数据导入到 Elasticsearch 中:
 
 
正如上面所显示的那样，我们可以通过：
 
Beats：我们可以通过 Beats 把数据导入到 Elasticsearch中
Logstash：我们可以 Logstash 把数据导入。Logstash的数据来源也可以是 Beats
REST API：我们可以通过 Elastic 所提供的丰富的API来把数据导入到 Elasticsearch 中。我们可以通过 Java, Python, Go, Nodejs 等各种 Elasticsearch API 来完成我们的数据导入。如果你对这些感兴趣的话，请阅读 “Elastic：菜鸟上手指南” 中的 “各类语言日志导入”。
那么针对 Beats 来说，Logstash 是如何和其它的 Elastic Stack 一起工作的呢？我们可以看如下的框图：
 
 
从上面我们可以看出来，Beats 的数据可以有如下的三种方式导入到 Elasticsearch 中：
 
Beats ==> Elasticsearch
Beats ==> Logstash ==> Elasticsearch
Beats ==> Kafka ==> Logstash ==> Elasticsearch
正如上面所显示的那样：
 
我们可以直接把 Beats 的数据传入到 Elasticsearch 中，甚至在现在的很多情况中，这也是一种比较受欢迎的一种方案。它甚至可以结合 Elasticsearch 所提供的 pipeline 一起完成更为强大的组合。
我们可以利用 Logstash 所提供的强大的 filter 组合对数据流进行处理：解析，丰富，转换，删除，添加等等。你可以参阅我之前的文章 “Data转换，分析，提取，丰富及核心操作”
针对有些情况，如果我们的数据流具有不确定性，比如可能在某个时刻生产大量的数据，从而导致 Logstash 不能及时处理，我们可以通过 Kafka 来做一个缓存。你可以参考我的文章 “使用 Kafka 部署 Elastic Stack”。
更多关于 Elastic Stack 的介绍，请阅读 “Elastic：菜鸟上手指南” 文章中的 Elastic Stack 架构 章节。
 
 
Logstash 是如何工作的？
 
 
 
 
Logstash 包含3个主要部分： 输入（inputs），过滤器（filters）和输出（outputs）。 你必须定义这些过程的配置才能使用 Logstash，尽管不是每一个都必须的。在有些情况下，我们可以甚至没有过滤器。在过滤器的部分，它可以对数据源的数据进行分析，丰富，处理等等。
 
 
在输出的部分，我们甚至可以有多于一个以上的输出。 
 
在下面的图中，我们可以看到一些常见的 inputs, filters 及 outputs:
 
 
你如果想了解更多关于 Logstash 的这些 Inputs, Filters 及 Outputs，那么请访问 Elastic 的官方网址Logstash Reference [7.14] | Elastic
 
 
好了，我在这篇文章中就先讲到这里。希望你到现在对 Logstash 有一个比初步的认识。在接下来的文章中，我们将具体讲述如何安装 Logstash，并运用一个具体的例子来展示 Logstash 是如何工作的。请参阅这个系列的第二篇文章 “Logstash：Logstash 入门教程 （二）”。