Logstash配置详解

一、Logstash配置文件详解

Logstash是一个开源的数据收集、处理和传输工具，可以从多个来源收集数据、对数据进行处理和转换，最后将数据传输到相应的目的地。Logstash的配置文件是实现这一过程的关键，下面就通过一个示例来详细阐述Logstash的配置文件。

input {
  stdin {}
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
}

output {
  stdout { codec => rubydebug }
}

该配置文件定义了三个部分：输入、过滤和输出。

输入部分指定了数据源，本例中为标准输入stdin。

过滤部分使用grok插件进行数据转换，将数据从输入格式转换为所需的格式。

输出部分指定了将转换后的数据发送到stdout，输出格式为Ruby的debug格式。

二、Logstash配置多个输入

Logstash可以从多个数据源中收集数据，下面的示例阐述了如何在Logstash中配置多个输入。

input {
  file {
    path => "/var/log/messages"
    type => "syslog"
  }
  syslog {
    port => "5514"
    type => "syslog"
  }
}

output {
  stdout { codec => rubydebug }
}

该配置文件定义了两个输入部分，一个是通过文件方式收集数据，路径为“/var/log/messages”，数据类型为“syslog”；另一个是通过syslog协议收集数据，监听端口号为“5514”，数据类型同样为“syslog”。

三、Logstash配置详解

Logstash的策略配置主要通过插件来实现，下面阐述了一些常用的插件:

• Grok插件：用于文本解析和分析。
• Date插件：用于将文本日期转换为日期对象。
• CSV插件：用于解析CSV格式数据。
• Json插件：用于解析JSON数据。
• HTTP插件：用于从HTTP源输入数据。
• Elasticsearch插件：用于将数据输出到Elasticsearch。

四、Logstash配置日志关键字段

在数据处理过程中，通常需要从原始数据中提取常用信息，下面的示例阐述了如何将日志数据中的关键字段提取出来。

假设原始数据格式为：[timestamp] [message]

filter {
  grok {
    match => { "message" => "\[%{TIMESTAMP_ISO8601:timestamp}\] %{GREEDYDATA:message}" }
  }
}

output {
  stdout { codec => rubydebug }
}

在上面的过滤部分，使用grok插件匹配一个包含时间戳和消息的字符串。并将TIMESTAMP_ISO8601匹配到的结果保存到"timestamp"字段，将GREEDYDATA匹配到的结果保存到"message"字段。

五、Logstash配置Grok

Grok插件是Logstash的核心插件之一，主要用于从原始文本中提取结构化的数据。下面的示例演示了如何使用Grok插件提取消息日志中的关键字段。

假设原始数据格式为：[timestamp] [message]

filter {
  grok {
    patterns_dir => "/etc/logstash/patterns"
    match => { "message" => "\[%{TIMESTAMP_ISO8601:timestamp}\] %{GREEDYDATA:message}" }
  }
}

output {
  stdout { codec => rubydebug }
}

该示例先从"/etc/logstash/patterns"目录中加载自定义模式，以便可以在匹配中使用。然后，使用Grok匹配从原始文本中提取事件时间戳和消息内容。

六、Logstash配置文件

在Logstash中，配置文件是实现数据流转和处理的关键，下面的示例展示了一个典型的Logstash配置文件。

input {
  syslog {
    port => "5514"
    type => "syslog"
  }
}

filter {
  grok {
    match => { "message" => "\[%{MONTHDAY:day}/%{MONTH:month}/%{YEAR:year}:%{TIME:time} %{ISO8601_TIMEZONE} \]%{GREEDYDATA:message}" }
  }
}

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

该配置文件定义了三个部分：输入、过滤和输出。

输入部分使用syslog协议收集数据，监听端口号为"5514"，数据类型为"syslog"。

过滤部分使用Grok插件对日志文本进行提取，将日期时间和消息内容提取出来。

输出部分将数据输出到一个Elasticsearch实例中，存储在名为"logstash-%{+YYYY.MM.dd}"的索引中。

七、Logstash配置文件位置

Logstash配置文件通常存储在"/etc/logstash/conf.d/"目录中，下面的示例展示了如何在Logstash中指定配置文件。

logstash --path.config /etc/logstash/conf.d/

该示例指定Logstash配置文件存储在"/etc/logstash/conf.d/"目录中。

八、Logstash配置ES集群

Elasticsearch是一个强大的搜索和分析引擎，可以作为Logstash数据的终点。下面的示例演示了如何在Logstash中配置Elasticsearch集群。

output {
  elasticsearch {
    hosts => [ "es-host-01.example.com:9200", "es-host-02.example.com:9200", "es-host-03.example.com:9200" ]
    index => "logstash-%{+YYYY.MM.dd}"
  }
}

该示例将输出数据发送到Elasticsearch集群中，由三个不同的节点处理。数据被存储在名为"logstash-%{+YYYY.MM.dd}"的索引中。

九、Logstash Filter

Logstash Filter能够对原始数据进行处理，过滤出所需数据，对数据进行格式化和转换等操作。下面是一个Filter示例：

filter {
  if "syslog" in [tags] {
    grok {
      match => { "message" => "%{SYSLOG5424SD:syslog5424_sd} %{HOSTNAME:hostname} %{SYSLOG5424PROC:syslog5424_proc} %{SYSLOG5424MSG:syslog5424_msg}" }
    }
    date {
      match => [ "syslog5424_sd", "ISO8601" ]
    }
    mutate {
      replace => { "message" => "%{syslog5424_msg}" }
      remove_field => [ "syslog5424_msg", "syslog5424_sd", "syslog5424_proc" ]
    }
  }
}

该Filter通过匹配[tags]字段中包含"syslog"的日志，使用grok插件解析匹配到的日志，使用date插件将日志时间转换为标准时间格式，使用mutate插件进行字段替换和删除。