embulkでwowzaのログをelasticsearchに流し込む為のconfigメモ


wowzaのログコンフィグは、[デフォルト](https://github.com/awead/WowzaMediaServer/blob/master/conf/log4j.properties)の場合の想定です。

```bash:log4j.properties
# Access appender
log4j.appender.serverAccess.layout.Fields=date,time,tz,x-event,x-category,x-severity,x-status,x-ctx,x-comment,x-vhost,x-app,x-appinst,x-duration,s-ip,s-port,s-uri,c-ip,c-proto,c-referrer,c-user-agent,c-client-id,cs-bytes,sc-bytes,x-stream-id,x-spos,cs-stream-bytes,sc-stream-bytes,x-sname,x-sname-query,x-file-name,x-file-ext,x-file-size,x-file-length,x-suri,x-suri-stem,x-suri-query,cs-uri-stem,cs-uri-query

# Error appender
log4j.appender.serverError.layout.Fields=x-severity,x-category,x-event;date,time,c-client-id,c-ip,c-port,cs-bytes,sc-bytes,x-duration,x-sname,x-stream-id,x-spos,sc-stream-bytes,cs-stream-bytes,x-file-size,x-file-length,x-ctx,x-comment
```

config例
====

embulkで流し込むコンフィグは下記の通り。
キモは、日時フィールドの作成部分です。
wowzaのログは、日付と時刻を別のフィールドに出力するため、
それを結合する必要があります。

そのために

* [embulk-filter-column](https://github.com/sonots/embulk-filter-column)
* [embulk-filter-ruby_proc](https://github.com/joker1007/embulk-filter-ruby_proc)

を利用します。
※プラグインのインストールは下記となります。

```
 embulk gem install embulk-filter-column embulk-filter-ruby_proc
```

#### アクセスログ

```yaml:embulk-config.yml
in:
  type: file
  path_prefix: /path/to/wowza-access-log.txt
  parser:
    charset: UTF-8
    newline: LF
    type: csv
    delimiter: "\t"
    quote: '"'
    escape: '"'
    trim_if_not_quoted: false
    skip_header_lines: 0
    allow_extra_columns: false
    allow_optional_columns: false
    columns:
    - {name: date        , type: string}
    - {name: time        , type: string}
    - {name: tz          , type: string}
    - {name: x-event     , type: string}
    - {name: x-category  , type: string}
    - {name: x-severity  , type: string}
    - {name: x-status    , type: long}
    - {name: x-ctx       , type: string}
    - {name: x-comment   , type: string}
    - {name: x-vhost     , type: string}
    - {name: x-app       , type: string}
    - {name: x-appinst   , type: string}
    - {name: x-duration  , type: string}
    - {name: s-ip        , type: string}
    - {name: s-port      , type: string}
    - {name: s-uri       , type: string}
    - {name: c-ip        , type: string}
    - {name: c-proto     , type: string}
    - {name: c-referer   , type: string}
    - {name: c-user-agent, type: string}
    - {name: client-id   , type: string}
    - {name: cs-bytes    , type: string}
    - {name: sc-bytes    , type: string}
    - {name: x-stream-id , type: string}
    - {name: x-spos          , type: string}
    - {name: cs-stream-bytes , type: string}
    - {name: sc-stream-bytes , type: string}
    - {name: x-sname         , type: string}
    - {name: x-sname-query   , type: string}
    - {name: x-file-name     , type: string}
    - {name: x-file-ext      , type: string}
    - {name: x-file-size     , type: string}
    - {name: x-file-length   , type: string}
    - {name: x-suri          , type: string}
    - {name: x-suri-stem     , type: string}
    - {name: x-suri-queri    , type: string}
    - {name: cs-uri-stem     , type: string}
    - {name: cs-uri-query    , type: string}

filters:
  - type: column
    add_columns:
    - {name : datetime, type: timestamp, default: "1970-01-01", format: "%Y-%m-%d"}
  - type: ruby_proc
    before:
      - proc: |
          -> do
            require "time"
          end
    rows:
      - proc: |
          ->(record) do
            record.tap do |r|
              r["datetime"] = Time.iso8601(r["date"] + "T" + r["time"] + "+09:00")
            end
          end

  - type: column
    drop_columns:
    - {name : date}
    - {name : time}

out:
  type: elasticsearch_ruby
  mode: normal
  nodes:
  - {host: localhost, port: 9200}
  index: logstash-wowza
  index_type: wow
```


#### エラーログ

```yaml:embulk-config-error.yml
in:
  type: file
  path_prefix: /path/to/wowza-error-log.txt
  parser:
    charset: UTF-8
    newline: LF
    type: csv
    delimiter: "\t"
    quote: '"'
    escape: '"'
    trim_if_not_quoted: false
    skip_header_lines: 0
    allow_extra_columns: false
    allow_optional_columns: false
    columns:
    - {name:  x-severity      , type: string}
    - {name:  x-category      , type: string}
    - {name:  x-event         , type: string}
    - {name:  date            , type: string}
    - {name:  time            , type: string}
    - {name:  c-client-id     , type: string}
    - {name:  c-ip            , type: string}
    - {name:  c-port          , type: string}
    - {name:  cs-bytes        , type: string}
    - {name:  sc-bytes        , type: string}
    - {name:  x-duration      , type: string}
    - {name:  x-sname         , type: string}
    - {name:  x-stream-id     , type: string}
    - {name:  x-spos          , type: string}
    - {name:  sc-stream-bytes , type: string}
    - {name:  cs-stream-bytes , type: string}
    - {name:  x-file-size     , type: string}
    - {name:  x-file-length   , type: string}
    - {name:  x-ctx           , type: string}
    - {name:  x-comment       , type: string}

filters:
  - type: column
    add_columns:
    - {name : datetime, type: timestamp, default: "1970-01-01", format: "%Y-%m-%d"}
  - type: ruby_proc
    before:
      - proc: |
          -> do
            require "time"
          end
    rows:
      - proc: |
          ->(record) do
            record.tap do |r|
              r["datetime"] = Time.iso8601(r["date"] + "T" + r["time"] + "+09:00")
            end
          end

  - type: column
    drop_columns:
    - {name : date}
    - {name : time}

out:
  type: elasticsearch_ruby
  mode: normal
  nodes:
  - {host: localhost, port: 9200}
  index: logstash-wowza-error-test
  index_type: wow
```