Logstash配置多入多出并互相隔离

Logstash可以作为一个强大的日志收集工具，能够从多种数据源中收集数据，并将该数据转换为可用于分析和存储的格式。

在本文中，我们将学习如何配置Logstash以实现多入多出，并确保每个数据流之间互相隔离。

多入多出架构

多入多出架构允许多个数据源同时输入数据到Logstash中，并让Logstash将数据输出到多个目的地。这里有一些常见的用例：

• 使用多个logstash模块收集不同类型的日志数据。
• 使用多个数据源，如主机、容器、云平台等，以及多种格式的日志。
• 多个Logstash部署在不同的位置以处理更大的数据负载。

实现多入多出架构需要使用Logstash配置文件和输入和输出插件。下面是一个例子：

input {
  tcp {
    port => 5000
    type => syslog
  }
  tcp {
    port => 5001
    type => apache
  }
}

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

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    document_type => "%{[@metadata][type]}"
  }
  stdout {}
}

在上面的例子中，我们配置了两个输入插件，分别监听5000和5001端口，并使用不同的类型(syslog与apache)来标识不同的数据源。

在filter部分，使用了grok插件将数据进行解析，并在必要时使用date插件处理日期时间信息。

最后，在输出部分，我们使用了elasticsearch插件，将数据存储到Elasticsearch中，并使用stdout插件将数据输出到控制台。在实际生产环境中，可以使用其他插件来将数据输出到不同的存储设备或数据处理管道中。

实现互相隔离

当使用多入多出架构时，应该确保每个数据源之间互相隔离。如果不做好这一点，就可能出现日志混乱和数据遗漏的情况。

为了实现互相隔离，可以为每个数据源设置独立的pipeline和worker。这里有一些需要注意的事项：

• 每个输入插件都应该使用不同的端口号，并且在配置文件中应该使用不同的type来识别它们。
• 使用不同的管道和工作线程来处理每个输入源。
• 应该为每个输入源配置独立的输出插件，这些插件应该将日志数据输出到独立的存储类型中。

下面是一个示例配置文件，演示如何为两个输入源配置独立的管道和工作线程。

input {
  beats {
    port => 5044
    type => "logs"
  }
  tcp {
    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}" }
    }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
  }
}

output {
  if [type] == "syslog" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "syslog-%{+YYYY.MM.dd}"
      document_type => "syslog"
    }
  }
  if [type] == "logs" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "filebeat-%{+YYYY.MM.dd}"
      document_type => "logs"
    }
  }
}

在上面的示例中，我们为两个输入源分别指定了不同的类型，并在output部分为每个类型分别配置了不同的elasticsearch插件，将日志分别输出到相应类型的索引中。

如此一来，我们就可以保证不同来源的日志之间不会混淆，也不会出现数据丢失的情况。

总之，为了实现多入多出架构并确保数据互相隔离，我们需要使用Logstash配置文件和输入输出插件，并为每个数据源设置独立的管道和工作线程。通过这些措施，我们可以更加有效地处理日志数据，并保证数据完整、准确地传输到目标地。