本文主要概述 Logstash 的一些最受欢迎的输入插件，以大致了解 Logstash 的用途；相关的环境及软件信息如下：CentOS 7.9、Logstash 8.2.2。

1、什么是 Logstash input 插件

Logstash 用作日志管道，用于侦听已配置日志源（例如，应用程序，数据库，消息代理）的事件，使用过滤器和编解码器对其进行转换和格式化，并运送到输出位置（例如，Elasticsearch 或 Kafka）。

Logstash 如此强大，因为它可以聚合来自多个源（例如 Redis，Apache HTTP 或 Apache Kafka）的日志，这些源位于多个节点上，并将它们放入由多个工作程序和线程管理的高效日志处理队列中。 Logstash 优化了输入和输出目标之间的日志流，从而确保了容错性能和数据完整性。 Logstash 的最大优点之一是可以使用众多过滤器和编解码器，这些过滤器和编解码器可以从日志中提取模式并将日志转换为适合在 Elasticsearch 和 Kibana 中进行分析的丰富数据对象。 这些功能使原始日志能够快速转换为可操作的对象，从而使你的业务受益。

输入插件是 Logstash 管道的重要组件，它们充当输入日志源和 Logstash 过滤功能之间的中间件。 通常，每个输入插件都允许连接到指定的日志源提供程序并使用其 API 提取日志。

2、查看 Logstash 的所有 input 插件

在 Logststash 的安装目录下，可通过如下的命令来查询所有内置的 Logstash 的输入插件：

./bin/logstash-plugin list --group input

3、常用 input 插件

3.1、Beats 家族

Elastic Beats 系列包括许多针对各种数据（日志，指标，事件等）的摄入。 Filebeat 用于日志传送，Packetbeat 用于网络数据传送以及 Metricbeat 用于系统和应用程序指标监视。 Elasticsearch 还提供 Windows 事件日志的 Winlogbeat，审计数据的 Auditbeat 和正常运行时间监视的 Heartbeat。其中的一些 Beat（例如 Filebeat）允许将日志直接传送到你的 Elasticsearch 索引。

随之而来的自然问题是：为什么要使用 Logstash 呢？

input {
     beats {
     port => 5044
  }
}
 
output {
    stdout {
    codec => rubydebug
  }
}

input {
  exec {
    interval => 10
    command => "top -n 1 -b|head -5"
  }
}
 
output {
  stdout {
    codec => rubydebug
  }
}

input {
  file {
    path => ["/home/hadoop/app/nginx-1.8.0/logs/access.log"]
    #type => "system"
    start_position => "beginning"
  }
}

output {
    stdout {
    codec => rubydebug
  }
}

input {
  udp {
    port => 25000
    workers => 4
    codec => json
  }
}
 
output {
  stdout { 
    codec => rubydebug 
  }
}

你可以选择使用 JSON 编解码器将 UDP 消息转换为 JSON 对象，还可以使用 queue_size 参数控制消息队列的大小，并使用 worker 参数指定用于处理 UDP 数据包的工作线程数。

另起一个终端，运行如下命令进行测试：

nc -u 127.0.0.1 25000

3.3.2、Unix Domain Sockets

UNIX 套接字是一种进程间通信（IPC）机制，它允许在同一计算机上运行的进程之间进行双向数据交换。 你可以使用此插件捕获应用程序发出的 Unix 域套接字事件（消息）。 与文件输入插件类似，每个事件等于套接字发出的一行文本。 该插件支持两种模式：服务器和客户端。 在服务器模式下，它将监听客户端连接，在客户端模式下，观察客户端何时连接到服务器。

input {
  unix {
    mode => "server"
    path => "/var/logstash/ls"
    data_timeout => 2
  }
}
 
output {
  stdout { 
    codec => rubydebug 
  }
}

该配置使用服务器模式，并将其配置为侦听在 /var/logstash/ls 路径中生成的 Unix 域套接字事件。

3.3.3、Websocket Input Plugin

WebSocket 协议使 Web 客户端（例如，浏览器）和 Web 服务器之间的交互具有较低的开销，从而实现了实时数据传输并允许在保持连接打开的同时来回传递消息。 Websocket 插件允许从打开的 Websocket 连接读取事件。 唯一必需的参数是打开 Websocket 连接的 URL。

websocket Input plugin 默认的情况下没有安装的，可按照如下的方法进行安装：

bin/logstash-plugin install logstash-input-websocket

 插件当前支持的唯一模式是客户端模式，在该模式下，插件连接到 Websocket 服务器并从服务器接收 Websocket 消息。

input {
  websocket {
    mode => "client"
    url => "ws://127.0.0.1:8080/websocket"
    codec => "plain"
  }
}
 
output {
  stdout { 
    codec => rubydebug 
  }
}

3.3.4、HTTP Input Plugin

HTTP Input Plugin 将 HTTP Post 请求（带有由应用程序发送的正文）转换为插件指定的端点，Logstash 会将消息转换为事件。 应用程序可以将 JSON，纯文本或任何格式的数据传递到端点，并使用相应的编解码器来转换消息。通过利用 Logstash 中可用的庞大插件生态系统，你可以直接在应用程序中触发 Logstash 中的可操作事件，并将其发送到 Elasticsearch。 该插件支持 HTTP 基本身份验证标头和 SSL，可通过 https 安全地发送数据，并提供验证客户端证书的选项。

input {
  http {
    host => "0.0.0.0"
    port => 8088
  }
}
 
output {
  stdout { 
    codec => rubydebug 
  }
}

启动 Logstash 后，可以使用客户端工具(如 Postman) 调用 http://10.49.196.11:8088 服务，给 Logstash 发送数据。

3.3.5、HTTP Poller

HTTP Poller 插件是 Logstash 中另一个基于 HTTP 的输入插件，它允许调用 HTTP API，将响应消息转换为事件并将消息沿管道发送（例如，发送到 Elasticsearch）。

input {
  http_poller {
    urls => {
      test1 => {
        url => "http://localhost:8080/test"
        method => get
      }
    }
    request_timeout => 60
    # Supports "cron", "every", "at" and "in" schedules by rufus scheduler
    # schedule => { every => "10s"}
    schedule => { cron => "* * * * * UTC"}
    codec => "json"
    # A hash of request metadata info (timing, response headers, etc.) will be sent here
    metadata_target => "http_poller_metadata"
  }
}

output {
  stdout { 
    codec => rubydebug 
  }
}

该配置每隔一分钟调用 http://localhost:8080/test 接口，并把请求的元信息存储在 http_poller_metadata 子段中。

3.4、发布/订阅管道

 在物联网，微服务和实时应用程序时代，应用程序之间的数据流和实时消息传递非常流行。 Logstash 对各种消息代理和数据流平台都有很好的支持。

3.4.1、Apache Kafka Plugin

Logstash Kafka 插件可轻松与 Kafka Producer 和 Consumer API 集成。你可以使用默认偏移量管理策略指定多个要订阅的主题。该插件的一个强大功能是，你可以运行多个读取同一主题的 Logstash 实例，以便在多个物理机之间分配负载。要使用此功能，你需要指定一个 group_id，该组创建一个由多个处理器组成的单个逻辑订户。主题中的消息将分发到具有相同 group_id 的所有 Logstash 实例。

input {
   kafka {
     bootstrap_servers => "localhost:9092"
     topics => "test"
     group_id => "mygroup"
  }
}

output {
  stdout { 
    codec => rubydebug 
  }
}

3.4.2、RabbitMQ Plugin

该插件使用 March Hare 包与 RabbitMQ 进行通信；大多数配置选项直接映射到标准的 RabbitMQ and AMQP 概念。

input {
  rabbitmq {
    host => "localhost"
    port => 5672
    user => "admin"
    password => "admin"
    queue => "test"
    durable => true
    codec => "plain" 
  }
}

output {
  stdout { 
    codec => rubydebug 
  }
}

4、总结

Logstash 支持各种流行的日志和事件源。 特别是，你可以将输入插件用于许多主要的网络协议，消息代理，IRC 服务器，数据库 和 Web 应用程序。更重要的是，Logstash 可以通过网络接收 bash 命令，系统日志和其他系统信息，本地文件和文件，将它们转换为有价值的事件，并过滤和丰富日志，以供 ELK 堆栈中的后续分析或你喜欢的任何其他日志分析解决方案 。 所有这些功能使 Logstash 成为用于传输和规范化数据的 ELK 堆栈的强大组件。

本文中介绍的插件只是 Logstash 输入插件的很小的一部分，可查看 Logstash 官方文档 了解更多信息。

参考：https://blog.csdn.net/UbuntuTouch/article/details/104260743