Merge pull request #22 from tdeo/master

Allow for Date in index

README.md

@@ -27,6 +27,7 @@ npm install pino-elasticsearch -g
   -H  | --host              the IP address of elasticsearch; default: 127.0.0.1
   -p  | --port              the port of elasticsearch; default: 9200
   -i  | --index             the name of the index to use; default: pino
+                            will replace %{DATE} with the YYYY-MM-DD date
   -t  | --type              the name of the type to use; default: log
   -b  | --size              the number of documents for each bulk insert
   -l  | --trace-level       trace level for the elasticsearch client, default 'error' (info, debug, trace).

pino-elasticsearch.js

@@ -58,7 +58,12 @@ function pinoElasticSearch (opts) {
       for (var i = 0; i < docs.length; i++) {
         if (i % 2 === 0) {
           // add the header
-          docs[i] = { index: { _index: index, _type: type } }
+          docs[i] = {
+            index: {
+              _type: type,
+              _index: index.replace('%{DATE}', chunks[Math.floor(i / 2)].chunk.time.substring(0, 10))
+            }
+          }
         } else {
           // add the chunk
           docs[i] = chunks[Math.floor(i / 2)].chunk
@@ -83,7 +88,8 @@ function pinoElasticSearch (opts) {
       })
     },
     write: function (body, enc, cb) {
-      const obj = { index, type, body }
+      var idx = index.replace('%{DATE}', body.time.substring(0, 10))
+      const obj = { index: idx, type, body }
       client.index(obj, function (err, data) {
         if (!err) {
           splitter.emit('insert', data, body)

test/acceptance.test.js

@@ -111,3 +111,54 @@ test('store lines in bulk', { timeout }, (t) => {
     }, refreshInterval)
   })
 })
+
+test('replaces date in index', { timeout }, (t) => {
+  t.plan(3)
+  const index = 'pinotest-%{DATE}'
+
+  const instance = elastic({ index, type, consistency, host, port })
+  const log = pino(instance)
+
+  log.info('hello world')
+
+  instance.on('insert', (obj, body) => {
+    t.ok('data uploaded')
+
+    client.get({
+      index: index.replace('%{DATE}', new Date().toISOString().substring(0, 10)),
+      type,
+      id: obj._id
+    }, (err, response) => {
+      t.error(err)
+      t.deepEqual(response._source, body, 'obj matches')
+    })
+  })
+})
+
+test('replaces date in index during bulk insert', { timeout }, (t) => {
+  t.plan(15)
+
+  const index = 'pinotest-%{DATE}'
+  const instance = elastic({ index, type, consistency, host, port })
+  const log = pino(instance)
+
+  log.info('hello world')
+  log.info('hello world')
+  log.info('hello world')
+  log.info('hello world')
+  log.info('hello world')
+
+  instance.on('insert', (obj, body) => {
+    t.ok(obj, 'data uploaded')
+    setTimeout(function () {
+      client.get({
+        index: index.replace('%{DATE}', new Date().toISOString().substring(0, 10)),
+        type,
+        id: obj._id
+      }, (err, response) => {
+        t.error(err)
+        t.deepEqual(response._source, body, 'obj matches')
+      })
+    }, refreshInterval)
+  })
+})

usage.txt

@@ -11,8 +11,8 @@
   -H  | --host              the IP address of elasticsearch; default: 127.0.0.1
   -p  | --port              the port of elasticsearch; default: 9200
   -i  | --index             the name of the index to use; default: pino
+                            will replace %{DATE} with the YYYY-MM-DD date
   -t  | --type              the name of the type to use; default: log
-  -c  | --consistency       the consistency of the write; default: one
   -b  | --size              the number of documents for each bulk insert
   -l  | --trace-level       trace level for the elasticsearch client, default 'error' (info, debug, trace).
   -c  | --aws-credentials   path to aws_config.json (is using AWS Elasticsearch)