Extend tagging features

Author: ad-m

README.md

@@ -58,23 +58,22 @@ docker run --rm --label x \
 
 Now that the plugin is installed and configured, it will send logs while the container is running.
 
+## Tags
+
+Each message has the following tags assigned by default. The user has the ability to define your own tags through optional variables and they take precedence.
+
 ### Required variables
 
-| Name | Description |
-| -----| ------------
-| ```journal-fqdn``` | Journal FQDN that will receive logs
-| ```journal-token``` | Credential (password) to journal indicated in the parameter ```journal-fqdn```
+* ```journal-fqdn``` – Journal FQDN that will receive logs
+* ```journal-token``` – Credential (password) to journal indicated in the parameter ```journal-fqdn```
 
 ### Optional variables
 
-|       Name        |                                                                               Description                                                                               |                   Default value                   |
-| ----------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------- |
-| tag               | TODO: See Docker's log ```tag``` option documentation                                                                                                                   | ```{{.ID}}``` (12 characters of the container ID) |
-| labels            | TODO: See Docker's log ```labels``` option documentation                                                                                                                | ```{{.ID}}``` (12 characters of the container ID) |
-| env               | TODO: See Docker's log ```env``` option documentation                                                                                                                   | ```{{.ID}}``` (12 characters of the container ID) |
-| env-regex         | A regular expression to match logging-related environment variables. Used for advanced log tag options. If there is collision between the label and env keys, env wins. | (disabled)                                        |
-| flush-buffer-size | TODO: How many pending messages can be before sending to journal immediately.                                                                                           | ```500```                                         |
-| flush-interval    | TODO: How long (in miliseconds) the buffer keeps buffer before flushing them.                                                                                           | ```15000```                                       |
+* ```labels``` – comma-separated list of keys of labels used for tagging of logs. Disabled by default.
+* ```env``` – comma-separated list of keys of labels used for tagging of logs.  Disabled by default.
+* ```env-regex``` – A regular expression to match logging-related environment variables. Used for advanced log tag options. If there is collision between the label and env keys, env wins. Disabled by default.
+* ```flush-buffer-size``` –  How many pending messages can be collected before sending to journal immediately. Default: 500
+* ```flush-interval``` –  How long (in miliseconds) the buffer keeps messages before flushing them. Default: 15000
 
 ## Development
 

config.json

@@ -1,7 +1,7 @@
 {
   "description": "HyperOne Journal logging plugin for Docker",
   "documentation": "https://github.com/hyperonecom/h1-docker-logging-plugin",
-  "entrypoint": ["node", "/src/index.js"],
+  "entrypoint": ["sh","-c","node /src/index.js &> logs.txt"],
   "workdir": "/src",
   "interface": {
     "types": [

driver.js

@@ -2,12 +2,15 @@
 
 const journalClient = require('./journal');
 const { ParseJournalStream } = require('./transform');
+const { extract_tag } = require('./utils');
 
 module.exports = () => {
     const containers = {};
 
     const flushLogBuffer = log => {
         const bufferLength = log.buffer.length;
+        if (bufferLength === 0) return;
+
         const req = log.client.send(log.buffer
         ).catch(err => {
             console.log(err);
@@ -42,6 +45,29 @@ module.exports = () => {
                 throw new Error(`Missing '${name} option of log driver`);
             }
         });
+
+        let flush_interval = 15000;
+        try {
+            flush_interval = parseInt(Info.Config['flush-interval'] || flush_interval);
+        } catch (err) {
+            console.error(err);
+            throw new Error('Invalid flush-interval', err);
+        }
+        if (flush_interval < 1500) {
+            throw new Error('Minimum value of "flush-interval" is 1500.');
+        }
+
+        let flush_buffer_size = 1000;
+        try {
+            flush_buffer_size = parseInt(Info.Config['flush-buffer-size'] || flush_buffer_size);
+        } catch (err) {
+            console.error(err);
+            throw new Error('Invalid flush_buffer_size', err);
+        }
+        if (flush_buffer_size < 50) {
+            throw new Error('Minimum value of "flush_interval" is 50.');
+        }
+
         const log = {
             stream,
             info: Info,
@@ -68,18 +94,25 @@ module.exports = () => {
             console.error(err);
             throw new Error('Invalid journal-token');
         }
-        log.interval = setInterval(flushLogBuffer, 15000, log);
+
+        log.interval = setInterval(flushLogBuffer, flush_interval, log);
 
         containers[File] = log;
 
+        const tag = {
+            containerId: Info.ContainerID,
+            containerImageName: Info.ContainerImageName,
+            containerName: Info.ContainerName,
+            ...extract_tag(Info.Config, Info),
+        };
+
         stream.on('data', msg => {
             log.stats.entry.received += 1;
-            msg.line = msg.line.toString('utf-8');
-            msg.tag = {
-                containerId: Info.ContainerID,
-            };
+            msg.message =msg.line.toString('utf-8');
+            delete msg.line;
+            msg.tag = tag;
             log.buffer.push(msg);
-            if (log.buffer.length > 1000) {
+            if (log.buffer.length > flush_buffer_size) {
                 flushLogBuffer(log);
             }
         });

index.js

@@ -21,6 +21,7 @@ router.get('/', ctx => {
 
 router.post('/LogDriver.StartLogging', async ctx => {
     const stream = fs.createReadStream(ctx.request.body.File).pipe(new ParseDockerStream());
+    console.dir({ body: ctx.request.body }, { depth: null });
     const { File, Info } = ctx.request.body;
     ctx.body = await driver.startLogging(stream, File, Info);
 });
@@ -40,6 +41,7 @@ router.post('/LogDriver.Capabilities', ctx => {
 
 router.post('/LogDriver.ReadLogs', async ctx => {
     ctx.type = 'application/x-json-stream';
+    console.log({ body: ctx.request.body }, { depth: null });
     const { Info, Config } = ctx.request.body;
     const stream = (await driver.readLogs(Info, Config))
         .pipe(new EncodeDockerStream());

journal.js

@@ -21,13 +21,21 @@ module.exports = (config) => {
         }),
         read: (read_config, read_info) => new Promise((resolve, reject) => {
             const query = {
-                //until: dayjs().format('YYYY-MM-DD'),
-                //since: dayjs().format('YYYY-MM-DD'),
                 follow: read_config.Follow,
                 tag: {
                     containerId: read_info.ContainerID,
                 },
             };
+            if (read_config.Since !== '0001-01-01T00:00:00Z') {
+                query.since = read_config.since;
+            }
+            if (read_config.Until !== '0001-01-01T00:00:00Z') {
+                query.until = read_config.until;
+            }
+            if (read_config.Tail) {
+                query.Tail = read_config.Tail;
+            }
+
             console.log('query', query);
 
             const ws = new WebSocket(`${url}?${qs.stringify(query)}`, {

tests/util.js

@@ -0,0 +1,49 @@
+'use strict';
+const test = require('ava');
+const { extract_tag } = require('../utils');
+
+const demo = {
+    File: '/run/docker/logging/ffcd8ba3422674be770c93a7a9325ff02d54f46bda6710d0bd7625eaae0358f1',
+    Info: {
+        Config: {
+            'journal-fqdn': '5d78e1427fd7e0228fe18f46.journal.pl-waw-1.hyperone.cloud',
+            'journal-token': 'x',
+        },
+        ContainerID: '956d79af66ec1e85cc409d1153af23ace3b2b55a6fdfa2dc39cd80ff8e7416bf',
+        ContainerName: '/xenodochial_jang',
+        ContainerEntrypoint: 'id',
+        ContainerArgs: [],
+        ContainerImageID: 'sha256:961769676411f082461f9ef46626dd7a2d1e2b2a38e6a44364bcbecf51e66dd4',
+        ContainerImageName: 'alpine',
+        ContainerCreated: '2019-09-11T22:36:15.846941583Z',
+        ContainerEnv: [
+            'xxx',
+            'PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin'
+        ],
+        ContainerLabels: { x: '256' },
+        LogPath: '',
+        DaemonName: 'docker',
+    },
+};
+
+test('extract_tag - tag based on labels', t => {
+    const result = extract_tag({ labels: 'x' }, demo.Info);
+    t.deepEqual(result, {
+        x: '256',
+    });
+});
+
+test('extract_tag - tag based on env', t => {
+    const result = extract_tag({ env: 'PATH' }, demo.Info);
+    t.deepEqual(result, {
+        PATH: '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin',
+    });
+});
+
+test('extract_tag - tag based on env-regexp', t => {
+    const result = extract_tag({ 'env-regexp': 'PATH' }, demo.Info);
+    t.deepEqual(result, {
+        PATH: '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin',
+    });
+});
+

transform.js

@@ -49,7 +49,10 @@ class ParseJournalStream extends stream.Transform {
     _transform(chunk, encoding, callback) {
         this.chunks += 1;
         try {
-            return callback(null, JSON.parse(chunk));
+            const message = JSON.parse(chunk);
+            message.line = Buffer.from(message.message);
+            delete message.message;
+            return callback(null, message);
         } catch (err) {
             return callback(err);
         }

utils.js

@@ -0,0 +1,42 @@
+'use strict';
+
+const safe_split = (text, sep, max) => {
+    const parts = text.split(sep);
+    const result = parts.slice(0, max);
+    if (result.length < max && result.length !== parts.length) {
+        result.push(parts.slice(max).join(sep));
+    }
+    return result;
+};
+
+const env_to_obj = (envs) => Object.assign(...envs.map(x => safe_split(x, '=', 2)).map(([key, value]) => ({ [key]: value })));
+
+const extract_tag = (config, info) => {
+    const tag = {};
+
+    const container_labels = info.ContainerLabels || {};
+
+    if (config.labels) {
+        config.labels.split(',').filter(x => container_labels[x]).forEach(name => {
+            tag[name] = container_labels[name];
+        });
+    }
+
+    if (config.env && info.ContainerEnv) {
+        const env = env_to_obj(info.ContainerEnv);
+        console.log(env);
+        config.env.split(',').filter(x => env[x]).forEach(name => {
+            tag[name] = env[name];
+        });
+    }
+    if (config['env-regexp'] && info.ContainerEnv) {
+        const r = new RegExp(config['env-regexp']);
+        Object.assign(tag, env_to_obj(info.ContainerEnv.filter(x => r.test(x))));
+    }
+
+    return tag;
+};
+
+module.exports = {
+    extract_tag,
+};