Logstash 有哪些常用的输出插件，如何配置 Elasticsearch 输出？

Logstash 支持多种输出插件，可以将处理后的数据发送到各种目标系统。以下是常用的输出插件及其配置方法。

1. Elasticsearch 输出插件

Elasticsearch 是 Logstash 最常用的输出目标。

基本配置

conf
output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "logstash-%{+YYYY.MM.dd}"
  }
}

重要参数

• hosts：Elasticsearch 节点地址列表
• index：索引名称，支持日期模式
• document_type：文档类型（ES 7.x 后已废弃）
• document_id：文档 ID
• action：操作类型（index、create、update、delete）
• pipeline：ES 管道名称

高级配置

conf
output {
  elasticsearch {
    hosts => ["http://es1:9200", "http://es2:9200"]
    index => "app-logs-%{[service]}-%{+YYYY.MM.dd}"
    document_id => "%{[@metadata][_id]}"
    action => "update"
    doc_as_upsert => true
    pipeline => "timestamp_pipeline"
    # 性能优化
    flush_size => 500
    idle_flush_time => 1
    retry_on_conflict => 3
    # SSL 配置
    ssl => true
    cacert => "/path/to/ca.crt"
    user => "elastic"
    password => "changeme"
  }
}

条件索引

conf
output {
  if [type] == "error" {
    elasticsearch {
      hosts => ["http://localhost:9200"]
      index => "error-logs-%{+YYYY.MM.dd}"
    }
  } else {
    elasticsearch {
      hosts => ["http://localhost:9200"]
      index => "access-logs-%{+YYYY.MM.dd}"
    }
  }
}

2. File 输出插件

File 插件将数据写入文件系统。

基本配置

conf
output {
  file {
    path => "/path/to/output.log"
  }
}

重要参数

• path：输出文件路径
• codec：编解码器
• flush_interval：刷新间隔
• gzip：启用 gzip 压缩

高级配置

conf
output {
  file {
    path => "/var/log/logstash/%{type}-%{+YYYY-MM-dd}.log"
    codec => line { format => "%{message}" }
    flush_interval => 5
    gzip => true
    file_mode => 0644
    dir_mode => 0755
  }
}

3. Kafka 输出插件

Kafka 插件将数据发送到 Kafka 消息队列。

基本配置

conf
output {
  kafka {
    bootstrap_servers => "localhost:9092"
    topic_id => "processed-logs"
  }
}

重要参数

• bootstrap_servers：Kafka 服务器地址
• topic_id：主题名称
• codec：编解码器
• compression_type：压缩类型（none、gzip、snappy、lz4、zstd）

高级配置

conf
output {
  kafka {
    bootstrap_servers => ["kafka1:9092", "kafka2:9092"]
    topic_id => "processed-logs"
    codec => "json"
    compression_type => "snappy"
    acks => "all"
    retries => 3
    batch_size => 16384
    linger_ms => 10
    buffer_memory => 33554432
    # SSL 配置
    security_protocol => "SSL"
    ssl_keystore_location => "/path/to/keystore.jks"
    ssl_keystore_password => "password"
    ssl_truststore_location => "/path/to/truststore.jks"
    ssl_truststore_password => "password"
  }
}

动态主题

conf
output {
  kafka {
    bootstrap_servers => "localhost:9092"
    topic_id => "%{[service]}-logs"
  }
}

4. Redis 输出插件

Redis 插件将数据发送到 Redis。

基本配置

conf
output {
  redis {
    host => "localhost"
    port => 6379
    data_type => "list"
    key => "logstash"
  }
}

数据类型

• list：列表类型
• channel：发布订阅频道
• set：集合类型

高级配置

conf
output {
  redis {
    host => "redis.example.com"
    port => 6379
    data_type => "list"
    key => "logstash-%{[type]}"
    codec => "json"
    db => 0
    password => "secret"
    timeout => 5
    reconnect_attempts => 3
    reconnect_interval => 2
  }
}

5. HTTP 输出插件

HTTP 插件通过 HTTP 接口发送数据。

基本配置

conf
output {
  http {
    url => "http://example.com/api/logs"
    http_method => "post"
    format => "json"
  }
}

重要参数

• url：目标 URL
• http_method：HTTP 方法（post、put、patch）
• format：数据格式（json、form、message）
• headers：HTTP 请求头

高级配置

conf
output {
  http {
    url => "http://api.example.com/v1/logs"
    http_method => "post"
    format => "json"
    headers => {
      "Content-Type" => "application/json"
      "Authorization" => "Bearer %{[api_token]}"
    }
    mapping => {
      "timestamp" => "%{@timestamp}"
      "message" => "%{message}"
      "level" => "%{[log_level]}"
    }
    pool_size => 50
    pool_max_per_route => 25
    keepalive => true
    retry_non_idempotent => true
  }
}

6. Stdout 输出插件

Stdout 插件将数据输出到标准输出，常用于调试。

基本配置

conf
output {
  stdout {
    codec => rubydebug
  }
}

编解码器选项

• rubydebug：格式化输出
• json：JSON 格式
• json_lines：每行一个 JSON
• dots：点号输出

7. 多输出配置

可以同时配置多个输出插件：

conf
output {
  # 输出到 Elasticsearch
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "logs-%{+YYYY.MM.dd}"
  }
  
  # 同时输出到文件备份
  file {
    path => "/backup/logs-%{+YYYY-MM-dd}.log"
  }
  
  # 错误日志发送到 Kafka
  if [level] == "ERROR" {
    kafka {
      bootstrap_servers => "localhost:9092"
      topic_id => "error-logs"
    }
  }
}

8. 条件输出

使用条件语句控制数据流向：

conf
output {
  if [type] == "apache" {
    elasticsearch {
      hosts => ["http://localhost:9200"]
      index => "apache-%{+YYYY.MM.dd}"
    }
  } else if [type] == "nginx" {
    elasticsearch {
      hosts => ["http://localhost:9200"]
      index => "nginx-%{+YYYY.MM.dd}"
    }
  } else {
    file {
      path => "/var/log/other-logs.log"
    }
  }
}

最佳实践

1. 批量写入：使用 flush_size 和 idle_flush_time 优化性能
2. 错误处理：配置重试机制和错误日志
3. 索引策略：合理设计索引命名和分片策略
4. 安全配置：使用 SSL/TLS 保护数据传输
5. 监控指标：监控输出插件的性能指标
6. 备份策略：重要数据配置多个输出目标