Fix e2e tests for journal

ad-m · ad-m · commit e8a9bd9a57b0 · 2019-09-17T16:58:57.000+02:00
diff --git a/driver.js b/driver.js
@@ -89,6 +89,7 @@ module.exports = () => {
             },
         };
         try {
+            console.log('Checking journal token');
             await log.client.checkJournalToken();
         } catch (err) {
             console.error(err);
@@ -108,7 +109,7 @@ module.exports = () => {
 
         stream.on('data', msg => {
             log.stats.entry.received += 1;
-            msg.message =msg.line.toString('utf-8');
+            msg.message = msg.line.toString('utf-8');
             delete msg.line;
             msg.tag = tag;
             log.buffer.push(msg);
@@ -169,7 +170,7 @@ module.exports = () => {
         // }
         const client = journalClient(Info.Config);
         const stream = await client.read(Config, Info);
-        return stream.pipe(new ParseJournalStream());
+        return stream;
     };
 
     return driver;
diff --git a/journal.js b/journal.js
@@ -2,12 +2,15 @@
 const qs = require('qs');
 const logger = require('superagent-logger');
 const WebSocket = require('ws');
+const { FilterJournalDockerStream, ParseJournalStream } = require('./transform');
 
 module.exports = (config) => {
-    const url = `http://${config['journal-fqdn']}/log`;
+    const url = `https://${config['journal-fqdn']}/log`;
     const agent = require('superagent').agent().use(logger);
     return {
-        checkJournalToken: () => agent.head(url).query({ follow: false })
+        checkJournalToken: () => agent
+            .head(url)
+            .query({ follow: 'false' })
             .set('x-auth-password', config['journal-token']),
         send: (messages) => new Promise((resolve, reject) => {
             const body = Array.isArray(messages) ? messages : [messages];
@@ -37,14 +40,18 @@ module.exports = (config) => {
             }
 
             console.log('query', query);
-
-            const ws = new WebSocket(`${url}?${qs.stringify(query)}`, {
+            const ws_url = `${url}?${qs.stringify(query)}`;
+            console.log('WS', ws_url);
+            const ws = new WebSocket(ws_url, {
                 headers: { 'x-auth-password': config['journal-token'] },
             });
 
             ws.on('open', () => {
                 console.log(config['journal-fqdn'], 'websocket opened');
-                const stream = WebSocket.createWebSocketStream(ws);
+                const stream = WebSocket.createWebSocketStream(ws).
+                    pipe(new ParseJournalStream()).
+                    pipe(new FilterJournalDockerStream());
+                stream.pause();
                 resolve(stream);
             });
 
diff --git a/tests/driver.js b/tests/driver.js
@@ -38,12 +38,12 @@ class LogGenerator extends Readable {
     }
 }
 
-const defaultInfo = {
+const defaultInfo = () => ({
     Config: {
         'journal-fqdn': `${process.env.JOURNAL_ID}.journal.pl-waw-1.hyperone.cloud`,
         'journal-token': process.env.JOURNAL_TOKEN,
     },
-    ContainerID: 'eddeb66fc259',
+    ContainerID: getRandom(),
     ContainerName: '/confident_carson',
     ContainerEntrypoint: 'sh',
     ContainerArgs: [],
@@ -54,12 +54,12 @@ const defaultInfo = {
     ContainerLabels: [],
     LogPath: '',
     DaemonName: 'docker',
-};
+});
 
 test.serial('driver.startLogging with credentials starts', hasCredentialEnv(async t => {
     const d = driver();
     const stream = new LogGenerator();
-    const resp = await d.startLogging(stream, '/tmp/file.sock', defaultInfo);
+    const resp = await d.startLogging(stream, '/tmp/file.sock', defaultInfo());
     t.true(!!resp);
     stream.destroy();
 }));
@@ -68,7 +68,7 @@ test.serial('driver.stopLogging', hasCredentialEnv(async t => {
     const d = driver();
     const stream = new LogGenerator();
     const file = '/tmp/file.sock';
-    await d.startLogging(stream, file, defaultInfo);
+    await d.startLogging(stream, file, defaultInfo());
     const resp = d.stopLogging(file);
     stream.destroy();
     t.true(!!resp);
@@ -78,14 +78,14 @@ test.serial('driver.stopLogging without credentials raise error', async t => {
     const d = driver();
     const stream = new LogGenerator();
     const file = '/tmp/file.sock';
-    await t.throwsAsync(() => d.startLogging(stream, file, { ...defaultInfo, Config: {} }));
+    await t.throwsAsync(() => d.startLogging(stream, file, { ...defaultInfo(), Config: {} }));
 });
 
 test.serial('driver.startLogging consume logs', hasCredentialEnv(async t => {
     const d = driver();
     const stream = new LogGenerator();
     const file = '/tmp/file.sock';
-    await d.startLogging(stream, file, defaultInfo);
+    await d.startLogging(stream, file, defaultInfo());
     await d.stopLogging(file);
     t.true(stream._index > 100);
 }));
@@ -95,19 +95,21 @@ test.serial('driver.startLogging consume & reads logs from journal', hasCredenti
     const token = getRandom();
     const instream = new Readable.from([
         {
+            time_nano: +new Date(),
             source: 'stdout',
-            line: token,
+            line: Buffer.from(token),
         },
     ]);
     const file = '/tmp/file.sock';
-    await d.startLogging(instream, file, defaultInfo);
+    const info = defaultInfo();
+    await d.startLogging(instream, file, info);
     await d.stopLogging(file);
-    const outstream = await d.readLogs(defaultInfo, {
+    const outstream = await d.readLogs(info, {
         Follow: false,
     });
     let found = false;
     outstream.on('data', msg => {
-        found = found || msg.line === token;
+        found = found || msg.line.toString() === token;
     });
     await new Promise(resolve => {
         outstream.on('end', resolve);
diff --git a/tests/e2e/e2e.bats b/tests/e2e/e2e.bats
@@ -16,7 +16,8 @@ teardown() {
 @test "plugin send logs" {
     run docker run \
 	--log-driver 'h1cr.io/h1-docker-logging-plugin:latest' \
-    --label dockerbats="$BATS_TEST_NAME" \
+  --label dockerbats="$BATS_TEST_NAME" \
+  --log-opt labels=dockerbats \
 	--log-opt journal-fqdn=${JOURNAL_ID}.journal.pl-waw-1.hyperone.cloud \
 	--log-opt journal-token=${JOURNAL_TOKEN} \
 	alpine sh -c 'echo $RANDOM';
@@ -30,7 +31,8 @@ teardown() {
 @test "plugin flush logs" {
     run docker run -d \
 	--log-driver 'h1cr.io/h1-docker-logging-plugin:latest' \
-    --label dockerbats="$BATS_TEST_NAME" \
+  --label dockerbats="$BATS_TEST_NAME" \
+  --log-opt labels=dockerbats \
 	--log-opt journal-fqdn=${JOURNAL_ID}.journal.pl-waw-1.hyperone.cloud \
 	--log-opt journal-token=${JOURNAL_TOKEN} \
 	alpine sh -c 'seq 1 10; sleep 30';
@@ -47,13 +49,16 @@ teardown() {
     token=${RANDOM};
     run docker run \
 	--log-driver 'h1cr.io/h1-docker-logging-plugin:latest' \
-    --label dockerbats="$BATS_TEST_NAME" \
+    --label dockerbats="$BATS_TEST_NAME-${token}" \
+    --log-opt labels=dockerbats \
 	--log-opt journal-fqdn=${JOURNAL_ID}.journal.pl-waw-1.hyperone.cloud \
 	--log-opt journal-token=${JOURNAL_TOKEN} \
 	alpine sh -c "seq 100 | while read line; do echo \"multiple-\${line}-${token}\"; done;";
     [ "$status" -eq 0 ]
-    containerId=$(docker container ls -a -q --filter label=dockerbats="$BATS_TEST_NAME");
+    containerId=$(docker container ls -a -q --filter label=dockerbats="${BATS_TEST_NAME}-${token}");
+    echo "Container id: ${containerId}";
     run docker logs "${containerId}";
+    echo "Output of logs: ${output}";
     [[ $output =~ "multiple-1-${token}" ]]
     [[ $output =~ "multiple-100-${token}" ]]
     [ "$status" -eq 0 ]
@@ -62,6 +67,7 @@ teardown() {
 @test "plugin require token" {
     run docker run -d \
 	--log-driver 'h1cr.io/h1-docker-logging-plugin:latest' \
+  --log-opt labels=dockerbats \
     --label dockerbats="$BATS_TEST_NAME" \
 	alpine id;
     [[ $output =~ "Missing 'journal-fqdn option of log driver." ]]
diff --git a/tests/protocol.js b/tests/protocol.js
@@ -29,7 +29,7 @@ test('parse journal stream', async t => new Promise((resolve, reject) => {
             count+=1;
         })
         .on('error', reject)
-        .on('end', () => {
+        .on('finish', () => {
             t.true(count == 2);
             return resolve();
         });
diff --git a/transform.js b/transform.js
@@ -5,7 +5,7 @@ const { messages } = require('./parse');
 
 class ParseDockerStream extends stream.Transform {
     constructor(options = {}) {
-        super({objectMode: true, ...options});
+        super({ objectMode: true, ...options });
         this._buf = Buffer.alloc(0);
         this.bytes = 0;
     }
@@ -42,16 +42,18 @@ class ParseDockerStream extends stream.Transform {
 
 class ParseJournalStream extends stream.Transform {
     constructor(options = {}) {
-        super({objectMode: true, ...options});
+        super({ objectMode: true, ...options });
         this.chunks = 0;
     }
 
     _transform(chunk, encoding, callback) {
         this.chunks += 1;
         try {
-            const message = JSON.parse(chunk);
-            message.line = Buffer.from(message.message);
-            delete message.message;
+            const message = JSON.parse(chunk.toString('utf-8'));
+            if (message.message) {
+                message.line = Buffer.from(message.message);
+                delete message.message;
+            }
             return callback(null, message);
         } catch (err) {
             return callback(err);
@@ -67,7 +69,7 @@ class FilterJournalDockerStream extends stream.Transform {
 
     _transform(chunk, encoding, callback) {
         this.chunks += 1;
-        if (!['source', 'time', 'line'].every(x => Object.keys(chunk).includes(x))) {
+        if (!['source', 'time_nano', 'line'].every(x => Object.keys(chunk).includes(x))) {
             return callback(null);
         }
         return callback(null, chunk);
diff --git a/utils.js b/utils.js
@@ -24,7 +24,6 @@ const extract_tag = (config, info) => {
 
     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];
         });
