SAN-5848 - Stream logs from s3 to the frontend

configs/.env.test

@@ -46,6 +46,7 @@ TOKEN_EXPIRES=30000
 USER_CONTENT_DOMAIN=runnableapp.com
 VALID_REDIR_TLDS=runnableapp.com,domains.net
 WEB_URL=https://web.runnable.dev
+S3_LOG_BUCKET=testing.container-logs
 RABBITMQ_PASSWORD=guest
 RABBITMQ_PORT=5672
 RABBITMQ_USERNAME=guest

lib/socket/common-s3.js

@@ -0,0 +1,38 @@
+const aws = require('aws-sdk')
+const logger = require('logger')
+const s3 = new aws.S3()
+const Promise = require('bluebird')
+
+const pipeLogsToClient = (targetStream, containerId) => {
+ const log = logger.child({
+ containerId,
+ method: 'pipeLogsToClient'
+ })
+ log.trace('called')
+
+ const s3Object = s3.getObject({
+ Bucket: process.env.S3_LOG_BUCKET,
+ Key: containerId
+ })
+
+ return Promise.fromCallback((cb) => {
+ s3Object
+ .createReadStream()
+ .on('data', (data) => {
+ targetStream.write(data.toString())
+ })
+ .on('error', (error) => {
+ log.trace({error}, 'Error while fetching logs from s3')
+ cb(error)
+ })
+ .on('end', () => {
+ log.trace('Finished serving logs from s3')
+ targetStream.end()
+ cb(null)
+ })
+ })
+}
+
+module.exports = {
+ pipeLogsToClient
+}

lib/socket/log-stream.js

@@ -8,6 +8,7 @@ const keypather = require('keypather')()
 const monitorDog = require('monitor-dog')
 
 const commonStream = require('./common-stream')
+const commonS3 = require('./common-s3')
 const Instance = require('models/mongo/instance')
 const logger = require('logger')
 const PermissionService = require('models/services/permission-service')
@@ -91,7 +92,22 @@ function setupLogs (socket, id, data, instance, tags) {
  }
  }
  rabbitMQ.publishLogStreamConnected(eventData)
- return commonStream.pipeLogsToClient(destLogStream, baseDataName, tags, data.containerId, { tailLimit })
+ let streamPromise
+ if (keypather.get(instance, 'container.inspect.State.Running')) {
+ streamPromise = commonStream.pipeLogsToClient(destLogStream, baseDataName, tags, data.containerId, { tailLimit })
+ } else {
+ streamPromise = commonS3.pipeLogsToClient(destLogStream, data.containerId)
+ .catch((err) => {
+ log.error({error: err}, 'Error piping logs from s3 to client')
+ if (err.code === 'NoSuchKey') {
+ // fallback on no file exists to go against docker directly
+ return commonStream.pipeLogsToClient(destLogStream, baseDataName, tags, data.containerId, { tailLimit })
+ }
+ throw err
+ })
+ }
+
+ return streamPromise
  .catch(function (err) {
  tags.result = 'failure'
  monitorDog.increment(baseDataName + '.err.getting_logs', tags)

lib/workers/container.logs.store.js

@@ -31,7 +31,7 @@ module.exports.task = (job) => {
  return docker.getLogsAsync(job.containerId)
  .then(function (dockerLogStream) {
  log.trace({
- bucket: `${process.env.NODE_ENV}.container-logs`,
+ bucket: process.env.S3_LOG_BUCKET,
  key: job.containerId
  }, 'Began piping logs')
  const cleanedStream = new Through2({}, commonStream.buff2StringTransform)
@@ -43,7 +43,7 @@ module.exports.task = (job) => {
  })
  // Send stream to s3
  return s3.upload({
- Bucket: `${process.env.NODE_ENV}.container-logs`,
+ Bucket: process.env.S3_LOG_BUCKET,
  Key: job.containerId,
  Body: cleanedStream
  })

unit/socket/log-stream.js

@@ -1,26 +1,27 @@
 'use strict'
 
-var Lab = require('lab')
-var lab = exports.lab = Lab.script()
-var describe = lab.describe
-var it = lab.it
-var beforeEach = lab.beforeEach
-// var after = lab.after
-var afterEach = lab.afterEach
-var Code = require('code')
-var expect = Code.expect
+const Lab = require('lab')
+const lab = exports.lab = Lab.script()
+const describe = lab.describe
+const it = lab.it
+const beforeEach = lab.beforeEach
+// const after = lab.after
+const afterEach = lab.afterEach
+const Code = require('code')
+const expect = Code.expect
 
-var sinon = require('sinon')
-var EventEmitter = require('events').EventEmitter
-var util = require('util')
+const sinon = require('sinon')
+const EventEmitter = require('events').EventEmitter
+const util = require('util')
 
-var logStream = require('socket/log-stream')
-var Instance = require('models/mongo/instance')
+const logStream = require('socket/log-stream')
+const Instance = require('models/mongo/instance')
 
-var Promise = require('bluebird')
+const Promise = require('bluebird')
 require('sinon-as-promised')(Promise)
-var commonStream = require('socket/common-stream')
-var PermissionService = require('models/services/permission-service')
+const commonStream = require('socket/common-stream')
+const commonS3 = require('socket/common-s3')
+const PermissionService = require('models/services/permission-service')
 
 function ClientStream () {
  EventEmitter.call(this)
@@ -80,7 +81,12 @@ describe('log stream: ' + moduleName, function () {
  github: 123
  },
  container: {
- dockerContainer: ctx.data.containerId
+ dockerContainer: ctx.data.containerId,
+ inspect: {
+ State: {
+ Running: true
+ }
+ }
  }
  }
  writeStream = new ClientStream()
@@ -220,6 +226,62 @@ describe('log stream: ' + moduleName, function () {
  })
  .catch(done)
  })
+
+ describe('When container is stopped', () => {
+ beforeEach((done) => {
+ ctx.instance.container.inspect.State.Running = false
+ sinon.stub(commonS3, 'pipeLogsToClient').resolves({})
+ done()
+ })
+ afterEach((done) => {
+ commonS3.pipeLogsToClient.restore()
+ done()
+ })
+ describe('file exists in s3', () => {
+ it('should stream logs from s3 file', (done) => {
+ logStream.logStreamHandler(ctx.socket, ctx.id, ctx.data)
+ .then(() => {
+ sinon.assert.notCalled(commonStream.pipeLogsToClient)
+ sinon.assert.calledOnce(commonS3.pipeLogsToClient)
+ sinon.assert.calledWith(
+ commonS3.pipeLogsToClient,
+ substream,
+ ctx.data.containerId
+ )
+ })
+ .asCallback(done)
+ })
+ })
+ describe('and file does not exist in s3', () => {
+ beforeEach((done) => {
+ commonS3.pipeLogsToClient.rejects({code: 'NoSuchKey'})
+ done()
+ })
+ it('should stream logs from docker file', (done) => {
+ logStream.logStreamHandler(ctx.socket, ctx.id, ctx.data)
+ .then(() => {
+ sinon.assert.calledOnce(commonS3.pipeLogsToClient)
+ sinon.assert.calledWith(
+ commonS3.pipeLogsToClient,
+ substream,
+ ctx.data.containerId
+ )
+ sinon.assert.calledOnce(commonStream.pipeLogsToClient)
+ sinon.assert.calledWith(
+ commonStream.pipeLogsToClient,
+ substream,
+ 'api.socket.log',
+ sinon.match.object,
+ ctx.data.containerId,
+ {
+ tailLimit: process.env.DOCKER_LOG_TAIL_LIMIT
+ }
+ )
+ })
+ .asCallback(done)
+ })
+ })
+ })
  })
  })
 })