mozilla-services/heka — pull request #283 by tlrx (2013-06-25 12:57:31+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}"
type_name = "message"
server = "http://es-server:9200"
format = "raw"
flush_interval = 5000
flush_count = 10
```

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": "a585f052-b312-44f6-ba94-18216fad54e8",
"Timestamp": "2013-06-25T14:51:18.354Z",
"Type": "hekabench",
"Logger": "flood",
"Severity": 0,
"EnvVersion": 0.2,
"Pid": 3070,
"Hostname": "laptop",
...
}
```

The messages are buffered and indexed every 1 second. This setting can be changed in configuration with *flush_interval*.

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~~
* ~~Use [Bulk UDP](http://www.elasticsearch.org/guide/reference/api/bulk-udp/) feature~~

Thanks to @rafrombrc for his help and tips!