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插件進行字段替換和刪除。