Using ElasticSearch with nAttrMon

nAttrMon captures monitoring inputs to be validated and output to where deem necessary. But you start having more than one instance or you are monitoring a lot of inputs you start having the need to build specific dashboards, visualizations, etc.
Additionally althougth nAttrMon lets you store input history, it's limited since it's not intended to be used for longer than a couple of days.
For all these challenges output and use ElasticSearch & Kibana provide you the answer:

• Elasticsearch let's you store more than just a couple of days of nAttrMon's input
• Kibana let's you visualize the data anyway you want from ElasticSearch.

How to set it up

So how to get nAttrMon to output to ElasticSearch? Simply create a new output (example in YAML):

output:
  name         : Output ES values
  chSubscribe  : nattrmon::cvals
  waitForFinish: true
  onlyOnEvent  : true
  execFrom     : nOutput_ES
  execArgs     : 
    url      : http://127.0.0.1:9200
    #user     : nouser
    #pass     : nopass
    #considerSetAll: false 
    #funcIndex: "ow.ch.utils.getElasticIndex('nattrmon-attrs',\"yyyy.ww\")" 
    ## note: Check getElasticIndex function, If a specific format is needed
    ## you can provide it as aFormat (see ow.format.fromDate)"
    funcIndex: "ow.ch.utils.getElasticIndex('nattrmon-attrs', 'yyyy.ww')"
    #index: nattrs
    stampMap : 
      environment: EnvA
      region     : EU
    #include  :
    #  - test/test 1
    # exclude  :

In most of the cases you only need to specify the ElasticSearch URL. But there are more parameters:

Exec arguments	Description
index	The ElasticSearch index where to store the current values (channel nattrmon::cvals).
funcIndex	In most of the cases you want to use a different ElasticSearch index on different times. You can specify a function for that.
considerSetAll	The nOutput_ES is based on channel subscription (chSubscribe) so it can potential receive a setAll operation. In case you use nAttrMon channel buffers you need to set this argument to true. Otherwise you are probably fine with false.
stamp	Since you might have several nAttrMon's instances dumping attribute values to the same ElasticSearch index you can "stamp" each output entries with specific entries (e.g. in the example environment and region)
include	You can specify that only some attributes' values will be sent to ElasticSearch.
exclude	Or you can specifiy that only some attributes' values shouldn't be sent to ElasticSearch.
user	The user credential if needed.
pass	The password credential if needed.

Since in most ElasticSearch configuration cases it's advisable to have different indexes per time you can provide nAttrMon with a function on funcIndex. Using the helper function ow.ch.utils.getElasticIndex:

ow.ch.utils.getElasticIndex('nattrmon-attrs', 'yyyy.ww')

How the output looks in ElasticSearch?

For each attribute value change a new entry will be created in ElasticSearch:

The key is id and it's a hash. Then you will have the name of the attribute and the date it was checked. The value will be in a key with the same name as the attribute. If the attribute has categories the '/' will be converted to a ''. In the example above: _Random_Number and Random_Dice. The stamp keys will also, of course, be part of the ElasticSearch entry.

And the warnings?

To add warnings just create a new similar output changing it's name, the ElasticSearch index where the warnings will be kept and replace on the chSubscribe to use nattrmon::warnings instead of nattrmon::cvals.

Example of an OpenAF channel accessing the ElasticSearch's nattrmon-warns index

The warnings instead of the name of the attribute have the title of the warning, level (e.g. High, Medium, Low, Info, Closed), the description, the date of the last update and date of creation. The stamp map will also be part of the ElasticSearch entry as expected.

Example of a Kibana dashboard using sample warnings from nAttrMon

Using ElasticSearch in OpenAF

OpenAF comes with builtin support for ElasticSearch. So, among other things, it can log directly to ElasticSearch whenever you use the log* functions. Nevertheless there is an oPack to make it all a little easier. Let's start by installing it:

$ opack install elasticsearch

The ElasticSearch oPack is basically a wrapper around some ElasticSearch functionality aiming to make easier daily ElasticSearch operations (e.g. creating/deleting indexes, export/import data, reindex indexes, etc...). We are going to describe some basic functionality. To start you need to instantiate an object poiting to your ElasticSearch cluster:

load("elasticsearch.js");
var es = new ElasticSearch("http://my.elastic.cluster:9200", "myUser", "myPassword");

Note: the user and password are optional and only needed if your ElasticSearch cluster is protected with user/password.

Creating an index

To create an index you just need to:

> es.createIndex("test", 1, 1);

This will create a new index "test" with 1 primary shard and 1 replica. This operation isn't usually necessary as ElasticSearch will just create any index you try to use.

You can check that it was created by executing:

> es.getIndices()

And checking the resulting array.

Adding/Changing data

To interact with ElasticSearch at the data level the easiest way in OpenAF is to use an OpenAF channel. To easily create an OpenAF channel to connect to the "test" index just execute:

es.createCh("test", ["canonicalPath"], "testCh");

This creates an OpenAF channel "testCh" that allows you to access the test index. You also need to provide a list of keys that can be used to retrieve an unique record (in this case "canonicalPath").

Note: You can also define a pattern of indexes instead of the exact name but you will be limited just to .get* functions. It's also possible instead of a string to provide a function that returns the name of the ElasticSearch index to use (e.g. for example, appending the current date).

To add new data just use the newly created channel:

$ch("testCh").set({
    canonicalPath: "/",
}, {
    isDirectory: yes,
    isFile: no,
    filename: "noname",
    filepath: "/",
    canonicalPath: "/"
    lastModified: 0,
    createTime: 0,
    size: 0
});

Retriving data

If you know how to use OpenAF channels now it becomes easier. To get a map value you just need:

var fileMap = $ch("test").get({ canonicalPath: "/" });

Batch get/set data

To batch insert/change data into ElasticSearch you need to divide all the requests in smaller chuncks of data. Then you just use the .setAll function:

$ch("test").setAll(["canonicalPath"], io.listFiles("/some/path").files);

To obtain a list of keys or values you can use getAll or getKeys:

var q = (m, s) => { return { size: s, query: { query_string: { query: m }}}; };
var listOfSmallFiles = $ch("test").getAll(q("size:<1 AND ", 1000));

Unlike the usual getAll and getKeys behaviour the elasticsearch OpenAF channel type will only retrieves a specific amount of records (default to 10). In this example we created a small function to allow you to query using Lucene query syntax and specifying the limit number of records you wish to retrieve (within the search API limits). Check the ElasticSearch search API for more.

Delete data

To delete data simply use the unset/unsetAll functions:

$ch("test").unset({ canonicalPath: "/" });