一、logstash跟es有版本对照关系

了解对照关系，决定要安装的logstash版本

二、ELK出现的原因

三、Logstash工作原理

Logstash事件处理管道有三个阶段：输入→过滤器→输出，输入生成事件，过滤器修改它们，然后输出将它们发送到其他地方。输入和输出支持编解码器，使你能够在数据进入或离开管道时对其进行编码或解码，而无需使用单独的过滤器。

• 输入

你使用输入将数据获取到Logstash中，一些比较常用的输入是：

属性	解析
file	从文件系统上的文件进行读取，非常类似于UNIX命令tail -0F。
syslog	在众所周知的端口514上监听syslog消息并根据RFC3164格式进行解析。
redis	从redis服务器读取数据，同时使用Redis通道和Redis列表，Redis通常被用作集中式Logstash安装中的“broker”，它将从远程Logstash “shipper”中的Logstash事件排队。
beats	处理Beats发送的事件。

有关可用输入的更多信息，请参见输入插件。

• 过滤器

过滤器是Logstash管道中的中间处理设备，如果事件符合一定的条件，你可以将过滤器与条件语句组合在一起，对其执行操作，一些有用的过滤器包括：

属性	解析
grok	解析和构造任意文本，Grok是目前Logstash中解析非结构化日志数据到结构化和可查询数据的最佳方式，使用内置的120种模式，你很可能会找到一个满足你的需要！
mutate	对事件字段执行一般的转换，你可以重命名、删除、替换和修改事件中的字段。
drop	完全删除事件，例如debug事件。
clone	复制事件，可能添加或删除字段。
geoip	添加关于IP地址地理位置的信息（在Kibana中还显示了令人惊叹的图表！）

有关可用过滤器的更多信息，请参见过滤器插件。

• 输出

输出是Logstash管道的最后阶段，事件可以通过多个输出，但是一旦所有的输出处理完成，事件就完成了它的执行，一些常用的输出包括：

属性	解析
elasticsearch	发送事件数据到Elasticsearch，如果你打算以一种高效、方便、易于查询的格式保存数据，那么使用Elasticsearch是可行的。
file	将事件数据写入磁盘上的文件。
graphite	将事件数据发送到graphite，这是一种流行的用于存储和绘制指标的开源工具。http://graphite.readthedocs.io/en/latest/
statsd	发送事件到statsd，“监听统计信息（如计数器和计时器）、通过UDP发送聚合并将聚合发送到一个或多个可插拔后端服务”的服务，如果你已经在使用statsd，这可能对你很有用！

有关可用输出的更多信息，请参见输出插件。

• 编解码器

Codecs是基本的流过滤器，可以作为输入或输出的一部分进行操作，Codecs使你能够轻松地将消息的传输与序列化过程分开，流行的codecs包括json、msgpack和plain（text）。

属性	解析
json	以JSON格式对数据进行编码或解码。
multiline	将多行文本事件（如java异常和stacktrace消息）合并到单个事件中。

有关可用编解码器的更多信息，请参见编解码器插件。

• 执行模型

Logstash事件处理管道协调输入、过滤器和输出的执行。Logstash管道中的每个输入阶段都在自己的线程中运行，输入将事件写入位于内存（默认）或磁盘上的中央队列，每个管道工作线程从这个队列中取出一批事件，通过配置的过滤器运行事件批处理，然后通过任何输出运行过滤的事件，可以配置批处理的大小和管道工作线程的数量（参见调优和分析Logstash性能）。默认情况下，Logstash使用内存有限队列在管道阶段之间（输入→过滤器和过滤器→输出）来缓冲事件，如果Logstash不安全的终止，则存储在内存中的任何事件都将丢失。为了防止数据丢失，你可以启用Logstash将运行中的事件持久化到磁盘上，有关更多信息，请参见持久队列。

四、logstash的配置文件

1. logstash.yml
   你可以在Logstash设置文件logstash.yml中设置选项来控制Logstash执行，例如，你可以指定管道设置、配置文件的位置、日志记录选项和其他设置。当你运行Logstash时，logstash.yml文件中的大多数设置都可以作为命令行标志使用，在命令行中设置的任何标志都会覆盖logstash.yml文件中的相应设置。logstash.yml文件是用YAML编写的，它的位置因平台而异（参见Logstash目录布局），你可以以层次结构形式指定设置或使用平面键，例如，要使用分层表单设置管道批处理大小和批延迟，你需要指定：

pipeline:
  batch:
    size: 125
    delay: 50

要表示与平面键相同的值，需要指定：

pipeline.batch.size: 125
pipeline.batch.delay: 50

logstash.yml文件还支持bash风格的环境变量插值设置值。

pipeline:
  batch:
    size: ${BATCH_SIZE}
    delay: ${BATCH_DELAY:50}
node:
  name: "node_${LS_NODE_NAME}"
path:
   queue: "/tmp/${QUEUE_DIR:queue}"

注意，${VAR_NAME:default_value}表示法是受支持的，在上面的示例中，它设置了一个默认的批延迟50和一个默认的path.queue为/tmp/queue的。

模块也可以在logstash.yml文件中指定，模块定义将具有这种格式：

modules:
  - name: MODULE_NAME1
    var.PLUGIN_TYPE1.PLUGIN_NAME1.KEY1: VALUE
    var.PLUGIN_TYPE1.PLUGIN_NAME1.KEY2: VALUE
    var.PLUGIN_TYPE2.PLUGIN_NAME2.KEY1: VALUE
    var.PLUGIN_TYPE3.PLUGIN_NAME3.KEY1: VALUE
  - name: MODULE_NAME2
    var.PLUGIN_TYPE1.PLUGIN_NAME1.KEY1: VALUE
    var.PLUGIN_TYPE1.PLUGIN_NAME1.KEY2: VALUE

如果使用命令行标志--modules，则忽略在logstash.yml文件中定义的任何模块。
logstash.yml文件包括以下设置，如果你使用的是X-Pack，请参阅Logstash中的X-Pack设置。

设置	描述	默认值
node.name	节点的描述性名称	机器的主机名
path.data	Logstash及其插件用于任何持久需求的目录	LOGSTASH_HOME/data
pipeline.id	管道的ID	main
pipeline.workers	将并行执行管道的过滤和输出阶段的工人数量，如果你发现事件正在备份，或者CPU没有饱和，请考虑增加这个数字，以更好地利用机器处理能力	主机CPU核心的数量
pipeline.batch.size	在尝试执行过滤器和输出之前，单个工作线程将从输入中收集的最大事件数，更大的批处理大小通常更高效，但代价是增加内存开销，你可能需要增加jvm.options配置文件中的JVM堆空间，有关更多信息，请参阅Logstash配置文件	125
pipeline.batch.delay	当创建管道事件批处理时，在向管道工作人员发送一个较小的批处理之前，等待每个事件的时间为多少毫秒	50
pipeline.unsafe_shutdown	当设置为true时，即使内存中仍然存在游离事件，也会在关闭期间强制Logstash退出，默认情况下，Logstash将拒绝退出，直到所有接收到的事件都被推送到输出，启用此选项可能导致关闭期间的数据丢失	false
path.config	主管道的Logstash配置路径，如果指定目录或通配符，配置文件将按字母顺序从目录中读取	None
config.test_and_exit	当设置为true时，检查配置是否有效，然后退出，注意，在此设置中没有检查grok模式的正确性，Logstash可以从一个目录中读取多个配置文件，如果你把这个设置和log.level: debug结合起来，Logstash将对合并后的配置文件进行日志记录，并用它来自的源文件注解每个配置块	false
config.reload.automatic	当设置为true时，定期检查配置是否已更改，并在更改配置时重新加载配置，这也可以通过SIGHUP信号手动触发	false
config.reload.interval	Logstash多久检查一次配置文件以查看更改	3s
config.debug	当设置为true时，将完整编译的配置显示为debug日志消息，你还必须设置log.level: debug，警告：日志消息将包含传递给插件配置的任意密码选项，可能会导致明文密码出现在日志中！	false
config.support_escapes	当设置为true时，引号中的字符串将处理以下转义序列：\n变成文字换行符（ASCII 10），\r变成文字回车（ASCII 13），\t变成文字制表符（ASCII 9），\变成字面反斜杠\，"变成一个文字双引号，'变成文字引号	false
modules	当配置时，modules必须位于上表中描述的嵌套YAML结构中	None
queue.type	用于事件缓冲的内部队列模型，为基于内存中的遗留队列指定memory，或者persisted基于磁盘的ACKed队列（持久队列）	memory
path.queue	启用持久队列时存储数据文件的目录路径（queue.type: persisted）	path.data/queue
queue.page_capacity	启用持久队列时使用的页面数据文件的大小（queue.type: persisted），队列数据由分隔成页面的仅追加的数据文件组成	64mb
queue.max_events	启用持久队列时队列中未读事件的最大数量（queue.type: persisted）	0（无限）
queue.max_bytes	队列的总容量（字节数），确保磁盘驱动器的容量大于这里指定的值，如果queue.max_events和queue.max_bytes都指定，Logstash使用最先达到的任何标准	1024mb（1g）
queue.checkpoint.acks	当启用持久队列时，在强制执行检查点之前的最大ACKed事件数（queue.type: persisted），指定queue.checkpoint.acks: 0设置此值为无限制	1024
queue.checkpoint.writes	启用持久队列时强制执行检查点之前的最大写入事件数（queue.type: persisted），指定queue.checkpoint.writes: 0设置此值为无限制	1024
queue.drain	启用时，Logstash会一直等到持久队列耗尽后才关闭	false
dead_letter_queue.enable	标记指示Logstash以插件支持的DLQ特性	false
dead_letter_queue.max_bytes	每个dead letter队列的最大大小，如果条目将增加dead letter队列的大小，超过此设置，则删除条目	1024mb
path.dead_letter_queue	存储dead letter队列数据文件的目录路径	path.data/dead_letter_queue
http.host	指标REST端点的绑定地址	“127.0.0.1”
http.port	指标REST端点的绑定端口	9600
log.level	日志级别，有效的选项是：fatal、error、warn、info、debug、trace	info
log.format	日志格式，设置为json日志以JSON格式，或plain使用Object#.inspect	plain
path.logs	Logstash将其日志写到的目录	LOGSTASH_HOME/logs
path.plugins	哪里可以找到自定义插件，你可以多次指定此设置以包含多个路径，插件应该在特定的目录层次结构中：PATH/logstash/TYPE/NAME.rb，TYPE是inputs、filters、outputs或codecs，NAME是插件的名称	特定于平台的

五、logstash配置

下面的示例演示如何配置Logstash来过滤事件，处理Apache日志和syslog消息，并使用条件来控制哪些事件由过滤器或输出处理。如果你需要帮助构建grok模式，请尝试grok调试器，Grok调试器是基本许可证下的X-Pack特性，因此可以免费使用。

• 配置过滤器

过滤器是一种在线处理机制，它提供了根据需要对数据进行切片和切割的灵活性，让我们看一下活动中的一些过滤器，下面的配置文件设置了grok和date过滤器。

input { stdin { } }

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  date {
    match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

output {
  elasticsearch { hosts => ["localhost:9200"] }
  stdout { codec => rubydebug }
}

使用此配置运行Logstash：

bin/logstash -f logstash-filter.conf
现在，将下面的行粘贴到你的终端并按Enter键，这样它就会被stdin输入处理：

127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] “GET /xampp/status.php HTTP/1.1” 200 3891 “http://cadenza/xampp/navi.php” “Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0”
你应该会看到返回到stdout的是这样的：

{
        "message" => "127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] \"GET /xampp/status.php HTTP/1.1\" 200 3891 \"http://cadenza/xampp/navi.php\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0\"",
     "@timestamp" => "2013-12-11T08:01:45.000Z",
       "@version" => "1",
           "host" => "cadenza",
       "clientip" => "127.0.0.1",
          "ident" => "-",
           "auth" => "-",
      "timestamp" => "11/Dec/2013:00:01:45 -0800",
           "verb" => "GET",
        "request" => "/xampp/status.php",
    "httpversion" => "1.1",
       "response" => "200",
          "bytes" => "3891",
       "referrer" => "\"http://cadenza/xampp/navi.php\"",
          "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0\""
}

如你所见，Logstash（在grok过滤器的帮助下）能够解析日志行（碰巧是Apache的“组合日志”格式），并将其分解为许多不同的离散信息，一旦开始查询和分析日志数据，这就非常有用。例如，你将能够轻松地在HTTP响应码、IP地址、referrers等上运行报表。Logstash有很多现成的grok模式，因此如果你需要解析通用的日志格式，很可能已经有人为你完成了这项工作，有关更多信息，请参阅GitHub上Logstash grok模式的列表。本例中使用的另一个过滤器是date过滤器，这个过滤器会解析一个时间戳，并将其用作事件的时间戳（不管你什么时候使用日志数据）。你将注意到，本例中的@timestamp字段设置为2013年12月11日，尽管Logstash在随后的某个时间摄取了该事件，这在备份日志时非常方便，它使你能够告诉Logstash“使用此值作为此事件的时间戳”。

• 处理Apache日志

让我们做一些有用的事情：处理apache访问日志文件！我们将从本地主机上的文件中读取输入，并根据需要使用条件处理事件。首先，创建一个名为logstash-apache.conf的文件包含以下内容（你可以根据需要更改日志文件路径）：

input {
  file {
    path => "/tmp/access_log"
    start_position => "beginning"
  }
}

filter {
  if [path] =~ "access" {
    mutate { replace => { "type" => "apache_access" } }
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
  }
  date {
    match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
  }
  stdout { codec => rubydebug }
}

然后，使用以下日志条目（或使用你自己的webserver中的一些日志条目）创建上面配置的输入文件（在本例中为“/tmp/access_log”）：

71.141.244.242 - kurt [18/May/2011:01:48:10 -0700] "GET /admin HTTP/1.1" 301 566 "-" "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.2.3) Gecko/20100401 Firefox/3.6.3"
134.39.72.245 - - [18/May/2011:12:40:18 -0700] "GET /favicon.ico HTTP/1.1" 200 1189 "-" "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; .NET CLR 2.0.50727; .NET CLR 3.0.4506.2152; .NET CLR 3.5.30729; InfoPath.2; .NET4.0C; .NET4.0E)"
98.83.179.51 - - [18/May/2011:19:35:08 -0700] "GET /css/main.css HTTP/1.1" 200 1837 "http://www.safesand.com/information.htm" "Mozilla/5.0 (Windows NT 6.0; WOW64; rv:2.0.1) Gecko/20100101 Firefox/4.0.1"

现在，使用-f标志运行Logstash以将其传递到配置文件：

bin/logstash -f logstash-apache.conf
现在你应该在Elasticsearch中看到你的apache日志数据了，Logstash打开并读取指定的输入文件，处理遇到的每个事件。记录到此文件的任何追加行也将被捕获，由Logstash作为事件处理，并存储在Elasticsearch中。还有一个额外的好处，他们储藏的字段“type”设置为“apache_access”（这是由输入配置中的type ⇒ "apache_access"行）。

在这个配置中，Logstash只查看apache access_log，但是通过更改上面配置中的一行就可以同时查看access_log和error_log（实际上是任何文件匹配*log）：

input {
  file {
    path => "/tmp/*_log"
...

当你重新启动Logstash时，它将同时处理error和access日志，但是，如果你检查数据（可能使用elasticsearch-kopf)，你会看到access_log被分解为离散字段，而error_log则不是。这是因为我们使用了grok过滤器来匹配标准的组合apache日志格式，并自动将数据分割为单独的字段。如果我们能根据它的格式来控制行是如何被解析的，不是很好吗？嗯，我们可以…
注意，Logstash不会重新处理access_log文件中已经查看的事件，从文件中读取数据时，Logstash保存其位置，并只在添加新行时处理它们。

• 使用条件

你可以使用条件来控制哪些事件由过滤器或输出处理，例如，你可以根据出现在哪个文件（access_log、error_log以及以“log”结尾的其他随机文件）中来标记每个事件。

input {
  file {
     #标签
    type => "systemlog-localhost"
    #采集点
    path => "/tmp/*_log"
    #开始收集点
    start_position => "beginning"
    #扫描间隔时间，默认是1s，建议5s
    stat_interval => "5"
  }
}
filter {
  if [path] =~ "access" {
    mutate { replace => { type => "apache_access" } }
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
    date {
      match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
    }
  } else if [path] =~ "error" {
    mutate { replace => { type => "apache_error" } }
  } else {
    mutate { replace => { type => "random_logs" } }
  }
}

output {
  elasticsearch { hosts => ["localhost:9200"] }
  stdout { codec => rubydebug }
}

这个示例使用type字段标记所有事件，但实际上不解析error或random文件，有很多类型的错误日志，它们应该如何标记取决于你使用的日志。

类似地，你可以使用条件将事件定向到特定的输出，例如，你可以：

• 警告nagios任何状态为5xx的apache事件
• 将任何4xx状态记录到Elasticsearch
• 通过statsd记录所有的状态代码

要告诉nagios任何具有5xx状态码的http事件，首先需要检查type字段的值，如果是apache，那么你可以检查status字段是否包含5xx错误，如果是，发送到nagios。如果不是5xx错误，检查status字段是否包含4xx错误，如果是，发送到Elasticsearch。最后，将所有apache状态码发送到statsd，无论状态字段包含什么：

output {
  if [type] == "apache" {
    if [status] =~ /^5\d\d/ {
      nagios { ...  }
    } else if [status] =~ /^4\d\d/ {
      elasticsearch { ... }
    }
    statsd { increment => "apache.%{status}" }
  }
}

• 处理Syslog消息

Syslog是Logstash最常见的用例之一，而且它处理得非常好（只要日志行大致符合RFC3164），Syslog实际上是UNIX网络日志记录标准，它将消息从客户端发送到本地文件，或通过rsyslog发送到集中式日志服务器。对于本例，你不需要一个功能正常的syslog实例；我们将从命令行中伪造它，这样你就可以了解发生了什么。

首先，让我们为Logstash + syslog创建一个简单的配置文件，名为logstash-syslog.conf。

input {
  tcp {
    port => 5000
    type => syslog
  }
  udp {
    port => 5000
    type => syslog
  }
}

filter {
  if [type] == "syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
  }
}

output {
  elasticsearch { hosts => ["localhost:9200"] }
  stdout { codec => rubydebug }
}

使用这个新配置运行Logstash：

bin/logstash -f logstash-syslog.conf

通常，客户端将连接到端口5000上的Logstash实例并发送消息，对于本例，我们将telnet到Logstash并输入一条日志行（类似于前面在STDIN中输入日志行），打开另一个shell窗口与Logstash syslog输入进行交互并输入以下命令：

telnet localhost 5000

复制粘贴以下行作为示例（你可以自己尝试一些，但是要记住，如果grok过滤器对你的数据不正确，它们可能无法解析）。

Dec 23 12:11:43 louis postfix/smtpd[31499]: connect from unknown[95.75.93.154]
Dec 23 14:42:56 louis named[16000]: client 199.48.164.7#64817: query (cache) 'amsterdamboothuren.com/MX/IN' denied
Dec 23 14:30:01 louis CRON[619]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)
Dec 22 18:28:06 louis rsyslogd: [origin software="rsyslogd" swVersion="4.2.0" x-pid="2253" x-info="http://www.rsyslog.com"] rsyslogd was HUPed, type 'lightweight'.

现在，当你的原始shell处理和解析消息时，你应该会看到Logstash的输出！

{
                 "message" => "Dec 23 14:30:01 louis CRON[619]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)",
              "@timestamp" => "2013-12-23T22:30:01.000Z",
                "@version" => "1",
                    "type" => "syslog",
                    "host" => "0:0:0:0:0:0:0:1:52617",
        "syslog_timestamp" => "Dec 23 14:30:01",
         "syslog_hostname" => "louis",
          "syslog_program" => "CRON",
              "syslog_pid" => "619",
          "syslog_message" => "(www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)",
             "received_at" => "2013-12-23 22:49:22 UTC",
           "received_from" => "0:0:0:0:0:0:0:1:52617",
    "syslog_severity_code" => 5,
    "syslog_facility_code" => 1,
         "syslog_facility" => "user-level",
         "syslog_severity" => "notice"
}

六、安装及例子

docker pull logstash:7.14.0

1. 启动

 docker run -d --name=logstash logstash:7.14.0

2. 复制文件，以便下次启动挂载

docker cp logstash:/usr/share/logstash /mydata/logstash

3. 授权挂载文件夹

chmod 777 -R /mydata/logstash

4. 在es正常运行的前提下

• 编辑/mydata/logstash/config/logstash.yml如下

http.host: "0.0.0.0"
#根据实际修改es的ip：port
xpack.monitoring.elasticsearch.hosts: [ "http://192.168.31.196:9200" ]
# 主管道的Logstash配置路径，如果指定目录或通配符，配置文件将按字母顺序从目录中读取
path.config: /usr/share/logstash/config/conf.d/*.conf
#Logstash将其日志写到的目录
path.logs: /usr/share/logstash/logs

5. 然后根据下面的内容编写logstash.conf文件，就可以启动logstash的收集日志功能

input {
  file {
    #标签
    type => "systemlog-localhost"
    #采集点(这里一定要注意，由于我是docker启动的logstash,这里是容器内的文件，如果要生成外面日志索引，那么文件的路径一定要挂载正确)
    path => "/usr/share/logstash/logs/access_log.2021-03-19.log"
    #开始收集点
    start_position => "beginning"
    #扫描间隔时间，默认是1s，建议5s
    stat_interval => "5"
  }
}

output {
  elasticsearch {
  #集群的话，直接添加多个url
    hosts => ["172.17.0.3:9200"]
    #es的用户名和密码
	user =>"elastic"
	password =>"elastic"
	#建立的索引以日期区分
    index => "logstash-system-localhost-%{+YYYY.MM.dd}"
 }
 #在控制台输出logstash的日志
 stdout { codec=> rubydebug }
}

控制台的输出，说明在es中已经生成索引

6. 在kibana中查看生成的索引
   

关于如何docker安装es，docker安装kibana

参考文章