上一篇中我们进行了logstash1.4.0的简单使用，和一些核心介绍。这篇我们来详细的进行实际运用。

配置文件

虽然使用-e 来配置很有用，但是多数我们都想使用配置文件来完成。使先我们来创建1个文件叫”logstash-simple.conf”,然后录入以下内容

input { stdin { } }
output {
  elasticsearch { host => localhost }
  stdout { codec => rubydebug }
}

然后运行命令

bin/logstash -f logstash-simple.conf

注 这个配置会依赖于上文配置运行的elasticsearch，请保证elasticsearch是运行着的

键入一点信息试一下

hello configuration
{
   "message" => "hello configuration",
  "@version" => "1",
"@timestamp" => "2014-04-03T05:49:43.430Z",
      "host" => "avalon"
}

访问()[]也可以得到以下信息

{
    _index: "logstash-2014.04.03",
    _type: "logs",
    _id: "1hfFPr7GSB61y5wgFGTEaQ",
    _score: 1,
    _source: {
    message: " "message" => "hello configuration",",
    @version: "1",
    @timestamp: "2014-04-03T05:49:49.257Z",
    host: "avalon"
    }
}

添加filter

filter是非常灵活的，它能够按照你的需求处理或者切分你的数据。来我们使用grok filter试一下。停止logstash，创建一个logstash-filter.conf文件录入以下配置

input { stdin { } }

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  date {
    match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

output {
  elasticsearch { host => localhost }
  stdout { codec => rubydebug }
}

然后运行以下命令

bin/logstash -f logstash-filter.conf

然后复制以下信息到控制台

127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] "GET /xampp/status.php HTTP/1.1" 200 3891 "http://cadenza/xampp/navi.php" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0"

你会看到以下输出

{
    "message" => "127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] \"GET /xampp/status.php HTTP/1.1\" 200 3891 \"http://cadenza/xampp/navi.php\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0\"",
   "@version" => "1",
 "@timestamp" => "2013-12-11T08:01:45.000Z",
       "host" => "avalon",
   "clientip" => "127.0.0.1",
      "ident" => "-",
       "auth" => "-",
  "timestamp" => "11/Dec/2013:00:01:45 -0800",
       "verb" => "GET",
    "request" => "/xampp/status.php",
"httpversion" => "1.1",
   "response" => "200",
      "bytes" => "3891",
   "referrer" => "\"http://cadenza/xampp/navi.php\"",
      "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0\""
}

看到没，logstash使用grok filter直接可以使用分析Apache日志(Apache正好是combined log格式)。这是非常有用的，当我们需要从apache中得到HTTP响应代码， IP地址，强求路径等等。grop中已经包含了很多正则，你看patterns下的grok-patterns文件或者看github 上的

还有注意到没有，我们也正对时间进行了格式化，格式了的时间是显示在@timestamp上。

采用文件读取数据(apache log)

多数情况下我们的日志文件都是file，那基于file读取时什么样的呢？下面我们基于apache log来练习一下，同时我们也会练习一下条件判断。创建一个配置文件logstash-apache.conf,然后录入以下信息

input {
  file {
    path => "/tmp/access_log"
    start_position => beginning
  }
}

filter {
  if [path] =~ "access" {
    mutate { replace => { "type" => "apache_access" } }
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
  }
  date {
    match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

output {
  elasticsearch {
    host => localhost
  }
  stdout { codec => rubydebug }
}

然后在“/tmp”下创建access_log录入

71.141.244.242 - kurt [18/May/2011:01:48:10 -0700] "GET /admin HTTP/1.1" 301 566 "-" "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.2.3) Gecko/20100401 Firefox/3.6.3"
134.39.72.245 - - [18/May/2011:12:40:18 -0700] "GET /favicon.ico HTTP/1.1" 200 1189 "-" "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; .NET CLR 2.0.50727; .NET CLR 3.0.4506.2152; .NET CLR 3.5.30729; InfoPath.2; .NET4.0C; .NET4.0E)"
98.83.179.51 - - [18/May/2011:19:35:08 -0700] "GET /css/main.css HTTP/1.1" 200 1837 "http://www.safesand.com/information.htm" "Mozilla/5.0 (Windows NT 6.0; WOW64; rv:2.0.1) Gecko/20100101 Firefox/4.0.1"

注 标准文档最后一行应该有回车，使用记得最后一行数据最后录入一个回车，否则会少一行数据。

好，运行

bin/logstash -f logstash-apache.conf

结果你看到了三条json数据输出。而且json中新增了一行数据“type” => “apache_access”,这个就是mutate的作用。简单说明一下type，如果你在input配置file的时候配置了type，type的值就是那个，mutate会吧结果换成舍得的，没有就添加上。

file的录入可以采用ant的方式匹配多个文件，比如

nput {
  file {
    path => "/tmp/*_log"
...

还有你注意了没有上面我们使用了if判断，判断就是path中有没有access。下面我们就来详细介绍一下条件判断

条件判断

条件判断支持以下条件

  • 判断是否相等:==, !=, <, >, <=, >=
  • 正则判断是否相等: =~, !~
  • 是否包含: in, not in

比如上面的例子，我们还有error_log还有sys_log都在同一位置而且我们的文件录入如下

input {
  file {
    path => "/tmp/*_log"
  }
}

这时候我们只需要简单的条件判断就可以，配置文件如下

input {
  file {
    path => "/tmp/*_log"
  }
}

filter {
  if [path] =~ "access" {
    mutate { replace => { type => "apache_access" } }
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
    date {
      match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
    }
  } else if [path] =~ "error" {
    mutate { replace => { type => "apache_error" } }
  } else {
    mutate { replace => { type => "random_logs" } }
  }
}

output {
  elasticsearch { host => localhost }
  stdout { codec => rubydebug }
}

参考文档