Camel-Elasticsearch: create timestamped indices
One nice feature of the logstash-elasticsearch integration is that, by default, logstash will use timestamped indices when feeding data to elasticsearch.
This means that yesterday's data is in a separate index from today's data and from each other day's data, simplifying index management. For instance, suppose you only want to keep the last 30 days:
elasticsearch-remove-old-indices.sh -i 30
The Apache Camel Elasticsearch component provides no such feature out of the box, but luckily it is quite easy to implement (when you know what to do. /grin ).
As a matter of fact, it is enough to define the proper header on the message and the elasticsearch component will then use that header as the index name. Unfortunately this is not documented anywhere, but it can be understood by looking at the source. Once again: use the source, Luke.
So, let's suppose the route is something as simple as:
Then all that is needed to is to define a content enricher bean as follows:
The bean is also pretty simple (imports omitted for brevity):
Update: get timestamp index name from the message itself.
If the data to be indexed contains, as it should, a @timestamp field then the content enricher bean can be imrproved to use it as follows:
This means that yesterday's data is in a separate index from today's data and from each other day's data, simplifying index management. For instance, suppose you only want to keep the last 30 days:
elasticsearch-remove-old-indices.sh -i 30
The Apache Camel Elasticsearch component provides no such feature out of the box, but luckily it is quite easy to implement (when you know what to do. /grin ).
As a matter of fact, it is enough to define the proper header on the message and the elasticsearch component will then use that header as the index name. Unfortunately this is not documented anywhere, but it can be understood by looking at the source. Once again: use the source, Luke.
So, let's suppose the route is something as simple as:
<route autostartup="true" id="processMirthMessages-route">
<from uri="sql:{{sql.selectMessage}}?consumer.delay=5000&consumer.onConsume={{sql.markMessage}}">
<to uri="elasticsearch://mirth?operation=INDEX&indexType=mmsg">
</to></from></route>
Then all that is needed to is to define a content enricher bean as follows:
<route autostartup="true" id="processMirthMessages-route">
<from uri="sql:{{sql.selectMessage}}?consumer.delay=5000&consumer.onConsume={{sql.markMessage}}">
<bean method="process" ref="eSheaders">
<to uri="elasticsearch://mirth?operation=INDEX&indexType=mmsg">
</to></bean></from></route>
The bean is also pretty simple (imports omitted for brevity):
public class ESHeaders {
public void process(Exchange exchange) {
Message in = exchange.getIn();
DateFormat df=new SimpleDateFormat("YYYY.MM.dd");
in.setHeader(ElasticsearchConfiguration.PARAM_INDEX_NAME, "mirth2-"+df.format(new Date()));
}
}
Update: get timestamp index name from the message itself.
If the data to be indexed contains, as it should, a @timestamp field then the content enricher bean can be imrproved to use it as follows:
public void process(Exchange exchange) {
Message in = exchange.getIn();
String indexName=null;
DateFormat df=new SimpleDateFormat("YYYY.MM.dd");
try {
Map body = (Map) in.getBody();
if(body.containsKey("@timestamp")) {
logger.trace("Computing indexName from @timestamp: "+body.get("@timestamp"));
indexName = "mirth2-"+df.format((Date) body.get("@timestamp")));
} else {
indexName = "mirth2-"+df.format(new Date()));
}
} catch(Exception e) {
logger.error("Cannot compute index name, failing back to default");
indexName = "mirth2-"+df.format(new Date()));
}
in.setHeader(ElasticsearchConfiguration.PARAM_INDEX_NAME, indexName);
}