本文主要是介绍怎么编写Logstash grok,其他部分不在讨论范围.

Configure the Logstash outputwww.elastic.co

一般来说,我们从filebeat或者其他地方拿到下面的日志:

2018-04-13 16:03:49.822 INFO  o.n.p.j.c.XXXXX - Star Calculator

然后想把它其中的一些信息提取出来再扔到es中存储,我们就需要grok match把日志信息切分成索引数据(match本质是一个正则匹配)

grok match:

match => { "message" => "%{DATA:log_date} %{TIME:log_localtime} %{WORD:log_type} %{JAVAFILE:log_file} - %{GREEDYDATA:log_content}"}

切出来的数据

{
  "log_date": [
    [
      "2018-04-13"
    ]
  ],
  "log_localtime": [
    [
      "16:03:49.822"
    ]
  ],
  "HOUR": [
    [
      "16"
    ]
  ],
  "MINUTE": [
    [
      "03"
    ]
  ],
  "SECOND": [
    [
      "49.822"
    ]
  ],
  "log_type": [
    [
      "INFO"
    ]
  ],
  "log_file": [
    [
      "o.n.p.j.c.XXXX"
    ]
  ],
  "log_content": [
    [
      "Star Calculator"
    ]
  ]
}

上面所有切出来的field都是es中mapping index,都可以在用来做条件查询.

grokdebug.herokuapp.com里面可以做测试.

grokdebug.herokuapp.com/patterns 所有可用的patterns都可以在这里查到.

现在我们在用的配置见/logstash/logstash-k8s.conf

Q: 需要指定mapping index的数据类型怎么办?

A: grok match本质是一个正则匹配,默认出来的数据都是String.有些时候我们知道某个值其实是个数据类型,这时候可以直接指定数据类型. 不过match中仅支持直接转换成int ,float,语法是 %{NUMBER:response_time:int}
完整配置:

match => {
            "message" => "%{DATA:log_date} %{TIME:log_localtime} %{WORD:log_type}  %{JAVAFILE:log_file} - %{WORD:method} %{URIPATHPARAM:uri} %{NUMBER:status:int} %{NUMBER:size:int} %{NUMBER:response_time:int}"}

Q: 索引文件想需要按日期分别存放,怎么办?

A: out中指定index格式,如 index=> "k8s-%{+YYYY.MM.dd}"

完整out如下:

output {
    elasticsearch {
      hosts => "${ES_URL}"
      manage_template => false
      index => "k8s-%{+YYYY.MM.dd}"
      }
  }

完整logstash.conf

input {
    beats {
      host => "0.0.0.0"
      port => 5043
    }
  }
  filter {
    if [type] == "kube-logs" {
      mutate {
        rename => ["log", "message"]
      }
      date {
        match => ["time", "ISO8601"]
        remove_field => ["time"]
      }
      grok {
          match => {
            "source" => "/var/log/containers/%{DATA:pod_name}_%{DATA:namespace}_%{GREEDYDATA:container_name}-%{DATA:container_id}.log"}
          match => {
            "message" => "%{DATA:log_date} %{TIME:log_localtime} %{WORD:log_type}  %{JAVAFILE:log_file} - %{WORD:method} %{URIPATHPARAM:uri} %{NUMBER:status:int} %{NUMBER:size:int} %{NUMBER:response_time:int}"}
          remove_field => ["source"]
          break_on_match => false
      }
    }
  }
  output {
    elasticsearch {
      hosts => "${ES_URL}"
      manage_template => false
      index => "k8s-%{+YYYY.MM.dd}"
      }
  }

ITPub博客

logstash grok配置规则

Q: 需要指定mapping index的数据类型怎么办?

Q: 索引文件想需要按日期分别存放,怎么办?

最新文章