mozilla-services/heka — pull request #258 by tlrx (2013-06-14 16:44:11+00:00)
Title: Add ElasticSearchOutput support to Heka

Hi all,

This pull request adds an ElasticSearch Output plugin to Heka. It allows to index messages directly into an [ElasticSearch](http://www.elasticsearch.org) cluster and to use a tool like [Kibana3](http://three.kibana.org/) to analyse them as quoted in issue #253.

This plugin is largely based on the FileOutput plugin.

The configuration is pretty simple:
```
[ElasticSearchOutput]
type = "ElasticSearchOutput"
message_matcher = "TRUE"
```
This configuration sample will index raw Heka messages into an elasticsearch instance that is accessible on "http://localhost:9200".

The messages will be indexed (with the [Bulk Index API] (http://www.elasticsearch.org/guide/reference/api/bulk/)) in an index named **"heka-YYYY.MM.DD"** (where *YYYY.MM.DD* stands for the current day) and with a document type named **"message"**.

For instance, a raw Heka message looks like:
```json
{
"uuid":"npmKbRZIQhaaGubKbfd0Ng==",
"timestamp":1371227304691742085,
"type":"hekabench",
"logger":"flood",
"severity":0,
"env_version":"0.2",
"pid":11322,
"hostname":"portable",
"fields":[...]
...
}
```

It's possible to override default settings, if one need to change the index name, the document type or the elasticsearch server URL:
```
[ElasticSearchOutput]
type = "ElasticSearchOutput"
message_matcher = "TRUE"
cluster = "elasticsearch-cluster"
index = "heka-%{2006.01.02.15.04.05}"
typeName = "message"
server = "http://es-server:9200"
format = "raw"
```

There's also a "clean" message format, that replaces the default JSON formating into something more ElasticSearch friendly. It also allows to include/exclude message fields:

```
[ElasticSearchOutput]
type = "ElasticSearchOutput"
message_matcher = "TRUE"
format = "clean"
fields = ["uuid", "logger", "type", "timestamp", "severity"]
```

The JSON document indexed in elasticsearch will look like:
```json
{
"uuid":"0cb34c90-379d-4ba7-b973-582ee8157199",
"logger":"flood",
"type":"hekabench",
"timestamp":"2013-06-14T18:34:22.095Z",
"severity":0
}
```

The messages are buffered and indexed every 5 seconds. This setting can be changed in configuration with *flushInterval*.

This pull request still need some tests, I'll try to attach them as soon as possible.

There are also some ideas for improvements:
* Use round robin over multiple elasticsearch server URLs
* Excludes unavailable server URLs
* Map index or type names to message fields or headers
* Flush bulk every 5 seconds or *n* items

Hope it will be as useful as it is for me :)